React chat interrupts preview

// eslint-disable-next-line @nx/enforce-module-boundaries -- The isolated consumer copies this internal example layout.
import { ExampleFooter, ExampleLayout } from '../../../../shared/react-example';
import { useEffect, useState } from 'react';
import { useAgent } from '@threadplane/react';
import {
  ChatInput,
  ApprovalCard,
  MessageList,
  Reasoning,
  Citations,
} from '@threadplane/react/chat';
import { Markdown } from '@threadplane/react/markdown';
import type { MessageRow } from '@threadplane/content/messages';
import {
  createConnectedApplication,
  type interruptsConnection,
} from './connection';
 
/** Stable row renderer; the application owns Markdown snapshot lifetimes. */
export function renderMessage(row: MessageRow) {
  if (row.role === 'tool') return null;
  if (
    row.role === 'assistant' &&
    !row.message.content &&
    !row.reasoning &&
    !row.message.citations?.length &&
    row.message.toolCallIds?.length
  )
    return null;
  const label =
    row.role === 'user'
      ? 'You'
      : row.role === 'assistant'
      ? 'Assistant'
      : 'System';
  return (
    <article
      aria-label={label}
      className={`tp-chat-message tp-chat-message--${row.role}`}
    >
      <p className="tp-chat-message__role">{label}</p>
      {row.role === 'assistant' && row.reasoning && (
        <Reasoning snapshot={row.reasoning} />
      )}
      <Markdown snapshot={row.markdown} />
      {row.role === 'assistant' && row.message.citations && (
        <Citations citations={row.message.citations} />
      )}
    </article>
  );
}
export function InterruptsDemo({
  connection,
  onReady,
}: {
  readonly connection: ReturnType<typeof interruptsConnection>;
  readonly onReady: () => void;
}) {
  const [application] = useState(() => createConnectedApplication(connection));
  const [draft, setDraft] = useState('');
  const snapshot = useAgent(application),
    decision = snapshot.decision;
  useEffect(() => {
    onReady();
  }, [onReady]);
  useEffect(
    () => () => {
      void application.dispose();
    },
    [application]
  );
  const status = snapshot.busy
    ? snapshot.activity === 'creating'
      ? 'Creating conversation…'
      : snapshot.activity === 'confirming'
      ? 'Confirming saved conversation…'
      : snapshot.activity === 'replacing'
      ? 'Releasing conversation…'
      : 'Receiving response…'
    : decision
    ? 'Review this saved booking request.'
    : snapshot.outcome === 'success'
    ? 'Response saved.'
    : !snapshot.canSubmit
    ? 'Start a new conversation to continue.'
    : 'Ready.';
  const decide = (value: 'confirm' | 'cancel') => {
    const current = application.getSnapshot();
    if (!decision || current.busy || current.decision !== decision) return;
    void application.decide(decision, value);
  };
  return (
    <ExampleLayout className="interrupts-demo">
      <header>
        <div>
          <h1>Chat interrupts</h1>
        </div>
        <button
          type="button"
          disabled={snapshot.busy}
          onClick={() => {
            if (application.getSnapshot().busy) return;
            setDraft('');
            void application.newConversation();
          }}
        >
          New conversation
        </button>
      </header>
      <p>
        Explore flight approvals with demo data. Confirm or cancel a saved
        booking request; no real ticket is purchased.
      </p>
      <div className="seed-prompts" aria-label="Example prompts">
        {['UA123', 'AA404'].map((flight) => (
          <button
            key={flight}
            type="button"
            disabled={!snapshot.canSubmit || snapshot.busy}
            onClick={() => {
              if (
                application.getSnapshot().canSubmit &&
                !application.getSnapshot().busy
              )
                setDraft(`Book flight ${flight}.`);
            }}
          >
            Book {flight}
          </button>
        ))}
      </div>
      <MessageList
        rows={snapshot.rows}
        renderMessage={renderMessage}
        label="Conversation"
      >
        {decision && (
          <ApprovalCard
            title="Approve demo booking"
            disabled={snapshot.busy}
            actions={[
              {
                id: 'confirm',
                label: 'Confirm',
                onSelect: () => decide('confirm'),
              },
              {
                id: 'cancel',
                label: 'Cancel',
                onSelect: () => decide('cancel'),
              },
            ]}
          >
            <p>{decision.approval.summary}</p>
            <dl className="flight-details">
              <div>
                <dt>Flight</dt>
                <dd>{decision.approval.flight.flight_number}</dd>
              </div>
              <div>
                <dt>Airline</dt>
                <dd>{decision.approval.flight.airline}</dd>
              </div>
              <div>
                <dt>From</dt>
                <dd>{decision.approval.flight.from}</dd>
              </div>
              <div>
                <dt>To</dt>
                <dd>{decision.approval.flight.to}</dd>
              </div>
              <div>
                <dt>Departure</dt>
                <dd>{decision.approval.flight.depart_local}</dd>
              </div>
              <div>
                <dt>Aircraft</dt>
                <dd>{decision.approval.flight.aircraft}</dd>
              </div>
            </dl>
          </ApprovalCard>
        )}
      </MessageList>
      <ExampleFooter>
        {snapshot.error && <p role="alert">{snapshot.error}</p>}
        <ChatInput
          key={snapshot.viewGeneration}
          label="Message"
          value={draft}
          onValueChange={setDraft}
          placeholder="Ask about a flight or request a booking…"
          hint="Enter to send. Shift+Enter adds a new line."
          busy={snapshot.busy}
          disabled={!snapshot.canSubmit && !snapshot.busy}
          onStop={
            snapshot.busy && snapshot.activity !== 'replacing'
              ? () => void application.stop()
              : undefined
          }
          onSubmit={(text) => {
            if (!application.getSnapshot().canSubmit) return false;
            void application.submit(text);
            return true;
          }}
        />
        <p role="status">{status}</p>
      </ExampleFooter>
    </ExampleLayout>
  );
}

import type { AgentSnapshot, Message, ToolCall } from '@threadplane/core';
 
export interface InterruptsState extends AgentSnapshot {
  readonly interrupts: readonly unknown[];
  readonly subgraphs: readonly unknown[];
  readonly history:
    | readonly {
        readonly checkpoint: unknown;
        readonly next: readonly string[];
        readonly parent_checkpoint?: unknown;
        readonly created_at?: unknown;
      }[]
    | undefined;
}
export interface Approval {
  readonly summary: string;
  readonly flight: Readonly<{
    flight_number: string;
    airline: string;
    from: string;
    to: string;
    depart_local: string;
    aircraft: string;
  }>;
}
export interface Authority {
  readonly threadId: string;
  readonly checkpoint: string;
  readonly signature: string;
  readonly batch: readonly unknown[];
  readonly messages: readonly Message[];
  readonly tools: readonly ToolCall[];
}
export interface PauseAuthority extends Authority {
  readonly approval: Approval;
}
/** Capture own data without evaluating backend-supplied accessors. */
function copy(value: unknown, parents = new Set<object>()): unknown {
  if (
    value === null ||
    value === undefined ||
    typeof value === 'string' ||
    typeof value === 'boolean'
  )
    return value;
  if (typeof value === 'number' && Number.isFinite(value)) return value;
  if (
    typeof value !== 'object' ||
    parents.has(value) ||
    Object.getOwnPropertySymbols(value).length
  )
    throw new Error('Unsupported data');
  const array = Array.isArray(value),
    prototype = Object.getPrototypeOf(value);
  if (array && prototype !== Array.prototype)
    throw new Error('Unsupported data');
  if (!array && prototype !== Object.prototype && prototype !== null)
    throw new Error('Unsupported data');
  parents.add(value);
  const descriptors = Object.getOwnPropertyDescriptors(value);
  const result: Record<string, unknown> | unknown[] = array
    ? []
    : Object.create(null);
  const keys = array
    ? Array.from({ length: value.length }, (_, index) => String(index))
    : Object.keys(descriptors).sort();
  if (
    array &&
    Object.keys(descriptors).some(
      (name) => name !== 'length' && !keys.includes(name)
    )
  )
    throw new Error('Unsupported data');
  for (const key of keys) {
    const descriptor = descriptors[key];
    if (!descriptor || !('value' in descriptor))
      throw new Error('Unsupported data');
    Object.defineProperty(result, key, {
      value: copy(descriptor.value, parents),
      enumerable: true,
    });
  }
  parents.delete(value);
  return Object.freeze(result);
}
const record = (value: unknown): value is Record<string, unknown> =>
  !!value && typeof value === 'object' && !Array.isArray(value);
const text = (value: unknown): value is string =>
  typeof value === 'string' && !!value.trim();
const key = (value: unknown) => JSON.stringify(copy(value));
export function messageKey(message: Message): string {
  return key({
    id: message.id,
    role: message.role,
    content: message.content,
    reasoning: message.reasoning,
    citations: message.citations,
    name: message.name,
    toolCallId: message.toolCallId,
    toolCallIds: message.toolCallIds,
  });
}
export function toolsKey(tools: readonly ToolCall[]): string {
  return key(tools);
}
 
function validArgs(tool: ToolCall) {
  const args = tool.args;
  if (!record(args)) return false;
  switch (tool.name) {
    case 'book_flight':
    case 'lookup_flight':
      return text(args['flight_number']);
    case 'get_airport_info':
      return text(args['airport_code']);
    case 'find_routes':
      return (
        text(args['from_code']) &&
        text(args['to_code']) &&
        (args['date_offset_days'] === undefined ||
          Number.isInteger(args['date_offset_days']))
      );
    default:
      return false;
  }
}
 
function capture(
  state: InterruptsState,
  threadId: string,
  paused: boolean
): Authority | null {
  if (
    state.status !== 'idle' ||
    state.error ||
    state.subgraphs.length ||
    !state.history?.length
  )
    return null;
  const history = copy(state.history[0]) as NonNullable<
    InterruptsState['history']
  >[number];
  const checkpoint = history.checkpoint;
  if (
    !record(checkpoint) ||
    checkpoint['thread_id'] !== threadId ||
    checkpoint['checkpoint_ns'] !== '' ||
    !text(checkpoint['checkpoint_id']) ||
    !Array.isArray(history.next) ||
    key(history.next) !== key(paused ? ['tools'] : [])
  )
    return null;
  const messages = copy(state.messages) as readonly Message[];
  const tools = copy(state.toolCalls) as readonly ToolCall[];
  if (
    !messages.length ||
    messages[0].role !== 'user' ||
    new Set(messages.map((m) => m.id)).size !== messages.length ||
    new Set(tools.map((tool) => tool.id)).size !== tools.length
  )
    return null;
  const catalog = new Map(tools.map((tool) => [tool.id, tool]));
  if (
    tools.some(
      (tool) =>
        !text(tool.id) ||
        !validArgs(tool) ||
        (tool.status !== 'pending' && tool.status !== 'complete')
    )
  )
    return null;
  const calls = new Set<string>(),
    results = new Set<string>();
  let human = -1,
    pendingOwner = -1;
  for (const [index, message] of messages.entries()) {
    if (
      !text(message.id) ||
      typeof message.content !== 'string' ||
      message.delivery.phase !== 'complete' ||
      message.delivery.generation !== message.id
    )
      return null;
    const pending =
      message.role === 'assistant' &&
      message.toolCallIds?.some((id) => catalog.get(id)?.status === 'pending');
    if (message.delivery.outcome !== (paused && pending ? 'paused' : 'success'))
      return null;
    if (message.role === 'user') {
      human = index;
      if (message.toolCallId !== undefined || message.toolCallIds?.length)
        return null;
    } else if (message.role === 'assistant') {
      if (message.toolCallId !== undefined) return null;
      for (const id of message.toolCallIds ?? []) {
        if (!catalog.has(id) || calls.has(id)) return null;
        calls.add(id);
        if (catalog.get(id)?.status === 'pending') pendingOwner = index;
      }
    } else if (message.role === 'tool') {
      const id = message.toolCallId,
        tool = id ? catalog.get(id) : undefined;
      if (
        !id ||
        !tool ||
        !calls.has(id) ||
        results.has(id) ||
        tool.status !== 'complete' ||
        tool.name !== message.name ||
        key(tool.result) !== key(message.content) ||
        message.toolCallIds?.length
      )
        return null;
      results.add(id);
    } else return null;
  }
  if (
    calls.size !== tools.length ||
    tools.some((tool) => tool.status === 'complete' && !results.has(tool.id))
  )
    return null;
  const pending = tools.filter((tool) => tool.status === 'pending');
  if (
    paused
      ? pending.length !== 1 ||
        pending[0].name !== 'book_flight' ||
        pendingOwner <= human ||
        pendingOwner !== messages.length - 1
      : pending.length ||
        state.interrupts.length ||
        messages.at(-1)?.role !== 'assistant' ||
        !text(messages.at(-1)?.content)
  )
    return null;
  return Object.freeze({
    threadId,
    checkpoint: key(checkpoint),
    signature: key({ history, messages, tools, interrupts: state.interrupts }),
    batch: state.interrupts,
    messages,
    tools,
  });
}
 
export function capturePause(
  state: InterruptsState,
  threadId: string
): PauseAuthority | null {
  try {
    if (state.interrupts.length !== 1) return null;
    const interrupt = copy(state.interrupts[0]);
    if (
      !record(interrupt) ||
      !text(interrupt['id']) ||
      !record(interrupt['value'])
    )
      return null;
    const payload = interrupt['value'],
      flight = payload['flight'];
    if (
      payload['type'] !== 'approval_request' ||
      !text(payload['summary']) ||
      !record(flight) ||
      ![
        'flight_number',
        'airline',
        'from',
        'to',
        'depart_local',
        'aircraft',
      ].every((field) => text(flight[field]))
    )
      return null;
    const authority = capture(state, threadId, true);
    if (!authority) return null;
    const pending = authority.tools.find((tool) => tool.status === 'pending');
    if (
      !pending ||
      !record(pending.args) ||
      (pending.args['flight_number'] as string).trim().toUpperCase() !==
        flight['flight_number']
    )
      return null;
    const approval: Approval = Object.freeze({
      summary: payload['summary'],
      flight: Object.freeze({
        flight_number: flight['flight_number'] as string,
        airline: flight['airline'] as string,
        from: flight['from'] as string,
        to: flight['to'] as string,
        depart_local: flight['depart_local'] as string,
        aircraft: flight['aircraft'] as string,
      }),
    });
    return Object.freeze({ ...authority, approval });
  } catch {
    return null;
  }
}
export function captureTerminal(
  state: InterruptsState,
  threadId: string
): Authority | null {
  try {
    return capture(state, threadId, false);
  } catch {
    return null;
  }
}
export function sameObservedAuthority(
  captured: PauseAuthority,
  state: InterruptsState
): boolean {
  const current = capturePause(state, captured.threadId);
  return (
    !!current &&
    current.batch === captured.batch &&
    current.checkpoint === captured.checkpoint &&
    current.signature === captured.signature
  );
}

import type {
  AgentSession,
  CompleteOutcome,
  Message,
  ToolCall,
} from '@threadplane/core';
import {
  createMessageContent,
  type MessageRow,
} from '@threadplane/content/messages';
import {
  capturePause,
  captureTerminal,
  sameObservedAuthority,
  messageKey,
  toolsKey,
  type InterruptsState,
  type PauseAuthority,
} from './authority';
 
export interface InterruptsSession extends Omit<AgentSession, 'getSnapshot'> {
  getSnapshot(): InterruptsState;
  load(options?: { readonly signal?: AbortSignal }): Promise<void>;
  resume(
    value: 'confirm' | 'cancel',
    options?: { readonly signal?: AbortSignal }
  ): Promise<CompleteOutcome>;
}
export interface InterruptsSnapshot {
  readonly threadId: string | null;
  readonly rows: readonly MessageRow[];
  readonly activity:
    | 'idle'
    | 'creating'
    | 'running'
    | 'confirming'
    | 'replacing';
  readonly busy: boolean;
  readonly canSubmit: boolean;
  readonly outcome: CompleteOutcome | null;
  readonly decision: PauseAuthority | null;
  readonly error: string | null;
  readonly viewGeneration: number;
}
const completed = (message: Message) => message.delivery.phase === 'complete';
const same = (left: Message, right: Message | undefined) =>
  !!right && messageKey(left) === messageKey(right);
 
/** Owns one conversation. Only an observed, saved pause grants a fixed decision. */
export function createInterruptsApplication(options: {
  readonly createThread: (signal: AbortSignal) => Promise<string>;
  readonly sessionFactory: (id: string) => InterruptsSession;
}) {
  type Turn = {
    text?: string;
    human?: Message;
    generation?: string;
    base: readonly Message[];
    positions: string[];
    completed: Map<string, string>;
    tools: Map<string, ToolCall>;
    expected?: { messages: readonly Message[]; tools: readonly ToolCall[] };
  };
  type Owner = {
    id: string;
    session: InterruptsSession;
    content: ReturnType<typeof createMessageContent>;
    release?: () => void;
    closed: boolean;
    unsafe: boolean;
    confirmed: readonly Message[];
    confirmedTools: readonly ToolCall[];
    generations: Set<string>;
    turn?: Turn;
  };
  let owner: Owner | undefined, operation: AbortController | undefined;
  let disposed = false,
    disposal: Promise<void> | undefined;
  const listeners = new Set<() => void>(),
    cleanups = new Set<Promise<void>>();
  let snapshot: InterruptsSnapshot = Object.freeze({
    threadId: null,
    rows: Object.freeze([]),
    activity: 'idle',
    busy: false,
    canSubmit: true,
    outcome: null,
    decision: null,
    error: null,
    viewGeneration: 0,
  });
  function publish(update: Partial<InterruptsSnapshot>) {
    if (disposed) return;
    snapshot = Object.freeze({ ...snapshot, ...update });
    for (const notify of listeners) notify();
  }
  function bounded(work: () => void | Promise<void>) {
    let timeout: ReturnType<typeof setTimeout>;
    return Promise.race([
      Promise.resolve()
        .then(work)
        .catch(() => undefined),
      new Promise<void>((resolve) => {
        timeout = setTimeout(resolve, 2000);
      }),
    ]).finally(() => clearTimeout(timeout));
  }
  function close(selected: Owner) {
    if (selected.closed) return Promise.resolve();
    selected.closed = true;
    selected.release?.();
    selected.content.dispose();
    const cleanup = bounded(() => selected.session.dispose());
    cleanups.add(cleanup);
    void cleanup.then(() => cleanups.delete(cleanup));
    return cleanup;
  }
  function detach() {
    const previous = owner;
    owner = undefined;
    return previous ? close(previous) : Promise.resolve();
  }
  function blocked() {
    publish({
      canSubmit: false,
      decision: null,
      outcome: 'paused',
      error:
        'This conversation could not be confirmed. Start a new conversation to continue.',
    });
  }
  function observe(selected: Owner, state: InterruptsState) {
    if (
      !state ||
      !Array.isArray(state.messages) ||
      !Array.isArray(state.toolCalls) ||
      !Array.isArray(state.interrupts) ||
      !Array.isArray(state.subgraphs) ||
      state.error ||
      state.status === 'error' ||
      state.subgraphs.length ||
      new Set(state.messages.map((m) => m.id)).size !== state.messages.length ||
      new Set(state.toolCalls.map((t) => t.id)).size !== state.toolCalls.length
    )
      throw new Error('Unconfirmed state');
    for (const tool of selected.confirmedTools) {
      const current = state.toolCalls.find(
        (candidate) => candidate.id === tool.id
      );
      if (
        !current ||
        current.name !== tool.name ||
        toolsKey([{ ...tool, args: current.args }]) !== toolsKey([tool]) ||
        (tool.status === 'complete' && toolsKey([current]) !== toolsKey([tool]))
      )
        throw new Error('Changed saved tool');
    }
    const turn = selected.turn,
      prefix = turn?.base ?? selected.confirmed;
    for (const tool of state.toolCalls) {
      if (
        ![
          'book_flight',
          'lookup_flight',
          'get_airport_info',
          'find_routes',
        ].includes(tool.name) ||
        tool.status === 'error'
      )
        throw new Error('Unsupported tool');
      const previous = turn?.tools.get(tool.id);
      if (
        previous &&
        (toolsKey([{ ...previous, name: tool.name, args: tool.args }]) !==
          toolsKey([previous]) ||
          (previous.status === 'complete' &&
            toolsKey([previous]) !== toolsKey([tool])))
      )
        throw new Error('Changed tool identity');
      turn?.tools.set(tool.id, tool);
    }
    if (
      turn &&
      [...turn.tools.keys()].some(
        (id) => !state.toolCalls.some((tool) => tool.id === id)
      )
    )
      throw new Error('Missing observed tool');
    if (
      state.messages.length < prefix.length ||
      prefix.some((message, index) => !same(message, state.messages[index]))
    )
      throw new Error('Changed prefix');
    if (snapshot.decision && !sameObservedAuthority(snapshot.decision, state))
      throw new Error('Changed decision');
    if (!turn) {
      if (state.messages.length !== prefix.length)
        throw new Error('Unowned turn');
      return;
    }
    if (turn.expected) {
      if (
        state.messages.length !== turn.expected.messages.length ||
        turn.expected.messages.some(
          (message, index) => !same(message, state.messages[index])
        ) ||
        toolsKey(state.toolCalls) !== toolsKey(turn.expected.tools)
      )
        throw new Error('Changed saved evidence');
      return;
    }
    if (
      state.messages.length < turn.positions.length ||
      turn.positions.some((id, index) => state.messages[index].id !== id)
    )
      throw new Error('Changed message identity');
    turn.positions = state.messages.map((message) => message.id);
    for (const message of state.messages) {
      if (
        completed(message) &&
        message.delivery.outcome !== 'success' &&
        message.delivery.outcome !== 'paused'
      )
        throw new Error('Failed message');
      const identity = messageKey(message),
        previous = turn.completed.get(message.id);
      if (previous !== undefined && previous !== identity)
        throw new Error('Changed completed message');
      if (completed(message)) turn.completed.set(message.id, identity);
    }
    const suffix = state.messages.slice(prefix.length);
    if (suffix.length) {
      turn.generation ??= suffix[0].delivery.generation;
      if (
        !turn.generation ||
        suffix.some(
          (message) => message.delivery.generation !== turn.generation
        )
      )
        throw new Error('Foreign run generation');
    }
    if (turn.text !== undefined && suffix.length) {
      const human = suffix[0];
      if (
        human.role !== 'user' ||
        human.content !== turn.text ||
        !completed(human) ||
        human.delivery.outcome !== 'success' ||
        selected.generations.has(human.delivery.generation) ||
        (turn.human &&
          (!same(turn.human, human) ||
            turn.human.delivery.generation !== human.delivery.generation)) ||
        suffix.slice(1).some((message) => message.role === 'user')
      )
        throw new Error('Unowned human turn');
      turn.human ??= human;
    } else if (
      turn.text === undefined &&
      suffix.some((message) => message.role === 'user')
    )
      throw new Error('Unowned resumed turn');
  }
  function read(selected: Owner, valid: () => boolean) {
    if (!valid()) return;
    const state = selected.session.getSnapshot();
    if (!valid()) return;
    try {
      observe(selected, state);
    } catch {
      selected.unsafe = true;
    }
    if (!valid()) return;
    if (Array.isArray(state?.messages) && Array.isArray(state?.toolCalls)) {
      const rows = selected.content.project(state);
      if (!valid()) return;
      publish({
        rows,
        ...(selected.unsafe ? { canSubmit: false, decision: null } : {}),
      });
    }
    if (!valid()) return;
    if (selected.session.getSnapshot() !== state) selected.unsafe = true;
    if (!valid()) return;
    return state;
  }
  async function execute(
    input:
      | { text: string }
      | { decision: PauseAuthority; value: 'confirm' | 'cancel' }
  ) {
    if (disposed || snapshot.busy) return;
    const decision = 'decision' in input ? input.decision : undefined;
    if (
      'text' in input
        ? !snapshot.canSubmit || !input.text.trim()
        : !decision ||
          decision !== snapshot.decision ||
          !owner ||
          (input.value !== 'confirm' && input.value !== 'cancel')
    )
      return;
    const admitted = new AbortController();
    operation = admitted;
    const current = () =>
      !disposed && operation === admitted && !admitted.signal.aborted;
    publish({
      busy: true,
      canSubmit: false,
      decision: null,
      activity: owner ? 'running' : 'creating',
      outcome: null,
      error: null,
    });
    try {
      if (!current()) return;
      if (!owner) {
        const id = await options.createThread(admitted.signal);
        if (!current()) return;
        if (!id || typeof id !== 'string')
          throw new Error('Unconfirmed conversation');
        const session = options.sessionFactory(id);
        const selected: Owner = {
          id,
          session,
          content: createMessageContent(),
          closed: false,
          unsafe: false,
          confirmed: [],
          confirmedTools: [],
          generations: new Set(),
        };
        if (!current()) {
          void close(selected);
          return;
        }
        owner = selected;
        const release = session.subscribe(() =>
          read(
            selected,
            () =>
              !disposed &&
              owner === selected &&
              !selected.closed &&
              !operation?.signal.aborted
          )
        );
        selected.release = release;
        if (selected.closed) release();
        if (!current()) return;
        publish({ threadId: id, activity: 'running' });
      }
      const selected = owner,
        valid = () => current() && owner === selected && !selected.closed;
      const before = read(selected, valid);
      if (!valid()) return;
      const admittedAuthority =
        !decision || (!!before && sameObservedAuthority(decision, before));
      if (!valid()) return;
      if (!before || selected.unsafe || !admittedAuthority) {
        blocked();
        return;
      }
      selected.turn = {
        ...('text' in input ? { text: input.text } : {}),
        base: selected.confirmed,
        positions: selected.confirmed.map((message) => message.id),
        completed: new Map(
          selected.confirmed.map((message) => [message.id, messageKey(message)])
        ),
        tools: new Map(selected.confirmedTools.map((tool) => [tool.id, tool])),
      };
      // Admission and row subscribers can replace observations; recheck directly before resume.
      if (!valid()) return;
      const dispatch = selected.session.getSnapshot();
      if (!valid()) return;
      const dispatchAuthority =
        !decision || sameObservedAuthority(decision, dispatch);
      if (!valid()) return;
      const currentDispatch = selected.session.getSnapshot();
      if (!valid()) return;
      if (
        !dispatchAuthority ||
        selected.unsafe ||
        currentDispatch !== dispatch
      ) {
        selected.unsafe = true;
        blocked();
        return;
      }
      const outcome = await ('text' in input
        ? selected.session.submit(input.text, { signal: admitted.signal })
        : selected.session.resume(input.value, { signal: admitted.signal }));
      if (!valid()) return;
      const live = read(selected, valid),
        turn = selected.turn;
      if (!valid()) return;
      if (
        !live ||
        selected.unsafe ||
        (outcome !== 'success' && outcome !== 'paused') ||
        live.status !== 'idle' ||
        !turn ||
        ('text' in input && !turn.human) ||
        live.messages.length <= turn.base.length ||
        live.messages.some((message) => !completed(message))
      ) {
        blocked();
        return;
      }
      turn.expected = {
        messages: Object.freeze([...live.messages]),
        tools: Object.freeze([...live.toolCalls]),
      };
      publish({ activity: 'confirming' });
      if (!valid()) return;
      if (!read(selected, valid) || selected.unsafe) {
        if (valid()) blocked();
        return;
      }
      await selected.session.load({ signal: admitted.signal });
      if (!valid()) return;
      const saved = read(selected, valid);
      if (!valid()) return;
      const authority =
        saved &&
        (outcome === 'paused'
          ? capturePause(saved, selected.id)
          : captureTerminal(saved, selected.id));
      if (!valid()) return;
      const currentSaved = selected.session.getSnapshot();
      if (!valid()) return;
      if (!authority || selected.unsafe || currentSaved !== saved) {
        blocked();
        return;
      }
      selected.confirmed = authority.messages;
      selected.confirmedTools = authority.tools;
      if (turn.human) selected.generations.add(turn.human.delivery.generation);
      selected.turn = undefined;
      publish({
        outcome,
        canSubmit: outcome === 'success',
        decision: outcome === 'paused' ? (authority as PauseAuthority) : null,
      });
      if (!valid()) return;
      const latest = read(selected, valid);
      if (valid() && (!latest || selected.unsafe)) blocked();
    } catch {
      if (current())
        publish({
          canSubmit: false,
          decision: null,
          outcome: 'error',
          error:
            'The conversation could not be confirmed. Start a new conversation.',
        });
    } finally {
      if (current()) {
        operation = undefined;
        publish({ busy: false, activity: 'idle' });
      }
    }
  }
  async function stop() {
    if (disposed || !operation || snapshot.activity === 'replacing') return;
    operation.abort();
    const admitted = new AbortController();
    operation = admitted;
    const selected = owner,
      cleanup = detach();
    publish({
      busy: true,
      activity: 'replacing',
      decision: null,
      canSubmit: false,
      outcome: 'aborted',
      error: null,
    });
    await Promise.all([cleanup, bounded(() => selected?.session.stop())]);
    if (!disposed && operation === admitted) {
      operation = undefined;
      publish({ busy: false, activity: 'idle' });
    }
  }
  async function newConversation() {
    if (disposed || snapshot.busy) return;
    const admitted = new AbortController();
    operation = admitted;
    const cleanup = detach();
    publish({
      busy: true,
      canSubmit: false,
      decision: null,
      activity: 'replacing',
    });
    await cleanup;
    if (disposed || operation !== admitted || admitted.signal.aborted) return;
    operation = undefined;
    publish({
      threadId: null,
      rows: Object.freeze([]),
      busy: false,
      canSubmit: true,
      decision: null,
      activity: 'idle',
      outcome: null,
      error: null,
      viewGeneration: snapshot.viewGeneration + 1,
    });
  }
  function dispose(): Promise<void> {
    if (disposal) return disposal;
    disposed = true;
    operation?.abort();
    operation = undefined;
    void detach();
    listeners.clear();
    disposal = Promise.all([...cleanups]).then(() => undefined);
    return disposal;
  }
  return {
    getSnapshot: () => snapshot,
    subscribe(notify: () => void) {
      if (disposed) return () => undefined;
      listeners.add(notify);
      return () => {
        listeners.delete(notify);
      };
    },
    submit: (text: string) => execute({ text }),
    decide: (decision: PauseAuthority, value: 'confirm' | 'cancel') =>
      execute({ decision, value }),
    stop,
    newConversation,
    dispose,
  };
}

import { Client } from '@langchain/langgraph-sdk';
import { createSession } from '@threadplane/langgraph';
// eslint-disable-next-line @nx/enforce-module-boundaries -- The isolated build copies this neutral internal bridge; it is not a published package.
import type { RuntimeBridgeConfiguration } from '../../../../../libs/cockpit-runtime-bridge/src/index';
import {
  createInterruptsApplication,
  type InterruptsSession,
} from './application';
 
export function interruptsConnection(
  configuration: RuntimeBridgeConfiguration
): {
  apiUrl: string;
  headers: Record<string, string>;
} {
  if (
    configuration.status === 'error' ||
    (configuration.status === 'configured' &&
      configuration.target.kind === 'ag-ui')
  )
    throw new Error('Runtime configuration failed.');
  const target =
    configuration.status === 'configured' &&
    configuration.target.kind === 'langsmith'
      ? configuration.target
      : null;
  return {
    apiUrl: target?.apiUrl ?? new URL('/api', window.location.origin).href,
    headers: target ? { 'x-api-key': target.apiKey } : {},
  };
}
export function createConnectedApplication(
  connection: ReturnType<typeof interruptsConnection>
) {
  const client = new Client({
    apiUrl: connection.apiUrl,
    apiKey: null,
    defaultHeaders: connection.headers,
    callerOptions: { maxRetries: 0 },
  });
  return createInterruptsApplication({
    async createThread(signal) {
      const threadId = crypto.randomUUID();
      const created = await client.threads.create({
        threadId,
        ifExists: 'raise',
        signal,
      });
      if (created.thread_id !== threadId)
        throw new Error('Conversation creation was not confirmed.');
      return threadId;
    },
    sessionFactory(threadId) {
      const session = createSession({
        assistantId: 'c-interrupts',
        threadId,
        apiUrl: connection.apiUrl,
        clientOptions: { maxRetries: 0, defaultHeaders: connection.headers },
      });
      try {
        const state = session.getSnapshot();
        if (
          !state ||
          !Array.isArray(state.messages) ||
          !Array.isArray(state.toolCalls) ||
          !Array.isArray(state.interrupts) ||
          !Array.isArray(state.subgraphs) ||
          ![
            'getSnapshot',
            'subscribe',
            'submit',
            'load',
            'resume',
            'stop',
            'dispose',
          ].every(
            (method) =>
              typeof session[method as keyof typeof session] === 'function'
          )
        )
          throw new Error('Runtime session is unavailable.');
      } catch {
        try {
          void Promise.resolve(session.dispose()).catch(() => undefined);
        } catch {
          /* A rejected session cannot be installed. */
        }
        throw new Error('Runtime session is unavailable.');
      }
      return session as InterruptsSession;
    },
  });
}