feat(native-chat): one structured-chat journal database per host, owned by one process

config/scripts/journal-replay-retention-benchmark.mjs

@@ -2,75 +2,86 @@
 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 { basename, dirname, join, resolve } 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.
+// Pass the root of a checkout (or `git archive <base> src` extract) of the base commit: each arm is
+// bundled whole from its own tree and writes its own fixture, since the two store history differently.
 const baselineDir = process.argv[2]
 assert.ok(
   baselineDir,
-  'Usage: node --expose-gc journal-replay-retention-benchmark.mjs BASELINE_DIR'
+  'Usage: node --expose-gc journal-replay-retention-benchmark.mjs BASELINE_CHECKOUT_ROOT'
 )
 assert.ok(global.gc, 'Run with --expose-gc to measure live backing memory during replay')
 const root = fileURLToPath(new URL('../..', import.meta.url))
+const journalSource = './src/main/native-chat/agent-session-journal'
+const entries = {
+  baseline: `export {openAgentSessionJournal} from '${journalSource}/journal-store-factory'; export {replayJournal} from '${journalSource}/journal-open'; export {openJournalDatabase} from '${journalSource}/journal-database'; export {journalDatabaseFile} from '${journalSource}/journal-paths';`,
+  current: `export {openAgentSessionJournal} from '${journalSource}/journal-store-factory'; export {replayJournal} from '${journalSource}/journal-open'; export {JournalHostDatabase, journalDatabasePath} from '${journalSource}/journal-host-database';`
+}
+const sourceRoots = { baseline: resolve(baselineDir), current: root }
+const identity = {
+  sessionId: 'benchmark',
+  workspaceId: 'fixture',
+  hostId: 'local',
+  agent: 'codex',
+  providerHandle: { kind: 'codex', threadId: 'thread' }
+}
 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) }
-              }
-            )
-          }
+// Released in `finally`, newest first: an open SQLite handle blocks the fixture's removal on Windows.
+const releases = []
+
+async function bundle(arm) {
+  const outfile = join(fixture, `${arm}.cjs`)
+  await build({
+    stdin: { contents: entries[arm], resolveDir: sourceRoots[arm] },
+    bundle: true,
+    platform: 'node',
+    format: 'cjs',
+    outfile,
+    // A bare base checkout has no node_modules of its own.
+    nodePaths: [join(root, 'node_modules')],
+    plugins: [
+      {
+        name: 'replay-memory-probe',
+        setup(plugin) {
+          plugin.onLoad({ filter: /journal-reducer\.ts$/ }, async ({ path }) => {
+            const marker =
+              'export function applyJournalRow(state: JournalReducerState, row: JournalRow): void {'
+            const source = await readFile(path, 'utf8')
+            assert.ok(source.includes(marker), `${basename(path)} lost the probe marker`)
+            return {
+              contents: source.replace(
+                marker,
+                `${marker}\nglobalThis.__replayMemoryProbe?.(row.seq);`
+              ),
+              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' }
+      }
+    ]
+  })
+  return createRequire(import.meta.url)(outfile)
+}
+
+/** One long-revised item, written by the arm's own store; returns the arm's replay of it. */
+async function openArm(arm) {
+  const implementation = await bundle(arm)
+  const stateDirectory = join(fixture, arm)
+  let journal
+  let database
+  if (arm === 'baseline') {
+    journal = await implementation.openAgentSessionJournal({ identity, journalDir: stateDirectory })
+  } else {
+    database = implementation.JournalHostDatabase.open(stateDirectory)
+    releases.push(() => database.close())
+    journal = await implementation.openAgentSessionJournal({ identity, database })
   }
-  const journalDir = join(fixture, 'session')
-  const journal = await implementations.current.openAgentSessionJournal({ identity, journalDir })
+  releases.push(() => journal.close())
   const item = { provider: 'codex', threadId: 'thread', turnId: 'turn', ordinal: 0 }
   const text = 'x'.repeat(32768)
   for (let revision = 0; revision < 2000; revision++) {
@@ -84,11 +95,29 @@ try {
       { fence: 1 }
     )
   }
-  await journal.close()
+  await releases.pop()()
+  if (arm === 'current') {
+    return {
+      path: implementation.journalDatabasePath(stateDirectory),
+      replay: () => implementation.replayJournal(database.db, identity.sessionId)
+    }
+  }
+  const path = implementation.journalDatabaseFile(stateDirectory)
+  const opened = implementation.openJournalDatabase(path)
+  releases.push(() => opened.db.close())
+  return {
+    path,
+    // The base replay takes the connection's read-only flag before the chat.
+    replay: () => implementation.replayJournal(opened.db, opened.readOnly, identity.sessionId)
+  }
+}
+
+try {
+  const arms = { baseline: await openArm('baseline'), current: await openArm('current') }
   for (const arm of ['baseline', 'current', 'current', 'baseline']) {
     global.gc()
     const start = performance.now()
-    let loaded = implementations[arm].loadJournal(journalDir, identity.sessionId)
+    let loaded = arms[arm].replay()
     const ms = performance.now() - start
     assert.equal(loaded.state.items.size, 1)
     assert.equal([...loaded.state.items.values()][0].revision, 2000)
@@ -103,20 +132,27 @@ try {
       global.gc()
       peakLiveHeap = Math.max(peakLiveHeap, process.memoryUsage().heapUsed)
     }
-    loaded = implementations[arm].loadJournal(journalDir, identity.sessionId)
+    loaded = arms[arm].replay()
     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,
+        databaseBytes: (await stat(arms[arm].path)).size,
         peakLiveHeapDelta: peakLiveHeap - initialHeap
       })
     )
   }
 } finally {
   delete globalThis.__replayMemoryProbe
+  for (const release of releases.toReversed()) {
+    try {
+      await release()
+    } catch (error) {
+      console.error('[journal-replay-retention-benchmark] release failed', error)
+    }
+  }
   await rm(fixture, { recursive: true, force: true })
 }

mobile/src/session/mobile-structured-agent-session-rpc.ts

@@ -15,8 +15,10 @@ import {
   agentSessionWriteNoticeParts
 } from '../../../src/shared/agent-session-refusal-notice'
 import {
-  agentSessionRpcErrorFailure,
+  agentSessionRefusalFailure,
+  agentSessionThrownFailure,
   agentSessionWriteKindForMethod,
+  readAgentSessionErrorRefusal,
   type AgentSessionWriteKind
 } from '../../../src/shared/agent-session-write-failure'
 import { structuredSessionOperationId } from './structured-session-operation-id'
@@ -51,12 +53,36 @@ export type StructuredAgentSessionMutate = <TValue>(
 class AgentSessionRpcResponseError extends Error {
   constructor(
     readonly code: string,
-    message: string
+    message: string,
+    /** A thrown refusal's reason rides here; its message is only the bare code. */
+    readonly data?: unknown
   ) {
     super(message)
   }
 }
 
+/** A failed read of a chat's history as the pane shows it, from a thrown error or a stream's error
+ *  frame (`{ message, error }`): a thrown refusal's message is its bare code, so its words come
+ *  from the refusal in the error's data. */
+export function agentSessionReadFailureText(failure: unknown): string {
+  const refusal = readAgentSessionErrorRefusal(
+    typeof failure === 'object' && failure !== null && 'error' in failure ? failure.error : failure
+  )
+  if (refusal) {
+    return agentSessionWriteNoticeEnglish(
+      agentSessionWriteNoticeParts(agentSessionRefusalFailure(refusal), 'read-history')
+    )
+  }
+  if (failure instanceof Error) {
+    return failure.message
+  }
+  return typeof failure === 'object' && failure !== null
+    ? 'message' in failure
+      ? String(failure.message ?? '')
+      : ''
+    : String(failure)
+}
+
 export async function callAgentSession<TResult>(
   client: RpcClient,
   method: string,
@@ -70,7 +96,11 @@ export async function callAgentSession<TResult>(
     ...(options?.failWhenDisconnected ? { failWhenDisconnected: true } : {})
   })
   if (!response.ok) {
-    throw new AgentSessionRpcResponseError(response.error.code, response.error.message)
+    throw new AgentSessionRpcResponseError(
+      response.error.code,
+      response.error.message,
+      response.error.data
+    )
   }
   return response.result as TResult
 }
@@ -186,7 +216,9 @@ export async function requestStructuredAgentSessionMutation<TValue>(args: {
         }
   } catch (error) {
     const answered =
-      error instanceof AgentSessionRpcResponseError ? agentSessionRpcErrorFailure(error.code) : null
+      error instanceof AgentSessionRpcResponseError
+        ? agentSessionThrownFailure(error, error.code)
+        : null
     if (answered && answered.kind !== 'unconfirmed') {
       // The host turned the request away before running it; its text is written for a log.
       return {

mobile/src/session/mobile-structured-agent-session-thrown-refusal.test.ts

@@ -0,0 +1,113 @@
+import { describe, expect, it } from 'vitest'
+import type { RpcClient } from '../transport/rpc-client'
+import {
+  agentSessionReadFailureText,
+  callAgentSession,
+  requestStructuredAgentSessionMutation
+} from './mobile-structured-agent-session-rpc'
+
+// As `mapRuntimeError` sends a thrown refusal (pinned in `src/main/runtime/rpc/errors.test.ts`):
+// its message is the bare code, and its reason rides in data.
+const THROWN_JOURNAL_REFUSAL = {
+  code: 'runtime_error',
+  message: 'agent_session_journal_unreadable',
+  data: {
+    refusal: {
+      code: 'agent_session_journal_unreadable',
+      details: { reason: 'journalCorrupt' }
+    }
+  }
+}
+
+function refusingClient(): RpcClient {
+  const sendRequest = async () => ({ id: 'req-1', ok: false, error: THROWN_JOURNAL_REFUSAL })
+  // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the paths under test reach only `sendRequest`.
+  return { sendRequest } as unknown as RpcClient
+}
+
+describe('a refusal the host threw', () => {
+  it("reads a chat's history failure in the refusal's words, from a stream or a request", async () => {
+    const words = 'Unable to load this chat.'
+    // The stream's error frame, as the RPC client hands it over.
+    expect(
+      agentSessionReadFailureText({
+        type: 'error',
+        message: THROWN_JOURNAL_REFUSAL.message,
+        error: THROWN_JOURNAL_REFUSAL
+      })
+    ).toBe(words)
+    const thrown = await callAgentSession(refusingClient(), 'agentSession.history', {}).catch(
+      (error: unknown) => error
+    )
+    expect(agentSessionReadFailureText(thrown)).toBe(words)
+    expect(agentSessionReadFailureText({ type: 'error', message: 'Connection interrupted' })).toBe(
+      'Connection interrupted'
+    )
+  })
+
+  it("fails a write with the refusal's words, not as unconfirmed", async () => {
+    const result = await requestStructuredAgentSessionMutation({
+      client: refusingClient(),
+      method: 'agentSession.cancel',
+      fingerprintMethod: 'agentSession.cancel',
+      sessionId: 'session-1',
+      expectedRuntimeFence: 3,
+      fields: { turnId: 'turn-1' },
+      clientOperationId: `1900000000000-${'c'.repeat(32)}`
+    })
+
+    expect(result).toEqual({
+      status: 'failed',
+      message: "Unable to load this chat. The agent wasn't stopped."
+    })
+  })
+})
+
+describe('a write refused on a journal a newer Orca wrote', () => {
+  // As the host answers it (pinned in `journal-open-failure.test.ts`).
+  const newerOrcaClient = (): RpcClient => {
+    const sendRequest = async () => ({
+      id: 'req-1',
+      ok: true,
+      result: {
+        ok: false,
+        refusal: {
+          code: 'agent_session_journal_unreadable',
+          message: 'Chats were saved by a newer Orca. Update Orca to keep using them.',
+          details: { reason: 'journalWrittenByNewerOrca' }
+        }
+      }
+    })
+    // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the paths under test reach only `sendRequest`.
+    return { sendRequest } as unknown as RpcClient
+  }
+
+  it.each([
+    [
+      'agentSession.send',
+      { body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hello' }] } },
+      'Chats were saved by a newer Orca. Your message was not sent. Update Orca to keep using them.'
+    ],
+    [
+      'agentSession.cancel',
+      { turnId: 'turn-1' },
+      "Chats were saved by a newer Orca. The agent wasn't stopped. Update Orca to keep using them."
+    ]
+  ] as const)('says to update Orca for %s', async (method, fields, message) => {
+    const result = await requestStructuredAgentSessionMutation({
+      client: newerOrcaClient(),
+      method,
+      fingerprintMethod: method,
+      sessionId: 'session-1',
+      expectedRuntimeFence: 3,
+      fields,
+      clientOperationId: `1900000000000-${'d'.repeat(32)}`
+    })
+
+    expect(result).toEqual({
+      status: 'refused',
+      code: 'agent_session_journal_unreadable',
+      message
+    })
+  })
+})