React durable execution preview

// eslint-disable-next-line @nx/enforce-module-boundaries -- The isolated consumer copies this internal example layout.
import { ExampleFooter, ExampleLayout } from '../../../../shared/react-example';
import { useEffect, useState } from 'react';
import { useAgent } from '@threadplane/react';
import { ChatInput, MessageList } from '@threadplane/react/chat';
import {
  createConnectedApplication,
  type durableConnection,
} from './connection';
export function DurableDemo({
  connection,
  onReady,
}: {
  readonly connection: ReturnType<typeof durableConnection>;
  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 === 'checking'
      ? 'Checking saved outcome…'
      : snapshot.activity === 'creating'
      ? 'Creating conversation…'
      : snapshot.activity === 'replacing'
      ? 'Starting new conversation…'
      : 'Receiving response…'
    : snapshot.outcome === 'aborted'
    ? 'Stopped. Start a new conversation to continue.'
    : snapshot.canCheck
    ? 'The outcome is uncertain. Check its saved status.'
    : !snapshot.canSubmit
    ? 'Start a new conversation to continue.'
    : snapshot.outcome === 'success'
    ? 'Response complete.'
    : 'Ready.';
  const latest = snapshot.checkpoints.find(
    (step) => step.id === snapshot.latestCheckpoint
  );
  return (
    <ExampleLayout className="durable-demo">
      <header>
        <div>
          <h1>LangGraph durable execution</h1>
        </div>
        <button
          type="button"
          disabled={snapshot.busy}
          onClick={() => void application.newConversation()}
        >
          New conversation
        </button>
      </header>
      <p>
        Follow a request through Analyze, Plan and Generate. Each completed node
        saves a checkpoint on the server.
      </p>
      <section
        className="saved-checkpoints"
        aria-label="Saved pipeline checkpoints"
      >
        <h2>Saved pipeline checkpoints</h2>
        <ol>
          {snapshot.checkpoints.map((step) => (
            <li key={step.id}>
              {step.label} — {step.status}
            </li>
          ))}
        </ol>
        <p>
          {latest
            ? `Latest checkpoint: ${latest.label}`
            : 'No checkpoint observed for this request yet.'}
        </p>
        <p>
          Checkpoints track completed nodes. A final reply replaces the
          intermediate drafts.
        </p>
      </section>
      <MessageList rows={snapshot.rows} label="Conversation" />
      {snapshot.canCheck && !snapshot.busy && (
        <button type="button" onClick={() => void application.checkStatus()}>
          Check status
        </button>
      )}
      <ExampleFooter>
        {snapshot.error && <p role="alert">{snapshot.error}</p>}
        <ChatInput
          key={snapshot.viewGeneration}
          label="Message"
          placeholder="Plan a fictional project…"
          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>
  );
}

import type {
  AgentSession,
  AgentSnapshot,
  CompleteOutcome,
} from '@threadplane/core';
import {
  createMessageContent,
  type MessageRow,
} from '@threadplane/content/messages';
 
export interface DurableSession extends Omit<AgentSession, 'getSnapshot'> {
  getSnapshot(): AgentSnapshot & {
    readonly values?: Readonly<Record<string, unknown>>;
    readonly interrupts: readonly { readonly value?: unknown }[];
  };
}
type Step = 'analyze' | 'plan' | 'generate';
const orderedSteps: readonly Step[] = ['analyze', 'plan', 'generate'];
function checkpoints(latest: Step | null) {
  const completed = latest === null ? -1 : orderedSteps.indexOf(latest);
  return Object.freeze(
    orderedSteps.map((id, index) =>
      Object.freeze({
        id,
        label: { analyze: 'Analyze', plan: 'Plan', generate: 'Generate' }[id],
        status:
          index <= completed ? ('complete' as const) : ('pending' as const),
      })
    )
  );
}
export interface DurableSnapshot {
  readonly threadId: string | null;
  readonly rows: readonly MessageRow[];
  readonly checkpoints: ReturnType<typeof checkpoints>;
  readonly latestCheckpoint: Step | null;
  readonly activity: 'idle' | 'creating' | 'running' | 'checking' | 'replacing';
  readonly busy: boolean;
  readonly canSubmit: boolean;
  readonly canCheck: boolean;
  readonly outcome: CompleteOutcome | null;
  readonly error: string | null;
  readonly viewGeneration: number;
}
/** Owns one confirmed thread; status reads never dispatch or replay work. */
export function createDurableApplication(options: {
  readonly createThread: (signal: AbortSignal) => Promise<string>;
  readonly sessionFactory: (id: string) => DurableSession;
}) {
  type Lifetime = {
    session: DurableSession;
    content: ReturnType<typeof createMessageContent>;
    release?: () => void;
    closed: boolean;
    unsafe: boolean;
    turn?: { id: string; generation: string };
    previousIds: ReadonlySet<string>;
  };
  const listeners = new Set<() => void>();
  const cleanups = new Set<Promise<void>>();
  let snapshot: DurableSnapshot = Object.freeze({
    threadId: null,
    rows: Object.freeze([]),
    checkpoints: checkpoints(null),
    latestCheckpoint: null,
    activity: 'idle',
    busy: false,
    canSubmit: true,
    canCheck: false,
    outcome: null,
    error: null,
    viewGeneration: 0,
  });
  let lifetime: Lifetime | undefined;
  let operation: AbortController | undefined;
  let disposed = false;
  let disposal: Promise<void> | undefined;
  function publish(update: Partial<DurableSnapshot>) {
    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 unsafe(state: ReturnType<DurableSession['getSnapshot']>) {
    return (
      !!state.interrupts.length ||
      !!state.toolCalls.length ||
      state.messages.some(
        (message) =>
          message.role === 'tool' ||
          message.toolCallId !== undefined ||
          !!message.toolCallIds?.length
      )
    );
  }
  function terminal(
    selected: Lifetime,
    state: ReturnType<DurableSession['getSnapshot']>
  ) {
    const turn = selected.turn;
    if (
      !turn ||
      selected.unsafe ||
      unsafe(state) ||
      state.error ||
      state.status !== 'idle' ||
      ownValue(state, 'step') !== 'generate' ||
      ownValue(state, 'completed_turn_id') !== turn.id
    )
      return;
    const answerId = ownValue(state, 'completed_answer_id');
    if (typeof answerId !== 'string' || !answerId || answerId === turn.id)
      return;
    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) return;
    const human = humans[0],
      answer = answers[0];
    if (
      human.role !== 'user' ||
      answer.role !== 'assistant' ||
      [human, answer].some(
        (message) =>
          message.delivery.generation !== turn.generation ||
          message.delivery.phase !== 'complete' ||
          message.delivery.outcome !== 'success'
      )
    )
      return;
    return [human, answer];
  }
  function ownValue(
    state: ReturnType<DurableSession['getSnapshot']>,
    key: string
  ) {
    const values = state.values;
    return values && Object.hasOwn(values, key) ? values[key] : undefined;
  }
  function checkable(
    selected: Lifetime,
    state: ReturnType<DurableSession['getSnapshot']>
  ) {
    return (
      !!selected.turn &&
      !selected.unsafe &&
      !unsafe(state) &&
      state.error?.recovery === 'check' &&
      typeof selected.session.checkStatus === 'function'
    );
  }
  function capture(selected: Lifetime, valid: () => boolean) {
    if (!valid()) return;
    const state = selected.session.getSnapshot();
    if (!valid()) return;
    const blocked = unsafe(state);
    if (!valid()) return;
    selected.unsafe ||= blocked;
    if (!selected.turn && snapshot.activity === 'running') {
      // The SDK admits a user row with successful complete delivery before I/O.
      // Its new identity and generation identify this turn, not delivery phase.
      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 finalMessages = terminal(selected, state);
    const complete = !!finalMessages;
    if (!valid()) return;
    const step = ownValue(state, 'step');
    if (!valid()) return;
    const latest: Step | null = complete
      ? 'generate'
      : selected.turn && (step === 'analyze' || step === 'plan')
      ? step
      : null;
    const rows = selected.content.project(
      finalMessages ? { ...state, messages: finalMessages } : state
    );
    if (!valid()) return;
    return { state, rows, complete, latest };
  }
  function observe(selected: Lifetime) {
    const valid = () => current(selected);
    const captured = capture(selected, valid);
    if (!captured || !valid()) return;
    const canCheck = !snapshot.busy && checkable(selected, captured.state);
    if (!valid()) return;
    publish({
      rows: captured.rows,
      latestCheckpoint: captured.latest,
      checkpoints: checkpoints(captured.latest),
      canCheck,
      ...(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(),
    };
    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: DurableSnapshot['activity']) {
    const admitted = new AbortController();
    operation = admitted;
    const valid = () =>
      !disposed && operation === admitted && !admitted.signal.aborted;
    publish({
      busy: true,
      canSubmit: false,
      canCheck: false,
      activity,
      outcome: null,
      error: null,
    });
    return { admitted, valid };
  }
  function finish(admitted: AbortController) {
    if (!disposed && operation === admitted) {
      operation = undefined;
      publish({ busy: false, activity: 'idle' });
    }
  }
  function settle(
    selected: Lifetime,
    valid: () => boolean,
    outcome?: CompleteOutcome
  ) {
    const captured = capture(selected, valid);
    if (!captured || !valid()) return;
    const safe =
      captured.complete && (outcome === undefined || outcome === 'success');
    const canCheck = !safe && checkable(selected, captured.state);
    if (!valid()) return;
    publish({
      rows: captured.rows,
      latestCheckpoint: captured.latest,
      checkpoints: checkpoints(captured.latest),
      canSubmit: safe,
      canCheck,
      outcome: safe ? 'success' : outcome ?? 'interrupted',
      error: safe ? null : 'The saved outcome could not be confirmed.',
    });
  }
  async function submit(text: string) {
    if (disposed || snapshot.busy || !snapshot.canSubmit || !text.trim())
      return;
    const previous = lifetime;
    const before = previous?.session.getSnapshot();
    if (
      disposed ||
      snapshot.busy ||
      !snapshot.canSubmit ||
      lifetime !== previous ||
      previous?.closed
    )
      return;
    const initiallyBlocked =
      previous && before && (previous.unsafe || unsafe(before) || before.error);
    // Safety fields are external reads too: replacement may occur in a getter.
    if (
      disposed ||
      snapshot.busy ||
      !snapshot.canSubmit ||
      lifetime !== previous ||
      previous?.closed
    )
      return;
    if (initiallyBlocked) {
      publish({ canSubmit: false });
      return;
    }
    const { admitted, valid } = admit(lifetime ? 'running' : 'creating');
    try {
      if (!valid()) return;
      publish({ checkpoints: checkpoints(null), latestCheckpoint: null });
      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;
      const state = selected.session.getSnapshot();
      if (!valid() || lifetime !== selected) return;
      selected.turn = undefined;
      selected.previousIds = new Set(
        state.messages
          .filter((message) => message.role === 'user')
          .map((message) => message.id)
      );
      if (!valid() || lifetime !== selected) return;
      const pending = selected.session.submit(text, {
        signal: admitted.signal,
      });
      if (!valid() || lifetime !== selected) return;
      observe(selected);
      const outcome = await pending;
      if (!valid() || lifetime !== selected) return;
      settle(selected, () => valid() && current(selected), outcome);
    } catch {
      if (valid())
        publish({
          canSubmit: false,
          canCheck: false,
          outcome: 'error',
          error: 'The LangGraph request failed.',
        });
    } finally {
      finish(admitted);
    }
  }
  async function checkStatus() {
    if (disposed || snapshot.busy || !snapshot.canCheck || !lifetime) return;
    const selected = lifetime,
      state = selected.session.getSnapshot();
    if (!current(selected) || snapshot.busy || !snapshot.canCheck) return;
    const allowed = checkable(selected, state);
    if (!current(selected) || snapshot.busy || !snapshot.canCheck) return;
    if (!allowed) {
      publish({ canCheck: false });
      return;
    }
    const { admitted, valid } = admit('checking');
    try {
      if (!valid() || !current(selected)) return;
      await selected.session.checkStatus!();
      if (valid() && current(selected))
        settle(selected, () => valid() && current(selected));
    } catch {
      if (valid() && current(selected)) {
        const captured = capture(selected, () => valid() && current(selected));
        if (!captured || !valid() || !current(selected)) return;
        const canCheck = checkable(selected, captured.state);
        if (!valid() || !current(selected)) return;
        publish({
          canSubmit: false,
          canCheck,
          error: 'The saved outcome could not be confirmed.',
        });
      }
    } 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([]),
        checkpoints: checkpoints(null),
        latestCheckpoint: 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 = lifetime,
      cleanup = detach();
    publish({
      busy: false,
      activity: 'idle',
      canSubmit: false,
      canCheck: false,
      outcome: 'aborted',
    });
    try {
      await selected?.session.stop();
    } catch {
      /* Detached commands cannot be reused. */
    }
    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,
    checkStatus,
    newConversation,
    stop,
    dispose,
  };
}

"""
LangGraph Durable Execution Graph
 
Demonstrates fault-tolerant multi-step execution with checkpointing.
The graph runs three nodes (analyze → plan → generate), saving state
after each one. If the process crashes mid-run, LangGraph can resume
from the last saved checkpoint rather than restarting from scratch.
 
The LangGraph API server provides checkpointing automatically,
enabling durable persistence across server restarts.
"""
 
from pathlib import Path
from typing import NotRequired, TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, convert_to_messages
 
PROMPTS_DIR = Path(__file__).parent.parent / "prompts"
 
 
# region state
class DurableState(TypedDict):
    messages: list
    step: str  # Most recently completed node
    completed_turn_id: NotRequired[str]  # Written only by the final node
    completed_answer_id: NotRequired[str]  # Paired with the final human identity
# endregion
 
 
def build_durable_execution_graph():
    """
    Constructs a multi-node StateGraph with checkpointing enabled.
 
    The graph runs three sequential nodes — analyze, plan, generate —
    and checkpoints state after each one. This means any node failure
    only requires replaying from the previous checkpoint, not the start.
    """
    llm = ChatOpenAI(model="gpt-5-mini", streaming=True)
    system_prompt = (PROMPTS_DIR / "durable-execution.md").read_text()
 
    # region analyze
    async def analyze(state: DurableState) -> dict:
        """
        Node 1: Analyze the user request.
 
        Examines the incoming messages to understand intent and key details.
        Checkpoints after completion so a failure in 'plan' doesn't rerun this.
        """
        messages = [SystemMessage(content=system_prompt + "\n\nYou are in the ANALYZE step. Briefly identify the key aspects of the request in 1-2 sentences.")] + state["messages"]
        response = await llm.ainvoke(messages)
        return {
            "messages": state["messages"] + [response],
            "step": "analyze",
        }
    # endregion
 
    async def plan(state: DurableState) -> dict:
        """
        Node 2: Plan a structured response.
 
        Uses the analysis from node 1 to outline the approach. State is
        checkpointed here before the more expensive generate step runs.
        """
        messages = [SystemMessage(content=system_prompt + "\n\nYou are in the PLAN step. Given the analysis so far, briefly outline the key points to address in 2-3 bullet points.")] + state["messages"]
        response = await llm.ainvoke(messages)
        return {
            "messages": state["messages"] + [response],
            "step": "plan",
        }
 
    # region generate
    async def generate(state: DurableState) -> dict:
        """
        Node 3: Generate the final response.
 
        Synthesises the analysis and plan into a complete, helpful reply.
        This is the last node before END; its output is the user-visible answer.
        """
        messages = [SystemMessage(content=system_prompt + "\n\nYou are in the GENERATE step. Synthesise the analysis and plan into a single, clear, helpful response for the user.")] + state["messages"]
        response = await llm.ainvoke(messages)
        # DurableState.messages has no reducer, so this return REPLACES the list.
        # Keep the user's original question (the newest human message) alongside
        # the final answer; the intermediate analyze/plan drafts are dropped.
        question = next(
            (m for m in reversed(convert_to_messages(state["messages"])) if m.type == "human"),
            None,
        )
        result = {
            "messages": ([question] if question is not None else []) + [response],
            "step": "generate",
        }
        question_id = getattr(question, "id", None)
        answer_id = getattr(response, "id", None)
        if (
            isinstance(question_id, str) and question_id
            and isinstance(answer_id, str) and answer_id
            and question_id != answer_id
        ):
            result["completed_turn_id"] = question_id
            result["completed_answer_id"] = answer_id
        return result
    # endregion
 
    # region graph
    graph = StateGraph(DurableState)
    graph.add_node("analyze", analyze)
    graph.add_node("plan", plan)
    graph.add_node("generate", generate)
 
    graph.set_entry_point("analyze")
    graph.add_edge("analyze", "plan")
    graph.add_edge("plan", "generate")
    graph.add_edge("generate", END)
 
    return graph.compile()
    # endregion
 
 
# The graph instance — referenced by langgraph.json
graph = build_durable_execution_graph()

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 { createDurableApplication } from './application';
export function durableConnection(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 durableConnection>
) {
  const client = new Client({
    apiUrl: connection.apiUrl,
    apiKey: null,
    defaultHeaders: connection.headers,
    callerOptions: { maxRetries: 0 },
  });
  return createDurableApplication({
    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: 'durable-execution',
        threadId,
        apiUrl: connection.apiUrl,
        clientOptions: { maxRetries: 0, defaultHeaders: connection.headers },
      });
      if (typeof session.checkStatus !== 'function') {
        void session.dispose().catch(() => undefined);
        throw new Error('Saved outcome reconciliation is unavailable.');
      }
      return session;
    },
  });
}