fix structured session journal durability

src/main/native-chat/agent-session-journal/journal-compaction.ts

@@ -22,6 +22,7 @@ import {
 } from './journal-log-file'
 import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types'
 import type { JournalRow } from './journal-row-schema'
+import { AgentSessionJournalError } from './journal-write-guards'
 
 export type JournalCompactionPolicy = {
   /** Always keep at least this many rows, however old they are. */
@@ -58,6 +59,7 @@ export async function compactJournal(input: {
   tailRows: readonly JournalRow[]
   policy?: JournalCompactionPolicy
   now: number
+  maxSessionBytes: number
 }): Promise<JournalCompactionResult> {
   const policy = input.policy ?? DEFAULT_JOURNAL_COMPACTION_POLICY
   const retained = retainTail(input.tailRows, policy, input.now)
@@ -89,6 +91,14 @@ export async function compactJournal(input: {
     tail: retained
   }
 
+  const snapshotBytes = Buffer.byteLength(JSON.stringify(snapshot), 'utf8')
+  if (snapshotBytes > input.maxSessionBytes) {
+    throw new AgentSessionJournalError(
+      'journal_bound_exceeded',
+      `agent-session journal snapshot reached its ${input.maxSessionBytes}-byte bound`
+    )
+  }
+
   await writeJournalSnapshotFile(input.journalDir, snapshot)
   await rewriteJournalLog(input.journalDir, retained)
   // Blobs are pruned last: a crash before this leaks bytes, whereas pruning

src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts

@@ -71,7 +71,8 @@ export async function replaceJournalEpoch(input: {
       state,
       tailRows: rows,
       policy: input.compaction,
-      now: input.now()
+      now: input.now(),
+      maxSessionBytes: input.budget.maxSessionBytes
     })
     await publishPreparedFile(stagingDir, input.journalDir, JOURNAL_SNAPSHOT_FILE)
     state.oldestSequence = compacted.oldestSequence

src/main/native-chat/agent-session-journal/journal-epoch-rollover.ts

@@ -12,6 +12,7 @@ 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'
+import { DEFAULT_JOURNAL_PAYLOAD_LIMITS } from './journal-payload-bounds'
 
 export async function publishNewEpoch(input: {
   journalDir: string
@@ -38,7 +39,8 @@ export async function publishNewEpoch(input: {
     state,
     tailRows: [row],
     policy: { minTailRows: 1, retainTailMs: Number.POSITIVE_INFINITY },
-    now: input.now
+    now: input.now,
+    maxSessionBytes: DEFAULT_JOURNAL_PAYLOAD_LIMITS.maxSessionBytes
   })
   applyJournalRow(state, row)
   state.oldestSequence = 1

src/main/native-chat/agent-session-journal/journal-log-file.ts

@@ -227,22 +227,7 @@ export async function appendJournalRows(
   // concatenate a new durable row onto that fragment: truncate the torn tail
   // first, then fsync the repair before acknowledging this append.
   try {
-    // Read as bytes, not text: `truncate`/`write` take byte offsets, and a
-    // transcript's multi-byte characters make string indices the wrong unit.
-    const existing = await readFile(path)
-    if (existing.length > 0 && existing.at(-1) !== NEWLINE_BYTE) {
-      const boundary = existing.lastIndexOf(NEWLINE_BYTE)
-      const finalLine = existing.subarray(boundary + 1).toString('utf-8')
-      // A whole row that merely lost its newline is kept; a real fragment goes.
-      const complete = parseJournalRow(finalLine).ok
-      const handle = await open(path, 'r+')
-      try {
-        await (complete ? handle.write('\n', existing.length) : handle.truncate(boundary + 1))
-        await handle.sync()
-      } finally {
-        await handle.close()
-      }
-    }
+    await repairJournalLogTail(path)
   } catch (error) {
     if ((error as NodeJS.ErrnoException).code !== 'ENOENT') {
       throw error
@@ -266,6 +251,47 @@ export async function appendJournalRows(
   }
 }
 
+/** Repair only a torn final row. The normal append path reads one byte; scanning
+ * backward is reserved for the crash-recovery case and never rereads the log. */
+async function repairJournalLogTail(path: string): Promise<void> {
+  const handle = await open(path, 'r+')
+  try {
+    const { size } = await handle.stat()
+    if (size === 0) {
+      return
+    }
+    const lastByte = Buffer.alloc(1)
+    await handle.read(lastByte, 0, 1, size - 1)
+    if (lastByte[0] === NEWLINE_BYTE) {
+      return
+    }
+
+    const scanChunkBytes = 64 * 1024
+    let scanEnd = size
+    let boundary = -1
+    while (scanEnd > 0 && boundary === -1) {
+      const scanStart = Math.max(0, scanEnd - scanChunkBytes)
+      const chunk = Buffer.alloc(scanEnd - scanStart)
+      await handle.read(chunk, 0, chunk.length, scanStart)
+      const newline = chunk.lastIndexOf(NEWLINE_BYTE)
+      if (newline !== -1) {
+        boundary = scanStart + newline
+      }
+      scanEnd = scanStart
+    }
+
+    const lineStart = boundary + 1
+    const finalLine = Buffer.alloc(size - lineStart)
+    await handle.read(finalLine, 0, finalLine.length, lineStart)
+    // A whole row that merely lost its newline is kept; a real fragment goes.
+    const complete = parseJournalRow(finalLine.toString('utf-8')).ok
+    await (complete ? handle.write('\n', size) : handle.truncate(lineStart))
+    await handle.sync()
+  } finally {
+    await handle.close()
+  }
+}
+
 export async function quarantineJournalRemainder(
   journalDir: string,
   remainder: string

src/main/native-chat/agent-session-journal/journal-store-contracts.ts

@@ -5,6 +5,7 @@ import type {
   AgentSessionJournalIdentity
 } from '../../../shared/agent-session-journal-types'
 import type { JournalCompactionPolicy } from './journal-compaction'
+import type { JournalLoad } from './journal-open'
 import type { JournalPayloadLimits } from './journal-payload-bounds'
 import type { JournalRow } from './journal-row-schema'
 
@@ -17,6 +18,8 @@ export type AgentSessionJournalOptions = {
   autoCompact?: boolean
   now?: () => number
   mintEpoch?: () => string
+  /** A caller that already loaded the journal can avoid reading the same files again. */
+  loaded?: JournalLoad | null
 }
 
 export type JournalReadSince =

src/main/native-chat/agent-session-journal/journal-store.test.ts

@@ -9,6 +9,7 @@ import type {
 } from '../../../shared/agent-session-journal-types'
 import { readJournalBlob } from './journal-blob-store'
 import { JOURNAL_LOG_FILE, JOURNAL_SNAPSHOT_FILE } from './journal-log-file'
+import { loadJournal } from './journal-open'
 import {
   boundInlineText,
   boundPayload,
@@ -111,6 +112,19 @@ describe('fences', () => {
 })
 
 describe('replay', () => {
+  it('adopts a caller-provided load without reading the journal files again', async () => {
+    const journal = await open()
+    await journal.appendItem(item(0), body('a'), { fence: 1 })
+    const loaded = await loadJournal(root, IDENTITY.sessionId)
+    expect(loaded).not.toBeNull()
+
+    await rm(join(root, JOURNAL_LOG_FILE), { force: true })
+    await rm(join(root, JOURNAL_SNAPSHOT_FILE), { force: true })
+
+    const reopened = await open({ loaded })
+    expect(reopened.snapshot()).toEqual(journal.snapshot())
+  })
+
   it('reopens to the same render model the live writer held', async () => {
     const journal = await open()
     await journal.appendItem(item(0), body('a'), { fence: 1 })
@@ -201,28 +215,14 @@ describe('automatic compaction', () => {
     expect(journal.compactionBoundary).toBe(0)
   })
 
-  it('sheds a tail shorter than the row floor before the size bound refuses the append', async () => {
+  it('refuses an append when a tail shorter than the row floor cannot make room', async () => {
     const journal = await open({
       limits: { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, maxSessionBytes: 900 },
       // The tail never reaches the floor, so honouring it would shed nothing.
       compaction: { minTailRows: 512, retainTailMs: 10_000 }
     })
-    for (let index = 0; index < 20; index += 1) {
-      await expect(
-        journal.appendItem(item(index), body('x'.repeat(96)), { fence: 1 })
-      ).resolves.toBeDefined()
-    }
-    expect(journal.compactionBoundary).toBeGreaterThan(0)
-    expect(journal.snapshot().items).toHaveLength(20)
-  })
-
-  it('sheds a tail that is entirely inside the retention window rather than refusing every append', async () => {
-    const journal = await open({
-      limits: { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, maxSessionBytes: 900 },
-      compaction: { minTailRows: 10, retainTailMs: 2 * 60 * 60 * 1000 }
-    })
     let rejected = 0
-    for (let index = 0; index < 50; index += 1) {
+    for (let index = 0; index < 20; index += 1) {
       try {
         await journal.appendItem(item(index), body('x'.repeat(96)), { fence: 1 })
       } catch (error) {
@@ -230,20 +230,35 @@ describe('automatic compaction', () => {
         rejected += 1
       }
     }
+    expect(rejected).toBeGreaterThan(0)
+  })
 
-    expect(rejected).toBe(0)
-    // Shed rows are folded into the snapshot, so the conversation is intact.
-    expect(journal.snapshot().items).toHaveLength(50)
-    expect(journal.compactionBoundary).toBeGreaterThan(0)
+  it('refuses once the retained snapshot itself reaches the session bound', async () => {
+    const journal = await open({
+      limits: { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, maxSessionBytes: 10_000 },
+      compaction: { minTailRows: 10, retainTailMs: 2 * 60 * 60 * 1000 }
+    })
+    let rejected = 0
+    for (let index = 0; index < 30; index += 1) {
+      try {
+        await journal.appendItem(item(index), body('x'.repeat(128)), { fence: 1 })
+      } catch (error) {
+        expect(error).toMatchObject({ code: 'journal_bound_exceeded' })
+        rejected += 1
+      }
+    }
+
+    expect(rejected).toBeGreaterThan(0)
+    expect(journal.snapshot().items.length).toBeLessThan(30)
   })
 
   it('keeps the newest rows resumable while shedding under budget pressure', async () => {
     const journal = await open({
-      limits: { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, maxSessionBytes: 900 },
+      limits: { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, maxSessionBytes: 10_000 },
       compaction: { minTailRows: 10, retainTailMs: 2 * 60 * 60 * 1000 }
     })
-    for (let index = 0; index < 50; index += 1) {
-      await journal.appendItem(item(index), body('x'.repeat(96)), { fence: 1 })
+    for (let index = 0; index < 30; index += 1) {
+      await journal.appendItem(item(index), body('x'.repeat(64)), { fence: 1 })
     }
 
     // The window yields oldest-first, never wholesale: the latest append is

src/main/native-chat/agent-session-journal/journal-store.ts

@@ -1,9 +1,4 @@
-// The append-only journal store for one agent session.
-//
-// Sequence numbers are assigned inside the same serialized append that makes
-// the row durable, so an `(epoch, sequence)` space has exactly one writer and
-// no gaps or reuse. Every mutating call carries the runtime fence; a stale
-// fence is rejected outright rather than merged or queued for the new owner.
+// Append-only journal store for one agent session.
 
 import { randomUUID } from 'node:crypto'
 import type {
@@ -86,6 +81,7 @@ export class AgentSessionJournal {
   private readonly autoCompact: boolean
   private readonly now: () => number
   private readonly mintEpoch: () => string
+  private readonly loaded: JournalLoad | null | undefined
 
   private state: JournalReducerState
   private tailRows: JournalRow[] = []
@@ -107,6 +103,7 @@ export class AgentSessionJournal {
     this.compaction = options.compaction ?? DEFAULT_JOURNAL_COMPACTION_POLICY
     this.now = options.now ?? (() => Date.now())
     this.mintEpoch = options.mintEpoch ?? randomUUID
+    this.loaded = options.loaded
     this.state = createJournalReducerState(options.identity.sessionId, '')
   }
 
@@ -130,7 +127,10 @@ export class AgentSessionJournal {
 
   async open(): Promise<void> {
     await ensureJournalDir(this.journalDir)
-    const loaded = await loadJournal(this.journalDir, this.identity.sessionId)
+    const loaded =
+      this.loaded !== undefined
+        ? this.loaded
+        : await loadJournal(this.journalDir, this.identity.sessionId)
     if (!loaded) {
       await this.startEpoch('session_created', 0)
       return
@@ -148,21 +148,17 @@ export class AgentSessionJournal {
     }
   }
 
-  cursor(): AgentJournalCursor {
-    return { epoch: this.state.epoch, sequence: this.state.lastSequence }
-  }
+  cursor = (): AgentJournalCursor => ({
+    epoch: this.state.epoch,
+    sequence: this.state.lastSequence
+  })
 
-  snapshot(): AgentJournalSnapshot {
-    return renderJournalState(this.state)
-  }
+  snapshot = (): AgentJournalSnapshot => renderJournalState(this.state)
 
-  submissions(): AgentJournalSubmission[] {
-    return [...this.state.submissions.values()]
-  }
+  submissions = (): AgentJournalSubmission[] => [...this.state.submissions.values()]
 
-  pendingSubmissions(): AgentJournalSubmission[] {
-    return this.submissions().filter((entry) => entry.dispatchState === 'pending')
-  }
+  pendingSubmissions = (): AgentJournalSubmission[] =>
+    this.submissions().filter((entry) => entry.dispatchState === 'pending')
 
   /** The durable answer to "did my send land?" — a reconnecting client asking
    *  again gets this instead of re-sending. */
@@ -170,9 +166,7 @@ export class AgentSessionJournal {
     return this.state.receipts.get(clientMessageId) ?? null
   }
 
-  canonicalItemId(itemId: string): string {
-    return resolveJournalItemId(this.state, itemId)
-  }
+  canonicalItemId = (itemId: string): string => resolveJournalItemId(this.state, itemId)
 
   referencedBlobDigests(): Set<string> {
     return referencedBlobDigests(this.state)
@@ -259,7 +253,8 @@ export class AgentSessionJournal {
       state: this.state,
       tailRows: this.tailRows,
       policy,
-      now
+      now,
+      maxSessionBytes: this.budget.maxSessionBytes
     })
     this.tailRows = result.tailRows
     this.compactedThrough = result.compactedThrough

src/main/native-chat/agent-session-journal/journal-write-guards.ts

@@ -56,6 +56,10 @@ export class JournalAppendBudget {
     return new JournalAppendBudget(this.sessionId, this.limits)
   }
 
+  get maxSessionBytes(): number {
+    return this.limits.maxSessionBytes
+  }
+
   wouldExceedSize(row: JournalRow, sizeBytes: number): boolean {
     return sizeBytes + journalRowByteLength(row) > this.limits.maxSessionBytes
   }

src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.test.ts

@@ -90,9 +90,8 @@ describe('structured agent-session lease renewal', () => {
         })
       )
     })
-    const renewLease = vi.fn(
-      async (renewal: { sessionId: string }) =>
-        records.find((record) => record.sessionId === renewal.sessionId)!
+    const renewLeases = vi.fn(async (renewals: readonly { sessionId: string }[]) =>
+      renewals.map((renewal) => records.find((record) => record.sessionId === renewal.sessionId)!)
     )
     const probeMany = vi.fn(
       async () =>
@@ -104,7 +103,7 @@ describe('structured agent-session lease renewal', () => {
         )
     )
     const renewer = new StructuredAgentSessionLeaseRenewer({
-      store: { listRecords: () => records, renewLease } as unknown as AgentSessionRecordStore,
+      store: { listRecords: () => records, renewLeases } as unknown as AgentSessionRecordStore,
       probe: vi.fn(),
       probeMany,
       now: () => NOW + 10_000
@@ -113,7 +112,8 @@ describe('structured agent-session lease renewal', () => {
     await renewer.renewNow()
 
     expect(probeMany).toHaveBeenCalledOnce()
-    expect(renewLease).toHaveBeenCalledTimes(2)
+    expect(renewLeases).toHaveBeenCalledOnce()
+    expect(renewLeases.mock.calls[0]?.[0]).toHaveLength(2)
   })
 
   it('keeps a healthy lease alive when a sibling renewal is superseded', async () => {
@@ -135,6 +135,9 @@ describe('structured agent-session lease renewal', () => {
         })
       )
     )
+    const renewLeases = vi.fn(async () => {
+      throw new Error('agent_session_checkpoint_stale')
+    })
     const renewLease = vi.fn(async (renewal: { sessionId: string }) => {
       if (renewal.sessionId === 'session-b') {
         throw new Error('agent_session_checkpoint_stale')
@@ -144,7 +147,11 @@ describe('structured agent-session lease renewal', () => {
     const onRenewed = vi.fn()
     const onError = vi.fn()
     const renewer = new StructuredAgentSessionLeaseRenewer({
-      store: { listRecords: () => records, renewLease } as unknown as AgentSessionRecordStore,
+      store: {
+        listRecords: () => records,
+        renewLeases,
+        renewLease
+      } as unknown as AgentSessionRecordStore,
       probe: async () => ({
         outcome: 'identity-matched' as const,
         matchedOn: ['spawn-token' as const]
@@ -156,7 +163,9 @@ describe('structured agent-session lease renewal', () => {
 
     await renewer.renewNow()
 
+    expect(renewLeases).toHaveBeenCalledOnce()
     expect(onRenewed).toHaveBeenCalledOnce()
+    expect(renewLease).toHaveBeenCalledTimes(2)
     expect(onRenewed).toHaveBeenCalledWith(records[0])
     expect(onError).toHaveBeenCalledWith({
       sessionId: 'session-b',

src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.ts

@@ -95,15 +95,22 @@ export class StructuredAgentSessionLeaseRenewer {
           now
         })
       }
-      // Renew each lease independently. A stale fence or an unverifiable owner must not
-      // poison one transaction and let otherwise healthy sessions expire with it.
-      const results = await Promise.allSettled(
-        renewals.map(async (renewal) => {
-          const renewed = await this.input.store.renewLease(renewal)
-          await this.input.onRenewed?.(renewed)
-          return renewed
-        })
-      )
+      // The store persists the whole record file per transaction, so keep the healthy path to
+      // one commit. If one renewal is superseded, retrying individually preserves isolation.
+      let results: PromiseSettledResult<AgentSessionRecord>[]
+      try {
+        const renewed = await this.input.store.renewLeases(renewals)
+        await Promise.all(renewed.map((record) => this.input.onRenewed?.(record)))
+        results = renewed.map((record) => ({ status: 'fulfilled', value: record }) as const)
+      } catch {
+        results = await Promise.allSettled(
+          renewals.map(async (renewal) => {
+            const renewed = await this.input.store.renewLease(renewal)
+            await this.input.onRenewed?.(renewed)
+            return renewed
+          })
+        )
+      }
       results.forEach((result, index) => {
         if (result.status === 'rejected') {
           const renewal = renewals[index]

src/main/native-chat/agent-session-wire/structured-agent-session-read-restore.ts

@@ -46,7 +46,8 @@ export async function restoreStructuredAgentSessionRead(
   }
   const journal = await openAgentSessionJournal({
     identity: journalIdentityFor(record, params),
-    journalDir
+    journalDir,
+    loaded
   })
   // Read restore opens the journal and nothing else: no adapter call, so no provider child.
   return { journal, params, fence: record.lease.runtimeFence, hasProviderChild: false }