refactor(native-chat): share the catch-up open runner across agents (#31622)

* refactor(native-chat): share the catch-up open runner across agents

The Codex catch-up's agent-neutral parts move to one module: the open-time
envelope (its own deadline and cancellation, never gating the open), the
bounded staged batch, the single settlement commit that re-checks the
journal before writing, and the "Continued outside this chat." divider.
Codex keeps only how it reads its saved turns and what the chat already
holds. No behavior change; the other agents' catch-up builds on this
instead of copying it.

* chore: remove unrelated pnpm executable metadata

* refactor(native-chat): the catch-up runner reads the chat's view only when the agent asks

* chore: remove regenerated pnpm executable metadata

* fix(native-chat): the catch-up deadline also bounds waiting for its commit to be written

A journal write barrier that never settled after the import was admitted held
the provider's open past the catch-up deadline. The wait now ends with the
deadline or the open's cancellation, like every other step.

* test: type the history catch-up sink fixture without assertions

* fix(native-chat): bound and revoke history preparation

* feat(native-chat): validate expected saved items during history catch-up

src/main/codex/codex-history-import-plan.ts

@@ -4,7 +4,6 @@ import {
   CodexTurnOrdinals,
   readCodexThreadItem
 } from './codex-structured-item-translation'
-import { createHash } from 'node:crypto'
 import {
   agentJournalItemKey,
   agentJournalSubmissionKey
@@ -13,20 +12,17 @@ import type {
   AgentJournalItemBody,
   AgentJournalItemIdentity
 } from '../../shared/agent-session-journal-types'
-import {
-  journalLifecycleItemMutation,
-  type JournalLifecycleMutationInput
-} from '../native-chat/agent-session-journal/journal-row-builders'
+import { journalLifecycleItemMutation } from '../native-chat/agent-session-journal/journal-row-builders'
 import type {
   StructuredAgentSessionEventSink,
   StructuredAgentSessionItemAppendOptions
 } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
 import type { StructuredAgentSessionTransitionJournal } from '../native-chat/agent-session-wire/structured-agent-session-transition'
-import { createCodexJournalTranslator } from './codex-structured-journal-translation'
 import {
-  CODEX_RESTORE_MAX_BYTES,
-  CODEX_RESTORE_MAX_OPERATIONS
-} from './codex-structured-journal-translation-restore'
+  continuedOutsideRecordId,
+  createHistoryCatchUpBatch
+} from '../native-chat/agent-session-wire/structured-agent-session-history-catch-up'
+import { createCodexJournalTranslator } from './codex-structured-journal-translation'
 
 /** Invocation-local shared translation: nothing reaches the live sink until this plan is whole. */
 export function createCodexHistoryImportPlan(input: {
@@ -37,10 +33,9 @@ export function createCodexHistoryImportPlan(input: {
   outsideNotice?: AgentJournalItemBody
   check: () => void
 }) {
-  const mutations: JournalLifecycleMutationInput[] = []
+  const batch = createHistoryCatchUpBatch()
   const encoded = new Map<string, string>()
   const turnIds: string[] = []
-  let bytes = 0
   let outsideNoted = false
   let omitted = new Map<string, string>()
   const append = (
@@ -72,11 +67,7 @@ export function createCodexHistoryImportPlan(input: {
       }
       return { accepted: true } as const
     }
-    bytes += Buffer.byteLength(serialized, 'utf8') + 1
-    if (mutations.length >= CODEX_RESTORE_MAX_OPERATIONS || bytes > CODEX_RESTORE_MAX_BYTES) {
-      throw new Error('too-large')
-    }
-    mutations.push(mutation)
+    batch.push(mutation, serialized)
     encoded.set(key, serialized)
     return { accepted: true } as const
   }
@@ -141,7 +132,7 @@ export function createCodexHistoryImportPlan(input: {
             provider: 'legacy',
             agent: 'codex',
             sessionId: input.sessionId,
-            recordId: `continued-outside:${turn.id}`
+            recordId: continuedOutsideRecordId(turn.id)
           },
           input.outsideNotice,
           { turnScope: AGENT_JOURNAL_THREAD_SCOPE }
@@ -157,17 +148,13 @@ export function createCodexHistoryImportPlan(input: {
     finish: () => {
       input.check()
       const keys = [...encoded.keys()]
-      const digest = createHash('sha256')
-        .update(JSON.stringify([input.journal.epoch, input.threadId, turnIds, keys]))
-        .digest('hex')
-      const settlementId = `codex-history-import:${digest}`
-      // Match the existing batch envelope charge, plus the transition's one publication byte.
-      const reservedBytes =
-        Buffer.byteLength(JSON.stringify({ settlementId, mutations }), 'utf8') + 512
-      if (reservedBytes + 1 > CODEX_RESTORE_MAX_BYTES) {
-        throw new Error('too-large')
-      }
-      return { settlementId, mutations, keys, turnIds, reservedBytes }
+      const plan = batch.finish('codex-history-import', [
+        input.journal.epoch,
+        input.threadId,
+        turnIds,
+        keys
+      ])
+      return { ...plan, turnIds }
     },
     dispose: () => translator.dispose()
   }

src/main/codex/codex-structured-thread-catch-up.ts

@@ -1,12 +1,13 @@
-import { agentSessionHostStatusBody } from '../../shared/agent-session-host-status-rows'
 import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
 import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger'
+import {
+  continuedOutsideNotice,
+  runHistoryCatchUp
+} from '../native-chat/agent-session-wire/structured-agent-session-history-catch-up'
 import {
   DEFAULT_CODEX_REQUEST_TIMEOUT_MS,
   type CodexAppServerConnection
 } from './codex-app-server-connection'
-import { setImmediate } from 'node:timers/promises'
-import { waitForPromiseWithSignal } from '../../shared/abort-signal-reason'
 import { createCodexHistoryImportPlan } from './codex-history-import-plan'
 import { discoverCodexHistory, loadCodexHistory } from './codex-structured-thread-history-pages'
 import {
@@ -28,127 +29,48 @@ export async function catchUpCodexThread(input: {
   timeoutMs?: number
   logger?: StructuredAgentSessionLogger
 }): Promise<void> {
-  const controller = new AbortController()
-  const abort = () => controller.abort(input.signal?.reason)
-  input.signal?.addEventListener('abort', abort, { once: true })
-  if (input.signal?.aborted) {
-    abort()
-  }
-  const timeoutMs = input.timeoutMs ?? DEFAULT_CODEX_REQUEST_TIMEOUT_MS
-  const deadline = Date.now() + timeoutMs
-  const timer = setTimeout(() => controller.abort(new Error('catch-up-deadline')), timeoutMs)
-  const check = () => {
-    if (controller.signal.aborted || Date.now() >= deadline) {
-      throw new Error('catch-up-cancelled')
-    }
-    input.assertCurrent?.()
-  }
-  const warn = (reason: string, error?: unknown) =>
-    input.logger?.warn('Codex chat did not catch up on turns made outside it', {
-      scope: 'codex-thread-catch-up',
-      sessionId: input.sessionId,
-      reason,
-      ...(error === undefined ? {} : { error })
-    })
-  let staging: ReturnType<typeof createCodexHistoryImportPlan> | undefined
-  try {
-    check()
-    const written = await waitForPromiseWithSignal(
-      input.sink.written?.() ?? Promise.resolve({ ok: true }),
-      controller.signal
-    )
-    check()
-    const history = input.sink.journalHistory?.()
-    const journal = input.sink.journalItems?.()
-    if (!history || !journal || !written.ok || !input.sink.tryAppendTransition) {
-      return
-    }
-    const view = readCodexHistoryJournalView(history, input.threadId)
-    const read = {
-      connection: input.connection,
-      threadId: input.threadId,
-      signal: controller.signal,
-      deadline,
-      check
-    }
-    const missing = await discoverCodexHistory(read, view)
-    if (missing.pages.length === 0) {
-      return
-    }
-    staging = createCodexHistoryImportPlan({
-      sessionId: input.sessionId,
-      threadId: input.threadId,
-      journal,
-      nativePromptKeys: view.nativePromptKeys,
-      outsideNotice: view.hasRows
-        ? agentSessionHostStatusBody(
-            missing.rewound ? 'continued-outside-rewound' : 'continued-outside'
-          )
-        : undefined,
-      check
-    })
-    for await (const turn of loadCodexHistory(read, missing.pages)) {
-      check()
-      staging.restore(turn)
-      await setImmediate()
-    }
-    const plan = staging.finish()
-    let resolved = false
-    const admission = input.sink.tryAppendTransition({
-      lifecycle: true,
-      publish: true,
-      steps: [
-        {
-          kind: 'settlement',
-          settlementId: plan.settlementId,
-          reservedBytes: plan.reservedBytes,
-          resolve: (current) => {
-            try {
-              check()
-            } catch {
-              return []
-            }
-            if (
-              current.epoch !== view.epoch ||
-              input.sink.journalHistory?.() !== history ||
-              !sameCodexHistoryJournalView(
-                view,
-                readCodexHistoryJournalView(history, input.threadId)
-              )
-            ) {
-              return []
-            }
-            if (plan.keys.some((key) => current.item(current.canonicalItemId(key)) !== null)) {
-              return []
-            }
-            // Canonical keys can also collide with one another through an existing submission alias.
-            if (
-              new Set(plan.keys.map((key) => current.canonicalItemId(key))).size !==
-              plan.keys.length
-            ) {
-              return []
-            }
-            resolved = true
-            return plan.mutations
-          }
+  await runHistoryCatchUp({
+    agentName: 'Codex',
+    scope: 'codex-thread-catch-up',
+    sessionId: input.sessionId,
+    sink: input.sink,
+    ...(input.signal ? { signal: input.signal } : {}),
+    ...(input.assertCurrent ? { assertCurrent: input.assertCurrent } : {}),
+    timeoutMs: input.timeoutMs ?? DEFAULT_CODEX_REQUEST_TIMEOUT_MS,
+    ...(input.logger ? { logger: input.logger } : {}),
+    readView: (history) => readCodexHistoryJournalView(history, input.threadId),
+    sameView: sameCodexHistoryJournalView,
+    prepare: async ({ view: readView, journal, signal, deadline, check, yieldTurn }) => {
+      const view = readView()
+      const read = {
+        connection: input.connection,
+        threadId: input.threadId,
+        signal,
+        deadline,
+        check
+      }
+      const missing = await discoverCodexHistory(read, view)
+      if (missing.pages.length === 0) {
+        return null
+      }
+      const staging = createCodexHistoryImportPlan({
+        sessionId: input.sessionId,
+        threadId: input.threadId,
+        journal,
+        nativePromptKeys: view.nativePromptKeys,
+        outsideNotice: view.hasRows ? continuedOutsideNotice(missing.rewound) : undefined,
+        check
+      })
+      try {
+        for await (const turn of loadCodexHistory(read, missing.pages)) {
+          check()
+          staging.restore(turn)
+          await yieldTurn()
         }
-      ]
-    })
-    if (!admission.accepted) {
-      warn(admission.reason)
-      return
+        return staging.finish()
+      } finally {
+        staging.dispose()
+      }
     }
-    const committed = await input.sink.written?.()
-    if (committed?.ok === false) {
-      warn('write-failed', committed.error)
-    } else if (!resolved) {
-      warn('stale-history')
-    }
-  } catch (error) {
-    warn('import-skipped', error)
-  } finally {
-    clearTimeout(timer)
-    input.signal?.removeEventListener('abort', abort)
-    staging?.dispose()
-  }
+  })
 }

src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.test.ts

@@ -0,0 +1,198 @@
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key'
+import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding'
+import { createTrackedJournalOpener } from '../agent-session-journal/journal-host-database-test-support'
+import type { StructuredAgentSessionEventSink } from './structured-agent-session-event-sink'
+import { createDeferredStructuredAgentSessionEventSink } from './structured-agent-session-event-sink'
+import { testEventSinkLogging } from './structured-agent-session-logger-test-support'
+import type { HistoryCatchUpPlan } from './structured-agent-session-history-catch-up'
+import {
+  createHistoryCatchUpBatch,
+  runHistoryCatchUp
+} from './structured-agent-session-history-catch-up'
+
+describe('history catch-up runner', () => {
+  afterEach(() => vi.useRealTimers())
+  function fixture() {
+    let barriers = 0
+    const journal = {
+      epoch: 'epoch-1',
+      item: () => null,
+      itemBody: () => null,
+      canonicalItemId: (key: string) => key,
+      visitItems: vi.fn(),
+      visitItemsWithLinkage: vi.fn(),
+      receiptFor: () => null,
+      submissions: () => [],
+      historyOrigin: () => null
+    }
+    const append = vi.fn<NonNullable<StructuredAgentSessionEventSink['tryAppendTransition']>>(
+      () => ({ accepted: true })
+    )
+    const sink: StructuredAgentSessionEventSink = {
+      appendItem: vi.fn(),
+      appendTombstone: vi.fn(),
+      publish: vi.fn(),
+      // The barrier before reading settles; the one after the commit never does.
+      written: () => (barriers++ === 0 ? Promise.resolve({ ok: true }) : new Promise(() => {})),
+      journalHistory: () => journal,
+      journalItems: () => journal,
+      tryAppendTransition: append
+    }
+    const logger = { warn: vi.fn(), error: vi.fn() }
+    const input: Parameters<typeof runHistoryCatchUp<string>>[0] = {
+      agentName: 'Test',
+      scope: 'test-catch-up',
+      sessionId: 'session-1',
+      sink,
+      timeoutMs: 50,
+      logger,
+      readView: () => 'view',
+      sameView: (a, b) => a === b,
+      prepare: async () => plan
+    }
+    return { input, append, journal, logger }
+  }
+  const plan: HistoryCatchUpPlan = {
+    settlementId: 's',
+    mutations: [{ kind: 'tombstone', identity: { provider: 'orca', clientMessageId: 'old' } }],
+    keys: [agentJournalItemKey({ provider: 'orca', clientMessageId: 'old' })],
+    reservedBytes: 1
+  }
+  it('gives up at the deadline when the commit never becomes durable', async () => {
+    const { input, append, journal, logger } = fixture()
+    const started = Date.now()
+    const outcome = await runHistoryCatchUp(input)
+    expect(outcome).toBe('skipped')
+    expect(Date.now() - started).toBeLessThan(5_000)
+    expect(logger.warn).toHaveBeenCalledWith(
+      'Test chat did not catch up on turns made outside it',
+      expect.objectContaining({ reason: 'import-skipped' })
+    )
+    expect(append.mock.calls[0]?.[0].steps[0]?.resolve(journal)).toEqual([])
+  })
+  it('abandons stalled preparation at its deadline or parent cancellation without late admission', async () => {
+    vi.useFakeTimers()
+    for (const parent of [undefined, new AbortController()]) {
+      const late = Promise.withResolvers<HistoryCatchUpPlan | null>()
+      const { input, append } = fixture()
+      input.prepare = () => late.promise
+      const outcome = runHistoryCatchUp({ ...input, signal: parent?.signal })
+      await vi.advanceTimersByTimeAsync(parent ? 0 : 50)
+      parent?.abort()
+      expect(await outcome).toBe('skipped')
+      if (parent) {
+        late.reject(new Error('late source failure'))
+      } else {
+        late.resolve(plan)
+      }
+      await vi.advanceTimersByTimeAsync(0)
+      expect(append).not.toHaveBeenCalled()
+    }
+  })
+  it('checks absolute time after preparation even before timer delivery, including a no-op', async () => {
+    vi.useFakeTimers()
+    for (const prepared of [plan, null]) {
+      const { input, append } = fixture()
+      input.prepare = async () => {
+        vi.setSystemTime(Date.now() + 51)
+        return prepared
+      }
+      expect(await runHistoryCatchUp(input)).toBe('skipped')
+      expect(append).not.toHaveBeenCalled()
+    }
+  })
+  it('revokes an admitted resolver after an early failed barrier', async () => {
+    const { input, append, journal } = fixture()
+    input.sink.written = vi
+      .fn()
+      .mockResolvedValueOnce({ ok: true })
+      .mockResolvedValue({ ok: false, error: new Error('write failed') })
+    expect(await runHistoryCatchUp(input)).toBe('skipped')
+    const step = append.mock.calls[0]?.[0].steps[0]
+    expect(step?.resolve(journal)).toEqual([])
+  })
+  it('completes saved items only while all saved reads match, and checks empty plans without writes', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'orca-31622-expected-items-'))
+    const journals = createTrackedJournalOpener()
+    const deferred = createDeferredStructuredAgentSessionEventSink(testEventSinkLogging())
+    try {
+      const journal = await journals.open({
+        stateDirectory: root,
+        identity: {
+          sessionId: 'expected-items',
+          workspaceId: 'ws',
+          hostId: 'local',
+          agent: 'codex',
+          providerHandle: codexProviderHandle('thread')
+        }
+      })
+      const publish = vi.fn()
+      deferred.bind({ journal, fence: 1, publish })
+      const identity = { provider: 'codex' as const, threadId: 't', turnId: 't', ordinal: 0 }
+      const parent = { ...identity, ordinal: 1 }
+      const key = agentJournalItemKey(identity)
+      const parentKey = agentJournalItemKey(parent)
+      const body = (text: string) => ({
+        kind: 'message' as const,
+        role: 'assistant' as const,
+        blocks: [{ type: 'text' as const, text }]
+      })
+      const options = { fence: 1, turnScope: { kind: 'thread' as const } }
+      await journal.appendItem(identity, body('prefix'), options)
+      await journal.appendItem(parent, body('eligible parent'), options)
+      const original = journal.item(key)
+      const expected = (itemKey: string) => {
+        const item = journal.item(itemKey)
+        if (!item) {
+          throw new Error('missing saved item')
+        }
+        return { canonicalItemId: item.itemId, revision: item.revision }
+      }
+      const make = (text: string) => {
+        const batch = createHistoryCatchUpBatch()
+        batch.revise(
+          { kind: 'item', identity, body: body(text), turnScope: options.turnScope },
+          expected(key)
+        )
+        batch.expectItem(parentKey, expected(parentKey))
+        return batch.finish('complete', ['same source keys'])
+      }
+      const { input } = fixture()
+      const run = (prepared: HistoryCatchUpPlan) =>
+        runHistoryCatchUp({
+          ...input,
+          sink: deferred.sink,
+          timeoutMs: 5_000,
+          prepare: async () => prepared
+        })
+      const first = make('complete')
+      expect(make('different contents').settlementId).not.toBe(first.settlementId)
+      expect(await run(first)).toBe('committed')
+      expect(journal.item(key)).toMatchObject({
+        ...original,
+        body: body('complete'),
+        revision: (original?.revision ?? -1) + 1
+      })
+      const stale = make('later completion')
+      await journal.appendItem(parent, body('changed parent'), options)
+      const before = journal.cursor()
+      expect(await run(stale)).toBe('skipped')
+      expect(journal.cursor()).toEqual(before)
+      expect(journal.itemBody(key)).toEqual(body('complete'))
+      const empty = createHistoryCatchUpBatch()
+      empty.expectItem(key, expected(key))
+      const publications = publish.mock.calls.length
+      expect(await run(empty.finish('checked', []))).toBe('checked')
+      expect(journal.cursor()).toEqual(before)
+      expect(publish).toHaveBeenCalledTimes(publications)
+    } finally {
+      deferred.close()
+      await journals.closeAll()
+      await rm(root, { recursive: true, force: true })
+    }
+  })
+})

src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.ts

@@ -0,0 +1,282 @@
+// Catching a chat up on turns its agent saved outside it, for any agent: the open-time envelope
+// that never gates the open, the bounded staged batch, and its one checked commit. Each agent adds
+// only how it reads its own history and what the chat already holds.
+
+import { createHash } from 'node:crypto'
+import { setImmediate } from 'node:timers/promises'
+import { waitForPromiseWithSignal } from '../../../shared/abort-signal-reason'
+import { agentSessionHostStatusBody } from '../../../shared/agent-session-host-status-rows'
+import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key'
+import type { AgentJournalItemBody } from '../../../shared/agent-session-journal-types'
+import type { JournalLifecycleIdentityMutationInput } from '../agent-session-journal/journal-lifecycle-mutation'
+import type {
+  StructuredAgentSessionEventSink,
+  StructuredAgentSessionHistoryJournal,
+  StructuredAgentSessionSinkBarrier
+} from './structured-agent-session-event-sink'
+import type { StructuredAgentSessionLogger } from './structured-agent-session-logger'
+import type { StructuredAgentSessionTransitionJournal } from './structured-agent-session-transition'
+
+/** One catch-up writes at most this much, the same budget a whole restored history has. */
+export const HISTORY_CATCH_UP_MAX_OPERATIONS = 1_024
+export const HISTORY_CATCH_UP_MAX_BYTES = 16 * 1024 * 1024
+
+/** The one divider before an imported batch, keyed by the first imported turn or entry. */
+export function continuedOutsideRecordId(anchor: string): string {
+  return `continued-outside:${anchor}`
+}
+
+export function continuedOutsideNotice(rewound: boolean): AgentJournalItemBody {
+  return agentSessionHostStatusBody(rewound ? 'continued-outside-rewound' : 'continued-outside')
+}
+
+export type ExpectedHistoryItem = {
+  key: string
+  canonicalItemId: string
+  /** Saved read precondition; the journal assigns the written revision. */
+  revision: number
+}
+
+export type HistoryCatchUpPlan = {
+  settlementId: string
+  mutations: JournalLifecycleIdentityMutationInput[]
+  keys: string[]
+  reservedBytes: number
+  expectedItems?: readonly ExpectedHistoryItem[]
+}
+
+/** Rows staged in memory against the batch budget; nothing reaches the sink until the plan is whole. */
+export function createHistoryCatchUpBatch() {
+  const mutations: JournalLifecycleIdentityMutationInput[] = []
+  const keys: string[] = []
+  const expectedItems: ExpectedHistoryItem[] = []
+  let bytes = 0
+  let admissions = 0
+  const admit = (serialized: string, expected?: ExpectedHistoryItem) => {
+    bytes += Buffer.byteLength(serialized, 'utf8') + 1
+    if (expected) {
+      bytes += Buffer.byteLength(JSON.stringify(expected), 'utf8') + 1
+    }
+    if (admissions >= HISTORY_CATCH_UP_MAX_OPERATIONS || bytes > HISTORY_CATCH_UP_MAX_BYTES) {
+      throw new Error('too-large')
+    }
+    admissions++
+    if (expected) {
+      expectedItems.push(expected)
+    }
+  }
+  const stage = (
+    mutation: JournalLifecycleIdentityMutationInput,
+    serialized: string,
+    expected?: ExpectedHistoryItem
+  ) => {
+    admit(serialized, expected)
+    mutations.push(mutation)
+    keys.push(agentJournalItemKey(mutation.identity))
+  }
+  return {
+    get size(): number {
+      return mutations.length
+    },
+    push(mutation: JournalLifecycleIdentityMutationInput, serialized = JSON.stringify(mutation)) {
+      stage(mutation, serialized)
+    },
+    revise(
+      mutation: Extract<JournalLifecycleIdentityMutationInput, { kind: 'item' }>,
+      expected: Omit<ExpectedHistoryItem, 'key'>
+    ) {
+      stage(mutation, JSON.stringify(mutation), {
+        ...expected,
+        key: agentJournalItemKey(mutation.identity)
+      })
+    },
+    expectItem(key: string, expected: Omit<ExpectedHistoryItem, 'key'>) {
+      const item = { ...expected, key }
+      admit(JSON.stringify(item))
+      expectedItems.push(item)
+    },
+    /** `digestOf` names what was imported, so the same import settles under one id. */
+    finish(settlementPrefix: string, digestOf: unknown): HistoryCatchUpPlan {
+      const expectations = expectedItems.length ? { expectedItems: [...expectedItems] } : {}
+      const content = expectedItems.length ? [digestOf, expectedItems, mutations] : digestOf
+      const digest = createHash('sha256').update(JSON.stringify(content)).digest('hex')
+      const settlementId = `${settlementPrefix}:${digest}`
+      // Match the existing batch envelope charge, plus the transition's one publication byte.
+      const reservedBytes =
+        Buffer.byteLength(JSON.stringify({ settlementId, mutations, ...expectations }), 'utf8') +
+        512
+      if (reservedBytes + 1 > HISTORY_CATCH_UP_MAX_BYTES) {
+        throw new Error('too-large')
+      }
+      return { settlementId, mutations, keys: [...keys], reservedBytes, ...expectations }
+    }
+  }
+}
+
+export type HistoryCatchUpContext<View> = {
+  /** What the chat already holds, read once on first use: an agent that reads its own history first
+   *  touches the journal only when there is something to add. */
+  view: () => View
+  history: StructuredAgentSessionHistoryJournal
+  journal: StructuredAgentSessionTransitionJournal
+  signal: AbortSignal
+  deadline: number
+  /** Throws once the open is cancelled, the deadline passed, or the caller is no longer current. */
+  check: () => void
+  /** Hands the event loop back between pieces of a long read. */
+  yieldTurn: () => Promise<void>
+  warn: (reason: string, error?: unknown) => void
+}
+
+/** `checked`: an explicit empty plan validated; `nothing`: no plan or coverage proof. */
+export type HistoryCatchUpOutcome = 'committed' | 'checked' | 'nothing' | 'skipped'
+
+/** Optional reconciliation never gates provider open; an abandoned invocation cannot later write. */
+export async function runHistoryCatchUp<View>(input: {
+  /** Names the agent in the log line. */
+  agentName: string
+  scope: string
+  sessionId: string
+  sink: StructuredAgentSessionEventSink
+  signal?: AbortSignal
+  assertCurrent?: () => void
+  timeoutMs: number
+  logger?: StructuredAgentSessionLogger
+  /** What the chat already holds, read from saved journal facts. */
+  readView: (history: StructuredAgentSessionHistoryJournal) => View
+  /** Planning and commit must see the same saved facts. */
+  sameView: (planned: View, current: View) => boolean
+  /** Reads the agent's history and stages what to add; null adds nothing. */
+  prepare: (context: HistoryCatchUpContext<View>) => Promise<HistoryCatchUpPlan | null>
+}): Promise<HistoryCatchUpOutcome> {
+  const controller = new AbortController()
+  const abort = () => controller.abort(input.signal?.reason)
+  input.signal?.addEventListener('abort', abort, { once: true })
+  if (input.signal?.aborted) {
+    abort()
+  }
+  const deadline = Date.now() + input.timeoutMs
+  const timer = setTimeout(() => controller.abort(new Error('catch-up-deadline')), input.timeoutMs)
+  const check = () => {
+    if (controller.signal.aborted || Date.now() >= deadline) {
+      throw new Error('catch-up-cancelled')
+    }
+    input.assertCurrent?.()
+  }
+  const warn = (reason: string, error?: unknown) =>
+    input.logger?.warn(`${input.agentName} chat did not catch up on turns made outside it`, {
+      scope: input.scope,
+      sessionId: input.sessionId,
+      reason,
+      ...(error === undefined ? {} : { error })
+    })
+  const { sink } = input
+  try {
+    check()
+    const written = await waitForPromiseWithSignal(
+      sink.written?.() ?? Promise.resolve({ ok: true }),
+      controller.signal
+    )
+    check()
+    const history = sink.journalHistory?.()
+    const journal = sink.journalItems?.()
+    if (!history || !journal || !written.ok || !sink.tryAppendTransition) {
+      return 'skipped'
+    }
+    const epoch = history.epoch
+    let planned: { view: View } | undefined
+    const view = (): View => (planned ??= { view: input.readView(history) }).view
+    const plan = await waitForPromiseWithSignal(
+      input.prepare({
+        view,
+        history,
+        journal,
+        signal: controller.signal,
+        deadline,
+        check,
+        yieldTurn: () => setImmediate(),
+        warn
+      }),
+      controller.signal
+    )
+    check()
+    if (!plan) {
+      return 'nothing'
+    }
+    let resolved = false
+    const admission = sink.tryAppendTransition({
+      lifecycle: true,
+      publish: true,
+      steps: [
+        {
+          kind: 'settlement',
+          settlementId: plan.settlementId,
+          reservedBytes: plan.reservedBytes,
+          resolve: (current) => {
+            try {
+              check()
+            } catch {
+              return []
+            }
+            if (
+              current.epoch !== epoch ||
+              sink.journalHistory?.() !== history ||
+              !input.sameView(view(), input.readView(history))
+            ) {
+              return []
+            }
+            const verified = new Set<string>()
+            for (const expected of plan.expectedItems ?? []) {
+              const canonical = current.canonicalItemId(expected.key)
+              if (
+                verified.has(expected.key) ||
+                canonical !== expected.canonicalItemId ||
+                current.item(canonical)?.revision !== expected.revision
+              ) {
+                return []
+              }
+              verified.add(expected.key)
+            }
+            // Canonical keys can also collide with one another through an existing submission alias.
+            const seen = new Set<string>()
+            for (const mutation of plan.mutations) {
+              const key = agentJournalItemKey(mutation.identity)
+              const canonical = current.canonicalItemId(key)
+              if (seen.has(canonical) || (!verified.has(key) && current.item(canonical) !== null)) {
+                return []
+              }
+              seen.add(canonical)
+            }
+            resolved = true
+            return plan.mutations
+          }
+        }
+      ]
+    })
+    if (!admission.accepted) {
+      warn(admission.reason)
+      return 'skipped'
+    }
+    // The deadline bounds the commit too: a stalled barrier must not hold the provider's open.
+    const committed = await waitForPromiseWithSignal(
+      sink.written?.() ?? Promise.resolve<StructuredAgentSessionSinkBarrier>({ ok: true }),
+      controller.signal
+    )
+    if (!committed.ok) {
+      warn('write-failed', committed.error)
+      return 'skipped'
+    }
+    if (!resolved) {
+      warn('stale-history')
+      return 'skipped'
+    }
+    return plan.mutations.length ? 'committed' : 'checked'
+  } catch (error) {
+    warn('import-skipped', error)
+    return 'skipped'
+  } finally {
+    controller.abort()
+    clearTimeout(timer)
+    input.signal?.removeEventListener('abort', abort)
+  }
+}