React chat threads 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 { createThreadsClient, type ThreadsConnection } from './connection';
import { createThreadsApplication } from './application';
 
/** Render existing native content stores without retaining them outside their session 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: ThreadsConnection;
  readonly onReady: () => void;
};
/** Give every changed runtime target a fresh application lifetime and an opaque React key. */
export function ThreadsDemo(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 <ThreadsOwner key={identity.current.version} {...props} />;
}
function ThreadsOwner({ connection, onReady }: Props) {
  const [application] = useState(() =>
    createThreadsApplication(createThreadsClient(connection))
  );
  const snapshot = useAgent(application);
  useEffect(() => {
    onReady();
  }, [onReady]);
  useEffect(
    () => () => {
      void application.dispose();
    },
    [application]
  );
  const status = snapshot.busy
    ? snapshot.activity === 'creating'
      ? 'Creating conversation…'
      : snapshot.activity === 'loading'
      ? 'Loading saved conversation…'
      : snapshot.activity === 'confirming'
      ? 'Confirming saved conversation…'
      : snapshot.activity === 'replacing'
      ? 'Releasing conversation…'
      : 'Receiving response…'
    : !snapshot.canSubmit
    ? 'Choose another conversation or start a new one.'
    : snapshot.confirmation === 'loaded'
    ? 'Conversation loaded.'
    : snapshot.confirmation === 'saved'
    ? 'Response saved.'
    : 'Ready.';
  return (
    <ExampleLayout
      className="threads-demo"
      about={
        <div className="seed-prompts" aria-label="Example prompts">
          <button
            type="button"
            disabled={snapshot.busy || !snapshot.canSubmit}
            onClick={() => {
              const current = application.getSnapshot();
              if (current.canSubmit && !current.busy)
                application.setDraft(
                  snapshot.viewGeneration,
                  'Explain how saved conversations work in two sentences.'
                );
            }}
          >
            Try a question
          </button>
        </div>
      }
    >
      <header>
        <div>
          <h1>Chat threads</h1>
        </div>
        <button
          type="button"
          disabled={snapshot.busy}
          onClick={() => void application.newConversation()}
        >
          New conversation
        </button>
      </header>
      <p>
        Switch between conversations created on this page. The list and drafts
        last while this page stays open; saved history is checked before
        continuing. Titles are optional.
      </p>
      <div className="threads-layout">
        <nav aria-label="Conversations" className="conversation-picker">
          <h2>Conversations</h2>
          {snapshot.conversations.length === 0 && (
            <p>No saved conversations on this page yet.</p>
          )}
          {snapshot.conversations.map((record) => (
            <button
              type="button"
              key={record.key}
              aria-pressed={snapshot.selectedKey === record.key}
              disabled={snapshot.busy || record.availability === 'unavailable'}
              onClick={() => void application.select(record.key)}
            >
              {record.label}
              {record.availability === 'unavailable' ? ' (unavailable)' : ''}
            </button>
          ))}
        </nav>
        <section
          className="conversation-panel"
          aria-label="Selected conversation"
        >
          <MessageList
            rows={snapshot.rows}
            renderMessage={renderMessage}
            label="Conversation"
          />
          <ExampleFooter>
            {snapshot.error && <p role="alert">{snapshot.error}</p>}
            <ChatInput
              key={snapshot.viewGeneration}
              label="Message"
              value={snapshot.draft}
              onValueChange={(text) =>
                application.setDraft(snapshot.viewGeneration, text)
              }
              placeholder="Write a message…"
              hint="Enter to send. Shift+Enter adds a new line."
              busy={snapshot.busy}
              disabled={
                snapshot.activity === 'loading' ||
                snapshot.activity === 'replacing' ||
                (!snapshot.canSubmit && !snapshot.busy)
              }
              onStop={
                snapshot.busy && snapshot.activity !== 'replacing'
                  ? () => void application.stop()
                  : undefined
              }
              onSubmit={(text) => {
                const current = application.getSnapshot();
                if (
                  !current.canSubmit ||
                  current.busy ||
                  current.viewGeneration !== snapshot.viewGeneration
                )
                  return false;
                void application.submit(text);
                return true;
              }}
            />
            <p role="status">{status}</p>
          </ExampleFooter>
        </section>
      </div>
    </ExampleLayout>
  );
}

import { copyData } from './authority';
/** Decode optional presentation metadata without granting conversation authority. */
export function decodeTitle(
  input: unknown,
  threadId: string
): string | undefined {
  try {
    const data = copyData(input) as {
      thread_id?: unknown;
      metadata?: { title?: unknown };
    };
    if (
      !threadId ||
      data?.thread_id !== threadId ||
      typeof data.metadata?.title !== 'string'
    )
      return undefined;
    const title = data.metadata.title.trim();
    return title ? Array.from(title).slice(0, 80).join('') : undefined;
  } catch {
    return undefined;
  }
}

import type { AgentSnapshot, Message } from '@threadplane/core';
export interface ThreadsState extends AgentSnapshot {
  readonly interrupts: readonly unknown[];
  readonly subgraphs: readonly unknown[];
  readonly history:
    | readonly {
        readonly checkpoint: unknown;
        readonly next: readonly string[];
      }[]
    | undefined;
}
export interface Canonical {
  readonly threadId: string;
  readonly checkpoint: string;
  readonly signature: string;
  readonly messages: readonly Message[];
}
/** Capture own data without evaluating backend-supplied accessors. */
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)
    throw new Error('Unsupported data');
  if (!array && prototype !== Object.prototype && prototype !== null)
    throw new Error('Unsupported data');
  parents.add(value);
  const descriptors = Object.getOwnPropertyDescriptors(value);
  const result: Record<string, unknown> | unknown[] = array
    ? []
    : Object.create(null);
  const keys = array
    ? Array.from({ length: value.length }, (_, index) => String(index))
    : Object.keys(descriptors).sort();
  if (
    array &&
    Object.keys(descriptors).some(
      (name) => name !== 'length' && !keys.includes(name)
    )
  )
    throw new Error('Unsupported data');
  for (const key of keys) {
    const descriptor = descriptors[key];
    if (!descriptor || !('value' in descriptor))
      throw new Error('Unsupported data');
    Object.defineProperty(result, key, {
      value: copyData(descriptor.value, parents),
      enumerable: true,
    });
  }
  parents.delete(value);
  return Object.freeze(result);
}
const record = (value: unknown): value is Record<string, unknown> =>
  !!value && typeof value === 'object' && !Array.isArray(value);
const text = (value: unknown): value is string =>
  typeof value === 'string' && !!value.trim();
/** Compare execution identity independently of optional display metadata. */
export function messageKey(input: Message): string {
  const message = copyData(input) as Message;
  return JSON.stringify({
    id: message.id,
    role: message.role,
    content: message.content,
  });
}
/** Capture exact saved root evidence; load fulfillment alone grants no authority. */
export function captureTerminal(
  input: ThreadsState,
  threadId: string
): Canonical | null {
  try {
    const state = copyData(input) as ThreadsState;
    if (
      !text(threadId) ||
      state.status !== 'idle' ||
      state.error ||
      !Array.isArray(state.interrupts) ||
      state.interrupts.length ||
      !Array.isArray(state.subgraphs) ||
      state.subgraphs.length ||
      !Array.isArray(state.toolCalls) ||
      state.toolCalls.length ||
      !Array.isArray(state.history) ||
      !state.history.length ||
      !Array.isArray(state.messages) ||
      !state.messages.length ||
      state.messages.length % 2
    )
      return null;
    const history = state.history[0],
      checkpoint = history.checkpoint;
    if (
      !record(checkpoint) ||
      checkpoint['thread_id'] !== threadId ||
      checkpoint['checkpoint_ns'] !== '' ||
      !text(checkpoint['checkpoint_id']) ||
      !Array.isArray(history.next) ||
      history.next.length
    )
      return null;
    const messages = state.messages;
    if (new Set(messages.map((m) => m.id)).size !== messages.length)
      return null;
    for (const [index, message] of messages.entries()) {
      if (
        !text(message.id) ||
        !text(message.content) ||
        message.role !== (index % 2 ? 'assistant' : 'user') ||
        message.toolCallId !== undefined ||
        (message.toolCallIds !== undefined &&
          (!Array.isArray(message.toolCallIds) ||
            message.toolCallIds.length)) ||
        message.delivery.phase !== 'complete' ||
        message.delivery.outcome !== 'success' ||
        message.delivery.generation !== message.id
      )
        return null;
    }
    const checkpointKey = JSON.stringify(checkpoint);
    return Object.freeze({
      threadId,
      checkpoint: checkpointKey,
      signature: JSON.stringify({
        checkpoint: checkpointKey,
        messages: messages.map(messageKey),
      }),
      messages,
    });
  } catch {
    return null;
  }
}

import { copyData, messageKey, type Canonical } from './authority';
export interface ConversationRecord {
  readonly key: string;
  readonly ordinal: number;
  readonly threadId: string | null;
  readonly label: string;
  readonly title?: string;
  readonly draft: string;
  readonly availability: 'available' | 'unavailable';
  readonly canonical: Canonical | null;
  readonly generations: readonly string[];
  readonly version: number;
}
/** Retains plain page-local records without keeping session or content owners alive. */
export class ConversationRecords {
  private readonly records = new Map<string, ConversationRecord>();
  private serial = 0;
  private require(key: string): ConversationRecord {
    const record = this.records.get(key);
    if (!record) throw new Error('Unknown conversation');
    return record;
  }
  private update(
    key: string,
    patch: Partial<ConversationRecord>
  ): ConversationRecord {
    const record = Object.freeze({ ...this.require(key), ...patch });
    this.records.set(key, record);
    return record;
  }
  /** Begin a blank local view while preserving every confirmed conversation. */
  create(): ConversationRecord {
    for (const [key, record] of this.records)
      if (!record.threadId) this.records.delete(key);
    const ordinal = ++this.serial;
    const record: ConversationRecord = Object.freeze({
      key: 'local-' + ordinal,
      ordinal,
      threadId: null,
      label: 'Conversation ' + ordinal,
      draft: '',
      availability: 'available',
      canonical: null,
      generations: Object.freeze([]),
      version: 0,
    });
    this.records.set(record.key, record);
    return record;
  }
  /** Look up only a key already owned by this page. */
  get(key: string): ConversationRecord | undefined {
    return this.records.get(key);
  }
  /** Return an immutable view of the page's local records. */
  list(): readonly ConversationRecord[] {
    return Object.freeze([...this.records.values()]);
  }
  /** Attach a confirmed remote ID without changing the record or draft identity. */
  confirmThread(key: string, id: string): void {
    const record = this.require(key);
    if (
      record.availability !== 'available' ||
      typeof id !== 'string' ||
      !id.trim() ||
      (record.threadId && record.threadId !== id) ||
      [...this.records.values()].some(
        (other) => other.key !== key && other.threadId === id
      )
    )
      throw new Error('Unconfirmed conversation');
    this.update(key, { threadId: id });
  }
  /** Retain draft text separately from canonical conversation evidence. */
  setDraft(key: string, draft: string): void {
    if (typeof draft !== 'string') throw new Error('Invalid draft');
    this.update(key, { draft });
  }
  /** Store a validated fresh turn that extends the previously confirmed prefix. */
  confirm(key: string, input: Canonical, generation: string): void {
    const canonical = copyData(input) as Canonical;
    const record = this.require(key),
      previous = record.canonical;
    if (
      record.availability !== 'available' ||
      !record.threadId ||
      canonical.threadId !== record.threadId ||
      typeof generation !== 'string' ||
      !generation.trim() ||
      record.generations.includes(generation) ||
      canonical.messages.length !== (previous?.messages.length ?? 0) + 2 ||
      (previous &&
        (previous.checkpoint === canonical.checkpoint ||
          previous.messages.some(
            (m, i) => messageKey(m) !== messageKey(canonical.messages[i])
          )))
    )
      throw new Error('Unconfirmed saved conversation');
    this.update(key, {
      canonical,
      generations: Object.freeze([...record.generations, generation]),
    });
  }
  /** Invalidate optional metadata work without changing conversation authority. */
  invalidate(key: string): number {
    const version = this.require(key).version + 1;
    this.update(key, { version });
    return version;
  }
  /** Apply presentation metadata only to its still-current known record version. */
  setTitle(key: string, title: string, version: number): boolean {
    const record = this.records.get(key);
    if (
      !record?.threadId ||
      record.availability !== 'available' ||
      record.version !== version ||
      typeof title !== 'string' ||
      !title.trim()
    )
      return false;
    this.update(key, { title, label: title });
    return true;
  }
  /** Permanently withhold continuation after uncertain work in this record. */
  markUnavailable(key: string): void {
    const record = this.require(key);
    this.update(key, {
      availability: 'unavailable',
      version: record.version + 1,
    });
  }
}

import type { CompleteOutcome, Message } from '@threadplane/core';
import {
  createMessageContent,
  type MessageRow,
} from '@threadplane/content/messages';
import type { ThreadsClient, ThreadsSession } from './connection';
import { ConversationRecords, type ConversationRecord } from './records';
export interface ThreadsSnapshot {
  readonly confirmation: 'saved' | 'loaded' | null;
  readonly selectedKey: string;
  readonly threadId: string | null;
  readonly conversations: readonly ConversationRecord[];
  readonly rows: readonly MessageRow[];
  readonly draft: string;
  readonly activity:
    | 'idle'
    | 'creating'
    | 'running'
    | 'confirming'
    | 'loading'
    | 'replacing';
  readonly busy: boolean;
  readonly canSubmit: boolean;
  readonly outcome: CompleteOutcome | null;
  readonly error: string | null;
  readonly viewGeneration: number;
}
import {
  captureTerminal,
  copyData,
  messageKey,
  type ThreadsState,
} from './authority';
import { decodeTitle } from './titles';
 
/** Own page-local records and one selected session; only matching saved evidence permits continuation. */
export function createThreadsApplication(options: ThreadsClient) {
  type Turn = {
    text: string;
    base: readonly Message[];
    generation?: string;
    positions: string[];
    completed: Map<string, string>;
    expected?: readonly Message[];
  };
  type Owner = {
    key: string;
    session: ThreadsSession;
    content: ReturnType<typeof createMessageContent>;
    closed: boolean;
    unsafe: boolean;
    release?: () => void;
    selecting: boolean;
    turn?: Turn;
  };
  type TitleRead = {
    key: string;
    version: number;
    view: number;
    controller: AbortController;
    timer: ReturnType<typeof setTimeout>;
  };
  const records = new ConversationRecords(),
    listeners = new Set<() => void>(),
    cleanups = new Set<Promise<void>>();
  let key = records.create().key,
    owner: Owner | undefined,
    operation: AbortController | undefined,
    titleRead: TitleRead | undefined;
  let disposed = false,
    disposal: Promise<void> | undefined;
  let snapshot: ThreadsSnapshot = Object.freeze({
    selectedKey: key,
    threadId: null,
    conversations: Object.freeze([]),
    rows: Object.freeze([]),
    draft: '',
    activity: 'idle',
    busy: false,
    canSubmit: true,
    outcome: null,
    error: null,
    viewGeneration: 0,
    confirmation: null,
  });
  function publish(update: Partial<ThreadsSnapshot> = {}) {
    if (disposed) return;
    const record = records.get(key);
    if (!record) return;
    snapshot = Object.freeze({
      ...snapshot,
      ...update,
      selectedKey: key,
      threadId: record.threadId,
      draft: record.draft,
      conversations: Object.freeze(
        records.list().filter((r) => r.threadId !== null)
      ),
    });
    for (const notify of [...listeners]) {
      if (disposed) break;
      notify();
    }
  }
  function bounded(work: () => 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(selected: Owner): Promise<void> {
    if (selected.closed) return Promise.resolve();
    selected.closed = true;
    try {
      selected.release?.();
    } catch {
      /* A failed unsubscribe cannot restore ownership. */
    }
    try {
      selected.content.dispose();
    } catch {
      /* Cleanup must still dispose the session. */
    }
    const cleanup = bounded(() => selected.session.dispose());
    cleanups.add(cleanup);
    void cleanup.then(() => cleanups.delete(cleanup));
    return cleanup;
  }
  function detach() {
    const previous = owner;
    owner = undefined;
    return previous ? close(previous) : Promise.resolve();
  }
  function cancelTitle() {
    const previous = titleRead;
    titleRead = undefined;
    if (records.get(key)) records.invalidate(key);
    if (previous) {
      clearTimeout(previous.timer);
      previous.controller.abort();
    }
  }
  function startTitle(selected: Owner) {
    const record = records.get(selected.key);
    if (
      disposed ||
      owner !== selected ||
      selected.closed ||
      snapshot.busy ||
      !snapshot.canSubmit ||
      !record?.threadId ||
      record.availability !== 'available'
    )
      return;
    const controller = new AbortController(),
      version = records.invalidate(selected.key),
      view = snapshot.viewGeneration;
    const pending: TitleRead = {
      key: selected.key,
      version,
      view,
      controller,
      timer: setTimeout(() => {
        if (titleRead === pending) {
          titleRead = undefined;
          controller.abort();
        }
      }, 5000),
    };
    titleRead = pending;
    const current = () =>
      !disposed &&
      titleRead === pending &&
      !controller.signal.aborted &&
      owner === selected &&
      !selected.closed &&
      key === pending.key &&
      snapshot.viewGeneration === view &&
      records.get(key)?.version === version;
    void Promise.resolve()
      .then(async () => {
        if (!current()) return;
        const raw = await options.readTitle(
          record.threadId as string,
          controller.signal
        );
        if (!current()) return;
        const title = decodeTitle(raw, record.threadId as string);
        if (!current()) return;
        if (title !== undefined && records.setTitle(key, title, version))
          publish();
      })
      .catch(() => undefined)
      .finally(() => {
        clearTimeout(pending.timer);
        if (titleRead === pending) titleRead = undefined;
      });
  }
  function block(selected?: Owner) {
    if (selected) selected.unsafe = true;
    records.markUnavailable(key);
    publish({
      canSubmit: false,
      outcome: 'error',
      confirmation: null,
      error:
        'This conversation could not be confirmed. Choose another conversation or start a new one.',
    });
  }
  const complete = (message: Message) =>
    message.delivery.phase === 'complete' &&
    message.delivery.outcome === 'success';
  const same = (left: Message, right: Message | undefined) =>
    !!right && messageKey(left) === messageKey(right);
  function inspect(selected: Owner, state: ThreadsState) {
    if (
      !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 ||
      new Set(state.messages.map((m) => m.id)).size !== state.messages.length
    )
      throw new Error('Unsupported conversation');
    for (const m of state.messages)
      if (
        typeof m.id !== 'string' ||
        !m.id.trim() ||
        typeof m.content !== 'string' ||
        !['user', 'assistant'].includes(m.role) ||
        m.toolCallId !== undefined ||
        (m.toolCallIds !== undefined &&
          (!Array.isArray(m.toolCallIds) || m.toolCallIds.length)) ||
        !['streaming', 'complete'].includes(m.delivery.phase) ||
        (m.delivery.phase === 'complete' && !complete(m))
      )
        throw new Error('Invalid message');
    const record = records.get(selected.key);
    if (!record) throw new Error('Unknown conversation');
    const prefix = record.canonical?.messages ?? [],
      turn = selected.turn;
    if (selected.selecting && state.messages.length === 0) return;
    if (
      state.messages.length < prefix.length ||
      prefix.some(
        (m, i) => !same(m, state.messages[i]) || !complete(state.messages[i])
      )
    )
      throw new Error('Changed saved prefix');
    if (!turn) {
      if (state.messages.length !== prefix.length)
        throw new Error('Unowned turn');
      if (!selected.selecting && record.canonical) {
        const canonical = captureTerminal(state, record.threadId ?? '');
        if (!canonical || canonical.signature !== record.canonical.signature)
          throw new Error('Changed idle saved authority');
      }
      return;
    }
    if (turn.expected) {
      if (
        state.messages.length !== turn.expected.length ||
        turn.expected.some(
          (m, i) =>
            !same(m, state.messages[i]) ||
            !complete(state.messages[i]) ||
            (state.messages[i].delivery.generation !== m.delivery.generation &&
              state.messages[i].delivery.generation !== state.messages[i].id)
        )
      )
        throw new Error('Changed saved response');
      return;
    }
    if (
      state.messages.length < turn.positions.length ||
      turn.positions.some((id, i) => state.messages[i].id !== id)
    )
      throw new Error('Changed message position');
    turn.positions = state.messages.map((m) => m.id);
    for (const m of state.messages) {
      const identity = messageKey(m),
        previous = turn.completed.get(m.id);
      if (previous !== undefined && previous !== identity)
        throw new Error('Changed completed message');
      if (complete(m)) turn.completed.set(m.id, identity);
    }
    const suffix = state.messages.slice(prefix.length);
    if (!suffix.length) return;
    const human = suffix[0];
    turn.generation ??= human.delivery.generation;
    if (
      suffix.length > 2 ||
      human.role !== 'user' ||
      human.content !== turn.text ||
      !complete(human) ||
      !turn.generation ||
      record.generations.includes(turn.generation) ||
      suffix.some((m) => m.delivery.generation !== turn.generation) ||
      suffix.some((m) => prefix.some((p) => p.id === m.id)) ||
      (suffix[1] && suffix[1].role !== 'assistant')
    )
      throw new Error('Foreign or changed live turn');
  }
  function read(
    selected: Owner,
    current: () => boolean
  ): ThreadsState | undefined {
    if (!current()) return;
    try {
      const raw = selected.session.getSnapshot();
      if (!current()) return;
      const state = copyData(raw) as ThreadsState;
      if (!current()) return;
      inspect(selected, state);
      if (!current()) return;
      const rows = selected.content.project(state);
      if (!current()) return;
      publish({ rows, ...(selected.unsafe ? { canSubmit: false } : {}) });
      if (!current()) return;
      const latest = selected.session.getSnapshot();
      if (!current()) return;
      if (latest !== raw) {
        block(selected);
        return;
      }
      return state;
    } catch {
      if (current()) block(selected);
      return;
    }
  }
  function install(
    id: string,
    selecting: boolean,
    current: () => boolean
  ): Owner | undefined {
    const session = options.sessionFactory(id),
      selected: Owner = {
        key,
        session,
        selecting,
        content: createMessageContent(),
        closed: false,
        unsafe: false,
      };
    if (!current()) {
      void close(selected);
      return;
    }
    owner = selected;
    const release = session.subscribe(() =>
      read(
        selected,
        () =>
          !disposed &&
          owner === selected &&
          !selected.closed &&
          selected.key === key &&
          !operation?.signal.aborted
      )
    );
    selected.release = release;
    if (selected.closed) release();
    if (!current() || owner !== selected) {
      void close(selected);
      return;
    }
    return selected;
  }
  function admit(activity: ThreadsSnapshot['activity']) {
    const admitted = new AbortController();
    operation = admitted;
    const current = () =>
      !disposed && operation === admitted && !admitted.signal.aborted;
    publish({
      activity,
      busy: true,
      canSubmit: false,
      error: null,
      outcome: null,
      confirmation: null,
    });
    if (current()) cancelTitle();
    return { admitted, current };
  }
  function finish(admitted: AbortController) {
    if (!disposed && operation === admitted) {
      operation = undefined;
      publish({ activity: 'idle', busy: false });
    }
  }
  async function submit(text: string) {
    if (
      disposed ||
      snapshot.busy ||
      !snapshot.canSubmit ||
      typeof text !== 'string' ||
      !text.trim()
    )
      return;
    records.setDraft(key, '');
    const { admitted, current } = admit(owner ? 'running' : 'creating');
    let succeeded: Owner | undefined;
    try {
      if (!current()) return;
      if (!owner) {
        const id = await options.createThread(admitted.signal);
        if (!current()) return;
        records.confirmThread(key, id);
        const selected = install(id, false, current);
        if (!selected || !current()) return;
        publish({ activity: 'running' });
      }
      const selected = owner;
      if (!selected || !current()) return;
      const valid = () =>
        current() &&
        owner === selected &&
        !selected.closed &&
        selected.key === key;
      const before = read(selected, valid);
      if (!valid()) return;
      if (!before || selected.unsafe || before.status !== 'idle') {
        block(selected);
        return;
      }
      selected.turn = {
        text,
        base: records.get(key)?.canonical?.messages ?? [],
        positions: before.messages.map((m) => m.id),
        completed: new Map(),
      };
      const result = await selected.session.submit(text, {
        signal: admitted.signal,
      });
      if (!valid()) return;
      const live = read(selected, valid);
      if (!valid()) return;
      const turn = selected.turn;
      if (
        result !== 'success' ||
        !live ||
        selected.unsafe ||
        live.status !== 'idle' ||
        !turn.generation ||
        live.messages.length !== turn.base.length + 2 ||
        live.messages.some((m) => !complete(m))
      ) {
        block(selected);
        return;
      }
      turn.expected = live.messages;
      publish({ activity: 'confirming' });
      if (!valid()) return;
      if (!read(selected, valid) || selected.unsafe) {
        if (valid()) block(selected);
        return;
      }
      if (!valid()) return;
      await selected.session.load({ signal: admitted.signal });
      if (!valid()) return;
      const saved = read(selected, valid);
      if (!valid()) return;
      const canonical = saved
        ? captureTerminal(saved, records.get(key)?.threadId ?? '')
        : null;
      if (!valid()) return;
      if (
        !canonical ||
        selected.unsafe ||
        canonical.messages.length !== turn.expected.length ||
        turn.expected.some((m, i) => !same(m, canonical.messages[i]))
      ) {
        block(selected);
        return;
      }
      records.confirm(key, canonical, turn.generation);
      selected.turn = undefined;
      publish({ canSubmit: true, outcome: 'success', confirmation: 'saved' });
      if (valid()) succeeded = selected;
    } catch {
      if (current()) block(owner);
    } finally {
      finish(admitted);
      if (succeeded) startTitle(succeeded);
    }
  }
  async function select(nextKey: string) {
    const next = records.get(nextKey);
    if (
      disposed ||
      snapshot.busy ||
      nextKey === key ||
      !next?.threadId ||
      !next.canonical ||
      next.availability !== 'available'
    )
      return;
    const { admitted, current } = admit('loading');
    let succeeded: Owner | undefined;
    try {
      if (!current()) return;
      const cleanup = detach();
      if (!current()) return;
      key = nextKey;
      publish({
        rows: Object.freeze([]),
        viewGeneration: snapshot.viewGeneration + 1,
      });
      if (!current()) return;
      await cleanup;
      if (!current()) return;
      const selected = install(next.threadId, true, current);
      if (!selected || !current()) return;
      const valid = () =>
        current() &&
        owner === selected &&
        !selected.closed &&
        key === selected.key;
      await selected.session.load({ signal: admitted.signal });
      if (!valid()) return;
      const state = read(selected, valid);
      if (!valid()) return;
      const canonical = state ? captureTerminal(state, next.threadId) : null;
      if (!valid()) return;
      if (
        !canonical ||
        selected.unsafe ||
        canonical.signature !== next.canonical.signature
      ) {
        block(selected);
        return;
      }
      selected.selecting = false;
      publish({ canSubmit: true, outcome: 'success', confirmation: 'loaded' });
      if (valid()) succeeded = selected;
    } catch {
      if (current()) block(owner);
    } finally {
      finish(admitted);
      if (succeeded) startTitle(succeeded);
    }
  }
  async function newConversation() {
    if (disposed || snapshot.busy) return;
    const { admitted, current } = admit('replacing');
    try {
      if (!current()) return;
      const cleanup = detach();
      if (!current()) return;
      await cleanup;
      if (!current()) return;
      key = records.create().key;
      publish({
        rows: Object.freeze([]),
        viewGeneration: snapshot.viewGeneration + 1,
        canSubmit: true,
        outcome: null,
        error: null,
        confirmation: null,
      });
    } finally {
      finish(admitted);
    }
  }
  async function stop() {
    if (disposed) return;
    const previous = operation;
    if (!previous) {
      cancelTitle();
      return;
    }
    const stopped = new AbortController();
    operation = stopped;
    previous.abort();
    cancelTitle();
    const selected = owner,
      cleanup = detach();
    records.markUnavailable(key);
    publish({
      busy: true,
      activity: 'replacing',
      canSubmit: false,
      outcome: 'aborted',
      error: null,
      confirmation: null,
    });
    await Promise.all([cleanup, bounded(() => selected?.session.stop())]);
    finish(stopped);
  }
  function setDraft(view: number, text: string) {
    if (
      disposed ||
      view !== snapshot.viewGeneration ||
      snapshot.activity === 'loading' ||
      snapshot.activity === 'replacing' ||
      records.get(key)?.availability !== 'available'
    )
      return;
    records.setDraft(key, text);
    publish();
  }
  function dispose(): Promise<void> {
    if (disposal) return disposal;
    disposed = true;
    const previous = operation;
    operation = undefined;
    previous?.abort();
    cancelTitle();
    void detach();
    listeners.clear();
    disposal = Promise.all([...cleanups]).then(() => undefined);
    return disposal;
  }
  return {
    getSnapshot: () => snapshot,
    subscribe(notify: () => void) {
      if (disposed) return () => undefined;
      listeners.add(notify);
      return () => {
        listeners.delete(notify);
      };
    },
    setDraft,
    submit,
    select,
    newConversation,
    stop,
    dispose,
  };
}