fix(chat): stream journal replay without retaining obsolete revisions

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",

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

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<JournalRow, { kind: 'epoch' }> | 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(

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

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