React subgraphs preview

A trip-planning agent can delegate research while keeping one conversation with the traveler. Subgraphs expose that child work so the UI can show what each specialist is doing.

The native React APIs on this page are experimental, private source-build packages. Run the repository example to use them; they are not an npm installation recipe.

Keep parent and child observations distinct

The native LangGraph snapshot includes namespaced child observations. The application decides which observations belong to the current run and how to present them beside the root transcript. A child response is not automatically another root assistant message, and child progress alone does not establish that the parent turn succeeded. The source walkthrough shows the matching and retention rules used by this example.

The React conversation walkthrough connects this capability to requests, tools, UI state, and saved history.

Follow the parent and child boundary

Choose Run and ask a factual question. The parent routes that question through a research child graph, then answers using its returned brief. A greeting answers directly. Send another message to reuse the same conversation; the parent retains its confirmed conversation history.

The Research boundary panel shows the topic sent to the child and the brief returned to the parent after the current result is confirmed. The child shares those two fields; it does not receive the parent conversation. Its streamed messages remain outside the parent chat transcript.

app.tsx

// 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, MessageList } from '@threadplane/react/chat';
import {
  createConnectedApplication,
  type subgraphsConnection,
} from './connection';
export function SubgraphsDemo({
  connection,
  onReady,
}: {
  readonly connection: ReturnType<typeof subgraphsConnection>;
  readonly onReady: () => void;
}) {
  const [application] = useState(() => createConnectedApplication(connection));
  const snapshot = useAgent(application);
  useEffect(() => {
    onReady();
  }, [onReady]);
  useEffect(
    () => () => {
      void application.dispose();
    },
    [application]
  );
  const status = snapshot.busy
    ? snapshot.activity === 'creating'
      ? 'Creating conversation…'
      : snapshot.activity === 'replacing'
      ? 'Starting new conversation…'
      : 'Receiving response…'
    : snapshot.outcome === 'aborted'
    ? 'Stopped. Start a new conversation to continue.'
    : !snapshot.canSubmit
    ? 'Start a new conversation to continue.'
    : snapshot.outcome === 'success'
    ? 'Response complete.'
    : 'Ready.';
  return (
    <ExampleLayout className="subgraphs-demo">
      <header>
        <div>
          <h1>LangGraph subgraphs</h1>
        </div>
        <button
          type="button"
          disabled={snapshot.busy}
          onClick={() => void application.newConversation()}
        >
          New conversation
        </button>
      </header>
      <p>
        A factual question enters a research child graph. A greeting answers
        directly. The child receives a topic, not the parent conversation.
      </p>
      <MessageList rows={snapshot.rows} label="Conversation" />
      <section aria-label="Research boundary">
        <h2>Research boundary</h2>
        <p>
          {snapshot.route === 'nested'
            ? 'Nested — research boundary confirmed.'
            : snapshot.route === 'direct'
            ? 'Direct — no child observed for this turn.'
            : snapshot.route === 'unconfirmed'
            ? 'Current route unconfirmed.'
            : 'Awaiting the current route.'}
        </p>
        {snapshot.route === 'nested' && (
          <dl>
            <div>
              <dt>Topic sent to child</dt>
              <dd>{snapshot.topic}</dd>
            </div>
            <div>
              <dt>Brief returned by child</dt>
              <dd>{snapshot.brief}</dd>
            </div>
          </dl>
        )}
        <p>
          The child shares only the topic and brief with the parent. Its
          internal stream remains separate from this conversation.
        </p>
      </section>
      <section aria-label="Child observations">
        <h2>Child observations</h2>
        {snapshot.children.length ? (
          snapshot.children.map((child) => (
            <div key={JSON.stringify(child.namespace)}>
              <p>Observed child stream</p>
              <ol>
                {child.namespace.map((segment, index) => (
                  <li key={index}>
                    <code>{segment}</code>
                  </li>
                ))}
              </ol>
            </div>
          ))
        ) : (
          <p>No child stream observed for this turn.</p>
        )}
        <p>
          Namespaces identify observed streams. They do not expose child
          execution status or controls.
        </p>
      </section>
      <ExampleFooter>
        {snapshot.error && <p role="alert">{snapshot.error}</p>}
        <ChatInput
          key={snapshot.viewGeneration}
          label="Message"
          placeholder="Ask a factual question or say hello…"
          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);
          }}
        />
        <p role="status">{status}</p>
      </ExampleFooter>
    </ExampleLayout>
  );
}

The Child observations panel displays the full namespace segments observed in the current child stream. An observation identifies a stream; it does not establish whether the child is running or complete, and it grants no child commands. Before the current result is confirmed, the route remains awaiting. Each new request clears the previous boundary and child observations.

Confirm the current result

The final parent answer records both the original question ID and its final answer ID. The application requires that exact pair, successful delivery for this request, and an idle safe session. A nested result also requires matching topic and brief from the current observed child. A direct result requires empty boundary fields and no child observed for this turn. Previous values and completion markers cannot confirm a new request.

application.ts

import type {
  AgentSession,
  AgentSnapshot,
  CompleteOutcome,
  Message,
} from '@threadplane/core';
import {
  createMessageContent,
  type MessageRow,
} from '@threadplane/content/messages';
 
interface Child {
  readonly namespace: readonly string[];
  readonly messages: readonly Message[];
  readonly values?: Readonly<Record<string, unknown>>;
  readonly interrupts: readonly { readonly value?: unknown }[];
  readonly error?: AgentSnapshot['error'];
}
export interface SubgraphsSession extends Omit<AgentSession, 'getSnapshot'> {
  getSnapshot(): AgentSnapshot & {
    readonly values?: Readonly<Record<string, unknown>>;
    readonly interrupts: readonly { readonly value?: unknown }[];
    readonly subgraphs: readonly Child[];
  };
}
export interface SubgraphsSnapshot {
  readonly threadId: string | null;
  readonly rows: readonly MessageRow[];
  readonly route: 'awaiting' | 'direct' | 'nested' | 'unconfirmed';
  readonly topic: string | null;
  readonly brief: string | null;
  readonly children: readonly { readonly namespace: readonly string[] }[];
  readonly activity: 'idle' | 'creating' | 'running' | 'replacing';
  readonly busy: boolean;
  readonly canSubmit: boolean;
  readonly outcome: CompleteOutcome | null;
  readonly error: string | null;
  readonly viewGeneration: number;
}
function own(
  values: Readonly<Record<string, unknown>> | undefined,
  key: string
) {
  return values && Object.hasOwn(values, key) ? values[key] : undefined;
}
function unexpectedTool(message: Message) {
  return (
    message.role === 'tool' ||
    message.toolCallId !== undefined ||
    !!message.toolCallIds?.length
  );
}
/** Owns parent conversation commands. Child observations never confer commands. */
export function createSubgraphsApplication(options: {
  readonly createThread: (signal: AbortSignal) => Promise<string>;
  readonly sessionFactory: (id: string) => SubgraphsSession;
}) {
  type ObservedChild = {
    namespace: readonly string[];
    topic: unknown;
    brief: unknown;
  };
  type Lifetime = {
    session: SubgraphsSession;
    content: ReturnType<typeof createMessageContent>;
    release?: () => void;
    closed: boolean;
    unsafe: boolean;
    turn?: { id: string; generation: string };
    previousIds: ReadonlySet<string>;
    previousNamespaces: ReadonlySet<string>;
    children: Map<string, ObservedChild>;
    confirmed: readonly Message[];
  };
  const listeners = new Set<() => void>(),
    cleanups = new Set<Promise<void>>();
  let snapshot: SubgraphsSnapshot = Object.freeze({
    threadId: null,
    rows: Object.freeze([]),
    route: 'awaiting',
    topic: null,
    brief: null,
    children: Object.freeze([]),
    activity: 'idle',
    busy: false,
    canSubmit: true,
    outcome: null,
    error: null,
    viewGeneration: 0,
  });
  let lifetime: Lifetime | undefined, operation: AbortController | undefined;
  let disposed = false,
    disposal: Promise<void> | undefined;
  function publish(update: Partial<SubgraphsSnapshot>) {
    if (disposed) return;
    snapshot = Object.freeze({ ...snapshot, ...update });
    for (const notify of listeners) notify();
  }
  function current(selected: Lifetime) {
    return (
      !disposed &&
      lifetime === selected &&
      !selected.closed &&
      !operation?.signal.aborted
    );
  }
  function read(selected: Lifetime, valid: () => boolean) {
    if (!valid()) return;
    const root = selected.session.getSnapshot();
    if (!valid()) return;
    // Capture external aggregate fields once before any application mutation.
    const state = {
      status: root.status,
      error: root.error,
      messages: root.messages,
      toolCalls: root.toolCalls,
      interrupts: root.interrupts,
      subgraphs: root.subgraphs,
      values: root.values,
    };
    if (!valid()) return;
    const children = state.subgraphs.map((child) => ({
      namespace: child.namespace,
      messages: child.messages,
      values: child.values,
      interrupts: child.interrupts,
      error: child.error,
    }));
    const blocked =
      !!state.error ||
      !!state.interrupts.length ||
      !!state.toolCalls.length ||
      state.messages.some(unexpectedTool) ||
      children.some(
        (child) =>
          !!child.error ||
          !!child.interrupts.length ||
          child.messages.some(unexpectedTool)
      );
    if (!valid()) return;
    return { ...state, subgraphs: children, blocked };
  }
  function capture(selected: Lifetime, valid: () => boolean) {
    const state = read(selected, valid);
    if (!state || !valid()) return;
    selected.unsafe ||= state.blocked;
    if (!selected.turn && snapshot.activity === 'running') {
      const human = [...state.messages]
        .reverse()
        .find(
          (message) =>
            message.role === 'user' && !selected.previousIds.has(message.id)
        );
      if (!valid()) return;
      if (human)
        selected.turn = { id: human.id, generation: human.delivery.generation };
    }
    const turn = selected.turn;
    if (turn && snapshot.activity === 'running') {
      for (const child of state.subgraphs) {
        const namespace = child.namespace;
        const wellFormed =
          Array.isArray(namespace) &&
          namespace.length > 0 &&
          namespace.every(
            (segment) => typeof segment === 'string' && segment.length > 0
          );
        if (!valid()) return;
        if (!wellFormed) {
          selected.unsafe = true;
          continue;
        }
        const key = JSON.stringify(namespace);
        if (selected.previousNamespaces.has(key)) continue;
        const topic = own(child.values, 'research_topic'),
          brief = own(child.values, 'research_brief');
        if (!valid()) return;
        if (
          child.values !== undefined &&
          (typeof topic !== 'string' || typeof brief !== 'string')
        )
          selected.unsafe = true;
        selected.children.set(
          key,
          Object.freeze({
            namespace: Object.freeze([...namespace]),
            topic,
            brief,
          })
        );
      }
    }
    const humanId = own(state.values, 'completed_turn_id'),
      answerId = own(state.values, 'completed_answer_id');
    const topic = own(state.values, 'research_topic'),
      brief = own(state.values, 'research_brief');
    if (!valid()) return;
    let final: readonly Message[] | undefined,
      route: 'direct' | 'nested' | undefined;
    if (
      turn &&
      !selected.unsafe &&
      state.status === 'idle' &&
      humanId === turn.id &&
      typeof answerId === 'string' &&
      answerId.length > 0 &&
      answerId !== turn.id &&
      typeof topic === 'string' &&
      typeof brief === 'string'
    ) {
      const humans = state.messages.filter((message) => message.id === turn.id);
      const answers = state.messages.filter(
        (message) => message.id === answerId
      );
      if (
        humans.length === 1 &&
        answers.length === 1 &&
        humans[0].role === 'user' &&
        answers[0].role === 'assistant' &&
        [humans[0], answers[0]].every(
          (message) =>
            message.delivery.generation === turn.generation &&
            message.delivery.phase === 'complete' &&
            message.delivery.outcome === 'success'
        )
      ) {
        const children = [...selected.children.values()];
        if (topic === '' && brief === '' && children.length === 0)
          route = 'direct';
        else if (
          topic.length > 0 &&
          brief.length > 0 &&
          children.length > 0 &&
          children.every(
            (child) => child.topic === topic && child.brief === brief
          )
        )
          route = 'nested';
        if (route) final = [humans[0], answers[0]];
      }
    }
    if (!valid()) return;
    const confirmedIds = new Set(
      selected.confirmed.map((message) => message.id)
    );
    const active = turn
      ? state.messages.filter(
          (message) =>
            !confirmedIds.has(message.id) &&
            message.delivery.generation === turn.generation
        )
      : [];
    const finalIds = final && new Set(final.map((message) => message.id));
    const messages = final
      ? [
          ...selected.confirmed.filter((message) => !finalIds!.has(message.id)),
          ...final,
        ]
      : [...selected.confirmed, ...active];
    const rows = selected.content.project({ ...state, messages });
    if (!valid()) return;
    const children = Object.freeze(
      [...selected.children.values()].map((child) =>
        Object.freeze({ namespace: child.namespace })
      )
    );
    return {
      state,
      rows,
      route,
      topic: typeof topic === 'string' ? topic : null,
      brief: typeof brief === 'string' ? brief : null,
      messages,
      children,
    };
  }
  function observe(selected: Lifetime) {
    const valid = () => current(selected),
      captured = capture(selected, valid);
    if (!captured || !valid()) return;
    publish({
      rows: captured.rows,
      children: captured.children,
      ...(captured.route
        ? {
            route: captured.route,
            topic: captured.topic,
            brief: captured.brief,
          }
        : {}),
      ...(selected.unsafe ? { canSubmit: false } : {}),
    });
  }
  function close(selected: Lifetime) {
    if (selected.closed) return Promise.resolve();
    selected.closed = true;
    selected.release?.();
    selected.content.dispose();
    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 = lifetime;
    lifetime = undefined;
    return previous ? close(previous) : Promise.resolve();
  }
  function install(id: string, valid: () => boolean) {
    const session = options.sessionFactory(id);
    const selected: Lifetime = {
      session,
      content: createMessageContent(),
      closed: false,
      unsafe: false,
      previousIds: new Set(),
      previousNamespaces: new Set(),
      children: new Map(),
      confirmed: [],
    };
    if (!valid()) {
      void close(selected);
      return;
    }
    lifetime = selected;
    const release = session.subscribe(() => observe(selected));
    selected.release = release;
    if (selected.closed) release();
    if (!valid() || lifetime !== selected) {
      if (!selected.closed) void close(selected);
      return;
    }
    return selected;
  }
  function admit(activity: SubgraphsSnapshot['activity']) {
    const admitted = new AbortController();
    operation = admitted;
    const valid = () =>
      !disposed && operation === admitted && !admitted.signal.aborted;
    publish({
      busy: true,
      canSubmit: false,
      activity,
      outcome: null,
      error: null,
      route: 'awaiting',
      topic: null,
      brief: null,
      children: Object.freeze([]),
    });
    return { admitted, valid };
  }
  function finish(admitted: AbortController) {
    if (!disposed && operation === admitted) {
      operation = undefined;
      publish({ busy: false, activity: 'idle' });
    }
  }
  async function submit(text: string) {
    if (disposed || snapshot.busy || !snapshot.canSubmit || !text.trim())
      return;
    const previous = lifetime;
    const initialValid = () =>
      !disposed &&
      !snapshot.busy &&
      snapshot.canSubmit &&
      lifetime === previous &&
      !previous?.closed;
    const before = previous && read(previous, initialValid);
    if (!initialValid()) return;
    if (previous && (previous.unsafe || before?.blocked)) {
      publish({ canSubmit: false });
      return;
    }
    if (previous && before) {
      const previousIds = new Set(
        before.messages
          .filter((message) => message.role === 'user')
          .map((message) => message.id)
      );
      const previousNamespaces = new Set(
        before.subgraphs.map((child) => JSON.stringify(child.namespace))
      );
      if (!initialValid()) return;
      // Reentrant observers of admission must not reuse the preceding turn.
      previous.turn = undefined;
      previous.previousIds = previousIds;
      previous.previousNamespaces = previousNamespaces;
      previous.children = new Map();
    }
    const { admitted, valid } = admit(lifetime ? 'running' : 'creating');
    try {
      if (!valid()) return;
      publish({
        route: 'awaiting',
        topic: null,
        brief: null,
        children: Object.freeze([]),
      });
      if (!valid()) return;
      if (!lifetime) {
        const id = await options.createThread(admitted.signal);
        if (!valid()) return;
        const selected = install(id, valid);
        if (!selected || !valid()) return;
        publish({ threadId: id, activity: 'running' });
      }
      if (!valid() || !lifetime) return;
      const selected = lifetime,
        ownedValid = () => valid() && current(selected);
      const state = read(selected, ownedValid);
      if (!state || !ownedValid()) return;
      if (state.blocked) {
        selected.unsafe = true;
        publish({ canSubmit: false });
        return;
      }
      selected.turn = undefined;
      selected.previousIds = new Set(
        state.messages
          .filter((message) => message.role === 'user')
          .map((message) => message.id)
      );
      selected.previousNamespaces = new Set(
        state.subgraphs.map((child) => JSON.stringify(child.namespace))
      );
      selected.children = new Map();
      if (!ownedValid()) return;
      const pending = selected.session.submit(text, {
        signal: admitted.signal,
      });
      if (!ownedValid()) return;
      observe(selected);
      const outcome = await pending;
      if (!ownedValid()) return;
      const captured = capture(selected, ownedValid);
      if (!captured || !ownedValid()) return;
      const safe =
        outcome === 'success' && !!captured.route && !selected.unsafe;
      if (safe) selected.confirmed = Object.freeze(captured.messages);
      publish({
        rows: captured.rows,
        children: captured.children,
        outcome,
        canSubmit: safe,
        route: safe ? captured.route! : 'unconfirmed',
        topic: safe ? captured.topic : null,
        brief: safe ? captured.brief : null,
        error: safe ? null : 'The current route could not be confirmed.',
      });
    } catch {
      if (valid())
        publish({
          canSubmit: false,
          route: 'unconfirmed',
          outcome: 'error',
          error: 'The LangGraph request failed.',
        });
    } finally {
      finish(admitted);
    }
  }
  async function newConversation() {
    if (disposed || snapshot.busy) return;
    const cleanup = detach(),
      { admitted, valid } = admit('replacing');
    try {
      if (!valid()) return;
      publish({
        threadId: null,
        rows: Object.freeze([]),
        route: 'awaiting',
        topic: null,
        brief: null,
        children: Object.freeze([]),
        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 = lifetime,
      cleanup = detach();
    publish({
      busy: false,
      activity: 'idle',
      canSubmit: false,
      route: 'unconfirmed',
      outcome: 'aborted',
    });
    try {
      await selected?.session.stop();
    } catch {
      /* Detached observations cannot grant authority. */
    }
    await cleanup;
  }
  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,
    newConversation,
    stop,
    dispose,
  };
}

Stop cancels active work and revokes its authority. Choose New conversation after stopping or an unconfirmed result. New clears both panels and the transcript; it creates a new thread only when you send a message. Unsupported pauses, tools, or errors block further messages in that conversation. Late results after cancellation or leaving the page cannot restore the old view.

Conditional composition and connection

The compiled research child runs as a parent graph node with a smaller state schema. The parent keeps its message reducer and clears the research fields when routing each request. Internal router output is not streamed into the chat.

graph.py

"""
LangGraph Subgraphs Graph — nested execution with a state boundary
 
Demonstrates LangGraph's composition primitive: a *compiled child graph added
to a parent graph as a plain node*. Three things make this a real subgraph
demo rather than a straight-line chain:
 
1. **A conditional route.** The `orchestrate` node classifies the request and
   the parent decides whether to enter the child graph at all. A greeting
   answers directly; a factual question routes through `research` first.
 
2. **A state boundary.** The parent state carries `messages`; the child state
   (`ResearchState`) does not. LangGraph wires a subgraph node through the
   keys the two schemas *share* — here `research_topic` and `research_brief` —
   so the child receives a topic and returns a brief, and can neither read the
   chat transcript nor append to it.
 
3. **Nested execution.** The child runs as its own graph with its own step
   sequence, and LangGraph emits its stream events under a namespace
   (`research:<uuid>`) rather than flattening them into the parent's.
 
**Two views of the child on the Angular side.** `agent.value()` reads the
parent's own state — the shared `research_topic` / `research_brief` keys are
the boundary made visible. `agent.subagents()` shows the child as a stream:
every namespaced child run appears there, plain subgraph nodes under their
namespace key (named by node) and tool-dispatched children under their
tool-call id (that shape is demonstrated in `cockpit/chat/subagents`).
"""
 
from pathlib import Path
from typing import Annotated, NotRequired, TypedDict
 
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field
 
PROMPTS_DIR = Path(__file__).parent.parent / "prompts"
 
# Gate phrase for the router's structured-output call. Kept distinct from the
# other two system prompts so e2e fixture replay can tell the calls apart.
ROUTER_PROMPT = (
    "Delegation Router.\n"
    "Decide whether answering the latest user turn requires a factual "
    "deep dive that should be handed to the research subgraph.\n"
    "Greetings, small talk, and questions about your own capabilities do "
    "NOT require research. Questions about facts, comparisons, or how a "
    "technology works DO.\n"
    "When research is required, restate the request as one focused topic "
    "line for the subgraph."
)
 
RESEARCH_PROMPT = (
    "Research Subgraph.\n"
    "You are a focused researcher running as a child graph. You are given a "
    "topic and nothing else — you cannot see the chat transcript, so never "
    "address the user or ask questions. Return 3-5 terse factual bullets. "
    "Your output is an internal brief the parent orchestrator will use to "
    "write the user-facing answer."
)
 
 
class DelegationDecision(BaseModel):
    """Structured routing decision produced by the `orchestrate` node."""
 
    needs_research: bool = Field(
        description="True when answering requires a factual deep dive that the research subgraph should handle."
    )
    topic: str = Field(
        default="",
        description="One focused topic line for the subgraph. Empty string when needs_research is false.",
    )
 
 
# region state
class ResearchState(TypedDict):
    """Child graph state — deliberately has no `messages` key.
 
    The only keys here are the ones shared with the parent, which is exactly
    the contract LangGraph uses to pass state into and out of a subgraph node.
    """
 
    research_topic: str
    research_brief: str
 
 
class OrchestratorState(TypedDict):
    """Parent graph state — the transcript plus the shared subgraph channel."""
 
    messages: Annotated[list, add_messages]
    research_topic: str
    research_brief: str
    # Final-answer identity binds the retained boundary to one original turn.
    completed_turn_id: NotRequired[str]
    completed_answer_id: NotRequired[str]
# endregion
 
 
def build_subgraphs_graph():
    """Constructs a parent graph that conditionally enters a child subgraph."""
    llm = ChatOpenAI(model="gpt-5-mini", streaming=True)
    router = ChatOpenAI(model="gpt-5-mini", tags=["nostream"]).with_structured_output(DelegationDecision)
    researcher = ChatOpenAI(model="gpt-5-mini")
 
    # region research-subgraph
    # ── Child: research subgraph ──────────────────────────────────────────────
 
    async def research_node(state: ResearchState) -> dict:
        """Turn a topic into an internal brief. No transcript access."""
        response = await researcher.ainvoke(
            [
                SystemMessage(content=RESEARCH_PROMPT),
                HumanMessage(content=f"Topic: {state['research_topic']}"),
            ]
        )
        text = response.content
        if isinstance(text, list):  # multi-part content
            text = "".join(part.get("text", "") for part in text if isinstance(part, dict))
        return {"research_brief": str(text).strip()}
 
    research_graph = StateGraph(ResearchState)
    research_graph.add_node("research", research_node)
    research_graph.add_edge(START, "research")
    research_graph.add_edge("research", END)
    compiled_research = research_graph.compile()
    # endregion
 
    # ── Parent: orchestrator graph ────────────────────────────────────────────
 
    # region orchestrate
    async def orchestrate_node(state: OrchestratorState) -> dict:
        """Classify the request. Writing a topic is what triggers delegation.
 
        Both shared keys are reset every turn so a topic left over from an
        earlier turn in the same thread can't re-trigger the subgraph.
        """
        decision = await router.ainvoke(
            [SystemMessage(content=ROUTER_PROMPT), *state["messages"]]
        )
        topic = decision.topic.strip() if decision.needs_research else ""
        return {"research_topic": topic, "research_brief": ""}
 
    def route_after_orchestrate(state: OrchestratorState) -> str:
        """The parent's decision: enter the child graph, or skip it."""
        return "research" if state.get("research_topic") else "answer"
    # endregion
 
    # region answer
    async def answer_node(state: OrchestratorState) -> dict:
        """The only node that writes to the transcript.
 
        `transcriptNodeNames: ['answer']` on the Angular side mirrors this:
        the router's and the subgraph's tokens never reach the chat UI.
        """
        system_prompt = (PROMPTS_DIR / "subgraphs.md").read_text()
        brief = state.get("research_brief") or ""
        context = (
            [SystemMessage(content=f"Internal brief returned by the research subgraph:\n{brief}")]
            if brief
            else []
        )
        response = await llm.ainvoke(
            [SystemMessage(content=system_prompt), *context, *state["messages"]]
        )
        question = next(
            (message for message in reversed(state["messages"])
             if getattr(message, "type", None) == "human"),
            None,
        )
        human_id = getattr(question, "id", None)
        answer_id = getattr(response, "id", None)
        completion = {}
        if (isinstance(human_id, str) and human_id
                and isinstance(answer_id, str) and answer_id
                and human_id != answer_id):
            completion = {
                "completed_turn_id": human_id,
                "completed_answer_id": answer_id,
            }
        return {"messages": [response], **completion}
    # endregion
 
    # region graph
    parent_graph = StateGraph(OrchestratorState)
    parent_graph.add_node("orchestrate", orchestrate_node)
    # The compiled child graph IS the node — no wrapper function. This is what
    # makes it a subgraph rather than an inline helper call.
    parent_graph.add_node("research", compiled_research)
    parent_graph.add_node("answer", answer_node)
    parent_graph.add_edge(START, "orchestrate")
    parent_graph.add_conditional_edges(
        "orchestrate",
        route_after_orchestrate,
        {"research": "research", "answer": "answer"},
    )
    parent_graph.add_edge("research", "answer")
    parent_graph.add_edge("answer", END)
    return parent_graph.compile()
    # endregion
 
 
# The graph instance — referenced by langgraph.json
graph = build_subgraphs_graph()

The shared backend is used by default. Developer runtime settings can provide a LangGraph target; credentials travel in SDK headers. Thread creation and execution use zero SDK retries. This preview does not expose child resume, retry, or recovery controls.

connection.ts

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 { createSubgraphsApplication } from './application';
export function subgraphsConnection(
  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 subgraphsConnection>
) {
  const client = new Client({
    apiUrl: connection.apiUrl,
    apiKey: null,
    defaultHeaders: connection.headers,
    callerOptions: { maxRetries: 0 },
  });
  return createSubgraphsApplication({
    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: 'subgraphs',
        threadId,
        apiUrl: connection.apiUrl,
        clientOptions: { maxRetries: 0, defaultHeaders: connection.headers },
      });
      try {
        const state = session.getSnapshot();
        if (
          !state ||
          !Array.isArray(state.subgraphs) ||
          !Array.isArray(state.interrupts)
        ) {
          throw new Error('Child observations are unavailable.');
        }
      } catch {
        void session.dispose().catch(() => undefined);
        throw new Error('Child observations are unavailable.');
      }
      return session;
    },
  });
}

Build from source

From a Threadplane checkout with dependencies installed:

npx nx build cockpit-langgraph-subgraphs-react
npx nx e2e cockpit-langgraph-subgraphs-react

The isolated build installs local candidate packages and checks strict declarations and emitted browser modules. Browser tests exercise real SDK HTTP and SSE against a local fixture, including conditional routing, namespace isolation, and cancellation. Docs, Code, and Run select this same React implementation. Streaming, interrupts, memory, client tools, persistence, durable execution, time travel and deployment runtime also have public React previews; AG-UI Streaming, Interrupts, Tool Views and JSON Render and Subagents also have React previews on Event Mapping and their canonical guides; other topics remain pending.