React AG-UI 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 { useEffect, useState } from 'react';
import { useAgent } from '@threadplane/react';
import { projectTextTranscript } from '@threadplane/ag-ui';
import {
  ChatInput,
  TextTranscript,
  ToolObservation,
} from '@threadplane/react/chat';
import {
  createConnectedApplication,
  type subagentsConnection,
} from './connection';
import { taskArguments, type ChildCard } from './children';
 
const titles = {
  research: 'Research specialist',
  booking: 'Booking specialist',
  itinerary: 'Itinerary specialist',
};
function SpecialistCard({
  card,
  active,
  stopped,
}: {
  readonly card: ChildCard;
  readonly active: boolean;
  readonly stopped: boolean;
}) {
  const captured = card.messages.map((message) => message.text).join('\n');
  const display = captured || card.answer;
  return (
    <section
      aria-label={titles[card.role]}
      className="specialist-card"
      data-child-id={card.id}
      data-child-call={card.callId}
      data-child-phase={card.phase}
    >
      <h3>{titles[card.role]}</h3>
      <p className="specialist-phase">
        {card.phase === 'complete'
          ? 'Complete'
          : card.phase === 'error'
          ? 'Could not finish'
          : active
          ? 'Working…'
          : stopped
          ? 'Stopped'
          : 'Incomplete'}
      </p>
      {card.phase === 'error' ? (
        <p>The specialist could not finish.</p>
      ) : display ? (
        <p className="specialist-text">{display}</p>
      ) : active ? (
        <p>Waiting for the specialist’s response…</p>
      ) : (
        <p>No completed specialist response was confirmed.</p>
      )}
    </section>
  );
}
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);
  useEffect(() => {
    onReady();
  }, [onReady]);
  useEffect(
    () => () => {
      void application.dispose();
    },
    [application]
  );
  useEffect(() => {
    setDraft('');
  }, [snapshot.viewGeneration]);
  const transcript = snapshot.native?.transcript ?? [];
  const rows = projectTextTranscript(transcript);
  const status = snapshot.busy
    ? snapshot.activity === 'replacing'
      ? 'Starting new conversation…'
      : 'Planning with specialists…'
    : !snapshot.canSubmit
    ? 'Start a new conversation to continue.'
    : snapshot.outcome === 'success'
    ? 'Response complete.'
    : 'Ready.';
  return (
    <ExampleLayout className="streaming-demo">
      <header>
        <div>
          <h1>AG-UI trip specialists</h1>
        </div>
        <button
          type="button"
          disabled={snapshot.busy}
          onClick={() => void application.newConversation()}
        >
          New conversation
        </button>
      </header>
      <p>
        Plan a fictional trip. Research, booking and itinerary specialists share
        their progress with the main agent.
      </p>
      <section aria-label="Runtime connection">
        <h2>Runtime connection</h2>
        <p>{connection.label}</p>
      </section>
      <section aria-label="Conversation" className="specialist-conversation">
        <ol>
          {transcript
            .filter((message) => !message.subagentRunId)
            .map((message) => {
              const row = rows.find((item) => item.id === message.id);
              const cards = snapshot.cards.filter(
                (card) => card.ownerId === message.id
              );
              const calls =
                message.role === 'assistant' ? message.toolCalls ?? [] : [];
              const resultCard =
                message.role === 'tool'
                  ? snapshot.cards.find(
                      (card) => card.callId === message.toolCallId
                    )
                  : undefined;
              return (
                <li key={message.id} data-parent-message={message.id}>
                  {row && (
                    <TextTranscript messages={[row]} label="Message text" />
                  )}
                  {cards.map((card) => (
                    <SpecialistCard
                      key={card.id}
                      card={card}
                      active={
                        snapshot.busy && snapshot.native?.run?.id === card.runId
                      }
                      stopped={snapshot.outcome === 'aborted'}
                    />
                  ))}
                  {calls.map((call) => {
                    const args = taskArguments(call.function.arguments);
                    const card = cards.find((item) => item.callId === call.id);
                    return args && call.function.name === 'task' ? (
                      <details key={call.id} data-parent-call={call.id}>
                        <summary>Specialist task</summary>
                        <ToolObservation
                          name="task"
                          argumentsText={JSON.stringify(args)}
                          resultText={
                            card?.phase === 'complete'
                              ? card.answer ?? undefined
                              : undefined
                          }
                        />
                      </details>
                    ) : null;
                  })}
                  {message.role === 'tool' && (
                    <details>
                      <summary>Specialist result</summary>
                      <p>
                        {resultCard?.phase === 'complete' && resultCard.answer
                          ? resultCard.answer
                          : 'The specialist result is unavailable.'}
                      </p>
                    </details>
                  )}
                </li>
              );
            })}
        </ol>
      </section>
      <ExampleFooter>
        {snapshot.error && <p role="alert">{snapshot.error}</p>}
        <ChatInput
          label="Message"
          placeholder="Plan a fictional trip…"
          value={draft}
          onValueChange={setDraft}
          disabled={!snapshot.canSubmit}
          busy={snapshot.busy}
          onStop={snapshot.busy ? () => void application.stop() : undefined}
          onSubmit={(text) => {
            if (!application.getSnapshot().canSubmit) return false;
            void application.submit(text);
            return application.getSnapshot().busy;
          }}
        />
        <p role="status">{status}</p>
      </ExampleFooter>
    </ExampleLayout>
  );
}

import type { Session } from '@threadplane/ag-ui';
 
export type NativeSnapshot = ReturnType<Session['getSnapshot']>;
export type SpecialistRole = 'research' | 'booking' | 'itinerary';
export interface ChildCard {
  readonly id: string;
  readonly runId: string;
  readonly ownerId: string;
  readonly callId: string;
  readonly role: SpecialistRole;
  readonly phase: 'running' | 'complete' | 'error';
  readonly messages: readonly Readonly<{ id: string; text: string }>[];
  readonly answer: string | null;
}
const roles: readonly string[] = ['research', 'booking', 'itinerary'];
const text = (value: unknown, maximum = 4096): value is string =>
  typeof value === 'string' && value.length > 0 && value.length <= maximum;
export function taskArguments(
  value: string
): Readonly<{ role: SpecialistRole; task_description: string }> | undefined {
  try {
    if (!text(value, 8192)) return;
    const args: unknown = JSON.parse(value);
    if (!args || typeof args !== 'object' || Array.isArray(args)) return;
    const record = args as Record<string, unknown>;
    if (
      Object.keys(record).length !== 2 ||
      !Object.hasOwn(record, 'role') ||
      !Object.hasOwn(record, 'task_description') ||
      typeof record.role !== 'string' ||
      !roles.includes(record.role) ||
      !text(record.task_description)
    )
      return;
    return Object.freeze({
      role: record.role as SpecialistRole,
      task_description: record.task_description,
    });
  } catch {
    return undefined;
  }
}
 
/** Display observations only. Captured child text never enters request history. */
export function observeChildren(
  snapshot: NativeSnapshot,
  previous: readonly ChildCard[],
  previousCalls: ReadonlySet<string>
): Readonly<{ cards: readonly ChildCard[]; unsupported: boolean }> {
  const cards = new Map(previous.map((card) => [card.id, card]));
  let unsupported = false;
  try {
    if (
      cards.size !== previous.length ||
      previous.length > 100 ||
      snapshot.subagents.length > 30
    )
      throw new Error();
    const attributed = new Set(
      snapshot.subagents.map((child) => child.started.subagentRunId)
    );
    if (
      snapshot.transcript.some(
        (message) =>
          typeof message.subagentRunId === 'string' &&
          !attributed.has(message.subagentRunId)
      )
    )
      throw new Error();
    const seen = new Set<string>();
    for (const child of snapshot.subagents) {
      const started = child.started;
      const id = started.subagentRunId,
        callId = started.parentToolCallId;
      const old = cards.get(id);
      const runId = snapshot.run?.id;
      if (
        !text(id) ||
        !text(callId) ||
        id !== callId + '-sub' ||
        seen.has(id) ||
        !text(runId) ||
        started.parentSubagentRunId !== undefined ||
        (previousCalls.has(callId) && !old)
      )
        throw new Error();
      seen.add(id);
      const owners = snapshot.transcript.filter(
        (message) =>
          message.role === 'assistant' &&
          !message.subagentRunId &&
          message.toolCalls?.some((call) => call.id === callId)
      );
      if (owners.length !== 1) throw new Error();
      const owner = owners[0];
      if (owner.role !== 'assistant') throw new Error();
      const calls = owner.toolCalls?.filter((call) => call.id === callId) ?? [];
      if (calls.length !== 1 || calls[0].function.name !== 'task')
        throw new Error();
      const args = taskArguments(calls[0].function.arguments);
      if (
        !roles.includes(started.name) ||
        (args && args.role !== started.name) ||
        (!args && snapshot.status !== 'running') ||
        (started.parentMessageId !== undefined &&
          started.parentMessageId !== owner.id)
      )
        throw new Error();
      // The bridge can defer full task arguments until a canonical snapshot.
      // A named child is display evidence only; root policy later needs args.
      const role = started.name as SpecialistRole;
      if (
        old &&
        (old.ownerId !== owner.id ||
          old.callId !== callId ||
          old.runId !== runId ||
          old.role !== role)
      )
        throw new Error();
      let phase: ChildCard['phase'] = 'running';
      const terminal = child.terminal;
      if (terminal) {
        if (terminal.subagentRunId !== id) throw new Error();
        // Error payloads are retained by the protocol owner, never copied here.
        phase =
          terminal.type === 'SUBAGENT_FINISHED' &&
          terminal.outcome?.type === 'success'
            ? 'complete'
            : 'error';
      }
      if (phase === 'error') unsupported = true;
      if (old && old.phase !== 'running' && old.phase !== phase)
        throw new Error();
      const observed = snapshot.transcript.filter(
        (message) => message.subagentRunId === id
      );
      if (observed.length > 8) throw new Error();
      const messages = observed.length
        ? observed.map((message) => {
            if (
              message.role !== 'assistant' ||
              message.toolCalls?.length ||
              (message.content !== undefined &&
                typeof message.content !== 'string') ||
              (message.content?.length ?? 0) > 65_536 ||
              !text(message.id) ||
              !message.id.startsWith(id + '-m')
            )
              throw new Error();
            return Object.freeze({
              id: message.id,
              text: message.content ?? '',
            });
          })
        : old?.messages ?? Object.freeze([]);
      if (
        new Set(messages.map((message) => message.id)).size !== messages.length
      )
        throw new Error();
      let answer: string | null = null;
      if (phase === 'complete') {
        const results = snapshot.transcript.filter(
          (message) => message.role === 'tool' && message.toolCallId === callId
        );
        if (results.length > 1) throw new Error();
        const result = results[0];
        if (result) {
          if (result.id !== callId || !text(result.content, 65_536))
            throw new Error();
          answer = result.content;
          const observedText = messages
            .map((message) => message.text)
            .join('\n');
          if (observedText && observedText !== answer) throw new Error();
        }
      }
      if (!old && cards.size >= 100) throw new Error();
      cards.set(
        id,
        Object.freeze({
          id,
          runId,
          ownerId: owner.id,
          callId,
          role,
          phase,
          messages: Object.freeze([...messages]),
          answer,
        })
      );
    }
  } catch {
    unsupported = true;
  }
  return Object.freeze({
    cards: Object.freeze([...cards.values()]),
    unsupported,
  });
}

import type { ChildCard, NativeSnapshot } from './children';
import { taskArguments } from './children';
 
export interface OwnedTurn {
  readonly threadId: string;
  readonly runId: string;
  readonly human: Readonly<{ id: string; content: string }>;
  readonly prefix: NativeSnapshot['transcript'];
}
function same(
  left: unknown,
  right: unknown,
  valid: () => boolean,
  depth = 0
): boolean {
  if (!valid() || depth > 100) return false;
  if (Object.is(left, right)) return true;
  if (Array.isArray(left) && Array.isArray(right))
    return (
      left.length === right.length &&
      left.every((item, index) => same(item, right[index], valid, depth + 1))
    );
  if (!left || !right || typeof left !== 'object' || typeof right !== 'object')
    return false;
  const keys = Object.keys(left);
  return (
    keys.length === Object.keys(right).length &&
    keys.every(
      (key) =>
        Object.hasOwn(right, key) &&
        same(
          (left as Record<string, unknown>)[key],
          (right as Record<string, unknown>)[key],
          valid,
          depth + 1
        )
    )
  );
}
 
/** Captures the native admission once; the confirmed prefix stays borrowed. */
export function captureTurn(
  snapshot: NativeSnapshot,
  threadId: string,
  previousRunId: string | undefined,
  prefix: NativeSnapshot['transcript'],
  text: string,
  valid: () => boolean
): OwnedTurn | undefined {
  if (
    !valid() ||
    snapshot.status !== 'running' ||
    !snapshot.run?.id ||
    snapshot.run.id === previousRunId ||
    snapshot.run.outcome !== undefined ||
    snapshot.transcript.length !== prefix.length + 1 ||
    !prefix.every((row, index) => same(row, snapshot.transcript[index], valid))
  )
    return;
  const human = snapshot.transcript.at(-1)!;
  if (
    !valid() ||
    human.role !== 'user' ||
    !human.id ||
    human.content !== text ||
    !threadId
  )
    return;
  return Object.freeze({
    threadId,
    runId: snapshot.run.id,
    human: Object.freeze({ id: human.id, content: text }),
    prefix,
  });
}
 
/** Confirms a borrowed canonical parent history; never prepares requests. */
export function confirmParent(
  snapshot: NativeSnapshot,
  turn: OwnedTurn,
  cards: readonly ChildCard[],
  valid: () => boolean = () => true
): NativeSnapshot['transcript'] | undefined {
  try {
    if (!valid()) return;
    const run = snapshot.run,
      transcript = snapshot.transcript,
      children = snapshot.subagents;
    if (
      !valid() ||
      snapshot.status !== 'idle' ||
      snapshot.decision ||
      run?.id !== turn.runId ||
      run.outcome !== 'success' ||
      run.legacyInterrupt ||
      run.terminal?.type !== 'RUN_FINISHED' ||
      run.terminal.threadId !== turn.threadId ||
      run.terminal.runId !== turn.runId ||
      (run.terminal.outcome && run.terminal.outcome.type !== 'success') ||
      (run.terminal.outcome?.type === 'success' &&
        run.terminal.outcome.pendingToolCallIds?.length) ||
      transcript.length > 500 ||
      transcript.length < turn.prefix.length + 2 ||
      !turn.prefix.every((message, index) =>
        same(message, transcript[index], valid)
      )
    )
      return;
    const human = transcript[turn.prefix.length];
    if (
      !valid() ||
      human.role !== 'user' ||
      human.id !== turn.human.id ||
      human.content !== turn.human.content
    )
      return;
    const ids = new Set<string>(),
      results = new Set<string>();
    const calls = new Map<
      string,
      Readonly<{ ownerId: string; role: string; current: boolean }>
    >();
    for (let index = 0; index < transcript.length; index++) {
      if (!valid()) return;
      const message = transcript[index];
      if (
        !message.id ||
        ids.has(message.id) ||
        typeof message.subagentRunId === 'string'
      )
        return;
      ids.add(message.id);
      if (message.role === 'user') {
        if (index > turn.prefix.length || typeof message.content !== 'string')
          return;
      } else if (message.role === 'assistant') {
        if (
          message.content !== undefined &&
          typeof message.content !== 'string'
        )
          return;
        for (const call of message.toolCalls ?? []) {
          const args = taskArguments(call.function.arguments);
          if (
            !call.id ||
            calls.has(call.id) ||
            call.function.name !== 'task' ||
            !args
          )
            return;
          calls.set(
            call.id,
            Object.freeze({
              ownerId: message.id,
              role: args.role,
              current: index > turn.prefix.length,
            })
          );
        }
      } else if (message.role === 'tool') {
        const call = calls.get(message.toolCallId);
        if (
          !call ||
          results.has(message.toolCallId) ||
          message.id !== message.toolCallId ||
          typeof message.content !== 'string' ||
          !message.content ||
          message.content.length > 65_536
        )
          return;
        results.add(message.toolCallId);
        if (call.current) {
          const matching = cards.filter(
            (card) =>
              card.callId === message.toolCallId && card.runId === turn.runId
          );
          if (matching.length !== 1) return;
          const card = matching[0];
          if (
            card.ownerId !== call.ownerId ||
            card.role !== call.role ||
            card.phase !== 'complete' ||
            card.answer !== message.content
          )
            return;
        }
      } else return;
    }
    if (calls.size !== results.size) return;
    const current = [...calls.entries()].filter(([, call]) => call.current);
    const currentCards = cards.filter((card) => card.runId === turn.runId);
    if (
      current.length !== currentCards.length ||
      current.length !== children.length
    )
      return;
    const seen = new Set<string>();
    for (const child of children) {
      if (!valid()) return;
      const start = child.started,
        terminal = child.terminal;
      const id = start.subagentRunId,
        callId = start.parentToolCallId;
      if (
        !callId ||
        id !== callId + '-sub' ||
        seen.has(id) ||
        start.parentSubagentRunId !== undefined
      )
        return;
      seen.add(id);
      const call = calls.get(callId);
      if (
        !call?.current ||
        call.role !== start.name ||
        (start.parentMessageId !== undefined &&
          start.parentMessageId !== call.ownerId) ||
        terminal?.type !== 'SUBAGENT_FINISHED' ||
        terminal.subagentRunId !== id ||
        terminal.outcome?.type !== 'success'
      )
        return;
      if (
        !currentCards.some(
          (card) =>
            card.id === id &&
            card.ownerId === call.ownerId &&
            card.callId === callId
        )
      )
        return;
    }
    const last = transcript.at(-1);
    if (
      !valid() ||
      last?.role !== 'assistant' ||
      last.toolCalls?.length ||
      typeof last.content !== 'string' ||
      !last.content.trim()
    )
      return;
    return transcript;
  } catch {
    return undefined;
  }
}

import type { Session } from '@threadplane/ag-ui';
import type { CompleteOutcome } from '@threadplane/core';
import {
  observeChildren,
  taskArguments,
  type ChildCard,
  type NativeSnapshot,
} from './children';
import { captureTurn, confirmParent, type OwnedTurn } from './policy';
function unsupportedSnapshot(root: NativeSnapshot, valid: () => boolean) {
  if (
    !valid() ||
    root.decision ||
    root.run?.legacyInterrupt ||
    root.transcript.length > 500
  )
    return true;
  const ids = new Set<string>();
  for (const message of root.transcript) {
    if (!valid() || !message.id || ids.has(message.id)) return true;
    ids.add(message.id);
    if (
      !['user', 'assistant', 'tool'].includes(message.role) ||
      (message.content !== undefined && typeof message.content !== 'string')
    )
      return true;
    if (message.role === 'assistant') {
      for (const call of message.toolCalls ?? []) {
        if (
          call.function.name !== 'task' ||
          !call.id ||
          (root.status !== 'running' && !taskArguments(call.function.arguments))
        )
          return true;
      }
    }
  }
  return !valid();
}
 
export interface SubagentsSnapshot {
  readonly threadId: string | null;
  readonly native: NativeSnapshot | undefined;
  readonly cards: readonly ChildCard[];
  readonly busy: boolean;
  readonly canSubmit: boolean;
  readonly activity: 'idle' | 'running' | 'replacing';
  readonly outcome: CompleteOutcome | null;
  readonly error: string | null;
  readonly viewGeneration: number;
}
 
/** Owns lifetime and admission; the native session owns all protocol execution. */
export function createSubagentsApplication(options: {
  readonly sessionFactory: (threadId: string) => Session;
}) {
  type Primary = {
    session: Session;
    threadId: string;
    closed: boolean;
    release?: () => void;
    unsafe: boolean;
    confirmed: NativeSnapshot['transcript'];
    previousRunId?: string;
    admittedText?: string;
    cards: readonly ChildCard[];
    turn?: OwnedTurn;
  };
  const listeners = new Set<() => void>(),
    cleanups = new Set<Promise<void>>();
  let primary: Primary | undefined;
  let operation: AbortController | undefined;
  let disposed = false,
    disposal: Promise<void> | undefined;
  let snapshot: SubagentsSnapshot = Object.freeze({
    threadId: null,
    native: undefined,
    cards: Object.freeze([]),
    busy: false,
    canSubmit: true,
    activity: 'idle',
    outcome: null,
    error: null,
    viewGeneration: 0,
  });
  function publish(update: Partial<SubagentsSnapshot>) {
    if (disposed) return;
    snapshot = Object.freeze({ ...snapshot, ...update });
    for (const notify of listeners) {
      try {
        notify();
      } catch {
        /* A view cannot interrupt native execution or cleanup. */
      }
    }
  }
  function current(selected: Primary) {
    return (
      !disposed &&
      primary === selected &&
      !selected.closed &&
      !operation?.signal.aborted
    );
  }
  function close(selected: Primary) {
    if (selected.closed) return Promise.resolve();
    selected.closed = true;
    try {
      selected.release?.();
    } catch {
      /* Revoked observation cannot prevent native cleanup. */
    }
    const cleanup = Promise.resolve()
      .then(() => selected.session.dispose())
      .then(
        () => undefined,
        () => undefined
      );
    cleanups.add(cleanup);
    void cleanup.then(() => cleanups.delete(cleanup));
    return cleanup;
  }
  function detach() {
    const previous = primary;
    primary = undefined;
    return previous ? close(previous) : Promise.resolve();
  }
  function read(selected: Primary, valid: () => boolean) {
    if (!valid()) return;
    const root = selected.session.getSnapshot();
    if (!valid()) return;
    const blocked = unsupportedSnapshot(root, valid);
    return valid() ? { root, blocked } : undefined;
  }
  function capture(selected: Primary, valid: () => boolean) {
    const state = read(selected, valid);
    if (!state || !valid()) return;
    selected.unsafe ||= state.blocked;
    if (!selected.turn && snapshot.activity === 'running') {
      const turn = captureTurn(
        state.root,
        selected.threadId,
        selected.previousRunId,
        selected.confirmed,
        selected.admittedText ?? '',
        valid
      );
      if (!valid()) return;
      if (turn) selected.turn = turn;
    }
    const previousCalls = new Set(
      selected.confirmed.flatMap((message) =>
        message.role === 'assistant'
          ? (message.toolCalls ?? []).map((call) => call.id)
          : []
      )
    );
    const projection = observeChildren(
      state.root,
      selected.cards,
      previousCalls
    );
    if (!valid()) return;
    selected.unsafe ||= projection.unsupported;
    selected.cards = projection.cards;
    const confirmed =
      selected.turn && !selected.unsafe
        ? confirmParent(state.root, selected.turn, selected.cards, valid)
        : undefined;
    return valid()
      ? { native: state.root, cards: selected.cards, confirmed }
      : undefined;
  }
 
  function observe(selected: Primary) {
    const valid = () => current(selected);
    try {
      const result = capture(selected, valid);
      if (!result || !valid()) return;
      publish({
        native: result.native,
        cards: result.cards,
        ...(selected.unsafe ? { canSubmit: false } : {}),
      });
    } catch {
      if (valid()) {
        selected.unsafe = true;
        publish({
          canSubmit: false,
          error:
            'The specialist result could not be confirmed. Start a new conversation.',
        });
      }
    }
  }
  function install(valid: () => boolean) {
    const threadId = crypto.randomUUID();
    const session = options.sessionFactory(threadId);
    const selected: Primary = {
      session,
      threadId,
      closed: false,
      unsafe: false,
      confirmed: [],
      cards: Object.freeze([]),
    };
    if (!valid()) {
      void close(selected);
      return;
    }
    primary = selected;
    const release = session.subscribe(() => observe(selected));
    selected.release = release;
    if (selected.closed) release();
    if (!valid() || primary !== selected) {
      void close(selected);
      return;
    }
    return selected;
  }
  function admit(activity: SubagentsSnapshot['activity'], execution: boolean) {
    const admitted = new AbortController();
    operation = admitted;
    const valid = () =>
      !disposed && operation === admitted && !admitted.signal.aborted;
    publish({
      busy: true,
      canSubmit: false,
      activity,
      ...(execution ? { outcome: null, error: null } : {}),
    });
    return { admitted, valid };
  }
  function finish(admitted: AbortController) {
    if (!disposed && operation === admitted) {
      operation = undefined;
      publish({ busy: false, activity: 'idle' });
    }
  }
  async function execute(text: string) {
    if (disposed || snapshot.busy || !snapshot.canSubmit || !text.trim())
      return;
    const before = primary;
    const initialValid = () =>
      !disposed &&
      !snapshot.busy &&
      snapshot.canSubmit &&
      primary === before &&
      !before?.closed;
    let state: ReturnType<typeof read>;
    try {
      state = before && read(before, initialValid);
    } catch {
      if (initialValid())
        publish({
          canSubmit: false,
          error:
            'The specialist session is unavailable. Start a new conversation.',
        });
      return;
    }
    if (!initialValid()) return;
    if (before && (before.unsafe || state?.blocked)) {
      publish({ canSubmit: false });
      return;
    }
    const { admitted, valid } = admit('running', true);
    try {
      if (!valid()) return;
      const selected = primary ?? install(valid);
      if (!selected || !valid()) return;
      publish({ threadId: selected.threadId });
      const ownedValid = () => valid() && current(selected);
      const root = read(selected, ownedValid);
      if (!root || !ownedValid()) return;
      if (root.blocked || selected.unsafe) {
        selected.unsafe = true;
        publish({ canSubmit: false });
        return;
      }
      selected.previousRunId = root.root.run?.id;
      selected.turn = undefined;
      selected.admittedText = text;
      if (!ownedValid()) return;
      const pending = selected.session.submit(text, {
        signal: admitted.signal,
      });
      void pending.catch(() => undefined);
      if (!ownedValid()) return;
      observe(selected);
      const outcome = await pending;
      if (!ownedValid()) return;
      const result = capture(selected, ownedValid);
      if (!result || !ownedValid()) return;
      const safe =
        outcome === 'success' && !!result.confirmed && !selected.unsafe;
      if (safe) selected.confirmed = result.confirmed!;
      publish({
        native: result.native,
        cards: result.cards,
        outcome,
        canSubmit: safe,
        error: safe
          ? null
          : 'The current parent outcome could not be confirmed. Start a new conversation.',
      });
    } catch {
      if (valid())
        publish({
          canSubmit: false,
          outcome: 'error',
          error: 'The AG-UI request failed. Start a new conversation.',
        });
    } finally {
      finish(admitted);
    }
  }
  async function newConversation() {
    if (disposed || snapshot.busy) return;
    const { admitted, valid } = admit('replacing', false);
    const cleanup = detach();
    try {
      if (!valid()) return;
      publish({
        threadId: null,
        native: undefined,
        cards: Object.freeze([]),
        outcome: null,
        error: null,
        viewGeneration: snapshot.viewGeneration + 1,
      });
      if (!valid()) return;
      await cleanup;
      if (valid()) publish({ canSubmit: true });
    } finally {
      finish(admitted);
    }
  }
  async function stop() {
    if (disposed || !operation) return;
    operation.abort();
    operation = undefined;
    const selected = primary,
      cleanup = detach();
    publish({
      busy: false,
      activity: 'idle',
      canSubmit: false,
      outcome: 'aborted',
      error: 'The operation stopped. Start a new conversation.',
    });
    try {
      await selected?.session.stop();
    } catch {
      /* Detached work has no publication authority. */
    }
    await cleanup;
  }
  function dispose(): Promise<void> {
    if (disposal) return disposal;
    disposed = true;
    operation?.abort();
    operation = undefined;
    void detach();
    listeners.clear();
    disposal = (async () => {
      do {
        await Promise.resolve();
        await Promise.all([...cleanups]);
      } while (cleanups.size);
    })();
    return disposal;
  }
  return {
    getSnapshot: () => snapshot,
    subscribe(notify: () => void) {
      if (disposed) return () => undefined;
      listeners.add(notify);
      return () => {
        listeners.delete(notify);
      };
    },
    submit: execute,
    newConversation,
    stop,
    dispose,
  };
}

import { createSession } from '@threadplane/ag-ui';
import { createSubagentsApplication } from './application';
// 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';
export function subagentsConnection(
  configuration: RuntimeBridgeConfiguration
): {
  url: string;
  label: 'Shared runtime' | 'Developer runtime';
} {
  if (
    configuration.status === 'error' ||
    (configuration.status === 'configured' &&
      configuration.target.kind === 'langsmith')
  )
    throw new Error('Runtime configuration failed.');
  const target =
    configuration.status === 'configured' &&
    configuration.target.kind === 'ag-ui'
      ? configuration.target
      : null;
  return {
    url:
      target?.endpoint ??
      new URL('/ag-ui/subagents/agent', window.location.origin).href,
    label: target ? 'Developer runtime' : 'Shared runtime',
  };
}
export function createConnectedApplication(
  connection: ReturnType<typeof subagentsConnection>
) {
  return createSubagentsApplication({
    sessionFactory(threadId) {
      const session = createSession({ threadId, url: connection.url });
      try {
        const snapshot = session.getSnapshot();
        if (
          !snapshot ||
          !Array.isArray(snapshot.transcript) ||
          !Array.isArray(snapshot.subagents) ||
          ![
            'getSnapshot',
            'subscribe',
            'submit',
            'resume',
            'stop',
            'dispose',
          ].every(
            (method) =>
              typeof session[method as keyof typeof session] === 'function'
          )
        )
          throw new Error('Runtime session is unavailable.');
      } catch {
        if (typeof session?.dispose === 'function') {
          try {
            void Promise.resolve(session.dispose()).catch(() => undefined);
          } catch {
            /* Rejected candidates have no authority. */
          }
        }
        throw new Error('Runtime session is unavailable.');
      }
      return session;
    },
  });
}

"""AG-UI Subagents Graph — orchestrator LLM with a single `task` tool that
dispatches to specialized subagents (research/booking/itinerary), surfacing
each dispatch as a native AG-UI subagent card.
 
Mirrors cockpit/chat/subagents' orchestrator + `task` tool + `_run_subagent`
structure, but each dispatch emits `subagent_activity` CUSTOM events (like the
examples/ag-ui `research` tool): `started {name}` before the run,
`message_start {message_id}` + `message {message_id, delta}` per streamed
token (via SubagentStreamHandler), `finished` after — or `error {message}` if
the child fails. The backend's SubagentEmittingAgent expands those CUSTOM
events into the protocol's standard SUBAGENT_STARTED / TEXT_MESSAGE_* (attributed
via subagentRunId) / SUBAGENT_FINISHED / SUBAGENT_ERROR events, which the
@threadplane/ag-ui reducer projects onto agent.subagents().
 
Self-contained: no imports from examples/ or other cockpit capabilities.
"""
 
import os
from pathlib import Path
from typing import Annotated, Literal
 
from langchain_core.callbacks import adispatch_custom_event
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool, InjectedToolCallId
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, END
from langgraph.prebuilt import ToolNode
from langgraph_sdk import get_client
 
from .streaming.subagent_stream_handler import SubagentStreamHandler
 
PROMPTS_DIR = Path(__file__).parent.parent / "prompts"
 
# ── generate_title node (background thread-title generation) ────────────────
 
_TITLE_PROMPT = (
    "In 3-5 words, summarize what the user is asking about. "
    "Output ONLY the title — no quotes, no period, no prefix."
)
_TITLE_MODEL = "gpt-5-mini"
 
 
async def generate_title(state: MessagesState, config) -> dict:
    """Background title generation: on the first turn, summarize the user's
    intent into 3-5 words and persist to LangGraph thread metadata.
 
    Idempotent — skips when metadata.title already exists. Errors are
    swallowed because the title is a UX nicety, never a blocker.
    """
    thread_id = (config.get("configurable") or {}).get("thread_id")
    if not thread_id:
        return {}
    sdk_url = os.environ.get("LANGGRAPH_API_URL")
    try:
        client = get_client(url=sdk_url)
        thread = await client.threads.get(thread_id)
        if (thread.get("metadata") or {}).get("title"):
            return {}
        first_user = next(
            (m for m in state["messages"] if getattr(m, "type", None) == "human"),
            None,
        )
        if not first_user or not isinstance(first_user.content, str):
            return {}
        if first_user.content.lstrip().startswith("{"):
            return {}
        llm = ChatOpenAI(model=_TITLE_MODEL, temperature=0)
        response = await llm.ainvoke([
            SystemMessage(content=_TITLE_PROMPT),
            HumanMessage(content=first_user.content),
        ])
        title = (response.content or "").strip().strip('"').strip("'")[:80]
        if title:
            await client.threads.update(thread_id, metadata={"title": title})
    except Exception as e:  # noqa: BLE001 — title is a UX nicety; never block
        print(
            f"[generate_title] failed for thread {thread_id}: "
            f"{type(e).__name__}: {e}",
            flush=True,
        )
    return {}
 
 
# ── Subagent role prompts ───────────────────────────────────────────────────
 
_RESEARCH_PROMPT = """You are a Research Agent for trip planning. Your job is to
gather destination intel about the airports the traveler is considering: city,
typical weather/conditions, terminals, and any practical travel notes for the
route described in the task.
 
Return a concise 2-4 sentence summary of what you found."""
 
_BOOKING_PROMPT = """You are a Booking Agent for trip planning. Your job is to
suggest flight options between the origin and destination airports described in
the task.
 
Return a concise summary listing 2-3 plausible flight options with airline,
approximate times, and a rough price estimate. Be helpful and concise."""
 
_ITINERARY_PROMPT = """You are an Itinerary Agent for trip planning. Your job is
to synthesize a final trip plan from the research + booking outputs you receive
in the task description.
 
Return a clean 3-5 sentence itinerary summarizing the recommended flight choice,
what to expect on arrival (weather), and any practical tips. Be helpful and
concise."""
 
 
# region run-subagent
async def _run_subagent(
    role: str,
    task_description: str,
    system_prompt: str,
    tool_call_id: str,
) -> str:
    """Run a single subagent LLM, streaming its tokens through
    SubagentStreamHandler so they surface as `subagent_activity`
    `message_start` / `message` (per-token delta) events keyed by the parent
    tool_call_id."""
    llm = ChatOpenAI(model="gpt-5-mini", streaming=True)
    messages = [
        SystemMessage(content=system_prompt),
        HumanMessage(content=task_description),
    ]
    response = await llm.ainvoke(
        messages,
        config={"callbacks": [SubagentStreamHandler(tool_call_id)]},
    )
    content = response.content
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = [
            b.get("text", "")
            for b in content
            if isinstance(b, dict) and b.get("type") == "text"
        ]
        return "\n".join(parts)
    return ""
# endregion
 
 
# region task-tool
@tool
async def task(
    role: Literal["research", "booking", "itinerary"],
    task_description: str,
    tool_call_id: Annotated[str, InjectedToolCallId],
) -> ToolMessage:
    """Delegate a subtask to a specialized subagent.
 
    Roles:
      - research: gathers destination intel (airports, weather, conditions)
      - booking: finds flight options between origin and destination
      - itinerary: synthesizes a final trip plan combining research + bookings
 
    Args:
        role: One of "research", "booking", "itinerary".
        task_description: Plain-English description of what the subagent
            should do (e.g., "Gather info on LAX and JFK airports", or
            "Find morning flights from LAX to JFK").
 
    Returns:
        A stable tool message containing the subagent's final answer.
 
    The subagent run is surfaced to the UI as the protocol's standard
    subagent events: SUBAGENT_STARTED → attributed TEXT_MESSAGE_* per token →
    SUBAGENT_FINISHED (or SUBAGENT_ERROR), with ids derived from this tool's
    own call id (`<tool_call_id>-sub`).
    """
 
    async def _emit(payload: dict) -> None:
        try:
            await adispatch_custom_event(
                "subagent_activity", {"subagent_id": tool_call_id, **payload}
            )
        except Exception:
            pass
 
    prompts = {
        "research": _RESEARCH_PROMPT,
        "booking": _BOOKING_PROMPT,
        "itinerary": _ITINERARY_PROMPT,
    }
    system_prompt = prompts.get(role)
    if system_prompt is None:
        return f"Unknown role: {role}"
 
    await _emit({"phase": "started", "name": role})
    try:
        result = await _run_subagent(role, task_description, system_prompt, tool_call_id)
    except Exception as exc:
        # Surface the failure on the subagent card (SUBAGENT_ERROR), then
        # re-raise so the bridge's own tool-error path still runs.
        await _emit({"phase": "error", "message": f"{type(exc).__name__}: {exc}"})
        raise
    await _emit({"phase": "finished", "status": "complete"})
    return ToolMessage(
        id=tool_call_id,
        tool_call_id=tool_call_id,
        name="task",
        content=result,
    )
# endregion
 
 
# region graph
def build_subagents_graph():
    """Orchestrator LLM with a single `task` tool that dispatches subagents."""
    llm = ChatOpenAI(model="gpt-5-mini", streaming=True).bind_tools([task])
 
    async def orchestrator(state: MessagesState) -> dict:
        system_prompt = (PROMPTS_DIR / "subagents.md").read_text()
        messages = [SystemMessage(content=system_prompt)] + state["messages"]
        response = await llm.ainvoke(messages)
        return {"messages": [response]}
 
    def should_continue(state: MessagesState) -> str:
        last = state["messages"][-1]
        if hasattr(last, "tool_calls") and last.tool_calls:
            return "tools"
        return "generate_title"
 
    graph = StateGraph(MessagesState)
    graph.add_node("orchestrator", orchestrator)
    graph.add_node("tools", ToolNode([task]))
    graph.add_node("generate_title", generate_title)
    graph.set_entry_point("orchestrator")
    graph.add_conditional_edges(
        "orchestrator",
        should_continue,
        {"tools": "tools", "generate_title": "generate_title"},
    )
    graph.add_edge("tools", "orchestrator")
    graph.add_edge("generate_title", END)
    return graph.compile(checkpointer=MemorySaver())
# endregion
 
 
# The graph instance — referenced by server.py
graph = build_subagents_graph()