React Deep Agents Memory 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 { useCallback, useEffect, useState } from 'react';
import { useAgent } from '@threadplane/react';
import {
  ChatInput,
  MessageList,
  ToolObservation,
  Reasoning,
  Citations,
} from '@threadplane/react/chat';
import { Markdown } from '@threadplane/react/markdown';
import type { MessageRow } from '@threadplane/content/messages';
import { MEMORY_PATH, type MemoryProjection } from './memory-projection';
import { copyData, LIMITS, type Authority } from './authority';
import {
  createConnectedApplication,
  type memoryConnection,
} from './connection';
 
function literal(value: unknown): string {
  try {
    const copied = copyData(value);
    const text =
      typeof copied === 'string'
        ? copied
        : JSON.stringify(copied, null, 2) ?? '';
    return text.length <= LIMITS.fieldText
      ? text
      : 'Observation text unavailable: display limit exceeded.';
  } catch {
    return 'Observation text unavailable.';
  }
}
 
/** Application owns row Markdown snapshots and tool/result association. */
export function renderMessage(
  row: MessageRow,
  authority: Authority = { kind: 'unconfirmed' }
) {
  if (row.role === 'tool') return null;
  const label =
    row.role === 'user'
      ? 'You'
      : row.role === 'assistant'
      ? 'Assistant'
      : 'System';
  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.toolCalls.map((call) => {
        const error =
          call.status === 'error' ||
          (authority.kind === 'terminal' &&
            authority.rawToolStatuses[call.id] === 'error');
        return (
          <div
            key={call.id}
            className="tool-observation"
            data-call-id={call.id}
          >
            <p>
              {error
                ? 'Tool result: error'
                : `Observed delivery: ${call.status}`}
            </p>
            <ToolObservation
              name={call.name}
              argumentsText={literal(call.args)}
              resultText={
                call.status === 'complete'
                  ? literal(call.result)
                  : call.status === 'error'
                  ? literal(call.error)
                  : undefined
              }
              label={error ? 'Observed tool error' : 'Observed tool delivery'}
            />
          </div>
        );
      })}
      {row.role === 'assistant' && row.message.citations && (
        <Citations citations={row.message.citations} />
      )}
    </article>
  );
}
 
export function MemoryPanel({ memory }: { memory: MemoryProjection }) {
  return (
    <section aria-label="Agent Memory" className="memory-panel">
      <h2>Agent Memory</h2>
      <p data-testid="memory-source" data-source={memory.kind}>
        {memory.kind === 'live'
          ? 'store observed during this run'
          : memory.kind === 'checkpoint'
          ? 'confirmed checkpoint at read time'
          : memory.kind === 'unavailable'
          ? 'memory observation unavailable'
          : 'no source yet'}
      </p>
      {(memory.kind === 'live' || memory.kind === 'checkpoint') &&
        (memory.file === 'missing' ? (
          <p>No memory file at this observation’s read time.</p>
        ) : (
          <div data-testid="memory-file">
            <h3>
              <code>{MEMORY_PATH}</code>
            </h3>
            {memory.contents === '' && (
              <p>The memory file is present and empty.</p>
            )}
            <pre>{memory.contents}</pre>
          </div>
        ))}
      <p>
        This panel shows one browser identity’s store at an observed read time.
        New conversations share that identity while browser storage is
        available. Checkpoints are historical; observations do not promise
        durability or that the store is current. Concurrent conversations are
        not serialized.
      </p>
    </section>
  );
}
 
export function MemoryDemo({
  connection,
  onReady,
}: {
  connection: ReturnType<typeof memoryConnection>;
  onReady: () => void;
}) {
  const [application] = useState(() => createConnectedApplication(connection));
  const [draft, setDraft] = useState('');
  const snapshot = useAgent(application);
  const renderRow = useCallback(
    (row: MessageRow) => renderMessage(row, snapshot.authority),
    [snapshot.authority]
  );
  useEffect(() => {
    onReady();
  }, [onReady]);
  useEffect(
    () => () => {
      void application.dispose();
    },
    [application]
  );
  useEffect(() => {
    setDraft('');
  }, [snapshot.viewGeneration]);
  const status = snapshot.busy
    ? snapshot.activity === 'confirming'
      ? 'Confirming saved conversation…'
      : snapshot.activity === 'restoring'
      ? 'Restoring conversation…'
      : snapshot.activity === 'creating'
      ? 'Creating conversation…'
      : 'Receiving response…'
    : !snapshot.canSubmit
    ? snapshot.scopeWarning
      ? 'Sending is unavailable. Reload after checking browser storage.'
      : 'Start a new conversation to continue.'
    : snapshot.outcome === 'success' &&
      snapshot.authority.kind === 'terminal' &&
      snapshot.memory.kind === 'checkpoint'
    ? 'Response saved.'
    : 'Ready.';
  return (
    <ExampleLayout className="memory-demo">
      <header>
        <div>
          <h1>Deep Agents Memory</h1>
        </div>
        <button
          type="button"
          onClick={() => {
            setDraft('');
            void application.newConversation();
          }}
        >
          New conversation
        </button>
      </header>
      <p>
        Teach the agent about your operation, then start a new conversation and
        ask what it remembers. Stop cancels local delivery; backend work may
        continue.
      </p>
      <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)
              setDraft(
                'I fly a Citation CJ3 out of KASE, and I always want briefings in bullet points. Please remember that.'
              );
          }}
        >
          Teach
        </button>
        <button
          type="button"
          disabled={snapshot.busy || !snapshot.canSubmit}
          onClick={() => {
            const current = application.getSnapshot();
            if (current.canSubmit && !current.busy)
              setDraft('What do you already know about my operation?');
          }}
        >
          Recall
        </button>
      </div>
      <label>
        Saved conversations{' '}
        <select
          aria-label="Saved conversations"
          value={snapshot.threadId ?? ''}
          disabled={snapshot.busy}
          onChange={(event) => {
            if (!event.target.value) return;
            setDraft('');
            void application.selectConversation(event.target.value);
          }}
        >
          <option value="">Choose a saved conversation</option>
          {snapshot.history.map((entry) => (
            <option key={entry.id} value={entry.id}>
              {entry.label}
            </option>
          ))}
        </select>
      </label>
      <div className="memory-workspace">
        <div className="memory-chat">
          <MessageList
            rows={snapshot.rows}
            renderMessage={renderRow}
            label="Conversation"
          />
          {snapshot.scopeWarning && (
            <p role="alert" data-testid="memory-scope-warning">
              {snapshot.scopeWarning}
            </p>
          )}
          <ExampleFooter>
            {snapshot.error && <p role="alert">{snapshot.error}</p>}
            <ChatInput
              key={snapshot.viewGeneration}
              label="Message"
              value={draft}
              onValueChange={setDraft}
              placeholder="Teach the agent or ask what it remembers…"
              hint="Enter to send. Shift+Enter adds a new line."
              busy={snapshot.busy}
              disabled={!snapshot.canSubmit && !snapshot.busy}
              onStop={snapshot.busy ? () => void application.stop() : undefined}
              onSubmit={(text) => {
                const current = application.getSnapshot();
                if (!current.canSubmit || current.busy || !text.trim())
                  return false;
                void application.submit(text);
                return true;
              }}
            />
            <p role="status">{status}</p>
          </ExampleFooter>
        </div>
        <aside className="memory-sidebar" aria-label="Memory sidebar">
          <MemoryPanel memory={snapshot.memory} />
        </aside>
      </div>
    </ExampleLayout>
  );
}

import { Client } from '@langchain/langgraph-sdk';
import { createSession } from '@threadplane/langgraph';
// eslint-disable-next-line @nx/enforce-module-boundaries -- Neutral bridge copied by isolated build.
import type { RuntimeBridgeConfiguration } from '../../../../../libs/cockpit-runtime-bridge/src/index';
import {
  createMemoryApplication,
  type MemoryClient,
  type MemorySession,
} from './application';
import { copyData, record, text, LIMITS, type Checkpoint } from './authority';
import { validScope } from './browser-scope';
export interface MemoryConnection {
  readonly apiUrl: string;
  readonly headers: Record<string, string>;
}
export function memoryConnection(
  configuration: RuntimeBridgeConfiguration
): MemoryConnection {
  if (
    configuration.status === 'error' ||
    (configuration.status === 'configured' &&
      configuration.target.kind === 'ag-ui')
  )
    throw Error('Runtime configuration failed.');
  const target =
    configuration.status === 'configured' &&
    configuration.target.kind === 'langsmith'
      ? configuration.target
      : null;
  return Object.freeze({
    apiUrl: target?.apiUrl ?? new URL('/api', window.location.origin).href,
    headers: Object.freeze(
      target ? { 'x-api-key': target.apiKey } : ({} as Record<string, string>)
    ),
  });
}
type Transport = NonNullable<Parameters<typeof createSession>[0]['transport']>;
type Event = ReturnType<Transport['stream']> extends AsyncIterable<infer T>
  ? T
  : never;
/** Routing comes only from the SDK event name and cursor, never payload fields. */
function normalize(event: string, data: unknown, id: unknown): Event {
  const [base, ...namespace] = event.split('|');
  const route = {
    type: event as Event['type'],
    ...(namespace.length ? { namespace } : {}),
    sseId: typeof id === 'string' ? id : undefined,
  };
  if (
    base === 'messages' &&
    Array.isArray(data) &&
    data.length === 2 &&
    record(data[1])
  )
    return { ...route, messages: [data[0]], messageMetadata: data[1], data };
  if (
    (base === 'messages' || base.startsWith('messages/')) &&
    Array.isArray(data)
  )
    return { ...route, messages: data, data };
  if (record(data))
    return {
      ...data,
      ...route,
      ...(Object.hasOwn(data, 'namespace') && !namespace.length
        ? { namespace: undefined }
        : {}),
      data,
    };
  return { ...route, data };
}
export function createMemoryClient(input: MemoryConnection): MemoryClient {
  const connection = Object.freeze({
    apiUrl: input.apiUrl,
    headers: Object.freeze({ ...input.headers }),
  });
  const sdk = new Client({
    apiUrl: connection.apiUrl,
    apiKey: null,
    defaultHeaders: connection.headers,
    callerOptions: { maxRetries: 0 },
  });
  const known = new Set<string>(),
    observers = new Map<string, (event: unknown) => void>();
  function owned(id: string | null, signal: AbortSignal): asserts id is string {
    if (!id || !known.has(id) || signal.aborted)
      throw Error('Conversation unavailable.');
  }
  const readHistory: MemoryClient['readHistory'] = async (id, signal) => {
    owned(id, signal);
    const result = await sdk.threads.getHistory(id, { limit: 1, signal });
    owned(id, signal);
    return result;
  };
  const readCheckpoint: MemoryClient['readCheckpoint'] = async (
    requested,
    signal
  ) => {
    const v = copyData(requested, false);
    if (
      !record(v) ||
      !text(v.thread_id) ||
      v.checkpoint_ns !== '' ||
      !text(v.checkpoint_id) ||
      v.checkpoint_id.length > LIMITS.fieldText ||
      Object.keys(v).some(
        (k) =>
          ![
            'thread_id',
            'checkpoint_ns',
            'checkpoint_id',
            'checkpoint_map',
          ].includes(k)
      ) ||
      (v.checkpoint_map !== undefined &&
        (!record(v.checkpoint_map) ||
          Object.keys(v.checkpoint_map).length !== 1 ||
          v.checkpoint_map[''] !== v.checkpoint_id))
    )
      throw Error('Checkpoint unavailable.');
    const cp = v as unknown as Checkpoint;
    owned(cp.thread_id, signal);
    const result = await sdk.threads.getState(
      cp.thread_id,
      cp as Parameters<typeof sdk.threads.getState>[1],
      { signal }
    );
    owned(cp.thread_id, signal);
    return result;
  };
  return {
    async createThread(signal) {
      if (signal.aborted) throw Error('Conversation unavailable.');
      const id = crypto.randomUUID();
      if (!validScope(id)) throw Error('Conversation unavailable.');
      const result = await sdk.threads.create({
        threadId: id,
        ifExists: 'raise',
        signal,
      });
      const captured = copyData(result, false);
      if (signal.aborted || !record(captured) || captured.thread_id !== id)
        throw Error('Conversation creation was not confirmed.');
      known.add(id);
      return id;
    },
    observeMemory(id, listener) {
      if (!known.has(id)) throw Error('Unknown conversation.');
      observers.set(id, listener);
      return () => {
        if (observers.get(id) === listener) observers.delete(id);
      };
    },
    sessionFactory(id) {
      if (!known.has(id)) throw Error('Unknown conversation.');
      const transport: Transport = {
        // Capture now, at physical invocation: a late event never adopts a later turn listener.
        stream(assistant, thread, payload, signal, options) {
          owned(thread, signal);
          const observe = observers.get(thread);
          const currentThread = thread;
          return (async function* () {
            try {
              for await (const packet of sdk.runs.stream(
                currentThread,
                assistant,
                {
                  input: payload as Record<string, unknown>,
                  context: options?.context,
                  config: options?.config,
                  metadata: options?.metadata,
                  streamMode: ['values', 'messages-tuple', 'updates', 'custom'],
                  streamSubgraphs: true,
                  signal,
                }
              )) {
                const event = normalize(packet.event, packet.data, packet.id);
                if (event.type === 'custom' && !event.namespace?.length)
                  observe?.(event.data);
                yield event;
              }
            } catch {
              throw Error('Conversation delivery unavailable.');
            }
          })();
        },
        async getHistory(thread, signal) {
          return (await readHistory(thread, signal)) as Awaited<
            ReturnType<NonNullable<Transport['getHistory']>>
          >;
        },
      };
      const session = createSession({
        assistantId: 'da-memory',
        threadId: id,
        transport,
      });
      try {
        const state = session.getSnapshot();
        if (
          ![
            'getSnapshot',
            'subscribe',
            'submit',
            'load',
            'stop',
            'dispose',
          ].every(
            (k) => typeof session[k as keyof typeof session] === 'function'
          ) ||
          ![
            state.messages,
            state.toolCalls,
            state.interrupts,
            state.subgraphs,
          ].every((x) => Array.isArray(x) && x.length === 0)
        )
          throw Error('Rejected session');
        return session as MemorySession;
      } catch {
        void Promise.resolve(session.dispose()).catch(() => undefined);
        throw Error('Runtime session is unavailable.');
      }
    },
    async readCurrent(id, signal) {
      owned(id, signal);
      const result = await sdk.threads.getState(id, undefined, { signal });
      owned(id, signal);
      return result;
    },
    readHistory,
    readCheckpoint,
  };
}
export function createConnectedApplication(connection: MemoryConnection) {
  return createMemoryApplication(createMemoryClient(connection));
}

import type {
  AgentSession,
  AgentStatus,
  CompleteOutcome,
  ToolCall,
} from '@threadplane/core';
import {
  createMessageContent,
  type MessageRow,
} from '@threadplane/content/messages';
import {
  captureAuthority,
  captureBaseline,
  captureTurn,
  copyData,
  record,
  sameData,
  LIMITS,
  type Authority,
  type Baseline,
  type Turn,
  type Checkpoint,
  type NativeSnapshot,
} from './authority';
import { projectMemory, type MemoryProjection } from './memory-projection';
import { createBrowserScope, type BrowserScope } from './browser-scope';
import type { LangGraphRunOptions } from '@threadplane/langgraph';
export interface MemorySession
  extends Omit<AgentSession, 'getSnapshot' | 'submit'> {
  getSnapshot(): NativeSnapshot;
  submit(
    input: string,
    options?: LangGraphRunOptions
  ): Promise<CompleteOutcome>;
  load(options?: { readonly signal?: AbortSignal }): Promise<void>;
}
export interface MemoryClient {
  observeMemory(id: string, listener: (event: unknown) => void): () => void;
  createThread(signal: AbortSignal): Promise<string>;
  sessionFactory(id: string): MemorySession;
  readCurrent(id: string, signal: AbortSignal): Promise<unknown>;
  readHistory(id: string, signal: AbortSignal): Promise<readonly unknown[]>;
  readCheckpoint(checkpoint: Checkpoint, signal: AbortSignal): Promise<unknown>;
}
export interface MemorySnapshot {
  readonly threadId: string | null;
  readonly rows: readonly MessageRow[];
  readonly toolCalls: readonly ToolCall[];
  readonly memory: MemoryProjection;
  readonly scopeWarning: string | null;
  readonly authority: Authority;
  readonly activity:
    | 'idle'
    | 'creating'
    | 'running'
    | 'confirming'
    | 'restoring';
  readonly busy: boolean;
  readonly canSubmit: boolean;
  readonly outcome: CompleteOutcome | null;
  readonly error: string | null;
  readonly viewGeneration: number;
  readonly history: readonly { id: string; label: string }[];
}
const initial = (
  viewGeneration = 0,
  history: MemorySnapshot['history'] = []
): MemorySnapshot => ({
  threadId: null,
  rows: [],
  toolCalls: [],
  memory: { kind: 'none' },
  scopeWarning: null,
  authority: { kind: 'unconfirmed' },
  activity: 'idle',
  busy: false,
  canSubmit: true,
  outcome: null,
  error: null,
  viewGeneration,
  history,
});
export function createMemoryApplication(
  client: MemoryClient,
  scope: BrowserScope = createBrowserScope({
    storage: () => window.localStorage,
    uuid: () => crypto.randomUUID(),
    events: window,
  })
) {
  type Terminal = Extract<Authority, { kind: 'terminal' }>;
  type Saved = { authority: Terminal; native: NativeSnapshot; label: string };
  type Owner = {
    id: string;
    key: string;
    generation: string;
    session: MemorySession;
    content: ReturnType<typeof createMessageContent>;
    baseline: Baseline;
    turn?: Turn;
    nativeGeneration?: string;
    expectedText?: string;
    confirmed?: NativeSnapshot;
    authority?: Terminal;
    release?: () => void;
    releaseMemory?: () => void;
    observation?: unknown;
    closed: boolean;
    unsafe: boolean;
    loading: boolean;
    resolved: Map<string, unknown>;
    completed: Map<string, unknown>;
  };
  let snapshot = initial(),
    owner: Owner | undefined,
    operation: AbortController | undefined,
    disposed = false,
    serial = 0,
    conversationSerial = 0,
    disposal: Promise<void> | undefined;
  const listeners = new Set<() => void>(),
    catalog = new Map<string, Saved>(),
    cleanups = new Set<Promise<void>>();
  const publish = (update: Partial<MemorySnapshot>) => {
    if (disposed) return;
    snapshot = Object.freeze({ ...snapshot, ...update });
    listeners.forEach((f) => f());
  };
  const scopeState = scope.getSnapshot();
  const memoryScope = scopeState.scope;
  const updateScope = () => {
    const state = scope.getSnapshot();
    publish({
      scopeWarning: state.status === 'blocked' ? state.warning : null,
      canSubmit:
        state.status === 'ready' &&
        !snapshot.busy &&
        !owner?.unsafe &&
        snapshot.outcome !== 'aborted' &&
        snapshot.outcome !== 'error',
    });
  };
  const releaseScope = scope.subscribe(updateScope);
  updateScope();
  const history = () => [...catalog].map(([id, s]) => ({ id, label: s.label }));
  function bounded(work: () => void | Promise<void>) {
    let timer: ReturnType<typeof setTimeout>;
    return Promise.race([
      Promise.resolve()
        .then(work)
        .catch(() => undefined),
      new Promise<void>((r) => {
        timer = setTimeout(r, 2000);
      }),
    ]).finally(() => clearTimeout(timer));
  }
  function close(o: Owner) {
    if (o.closed) return Promise.resolve();
    o.closed = true;
    // Each owned resource gets its release attempt even if another cleanup fails.
    try {
      o.release?.();
    } catch {
      /* Continue releasing the session. */
    }
    try {
      o.releaseMemory?.();
    } catch {
      /* Continue all cleanup attempts. */
    }
    try {
      o.content.dispose();
    } catch {
      /* Continue releasing the session. */
    }
    const p = bounded(() => o.session.dispose());
    cleanups.add(p);
    void p.finally(() => cleanups.delete(p));
    return p;
  }
  function detach() {
    const o = owner;
    owner = undefined;
    return o ? close(o) : Promise.resolve();
  }
  function block(o?: Owner) {
    if (o) {
      o.unsafe = true;
      catalog.delete(o.id);
    }
    publish({
      canSubmit: false,
      memory: { kind: 'unavailable' },
      authority: { kind: 'unconfirmed' },
      outcome: 'error',
      error:
        'This conversation could not be confirmed. Start a new conversation to continue.',
      history: history(),
    });
  }
  function install(
    o: Owner,
    s: NativeSnapshot,
    authority: Authority,
    source: 'observed-root' | 'checkpoint'
  ) {
    const memory = projectMemory({
      owner: o.key,
      generation: o.generation,
      currentOwner: o.key,
      currentGeneration: o.generation,
      scope: memoryScope,
      source:
        source === 'checkpoint'
          ? 'checkpoint'
          : o.observation
          ? 'custom'
          : 'none',
      event: o.observation,
      authority,
    });
    const rows = o.content.project({
      status: s.status as AgentStatus,
      messages: s.messages,
      toolCalls: s.toolCalls,
    });
    publish({
      rows,
      toolCalls: s.toolCalls,
      memory:
        source === 'checkpoint' || o.observation ? memory : { kind: 'none' },
      authority,
    });
  }
  function read(o: Owner, valid: () => boolean): NativeSnapshot | undefined {
    if (!valid() || o.loading || o.unsafe) return;
    try {
      const source = o.session.getSnapshot(),
        s = copyData(source) as NativeSnapshot;
      if (
        !record(s) ||
        ![s.messages, s.toolCalls, s.interrupts, s.subgraphs].every(
          Array.isArray
        )
      )
        throw Error('Malformed snapshot');
      if (s.subgraphs.length) throw Error('Unsupported child state');
      if (
        o.confirmed &&
        (s.messages.length < o.confirmed.messages.length ||
          o.confirmed.messages.some(
            (m, i) =>
              !sameData(
                { ...m, delivery: undefined },
                { ...s.messages[i], delivery: undefined }
              )
          ) ||
          o.confirmed.toolCalls.some(
            (t) =>
              !sameData(
                t,
                s.toolCalls.find((x) => x.id === t.id)
              )
          ))
      )
        throw Error('Changed prior baseline');
      if (
        o.expectedText !== undefined &&
        s.messages.length > o.baseline.messages.length
      ) {
        const h = s.messages[o.baseline.messages.length];
        if (
          h.role !== 'user' ||
          h.content !== o.expectedText ||
          !h.delivery?.generation ||
          s.messages
            .slice(o.baseline.messages.length + 1)
            .some((m) => m.role === 'user')
        )
          throw Error('Unowned human');
        if (!o.turn) {
          o.turn = captureTurn(o.baseline, {
            memoryScope: memoryScope!,
            owner: o.key,
            generation: o.generation,
            threadId: o.id,
            humanId: h.id,
            humanContent: o.expectedText,
          });
          o.nativeGeneration = h.delivery.generation;
          if (!o.turn) throw Error('Unavailable turn');
        }
        if (
          s.messages
            .slice(o.baseline.messages.length)
            .some((m) => m.delivery?.generation !== o.nativeGeneration)
        )
          throw Error('Foreign generation');
      }
      if (o.turn) {
        const h = s.messages[o.baseline.messages.length];
        if (
          !h ||
          h.id !== o.turn.humanId ||
          h.delivery.generation !== o.nativeGeneration
        )
          throw Error('Foreign generation');
      }
      for (const m of s.messages) {
        const semantic = { ...m, delivery: undefined };
        if (o.completed.has(m.id) && !sameData(o.completed.get(m.id), semantic))
          throw Error('Changed complete message');
        if (m.delivery?.phase === 'complete') o.completed.set(m.id, semantic);
      }
      for (const t of s.toolCalls) {
        if (o.resolved.has(t.id) && !sameData(o.resolved.get(t.id), t))
          throw Error('Changed resolved tool');
        if (t.status === 'complete' || t.status === 'error')
          o.resolved.set(t.id, t);
      }
      if (
        [...o.completed.keys()].some(
          (id) => !s.messages.some((m) => m.id === id)
        ) ||
        [...o.resolved.keys()].some(
          (id) => !s.toolCalls.some((t) => t.id === id)
        )
      )
        throw Error('Missing completed work');
      if (!valid()) return;
      if (o.session.getSnapshot() !== source) throw Error('Changed source');
      install(o, s, { kind: 'unconfirmed' }, 'observed-root');
      return s;
    } catch {
      if (valid()) block(o);
      return;
    }
  }
  function make(id: string): Owner {
    const o: Owner = {
      id,
      key: String(++serial),
      generation: String(++serial),
      session: client.sessionFactory(id),
      content: createMessageContent(),
      baseline: { messages: [] },
      closed: false,
      unsafe: false,
      loading: false,
      resolved: new Map(),
      completed: new Map(),
    };
    try {
      o.release = o.session.subscribe(() => {
        const active = operation;
        if (active && !o.loading)
          read(
            o,
            () =>
              !disposed &&
              operation === active &&
              !active.signal.aborted &&
              owner === o &&
              !o.closed
          );
      });
    } catch {
      void close(o);
      throw Error('Runtime subscription unavailable.');
    }
    return o;
  }
  async function confirm(
    o: Owner,
    controller: AbortController,
    live: NativeSnapshot,
    valid: () => boolean,
    retained?: Saved
  ) {
    const observedState = copyData(
      await client.readCurrent(o.id, controller.signal),
      false
    );
    if (!valid()) return;
    const h = copyData(
      await client.readHistory(o.id, controller.signal),
      false
    );
    if (!valid()) return;
    if (!Array.isArray(h) || !record(h[0])) throw Error('Missing history');
    const requested = copyData(h[0].checkpoint, false) as Checkpoint;
    const loadedCheckpoint = copyData(
      await client.readCheckpoint(requested, controller.signal),
      false
    );
    if (!valid()) return;
    o.loading = true;
    await o.session.load({ signal: controller.signal });
    if (!valid()) return;
    const source = o.session.getSnapshot(),
      loaded = copyData(source) as NativeSnapshot;
    if (!Array.isArray(loaded.subgraphs) || loaded.subgraphs.length)
      throw Error('Unsupported child state');
    const authority = captureAuthority({
      memoryScope,
      turn: o.turn,
      owner: o.key,
      generation: o.generation,
      threadId: o.id,
      outcome: 'success',
      observedState,
      history: h,
      loadedCheckpoint,
      snapshot: retained ? loaded : live,
      loadedSnapshot: loaded,
    });
    if (
      authority.kind !== 'terminal' ||
      (retained &&
        (!sameData(authority.checkpoint, retained.authority.checkpoint) ||
          !sameData(authority.values, retained.authority.values)))
    )
      throw Error('Unconfirmed root');
    if (!valid()) return;
    if (o.session.getSnapshot() !== source) throw Error('Changed publication');
    const projection = projectMemory({
      owner: o.key,
      currentOwner: o.key,
      generation: o.generation,
      currentGeneration: o.generation,
      scope: memoryScope,
      source: 'checkpoint',
      authority,
    });
    if (projection.kind !== 'checkpoint')
      throw Error('Memory confirmation unavailable.');
    const baseline = captureBaseline({ messages: authority.messages });
    if (!baseline) throw Error('Missing baseline');
    o.confirmed = loaded;
    o.authority = authority;
    o.baseline = baseline;
    o.loading = false;
    o.turn = undefined;
    o.expectedText = undefined;
    const saved: Saved = {
      authority,
      native: loaded,
      label: catalog.get(o.id)?.label ?? `Conversation ${++conversationSerial}`,
    };
    // Bound the entire page-local catalog, including repeated raw/native text.
    // Oldest entries leave the selectable catalog rather than remaining usable.
    copyData(saved);
    catalog.delete(o.id);
    catalog.set(o.id, saved);
    while (catalog.size > LIMITS.entries) {
      const oldest = catalog.keys().next().value;
      if (oldest !== undefined) catalog.delete(oldest);
    }
    while (true) {
      try {
        copyData([...catalog.values()]);
        break;
      } catch {
        const oldest = catalog.keys().next().value;
        if (oldest === undefined || oldest === o.id) {
          catalog.delete(o.id);
          throw Error('Unavailable catalog budget');
        }
        catalog.delete(oldest);
      }
    }
    install(o, loaded, authority, 'checkpoint');
    if (!valid()) return;
    publish({
      canSubmit: scope.getSnapshot().status === 'ready',
      outcome: 'success',
      history: history(),
    });
  }
  async function submit(text: string) {
    if (disposed || snapshot.busy || !snapshot.canSubmit || !text.trim())
      return;
    if (text.length > LIMITS.fieldText) {
      block(owner);
      return;
    }
    const context = scope.admit();
    if (!context) {
      updateScope();
      return;
    }
    const controller = new AbortController();
    operation = controller;
    const current = () =>
      !disposed && operation === controller && !controller.signal.aborted;
    publish({
      busy: true,
      canSubmit: false,
      activity: owner ? 'running' : 'creating',
      outcome: null,
      error: null,
      memory: { kind: 'none' },
      authority: { kind: 'unconfirmed' },
    });
    try {
      if (!current()) return;
      if (!owner) {
        const id = await client.createThread(controller.signal);
        if (!current()) return;
        if (typeof id !== 'string' || !id || id.length > LIMITS.fieldText)
          throw Error('Unknown thread');
        const o = make(id);
        if (!current()) {
          void close(o);
          return;
        }
        owner = o;
        publish({ threadId: id, activity: 'running' });
      }
      const o = owner,
        valid = () => current() && owner === o && !o.closed;
      const admitted = scope.admit();
      if (!admitted || admitted.memory_scope !== context.memory_scope) {
        updateScope();
        return;
      }
      o.generation = String(++serial);
      o.observation = undefined;
      o.turn = undefined;
      o.nativeGeneration = undefined;
      o.expectedText = text;
      o.completed = new Map();
      o.resolved = new Map();
      const before = read(o, valid);
      if (!valid()) return;
      if (
        !before ||
        o.unsafe ||
        before.messages.length !== o.baseline.messages.length
      )
        throw Error('Unowned baseline');
      o.releaseMemory?.();
      const observedGeneration = o.generation;
      o.releaseMemory = client.observeMemory(o.id, (event) => {
        if (!valid() || o.generation !== observedGeneration || o.unsafe) return;
        // Only actual matching custom observations are a Memory source.
        let captured: unknown;
        try {
          captured = copyData(event, false);
        } catch {
          block(o);
          return;
        }
        if (!record(captured) || captured.name !== 'deep_agents.memory') return;
        const projection = projectMemory({
          owner: o.key,
          currentOwner: o.key,
          generation: observedGeneration,
          currentGeneration: o.generation,
          scope: memoryScope,
          source: 'custom',
          event: captured,
        });
        if (projection.kind === 'unavailable') {
          block(o);
          return;
        }
        o.observation = captured;
        publish({ memory: projection });
      });
      const outcome = await o.session.submit(text, {
        context,
        signal: controller.signal,
      });
      if (!valid()) return;
      const live = read(o, valid);
      if (!valid()) return;
      const human = live?.messages[o.baseline.messages.length];
      if (
        !live ||
        o.unsafe ||
        outcome !== 'success' ||
        !human ||
        human.role !== 'user' ||
        human.content !== text ||
        !human.delivery?.generation ||
        live.messages
          .slice(o.baseline.messages.length + 1)
          .some((m) => m.role === 'user')
      )
        throw Error('Unconfirmed delivery');
      o.turn = captureTurn(o.baseline, {
        memoryScope: memoryScope!,
        owner: o.key,
        generation: o.generation,
        threadId: o.id,
        humanId: human.id,
        humanContent: text,
      });
      o.nativeGeneration = human.delivery.generation;
      if (
        !o.turn ||
        live.messages
          .slice(o.baseline.messages.length)
          .some((m) => m.delivery.generation !== o.nativeGeneration)
      )
        throw Error('Foreign generation');
      publish({ activity: 'confirming' });
      await confirm(o, controller, live, valid);
    } catch {
      if (current()) block(owner);
    } finally {
      if (current()) {
        operation = undefined;
        publish({ busy: false, activity: 'idle' });
        updateScope();
      }
    }
  }
  async function selectConversation(id: string) {
    if (disposed || !catalog.has(id)) return;
    if (operation && snapshot.threadId === id) {
      await stop();
      return;
    }
    const saved = catalog.get(id);
    if (!saved) return;
    if (operation && snapshot.threadId) catalog.delete(snapshot.threadId);
    operation?.abort();
    operation = undefined;
    const cleanup = detach();
    const controller = new AbortController();
    operation = controller;
    const current = () =>
      !disposed && operation === controller && !controller.signal.aborted;
    snapshot = initial(snapshot.viewGeneration + 1, history());
    publish({
      busy: true,
      canSubmit: false,
      activity: 'restoring',
      threadId: id,
    });
    updateScope();
    try {
      await cleanup;
      if (!current()) return;
      const o = make(id);
      owner = o;
      o.loading = true;
      o.turn = captureTurn(saved.authority.turn.baseline, {
        ...saved.authority.turn,
        owner: o.key,
        generation: o.generation,
      });
      const valid = () => current() && owner === o && !o.closed;
      await confirm(o, controller, saved.native, valid, saved);
    } catch {
      if (current()) {
        catalog.delete(id);
        block(owner);
      }
    } finally {
      if (current()) {
        operation = undefined;
        publish({ busy: false, activity: 'idle' });
        updateScope();
      }
    }
  }
  async function stop() {
    if (disposed || !operation) return;
    operation.abort();
    operation = undefined;
    const o = owner;
    const invalidated = o?.id ?? snapshot.threadId;
    if (invalidated) catalog.delete(invalidated);
    const p = detach();
    publish({
      busy: false,
      canSubmit: false,
      activity: 'idle',
      outcome: 'aborted',
      error: null,
      memory: { kind: 'none' },
      authority: { kind: 'unconfirmed' },
      history: history(),
    });
    await Promise.all([p, bounded(() => o?.session.stop())]);
  }
  async function newConversation() {
    if (disposed) return;
    if (operation && snapshot.threadId) catalog.delete(snapshot.threadId);
    operation?.abort();
    operation = undefined;
    const p = detach();
    snapshot = initial(snapshot.viewGeneration + 1, history());
    publish({});
    updateScope();
    await p;
  }
  function dispose() {
    if (disposal) return disposal;
    disposed = true;
    operation?.abort();
    operation = undefined;
    void detach();
    try {
      releaseScope();
    } catch {
      /* Finish cleanup. */
    }
    try {
      scope.dispose();
    } catch {
      /* Finish cleanup. */
    }
    listeners.clear();
    return (disposal = Promise.all([...cleanups]).then(() => undefined));
  }
  return {
    getSnapshot: () => snapshot,
    subscribe: (f: () => void) => {
      if (disposed) return () => undefined;
      listeners.add(f);
      return () => listeners.delete(f);
    },
    submit,
    stop,
    newConversation,
    selectConversation,
    dispose,
  };
}
export type MemoryApplication = ReturnType<typeof createMemoryApplication>;