React chat A2UI 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, useRef, useState } from 'react';
import { useAgent } from '@threadplane/react';
import {
  ChatInput,
  MessageList,
  type ChatInputHandle,
} from '@threadplane/react/chat';
import type { A2uiApplication } from './application';
// eslint-disable-next-line @nx/enforce-module-boundaries -- Same-layout private presentation source in installed consumers.
import { A2uiSurface } from '../../../../../libs/cockpit-a2ui-react/src/index';
 
function humanText(content: string) {
  try {
    const value = JSON.parse(content);
    if (value.version === 'v0.9') {
      const labels: Record<string, string> = {
        bookingSubmit: 'Search flights',
        flightSelect: 'Select demonstration flight',
        modifySearch: 'Modify search',
      };
      if (Object.hasOwn(labels, value.action?.name))
        return labels[value.action.name];
    }
  } catch {
    /* Ordinary human text. */
  }
  return content;
}
function SavedText({ content }: { readonly content: string }) {
  const at = content.indexOf('---a2ui_JSON---');
  return at < 0 ? (
    <p className="a2ui-prose">{content}</p>
  ) : (
    <>
      {content.slice(0, at).trim() && (
        <p className="a2ui-prose">{content.slice(0, at).trim()}</p>
      )}
      <details className="a2ui-wire">
        <summary>Response details</summary>
        <pre>{content.slice(at)}</pre>
      </details>
    </>
  );
}
/** Main owns the application. React only observes it and borrows its commands. */
export function A2uiDemo({
  application,
  onReady,
}: {
  readonly application: A2uiApplication;
  readonly onReady: () => void;
}) {
  const snapshot = useAgent(application);
  const [draft, setDraft] = useState('');
  const composer = useRef<ChatInputHandle>(null);
  useEffect(onReady, [onReady]);
  const status = snapshot.busy
    ? snapshot.activity === 'creating'
      ? 'Creating conversation…'
      : snapshot.activity === 'confirming'
      ? 'Confirming saved response…'
      : snapshot.activity === 'replacing'
      ? 'Starting a new conversation…'
      : 'Waiting for the saved response…'
    : snapshot.error
    ? 'Start a new conversation to continue.'
    : snapshot.messages.length
    ? 'Response saved.'
    : 'Ready.';
  return (
    <ExampleLayout className="a2ui-demo">
      <header>
        <div>
          <h1>Plan a flight</h1>
        </div>
        <button
          type="button"
          onClick={() => {
            setDraft('');
            void application.newConversation().catch(() => undefined);
          }}
        >
          New conversation
        </button>
      </header>
      <p>
        Flight preview: request a flight form, edit your details, and select a
        demonstration flight. This preview makes no booking or payment.
      </p>
      {snapshot.messages.length === 0 && (
        <button
          className="a2ui-suggestion"
          type="button"
          disabled={snapshot.busy || !snapshot.canSubmit}
          onClick={() => {
            setDraft('I want to fly LAX to JFK');
            composer.current?.focus();
          }}
        >
          Try a flight request
        </button>
      )}
      <MessageList
        rows={[]}
        label="Conversation"
        className="a2ui-conversation"
        key={`messages:${snapshot.viewGeneration}`}
      >
        {!snapshot.messages.length && (
          <p className="a2ui-empty">
            Your saved flight previews will appear here.
          </p>
        )}
        {snapshot.messages.map((message) => {
          const surface = snapshot.surfaces.find(
            (s) => s.messageId === message.id
          );
          return (
            <article
              key={message.id}
              className={`tp-chat-message tp-chat-message--${message.role} a2ui-message a2ui-message--${message.role}`}
              aria-label={message.role === 'user' ? 'You' : 'Assistant'}
            >
              <p className="tp-chat-message__role a2ui-message-role">
                {message.role === 'user' ? 'You' : 'Assistant'}
              </p>
              {surface ? (
                <A2uiSurface
                  surface={surface}
                  active={
                    snapshot.activeSurface === surface.id &&
                    !snapshot.busy &&
                    snapshot.canSubmit
                  }
                />
              ) : message.role === 'user' ? (
                <p className="a2ui-prose">{humanText(message.content)}</p>
              ) : (
                <SavedText content={message.content} />
              )}
            </article>
          );
        })}
      </MessageList>
      <ExampleFooter>
        {snapshot.error && <p role="alert">{snapshot.error}</p>}
        <ChatInput
          ref={composer}
          key={snapshot.viewGeneration}
          label="Message"
          value={draft}
          onValueChange={setDraft}
          placeholder="Ask for a flight form…"
          hint="Enter to send. Shift+Enter adds a new line."
          busy={snapshot.busy}
          disabled={!snapshot.canSubmit && !snapshot.busy}
          onStop={
            snapshot.busy
              ? () => {
                  void application.stop().catch(() => undefined);
                }
              : undefined
          }
          onSubmit={(text) => {
            const state = application.getSnapshot();
            if (state.busy || !state.canSubmit) return false;
            // ChatInput clears this admitted draft synchronously, independently of settlement.
            void application.submit(text).catch(() => undefined);
            return true;
          }}
        />
        <p role="status">{status}</p>
      </ExampleFooter>
    </ExampleLayout>
  );
}

import { StrictMode } from 'react';
import { createRoot, type Root } from 'react-dom/client';
// eslint-disable-next-line @nx/enforce-module-boundaries -- The isolated consumer copies this private bridge.
import {
  GENERATED_RUNTIME_PARENT_ORIGINS,
  installRuntimeBridge,
  type InstalledRuntimeBridge,
} from '../../../../../libs/cockpit-runtime-bridge/src/index';
import { A2uiDemo } from './app';
import {
  a2uiConnection,
  createConnectedApplication,
  type A2uiConnection,
} from './connection';
import type { A2uiApplication } from './application';
import '@threadplane/react/chat/styles.css';
import './styles.css';
// eslint-disable-next-line @nx/enforce-module-boundaries -- The isolated consumer copies this internal example layout.
import '../../../../shared/react-example.css';
 
/** Own bootstrap and teardown outside React's replayable presentation lifetime. */
export function startA2uiDemo(
  element: HTMLElement,
  dependencies: {
    readonly bridge?: InstalledRuntimeBridge;
    readonly root?: Pick<Root, 'render' | 'unmount'>;
    readonly createApplication?: (
      connection: A2uiConnection
    ) => A2uiApplication;
  } = {}
) {
  const bridge =
    dependencies.bridge ??
    installRuntimeBridge(undefined, {
      allowedParentOrigins: GENERATED_RUNTIME_PARENT_ORIGINS,
    });
  const root = dependencies.root ?? createRoot(element);
  const createApplication =
    dependencies.createApplication ?? createConnectedApplication;
  let active = true,
    markedReady = false,
    application: A2uiApplication | undefined,
    cleanup: Promise<void> | undefined;
  function dispose(): Promise<void> {
    if (!active) return cleanup ?? Promise.resolve();
    active = false;
    window.removeEventListener('pagehide', pagehide);
    // Revoke work/stop first, then remove borrowed views, then dispose owned data.
    application?.revoke();
    root.unmount();
    cleanup = Promise.resolve(application?.dispose())
      .catch(() => undefined)
      .then(() => {
        bridge.dispose();
      });
    return cleanup;
  }
  function pagehide() {
    void dispose();
  }
  window.addEventListener('pagehide', pagehide);
  const ready = bridge
    .awaitConfiguration()
    .then(async (configuration) => {
      if (!active) return;
      const connection = a2uiConnection(configuration);
      application = createApplication(connection);
      if (!active) {
        application.revoke();
        await application.dispose();
        return;
      }
      root.render(
        <StrictMode>
          <A2uiDemo
            application={application}
            onReady={() => {
              if (active && !markedReady) {
                markedReady = true;
                bridge.markReady();
              }
            }}
          />
        </StrictMode>
      );
    })
    .catch(() => {
      if (!active) return;
      bridge.markError('bootstrap_failed');
      root.render(<p role="alert">The flight preview could not start.</p>);
    });
  return Object.freeze({ ready, dispose });
}
const element = document.getElementById('root');
if (element) startA2uiDemo(element);

import { Client } from '@langchain/langgraph-sdk';
import { createSession } from '@threadplane/langgraph';
import {
  createA2uiApplication,
  type A2uiClient,
  type A2uiSession,
} from './application';
// eslint-disable-next-line @nx/enforce-module-boundaries -- Same-layout private data source in installed consumers.
import { copyData } from '../../../../../libs/cockpit-a2ui-react/src/data';
import { checkpointSource, type Checkpoint } from './terminal';
// eslint-disable-next-line @nx/enforce-module-boundaries -- Internal bridge is copied by the isolated build.
import type { RuntimeBridgeConfiguration } from '../../../../../libs/cockpit-runtime-bridge/src/index';
export interface A2uiConnection {
  readonly apiUrl: string;
  readonly headers: Record<string, string>;
}
/** Resolve the shared proxy or the explicitly authored developer target. */
export function a2uiConnection(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 } : {},
  };
}
/** Connect only this page's confirmed IDs, with retries disabled for owned requests. */
export function createA2uiClient(connection: A2uiConnection): A2uiClient {
  connection = Object.freeze({
    apiUrl: connection.apiUrl,
    headers: Object.freeze({ ...connection.headers }),
  });
  const known = new Set<string>();
  const client = new Client({
    apiUrl: connection.apiUrl,
    apiKey: null,
    defaultHeaders: connection.headers,
    callerOptions: { maxRetries: 0 },
  });
  return {
    async createThread(signal) {
      try {
        if (signal.aborted) throw new Error('Aborted');
        const threadId = crypto.randomUUID();
        const created = await client.threads.create({
          threadId,
          ifExists: 'raise',
          signal,
        });
        if (signal.aborted || created.thread_id !== threadId)
          throw new Error('Unconfirmed');
        known.add(threadId);
        return threadId;
      } catch {
        throw new Error('Conversation creation was not confirmed.');
      }
    },
    sessionFactory(threadId) {
      if (!known.has(threadId)) throw new Error('Unknown conversation.');
      let session: ReturnType<typeof createSession> | undefined;
      try {
        session = createSession({
          assistantId: 'c-a2ui',
          threadId,
          apiUrl: connection.apiUrl,
          clientOptions: { maxRetries: 0, defaultHeaders: connection.headers },
        });
        const selected = session;
        const state = selected.getSnapshot();
        if (
          !state ||
          !Array.isArray(state.messages) ||
          !Array.isArray(state.toolCalls) ||
          !Array.isArray(state.interrupts) ||
          !Array.isArray(state.subgraphs) ||
          ![
            'getSnapshot',
            'subscribe',
            'submit',
            'load',
            'stop',
            'dispose',
          ].every(
            (method) =>
              typeof selected[method as keyof typeof selected] === 'function'
          )
        )
          throw new Error('Rejected session');
        return session as A2uiSession;
      } catch {
        try {
          void Promise.resolve(session?.dispose()).catch(() => undefined);
        } catch {
          /* Rejected sessions cannot be installed. */
        }
        throw new Error('Runtime session is unavailable.');
      }
    },
    async readCheckpoint(requested, signal) {
      try {
        const captured = copyData(requested) as unknown as Checkpoint;
        const source = checkpointSource(captured, captured.thread_id);
        if (!source || !known.has(source.thread_id) || signal.aborted)
          throw new Error('Unavailable');
        // SDK declares checkpoint_map required-but-possibly-undefined. Preserve its absence on the wire.
        const state = await client.threads.getState(
          source.thread_id,
          source as Parameters<typeof client.threads.getState>[1],
          { signal }
        );
        if (signal.aborted) throw new Error('Aborted');
        return state;
      } catch {
        throw new Error('Checkpoint preview is unavailable.');
      }
    },
  };
}
 
export function createConnectedApplication(connection: A2uiConnection) {
  return createA2uiApplication(createA2uiClient(connection));
}

import type { AgentSession, AgentSnapshot, Message } from '@threadplane/core';
import {
  createObservation,
  prepareTerminal,
  type Checkpoint,
  type Observation,
} from './terminal';
// Same-layout private source import is copied unchanged into installed consumers.
// eslint-disable-next-line @nx/enforce-module-boundaries -- Internal package-less aviation presentation.
import {
  createFlightSurface,
  type FlightSurface,
  type SurfaceView,
  type SurfaceKind,
} from '../../../../../libs/cockpit-a2ui-react/src/index';
 
export interface A2uiSession extends Omit<AgentSession, 'getSnapshot'> {
  getSnapshot(): AgentSnapshot & {
    readonly interrupts: readonly unknown[];
    readonly subgraphs: readonly unknown[];
    readonly history: readonly unknown[] | undefined;
  };
  load(options?: { readonly signal?: AbortSignal }): Promise<void>;
}
export interface A2uiClient {
  createThread(signal: AbortSignal): Promise<string>;
  sessionFactory(threadId: string): A2uiSession;
  readCheckpoint(checkpoint: Checkpoint, signal: AbortSignal): Promise<unknown>;
}
export interface A2uiSnapshot {
  readonly threadId: string | null;
  readonly messages: readonly Message[];
  readonly surfaces: readonly SurfaceView[];
  readonly activeSurface: string | null;
  readonly activity:
    | 'idle'
    | 'creating'
    | 'running'
    | 'confirming'
    | 'replacing';
  readonly busy: boolean;
  readonly canSubmit: boolean;
  readonly error: string | null;
  readonly viewGeneration: number;
}
type SurfaceOwner = FlightSurface;
type Conversation = {
  id: string;
  session: A2uiSession;
  release?: () => void;
  observation?: Observation;
  closed: boolean;
};
/** No streamed assistant row can create presentation or interactive authority. */
export function createA2uiApplication(client: A2uiClient) {
  let conversation: Conversation | undefined,
    operation: AbortController | undefined;
  let active: SurfaceOwner | undefined,
    sequence = 0,
    revoked = false;
  let disposal: Promise<void> | undefined;
  const surfaces = new Set<SurfaceOwner>(),
    listeners = new Set<{ notify: () => void }>();
  const cleanups = new Set<Promise<void>>();
  let snapshot: A2uiSnapshot = Object.freeze({
    threadId: null,
    messages: Object.freeze([]),
    surfaces: Object.freeze([]),
    activeSurface: null,
    activity: 'idle',
    busy: false,
    canSubmit: true,
    error: null,
    viewGeneration: 0,
  });
  function publish(change: Partial<A2uiSnapshot>) {
    if (revoked) return;
    snapshot = Object.freeze({ ...snapshot, ...change });
    for (const listener of [...listeners]) {
      if (revoked) break;
      if (listeners.has(listener))
        try {
          listener.notify();
        } catch {
          /* Observation cannot reject admission. */
        }
    }
  }
  function retire(owner: SurfaceOwner) {
    owner.dispose();
  }
  function revokeActive() {
    if (active) retire(active);
    active = undefined;
  }
  function track(cleanup: Promise<void>) {
    cleanups.add(cleanup);
    void cleanup.then(() => cleanups.delete(cleanup));
    return cleanup;
  }
  function detach(stop: boolean) {
    const selected = conversation;
    conversation = undefined;
    if (!selected || selected.closed) return Promise.resolve();
    selected.closed = true;
    selected.observation = undefined;
    selected.release?.();
    let stopped: Promise<unknown>;
    try {
      stopped = stop
        ? Promise.resolve(selected.session.stop())
        : Promise.resolve();
    } catch {
      stopped = Promise.resolve();
    }
    return track(
      stopped
        .catch(() => undefined)
        .then(async () => {
          try {
            await selected.session.dispose();
          } catch {
            /* An uncertain operation is never replayed. */
          }
        })
    );
  }
  function prepare(content: string, kind: SurfaceKind, messageId: string) {
    const owner: SurfaceOwner = createFlightSurface({
      content,
      expected: kind,
      messageId,
      id: 'surface-' + ++sequence,
      isCurrent: () =>
        !revoked && active === owner && !snapshot.busy && snapshot.canSubmit,
      onAction(payload) {
        const next: SurfaceKind =
          payload.action.name === 'bookingSubmit'
            ? 'results'
            : payload.action.name === 'flightSelect'
            ? 'confirmation'
            : 'booking';
        // submit closes the epoch synchronously before transport or observer reentry.
        void submit(JSON.stringify(payload), next).catch(() => undefined);
      },
    });
    surfaces.add(owner);
    return owner;
  }
  async function submit(
    text: string,
    expected: SurfaceKind = 'booking'
  ): Promise<boolean> {
    if (
      revoked ||
      snapshot.busy ||
      !snapshot.canSubmit ||
      typeof text !== 'string' ||
      !text.trim()
    )
      return false;
    const admitted = new AbortController();
    operation = admitted;
    // Reentrant observers see the closed epoch and the same busy operation.
    revokeActive();
    const current = () =>
      !revoked && operation === admitted && !admitted.signal.aborted;
    publish({
      busy: true,
      canSubmit: false,
      activeSurface: null,
      activity: conversation ? 'running' : 'creating',
      error: null,
    });
    try {
      if (!current()) return false;
      if (!conversation) {
        const id = await client.createThread(admitted.signal);
        if (!current()) return false;
        if (typeof id !== 'string' || !id)
          throw Error('Unconfirmed conversation');
        const session = client.sessionFactory(id);
        if (!current()) {
          await track(
            Promise.resolve()
              .then(() => session.dispose())
              .catch(() => undefined)
          );
          return false;
        }
        const selected: Conversation = { id, session, closed: false };
        conversation = selected;
        const release = session.subscribe(() => {
          if (
            !revoked &&
            operation &&
            !operation.signal.aborted &&
            conversation === selected &&
            !selected.closed &&
            snapshot.activity === 'running'
          )
            selected.observation?.observe(session.getSnapshot());
        });
        selected.release = release;
        if (selected.closed) release();
        if (!current()) return false;
        publish({ threadId: id, activity: 'running' });
      }
      const selected = conversation;
      if (!current() || !selected || selected.closed) return false;
      const valid = () =>
        current() && conversation === selected && !selected.closed;
      const observation = createObservation({
        threadId: selected.id,
        input: text,
        previous: snapshot.messages,
      });
      selected.observation = observation;
      const outcome = await selected.session.submit(text, {
        signal: admitted.signal,
      });
      if (!valid()) return false;
      publish({ activity: 'confirming' });
      const terminal = await prepareTerminal({
        operation: admitted,
        observation,
        outcome,
        getOperation: () => (valid() ? operation : undefined),
        getSnapshot: () => selected.session.getSnapshot(),
        load: () => selected.session.load({ signal: admitted.signal }),
        read: (cp) => client.readCheckpoint(cp, admitted.signal),
      });
      if (!valid()) return false;
      if (terminal.kind !== 'prepared') throw Error('Unconfirmed response');
      const messages = terminal.confirmation.messages;
      const last = messages.at(-1)!;
      let prepared: SurfaceOwner;
      try {
        prepared = prepare(last.content, expected, last.id);
      } catch {
        // The saved response remains inspectable as text, with no partial executable tree.
        publish({
          messages,
          error: 'This response is unavailable. Start a new conversation.',
          canSubmit: false,
        });
        return false;
      }
      if (!valid()) {
        retire(prepared);
        return false;
      }
      active = prepared;
      selected.observation = undefined;
      operation = undefined;
      // Transcript, view and authority become observable in one immutable publication.
      publish({
        messages,
        surfaces: Object.freeze([...snapshot.surfaces, prepared.view]),
        activeSurface: prepared.view.id,
        busy: false,
        activity: 'idle',
        canSubmit: true,
        error: null,
      });
      return true;
    } catch {
      if (current())
        publish({
          canSubmit: false,
          error:
            'The response could not be confirmed. Start a new conversation.',
        });
      return false;
    } finally {
      if (current()) {
        operation = undefined;
        if (conversation) conversation.observation = undefined;
        publish({ busy: false, activity: 'idle' });
      }
    }
  }
  async function stop() {
    if (revoked || !operation) return;
    operation.abort();
    operation = undefined;
    revokeActive();
    const cleanup = detach(true);
    publish({
      busy: false,
      activity: 'idle',
      activeSurface: null,
      canSubmit: false,
      error: 'Stopped. Start a new conversation to continue.',
    });
    await cleanup;
  }
  async function newConversation() {
    if (revoked) return;
    const running = !!operation;
    operation?.abort();
    revokeActive();
    const admitted = new AbortController();
    operation = admitted;
    const cleanup = detach(running);
    for (const owner of surfaces) retire(owner);
    surfaces.clear();
    publish({
      threadId: null,
      messages: Object.freeze([]),
      surfaces: Object.freeze([]),
      activeSurface: null,
      busy: true,
      activity: 'replacing',
      canSubmit: false,
      error: null,
      viewGeneration: snapshot.viewGeneration + 1,
    });
    await cleanup;
    if (revoked || operation !== admitted || admitted.signal.aborted) return;
    operation = undefined;
    publish({ busy: false, activity: 'idle', canSubmit: true });
  }
  /** Main calls this before unmount; owner disposal occurs afterward. */
  function revoke() {
    if (revoked) return;
    revoked = true;
    const running = !!operation;
    operation?.abort();
    operation = undefined;
    for (const owner of surfaces) owner.revoke();
    conversation?.release?.();
    if (conversation) {
      conversation.release = undefined;
      conversation.observation = undefined;
    }
    if (running && conversation) {
      try {
        track(
          Promise.resolve(conversation.session.stop()).catch(() => undefined)
        );
      } catch {
        /* Cleanup continues. */
      }
    }
    listeners.clear();
  }
  function dispose(): Promise<void> {
    if (disposal) return disposal;
    revoke();
    for (const owner of surfaces) retire(owner);
    surfaces.clear();
    active = undefined;
    void detach(false);
    disposal = Promise.all([...cleanups]).then(() => undefined);
    return disposal;
  }
  return Object.freeze({
    getSnapshot: () => snapshot,
    subscribe(notify: () => void) {
      if (revoked) return () => undefined;
      const registration = { notify };
      listeners.add(registration);
      return () => {
        listeners.delete(registration);
      };
    },
    submit: (text: string) => submit(text),
    stop,
    newConversation,
    revoke,
    dispose,
  });
}
export type A2uiApplication = ReturnType<typeof createA2uiApplication>;

import type { Message } from '@threadplane/core';
// Keep the terminal proof transport-only: the private leaf avoids the React view barrel.
// eslint-disable-next-line @nx/enforce-module-boundaries -- Same-layout copied private data leaf.
import { copyData, equalData, plainRecord, closedFields } from '../../../../../libs/cockpit-a2ui-react/src/data.js';
 
export interface Checkpoint {
  readonly thread_id: string;
  readonly checkpoint_ns: '';
  readonly checkpoint_id: string;
  readonly checkpoint_map?: Readonly<Record<string, string>>;
}
interface Semantic {
  readonly id: string;
  readonly role: 'user' | 'assistant';
  readonly content: string;
  readonly name: string | undefined;
  readonly reasoning: string | undefined;
}
export interface Owned {
  readonly threadId: string;
  readonly previous: readonly Semantic[];
  readonly human: Semantic;
  readonly generation: string;
}
export interface Confirmation {
  readonly checkpoint: Checkpoint;
  readonly messages: readonly Message[];
}
const empty = (value: unknown) =>
  value === undefined ||
  value === null ||
  (Array.isArray(value) && value.length === 0);
const text = (value: unknown): value is string =>
  typeof value === 'string' && value.length > 0;
function clean(value: Record<string, unknown>): boolean {
  return (
    (value.error === undefined || value.error === null) &&
    (value.pause === undefined || value.pause === null) &&
    (value.paused === undefined || value.paused === false) &&
    ['interrupts', '__interrupt__', 'subgraphs', 'tasks', 'next'].every((k) =>
      empty(value[k])
    )
  );
}
function semantic(input: unknown): Semantic | undefined {
  if (
    !plainRecord(input) ||
    !closedFields(input, [
      'id',
      'role',
      'content',
      'delivery',
      'name',
      'reasoning',
      'citations',
      'toolCallId',
      'toolCallIds',
    ]) ||
    !text(input.id) ||
    !['user', 'assistant'].includes(input.role as string) ||
    typeof input.content !== 'string' ||
    !empty(input.toolCallId) ||
    !empty(input.toolCallIds) ||
    !empty(input.citations) ||
    (input.name !== undefined && typeof input.name !== 'string') ||
    (input.reasoning !== undefined && typeof input.reasoning !== 'string')
  )
    return;
  return Object.freeze({
    id: input.id,
    role: input.role as 'user' | 'assistant',
    content: input.content,
    name: input.name as string | undefined,
    reasoning: input.reasoning as string | undefined,
  });
}
function canonical(input: unknown): readonly Message[] | undefined {
  if (!Array.isArray(input)) return;
  const seen = new Set<string>();
  for (const m of input) {
    const s = semantic(m);
    if (
      !s ||
      seen.has(s.id) ||
      !plainRecord(m.delivery) ||
      !closedFields(m.delivery, ['generation', 'phase', 'outcome']) ||
      !text(m.delivery.generation) ||
      m.delivery.phase !== 'complete' ||
      m.delivery.outcome !== 'success'
    )
      return;
    seen.add(s.id);
  }
  return input as readonly Message[];
}
export function checkpointSource(
  input: unknown,
  threadId: string
): Checkpoint | undefined {
  try {
    const cp = copyData(input);
    if (
      !plainRecord(cp) ||
      !closedFields(cp, [
        'thread_id',
        'checkpoint_ns',
        'checkpoint_id',
        'checkpoint_map',
      ]) ||
      cp.thread_id !== threadId ||
      cp.checkpoint_ns !== '' ||
      !text(cp.checkpoint_id)
    )
      return;
    if (
      cp.checkpoint_map !== undefined &&
      (!plainRecord(cp.checkpoint_map) ||
        !Object.hasOwn(cp.checkpoint_map, '') ||
        Object.entries(cp.checkpoint_map).some(
          ([key, value]) => key !== '' || value !== cp.checkpoint_id
        ))
    )
      return;
    return cp as unknown as Checkpoint;
  } catch {
    return;
  }
}
 
export interface Observation {
  observe(snapshot: unknown): void;
  capture(outcome: unknown, snapshot: unknown): Owned | undefined;
}
/** Created before submit. Only matching human rows observed during that run can be confirmed. */
export function createObservation(input: {
  threadId: string;
  input: string;
  previous: unknown;
}): Observation {
  const { threadId, input: humanInput } = input;
  let previous: readonly Semantic[] = [],
    priorGenerations: readonly string[] = [],
    human: Semantic | undefined,
    generation: string | undefined,
    invalid = false;
  try {
    const rows = canonical(copyData(input.previous));
    if (!rows || !text(threadId) || typeof humanInput !== 'string')
      invalid = true;
    else {
      previous = Object.freeze(rows.map((m) => semantic(m)!));
      priorGenerations = rows.map((m) => m.delivery.generation);
    }
  } catch {
    invalid = true;
  }
  function matching(snapshot: unknown, mayCapture: boolean) {
    const state = copyData(snapshot);
    if (!plainRecord(state) || !Array.isArray(state.messages)) return false;
    const rows = state.messages;
    if (
      rows.length < previous.length ||
      !equalData(rows.slice(0, previous.length).map(semantic), previous)
    )
      return false;
    const suffix = rows.slice(previous.length);
    // Before the admitted human appears, ordinary subscription updates retain just the prefix.
    if (!suffix.length) return mayCapture && !human;
    const first = suffix[0],
      candidate = semantic(first);
    if (
      !candidate ||
      candidate.role !== 'user' ||
      candidate.content !== humanInput ||
      previous.some((m) => m.id === candidate.id) ||
      suffix.slice(1).some((m) => plainRecord(m) && m.role === 'user') ||
      !plainRecord(first) ||
      !plainRecord(first.delivery) ||
      !text(first.delivery.generation) ||
      priorGenerations.includes(first.delivery.generation)
    )
      return false;
    if (human)
      return (
        equalData(human, candidate) && generation === first.delivery.generation
      );
    if (!mayCapture) return false;
    human = candidate;
    generation = first.delivery.generation;
    return true;
  }
  return Object.freeze({
    observe(snapshot: unknown) {
      if (!invalid)
        try {
          if (!matching(snapshot, true)) invalid = true;
        } catch {
          invalid = true;
        }
    },
    capture(outcome: unknown, snapshot: unknown): Owned | undefined {
      try {
        const state = copyData(snapshot);
        if (
          invalid ||
          !human ||
          !generation ||
          outcome !== 'success' ||
          !plainRecord(state) ||
          state.status !== 'idle' ||
          !clean(state) ||
          !empty(state.toolCalls) ||
          !matching(state, false) ||
          !Array.isArray(state.messages) ||
          !canonical([state.messages[previous.length]])
        )
          return;
        return Object.freeze({ threadId, previous, human, generation });
      } catch {
        return;
      }
    },
  });
}
export function captureOwned(input: {
  observation: Observation;
  outcome: unknown;
  snapshot: unknown;
}): Owned | undefined {
  return input.observation.capture(input.outcome, input.snapshot);
}
function rawMatches(wire: unknown, message: Message): boolean {
  if (
    !plainRecord(wire) ||
    !clean(wire) ||
    wire.id !== message.id ||
    wire.content !== message.content ||
    wire.type !== (message.role === 'user' ? 'human' : 'ai') ||
    !empty(wire.tool_calls) ||
    !empty(wire.invalid_tool_calls) ||
    !empty(wire.tool_call_chunks) ||
    !empty(wire.tool_call_id)
  )
    return false;
  const extra = wire.additional_kwargs ?? {};
  if (
    !plainRecord(extra) ||
    !empty(extra.function_call) ||
    !empty(extra.tool_calls) ||
    !empty(extra.tool_call_chunks) ||
    !empty(extra.citations) ||
    !empty(extra.sources)
  )
    return false;
  const name = wire.name === null ? undefined : wire.name;
  const reasoning = wire.reasoning ?? extra.reasoning_content;
  return name === message.name && reasoning === message.reasoning;
}
/** Raw graph messages and the exact saved root tuple are the only canonical authority. */
export function captureTerminal(
  owned: Owned | undefined,
  input: unknown,
  checkpoint: unknown
): Confirmation | undefined {
  try {
    if (!owned) return;
    const state = copyData(input),
      raw = copyData(checkpoint);
    if (
      !plainRecord(state) ||
      !plainRecord(raw) ||
      state.status !== 'idle' ||
      !clean(state) ||
      !empty(state.toolCalls) ||
      !Array.isArray(state.interrupts) ||
      !Array.isArray(state.subgraphs) ||
      !Array.isArray(raw.next) ||
      raw.next.length ||
      !Array.isArray(raw.tasks) ||
      raw.tasks.length ||
      !clean(raw) ||
      !Array.isArray(state.history) ||
      !plainRecord(state.history[0]) ||
      !clean(state.history[0]) ||
      !Array.isArray(state.history[0].next)
    )
      return;
    const cp = checkpointSource(state.history[0].checkpoint, owned.threadId);
    if (!cp || !equalData(cp, checkpointSource(raw.checkpoint, owned.threadId)))
      return;
    const messages = canonical(state.messages),
      values = raw.values;
    if (
      !messages ||
      messages.length !== owned.previous.length + 2 ||
      !plainRecord(values) ||
      !clean(values) ||
      !Array.isArray(values.messages) ||
      values.messages.length !== messages.length ||
      !equalData(
        messages.slice(0, owned.previous.length).map(semantic),
        owned.previous
      ) ||
      !equalData(semantic(messages.at(-2)), owned.human) ||
      messages.at(-1)?.role !== 'assistant' ||
      !messages.at(-1)?.content.trim()
    )
      return;
    const wireMessages = values.messages;
    if (!messages.every((m, i) => rawMatches(wireMessages[i], m))) return;
    return Object.freeze({ checkpoint: cp, messages });
  } catch {
    return;
  }
}
const unavailable = Object.freeze({ kind: 'unavailable' as const });
export type Prepared =
  | typeof unavailable
  | { readonly kind: 'prepared'; readonly confirmation: Confirmation };
/** App owner fencing: never load after failed submit, never rescue a failed exact read. */
export async function prepareTerminal(input: {
  readonly operation: object;
  readonly observation: Observation;
  readonly outcome: unknown;
  readonly getOperation: () => object | undefined;
  readonly getSnapshot: () => unknown;
  readonly load: () => Promise<unknown>;
  readonly read: (checkpoint: Checkpoint) => Promise<unknown>;
}): Promise<Prepared> {
  const operation = input.operation;
  try {
    if (input.getOperation() !== operation) return unavailable;
    const owned = captureOwned({
      observation: input.observation,
      outcome: input.outcome,
      snapshot: input.getSnapshot(),
    });
    if (!owned) return unavailable;
    await input.load();
    if (input.getOperation() !== operation) return unavailable;
    const reference = input.getSnapshot(),
      snapshot = copyData(reference);
    if (
      !plainRecord(snapshot) ||
      !Array.isArray(snapshot.history) ||
      !plainRecord(snapshot.history[0])
    )
      return unavailable;
    const cp = checkpointSource(snapshot.history[0].checkpoint, owned.threadId);
    if (!cp) return unavailable;
    const exact = await input.read(cp);
    if (
      input.getOperation() !== operation ||
      input.getSnapshot() !== reference ||
      !equalData(input.getSnapshot(), snapshot)
    )
      return unavailable;
    const confirmation = captureTerminal(owned, snapshot, exact);
    return confirmation
      ? Object.freeze({ kind: 'prepared', confirmation })
      : unavailable;
  } catch {
    return unavailable;
  }
}

/** Application-private bounded, descriptor-only immutable data capture. */
export type Data =
  | undefined
  | null
  | boolean
  | number
  | string
  | readonly Data[]
  | { readonly [key: string]: Data };
const unsafeKeys = new Set(['__proto__', 'prototype', 'constructor']);
export function plainRecord(value: unknown): value is Record<string, Data> {
  if (!value || typeof value !== 'object' || Array.isArray(value)) return false;
  const prototype = Object.getPrototypeOf(value);
  if (prototype !== null && prototype !== Object.prototype) return false;
  return Reflect.ownKeys(value).every((key) => {
    if (typeof key !== 'string' || unsafeKeys.has(key)) return false;
    const d = Object.getOwnPropertyDescriptor(value, key);
    return !!d && 'value' in d && d.enumerable === true;
  });
}
export function closedFields(value: unknown, keys: readonly string[]): boolean {
  return (
    plainRecord(value) && Object.keys(value).every((key) => keys.includes(key))
  );
}
export function copyData(value: unknown): Data {
  let bytes = 0,
    nodes = 0;
  const parents = new Set<object>();
  function text(value: string) {
    if (value.length > 1048576) throw Error('Data byte bound');
    bytes += new TextEncoder().encode(value).length;
    if (bytes > 1048576) throw Error('Data byte bound');
    return value;
  }
  function visit(value: unknown, depth: number): Data {
    if (depth > 64 || ++nodes > 65536) throw Error('Data structure bound');
    if (value === null || value === undefined || typeof value === 'boolean')
      return value;
    if (typeof value === 'string') return text(value);
    if (typeof value === 'number' && Number.isFinite(value)) return value;
    if (typeof value !== 'object' || !value) throw Error('Unsupported data');
    if (parents.has(value)) throw Error('Cyclic data');
    parents.add(value);
    try {
      if (Array.isArray(value)) {
        if (
          Object.getPrototypeOf(value) !== Array.prototype ||
          value.length > 65536 ||
          Reflect.ownKeys(value).length !== value.length + 1
        )
          throw Error('Invalid array');
        const result: Data[] = [];
        for (let i = 0; i < value.length; i++) {
          const d = Object.getOwnPropertyDescriptor(value, String(i));
          if (!d || !('value' in d) || !d.enumerable)
            throw Error('Invalid array member');
          result.push(visit(d.value, depth + 1));
        }
        return Object.freeze(result);
      }
      if (!plainRecord(value)) throw Error('Invalid plain record');
      const result: Record<string, Data> = {};
      for (const key of Object.keys(value)) {
        text(key);
        result[key] = visit(
          Object.getOwnPropertyDescriptor(value, key)!.value,
          depth + 1
        );
      }
      return Object.freeze(result);
    } finally {
      parents.delete(value);
    }
  }
  return visit(value, 0);
}
/** Structural equality only for capturable finite plain data; key order is immaterial. */
export function equalData(left: unknown, right: unknown): boolean {
  try {
    const a = copyData(left),
      b = copyData(right);
    function same(a: Data, b: Data): boolean {
      if (Object.is(a, b)) return true;
      if (Array.isArray(a) || Array.isArray(b))
        return (
          Array.isArray(a) &&
          Array.isArray(b) &&
          a.length === b.length &&
          a.every((v, i) => same(v, b[i]))
        );
      if (!plainRecord(a) || !plainRecord(b)) return false;
      return (
        Object.keys(a).length === Object.keys(b).length &&
        Object.keys(a).every((k) => Object.hasOwn(b, k) && same(a[k], b[k]))
      );
    }
    return same(a, b);
  } catch {
    return false;
  }
}