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 52603248426..15ecbf0ea6d 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 @@ -145,6 +145,22 @@ describe('replay', () => { expect(reopened.snapshot()).toEqual(live) }) + it('refuses rows whose JSON form would change the live render model', async () => { + const journal = await open() + for (const input of [undefined, Number.NaN]) { + await expect( + journal.appendItem( + item(0), + { kind: 'tool-call', name: 'bash', input, state: 'running' }, + { fence: 1 } + ) + ).rejects.toMatchObject({ code: 'journal_invalid_row' }) + } + + await journal.appendItem(item(0), body('persisted'), { fence: 1 }) + expect((await open()).snapshot()).toEqual(journal.snapshot()) + }) + it('serves a resume from a cursor and refuses one from a stale epoch', async () => { const journal = await open() await journal.appendItem(item(0), body('a'), { fence: 1 }) 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 f6881290d21..cc7ee6d8fc9 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -30,12 +30,7 @@ import { renderJournalState, type JournalReducerState } from './journal-reducer' -import { - type AgentJournalEpochReason, - journalRowBase, - journalRowByteLength, - type JournalRow -} from './journal-row-schema' +import { type AgentJournalEpochReason, journalRowBase, type JournalRow } from './journal-row-schema' import type { AgentSessionJournalOptions, JournalAppendResult, @@ -44,6 +39,7 @@ import type { } from './journal-store-contracts' import { assertJournalFence, + assertJournalRowPersistable, assertNewSubmission, assertJournalWritable, JournalAppendBudget @@ -342,11 +338,12 @@ export class AgentSessionJournal { const ts = this.now() const row = build(this.state.lastSequence + 1, ts) assertJournalFence(row.fence, this.state.highestFence) - this.budget.assert(row, ts, this.sizeBytes) + const rowSizeBytes = assertJournalRowPersistable(row) + this.budget.assert(rowSizeBytes, ts, this.sizeBytes) await appendJournalRows(this.journalDir, [row]) applyJournalRow(this.state, row) this.tailRows.push(row) - this.sizeBytes += journalRowByteLength(row) + this.sizeBytes += rowSizeBytes return 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 0274cecea6a..240455fc537 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 @@ -1,10 +1,11 @@ // Guards an append clears before it becomes durable. // -// All four refuse loudly rather than degrade: a silent drop here is a message +// These guards refuse loudly rather than degrade: a silent drop here is a message // missing from the transcript with nothing to explain it. +import { isDeepStrictEqual } from 'node:util' import type { JournalPayloadLimits } from './journal-payload-bounds' -import { journalRowByteLength, type JournalRow } from './journal-row-schema' +import { parseJournalRow, serializeJournalRow, type JournalRow } from './journal-row-schema' export class AgentSessionJournalError extends Error { constructor( @@ -12,6 +13,7 @@ export class AgentSessionJournalError extends Error { | 'journal_read_only' | 'journal_stale_fence' | 'journal_duplicate_submission' + | 'journal_invalid_row' | 'journal_bound_exceeded' | 'journal_rate_exceeded', message: string @@ -21,6 +23,25 @@ export class AgentSessionJournalError extends Error { } } +/** Refuse values JSON would drop or coerce, keeping live state identical to replay. */ +export function assertJournalRowPersistable(row: JournalRow): number { + let serialized = '' + let parsed: ReturnType + try { + serialized = serializeJournalRow(row) + parsed = parseJournalRow(serialized) + } catch { + parsed = { ok: false, unreadable: false } + } + if (!parsed.ok || !isDeepStrictEqual(parsed.row, row)) { + throw new AgentSessionJournalError( + 'journal_invalid_row', + `journal ${row.kind} row contains a value that cannot be persisted losslessly` + ) + } + return Buffer.byteLength(serialized, 'utf8') + 1 +} + export function assertNewSubmission(exists: boolean, clientMessageId: string): void { if (exists) { throw new AgentSessionJournalError( @@ -62,8 +83,8 @@ export class JournalAppendBudget { private readonly limits: JournalPayloadLimits ) {} - assert(row: JournalRow, ts: number, sizeBytes: number): void { - if (sizeBytes + journalRowByteLength(row) > this.limits.maxSessionBytes) { + assert(rowSizeBytes: number, ts: number, sizeBytes: number): void { + if (sizeBytes + rowSizeBytes > this.limits.maxSessionBytes) { throw new AgentSessionJournalError( 'journal_bound_exceeded', `agent-session journal for ${this.sessionId} reached its ${this.limits.maxSessionBytes}-byte bound`