fix: page journal replay reads so no SQLite snapshot outlives its statement

- iterateJournalEpochRows fetches one completed LIMIT statement per page
  instead of a lazily consumed .iterate() cursor, so reduction never runs
  inside an open read snapshot and a WAL checkpoint can pass mid-replay.
  Regression test: a checkpoint issued from inside the reducer is not busy.
- The retention test now asserts the applyJournalRow spy observed every
  row, so the 8 MiB bound cannot pass vacuously if the spy stops
  intercepting.
- Reliability gate manifest records the new assertion and the paged
  read design.

config/reliability-gates.jsonc

@@ -27,7 +27,7 @@
         "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.",
+      "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 hold no SQLite read snapshot across reduction (a mid-replay checkpoint is not busy). 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"
@@ -41,6 +41,7 @@
           "file": "src/main/native-chat/agent-session-journal/journal-streaming-replay.test.ts",
           "assertions": [
             "releases superseded revision bodies while reducing a long journal",
+            "holds no read snapshot while reducing, so a checkpoint can pass mid-replay",
             "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"
@@ -72,7 +73,7 @@
       },
       "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."
+        "evidence": "Paged SQLite reads (one completed statement per page) and immediate reduction retain reduced state plus one page of rows. No cursor or read snapshot outlives its statement, so a WAL checkpoint can pass mid-replay; no new polling, subprocess, provider call or wire change."
       },
       "knownGaps": [
         "No Windows/Linux runtime execution or real remote-host validation.",

src/main/native-chat/agent-session-journal/journal-row-table.ts

@@ -52,17 +52,29 @@ export function readJournalEpochRows(
   sessionId: string,
   epoch: string
 ): JournalStoredRow[] {
-  return [...iterateJournalEpochRows(db, sessionId, epoch)]
+  return toStoredRows(db.prepare(SELECT_EPOCH_ROWS).all(sessionId, epoch))
 }
 
-/** Consume synchronously: closing or breaking the loop releases SQLite's read snapshot. */
+// Why pages, not `.iterate()`: a lazily consumed cursor pins a read snapshot for as long as the
+// consumer reduces, and a WAL checkpoint cannot pass an open snapshot. Each page is one completed
+// statement, so the consumer's memory is bounded by a page while no snapshot outlives a fetch.
+const EPOCH_ROW_PAGE_SIZE = 128
+
+/** Epoch rows in sequence order, fetched one completed statement at a time. */
 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)
+  let afterSeq = Number.MIN_SAFE_INTEGER
+  for (;;) {
+    const page = readJournalRowsAfter(db, sessionId, epoch, afterSeq, EPOCH_ROW_PAGE_SIZE)
+    yield* page
+    const last = page.at(-1)
+    if (page.length < EPOCH_ROW_PAGE_SIZE || last === undefined) {
+      return
+    }
+    afterSeq = last.seq
   }
 }
 
@@ -103,10 +115,8 @@ export function deleteJournalRowSuffix(
 }
 
 function toStoredRows(rows: readonly unknown[]): JournalStoredRow[] {
-  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 }
+  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 }
+  })
 }

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

@@ -73,11 +73,13 @@ describe('streaming journal replay', () => {
     gc()
     const initial = process.memoryUsage().heapUsed
     let peak = initial
+    let applied = 0
     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()
+      applied += 1
       if (row.seq % 256 === 0) {
         gc()
         peak = Math.max(peak, process.memoryUsage().heapUsed)
@@ -88,9 +90,29 @@ describe('streaming journal replay', () => {
     expect(loaded.state.items.size).toBe(1)
     expect(loaded.state.items.get('message-1')?.revision).toBe(2049)
     expect(loaded.state.lastSequence).toBe(2049)
+    // The probe must have measured every row, or the heap bound above is vacuous.
+    expect(applied).toBe(2049)
     expect(peak - initial).toBeLessThan(8 * 1024 * 1024)
   })
 
+  it('holds no read snapshot while reducing, so a checkpoint can pass mid-replay', () => {
+    put(anchor())
+    for (let seq = 2; seq <= 300; seq++) {
+      put(revision(seq))
+    }
+    const apply = reducer.applyJournalRow
+    const checkpoints: { busy: number }[] = []
+    vi.spyOn(reducer, 'applyJournalRow').mockImplementation((state, row) => {
+      if (row.seq === 2 || row.seq === 200) {
+        checkpoints.push(...(opened.db.pragma('wal_checkpoint(PASSIVE)') as { busy: number }[]))
+      }
+      apply(state, row)
+    })
+    const loaded = replayJournal(opened.db, false, sessionId)!
+    expect(loaded.state.lastSequence).toBe(300)
+    expect(checkpoints.map((entry) => entry.busy)).toEqual([0, 0])
+  })
+
   it('keeps the prefix but latches read-only for a future row beyond a gap', () => {
     put(anchor())
     put(revision(2))