fix(agent-session): harden journal recovery boundaries

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

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<void> {
+  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<void> {
+  const copied = await copyFileDurable(join(stagingDir, fileName), join(journalDir, fileName))
+  if (!copied) {
+    throw new Error(`prepared journal file disappeared before publish: ${fileName}`)
+  }
+}

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<JournalLoad> {
   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)
+  }
 }

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<AgentSessionJournal> {
+async function open(
+  agent: 'claude' | 'codex',
+  sessionId: string,
+  overrides: Partial<Parameters<typeof openAgentSessionJournal>[0]> = {}
+): Promise<AgentSessionJournal> {
   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

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 = {

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<void> {
@@ -67,15 +74,35 @@ export async function ensureJournalDir(journalDir: string): Promise<void> {
 export async function readJournalSnapshotFile(
   journalDir: string
 ): Promise<JournalSnapshotFile | null> {
+  const result = await readJournalSnapshot(journalDir)
+  return result.status === 'valid' ? result.snapshot : null
+}
+
+export async function readJournalSnapshot(journalDir: string): Promise<JournalSnapshotReadResult> {
   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<string> {
+  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<JournalReadRes
   let raw: string
   try {
     raw = await readFile(join(journalDir, JOURNAL_LOG_FILE), 'utf-8')
-  } catch {
-    return { rows: [], unreadable: false, malformed: 0 }
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
+      return { rows: [], unreadable: false, malformed: 0, hasBytes: false }
+    }
+    throw error
   }
   const rows: JournalRow[] = []
   let unreadable = false
@@ -109,12 +139,75 @@ export async function readJournalLog(journalDir: string): Promise<JournalReadRes
     }
     if (parsed.unreadable) {
       unreadable = true
-      break
+      return { rows, unreadable, malformed, remainder: raw.slice(offset), hasBytes: raw.length > 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<string, unknown>
+  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<string, unknown> | null {
+  return value && typeof value === 'object' && !Array.isArray(value)
+    ? (value as Record<string, unknown>)
+    : 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')

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<JournalLoad | null> {
-  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(

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<string[]> {
+  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
+}

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<T> = (seq: number, ts: number) => T
+
+export function journalItemRowBuilder(
+  state: () => JournalReducerState,
+  identity: AgentJournalItemIdentity,
+  body: AgentJournalItemBody,
+  options: { fence: number; observedAt?: number; recovered?: true }
+): RowBuilder<JournalItemRow> {
+  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<JournalTombstoneRow> {
+  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<JournalSubmissionRow> {
+  return (seq, ts) =>
+    buildJournalSubmissionRow({ state: state(), providerHandle, ...input, seq, ts })
+}
+
+export function journalDispatchRowBuilder(
+  state: () => JournalReducerState,
+  input: ResolveDispatchInput
+): RowBuilder<JournalDispatchRow> {
+  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<AgentJournalDispatchState, 'pending'>
+  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 } : {})
+  }
+}

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', () => {

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<JournalAppendResult> {
     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<JournalRow, { kind: 'item' }>).revision
-    }))
+        revision: (row as Extract<JournalRow, { kind: 'item' }>).revision
+      })
+    )
   }
 
   appendTombstone(
@@ -213,11 +207,9 @@ export class AgentSessionJournal {
     options: { fence: number }
   ): Promise<AgentJournalCursor> {
     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<AgentJournalCursor> {
-    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<AgentJournalCursor> {
-    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<string[]> {
-    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<void> {
+  async compact(
+    now = this.now(),
+    policy: JournalCompactionPolicy = this.compaction
+  ): Promise<void> {
     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<void> {
-    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<AgentJournalCursor> {
+    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<void> {
+    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)

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`