diff --git a/config/scripts/mobile-web-app-render-harness.mjs b/config/scripts/mobile-web-app-render-harness.mjs index 5675305d9ff..ffdfc690da2 100644 --- a/config/scripts/mobile-web-app-render-harness.mjs +++ b/config/scripts/mobile-web-app-render-harness.mjs @@ -103,7 +103,7 @@ export async function readBridgeProtocolVersion() { export async function readBridgeWindowCaps() { const sources = await Promise.all( [ - 'mobile/src/mobile-web-shell/bridge/bridge-caps.ts', + 'src/shared/mobile-bridge-frame-bytes.ts', 'mobile/src/mobile-web-shell/bridge-host-subscriptions.ts' ].map((path) => readFile(join(projectDir, path), 'utf8')) ) @@ -119,7 +119,7 @@ export async function readBridgeWindowCaps() { .reduce((product, factor) => product * factor, 1) } return { - maxMessageBytes: read('BRIDGE_MAX_MESSAGE_BYTES'), + maxMessageBytes: read('MOBILE_BRIDGE_MAX_FRAME_BYTES'), maxUnackedFrames: read('BRIDGE_MAX_UNACKED_FRAMES'), maxUnackedBytes: read('BRIDGE_MAX_UNACKED_BYTES') } diff --git a/config/scripts/vitest-sqlite-runtime-files.mjs b/config/scripts/vitest-sqlite-runtime-files.mjs index 7dba547d6ff..f4fdaeeff5e 100644 --- a/config/scripts/vitest-sqlite-runtime-files.mjs +++ b/config/scripts/vitest-sqlite-runtime-files.mjs @@ -58,6 +58,7 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/ipc/ssh-host-server-on-connect-wiring.test.ts', 'src/main/native-chat/agent-session-attachments/agent-session-attachment-claims.test.ts', 'src/main/native-chat/agent-session-attachments/agent-session-attachment-sweep.test.ts', + 'src/main/native-chat/agent-session-wire/agent-session-bounded-frames.test.ts', 'src/main/native-chat/agent-session-wire/agent-session-history-byte-accounting.test.ts', 'src/main/native-chat/agent-session-wire/agent-session-history-conversation-window.test.ts', 'src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts', diff --git a/mobile/src/mobile-web-shell/bridge/bridge-caps.ts b/mobile/src/mobile-web-shell/bridge/bridge-caps.ts index 4e625884b84..a8b0c145d8f 100644 --- a/mobile/src/mobile-web-shell/bridge/bridge-caps.ts +++ b/mobile/src/mobile-web-shell/bridge/bridge-caps.ts @@ -1,3 +1,5 @@ +import { MOBILE_BRIDGE_MAX_FRAME_BYTES } from '../../../../src/shared/mobile-bridge-frame-bytes' + /** * The bridge's caps, its refusal vocabulary, and the single point that enforces them. * @@ -16,8 +18,9 @@ export const BRIDGE_DIRECTIONS = ['page-to-shell', 'shell-to-page'] as const export type BridgeDirection = (typeof BRIDGE_DIRECTIONS)[number] -/** Frame ceiling, in UTF-8 bytes of the raw string, checked before `JSON.parse` sees it. */ -export const BRIDGE_MAX_MESSAGE_BYTES = 640 * 1024 +/** Frame ceiling, in UTF-8 bytes of the raw string, checked before `JSON.parse` sees it. Never + * raise it alone: hosts size frames to what released shells accept. */ +export const BRIDGE_MAX_MESSAGE_BYTES = MOBILE_BRIDGE_MAX_FRAME_BYTES /** Nesting levels a page-to-shell frame may carry, counting the frame object itself as one. */ export const BRIDGE_MAX_DEPTH = 16 diff --git a/src/main/native-chat/agent-session-journal/journal-dispatch-legacy-reason.test.ts b/src/main/native-chat/agent-session-journal/journal-dispatch-legacy-reason.test.ts new file mode 100644 index 00000000000..75783309d18 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-dispatch-legacy-reason.test.ts @@ -0,0 +1,52 @@ +// A dispatch row written before the write-side reason bound (#20654) still carries the provider's +// full error text; the submission every page and frame attaches must read it bounded. + +import { describe, expect, it } from 'vitest' +import { applyJournalRow, createJournalReducerState } from './journal-reducer' +import { DEFAULT_JOURNAL_PAYLOAD_LIMITS } from './journal-payload-bounds' +import type { JournalRow } from './journal-row-schema' + +const EPOCH = 'epoch-1' + +function base(seq: number): { v: number; epoch: string; seq: number; fence: number; ts: number } { + return { v: 1, epoch: EPOCH, seq, fence: 1, ts: 1_000 + seq } +} + +function reasonAfter(reason: string | null) { + const state = createJournalReducerState('session-1', EPOCH) + applyJournalRow(state, { + kind: 'submission', + clientMessageId: 'cm_1', + payloadFingerprint: 'fp_1', + providerHandle: { kind: 'codex', threadId: 'thread-1' }, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] }, + ...base(1) + }) + const dispatch: JournalRow = { + kind: 'dispatch', + clientMessageId: 'cm_1', + state: 'rejected', + providerItemId: null, + reason, + ...base(2) + } + applyJournalRow(state, dispatch) + return state.submissions.get('cm_1')?.reason +} + +describe("a dispatch row's reason", () => { + it('reads bounded, head first, when an older build stored megabytes of it', () => { + const reason = reasonAfter(`Write failed: ${'x'.repeat(4 * 1024 * 1024)}`) + + expect(reason?.startsWith('Write failed: ')).toBe(true) + expect(reason).toContain('[Orca: output truncated') + expect(Buffer.byteLength(reason ?? '')).toBeLessThan( + DEFAULT_JOURNAL_PAYLOAD_LIMITS.inlineHeadBytes + 1024 + ) + }) + + it('reads as stored when it already fits, or is absent', () => { + expect(reasonAfter('Provider rejected the message.')).toBe('Provider rejected the message.') + expect(reasonAfter(null)).toBeNull() + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts b/src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts index 305b7383ad9..60cfe328ec5 100644 --- a/src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts @@ -8,6 +8,7 @@ import { import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' import type { AgentJournalAnsweredTurn } from '../../../shared/agent-session-journal-types' import { journalDispatchRowApplies } from './journal-dispatch-settlement' +import { boundInlineText, DEFAULT_JOURNAL_PAYLOAD_LIMITS } from './journal-payload-bounds' import type { JournalReducerState } from './journal-reducer' import { noteTurnAccepted, @@ -28,7 +29,9 @@ export function applyJournalDispatchRow( submission.fence = row.fence submission.dispatchState = row.state submission.providerItemId = row.providerItemId - submission.reason = row.reason + // Rows written before the write-side bound (#20654) carry the full text; bound them on read too. + submission.reason = + row.reason === null ? null : boundInlineText(row.reason, DEFAULT_JOURNAL_PAYLOAD_LIMITS).text const rejection = row.state === 'rejected' ? readStoredRejectionFact(row.rejection) : undefined if (rejection) { submission.rejection = rejection diff --git a/src/main/native-chat/agent-session-wire/agent-session-bounded-frame-item.ts b/src/main/native-chat/agent-session-wire/agent-session-bounded-frame-item.ts new file mode 100644 index 00000000000..d1dc03b9456 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-bounded-frame-item.ts @@ -0,0 +1,62 @@ +// One journal item as a bounded-frame reader receives it: never more than one row's share of a +// frame. + +import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' +import { boundJournalKeyComponent } from '../../../shared/agent-session-journal-item-key' +import { + BOUNDED_FRAME_CONTENT_BUDGET_BYTES, + oversizedHistoryItem +} from './agent-session-history-page-bounds' +import { jsonBytes, shortenJsonStringsToFit } from './json-string-shortening' + +/** One row's share of a page, so a page always carries several rows. */ +export const BOUNDED_FRAME_ROW_BYTES = BOUNDED_FRAME_CONTENT_BUDGET_BYTES / 8 + +/** What the body may weigh once the rest of the row is paid for. */ +function bodyBudget(item: AgentJournalRenderItem): number { + return BOUNDED_FRAME_ROW_BYTES - (jsonBytes({ ...item, body: null }) - jsonBytes(null)) +} + +/** The too-large marker with every identifier it carries held to a key's bound, so the marker + * itself always fits a row. Provenance a reader never joins on is dropped. */ +function oversizedFrameItem(item: AgentJournalRenderItem): AgentJournalRenderItem { + const { + providerParentRef: _provenance, + agentId, + parentAgentId, + turnScope, + ...marker + } = oversizedHistoryItem(item) + return { + ...marker, + ...(agentId === undefined ? {} : { agentId: boundJournalKeyComponent(agentId) }), + ...(parentAgentId === undefined + ? {} + : { parentAgentId: boundJournalKeyComponent(parentAgentId) }), + ...(turnScope?.kind === 'turn' + ? { turnScope: { kind: 'turn', turnItemId: boundJournalKeyComponent(turnScope.turnItemId) } } + : turnScope === undefined + ? {} + : { turnScope }) + } +} + +function boundFrameItem(item: AgentJournalRenderItem): AgentJournalRenderItem { + const body = shortenJsonStringsToFit(item.body, bodyBudget(item)) + if (body === null) { + return oversizedFrameItem(item) + } + return body === item.body ? item : { ...item, body } +} + +// Items are immutable per revision, so a catch-up or a second subscriber reuses the bound form. +const bounded = new WeakMap() + +export function boundedFrameItem(item: AgentJournalRenderItem): AgentJournalRenderItem { + let result = bounded.get(item) + if (result === undefined) { + result = boundFrameItem(item) + bounded.set(item, result) + } + return result +} diff --git a/src/main/native-chat/agent-session-wire/agent-session-bounded-frames.test.ts b/src/main/native-chat/agent-session-wire/agent-session-bounded-frames.test.ts new file mode 100644 index 00000000000..5a9b9b03818 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-bounded-frames.test.ts @@ -0,0 +1,356 @@ +// The bounded-frame invariant: no frame the host emits to a `boundedFrames` subscriber is larger +// than a phone's bridge frame, whatever the chat holds. Each case below is a worst case for one +// thing a frame can carry; the last puts them all in one chat. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { + AGENT_JOURNAL_THREAD_SCOPE, + type AgentJournalItemBody, + type AgentJournalItemIdentity, + type AgentJournalRenderItem +} from '../../../shared/agent-session-journal-types' +import type { + AgentSessionBackgroundTaskState, + AgentSessionQueuedMessage, + AgentSessionSlashCommand, + AgentSessionSubscribeEvent +} from '../../../shared/agent-session-wire' +import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' +import { MOBILE_BRIDGE_MAX_FRAME_BYTES } from '../../../shared/mobile-bridge-frame-bytes' +import { createTrackedJournalOpener } from '../agent-session-journal/journal-host-database-test-support' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { BOUNDED_FRAME_ROW_BYTES, boundedFrameItem } from './agent-session-bounded-frame-item' +import { + EMPTY_STRUCTURED_AGENT_SESSION, + oldestStructuredAgentSessionCursor, + reduceStructuredAgentSession +} from '../../../shared/structured-agent-session-reducer' +import { readAgentSessionHistory } from './agent-session-history-page' +import { jsonBytes } from './json-string-shortening' +import { AgentSessionSubscribers } from './structured-agent-session-subscribers' +import type { QueuePublication } from './structured-agent-session-queued-publication' + +// Longest id a client may name a session by, so the ids every frame repeats are at their widest. +const SESSION = `s${'x'.repeat(511)}` +// Escaped (6 bytes), multi-byte (3 bytes) and astral (4 bytes) text, so a budget counted in +// characters rather than serialized bytes would overflow. +const WIDE_TEXT = '\u0001한😀' + +let root: string +let journal: AgentSessionJournal +const journals = createTrackedJournalOpener() + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-bounded-frames-')) + journal = await journals.open({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: codexProviderHandle('thread-1') + }, + stateDirectory: root + }) +}) + +afterEach(async () => { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) +}) + +let ordinal = 0 +function identity(): AgentJournalItemIdentity { + ordinal += 1 + return { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal } +} + +async function append(body: AgentJournalItemBody): Promise { + await journal.appendItem(identity(), body, { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE }) +} + +const message = (text: string): AgentJournalItemBody => ({ + kind: 'message', + role: 'assistant', + blocks: [{ type: 'text', text }] +}) + +const output = (head: string) => ({ + head, + byteLength: 40 * 1024 * 1024, + digest: 'f'.repeat(64), + truncated: true +}) + +/** The frame a phone's bridge carries for `payload`: its envelope around the event. */ +function bridgeFrameBytes(payload: AgentSessionSubscribeEvent): number { + return jsonBytes({ v: 1, type: 'event', id: crypto.randomUUID(), seq: 2 ** 31, payload }) +} + +function rendered(item: Partial & { body: AgentJournalItemBody }) { + return { itemId: 'item-1', revision: 1, sequence: 1, observedAt: 1, ...item } +} + +describe('a bounded row', () => { + it('never weighs more than its share of a page, whatever its fields hold', () => { + const rows = [ + rendered({ body: message(WIDE_TEXT.repeat(400_000)) }), + rendered({ + body: { + kind: 'tool-call', + name: 'Write', + state: 'completed', + input: { file_path: '/repo/a.ts', content: WIDE_TEXT.repeat(400_000) }, + output: output(WIDE_TEXT.repeat(4_000)) + } + }), + // Too many short strings to shorten: the row becomes the too-large marker. + rendered({ + body: { + kind: 'question', + question: 'Pick one', + options: Array.from({ length: 20_000 }, (_, index) => ({ id: `o${index}`, label: 'x' })), + resolution: { + state: 'pending', + selectedOptionId: null, + resolvedBy: null, + resolvedAt: null + } + } + }), + // Identifiers are never shortened, so they are what the marker must bound itself. + rendered({ + itemId: 'i'.repeat(5 * 1024 * 1024), + agentId: 'a'.repeat(5 * 1024 * 1024), + parentAgentId: 'p'.repeat(5 * 1024 * 1024), + providerParentRef: 'r'.repeat(5 * 1024 * 1024), + turnScope: { kind: 'turn', turnItemId: 't'.repeat(5 * 1024 * 1024) }, + body: message('short') + }) + ] + for (const row of rows) { + expect(jsonBytes(boundedFrameItem(row))).toBeLessThanOrEqual(BOUNDED_FRAME_ROW_BYTES) + } + }) + + it('is the stored row when it already fits', () => { + const row = rendered({ body: message('hello') }) + expect(boundedFrameItem(row)).toBe(row) + }) +}) + +describe('frames to a bounded-frame subscriber', () => { + function largeQueue(): QueuePublication { + // What an older build admitted under its 1 MiB bound. + const queuedMessages: AgentSessionQueuedMessage[] = Array.from({ length: 40 }, (_, index) => ({ + messageId: `draft-${index}`, + position: index, + state: 'waiting', + body: { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: WIDE_TEXT.repeat(4_000) }] + } + })) + return { queuedMessages, queuePause: null, nextQueuedMessageId: 'draft-0' } + } + + const largeRoster: AgentSessionBackgroundTaskState = { + state: 'monitoring', + tasks: Array.from({ length: 2_000 }, (_, index) => ({ + id: `task-${index}`, + kind: 'command', + description: WIDE_TEXT.repeat(200) + })) + } + + const largeCommands: AgentSessionSlashCommand[] = Array.from({ length: 20_000 }, (_, index) => ({ + name: `command-${index}`, + kind: 'command', + description: WIDE_TEXT.repeat(50) + })) + + async function fillChat(): Promise { + for (let index = 0; index < 30; index += 1) { + await append(message(WIDE_TEXT.repeat(30_000))) + await append({ + kind: 'tool-call', + name: 'Bash', + state: 'completed', + input: { command: WIDE_TEXT.repeat(3_000) }, + output: output(WIDE_TEXT.repeat(4_000)) + }) + } + } + + it('each fit a phone bridge frame, from the opening snapshot through catch-up and live rows', async () => { + await fillChat() + const queue = largeQueue() + const subscribers = new AgentSessionSubscribers({ + readQueuePublication: () => queue, + readBackgroundTasks: () => largeRoster, + readCommands: () => largeCommands + }) + const frames: AgentSessionSubscribeEvent[] = [] + const emit = (event: AgentSessionSubscribeEvent): void => { + frames.push(event) + } + subscribers.open({ + id: 'opening', + sessionId: SESSION, + journal, + fence: 1, + emit, + boundedFrames: true + }) + subscribers.open({ + id: 'catch-up', + sessionId: SESSION, + journal, + fence: 1, + emit, + cursor: { epoch: journal.epoch, sequence: 0 }, + boundedFrames: true + }) + await fillChat() + subscribers.publish(SESSION, journal, { turnId: 't'.repeat(100_000), text: 'working' }) + subscribers.snapshot(SESSION, journal, 1) + + expect(frames.length).toBeGreaterThan(4) + for (const frame of frames) { + expect(bridgeFrameBytes(frame)).toBeLessThanOrEqual(MOBILE_BRIDGE_MAX_FRAME_BYTES) + expect(frame).not.toHaveProperty('commands') + } + // The side lists still arrive, each in a frame of its own. + const cards = frames.flatMap((frame) => + frame.type === 'batch' && frame.queuedMessages ? [frame.queuedMessages] : [] + ) + // Whole cards from the front: an older build's 1 MiB queue lists what fits. + const listed = cards.at(-1) ?? [] + expect(listed.length).toBeGreaterThan(0) + expect(listed).toEqual(queue.queuedMessages.slice(0, listed.length)) + expect(frames.some((frame) => frame.type === 'batch' && frame.backgroundTasks)).toBe(true) + }) + + it('keep a burst of rows to one page, so no publish outruns the bridge window', async () => { + const queue = largeQueue() + const subscribers = new AgentSessionSubscribers({ + readQueuePublication: () => queue, + readBackgroundTasks: () => largeRoster + }) + const frames: AgentSessionSubscribeEvent[] = [] + subscribers.open({ + id: 'phone', + sessionId: SESSION, + journal, + fence: 1, + emit: (event) => frames.push(event), + boundedFrames: true + }) + // About 10 MB of rows land before one publish, as a drain of buffered events can. + for (let index = 0; index < 40; index += 1) { + await append(message('x'.repeat(256 * 1024))) + } + frames.length = 0 + subscribers.publish(SESSION, journal) + + // The phone's shell ends a stream past 4 MiB unacked. + const bytes = frames.reduce((sum, frame) => sum + bridgeFrameBytes(frame), 0) + expect(bytes).toBeLessThan(4 * 1024 * 1024) + // One page of rows; the rest are the side lists a hydrating frame owes. + const [tail, ...side] = frames + expect(tail?.type === 'snapshot' && tail.page).toMatchObject({ hasOlder: true }) + expect(side.every((frame) => frame.type === 'batch' && frame.batch.items.length === 0)).toBe( + true + ) + // Live rows after it arrive as batches again. + await append(message('next')) + const live = frames.length + subscribers.publish(SESSION, journal) + expect(frames.slice(live).map((frame) => frame.type)).toEqual(['batch']) + + // A phone that applied all of it pages back to the first row with no gap. + let state = frames.reduce( + (current, event) => reduceStructuredAgentSession(current, { type: 'event', event }), + EMPTY_STRUCTURED_AGENT_SESSION + ) + for (let cursor = oldestStructuredAgentSessionCursor(state); state.hasOlder && cursor;) { + const older = readAgentSessionHistory(journal, { + sessionId: SESSION, + direction: 'before', + cursor + }) + state = reduceStructuredAgentSession(state, { + type: 'older-page', + requestedCursor: cursor, + page: older.page + }) + cursor = oldestStructuredAgentSessionCursor(state) + } + expect(state.items.map((item) => item.sequence)).toEqual( + journal.snapshot().items.map((item) => item.sequence) + ) + }) + + it('leave out a turn record no real provider id could make too large, rather than outgrow', async () => { + await journal.appendItem( + identity(), + { kind: 'turn', turnId: 'u'.repeat(100_000), state: 'running' }, + { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + const frames: AgentSessionSubscribeEvent[] = [] + new AgentSessionSubscribers().open({ + id: 'phone', + sessionId: SESSION, + journal, + fence: 1, + emit: (event) => frames.push(event), + boundedFrames: true + }) + const [snapshot] = frames + expect(bridgeFrameBytes(snapshot!)).toBeLessThanOrEqual(MOBILE_BRIDGE_MAX_FRAME_BYTES) + expect(snapshot?.type === 'snapshot' && snapshot.page).not.toHaveProperty('latestTurn') + }) + + it('leave every other subscriber the frames it had', async () => { + const stored = message(WIDE_TEXT.repeat(30_000)) + await append(stored) + const queue = largeQueue() + const frames: AgentSessionSubscribeEvent[] = [] + const subscribers = new AgentSessionSubscribers({ + readQueuePublication: () => queue, + readBackgroundTasks: () => largeRoster, + readCommands: () => largeCommands + }) + subscribers.open({ + id: 'desktop', + sessionId: SESSION, + journal, + fence: 1, + emit: (event) => frames.push(event) + }) + await append(stored) + subscribers.publish(SESSION, journal) + + expect(frames).toHaveLength(2) + expect(frames[0]).toMatchObject({ + type: 'snapshot', + commands: largeCommands, + queuedMessages: queue.queuedMessages, + backgroundTasks: largeRoster + }) + // Rows as stored, whole. + for (const frame of frames) { + const items = + frame.type === 'snapshot' + ? frame.page.items + : frame.type === 'batch' + ? frame.batch.items + : [] + expect(items.map((item) => item.body)).toEqual(items.map(() => stored)) + } + }) +}) diff --git a/src/main/native-chat/agent-session-wire/agent-session-bounded-frames.ts b/src/main/native-chat/agent-session-wire/agent-session-bounded-frames.ts new file mode 100644 index 00000000000..a42476e6171 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-bounded-frames.ts @@ -0,0 +1,156 @@ +// What a bounded-frame subscriber is sent: every frame fits a phone's bridge frame. +// +// A history frame carries rows its reader already sized in bounded form, and metadata held to +// `FRAME_METADATA_CEILING_BYTES`. The lists that ride beside history on other subscribers' frames +// (the draft queue and the background-task roster) each ride an empty batch of their own, fitted to +// the same content budget; the slash-command list is not sent, since a phone reads it from +// `agentSession.options`. + +import type { AgentJournalCursor } from '../../../shared/agent-session-journal-types' +import type { + AgentSessionBackgroundTaskState, + AgentSessionSubscribeEvent +} from '../../../shared/agent-session-wire' +import { boundedFrameItem } from './agent-session-bounded-frame-item' +import { boundedQueuePublication } from './agent-session-bounded-queue' +import { emptyAgentSessionBatch } from './agent-session-empty-batch' +import { + BOUNDED_FRAME_CONTENT_BUDGET_BYTES, + STORED_HISTORY_READER, + type HistoryReader +} from './agent-session-history-page-bounds' +import { jsonBytes, shortenJsonStringsToFit } from './json-string-shortening' +import type { QueuePublication } from './structured-agent-session-queued-publication' + +const BOUNDED_FRAME_READER: HistoryReader = { + form: boundedFrameItem, + budgetBytes: BOUNDED_FRAME_CONTENT_BUDGET_BYTES +} + +/** How a subscriber's rows are read, and so how its pages are sized. */ +export function subscriberHistoryReader(subscriber: { boundedFrames: boolean }): HistoryReader { + return subscriber.boundedFrames ? BOUNDED_FRAME_READER : STORED_HISTORY_READER +} + +/** The latest turn and the activity line are each a few hundred bytes, their ids and text bounded + * where they are made. Only a provider id far past any real one could pass these; the frame then + * leaves the field out, which a reader takes as a host that did not say. */ +const BOUNDED_FRAME_LATEST_TURN_BYTES = 8 * 1024 +const BOUNDED_FRAME_ACTIVITY_BYTES = 2 * 1024 + +type HistoryFrame = Exclude +type BatchFrame = Extract +type SideFields = Pick< + BatchFrame, + 'queuedMessages' | 'queuePause' | 'nextQueuedMessageId' | 'backgroundTasks' +> + +function within(value: unknown, maxBytes: number): boolean { + return value === null || jsonBytes(value) <= maxBytes +} + +/** The cursor a reader holds once it has applied `event`, which a side frame must name. */ +function cursorAfter(event: HistoryFrame): AgentJournalCursor { + return event.type === 'batch' + ? event.batch.cursor + : (event.page.liveCursor ?? event.page.window.nextCursor) +} + +/** `event` without the roster, which rides its own frame, and with metadata past its bound left + * out. The other side lists are attached only after this, so an event never carries them. */ +function boundedHistoryFrame(event: HistoryFrame): HistoryFrame { + const { backgroundTasks: _backgroundTasks, activity, ...rest } = event + const frame = { + ...rest, + ...(activity !== undefined && within(activity, BOUNDED_FRAME_ACTIVITY_BYTES) + ? { activity } + : {}) + } + if (frame.type === 'batch') { + const { latestTurn, ...batch } = frame + return latestTurn === undefined || within(latestTurn, BOUNDED_FRAME_LATEST_TURN_BYTES) + ? frame + : batch + } + const { latestTurn, ...page } = frame.page + return latestTurn === undefined || within(latestTurn, BOUNDED_FRAME_LATEST_TURN_BYTES) + ? frame + : { ...frame, page } +} + +function isEmptyBatch(frame: HistoryFrame): frame is BatchFrame { + return ( + frame.type === 'batch' && + frame.batch.items.length === 0 && + frame.batch.removedItemIds.length === 0 && + frame.batch.submissions.length === 0 + ) +} + +/** The newer half of each list in turn until the roster fits, least telling first; `incomplete` + * tells a reader that what is listed is not everything. */ +function withoutOlderRows(state: AgentSessionBackgroundTaskState): AgentSessionBackgroundTaskState { + let fitted = state + for (const list of ['settledTasks', 'children', 'tasks'] as const) { + while ( + (fitted[list]?.length ?? 0) > 0 && + jsonBytes(fitted) > BOUNDED_FRAME_CONTENT_BUDGET_BYTES + ) { + const rows = fitted[list] ?? [] + fitted = { ...fitted, [list]: rows.slice(Math.ceil(rows.length / 2)), incomplete: true } + } + } + return fitted +} + +/** The roster fitted to a frame: its text shortened, then older rows left out. */ +function boundedBackgroundTasks( + state: AgentSessionBackgroundTaskState | null +): AgentSessionBackgroundTaskState | null { + if (state === null) { + return null + } + return ( + shortenJsonStringsToFit(state, BOUNDED_FRAME_CONTENT_BUDGET_BYTES) ?? + shortenJsonStringsToFit(withoutOlderRows(state), BOUNDED_FRAME_CONTENT_BUDGET_BYTES) ?? { + state: state.state, + incomplete: true + } + ) +} + +type BoundedSideLists = { + queued: QueuePublication | undefined + backgroundTasks: AgentSessionBackgroundTaskState | null | undefined +} + +/** One event as a bounded-frame subscriber receives it: the history frame, then an empty batch per + * side list it would have carried. An empty history batch is one already, so the first rides it. */ +export function boundedFrames( + event: AgentSessionSubscribeEvent, + side: BoundedSideLists +): AgentSessionSubscribeEvent[] { + if (event.type === 'end') { + return [event] + } + const lists: SideFields[] = [ + ...(side.queued === undefined ? [] : [boundedQueuePublication(side.queued)]), + ...(side.backgroundTasks === undefined + ? [] + : [{ backgroundTasks: boundedBackgroundTasks(side.backgroundTasks) }]) + ] + const history = boundedHistoryFrame(event) + const carrier = (fields: SideFields): BatchFrame => ({ + type: 'batch', + sessionId: event.sessionId, + batch: emptyAgentSessionBatch(cursorAfter(event)), + ...(event.fence === undefined ? {} : { fence: event.fence }), + ...(event.hostNow === undefined ? {} : { hostNow: event.hostNow }), + ...fields + }) + if (lists.length > 0 && isEmptyBatch(history)) { + const [first, ...others] = lists + return [{ ...history, ...first }, ...others.map(carrier)] + } + return [history, ...lists.map(carrier)] +} diff --git a/src/main/native-chat/agent-session-wire/agent-session-bounded-queue.ts b/src/main/native-chat/agent-session-wire/agent-session-bounded-queue.ts new file mode 100644 index 00000000000..eab7fc5fbfc --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-bounded-queue.ts @@ -0,0 +1,26 @@ +// The draft queue as a bounded-frame subscriber receives it: whole, or when the list outgrows the +// rows' share of a frame, the cards it sends first, whole; the rest show once those leave. A card is +// never shortened: a reader that edits one replaces the draft with what it holds. + +import { BOUNDED_FRAME_CONTENT_BUDGET_BYTES } from './agent-session-history-page-bounds' +import { jsonBytes } from './json-string-shortening' +import type { QueuePublication } from './structured-agent-session-queued-publication' + +export function boundedQueuePublication(publication: QueuePublication): QueuePublication { + const cards = publication.queuedMessages + if (jsonBytes(cards) <= BOUNDED_FRAME_CONTENT_BUDGET_BYTES) { + return publication + } + // The longest run from the front that fits. + let fits = 0 + let fitsNot = cards.length + while (fitsNot - fits > 1) { + const middle = Math.floor((fits + fitsNot) / 2) + if (jsonBytes(cards.slice(0, middle)) <= BOUNDED_FRAME_CONTENT_BUDGET_BYTES) { + fits = middle + } else { + fitsNot = middle + } + } + return { ...publication, queuedMessages: cards.slice(0, fits) } +} diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-byte-accounting.test.ts b/src/main/native-chat/agent-session-wire/agent-session-history-byte-accounting.test.ts index 115e2248ea0..0569cdfe87a 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-history-byte-accounting.test.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-history-byte-accounting.test.ts @@ -94,7 +94,7 @@ it.each([1, 100, 200])('serializes each of %i unchanged forward page items once' limit: count, cursor }, - snapshot + { snapshot } ) expect(result.ok).toBe(true) expect(result.page.items).toHaveLength(count) diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-page-bounds.ts b/src/main/native-chat/agent-session-wire/agent-session-history-page-bounds.ts index 7157f1799b1..b0694726f78 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-history-page-bounds.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-history-page-bounds.ts @@ -7,6 +7,7 @@ import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' import { REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES } from '../../../shared/remote-runtime-memory-limits' +import { MOBILE_BRIDGE_MAX_FRAME_BYTES } from '../../../shared/mobile-bridge-frame-bytes' import { agentSessionHostStatusBody } from '../../../shared/agent-session-host-status-rows' import { subagentFoldKey, subagentItemBlock } from '../../../shared/native-chat-subagent-items' @@ -17,6 +18,14 @@ const HISTORY_PAGE_ENVELOPE_RESERVE_BYTES = 64 * 1024 export const HISTORY_PAGE_CONTENT_BUDGET_BYTES = AGENT_SESSION_HISTORY_MAX_PAGE_BYTES - HISTORY_PAGE_ENVELOPE_RESERVE_BYTES +/** Everything a bounded frame carries beside its rows: the bridge envelope, ids and cursors, the + * subagent roster, the latest turn and the activity line, each held to its own bound. */ +const FRAME_METADATA_CEILING_BYTES = 64 * 1024 + +/** The rows' share of a phone's bridge frame, which its app ends a stream over. */ +export const BOUNDED_FRAME_CONTENT_BUDGET_BYTES = + MOBILE_BRIDGE_MAX_FRAME_BYTES - FRAME_METADATA_CEILING_BYTES + export function historyEntryBytes( item: AgentJournalRenderItem, submissionBytes: ReadonlyMap @@ -56,16 +65,31 @@ export function oversizedHistoryItem(item: AgentJournalRenderItem): AgentJournal } } +/** How a reader receives each item, and the bytes its page may hold of them. A page is measured + * in that form, so it carries what it will actually send. */ +export type HistoryReader = { + form: (item: AgentJournalRenderItem) => AgentJournalRenderItem + budgetBytes: number +} + +export const STORED_HISTORY_READER: HistoryReader = { + form: (item) => item, + budgetBytes: HISTORY_PAGE_CONTENT_BUDGET_BYTES +} + +/** Rows are put in `form` only as the page takes them. */ export function boundHistoryItemsByBytes( items: AgentJournalRenderItem[], keep: 'newest' | 'oldest', submissionBytes: ReadonlyMap, - maxBytes: number + maxBytes: number, + form: HistoryReader['form'] = STORED_HISTORY_READER.form ): { items: AgentJournalRenderItem[]; dropped: number } { const ordered = groupItemsBySequence(items, keep) const kept: AgentJournalRenderItem[][] = [] let total = 0 - for (const group of ordered) { + for (const stored of ordered) { + const group = stored.map(form) const bytes = group.reduce((sum, item) => sum + historyEntryBytes(item, submissionBytes), 0) if (kept.length === 0 && bytes > maxBytes) { kept.push(group.map((item) => oversizedHistoryItem(item))) diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-page.ts b/src/main/native-chat/agent-session-wire/agent-session-history-page.ts index 64e9070d93c..80bbb4b5d7f 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-history-page.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-history-page.ts @@ -30,7 +30,8 @@ import type { AgentSessionJournal } from '../agent-session-journal/journal-store import { projectJournalBatch } from './agent-session-journal-batch' import { boundHistoryItemsByBytes, - HISTORY_PAGE_CONTENT_BUDGET_BYTES, + STORED_HISTORY_READER, + type HistoryReader, historyEntryBytes, newestWholeSequenceGroups, wholeSubagentGroupsStart, @@ -91,22 +92,29 @@ function isSubagentRow(item: AgentJournalRenderItem): boolean { export function readAgentSessionHistory( journal: AgentSessionJournal, request: AgentSessionHistoryRequest, - /** Reduced state to read against. A synchronous multi-page catch-up passes one - * snapshot for the whole run so each page costs its own rows, not the timeline. */ - snapshot: AgentJournalSnapshot = journal.snapshot(), - scope: AgentSessionHistoryScope = 'every-agent' + { + snapshot = journal.snapshot(), + scope = 'every-agent', + reader = STORED_HISTORY_READER + }: { + /** Reduced state to read against. A synchronous multi-page catch-up passes one + * snapshot for the whole run so each page costs its own rows, not the timeline. */ + snapshot?: AgentJournalSnapshot + scope?: AgentSessionHistoryScope + reader?: HistoryReader + } = {} ): AgentSessionHistoryResult { const limit = resolveHistoryLimit(request.limit) if (request.direction === 'after') { - return readForward(journal, snapshot, request.cursor, limit) + return readForward(journal, snapshot, request.cursor, limit, reader) } const cursor = request.direction === 'before' ? request.cursor : undefined if (cursor) { if (cursor.epoch !== snapshot.cursor.epoch) { - return historyReset(snapshot, 'epoch_changed') + return historyReset(snapshot, 'epoch_changed', reader) } if (cursor.sequence > snapshot.cursor.sequence) { - return historyReset(snapshot, 'cursor_ahead') + return historyReset(snapshot, 'cursor_ahead', reader) } } const scoped = @@ -117,7 +125,8 @@ export function readAgentSessionHistory( windowed, 'newest', submissionBytesByItemId(snapshot.submissions), - HISTORY_PAGE_CONTENT_BUDGET_BYTES + reader.budgetBytes, + reader.form ) return { ok: true, @@ -142,7 +151,8 @@ export function readAgentSessionHistory( * check re-reduces if anything did advance the journal between pages. */ export function createAgentSessionCatchUpReader( - journal: AgentSessionJournal + journal: AgentSessionJournal, + reader: HistoryReader = STORED_HISTORY_READER ): (request: AgentSessionHistoryRequest) => AgentSessionHistoryResult { let snapshot = journal.snapshot() return (request) => { @@ -150,27 +160,30 @@ export function createAgentSessionCatchUpReader( if (live.epoch !== snapshot.cursor.epoch || live.sequence !== snapshot.cursor.sequence) { snapshot = journal.snapshot() } - return readAgentSessionHistory(journal, request, snapshot) + return readAgentSessionHistory(journal, request, { snapshot, reader }) } } export function readAgentSessionHydrationPage( journal: AgentSessionJournal, - fence?: number + fence?: number, + reader: HistoryReader = STORED_HISTORY_READER ): AgentSessionHistoryPage { - return buildHydrationPage(journal.snapshot(), fence) + return buildHydrationPage(journal.snapshot(), fence, reader) } function buildHydrationPage( snapshot: AgentJournalSnapshot, - fence?: number + fence: number | undefined, + reader: HistoryReader ): AgentSessionHistoryPage { const items = conversationWindow(snapshot.items, AGENT_SESSION_HISTORY_MAX_LIMIT) const bounded = boundHistoryItemsByBytes( items, 'newest', submissionBytesByItemId(snapshot.submissions), - HISTORY_PAGE_CONTENT_BUDGET_BYTES + reader.budgetBytes, + reader.form ) return buildPage({ snapshot, @@ -189,12 +202,13 @@ function buildHydrationPage( function historyReset( snapshot: AgentJournalSnapshot, - reset: Extract['reset'] + reset: Extract['reset'], + reader: HistoryReader ): AgentSessionHistoryResult { return { ok: false, reset, - page: buildHydrationPage(snapshot) + page: buildHydrationPage(snapshot, undefined, reader) } } @@ -202,18 +216,19 @@ function readForward( journal: AgentSessionJournal, snapshot: AgentJournalSnapshot, cursor: AgentJournalCursor | undefined, - limit: number + limit: number, + reader: HistoryReader ): AgentSessionHistoryResult { if (!cursor) { // Why: forward paging replays rows after a position; without one there is // nothing to be after, and silently serving the tail would hand the client // a page it cannot place. - return historyReset(snapshot, 'cursor_ahead') + return historyReset(snapshot, 'cursor_ahead', reader) } // One lookahead preserves hasNewer without rereading the entire remaining journal per page. const since = journal.readSince(cursor, limit + 1) if (!since.ok) { - return historyReset(snapshot, since.reset) + return historyReset(snapshot, since.reset, reader) } const submissionBytes = submissionBytesByItemId(snapshot.submissions) // The page cost is EVERYTHING variable it carries: items with their @@ -232,46 +247,47 @@ function readForward( // clipping projected items: dropping an item while advancing the cursor past // the rows that touched it would lose that revision for good. let rows = since.rows.slice(0, limit) - let projected = projectJournalBatch({ - rows, - snapshot, - afterSequence: cursor.sequence, - canonicalItemId: (itemId) => journal.canonicalItemId(itemId) - }) - if (!projected.ok) { - return historyReset(snapshot, projected.reset) - } - let contentBytes = pageContentBytes(projected.batch.items, projected.batch.removedItemIds) - while (rows.length > 1 && contentBytes > HISTORY_PAGE_CONTENT_BUDGET_BYTES) { - rows = rows.slice(0, Math.ceil(rows.length / 2)) - const shrunk = projectJournalBatch({ - rows, + const project = (window: typeof rows): ReturnType => { + const result = projectJournalBatch({ + rows: window, snapshot, afterSequence: cursor.sequence, canonicalItemId: (itemId) => journal.canonicalItemId(itemId) }) + return result.ok + ? { ...result, batch: { ...result.batch, items: result.batch.items.map(reader.form) } } + : result + } + let projected = project(rows) + if (!projected.ok) { + return historyReset(snapshot, projected.reset, reader) + } + let contentBytes = pageContentBytes(projected.batch.items, projected.batch.removedItemIds) + while (rows.length > 1 && contentBytes > reader.budgetBytes) { + rows = rows.slice(0, Math.ceil(rows.length / 2)) + const shrunk = project(rows) if (!shrunk.ok) { - return historyReset(snapshot, shrunk.reset) + return historyReset(snapshot, shrunk.reset, reader) } projected = shrunk contentBytes = pageContentBytes(projected.batch.items, projected.batch.removedItemIds) } // One row can still touch an over-budget item; degrade it visibly. let items = projected.batch.items - if (contentBytes > HISTORY_PAGE_CONTENT_BUDGET_BYTES) { + if (contentBytes > reader.budgetBytes) { items = items.map((item) => { const bytes = historyEntryBytes(item, submissionBytes) - return bytes > HISTORY_PAGE_CONTENT_BUDGET_BYTES ? oversizedHistoryItem(item) : item + return bytes > reader.budgetBytes ? oversizedHistoryItem(item) : item }) contentBytes = pageContentBytes(items, projected.batch.removedItemIds) } - if (contentBytes > HISTORY_PAGE_CONTENT_BUDGET_BYTES) { + if (contentBytes > reader.budgetBytes) { // A single row's semantic payload — in practice a pre-bounding oversized // removal id — can never fit any page, and truncating a removal id would // break the client's keying. A bounded tail replaces the client's state // wholesale, which applies the removal without carrying the id, and the // client resumes from the live cursor past this row. - return historyReset(snapshot, 'cursor_compacted') + return historyReset(snapshot, 'cursor_compacted', reader) } const lastSequence = rows.at(-1)?.seq ?? cursor.sequence return { diff --git a/src/main/native-chat/agent-session-wire/agent-session-stop-note-projection.test.ts b/src/main/native-chat/agent-session-wire/agent-session-stop-note-projection.test.ts index d63ce1c4d1c..54845667927 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-stop-note-projection.test.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-stop-note-projection.test.ts @@ -24,6 +24,7 @@ import { readAgentSessionHistory } from './agent-session-history-page' import { AgentSessionSubscribers } from './structured-agent-session-subscribers' import { structuredAgentSessionStopNoteIdentity } from './structured-agent-session-command-turn' import { HISTORY_PAGE_CONTENT_BUDGET_BYTES } from './agent-session-history-page-bounds' +import { MOBILE_BRIDGE_MAX_FRAME_BYTES } from '../../../shared/mobile-bridge-frame-bytes' const journals = createTrackedJournalOpener() const turnIdentity = { provider: 'orca', clientMessageId: 'turn-1' } as const @@ -387,3 +388,32 @@ it('refreshes at the exact tip using bounded frames and no durable writes', asyn ).toBe(true) expect(rows()).toEqual(stored) }) + +it('refreshes a phone within its frames, in bounded pages', async () => { + await turn('interrupted') + for (let index = 0; index < 205; index += 1) { + await journal.appendItem(structuredAgentSessionStopNoteIdentity(`op-${index}`), unconfirmed, { + fence: 1, + turnScope: scope + }) + } + const events: AgentSessionSubscribeEvent[] = [] + new AgentSessionSubscribers().open({ + id: 'phone', + sessionId: 'session-1', + journal, + fence: 1, + cursor: journal.cursor(), + emit: (event) => events.push(event), + boundedFrames: true + }) + const batches = events.flatMap((event) => (event.type === 'batch' ? [event.batch] : [])) + // The caught-up checkpoint, then the 205 notes in two pages. + expect(batches.filter((batch) => batch.items.length > 0)).toHaveLength(2) + expect(batches.flatMap((batch) => batch.items)).toHaveLength(205) + for (const event of events) { + expect(Buffer.byteLength(JSON.stringify(event), 'utf8')).toBeLessThan( + MOBILE_BRIDGE_MAX_FRAME_BYTES + ) + } +}) diff --git a/src/main/native-chat/agent-session-wire/agent-session-stop-note-refresh.ts b/src/main/native-chat/agent-session-wire/agent-session-stop-note-refresh.ts index 5f86048af0c..0c1610dd728 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-stop-note-refresh.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-stop-note-refresh.ts @@ -1,11 +1,9 @@ import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' import { AGENT_SESSION_HISTORY_MAX_LIMIT } from '../../../shared/agent-session-wire' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' -import { - HISTORY_PAGE_CONTENT_BUDGET_BYTES, - oversizedHistoryItem -} from './agent-session-history-page-bounds' +import { oversizedHistoryItem } from './agent-session-history-page-bounds' import { emptyAgentSessionBatch } from './agent-session-empty-batch' +import { subscriberHistoryReader } from './agent-session-bounded-frames' import type { SubscriberDeliveryPort } from './agent-session-subscriber-catch-up' import type { Subscriber } from './structured-agent-session-subscribers' import { isStructuredAgentSessionStopNote } from './structured-agent-session-command-turn' @@ -31,6 +29,7 @@ export function refreshDerivedStopNotes( return } const snapshot = journal.snapshot() + const { form, budgetBytes } = subscriberHistoryReader(subscriber) let pending: AgentJournalRenderItem[] = [] let bytes = 0 const emit = (): void => { @@ -57,14 +56,13 @@ export function refreshDerivedStopNotes( ) { continue } - const projectedBytes = Buffer.byteLength(JSON.stringify(item), 'utf8') - const bounded = - projectedBytes > HISTORY_PAGE_CONTENT_BUDGET_BYTES ? oversizedHistoryItem(item) : item + const formed = form(item) + const projectedBytes = Buffer.byteLength(JSON.stringify(formed), 'utf8') + const bounded = projectedBytes > budgetBytes ? oversizedHistoryItem(formed) : formed const itemBytes = Buffer.byteLength(JSON.stringify(bounded), 'utf8') + 1 if ( pending.length > 0 && - (pending.length === AGENT_SESSION_HISTORY_MAX_LIMIT || - bytes + itemBytes > HISTORY_PAGE_CONTENT_BUDGET_BYTES) + (pending.length === AGENT_SESSION_HISTORY_MAX_LIMIT || bytes + itemBytes > budgetBytes) ) { emit() } diff --git a/src/main/native-chat/agent-session-wire/agent-session-subscriber-catch-up.ts b/src/main/native-chat/agent-session-wire/agent-session-subscriber-catch-up.ts index df98d8ba902..3546a7a6b8f 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-subscriber-catch-up.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-subscriber-catch-up.ts @@ -12,8 +12,13 @@ import { import { sameJournalCursor } from '../agent-session-journal/journal-cursor' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { emptyAgentSessionBatch } from './agent-session-empty-batch' -import { createAgentSessionCatchUpReader } from './agent-session-history-page' import { + createAgentSessionCatchUpReader, + readAgentSessionHydrationPage +} from './agent-session-history-page' +import { subscriberHistoryReader } from './agent-session-bounded-frames' +import { + subscriberCommandsChanged, subscriberQueuedMessagesChanged, type SubscriberFieldHooks } from './agent-session-subscriber-frame-fields' @@ -54,7 +59,8 @@ export function deliverToSubscriber( emitCaughtUp(port, subscriber, emitCheckpoint, shared) return } - const readPage = createAgentSessionCatchUpReader(journal) + const reader = subscriberHistoryReader(subscriber) + const readPage = createAgentSessionCatchUpReader(journal, reader) while (true) { const result = readPage({ sessionId: subscriber.sessionId, @@ -81,6 +87,21 @@ export function deliverToSubscriber( emitCaughtUp(port, subscriber, emitCheckpoint, shared) return } + // A phone's app ends a stream that outruns its unacked window, so a gap of more than one page + // arrives as the newest page instead; older rows load on scroll. + if (page.hasNewer && subscriber.boundedFrames) { + const tail = readAgentSessionHydrationPage(journal, subscriber.fence, reader) + port.emit(subscriber, { + type: 'snapshot', + sessionId: subscriber.sessionId, + page: tail, + fence: subscriber.fence, + hostNow, + activity: port.activity(subscriber.sessionId) + }) + subscriber.cursor = tail.liveCursor ?? tail.window.nextCursor + return + } port.emit( subscriber, { @@ -116,9 +137,7 @@ function emitCaughtUp( activity?: AgentSessionTurnActivity | null } ): void { - const commandsChanged = - port.hooks.readCommands !== undefined && - (port.hooks.readCommands(subscriber.sessionId) ?? null) !== subscriber.commands + const commandsChanged = subscriberCommandsChanged(port.hooks, subscriber) const queuedChanged = subscriberQueuedMessagesChanged(port.hooks, subscriber) if (emitCheckpoint || shared.activity !== undefined || commandsChanged || queuedChanged) { port.emit(subscriber, { 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 07e197ceb6a..a5535ea294a 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 @@ -10,9 +10,12 @@ import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire' import type { QueuePublication } from './structured-agent-session-queued-publication' +import { boundedFrames } from './agent-session-bounded-frames' export type SubscriberFieldState = { sessionId: string + /** Every frame to fit a phone's bridge frame; see `boundedFrames`. */ + boundedFrames: boolean commands?: AgentSessionSlashCommand[] | null /** The last queue publication actually SENT. */ queuePublication?: QueuePublication @@ -28,8 +31,10 @@ export type SubscriberFieldHooks = { } export type SubscriberFrame = { - frame: AgentSessionSubscribeEvent - commands: AgentSessionSlashCommand[] | null + /** In order: one for most subscribers; for a bounded-frame one, its side lists follow. */ + frames: AgentSessionSubscribeEvent[] + /** Undefined for a bounded-frame subscriber, which is never sent the list. */ + commands: AgentSessionSlashCommand[] | null | undefined attachedQueued: boolean queued: QueuePublication | undefined /** The attached roster's fingerprint; undefined when the frame carries none. */ @@ -48,7 +53,9 @@ export function buildSubscriberFrame( event: AgentSessionSubscribeEvent, withholdQueued: boolean ): SubscriberFrame { - const commands = hooks.readCommands?.(subscriber.sessionId) ?? null + const commands = subscriber.boundedFrames + ? undefined + : (hooks.readCommands?.(subscriber.sessionId) ?? null) const includeCommands = hooks.readCommands !== undefined && event.type !== 'end' && @@ -61,19 +68,24 @@ export function buildSubscriberFrame( event.type !== 'end' && (event.type !== 'batch' || queued !== subscriber.queuePublication) const backgroundTasks = subscriberBackgroundTasks(hooks, subscriber, event) + const sideQueued = attachedQueued ? queued : undefined return { - frame: { - ...event, - ...(includeCommands ? { commands: commands ?? null } : {}), - ...(attachedQueued && queued - ? { - queuedMessages: queued.queuedMessages, - queuePause: queued.queuePause, - nextQueuedMessageId: queued.nextQueuedMessageId + frames: subscriber.boundedFrames + ? boundedFrames(event, { queued: sideQueued, backgroundTasks }) + : [ + { + ...event, + ...(includeCommands ? { commands: commands ?? null } : {}), + ...(sideQueued + ? { + queuedMessages: sideQueued.queuedMessages, + queuePause: sideQueued.queuePause, + nextQueuedMessageId: sideQueued.nextQueuedMessageId + } + : {}), + ...(backgroundTasks !== undefined ? { backgroundTasks } : {}) } - : {}), - ...(backgroundTasks !== undefined ? { backgroundTasks } : {}) - }, + ], commands, attachedQueued, queued, @@ -101,6 +113,19 @@ function subscriberBackgroundTasks( : undefined } +/** Whether the `/` list changed since this subscriber was last sent it. A bounded-frame subscriber + * is never sent it: a phone reads it from `agentSession.options`. */ +export function subscriberCommandsChanged( + hooks: SubscriberFieldHooks, + subscriber: SubscriberFieldState +): boolean { + return ( + !subscriber.boundedFrames && + hooks.readCommands !== undefined && + (hooks.readCommands(subscriber.sessionId) ?? null) !== subscriber.commands + ) +} + /** Whether a caught-up publish with no rows still owes this subscriber a frame: * draft inserts and pause changes write no journal row, so an unchanged cursor * must still deliver the changed publication. */ diff --git a/src/main/native-chat/agent-session-wire/json-string-shortening.test.ts b/src/main/native-chat/agent-session-wire/json-string-shortening.test.ts new file mode 100644 index 00000000000..d1c3d6bf695 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/json-string-shortening.test.ts @@ -0,0 +1,28 @@ +import { expect, it } from 'vitest' +import { jsonBytes, shortenJsonStringsToFit } from './json-string-shortening' + +it('returns the value itself when it already fits', () => { + const value = { text: 'short' } + expect(shortenJsonStringsToFit(value, 1_000)).toBe(value) +}) + +it('cuts long strings to fit and marks them, never splitting a surrogate pair', () => { + const value = { text: '😀'.repeat(2_000), other: 'x'.repeat(4_000) } + const fitted = shortenJsonStringsToFit(value, 2_000) + expect(fitted).not.toBeNull() + expect(jsonBytes(fitted)).toBeLessThanOrEqual(2_000) + expect(fitted?.text.endsWith('…')).toBe(true) + // A lone high surrogate would serialize as an escape and render as a broken glyph. + expect(fitted?.text).not.toMatch(/[\uD800-\uDBFF](?![\uDC00-\uDFFF])/) +}) + +it('leaves identifiers whole', () => { + const id = 'i'.repeat(3_000) + const fitted = shortenJsonStringsToFit({ id, itemId: id, text: 'x'.repeat(3_000) }, 7_000) + expect(fitted).toMatchObject({ id, itemId: id }) +}) + +it('is null when even the shortest cut cannot fit', () => { + const many = Array.from({ length: 1_000 }, () => 'x'.repeat(200)) + expect(shortenJsonStringsToFit(many, 1_000)).toBeNull() +}) diff --git a/src/main/native-chat/agent-session-wire/json-string-shortening.ts b/src/main/native-chat/agent-session-wire/json-string-shortening.ts new file mode 100644 index 00000000000..a53a3e15dc7 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/json-string-shortening.ts @@ -0,0 +1,109 @@ +// Fits a JSON value under a byte budget by cutting its long strings to one common length: the +// longest length that fits, so short strings keep everything and long ones keep what room allows. +// +// Shortening never touches an identifier (`id` or a key ending in `Id`): a reader keys and answers +// by them. A bounded payload's `head` is cut and flagged through its own `truncated` field; any +// other string ends in `…`, the mark the chat already draws for text it did not show in full. + +import type { AgentJournalBoundedPayload } from '../../../shared/agent-session-journal-types' + +const ELLIPSIS = '…' +const ELLIPSIS_BYTES = Buffer.byteLength(ELLIPSIS, 'utf8') +/** Strings this short are left alone: cutting them saves little and reads as noise. */ +const KEEP_AT_LEAST = 64 + +export function jsonBytes(value: unknown): number { + return Buffer.byteLength(JSON.stringify(value), 'utf8') +} + +function isIdentifierKey(key: string): boolean { + return key === 'id' || key.endsWith('Id') +} + +function isBoundedPayload(value: unknown): value is AgentJournalBoundedPayload { + return ( + typeof value === 'object' && + value !== null && + 'head' in value && + typeof value.head === 'string' && + 'byteLength' in value && + 'truncated' in value + ) +} + +/** The first `length` UTF-16 units of `text`, never splitting a surrogate pair. */ +function headOf(text: string, length: number): string { + const last = text.charCodeAt(length - 1) + return text.slice(0, last >= 0xd800 && last <= 0xdbff ? length - 1 : length) +} + +/** + * `value` with every shortenable string over `length` cut to it; `value` itself when none is. A + * string is cut only when that removes more units than its mark adds bytes, so a smaller `length` + * never yields a larger value: every unit is at least one byte, and the mark is three. + */ +function cutLongStrings(value: T, length: number): T { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: `cutTo` keeps every key and shape and only shortens strings. + return cutTo(value, length, '') as T +} + +function cutTo(value: unknown, length: number, key: string): unknown { + if (typeof value === 'string') { + return value.length > length + ELLIPSIS_BYTES && !isIdentifierKey(key) + ? `${headOf(value, length)}${ELLIPSIS}` + : value + } + if (typeof value !== 'object' || value === null) { + return value + } + if (isBoundedPayload(value)) { + return value.head.length > length + ? { ...value, head: headOf(value.head, length), truncated: true } + : value + } + const entries = Object.entries(value) + const cut = entries.map(([entryKey, entry]) => cutTo(entry, length, entryKey)) + if (cut.every((entry, index) => entry === entries[index]![1])) { + return value + } + return Array.isArray(value) + ? cut + : Object.fromEntries(entries.map(([entryKey], index) => [entryKey, cut[index]])) +} + +function longestStringLength(value: unknown): number { + if (typeof value === 'string') { + return value.length + } + if (typeof value !== 'object' || value === null) { + return 0 + } + return Object.values(value).reduce( + (longest, entry) => Math.max(longest, longestStringLength(entry)), + 0 + ) +} + +/** `value` within `maxBytes`, or null when cutting every long string to `KEEP_AT_LEAST` is not + * enough. `value` itself when it already fits; never modified. */ +export function shortenJsonStringsToFit(value: T, maxBytes: number): T | null { + if (jsonBytes(value) <= maxBytes) { + return value + } + const at = (length: number): T => cutLongStrings(value, length) + if (jsonBytes(at(KEEP_AT_LEAST)) > maxBytes) { + return null + } + // The longest common length that fits; `low` always fits, `high` never does. + let low = KEEP_AT_LEAST + let high = longestStringLength(value) + while (high - low > 1) { + const middle = Math.floor((low + high) / 2) + if (jsonBytes(at(middle)) <= maxBytes) { + low = middle + } else { + high = middle + } + } + return at(low) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts index 49fb06e83ba..715a4a60422 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts @@ -24,7 +24,7 @@ export function readStructuredAgentSessionHistoryResult(input: { request: AgentSessionHistoryRequest scope?: AgentSessionHistoryScope }): AgentSessionHistoryResult { - const result = readAgentSessionHistory(input.journal, input.request, undefined, input.scope) + const result = readAgentSessionHistory(input.journal, input.request, { scope: input.scope }) const fence = input.record?.lease.runtimeFence const providerSession = structuredAgentSessionProviderSessionMetadata(input.record) if (fence === undefined) { 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 dcd3e07c4f2..705592d414f 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 @@ -6,6 +6,7 @@ import type { AgentJournalCursor } from '../../../shared/agent-session-journal-types' import type { + AgentSessionHistoryPage, AgentSessionSlashCommand, AgentSessionSubscribeEvent, AgentSessionTurnActivity @@ -13,6 +14,7 @@ import type { import { backgroundTaskFingerprint, buildSubscriberFrame, + subscriberCommandsChanged, type SubscriberFieldHooks } from './agent-session-subscriber-frame-fields' import type { QueuePublication } from './structured-agent-session-queued-publication' @@ -20,6 +22,7 @@ import { deliverToSubscriber } from './agent-session-subscriber-catch-up' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { emptyAgentSessionBatch } from './agent-session-empty-batch' import { readAgentSessionHydrationPage } from './agent-session-history-page' +import { subscriberHistoryReader } from './agent-session-bounded-frames' import { rememberSessionActivity } from './structured-agent-session-activity-retention' import { refreshDerivedStopNotes } from './agent-session-stop-note-refresh' @@ -29,6 +32,8 @@ export type AgentSessionSubscribeInput = { sessionId: string emit: AgentSessionSubscriberEmit cursor?: AgentJournalCursor + /** Every frame to fit a phone's bridge frame; the RPC sets it for every phone. */ + boundedFrames?: boolean } export type Subscriber = { @@ -37,6 +42,7 @@ export type Subscriber = { emit: AgentSessionSubscriberEmit cursor: AgentJournalCursor fence: number + boundedFrames: boolean 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. */ @@ -77,6 +83,7 @@ export class AgentSessionSubscribers { fence: number emit: AgentSessionSubscriberEmit cursor?: AgentJournalCursor + boundedFrames?: boolean }): () => void { const liveCursor = input.journal.cursor() const subscriber: Subscriber = { @@ -84,7 +91,8 @@ export class AgentSessionSubscribers { sessionId: input.sessionId, emit: input.emit, cursor: input.cursor ?? { epoch: liveCursor.epoch, sequence: 0 }, - fence: input.fence + fence: input.fence, + boundedFrames: input.boundedFrames === true } const session = this.bySession.get(input.sessionId) ?? new Map() session.set(input.id, subscriber) @@ -103,7 +111,11 @@ export class AgentSessionSubscribers { hostNow ) } else { - const page = readAgentSessionHydrationPage(input.journal, input.fence) + const page = readAgentSessionHydrationPage( + input.journal, + input.fence, + subscriberHistoryReader(subscriber) + ) this.emit(subscriber, { type: 'snapshot', sessionId: input.sessionId, @@ -155,9 +167,18 @@ export class AgentSessionSubscribers { /** Every subscriber back to a bounded tail page. */ snapshot(sessionId: string, journal: AgentSessionJournal, fence: number): void { - const page = readAgentSessionHydrationPage(journal, fence) + // Read once per item form, since bounded-frame subscribers size their page by theirs. + let storedPage: AgentSessionHistoryPage | undefined + let boundedPage: AgentSessionHistoryPage | undefined + const read = (subscriber: Subscriber): AgentSessionHistoryPage => + readAgentSessionHydrationPage(journal, fence, subscriberHistoryReader(subscriber)) + const pageFor = (subscriber: Subscriber): AgentSessionHistoryPage => + subscriber.boundedFrames + ? (boundedPage ??= read(subscriber)) + : (storedPage ??= read(subscriber)) const hostNow = this.now() for (const subscriber of this.subscribers(sessionId)) { + const page = pageFor(subscriber) this.emit(subscriber, { type: 'snapshot', sessionId, @@ -202,7 +223,7 @@ export class AgentSessionSubscribers { const hostNow = this.now() for (const sessionId of this.bySession.keys()) { for (const subscriber of this.subscribers(sessionId)) { - if ((this.hooks.readCommands?.(sessionId) ?? null) !== subscriber.commands) { + if (subscriberCommandsChanged(this.hooks, subscriber)) { this.emit(subscriber, { type: 'batch', sessionId, @@ -256,7 +277,9 @@ export class AgentSessionSubscribers { event, options?.withholdQueued === true ) - subscriber.emit(built.frame) + for (const frame of built.frames) { + subscriber.emit(frame) + } subscriber.commands = built.commands if (built.attachedQueued) { subscriber.queuePublication = built.queued diff --git a/src/main/runtime/rpc/methods/structured-agent-session-bounded-frames-rpc.test.ts b/src/main/runtime/rpc/methods/structured-agent-session-bounded-frames-rpc.test.ts new file mode 100644 index 00000000000..0d96a3861ac --- /dev/null +++ b/src/main/runtime/rpc/methods/structured-agent-session-bounded-frames-rpc.test.ts @@ -0,0 +1,33 @@ +// Every phone gets bounded frames, whatever version its shell or page is; other clients keep the +// frames they had, `commands` included. + +import { afterEach, beforeEach, expect, it } from 'vitest' +import type { AgentSessionSubscribeInput } from '../../../native-chat/agent-session-wire/structured-agent-session-subscribers' +import { + call, + clearStructuredHostStub, + hostCalls, + installStructuredHostStub, + SESSION, + STRUCTURED_CLIENT, + STRUCTURED_MOBILE_CLIENT +} from './structured-agent-session-rpc.test-fixture' + +beforeEach(installStructuredHostStub) +afterEach(clearStructuredHostStub) + +it.each([ + // Being a phone is enough: no capability or param is asked of it. + ['a phone of any version', STRUCTURED_MOBILE_CLIENT, true], + ['a paired desktop', STRUCTURED_CLIENT, false], + ['an in-process reader', undefined, false] +] as const)('subscribes %s with bounded frames: %s', async (_label, client, bounded) => { + let input: AgentSessionSubscribeInput | undefined + hostCalls.subscribe.mockImplementation((opened: AgentSessionSubscribeInput) => { + input = opened + opened.emit({ type: 'end' }) + return () => {} + }) + await call('agentSession.subscribe', { sessionId: SESSION }, client) + expect(input?.boundedFrames === true).toBe(bounded) +}) diff --git a/src/main/runtime/rpc/methods/structured-agent-session.ts b/src/main/runtime/rpc/methods/structured-agent-session.ts index 37b9e62f960..a25d40b52b2 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.ts @@ -181,7 +181,9 @@ export const STRUCTURED_AGENT_SESSION_METHODS = [ host.sessionAgent(params.sessionId) ) ), - ...(params.cursor ? { cursor: params.cursor } : {}) + ...(params.cursor ? { cursor: params.cursor } : {}), + // Every phone's app, whatever its version, ends a stream over a frame its bridge cannot carry. + ...(ctx.clientKind === 'mobile' ? { boundedFrames: true } : {}) }) if (stream.isClosed()) { dispose() diff --git a/src/shared/mobile-bridge-frame-bytes.ts b/src/shared/mobile-bridge-frame-bytes.ts new file mode 100644 index 00000000000..10980a8cd3c --- /dev/null +++ b/src/shared/mobile-bridge-frame-bytes.ts @@ -0,0 +1,4 @@ +/** The mobile app's page <-> shell frame ceiling, in UTF-8 bytes. The installed app ends any stream + * whose event frame is larger, and released apps keep this value, so hosts size streamed frames to + * fit it rather than to the relay's larger cap. */ +export const MOBILE_BRIDGE_MAX_FRAME_BYTES = 640 * 1024