fix(native-chat): keep committed epoch after cleanup failure

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<JournalCompactionResult> {
   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(

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])
+  })
+})

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<PublishedJournalEpoch> {
   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
 }

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<ReturnType<typeof publishNewEpoch>>): void {
     this.state = published.state
     this.tailRows = [published.row]
     this.compactedThrough = 0