From 4160c055d915aed891fff841ac7274500f2cd8e5 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Tue, 18 Aug 2026 21:45:31 -0700 Subject: [PATCH] fix(agent-session): harden journal recovery boundaries --- .../journal-compaction.ts | 9 + .../journal-epoch-replacement.ts | 102 +++++++++++ .../journal-epoch-rollover.ts | 20 ++- .../journal-legacy-import.test.ts | 65 ++++++- .../journal-legacy-import.ts | 13 +- .../agent-session-journal/journal-log-file.ts | 120 +++++++++++-- .../agent-session-journal/journal-open.ts | 21 ++- .../journal-pending-submission-recovery.ts | 18 ++ .../journal-row-builders.ts | 168 +++++++++++++++++ .../journal-store.test.ts | 71 ++++++++ .../agent-session-journal/journal-store.ts | 170 +++++++++--------- .../journal-write-guards.ts | 10 +- 12 files changed, 662 insertions(+), 125 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts create mode 100644 src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts create mode 100644 src/main/native-chat/agent-session-journal/journal-row-builders.ts diff --git a/src/main/native-chat/agent-session-journal/journal-compaction.ts b/src/main/native-chat/agent-session-journal/journal-compaction.ts index 9bc5dd21941..c45da2795a1 100644 --- a/src/main/native-chat/agent-session-journal/journal-compaction.ts +++ b/src/main/native-chat/agent-session-journal/journal-compaction.ts @@ -127,3 +127,12 @@ export function journalTailIsReadyToCompact( } return (tailRows[0]?.ts ?? now) < now - policy.retainTailMs } + +/** Budget pressure may need to shed rows before the ordinary batching threshold. */ +export function journalTailCanShedRows( + tailRows: readonly JournalRow[], + policy: JournalCompactionPolicy, + now: number +): boolean { + return retainTail(tailRows, policy, now).length < tailRows.length +} diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts new file mode 100644 index 00000000000..22a05e5c4fd --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts @@ -0,0 +1,102 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { join } from 'node:path' +import { copyFileDurable } from '../../durable-file-write' +import type { + AgentJournalItemBody, + AgentJournalItemIdentity, + AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import { compactJournal, type JournalCompactionPolicy } from './journal-compaction' +import { JOURNAL_LOG_FILE, JOURNAL_SNAPSHOT_FILE, appendJournalRows } from './journal-log-file' +import { applyJournalRow, createJournalReducerState } from './journal-reducer' +import { buildJournalItemRow, journalRowBase } from './journal-row-builders' +import type { AgentJournalEpochReason, JournalRow } from './journal-row-schema' +import { journalRowByteLength } from './journal-row-schema' +import { assertJournalFence, type JournalAppendBudget } from './journal-write-guards' +import type { JournalLoad } from './journal-open' + +export type JournalReplacementItem = { + identity: AgentJournalItemIdentity + body: AgentJournalItemBody + observedAt?: number +} + +export async function replaceJournalEpoch(input: { + journalDir: string + identity: AgentSessionJournalIdentity + reason: AgentJournalEpochReason + fence: number + items: readonly JournalReplacementItem[] + budget: JournalAppendBudget + compaction: JournalCompactionPolicy + now: () => number + mintEpoch: () => string + onSnapshotPublished: (loaded: JournalLoad) => void +}): Promise { + const stagingDir = await mkdtemp(join(input.journalDir, '.epoch-replacement-')) + try { + const epoch = input.mintEpoch() + const state = createJournalReducerState(input.identity.sessionId, epoch) + const epochRow: JournalRow = { + kind: 'epoch', + reason: input.reason, + providerHandle: input.identity.providerHandle, + ...journalRowBase(epoch, 1, input.fence, input.now()) + } + const rows: JournalRow[] = [epochRow] + applyJournalRow(state, epochRow) + let sizeBytes = journalRowByteLength(epochRow) + await appendJournalRows(stagingDir, [epochRow]) + + for (const item of input.items) { + const appendTime = input.now() + const row = buildJournalItemRow({ + state, + identity: item.identity, + body: item.body, + seq: state.lastSequence + 1, + fence: input.fence, + ts: item.observedAt ?? appendTime + }) + assertJournalFence(row.fence, state.highestFence) + input.budget.assert(row, appendTime, sizeBytes) + await appendJournalRows(stagingDir, [row]) + applyJournalRow(state, row) + rows.push(row) + sizeBytes += journalRowByteLength(row) + } + + const compacted = await compactJournal({ + journalDir: stagingDir, + state, + tailRows: rows, + policy: input.compaction, + now: input.now() + }) + await publishPreparedFile(stagingDir, input.journalDir, JOURNAL_SNAPSHOT_FILE) + state.oldestSequence = compacted.oldestSequence + input.onSnapshotPublished({ + state, + tailRows: compacted.tailRows, + compactedThrough: compacted.compactedThrough, + readOnly: false, + corrupt: false, + malformedRows: 0, + sizeBytes: compacted.tailRows.reduce((total, row) => total + journalRowByteLength(row), 0) + }) + await publishPreparedFile(stagingDir, input.journalDir, JOURNAL_LOG_FILE) + } finally { + await rm(stagingDir, { recursive: true, force: true }) + } +} + +async function publishPreparedFile( + stagingDir: string, + journalDir: string, + fileName: string +): Promise { + const copied = await copyFileDurable(join(stagingDir, fileName), join(journalDir, fileName)) + if (!copied) { + throw new Error(`prepared journal file disappeared before publish: ${fileName}`) + } +} diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-rollover.ts b/src/main/native-chat/agent-session-journal/journal-epoch-rollover.ts index b6b9a5ea9be..06379161c1c 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-rollover.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-rollover.ts @@ -8,12 +8,10 @@ import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' import type { AgentSessionProviderHandle } from '../../../shared/agent-session-journal-types' import { compactJournal } from './journal-compaction' -import { - applyJournalRow, - createJournalReducerState, - type JournalReducerState -} from './journal-reducer' +import { applyJournalRow, createJournalReducerState } from './journal-reducer' import type { AgentJournalEpochReason, JournalRow } from './journal-row-schema' +import { journalRowByteLength } from './journal-row-schema' +import type { JournalLoad } from './journal-open' export async function publishNewEpoch(input: { journalDir: string @@ -23,7 +21,7 @@ export async function publishNewEpoch(input: { reason: AgentJournalEpochReason fence: number now: number -}): Promise<{ state: JournalReducerState; row: JournalRow }> { +}): Promise { const row: JournalRow = { kind: 'epoch', reason: input.reason, @@ -44,5 +42,13 @@ export async function publishNewEpoch(input: { }) applyJournalRow(state, row) state.oldestSequence = 1 - return { state, row } + return { + state, + tailRows: [row], + compactedThrough: 0, + readOnly: false, + corrupt: false, + malformedRows: 0, + sizeBytes: journalRowByteLength(row) + } } diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts index 044315e81aa..dba8bcddd28 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts @@ -2,7 +2,7 @@ // results by identity read off the same raw lines. Fixtures are shaped like the // files the providers actually write. -import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it } from 'vitest' @@ -41,12 +41,17 @@ function identity(agent: 'claude' | 'codex', sessionId: string): AgentSessionJou } } -async function open(agent: 'claude' | 'codex', sessionId: string): Promise { +async function open( + agent: 'claude' | 'codex', + sessionId: string, + overrides: Partial[0]> = {} +): Promise { return openAgentSessionJournal({ identity: identity(agent, sessionId), journalDir: root, now: tick, - mintEpoch: () => `epoch-${clock}` + mintEpoch: () => `epoch-${clock}`, + ...overrides }) } @@ -420,6 +425,60 @@ describe('payload bounds on import', () => { }) describe('import failures', () => { + it('keeps the live epoch intact when a staged rebuild runs out of budget', async () => { + const limits = { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, maxSessionBytes: 2_000 } + const journal = await open('codex', CODEX_SESSION, { limits }) + await appendLegacyTranscriptMessages({ + journal, + agent: 'codex', + sessionId: CODEX_SESSION, + fence: 1, + messages: [ + { + id: 'durable-prefix', + role: 'assistant', + blocks: [{ type: 'text', text: 'keep me' }], + timestamp: 1_800_000_000_000, + source: 'transcript' + } + ] + }) + const filePath = await writeFixture('oversized-rollout.jsonl', [ + CODEX_LINES[0], + CODEX_LINES[1], + CODEX_LINES[2], + { + type: 'event_msg', + timestamp: '2026-08-05T10:00:03.000Z', + payload: { type: 'agent_message', message: 'x'.repeat(2_000) } + } + ]) + const epoch = journal.epoch + const snapshotPath = join(root, 'snapshot.json') + const logPath = join(root, 'log.jsonl') + const before = { + snapshot: await readFile(snapshotPath, 'utf-8'), + log: await readFile(logPath, 'utf-8') + } + + await expect( + importLegacyTranscriptIntoJournal({ + journal, + agent: 'codex', + sessionId: CODEX_SESSION, + fence: 1, + options: { filePath, limits } + }) + ).rejects.toMatchObject({ code: 'journal_bound_exceeded' }) + expect(journal.epoch).toBe(epoch) + expect(await readFile(snapshotPath, 'utf-8')).toBe(before.snapshot) + expect(await readFile(logPath, 'utf-8')).toBe(before.log) + expect(journal.snapshot().items[0]?.body).toMatchObject({ + kind: 'message', + blocks: [{ type: 'text', text: 'keep me' }] + }) + }) + it('reports a missing transcript without touching the journal', async () => { const journal = await open('claude', CLAUDE_SESSION) const before = journal.epoch diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts index 967704af5e2..f72e34c937a 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts @@ -29,6 +29,7 @@ import { import { decodeTranscriptStream } from '../transcript-stream-lines' import { putJournalBlob } from './journal-blob-store' import { createLegacyIdentityTracker } from './journal-legacy-identity' +import type { JournalReplacementItem } from './journal-epoch-replacement' import { boundInlineText, boundPayload, @@ -108,8 +109,7 @@ export async function importLegacyTranscriptIntoJournal(input: { return { ok: false, error: err instanceof Error ? err.message : String(err) } } - await input.journal.rollEpoch('legacy_import', input.fence) - let cursor = input.journal.cursor() + const replacement: JournalReplacementItem[] = [] for (const [index, message] of decoded.messages.entries()) { const identity = decoded.identities[index] if (!identity) { @@ -119,13 +119,14 @@ export async function importLegacyTranscriptIntoJournal(input: { for (const blob of mapped.blobs) { await putJournalBlob(input.journal.directory, blob.digest, blob.payload) } - const appended = await input.journal.appendItem(identity, mapped.body, { - fence: input.fence, + replacement.push({ + identity, + body: mapped.body, observedAt: message.timestamp ?? undefined }) - cursor = appended.cursor } - return { ok: true, epoch: input.journal.epoch, cursor, imported: decoded.messages.length } + const cursor = await input.journal.replaceEpochItems('legacy_import', input.fence, replacement) + return { ok: true, epoch: cursor.epoch, cursor, imported: decoded.messages.length } } const TRANSCRIPT_DECODERS = { diff --git a/src/main/native-chat/agent-session-journal/journal-log-file.ts b/src/main/native-chat/agent-session-journal/journal-log-file.ts index 19ee622303c..3d70d0a4ed6 100644 --- a/src/main/native-chat/agent-session-journal/journal-log-file.ts +++ b/src/main/native-chat/agent-session-journal/journal-log-file.ts @@ -11,7 +11,7 @@ import { appendFile, mkdir, open, readFile } from 'node:fs/promises' import { randomUUID } from 'node:crypto' import { join } from 'node:path' -import { durableWriteTempPath, writeFileDurable } from '../../durable-file-write' +import { durableWriteTempPath, renameDurable, writeFileDurable } from '../../durable-file-write' import type { AgentJournalRenderItem, AgentJournalSubmission @@ -56,8 +56,15 @@ export type JournalReadResult = { malformed: number /** Raw suffix beginning at the first malformed line, if any. */ remainder?: string + /** Distinguishes an absent/empty log from bytes that could not name an epoch. */ + hasBytes: boolean } +export type JournalSnapshotReadResult = + | { status: 'missing' } + | { status: 'valid'; snapshot: JournalSnapshotFile } + | { status: 'invalid' } + const NEWLINE_BYTE = 0x0a export async function ensureJournalDir(journalDir: string): Promise { @@ -67,15 +74,35 @@ export async function ensureJournalDir(journalDir: string): Promise { export async function readJournalSnapshotFile( journalDir: string ): Promise { + const result = await readJournalSnapshot(journalDir) + return result.status === 'valid' ? result.snapshot : null +} + +export async function readJournalSnapshot(journalDir: string): Promise { try { const raw = await readFile(join(journalDir, JOURNAL_SNAPSHOT_FILE), 'utf-8') - const parsed = JSON.parse(raw) as JournalSnapshotFile - return typeof parsed?.epoch === 'string' ? parsed : null - } catch { - return null + const parsed: unknown = JSON.parse(raw) + return isJournalSnapshotFile(parsed) + ? { status: 'valid', snapshot: parsed } + : { status: 'invalid' } + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + return { status: 'missing' } + } + if (error instanceof SyntaxError) { + return { status: 'invalid' } + } + throw error } } +export async function quarantineInvalidJournalSnapshot(journalDir: string): Promise { + const source = join(journalDir, JOURNAL_SNAPSHOT_FILE) + const target = join(journalDir, `quarantine-snapshot-${Date.now()}-${randomUUID()}.json`) + await renameDurable(source, target) + return target +} + export async function writeJournalSnapshotFile( journalDir: string, snapshot: JournalSnapshotFile @@ -88,8 +115,11 @@ export async function readJournalLog(journalDir: string): Promise 0 } } malformed += 1 - return { rows, unreadable, malformed, remainder: raw.slice(offset) } + return { rows, unreadable, malformed, remainder: raw.slice(offset), hasBytes: raw.length > 0 } } - return { rows, unreadable, malformed } + return { rows, unreadable, malformed, hasBytes: raw.length > 0 } +} + +function isJournalSnapshotFile(value: unknown): value is JournalSnapshotFile { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + return false + } + const snapshot = value as Record + return ( + typeof snapshot.v === 'number' && + typeof snapshot.epoch === 'string' && + snapshot.epoch.length > 0 && + typeof snapshot.compactedThrough === 'number' && + typeof snapshot.highestFence === 'number' && + arrayOf(snapshot.items, isRenderItem) && + arrayOf(snapshot.submissions, isSubmission) && + arrayOf(snapshot.receipts, isReceipt) && + arrayOf(snapshot.aliases, isAlias) && + arrayOf(snapshot.tail, (row) => parseJournalRow(JSON.stringify(row)).ok) + ) +} + +function arrayOf(value: unknown, predicate: (entry: unknown) => boolean): value is unknown[] { + return Array.isArray(value) && value.every(predicate) +} + +function recordOf(value: unknown): Record | null { + return value && typeof value === 'object' && !Array.isArray(value) + ? (value as Record) + : null +} + +function isRenderItem(value: unknown): boolean { + const item = recordOf(value) + return Boolean( + item && + typeof item.itemId === 'string' && + typeof item.revision === 'number' && + recordOf(item.body) + ) +} + +function isSubmission(value: unknown): boolean { + const submission = recordOf(value) + return Boolean(submission && typeof submission.clientMessageId === 'string') +} + +function isReceipt(value: unknown): boolean { + const receipt = recordOf(value) + return Boolean( + receipt && + typeof receipt.clientMessageId === 'string' && + typeof receipt.providerItemId === 'string' && + typeof receipt.epoch === 'string' && + typeof receipt.sequence === 'number' && + typeof receipt.acceptedAt === 'number' + ) +} + +function isAlias(value: unknown): boolean { + const alias = recordOf(value) + return Boolean( + alias && typeof alias.providerItemId === 'string' && typeof alias.itemId === 'string' + ) } /** @@ -150,8 +243,11 @@ export async function appendJournalRows( await handle.close() } } - } catch { - // The append below creates a missing log; other read errors remain visible. + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') { + throw error + } + // The append below creates a missing log. } const payload = `${rows.map(serializeJournalRow).join('\n')}\n` await appendFile(path, payload, 'utf-8') diff --git a/src/main/native-chat/agent-session-journal/journal-open.ts b/src/main/native-chat/agent-session-journal/journal-open.ts index 26facc1f563..1ab61240ac1 100644 --- a/src/main/native-chat/agent-session-journal/journal-open.ts +++ b/src/main/native-chat/agent-session-journal/journal-open.ts @@ -9,8 +9,9 @@ import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' import { findSequenceGap } from './journal-cursor' import { + quarantineInvalidJournalSnapshot, readJournalLog, - readJournalSnapshotFile, + readJournalSnapshot, type JournalSnapshotFile } from './journal-log-file' import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' @@ -44,11 +45,15 @@ export async function loadJournal( journalDir: string, sessionId: string ): Promise { - const snapshot = await readJournalSnapshotFile(journalDir) + const snapshotRead = await readJournalSnapshot(journalDir) + if (snapshotRead.status === 'invalid') { + await quarantineInvalidJournalSnapshot(journalDir) + } + const snapshot = snapshotRead.status === 'valid' ? snapshotRead.snapshot : null const log = await readJournalLog(journalDir) const epoch = resolveEpoch(snapshot, log.rows) if (!epoch) { - return log.unreadable ? emptyReadOnlyLoad(sessionId) : null + return snapshotRead.status === 'invalid' || log.hasBytes ? emptyReadOnlyLoad(sessionId) : null } const compactedThrough = snapshot?.epoch === epoch ? snapshot.compactedThrough : 0 @@ -110,18 +115,12 @@ function emptyReadOnlyLoad(sessionId: string): JournalLoad { } } -/** The snapshot names the live epoch; without one, the newest epoch row does. */ +/** The snapshot names the live epoch; without one, the newest valid row does. */ function resolveEpoch(snapshot: JournalSnapshotFile | null, rows: JournalRow[]): string | null { if (snapshot?.epoch) { return snapshot.epoch } - for (let index = rows.length - 1; index >= 0; index -= 1) { - const row = rows[index] - if (row?.kind === 'epoch') { - return row.epoch - } - } - return null + return rows.at(-1)?.epoch ?? null } function seedState( diff --git a/src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts b/src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts new file mode 100644 index 00000000000..76bc00394f2 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts @@ -0,0 +1,18 @@ +import type { AgentSessionJournal } from './journal-store' + +export async function markJournalPendingSubmissionsUnknown( + journal: AgentSessionJournal, + fence: number +): Promise { + const pending = journal.pendingSubmissions().map((entry) => entry.clientMessageId) + for (const clientMessageId of pending) { + await journal.resolveDispatch({ + clientMessageId, + state: 'unknown', + reason: 'host_restarted_before_acknowledgement', + fence, + recovered: true + }) + } + return pending +} diff --git a/src/main/native-chat/agent-session-journal/journal-row-builders.ts b/src/main/native-chat/agent-session-journal/journal-row-builders.ts new file mode 100644 index 00000000000..89a96465187 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-row-builders.ts @@ -0,0 +1,168 @@ +import type { + AgentJournalDispatchState, + AgentJournalItemBody, + AgentJournalItemIdentity, + AgentJournalMessageItem, + AgentSessionProviderHandle +} from '../../../shared/agent-session-journal-types' +import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import type { JournalReducerState } from './journal-reducer' +import type { + JournalDispatchRow, + JournalItemRow, + JournalSubmissionRow, + JournalTombstoneRow +} from './journal-row-schema' +import type { ResolveDispatchInput } from './journal-store-contracts' + +type RowBuilder = (seq: number, ts: number) => T + +export function journalItemRowBuilder( + state: () => JournalReducerState, + identity: AgentJournalItemIdentity, + body: AgentJournalItemBody, + options: { fence: number; observedAt?: number; recovered?: true } +): RowBuilder { + return (seq, ts) => + buildJournalItemRow({ + state: state(), + identity, + body, + seq, + fence: options.fence, + ts: options.observedAt ?? ts, + recovered: options.recovered + }) +} + +export function journalTombstoneRowBuilder( + state: () => JournalReducerState, + itemId: string, + fence: number +): RowBuilder { + return (seq, ts) => buildJournalTombstoneRow({ state: state(), itemId, seq, fence, ts }) +} + +export function journalSubmissionRowBuilder( + state: () => JournalReducerState, + providerHandle: AgentSessionProviderHandle, + input: { + clientMessageId: string + payloadFingerprint: string + body: AgentJournalMessageItem + fence: number + } +): RowBuilder { + return (seq, ts) => + buildJournalSubmissionRow({ state: state(), providerHandle, ...input, seq, ts }) +} + +export function journalDispatchRowBuilder( + state: () => JournalReducerState, + input: ResolveDispatchInput +): RowBuilder { + const providerItemId = + input.state === 'accepted' ? agentJournalItemKey(input.providerIdentity) : null + return (seq, ts) => + buildJournalDispatchRow({ + state: state(), + clientMessageId: input.clientMessageId, + dispatchState: input.state, + providerItemId, + reason: input.state === 'accepted' ? null : (input.reason ?? null), + seq, + fence: input.fence, + ts, + recovered: input.recovered + }) +} + +export function journalRowBase( + epoch: string, + seq: number, + fence: number, + ts: number +): { v: number; epoch: string; seq: number; fence: number; ts: number } { + return { v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, epoch, seq, fence, ts } +} + +export function buildJournalItemRow(input: { + state: JournalReducerState + identity: AgentJournalItemIdentity + body: AgentJournalItemBody + seq: number + fence: number + ts: number + recovered?: true +}): JournalItemRow { + const itemId = agentJournalItemKey(input.identity) + const resolved = input.state.aliases.get(itemId) ?? itemId + const revision = (input.state.items.get(resolved)?.revision ?? 0) + 1 + return { + kind: 'item', + itemId, + revision, + body: input.body, + ...journalRowBase(input.state.epoch, input.seq, input.fence, input.ts), + ...(input.recovered ? { recovered: input.recovered } : {}) + } +} + +export function buildJournalTombstoneRow(input: { + state: JournalReducerState + itemId: string + seq: number + fence: number + ts: number +}): JournalTombstoneRow { + const resolved = input.state.aliases.get(input.itemId) ?? input.itemId + return { + kind: 'tombstone', + itemId: input.itemId, + revision: (input.state.items.get(resolved)?.revision ?? 0) + 1, + ...journalRowBase(input.state.epoch, input.seq, input.fence, input.ts) + } +} + +export function buildJournalSubmissionRow(input: { + state: JournalReducerState + clientMessageId: string + payloadFingerprint: string + providerHandle: AgentSessionProviderHandle + body: AgentJournalMessageItem + seq: number + fence: number + ts: number +}): JournalSubmissionRow { + return { + kind: 'submission', + clientMessageId: input.clientMessageId, + payloadFingerprint: input.payloadFingerprint, + providerHandle: input.providerHandle, + body: input.body, + ...journalRowBase(input.state.epoch, input.seq, input.fence, input.ts) + } +} + +export function buildJournalDispatchRow(input: { + state: JournalReducerState + clientMessageId: string + dispatchState: Exclude + providerItemId: string | null + reason: string | null + seq: number + fence: number + ts: number + recovered?: true +}): JournalDispatchRow { + return { + kind: 'dispatch', + clientMessageId: input.clientMessageId, + state: input.dispatchState, + providerItemId: input.providerItemId, + reason: input.reason, + ...journalRowBase(input.state.epoch, input.seq, input.fence, input.ts), + ...(input.recovered ? { recovered: input.recovered } : {}) + } +} diff --git a/src/main/native-chat/agent-session-journal/journal-store.test.ts b/src/main/native-chat/agent-session-journal/journal-store.test.ts index 988820bd36d..08deef77163 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.test.ts @@ -200,6 +200,30 @@ describe('automatic compaction', () => { } expect(journal.compactionBoundary).toBe(0) }) + + it('compacts an aged tail before the size bound can latch out every append', async () => { + const journal = await open({ + limits: { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, maxSessionBytes: 900 }, + compaction: { minTailRows: 10, retainTailMs: 10_000 } + }) + let blockedIndex = 0 + for (; blockedIndex < 20; blockedIndex += 1) { + try { + await journal.appendItem(item(blockedIndex), body('x'.repeat(96)), { fence: 1 }) + } catch (error) { + expect(error).toMatchObject({ code: 'journal_bound_exceeded' }) + break + } + } + expect(blockedIndex).toBeLessThan(20) + + clock += 10_001 + await expect( + journal.appendItem(item(blockedIndex), body('after retention'), { fence: 1 }) + ).resolves.toBeDefined() + expect(journal.compactionBoundary).toBeGreaterThan(0) + expect(journal.snapshot().items).toHaveLength(blockedIndex + 1) + }) }) describe('compaction and retention', () => { @@ -352,6 +376,30 @@ describe('bounds', () => { }) describe('schema', () => { + it('quarantines an invalid compacted snapshot without replacing its tail', async () => { + const journal = await open({ compaction: { minTailRows: 2, retainTailMs: 0 } }) + for (let index = 0; index < 6; index += 1) { + await journal.appendItem(item(index), body(`m${index}`), { fence: 1 }) + } + await journal.compact() + const epoch = journal.epoch + const snapshotPath = join(root, JOURNAL_SNAPSHOT_FILE) + const logPath = join(root, JOURNAL_LOG_FILE) + const invalidSnapshot = '{"folded history":' + await writeFile(snapshotPath, invalidSnapshot, 'utf-8') + const retainedTail = await readFile(logPath, 'utf-8') + expect(retainedTail).not.toContain('"kind":"epoch"') + + const reopened = await open() + expect(reopened.epoch).toBe(epoch) + expect(await readFile(logPath, 'utf-8')).toBe(retainedTail) + const quarantined = (await readdir(root)).find((name) => + name.startsWith('quarantine-snapshot-') + ) + expect(quarantined).toBeDefined() + expect(await readFile(join(root, quarantined!), 'utf-8')).toBe(invalidSnapshot) + }) + it('degrades to read-only on a row from a newer build, without skipping or deleting it', async () => { const journal = await open() await journal.appendItem(item(0), body('a'), { fence: 1 }) @@ -480,6 +528,29 @@ describe('schema', () => { expect(reopened.snapshot().items).toHaveLength(0) expect((await readdir(root)).some((name) => name.startsWith('quarantine-'))).toBe(true) }) + + it('keeps the unreadable log suffix in the schema escape quarantine', async () => { + const journal = await open() + const logPath = join(root, JOURNAL_LOG_FILE) + const future = JSON.stringify({ + v: 99, + kind: 'item', + epoch: journal.epoch, + seq: 2, + fence: 1, + ts: 1, + itemId: 'future', + revision: 1, + body: { kind: 'status', text: 'preserve these bytes' } + }) + await writeFile(logPath, `${await readFile(logPath, 'utf-8')}${future}\n`, 'utf-8') + const reopened = await open() + + await reopened.rollEpoch('schema_unreadable', 2) + const quarantine = (await readdir(root)).find((name) => name.startsWith('quarantine-')) + expect(quarantine).toBeDefined() + expect(await readFile(join(root, quarantine!), 'utf-8')).toContain('preserve these bytes') + }) }) describe('journal location', () => { diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index b3d735485f5..27377aabd9c 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -16,14 +16,15 @@ import type { AgentJournalSubmission, AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' -import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' import { compactJournal, DEFAULT_JOURNAL_COMPACTION_POLICY, + journalTailCanShedRows, journalTailIsReadyToCompact, type JournalCompactionPolicy } from './journal-compaction' +import { replaceJournalEpoch, type JournalReplacementItem } from './journal-epoch-replacement' import { readJournalSince } from './journal-cursor' import { publishNewEpoch } from './journal-epoch-rollover' import { appendJournalRows, ensureJournalDir } from './journal-log-file' @@ -32,8 +33,9 @@ import { quarantineCorruptSuffix, quarantineUnreadableSchema } from './journal-corruption-quarantine' -import { loadJournal } from './journal-open' +import { loadJournal, type JournalLoad } from './journal-open' import { DEFAULT_JOURNAL_PAYLOAD_LIMITS } from './journal-payload-bounds' +import { markJournalPendingSubmissionsUnknown } from './journal-pending-submission-recovery' import { applyJournalRow, createJournalReducerState, @@ -42,6 +44,12 @@ import { resolveJournalItemId, type JournalReducerState } from './journal-reducer' +import { + journalDispatchRowBuilder, + journalItemRowBuilder, + journalSubmissionRowBuilder, + journalTombstoneRowBuilder +} from './journal-row-builders' import type { AgentSessionJournalOptions, JournalAppendResult, @@ -126,12 +134,7 @@ export class AgentSessionJournal { await this.startEpoch('session_created', 0) return } - this.state = loaded.state - this.tailRows = loaded.tailRows - this.compactedThrough = loaded.compactedThrough - this.sizeBytes = loaded.sizeBytes - this.readOnly = loaded.readOnly - this.malformedRows = loaded.malformedRows + this.adoptLoadedJournal(loaded) if (loaded.corrupt && !loaded.readOnly) { // The epoch stays put: no intact history is discarded to recover. await quarantineCorruptSuffix(this.journalDir, this.tailRows, loaded.quarantineRemainder) @@ -190,22 +193,13 @@ export class AgentSessionJournal { options: { fence: number; observedAt?: number; recovered?: true } = { fence: 0 } ): Promise { const itemId = agentJournalItemKey(identity) - return this.enqueue((seq, ts) => { - const resolved = this.state.aliases.get(itemId) ?? itemId - const revision = (this.state.items.get(resolved)?.revision ?? 0) + 1 - return { - kind: 'item', + return this.enqueue(journalItemRowBuilder(() => this.state, identity, body, options)).then( + (row) => ({ + cursor: { epoch: row.epoch, sequence: row.seq }, itemId, - revision, - body, - ...this.rowBase(seq, options.fence, options.observedAt ?? ts), - ...(options.recovered ? { recovered: options.recovered } : {}) - } - }).then((row) => ({ - cursor: { epoch: row.epoch, sequence: row.seq }, - itemId, - revision: (row as Extract).revision - })) + revision: (row as Extract).revision + }) + ) } appendTombstone( @@ -213,11 +207,9 @@ export class AgentSessionJournal { options: { fence: number } ): Promise { const itemId = agentJournalItemKey(identity) - return this.enqueue((seq, ts) => { - const resolved = this.state.aliases.get(itemId) ?? itemId - const revision = (this.state.items.get(resolved)?.revision ?? 0) + 1 - return { kind: 'tombstone', itemId, revision, ...this.rowBase(seq, options.fence, ts) } - }).then((row) => ({ epoch: row.epoch, sequence: row.seq })) + return this.enqueue(journalTombstoneRowBuilder(() => this.state, itemId, options.fence)).then( + (row) => ({ epoch: row.epoch, sequence: row.seq }) + ) } /** @@ -231,14 +223,9 @@ export class AgentSessionJournal { body: AgentJournalMessageItem fence: number }): Promise { - return this.enqueue((seq, ts) => ({ - kind: 'submission', - clientMessageId: input.clientMessageId, - payloadFingerprint: input.payloadFingerprint, - providerHandle: this.identity.providerHandle, - body: input.body, - ...this.rowBase(seq, input.fence, ts) - })).then((row) => ({ epoch: row.epoch, sequence: row.seq })) + return this.enqueue( + journalSubmissionRowBuilder(() => this.state, this.identity.providerHandle, input) + ).then((row) => ({ epoch: row.epoch, sequence: row.seq })) } /** @@ -249,42 +236,28 @@ export class AgentSessionJournal { * string here would silently give the user a second copy of their own message. */ resolveDispatch(input: ResolveDispatchInput): Promise { - const providerItemId = - input.state === 'accepted' ? agentJournalItemKey(input.providerIdentity) : null - return this.enqueue((seq, ts) => ({ - kind: 'dispatch', - clientMessageId: input.clientMessageId, - state: input.state, - providerItemId, - reason: input.state === 'accepted' ? null : (input.reason ?? null), - ...this.rowBase(seq, input.fence, ts), - ...(input.recovered ? { recovered: input.recovered } : {}) - })).then((row) => ({ epoch: row.epoch, sequence: row.seq })) + return this.enqueue(journalDispatchRowBuilder(() => this.state, input)).then((row) => ({ + epoch: row.epoch, + sequence: row.seq + })) } /** On restart every `pending` submission becomes `unknown` before the session * accepts a writer. Orca never re-sends on the user's behalf. */ async markPendingSubmissionsUnknown(fence: number): Promise { - const pending = this.pendingSubmissions().map((entry) => entry.clientMessageId) - for (const clientMessageId of pending) { - await this.resolveDispatch({ - clientMessageId, - state: 'unknown', - reason: 'host_restarted_before_acknowledgement', - fence, - recovered: true - }) - } - return pending + return markJournalPendingSubmissionsUnknown(this, fence) } - async compact(now = this.now()): Promise { + async compact( + now = this.now(), + policy: JournalCompactionPolicy = this.compaction + ): Promise { assertJournalWritable(this.readOnly, this.identity.sessionId) const result = await compactJournal({ journalDir: this.journalDir, state: this.state, tailRows: this.tailRows, - policy: this.compaction, + policy, now }) this.tailRows = result.tailRows @@ -306,34 +279,53 @@ export class AgentSessionJournal { return this.cursor() } - private async startEpoch(reason: AgentJournalEpochReason, fence: number): Promise { - const published = await publishNewEpoch({ - journalDir: this.journalDir, - sessionId: this.identity.sessionId, - providerHandle: this.identity.providerHandle, - epoch: this.mintEpoch(), - reason, - fence, - now: this.now() + replaceEpochItems( + reason: AgentJournalEpochReason, + fence: number, + items: readonly JournalReplacementItem[] + ): Promise { + const run = this.writes.then(async () => { + assertJournalWritable(this.readOnly, this.identity.sessionId) + assertJournalFence(fence, this.state.highestFence) + await replaceJournalEpoch({ + journalDir: this.journalDir, + identity: this.identity, + reason, + fence, + items, + budget: this.budget.fork(), + compaction: this.compaction, + now: this.now, + mintEpoch: this.mintEpoch, + onSnapshotPublished: (loaded) => this.adoptLoadedJournal(loaded) + }) + return this.cursor() }) - this.state = published.state - this.tailRows = [published.row] - this.compactedThrough = 0 - this.sizeBytes = journalRowByteLength(published.row) + this.writes = run.catch(() => undefined) + return run } - private rowBase( - seq: number, - fence: number, - ts: number - ): { v: number; epoch: string; seq: number; fence: number; ts: number } { - return { - v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, - epoch: this.state.epoch, - seq, - fence, - ts - } + private async startEpoch(reason: AgentJournalEpochReason, fence: number): Promise { + this.adoptLoadedJournal( + await publishNewEpoch({ + journalDir: this.journalDir, + sessionId: this.identity.sessionId, + providerHandle: this.identity.providerHandle, + epoch: this.mintEpoch(), + reason, + fence, + now: this.now() + }) + ) + } + + private adoptLoadedJournal(loaded: JournalLoad): void { + this.state = loaded.state + this.tailRows = loaded.tailRows + this.compactedThrough = loaded.compactedThrough + this.sizeBytes = loaded.sizeBytes + this.readOnly = loaded.readOnly + this.malformedRows = loaded.malformedRows } /** @@ -347,6 +339,14 @@ export class AgentSessionJournal { const ts = this.now() const row = build(this.state.lastSequence + 1, ts) assertJournalFence(row.fence, this.state.highestFence) + const budgetCompaction = { ...this.compaction, minTailRows: 0 } + if ( + this.autoCompact && + this.budget.wouldExceedSize(row, this.sizeBytes) && + journalTailCanShedRows(this.tailRows, budgetCompaction, ts) + ) { + await this.compact(ts, budgetCompaction) + } this.budget.assert(row, ts, this.sizeBytes) await appendJournalRows(this.journalDir, [row]) applyJournalRow(this.state, row) diff --git a/src/main/native-chat/agent-session-journal/journal-write-guards.ts b/src/main/native-chat/agent-session-journal/journal-write-guards.ts index eabdc9d750c..4914c6dcf61 100644 --- a/src/main/native-chat/agent-session-journal/journal-write-guards.ts +++ b/src/main/native-chat/agent-session-journal/journal-write-guards.ts @@ -52,8 +52,16 @@ export class JournalAppendBudget { private readonly limits: JournalPayloadLimits ) {} + fork(): JournalAppendBudget { + return new JournalAppendBudget(this.sessionId, this.limits) + } + + wouldExceedSize(row: JournalRow, sizeBytes: number): boolean { + return sizeBytes + journalRowByteLength(row) > this.limits.maxSessionBytes + } + assert(row: JournalRow, ts: number, sizeBytes: number): void { - if (sizeBytes + journalRowByteLength(row) > this.limits.maxSessionBytes) { + if (this.wouldExceedSize(row, sizeBytes)) { throw new AgentSessionJournalError( 'journal_bound_exceeded', `agent-session journal for ${this.sessionId} reached its ${this.limits.maxSessionBytes}-byte bound`