perf(native-chat): bound journal reads during paged catch-up (#19360)

* perf(native-chat): bound journal reads during paged catch-up * perf(native-chat): reduce the journal once per catch-up run, not per page Bounding the SQL read per page left the JS side still O(total items) per page: every page re-reduced the whole timeline and rebuilt the live-item map, and the byte-shrink loop rebuilt it again on each halving. A catch-up run is a synchronous loop with no await between pages, so the reduced timeline is loop-invariant. `createAgentSessionCatchUpReader` holds one snapshot for the run and re-reduces only if the journal cursor actually moved, and the projection's live-item / alias / submission-byte indexes memoize on the snapshot arrays the reducer rebuilds on change. Per catch-up over a 8,000-message backlog: 40 timeline reductions to 1, reduce+project time 24.3ms to 3.5ms, end-to-end 134.6ms to 111.3ms.
2026-09-22 16:02:32 +00:00 · 2026-09-07 23:24:25 -07:00
parent 668946345a
commit 28d936e030
9 changed files with 401 additions and 23 deletions
@@ -10,6 +10,79 @@
    }
  },
  "gates": [
+    {
+      "id": "agent-session.history-forward-read-budget",
+      "title": "Journal catch-up reads only the next page and one lookahead row",
+      "maturity": "experimental",
+      "protection": "partial",
+      "owner": "agent-session-runtime",
+      "layer": "runtime-unit",
+      "surfaces": ["structured agent history", "structured agent subscriptions"],
+      "platforms": ["macos", "linux", "windows"],
+      "providers": ["local", "ssh", "remote-runtime"],
+      "coveredPlatforms": ["macos"],
+      "coveredProviders": ["local", "ssh", "remote-runtime"],
+      "coverageNotes": "The real SQLite journal and production subscriber delivery are exercised with a folder workspace and remote host identity. The SQL and pagination code is shared across execution hosts; live SSH transport and Linux/Windows runtime execution are not exercised. PTY, daemon, WSL execution, and mobile rendering are unaffected.",
+      "motivatingLinks": [
+        "https://github.com/stablyai/orca/blob/main/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts"
+      ],
+      "invariant": "Forward catch-up preserves every item, revision, tombstone, sequence cursor, page byte bound, and reset behavior while reading at most the requested row count plus one from SQLite for each page.",
+      "oracle": "Reconnect a real subscriber to a 2,000-row journal and receive all 2,000 item identities in order through the live cursor; count the actual SQL rows returned and parsed as 2,009 instead of 11,000. Assert exact final-page hasNewer, unlimited reader compatibility, gap detection at the next page, and parse-stop behavior at the lookahead row. Existing history tests cover revisions, tombstones, byte-bound shrinking, epochs, and schema resets.",
+      "commands": [
+        "ORCA_BACKGROUND_LAUNCH=1 pnpm exec vitest run --config config/vitest.config.ts src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts src/main/native-chat/agent-session-journal"
+      ],
+      "testFiles": [
+        "src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts",
+        "src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts",
+        "src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts"
+      ],
+      "assertionRefs": [
+        {
+          "file": "src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts",
+          "assertions": [
+            "reconnects through every page with one lookahead row per page",
+            "keeps an exact final page final and preserves unlimited journal readers",
+            "reports a sequence gap when the next page reaches it",
+            "preserves parse-stop behavior at lookahead: %s"
+          ]
+        }
+      ],
+      "evidenceRuns": [
+        {
+          "date": "2026-09-07",
+          "runner": "local",
+          "platform": "macos",
+          "command": "ORCA_BACKGROUND_LAUNCH=1 pnpm exec vitest run --config config/vitest.config.ts src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts src/main/native-chat/agent-session-journal",
+          "result": "passed",
+          "durationSeconds": 9.94,
+          "summary": "214 tests passed across 19 files, including actual SQLite row and JSON parse counts through production subscriber catch-up."
+        }
+      ],
+      "runtimeBudget": {
+        "p95Seconds": 30,
+        "scope": "Real SQLite journal unit and production subscriber tests; no launched app."
+      },
+      "flakeHistory": {
+        "status": "not-started",
+        "evidence": "Initial deterministic local validation; CI soak has not started."
+      },
+      "redGreenEvidence": {
+        "status": "complete",
+        "evidence": "Before the change, SQL returned 2,000, 1,800, 1,600 through 200 rows across ten pages, failing the count assertion. The bounded query returns nine pages of 201 rows and a final 200, with exactly 2,009 row parses and identical item delivery."
+      },
+      "performanceBudget": {
+        "required": true,
+        "evidence": "Catch-up materialization and JSON parsing are linear in unseen journal rows plus page lookaheads. A cached parameterized LIMIT adds no polling, cache invalidation, output loss, protocol change, or provider calls."
+      },
+      "knownGaps": [
+        "Linux and Windows execution and live SSH transport have not been exercised.",
+        "The existing full reduced-state snapshot and batch projection cost are outside this SQL read budget."
+      ],
+      "promotionCriteria": [
+        "Complete CI soak requirements while preserving the deterministic row budget and pagination oracles."
+      ],
+      "demotionRule": "Keep experimental until CI soak; investigate fidelity or count failures without relaxing the row budget."
+    },
    {
      "id": "terminal-performance.padded-fullscreen-redraw",
      "title": "Fullscreen redraw padding does not stall terminal delivery",
@@ -167,10 +167,11 @@ export function readJournalRowsAfterCursor(
  db: Database.Database,
  sessionId: string,
  epoch: string,
-  afterSequence: number
+  afterSequence: number,
+  limit?: number
 ): JournalRow[] {
  const rows: JournalRow[] = []
-  for (const stored of readJournalRowsAfter(db, sessionId, epoch, afterSequence)) {
+  for (const stored of readJournalRowsAfter(db, sessionId, epoch, afterSequence, limit)) {
    const parsed = parseJournalRow(stored.rowJson)
    if (!parsed.ok) {
      break
@@ -20,6 +20,7 @@ const SELECT_EPOCH_ROWS = `SELECT epoch, seq, ts, row_json FROM journal_rows
 WHERE session_id = ? AND epoch = ? ORDER BY seq ASC`
 const SELECT_ROWS_AFTER = `SELECT epoch, seq, ts, row_json FROM journal_rows
 WHERE session_id = ? AND epoch = ? AND seq > ? ORDER BY seq ASC`
+const SELECT_ROWS_AFTER_LIMITED = `${SELECT_ROWS_AFTER} LIMIT ?`
 const DELETE_SUFFIX = 'DELETE FROM journal_rows WHERE session_id = ? AND epoch = ? AND seq >= ?'

 export function readJournalSessionEpoch(db: Database.Database, sessionId: string): string | null {
@@ -58,8 +59,14 @@ export function readJournalRowsAfter(
  db: Database.Database,
  sessionId: string,
  epoch: string,
-  afterSeq: number
+  afterSeq: number,
+  limit?: number
 ): JournalStoredRow[] {
+  if (limit !== undefined) {
+    return toStoredRows(
+      db.prepare(SELECT_ROWS_AFTER_LIMITED).all(sessionId, epoch, afterSeq, limit)
+    )
+  }
  return toStoredRows(db.prepare(SELECT_ROWS_AFTER).all(sessionId, epoch, afterSeq))
 }

@@ -177,7 +177,7 @@ export class AgentSessionJournal {

  canonicalItemId = (itemId: string): string => resolveJournalItemId(this.state, itemId)

-  readSince(cursor: AgentJournalCursor): JournalReadSince {
+  readSince(cursor: AgentJournalCursor, limit?: number): JournalReadSince {
    return readJournalSince(
      {
        state: this.state,
@@ -186,7 +186,8 @@ export class AgentSessionJournal {
            this.requireDatabase().db,
            this.identity.sessionId,
            this.state.epoch,
-            afterSequence
+            afterSequence,
+            limit
          ),
        readOnly: this.readOnly
      },
@@ -0,0 +1,228 @@
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import Database from '../../sqlite/sync-database'
+import {
+  AGENT_SESSION_JOURNAL_SCHEMA_VERSION,
+  type AgentSessionJournalIdentity
+} from '../../../shared/agent-session-journal-types'
+import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire'
+import { openJournalDatabase } from '../agent-session-journal/journal-database'
+import { journalDatabaseFile } from '../agent-session-journal/journal-paths'
+import {
+  insertJournalRow,
+  upsertJournalSessionRow
+} from '../agent-session-journal/journal-row-table'
+import * as journalReducer from '../agent-session-journal/journal-reducer'
+import * as rowSchema from '../agent-session-journal/journal-row-schema'
+import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open'
+import { AgentSessionSubscribers } from './structured-agent-session-subscribers'
+import { readAgentSessionHistory } from './agent-session-history-page'
+
+const identity: AgentSessionJournalIdentity = {
+  sessionId: 'bounded-catch-up',
+  workspaceId: 'folder-workspace',
+  hostId: 'remote-host',
+  agent: 'codex',
+  providerHandle: { kind: 'codex', threadId: 'thread-1' }
+}
+const journals = createTrackedJournalOpener()
+let root: string | undefined
+
+afterEach(async () => {
+  vi.restoreAllMocks()
+  await journals.closeAll()
+  if (root) {
+    await rm(root, { recursive: true, force: true })
+  }
+})
+
+async function seedJournal(count: number) {
+  root = await mkdtemp(join(tmpdir(), 'orca-history-read-budget-'))
+  const { db } = openJournalDatabase(journalDatabaseFile(root))
+  const base = {
+    v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION,
+    epoch: 'epoch-1',
+    fence: 1,
+    ts: 1_000
+  }
+  try {
+    db.exec('BEGIN IMMEDIATE')
+    upsertJournalSessionRow(db, identity.sessionId, base.epoch, base.ts)
+    insertJournalRow(db, identity.sessionId, {
+      ...base,
+      kind: 'epoch',
+      seq: 1,
+      reason: 'session_created',
+      providerHandle: identity.providerHandle
+    })
+    for (let index = 0; index < count; index += 1) {
+      insertJournalRow(db, identity.sessionId, {
+        ...base,
+        kind: 'item',
+        seq: index + 2,
+        itemId: `item-${index}`,
+        revision: 1,
+        body: {
+          kind: 'message',
+          role: 'assistant',
+          blocks: [{ type: 'text', text: `${index}:${'x'.repeat(4096)}` }]
+        }
+      })
+    }
+    db.exec('COMMIT')
+  } finally {
+    db.close()
+  }
+  return journals.open({ identity, journalDir: root })
+}
+
+function observeForwardReads() {
+  const returnedRows: number[] = []
+  const observed = new WeakSet<object>()
+  const prepare = Database.prototype.prepare
+  vi.spyOn(Database.prototype, 'prepare').mockImplementation(function (this: Database, sql) {
+    const statement = prepare.call(this, sql)
+    if (sql.includes('seq > ?') && !observed.has(statement)) {
+      observed.add(statement)
+      const all = statement.all.bind(statement)
+      vi.spyOn(statement, 'all').mockImplementation((...args) => {
+        const rows = all(...args)
+        returnedRows.push(rows.length)
+        return rows
+      })
+    }
+    return statement
+  })
+  const parse = vi.spyOn(rowSchema, 'parseJournalRow')
+  return { returnedRows, parse }
+}
+
+describe('forward history SQL read budget', () => {
+  it('reconnects through every page with one lookahead row per page', async () => {
+    const count = 2_000
+    const journal = await seedJournal(count)
+    const { returnedRows, parse } = observeForwardReads()
+    const events: AgentSessionSubscribeEvent[] = []
+    new AgentSessionSubscribers().open({
+      id: 'reader',
+      sessionId: identity.sessionId,
+      journal,
+      fence: 1,
+      cursor: { epoch: journal.epoch, sequence: 1 },
+      emit: (event) => events.push(event)
+    })
+    const batches = events.filter((event) => event.type === 'batch')
+    expect(batches.flatMap((event) => event.batch.items.map((item) => item.itemId))).toEqual(
+      Array.from({ length: count }, (_, index) => `item-${index}`)
+    )
+    expect(batches.at(-1)?.batch.cursor).toEqual(journal.cursor())
+    expect(returnedRows).toEqual([...Array<number>(9).fill(201), 200])
+    expect(parse).toHaveBeenCalledTimes(2_009)
+  })
+
+  it('reduces the timeline once for the whole catch-up, not once per page', async () => {
+    const journal = await seedJournal(2_000)
+    const render = vi.spyOn(journalReducer, 'renderJournalState')
+    const events: AgentSessionSubscribeEvent[] = []
+    new AgentSessionSubscribers().open({
+      id: 'reader',
+      sessionId: identity.sessionId,
+      journal,
+      fence: 1,
+      cursor: { epoch: journal.epoch, sequence: 1 },
+      emit: (event) => events.push(event)
+    })
+    // Catch-up is synchronous, so the reduced timeline cannot change between pages.
+    expect(events.filter((event) => event.type === 'batch')).toHaveLength(10)
+    expect(render).toHaveBeenCalledTimes(1)
+  })
+
+  it('keeps an exact final page final and preserves unlimited journal readers', async () => {
+    const journal = await seedJournal(6)
+    const cursor = { epoch: journal.epoch, sequence: 1 }
+    expect(journal.readSince(cursor)).toMatchObject({ ok: true, rows: expect.any(Array) })
+    const first = readAgentSessionHistory(journal, {
+      sessionId: identity.sessionId,
+      direction: 'after',
+      cursor,
+      limit: 3
+    })
+    expect(first).toMatchObject({ ok: true, page: { hasNewer: true } })
+    if (!first.ok) {
+      throw new Error('Expected first page')
+    }
+    const last = readAgentSessionHistory(journal, {
+      sessionId: identity.sessionId,
+      direction: 'after',
+      cursor: first.page.window.nextCursor,
+      limit: 3
+    })
+    expect(last).toMatchObject({ ok: true, page: { hasNewer: false } })
+    const unlimited = journal.readSince(cursor)
+    expect(unlimited.ok && unlimited.rows).toHaveLength(6)
+  })
+
+  it('reports a sequence gap when the next page reaches it', async () => {
+    const journal = await seedJournal(6)
+    const { db } = openJournalDatabase(journalDatabaseFile(root!))
+    try {
+      db.prepare('DELETE FROM journal_rows WHERE session_id = ? AND seq = ?').run(
+        identity.sessionId,
+        4
+      )
+    } finally {
+      db.close()
+    }
+    const first = readAgentSessionHistory(journal, {
+      sessionId: identity.sessionId,
+      direction: 'after',
+      cursor: { epoch: journal.epoch, sequence: 1 },
+      limit: 2
+    })
+    expect(first).toMatchObject({ ok: true, page: { hasNewer: true } })
+    if (!first.ok) {
+      throw new Error('Expected first page')
+    }
+    expect(
+      readAgentSessionHistory(journal, {
+        sessionId: identity.sessionId,
+        direction: 'after',
+        cursor: first.page.window.nextCursor,
+        limit: 2
+      })
+    ).toMatchObject({ ok: false, reset: 'journal_gap' })
+  })
+
+  it.each(['{', '{"v":9999}'])(
+    'preserves parse-stop behavior at lookahead: %s',
+    async (rowJson) => {
+      const journal = await seedJournal(6)
+      const { db } = openJournalDatabase(journalDatabaseFile(root!))
+      try {
+        db.prepare('UPDATE journal_rows SET row_json = ? WHERE session_id = ? AND seq = ?').run(
+          rowJson,
+          identity.sessionId,
+          4
+        )
+      } finally {
+        db.close()
+      }
+      const page = readAgentSessionHistory(journal, {
+        sessionId: identity.sessionId,
+        direction: 'after',
+        cursor: { epoch: journal.epoch, sequence: 1 },
+        limit: 2
+      })
+      expect(page).toMatchObject({
+        ok: true,
+        page: { hasNewer: false, window: { nextCursor: { sequence: 3 } } }
+      })
+      if (!page.ok) {
+        throw new Error('Expected valid prefix')
+      }
+      expect(page.page.items.map((item) => item.itemId)).toEqual(['item-0', 'item-1'])
+    }
+  )
+})
@@ -22,15 +22,28 @@ export function historyEntryBytes(
  return Buffer.byteLength(JSON.stringify(item), 'utf8') + (submissionBytes.get(item.itemId) ?? 0)
 }

+// Keyed on the snapshot's own submissions array, which the reducer rebuilds on
+// every change, so a paged read over one snapshot serializes submissions once.
+const bytesBySubmissions = new WeakMap<
+  readonly AgentJournalSubmission[],
+  ReadonlyMap<string, number>
+>()
+
 export function submissionBytesByItemId(
  submissions: readonly AgentJournalSubmission[]
-): Map<string, number> {
-  return new Map(
+): ReadonlyMap<string, number> {
+  const cached = bytesBySubmissions.get(submissions)
+  if (cached) {
+    return cached
+  }
+  const bytes = new Map(
    submissions.map((submission) => [
      agentJournalSubmissionKey(submission.clientMessageId),
      Buffer.byteLength(JSON.stringify(submission), 'utf8')
    ])
  )
+  bytesBySubmissions.set(submissions, bytes)
+  return bytes
 }

 export function oversizedHistoryItem(
@@ -45,9 +45,11 @@ export function resolveHistoryLimit(limit: number | undefined): number {

 export function readAgentSessionHistory(
  journal: AgentSessionJournal,
-  request: AgentSessionHistoryRequest
+  request: AgentSessionHistoryRequest,
+  /** Reduced state to read against. A synchronous multi-page catch-up passes one
+   *  snapshot for the whole run so each page costs its own rows, not the timeline. */
+  snapshot: AgentJournalSnapshot = journal.snapshot()
 ): AgentSessionHistoryResult {
-  const snapshot = journal.snapshot()
  if (journal.isReadOnly) {
    return historyReset(snapshot, 'schema_unreadable')
  }
@@ -90,6 +92,24 @@ export function readAgentSessionHistory(
  }
 }

+/**
+ * A catch-up run over one journal. Pages share one reduced timeline, so the run
+ * costs its own rows instead of re-reducing every item per page; the cursor
+ * check re-reduces if anything did advance the journal between pages.
+ */
+export function createAgentSessionCatchUpReader(
+  journal: AgentSessionJournal
+): (request: AgentSessionHistoryRequest) => AgentSessionHistoryResult {
+  let snapshot = journal.snapshot()
+  return (request) => {
+    const live = journal.cursor()
+    if (live.epoch !== snapshot.cursor.epoch || live.sequence !== snapshot.cursor.sequence) {
+      snapshot = journal.snapshot()
+    }
+    return readAgentSessionHistory(journal, request, snapshot)
+  }
+}
+
 export function readAgentSessionHydrationPage(
  journal: AgentSessionJournal,
  fence?: number
@@ -145,7 +165,8 @@ function readForward(
    // a page it cannot place.
    return historyReset(snapshot, 'cursor_ahead')
  }
-  const since = journal.readSince(cursor)
+  // One lookahead preserves hasNewer without rereading the entire remaining journal per page.
+  const since = journal.readSince(cursor, limit + 1)
  if (!since.ok) {
    return historyReset(snapshot, since.reset)
  }
@@ -6,7 +6,11 @@
 // key instead of appearing as a second copy of the user's own message.

 import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key'
-import type { AgentJournalSnapshot } from '../../../shared/agent-session-journal-types'
+import type {
+  AgentJournalRenderItem,
+  AgentJournalSnapshot,
+  AgentJournalSubmission
+} from '../../../shared/agent-session-journal-types'
 import type { AgentSessionJournalBatch } from '../../../shared/agent-session-wire'
 import { findSequenceGap } from '../agent-session-journal/journal-cursor'
 import type { JournalRow } from '../agent-session-journal/journal-row-schema'
@@ -31,7 +35,7 @@ export function projectJournalBatch(input: {
  if (gap) {
    return { ok: false, reset: 'journal_gap' }
  }
-  const aliases = submissionAliases(input.snapshot)
+  const aliases = submissionAliases(input.snapshot.submissions)
  const touchedItemIds = new Set<string>()
  const touchedClientMessageIds = new Set<string>()
  for (const row of input.rows) {
@@ -57,7 +61,7 @@ export function projectJournalBatch(input: {
    }
  }

-  const live = new Map(input.snapshot.items.map((item) => [item.itemId, item]))
+  const live = liveItemsById(input.snapshot.items)
  const items = [...touchedItemIds]
    .map((itemId) => live.get(itemId))
    .filter((item) => item !== undefined)
@@ -75,18 +79,49 @@ export function projectJournalBatch(input: {
  }
 }

+// Both indexes are keyed on the snapshot arrays themselves, which the reducer
+// rebuilds on every change, so a paged catch-up over one snapshot pays for them
+// once instead of once per page — including the byte-shrink loop's re-projections.
+const liveItemsByTimeline = new WeakMap<
+  readonly AgentJournalRenderItem[],
+  ReadonlyMap<string, AgentJournalRenderItem>
+>()
+const aliasesBySubmissions = new WeakMap<
+  readonly AgentJournalSubmission[],
+  ReadonlyMap<string, string>
+>()
+
+function liveItemsById(
+  items: readonly AgentJournalRenderItem[]
+): ReadonlyMap<string, AgentJournalRenderItem> {
+  const cached = liveItemsByTimeline.get(items)
+  if (cached) {
+    return cached
+  }
+  const live = new Map(items.map((item) => [item.itemId, item]))
+  liveItemsByTimeline.set(items, live)
+  return live
+}
+
 /**
 * Provider item id → the submission slot that adopted it, rebuilt from the
 * snapshot's own accepted submissions. This mirrors the alias the reducer
 * writes on an accepted dispatch; deriving it here keeps the projection a pure
 * function of published state instead of reaching into reducer internals.
 */
-function submissionAliases(snapshot: AgentJournalSnapshot): Map<string, string> {
+function submissionAliases(
+  submissions: readonly AgentJournalSubmission[]
+): ReadonlyMap<string, string> {
+  const cached = aliasesBySubmissions.get(submissions)
+  if (cached) {
+    return cached
+  }
  const aliases = new Map<string, string>()
-  for (const submission of snapshot.submissions) {
+  for (const submission of submissions) {
    if (submission.dispatchState === 'accepted' && submission.providerItemId) {
      aliases.set(submission.providerItemId, agentJournalSubmissionKey(submission.clientMessageId))
    }
  }
+  aliasesBySubmissions.set(submissions, aliases)
  return aliases
 }
@@ -18,7 +18,7 @@ import {
 } from '../../../shared/agent-session-wire'
 import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
 import {
-  readAgentSessionHistory,
+  createAgentSessionCatchUpReader,
  readAgentSessionHydrationPage
 } from './agent-session-history-page'

@@ -229,14 +229,13 @@ export class AgentSessionSubscribers {
    backgroundTasks?: AgentSessionBackgroundTaskState | null,
    activity?: AgentSessionTurnActivity | null
  ): void {
-    const publishedActivity =
-      activity !== undefined
-        ? activity
-        : emitCheckpoint
-          ? (this.activityBySession.get(subscriber.sessionId) ?? null)
-          : undefined
+    const checkpointActivity = emitCheckpoint
+      ? this.activityField(subscriber.sessionId).activity
+      : undefined
+    const publishedActivity = activity !== undefined ? activity : checkpointActivity
+    const readPage = createAgentSessionCatchUpReader(journal)
    while (true) {
-      const result = readAgentSessionHistory(journal, {
+      const result = readPage({
        sessionId: subscriber.sessionId,
        direction: 'after',
        cursor: subscriber.cursor,