From ecc5b499ff177bc9d9d1489faa0a474e9f01b498 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 8 Oct 2026 17:03:05 -0700 Subject: [PATCH 1/2] feat(native-chat): add bounded queue reads beside chat history Hosts advertise agent-session.queue-pages.v1 and serve the queue as its own read: a summary on history/subscribe frames for readers that opt in with queueView 'paged-v1', plus byte-bounded queuedMessagesPage and queuedMessageRead methods. Shipped readers keep the full v1 list. --- .../runtime-environments-call-routing.test.ts | 31 +- .../agent-session-journal/journal-database.ts | 7 +- .../journal-host-database.ts | 12 + .../journal-queued-message-pages.ts | 158 +++++++++ .../journal-queued-messages.ts | 11 +- .../queued-message-schema.ts | 10 + .../queued-message-source-index.ts | 10 + .../agent-session-queue-page-cursor.ts | 42 +++ .../agent-session-queue-pages.test-fixture.ts | 101 ++++++ .../agent-session-queue-preview.ts | 61 ++++ .../agent-session-queue-reply-budget.ts | 31 ++ .../agent-session-subscriber-frame-fields.ts | 31 +- .../agent-session-subscriber-output.ts | 87 +++++ ...d-agent-session-background-task-channel.ts | 19 +- ...tructured-agent-session-client-delivery.ts | 9 +- .../structured-agent-session-host-reads.ts | 61 ++++ .../structured-agent-session-host.ts | 43 ++- ...tructured-agent-session-queue-body.test.ts | 114 +++++++ .../structured-agent-session-queue-body.ts | 105 ++++++ ...ctured-agent-session-queue-history.test.ts | 97 ++++++ ...tructured-agent-session-queue-page.test.ts | 310 ++++++++++++++++++ .../structured-agent-session-queue-page.ts | 125 +++++++ ...d-agent-session-queue-subscription.test.ts | 303 +++++++++++++++++ .../structured-agent-session-queue-summary.ts | 82 +++++ ...ctured-agent-session-queued-publication.ts | 32 +- ...ssion-subscriber-journal-retention.test.ts | 109 +++--- .../structured-agent-session-subscribers.ts | 103 ++++-- src/main/runtime/mobile-rpc-allowlist.test.ts | 2 + ...ession-gate-classification.test-fixture.ts | 5 + ...tured-agent-session-queue-envelope.test.ts | 72 ++++ ...ructured-agent-session-queue-pages.test.ts | 88 +++++ ...structured-agent-session-queued-methods.ts | 27 +- ...ructured-agent-session-rpc.test-fixture.ts | 6 +- .../structured-agent-session-schemas.ts | 2 + .../methods/structured-agent-session.test.ts | 2 +- .../rpc/methods/structured-agent-session.ts | 3 +- ...untime-rpc-mobile-agent-session-methods.ts | 2 + src/shared/agent-session-queue-pages.ts | 86 +++++ .../agent-session-queued-message-wire.ts | 3 + src/shared/agent-session-wire.ts | 4 + src/shared/protocol-version.ts | 13 +- .../rpc-params-catalog.generated.ts | 4 + .../structured-agent-session-params.ts | 28 +- ...uctured-agent-session-queue-page-params.ts | 28 ++ ...ss-version-agent-session-wire.unit.test.ts | 7 + .../structured-agent-session-host-fixture.ts | 2 + .../structured-agent-session-surface-calls.ts | 10 + ...ructured-agent-session-surface-manifest.ts | 2 + 48 files changed, 2321 insertions(+), 179 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-queued-message-pages.ts create mode 100644 src/main/native-chat/agent-session-journal/queued-message-source-index.ts create mode 100644 src/main/native-chat/agent-session-wire/agent-session-queue-page-cursor.ts create mode 100644 src/main/native-chat/agent-session-wire/agent-session-queue-pages.test-fixture.ts create mode 100644 src/main/native-chat/agent-session-wire/agent-session-queue-preview.ts create mode 100644 src/main/native-chat/agent-session-wire/agent-session-queue-reply-budget.ts create mode 100644 src/main/native-chat/agent-session-wire/agent-session-subscriber-output.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-host-reads.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-queue-body.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-queue-body.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-queue-history.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-queue-page.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-queue-page.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-queue-subscription.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-queue-summary.ts create mode 100644 src/main/runtime/rpc/methods/structured-agent-session-queue-envelope.test.ts create mode 100644 src/main/runtime/rpc/methods/structured-agent-session-queue-pages.test.ts create mode 100644 src/shared/agent-session-queue-pages.ts create mode 100644 src/shared/rpc-contract/structured-agent-session-queue-page-params.ts diff --git a/src/main/ipc/runtime-environments-call-routing.test.ts b/src/main/ipc/runtime-environments-call-routing.test.ts index 643a1dd10e9b..0a36f1eaa736 100644 --- a/src/main/ipc/runtime-environments-call-routing.test.ts +++ b/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, diff --git a/src/main/native-chat/agent-session-journal/journal-database.ts b/src/main/native-chat/agent-session-journal/journal-database.ts index ba961541a7e9..9de3955ef17a 100644 --- a/src/main/native-chat/agent-session-journal/journal-database.ts +++ b/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( 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) { diff --git a/src/main/native-chat/agent-session-journal/journal-host-database.ts b/src/main/native-chat/agent-session-journal/journal-host-database.ts index aaea0228b6e9..ad729bc59f27 100644 --- a/src/main/native-chat/agent-session-journal/journal-host-database.ts +++ b/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(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 { diff --git a/src/main/native-chat/agent-session-journal/journal-queued-message-pages.ts b/src/main/native-chat/agent-session-journal/journal-queued-message-pages.ts new file mode 100644 index 000000000000..f318e8154a6e --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-queued-message-pages.ts @@ -0,0 +1,158 @@ +import { randomUUID } from 'node:crypto' +import 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(run: () => T): T { + const database = this.database() + return database.db.isTransaction ? run() : database.readTransaction(run) + } + + counts(): Record { + 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 ( + (row.source === 'person' || row.source === 'agent' || row.source === 'unknown') && + 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' && + (source === 'person' || source === 'agent' || source === 'unknown') + ) { + 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 + } + } +} diff --git a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts b/src/main/native-chat/agent-session-journal/journal-queued-messages.ts index 9e35f65542ce..19cfa9f87848 100644 --- a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts +++ b/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 @@ -83,15 +84,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 { diff --git a/src/main/native-chat/agent-session-journal/queued-message-schema.ts b/src/main/native-chat/agent-session-journal/queued-message-schema.ts index b343b2f0b508..003c384f39d2 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-schema.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-schema.ts @@ -1,6 +1,7 @@ // The draft table's shape, created and healed at every writable open. import type Database from '../../sqlite/sync-database' +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])[] = [ @@ -73,5 +74,14 @@ CREATE INDEX IF NOT EXISTS queued_messages_unsettled_position ON queued_messages (session_id, position) WHERE state IN ('waiting', 'returned'); 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 state IN ('waiting', 'returned') AND json_valid(body_json); +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 state IN ('waiting', 'returned') AND json_valid(body_json); +CREATE INDEX IF NOT EXISTS queued_messages_unsettled_keyset + ON queued_messages (session_id, position, message_id) + WHERE state IN ('waiting', 'returned') AND json_valid(body_json); `) } diff --git a/src/main/native-chat/agent-session-journal/queued-message-source-index.ts b/src/main/native-chat/agent-session-journal/queued-message-source-index.ts new file mode 100644 index 000000000000..4087fdd4816e --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-source-index.ts @@ -0,0 +1,10 @@ +// 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` + +export const QUEUED_MESSAGE_UNSETTLED_SQL = + "state IN ('waiting', 'returned') AND json_valid(body_json)" diff --git a/src/main/native-chat/agent-session-wire/agent-session-queue-page-cursor.ts b/src/main/native-chat/agent-session-wire/agent-session-queue-page-cursor.ts new file mode 100644 index 000000000000..68b2f3332f41 --- /dev/null +++ b/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 | z.infer +): 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 +} diff --git a/src/main/native-chat/agent-session-wire/agent-session-queue-pages.test-fixture.ts b/src/main/native-chat/agent-session-wire/agent-session-queue-pages.test-fixture.ts new file mode 100644 index 000000000000..443e62e7c7ed --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-queue-pages.test-fixture.ts @@ -0,0 +1,101 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import { claudeProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' +import { + createTrackedJournalOpener, + openTestJournalHostDatabase +} from '../agent-session-journal/journal-host-database-test-support' +import { readQueuedMessagesPage } from './structured-agent-session-queue-page' +import { readQueuedMessageBody } from './structured-agent-session-queue-body' +import { readQueueSummary } from './structured-agent-session-queue-summary' + +export const PAGE_SESSION = 'queue-page-tests' +export const PAGE_GATE = () => ({ record: null, fence: 0 }) +export const PAGE_ENVELOPE = { id: 'request-"\\🦦', runtimeId: 'runtime-漢' } +export function pageMessage(text: string): AgentJournalMessageItem { + return { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] } +} + +export async function queuePageRig() { + const root = await mkdtemp(join(tmpdir(), 'orca-queue-pages-')) + const journals = createTrackedJournalOpener() + let epoch = 0 + const open = () => + journals.open({ + identity: { + sessionId: PAGE_SESSION, + workspaceId: 'folder', + hostId: 'host', + agent: 'claude', + providerHandle: claudeProviderHandle('native', null) + }, + stateDirectory: root, + mintEpoch: () => `epoch-${++epoch}` + }) + let journal = await open() + let db = openTestJournalHostDatabase(root).db + const insert = (messageId: string, body = pageMessage(messageId)) => + journal.queuedMessages.insert({ + messageId, + body, + fingerprint: `fp-${messageId}`, + hostInstance: 'host' + }) + const seed = (count: number, body = pageMessage('x')) => { + db.prepare(`WITH RECURSIVE cards(n) AS (VALUES(1) UNION ALL SELECT n + 1 FROM cards WHERE n < ?) + INSERT INTO queued_messages (session_id, message_id, position, body_json, fingerprint, created_at, host_instance, state) + SELECT ?, printf('card-%05d', n), n / 3, ?, 'fp', 1, 'host', 'waiting' FROM cards`).run( + count, + PAGE_SESSION, + JSON.stringify(body) + ) + journal.queuedMessages.invalidate() + } + const wrapped = (result: unknown) => ({ + id: PAGE_ENVELOPE.id, + ok: true, + result, + _meta: { runtimeId: PAGE_ENVELOPE.runtimeId } + }) + return { + get journal() { + return journal + }, + get db() { + return db + }, + reopen: async () => { + await journals.closeAll() + journal = await open() + db = openTestJournalHostDatabase(root).db + }, + insert, + seed, + wrapped, + page: ( + params: Omit[2], 'sessionId'> = {}, + envelope = PAGE_ENVELOPE + ) => + readQueuedMessagesPage(journal, PAGE_GATE, { sessionId: PAGE_SESSION, ...params }, envelope), + body: ( + messageId: string, + params: Omit[2], 'sessionId' | 'messageId'> = {}, + envelope = PAGE_ENVELOPE + ) => + readQueuedMessageBody( + journal, + PAGE_GATE, + { sessionId: PAGE_SESSION, messageId, ...params }, + envelope + ), + summary: () => readQueueSummary(journal, PAGE_GATE).queueSummary, + close: async () => { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) + } + } +} + +export type QueuePageRig = Awaited> diff --git a/src/main/native-chat/agent-session-wire/agent-session-queue-preview.ts b/src/main/native-chat/agent-session-wire/agent-session-queue-preview.ts new file mode 100644 index 000000000000..e25e345e521e --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-queue-preview.ts @@ -0,0 +1,61 @@ +import { + AGENT_SESSION_QUEUE_PREVIEW_MAX_BYTES, + type AgentSessionQueuedMessagePreview, + type AgentSessionQueueSource +} from '../../../shared/agent-session-queue-pages' +import { readAgentMessageSource } from '../../../shared/agent-session-message-source' +import { QUEUED_MESSAGE_PAUSED_SEND_FAILED } from '../../../shared/agent-session-queued-message-wire' +import { clampUtf8TextPrefix } from '../../../shared/utf8-byte-limits' +import type { QueuedMessageRow } from '../agent-session-journal/queued-message-table' + +export function queuedMessagePreview(input: { + row: QueuedMessageRow + source: AgentSessionQueueSource + bodyBytes: number +}): AgentSessionQueuedMessagePreview | null { + const { row, source, bodyBytes } = input + if (row.state !== 'waiting' && row.state !== 'returned') { + return null + } + let preview = '' + let truncated = false + for (const block of row.body.blocks) { + if (block.type !== 'text') { + continue + } + const full = preview + (preview ? '\n' : '') + block.text + preview = clampUtf8TextPrefix(full, AGENT_SESSION_QUEUE_PREVIEW_MAX_BYTES) + if (preview !== full) { + truncated = true + break + } + } + const senders = readAgentMessageSource(row.body.from)?.senders ?? [] + const paused = row.state === 'waiting' && row.holdReason !== null + return { + messageId: row.messageId, + position: row.position, + state: row.state, + ...(paused ? { paused: true as const } : {}), + ...(paused && row.holdReason === QUEUED_MESSAGE_PAUSED_SEND_FAILED + ? { pausedReason: row.holdReason } + : {}), + ...(row.state === 'returned' + ? { + returnedReason: + row.returnedReason === null ? null : clampUtf8TextPrefix(row.returnedReason, 512), + ...(row.returnedRejection + ? { returnedFailure: { kind: clampUtf8TextPrefix(row.returnedRejection.kind, 128) } } + : {}) + } + : {}), + source, + senderLabels: senders + .slice(0, 3) + .map((sender) => clampUtf8TextPrefix(sender.name ?? sender.party.address, 128)), + senderCount: senders.length, + preview, + truncated, + bodyBytes + } +} diff --git a/src/main/native-chat/agent-session-wire/agent-session-queue-reply-budget.ts b/src/main/native-chat/agent-session-wire/agent-session-queue-reply-budget.ts new file mode 100644 index 000000000000..cdfe011f4044 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-queue-reply-budget.ts @@ -0,0 +1,31 @@ +import { AGENT_SESSION_QUEUE_REPLY_MAX_BYTES } from '../../../shared/agent-session-queue-pages' +import { + JsonStringifyByteLimitError, + stringifyJsonWithinByteLimit +} from '../../../shared/node-bounded-json-stringify' + +export type QueueReplyEnvelope = { id: string; runtimeId: string } +export const LOCAL_QUEUE_REPLY_ENVELOPE: QueueReplyEnvelope = { id: '', runtimeId: '' } + +/** The dispatcher uses these exact fields for unary replies, including on the local IPC path. */ +export function queueReplyFits(result: unknown, envelope: QueueReplyEnvelope): boolean { + try { + stringifyJsonWithinByteLimit( + { id: envelope.id, ok: true, result, _meta: { runtimeId: envelope.runtimeId } }, + AGENT_SESSION_QUEUE_REPLY_MAX_BYTES + ) + return true + } catch (error) { + if (error instanceof JsonStringifyByteLimitError) { + return false + } + throw error + } +} + +export function boundedQueueReply(result: T, envelope: QueueReplyEnvelope): T { + if (!queueReplyFits(result, envelope)) { + throw new Error('Queue reply envelope exceeds its byte budget') + } + return result +} diff --git a/src/main/native-chat/agent-session-wire/agent-session-subscriber-frame-fields.ts b/src/main/native-chat/agent-session-wire/agent-session-subscriber-frame-fields.ts index 07e197ceb6a8..ea5fee14965a 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-subscriber-frame-fields.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-subscriber-frame-fields.ts @@ -9,20 +9,28 @@ import type { AgentSessionSlashCommand, AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire' -import type { QueuePublication } from './structured-agent-session-queued-publication' +import { + queuePublicationFields, + type ReaderQueuePublication +} from './structured-agent-session-queue-summary' +import type { AgentSessionQueueView } from '../../../shared/agent-session-queue-pages' export type SubscriberFieldState = { + queueView?: AgentSessionQueueView sessionId: string commands?: AgentSessionSlashCommand[] | null /** The last queue publication actually SENT. */ - queuePublication?: QueuePublication + queuePublication?: ReaderQueuePublication /** Fingerprint of the roster last SENT; absent until this subscriber's first frame. */ backgroundTasks?: string } export type SubscriberFieldHooks = { readCommands?: (sessionId: string) => AgentSessionSlashCommand[] | undefined - readQueuePublication?: (sessionId: string) => QueuePublication | undefined + readQueuePublication?: ( + sessionId: string, + queueView?: AgentSessionQueueView + ) => ReaderQueuePublication | undefined /** Built from the host's child records, so it is read only when a frame owes it: never per token. */ readBackgroundTasks?: (sessionId: string) => AgentSessionBackgroundTaskState | null } @@ -31,7 +39,7 @@ export type SubscriberFrame = { frame: AgentSessionSubscribeEvent commands: AgentSessionSlashCommand[] | null attachedQueued: boolean - queued: QueuePublication | undefined + queued: ReaderQueuePublication | undefined /** The attached roster's fingerprint; undefined when the frame carries none. */ backgroundTasks: string | undefined } @@ -55,7 +63,9 @@ export function buildSubscriberFrame( (event.type !== 'batch' || commands !== subscriber.commands) // Withheld on intermediate catch-up pages (the caller says so), attached to // every hydrating frame, and to batches only when the list changed. - const queued = withholdQueued ? undefined : hooks.readQueuePublication?.(subscriber.sessionId) + const queued = withholdQueued + ? undefined + : hooks.readQueuePublication?.(subscriber.sessionId, subscriber.queueView) const attachedQueued = queued !== undefined && event.type !== 'end' && @@ -65,13 +75,7 @@ export function buildSubscriberFrame( frame: { ...event, ...(includeCommands ? { commands: commands ?? null } : {}), - ...(attachedQueued && queued - ? { - queuedMessages: queued.queuedMessages, - queuePause: queued.queuePause, - nextQueuedMessageId: queued.nextQueuedMessageId - } - : {}), + ...(attachedQueued && queued ? queuePublicationFields(queued) : {}), ...(backgroundTasks !== undefined ? { backgroundTasks } : {}) }, commands, @@ -110,6 +114,7 @@ export function subscriberQueuedMessagesChanged( ): boolean { return ( hooks.readQueuePublication !== undefined && - hooks.readQueuePublication(subscriber.sessionId) !== subscriber.queuePublication + hooks.readQueuePublication(subscriber.sessionId, subscriber.queueView) !== + subscriber.queuePublication ) } diff --git a/src/main/native-chat/agent-session-wire/agent-session-subscriber-output.ts b/src/main/native-chat/agent-session-wire/agent-session-subscriber-output.ts new file mode 100644 index 000000000000..2d3475bd6361 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-subscriber-output.ts @@ -0,0 +1,87 @@ +import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire' +import { + buildSubscriberFrame, + type SubscriberFieldHooks +} from './agent-session-subscriber-frame-fields' +import type { Subscriber } from './structured-agent-session-subscribers' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' + +export function resumeSubscriberOutput(input: { + subscriber: Subscriber + readJournal?: (sessionId: string) => Promise + active: () => boolean + ready: (journal: AgentSessionJournal) => void + drop: () => void +}): void { + const { subscriber } = input + if (!subscriber.pendingOutput) { + return + } + // Why: a live reader must not pin an idle fold; reopen through its execution host when ready. + if (input.readJournal) { + subscriber.outputBlocked = true + void input + .readJournal(subscriber.sessionId) + .then((journal) => { + if (input.active()) { + subscriber.journal = new WeakRef(journal) + subscriber.outputBlocked = undefined + input.ready(journal) + } + }) + .catch(input.drop) + } else { + const journal = subscriber.journal.deref() + if (journal) { + input.ready(journal) + } else { + input.drop() + } + } +} + +export function emitSubscriberFrame(input: { + hooks: SubscriberFieldHooks + subscriber: Subscriber + event: AgentSessionSubscribeEvent + withholdQueued: boolean + active: () => boolean + ready: () => void + drop: () => void +}): void { + const { subscriber } = input + if (subscriber.outputBlocked) { + subscriber.pendingOutput = true + return + } + try { + const built = buildSubscriberFrame(input.hooks, subscriber, input.event, input.withholdQueued) + const sent = (): void => { + subscriber.commands = built.commands + if (built.attachedQueued) { + subscriber.queuePublication = built.queued + } + if (built.backgroundTasks !== undefined) { + subscriber.backgroundTasks = built.backgroundTasks + } + } + const delivery: unknown = subscriber.emit(built.frame) + if (subscriber.queueView && delivery instanceof Promise) { + subscriber.outputBlocked = true + void delivery + .then(() => { + if (!input.active()) { + return + } + sent() + subscriber.outputBlocked = undefined + input.ready() + }) + .catch(input.drop) + } else { + sent() + } + } catch { + input.drop() + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.ts index cf44fc3d66b5..26f9c388ae79 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.ts @@ -1,3 +1,7 @@ +import { + queuePublicationFields, + tryReadQueueSummary +} from './structured-agent-session-queue-summary' import type { AgentSessionBackgroundTaskState, AgentSessionHistoryRequest, @@ -52,10 +56,9 @@ export class StructuredAgentSessionBackgroundTaskChannel { scope }) const backgroundTasks = this.read(request.sessionId) - const queue = tryReadQueuePublication( - journal, - structuredQueueSendGate(this.deps.store, request.sessionId) - ) + const queue = ( + request.queueView === 'paged-v1' ? tryReadQueueSummary : tryReadQueuePublication + )(journal, structuredQueueSendGate(this.deps.store, request.sessionId)) const hostNow = this.deps.now?.() ?? Date.now() return { ...result, @@ -64,13 +67,7 @@ export class StructuredAgentSessionBackgroundTaskChannel { hostNow, // A stale history answer never replaces newer live subscription state; // the client's reducer keeps live-over-history precedence. - ...(queue !== undefined - ? { - queuedMessages: queue.queuedMessages, - queuePause: queue.queuePause, - nextQueuedMessageId: queue.nextQueuedMessageId - } - : {}), + ...(queue !== undefined ? queuePublicationFields(queue) : {}), backgroundTasks } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts index c369ce1a5602..62217c74e6af 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts @@ -1,3 +1,4 @@ +import { tryReadQueueSummary } from './structured-agent-session-queue-summary' import type { AgentSessionStatusSummary } from '../../../shared/agent-session-wire' import { AgentSessionRefusalError } from '../../../shared/agent-session-wire-refusals' import type { AgentChildWorkEvidence } from '../../../shared/agent-status-child-work-evidence' @@ -40,7 +41,8 @@ export class StructuredAgentSessionClientDelivery { /** A session's child records changed; the chat strip republishes from them. */ onChildWorkChanged: (sessionId: string) => void, // Required: an opening frame without the roster reads as "no tasks" to current clients. - readBackgroundTasks: NonNullable + readBackgroundTasks: NonNullable, + readJournal: (sessionId: string) => Promise ) { this.statusFeed = createStructuredAgentSessionHostStatusFeed({ sessions, @@ -60,9 +62,10 @@ export class StructuredAgentSessionClientDelivery { ) this.waitForSendSettlement = this.sendSettlement.wait this.subscribers = new AgentSessionSubscribers({ + readJournal, readCommands: (sessionId) => this.readCommands(sessionId), - readQueuePublication: (sessionId) => - tryReadQueuePublication( + readQueuePublication: (sessionId, queueView) => + (queueView === 'paged-v1' ? tryReadQueueSummary : tryReadQueuePublication)( sessions.get(sessionId)?.journal, structuredQueueSendGate(this.deps().store, sessionId) ), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-reads.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-reads.ts new file mode 100644 index 000000000000..1d0ee9bb41e7 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-reads.ts @@ -0,0 +1,61 @@ +import type { AgentJournalSnapshot } from '../../../shared/agent-session-journal-types' +import type { + AgentSessionQueuedMessagesPageRequest, + AgentSessionQueuedMessageReadRequest, + AgentSessionSlashCommand +} from '../../../shared/agent-session-wire' +import { readQueuedMessagesPage } from './structured-agent-session-queue-page' +import { readQueuedMessageBody } from './structured-agent-session-queue-body' +import { structuredQueueSendGate } from './structured-agent-session-queued-publication' +import { structuredAgentSessionOwnerStatus } from './structured-agent-session-owner-status' +import type { QueueReplyEnvelope } from './agent-session-queue-reply-budget' +import type { StructuredAgentSessionBackgroundTaskChannel } from './structured-agent-session-background-task-channel' +import type { + StructuredAgentSessionHostDeps, + StructuredAgentSessionHostSession +} from './structured-agent-session-host-types' + +export function structuredAgentSessionHostReads( + context: () => { + deps: StructuredAgentSessionHostDeps + conversation: (sessionId: string) => Promise + backgroundTasks: StructuredAgentSessionBackgroundTaskChannel + readCommands: (sessionId: string) => AgentSessionSlashCommand[] | undefined + } +) { + return { + readCommands: (sessionId: string) => ({ commands: context().readCommands(sessionId) }), + handoffStatus: (sessionId: string) => + structuredAgentSessionOwnerStatus(context().deps, sessionId), + history: (...args: Parameters) => + context().backgroundTasks.history(...args), + journalSnapshot: async (sessionId: string): Promise => + (await context().conversation(sessionId)).journal.snapshot(), + subscribe: (...args: Parameters) => + context().backgroundTasks.subscribe(...args), + queuedMessagesPage: async ( + request: AgentSessionQueuedMessagesPageRequest, + envelope?: QueueReplyEnvelope + ) => { + const { conversation, deps } = context() + return readQueuedMessagesPage( + (await conversation(request.sessionId)).journal, + structuredQueueSendGate(deps.store, request.sessionId), + request, + envelope + ) + }, + queuedMessageRead: async ( + request: AgentSessionQueuedMessageReadRequest, + envelope?: QueueReplyEnvelope + ) => { + const { conversation, deps } = context() + return readQueuedMessageBody( + (await conversation(request.sessionId)).journal, + structuredQueueSendGate(deps.store, request.sessionId), + request, + envelope + ) + } + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index 9a033b5f831b..00c50081511d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -1,3 +1,4 @@ +import { structuredAgentSessionHostReads } from './structured-agent-session-host-reads' import type { AgentSessionRewindParams } from '../../../shared/agent-session-rewind' import { rewindStructuredAgentSession } from './structured-agent-session-rewind' import { StructuredConversationCommandController } from './structured-conversation-command-controller' @@ -6,16 +7,13 @@ import { StructuredConversationCommandController } from './structured-conversati // only through `conversation`, which opens it at rest; an agent is started only by work that needs // it, and the idle sweep is the one thing that puts it to rest. -import type { AgentJournalSnapshot } from '../../../shared/agent-session-journal-types' import type { AgentSessionExecutionLocation } from '../../../shared/agent-session-record' import type * as SessionWire from '../../../shared/agent-session-wire' import type { AgentSessionAttachParams } from './structured-agent-session-attach' import { createRestartReconciler } from './structured-agent-session-restart-reconcile' -import type { AgentSessionSubscribeInput } from './structured-agent-session-subscribers' import { StructuredAgentSessionTaskQueue } from './structured-agent-session-task-queue' import * as providerSupport from './structured-agent-session-provider-support' import * as reveal from './structured-agent-session-reveal' -import { structuredAgentSessionOwnerStatus } from './structured-agent-session-owner-status' import { StructuredAgentSessionHostRuntimeState } from './structured-agent-session-host-runtime-state' import { attachStructuredAgentSession } from './structured-agent-session-attach-orchestration' import type { StructuredAgentSessionLifetimeContext } from './structured-agent-session-host-lifetime' @@ -79,7 +77,8 @@ export class StructuredAgentSessionHost { (sessionId) => this.queued.onJournalActivity(sessionId), (sessionId) => this.restartResume.onAgentStarted(sessionId), (sessionId) => this.backgroundTasks.publish(sessionId), - (sessionId) => this.backgroundTasks.read(sessionId) + (sessionId) => this.backgroundTasks.read(sessionId), + async (sessionId) => (await this.lifetime.conversation(sessionId)).journal ) private readonly subscribers = this.clientDelivery.subscribers private readonly tasks = new StructuredAgentSessionTaskQueue() @@ -245,13 +244,12 @@ export class StructuredAgentSessionHost { private serialize = this.tasks.serialize.bind(this.tasks) - attach( + attach = ( caller: StructuredAgentSessionCaller, params: AgentSessionAttachParams, options?: Parameters[3] - ): Promise> { - return attachStructuredAgentSession(this.attachContext(), caller.callerKey, params, options) - } + ): Promise> => + attachStructuredAgentSession(this.attachContext(), caller.callerKey, params, options) /** Test barrier: every write has landed by its call's return, so no production path needs it. */ flushStreamedEvents = (id: string): Promise => this.runtimeState.flushEventSink(id) @@ -317,22 +315,19 @@ export class StructuredAgentSessionHost { conversationCommand = (...args: Parameters) => this.conversationCommands.run(...args) conversationReplacements = () => this.conversationCommands.replacements() - /** Undefined means unavailable; an empty array is an authoritative catalog. */ - readCommands = (sessionId: string) => ({ commands: this.clientDelivery.readCommands(sessionId) }) - - /** From the record store, never the session map: an idle-released chat has no map entry. */ - handoffStatus = (sessionId: string): SessionWire.AgentSessionHandoffStatus => - structuredAgentSessionOwnerStatus(this.deps, sessionId) - - history: StructuredAgentSessionBackgroundTaskChannel['history'] = (request, scope) => - this.backgroundTasks.history(request, scope) - - /** The fully reduced timeline, for readers that cannot tolerate a page's ambiguity — rows are - * revised or tombstoned in place, so an item's ABSENCE from a bounded page proves nothing. */ - journalSnapshot = async (sessionId: string): Promise => - (await this.lifetime.conversation(sessionId)).journal.snapshot() - - subscribe = (input: AgentSessionSubscribeInput) => this.backgroundTasks.subscribe(input) + private readonly reads = structuredAgentSessionHostReads(() => ({ + deps: this.deps, + conversation: this.lifetime.conversation, + backgroundTasks: this.backgroundTasks, + readCommands: (sessionId) => this.clientDelivery.readCommands(sessionId) + })) + readCommands = this.reads.readCommands + handoffStatus = this.reads.handoffStatus + history = this.reads.history + journalSnapshot = this.reads.journalSnapshot + subscribe = this.reads.subscribe + queuedMessagesPage = this.reads.queuedMessagesPage + queuedMessageRead = this.reads.queuedMessageRead settleLateDispatch = (input: Parameters[1]) => settleStructuredAgentSessionLateDispatch(this.mutationContext(), input) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queue-body.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-body.test.ts new file mode 100644 index 000000000000..9bb00b88783c --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-body.test.ts @@ -0,0 +1,114 @@ +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { AGENT_SESSION_QUEUE_REPLY_MAX_BYTES } from '../../../shared/agent-session-queue-pages' +import { + PAGE_SESSION, + pageMessage, + queuePageRig, + type QueuePageRig +} from './agent-session-queue-pages.test-fixture' + +let rig: QueuePageRig +beforeEach(async () => { + rig = await queuePageRig() +}) +afterEach(async () => { + vi.restoreAllMocks() + await rig.close() +}) + +it.each(['x'.repeat(256 * 1024), '\u0000"\\🦦漢'.repeat(150000), '\ud800'.repeat(90000)])( + 'reassembles the complete stored body losslessly within each actual reply budget', + async (text) => { + const body = { + ...pageMessage(text), + blocks: [ + { type: 'text' as const, text }, + { type: 'text' as const, text: 'end 🦦' } + ], + from: { + kind: 'agent' as const, + senders: [], + orchestration: { + message: 'task' as const, + taskId: 'task'.repeat(100000), + runId: 'run', + dispatchId: 'dispatch' + } + } + } + const id = '"\\'.repeat(250) + await rig.insert(id, body) + let current = rig.body(id) + let json = '' + let parts = 0 + for (;;) { + expect(Buffer.byteLength(JSON.stringify(rig.wrapped(current)))).toBeLessThanOrEqual( + AGENT_SESSION_QUEUE_REPLY_MAX_BYTES + ) + if (current.status !== 'body') { + throw new Error('body read did not advance') + } + expect(current.part.offset).toBe(json.length) + expect(current.part.json.length).toBeGreaterThan(0) + json += current.part.json + parts++ + if (!current.nextCursor) { + break + } + current = rig.body(id, { cursor: current.nextCursor }) + } + expect(parts).toBeGreaterThan(1) + expect(JSON.parse(json)).toEqual(body) + expect(json).toBe( + rig.db + .prepare('SELECT body_json FROM queued_messages WHERE session_id = ? AND message_id = ?') + .get(PAGE_SESSION, id)?.body_json + ) + } +) + +it('reads a normal body in one part without parsing it, through one short transaction', async () => { + const body = pageMessage('exact original 🦦') + await rig.insert('normal', body) + const parse = vi.spyOn(JSON, 'parse') + const exec = vi.spyOn(rig.db, 'exec') + const result = rig.body('normal') + expect(result).toMatchObject({ + status: 'body', + part: { offset: 0, json: JSON.stringify(body) }, + nextCursor: null + }) + expect(parse).not.toHaveBeenCalled() + expect(exec.mock.calls.map(([sql]) => sql)).toEqual(['BEGIN', 'COMMIT']) +}) + +it('answers gone, handed-off and state-changed as readable outcomes', async () => { + await rig.insert('draft') + expect(rig.body('draft', { expectedState: 'returned' })).toMatchObject({ + status: 'state-changed', + state: 'waiting' + }) + rig.db + .prepare( + "UPDATE queued_messages SET state = 'dispatched', consumed_as = 'submission' WHERE message_id = 'draft'" + ) + .run() + expect(rig.body('draft')).toMatchObject({ status: 'handed-off' }) + rig.db.prepare("UPDATE queued_messages SET state = 'withdrawn' WHERE message_id = 'draft'").run() + expect(rig.body('draft')).toMatchObject({ status: 'gone' }) + expect(rig.body('missing')).toMatchObject({ status: 'gone' }) +}) + +it('resets foreign and retired body cursors rather than returning truncated editable text', async () => { + await rig.insert('huge', pageMessage('🦦'.repeat(300000))) + await rig.insert('other') + const first = rig.body('huge') + if (first.status !== 'body' || !first.nextCursor) { + throw new Error('expected body parts') + } + expect(rig.body('other', { cursor: first.nextCursor }).status).toBe('stale') + expect(rig.body('huge', { cursor: 'bad' }).status).toBe('stale') + await rig.journal.rollEpoch('handle_forked', 1) + expect(rig.body('huge', { cursor: first.nextCursor })).toMatchObject({ status: 'stale' }) + expect(rig.body('huge').generation).not.toBe(first.generation) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queue-body.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-body.ts new file mode 100644 index 000000000000..d4dfdb2a2123 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-body.ts @@ -0,0 +1,105 @@ +import { + AGENT_SESSION_QUEUE_REPLY_MAX_BYTES, + type AgentSessionQueuedMessageReadRequest, + type AgentSessionQueuedMessageReadResult +} from '../../../shared/agent-session-queue-pages' +import { getUtf8ChunkEndIndex } from '../../../shared/utf8-byte-limits' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { + boundedQueueReply, + LOCAL_QUEUE_REPLY_ENVELOPE, + queueReplyFits, + type QueueReplyEnvelope +} from './agent-session-queue-reply-budget' +import { queueCursor, readQueueBodyCursor } from './agent-session-queue-page-cursor' +import { readQueueSummary } from './structured-agent-session-queue-summary' +import type { QueueSendGate } from './structured-agent-session-queued-publication' + +export function readQueuedMessageBody( + journal: AgentSessionJournal, + gate: QueueSendGate, + request: AgentSessionQueuedMessageReadRequest, + envelope: QueueReplyEnvelope = LOCAL_QUEUE_REPLY_ENVELOPE +): AgentSessionQueuedMessageReadResult { + return journal.queuedMessages.pages.read(() => { + const { generation, revision } = readQueueSummary(journal, gate).queueSummary + const reset = () => + boundedQueueReply({ status: 'stale' as const, generation, revision }, envelope) + const cursor = request.cursor ? readQueueBodyCursor(request.cursor) : null + if ( + request.cursor && + (!cursor || + cursor.generation !== generation || + cursor.sessionId !== request.sessionId || + cursor.messageId !== request.messageId) + ) { + return reset() + } + const { header, json, bodyBytes } = journal.queuedMessages.pages.body(request.messageId) + const common = { generation, revision, messageId: request.messageId } + if (!header || json === null || (header.state !== 'waiting' && header.state !== 'returned')) { + return boundedQueueReply( + { + ...common, + status: header?.state === 'dispatched' ? ('handed-off' as const) : ('gone' as const) + }, + envelope + ) + } + if (request.expectedState && request.expectedState !== header.state) { + return boundedQueueReply( + { ...common, status: 'state-changed' as const, state: header.state }, + envelope + ) + } + const offset = cursor?.offset ?? 0 + if ( + (cursor && cursor.fingerprint !== header.fingerprint) || + offset >= json.length || + splitsSurrogate(json, offset) + ) { + return reset() + } + const reply = (end: number): AgentSessionQueuedMessageReadResult => ({ + ...common, + status: 'body', + state: header.state === 'returned' ? 'returned' : 'waiting', + bodyBytes, + part: { offset, json: json.slice(offset, end) }, + nextCursor: + end < json.length + ? queueCursor({ + sessionId: request.sessionId, + generation, + messageId: request.messageId, + fingerprint: header.fingerprint, + offset: end + }) + : null + }) + let end = getUtf8ChunkEndIndex(json, offset, AGENT_SESSION_QUEUE_REPLY_MAX_BYTES) + if (!queueReplyFits(reply(end), envelope)) { + let low = offset + let high = end + while (low < high) { + const middle = Math.ceil((low + high) / 2) + if (queueReplyFits(reply(middle), envelope)) { + low = middle + } else { + high = middle - 1 + } + } + end = splitsSurrogate(json, low) ? low - 1 : low + } + if (end <= offset) { + throw new Error('Queue reply envelope leaves no room for a body part') + } + return boundedQueueReply(reply(end), envelope) + }) +} + +function splitsSurrogate(text: string, offset: number): boolean { + const lead = text.charCodeAt(offset - 1) + const trail = text.charCodeAt(offset) + return lead >= 0xd800 && lead <= 0xdbff && trail >= 0xdc00 && trail <= 0xdfff +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queue-history.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-history.test.ts new file mode 100644 index 000000000000..3975a94a5032 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-history.test.ts @@ -0,0 +1,97 @@ +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire' +import { JournalQueuedMessages } from '../agent-session-journal/journal-queued-messages' +import { readStructuredAgentSessionHistoryResult } from './structured-agent-session-history-result' +import { + readQueuePublication, + structuredQueueSendGate +} from './structured-agent-session-queued-publication' +import { + createQueuedMessageTestRig, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { HOST_TEST_NOW, HOST_TEST_SESSION } from './structured-agent-session-host-test-data' + +let rig: QueuedMessageTestRig +beforeEach(async () => { + rig = await createQueuedMessageTestRig() +}) +afterEach(async () => { + vi.restoreAllMocks() + await rig.dispose() +}) + +it('keeps the complete legacy history reply byte-identical and replaces only its queue list after opt-in', async () => { + await rig.workingSend() + await rig.send('exact draft "\\🦦', 'queue-if-active').result + const request = { sessionId: HOST_TEST_SESSION, direction: 'tail' as const } + const session = rig.host.collaboratorsForTests().sessions.get(HOST_TEST_SESSION) + if (!session) { + throw new Error('missing conversation') + } + const journal = session.journal + const base = readStructuredAgentSessionHistoryResult({ + journal, + record: rig.store.getRecord(HOST_TEST_SESSION), + request + }) + const queue = readQueuePublication(journal, structuredQueueSendGate(rig.store, HOST_TEST_SESSION)) + const expected = { + ...base, + page: { + ...base.page, + hostNow: HOST_TEST_NOW, + queuedMessages: queue.queuedMessages, + queuePause: queue.queuePause, + nextQueuedMessageId: queue.nextQueuedMessageId, + backgroundTasks: null + } + } + expect(JSON.stringify(await rig.host.history(request))).toBe(JSON.stringify(expected)) + const list = vi.spyOn(JournalQueuedMessages.prototype, 'list') + const paged = await rig.host.history({ ...request, queueView: 'paged-v1' }) + expect(paged.page).not.toHaveProperty('queuedMessages') + expect(paged.page.queueSummary).toMatchObject({ + total: 1, + counts: { person: 1, agent: 0, unknown: 0 } + }) + expect(paged.page.queuePause).toEqual(queue.queuePause) + expect(paged.page.nextQueuedMessageId).toEqual(queue.nextQueuedMessageId) + expect(list).not.toHaveBeenCalled() +}) + +it('publishes host-derived Resume availability and off-page blockers on commit notifications', async () => { + await rig.workingSend() + await rig.send('queued', 'queue-if-active').result + const frames: AgentSessionSubscribeEvent[] = [] + await rig.host.subscribe({ + id: 'paged', + sessionId: HOST_TEST_SESSION, + queueView: 'paged-v1', + emit: (event) => frames.push(event) + }) + await rig.stop() + const history = await rig.host.history({ + sessionId: HOST_TEST_SESSION, + direction: 'tail', + queueView: 'paged-v1' + }) + expect(history.page.queueSummary?.resumeAvailable).toBe(history.page.queuePause !== null) + expect(history.page.queueSummary?.resumeAvailable).toBe(true) + expect(history.page.queueSummary?.total).toBe(1) + for (const event of frames) { + expect(event).not.toHaveProperty('queuedMessages') + if (event.type === 'snapshot' || event.type === 'reset') { + expect(event.queueSummary).toBeDefined() + } + } + const afterStop = history.page.queueSummary?.revision ?? 0 + await rig.deleteQueued(history.page.queueSummary?.newestPersonMessageId ?? '') + const afterDelete = await rig.host.history({ + sessionId: HOST_TEST_SESSION, + direction: 'tail', + queueView: 'paged-v1' + }) + expect(afterDelete.page.queueSummary).toMatchObject({ total: 0, resumeAvailable: false }) + expect(afterDelete.page.queueSummary?.revision).toBeGreaterThan(afterStop) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queue-page.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-page.test.ts new file mode 100644 index 000000000000..27a4bb2d5101 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-page.test.ts @@ -0,0 +1,310 @@ +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { + AGENT_SESSION_QUEUE_REPLY_MAX_BYTES, + AGENT_SESSION_QUEUE_SUMMARY_MAX_BYTES, + type AgentSessionQueuedMessagesPageResult +} from '../../../shared/agent-session-queue-pages' +import { + QUEUED_MESSAGE_SOURCE_SQL, + QUEUED_MESSAGE_UNSETTLED_SQL +} from '../agent-session-journal/queued-message-source-index' +import { JournalQueuedMessages } from '../agent-session-journal/journal-queued-messages' +import { + PAGE_SESSION, + pageMessage, + queuePageRig, + type QueuePageRig +} from './agent-session-queue-pages.test-fixture' + +let rig: QueuePageRig +beforeEach(async () => { + rig = await queuePageRig() +}) +afterEach(async () => { + vi.restoreAllMocks() + await rig.close() +}) + +function page(result: AgentSessionQueuedMessagesPageResult) { + expect(Buffer.byteLength(JSON.stringify(rig.wrapped(result)))).toBeLessThanOrEqual( + AGENT_SESSION_QUEUE_REPLY_MAX_BYTES + ) + if (result.status !== 'page') { + throw new Error('unexpected queue reset') + } + return result +} + +it('walks every tiny card in both directions, including tied positions, with a host size cap', () => { + rig.seed(1001) + const first = page(rig.page({ size: 100000 })) + expect(first.rows).toHaveLength(200) + let current = first + const forward = first.rows.map((row) => row.messageId) + while (current.nextCursor) { + current = page(rig.page({ cursor: current.nextCursor, size: 137 })) + forward.push(...current.rows.map((row) => row.messageId)) + } + const backwards = current.rows.map((row) => row.messageId) + while (current.previousCursor) { + current = page(rig.page({ cursor: current.previousCursor, size: 113 })) + backwards.unshift(...current.rows.map((row) => row.messageId)) + } + expect(new Set(forward).size).toBe(1001) + expect(backwards).toEqual(forward) +}) + +it('counts 10k rows and finds the newest person and returned blocker using indexes without hydration', () => { + rig.seed(10000) + rig.db + .prepare("UPDATE queued_messages SET state = 'returned' WHERE message_id = 'card-00003'") + .run() + const list = vi.spyOn(JournalQueuedMessages.prototype, 'list') + const parse = vi.spyOn(JSON, 'parse') + const prepare = vi.spyOn(rig.db, 'prepare') + const summary = rig.summary() + expect(summary).toMatchObject({ + total: 10000, + counts: { person: 10000, agent: 0, unknown: 0 }, + newestPersonMessageId: 'card-10000', + blockingReturnedMessageId: 'card-00003', + resumeAvailable: false + }) + expect(Buffer.byteLength(JSON.stringify(summary))).toBeLessThanOrEqual( + AGENT_SESSION_QUEUE_SUMMARY_MAX_BYTES + ) + expect(parse).not.toHaveBeenCalled() + expect(list).not.toHaveBeenCalled() + for (const [sql] of prepare.mock.calls) { + expect(sql.split('FROM')[0]).not.toMatch(/SELECT .*body_json|returned_rejection/) + } + const plan = rig.db + .prepare(`EXPLAIN QUERY PLAN 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`) + .all(PAGE_SESSION) + expect(plan.map((row) => row.detail).join(' ')).toContain( + 'queued_messages_unsettled_source_created' + ) + const countPlan = rig.db + .prepare(`EXPLAIN QUERY PLAN SELECT ${QUEUED_MESSAGE_SOURCE_SQL}, COUNT(*) + FROM queued_messages WHERE session_id = ? AND ${QUEUED_MESSAGE_UNSETTLED_SQL} + GROUP BY ${QUEUED_MESSAGE_SOURCE_SQL}`) + .all(PAGE_SESSION) + expect(countPlan.map((row) => row.detail).join(' ')).toContain( + 'queued_messages_unsettled_source_position' + ) +}) + +it('never counts malformed or unknown provenance as a person and anchors independently of agent backlog', async () => { + rig.seed(300, { + ...pageMessage('agent'), + from: { kind: 'agent', senders: [], orchestration: null } + }) + await rig.insert('person-newest') + const unknown = ['null', '7', '{}', '{"kind":7}', '{"kind":"future"}'] + for (const [index, from] of unknown.entries()) { + await rig.insert(`unknown-${index}`) + rig.db + .prepare('UPDATE queued_messages SET body_json = ? WHERE message_id = ?') + .run(`{"kind":"message","role":"user","blocks":[],"from":${from}}`, `unknown-${index}`) + } + rig.journal.queuedMessages.invalidate() + expect(rig.summary()).toMatchObject({ + counts: { person: 1, agent: 301, unknown: 4 }, + newestPersonMessageId: 'person-newest' + }) + expect( + page(rig.page({ source: 'person', aroundMessageId: 'person-newest' })).rows.map( + (row) => row.messageId + ) + ).toEqual(['person-newest']) + expect(page(rig.page({ aroundMessageId: 'person-newest', size: 1 })).rows[0]?.messageId).toBe( + 'person-newest' + ) +}) + +it('returns a reset for a vanished, moved, foreign-filter or foreign-generation anchor', async () => { + rig.seed(4) + const first = page(rig.page({ size: 1 })) + const cursor = first.nextCursor ?? '' + expect(rig.page({ cursor, source: 'person' })).toMatchObject({ + status: 'stale', + generation: first.generation + }) + rig.db + .prepare('UPDATE queued_messages SET position = 20 WHERE message_id = ?') + .run(first.rows[0]?.messageId ?? '') + expect(rig.page({ cursor })).toMatchObject({ status: 'stale' }) + rig.db + .prepare('UPDATE queued_messages SET position = 0 WHERE message_id = ?') + .run(first.rows[0]?.messageId ?? '') + await rig.journal.queuedMessages.withdraw({ + messageIds: [first.rows[0]?.messageId ?? ''], + settledByOp: 'op' + }) + expect(rig.page({ cursor })).toMatchObject({ status: 'stale' }) + expect(rig.page({ aroundMessageId: 'vanished' })).toMatchObject({ status: 'stale' }) + const before = page(rig.page({ size: 1 })) + await rig.journal.rollEpoch('handle_forked', 1) + const reset = rig.page({ cursor: before.nextCursor ?? '' }) + expect(reset.status).toBe('stale') + expect(reset.generation).not.toBe(before.generation) +}) + +it('bounds escaped previews, sender labels and failure metadata in the actual envelope, and always advances', async () => { + const text = '\u0000"\\🦦漢'.repeat(24000) + const body = { + ...pageMessage(text), + from: { + kind: 'agent' as const, + senders: Array.from({ length: 80 }, () => ({ + name: 'name'.repeat(1000), + party: { address: 'address'.repeat(1000), terminalHandle: null, orcaSessionId: null } + })), + orchestration: { + message: 'mail-notice' as const, + mailbox: 'mail', + dispatchId: null, + messages: Array.from({ length: 1000 }, () => ({ + messageId: 'source', + runId: 'run', + from: 'sender' + })) + } + } + } + rig.seed(150, body) + rig.db + .prepare( + "UPDATE queued_messages SET state = 'returned', returned_reason = ?, returned_rejection = ?" + ) + .run( + '\u0000'.repeat(100000), + JSON.stringify({ + kind: '\u0000'.repeat(1000), + detail: { text, audience: 'log' }, + taskSpec: text + }) + ) + const envelope = { id: '\u0000'.repeat(10000), runtimeId: 'runtime' } + let current = rig.page({ size: 200 }, envelope) + const ids = new Set() + for (;;) { + expect( + Buffer.byteLength( + JSON.stringify({ + id: envelope.id, + ok: true, + result: current, + _meta: { runtimeId: envelope.runtimeId } + }) + ) + ).toBeLessThanOrEqual(AGENT_SESSION_QUEUE_REPLY_MAX_BYTES) + if (current.status !== 'page') { + throw new Error('unexpected reset') + } + expect(current.rows.length).toBeGreaterThan(0) + for (const row of current.rows) { + ids.add(row.messageId) + expect(Buffer.byteLength(row.preview)).toBeLessThanOrEqual(512) + expect(row.truncated).toBe(true) + expect(row.senderLabels).toHaveLength(3) + expect(row.senderCount).toBe(80) + expect(JSON.stringify(row)).not.toContain('taskSpec') + expect(JSON.stringify(row)).not.toContain('mailbox') + } + if (!current.nextCursor) { + break + } + current = rig.page({ size: 200, cursor: current.nextCursor }, envelope) + } + expect(ids.size).toBe(150) +}) + +it.each([`${'a'.repeat(511)}🦦`, '漢'.repeat(171), 'é'.repeat(257), '"\\\n'.repeat(200)])( + 'cuts preview on a UTF-8 code point boundary', + async (text) => { + await rig.insert('unicode', pageMessage(text)) + const row = page(rig.page()).rows[0] + expect(Buffer.byteLength(row?.preview ?? '')).toBeLessThanOrEqual(512) + expect(text.startsWith(row?.preview ?? '')).toBe(true) + expect(row?.preview.endsWith('\ud83e')).toBe(false) + } +) + +it('uses one short read transaction per page and releases it on a reset', () => { + rig.seed(3) + const exec = vi.spyOn(rig.db, 'exec') + page(rig.page()) + expect(exec.mock.calls.map(([sql]) => sql)).toEqual(['BEGIN', 'COMMIT']) + exec.mockClear() + expect(rig.page({ cursor: 'invalid' }).status).toBe('stale') + expect(exec.mock.calls.map(([sql]) => sql)).toEqual(['BEGIN', 'COMMIT']) + expect(rig.db.isTransaction).toBe(false) +}) + +it('keeps the newest person shortcut tied to creation when queue order changes', async () => { + await rig.insert('older') + await rig.insert('newer') + rig.db + .prepare("UPDATE queued_messages SET created_at = 2, position = -1 WHERE message_id = 'newer'") + .run() + rig.db.prepare("UPDATE queued_messages SET created_at = 1 WHERE message_id = 'older'").run() + rig.journal.queuedMessages.invalidate() + expect(rig.summary().newestPersonMessageId).toBe('newer') +}) + +it('resets cursors after the same saved journal is reopened with a new handle', async () => { + rig.seed(3) + const first = page(rig.page({ size: 1 })) + const epoch = rig.journal.epoch + await rig.reopen() + expect(rig.journal.epoch).toBe(epoch) + expect(rig.page({ cursor: first.nextCursor ?? '' })).toMatchObject({ status: 'stale' }) + expect(rig.summary().generation).not.toBe(first.generation) + expect(rig.summary().total).toBe(3) +}) + +it('walks long escaped ids both ways when the byte budget limits page size', () => { + rig.seed(220, pageMessage('\u0000'.repeat(1000))) + rig.db + .prepare('UPDATE queued_messages SET message_id = message_id || ?') + .run('\u0000'.repeat(490)) + let current = page(rig.page({ size: 200 })) + expect(current.rows.length).toBeLessThan(200) + const forward = current.rows.map((row) => row.messageId) + while (current.nextCursor) { + current = page(rig.page({ cursor: current.nextCursor, size: 200 })) + forward.push(...current.rows.map((row) => row.messageId)) + } + const backward = current.rows.map((row) => row.messageId) + while (current.previousCursor) { + current = page(rig.page({ cursor: current.previousCursor, size: 200 })) + backward.unshift(...current.rows.map((row) => row.messageId)) + } + expect(new Set(forward).size).toBe(220) + expect(backward).toEqual(forward) +}) + +it('shrinks optional row metadata before returning an eligible first row', () => { + rig.seed(2, pageMessage('\u0000'.repeat(1000))) + const envelope = { id: 'x'.repeat(510 * 1024), runtimeId: 'host' } + const result = rig.page({ size: 1 }, envelope) + if (result.status !== 'page') { + throw new Error('unexpected reset') + } + expect(result.rows).toHaveLength(1) + expect(result.rows[0]).toMatchObject({ preview: '', truncated: true }) + expect(result.nextCursor).not.toBeNull() + expect( + Buffer.byteLength( + JSON.stringify({ + id: envelope.id, + ok: true, + result, + _meta: { runtimeId: envelope.runtimeId } + }) + ) + ).toBeLessThanOrEqual(AGENT_SESSION_QUEUE_REPLY_MAX_BYTES) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queue-page.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-page.ts new file mode 100644 index 000000000000..84073b813050 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-page.ts @@ -0,0 +1,125 @@ +import { + AGENT_SESSION_QUEUE_PAGE_DEFAULT_SIZE, + AGENT_SESSION_QUEUE_PAGE_MAX_SIZE, + type AgentSessionQueuedMessagesPageRequest, + type AgentSessionQueuedMessagesPageResult, + type AgentSessionQueuedMessagePreview +} from '../../../shared/agent-session-queue-pages' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import type { QueuePageRange } from '../agent-session-journal/journal-queued-message-pages' +import { + boundedQueueReply, + LOCAL_QUEUE_REPLY_ENVELOPE, + queueReplyFits, + type QueueReplyEnvelope +} from './agent-session-queue-reply-budget' +import { queueCursor, readQueuePageCursor } from './agent-session-queue-page-cursor' +import { queuedMessagePreview } from './agent-session-queue-preview' +import { readQueueSummary } from './structured-agent-session-queue-summary' +import type { QueueSendGate } from './structured-agent-session-queued-publication' + +export function readQueuedMessagesPage( + journal: AgentSessionJournal, + gate: QueueSendGate, + request: AgentSessionQueuedMessagesPageRequest, + envelope: QueueReplyEnvelope = LOCAL_QUEUE_REPLY_ENVELOPE +): AgentSessionQueuedMessagesPageResult { + return journal.queuedMessages.pages.read(() => { + const store = journal.queuedMessages.pages + const { generation, revision } = readQueueSummary(journal, gate).queueSummary + const reset = () => + boundedQueueReply({ status: 'stale' as const, generation, revision }, envelope) + const size = Math.max( + 1, + Math.min( + AGENT_SESSION_QUEUE_PAGE_MAX_SIZE, + request.size ?? AGENT_SESSION_QUEUE_PAGE_DEFAULT_SIZE + ) + ) + let range: QueuePageRange = { source: request.source, direction: 'after' } + if (request.cursor) { + const cursor = readQueuePageCursor(request.cursor) + if ( + !cursor || + cursor.generation !== generation || + cursor.sessionId !== request.sessionId || + cursor.source !== (request.source ?? null) + ) { + return reset() + } + const anchor = store.anchor(cursor.messageId, request.source) + if (!anchor || anchor.position !== cursor.position) { + return reset() + } + range = { source: request.source, direction: cursor.direction, anchor } + } else if (request.aroundMessageId) { + const anchor = store.anchor(request.aroundMessageId, request.source) + if (!anchor) { + return reset() + } + // Why: starting at the requested card guarantees it fits before any preceding backlog. + range = { source: request.source, direction: 'after', anchor, inclusive: true } + } + // Why: two cursors and false flags conservatively budget each row; probe the edges only once. + const reply = ( + rows: AgentSessionQueuedMessagePreview[], + probe = false + ): AgentSessionQueuedMessagesPageResult => { + const first = rows[0] + const last = rows.at(-1) + const before = + probe && first + ? store.hasRows({ source: request.source, direction: 'before', anchor: first }) + : false + const after = + probe && last + ? store.hasRows({ source: request.source, direction: 'after', anchor: last }) + : false + const cursor = (row: AgentSessionQueuedMessagePreview, direction: 'before' | 'after') => + queueCursor({ + sessionId: request.sessionId, + generation, + source: request.source ?? null, + position: row.position, + messageId: row.messageId, + direction + }) + return { + status: 'page', + generation, + revision, + rows, + hasMore: { before, after }, + previousCursor: (!probe || before) && first ? cursor(first, 'before') : null, + nextCursor: (!probe || after) && last ? cursor(last, 'after') : null + } + } + let rows: AgentSessionQueuedMessagePreview[] = [] + for (const stored of store.rows(range, size)) { + const preview = queuedMessagePreview(stored) + if (!preview) { + continue + } + let row = preview + const append = () => (range.direction === 'before' ? [row, ...rows] : [...rows, row]) + let candidate = append() + if (!queueReplyFits(reply(candidate), envelope)) { + if (rows.length > 0) { + break + } + row = { + ...row, + senderLabels: [], + preview: '', + truncated: row.truncated || row.preview.length > 0, + returnedReason: null, + returnedFailure: undefined + } + candidate = append() + } + boundedQueueReply(reply(candidate), envelope) + rows = candidate + } + return boundedQueueReply(reply(rows, true), envelope) + }) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queue-subscription.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-subscription.test.ts new file mode 100644 index 000000000000..7cd1c1cf2bb8 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-subscription.test.ts @@ -0,0 +1,303 @@ +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire' +import { JournalQueuedMessages } from '../agent-session-journal/journal-queued-messages' +import { readQueuePublication } from './structured-agent-session-queued-publication' +import { readQueueSummary } from './structured-agent-session-queue-summary' +import { AgentSessionSubscribers } from './structured-agent-session-subscribers' +import { buildSubscriberFrame } from './agent-session-subscriber-frame-fields' +import { + PAGE_GATE, + PAGE_SESSION, + queuePageRig, + type QueuePageRig +} from './agent-session-queue-pages.test-fixture' + +let rig: QueuePageRig +beforeEach(async () => { + rig = await queuePageRig() +}) +afterEach(async () => { + vi.restoreAllMocks() + await rig.close() +}) + +function subscribers() { + return new AgentSessionSubscribers({ + readQueuePublication: (_id, view) => + view === 'paged-v1' + ? readQueueSummary(rig.journal, PAGE_GATE) + : readQueuePublication(rig.journal, PAGE_GATE) + }) +} + +function summary(event: AgentSessionSubscribeEvent | undefined) { + if (!event || event.type === 'end') { + throw new Error('expected a queue frame') + } + expect(event).not.toHaveProperty('queuedMessages') + if (event.type === 'snapshot' || event.type === 'reset') { + expect(event.page).not.toHaveProperty('queuedMessages') + } + return event.queueSummary +} + +it('pins byte-identical legacy frames and uses summary-only snapshots, resets and changed batches for opt-in', async () => { + await rig.insert('first') + const registry = subscribers() + const legacy: AgentSessionSubscribeEvent[] = [] + const paged: AgentSessionSubscribeEvent[] = [] + registry.open({ + id: 'legacy', + sessionId: PAGE_SESSION, + journal: rig.journal, + fence: 1, + emit: (event) => legacy.push(event) + }) + registry.open({ + id: 'paged', + sessionId: PAGE_SESSION, + journal: rig.journal, + fence: 1, + queueView: 'paged-v1', + emit: (event) => paged.push(event) + }) + expect(summary(paged[0])?.total).toBe(1) + const publication = readQueuePublication(rig.journal, PAGE_GATE) + const event = { + type: 'batch' as const, + sessionId: PAGE_SESSION, + batch: { cursor: rig.journal.cursor(), items: [], removedItemIds: [], submissions: [] } + } + const expected = { + ...event, + queuedMessages: publication.queuedMessages, + queuePause: publication.queuePause, + nextQueuedMessageId: publication.nextQueuedMessageId + } + const actual = buildSubscriberFrame( + { readQueuePublication: () => publication }, + { sessionId: PAGE_SESSION }, + event, + false + ).frame + expect(JSON.stringify(actual)).toBe(JSON.stringify(expected)) + const initialLegacy = legacy[0] + if (!initialLegacy || initialLegacy.type !== 'snapshot') { + throw new Error('missing snapshot') + } + const { + queuedMessages: _list, + queuePause: _pause, + nextQueuedMessageId: _next, + ...base + } = initialLegacy + expect(JSON.stringify(initialLegacy)).toBe( + JSON.stringify({ + ...base, + queuedMessages: publication.queuedMessages, + queuePause: publication.queuePause, + nextQueuedMessageId: publication.nextQueuedMessageId + }) + ) + const list = vi.spyOn(JournalQueuedMessages.prototype, 'list') + // Disconnect the legacy reader before asserting the paged path does not hydrate. + registry.close(PAGE_SESSION, 'legacy') + rig.journal.observeCommits(() => registry.publish(PAGE_SESSION, rig.journal)) + await rig.insert('second') + expect(summary(paged.at(-1))?.total).toBe(2) + const count = paged.length + registry.publish(PAGE_SESSION, rig.journal) + expect(paged).toHaveLength(count) + registry.open({ + id: 'reset', + sessionId: PAGE_SESSION, + journal: rig.journal, + fence: 1, + queueView: 'paged-v1', + cursor: { epoch: 'retired', sequence: 0 }, + emit: (frame) => { + expect(frame.type).toBe('reset') + expect(summary(frame)?.total).toBe(2) + } + }) + expect(list).not.toHaveBeenCalled() +}) + +it('registers before the initial queue read and observes an insert issued during attach', async () => { + const frames: AgentSessionSubscribeEvent[] = [] + let registry: AgentSessionSubscribers + let insert: Promise | undefined + registry = new AgentSessionSubscribers({ + readQueuePublication: () => { + expect(registry.subscriberCountForTests(PAGE_SESSION)).toBe(1) + insert ??= rig.insert('concurrent') + return readQueueSummary(rig.journal, PAGE_GATE) + } + }) + rig.journal.observeCommits(() => registry.publish(PAGE_SESSION, rig.journal)) + registry.open({ + id: 'paged', + sessionId: PAGE_SESSION, + journal: rig.journal, + fence: 1, + queueView: 'paged-v1', + emit: (event) => frames.push(event) + }) + await insert + expect(summary(frames.at(-1))?.total).toBe(1) +}) + +it('coalesces mutations while output is blocked, keeps the newest revision, and owes nothing after detach', async () => { + const frames: AgentSessionSubscribeEvent[] = [] + let ready: (() => void) | undefined + const blocked = new Promise((resolve) => { + ready = resolve + }) + const registry = subscribers() + rig.journal.observeCommits(() => registry.publish(PAGE_SESSION, rig.journal)) + registry.open({ + id: 'paged', + sessionId: PAGE_SESSION, + journal: rig.journal, + fence: 1, + queueView: 'paged-v1', + emit: (event) => { + frames.push(event) + if (frames.length === 1) { + return blocked + } + } + }) + const revision = summary(frames[0])?.revision ?? 0 + await rig.insert('one') + await rig.insert('two') + await rig.insert('three') + expect(frames).toHaveLength(1) + ready?.() + await blocked + await Promise.resolve() + expect(frames).toHaveLength(2) + expect(summary(frames[1])?.total).toBe(3) + expect(summary(frames[1])?.revision).toBeGreaterThan(revision) + registry.close(PAGE_SESSION, 'paged') + await rig.insert('four') + expect(frames).toHaveLength(3) + expect(frames[2]).toEqual({ type: 'end' }) +}) + +it('coalesces multi-page journal catch-up without advancing past an unsent batch', async () => { + for (let index = 0; index < 210; index++) { + await rig.journal.appendItem( + { provider: 'claude', sessionId: 'native', uuid: `row-${index}` }, + { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'row' }] }, + { fence: 1 } + ) + } + const frames: AgentSessionSubscribeEvent[] = [] + let ready: (() => void) | undefined + const blocked = new Promise((resolve) => { + ready = resolve + }) + subscribers().open({ + id: 'paged', + sessionId: PAGE_SESSION, + journal: rig.journal, + fence: 1, + queueView: 'paged-v1', + cursor: { epoch: rig.journal.epoch, sequence: 0 }, + emit: (event) => { + frames.push(event) + if (frames.length === 1) { + return blocked + } + } + }) + expect(frames).toHaveLength(1) + ready?.() + await blocked + await Promise.resolve() + const ids = frames.flatMap((frame) => + frame.type === 'batch' ? frame.batch.items.map((item) => item.itemId) : [] + ) + expect(new Set(ids).size).toBe(210) + expect(summary(frames.at(-1))?.total).toBe(0) +}) + +it('replays a coalesced snapshot only to the subscriber whose output was blocked', async () => { + const blockedFrames: AgentSessionSubscribeEvent[] = [] + const readyFrames: AgentSessionSubscribeEvent[] = [] + let ready: (() => void) | undefined + const blocked = new Promise((resolve) => { + ready = resolve + }) + const registry = subscribers() + registry.open({ + id: 'blocked', + sessionId: PAGE_SESSION, + journal: rig.journal, + fence: 1, + queueView: 'paged-v1', + emit: (frame) => { + blockedFrames.push(frame) + if (blockedFrames.length === 1) { + return blocked + } + } + }) + registry.open({ + id: 'ready', + sessionId: PAGE_SESSION, + journal: rig.journal, + fence: 1, + queueView: 'paged-v1', + emit: (frame) => { + readyFrames.push(frame) + } + }) + await rig.insert('newest') + registry.snapshot(PAGE_SESSION, rig.journal, 2) + expect(blockedFrames).toHaveLength(1) + expect(readyFrames).toHaveLength(2) + ready?.() + await blocked + await Promise.resolve() + expect(blockedFrames).toHaveLength(2) + expect(blockedFrames[1]).toMatchObject({ type: 'snapshot', fence: 2 }) + expect(summary(blockedFrames[1])?.total).toBe(1) + expect(readyFrames).toHaveLength(2) +}) + +it('re-derives a blocked reader from the execution host after journal eviction', async () => { + const frames: AgentSessionSubscribeEvent[] = [] + let ready: (() => void) | undefined + const blocked = new Promise((resolve) => { + ready = resolve + }) + const readJournal = vi.fn(async () => rig.journal) + const registry = new AgentSessionSubscribers({ + readJournal, + readQueuePublication: () => readQueueSummary(rig.journal, PAGE_GATE) + }) + registry.open({ + id: 'paged', + sessionId: PAGE_SESSION, + journal: rig.journal, + fence: 1, + queueView: 'paged-v1', + emit: (frame) => { + frames.push(frame) + if (frames.length === 1) { + return blocked + } + } + }) + const generation = summary(frames[0])?.generation + await rig.insert('new') + registry.publish(PAGE_SESSION, rig.journal) + await rig.reopen() + ready?.() + await vi.waitFor(() => expect(frames).toHaveLength(2)) + expect(readJournal).toHaveBeenCalledWith(PAGE_SESSION) + expect(summary(frames[1])?.total).toBe(1) + expect(summary(frames[1])?.generation).not.toBe(generation) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queue-summary.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-summary.ts new file mode 100644 index 000000000000..560be8cfb2cd --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-summary.ts @@ -0,0 +1,82 @@ +import { + AGENT_SESSION_QUEUE_SUMMARY_MAX_BYTES, + type AgentSessionQueueSummary +} from '../../../shared/agent-session-queue-pages' +import { stringifyJsonWithinByteLimit } from '../../../shared/node-bounded-json-stringify' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { + readQueuePublicationFacts, + sameQueuePause, + type QueueSendGate, + type QueuePublication +} from './structured-agent-session-queued-publication' + +export type PagedQueuePublication = Pick & { + queueSummary: AgentSessionQueueSummary +} +export type ReaderQueuePublication = QueuePublication | PagedQueuePublication +type SummaryMemo = { repositoryRevision: number; publication: PagedQueuePublication } +const summaries = new WeakMap() + +export function readQueueSummary( + journal: AgentSessionJournal, + gate: QueueSendGate +): PagedQueuePublication { + const store = journal.queuedMessages.pages + const generation = store.generation(journal.epoch) + const repositoryRevision = journal.queuedMessages.revision() + const previous = summaries.get(journal) + const facts = readQueuePublicationFacts(journal, gate) + const unchangedRows = + previous?.repositoryRevision === repositoryRevision && + previous.publication.queueSummary.generation === generation + if ( + unchangedRows && + sameQueuePause(previous.publication.queuePause, facts.queuePause) && + previous.publication.nextQueuedMessageId === facts.nextQueuedMessageId + ) { + return previous.publication + } + const counts = unchangedRows ? previous.publication.queueSummary.counts : store.counts() + const queueSummary: AgentSessionQueueSummary = { + generation, + revision: + previous && previous.publication.queueSummary.generation === generation + ? Math.max(repositoryRevision, previous.publication.queueSummary.revision + 1) + : repositoryRevision, + total: counts.person + counts.agent + counts.unknown, + counts, + newestPersonMessageId: unchangedRows + ? previous.publication.queueSummary.newestPersonMessageId + : store.newestPersonMessageId(), + blockingReturnedMessageId: unchangedRows + ? previous.publication.queueSummary.blockingReturnedMessageId + : store.blockingReturnedMessageId(), + resumeAvailable: facts.queuePause !== null + } + stringifyJsonWithinByteLimit(queueSummary, AGENT_SESSION_QUEUE_SUMMARY_MAX_BYTES) + const publication = { queueSummary, ...facts } + summaries.set(journal, { repositoryRevision, publication }) + return publication +} + +export function tryReadQueueSummary( + journal: AgentSessionJournal | undefined, + gate: QueueSendGate +): PagedQueuePublication | undefined { + try { + return journal ? readQueueSummary(journal, gate) : undefined + } catch { + return undefined + } +} + +export function queuePublicationFields(publication: ReaderQueuePublication) { + return { + ...('queueSummary' in publication + ? { queueSummary: publication.queueSummary } + : { queuedMessages: publication.queuedMessages }), + queuePause: publication.queuePause, + nextQueuedMessageId: publication.nextQueuedMessageId + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts index d970940be949..77ed3f818010 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts @@ -112,6 +112,25 @@ export function readQueuePublication( gate: QueueSendGate ): QueuePublication { const queuedMessages = readPublishedQueuedMessages(journal) + const { queuePause, nextQueuedMessageId } = readQueuePublicationFacts(journal, gate) + const previous = publications.get(journal) + if ( + previous && + previous.queuedMessages === queuedMessages && + sameQueuePause(previous.queuePause, queuePause) && + previous.nextQueuedMessageId === nextQueuedMessageId + ) { + return previous + } + const publication = { queuedMessages, queuePause, nextQueuedMessageId } + publications.set(journal, publication) + return publication +} + +export function readQueuePublicationFacts( + journal: AgentSessionJournal, + gate: QueueSendGate +): Pick { // Read per emit: the pause also turns on submissions (a turn starting). Only a person's Stop is // shown, and only over a card Resume would send, so its header never offers to send nothing; // deleting a blocking returned card shows it again. After a /clear, a restart or a close nothing @@ -138,18 +157,7 @@ export function readQueuePublication( // The test narrows the type. const queuePause = pause?.reason === 'stopped' ? { reason: pause.reason } : null const nextQueuedMessageId = nextStructuredQueuedMessage({ journal, ...gate() })?.messageId ?? null - const previous = publications.get(journal) - if ( - previous && - previous.queuedMessages === queuedMessages && - sameQueuePause(previous.queuePause, queuePause) && - previous.nextQueuedMessageId === nextQueuedMessageId - ) { - return previous - } - const publication = { queuedMessages, queuePause, nextQueuedMessageId } - publications.set(journal, publication) - return publication + return { queuePause, nextQueuedMessageId } } /** For readers that must never fail on drafts — a subscriber stream, a history diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subscriber-journal-retention.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subscriber-journal-retention.test.ts index 8760e378525f..16940edf2b8f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subscriber-journal-retention.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subscriber-journal-retention.test.ts @@ -4,6 +4,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' import { STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY } from '../../../shared/protocol-version' import { MAX_TIMER_DELAY_MS } from '../../../shared/timer-delay' +import type { AgentSessionQueueView } from '../../../shared/agent-session-queue-pages' import { codexItemBody } from '../../codex/codex-structured-item-translation' import { OrcaRuntimeService } from '../../runtime/orca-runtime' import { RpcDispatcher } from '../../runtime/rpc/dispatcher' @@ -110,7 +111,10 @@ async function seedFold() { return { journal: new WeakRef(journal), digest: digest.digest('hex') } } -async function subscribeOverRpc(reader: ReturnType): Promise { +async function subscribeOverRpc( + reader: ReturnType, + queueView?: AgentSessionQueueView +): Promise { const controller = new AbortController() controllers.push(controller) const dispatcher = new RpcDispatcher({ @@ -122,7 +126,7 @@ async function subscribeOverRpc(reader: ReturnType): Promise { const frame: unknown = JSON.parse(raw) @@ -205,58 +209,61 @@ afterEach(async () => { }) describe('subscriber journal lifetime', () => { - it('releases an idle fold while its owned RPC reader stays live and resumes on the durable journal', async () => { - const seeded = await seedFold() - const reader = observer() - const controller = await subscribeOverRpc(reader) - expect(reader.state.sawSeed).toBe(true) - expect(subscriberCount()).toBe(1) - // Provider mock call history owns the event sink; remove that fixture root before GC. - vi.clearAllMocks() - rig.statusEvents.length = 0 - await collect() - expect(seeded.journal.deref() !== undefined).toBe(true) + it.each([undefined, 'paged-v1'] as const)( + 'releases an idle fold while its owned RPC reader (%s) stays live and resumes on the durable journal', + async (queueView) => { + const seeded = await seedFold() + const reader = observer() + const controller = await subscribeOverRpc(reader, queueView) + expect(reader.state.sawSeed).toBe(true) + expect(subscriberCount()).toBe(1) + // Provider mock call history owns the event sink; remove that fixture root before GC. + vi.clearAllMocks() + rig.statusEvents.length = 0 + await collect() + expect(seeded.journal.deref() !== undefined).toBe(true) - rig.clock.now += IDLE_MS + 1 - await sweepOnce(rig.host) - expect(rig.host.hasSession(SESSION)).toBe(false) - expect(rig.adapter.closeSession).toHaveBeenCalledWith(SESSION) - expect(subscriberCount()).toBe(1) - expect(reader.state.types).not.toContain('end') - vi.clearAllMocks() - rig.statusEvents.length = 0 - await collect() - const releasedWhileSubscribed = seeded.journal.deref() === undefined + rig.clock.now += IDLE_MS + 1 + await sweepOnce(rig.host) + expect(rig.host.hasSession(SESSION)).toBe(false) + expect(rig.adapter.closeSession).toHaveBeenCalledWith(SESSION) + expect(subscriberCount()).toBe(1) + expect(reader.state.types).not.toContain('end') + vi.clearAllMocks() + rig.statusEvents.length = 0 + await collect() + const releasedWhileSubscribed = seeded.journal.deref() === undefined - await restoreDurableHistory(seeded.digest) - expect(rig.adapter.acquire).not.toHaveBeenCalled() - const sibling = observer() - const disposeSibling = await rig.host.subscribe({ - id: 'new-reader', - sessionId: SESSION, - emit: sibling.emit - }) - expect(sibling.state.sawSeed).toBe(true) - expect(subscriberCount()).toBe(2) - await publishAfterIdle('first-publication') - await vi.waitFor(() => expect(reader.state.sawFuture && sibling.state.sawFuture).toBe(true)) + await restoreDurableHistory(seeded.digest) + expect(rig.adapter.acquire).not.toHaveBeenCalled() + const sibling = observer() + const disposeSibling = await rig.host.subscribe({ + id: 'new-reader', + sessionId: SESSION, + emit: sibling.emit + }) + expect(sibling.state.sawSeed).toBe(true) + expect(subscriberCount()).toBe(2) + await publishAfterIdle('first-publication') + await vi.waitFor(() => expect(reader.state.sawFuture && sibling.state.sawFuture).toBe(true)) - controller.abort() - expect(subscriberCount()).toBe(1) - expect(reader.state.types.filter((type) => type === 'end')).toHaveLength(1) - const endedFrames = reader.state.types.length - const siblingFrames = sibling.state.types.length - await publishAfterIdle('second-publication') - await vi.waitFor(() => expect(sibling.state.types.length).toBeGreaterThan(siblingFrames)) - expect(reader.state.types).toHaveLength(endedFrames) - disposeSibling() - disposeSibling() - expect(subscriberCount()).toBe(0) - expect(sibling.state.types.filter((type) => type === 'end')).toHaveLength(1) - await collect() - expect(seeded.journal.deref() === undefined).toBe(true) - expect(releasedWhileSubscribed).toBe(true) - }) + controller.abort() + expect(subscriberCount()).toBe(1) + expect(reader.state.types.filter((type) => type === 'end')).toHaveLength(1) + const endedFrames = reader.state.types.length + const siblingFrames = sibling.state.types.length + await publishAfterIdle('second-publication') + await vi.waitFor(() => expect(sibling.state.types.length).toBeGreaterThan(siblingFrames)) + expect(reader.state.types).toHaveLength(endedFrames) + disposeSibling() + disposeSibling() + expect(subscriberCount()).toBe(0) + expect(sibling.state.types.filter((type) => type === 'end')).toHaveLength(1) + await collect() + expect(seeded.journal.deref() === undefined).toBe(true) + expect(releasedWhileSubscribed).toBe(true) + } + ) it('keeps direct cleanup tied to its original subscriber after the caller changes its input', async () => { const seeded = await seedFold() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts index dcd3e07c4f2a..de44d72479bb 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts @@ -1,3 +1,5 @@ +import { emitSubscriberFrame, resumeSubscriberOutput } from './agent-session-subscriber-output' +import { sameJournalCursor } from '../agent-session-journal/journal-cursor' // Per-subscriber cursors over one session's journal. // // Each subscriber advances independently: a client that connected two epochs @@ -12,10 +14,10 @@ import type { } from '../../../shared/agent-session-wire' import { backgroundTaskFingerprint, - buildSubscriberFrame, type SubscriberFieldHooks } from './agent-session-subscriber-frame-fields' -import type { QueuePublication } from './structured-agent-session-queued-publication' +import type { ReaderQueuePublication } from './structured-agent-session-queue-summary' +import type { AgentSessionQueueView } from '../../../shared/agent-session-queue-pages' import { deliverToSubscriber } from './agent-session-subscriber-catch-up' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { emptyAgentSessionBatch } from './agent-session-empty-batch' @@ -29,9 +31,15 @@ export type AgentSessionSubscribeInput = { sessionId: string emit: AgentSessionSubscriberEmit cursor?: AgentJournalCursor + queueView?: AgentSessionQueueView } export type Subscriber = { + journal: WeakRef + outputBlocked?: true + pendingOutput?: true + pendingSnapshot?: number + queueView?: AgentSessionQueueView id: string sessionId: string emit: AgentSessionSubscriberEmit @@ -40,16 +48,20 @@ export type Subscriber = { commands?: AgentSessionSlashCommand[] | null /** The last draft list actually SENT — never advanced on a page that withheld * it, or the final replacement would be suppressed by the identity dedup. */ - queuePublication?: QueuePublication + queuePublication?: ReaderQueuePublication /** Fingerprint of the background-task roster last SENT. */ backgroundTasks?: string } export type AgentSessionSubscribersHooks = { + readJournal?: (sessionId: string) => Promise readCommands?: (sessionId: string) => AgentSessionSlashCommand[] | undefined /** Revision-stable per emit: an unchanged list keeps its reference, so token * streams never re-serialize it; any draft-table write changes it. */ - readQueuePublication?: (sessionId: string) => QueuePublication | undefined + readQueuePublication?: ( + sessionId: string, + queueView?: AgentSessionQueueView + ) => ReaderQueuePublication | undefined readBackgroundTasks?: SubscriberFieldHooks['readBackgroundTasks'] /** Fires after publications that can change journal content. */ onJournalPublished?: (sessionId: string, journal: AgentSessionJournal) => void @@ -77,14 +89,17 @@ export class AgentSessionSubscribers { fence: number emit: AgentSessionSubscriberEmit cursor?: AgentJournalCursor + queueView?: AgentSessionQueueView }): () => void { const liveCursor = input.journal.cursor() const subscriber: Subscriber = { id: input.id, + journal: new WeakRef(input.journal), sessionId: input.sessionId, emit: input.emit, cursor: input.cursor ?? { epoch: liveCursor.epoch, sequence: 0 }, - fence: input.fence + fence: input.fence, + ...(input.queueView ? { queueView: input.queueView } : {}) } const session = this.bySession.get(input.sessionId) ?? new Map() session.set(input.id, subscriber) @@ -114,8 +129,7 @@ export class AgentSessionSubscribers { }) subscriber.cursor = page.liveCursor ?? page.window.nextCursor } - const { sessionId, id } = subscriber - return () => this.close(sessionId, id) + return () => this.close(subscriber.sessionId, subscriber.id) } close(sessionId: string, id: string): void { @@ -158,6 +172,12 @@ export class AgentSessionSubscribers { const page = readAgentSessionHydrationPage(journal, fence) const hostNow = this.now() for (const subscriber of this.subscribers(sessionId)) { + subscriber.journal = new WeakRef(journal) + if (subscriber.outputBlocked) { + subscriber.pendingOutput = true + subscriber.pendingSnapshot = fence + continue + } this.emit(subscriber, { type: 'snapshot', sessionId, @@ -226,18 +246,26 @@ export class AgentSessionSubscribers { emitCheckpoint = false, activity?: AgentSessionTurnActivity | null ): void { + subscriber.journal = new WeakRef(journal) + if (subscriber.outputBlocked) { + subscriber.pendingOutput = true + return + } deliverToSubscriber( { hooks: this.hooks, emit: (target, event, options) => this.emit(target, event, options), - isActive: (target) => this.isActive(target), + isActive: (target) => this.isActive(target) && !target.outputBlocked, activity: (sessionId) => this.activityField(sessionId).activity }, { subscriber, journal, hostNow, emitCheckpoint, activity } ) + if (subscriber.outputBlocked && !sameJournalCursor(subscriber.cursor, journal.cursor())) { + subscriber.pendingOutput = true + } } - private now = (): number => this.hooks.now?.() ?? Date.now() + private now = () => this.hooks.now?.() ?? Date.now() private isActive = (subscriber: Subscriber): boolean => this.bySession.get(subscriber.sessionId)?.get(subscriber.id) === subscriber @@ -249,23 +277,46 @@ export class AgentSessionSubscribers { event: AgentSessionSubscribeEvent, options?: { withholdQueued?: boolean } ): void { - try { - const built = buildSubscriberFrame( - this.hooks, - subscriber, - event, - options?.withholdQueued === true - ) - subscriber.emit(built.frame) - subscriber.commands = built.commands - if (built.attachedQueued) { - subscriber.queuePublication = built.queued - } - if (built.backgroundTasks !== undefined) { - subscriber.backgroundTasks = built.backgroundTasks - } - } catch { - this.drop(subscriber) + emitSubscriberFrame({ + hooks: this.hooks, + subscriber, + event, + withholdQueued: options?.withholdQueued === true, + active: () => this.isActive(subscriber), + ready: () => + resumeSubscriberOutput({ + subscriber, + readJournal: this.hooks.readJournal, + active: () => this.isActive(subscriber), + ready: (journal) => this.resumeOutput(subscriber, journal), + drop: () => this.drop(subscriber) + }), + drop: () => this.drop(subscriber) + }) + } + + private resumeOutput(subscriber: Subscriber, journal: AgentSessionJournal): void { + if (!subscriber.pendingOutput) { + return + } + subscriber.pendingOutput = undefined + const fence = subscriber.pendingSnapshot + subscriber.pendingSnapshot = undefined + if (fence !== undefined) { + const page = readAgentSessionHydrationPage(journal, fence) + this.emit(subscriber, { + type: 'snapshot', + sessionId: subscriber.sessionId, + page, + fence, + hostNow: this.now(), + ...this.activityField(subscriber.sessionId) + }) + subscriber.cursor = page.liveCursor ?? page.window.nextCursor + subscriber.fence = fence + } else { + this.deliver(subscriber, journal, this.now(), true) + this.republishBackgroundTasks(subscriber.sessionId, subscriber.fence) } } diff --git a/src/main/runtime/mobile-rpc-allowlist.test.ts b/src/main/runtime/mobile-rpc-allowlist.test.ts index e761fd9e04c4..93542cc4abb8 100644 --- a/src/main/runtime/mobile-rpc-allowlist.test.ts +++ b/src/main/runtime/mobile-rpc-allowlist.test.ts @@ -157,6 +157,8 @@ describe('mobile RPC allowlist', () => { 'agentSession.queuedMessageSend', 'agentSession.queuedMessageDelete', 'agentSession.queuedMessagesResume', + 'agentSession.queuedMessagesPage', + 'agentSession.queuedMessageRead', 'agentSession.close', 'agentSession.respondToApproval', 'agentSession.respondToQuestion', diff --git a/src/main/runtime/rpc/methods/structured-agent-session-gate-classification.test-fixture.ts b/src/main/runtime/rpc/methods/structured-agent-session-gate-classification.test-fixture.ts index 12813a8f66ea..d504f549dd0f 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-gate-classification.test-fixture.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-gate-classification.test-fixture.ts @@ -65,6 +65,11 @@ export const WORK_METHODS = [ params: { envelope: envelope(), messageId: 'queued-1' } }, { method: 'agentSession.queuedMessagesResume', params: { envelope: envelope() } }, + { method: 'agentSession.queuedMessagesPage', params: { sessionId: SESSION } }, + { + method: 'agentSession.queuedMessageRead', + params: { sessionId: SESSION, messageId: 'queued-1' } + }, { method: 'agentSession.rewind', params: { envelope: envelope(), itemId: 'chosen', expectedEpoch: 'epoch' } diff --git a/src/main/runtime/rpc/methods/structured-agent-session-queue-envelope.test.ts b/src/main/runtime/rpc/methods/structured-agent-session-queue-envelope.test.ts new file mode 100644 index 000000000000..cf8751554ad3 --- /dev/null +++ b/src/main/runtime/rpc/methods/structured-agent-session-queue-envelope.test.ts @@ -0,0 +1,72 @@ +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { AGENT_SESSION_QUEUE_REPLY_MAX_BYTES } from '../../../../shared/agent-session-queue-pages' +import type { + AgentSessionQueuedMessagesPageRequest, + AgentSessionQueuedMessageReadRequest +} from '../../../../shared/agent-session-queue-pages' +import { readQueuedMessagesPage } from '../../../native-chat/agent-session-wire/structured-agent-session-queue-page' +import { readQueuedMessageBody } from '../../../native-chat/agent-session-wire/structured-agent-session-queue-body' +import type { QueueReplyEnvelope } from '../../../native-chat/agent-session-wire/agent-session-queue-reply-budget' +import { + PAGE_SESSION, + PAGE_GATE, + pageMessage, + queuePageRig, + type QueuePageRig +} from '../../../native-chat/agent-session-wire/agent-session-queue-pages.test-fixture' +import { + clearStructuredHostStub, + dispatcher, + hostCalls, + installStructuredHostStub, + STRUCTURED_CLIENT +} from './structured-agent-session-rpc.test-fixture' + +let rig: QueuePageRig +beforeEach(async () => { + rig = await queuePageRig() + installStructuredHostStub() + hostCalls.queuedMessagesPage = vi.fn( + (request: AgentSessionQueuedMessagesPageRequest, envelope: QueueReplyEnvelope) => + readQueuedMessagesPage(rig.journal, PAGE_GATE, request, envelope) + ) + hostCalls.queuedMessageRead = vi.fn( + (request: AgentSessionQueuedMessageReadRequest, envelope: QueueReplyEnvelope) => + readQueuedMessageBody(rig.journal, PAGE_GATE, request, envelope) + ) +}) +afterEach(async () => { + clearStructuredHostStub() + await rig.close() +}) + +it.each(['unary', 'streaming'] as const)( + 'measures the actual %s dispatcher envelope on local and paired/mobile paths', + async (transport) => { + const text = '\u0000"\\🦦'.repeat(70000) + rig.seed(240, pageMessage('preview'.repeat(100))) + await rig.insert('large', pageMessage(text)) + const rpc = dispatcher() + for (const [method, params] of [ + ['agentSession.queuedMessagesPage', { sessionId: PAGE_SESSION, size: 200 }], + ['agentSession.queuedMessageRead', { sessionId: PAGE_SESSION, messageId: 'large' }] + ] as const) { + const request = { id: '\u0000'.repeat(10000), authToken: 'test', method, params } + if (transport === 'unary') { + const reply = await rpc.dispatch(request) + expect(reply.ok).toBe(true) + expect(Buffer.byteLength(JSON.stringify(reply))).toBeLessThanOrEqual( + AGENT_SESSION_QUEUE_REPLY_MAX_BYTES + ) + } else { + const replies: string[] = [] + await rpc.dispatchStreaming(request, (reply) => replies.push(reply), STRUCTURED_CLIENT) + expect(replies).toHaveLength(1) + expect(JSON.parse(replies[0] ?? '').ok).toBe(true) + expect(Buffer.byteLength(replies[0] ?? '')).toBeLessThanOrEqual( + AGENT_SESSION_QUEUE_REPLY_MAX_BYTES + ) + } + } + } +) diff --git a/src/main/runtime/rpc/methods/structured-agent-session-queue-pages.test.ts b/src/main/runtime/rpc/methods/structured-agent-session-queue-pages.test.ts new file mode 100644 index 000000000000..42a9d5f3ea3b --- /dev/null +++ b/src/main/runtime/rpc/methods/structured-agent-session-queue-pages.test.ts @@ -0,0 +1,88 @@ +import { afterEach, beforeEach, expect, it } from 'vitest' +import { + AGENT_SESSION_QUEUE_PAGES_RUNTIME_CAPABILITY, + AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY, + RUNTIME_CAPABILITIES +} from '../../../../shared/protocol-version' +import { MOBILE_RPC_METHOD_ALLOWLIST } from '../../runtime-rpc/runtime-rpc-mobile-method-allowlist' +import { + call, + clearStructuredHostStub, + hostCalls, + installStructuredHostStub, + openStream, + SESSION, + STRUCTURED_CLIENT, + STRUCTURED_MOBILE_CLIENT +} from './structured-agent-session-rpc.test-fixture' + +beforeEach(installStructuredHostStub) +afterEach(clearStructuredHostStub) + +it('advertises queue reads independently of person-typed queueing', () => { + expect(RUNTIME_CAPABILITIES).toContain(AGENT_SESSION_QUEUE_PAGES_RUNTIME_CAPABILITY) + expect(RUNTIME_CAPABILITIES).not.toContain(AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY) +}) + +it.each([ + { + method: 'agentSession.queuedMessagesPage', + host: 'queuedMessagesPage', + params: { sessionId: SESSION, source: 'person', size: 900 } + }, + { + method: 'agentSession.queuedMessageRead', + host: 'queuedMessageRead', + params: { sessionId: SESSION, messageId: 'draft', expectedState: 'waiting' } + } +])( + 'registers and mobile-allowlists $method behind the structured gate', + async ({ method, host, params }) => { + expect(MOBILE_RPC_METHOD_ALLOWLIST.has(method)).toBe(true) + expect( + await call(method, params, { clientKind: 'runtime', clientCapabilities: [] }) + ).toMatchObject({ ok: false, error: { message: 'structured_agent_session_unsupported' } }) + expect(hostCalls[host]).not.toHaveBeenCalled() + expect(await call(method, params, STRUCTURED_MOBILE_CLIENT)).toMatchObject({ ok: true }) + expect(hostCalls[host]).toHaveBeenCalledWith(params, { + id: 'request-1', + runtimeId: 'runtime-1' + }) + } +) + +it.each([ + ['agentSession.queuedMessagesPage', { sessionId: SESSION, size: 0 }], + ['agentSession.queuedMessagesPage', { sessionId: SESSION, source: 'future' }], + [ + 'agentSession.queuedMessagesPage', + { sessionId: SESSION, cursor: 'cursor', aroundMessageId: 'anchor' } + ], + ['agentSession.queuedMessagesPage', { sessionId: SESSION, extra: true }], + [ + 'agentSession.queuedMessageRead', + { sessionId: SESSION, messageId: 'draft', expectedState: 'dispatched' } + ], + ['agentSession.queuedMessageRead', { sessionId: SESSION, messageId: 'draft', extra: true }], + ['agentSession.history', { sessionId: SESSION, direction: 'tail', queueView: 'future' }], + ['agentSession.subscribe', { sessionId: SESSION, queueView: 'future' }] +])('keeps %s params strict', async (method, params) => { + expect(await call(method, params, STRUCTURED_CLIENT)).toMatchObject({ + ok: false, + error: { code: 'invalid_argument' } + }) +}) + +it('passes the explicit queue view to history and subscription, including trusted local calls', async () => { + const params = { sessionId: SESSION, direction: 'tail', queueView: 'paged-v1' } + expect(await call('agentSession.history', params)).toMatchObject({ ok: true }) + expect(hostCalls.history).toHaveBeenCalledWith(params) + await openStream( + 'agentSession.subscribe', + { sessionId: SESSION, queueView: 'paged-v1' }, + STRUCTURED_CLIENT + ) + expect(hostCalls.subscribe).toHaveBeenCalledWith( + expect.objectContaining({ queueView: 'paged-v1' }) + ) +}) diff --git a/src/main/runtime/rpc/methods/structured-agent-session-queued-methods.ts b/src/main/runtime/rpc/methods/structured-agent-session-queued-methods.ts index 5689330e3c3b..799448032f4b 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-queued-methods.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-queued-methods.ts @@ -5,14 +5,37 @@ import { defineMethod } from '../core' import { requireStructuredSessionHost as requireSessionHost, - structuredCallerFor as callerFor + structuredCallerFor as callerFor, + requireInstalledStructuredHost } from './structured-agent-session-gate' import { QueuedMessageActionParams, - QueuedMessagesResumeParams + QueuedMessagesResumeParams, + QueuedMessagesPageParams, + QueuedMessageReadParams } from './structured-agent-session-schemas' export const STRUCTURED_AGENT_SESSION_QUEUED_METHODS = [ + defineMethod({ + name: 'agentSession.queuedMessagesPage', + permission: 'workspace', + params: QueuedMessagesPageParams, + handler: async (params, ctx) => + (await requireInstalledStructuredHost(ctx, params.sessionId)).queuedMessagesPage(params, { + id: ctx.requestId ?? '', + runtimeId: ctx.runtime.getRuntimeId() + }) + }), + defineMethod({ + name: 'agentSession.queuedMessageRead', + permission: 'workspace', + params: QueuedMessageReadParams, + handler: async (params, ctx) => + (await requireInstalledStructuredHost(ctx, params.sessionId)).queuedMessageRead(params, { + id: ctx.requestId ?? '', + runtimeId: ctx.runtime.getRuntimeId() + }) + }), defineMethod({ name: 'agentSession.queuedMessageSend', permission: 'workspace', diff --git a/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts b/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts index c4b3f62700b6..07fa15bea4cd 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts @@ -194,6 +194,8 @@ export function hostStub(): StructuredAgentSessionHost { models: [{ id: 'gpt-live', label: 'GPT Live', isDefault: true, efforts: [] }], current: { model: 'gpt-live' } })), + queuedMessagesPage: vi.fn(async () => ({ status: 'page', rows: [] })), + queuedMessageRead: vi.fn(async () => ({ status: 'gone' })), history: vi.fn(() => ({ ok: true, page: { items: [] } })), sessionAgent: vi.fn(() => null), journalSnapshot: vi.fn((sessionId: string) => ({ @@ -323,6 +325,4 @@ export function installStructuredHostStub(): void { setStructuredAgentSessionHost(hostStub()) } -export function clearStructuredHostStub(): void { - setStructuredAgentSessionHost(null) -} +export const clearStructuredHostStub = (): void => setStructuredAgentSessionHost(null) diff --git a/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts b/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts index 1f749d50490d..2c99720cd090 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts @@ -20,6 +20,8 @@ export { OptionsParams, QueuedMessageActionParams, QueuedMessagesResumeParams, + QueuedMessagesPageParams, + QueuedMessageReadParams, RespondParams, RespondToQuestionParams, RestartResumableParams, diff --git a/src/main/runtime/rpc/methods/structured-agent-session.test.ts b/src/main/runtime/rpc/methods/structured-agent-session.test.ts index 4bbd415cd025..5d6f6f3cd5f4 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.test.ts @@ -175,7 +175,7 @@ describe('capability gating', () => { } // Bump deliberately: the whole agentSession.* surface is behind the structured capability, // so an additive method is invisible to old clients and needs no protocol bump. - expect(STRUCTURED_AGENT_SESSION_METHODS).toHaveLength(34) + expect(STRUCTURED_AGENT_SESSION_METHODS).toHaveLength(36) }) it('hides the surface from a declared client that did not advertise it', async () => { diff --git a/src/main/runtime/rpc/methods/structured-agent-session.ts b/src/main/runtime/rpc/methods/structured-agent-session.ts index 7324ae7ffd2e..b4ef7f0e64e4 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.ts @@ -274,7 +274,8 @@ export const STRUCTURED_AGENT_SESSION_METHODS = [ host.sessionAgent(params.sessionId) ) ), - ...(params.cursor ? { cursor: params.cursor } : {}) + ...(params.cursor ? { cursor: params.cursor } : {}), + ...(params.queueView ? { queueView: params.queueView } : {}) }) if (stream.isClosed()) { dispose() diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-agent-session-methods.ts b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-agent-session-methods.ts index 56b30d837a5d..2cd235990cfa 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-agent-session-methods.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-agent-session-methods.ts @@ -9,6 +9,8 @@ export const MOBILE_AGENT_SESSION_RPC_METHODS = [ 'agentSession.queuedMessageSend', 'agentSession.queuedMessageDelete', 'agentSession.queuedMessagesResume', + 'agentSession.queuedMessagesPage', + 'agentSession.queuedMessageRead', 'agentSession.close', 'agentSession.respondToApproval', 'agentSession.respondToQuestion', diff --git a/src/shared/agent-session-queue-pages.ts b/src/shared/agent-session-queue-pages.ts new file mode 100644 index 000000000000..e53f1957fa70 --- /dev/null +++ b/src/shared/agent-session-queue-pages.ts @@ -0,0 +1,86 @@ +import type { AgentSessionQueuedMessagePausedReason } from './agent-session-queued-message-wire' + +export const AGENT_SESSION_QUEUE_REPLY_MAX_BYTES = 512 * 1024 +export const AGENT_SESSION_QUEUE_SUMMARY_MAX_BYTES = 8 * 1024 +export const AGENT_SESSION_QUEUE_PREVIEW_MAX_BYTES = 512 +export const AGENT_SESSION_QUEUE_PAGE_MAX_SIZE = 200 +export const AGENT_SESSION_QUEUE_PAGE_DEFAULT_SIZE = 40 +export const AGENT_SESSION_QUEUE_SOURCES = ['person', 'agent', 'unknown'] as const +export type AgentSessionQueueSource = (typeof AGENT_SESSION_QUEUE_SOURCES)[number] +export type AgentSessionQueueView = 'paged-v1' + +export type AgentSessionQueueSummary = { + generation: string + revision: number + total: number + counts: Record + newestPersonMessageId: string | null + blockingReturnedMessageId: string | null + resumeAvailable: boolean +} + +export type AgentSessionQueuedMessagePreview = { + messageId: string + position: number + state: 'waiting' | 'returned' + paused?: true + pausedReason?: AgentSessionQueuedMessagePausedReason + returnedReason?: string | null + returnedFailure?: { kind: string } + source: AgentSessionQueueSource + senderLabels: string[] + senderCount: number + preview: string + truncated: boolean + bodyBytes: number +} + +export type AgentSessionQueuedMessagesPageRequest = { + sessionId: string + source?: AgentSessionQueueSource + cursor?: string + aroundMessageId?: string + size?: number +} + +export type AgentSessionQueueReset = { status: 'stale'; generation: string; revision: number } +export type AgentSessionQueuedMessagesPageResult = + | AgentSessionQueueReset + | { + status: 'page' + generation: string + revision: number + rows: AgentSessionQueuedMessagePreview[] + hasMore: { before: boolean; after: boolean } + previousCursor: string | null + nextCursor: string | null + } + +export type AgentSessionQueuedMessageReadRequest = { + sessionId: string + messageId: string + expectedState?: 'waiting' | 'returned' + cursor?: string +} + +// Concatenate json in offset order, then JSON.parse once complete; never edit a partial body. +export type AgentSessionQueuedMessageReadResult = + | AgentSessionQueueReset + | { + status: 'body' + generation: string + revision: number + messageId: string + state: 'waiting' | 'returned' + bodyBytes: number + /** Offset in UTF-16 code units of the stored JSON. */ + part: { offset: number; json: string } + nextCursor: string | null + } + | { + status: 'gone' | 'handed-off' | 'state-changed' + generation: string + revision: number + messageId: string + state?: string + } diff --git a/src/shared/agent-session-queued-message-wire.ts b/src/shared/agent-session-queued-message-wire.ts index d10d263aa024..5292fd61de3f 100644 --- a/src/shared/agent-session-queued-message-wire.ts +++ b/src/shared/agent-session-queued-message-wire.ts @@ -3,6 +3,7 @@ import type { UnreadAgentSessionFailureFact } from './agent-session-failure' import type { AgentJournalMessageItem } from './agent-session-journal-types' +import type { AgentSessionQueueSummary } from './agent-session-queue-pages' /** The draft could not be converted into a send; an explicit Send retries it. */ export const QUEUED_MESSAGE_PAUSED_SEND_FAILED = 'send_failed' as const @@ -19,6 +20,8 @@ export type AgentSessionQueuePause = { reason: 'stopped' } /** What rides beside a frame's `queuedMessages`, published together with the list. */ export type AgentSessionQueuePublicationFields = { + /** Only readers opting into paged-v1 receive this instead of queuedMessages. */ + queueSummary?: AgentSessionQueueSummary /** Null when the queue sends on its own. */ queuePause?: AgentSessionQueuePause | null /** The card the queue sends next once nothing runs, null while anything holds the queue. diff --git a/src/shared/agent-session-wire.ts b/src/shared/agent-session-wire.ts index bfcd45c7cf34..48543592dd70 100644 --- a/src/shared/agent-session-wire.ts +++ b/src/shared/agent-session-wire.ts @@ -1,4 +1,6 @@ import type { AgentSessionUnavailable } from './agent-session-availability' +import type { AgentSessionQueueSummary, AgentSessionQueueView } from './agent-session-queue-pages' +export * from './agent-session-queue-pages' import type { AgentSessionBackgroundTask, AgentSessionBackgroundTaskState @@ -93,6 +95,7 @@ export type AgentSessionHistoryRequest = { /** Required for `before` and `after`; ignored for `tail`. */ cursor?: AgentJournalCursor limit?: number + queueView?: AgentSessionQueueView } /** A subagent named by a roster row the page does not carry, while its own rows are on it. */ @@ -134,6 +137,7 @@ export type AgentSessionHistoryPage = { /** The host's queued drafts. Absent = no claim (older host); `[]`/null = empty. * Live subscription state stays authoritative over a stale history answer. */ queuedMessages?: AgentSessionQueuedMessage[] | null + queueSummary?: AgentSessionQueueSummary /** The queue's pause, published with the list: present whenever `queuedMessages` is, null * when the queue sends on its own. */ queuePause?: AgentSessionQueuePause | null diff --git a/src/shared/protocol-version.ts b/src/shared/protocol-version.ts index cac746323b86..c69af2d66e71 100644 --- a/src/shared/protocol-version.ts +++ b/src/shared/protocol-version.ts @@ -58,10 +58,7 @@ import { SKILL_MANAGEMENT_CAPABILITY, SKILL_UPLOAD_CAPABILITY } from './skill-install-capability' -import { - AGENT_LAUNCH_RUNTIME_CAPABILITIES, - AGENT_LAUNCH_RUNTIME_CAPABILITY -} from './agent-launch-runtime-capability' +import * as launch from './agent-launch-runtime-capability' // Why: declares the Orca runtime RPC compatibility contract. Desktop, // headless server, CLI, and mobile builds may drift in app version, but @@ -250,6 +247,9 @@ export const AGENT_SESSION_SEND_ANSWERS_PROOF_RUNTIME_CAPABILITY = // gates the rollout. export const AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY = 'agent-session.queued-messages.v1' as const +// Why: queue pages and lossless body parts are separate from typed queueing. History/subscribe +// params are strict on old hosts: send queueView: 'paged-v1' only after this host capability. +export const AGENT_SESSION_QUEUE_PAGES_RUNTIME_CAPABILITY = 'agent-session.queue-pages.v1' as const // Why: paired clients advertise Claude-structured support so the host can gate its agent-specific // journal and lifecycle surfaces independently from Codex support. export const CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY = @@ -330,7 +330,7 @@ export const NATIVE_REMOTE_RUNTIME_CLIENT_CAPABILITIES = [ REPO_SEARCH_QUALIFIED_REFS_RUNTIME_CAPABILITY, AUTOMATION_OWNER_FENCING_RUNTIME_CAPABILITY, AUTOMATION_CREATE_IDEMPOTENCY_RUNTIME_CAPABILITY, - AGENT_LAUNCH_RUNTIME_CAPABILITY + launch.AGENT_LAUNCH_RUNTIME_CAPABILITY ] as const export const ANTIGRAVITY_CONFIGURED_MODEL_RUNTIME_CAPABILITY = @@ -405,6 +405,7 @@ export const RUNTIME_CAPABILITIES = [ AGENT_SESSION_REWIND_RUNTIME_CAPABILITY, AGENT_SESSION_REWIND_RECOVERY_CAPABILITY, AGENT_SESSION_CONVERSATION_OUTLINE_RUNTIME_CAPABILITY, + AGENT_SESSION_QUEUE_PAGES_RUNTIME_CAPABILITY, AGENT_SESSION_BACKGROUND_TASK_STOP_CAPABILITY, AGENT_SESSION_QUESTION_ANSWERS_RUNTIME_CAPABILITY, AGENT_SESSION_ATTACHMENTS_RUNTIME_CAPABILITY, @@ -435,7 +436,7 @@ export const RUNTIME_CAPABILITIES = [ AUTOMATION_OWNER_FENCING_RUNTIME_CAPABILITY, AUTOMATION_CREATE_IDEMPOTENCY_RUNTIME_CAPABILITY, NOTIFICATIONS_REMOTE_PUSH_RUNTIME_CAPABILITY, - ...AGENT_LAUNCH_RUNTIME_CAPABILITIES + ...launch.AGENT_LAUNCH_RUNTIME_CAPABILITIES ] as const export type RuntimeCapability = (typeof RUNTIME_CAPABILITIES)[number] | (string & {}) diff --git a/src/shared/rpc-contract/rpc-params-catalog.generated.ts b/src/shared/rpc-contract/rpc-params-catalog.generated.ts index 27c4ee41ded5..376730171da2 100644 --- a/src/shared/rpc-contract/rpc-params-catalog.generated.ts +++ b/src/shared/rpc-contract/rpc-params-catalog.generated.ts @@ -504,6 +504,8 @@ import { ModelCatalogParams, OptionsParams, QueuedMessageActionParams, + QueuedMessageReadParams, + QueuedMessagesPageParams, QueuedMessagesResumeParams, RespondParams, RespondToQuestionParams, @@ -627,7 +629,9 @@ export const RPC_PARAMS_BY_METHOD = { 'agentSession.modelCatalog': ModelCatalogParams, 'agentSession.options': OptionsParams, 'agentSession.queuedMessageDelete': QueuedMessageActionParams, + 'agentSession.queuedMessageRead': QueuedMessageReadParams, 'agentSession.queuedMessageSend': QueuedMessageActionParams, + 'agentSession.queuedMessagesPage': QueuedMessagesPageParams, 'agentSession.queuedMessagesResume': QueuedMessagesResumeParams, 'agentSession.readVisual': ReadVisualParams, 'agentSession.release': HoldParams, diff --git a/src/shared/rpc-contract/structured-agent-session-params.ts b/src/shared/rpc-contract/structured-agent-session-params.ts index d3ed6283c48a..628cc7e52d7d 100644 --- a/src/shared/rpc-contract/structured-agent-session-params.ts +++ b/src/shared/rpc-contract/structured-agent-session-params.ts @@ -223,6 +223,11 @@ export const QueuedMessageActionParams = z * restart's) so the cards send again. Gated like the draft actions above. */ export const QueuedMessagesResumeParams = z.object({ envelope: MutationEnvelope }).strict() +export { + QueuedMessagesPageParams, + QueuedMessageReadParams +} from './structured-agent-session-queue-page-params' + export const RespondParams = z .object({ envelope: MutationEnvelope, @@ -350,18 +355,19 @@ export const RestartResumeParams = z .object({ sessionIds: z.array(SessionId).max(MAX_RESTART_RESUME_SESSIONS).optional() }) .strict() -export const HistoryParams = z - .object({ - sessionId: SessionId, - direction: z.enum(AGENT_SESSION_HISTORY_DIRECTIONS), - cursor: JournalCursor.optional(), - limit: z.number().int().positive().max(AGENT_SESSION_HISTORY_MAX_LIMIT).optional() - }) - .strict() +export const HistoryParams = z.strictObject({ + sessionId: SessionId, + direction: z.enum(AGENT_SESSION_HISTORY_DIRECTIONS), + cursor: JournalCursor.optional(), + limit: z.number().int().positive().max(AGENT_SESSION_HISTORY_MAX_LIMIT).optional(), + queueView: z.literal('paged-v1').optional() +}) -export const SubscribeParams = z - .object({ sessionId: SessionId, cursor: JournalCursor.optional() }) - .strict() +export const SubscribeParams = z.strictObject({ + sessionId: SessionId, + cursor: JournalCursor.optional(), + queueView: z.literal('paged-v1').optional() +}) // Not strict: an older host ignores these params entirely, and this host must ignore a newer client's. export const SubscribeTurnCompletionsParams = z.object({ includePrompts: z.boolean().optional() }) diff --git a/src/shared/rpc-contract/structured-agent-session-queue-page-params.ts b/src/shared/rpc-contract/structured-agent-session-queue-page-params.ts new file mode 100644 index 000000000000..2e5f35ec8958 --- /dev/null +++ b/src/shared/rpc-contract/structured-agent-session-queue-page-params.ts @@ -0,0 +1,28 @@ +import { z } from 'zod' +import { Identifier, SessionId } from './structured-agent-session-identifiers' +import { AGENT_SESSION_QUEUE_SOURCES } from '../agent-session-queue-pages' + +export const QueuedMessagesPageParams = z + .strictObject({ + sessionId: SessionId, + source: z.enum(AGENT_SESSION_QUEUE_SOURCES).optional(), + cursor: z + .string() + .min(1) + .max(16 * 1024) + .optional(), + aroundMessageId: Identifier('Invalid queued message id').optional(), + size: z.number().int().positive().optional() + }) + .refine((value) => !(value.cursor && value.aroundMessageId), 'Choose a cursor or an anchor') + +export const QueuedMessageReadParams = z.strictObject({ + sessionId: SessionId, + messageId: Identifier('Invalid queued message id'), + expectedState: z.enum(['waiting', 'returned']).optional(), + cursor: z + .string() + .min(1) + .max(16 * 1024) + .optional() +}) diff --git a/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts b/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts index 12fd794ff2d9..7fe98fcd2456 100644 --- a/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts +++ b/tests/e2e/cross-version-wire/cross-version-agent-session-wire.unit.test.ts @@ -27,6 +27,7 @@ import { AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY, AGENT_SESSION_QUESTION_ANSWERS_RUNTIME_CAPABILITY, AGENT_SESSION_REWIND_RUNTIME_CAPABILITY, + AGENT_SESSION_QUEUE_PAGES_RUNTIME_CAPABILITY, AGENT_SESSION_CONVERSATION_OUTLINE_RUNTIME_CAPABILITY, AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY, STRUCTURED_AGENT_SESSION_CLIENT_LAUNCH_MODE_CAPABILITY, @@ -257,6 +258,12 @@ describe('cross-version structured agent sessions', () => { }) it('can detect the absence during negotiation instead of by calling', () => { + for (const build of [current, baseline]) { + expect(build.capabilities.includes(AGENT_SESSION_QUEUE_PAGES_RUNTIME_CAPABILITY)).toBe( + build.methodNames.includes('agentSession.queuedMessagesPage') && + build.methodNames.includes('agentSession.queuedMessageRead') + ) + } // The invariant that survives a release cut: each build's advertised list and // its registered methods agree. "The old build has neither" is only true // until a release ships the surface, and pinning it turns this red on the cut diff --git a/tests/e2e/cross-version-wire/structured-agent-session-host-fixture.ts b/tests/e2e/cross-version-wire/structured-agent-session-host-fixture.ts index 949457d4f63a..10b5d76eece5 100644 --- a/tests/e2e/cross-version-wire/structured-agent-session-host-fixture.ts +++ b/tests/e2e/cross-version-wire/structured-agent-session-host-fixture.ts @@ -57,6 +57,8 @@ export function structuredHostStub( cancel: vi.fn(async () => ({ ok: true, replayed: false })), queuedMessageSend: vi.fn(async () => ({ ok: true, replayed: false })), queuedMessageDelete: vi.fn(async () => ({ ok: true, replayed: false })), + queuedMessagesPage: vi.fn(async () => ({ status: 'page', rows: [] })), + queuedMessageRead: vi.fn(async () => ({ status: 'gone' })), queuedMessagesResume: vi.fn(async () => ({ ok: true, replayed: false })), rewind: vi.fn(async () => ({ ok: true, diff --git a/tests/e2e/cross-version-wire/structured-agent-session-surface-calls.ts b/tests/e2e/cross-version-wire/structured-agent-session-surface-calls.ts index f24ea7224509..454fc3ca0af7 100644 --- a/tests/e2e/cross-version-wire/structured-agent-session-surface-calls.ts +++ b/tests/e2e/cross-version-wire/structured-agent-session-surface-calls.ts @@ -42,6 +42,16 @@ export const STRUCTURED_CALLS: { }, { method: 'agentSession.send', hostMethod: 'send', result: { ok: true, replayed: false } }, { method: 'agentSession.cancel', hostMethod: 'cancel', result: { ok: true, replayed: false } }, + { + method: 'agentSession.queuedMessagesPage', + hostMethod: 'queuedMessagesPage', + result: { status: 'page', rows: [] } + }, + { + method: 'agentSession.queuedMessageRead', + hostMethod: 'queuedMessageRead', + result: { status: 'gone' } + }, // Draft mutations for mid-turn queueing. Methods exist ahead of the // capability's advertisement; only capability-gated clients ever call them. { diff --git a/tests/e2e/cross-version-wire/structured-agent-session-surface-manifest.ts b/tests/e2e/cross-version-wire/structured-agent-session-surface-manifest.ts index 3aac5e704dc1..45222c7e1bc3 100644 --- a/tests/e2e/cross-version-wire/structured-agent-session-surface-manifest.ts +++ b/tests/e2e/cross-version-wire/structured-agent-session-surface-manifest.ts @@ -149,6 +149,8 @@ export function paramsFor(method: string): unknown { const fields = { messageId: 'queued-1' } return { envelope: envelope({ method, fields, fence }), ...fields } } + case 'agentSession.queuedMessageRead': + return { sessionId: SESSION, messageId: 'queued-1' } case 'agentSession.queuedMessagesResume': return { envelope: envelope({ method, fields: {}, fence }) } case 'agentSession.respondToApproval': From 1a64dd7177c57465b1da34a665057956f04e5d42 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 8 Oct 2026 17:20:15 -0700 Subject: [PATCH 2/2] fix(native-chat): typecheck queue page sources and subscription tests --- .../journal-queued-message-pages.ts | 16 ++++++---------- ...ured-agent-session-queue-subscription.test.ts | 6 +++++- src/shared/agent-session-queue-pages.ts | 3 +++ 3 files changed, 14 insertions(+), 11 deletions(-) diff --git a/src/main/native-chat/agent-session-journal/journal-queued-message-pages.ts b/src/main/native-chat/agent-session-journal/journal-queued-message-pages.ts index f318e8154a6e..ef5524f670f8 100644 --- a/src/main/native-chat/agent-session-journal/journal-queued-message-pages.ts +++ b/src/main/native-chat/agent-session-journal/journal-queued-message-pages.ts @@ -1,5 +1,8 @@ import { randomUUID } from 'node:crypto' -import type { AgentSessionQueueSource } from '../../../shared/agent-session-queue-pages' +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' @@ -63,10 +66,7 @@ export class JournalQueuedMessagePages { FROM queued_messages WHERE session_id = ? AND ${QUEUED_MESSAGE_UNSETTLED_SQL} GROUP BY ${QUEUED_MESSAGE_SOURCE_SQL}`) .all(this.sessionId)) { - if ( - (row.source === 'person' || row.source === 'agent' || row.source === 'unknown') && - typeof row.count === 'number' - ) { + if (isAgentSessionQueueSource(row.source) && typeof row.count === 'number') { counts[row.source] = row.count } } @@ -129,11 +129,7 @@ export class JournalQueuedMessagePages { .iterate(this.sessionId, ...bindings, size)) { const row = readStoredQueuedMessageRow(stored) const source = stored.source - if ( - row && - typeof stored.body_bytes === 'number' && - (source === 'person' || source === 'agent' || source === 'unknown') - ) { + if (row && typeof stored.body_bytes === 'number' && isAgentSessionQueueSource(source)) { yield { row, source, bodyBytes: stored.body_bytes } } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queue-subscription.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-subscription.test.ts index 7cd1c1cf2bb8..06b88c6f82e9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queue-subscription.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queue-subscription.test.ts @@ -166,6 +166,7 @@ it('coalesces mutations while output is blocked, keeps the newest revision, and if (frames.length === 1) { return blocked } + return undefined } }) const revision = summary(frames[0])?.revision ?? 0 @@ -190,7 +191,7 @@ it('coalesces multi-page journal catch-up without advancing past an unsent batch await rig.journal.appendItem( { provider: 'claude', sessionId: 'native', uuid: `row-${index}` }, { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'row' }] }, - { fence: 1 } + { fence: 1, turnScope: { kind: 'thread' } } ) } const frames: AgentSessionSubscribeEvent[] = [] @@ -210,6 +211,7 @@ it('coalesces multi-page journal catch-up without advancing past an unsent batch if (frames.length === 1) { return blocked } + return undefined } }) expect(frames).toHaveLength(1) @@ -242,6 +244,7 @@ it('replays a coalesced snapshot only to the subscriber whose output was blocked if (blockedFrames.length === 1) { return blocked } + return undefined } }) registry.open({ @@ -289,6 +292,7 @@ it('re-derives a blocked reader from the execution host after journal eviction', if (frames.length === 1) { return blocked } + return undefined } }) const generation = summary(frames[0])?.generation diff --git a/src/shared/agent-session-queue-pages.ts b/src/shared/agent-session-queue-pages.ts index e53f1957fa70..09560c7a8564 100644 --- a/src/shared/agent-session-queue-pages.ts +++ b/src/shared/agent-session-queue-pages.ts @@ -7,6 +7,9 @@ export const AGENT_SESSION_QUEUE_PAGE_MAX_SIZE = 200 export const AGENT_SESSION_QUEUE_PAGE_DEFAULT_SIZE = 40 export const AGENT_SESSION_QUEUE_SOURCES = ['person', 'agent', 'unknown'] as const export type AgentSessionQueueSource = (typeof AGENT_SESSION_QUEUE_SOURCES)[number] +export function isAgentSessionQueueSource(value: unknown): value is AgentSessionQueueSource { + return AGENT_SESSION_QUEUE_SOURCES.some((source) => source === value) +} export type AgentSessionQueueView = 'paged-v1' export type AgentSessionQueueSummary = {