React chat timeline 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,
  Reasoning,
  Citations,
} from '@threadplane/react/chat';
import { Markdown } from '@threadplane/react/markdown';
import type { MessageRow } from '@threadplane/content/messages';
import { createTimelineClient, type TimelineConnection } from './connection';
import { createTimelineApplication } from './application';
 
/** Content stores stay with their current or preview owner. */
export function renderMessage(row: MessageRow) {
  const label = row.role === 'user' ? 'You' : 'Assistant';
  return (
    <article
      aria-label={label}
      className={`tp-chat-message tp-chat-message--${row.role}`}
    >
      <p className="tp-chat-message__role">{label}</p>
      {row.role === 'assistant' && row.reasoning && (
        <Reasoning snapshot={row.reasoning} />
      )}
      <Markdown snapshot={row.markdown} />
      {row.role === 'assistant' && row.message.citations && (
        <Citations citations={row.message.citations} />
      )}
    </article>
  );
}
type Props = {
  readonly connection: TimelineConnection;
  readonly onReady: () => void;
};
/** Configuration changes replace the entire owner; credentials never become React keys. */
export function TimelineDemo(props: Props) {
  const fingerprint = JSON.stringify([
    props.connection.apiUrl,
    Object.entries(props.connection.headers).sort(([a], [b]) =>
      a.localeCompare(b)
    ),
  ]);
  const identity = useRef({ fingerprint, version: 0 });
  if (identity.current.fingerprint !== fingerprint)
    identity.current = { fingerprint, version: identity.current.version + 1 };
  return <TimelineOwner key={identity.current.version} {...props} />;
}
function TimelineOwner({ connection, onReady }: Props) {
  const [application] = useState(() =>
    createTimelineApplication(createTimelineClient(connection))
  );
  const snapshot = useAgent(application);
  useEffect(() => {
    onReady();
  }, [onReady]);
  useEffect(
    () => () => {
      void application.dispose();
    },
    [application]
  );
  const currentView = () =>
    application.getSnapshot().viewGeneration === snapshot.viewGeneration;
  const status = snapshot.busy
    ? snapshot.activity === 'creating'
      ? 'Creating conversation…'
      : snapshot.activity === 'replacing'
      ? 'Releasing conversation…'
      : 'Receiving response…'
    : !snapshot.canSubmit
    ? 'Start a new conversation to continue.'
    : snapshot.outcome === 'success'
    ? 'Response saved.'
    : 'Ready.';
  return (
    <ExampleLayout className="timeline-demo">
      <header>
        <div>
          <h1>Chat timeline</h1>
        </div>
        <button
          type="button"
          disabled={snapshot.busy}
          onClick={() => {
            if (currentView()) void application.newConversation();
          }}
        >
          New conversation
        </button>
      </header>
      <p>
        Select a saved checkpoint to preview it. Send always continues the
        current conversation. Fork from here uses the message in the composer to
        continue from the selected checkpoint.
      </p>
      <div className="timeline-layout">
        <section className="current-panel" aria-label="Current conversation">
          <h2>Current conversation</h2>
          <MessageList
            key={`current:${snapshot.viewGeneration}`}
            rows={snapshot.rows}
            renderMessage={renderMessage}
            label="Current messages"
          />
          <ExampleFooter>
            {snapshot.error && <p role="alert">{snapshot.error}</p>}
            <ChatInput
              key={`composer:${snapshot.viewGeneration}`}
              label="Message"
              value={snapshot.draft}
              onValueChange={(text) => {
                if (currentView()) application.setDraft(text);
              }}
              placeholder="Write a message…"
              hint="Enter to send. Shift+Enter adds a new line."
              busy={snapshot.busy}
              disabled={snapshot.busy || !snapshot.canSubmit}
              onStop={
                snapshot.busy && snapshot.activity !== 'replacing'
                  ? () => {
                      if (currentView()) void application.stop();
                    }
                  : undefined
              }
              onSubmit={(text) => {
                const current = application.getSnapshot();
                if (!currentView() || !current.canSubmit || current.busy)
                  return false;
                void application.submit(text);
                return true;
              }}
            />
            <p role="status">{status}</p>
          </ExampleFooter>
        </section>
        <aside className="history-panel" aria-label="Timeline">
          <nav className="checkpoint-picker" aria-label="Saved checkpoints">
            <div className="panel-heading">
              <h2>Saved checkpoints</h2>
              <button
                type="button"
                disabled={!snapshot.canRefresh || snapshot.readingHistory}
                onClick={() => {
                  if (currentView()) void application.refreshHistory();
                }}
              >
                Refresh checkpoints
              </button>
            </div>
            {!snapshot.threadId && (
              <p>Send a message to begin saving checkpoints.</p>
            )}
            {snapshot.threadId &&
              snapshot.historyPage === undefined &&
              !snapshot.readingHistory && (
                <p>Refresh to see saved checkpoints.</p>
              )}
            {snapshot.readingHistory && (
              <p role="status">Loading checkpoints…</p>
            )}
            {snapshot.historyError && (
              <p role="alert">{snapshot.historyError}</p>
            )}
            {snapshot.historyPage?.length === 0 && (
              <p>No checkpoints available.</p>
            )}
            {snapshot.historyPage?.map((row, index) => (
              <button
                key={index}
                type="button"
                aria-pressed={
                  !!row.source && row.source === snapshot.selectedSource
                }
                disabled={
                  !row.source ||
                  snapshot.busy ||
                  snapshot.readingHistory ||
                  !snapshot.canSubmit
                }
                onClick={() => {
                  if (
                    currentView() &&
                    application.getSnapshot().historyPage?.[index] === row
                  )
                    void application.selectCheckpoint(index);
                }}
              >
                Checkpoint {index + 1}
                {!row.source ? ' (unavailable)' : ''}
                {row.createdAt && (
                  <span className="checkpoint-time">{row.createdAt}</span>
                )}
              </button>
            ))}
          </nav>
          <section className="preview-panel" aria-label="Historical preview">
            <h2>Historical preview</h2>
            <p>
              Read-only. Selecting a checkpoint does not change where Send
              continues.
            </p>
            {!snapshot.selectedSource && (
              <p>Select a checkpoint to preview its messages.</p>
            )}
            {snapshot.readingPreview && <p role="status">Loading preview…</p>}
            {snapshot.previewError && (
              <p role="alert">{snapshot.previewError}</p>
            )}
            <MessageList
              key={`${snapshot.viewGeneration}:${
                snapshot.historyPage?.findIndex(
                  (row) => row.source === snapshot.selectedSource
                ) ?? -1
              }`}
              rows={snapshot.previewRows}
              renderMessage={renderMessage}
              label="Preview messages"
            />
            <div className="preview-actions">
              <button
                type="button"
                disabled={!snapshot.canFork || !snapshot.draft.trim()}
                onClick={() => {
                  const current = application.getSnapshot();
                  if (
                    currentView() &&
                    current.selectedSource === snapshot.selectedSource &&
                    current.canFork &&
                    current.draft.trim()
                  )
                    void application.forkSelected(current.draft);
                }}
              >
                Fork from here
              </button>
              {snapshot.readingPreview ? (
                <button
                  type="button"
                  onClick={() => {
                    if (
                      currentView() &&
                      application.getSnapshot().selectedSource ===
                        snapshot.selectedSource
                    )
                      application.cancelPreview();
                  }}
                >
                  Cancel preview
                </button>
              ) : (
                <button
                  type="button"
                  disabled={!snapshot.selectedSource}
                  onClick={() => {
                    if (
                      currentView() &&
                      application.getSnapshot().selectedSource ===
                        snapshot.selectedSource
                    )
                      application.clearSelection();
                  }}
                >
                  Clear selection
                </button>
              )}
            </div>
            <p>
              Fork from here sends the composer message from this checkpoint. A
              successful fork becomes the current conversation.
            </p>
          </section>
        </aside>
      </div>
    </ExampleLayout>
  );
}

/** Descriptor capture never evaluates accessors; every returned container is owned. */
export function copyData(value: unknown, parents = new Set<object>()): unknown {
  if (
    value === null ||
    value === undefined ||
    typeof value === 'string' ||
    typeof value === 'boolean'
  )
    return value;
  if (typeof value === 'number' && Number.isFinite(value)) return value;
  if (
    typeof value !== 'object' ||
    parents.has(value) ||
    Object.getOwnPropertySymbols(value).length
  )
    throw new Error('Unsupported data');
  const array = Array.isArray(value),
    prototype = Object.getPrototypeOf(value);
  if (
    array
      ? prototype !== Array.prototype
      : prototype !== Object.prototype && prototype !== null
  )
    throw new Error('Unsupported data');
  const descriptors = Object.getOwnPropertyDescriptors(value);
  const keys = Object.keys(descriptors);
  if (
    array &&
    (keys.length !== descriptors['length'].value + 1 ||
      keys.some(
        (key) =>
          key !== 'length' &&
          (!/^(0|[1-9]\d*)$/.test(key) ||
            Number(key) >= descriptors['length'].value)
      ))
  )
    throw new Error('Malformed array');
  parents.add(value);
  const result = array ? [] : Object.create(null);
  for (const key of keys.sort()) {
    if (array && key === 'length') continue;
    const descriptor = descriptors[key];
    if (!('value' in descriptor)) throw new Error('Accessor data');
    Object.defineProperty(result, key, {
      value: copyData(descriptor.value, parents),
      enumerable: true,
    });
  }
  parents.delete(value);
  return Object.freeze(result);
}
export const record = (value: unknown): value is Record<string, unknown> =>
  !!value && typeof value === 'object' && !Array.isArray(value);
export const text = (value: unknown): value is string =>
  typeof value === 'string' && value.trim().length > 0;
export interface ForkSource {
  readonly thread_id: string;
  readonly checkpoint_ns: '';
  readonly checkpoint_id: string;
  readonly checkpoint_map?: Readonly<Record<string, string>>;
}
/** Called on captured data only. A missing map differs from an explicit map. */
export function checkpointSource(
  value: unknown,
  threadId: string
): ForkSource | null {
  if (
    !text(threadId) ||
    !record(value) ||
    value['thread_id'] !== threadId ||
    value['checkpoint_ns'] !== '' ||
    !text(value['checkpoint_id'])
  )
    return null;
  const map = value['checkpoint_map'];
  if (
    map !== undefined &&
    (!record(map) || Object.values(map).some((item) => !text(item)))
  )
    return null;
  return Object.freeze({
    thread_id: threadId,
    checkpoint_ns: '',
    checkpoint_id: value['checkpoint_id'],
    ...(map === undefined
      ? {}
      : { checkpoint_map: map as Readonly<Record<string, string>> }),
  });
}
export interface CheckpointRow {
  readonly id: string | null;
  readonly namespace: string | null;
  readonly createdAt: string | null;
  readonly parentId: string | null;
  readonly next: readonly string[] | null;
  readonly source: ForkSource | null;
  readonly unavailable: string | null;
}
/** Retains metadata only, in server order. Malformed pages grant no authority. */
export function captureHistoryPage(
  input: unknown,
  threadId: string
): readonly CheckpointRow[] | undefined {
  try {
    const page = copyData(input);
    if (!Array.isArray(page)) return;
    const counts = new Map<string, number>();
    const rows = page.map((entry) => {
      const item = record(entry) ? entry : {};
      const checkpoint = record(item['checkpoint']) ? item['checkpoint'] : {};
      const id =
        typeof checkpoint['checkpoint_id'] === 'string'
          ? checkpoint['checkpoint_id']
          : null;
      if (id !== null) counts.set(id, (counts.get(id) ?? 0) + 1);
      const next =
        Array.isArray(item['next']) &&
        item['next'].every((node) => typeof node === 'string')
          ? (item['next'] as readonly string[])
          : null;
      const source =
        next?.length === 0 ? checkpointSource(checkpoint, threadId) : null;
      const parent = item['parent_checkpoint'];
      return {
        id,
        namespace:
          typeof checkpoint['checkpoint_ns'] === 'string'
            ? checkpoint['checkpoint_ns']
            : null,
        createdAt:
          typeof item['created_at'] === 'string' ? item['created_at'] : null,
        parentId:
          record(parent) && typeof parent['checkpoint_id'] === 'string'
            ? parent['checkpoint_id']
            : null,
        next,
        source,
        unavailable: source ? null : 'Checkpoint is unavailable.',
      };
    });
    return Object.freeze(
      rows.map((row) =>
        Object.freeze(
          row.source && counts.get(row.id!) !== 1
            ? {
                ...row,
                source: null,
                unavailable: 'Duplicate checkpoint identity.',
              }
            : row
        )
      )
    );
  } catch {
    return;
  }
}

import type { Message } from '@threadplane/core';
import {
  checkpointSource,
  copyData,
  record,
  text,
  type ForkSource,
} from './checkpoints';
export interface Preview {
  readonly source: ForkSource;
  readonly messages: readonly Message[];
}
const emptyOptional = (value: unknown) =>
  value === undefined || (Array.isArray(value) && value.length === 0);
/** Known-checkpoint reads must prove the exact requested source before conversion. */
export function capturePreview(
  input: unknown,
  requested: ForkSource
): Preview | null {
  try {
    const state = copyData(input),
      request = copyData(requested);
    if (!record(state) || !record(request) || !text(request['thread_id']))
      return null;
    const source = checkpointSource(request, request['thread_id']);
    const actual = checkpointSource(state['checkpoint'], request['thread_id']);
    if (
      !source ||
      !actual ||
      JSON.stringify(source) !== JSON.stringify(actual) ||
      !Array.isArray(state['next']) ||
      state['next'].length ||
      !Array.isArray(state['tasks']) ||
      state['tasks'].length ||
      (state['error'] !== undefined && state['error'] !== null)
    )
      return null;
    for (const key of ['interrupts', 'subgraphs', 'tool_calls', 'toolCalls'])
      if (!emptyOptional(state[key])) return null;
    const values = state['values'];
    if (
      !record(values) ||
      !Array.isArray(values['messages']) ||
      !values['messages'].length ||
      values['messages'].length % 2
    )
      return null;
    if (values['error'] !== undefined && values['error'] !== null) return null;
    for (const key of [
      '__interrupt__',
      'interrupts',
      'subgraphs',
      'tool_calls',
      'toolCalls',
    ])
      if (!emptyOptional(values[key])) return null;
    const messages: Message[] = [],
      ids = new Set<string>();
    for (const [index, raw] of values['messages'].entries()) {
      if (
        !record(raw) ||
        !text(raw['id']) ||
        ids.has(raw['id']) ||
        raw['type'] !== (index % 2 ? 'ai' : 'human') ||
        typeof raw['content'] !== 'string'
      )
        return null;
      if (
        raw['tool_call_id'] !== undefined ||
        raw['toolCallId'] !== undefined ||
        raw['function_call'] !== undefined ||
        !emptyOptional(raw['toolCallIds']) ||
        !emptyOptional(raw['interrupts']) ||
        !emptyOptional(raw['subgraphs']) ||
        !emptyOptional(raw['tool_calls']) ||
        !emptyOptional(raw['invalid_tool_calls'])
      )
        return null;
      const extra = raw['additional_kwargs'];
      if (
        extra !== undefined &&
        (!record(extra) ||
          extra['function_call'] !== undefined ||
          !emptyOptional(extra['tool_calls']))
      )
        return null;
      ids.add(raw['id']);
      messages.push(
        Object.freeze({
          id: raw['id'],
          role: index % 2 ? 'assistant' : 'user',
          content: raw['content'],
          delivery: Object.freeze({
            generation: raw['id'],
            phase: 'complete',
            outcome: 'success',
          }),
        })
      );
    }
    return Object.freeze({ source, messages: Object.freeze(messages) });
  } catch {
    return null;
  }
}

import type { CompleteOutcome, Message } from '@threadplane/core';
import {
  createMessageContent,
  type MessageRow,
} from '@threadplane/content/messages';
import {
  captureTurn,
  confirmCanonicalHistory,
  type OwnedTurn,
} from './authority';
import {
  captureHistoryPage,
  copyData,
  type CheckpointRow,
  type ForkSource,
} from './checkpoints';
import { capturePreview, type Preview } from './preview';
import type {
  TimelineClient,
  TimelineSession,
  TimelineState,
} from './connection';
export interface TimelineSnapshot {
  readonly threadId: string | null;
  readonly rows: readonly MessageRow[];
  readonly previewRows: readonly MessageRow[];
  readonly historyPage: readonly CheckpointRow[] | undefined;
  readonly selectedSource: ForkSource | null;
  readonly draft: string;
  readonly activity: 'idle' | 'creating' | 'running' | 'replacing';
  readonly busy: boolean;
  readonly readingHistory: boolean;
  readonly readingPreview: boolean;
  readonly canSubmit: boolean;
  readonly canRefresh: boolean;
  readonly canFork: boolean;
  readonly outcome: CompleteOutcome | null;
  readonly error: string | null;
  readonly historyError: string | null;
  readonly previewError: string | null;
  readonly viewGeneration: number;
}
/** Execution, metadata reads and checkpoint previews have independent lifetimes. */
export function createTimelineApplication(options: TimelineClient) {
  type Owner = {
    session: TimelineSession;
    content: ReturnType<typeof createMessageContent>;
    release?: () => void;
    closed: boolean;
    unsafe: boolean;
    confirmed: readonly Message[];
    proof?: OwnedTurn;
    admission?: {
      input: string;
      prefix: readonly Message[];
      generations: Set<string>;
      turn?: OwnedTurn;
    };
  };
  type Read = { controller: AbortController; owner?: Owner };
  const listeners = new Set<() => void>(),
    cleanups = new Set<Promise<void>>();
  let primary: Owner | undefined,
    execution: AbortController | undefined,
    historyRead: Read | undefined,
    previewRead: Read | undefined;
  let preview: Preview | null = null,
    previewContent: ReturnType<typeof createMessageContent> | undefined;
  let disposed = false,
    disposal: Promise<void> | undefined;
  let snapshot: TimelineSnapshot = Object.freeze({
    threadId: null,
    rows: [],
    previewRows: [],
    historyPage: undefined,
    selectedSource: null,
    draft: '',
    activity: 'idle',
    busy: false,
    readingHistory: false,
    readingPreview: false,
    canSubmit: true,
    canRefresh: false,
    canFork: false,
    outcome: null,
    error: null,
    historyError: null,
    previewError: null,
    viewGeneration: 0,
  });
  function publish(
    update: Partial<TimelineSnapshot> = {},
    notifyListeners = true
  ) {
    if (disposed) return;
    const next = { ...snapshot, ...update };
    snapshot = Object.freeze({
      ...next,
      canRefresh: !!next.threadId && !next.busy && next.canSubmit,
      canFork:
        !!preview && !next.busy && next.canSubmit && !next.readingPreview,
    });
    if (!notifyListeners) return;
    for (const notify of [...listeners]) {
      if (disposed) break;
      notify();
    }
  }
  function bounded(work: () => void | Promise<void>): Promise<void> {
    let timer: ReturnType<typeof setTimeout> | undefined;
    return Promise.race([
      Promise.resolve()
        .then(work)
        .catch(() => undefined),
      new Promise<void>((resolve) => {
        timer = setTimeout(resolve, 2000);
      }),
    ]).finally(() => {
      if (timer) clearTimeout(timer);
    });
  }
  function close(owner: Owner) {
    if (owner.closed) return Promise.resolve();
    owner.closed = true;
    // Register retirement before unsubscribe/content cleanup can call back into commands.
    const pending = track(() => owner.session.dispose());
    try {
      owner.release?.();
    } catch {
      /* Ownership is already revoked. */
    }
    try {
      owner.content.dispose();
    } catch {
      /* Session cleanup still runs. */
    }
    return pending;
  }
  function track(work: () => void | Promise<void>) {
    const pending = bounded(work);
    cleanups.add(pending);
    void pending.then(() => cleanups.delete(pending));
    return pending;
  }
  function cancelHistory(notifyListeners = true) {
    const read = historyRead;
    historyRead = undefined;
    read?.controller.abort();
    if (read?.owner) void close(read.owner);
    publish({ readingHistory: false }, notifyListeners);
  }
  function clearPreview() {
    preview = null;
    previewContent?.dispose();
    previewContent = undefined;
  }
  function cancelPreview(notifyListeners = true) {
    const read = previewRead;
    previewRead = undefined;
    read?.controller.abort();
    clearPreview();
    publish(
      {
        readingPreview: false,
        selectedSource: null,
        previewRows: [],
        previewError: null,
      },
      notifyListeners
    );
  }
  function unsafe(state: TimelineState) {
    return (
      !state ||
      !Array.isArray(state.messages) ||
      !Array.isArray(state.toolCalls) ||
      !Array.isArray(state.interrupts) ||
      !Array.isArray(state.subgraphs) ||
      !!state.error ||
      state.status === 'error' ||
      !!state.toolCalls.length ||
      !!state.interrupts.length ||
      !!state.subgraphs.length
    );
  }
  function block(owner?: Owner) {
    if (owner) owner.unsafe = true;
    publish({
      canSubmit: false,
      outcome: 'error',
      error:
        'The current conversation could not be confirmed. Start a new conversation.',
    });
  }
  function observe(owner: Owner): TimelineState | undefined {
    if (disposed || primary !== owner || owner.closed) return;
    try {
      const raw = owner.session.getSnapshot(),
        state = copyData(raw) as TimelineState;
      if (unsafe(state)) throw new Error('Unsafe session');
      const admission = owner.admission;
      if (admission && !admission.turn)
        admission.turn = captureTurn(
          state.messages,
          admission.generations,
          admission.prefix,
          admission.input
        );
      const visible = admission
        ? [
            ...admission.prefix,
            ...(admission.turn
              ? state.messages.filter(
                  (message) =>
                    message.delivery.generation === admission.turn!.generation
                )
              : []),
          ]
        : owner.confirmed;
      const rows = owner.content.project({ ...state, messages: visible });
      if (disposed || primary !== owner || owner.closed) return;
      publish({ rows });
      if (disposed || primary !== owner || owner.closed) return;
      if (owner.session.getSnapshot() !== raw)
        throw new Error('Snapshot changed during observation');
      return state;
    } catch {
      if (!disposed && primary === owner && !owner.closed) block(owner);
      return;
    }
  }
  function makeOwner(id: string): Owner {
    return {
      session: options.sessionFactory(id),
      content: createMessageContent(),
      closed: false,
      unsafe: false,
      confirmed: [],
    };
  }
  async function execute(input: string, selected: Preview | null) {
    if (
      disposed ||
      snapshot.busy ||
      !snapshot.canSubmit ||
      typeof input !== 'string' ||
      !input.trim()
    )
      return;
    const controller = new AbortController();
    execution = controller;
    const current = () =>
      !disposed && execution === controller && !controller.signal.aborted;
    // Capture authority before publication can synchronously trigger another command.
    const prefix = selected?.messages ?? primary?.confirmed ?? [],
      source = selected?.source;
    publish({
      busy: true,
      activity: primary ? 'running' : 'creating',
      draft: '',
      error: null,
      outcome: null,
    });
    if (!current()) return;
    cancelHistory(false);
    // A completed preview survives Send; a pending preview has no validated authority.
    if (previewRead) cancelPreview(false);
    try {
      if (!current()) return;
      if (!primary) {
        const id = await options.createThread(controller.signal);
        if (!current()) return;
        const owner = makeOwner(id);
        if (!current()) {
          void close(owner);
          return;
        }
        primary = owner;
        owner.release = owner.session.subscribe(() => observe(owner));
        if (owner.closed) owner.release();
        if (!current()) return;
        publish({ threadId: id, activity: 'running' });
      }
      const owner = primary;
      if (!owner || !current()) return;
      const valid = () => current() && primary === owner && !owner.closed;
      const before = observe(owner);
      if (!valid()) return;
      if (
        !before ||
        (!owner.proof && before.messages.length > 0) ||
        owner.unsafe ||
        before.status !== 'idle' ||
        (owner.proof &&
          !confirmCanonicalHistory(before.messages, before.values, owner.proof))
      ) {
        block(owner);
        return;
      }
      owner.admission = {
        input,
        prefix,
        generations: new Set(before.messages.map((m) => m.delivery.generation)),
      };
      const pending = source
        ? owner.session.fork(source, input, { signal: controller.signal })
        : owner.session.submit(input, { signal: controller.signal });
      void pending.catch(() => undefined);
      if (valid()) observe(owner);
      const outcome = await pending;
      if (!valid()) return;
      const terminal = observe(owner);
      if (!valid()) return;
      const turn = owner.admission?.turn;
      const canonical =
        terminal && terminal.status === 'idle' && !unsafe(terminal) && turn
          ? confirmCanonicalHistory(terminal.messages, terminal.values, turn)
          : undefined;
      if (outcome !== 'success' || owner.unsafe || !canonical) {
        block(owner);
        return;
      }
      owner.confirmed = canonical;
      owner.proof = turn;
      owner.admission = undefined;
      const rows = owner.content.project({ ...terminal!, messages: canonical });
      if (source) cancelPreview(false);
      if (valid())
        publish({ rows, canSubmit: true, outcome: 'success', error: null });
    } catch {
      if (current()) block(primary);
    } finally {
      if (current()) {
        execution = undefined;
        publish({ busy: false, activity: 'idle' });
      }
    }
  }
  async function refreshHistory() {
    if (disposed || !snapshot.canRefresh || !snapshot.threadId) return;
    cancelHistory(false);
    cancelPreview(false);
    const id = snapshot.threadId,
      read: Read = { controller: new AbortController() };
    historyRead = read;
    const current = () =>
      !disposed &&
      historyRead === read &&
      !read.controller.signal.aborted &&
      snapshot.threadId === id;
    publish({
      readingHistory: true,
      historyError: null,
      historyPage: undefined,
    });
    try {
      if (!current()) return;
      const owner = makeOwner(id);
      read.owner = owner;
      if (!current()) {
        void close(owner);
        return;
      }
      await owner.session.load({ signal: read.controller.signal });
      if (!current()) return;
      const state = copyData(owner.session.getSnapshot()) as TimelineState;
      const page = captureHistoryPage(state.history, id);
      if (!current()) return;
      if (!page) throw new Error('Unavailable history');
      publish({ historyPage: page });
    } catch {
      if (current())
        publish({ historyError: 'Checkpoint history is unavailable.' });
    } finally {
      if (read.owner) void close(read.owner);
      if (current()) {
        historyRead = undefined;
        publish({ readingHistory: false });
      }
    }
  }
  async function selectCheckpoint(index: number) {
    if (
      disposed ||
      snapshot.busy ||
      snapshot.readingHistory ||
      !snapshot.canSubmit
    )
      return;
    const source = snapshot.historyPage?.[index]?.source;
    if (!source) return;
    cancelPreview(false);
    const read: Read = { controller: new AbortController() };
    previewRead = read;
    const current = () =>
      !disposed &&
      previewRead === read &&
      !read.controller.signal.aborted &&
      snapshot.threadId === source.thread_id;
    publish({
      selectedSource: source,
      readingPreview: true,
      previewError: null,
    });
    try {
      if (!current()) return;
      const raw = await options.readCheckpoint(source, read.controller.signal);
      if (!current()) return;
      const captured = capturePreview(raw, source);
      if (!captured) throw new Error('Unavailable preview');
      const content = createMessageContent();
      const rows = content.project({
        status: 'idle',
        messages: captured.messages,
        toolCalls: [],
      });
      if (!current()) {
        content.dispose();
        return;
      }
      preview = captured;
      previewContent = content;
      publish({ previewRows: rows });
    } catch {
      if (current())
        publish({ previewError: 'Checkpoint preview is unavailable.' });
    } finally {
      if (current()) {
        previewRead = undefined;
        publish({ readingPreview: false });
      }
    }
  }
  async function newConversation() {
    if (disposed) return;
    const previous = execution,
      controller = new AbortController();
    execution = controller;
    const owner = primary;
    primary = undefined;
    publish({ busy: true, activity: 'replacing', canSubmit: false }, false);
    if (owner) void close(owner);
    previous?.abort();
    if (disposed || execution !== controller) return;
    cancelHistory(false);
    cancelPreview(false);
    if (disposed || execution !== controller) return;
    publish();
    await Promise.all([...cleanups]);
    if (disposed || execution !== controller) return;
    execution = undefined;
    publish({
      threadId: null,
      rows: [],
      draft: '',
      historyPage: undefined,
      historyError: null,
      error: null,
      outcome: null,
      canSubmit: true,
      busy: false,
      activity: 'idle',
      viewGeneration: snapshot.viewGeneration + 1,
    });
  }
  async function stop() {
    if (disposed || !execution || snapshot.activity === 'replacing') return;
    const previous = execution,
      controller = new AbortController();
    execution = controller;
    const owner = primary;
    primary = undefined;
    publish(
      {
        busy: true,
        activity: 'replacing',
        canSubmit: false,
        outcome: 'aborted',
        error: 'Execution stopped. Start a new conversation.',
      },
      false
    );
    // Both asynchronous retirement operations belong to New/dispose's cleanup barrier.
    const stopping = track(() => owner?.session.stop());
    const closing = owner ? close(owner) : Promise.resolve();
    previous.abort();
    if (disposed || execution !== controller) return;
    cancelHistory(false);
    cancelPreview(false);
    if (disposed || execution !== controller) return;
    publish();
    await Promise.all([closing, stopping]);
    if (!disposed && execution === controller) {
      execution = undefined;
      publish({ busy: false, activity: 'idle' });
    }
  }
  function dispose(): Promise<void> {
    if (disposal) return disposal;
    disposed = true;
    execution?.abort();
    execution = undefined;
    cancelHistory(false);
    cancelPreview(false);
    const owner = primary;
    primary = undefined;
    if (owner) void close(owner);
    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);
      };
    },
    setDraft(text: string) {
      if (
        !disposed &&
        !snapshot.busy &&
        snapshot.canSubmit &&
        typeof text === 'string'
      )
        publish({ draft: text });
    },
    submit: (text = snapshot.draft) => execute(text, null),
    forkSelected: (text = snapshot.draft) =>
      preview ? execute(text, preview) : Promise.resolve(),
    refreshHistory,
    selectCheckpoint,
    cancelPreview: () => cancelPreview(),
    clearSelection: () => cancelPreview(),
    newConversation,
    stop,
    dispose,
  };
}

import type { Message } from '@threadplane/core';
import { copyData, record, text } from './checkpoints';
export interface OwnedTurn {
  readonly id: string;
  readonly generation: string;
  readonly input: string;
  /** Supplied independently: confirmed current history for Send, preview for Fork. */
  readonly prefix: readonly Message[];
}
function complete(value: unknown): value is Message {
  if (
    !record(value) ||
    !text(value['id']) ||
    typeof value['content'] !== 'string' ||
    !['user', 'assistant'].includes(value['role'] as string) ||
    value['toolCallId'] !== undefined ||
    (value['toolCallIds'] !== undefined &&
      (!Array.isArray(value['toolCallIds']) || value['toolCallIds'].length))
  )
    return false;
  const delivery = value['delivery'];
  return (
    record(delivery) &&
    text(delivery['generation']) &&
    delivery['phase'] === 'complete' &&
    delivery['outcome'] === 'success'
  );
}
function validPrefix(value: unknown): value is readonly Message[] {
  return (
    Array.isArray(value) &&
    value.length % 2 === 0 &&
    value.every(
      (message, index) =>
        complete(message) && message.role === (index % 2 ? 'assistant' : 'user')
    ) &&
    new Set(value.map((message) => message.id)).size === value.length
  );
}
/** Admission never derives the expected prefix from the session being verified. */
export function captureTurn(
  inputMessages: readonly Message[],
  previousGenerations: ReadonlySet<string>,
  expectedPrefix: readonly Message[],
  submittedInput: string
): OwnedTurn | undefined {
  try {
    const messages = copyData(inputMessages),
      prefix = copyData(expectedPrefix);
    if (
      !Array.isArray(messages) ||
      !validPrefix(prefix) ||
      typeof submittedInput !== 'string'
    )
      return;
    const candidates = messages.filter(
      (message) =>
        record(message) &&
        message['role'] === 'user' &&
        record(message['delivery']) &&
        message['delivery']['generation'] !== message['id'] &&
        !previousGenerations.has(message['delivery']['generation'] as string)
    );
    if (candidates.length !== 1 || !complete(candidates[0])) return;
    const human = candidates[0];
    if (
      human.content !== submittedInput ||
      prefix.some((message) => message.id === human.id) ||
      messages.filter(
        (message) => record(message) && message['id'] === human.id
      ).length !== 1
    )
      return;
    return Object.freeze({
      id: human.id,
      generation: human.delivery.generation,
      input: submittedInput,
      prefix,
    });
  } catch {
    return;
  }
}
/** Call only after the controller proves idle, error-free terminal session state. */
export function confirmCanonicalHistory(
  inputMessages: readonly Message[],
  inputValues: unknown,
  inputTurn: OwnedTurn
): readonly Message[] | undefined {
  try {
    const messages = copyData(inputMessages),
      values = copyData(inputValues),
      turn = copyData(inputTurn) as OwnedTurn;
    if (
      !Array.isArray(messages) ||
      !record(values) ||
      !validPrefix(turn.prefix) ||
      !text(turn.id) ||
      !text(turn.generation) ||
      typeof turn.input !== 'string' ||
      values['completed_turn_id'] !== turn.id ||
      !text(values['completed_answer_id'])
    )
      return;
    const expected = [
      ...turn.prefix.map((message) => message.id),
      turn.id,
      values['completed_answer_id'],
    ];
    const ids = values['completed_message_ids'];
    if (
      !Array.isArray(ids) ||
      ids.length !== expected.length ||
      new Set(ids).size !== ids.length ||
      !ids.every((id, index) => id === expected[index])
    )
      return;
    const canonical: Message[] = [];
    for (const id of ids) {
      const matches = messages.filter(
        (message) => record(message) && message['id'] === id
      );
      if (matches.length !== 1 || !complete(matches[0])) return;
      canonical.push(matches[0]);
    }
    if (
      !turn.prefix.every(
        (message, index) =>
          message.id === canonical[index].id &&
          message.role === canonical[index].role &&
          message.content === canonical[index].content
      )
    )
      return;
    const human = canonical.at(-2)!,
      answer = canonical.at(-1)!;
    if (
      human.role !== 'user' ||
      answer.role !== 'assistant' ||
      human.content !== turn.input ||
      human.delivery.generation !== turn.generation ||
      answer.delivery.generation !== turn.generation
    )
      return;
    return Object.freeze(canonical);
  } catch {
    return;
  }
}