diff --git a/.changeset/681-bound-invocation-render-history.md b/.changeset/681-bound-invocation-render-history.md new file mode 100644 index 000000000..c9b2a0310 --- /dev/null +++ b/.changeset/681-bound-invocation-render-history.md @@ -0,0 +1,5 @@ +--- +'agent-bundle': patch +--- + +Bound every retained copy of a Workbench route invocation's render stream under one render-history window — at most 256 events and 2 MiB serialized, always keeping the newest event and the newest `shell`/`replace`/`complete` event, which alone may exceed 2 MiB — applied alike to the live `GET /api/routes/invocations//stream` replay, the completed and cancelled envelopes, `GET /api/routes/invocations/`, the terminal `final` message, the Workbench's live view, and the envelopes the Workbench synthesizes from runtime runs. Stream replay is paced by socket drain instead of queued, so a reconnect can take a whole retained window through a backpressured socket. The render child no longer returns the whole event stream over IPC and the compiled-route producer keeps only the `complete` document. Envelopes whose events were evicted carry a new optional `retention` field (`producedEvents`, `evictedEvents`, `evictedBytes`, `retainedBytes`), cancelled and failed envelopes keep the retained window and its latest document, compiled routes' progress reports now reach the Workbench as `progress` render events, and `document`, `result`, `outcome`, and `correlationId` are never truncated. (#715) diff --git a/packages/agent-bundle/src/contracts/invocations.ts b/packages/agent-bundle/src/contracts/invocations.ts index b6be95b2c..eb9e8d411 100644 --- a/packages/agent-bundle/src/contracts/invocations.ts +++ b/packages/agent-bundle/src/contracts/invocations.ts @@ -1,7 +1,8 @@ /** * Browser-consumable contract surface for dev-server route invocations — the - * one execution path behind the Workbench route workspace. Type-only: routes - * render on the server through the production runtime. + * one execution path behind the Workbench route workspace. Types, plus the + * render-history retention policy the browser's live window shares with the + * server; routes render on the server through the production runtime. */ export type { RouteInvocationCliProjection, @@ -29,4 +30,15 @@ export type { RouteInvocationStreamMessage, RunningRouteInvocationResponse, } from '../dev/routes/route-invocation-result.ts'; +export { + emptyRetainedRenderEvents, + renderRetention, + retainedLatestDocument, + retainedRenderEvents, + retainRenderEvent, + routeInvocationRenderHistoryLimits, + type RetainedRenderEvents, + type RouteInvocationRenderHistoryLimits, + type RouteInvocationRenderRetention, +} from '../dev/routes/route-invocation-render-history.ts'; export type { EventTraceEvent } from '../events/trace.ts'; diff --git a/packages/agent-bundle/src/dev/routes/route-invocation-child.ts b/packages/agent-bundle/src/dev/routes/route-invocation-child.ts index 5eb4e25f3..52ca41a01 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-child.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-child.ts @@ -68,9 +68,9 @@ const forwardEventTrace = (event: EventTraceEvent): void => { process.send?.({ event, type: 'trace' } satisfies RouteInvocationChildResponse); }; -const forwardRenderEvent = (event: AgentRenderEvent): void => { - process.send?.({ event, type: 'render' } satisfies RouteInvocationChildResponse); -}; +/** Awaited per event so the IPC channel, not an in-child queue, paces a fast producer. */ +const forwardRenderEvent = (event: AgentRenderEvent): Promise => + respond({ event, type: 'render' }); /** * The exit code a generated executable would set for this unit render. There @@ -134,10 +134,12 @@ const renderUnitRoute = async (request: RouteInvocationChildRequest): Promise => { - const result = request.surface.kind === 'unit-render' - ? await renderUnitRoute(request) - : await renderProductionRoute(request, forwardEventTrace, forwardRenderEvent); - if (request.surface.kind === 'unit-render') { - for (const event of result.events) forwardRenderEvent(event); - } - return result; -}; +const render = (request: RouteInvocationChildRequest): Promise => + request.surface.kind === 'unit-render' + ? renderUnitRoute(request) + : renderProductionRoute(request, forwardEventTrace, forwardRenderEvent); process.once('message', (request: RouteInvocationChildRequest) => { const disposeTraceObserver = installEventTraceObserver(forwardEventTrace); diff --git a/packages/agent-bundle/src/dev/routes/route-invocation-production.ts b/packages/agent-bundle/src/dev/routes/route-invocation-production.ts index bcc1b5a0a..ee17c2495 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-production.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-production.ts @@ -10,6 +10,9 @@ import { createAgentRenderDispatcher, documentToCallToolResult, type AgentDocument, + type AgentProgressReporter, + type AgentProgressUpdate, + type AgentRenderDispatch, type AgentRenderEvent, type AgentRenderInvocation, } from '@agent-bundle/runtime'; @@ -287,6 +290,7 @@ const streamFromWorker = ( readonly abort: () => void; readonly controller: ReadableStreamDefaultController; readonly dispatchSignal: AbortSignal; + readonly progress: AgentProgressReporter | undefined; }>(); const failAll = (error: Error): void => { for (const [id, entry] of pending) { @@ -302,7 +306,18 @@ const streamFromWorker = ( worker.on('message', (message: WorkerMessage) => { const entry = pending.get(message.id); if (entry === undefined) return; - if (message.type === 'progress') return; + if (message.type === 'progress') { + // The route's reported progress becomes `progress` render events through + // the dispatcher's reporter, as the generated CLI session forwards it. + Promise.resolve() + .then(() => entry.progress?.report(message.update as AgentProgressUpdate)) + .catch((error: unknown) => { + pending.delete(message.id); + entry.dispatchSignal.removeEventListener('abort', entry.abort); + entry.controller.error(error); + }); + return; + } if (message.type === 'observed-providers-start') { trace?.providersStart(); return; @@ -372,10 +387,7 @@ const streamFromWorker = ( entry.controller.error(new Error(message.message ?? 'Compiled route worker failed.')); }); const host = Object.freeze({ - execute: async (dispatch: Readonly<{ - readonly invocation: AgentRenderInvocation; - readonly signal: AbortSignal; - }>): Promise> => { + execute: async (dispatch: AgentRenderDispatch): Promise> => { const id = ++sequence; let controller!: ReadableStreamDefaultController; const cancelRender = (): void => { @@ -396,7 +408,7 @@ const streamFromWorker = ( }, start: (opened) => { controller = opened; }, }); - pending.set(id, { abort, controller, dispatchSignal: dispatch.signal }); + pending.set(id, { abort, controller, dispatchSignal: dispatch.signal, progress: dispatch.progress }); dispatch.signal.addEventListener('abort', abort, { once: true }); worker.postMessage({ actor: request.context.actor, @@ -448,17 +460,21 @@ const missingRouteWorkerError = (error: unknown): boolean => || error.message.includes('Generated rendered route must default-export') ); +/** + * Drives one compiled worker's render stream. Each event is handed to + * `publishRender` as it arrives and then dropped; only the `complete` event's + * document is kept, so the producer holds one document, not the stream. + */ const renderCompiled = async ( request: ProductionRequest, input: JsonValue, signal: AbortSignal, env: NodeJS.ProcessEnv, trace?: EventTracer, - publishRender?: (event: AgentRenderEvent) => void, + publishRender?: (event: AgentRenderEvent) => Promise | void, ): Promise event.type === 'complete'); - if (complete === undefined) throw new Error('Compiled route render ended without a complete event.'); + if (document === undefined) throw new Error('Compiled route render ended without a complete event.'); return Object.freeze({ - document: complete.document, + document, durationMs: performance.now() - startedAt, - events: Object.freeze(events), observed: { providers: Object.freeze([...session.observed.providers]), timings: Object.freeze([...session.observed.timings]), @@ -504,7 +518,7 @@ const renderCompiled = async ( export const renderProductionRoute = async ( request: RouteInvocationChildRequest, publishTrace?: EventTraceObserver, - publishRender?: (event: AgentRenderEvent) => void, + publishRender?: (event: AgentRenderEvent) => Promise | void, ): Promise => { if (request.artifactEpoch === undefined || request.artifactRoot === undefined) { throw new ProductionRouteInvocationError( @@ -539,7 +553,6 @@ export const renderProductionRoute = async ( const value = prepared.preflight.gate as JsonValue; return Object.freeze({ document: completeDocument(value), - events: Object.freeze([]), input: prepared.input, result: value, trace: Object.freeze(traceEvents), @@ -569,7 +582,6 @@ export const renderProductionRoute = async ( : undefined; return Object.freeze({ document: rendered.document, - events: rendered.events, ...(exitCode === undefined ? {} : { exitCode }), input: prepared.input, ...(kind === 'tool' diff --git a/packages/agent-bundle/src/dev/routes/route-invocation-render-history.ts b/packages/agent-bundle/src/dev/routes/route-invocation-render-history.ts new file mode 100644 index 000000000..b164dd736 --- /dev/null +++ b/packages/agent-bundle/src/dev/routes/route-invocation-render-history.ts @@ -0,0 +1,143 @@ +/** + * The one retention policy for a route invocation's render-event history. + * + * Every place a render stream is retained — the service's live replay buffer, + * the completed envelope (`events`), `GET /api/routes/invocations/`, the + * terminal `final` stream message, and the Workbench's live window — applies + * this window, so no reader can rehydrate what another evicted. The window is + * bounded by event count and by serialized bytes; the newest event and the + * newest document-bearing event (`shell`, `replace`, or `complete`) are never + * evicted, so a reader always folds to a coherent latest document. Everything + * older is disposable intermediate history. The final Agent Document itself + * is retained separately on the envelope (`document`) and is never truncated + * here: the runtime bounds it (`maxDocumentBytes`) before it reaches us. + * + * Browser-safe: no Node imports. The Workbench reaches it through + * `contracts/invocations.ts`. + */ +import type { AgentDocument, AgentRenderEvent } from '@agent-bundle/runtime'; + +import { deepFreeze } from '../../core/freeze.ts'; + +export interface RouteInvocationRenderHistoryLimits { + /** Serialized UTF-8 bytes of the retained events, summed. */ + readonly maxBytes: number; + readonly maxEvents: number; +} + +/** + * 256 events and 2 MiB. The window exceeds `maxBytes` only by what its two + * pinned events need; the runtime caps one event at `maxEventBytes` + * (1 MiB + 1 KiB), so a runtime-produced window never holds more than + * 2 MiB + 2 KiB. + */ +export const routeInvocationRenderHistoryLimits: RouteInvocationRenderHistoryLimits = Object.freeze({ + maxBytes: 2 * 1024 * 1024, + maxEvents: 256, +}); + +/** What the policy evicted from a completed run; present on the envelope only when something was. */ +export interface RouteInvocationRenderRetention { + /** Serialized bytes of the evicted events. */ + readonly evictedBytes: number; + /** Render events evicted, oldest first; `events` holds the remaining `producedEvents - evictedEvents`. */ + readonly evictedEvents: number; + /** Every render event the run published, retained or not. */ + readonly producedEvents: number; + /** Serialized bytes of the retained `events`. */ + readonly retainedBytes: number; +} + +interface RetainedRenderEvent { + readonly bytes: number; + readonly event: AgentRenderEvent; +} + +/** An immutable retained window; `retainRenderEvent` derives the next one. */ +export interface RetainedRenderEvents { + readonly entries: readonly RetainedRenderEvent[]; + readonly evictedBytes: number; + readonly evictedEvents: number; + /** The newest `shell`, `replace`, or `complete` entry, pinned against eviction. */ + readonly latestDocument?: RetainedRenderEvent; + readonly producedEvents: number; + readonly retainedBytes: number; +} + +export const emptyRetainedRenderEvents: RetainedRenderEvents = Object.freeze({ + entries: Object.freeze([]), + evictedBytes: 0, + evictedEvents: 0, + producedEvents: 0, + retainedBytes: 0, +}); + +const encoder = new TextEncoder(); + +export const renderEventBytes = (event: AgentRenderEvent): number => + encoder.encode(JSON.stringify(event)).byteLength; + +const bearsDocument = (event: AgentRenderEvent): boolean => + event.type === 'shell' || event.type === 'replace' || event.type === 'complete'; + +/** + * Appends `event` and evicts the oldest disposable entries until the window + * fits both bounds again. The newest entry and the pinned document entry are + * never evicted, so a window can exceed `maxBytes` only when those two alone + * do. Returns the next window and the evicted events, oldest first. + */ +export const retainRenderEvent = ( + retained: RetainedRenderEvents, + event: AgentRenderEvent, + limits: RouteInvocationRenderHistoryLimits = routeInvocationRenderHistoryLimits, +): Readonly<{ readonly evicted: readonly AgentRenderEvent[]; readonly retained: RetainedRenderEvents }> => { + const entry: RetainedRenderEvent = Object.freeze({ bytes: renderEventBytes(event), event }); + const latestDocument = bearsDocument(event) ? entry : retained.latestDocument; + const entries = [...retained.entries, entry]; + const evicted: AgentRenderEvent[] = []; + let retainedBytes = retained.retainedBytes + entry.bytes; + let evictedBytes = retained.evictedBytes; + let index = 0; + while ((entries.length > limits.maxEvents || retainedBytes > limits.maxBytes) && index < entries.length - 1) { + const candidate = entries[index]!; + if (candidate === latestDocument) { + index += 1; + continue; + } + entries.splice(index, 1); + retainedBytes -= candidate.bytes; + evictedBytes += candidate.bytes; + evicted.push(candidate.event); + } + return Object.freeze({ + evicted: Object.freeze(evicted), + retained: Object.freeze({ + entries: Object.freeze(entries), + evictedBytes, + evictedEvents: retained.evictedEvents + evicted.length, + ...(latestDocument === undefined ? {} : { latestDocument }), + producedEvents: retained.producedEvents + 1, + retainedBytes, + }), + }); +}; + +export const retainedRenderEvents = (retained: RetainedRenderEvents): readonly AgentRenderEvent[] => + Object.freeze(retained.entries.map((entry) => entry.event)); + +/** The document a reader folds to after eviction: the pinned newest `shell`, `replace`, or `complete`. */ +export const retainedLatestDocument = (retained: RetainedRenderEvents): AgentDocument | undefined => { + const event = retained.latestDocument?.event; + return event === undefined || event.type === 'progress' || event.type === 'error' ? undefined : event.document; +}; + +/** The envelope's truncation indication; `undefined` while nothing has been evicted. */ +export const renderRetention = (retained: RetainedRenderEvents): RouteInvocationRenderRetention | undefined => + retained.evictedEvents === 0 + ? undefined + : deepFreeze({ + evictedBytes: retained.evictedBytes, + evictedEvents: retained.evictedEvents, + producedEvents: retained.producedEvents, + retainedBytes: retained.retainedBytes, + }); diff --git a/packages/agent-bundle/src/dev/routes/route-invocation-result.ts b/packages/agent-bundle/src/dev/routes/route-invocation-result.ts index 745d55ced..5098a1515 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-result.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-result.ts @@ -3,6 +3,7 @@ import type { AgentDocument, AgentRenderEvent } from '@agent-bundle/runtime'; import type { JsonValue } from '../../core/strict-json.ts'; import type { RequestContextProvenance } from '../../contracts/request-provenance.ts'; import type { EventTraceEvent } from '../../events/trace.ts'; +import type { RouteInvocationRenderRetention } from './route-invocation-render-history.ts'; import type { RunningRouteInvocation, RouteInvocationProjection, @@ -12,14 +13,25 @@ import type { export interface RouteInvocation extends RouteInvocationSummary { readonly context: RequestContextProvenance; - /** The final Agent Document; absent when rendering failed before a document existed. */ + /** + * The final Agent Document of a `succeeded` run. A `cancelled` or `failed` + * run carries the latest document its retained stream reached, absent when + * none did. Never truncated by `retention`. + */ readonly document?: AgentDocument; - /** The production `shell | progress | replace | error | complete` stream, in order. */ + /** + * The production `shell | progress | replace | error | complete` stream, in + * order, as retained by the render-history window + * (`routeInvocationRenderHistoryLimits`): the newest events plus the newest + * document-bearing event. Complete unless `retention` is present. + */ readonly events: readonly AgentRenderEvent[]; readonly projection: RouteInvocationProjection; readonly providers: readonly RouteInvocationProvider[]; /** Structured value recorded by the selected surface; its presence alone proves neither a `resultSchema` declaration nor validation. */ readonly result?: JsonValue; + /** Present when the render-history window evicted events; the one truthful account of what `events` no longer holds. */ + readonly retention?: RouteInvocationRenderRetention; /** Event-kernel phase events emitted by a compiled preflight execution. */ readonly trace?: readonly EventTraceEvent[]; } diff --git a/packages/agent-bundle/src/dev/routes/route-invocation-routes.ts b/packages/agent-bundle/src/dev/routes/route-invocation-routes.ts index 8caef040c..a4593e90e 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-routes.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-routes.ts @@ -1,3 +1,4 @@ +import { Buffer } from 'node:buffer'; import type { IncomingMessage, ServerResponse } from 'node:http'; import type { ProjectEventHub } from '../events.ts'; @@ -27,6 +28,7 @@ import type { RouteInvocationListResponse, RouteInvocationRequest, } from './route-invocation.ts'; +import { routeInvocationRenderHistoryLimits } from './route-invocation-render-history.ts'; import { invocationSummary, parseRouteInvocationRequest, @@ -35,8 +37,29 @@ import { type RouteInvocationRequestError, } from './route-invocation-service.ts'; -const streamQueueByteLimit = 256 * 1024; -const streamQueueEntryLimit = 128; +/** + * The live backlog one backpressured socket may hold before it is cut off. + * Replay never enters this queue (`#stream` paces it by drain), so the limit + * only has to fit what a live run can still send a slow consumer: the window + * as `render` frames plus the `final` envelope, which repeats that window + * beside a document, result, and projections the runtime each bounds near + * 1 MiB. + */ +const streamQueueByteLimit = 8 * routeInvocationRenderHistoryLimits.maxBytes; +const streamQueueEntryLimit = 2 * routeInvocationRenderHistoryLimits.maxEvents; + +interface PendingFrame { + readonly bytes: number; + readonly final: boolean; + /** Arrived after the replay snapshot was taken. */ + readonly live: boolean; + readonly text: string; +} + +const pendingFrame = (message: RouteInvocationStreamMessage, live: boolean): PendingFrame => { + const text = `event: ${message.type}\ndata: ${JSON.stringify(message)}\n\n`; + return { bytes: Buffer.byteLength(text, 'utf8'), final: message.type === 'final', live, text }; +}; const invalidShape = badRequest( ROUTE_INVOCATION_MALFORMED_REQUEST_CODE, 'Route invocation request has an invalid shape.', @@ -258,6 +281,12 @@ export class RouteInvocationRoutes { return true; } + /** + * Replays the retained window one frame per socket drain, then delivers + * live messages through the bounded queue. The replay is bounded by + * retention, not by the queue: bursting it into the queue would destroy + * every reconnect whose window outgrew the queue's live-consumer limits. + */ #stream(service: RouteInvocationRouteService, id: string, response: ServerResponse): void { let terminal = false; const stream = { unsubscribe: undefined as (() => void) | undefined }; @@ -266,30 +295,56 @@ export class RouteInvocationRoutes { stream.unsubscribe?.(); response.end(); }; + const deliver = (frame: PendingFrame): void => { + if (writer.enqueue(frame.text) === 'overflow') response.destroy(); + if (frame.final) terminal = true; + finish(); + }; + // Live frames that arrive during the replay wait behind it, held to the + // live queue's own limits. + const pending: PendingFrame[] = []; + let replaying = true; + let subscribed = false; + let liveBytes = 0; + let liveRecords = 0; + const pump = (): void => { + while (replaying && writer.idle && !response.destroyed) { + const next = pending.shift(); + if (next === undefined) { + replaying = false; + break; + } + if (next.live) { + liveBytes -= next.bytes; + liveRecords -= 1; + } + deliver(next); + } + finish(); + }; const writer = createBackpressuredWriter(response, { byteLimit: streamQueueByteLimit, - onIdle: finish, + onIdle: pump, recordLimit: streamQueueEntryLimit, }); - const deliver = (message: RouteInvocationStreamMessage): void => { - const result = writer.enqueue(`event: ${message.type}\ndata: ${JSON.stringify(message)}\n\n`); - if (result === 'overflow') response.destroy(); - if (message.type === 'final') terminal = true; - finish(); - }; response.once('close', () => { writer.markClosed(); stream.unsubscribe?.(); }); - const replay: RouteInvocationStreamMessage[] = []; - let replaying = true; - stream.unsubscribe = service.subscribe(id, (message) => replaying ? replay.push(message) : deliver(message)); + stream.unsubscribe = service.subscribe(id, (message) => { + const frame = pendingFrame(message, subscribed); + if (!replaying) return deliver(frame); + pending.push(frame); + if (!frame.live) return; + liveBytes += frame.bytes; + liveRecords += 1; + if (liveBytes > streamQueueByteLimit || liveRecords > streamQueueEntryLimit) response.destroy(); + }); + subscribed = true; writeKeepAliveStreamHead(response, { cacheControl: 'no-cache', contentType: 'text/event-stream; charset=utf-8', }); - replaying = false; - for (const message of replay) deliver(message); - finish(); + pump(); } } diff --git a/packages/agent-bundle/src/dev/routes/route-invocation-service.ts b/packages/agent-bundle/src/dev/routes/route-invocation-service.ts index 697d394fa..ec2d6bbbe 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-service.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-service.ts @@ -47,6 +47,14 @@ import { isProductionRouteInvocationCode, ProductionRouteInvocationError, } from './route-invocation-production-error.ts'; +import { + emptyRetainedRenderEvents, + renderRetention, + retainedLatestDocument, + retainedRenderEvents, + retainRenderEvent, + type RetainedRenderEvents, +} from './route-invocation-render-history.ts'; import type { RouteInvocation, RouteInvocationStart, @@ -151,9 +159,14 @@ export interface RouteInvocationChildRequest { readonly surface: RouteInvocationSurface; } +/** + * What the child reports once. Render events never travel here: the child + * publishes each one as it happens (`type: 'render'`), and the service retains + * them under the render-history window, so neither IPC nor the completed + * envelope carries an unbounded copy of the stream. + */ export interface RouteInvocationChildResult { readonly document: NonNullable; - readonly events: RouteInvocation['events']; /** * Process surfaces only: the exit code the generated executable sets for * this completed run — a plain script's real exit status, the generated @@ -294,6 +307,7 @@ export const invocationSummary = (invocation: RouteInvocation): RouteInvocationS projection: _projection, providers: _providers, result: _result, + retention: _retention, trace: _trace, ...summary } = invocation; @@ -529,6 +543,7 @@ const runPlainScript = async ( script: TestableScriptDescriptor, input: JsonValue, signal: AbortSignal, + publishRenderEvent: (event: AgentRenderEvent) => void, ): Promise => { if (scripts === undefined) throw new Error('No script runner is available for a plain script.'); if (prepared.artifact === undefined) throw new Error('A plain script runs from the published build; none is published.'); @@ -544,9 +559,9 @@ const runPlainScript = async ( status: run.exitCode === 0 ? 'success' : 'represented-error', version: 1, }; + publishRenderEvent({ document, sequence: 1, type: 'complete' }); return deepFreeze({ document, - events: [{ document, sequence: 1, type: 'complete' }], exitCode: run.exitCode, input, renderDurationMs: performance.now() - startedAt, @@ -1133,10 +1148,24 @@ const invocationProjection = ( return {}; }; +/** The retained render stream and its truncation account, shared by every envelope a stream record settles into. */ +const retainedHistory = ( + history: RetainedRenderEvents, +): Pick => { + const document = retainedLatestDocument(history); + const retention = renderRetention(history); + return { + ...(document === undefined ? {} : { document }), + events: retainedRenderEvents(history), + ...(retention === undefined ? {} : { retention }), + }; +}; + const failedInvocation = (input: { readonly code: string; readonly completedAt: Date; readonly context: RequestContextProvenance; + readonly history: RetainedRenderEvents; readonly id: string; readonly manifest: RouteManifest; readonly message: string; @@ -1154,7 +1183,7 @@ const failedInvocation = (input: { context: input.context, ...(input.request.correlationId === undefined ? {} : { correlationId: input.request.correlationId }), diagnostics: [diagnostic(input.code, input.message)], - events: [], + ...retainedHistory(input.history), id: input.id, input: canonical ?? renderedInput ?? {}, kind: input.route.kind as RouteInvocationKind, @@ -1174,61 +1203,46 @@ const failedInvocation = (input: { interface InvocationStreamRecord { readonly controller: AbortController; readonly listeners: Set<(message: RouteInvocationStreamMessage) => void>; + /** Replay order: the `truncated` marker first once eviction happened, then retained `render` and `trace` messages as they arrived, then `final`. */ readonly messages: RouteInvocationStreamMessage[]; readonly running: RunningRouteInvocation; cancelRequested: boolean; final?: RouteInvocation; + /** The render-history window; `messages` holds exactly its render events. */ + history: RetainedRenderEvents; result?: Promise; } -const latestDocument = ( - messages: readonly RouteInvocationStreamMessage[], -): AgentDocument | undefined => { - for (let index = messages.length - 1; index >= 0; index -= 1) { - const message = messages[index]!; - if (message.type !== 'render') continue; - if (message.event.type === 'shell' || message.event.type === 'replace' || message.event.type === 'complete') { - return message.event.document; - } - } - return undefined; -}; - const cancelledInvocation = (input: { readonly context: RequestContextProvenance; + readonly history: RetainedRenderEvents; readonly id: string; readonly manifest: RouteManifest; - readonly messages: readonly RouteInvocationStreamMessage[]; readonly request: RouteInvocationRequest; readonly route: RouteManifestRoute; readonly startedAt: Date; readonly surface: RouteInvocationSurface; readonly completedAt: Date; -}): RouteInvocation => { - const events = input.messages.flatMap((message) => message.type === 'render' ? [message.event] : []); - const document = latestDocument(input.messages); - return deepFreeze({ - completedAt: input.completedAt.toISOString(), - context: input.context, - ...(input.request.correlationId === undefined ? {} : { correlationId: input.request.correlationId }), - diagnostics: [], - ...(document === undefined ? {} : { document }), - events, - id: input.id, - input: input.request.input ?? {}, - kind: input.route.kind as RouteInvocationKind, - manifestDigest: input.manifest.digest, - projection: {}, - providers: unobservedProviders(input.manifest), - routeId: input.route.id, - source: input.route.source, - sourceRevision: input.manifest.sourceRevision, - startedAt: input.startedAt.toISOString(), - status: 'cancelled', - surface: input.surface, - timings: [timing('elapsed', input.startedAt, input.completedAt.getTime() - input.startedAt.getTime())], - }); -}; +}): RouteInvocation => deepFreeze({ + completedAt: input.completedAt.toISOString(), + context: input.context, + ...(input.request.correlationId === undefined ? {} : { correlationId: input.request.correlationId }), + diagnostics: [], + ...retainedHistory(input.history), + id: input.id, + input: input.request.input ?? {}, + kind: input.route.kind as RouteInvocationKind, + manifestDigest: input.manifest.digest, + projection: {}, + providers: unobservedProviders(input.manifest), + routeId: input.route.id, + source: input.route.source, + sourceRevision: input.manifest.sourceRevision, + startedAt: input.startedAt.toISOString(), + status: 'cancelled', + surface: input.surface, + timings: [timing('elapsed', input.startedAt, input.completedAt.getTime() - input.startedAt.getTime())], +}); export class RouteInvocationService { readonly #completedStreams: string[] = []; @@ -1315,21 +1329,23 @@ export class RouteInvocationService { } #publishStream(record: InvocationStreamRecord, message: RouteInvocationStreamMessage): void { - if (message.type === 'render') { - const renderCount = record.messages.reduce((count, retained) => - count + (retained.type === 'render' ? 1 : 0), 0); - if (renderCount === 256) { - const oldest = record.messages.findIndex((retained) => retained.type === 'render'); - if (oldest !== -1) record.messages.splice(oldest, 1); - const markerIndex = record.messages.findIndex((retained) => retained.type === 'truncated'); - if (markerIndex === -1) { + const frozen = deepFreeze(message); + if (frozen.type === 'render') { + const next = retainRenderEvent(record.history, frozen.event); + record.history = next.retained; + if (next.evicted.length > 0) { + const evicted = new Set(next.evicted); + for (let index = record.messages.length - 1; index >= 0; index -= 1) { + const retained = record.messages[index]!; + if (retained.type === 'render' && evicted.has(retained.event)) record.messages.splice(index, 1); + } + if (!record.messages.some((retained) => retained.type === 'truncated')) { const marker = deepFreeze({ type: 'truncated' }); record.messages.unshift(marker); for (const listener of record.listeners) listener(marker); } } } - const frozen = deepFreeze(message); record.messages.push(frozen); for (const listener of record.listeners) listener(frozen); } @@ -1407,6 +1423,7 @@ export class RouteInvocationService { const streamRecord: InvocationStreamRecord = { cancelRequested: false, controller: operationController, + history: emptyRetainedRenderEvents, listeners: new Set(), messages: [], running: runningInvocation, @@ -1522,6 +1539,9 @@ export class RouteInvocationService { const timeout = setTimeout(() => controller.abort(new DOMException('Route invocation timed out.', 'TimeoutError')), this.#timeoutMs); let child: RouteInvocationChildResult; const plainScript = plainScriptFor(prepared, route); + const publishRenderEvent = (event: AgentRenderEvent): void => { + this.#publishStream(streamRecord, { event, type: 'render' }); + }; try { child = plainScript === undefined ? await this.#renderChild({ @@ -1537,19 +1557,17 @@ export class RouteInvocationService { routeId: route.id, stateRoot: prepared.stateRoot, surface, - }, controller.signal, publishKernelEvent, (event) => { - this.#publishStream(streamRecord, { event, type: 'render' }); - }) - : await runPlainScript(this.#scripts, prepared, plainScript, input, controller.signal); + }, controller.signal, publishKernelEvent, publishRenderEvent) + : await runPlainScript(this.#scripts, prepared, plainScript, input, controller.signal, publishRenderEvent); } catch (error) { const completedAt = this.#now(); if (streamRecord.cancelRequested) { return cancelledInvocation({ completedAt, context, + history: streamRecord.history, id, manifest, - messages: streamRecord.messages, request: { ...request, input }, route, startedAt, @@ -1563,6 +1581,7 @@ export class RouteInvocationService { code: childCode, completedAt, context, + history: streamRecord.history, id, manifest, message: controller.signal.reason instanceof DOMException && controller.signal.reason.name === 'TimeoutError' @@ -1593,7 +1612,6 @@ export class RouteInvocationService { context, ...(request.correlationId === undefined ? {} : { correlationId: request.correlationId }), diagnostics: [], - document: child.document, ...(canonical !== undefined && isJsonRecord(canonical) ? { event: { @@ -1607,7 +1625,8 @@ export class RouteInvocationService { }, } : {}), - events: child.events, + ...retainedHistory(streamRecord.history), + document: child.document, id, input: canonical ?? child.input, kind: route.kind as RouteInvocationKind, @@ -1634,9 +1653,9 @@ export class RouteInvocationService { return cancelledInvocation({ completedAt, context: cancellationContext, + history: streamRecord.history, id, manifest: queued, - messages: streamRecord.messages, request, route, startedAt, @@ -1651,6 +1670,7 @@ export class RouteInvocationService { code: error instanceof RouteInvocationRequestError ? error.code : ROUTE_INVOCATION_CHILD_FAILURE_CODE, completedAt, context: cancellationContext, + history: streamRecord.history, id, manifest: queued, message: error instanceof Error ? error.message : String(error), diff --git a/packages/agent-bundle/tests/route-invocation-dev-server.test.ts b/packages/agent-bundle/tests/route-invocation-dev-server.test.ts index 862f1a0e2..c01ee70b5 100644 --- a/packages/agent-bundle/tests/route-invocation-dev-server.test.ts +++ b/packages/agent-bundle/tests/route-invocation-dev-server.test.ts @@ -7,7 +7,11 @@ import { expect, it } from '@rstest/core'; import { Client } from '@modelcontextprotocol/client'; import { StdioClientTransport } from '@modelcontextprotocol/client/stdio'; -import type { RouteInvocationResponse } from '../src/dev/routes/route-invocation-result.ts'; +import { + renderEventBytes, + routeInvocationRenderHistoryLimits, +} from '../src/dev/routes/route-invocation-render-history.ts'; +import type { RouteInvocation, RouteInvocationResponse } from '../src/dev/routes/route-invocation-result.ts'; import type { RouteInvocationListResponse } from '../src/dev/routes/route-invocation.ts'; import type { RouteManifestResponse } from '../src/dev/routes/route-manifest.ts'; import type { TraceReplay } from '../src/dev/trace/trace-entry.ts'; @@ -92,6 +96,16 @@ const isFallbackRender = (fallback: string) => (message: StreamMessage): boolean const isFinal = (message: StreamMessage): boolean => message.type === 'final'; +/** Reads through `final` and returns every message seen. */ +const readInvocationStream = async (response: Response): Promise => { + const stream = invocationMessages(response); + await stream.next(isFinal); + return stream.seen; +}; + +const renderBytes = (events: readonly AgentRenderEvent[]): number => + events.reduce((sum, event) => sum + renderEventBytes(event), 0); + it('invokes compiled tool and event routes through the foreground server', { timeout: 180_000 }, async () => { const project = await createProjectFixture({ config: [ @@ -271,6 +285,23 @@ it('invokes compiled tool and event routes through the foreground server', { tim '}', '', ].join('\n'), + 'src/mcp/status/tools/burst.tsx': [ + "import { Agent, agent } from '@agent-bundle/runtime';", + "import { createElement } from 'react';", + "import { z } from 'zod';", + '', + 'export const inputSchema = z.object({ messageBytes: z.number().int().min(1).max(65_536), ticks: z.number().int().min(1).max(900) }).strict();', + 'export const resultSchema = z.object({ ticks: z.number() }).strict();', + '', + 'export default async function Burst({ input }) {', + ' const { progress } = await agent();', + ' for (let step = 1; step <= input.ticks; step += 1) {', + " await progress.report({ completed: step, message: `${step}:${'p'.repeat(input.messageBytes)}`, total: input.ticks });", + ' }', + " return createElement(Agent.Result, { value: { ticks: input.ticks } }, createElement(Agent.Text, null, `Reported ${input.ticks} ticks.`));", + '}', + '', + ].join('\n'), 'src/mcp/status/tools/report.cli.ts': [ "export const config = { command: ['report'], confirm: true, flags: { service: { name: 'name' }, source: { required: false } } };", "export const mapInput = (input) => ({ ...input, source: input.source ?? 'cli-projection' });", @@ -449,6 +480,48 @@ it('invokes compiled tool and event routes through the foreground server', { tim const unknownStream = await fetch(`${server.url}/api/routes/invocations/inv_missing/stream`, { headers }); expect(unknownStream.status).toBe(404); + // A compiled route that reports 400 progress ticks of 16 KiB each produces + // far more render events, and far more bytes (6.4 MiB), than the + // render-history window holds. Every reader must see the same bounded window. + const burstResponse = await fetch(`${server.url}/api/routes/invocations`, { + body: JSON.stringify({ correlationId: 'burst-1', input: { messageBytes: 16 * 1024, ticks: 400 }, routeId: 'tool:status/burst' }), + headers, + method: 'POST', + }); + expect(burstResponse.status).toBe(200); + const burst = (await burstResponse.json() as RouteInvocationResponse).invocation; + expect(burst.status, JSON.stringify(burst.diagnostics)).toBe('succeeded'); + expect(burst).toMatchObject({ correlationId: 'burst-1', outcome: { kind: 'success' }, result: { ticks: 400 } }); + const burstComplete = burst.events.at(-1); + expect(burstComplete?.type).toBe('complete'); + expect(burst.document).toEqual(burstComplete?.type === 'complete' ? burstComplete.document : undefined); + expect(JSON.stringify(burst.document)).toContain('Reported 400 ticks.'); + expect(burst.retention).toBeDefined(); + expect(burst.retention!.producedEvents).toBeGreaterThan(400); + expect(burst.retention!.producedEvents).toBe(burst.retention!.evictedEvents + burst.events.length); + expect(burst.events.length).toBeLessThan(routeInvocationRenderHistoryLimits.maxEvents); + expect(burst.events.filter((event) => event.type === 'progress').length).toBeGreaterThan(64); + expect(burst.events.at(-2)).toMatchObject({ completed: 400, total: 400, type: 'progress' }); + const burstBytes = burst.events.reduce((sum, event) => sum + renderEventBytes(event), 0); + expect(burstBytes).toBeLessThanOrEqual(routeInvocationRenderHistoryLimits.maxBytes); + expect(burst.retention!.retainedBytes).toBe(burstBytes); + expect(burst.retention!.evictedBytes).toBeGreaterThan(4 * 1024 * 1024); + const burstRead = await fetch(`${server.url}/api/routes/invocations/${burst.id}`, { headers }); + expect(burstRead.status).toBe(200); + expect(await burstRead.json()).toEqual({ invocation: burst }); + const burstReplay = await fetch(`${server.url}/api/routes/invocations/${burst.id}/stream`, { headers }); + expect(burstReplay.status).toBe(200); + const burstMessages = await readInvocationStream(burstReplay); + expect(burstMessages.filter((message) => message.type === 'truncated')).toEqual([{ type: 'truncated' }]); + expect(burstMessages[0]).toEqual({ type: 'truncated' }); + expect(burstMessages.flatMap((message) => message.type === 'render' ? [message.event] : [])).toEqual(burst.events); + expect(burstMessages.at(-1)).toEqual({ invocation: burst, type: 'final' }); + const burstList = await fetch(`${server.url}/api/routes/invocations?limit=5`, { headers }); + const burstSummary = (await burstList.json() as RouteInvocationListResponse).invocations.find((entry) => entry.id === burst.id); + expect(burstSummary).toBeDefined(); + expect(burstSummary).not.toHaveProperty('events'); + expect(burstSummary).not.toHaveProperty('retention'); + const activeEpoch = server.status().artifact; if (activeEpoch.state !== 'active') throw new Error('Expected an active compiled epoch.'); const artifactRoot = join(project.root, '.agent-bundle', 'epochs', activeEpoch.activeEpoch.id); @@ -1352,3 +1425,146 @@ it('publishes invocation routes only after a successful initial or recovered bui await rm(project.root, { force: true, maxRetries: 5, recursive: true, retryDelay: 50 }); } }); + +it('bounds the render history a compiled child produces by count and bytes across the envelope, reads, replay, and cancellation', { timeout: 180_000 }, async () => { + const project = await createProjectFixture({ + config: [ + 'export default {', + " plugin: { name: 'route-invocation-retention', version: '1.0.0' },", + " targets: ['claude'],", + '};', + '', + ].join('\n'), + files: { + 'package.json': '{"dependencies":{"@agent-bundle/runtime":"workspace:*","react":"19.2.8","zod":"4.5.4"},"type":"module"}\n', + // One Suspense boundary per cell settles per macrotask, so the stream + // grows one `replace` snapshot at a time; `gate` names a file the last + // boundary waits for, or `none`. + 'src/cli/flood.tsx': [ + "import { existsSync } from 'node:fs';", + "import { Agent } from '@agent-bundle/runtime';", + "import { createElement, Suspense } from 'react';", + "import { z } from 'zod';", + '', + "export const config = { description: 'Streams many boundaries.', positionals: ['boundaries', 'bytes', 'gate'] };", + 'export const inputSchema = z.object({ boundaries: z.number().int().min(1).max(900), bytes: z.number().int().min(1).max(4096), gate: z.string() }).strict();', + 'export const resultSchema = z.object({ boundaries: z.number() }).strict();', + '', + 'let turn = Promise.resolve();', + 'const Cell = async ({ bytes, index }) => {', + ' const mine = turn.then(() => new Promise((resolve) => setImmediate(resolve)));', + ' turn = mine;', + ' await mine;', + " return createElement(Agent.Text, null, `${index}:`.padEnd(bytes, 'x'));", + '};', + 'const Gate = async ({ path }) => {', + ' while (!existsSync(path)) await new Promise((resolve) => setTimeout(resolve, 20));', + " return createElement(Agent.Text, null, 'released');", + '};', + '', + 'export default async function Flood({ input }) {', + " const cells = Array.from({ length: input.boundaries }, (_, index) => createElement(Suspense, { fallback: createElement(Agent.Text, null, 'pending'), key: index }, createElement(Cell, { bytes: input.bytes, index })));", + " const gate = input.gate === 'none' ? [] : [createElement(Suspense, { fallback: createElement(Agent.Text, null, 'waiting'), key: 'gate' }, createElement(Gate, { path: input.gate }))];", + ' return createElement(Agent.Result, { value: { boundaries: input.boundaries } }, ...cells, ...gate);', + '}', + '', + ].join('\n'), + }, + prefix: 'agent-bundle-route-invocation-retention-', + }); + const assetsRoot = join(project.root, 'workbench'); + let server: Awaited> | undefined; + await mkdir(assetsRoot, { recursive: true }); + await Promise.all([ + symlink(agentBundleNodeModules, join(project.root, 'node_modules'), 'dir'), + writeFile(join(assetsRoot, 'index.html'), 'Route invocation retention'), + ]); + try { + server = await startDevServer({ + assets: createWorkbenchAssetSource({ root: assetsRoot }), + open: false, + port: 0, + root: project.root, + }); + const bootstrap = await fetch(`${server.url}/api/project/session`, { + headers: { 'sec-fetch-site': 'same-origin' }, + }); + const session = await bootstrap.json() as { readonly token: string }; + const headers = { + 'content-type': 'application/json', + origin: server.url, + 'x-agent-bundle-session': session.token, + }; + await expect.poll( + async () => fetch(`${server!.url}/api/routes/manifest`, { headers }).then((response) => response.status), + { timeout: 20_000 }, + ).toBe(200); + const boundaries = 300; + const flood = (args: readonly string[], stream: boolean) => fetch(`${server!.url}/api/routes/invocations`, { + body: JSON.stringify({ + correlationId: 'retention-1', + routeId: 'cli:flood', + ...(stream ? { stream: true } : {}), + surface: { args, command: 'flood', kind: 'cli' }, + }), + headers, + method: 'POST', + }); + const expectBounded = (invocation: RouteInvocation): void => { + expect(invocation.events.length).toBeLessThanOrEqual(routeInvocationRenderHistoryLimits.maxEvents); + expect(renderBytes(invocation.events)).toBeLessThanOrEqual(routeInvocationRenderHistoryLimits.maxBytes); + expect(invocation.retention?.evictedEvents).toBeGreaterThan(0); + expect(invocation.retention?.producedEvents).toBe(invocation.retention!.evictedEvents + invocation.events.length); + // shell, one replace per settled boundary, then complete or the pending gate + expect(invocation.retention?.producedEvents).toBeGreaterThanOrEqual(boundaries + 1); + expect(invocation.correlationId).toBe('retention-1'); + }; + + // Large intermediate snapshots: every replace carries the whole document, + // so bytes bound the window long before the event count does. + const completedResponse = await flood([String(boundaries), '256', 'none'], false); + expect(completedResponse.status).toBe(200); + const completed = (await completedResponse.json() as RouteInvocationResponse).invocation; + expect(completed.status, JSON.stringify(completed.diagnostics)).toBe('succeeded'); + expectBounded(completed); + expect(completed.events.length).toBeLessThan(routeInvocationRenderHistoryLimits.maxEvents); + expect(completed.events.at(-1)?.type).toBe('complete'); + expect(completed.outcome).toEqual({ kind: 'success' }); + expect(completed.result).toEqual({ boundaries }); + expect(JSON.stringify(completed.document)).toContain(`${String(boundaries - 1)}:`); + const read = (await (await fetch(`${server.url}/api/routes/invocations/${completed.id}`, { headers })).json() as RouteInvocationResponse).invocation; + expect(read).toEqual(completed); + const replayed = await readInvocationStream(await fetch(`${server.url}/api/routes/invocations/${completed.id}/stream`, { headers })); + expect(replayed[0]).toEqual({ type: 'truncated' }); + expect(renderEvents(replayed)).toEqual(completed.events); + expect(replayed.at(-1)).toEqual({ invocation: completed, type: 'final' }); + + // Reconnect once the shell has been evicted, then cancel behind the gate. + const gate = join(project.root, '.agent-bundle', 'flood-gate'); + const startedResponse = await flood([String(boundaries), '16', gate], true); + expect(startedResponse.status).toBe(202); + const started = await startedResponse.json() as { readonly invocation: { readonly id: string } }; + const live = invocationMessages(await fetch(`${server.url}/api/routes/invocations/${started.invocation.id}/stream`, { headers })); + await live.next(() => renderEvents(live.seen).length >= boundaries + 1); + const reconnectedResponse = await fetch(`${server.url}/api/routes/invocations/${started.invocation.id}/stream`, { headers }); + expect(reconnectedResponse.status).toBe(200); + const cancelResponse = await fetch(`${server.url}/api/routes/invocations/${started.invocation.id}/cancel`, { headers, method: 'POST' }); + expect(cancelResponse.status).toBe(202); + const cancelled = (await cancelResponse.json() as RouteInvocationResponse).invocation; + expect(cancelled.status).toBe('cancelled'); + expect(cancelled).not.toHaveProperty('outcome'); + expectBounded(cancelled); + expect(cancelled.events[0]?.type).not.toBe('shell'); + expect(JSON.stringify(cancelled.document)).toContain(`${String(boundaries - 1)}:`); + const reconnected = await readInvocationStream(reconnectedResponse); + expect(reconnected[0]).toEqual({ type: 'truncated' }); + expect(renderEvents(reconnected)).toEqual(cancelled.events); + expect(reconnected.at(-1)).toEqual({ invocation: cancelled, type: 'final' }); + await live.next(isFinal); + expect(live.seen.filter((message) => message.type === 'truncated')).toHaveLength(1); + expect(renderEvents(live.seen).length).toBeGreaterThan(routeInvocationRenderHistoryLimits.maxEvents); + } finally { + await server?.close().catch(() => undefined); + await rm(project.root, { force: true, maxRetries: 5, recursive: true, retryDelay: 50 }); + } +}); diff --git a/packages/agent-bundle/tests/route-invocation-retention.test.ts b/packages/agent-bundle/tests/route-invocation-retention.test.ts new file mode 100644 index 000000000..812817180 --- /dev/null +++ b/packages/agent-bundle/tests/route-invocation-retention.test.ts @@ -0,0 +1,143 @@ +import { createServer } from 'node:http'; +import { once } from 'node:events'; + +import { expect, it } from '@rstest/core'; + +import type { AgentDocument, AgentRenderEvent } from '@agent-bundle/runtime'; + +import { + emptyRetainedRenderEvents, + retainedLatestDocument, + retainedRenderEvents, + retainRenderEvent, + routeInvocationRenderHistoryLimits, +} from '../src/dev/routes/route-invocation-render-history.ts'; +import { RouteInvocationRoutes, type RouteInvocationRouteService } from '../src/dev/routes/route-invocation-routes.ts'; +import { + RouteInvocationService, + type RouteInvocationChildRequest, + type RouteInvocationChildResult, + type RouteInvocationServiceOptions, +} from '../src/dev/routes/route-invocation-service.ts'; +import type { RouteManifest } from '../src/dev/routes/route-manifest.ts'; + +const echoRoute = { + config: [], + id: 'tool:fixture/echo', + kind: 'tool', + provenance: { kind: 'conventional' }, + serverId: 'mcp:fixture', + source: 'src/mcp/fixture/tools/echo.tsx', +} as const; + +const manifest: RouteManifest = { + diagnostics: [], + digest: 'digest', + events: [], + providers: [], + scripts: [], + servers: [{ id: 'mcp:fixture', mode: 'generated', name: 'fixture', routes: [echoRoute] }], + sourceRevision: 'revision', +}; + +const documentOf = (text: string): AgentDocument => ({ root: { kind: 'text', text }, status: 'success', version: 1 }); +const finalDocument = documentOf('final'); + +/** A shell, then `replace` snapshots, then `complete`; `bytes` pads every snapshot's text. */ +const stream = (length: number, bytes = 0): readonly AgentRenderEvent[] => Array.from({ length }, (_, sequence): AgentRenderEvent => + sequence === 0 + ? { document: documentOf('shell'), sequence, type: 'shell' } + : sequence === length - 1 + ? { document: finalDocument, sequence, type: 'complete' } + : { boundaryId: 'b', document: documentOf(`${String(sequence)}:`.padEnd(bytes, 'x')), sequence, type: 'replace' }); + +const service = ( + renderChild: NonNullable, + historyLimit = 2, +): RouteInvocationService => new RouteInvocationService({ + historyLimit, + manifest: { manifest: () => manifest }, + prepared: async () => ({ + project: { + manifest: { projectRoot: '/project' } as never, + stateRoot: '/project/.agent-bundle/state', + targets: ['claude'], + }, + release: () => undefined, + }), + renderChild, +}); + +const childResult = (request: RouteInvocationChildRequest): RouteInvocationChildResult => ({ + document: finalDocument, + input: request.input, + mcp: {}, + renderDurationMs: 1, +}); + +const publishing = (events: readonly AgentRenderEvent[]): NonNullable => + async (request, _signal, _kernel, publishRender) => { + for (const event of events) publishRender(event); + return childResult(request); + }; + +it('never evicts the newest event or the pinned document, however large', () => { + const oversized = stream(4, routeInvocationRenderHistoryLimits.maxBytes); + let retained = emptyRetainedRenderEvents; + for (const event of oversized.slice(0, 3)) retained = retainRenderEvent(retained, event).retained; + expect(retainedRenderEvents(retained)).toEqual([oversized[2]]); + expect(retained.evictedEvents).toBe(2); + + // A progress event newer than the last snapshot keeps that snapshot pinned beside it. + const progress: AgentRenderEvent = { completed: 1, sequence: 3, type: 'progress' }; + retained = retainRenderEvent(retained, progress).retained; + expect(retainedRenderEvents(retained)).toEqual([oversized[2], progress]); + expect(retainedLatestDocument(retained)).toEqual(documentOf('2:'.padEnd(routeInvocationRenderHistoryLimits.maxBytes, 'x'))); +}); + +it('drops an evicted run\'s replay with its history record', async () => { + const events = stream(routeInvocationRenderHistoryLimits.maxEvents + 8); + const invocations = service(publishing(events)); + const first = await invocations.invoke({ correlationId: 'browser-1', input: {}, routeId: echoRoute.id }); + expect(first).toMatchObject({ + correlationId: 'browser-1', + document: finalDocument, + outcome: { kind: 'success' }, + retention: { evictedEvents: 8, producedEvents: events.length }, + status: 'succeeded', + }); + expect(invocations.read(first.id)).toBe(first); + + const second = await invocations.invoke({ input: {}, routeId: echoRoute.id }); + const third = await invocations.invoke({ input: {}, routeId: echoRoute.id }); + expect(invocations.read(first.id)).toBeUndefined(); + expect(() => invocations.subscribe(first.id, () => undefined)).toThrow(/was not found/); + expect(invocations.list().map((entry) => entry.id)).toEqual([third.id, second.id]); +}); + +it('serves a full count-bounded replay over the stream route without tripping the live-consumer queue', async () => { + const events = stream(routeInvocationRenderHistoryLimits.maxEvents * 2); + const invocations = service(publishing(events)); + const invocation = await invocations.invoke({ input: {}, routeId: echoRoute.id }); + const routes = new RouteInvocationRoutes({ + authorize: () => undefined, + eventHub: { publish: () => undefined } as never, + service: invocations as RouteInvocationRouteService, + }); + const server = createServer((request, response) => void routes.handle(request, response)); + server.listen(0, '127.0.0.1'); + await once(server, 'listening'); + const address = server.address(); + if (address === null || typeof address === 'string') throw new Error('The test server did not bind a port.'); + try { + const response = await fetch(`http://127.0.0.1:${String(address.port)}/api/routes/invocations/${invocation.id}/stream`); + expect(response.status).toBe(200); + const body = await response.text(); + const types = body.split('\n\n').filter((frame) => frame.startsWith('event: ')).map((frame) => frame.slice('event: '.length, frame.indexOf('\n'))); + expect(types[0]).toBe('truncated'); + expect(types.filter((type) => type === 'render')).toHaveLength(invocation.events.length); + expect(types.at(-1)).toBe('final'); + } finally { + server.close(); + } +}); diff --git a/packages/agent-bundle/tests/route-invocation-service.test.ts b/packages/agent-bundle/tests/route-invocation-service.test.ts index e5c5c9e89..0de67355c 100644 --- a/packages/agent-bundle/tests/route-invocation-service.test.ts +++ b/packages/agent-bundle/tests/route-invocation-service.test.ts @@ -22,6 +22,10 @@ import { type RouteInvocationPreparedProject, type RouteInvocationServiceOptions, } from '../src/dev/routes/route-invocation-service.ts'; +import { + renderEventBytes, + routeInvocationRenderHistoryLimits, +} from '../src/dev/routes/route-invocation-render-history.ts'; import type { RouteManifest } from '../src/dev/routes/route-manifest.ts'; import type { CompiledRouteGraph } from '../src/routes/types.ts'; import { testManifestFromRouteGraph } from '../src/test/manifest.ts'; @@ -228,7 +232,6 @@ it('publishes correlated invocation and kernel entries with slim details', async }; return { document, - events: [{ document, sequence: 1, type: 'complete' }], input: { value: 'echo' }, mcp: { content: [] }, renderDurationMs: 4, @@ -418,7 +421,6 @@ const childResult = (request: RouteInvocationChildRequest, text = 'ok'): RouteIn status: 'success', version: 1, }, - events: [], input: request.input, mcp: {}, renderDurationMs: 1, @@ -459,10 +461,7 @@ it('publishes render events before the final invocation', async () => { rendered.resolve(); await release.promise; publishRender({ document, sequence: 1, type: 'complete' }); - return { ...childResult(request), events: [ - { document, sequence: 0, type: 'shell' }, - { document, sequence: 1, type: 'complete' }, - ] }; + return childResult(request); }); const started = service.start({ input: {}, routeId: echoRoute.id }); @@ -471,24 +470,30 @@ it('publishes render events before the final invocation', async () => { service.subscribe(started.invocation.id, (message) => messages.push(message)); expect(messages.map((message) => message.type)).toEqual(['render']); release.resolve(); - await started.result; + const invocation = await started.result; expect(messages.map((message) => message.type)).toEqual(['render', 'render', 'final']); + expect(invocation.events).toEqual([ + { document, sequence: 0, type: 'shell' }, + { document, sequence: 1, type: 'complete' }, + ]); + expect(invocation).not.toHaveProperty('retention'); }); +const shellDocument = childResult({ + context: {} as never, + input: {}, + manifest: {} as never, + routeId: echoRoute.id, + stateRoot: '/project/state', + surface: { kind: 'unit-render' }, +}).document; + it('retains only the newest 256 render events and one truncation marker', async () => { const release = deferred(); const rendered = deferred(); - const document = childResult({ - context: {} as never, - input: {}, - manifest: {} as never, - routeId: echoRoute.id, - stateRoot: '/project/state', - surface: { kind: 'unit-render' }, - }).document; const service = streamingService(async (request, _signal, _trace, publishRender) => { for (let sequence = 0; sequence < 300; sequence += 1) { - publishRender({ document, sequence, type: 'shell' }); + publishRender({ document: shellDocument, sequence, type: 'shell' }); } rendered.resolve(); await release.promise; @@ -507,10 +512,134 @@ it('retains only the newest 256 render events and one truncation marker', async event: { sequence: 44 }, }); release.resolve(); - await started.result; + const invocation = await started.result; expect(messages.findLast((message) => message.type === 'final')).toMatchObject({ - invocation: { document }, + invocation: { document: shellDocument }, }); + expect(invocation.events).toHaveLength(256); + expect(invocation.events[0]).toMatchObject({ sequence: 44 }); + expect(invocation.retention).toMatchObject({ evictedEvents: 44, producedEvents: 300 }); +}); + +it('publishes the same bounded history to the final envelope, history reads, and the stream replay', async () => { + const shell = { document: shellDocument, sequence: 0, type: 'shell' as const }; + const progress = (sequence: number) => ({ completed: sequence, sequence, total: 1_000, type: 'progress' as const }); + const complete = { document: shellDocument, sequence: 1_000, type: 'complete' as const }; + const service = streamingService(async (request, _signal, _trace, publishRender) => { + publishRender(shell); + for (let sequence = 1; sequence < 1_000; sequence += 1) publishRender(progress(sequence)); + publishRender(complete); + return childResult(request); + }); + + const started = service.start({ correlationId: 'browser-7', input: {}, routeId: echoRoute.id }); + const invocation = await started.result; + const messages: Parameters[1]>[0][] = []; + service.subscribe(started.invocation.id, (message) => messages.push(message)); + + expect(invocation).toMatchObject({ + correlationId: 'browser-7', + document: shellDocument, + outcome: { kind: 'success' }, + retention: { evictedEvents: 745, producedEvents: 1_001 }, + status: 'succeeded', + }); + expect(invocation.events).toHaveLength(routeInvocationRenderHistoryLimits.maxEvents); + expect(invocation.events.at(-1)).toEqual(complete); + expect(invocation.events[0]).toEqual(progress(745)); + expect(invocation.retention?.retainedBytes).toBe(invocation.events.reduce((sum, event) => sum + renderEventBytes(event), 0)); + expect(service.read(started.invocation.id)).toBe(invocation); + expect(messages[0]).toEqual({ type: 'truncated' }); + expect(messages.flatMap((message) => message.type === 'render' ? [message.event] : [])).toEqual(invocation.events); + expect(messages.at(-1)).toEqual({ invocation, type: 'final' }); +}); + +it('bounds retained bytes with large intermediate snapshots and pins the latest document', async () => { + const rendered = deferred(); + const snapshot = (sequence: number) => ({ + boundaryId: 'b', + document: { root: { kind: 'text' as const, text: `${String(sequence)}:${'x'.repeat(300 * 1024)}` }, status: 'success' as const, version: 1 as const }, + sequence, + type: 'replace' as const, + }); + const service = streamingService((_request, signal, _trace, publishRender) => new Promise((_resolve, reject) => { + publishRender({ document: shellDocument, sequence: 0, type: 'shell' }); + for (let sequence = 1; sequence <= 12; sequence += 1) publishRender(snapshot(sequence)); + for (let sequence = 13; sequence <= 20; sequence += 1) publishRender({ completed: sequence, sequence, type: 'progress' }); + rendered.resolve(); + signal.addEventListener('abort', () => reject(signal.reason), { once: true }); + })); + + const started = service.start({ input: {}, routeId: echoRoute.id }); + await rendered.promise; + const messages: Parameters[1]>[0][] = []; + service.subscribe(started.invocation.id, (message) => messages.push(message)); + const replayed = messages.flatMap((message) => message.type === 'render' ? [message.event] : []); + const replayedBytes = replayed.reduce((sum, event) => sum + renderEventBytes(event), 0); + expect(replayedBytes).toBeLessThanOrEqual(routeInvocationRenderHistoryLimits.maxBytes); + expect(replayed.map((event) => event.sequence)).toEqual([7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20]); + expect(messages.filter((message) => message.type === 'truncated')).toHaveLength(1); + + const cancelled = await service.cancel(started.invocation.id); + expect(cancelled.status).toBe('cancelled'); + expect(cancelled.document).toEqual(snapshot(12).document); + expect(cancelled.events.map((event) => event.sequence)).toEqual([7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20]); + expect(cancelled.retention).toMatchObject({ evictedEvents: 7, producedEvents: 21, retainedBytes: replayedBytes }); + expect(cancelled.retention!.evictedBytes).toBeGreaterThan(6 * 300 * 1024); +}); + +it('keeps the retained window and latest document when the child fails after eviction', async () => { + const service = streamingService(async (_request, _signal, _trace, publishRender) => { + publishRender({ document: shellDocument, sequence: 0, type: 'shell' }); + for (let sequence = 1; sequence <= 300; sequence += 1) publishRender({ completed: sequence, sequence, type: 'progress' }); + throw new Error('render exploded'); + }); + + const started = service.start({ input: {}, routeId: echoRoute.id }); + const failed = await started.result; + const replay: Parameters[1]>[0][] = []; + service.subscribe(started.invocation.id, (message) => replay.push(message)); + + expect(failed).toMatchObject({ + diagnostics: [{ code: 'AB8236', message: 'Route invocation child failed: render exploded' }], + document: shellDocument, + retention: { evictedEvents: 45, producedEvents: 301 }, + status: 'failed', + }); + expect(failed).not.toHaveProperty('outcome'); + expect(failed.events).toHaveLength(routeInvocationRenderHistoryLimits.maxEvents); + expect(failed.events[0]).toEqual({ document: shellDocument, sequence: 0, type: 'shell' }); + expect(replay[0]).toEqual({ type: 'truncated' }); + expect(replay.flatMap((message) => message.type === 'render' ? [message.event] : [])).toEqual(failed.events); +}); + +it('cancels after the shell was evicted with the pinned document and the truncation account', async () => { + const startedChild = deferred(); + const rendered = deferred(); + const service = streamingService((_request, signal, _trace, publishRender) => new Promise((_resolve, reject) => { + startedChild.resolve(); + publishRender({ document: shellDocument, sequence: 0, type: 'shell' }); + for (let sequence = 1; sequence <= 600; sequence += 1) { + publishRender({ completed: sequence, message: `step ${String(sequence)}`, sequence, total: 1_000, type: 'progress' }); + } + rendered.resolve(); + signal.addEventListener('abort', () => reject(signal.reason), { once: true }); + })); + + const started = service.start({ input: {}, routeId: echoRoute.id }); + await startedChild.promise; + await rendered.promise; + const cancelled = await service.cancel(started.invocation.id); + + expect(cancelled).toMatchObject({ id: started.invocation.id, status: 'cancelled' }); + expect(cancelled).not.toHaveProperty('outcome'); + expect(cancelled.document).toEqual(shellDocument); + expect(cancelled.events).toHaveLength(routeInvocationRenderHistoryLimits.maxEvents); + expect(cancelled.events[0]).toEqual({ document: shellDocument, sequence: 0, type: 'shell' }); + expect(cancelled.events[1]).toMatchObject({ sequence: 346, type: 'progress' }); + expect(cancelled.events.at(-1)).toMatchObject({ sequence: 600, type: 'progress' }); + expect(cancelled.retention).toMatchObject({ evictedEvents: 345, producedEvents: 601 }); + expect(service.read(started.invocation.id)).toBe(cancelled); }); it('cancels a running invocation without an outcome and publishes cancellation', async () => { @@ -987,6 +1116,69 @@ it('resolves a `.js` import of a `.tsx` sibling without rewriting the same strin } }); +/** + * A tool that reports far more progress than the render-history window holds + * and streams five Suspense chunks whose `replace` snapshots sum past the byte + * bound, while the final document (750 KiB) stays under the runtime's 1 MiB. + */ +const burstRouteProject = async (): Promise => routeProject( + await mkdtemp(join(tmpdir(), 'agent-bundle-route-invocation-burst-')), + 'burst', + { + 'src/mcp/fixture/tools/burst.tsx': [ + "import { Agent, agent } from '@agent-bundle/runtime';", + "import { createElement, Suspense } from 'react';", + '', + 'const Chunk = async ({ index }) => {', + ' await new Promise((resolve) => setTimeout(resolve, 20 * (index + 1)));', + " return createElement(Agent.Text, null, `${index}:${'x'.repeat(150 * 1024)}`);", + '};', + '', + 'export default async function Burst() {', + ' const { progress } = await agent();', + ' for (let step = 1; step <= 400; step += 1) await progress.report({ completed: step, total: 400 });', + ' return createElement(Agent.Result, null, ...Array.from({ length: 5 }, (_, index) =>', + ' createElement(Suspense, { fallback: createElement(Agent.Progress, { completed: index, total: 5 }), key: index }, createElement(Chunk, { index }))));', + '}', + '', + ].join('\n'), + }, +); + +it('bounds a real child\'s long, heavy render stream end to end', { timeout: 60_000 }, async () => { + const project = await burstRouteProject(); + const service = project.service(); + try { + const started = service.start({ correlationId: 'burst-1', input: {}, routeId: 'tool:fixture/burst', surface: { kind: 'unit-render' } }); + const invocation = await started.result; + expect(invocation.status, JSON.stringify(invocation.diagnostics)).toBe('succeeded'); + + const complete = invocation.events.at(-1); + expect(complete?.type).toBe('complete'); + expect(invocation.document).toEqual(complete?.type === 'complete' ? complete.document : undefined); + expectDocument(invocation.document!).toContainText('4:xxxx'); + expect(invocation).toMatchObject({ correlationId: 'burst-1', outcome: { kind: 'success' } }); + expect(invocation.events.length).toBeLessThanOrEqual(routeInvocationRenderHistoryLimits.maxEvents); + const retainedBytes = invocation.events.reduce((sum, event) => sum + renderEventBytes(event), 0); + expect(retainedBytes).toBeLessThanOrEqual(routeInvocationRenderHistoryLimits.maxBytes); + expect(invocation.retention).toMatchObject({ retainedBytes }); + expect(invocation.retention!.producedEvents).toBeGreaterThan(400); + expect(invocation.retention!.producedEvents).toBe(invocation.retention!.evictedEvents + invocation.events.length); + expect(invocation.retention!.evictedBytes).toBeGreaterThan(150 * 1024); + expect(invocation.events.filter((event) => event.type === 'replace').length).toBeLessThan(5); + + expect(service.read(started.invocation.id)).toBe(invocation); + const replay: Parameters[1]>[0][] = []; + service.subscribe(started.invocation.id, (message) => replay.push(message)); + expect(replay.filter((message) => message.type === 'truncated')).toHaveLength(1); + expect(replay.flatMap((message) => message.type === 'render' ? [message.event] : [])).toEqual(invocation.events); + expect(replay.at(-1)).toEqual({ invocation, type: 'final' }); + } finally { + await service.close(); + await rm(project.root, { force: true, recursive: true }); + } +}); + /** A zombie has exited; only a process still scheduled counts as alive. */ const alive = (pid: number): boolean => { if (isProcessGone(pid)) return false; @@ -1232,15 +1424,6 @@ const succeededChild = (observed?: RouteInvocationChildResult['observed']): Rout status: 'success', version: 1, }, - events: [{ - document: { - root: { children: [{ kind: 'text', text: 'ok' }], kind: 'result' }, - status: 'success', - version: 1, - }, - sequence: 1, - type: 'complete', - }], input: {}, mcp: { content: [] }, ...(observed === undefined ? {} : { observed }), diff --git a/packages/workbench/src/application/invocation-client.ts b/packages/workbench/src/application/invocation-client.ts index 2f6dee33f..0a3f9a98b 100644 --- a/packages/workbench/src/application/invocation-client.ts +++ b/packages/workbench/src/application/invocation-client.ts @@ -123,6 +123,12 @@ const eventTraceWireSchema = z.strictObject({ const eventTraceSchema = z.custom( (value) => eventTraceWireSchema.safeParse(value).success, ); +const retentionSchema = z.strictObject({ + evictedBytes: z.number().int().nonnegative(), + evictedEvents: z.number().int().positive(), + producedEvents: z.number().int().positive(), + retainedBytes: z.number().int().nonnegative(), +}); const outcomeSchema = z.discriminatedUnion('kind', [ z.strictObject({ kind: z.literal('success') }), z.strictObject({ kind: z.literal('represented-error'), summary: z.string() }), @@ -152,6 +158,9 @@ const outcomeMatchesStatus = (value: Pick = z.strictObject(invocationSummaryFields).refine(outcomeMatchesStatus); +// A retention account must describe exactly the events it sits beside. +const retentionMatchesEvents = (value: Pick): boolean => + value.retention === undefined || value.retention.producedEvents === value.retention.evictedEvents + value.events.length; const invocationSchema: z.ZodType = z.strictObject({ ...invocationSummaryFields, context: requestContextProvenanceSchema, @@ -160,8 +169,9 @@ const invocationSchema: z.ZodType = z.strictObject({ projection: projectionSchema, providers: z.array(providerSchema), result: z.json().optional(), + retention: retentionSchema.optional(), trace: z.array(eventTraceSchema).optional(), -}).refine(outcomeMatchesStatus); +}).refine(outcomeMatchesStatus).refine(retentionMatchesEvents); const invocationResponseSchema = z.strictObject({ invocation: invocationSchema }); const runningInvocationSchema: z.ZodType = z.strictObject({ id: textSchema, diff --git a/packages/workbench/src/application/invocation-model.ts b/packages/workbench/src/application/invocation-model.ts index 4ff670c08..7f901d7bc 100644 --- a/packages/workbench/src/application/invocation-model.ts +++ b/packages/workbench/src/application/invocation-model.ts @@ -4,11 +4,14 @@ * the summary projection of an envelope. */ import type { Diagnostic } from '../../../agent-bundle/src/contracts/diagnostics.ts'; -import type { - RouteInvocation, - RouteInvocationOutcome, - RouteInvocationStatus, - RouteInvocationSummary, +import { + emptyRetainedRenderEvents, + retainRenderEvent, + type RetainedRenderEvents, + type RouteInvocation, + type RouteInvocationOutcome, + type RouteInvocationStatus, + type RouteInvocationSummary, } from '../../../agent-bundle/src/contracts/invocations.ts'; import { parseJsonWithoutDuplicateKeys, @@ -29,7 +32,8 @@ export type InvocationState = | Readonly<{ readonly phase: 'idle' }> | Readonly<{ readonly correlationId: string; - readonly events?: readonly AgentRenderEvent[]; + /** The live render window, held under the same retention policy the server applies to its replay and envelope. */ + readonly history?: RetainedRenderEvents; readonly invocationId?: string; readonly phase: 'running'; readonly startedAt: number; @@ -61,9 +65,6 @@ const settled = (invocation: RouteInvocation, durationMs?: number): InvocationSt export const idleInvocationState: InvocationState = Object.freeze({ phase: 'idle' }); -/** Matches RouteInvocationService's retained render-event bound. */ -const maximumLiveRenderEvents = 256; - export const reduceInvocationState = (state: InvocationState, action: InvocationAction): InvocationState => { switch (action.type) { case 'start': @@ -79,7 +80,7 @@ export const reduceInvocationState = (state: InvocationState, action: Invocation }); case 'render': return state.phase === 'running' - ? Object.freeze({ ...state, events: Object.freeze([...(state.events ?? []), action.event].slice(-maximumLiveRenderEvents)) }) + ? Object.freeze({ ...state, history: retainRenderEvent(state.history ?? emptyRetainedRenderEvents, action.event).retained }) : state; case 'stream.start': return state.phase === 'running' diff --git a/packages/workbench/src/application/result-tabs.tsx b/packages/workbench/src/application/result-tabs.tsx index 531bb724f..993f3330c 100644 --- a/packages/workbench/src/application/result-tabs.tsx +++ b/packages/workbench/src/application/result-tabs.tsx @@ -7,10 +7,12 @@ */ import React, { useEffect, useState } from 'react'; -import type { - RouteInvocation, - RouteInvocationOutcome, - RouteInvocationStatus, +import { + emptyRetainedRenderEvents, + retainedRenderEvents, + type RouteInvocation, + type RouteInvocationOutcome, + type RouteInvocationStatus, } from '../../../agent-bundle/src/contracts/invocations.ts'; import { isTraceReplayGap, @@ -93,6 +95,13 @@ const RawDocument = ({ invocation }: { readonly invocation?: RouteInvocation }):

Render events ({String(invocation.events.length)})

+ {invocation.retention === undefined + ? undefined + :

+ Retained {String(invocation.events.length)} of {String(invocation.retention.producedEvents)} events; + {' '}{String(invocation.retention.evictedEvents)} older events ({String(invocation.retention.evictedBytes)} B) were evicted. + The Agent Document above is never truncated. +

} {invocation.events.length === 0 ?

The stream carried no events.

:
    {invocation.events.map((event) =>
  1. @@ -207,7 +216,9 @@ const useTraceEntries = (trace: TraceClient | undefined): TraceLoadState => { export const ResultTabs = ({ controller, extraTabs = [], leaf, onNavigate, onTabChange, tab, trace }: ResultTabsProps): React.ReactNode => { const invocation = invocationOf(controller.state); const running = controller.state.phase === 'running'; - const events = running ? controller.state.events ?? [] : invocation?.events ?? []; + const events = running + ? retainedRenderEvents(controller.state.history ?? emptyRetainedRenderEvents) + : invocation?.events ?? []; const traceState = useTraceEntries(trace); const definitions: readonly ResultTabDefinition[] = [ { id: 'rendered', label: coreTabLabels.rendered, render: () => { - for (let index = events.length - 1; index >= 0; index -= 1) { - const event = events[index]!; - switch (event.type) { - case 'shell': - case 'replace': - case 'complete': - return event.document; - case 'progress': - case 'error': - break; - default: { - const exhaustive: never = event; - return exhaustive; - } - } - } - return undefined; -}; - const invocationKind = (kind: RouteInvocationKind) => { switch (kind) { case 'tool': @@ -216,7 +199,11 @@ const invocationForRun = ( const diagnostics = run.status === 'failed' ? Object.freeze(run.diagnostics.map(diagnosticFor)) : Object.freeze([]); - const document = documentFor(events); + // The run document arrives whole from the runtime; hold it under the same + // window every other RouteInvocation applies. + const history = events.reduce((retained, event) => retainRenderEvent(retained, event).retained, emptyRetainedRenderEvents); + const document = retainedLatestDocument(history); + const retention = renderRetention(history); const timings = run.status === 'succeeded' ? Object.freeze(run.result.trace.flatMap((span) => span.durationMs === undefined ? [] @@ -244,7 +231,7 @@ const invocationForRun = ( ...(correlationId === undefined ? {} : { correlationId }), diagnostics, ...(document === undefined ? {} : { document }), - events, + events: retainedRenderEvents(history), id: run.id, input: run.input, kind, @@ -253,6 +240,7 @@ const invocationForRun = ( projection: Object.freeze({}), providers: Object.freeze([]), ...(result === undefined ? {} : { result }), + ...(retention === undefined ? {} : { retention }), routeId: leaf.routeId, source: leaf.source ?? '', sourceRevision: run.vector.sourceRevision, diff --git a/packages/workbench/tests/invocation-client.test.ts b/packages/workbench/tests/invocation-client.test.ts index a17a0cdbe..a814a2b70 100644 --- a/packages/workbench/tests/invocation-client.test.ts +++ b/packages/workbench/tests/invocation-client.test.ts @@ -137,6 +137,26 @@ it('sends and decodes the optional correlationId on invocations and summaries', await expect(rejecting.invoke({ routeId: invocation.routeId })).rejects.toMatchObject({ code: 'AB8230' }); }); +it('decodes the render-history retention account and rejects an impossible one', async () => { + const retained = { + ...invocation, + retention: Object.freeze({ evictedBytes: 4_096, evictedEvents: 44, producedEvents: 45, retainedBytes: 512 }), + } satisfies RouteInvocation; + const client = new InvocationClient({ foreground: foreground(() => Response.json({ invocation: retained })) }); + await expect(client.read(invocation.id)).resolves.toEqual(retained); + + for (const retention of [ + { evictedBytes: 4_096, evictedEvents: 0, producedEvents: 1, retainedBytes: 512 }, + { evictedBytes: -1, evictedEvents: 44, producedEvents: 45, retainedBytes: 512 }, + { evictedEvents: 44, producedEvents: 45, retainedBytes: 512 }, + { evictedBytes: 4_096, evictedEvents: 44, producedEvents: 300, retainedBytes: 512 }, + { evictedBytes: 4_096, evictedEvents: 44, producedEvents: 45, retainedBytes: 512, truncated: true }, + ]) { + const rejecting = new InvocationClient({ foreground: foreground(() => Response.json({ invocation: { ...invocation, retention } })) }); + await expect(rejecting.read(invocation.id)).rejects.toMatchObject({ code: 'AB8230' }); + } +}); + it('preserves coded HTTP diagnostics', async () => { const client = new InvocationClient({ foreground: foreground(() => Response.json({ diagnostic: { code: 'AB8232', message: 'No published build.' }, diff --git a/packages/workbench/tests/invocation-model.test.ts b/packages/workbench/tests/invocation-model.test.ts index e87fad40c..4f0ba15cb 100644 --- a/packages/workbench/tests/invocation-model.test.ts +++ b/packages/workbench/tests/invocation-model.test.ts @@ -1,6 +1,11 @@ import { afterEach, beforeEach, expect, it } from '@rstest/core'; -import type { RouteInvocation } from '../../agent-bundle/src/contracts/invocations.ts'; +import { + emptyRetainedRenderEvents, + retainedRenderEvents, + routeInvocationRenderHistoryLimits, + type RouteInvocation, +} from '../../agent-bundle/src/contracts/invocations.ts'; import type { ApplicationLeaf } from '../src/application/application-tree-model.ts'; import type { InvocationBackend } from '../src/application/invocation-backend.ts'; import { @@ -12,7 +17,10 @@ import { selectBackend, statusLabel, writeLastInput, + type InvocationState, } from '../src/application/invocation-model.ts'; +import { foldAgentDocumentEvents } from '../src/application/rendered-document.tsx'; +import type { AgentRenderEvent } from '../src/runtime/agent-document-client.ts'; const invocation = Object.freeze({ completedAt: '2026-09-05T07:00:01.000Z', @@ -96,6 +104,11 @@ it('reduces invocation lifecycle states without retaining stale failures', () => expect(reduceInvocationState(running, { type: 'reset' })).toBe(idleInvocationState); }); +const liveEvents = (state: InvocationState): readonly AgentRenderEvent[] => { + if (state.phase !== 'running') throw new Error('Expected a running invocation.'); + return retainedRenderEvents(state.history ?? emptyRetainedRenderEvents); +}; + it('retains only the newest 256 live render events', () => { let state = reduceInvocationState(idleInvocationState, { correlationId: 'c1', startedAt: 1_000, type: 'start' }); for (let sequence = 0; sequence < 300; sequence += 1) { @@ -110,10 +123,54 @@ it('retains only the newest 256 live render events', () => { } expect(state).toMatchObject({ phase: 'running' }); + const events = liveEvents(state); + expect(events).toHaveLength(256); + expect(events[0]?.sequence).toBe(44); + expect(events.at(-1)?.sequence).toBe(299); + if (state.phase !== 'running') throw new Error('Expected a running invocation.'); + expect(state.history).toMatchObject({ evictedEvents: 44, producedEvents: 300 }); +}); + +it('pins the latest document event while a long progress sequence evicts the rest', () => { + const shell = { + document: { root: { kind: 'text' as const, text: 'shell' }, status: 'success' as const, version: 1 as const }, + sequence: 0, + type: 'shell' as const, + }; + let state = reduceInvocationState(idleInvocationState, { correlationId: 'c1', startedAt: 1_000, type: 'start' }); + state = reduceInvocationState(state, { event: shell, type: 'render' }); + for (let sequence = 1; sequence <= 400; sequence += 1) { + state = reduceInvocationState(state, { event: { completed: sequence, sequence, total: 400, type: 'progress' }, type: 'render' }); + } + + const events = liveEvents(state); + expect(events).toHaveLength(routeInvocationRenderHistoryLimits.maxEvents); + expect(events[0]).toEqual(shell); + expect(events[1]?.sequence).toBe(146); + expect(events.at(-1)?.sequence).toBe(400); + expect(foldAgentDocumentEvents(events)).toMatchObject({ + complete: false, + document: shell.document, + progress: { completed: 400 }, + }); +}); + +it('bounds the live window by retained bytes, not only by event count', () => { + const snapshot = (sequence: number) => ({ + boundaryId: 'b', + document: { root: { kind: 'text' as const, text: 'x'.repeat(256 * 1024) }, status: 'success' as const, version: 1 as const }, + sequence, + type: 'replace' as const, + }); + let state = reduceInvocationState(idleInvocationState, { correlationId: 'c1', startedAt: 1_000, type: 'start' }); + for (let sequence = 0; sequence < 24; sequence += 1) { + state = reduceInvocationState(state, { event: snapshot(sequence), type: 'render' }); + } + if (state.phase !== 'running') throw new Error('Expected a running invocation.'); - expect(state.events).toHaveLength(256); - expect(state.events?.[0]?.sequence).toBe(44); - expect(state.events?.at(-1)?.sequence).toBe(299); + expect(state.history?.retainedBytes).toBeLessThanOrEqual(routeInvocationRenderHistoryLimits.maxBytes); + expect(state.history?.evictedEvents).toBe(17); + expect(liveEvents(state).map((event) => event.sequence)).toEqual([17, 18, 19, 20, 21, 22, 23]); }); it('stores strict JSON last-input snapshots by leaf key and tolerates unavailable storage', () => { diff --git a/packages/workbench/tests/route-workspace.test.ts b/packages/workbench/tests/route-workspace.test.ts index 069b9a55e..fd4fd53ec 100644 --- a/packages/workbench/tests/route-workspace.test.ts +++ b/packages/workbench/tests/route-workspace.test.ts @@ -3,7 +3,7 @@ import { renderToStaticMarkup } from 'react-dom/server'; import { describe, expect, it } from '@rstest/core'; -import type { RouteInvocation } from '../../agent-bundle/src/contracts/invocations.ts'; +import { retainedRenderEvents, type RouteInvocation } from '../../agent-bundle/src/contracts/invocations.ts'; import type { TraceEntry } from '../../agent-bundle/src/contracts/trace.ts'; import { appResourceUriFor, appToolCallRequest, catalogToolsFor, orderedToolsForApp } from '../src/application/app-route-workspace.tsx'; import type { ApplicationLeaf } from '../src/application/application-tree-model.ts'; @@ -92,10 +92,15 @@ describe('invocation state contract', () => { type: 'render', }); expect(progressed).toMatchObject({ - events: [expect.objectContaining({ type: 'shell' }), expect.objectContaining({ type: 'progress' })], + history: { evictedEvents: 0, producedEvents: 2 }, invocationId: 'inv-live', phase: 'running', }); + if (progressed.phase !== 'running') throw new Error('Expected a running invocation.'); + expect(retainedRenderEvents(progressed.history!)).toEqual([ + expect.objectContaining({ type: 'shell' }), + expect.objectContaining({ type: 'progress' }), + ]); const { outcome: _outcome, ...withoutOutcome } = invocation; const cancelled = { ...withoutOutcome, status: 'cancelled' as const }; diff --git a/packages/workbench/tests/runtime-backend.test.ts b/packages/workbench/tests/runtime-backend.test.ts index 50e8cea33..6a74c4ccd 100644 --- a/packages/workbench/tests/runtime-backend.test.ts +++ b/packages/workbench/tests/runtime-backend.test.ts @@ -5,6 +5,7 @@ import type { DevRuntimeRun, DevRuntimeSurface, } from '../../agent-bundle/src/contracts/runtime.ts'; +import { routeInvocationRenderHistoryLimits } from '../../agent-bundle/src/contracts/invocations.ts'; import type { ApplicationLeaf } from '../src/application/application-tree-model.ts'; import { createRuntimeBackend, type RuntimeInvocationClient } from '../src/application/runtime-backend.ts'; import type { RuntimePlaygroundController } from '../src/runtime-controller.ts'; @@ -185,3 +186,23 @@ it('keeps the succeeded outcome invariant when a runtime run has no document eve expect(invocation.document).toBeUndefined(); expect(invocation.outcome).toEqual({ kind: 'success' }); }); + +it('holds a long runtime run document under the shared render-history window', async () => { + const setup = fixture(); + const flood = Object.freeze([ + { document, sequence: 0, type: 'shell' as const }, + ...Array.from({ length: 300 }, (_, index) => ({ completed: index, sequence: index + 1, type: 'progress' as const })), + ]); + const backend = createRuntimeBackend({ + ...setup, + runtimeClient: { ...setup.runtimeClient, readRunDocument: async () => flood }, + }); + + const invocation = await backend.invoke(leaf, { input: { title: 'Dune' }, routeId: leaf.routeId }); + + expect(invocation.events).toHaveLength(routeInvocationRenderHistoryLimits.maxEvents); + expect(invocation.events[0]).toEqual(flood[0]); + expect(invocation.events.at(-1)).toEqual(flood.at(-1)); + expect(invocation.document).toEqual(document); + expect(invocation.retention).toMatchObject({ evictedEvents: 45, producedEvents: 301 }); +}); diff --git a/website/docs/en/guide/development/workbench.mdx b/website/docs/en/guide/development/workbench.mdx index 92eeeead1..7426ddda0 100644 --- a/website/docs/en/guide/development/workbench.mdx +++ b/website/docs/en/guide/development/workbench.mdx @@ -327,8 +327,9 @@ The route workspace uses one authenticated, origin-guarded foreground API: - `POST /api/routes/invocations` accepts a route invocation request and returns its completed invocation envelope. With `stream: true`, it returns `202` immediately with the running id. - `GET /api/routes/invocations//stream` streams retained and live `render`, `trace`, - `truncated`, and terminal `final` messages as server-sent events. It retains the newest 256 - render events and signals truncation once. + `truncated`, and terminal `final` messages as server-sent events. Retained render events are + the render-history window described below, and a replay that starts with `truncated` has lost + older events to it. - `POST /api/routes/invocations//cancel` stops a running or queued invocation and returns its cancelled final envelope; cancelling an already-final invocation reports `AB8256` (409). - `GET /api/routes/invocations?limit=50` returns newest-first summaries for Trace. @@ -358,7 +359,18 @@ survives a successful republish; `unit-render` still uses a fresh temporary stat The envelope records the resolved `surface` beside `routeId` and carries canonical input, request context, providers, ordered render events, the final Agent Document, structured result, -projections, diagnostics, and execution timings. When the worker reports observations, +projections, diagnostics, and execution timings. Render events are retained under one +render-history window, applied identically to the live stream replay, the completed and +cancelled envelopes, `GET /api/routes/invocations/`, the `final` stream message, and the +Workbench's live view: at most 256 events and 2 MiB serialized, with the newest event +and the newest document-bearing event (`shell`, `replace`, or `complete`) always kept — those two +alone may carry the window past 2 MiB — so `events` folds to a coherent latest document even +after eviction. Older intermediate events are dropped; the child never returns the stream a +second time. When anything was evicted the envelope carries `retention` — `producedEvents`, +`evictedEvents`, `evictedBytes`, and `retainedBytes` — and the stream signals `truncated` once. A +cancelled or failed run keeps the same retained window and the latest document it reached. The +final `document`, `result`, `outcome`, and `correlationId` are never truncated by this window; +the runtime's own `maxDocumentBytes` bounds the document before it reaches the service. When the worker reports observations, `providers` contains only providers it selected and reports each as `mounted` or `failed`; unselected catalog providers are absent. A `durationMs` appears only when measured. When the child reports no observations — a plain script, a preflight short-circuit, or a failure before diff --git a/website/docs/en/reference/dev-server-http.mdx b/website/docs/en/reference/dev-server-http.mdx index 7e1f2658a..4c588f64f 100644 --- a/website/docs/en/reference/dev-server-http.mdx +++ b/website/docs/en/reference/dev-server-http.mdx @@ -25,8 +25,13 @@ The POST body is a `RouteInvocationRequest`: `routeId` plus optional `input`, `a identifiers on the invocation so the route workspace and trace can join the run. The completed envelope holds the input, provenance context, providers, render events, Agent Document, result, projections, diagnostics, and timings. Trace entries link to that full snapshot instead of -duplicating it. A cancelled invocation has `status: "cancelled"` and no `outcome`; cancelling an -already-final invocation returns `AB8256` (409). +duplicating it. Render events are the render-history window (at most 256 events and 2 MiB, always +keeping the newest event and the newest document-bearing event, which alone may exceed 2 MiB) +shared by the envelope, the item route, and the stream replay; when events were evicted the envelope carries `retention` with `producedEvents`, +`evictedEvents`, `evictedBytes`, and `retainedBytes`, and the stream sends `truncated` once. +The Agent Document, result, and outcome are never truncated by it. A cancelled invocation has +`status: "cancelled"` and no `outcome`; cancelling an already-final invocation returns `AB8256` +(409). ## Host sessions diff --git a/website/docs/zh/guide/development/workbench.mdx b/website/docs/zh/guide/development/workbench.mdx index 43b94d2fa..890056495 100644 --- a/website/docs/zh/guide/development/workbench.mdx +++ b/website/docs/zh/guide/development/workbench.mdx @@ -274,7 +274,8 @@ Workbench 外壳，因此路由、trace、Problems 与 Advanced 的深度链接 - `POST /api/routes/invocations` 接受一条路由调用请求，并返回其已完成的调用信封。传入 `stream: true` 时会立即以 `202` 返回运行中的 id。 - `GET /api/routes/invocations//stream` 通过服务器发送事件重放并实时发送 `render`、`trace`、 - `truncated` 与终止 `final` 消息。它保留最新 256 条 render 事件，并用一条标记表示发生了截断。 + `truncated` 与终止 `final` 消息。重放的 render 事件即下文所述的渲染历史窗口；以 `truncated` + 开头的重放表示更早的事件已被该窗口淘汰。 - `POST /api/routes/invocations//cancel` 停止运行中或排队中的调用并返回取消后的最终信封；取消已经 终止的调用会报告 `AB8256`（409）。 - `GET /api/routes/invocations?limit=50` 为 Trace 返回按最新优先的摘要。 @@ -301,7 +302,15 @@ providers、持久状态、CLI `mapInput` 与确认、事件 preflight 以及渲 仍为每次运行使用新的临时状态根。 该信封在 `routeId` 旁记录解析后的 `surface`，并携带规范输入、请求上下文、providers、有序渲染事件、 -最终的 Agent Document、结构化结果、投影、诊断与执行计时。worker 报告观测值时，`providers` 只包含 +最终的 Agent Document、结构化结果、投影、诊断与执行计时。渲染事件由同一个渲染历史窗口保留，该窗口 +同样适用于实时流重放、已完成与已取消的信封、`GET /api/routes/invocations/`、`final` 流消息以及 +Workbench 的实时视图：至多 256 条事件、序列化后至多 2 MiB，且始终保留最新一条事件与最新一条携带文档 +的事件（`shell`、`replace` 或 `complete`）——仅这两条事件本身可能使窗口超过 2 MiB——因此即使发生 +淘汰，`events` 折叠后仍是连贯的最新文档。更早的中间事件会被丢弃；子进程不会再次返回整条事件流。 +一旦有事件被淘汰，信封会携带 `retention`——`producedEvents`、`evictedEvents`、`evictedBytes` 与 +`retainedBytes`——流也会发送一次 `truncated`。已取消或失败的运行保留同一窗口以及它到达的最新文档。 +最终的 `document`、`result`、`outcome` 与 `correlationId` 永不被该窗口截断；文档在到达服务之前已由 +运行时自身的 `maxDocumentBytes` 约束。worker 报告观测值时，`providers` 只包含 它实际选择的 provider，并将状态报告为 `mounted` 或 `failed`；未选择的目录 provider 不会出现。只有 实际测到时长时才带 `durationMs`。子进程未报告观测值时（普通脚本、preflight 短路，或在观测开始之前 失败），每个目录 provider 为 `unobserved` 且省略 `durationMs`；`0` 是测得的零，不是「未知」。 diff --git a/website/docs/zh/reference/dev-server-http.mdx b/website/docs/zh/reference/dev-server-http.mdx index f7d383d10..2644e5474 100644 --- a/website/docs/zh/reference/dev-server-http.mdx +++ b/website/docs/zh/reference/dev-server-http.mdx @@ -21,8 +21,12 @@ description: 'Workbench 调用、统一 Trace、Raw logs 与宿主钩子收据 POST 正文是一份 `RouteInvocationRequest`：包含 `routeId`，以及可选的 `input`、`args`、 `correlationId`、调用方 `requestId` 与事件夹具选项。前台会把这两个标识回显到调用上，使路由工作区与 Trace 可以关联此次运行。完成后的信封包含输入、provenance 上下文、providers、渲染事件、Agent -Document、结果、投影、诊断与计时。Trace 条目链接到这份完整快照，而不会复制它。取消后的调用带有 -`status: "cancelled"` 且不含 `outcome`；取消已终止的调用会返回 `AB8256`（409）。 +Document、结果、投影、诊断与计时。Trace 条目链接到这份完整快照，而不会复制它。渲染事件即渲染历史窗口 +（至多 256 条事件与 2 MiB，并始终保留最新一条事件与最新一条携带文档的事件——仅这两条本身可能 +超过 2 MiB），由信封、单项路由与流重放共享； +一旦有事件被淘汰，信封会携带 `retention`（`producedEvents`、`evictedEvents`、`evictedBytes` 与 +`retainedBytes`），流也会发送一次 `truncated`。Agent Document、结果与 outcome 永不被该窗口截断。 +取消后的调用带有 `status: "cancelled"` 且不含 `outcome`；取消已终止的调用会返回 `AB8256`（409）。 ## 宿主会话