diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index 5b21fa9b0ad..f1218ccdd07 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -10,6 +10,77 @@ } }, "gates": [ + { + "id": "agent-session.journal-streaming-replay", + "title": "Journal replay bounds obsolete revision memory without changing recovery", + "maturity": "experimental", + "protection": "partial", + "owner": "agent-session-runtime", + "layer": "runtime-unit", + "surfaces": ["structured chat journal replay", "structured chat recovery"], + "platforms": ["macos", "linux", "windows"], + "providers": ["local", "remote-runtime"], + "coveredPlatforms": ["macos"], + "coveredProviders": ["local"], + "coverageNotes": "Production SQLite and reducer tests on macOS. Execution-host-local storage behavior is shared by remote runtimes; no live SSH or Windows/Linux run. PTY, daemon, WSL process launch, transport framing and mobile rendering are unaffected.", + "motivatingLinks": [ + "https://github.com/stablyai/orca/blob/main/src/main/native-chat/agent-session-journal/journal-open.ts" + ], + "invariant": "Replay preserves latest revisions, original item order, fences, aliases, submissions, repair precedence, read-only schema latching and cursor cleanup while retaining live items rather than all historical bodies.", + "oracle": "Replay 2,048 16 KiB revisions into one latest item with less than 8 MiB sampled live heap growth; preserve prefix and future-schema latching after a gap, malformed suffix repair precedence, and release SQLite read cursors on early exit. Existing journal and subscriber tests cover replayed content and recovery.", + "commands": [ + "ORCA_BACKGROUND_LAUNCH=1 pnpm test src/main/native-chat/agent-session-journal src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts src/main/native-chat/agent-session-wire/agent-session-journal-recovery.test.ts", + "ORCA_BACKGROUND_LAUNCH=1 pnpm test src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts src/main/native-chat/agent-session-journal/journal-corruption-repair.test.ts" + ], + "testFiles": [ + "src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts", + "src/main/native-chat/agent-session-journal/journal-corruption-repair.test.ts" + ], + "assertionRefs": [ + { + "file": "src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts", + "assertions": [ + "releases superseded revision bodies while reducing a long journal", + "keeps the prefix but latches read-only for a future row beyond a gap", + "keeps gap repair precedence when a later row is malformed", + "rejects an unanchored prefix before a later gap" + ] + } + ], + "evidenceRuns": [ + { + "date": "2026-09-11", + "runner": "local", + "platform": "macos", + "command": "ORCA_BACKGROUND_LAUNCH=1 pnpm test src/main/native-chat/agent-session-journal src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts src/main/native-chat/agent-session-wire/agent-session-journal-recovery.test.ts", + "result": "passed", + "durationSeconds": 7.71, + "summary": "245 tests passed across 22 files. Retained-heap oracle fails on baseline at 68.6 MB and passes under 8 MiB with streaming; gap/schema and cursor-cleanup assertions passed." + } + ], + "runtimeBudget": { + "p95Seconds": 30, + "scope": "Unit fixtures; p95 not established." + }, + "flakeHistory": { + "status": "not-started", + "evidence": "Local candidate validation only; no CI soak." + }, + "redGreenEvidence": { + "status": "complete", + "evidence": "Production-function AB/BA benchmark samples 132.6 MB live heap in baseline vs 88-92 KB with streaming on a 66.7 MB revision-heavy journal. The retained-heap unit test fails against the original code and passes with streaming; value and cursor assertions pass in both implementations." + }, + "performanceBudget": { + "required": true, + "evidence": "Synchronous SQLite iteration and immediate reduction retain reduced state plus the current row. No cooperative await pins the cursor; no new polling, subprocess, provider call or wire change." + }, + "knownGaps": [ + "No Windows/Linux runtime execution or real remote-host validation.", + "Latest live message bodies still require memory proportional to their total size; this removes superseded-history retention, not live-history storage." + ], + "promotionCriteria": ["Retain red/green heap and value assertions and complete CI soak."], + "demotionRule": "Keep experimental until cross-platform and soak evidence; investigate failures without weakening content or memory assertions." + }, { "id": "runtime.connection-owned-host-status", "title": "Host status recovers with its owning connection", diff --git a/config/scripts/journal-replay-retention-benchmark.mjs b/config/scripts/journal-replay-retention-benchmark.mjs new file mode 100644 index 00000000000..3991a886ab5 --- /dev/null +++ b/config/scripts/journal-replay-retention-benchmark.mjs @@ -0,0 +1,122 @@ +#!/usr/bin/env node +import assert from 'node:assert/strict' +import { mkdtemp, readFile, rm, stat } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { basename, dirname, join } from 'node:path' +import { createRequire } from 'node:module' +import { fileURLToPath } from 'node:url' +import { build } from 'esbuild' + +// Pass a directory containing journal-open.ts and journal-row-table.ts from the base commit. +const baselineDir = process.argv[2] +assert.ok( + baselineDir, + 'Usage: node --expose-gc journal-replay-retention-benchmark.mjs BASELINE_DIR' +) +assert.ok(global.gc, 'Run with --expose-gc to measure live backing memory during replay') +const root = fileURLToPath(new URL('../..', import.meta.url)) +const fixture = await mkdtemp(join(tmpdir(), 'orca-journal-replay-bench-')) +try { + const implementations = {} + for (const arm of ['baseline', 'current']) { + const outfile = join(fixture, `${arm}.cjs`) + await build({ + stdin: { + contents: + "export {openAgentSessionJournal} from './src/main/native-chat/agent-session-journal/journal-store-factory'; export {loadJournal} from './src/main/native-chat/agent-session-journal/journal-open'; export {journalDatabaseFile} from './src/main/native-chat/agent-session-journal/journal-paths';", + resolveDir: root + }, + bundle: true, + platform: 'node', + format: 'cjs', + outfile, + plugins: [ + { + name: 'replay-memory-probe', + setup(plugin) { + plugin.onLoad( + { filter: /journal-(?:open|row-table|reducer)\.ts$/ }, + async ({ path }) => { + const leaf = basename(path) + let source = await readFile( + arm === 'baseline' && leaf !== 'journal-reducer.ts' + ? join(baselineDir, leaf) + : path, + 'utf8' + ) + if (leaf === 'journal-reducer.ts') { + const marker = + 'export function applyJournalRow(state: JournalReducerState, row: JournalRow): void {' + assert.ok(source.includes(marker)) + source = source.replace( + marker, + `${marker}\nglobalThis.__replayMemoryProbe?.(row.seq);` + ) + } + return { contents: source, loader: 'ts', resolveDir: dirname(path) } + } + ) + } + } + ] + }) + implementations[arm] = createRequire(import.meta.url)(outfile) + } + const identity = { + sessionId: 'benchmark', + workspaceId: 'fixture', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread' } + } + const journalDir = join(fixture, 'session') + const journal = await implementations.current.openAgentSessionJournal({ identity, journalDir }) + const item = { provider: 'codex', threadId: 'thread', turnId: 'turn', ordinal: 0 } + const text = 'x'.repeat(32768) + for (let revision = 0; revision < 2000; revision++) { + await journal.appendItem( + item, + { + kind: 'message', + role: 'assistant', + blocks: [{ type: 'text', text: `${text}${revision}` }] + }, + { fence: 1 } + ) + } + await journal.close() + for (const arm of ['baseline', 'current', 'current', 'baseline']) { + global.gc() + const start = performance.now() + let loaded = implementations[arm].loadJournal(journalDir, identity.sessionId) + const ms = performance.now() - start + assert.equal(loaded.state.items.size, 1) + assert.equal([...loaded.state.items.values()][0].revision, 2000) + loaded = null + global.gc() + const initialHeap = process.memoryUsage().heapUsed + let peakLiveHeap = initialHeap + globalThis.__replayMemoryProbe = (sequence) => { + if (sequence !== 1 && sequence % 256 !== 0) { + return + } + global.gc() + peakLiveHeap = Math.max(peakLiveHeap, process.memoryUsage().heapUsed) + } + loaded = implementations[arm].loadJournal(journalDir, identity.sessionId) + delete globalThis.__replayMemoryProbe + assert.equal(loaded.state.items.size, 1) + loaded = null + console.log( + JSON.stringify({ + arm, + ms, + databaseBytes: (await stat(implementations[arm].journalDatabaseFile(journalDir))).size, + peakLiveHeapDelta: peakLiveHeap - initialHeap + }) + ) + } +} finally { + delete globalThis.__replayMemoryProbe + await rm(fixture, { recursive: true, force: true }) +} diff --git a/src/main/native-chat/agent-session-journal/journal-open.ts b/src/main/native-chat/agent-session-journal/journal-open.ts index e4cc5e0f73e..d86da773f71 100644 --- a/src/main/native-chat/agent-session-journal/journal-open.ts +++ b/src/main/native-chat/agent-session-journal/journal-open.ts @@ -8,7 +8,6 @@ import { existsSync } from 'node:fs' import type Database from '../../sqlite/sync-database' -import { findSequenceGap } from './journal-cursor' import { openJournalDatabase } from './journal-database' import { journalDatabaseFile } from './journal-paths' import { @@ -17,7 +16,7 @@ import { type JournalReducerState } from './journal-reducer' import { - readJournalEpochRows, + iterateJournalEpochRows, readJournalRowsAfter, readJournalSessionEpoch } from './journal-row-table' @@ -59,108 +58,69 @@ export function replayJournal( return null } const state = createJournalReducerState(sessionId, epoch) - const stored = readJournalEpochRows(db, sessionId, epoch) - // A partial repair keeps its prefix, so the surviving rows look contiguous and - // anchored however much of the timeline it deleted. Its marker is what still - // says otherwise, naming the sequence past which the epoch would be its own - // history again. const repairedFrom = pendingJournalRepairSequence(db, sessionId, epoch) - const rows: JournalRow[] = [] + let expectedSequence = FIRST_JOURNAL_SEQUENCE + let gapSequence: number | undefined + let unanchoredSequence: number | undefined + let anchor: Extract | undefined + let repairHasContent = false + let providerHasContent = false let malformedRows = 0 let latched = false let truncateFrom: number | undefined - for (const entry of stored) { + + for (const entry of iterateJournalEpochRows(db, sessionId, epoch)) { const parsed = parseJournalRow(entry.rowJson) - if (parsed.ok) { - rows.push(parsed.row) + if (!parsed.ok) { + truncateFrom = entry.seq + latched = parsed.unreadable + malformedRows = parsed.unreadable ? 0 : 1 + break + } + const row = parsed.row + // Parse past a gap so an unreadable future row still latches read-only. + if (gapSequence !== undefined) { continue } - // Reading STOPS at the first row this build cannot represent. A future - // version latches read-only; anything else is one skipped row, disclosed. - truncateFrom = entry.seq - if (parsed.unreadable) { - latched = true - } else { - malformedRows = 1 + if (row.seq !== expectedSequence) { + gapSequence = row.seq + continue } - break - } - - // Anchored at 1, never at the first row that HAPPENS to remain: nothing trims - // a prefix, so a missing epoch row is a hole like any other and everything - // behind it is unanchored. Validating from `rows[0].seq` would call the - // leftovers contiguous and leave them out of the repair that runs before - // provider history replaces the epoch. - const gap = findSequenceGap( - rows.map((row) => row.seq), - FIRST_JOURNAL_SEQUENCE - ) - if (gap) { - const firstBad = rows.findIndex((row, index) => row.seq !== FIRST_JOURNAL_SEQUENCE + index) - if (firstBad !== -1) { - truncateFrom = rows[firstBad]?.seq ?? truncateFrom - rows.length = firstBad + expectedSequence += 1 + if (row.seq === FIRST_JOURNAL_SEQUENCE) { + if (row.kind === 'epoch') { + anchor = row + } else { + unanchoredSequence = row.seq + } + } + if (!anchor) { + continue } - } - // Contiguity from 1 is not the whole invariant: sequence 1 has to BE the epoch - // row. An ordinary row there is an epoch nothing anchors, and replaying it as - // clean is how a repaired journal silently adopts a timeline whose real - // history was never rebuilt. - if (rows.length > 0 && rows[0]?.kind !== 'epoch') { - truncateFrom = rows[0]?.seq ?? truncateFrom - rows.length = 0 - } - for (const row of rows) { applyJournalRow(state, row) + const disclosure = row.kind === 'item' && row.itemId === JOURNAL_REPAIR_DISCLOSURE_ITEM_ID + if (!disclosure) { + repairHasContent ||= repairedFrom !== null && row.seq >= repairedFrom + providerHasContent ||= row.seq >= FIRST_JOURNAL_SEQUENCE + 1 + } } + // Anchor rejection takes precedence over a gap, which takes precedence over malformed rows. + truncateFrom = unanchoredSequence ?? gapSequence ?? truncateFrom state.oldestSequence = FIRST_JOURNAL_SEQUENCE - - // A latched journal reduces to nothing by design; only a writable one can be - // held to the anchor. - const unanchored = !latched && rows[0]?.kind !== 'epoch' return { state, readOnly: latched, corrupt: - Boolean(gap) || + gapSequence !== undefined || malformedRows > 0 || - unanchored || - (repairedFrom !== null && awaitsRebuild(rows, repairedFrom)) || - awaitsProviderHistory(rows), + (!latched && !anchor) || + (repairedFrom !== null && !repairHasContent) || + (anchor?.reason === 'unreconcilable_prefix' && !providerHasContent), malformedRows, ...(truncateFrom !== undefined && !latched ? { truncateFrom } : {}) } } -/** - * The epoch a total repair published, still holding nothing but its own anchor - * and disclosure. The rows it dropped were never reconstructed, so provider - * history has to be retried rather than this being called a clean timeline. - */ -function awaitsProviderHistory(rows: readonly JournalRow[]): boolean { - const anchor = rows[0] - if (anchor?.kind !== 'epoch' || anchor.reason !== 'unreconcilable_prefix') { - return false - } - // The anchor sits at sequence 1, so content of the epoch's own starts at 2. - return awaitsRebuild(rows, FIRST_JOURNAL_SEQUENCE + 1) -} - -/** - * True while everything at or above `contentFrom` is the repair's own - * bookkeeping: the deleted history was never rebuilt, so the provider has to be - * asked again. The moment the session writes content of its own past that - * sequence the epoch IS its own history, and the retry stops rather than a - * later import replacing rows the user has since seen. - */ -function awaitsRebuild(rows: readonly JournalRow[], contentFrom: number): boolean { - return rows.every( - (row) => - row.seq < contentFrom || - (row.kind === 'item' && row.itemId === JOURNAL_REPAIR_DISCLOSURE_ITEM_ID) - ) -} - /** Rows after a cursor, in sequence order. Stops at the first row this build * cannot parse, exactly as replay does. */ export function readJournalRowsAfterCursor( diff --git a/src/main/native-chat/agent-session-journal/journal-row-table.ts b/src/main/native-chat/agent-session-journal/journal-row-table.ts index a6689f2040a..fa566b2060d 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-table.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-table.ts @@ -52,7 +52,18 @@ export function readJournalEpochRows( sessionId: string, epoch: string ): JournalStoredRow[] { - return toStoredRows(db.prepare(SELECT_EPOCH_ROWS).all(sessionId, epoch)) + return [...iterateJournalEpochRows(db, sessionId, epoch)] +} + +/** Consume synchronously: closing or breaking the loop releases SQLite's read snapshot. */ +export function* iterateJournalEpochRows( + db: Database.Database, + sessionId: string, + epoch: string +): Generator { + for (const row of db.prepare(SELECT_EPOCH_ROWS).iterate(sessionId, epoch)) { + yield toStoredRow(row) + } } export function readJournalRowsAfter( @@ -92,8 +103,10 @@ export function deleteJournalRowSuffix( } function toStoredRows(rows: readonly unknown[]): JournalStoredRow[] { - return rows.map((entry) => { - const record = entry as { epoch: string; seq: number; ts: number; row_json: string } - return { epoch: record.epoch, seq: record.seq, ts: record.ts, rowJson: record.row_json } - }) + return rows.map(toStoredRow) +} + +function toStoredRow(entry: unknown): JournalStoredRow { + const record = entry as { epoch: string; seq: number; ts: number; row_json: string } + return { epoch: record.epoch, seq: record.seq, ts: record.ts, rowJson: record.row_json } } diff --git a/src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts b/src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts new file mode 100644 index 00000000000..39eb70ee172 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts @@ -0,0 +1,133 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' +import { openJournalDatabase, type OpenJournalDatabase } from './journal-database' +import { journalDatabaseFile } from './journal-paths' +import { replayJournal } from './journal-open' +import { insertJournalRow, upsertJournalSessionRow } from './journal-row-table' +import type { JournalRow } from './journal-row-schema' +import * as reducer from './journal-reducer' + +let root: string +let opened: OpenJournalDatabase +const sessionId = 'streaming-session' +const epoch = 'epoch-1' + +function anchor(): JournalRow { + return { + v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + kind: 'epoch', + epoch, + seq: 1, + ts: 1, + fence: 1, + reason: 'session_created', + providerHandle: { kind: 'codex', threadId: 'thread-1' } + } +} + +function revision(seq: number, text = 'content'): JournalRow { + return { + v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + kind: 'item', + epoch, + seq, + ts: seq, + fence: 1, + itemId: 'message-1', + revision: seq, + body: { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text }] } + } +} + +function put(row: JournalRow): void { + insertJournalRow(opened.db, sessionId, row) +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-stream-replay-')) + opened = openJournalDatabase(journalDatabaseFile(root)) + upsertJournalSessionRow(opened.db, sessionId, epoch, 1) +}) + +afterEach(async () => { + vi.restoreAllMocks() + opened.db.close() + await rm(root, { recursive: true, force: true }) +}) + +describe('streaming journal replay', () => { + it('releases superseded revision bodies while reducing a long journal', () => { + const gc = global.gc + if (!gc) { + throw new Error('Run retention tests with --expose-gc') + } + opened.db.exec('BEGIN') + put(anchor()) + for (let seq = 2; seq <= 2049; seq++) { + put(revision(seq, `${'x'.repeat(16384)}:${seq}`)) + } + opened.db.exec('COMMIT') + gc() + const initial = process.memoryUsage().heapUsed + let peak = initial + const apply = reducer.applyJournalRow + const spy = vi.spyOn(reducer, 'applyJournalRow') + spy.mockImplementation((state, row) => { + // The probe must not retain old row bodies in Vitest's call history. + spy.mockClear() + if (row.seq % 256 === 0) { + gc() + peak = Math.max(peak, process.memoryUsage().heapUsed) + } + apply(state, row) + }) + const loaded = replayJournal(opened.db, false, sessionId)! + expect(loaded.state.items.size).toBe(1) + expect(loaded.state.items.get('message-1')?.revision).toBe(2049) + expect(loaded.state.lastSequence).toBe(2049) + expect(peak - initial).toBeLessThan(8 * 1024 * 1024) + }) + + it('keeps the prefix but latches read-only for a future row beyond a gap', () => { + put(anchor()) + put(revision(2)) + put(revision(4)) + put({ ...revision(5), v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION + 1 }) + const loaded = replayJournal(opened.db, false, sessionId)! + expect(loaded).toMatchObject({ readOnly: true, corrupt: true, malformedRows: 0 }) + expect(loaded.truncateFrom).toBeUndefined() + expect(loaded.state.items.get('message-1')?.revision).toBe(2) + expect(loaded.state.lastSequence).toBe(2) + const checkpoint = opened.db.pragma('wal_checkpoint(TRUNCATE)') as { busy: number }[] + expect(checkpoint[0].busy).toBe(0) + }) + + it('keeps gap repair precedence when a later row is malformed', () => { + put(anchor()) + put(revision(2)) + put(revision(4)) + opened.db + .prepare('INSERT INTO journal_rows VALUES (?, ?, ?, ?, ?)') + .run(sessionId, epoch, 5, 5, '{') + const loaded = replayJournal(opened.db, false, sessionId)! + expect(loaded).toMatchObject({ + readOnly: false, + corrupt: true, + malformedRows: 1, + truncateFrom: 4 + }) + expect(loaded.state.lastSequence).toBe(2) + }) + + it('rejects an unanchored prefix before a later gap', () => { + put(revision(1)) + put(revision(3)) + const loaded = replayJournal(opened.db, false, sessionId)! + expect(loaded).toMatchObject({ readOnly: false, corrupt: true, truncateFrom: 1 }) + expect(loaded.state.items.size).toBe(0) + expect(loaded.state.lastSequence).toBe(0) + }) +})