From 4ff6d4c2ad7a040f7b5a56c09e269165f2648799 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 26 Aug 2026 20:19:19 -0700 Subject: [PATCH] fix structured session journal durability --- .../journal-compaction.ts | 10 +++ .../journal-epoch-replacement.ts | 3 +- .../journal-epoch-rollover.ts | 4 +- .../agent-session-journal/journal-log-file.ts | 58 +++++++++++++----- .../journal-store-contracts.ts | 3 + .../journal-store.test.ts | 61 ++++++++++++------- .../agent-session-journal/journal-store.ts | 41 ++++++------- .../journal-write-guards.ts | 4 ++ ...ctured-agent-session-lease-renewer.test.ts | 21 +++++-- .../structured-agent-session-lease-renewer.ts | 25 +++++--- .../structured-agent-session-read-restore.ts | 3 +- 11 files changed, 153 insertions(+), 80 deletions(-) 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 60e0b91bd81..6533ceda5ab 100644 --- a/src/main/native-chat/agent-session-journal/journal-compaction.ts +++ b/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 { 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 diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts index 22a05e5c4fd..8cc12c5ee66 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts +++ b/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 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 06379161c1c..0cf3f9d6cda 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 @@ -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 diff --git a/src/main/native-chat/agent-session-journal/journal-log-file.ts b/src/main/native-chat/agent-session-journal/journal-log-file.ts index 3d70d0a4ed6..8acc7743496 100644 --- a/src/main/native-chat/agent-session-journal/journal-log-file.ts +++ b/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 { + 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 diff --git a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts index 41b278ac94a..4a864315c40 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts +++ b/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 = diff --git a/src/main/native-chat/agent-session-journal/journal-store.test.ts b/src/main/native-chat/agent-session-journal/journal-store.test.ts index 79100b67aba..ad48723604e 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.test.ts +++ b/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 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 bbc72c52a96..52a47ae2561 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/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 { 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 { 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 diff --git a/src/main/native-chat/agent-session-journal/journal-write-guards.ts b/src/main/native-chat/agent-session-journal/journal-write-guards.ts index 4914c6dcf61..83071595f8b 100644 --- a/src/main/native-chat/agent-session-journal/journal-write-guards.ts +++ b/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 } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.test.ts index 250f656d927..e5b97dea189 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.test.ts +++ b/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', diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.ts index 85db0ba6420..570914d7ba1 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.ts +++ 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[] + 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] diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-read-restore.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-read-restore.ts index e8b06535329..3b4ffa95e54 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-read-restore.ts +++ b/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 }