From 72b6843a4c0bbff372bf48027b039b046bf1b2bb Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 7 Aug 2026 14:17:44 -0700 Subject: [PATCH] fix(native-chat): close journal replay edge cases --- .../journal-crash-boundary.test.ts | 27 ++++++++++ .../agent-session-journal/journal-log-file.ts | 12 +++-- .../agent-session-journal/journal-open.ts | 2 +- .../journal-reducer.test.ts | 43 +++++++++++++++ .../agent-session-journal/journal-reducer.ts | 18 ++++++- .../journal-row-schema.ts | 11 +++- .../journal-store.test.ts | 45 +++++++++++++++- .../agent-session-journal/journal-store.ts | 54 +++++++++---------- .../journal-write-guards.ts | 10 ++++ 9 files changed, 185 insertions(+), 37 deletions(-) diff --git a/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts b/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts index f34f2ce171c..795c2651bef 100644 --- a/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts @@ -195,6 +195,33 @@ describe('crash between provider accept and journal commit', () => { ).toEqual([]) }) + it('refuses to append a duplicate client message id after acceptance', async () => { + const journal = await open() + await journal.appendSubmission({ + clientMessageId: 'cm_1', + payloadFingerprint: digestPayload('once'), + body: userMessage('once'), + fence: 1 + }) + await journal.resolveDispatch({ + clientMessageId: 'cm_1', + state: 'accepted', + providerIdentity: ACCEPTED_IDENTITY, + fence: 1 + }) + + await expect( + journal.appendSubmission({ + clientMessageId: 'cm_1', + payloadFingerprint: digestPayload('twice'), + body: userMessage('twice'), + fence: 1 + }) + ).rejects.toMatchObject({ code: 'journal_duplicate_submission' }) + expect(journal.submissions()[0]?.dispatchState).toBe('accepted') + expect(journal.receiptFor('cm_1')).not.toBeNull() + }) + it('keeps the receipt after the row that minted it was compacted away', async () => { const journal = await openAgentSessionJournal({ identity: IDENTITY, 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 a951954bb0b..c2c58471322 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 @@ -8,7 +8,7 @@ // between publishing the snapshot and truncating the log leaves the log a // superset of the tail, and recovery unions the two by sequence — never a hole. -import { appendFile, mkdir, open, readFile } from 'node:fs/promises' +import { mkdir, open, readFile } from 'node:fs/promises' import { join } from 'node:path' import { durableWriteTempPath, writeFileDurable } from '../../durable-file-write' import type { @@ -139,9 +139,15 @@ export async function appendJournalRows( } const path = join(journalDir, JOURNAL_LOG_FILE) const payload = `${rows.map(serializeJournalRow).join('\n')}\n` - await appendFile(path, payload, 'utf-8') - const handle = await open(path, 'r+') + const handle = await open(path, 'a+') try { + const { size } = await handle.stat() + const lastByte = Buffer.alloc(1) + const needsBoundary = size > 0 && (await handle.read(lastByte, 0, 1, size - 1)).bytesRead === 1 + await handle.writeFile( + `${needsBoundary && lastByte[0] !== 0x0a ? '\n' : ''}${payload}`, + 'utf-8' + ) await handle.sync() } finally { await handle.close() 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 e51918f3138..983d1a85d32 100644 --- a/src/main/native-chat/agent-session-journal/journal-open.ts +++ b/src/main/native-chat/agent-session-journal/journal-open.ts @@ -61,7 +61,7 @@ export async function loadJournal( // covers `compactedThrough`, so a tail that starts above it lost rows. const corrupt = Boolean(gap) || oldest > compactedThrough + 1 - for (const row of liveRows) { + for (const row of tailRows) { if (row.seq > compactedThrough) { applyJournalRow(state, row) } diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.test.ts b/src/main/native-chat/agent-session-journal/journal-reducer.test.ts index 3ebb1dffc85..ea8c22f73d5 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.test.ts @@ -178,6 +178,49 @@ describe('submission and dispatch state machine', () => { expect(items[0]?.revision).toBe(1) }) + it('folds an early provider echo into the submission bubble when acceptance follows it', () => { + const state = fold([ + submission, + { + kind: 'item', + itemId: 'codex:thread-1:turn-1:0', + revision: 1, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] }, + ...base(2) + }, + { + kind: 'dispatch', + clientMessageId: 'cm_1', + state: 'accepted', + providerItemId: 'codex:thread-1:turn-1:0', + reason: null, + ...base(3) + } + ]) + const items = renderJournalState(state).items + expect(items).toHaveLength(1) + expect(items[0]?.itemId).toBe(agentJournalSubmissionKey('cm_1')) + expect(items[0]?.sequence).toBe(1) + expect(items[0]?.revision).toBe(1) + }) + + it('does not let a duplicate submission reopen an accepted dispatch', () => { + const state = fold([ + submission, + { + kind: 'dispatch', + clientMessageId: 'cm_1', + state: 'accepted', + providerItemId: 'codex:thread-1:turn-1:0', + reason: null, + ...base(2) + }, + { ...submission, payloadFingerprint: 'different', ...base(3) } + ]) + expect(state.submissions.get('cm_1')?.dispatchState).toBe('accepted') + expect(state.receipts.has('cm_1')).toBe(true) + }) + it('treats rejected as terminal', () => { const state = fold([ submission, diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts index 31898d4c5e7..91cd2de368e 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -48,6 +48,10 @@ export function createJournalReducerState(sessionId: string, epoch: string): Jou } } +export function nextJournalItemRevision(state: JournalReducerState, itemId: string): number { + return Math.max(state.items.get(itemId)?.revision ?? 0, state.tombstones.get(itemId) ?? 0) + 1 +} + export function applyJournalRow(state: JournalReducerState, row: JournalRow): void { state.lastSequence = Math.max(state.lastSequence, row.seq) state.highestFence = Math.max(state.highestFence, row.fence) @@ -122,6 +126,9 @@ function applySubmission( state: JournalReducerState, row: Extract ): void { + if (state.submissions.has(row.clientMessageId)) { + return + } state.submissions.set(row.clientMessageId, { clientMessageId: row.clientMessageId, fence: row.fence, @@ -161,7 +168,16 @@ function applyDispatch( if (row.state !== 'accepted' || !row.providerItemId) { return } - state.aliases.set(row.providerItemId, agentJournalSubmissionKey(row.clientMessageId)) + const submissionItemId = agentJournalSubmissionKey(row.clientMessageId) + const providerItem = state.items.get(row.providerItemId) + state.aliases.set(row.providerItemId, submissionItemId) + if (providerItem) { + upsertItem(state, submissionItemId, providerItem.revision, { + ...providerItem, + itemId: submissionItemId + }) + state.items.delete(row.providerItemId) + } state.receipts.set(row.clientMessageId, { clientMessageId: row.clientMessageId, providerItemId: row.providerItemId, diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index 6f89d9e6395..f7c15afe1ee 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -13,7 +13,7 @@ import { type AgentSessionProviderHandle } from '../../../shared/agent-session-journal-types' -type JournalRowBase = { +export type JournalRowBase = { /** Schema version of THIS row. */ v: number epoch: string @@ -92,6 +92,15 @@ export type JournalRowParse = const ROW_KINDS = new Set(['epoch', 'item', 'tombstone', 'submission', 'dispatch']) +export function journalRowBase( + epoch: string, + seq: number, + fence: number, + ts: number +): JournalRowBase { + return { v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, epoch, seq, fence, ts } +} + export function serializeJournalRow(row: JournalRow): string { return JSON.stringify(row) } 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 8475e4c4349..ae5d613d469 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 @@ -90,6 +90,17 @@ describe('sequences', () => { expect(journal.snapshot().items).toHaveLength(1) expect(journal.snapshot().items[0]?.revision).toBe(3) }) + + it('assigns a revision above a tombstone when an item is re-created', async () => { + const journal = await open() + await journal.appendItem(item(0), body('first'), { fence: 1 }) + await journal.appendTombstone(item(0), { fence: 1 }) + + const recreated = await journal.appendItem(item(0), body('back'), { fence: 1 }) + expect(recreated.revision).toBe(3) + expect(journal.snapshot().items.map((entry) => entry.body)).toEqual([body('back')]) + expect((await open()).snapshot()).toEqual(journal.snapshot()) + }) }) describe('fences', () => { @@ -170,6 +181,21 @@ describe('replay', () => { expect((await open()).snapshot()).toEqual(journal.snapshot()) }) + it('does not reuse the epoch sequence when rollover crashes before rewriting the log', async () => { + const journal = await open() + await journal.appendItem(item(0), body('old epoch'), { fence: 1 }) + const logBefore = await readFile(join(root, JOURNAL_LOG_FILE), 'utf-8') + + await journal.rollEpoch('handle_forked', 2) + await writeFile(join(root, JOURNAL_LOG_FILE), logBefore, 'utf-8') + + const reopened = await open() + expect(reopened.cursor()).toEqual({ epoch: journal.epoch, sequence: 1 }) + const appended = await reopened.appendItem(item(1), body('new epoch'), { fence: 2 }) + expect(appended.cursor.sequence).toBe(2) + expect((await open()).snapshot()).toEqual(reopened.snapshot()) + }) + it('rolls the epoch when the log lost a row', async () => { const journal = await open() for (let index = 0; index < 4; index += 1) { @@ -250,8 +276,9 @@ describe('compaction and retention', () => { await expect( reopened.appendItem(item(1), body('stale writer'), { fence: 6 }) ).rejects.toMatchObject({ code: 'journal_stale_fence' }) - await reopened.appendItem(item(0), body('late stale revision'), { fence: 7 }) - expect(reopened.snapshot().items).toHaveLength(0) + const recreated = await reopened.appendItem(item(0), body('re-created'), { fence: 7 }) + expect(recreated.revision).toBe(3) + expect(reopened.snapshot().items.map((entry) => entry.body)).toEqual([body('re-created')]) }) it('prunes blobs no live row references and keeps the ones that survive', async () => { @@ -419,6 +446,20 @@ describe('schema', () => { expect(reopened.isReadOnly).toBe(false) expect(reopened.snapshot().items).toHaveLength(1) }) + + it('separates the next append from a malformed non-newline tail', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 1 }) + const logPath = join(root, JOURNAL_LOG_FILE) + await writeFile(logPath, `${await readFile(logPath, 'utf-8')}{"v":1`, 'utf-8') + + const reopened = await open() + await reopened.appendItem(item(1), body('b'), { fence: 1 }) + + const replayed = await open() + expect(replayed.snapshot().items.map((entry) => entry.body)).toEqual([body('a'), body('b')]) + expect(replayed.cursor()).toEqual(reopened.cursor()) + }) }) 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 6ea1220b423..f6881290d21 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -16,7 +16,6 @@ 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, type JournalCompactionPolicy } from './journal-compaction' import { resolveJournalResume, type JournalResume } from './journal-cursor' @@ -27,12 +26,14 @@ import { DEFAULT_JOURNAL_PAYLOAD_LIMITS } from './journal-payload-bounds' import { applyJournalRow, createJournalReducerState, + nextJournalItemRevision, renderJournalState, type JournalReducerState } from './journal-reducer' import { - journalRowByteLength, type AgentJournalEpochReason, + journalRowBase, + journalRowByteLength, type JournalRow } from './journal-row-schema' import type { @@ -43,6 +44,7 @@ import type { } from './journal-store-contracts' import { assertJournalFence, + assertNewSubmission, assertJournalWritable, JournalAppendBudget } from './journal-write-guards' @@ -189,13 +191,13 @@ export class AgentSessionJournal { 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 + const revision = nextJournalItemRevision(this.state, resolved) return { kind: 'item', itemId, revision, body, - ...this.rowBase(seq, options.fence, options.observedAt ?? ts), + ...journalRowBase(this.state.epoch, seq, options.fence, options.observedAt ?? ts), ...(options.recovered ? { recovered: options.recovered } : {}) } }).then((row) => ({ @@ -212,8 +214,13 @@ export class AgentSessionJournal { 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) } + const revision = nextJournalItemRevision(this.state, resolved) + return { + kind: 'tombstone', + itemId, + revision, + ...journalRowBase(this.state.epoch, seq, options.fence, ts) + } }).then((row) => ({ epoch: row.epoch, sequence: row.seq })) } @@ -228,14 +235,17 @@ 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((seq, ts) => { + assertNewSubmission(this.state.submissions.has(input.clientMessageId), input.clientMessageId) + return { + kind: 'submission', + clientMessageId: input.clientMessageId, + payloadFingerprint: input.payloadFingerprint, + providerHandle: this.identity.providerHandle, + body: input.body, + ...journalRowBase(this.state.epoch, seq, input.fence, ts) + } + }).then((row) => ({ epoch: row.epoch, sequence: row.seq })) } /** @@ -254,7 +264,7 @@ export class AgentSessionJournal { state: input.state, providerItemId, reason: input.state === 'accepted' ? null : (input.reason ?? null), - ...this.rowBase(seq, input.fence, ts), + ...journalRowBase(this.state.epoch, seq, input.fence, ts), ...(input.recovered ? { recovered: input.recovered } : {}) })).then((row) => ({ epoch: row.epoch, sequence: row.seq })) } @@ -321,20 +331,6 @@ export class AgentSessionJournal { this.sizeBytes = published.sizeBytes } - 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 - } - } - /** * Assign the next sequence, make the row durable, and fold it through the * SAME reducer replay uses — all inside one serialized step, so concurrent 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..0274cecea6a 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 @@ -11,6 +11,7 @@ export class AgentSessionJournalError extends Error { readonly code: | 'journal_read_only' | 'journal_stale_fence' + | 'journal_duplicate_submission' | 'journal_bound_exceeded' | 'journal_rate_exceeded', message: string @@ -20,6 +21,15 @@ export class AgentSessionJournalError extends Error { } } +export function assertNewSubmission(exists: boolean, clientMessageId: string): void { + if (exists) { + throw new AgentSessionJournalError( + 'journal_duplicate_submission', + `submission ${clientMessageId} already exists in this journal epoch` + ) + } +} + /** A journal written by a newer schema is readable but never writable: this * host cannot represent rows it does not understand. */ export function assertJournalWritable(readOnly: boolean, sessionId: string): void {