feat(native-chat): serve the chat queue as its own bounded read

src/main/ipc/runtime-environments-call-routing.test.ts

@@ -121,12 +121,29 @@ describe('registerRuntimeEnvironmentHandlers', () => {
     rmSync(userDataPath, { recursive: true, force: true })
   })
 
-  it('proxies generic one-shot RPC calls to the saved remote runtime', async () => {
+  it.each([
+    { method: 'repo.list', params: undefined, result: { repos: [{ id: 'repo-1' }] } },
+    {
+      method: 'agentSession.history',
+      params: { sessionId: 'session-remote', direction: 'tail', queueView: 'paged-v1' },
+      result: { ok: true, page: { queueSummary: { total: 1 } } }
+    },
+    {
+      method: 'agentSession.queuedMessagesPage',
+      params: { sessionId: 'session-remote', source: 'person', aroundMessageId: 'draft' },
+      result: { status: 'page', rows: [{ messageId: 'draft' }] }
+    },
+    {
+      method: 'agentSession.queuedMessageRead',
+      params: { sessionId: 'session-remote', messageId: 'draft', cursor: 'opaque' },
+      result: { status: 'body', part: { offset: 0, json: '{}' } }
+    }
+  ])('proxies $method to the saved execution host', async ({ method, params, result }) => {
     registerRuntimeEnvironmentHandlers(store as never)
     sendRemoteRuntimeRequestMock.mockResolvedValue({
       id: 'rpc-2',
       ok: true,
-      result: { repos: [{ id: 'repo-1' }] },
+      result,
       _meta: { runtimeId: 'runtime-remote' }
     })
 
@@ -140,16 +157,14 @@ describe('registerRuntimeEnvironmentHandlers', () => {
       { selector: string; method: string; params?: unknown; timeoutMs?: number },
       { ok: true; result: unknown }
     >('runtimeEnvironments:call')
-    expect(
-      await call(null, { selector: 'desk', method: 'repo.list', timeoutMs: 75 })
-    ).toMatchObject({
+    expect(await call(null, { selector: 'desk', method, params, timeoutMs: 75 })).toMatchObject({
       ok: true,
-      result: { repos: [{ id: 'repo-1' }] }
+      result
     })
     expect(sendRemoteRuntimeRequestMock).toHaveBeenCalledWith(
       expect.objectContaining({ endpoint: 'ws://127.0.0.1:6768' }),
-      'repo.list',
-      undefined,
+      method,
+      params,
       75,
       undefined,
       undefined,

src/main/native-chat/agent-session-journal/journal-database.ts

@@ -128,7 +128,7 @@ function migrateJournalSchema(db: Database.Database, stored: number): void {
 }
 
 /**
- * One IMMEDIATE transaction, its COMMIT inside the failure boundary: SQLite can leave a transaction
+ * One synchronous transaction, its COMMIT inside the failure boundary: SQLite can leave a transaction
  * open after a failed COMMIT, and on the shared connection every later BEGIN would then fail. The
  * caller gets the original error; a ROLLBACK that fails too goes to `onStranded`. `run` is
  * synchronous by contract: an await inside it would let another chat's statements land in this
@@ -137,9 +137,10 @@ function migrateJournalSchema(db: Database.Database, stored: number): void {
 export function runJournalTransaction<T>(
   db: Database.Database,
   run: (db: Database.Database) => T,
-  onStranded: () => void = () => undefined
+  onStranded: () => void = () => undefined,
+  mode: 'write' | 'read' = 'write'
 ): T {
-  db.exec('BEGIN IMMEDIATE')
+  db.exec(mode === 'read' ? 'BEGIN' : 'BEGIN IMMEDIATE')
   try {
     const result = run(db)
     if (result instanceof Promise) {

src/main/native-chat/agent-session-journal/journal-host-database.ts

@@ -67,6 +67,18 @@ export class JournalHostDatabase {
     })
   }
 
+  /** A short read snapshot, without reserving the writer slot. */
+  readTransaction<T>(run: () => T): T {
+    return runJournalTransaction(
+      this.db,
+      run,
+      () => {
+        this.stranded = true
+      },
+      'read'
+    )
+  }
+
   /** Last, after every store has drained. A close that fails keeps the handle, so the retried
    *  teardown closes this same connection. */
   close(): void {

src/main/native-chat/agent-session-journal/journal-queued-message-pages.ts

@@ -0,0 +1,154 @@
+import { randomUUID } from 'node:crypto'
+import {
+  isAgentSessionQueueSource,
+  type AgentSessionQueueSource
+} from '../../../shared/agent-session-queue-pages'
+import type { SqliteBindings } from '../../sqlite/sqlite-statement'
+import type { JournalHostDatabase } from './journal-host-database'
+import { getQueuedMessageHeader } from './queued-message-headers'
+import { readStoredQueuedMessageRow } from './queued-message-stored-row'
+import {
+  QUEUED_MESSAGE_SOURCE_SQL,
+  QUEUED_MESSAGE_UNSETTLED_SQL
+} from './queued-message-source-index'
+
+export type QueuePageAnchor = { position: number; messageId: string }
+export type QueuePageRange = {
+  source?: AgentSessionQueueSource
+  anchor?: QueuePageAnchor
+  direction: 'before' | 'after'
+  inclusive?: boolean
+}
+
+function sourceBindings(source: AgentSessionQueueSource | undefined): SqliteBindings {
+  return source === undefined ? [] : [source]
+}
+
+function rangeSql(range: QueuePageRange): { sql: string; bindings: SqliteBindings } {
+  const comparison = range.direction === 'before' ? '<' : '>'
+  return {
+    sql: `${range.source === undefined ? '' : `AND ${QUEUED_MESSAGE_SOURCE_SQL} = ?`}
+      ${range.anchor ? `AND (position, message_id) ${comparison}${range.inclusive ? '=' : ''} (?, ?)` : ''}`,
+    bindings: [
+      ...sourceBindings(range.source),
+      ...(range.anchor ? [range.anchor.position, range.anchor.messageId] : [])
+    ]
+  }
+}
+
+export class JournalQueuedMessagePages {
+  private incarnation = randomUUID()
+  private epoch: string | undefined
+
+  constructor(
+    private readonly database: () => JournalHostDatabase,
+    private readonly sessionId: string
+  ) {}
+
+  generation(epoch: string): string {
+    if (this.epoch !== undefined && this.epoch !== epoch) {
+      this.incarnation = randomUUID()
+    }
+    this.epoch = epoch
+    return this.incarnation
+  }
+
+  read<T>(run: () => T): T {
+    const database = this.database()
+    return database.db.isTransaction ? run() : database.readTransaction(run)
+  }
+
+  counts(): Record<AgentSessionQueueSource, number> {
+    const counts = { person: 0, agent: 0, unknown: 0 }
+    for (const row of this.database()
+      .db.prepare(`
+      SELECT ${QUEUED_MESSAGE_SOURCE_SQL} AS source, COUNT(*) AS count
+      FROM queued_messages WHERE session_id = ? AND ${QUEUED_MESSAGE_UNSETTLED_SQL}
+      GROUP BY ${QUEUED_MESSAGE_SOURCE_SQL}`)
+      .all(this.sessionId)) {
+      if (isAgentSessionQueueSource(row.source) && typeof row.count === 'number') {
+        counts[row.source] = row.count
+      }
+    }
+    return counts
+  }
+
+  newestPersonMessageId(): string | null {
+    const row = this.database()
+      .db.prepare(`
+      SELECT message_id FROM queued_messages
+      WHERE session_id = ? AND ${QUEUED_MESSAGE_UNSETTLED_SQL} AND ${QUEUED_MESSAGE_SOURCE_SQL} = 'person'
+      ORDER BY created_at DESC, position DESC, message_id DESC LIMIT 1`)
+      .get(this.sessionId)
+    return typeof row?.message_id === 'string' ? row.message_id : null
+  }
+
+  blockingReturnedMessageId(): string | null {
+    const row = this.database()
+      .db.prepare(`SELECT message_id FROM queued_messages
+      WHERE session_id = ? AND state = 'returned' AND ${QUEUED_MESSAGE_UNSETTLED_SQL}
+      ORDER BY position, message_id LIMIT 1`)
+      .get(this.sessionId)
+    return typeof row?.message_id === 'string' ? row.message_id : null
+  }
+
+  anchor(messageId: string, source?: AgentSessionQueueSource): QueuePageAnchor | null {
+    const row = this.database()
+      .db.prepare(`
+      SELECT message_id, position FROM queued_messages
+      WHERE session_id = ? AND message_id = ? AND ${QUEUED_MESSAGE_UNSETTLED_SQL}
+      ${source === undefined ? '' : `AND ${QUEUED_MESSAGE_SOURCE_SQL} = ?`}`)
+      .get(this.sessionId, messageId, ...sourceBindings(source))
+    return typeof row?.message_id === 'string' && typeof row.position === 'number'
+      ? { messageId: row.message_id, position: row.position }
+      : null
+  }
+
+  hasRows(range: QueuePageRange): boolean {
+    const { sql, bindings } = rangeSql(range)
+    const row = this.database()
+      .db.prepare(`
+      SELECT 1 FROM queued_messages
+      WHERE session_id = ? AND ${QUEUED_MESSAGE_UNSETTLED_SQL} ${sql}
+      LIMIT 1`)
+      .get(this.sessionId, ...bindings)
+    return row !== undefined
+  }
+
+  *rows(range: QueuePageRange, size: number) {
+    const { sql, bindings } = rangeSql(range)
+    const order = range.direction === 'before' ? 'DESC' : 'ASC'
+    for (const stored of this.database()
+      .db.prepare(`
+      SELECT session_id, message_id, position, body_json, fingerprint, created_at, host_instance,
+        state, hold_reason, returned_reason, returned_rejection, settled_at, settled_by_op,
+        consumed_as, carried_from, queued_epoch, queued_sequence,
+        length(CAST(body_json AS BLOB)) AS body_bytes, ${QUEUED_MESSAGE_SOURCE_SQL} AS source
+      FROM queued_messages WHERE session_id = ? AND ${QUEUED_MESSAGE_UNSETTLED_SQL} ${sql}
+      ORDER BY position ${order}, message_id ${order} LIMIT ?`)
+      .iterate(this.sessionId, ...bindings, size)) {
+      const row = readStoredQueuedMessageRow(stored)
+      const source = stored.source
+      if (row && typeof stored.body_bytes === 'number' && isAgentSessionQueueSource(source)) {
+        yield { row, source, bodyBytes: stored.body_bytes }
+      }
+    }
+  }
+
+  body(messageId: string) {
+    const db = this.database().db
+    const header = getQueuedMessageHeader(db, this.sessionId, messageId)
+    if (!header || (header.state !== 'waiting' && header.state !== 'returned')) {
+      return { header, json: null, bodyBytes: 0 }
+    }
+    const row = db
+      .prepare(`SELECT body_json, length(CAST(body_json AS BLOB)) AS bytes
+      FROM queued_messages WHERE session_id = ? AND message_id = ?`)
+      .get(this.sessionId, messageId)
+    return {
+      header,
+      json: typeof row?.body_json === 'string' ? row.body_json : null,
+      bodyBytes: typeof row?.bytes === 'number' ? row.bytes : 0
+    }
+  }
+}

src/main/native-chat/agent-session-journal/journal-queued-messages.ts

@@ -5,6 +5,7 @@
 // working status, teardown, or the idle sweep.
 
 import type Database from '../../sqlite/sync-database'
+import { JournalQueuedMessagePages } from './journal-queued-message-pages'
 import type {
   AgentJournalCursor,
   AgentJournalMessageItem
@@ -79,15 +80,17 @@ export class JournalQueuedMessages {
   /** Bumped on every draft-table write, so publication memos recompute only when they must. */
   private changeRevision = 0
 
-  constructor(private readonly deps: JournalQueuedMessagesDeps) {}
+  readonly pages: JournalQueuedMessagePages
+
+  constructor(private readonly deps: JournalQueuedMessagesDeps) {
+    this.pages = new JournalQueuedMessagePages(deps.database, deps.sessionId)
+  }
 
   get sessionId(): string {
     return this.deps.sessionId
   }
 
-  revision(): number {
-    return this.changeRevision
-  }
+  revision = (): number => this.changeRevision
 
   /** A journal transaction rolled back: nothing read inside it may stay cached. */
   invalidate(): void {
@@ -103,9 +106,8 @@ export class JournalQueuedMessages {
     return queuedMessageHeaders(this.deps.database().db, this.deps.sessionId, selection)
   }
 
-  publishedBodyBytes(): number {
-    return unsettledQueuedMessageBodyBytes(this.deps.database().db, this.deps.sessionId)
-  }
+  publishedBodyBytes = (): number =>
+    unsettledQueuedMessageBodyBytes(this.deps.database().db, this.deps.sessionId)
 
   nextSendable() {
     return hasWaitingQueuedMessage(this.deps.database().db, this.deps.sessionId)

src/main/native-chat/agent-session-journal/queued-message-schema.ts

@@ -3,6 +3,7 @@
 import type Database from '../../sqlite/sync-database'
 import { classifyJournalOpenFailure } from './journal-open-failure'
 import { READABLE_UNSETTLED_QUEUED_MESSAGE } from './queued-message-readability'
+import { QUEUED_MESSAGE_SOURCE_SQL } from './queued-message-source-index'
 
 /** Columns a later build added, so an older draft table can gain them in place. */
 const NULLABLE_COLUMNS: readonly (readonly [name: string, type: string])[] = [
@@ -80,6 +81,15 @@ CREATE INDEX IF NOT EXISTS queued_messages_readable_unsettled_position
   ON queued_messages (session_id, position) WHERE ${READABLE_UNSETTLED_QUEUED_MESSAGE};
 CREATE INDEX IF NOT EXISTS queued_messages_state_settled
   ON queued_messages (session_id, state, settled_at, message_id);
+CREATE INDEX IF NOT EXISTS queued_messages_unsettled_source_position
+  ON queued_messages (session_id, ${QUEUED_MESSAGE_SOURCE_SQL}, position, message_id)
+  WHERE ${READABLE_UNSETTLED_QUEUED_MESSAGE};
+CREATE INDEX IF NOT EXISTS queued_messages_unsettled_source_created
+  ON queued_messages (session_id, ${QUEUED_MESSAGE_SOURCE_SQL}, created_at, position, message_id)
+  WHERE ${READABLE_UNSETTLED_QUEUED_MESSAGE};
+CREATE INDEX IF NOT EXISTS queued_messages_unsettled_keyset
+  ON queued_messages (session_id, position, message_id)
+  WHERE ${READABLE_UNSETTLED_QUEUED_MESSAGE};
 `)
   } catch (error) {
     console.warn('[journal-open] queued-message indexes skipped:', {

src/main/native-chat/agent-session-journal/queued-message-source-index.ts

@@ -0,0 +1,12 @@
+import { READABLE_UNSETTLED_QUEUED_MESSAGE } from './queued-message-readability'
+
+// Why: an expression index follows old and new writers without another persisted source truth.
+export const QUEUED_MESSAGE_SOURCE_SQL = `CASE
+  WHEN NOT json_valid(body_json) THEN 'unknown'
+  WHEN json_type(body_json, '$.from') IS NULL THEN 'person'
+  WHEN json_type(body_json, '$.from') = 'object'
+    AND json_type(body_json, '$.from.kind') = 'text' THEN 'agent'
+  ELSE 'unknown' END`
+
+// Same text as the partial indexes' WHERE, so the planner can use them.
+export const QUEUED_MESSAGE_UNSETTLED_SQL = READABLE_UNSETTLED_QUEUED_MESSAGE

src/main/native-chat/agent-session-wire/agent-session-queue-page-cursor.ts

@@ -0,0 +1,42 @@
+import { z } from 'zod'
+import { AGENT_SESSION_QUEUE_SOURCES } from '../../../shared/agent-session-queue-pages'
+
+const Scope = z.object({ sessionId: z.string(), generation: z.string() })
+const PageCursor = Scope.extend({
+  source: z.enum(AGENT_SESSION_QUEUE_SOURCES).nullable(),
+  position: z.number().finite(),
+  messageId: z.string(),
+  direction: z.enum(['before', 'after'])
+}).strict()
+const BodyCursor = Scope.extend({
+  messageId: z.string(),
+  offset: z.number().int().nonnegative(),
+  fingerprint: z.string()
+}).strict()
+
+export function queueCursor(
+  value: z.infer<typeof PageCursor> | z.infer<typeof BodyCursor>
+): string {
+  return Buffer.from(JSON.stringify(value), 'utf8').toString('base64url')
+}
+
+function decodedCursor(cursor: string): unknown {
+  if (cursor.length > 16 * 1024) {
+    return null
+  }
+  try {
+    return JSON.parse(Buffer.from(cursor, 'base64url').toString('utf8'))
+  } catch {
+    return null
+  }
+}
+
+export function readQueuePageCursor(cursor: string) {
+  const parsed = PageCursor.safeParse(decodedCursor(cursor))
+  return parsed.success ? parsed.data : null
+}
+
+export function readQueueBodyCursor(cursor: string) {
+  const parsed = BodyCursor.safeParse(decodedCursor(cursor))
+  return parsed.success ? parsed.data : null
+}