React chat subagents 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 { useCallback, useEffect, useState } from 'react';
import { useAgent } from '@threadplane/react';
import {
  ChatInput,
  ToolObservation,
  MessageList,
  Reasoning,
  Citations,
} from '@threadplane/react/chat';
import { Markdown } from '@threadplane/react/markdown';
import type { MessageRow } from '@threadplane/content/messages';
import {
  createConnectedApplication,
  type subagentsConnection,
} from './connection';
import type { ToolCard } from './projection';
 
/** Stable row renderer; the application owns Markdown snapshot lifetimes. */
export function renderMessage(
  row: MessageRow,
  cards: readonly ToolCard[] = []
) {
  if (row.role === 'tool') return null;
  if (
    row.role === 'assistant' &&
    !row.message.content &&
    !row.reasoning &&
    !row.message.citations?.length &&
    !cards.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} />
      {cards.map((card) => (
        <div className="tool-observation" key={card.id}>
          <h3>{card.role[0].toUpperCase() + card.role.slice(1)} specialist</h3>
          <ToolObservation
            name={card.name}
            argumentsText={card.argumentsText}
            resultText={card.resultText}
            label={
              card.resultText === undefined
                ? 'Observed specialist request'
                : 'Observed specialist result'
            }
          />
        </div>
      ))}
      {row.role === 'assistant' && row.message.citations && (
        <Citations citations={row.message.citations} />
      )}
    </article>
  );
}
export function SubagentsDemo({
  connection,
  onReady,
}: {
  readonly connection: ReturnType<typeof subagentsConnection>;
  readonly onReady: () => void;
}) {
  const [application] = useState(() => createConnectedApplication(connection));
  const [draft, setDraft] = useState('');
  const snapshot = useAgent(application);
  const renderRow = useCallback(
    (row: MessageRow) => renderMessage(row, snapshot.observations.get(row.id)),
    [snapshot.observations]
  );
  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…'
    : snapshot.outcome === 'success'
    ? 'Response saved.'
    : !snapshot.canSubmit
    ? 'Start a new conversation to continue.'
    : 'Ready.';
  return (
    <ExampleLayout className="subagents-demo">
      <header>
        <div>
          <h1>Chat subagents</h1>
        </div>
        <button
          type="button"
          disabled={snapshot.busy}
          onClick={() => {
            if (application.getSnapshot().busy) return;
            setDraft('');
            void application.newConversation();
          }}
        >
          New conversation
        </button>
      </header>
      <p>
        Explore model-generated planning suggestions from research, booking, and
        itinerary specialists. These are suggestions only: no live availability
        or prices, and no actual booking or purchase.
      </p>
      <div className="seed-prompts" aria-label="Example prompts">
        {[
          {
            label: 'Plan LAX to JFK',
            text: 'Plan a trip from LAX to JFK with research, flight suggestions, and an itinerary.',
          },
        ].map((prompt) => (
          <button
            key={prompt.label}
            type="button"
            disabled={!snapshot.canSubmit || snapshot.busy}
            onClick={() => {
              if (
                application.getSnapshot().canSubmit &&
                !application.getSnapshot().busy
              )
                setDraft(prompt.text);
            }}
          >
            {prompt.label}
          </button>
        ))}
      </div>
      <MessageList
        rows={snapshot.rows}
        renderMessage={renderRow}
        label="Conversation"
      />
      {snapshot.childObservations.length > 0 && (
        <details className="child-observations">
          <summary>Child response observations</summary>
          <p>
            Local observations from this open conversation. These are not
            restored execution history and are not linked to individual task
            cards.
          </p>
          {snapshot.childObservations.map((child) => (
            <section
              key={child.key}
              aria-label={`${child.role} response observation`}
            >
              <h3>
                {child.role[0].toUpperCase() + child.role.slice(1)} response
              </h3>
              <p className="child-observation-status">{child.status}</p>
              <pre>{child.text}</pre>
            </section>
          ))}
        </details>
      )}
      <ExampleFooter>
        {snapshot.error && <p role="alert">{snapshot.error}</p>}
        <ChatInput
          key={snapshot.viewGeneration}
          label="Message"
          value={draft}
          onValueChange={setDraft}
          placeholder="Describe a trip you would like to plan…"
          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 { Message, ToolCall } from '@threadplane/core';
import { copyData, validToolArgs } from './authority';
 
export interface ToolCard {
  readonly id: string;
  readonly name: string;
  readonly role: string;
  readonly argumentsText: string;
  readonly resultText?: string;
}
 
/** Associate literal observations with their one requesting assistant, never a result row. */
export function projectToolObservations(
  observedMessages: readonly Message[],
  observedTools: readonly ToolCall[]
): ReadonlyMap<string, readonly ToolCard[]> {
  const cards = new Map<string, readonly ToolCard[]>();
  try {
    const messages = copyData(observedMessages) as readonly Message[];
    const tools = copyData(observedTools) as readonly ToolCall[];
    if (
      new Set(messages.map((message) => message.id)).size !== messages.length ||
      new Set(tools.map((tool) => tool.id)).size !== tools.length
    )
      return cards;
    for (const tool of tools) {
      if (
        !tool.id ||
        !validToolArgs(tool) ||
        !['pending', 'complete'].includes(tool.status)
      )
        continue;
      const owners = messages.flatMap((message, index) =>
        message.role === 'assistant'
          ? (message.toolCallIds ?? [])
              .filter((id) => id === tool.id)
              .map(() => ({ message, index }))
          : []
      );
      if (owners.length !== 1) continue;
      const owner = owners[0];
      const results = messages.flatMap((message, index) =>
        message.toolCallId === tool.id ? [{ message, index }] : []
      );
      let resultText: string | undefined;
      if (tool.status === 'complete') {
        if (results.length !== 1 || typeof tool.result !== 'string') continue;
        const result = results[0];
        if (
          result.index <= owner.index ||
          result.message.role !== 'tool' ||
          result.message.name !== tool.name ||
          result.message.content !== tool.result ||
          result.message.toolCallIds?.length ||
          messages
            .slice(owner.index + 1, result.index)
            .some((message) => message.role === 'user')
        )
          continue;
        resultText = tool.result;
      } else if (
        results.length ||
        ('result' in tool && tool.result !== undefined) ||
        messages
          .slice(owner.index + 1)
          .some((message) => message.role === 'user')
      )
        continue;
      const card = Object.freeze({
        id: tool.id,
        name: tool.name,
        role: (tool.args as { subagent_type: string }).subagent_type,
        argumentsText: JSON.stringify(tool.args, null, 2),
        ...(resultText === undefined ? {} : { resultText }),
      });
      cards.set(
        owner.message.id,
        Object.freeze([...(cards.get(owner.message.id) ?? []), card])
      );
    }
  } catch {
    return new Map();
  }
  return cards;
}

import type { Message } from '@threadplane/core';
import { copyData, messageKey } from './authority';
 
export type ChildStatus =
  | 'Receiving'
  | 'Observed response'
  | 'Stopped'
  | 'Incomplete';
export interface ChildRow {
  readonly key: string;
  readonly role: string;
  readonly text: string;
  readonly status: ChildStatus;
}
interface Entry {
  readonly namespace: readonly string[];
  readonly role?: string;
  readonly task?: string;
  readonly messages: readonly Message[];
}
interface Context {
  readonly generation: string;
  readonly rootIds: ReadonlySet<string>;
  readonly terminal: boolean;
}
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 roles = ['research', 'booking', 'itinerary'];
 
/** One physical turn's local observations; never associates a child with a root call. */
export class ChildObservations {
  private entries = new Map<string, Entry>();
  private generation?: string;
  private unsafe = false;
  private ended = false;
  private status: ChildStatus = 'Receiving';
 
  constructor(private readonly view: string, private readonly turn: string) {}
 
  observe(input: readonly unknown[], context: Context): boolean {
    if (this.unsafe || this.ended) return false;
    try {
      const children = copyData(input);
      if (
        !Array.isArray(children) ||
        !text(context.generation) ||
        (this.generation !== undefined &&
          this.generation !== context.generation)
      )
        throw new Error('Invalid child generation');
      const next = new Map<string, Entry>();
      for (const child of children) {
        if (
          !record(child) ||
          !Array.isArray(child['namespace']) ||
          !child['namespace'].length ||
          !child['namespace'].every(text) ||
          !Array.isArray(child['messages']) ||
          !Array.isArray(child['interrupts']) ||
          child['interrupts'].length ||
          child['error']
        )
          throw new Error('Invalid child observation');
        const namespace = child['namespace'] as string[];
        const key = JSON.stringify(namespace);
        if (next.has(key)) throw new Error('Duplicate child namespace');
        const prior = this.entries.get(key);
        const values = child['values'];
        if (
          values !== undefined &&
          (!record(values) ||
            Object.keys(values).some(
              (key) => !['subagent_type', 'task_description'].includes(key)
            ))
        )
          throw new Error('Invalid child values');
        const role = record(values) ? values['subagent_type'] : undefined;
        const task = record(values) ? values['task_description'] : undefined;
        if (
          (record(values) &&
            Object.hasOwn(values, 'subagent_type') &&
            (!text(role) || !roles.includes(role))) ||
          (record(values) &&
            Object.hasOwn(values, 'task_description') &&
            !text(task)) ||
          (prior?.role !== undefined && prior.role !== role) ||
          (prior?.task !== undefined && prior.task !== task)
        )
          throw new Error('Changed child identity');
        const messages = child['messages'] as Message[];
        const ids = new Set<string>();
        for (const [index, message] of messages.entries()) {
          if (
            !record(message) ||
            !text(message.id) ||
            ids.has(message.id) ||
            context.rootIds.has(message.id) ||
            message.role !== 'assistant' ||
            typeof message.content !== 'string' ||
            message.toolCallId !== undefined ||
            (message.toolCallIds !== undefined &&
              (!Array.isArray(message.toolCallIds) ||
                message.toolCallIds.length)) ||
            !record(message.delivery) ||
            message.delivery.generation !== context.generation ||
            !['streaming', 'complete'].includes(message.delivery.phase) ||
            (message.delivery.phase === 'complete'
              ? message.delivery.outcome !== 'success'
              : 'outcome' in message.delivery &&
                message.delivery.outcome !== undefined)
          )
            throw new Error('Invalid child message');
          ids.add(message.id);
          const previous = prior?.messages[index];
          if (
            previous &&
            (previous.id !== message.id ||
              (previous.delivery.phase === 'complete' &&
                (message.delivery.phase !== 'complete' ||
                  messageKey(previous) !== messageKey(message))))
          )
            throw new Error('Changed child message');
        }
        if (prior && messages.length < prior.messages.length)
          throw new Error('Missing child message');
        if (
          context.terminal &&
          (!text(role) ||
            !text(task) ||
            !messages.length ||
            messages.some((message) => message.delivery.phase !== 'complete'))
        )
          throw new Error('Incomplete child observation');
        next.set(
          key,
          Object.freeze({
            namespace,
            role: role as string | undefined,
            task: task as string | undefined,
            messages,
          })
        );
      }
      for (const key of this.entries.keys())
        if (!next.has(key)) throw new Error('Missing child namespace');
      this.entries = next;
      this.generation = context.generation;
      return true;
    } catch {
      this.unsafe = true;
      return false;
    }
  }
 
  finish(status: Exclude<ChildStatus, 'Receiving'>): boolean {
    if (this.ended) return this.status === status && !this.unsafe;
    if (
      status === 'Observed response' &&
      (this.unsafe ||
        [...this.entries.values()].some(
          (entry) =>
            !entry.role ||
            !entry.task ||
            !entry.messages.length ||
            entry.messages.some(
              (message) =>
                message.delivery.phase !== 'complete' ||
                message.delivery.outcome !== 'success'
            )
        ))
    ) {
      this.unsafe = true;
      return false;
    }
    this.status = status;
    this.ended = true;
    return !this.unsafe;
  }
 
  rows(): readonly ChildRow[] {
    return Object.freeze(
      [...this.entries.values()].flatMap((entry) => {
        if (!entry.role || !entry.task) return [];
        const content = entry.messages
          .map((message) => message.content)
          .filter(text)
          .join('\n\n');
        return content.trim()
          ? [
              Object.freeze({
                key: JSON.stringify([this.view, this.turn, entry.namespace]),
                role: entry.role,
                text: content,
                status: this.status,
              }),
            ]
          : [];
      })
    );
  }
}

import type { AgentSnapshot, Message, ToolCall } from '@threadplane/core';
 
export interface SubagentsState 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 Authority {
  readonly threadId: string;
  readonly checkpoint: string;
  readonly signature: string;
  readonly messages: readonly Message[];
  readonly tools: readonly ToolCall[];
}
/** Capture own data without evaluating backend-supplied accessors. */
export function copyData(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: copyData(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(copyData(value));
export function messageKey(input: Message): string {
  const message = copyData(input) as Message;
  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);
}
export function validToolArgs(input: Pick<ToolCall, 'name' | 'args'>): boolean {
  try {
    const tool = copyData(input) as Pick<ToolCall, 'name' | 'args'>;
    const args = tool.args;
    if (!record(args)) return false;
    return (
      tool.name === 'task' &&
      Object.keys(args).length === 2 &&
      ['research', 'booking', 'itinerary'].includes(
        args['subagent_type'] as string
      ) &&
      text(args['task_description'])
    );
  } catch {
    return false;
  }
}
/** Only exact saved root evidence grants permission for a follow-up turn. */
export function captureTerminal(
  input: SubagentsState,
  threadId: string
): Authority | null {
  try {
    const state = copyData(input) as SubagentsState;
    if (
      state.status !== 'idle' ||
      state.error ||
      state.interrupts.length ||
      state.subgraphs.length ||
      !state.history?.length
    )
      return null;
    const history = state.history[0],
      checkpoint = history.checkpoint;
    if (
      !record(checkpoint) ||
      checkpoint['thread_id'] !== threadId ||
      checkpoint['checkpoint_ns'] !== '' ||
      !text(checkpoint['checkpoint_id']) ||
      !Array.isArray(history.next) ||
      history.next.length
    )
      return null;
    const messages = state.messages,
      tools = state.toolCalls;
    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 ||
      tools.some(
        (tool) =>
          !text(tool.id) ||
          !validToolArgs(tool) ||
          tool.status !== 'complete' ||
          typeof tool.result !== 'string'
      )
    )
      return null;
    const catalog = new Map(tools.map((tool) => [tool.id, tool]));
    const calls = new Set<string>(),
      results = new Set<string>(),
      pending = new Set<string>();
    for (const message of messages) {
      if (
        !text(message.id) ||
        typeof message.content !== 'string' ||
        message.delivery.phase !== 'complete' ||
        message.delivery.outcome !== 'success' ||
        message.delivery.generation !== message.id
      )
        return null;
      if (message.role === 'user') {
        if (
          pending.size ||
          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);
          pending.add(id);
        }
      } else if (message.role === 'tool') {
        const id = message.toolCallId,
          tool = id ? catalog.get(id) : undefined;
        if (
          !id ||
          !tool ||
          !pending.has(id) ||
          results.has(id) ||
          tool.status !== 'complete' ||
          tool.name !== message.name ||
          tool.result !== message.content ||
          message.toolCallIds?.length
        )
          return null;
        results.add(id);
        pending.delete(id);
      } else return null;
    }
    if (
      calls.size !== tools.length ||
      results.size !== tools.length ||
      pending.size ||
      messages.at(-1)?.role !== 'assistant' ||
      !text(messages.at(-1)?.content)
    )
      return null;
    return Object.freeze({
      threadId,
      checkpoint: key(checkpoint),
      signature: key({ history, messages, tools }),
      messages,
      tools,
    });
  } catch {
    return null;
  }
}

import type {
  AgentSession,
  CompleteOutcome,
  Message,
  ToolCall,
} from '@threadplane/core';
import {
  createMessageContent,
  type MessageRow,
} from '@threadplane/content/messages';
import {
  captureTerminal,
  messageKey,
  toolsKey,
  copyData,
  validToolArgs,
  type SubagentsState,
} from './authority';
import { projectToolObservations, type ToolCard } from './projection';
import { ChildObservations, type ChildRow } from './children';
 
export interface SubagentsSession extends Omit<AgentSession, 'getSnapshot'> {
  getSnapshot(): SubagentsState;
  load(options?: { readonly signal?: AbortSignal }): Promise<void>;
}
export interface SubagentsSnapshot {
  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 observations: ReadonlyMap<string, readonly ToolCard[]>;
  readonly childObservations: readonly ChildRow[];
  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 a matching saved terminal transcript grants a follow-up. */
export function createSubagentsApplication(options: {
  readonly createThread: (signal: AbortSignal) => Promise<string>;
  readonly sessionFactory: (id: string) => SubagentsSession;
}) {
  type Turn = {
    children: ChildObservations;
    loading: boolean;
    childrenCleared: boolean;
    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: SubagentsSession;
    content: ReturnType<typeof createMessageContent>;
    release?: () => void;
    closed: boolean;
    unsafe: boolean;
    confirmed: readonly Message[];
    confirmedTools: readonly ToolCall[];
    generations: Set<string>;
    retainedChildren: readonly ChildRow[];
    turn?: Turn;
  };
  let owner: Owner | undefined, operation: AbortController | undefined;
  let turnSerial = 0;
  let disposed = false,
    disposal: Promise<void> | undefined;
  const listeners = new Set<() => void>(),
    cleanups = new Set<Promise<void>>();
  let snapshot: SubagentsSnapshot = Object.freeze({
    threadId: null,
    rows: Object.freeze([]),
    activity: 'idle',
    busy: false,
    canSubmit: true,
    outcome: null,
    observations: new Map(),
    childObservations: Object.freeze([]),
    error: null,
    viewGeneration: 0,
  });
  function publish(update: Partial<SubagentsSnapshot>) {
    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 childRows(selected: Owner): readonly ChildRow[] {
    return Object.freeze([
      ...selected.retainedChildren,
      ...(selected.turn?.children.rows() ?? []),
    ]);
  }
  function blocked() {
    owner?.turn?.children.finish('Incomplete');
    publish({
      ...(owner ? { childObservations: childRows(owner) } : {}),
      canSubmit: false,
      outcome: 'error',
      error:
        'This conversation could not be confirmed. Start a new conversation to continue.',
    });
  }
  function observe(selected: Owner, state: SubagentsState) {
    state = copyData(state) as SubagentsState;
    if (
      !state ||
      !Array.isArray(state.messages) ||
      !Array.isArray(state.toolCalls) ||
      !Array.isArray(state.interrupts) ||
      !Array.isArray(state.subgraphs) ||
      state.error ||
      state.status === 'error' ||
      state.interrupts.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 message of state.messages) {
      if (
        !['user', 'assistant', 'tool'].includes(message.role) ||
        (message.role !== 'tool' && message.toolCallId !== undefined) ||
        (message.role !== 'assistant' && message.toolCallIds?.length) ||
        (message.role === 'tool' && !message.toolCallId) ||
        !['streaming', 'complete'].includes(message.delivery.phase) ||
        (completed(message) && message.delivery.outcome !== 'success')
      )
        throw new Error('Invalid observed message');
    }
    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 (
        !validToolArgs(tool) ||
        !['pending', 'complete'].includes(tool.status)
      )
        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, copyData(tool) as ToolCall);
    }
    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 (!turn) {
      if (state.messages.length !== prefix.length || state.subgraphs.length)
        throw new Error('Unowned turn');
      return;
    }
    if (turn.expected) {
      if (
        state.messages.length !== turn.expected.messages.length ||
        turn.expected.messages.some((message, index) => {
          const current = state.messages[index];
          return (
            !same(message, current) ||
            current.delivery.phase !== 'complete' ||
            (current.delivery.generation !== message.delivery.generation &&
              current.delivery.generation !== current.id)
          );
        }) ||
        toolsKey(state.toolCalls) !== toolsKey(turn.expected.tools)
      )
        throw new Error('Changed saved evidence');
      if (turn.loading && !state.subgraphs.length) {
        turn.childrenCleared = true;
      } else if (
        turn.childrenCleared ||
        !turn.generation ||
        !turn.children.observe(state.subgraphs, {
          generation: turn.generation,
          rootIds: new Set(state.messages.map((message) => message.id)),
          terminal: true,
        })
      )
        throw new Error('Changed child confirmation');
      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')
        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');
    if (turn.generation) {
      if (
        !turn.children.observe(state.subgraphs, {
          generation: turn.generation,
          rootIds: new Set(state.messages.map((message) => message.id)),
          terminal: false,
        })
      )
        throw new Error('Invalid child observation');
    } else if (state.subgraphs.length)
      throw new Error('Unowned child observation');
  }
  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)) {
      let rows: readonly MessageRow[],
        observations: ReadonlyMap<string, readonly ToolCard[]>;
      try {
        const captured = copyData(state) as SubagentsState;
        rows = selected.content.project(captured);
        observations = projectToolObservations(
          captured.messages,
          captured.toolCalls
        );
      } catch {
        selected.unsafe = true;
        return state;
      }
      if (!valid()) return;
      if (selected.unsafe) selected.turn?.children.finish('Incomplete');
      publish({
        rows,
        observations,
        childObservations: childRows(selected),
        ...(selected.unsafe
          ? {
              canSubmit: false,
              outcome: 'error' as const,
              error:
                'This conversation could not be confirmed. Start a new conversation to continue.',
            }
          : {}),
      });
    }
    if (!valid()) return;
    if (selected.session.getSnapshot() !== state) selected.unsafe = true;
    if (!valid()) return;
    return state;
  }
  async function execute(input: { text: string }) {
    if (disposed || snapshot.busy || !snapshot.canSubmit || !input.text.trim())
      return;
    const admitted = new AbortController();
    operation = admitted;
    const current = () =>
      !disposed && operation === admitted && !admitted.signal.aborted;
    publish({
      busy: true,
      canSubmit: false,
      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(),
          retainedChildren: Object.freeze([]),
        };
        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;
      if (!before || selected.unsafe) {
        blocked();
        return;
      }
      selected.turn = {
        children: new ChildObservations(
          String(snapshot.viewGeneration),
          String(++turnSerial)
        ),
        loading: false,
        childrenCleared: false,
        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 dispatch.
      if (!valid()) return;
      const dispatch = selected.session.getSnapshot();
      if (!valid()) return;
      try {
        observe(selected, dispatch);
      } catch {
        selected.unsafe = true;
      }
      if (!valid()) return;
      const currentDispatch = selected.session.getSnapshot();
      if (!valid()) return;
      if (selected.unsafe || currentDispatch !== dispatch) {
        selected.unsafe = true;
        blocked();
        return;
      }
      const outcome = await selected.session.submit(input.text, {
        signal: admitted.signal,
      });
      if (!valid()) return;
      const live = read(selected, valid),
        turn = selected.turn;
      if (!valid()) return;
      const terminalChildren =
        live &&
        turn?.generation &&
        turn.children.observe(live.subgraphs, {
          generation: turn.generation,
          rootIds: new Set(live.messages.map((message) => message.id)),
          terminal: true,
        });
      if (!valid()) return;
      if (
        !live ||
        selected.unsafe ||
        outcome !== 'success' ||
        live.status !== 'idle' ||
        !turn ||
        !turn.human ||
        live.messages.length <= turn.base.length ||
        live.messages.some((message) => !completed(message)) ||
        !terminalChildren
      ) {
        blocked();
        return;
      }
      const expected = {
        messages: copyData(live.messages) as readonly Message[],
        tools: copyData(live.toolCalls) as readonly ToolCall[],
      };
      if (!valid()) return;
      turn.expected = expected;
      publish({ activity: 'confirming' });
      if (!valid()) return;
      if (!read(selected, valid) || selected.unsafe) {
        if (valid()) blocked();
        return;
      }
      turn.loading = true;
      await selected.session.load({ signal: admitted.signal });
      if (!valid()) return;
      const saved = read(selected, valid);
      if (!valid()) return;
      const authority = saved && captureTerminal(saved, selected.id);
      if (!valid()) return;
      const currentSaved = selected.session.getSnapshot();
      if (!valid()) return;
      if (!authority || selected.unsafe || currentSaved !== saved) {
        blocked();
        return;
      }
      if (!turn.children.finish('Observed response')) {
        blocked();
        return;
      }
      selected.retainedChildren = childRows(selected);
      selected.confirmed = authority.messages;
      selected.confirmedTools = authority.tools;
      if (turn.human) selected.generations.add(turn.human.delivery.generation);
      selected.turn = undefined;
      publish({
        outcome,
        childObservations: selected.retainedChildren,
        canSubmit: true,
      });
      if (!valid()) return;
      const latest = read(selected, valid);
      if (valid() && (!latest || selected.unsafe)) blocked();
    } catch {
      if (current()) blocked();
    } 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;
    selected?.turn?.children.finish('Stopped');
    const stoppedChildren = selected
      ? childRows(selected)
      : snapshot.childObservations;
    const cleanup = detach();
    publish({
      busy: true,
      activity: 'replacing',
      canSubmit: false,
      outcome: 'aborted',
      childObservations: stoppedChildren,
      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,
      activity: 'replacing',
    });
    await cleanup;
    if (disposed || operation !== admitted || admitted.signal.aborted) return;
    operation = undefined;
    publish({
      threadId: null,
      rows: Object.freeze([]),
      observations: new Map(),
      childObservations: Object.freeze([]),
      busy: false,
      canSubmit: true,
      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 }),
    stop,
    newConversation,
    dispose,
  };
}