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 26af61021e7..46a725fb99d 100644 --- a/src/main/native-chat/agent-session-journal/journal-compaction.ts +++ b/src/main/native-chat/agent-session-journal/journal-compaction.ts @@ -51,6 +51,7 @@ export async function compactJournal(input: { tailRows: readonly JournalRow[] policy?: JournalCompactionPolicy now: number + onSnapshotPublished?: (result: JournalCompactionResult) => void }): Promise { const policy = input.policy ?? DEFAULT_JOURNAL_COMPACTION_POLICY const retained = retainTail(input.tailRows, policy, input.now) @@ -82,7 +83,17 @@ export async function compactJournal(input: { tail: retained } + const result: JournalCompactionResult = { + tailRows: retained, + compactedThrough, + oldestSequence: retained[0]?.seq ?? compactedThrough + 1, + sizeBytes: + journalSnapshotByteLength(snapshot) + + retained.reduce((total, row) => total + journalRowByteLength(row), 0) + } + await writeJournalSnapshotFile(input.journalDir, snapshot) + input.onSnapshotPublished?.(result) await rewriteJournalLog(input.journalDir, retained) // Blobs are pruned last: a crash before this leaks bytes, whereas pruning // first would strand a snapshot pointing at a payload that no longer exists. @@ -92,14 +103,7 @@ export async function compactJournal(input: { } await pruneJournalBlobs(input.journalDir, retainedDigests) - return { - tailRows: retained, - compactedThrough, - oldestSequence: retained[0]?.seq ?? compactedThrough + 1, - sizeBytes: - journalSnapshotByteLength(snapshot) + - retained.reduce((total, row) => total + journalRowByteLength(row), 0) - } + return result } function retainTail( diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-failure.test.ts b/src/main/native-chat/agent-session-journal/journal-epoch-failure.test.ts new file mode 100644 index 00000000000..fc443201910 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-epoch-failure.test.ts @@ -0,0 +1,76 @@ +import { mkdir, 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' +import type { + AgentJournalItemBody, + AgentJournalItemIdentity, + AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import { JOURNAL_LOG_FILE, JOURNAL_SNAPSHOT_FILE } from './journal-log-file' +import { openAgentSessionJournal } from './journal-store' + +const IDENTITY: AgentSessionJournalIdentity = { + sessionId: 'session-1', + workspaceId: 'ws-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } +} + +const ITEM: AgentJournalItemIdentity = { + provider: 'codex', + threadId: 'thread-1', + turnId: 'turn-1', + ordinal: 0 +} + +const BODY: AgentJournalItemBody = { + kind: 'message', + role: 'assistant', + blocks: [{ type: 'text', text: 'new epoch' }] +} + +let root: string +let clock = 1_000 + +async function open() { + return openAgentSessionJournal({ + identity: IDENTITY, + journalDir: root, + now: () => (clock += 1), + mintEpoch: () => `epoch-${clock}` + }) +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-journal-epoch-failure-')) + clock = 1_000 +}) + +afterEach(async () => { + await rm(root, { recursive: true, force: true }) +}) + +describe('epoch publication failures', () => { + it('keeps later appends on the authoritative epoch when log cleanup fails', async () => { + const journal = await open() + await journal.appendItem(ITEM, { ...BODY, blocks: [] }, { fence: 1 }) + const logPath = join(root, JOURNAL_LOG_FILE) + const oldLog = await readFile(logPath, 'utf-8') + + await rm(logPath) + await mkdir(logPath) + await expect(journal.rollEpoch('handle_forked', 2)).rejects.toThrow() + + const snapshot = JSON.parse(await readFile(join(root, JOURNAL_SNAPSHOT_FILE), 'utf-8')) as { + epoch: string + } + await rm(logPath, { recursive: true }) + await writeFile(logPath, oldLog, 'utf-8') + + const appended = await journal.appendItem(ITEM, BODY, { fence: 2 }) + expect(appended.cursor.epoch).toBe(snapshot.epoch) + expect((await open()).snapshot().items.map((entry) => entry.body)).toEqual([BODY]) + }) +}) 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 6f35a71da39..5363f239870 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 @@ -7,7 +7,7 @@ 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 { compactJournal, type JournalCompactionResult } from './journal-compaction' import { applyJournalRow, createJournalReducerState, @@ -15,6 +15,12 @@ import { } from './journal-reducer' import type { AgentJournalEpochReason, JournalRow } from './journal-row-schema' +export type PublishedJournalEpoch = { + state: JournalReducerState + row: JournalRow + sizeBytes: number +} + export async function publishNewEpoch(input: { journalDir: string sessionId: string @@ -23,7 +29,8 @@ export async function publishNewEpoch(input: { reason: AgentJournalEpochReason fence: number now: number -}): Promise<{ state: JournalReducerState; row: JournalRow; sizeBytes: number }> { + onSnapshotPublished?: (published: PublishedJournalEpoch) => void +}): Promise { const row: JournalRow = { kind: 'epoch', reason: input.reason, @@ -35,14 +42,23 @@ export async function publishNewEpoch(input: { ts: input.now } const state = createJournalReducerState(input.sessionId, input.epoch) - const compacted = await compactJournal({ + let published: PublishedJournalEpoch | null = null + const capturePublication = (result: JournalCompactionResult): void => { + applyJournalRow(state, row) + state.oldestSequence = 1 + published = { state, row, sizeBytes: result.sizeBytes } + input.onSnapshotPublished?.(published) + } + await compactJournal({ journalDir: input.journalDir, state, tailRows: [row], policy: { minTailRows: 1, retainTailMs: Number.POSITIVE_INFINITY }, - now: input.now + now: input.now, + onSnapshotPublished: capturePublication }) - applyJournalRow(state, row) - state.oldestSequence = 1 - return { state, row, sizeBytes: compacted.sizeBytes } + if (!published) { + throw new Error('journal epoch snapshot publication did not complete') + } + return published } 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 cc7ee6d8fc9..e7146d75df6 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -319,8 +319,15 @@ export class AgentSessionJournal { epoch: this.mintEpoch(), reason, fence, - now: this.now() + now: this.now(), + // The snapshot commits the epoch; later log cleanup is recoverable and + // must not leave subsequent appends writing the superseded epoch. + onSnapshotPublished: (committed) => this.adoptEpoch(committed) }) + this.adoptEpoch(published) + } + + private adoptEpoch(published: Awaited>): void { this.state = published.state this.tailRows = [published.row] this.compactedThrough = 0