fix(native-chat): close journal replay edge cases

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,

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()

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

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,

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<JournalRow, { kind: 'submission' }>
 ): 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,

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

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

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

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 {