React deployment runtime 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 deploymentConnection,
} from './connection';
 
export function DeploymentDemo({
  connection,
  onReady,
}: {
  readonly connection: ReturnType<typeof deploymentConnection>;
  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 status = snapshot.busy
    ? snapshot.activity === 'creating'
      ? 'Creating conversation…'
      : snapshot.activity === 'replacing'
      ? 'Starting new conversation…'
      : 'Receiving response…'
    : !snapshot.canSubmit
    ? 'Start a new conversation to continue.'
    : snapshot.outcome === 'success'
    ? 'Response complete.'
    : 'Ready.';
  return (
    <ExampleLayout className="deployment-demo">
      <header>
        <div>
          <h1>LangGraph deployment runtime</h1>
        </div>
        <button
          type="button"
          disabled={snapshot.busy}
          onClick={() => void application.newConversation()}
        >
          New conversation
        </button>
      </header>
      <p>
        Send a message to the connected LangGraph runtime. Completed replies
        stay in this conversation until you start a new one.
      </p>
      <section aria-label="Runtime connection">
        <h2>Runtime connection</h2>
        <p>{connection.label}</p>
        <p>
          Assistant: <code>deployment-runtime</code>
        </p>
      </section>
      <MessageList rows={snapshot.rows} label="Conversation" />
      <ExampleFooter>
        {snapshot.error && <p role="alert">{snapshot.error}</p>}
        <ChatInput
          label="Message"
          placeholder="Ask a fictional deployment question…"
          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);
            if (!application.getSnapshot().busy) return false;
          }}
        />
        <p role="status">{status}</p>
      </ExampleFooter>
    </ExampleLayout>
  );
}

import { Client } from '@langchain/langgraph-sdk';
import { createSession } from '@threadplane/langgraph';
import {
  createDeploymentApplication,
  type DeploymentSession,
} 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 deploymentConnection(
  configuration: RuntimeBridgeConfiguration
): {
  apiUrl: string;
  headers: Record<string, string>;
  label: 'Shared runtime' | 'Developer runtime';
} {
  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 } : {},
    label: target ? 'Developer runtime' : 'Shared runtime',
  };
}
export function createConnectedApplication(
  connection: ReturnType<typeof deploymentConnection>
) {
  const client = new Client({
    apiUrl: connection.apiUrl,
    apiKey: null,
    defaultHeaders: connection.headers,
    callerOptions: { maxRetries: 0 },
  });
  return createDeploymentApplication({
    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: 'deployment-runtime',
        threadId,
        apiUrl: connection.apiUrl,
        clientOptions: { maxRetries: 0, defaultHeaders: connection.headers },
      });
      try {
        const state = session.getSnapshot();
        if (
          !state ||
          !Array.isArray(state.messages) ||
          !Array.isArray(state.toolCalls) ||
          !Array.isArray(state.interrupts) ||
          !['getSnapshot', 'subscribe', 'submit', '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 {
            /* A rejected candidate cannot become the active session. */
          }
        }
        throw new Error('Runtime session is unavailable.');
      }
      return session as DeploymentSession;
    },
  });
}

"""
LangGraph Deployment Runtime Graph
 
Standard chat graph demonstrating production deployment patterns.
The graph itself is intentionally simple — the focus of this example
is the deployment configuration, not a unique graph pattern.
 
Deploy with:
    langgraph deploy
 
Or run locally for development:
    langgraph dev
"""
 
from pathlib import Path
from typing import NotRequired
from langgraph.graph import StateGraph, MessagesState, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
 
PROMPTS_DIR = Path(__file__).parent.parent / "prompts"
 
 
class DeploymentRuntimeState(MessagesState):
    completed_turn_id: NotRequired[str]
    completed_answer_id: NotRequired[str]
    completed_message_ids: NotRequired[list[str]]
 
 
def build_deployment_runtime_graph():
    """
    Constructs a standard chat StateGraph suitable for production deployment.
 
    This graph is designed to be deployed via `langgraph deploy` to
    LangGraph Cloud. The assistantId in the Angular component must match
    the graph key in langgraph.json.
    """
    llm = ChatOpenAI(model="gpt-5-mini", streaming=True)
 
    async def generate(state: DeploymentRuntimeState) -> dict:
        """Generate a response using the full message history."""
        system_prompt = (PROMPTS_DIR / "deployment-runtime.md").read_text()
        messages = [SystemMessage(content=system_prompt)] + state["messages"]
        response = await llm.ainvoke(messages)
        result = {"messages": [response]}
        canonical = [*state["messages"], response]
        ids = [getattr(message, "id", None) for message in canonical]
        if (
            len(canonical) >= 2
            and getattr(canonical[-2], "type", None) == "human"
            and getattr(canonical[-1], "type", None) == "ai"
            and all(isinstance(identity, str) and identity for identity in ids)
            and len(ids) == len(set(ids))
        ):
            result["completed_turn_id"] = ids[-2]
            result["completed_answer_id"] = ids[-1]
            result["completed_message_ids"] = ids
        return result
 
    graph = StateGraph(DeploymentRuntimeState)
    graph.add_node("generate", generate)
    graph.set_entry_point("generate")
    graph.add_edge("generate", END)
    return graph.compile()
 
 
# The graph instance — referenced by langgraph.json
graph = build_deployment_runtime_graph()

import type { Message } from '@threadplane/core';
 
export interface OwnedTurn {
  readonly id: string;
  readonly generation: string;
  readonly prefix: readonly Message[];
}
 
function complete(message: Message) {
  return (
    typeof message.id === 'string' &&
    message.id.length > 0 &&
    (message.role === 'user' || message.role === 'assistant') &&
    message.toolCallId === undefined &&
    !message.toolCallIds?.length &&
    message.delivery.phase === 'complete' &&
    message.delivery.outcome === 'success'
  );
}
 
/** Only a newly admitted human can extend the preceding confirmed conversation. */
export function captureTurn(
  messages: readonly Message[],
  previousGenerations: ReadonlySet<string>,
  confirmed: readonly Message[],
  valid: () => boolean = () => true
): OwnedTurn | undefined {
  if (!valid()) return;
  const reverseIndex = [...messages]
    .reverse()
    .findIndex(
      (message) =>
        message.role === 'user' &&
        !previousGenerations.has(message.delivery.generation) &&
        message.delivery.generation !== message.id
    );
  if (!valid() || reverseIndex < 0) return;
  const index = messages.length - 1 - reverseIndex;
  const human = messages[index];
  if (!complete(human) || !valid()) return;
  const prefix = [...confirmed];
  if (
    prefix.some((message) => !complete(message)) ||
    new Set(prefix.map((message) => message.id)).size !== prefix.length ||
    prefix.some((message) => message.id === human.id) ||
    !valid()
  )
    return;
  return Object.freeze({
    id: human.id,
    generation: human.delivery.generation,
    prefix: Object.freeze(prefix),
  });
}
 
/** Exact prefix equality prevents a valid current pair from hiding missing history. */
export function confirmCanonicalHistory(
  messages: readonly Message[],
  values: Readonly<Record<string, unknown>> | undefined,
  turn: OwnedTurn,
  valid: () => boolean = () => true
): readonly Message[] | undefined {
  const read = (key: string) => {
    if (!valid()) return;
    return values && Object.hasOwn(values, key) ? values[key] : undefined;
  };
  const humanId = read('completed_turn_id');
  if (!valid()) return;
  const answerId = read('completed_answer_id');
  if (!valid()) return;
  const rawIds = read('completed_message_ids');
  if (
    !valid() ||
    humanId !== turn.id ||
    typeof answerId !== 'string' ||
    !answerId ||
    answerId === turn.id ||
    !Array.isArray(rawIds)
  )
    return;
  const ids = [...rawIds];
  const expected = [
    ...turn.prefix.map((message) => message.id),
    turn.id,
    answerId,
  ];
  if (
    ids.length !== expected.length ||
    !ids.every(
      (id, index) =>
        typeof id === 'string' && id.length > 0 && id === expected[index]
    ) ||
    new Set(ids).size !== ids.length ||
    !valid()
  )
    return;
  const canonical: Message[] = [];
  for (const id of ids) {
    const rows = messages.filter((message) => message.id === id);
    if (!valid() || rows.length !== 1 || !complete(rows[0])) return;
    canonical.push(rows[0]);
  }
  const human = canonical.at(-2)!,
    answer = canonical.at(-1)!;
  if (
    human.role !== 'user' ||
    answer.role !== 'assistant' ||
    human.delivery.generation !== turn.generation ||
    answer.delivery.generation !== turn.generation ||
    !valid()
  )
    return;
  return Object.freeze(canonical);
}

import type {
  AgentSession,
  AgentSnapshot,
  CompleteOutcome,
  Message,
} from '@threadplane/core';
import {
  createMessageContent,
  type MessageRow,
} from '@threadplane/content/messages';
import {
  captureTurn,
  confirmCanonicalHistory,
  type OwnedTurn,
} from './canonical-history';
 
export interface DeploymentSession extends Omit<AgentSession, 'getSnapshot'> {
  getSnapshot(): AgentSnapshot & {
    readonly values?: Readonly<Record<string, unknown>>;
    readonly interrupts: readonly unknown[];
  };
}
export interface DeploymentSnapshot {
  readonly threadId: string | null;
  readonly rows: readonly MessageRow[];
  readonly activity: 'idle' | 'creating' | 'running' | 'replacing';
  readonly busy: boolean;
  readonly canSubmit: boolean;
  readonly outcome: CompleteOutcome | null;
  readonly error: string | null;
  readonly viewGeneration: number;
}
 
/** Owns one confirmed conversation; only the exact current canonical result permits continuation. */
export function createDeploymentApplication(options: {
  readonly createThread: (signal: AbortSignal) => Promise<string>;
  readonly sessionFactory: (id: string) => DeploymentSession;
}) {
  type Primary = {
    session: DeploymentSession;
    closed: boolean;
    release?: () => void;
    content: ReturnType<typeof createMessageContent>;
    unsafe: boolean;
    confirmed: readonly Message[];
    turn?: OwnedTurn;
    previousGenerations: ReadonlySet<string>;
  };
  const listeners = new Set<() => void>(),
    cleanups = new Set<Promise<void>>();
  let primary: Primary | undefined;
  let operation: AbortController | undefined,
    disposed = false,
    disposal: Promise<void> | undefined;
  let snapshot: DeploymentSnapshot = Object.freeze({
    threadId: null,
    rows: Object.freeze([]),
    activity: 'idle',
    busy: false,
    canSubmit: true,
    outcome: null,
    error: null,
    viewGeneration: 0,
  });
  function publish(update: Partial<DeploymentSnapshot>) {
    if (disposed) return;
    snapshot = Object.freeze({ ...snapshot, ...update });
    for (const notify of listeners) notify();
  }
  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;
    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 old = primary;
    primary = undefined;
    return old ? close(old) : Promise.resolve();
  }
  function read(selected: Primary, valid: () => boolean) {
    if (!valid()) return;
    const root = selected.session.getSnapshot();
    if (!valid()) return;
    const status = root.status;
    if (!valid()) return;
    const error = root.error;
    if (!valid()) return;
    const messages = root.messages;
    if (!valid()) return;
    const toolCalls = root.toolCalls;
    if (!valid()) return;
    const interrupts = root.interrupts;
    if (!valid()) return;
    const values = root.values;
    if (!valid()) return;
    if (
      !Array.isArray(messages) ||
      !Array.isArray(toolCalls) ||
      !Array.isArray(interrupts)
    )
      throw new Error('The execution snapshot is unavailable.');
    const blocked =
      !!error ||
      !!toolCalls.length ||
      !!interrupts.length ||
      messages.some(
        (message) =>
          message.role === 'tool' ||
          message.toolCallId !== undefined ||
          !!message.toolCallIds?.length
      );
    if (!valid()) return;
    return { status, error, messages, toolCalls, interrupts, values, blocked };
  }
  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.messages,
        selected.previousGenerations,
        selected.confirmed,
        valid
      );
      if (!valid()) return;
      if (turn) selected.turn = turn;
    }
    const turn = selected.turn;
    const canonical =
      turn && !selected.unsafe && state.status === 'idle'
        ? confirmCanonicalHistory(state.messages, state.values, turn, valid)
        : undefined;
    if (!valid()) return;
    const messages =
      canonical ??
      (turn
        ? [
            ...turn.prefix,
            ...state.messages.filter(
              (message) => message.delivery.generation === turn.generation
            ),
          ]
        : selected.confirmed);
    if (!valid()) return;
    const rows = selected.content.project({ ...state, messages });
    if (!valid()) return;
    return { rows, canonical };
  }
  function observe(selected: Primary) {
    const valid = () => current(selected);
    try {
      const result = capture(selected, valid);
      if (!result || !valid()) return;
      publish({
        rows: result.rows,
        ...(selected.unsafe ? { canSubmit: false } : {}),
      });
    } catch {
      if (valid()) {
        selected.unsafe = true;
        publish({
          canSubmit: false,
          error: 'The execution snapshot could not be confirmed.',
        });
      }
    }
  }
  function install(id: string, valid: () => boolean) {
    const session = options.sessionFactory(id);
    const selected: Primary = {
      session,
      closed: false,
      content: createMessageContent(),
      unsafe: false,
      confirmed: [],
      previousGenerations: new Set(),
    };
    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: DeploymentSnapshot['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 previous = primary;
    const initialValid = () =>
      !disposed &&
      !snapshot.busy &&
      snapshot.canSubmit &&
      primary === previous &&
      !previous?.closed;
    let before: ReturnType<typeof read>;
    try {
      before = previous && read(previous, initialValid);
    } catch {
      if (initialValid())
        publish({
          canSubmit: false,
          error: 'The execution snapshot is unavailable.',
        });
      return;
    }
    if (!initialValid()) return;
    if (previous && (previous.unsafe || before?.blocked)) {
      publish({ canSubmit: false });
      return;
    }
    const generations = new Set(
      before?.messages.map((message) => message.delivery.generation)
    );
    if (!initialValid()) return;
    if (previous) {
      previous.turn = undefined;
      previous.previousGenerations = generations;
    }
    const { admitted, valid } = admit(primary ? 'running' : 'creating', true);
    try {
      if (!valid()) return;
      if (!primary) {
        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() || !primary) return;
      const selected = primary,
        ownedValid = () => valid() && current(selected);
      const state = read(selected, ownedValid);
      if (!state || !ownedValid()) return;
      if (state.blocked || selected.unsafe) {
        selected.unsafe = true;
        publish({ canSubmit: false });
        return;
      }
      selected.turn = undefined;
      selected.previousGenerations = new Set(
        state.messages.map((message) => message.delivery.generation)
      );
      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.canonical && !selected.unsafe;
      if (safe) selected.confirmed = result.canonical!;
      publish({
        rows: result.rows,
        outcome,
        canSubmit: safe,
        error: safe
          ? null
          : 'The current conversation history could not be confirmed. Start a new conversation.',
      });
    } catch {
      if (valid())
        publish({
          canSubmit: false,
          outcome: 'error',
          error: 'The LangGraph 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,
        rows: 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 cannot grant 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: (text: string) => execute(text),
    newConversation,
    stop,
    dispose,
  };
}