From 6cd2550e61348a86ca1a95f0d2f8dbb30e5db8f9 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 22 Oct 2026 01:08:46 -0700 Subject: [PATCH] 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 | 43 +-- .../codex/codex-structured-thread-catch-up.ts | 170 +++-------- ...red-agent-session-history-catch-up.test.ts | 198 ++++++++++++ ...ructured-agent-session-history-catch-up.ts | 282 ++++++++++++++++++ 4 files changed, 541 insertions(+), 152 deletions(-) create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.ts diff --git a/src/main/codex/codex-history-import-plan.ts b/src/main/codex/codex-history-import-plan.ts index 8f60858b4df..919f8144a57 100644 --- a/src/main/codex/codex-history-import-plan.ts +++ b/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() const turnIds: string[] = [] - let bytes = 0 let outsideNoted = false let omitted = new Map() 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() } diff --git a/src/main/codex/codex-structured-thread-catch-up.ts b/src/main/codex/codex-structured-thread-catch-up.ts index 02b90f94b59..3cb1a73e5b0 100644 --- a/src/main/codex/codex-structured-thread-catch-up.ts +++ b/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 { - 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 | 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() - } + }) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.test.ts new file mode 100644 index 00000000000..166990cfca7 --- /dev/null +++ b/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>( + () => ({ 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>[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() + 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 }) + } + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-history-catch-up.ts new file mode 100644 index 00000000000..651020e5f96 --- /dev/null +++ b/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, + expected: Omit + ) { + stage(mutation, JSON.stringify(mutation), { + ...expected, + key: agentJournalItemKey(mutation.identity) + }) + }, + expectItem(key: string, expected: Omit) { + 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 = { + /** 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 + 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(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) => Promise +}): Promise { + 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() + 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() + 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({ 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) + } +}