diff --git a/.changeset/681-bounded-invocation-render-history.md b/.changeset/681-bounded-invocation-render-history.md new file mode 100644 index 000000000..ce1c6e50b --- /dev/null +++ b/.changeset/681-bounded-invocation-render-history.md @@ -0,0 +1,5 @@ +--- +'agent-bundle': patch +--- + +Bound route-invocation render history end to end with one policy (`RENDER_EVENT_RETENTION`, exported from `contracts/invocations`): the render child, its IPC reply, the `GET /api/routes/invocations//stream` replay, the completed `RouteInvocation` envelope returned by `POST /api/routes/invocations` and `GET /api/routes/invocations/`, and the Workbench live buffer each keep the newest 256 render events totalling at most 1 MiB of JSON, evicting oldest-first while the newest event, `document`, `result`, `outcome`, and correlation identifiers always survive. Envelopes that lost events carry `evictedEvents`; cancelled and failed runs keep their retained window and the newest document a render event carried; stream replays are paced by socket drain instead of the live-consumer queue, so a reconnect whose window outgrows that queue no longer disconnects. (#699) diff --git a/packages/agent-bundle/src/contracts/invocations.ts b/packages/agent-bundle/src/contracts/invocations.ts index b6be95b2c..d251b6b5a 100644 --- a/packages/agent-bundle/src/contracts/invocations.ts +++ b/packages/agent-bundle/src/contracts/invocations.ts @@ -1,8 +1,15 @@ /** * 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. Routes render on + * the server through the production runtime; the only runtime export is the + * render-history retention the browser's live buffer shares with the server. */ +export { + RENDER_EVENT_RETENTION, + emptyRetainedRenderEvents, + retainRenderEvent, + type RetainedRenderEvents, +} from '../dev/routes/route-invocation-result.ts'; export type { RouteInvocationCliProjection, RouteInvocationEvent, 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..a12f60d7e 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-child.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-child.ts @@ -26,6 +26,7 @@ import type { } from './route-invocation-service.ts'; import { ProductionRouteInvocationError } from './route-invocation-production-error.ts'; import { renderProductionRoute } from './route-invocation-production.ts'; +import { retainRenderEvents } from './route-invocation-result.ts'; import { createRouteModuleLoader } from './route-module-loader.ts'; const { load } = createRouteModuleLoader(); @@ -153,13 +154,14 @@ 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); + if (request.surface.kind !== 'unit-render') { + return renderProductionRoute(request, forwardEventTrace, forwardRenderEvent); } - return result; + const result = await renderUnitRoute(request); + for (const event of result.events) forwardRenderEvent(event); + // The harness collects the whole stream; only the retained window crosses IPC. + const retained = retainRenderEvents(result.events); + return { ...result, events: retained.events, evictedEvents: retained.evicted }; }; process.once('message', (request: RouteInvocationChildRequest) => { 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 7263bcdc7..614375ceb 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-production.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-production.ts @@ -30,6 +30,7 @@ import { ROUTE_INVOCATION_COMPILED_ROUTE_UNAVAILABLE_CODE, ROUTE_INVOCATION_PREPARATION_FAILURE_CODE, } from './route-invocation-production-error.ts'; +import { emptyRetainedRenderEvents, retainRenderEvent } from './route-invocation-result.ts'; import type { RouteInvocationProvider, RouteInvocationTiming } from './route-invocation.ts'; interface CompiledCliInvocationModule { @@ -447,6 +448,7 @@ const renderCompiled = async ( readonly document: AgentDocument; readonly durationMs: number; readonly events: readonly AgentRenderEvent[]; + readonly evictedEvents: number; readonly observed: { readonly providers: readonly RouteInvocationProvider[]; readonly timings: readonly RouteInvocationTiming[]; @@ -457,21 +459,22 @@ const renderCompiled = async ( for (const workerPath of candidates) { const startedAt = performance.now(); const session = streamFromWorker(workerPath, request, invocation, input, signal, env, trace); - const events: AgentRenderEvent[] = []; + const retained = emptyRetainedRenderEvents(); try { const reader = session.events.getReader(); for (;;) { const next = await reader.read(); if (next.done) break; - events.push(next.value); + retainRenderEvent(retained, next.value); publishRender?.(next.value); } - const complete = events.findLast((event) => event.type === 'complete'); - if (complete === undefined) throw new Error('Compiled route render ended without a complete event.'); + const complete = retained.events.at(-1); + if (complete?.type !== 'complete') throw new Error('Compiled route render ended without a complete event.'); return Object.freeze({ document: complete.document, durationMs: performance.now() - startedAt, - events: Object.freeze(events), + events: Object.freeze(retained.events), + evictedEvents: retained.evicted, observed: { providers: Object.freeze([...session.observed.providers]), timings: Object.freeze([...session.observed.timings]), @@ -558,6 +561,7 @@ export const renderProductionRoute = async ( return Object.freeze({ document: rendered.document, events: rendered.events, + evictedEvents: rendered.evictedEvents, ...(exitCode === undefined ? {} : { exitCode }), input: prepared.input, ...(kind === 'tool' 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..9be0c9af6 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-result.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-result.ts @@ -10,12 +10,62 @@ import type { RouteInvocationSummary, } from './route-invocation.ts'; +/** + * The one retention policy for an invocation's render history, applied by + * every keeper of it: the render child while it streams, the child's IPC + * reply, the live stream replay, the completed envelope (and so history reads + * and the final stream message), and the browser's live buffer. Past either + * bound the oldest event is evicted first; the newest event always survives, + * so a completed stream keeps its `complete` event whatever its size. The + * final `document`, `outcome`, and correlation identifiers live outside this + * window and are never evicted. + */ +export const RENDER_EVENT_RETENTION = Object.freeze({ + /** JSON text of the retained events, in UTF-16 code units. */ + maxBytes: 1024 * 1024, + maxEvents: 256, +}); + +export interface RetainedRenderEvents { + /** JSON text size of `events`. */ + bytes: number; + /** Older events the bounds evicted — the truncation indication. */ + evicted: number; + readonly events: AgentRenderEvent[]; +} + +export const renderEventBytes = (event: AgentRenderEvent): number => JSON.stringify(event).length; + +export const emptyRetainedRenderEvents = (): RetainedRenderEvents => ({ bytes: 0, events: [], evicted: 0 }); + +/** Appends `event`, then evicts oldest-first while `RENDER_EVENT_RETENTION` is exceeded. */ +export const retainRenderEvent = (retained: RetainedRenderEvents, event: AgentRenderEvent): void => { + retained.events.push(event); + retained.bytes += renderEventBytes(event); + while ( + retained.events.length > 1 + && (retained.events.length > RENDER_EVENT_RETENTION.maxEvents || retained.bytes > RENDER_EVENT_RETENTION.maxBytes) + ) { + retained.bytes -= renderEventBytes(retained.events.shift()!); + retained.evicted += 1; + } +}; + +/** The retained window of an already-collected stream. */ +export const retainRenderEvents = (events: readonly AgentRenderEvent[]): RetainedRenderEvents => { + const retained = emptyRetainedRenderEvents(); + for (const event of events) retainRenderEvent(retained, event); + return retained; +}; + export interface RouteInvocation extends RouteInvocationSummary { readonly context: RequestContextProvenance; /** The final Agent Document; absent when rendering failed before a document existed. */ readonly document?: AgentDocument; - /** The production `shell | progress | replace | error | complete` stream, in order. */ + /** The retained window of the production `shell | progress | replace | error | complete` stream, in order (`RENDER_EVENT_RETENTION`). */ readonly events: readonly AgentRenderEvent[]; + /** Older render events `RENDER_EVENT_RETENTION` evicted from `events`; absent when none were. */ + readonly evictedEvents?: number; readonly projection: RouteInvocationProjection; readonly providers: readonly RouteInvocationProvider[]; /** Structured value recorded by the selected surface; its presence alone proves neither a `resultSchema` declaration nor validation. */ 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..6a2223e2a 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'; @@ -37,6 +38,19 @@ import { const streamQueueByteLimit = 256 * 1024; const streamQueueEntryLimit = 128; + +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 +272,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 +286,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(); + }; + // The replay drains one frame per socket drain. Live frames that arrive + // meanwhile 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..9ddf6b362 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-service.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-service.ts @@ -47,10 +47,14 @@ import { isProductionRouteInvocationCode, ProductionRouteInvocationError, } from './route-invocation-production-error.ts'; -import type { - RouteInvocation, - RouteInvocationStart, - RouteInvocationStreamMessage, +import { + emptyRetainedRenderEvents, + retainRenderEvent, + retainRenderEvents, + type RetainedRenderEvents, + type RouteInvocation, + type RouteInvocationStart, + type RouteInvocationStreamMessage, } from './route-invocation-result.ts'; import { nativeEventRequestContext } from './route-invocation.ts'; import type { @@ -153,7 +157,10 @@ export interface RouteInvocationChildRequest { export interface RouteInvocationChildResult { readonly document: NonNullable; + /** The retained window of the render stream (`RENDER_EVENT_RETENTION`); the service re-applies the bound. */ readonly events: RouteInvocation['events']; + /** Render events the child evicted before replying. */ + readonly evictedEvents?: number; /** * Process surfaces only: the exit code the generated executable sets for * this completed run — a plain script's real exit status, the generated @@ -291,6 +298,7 @@ export const invocationSummary = (invocation: RouteInvocation): RouteInvocationS context: _context, document: _document, events: _events, + evictedEvents: _evictedEvents, projection: _projection, providers: _providers, result: _result, @@ -1140,6 +1148,7 @@ const failedInvocation = (input: { readonly id: string; readonly manifest: RouteManifest; readonly message: string; + readonly record: InvocationStreamRecord; readonly request: RouteInvocationRequest; readonly route: RouteManifestRoute; readonly startedAt: Date; @@ -1154,7 +1163,7 @@ const failedInvocation = (input: { context: input.context, ...(input.request.correlationId === undefined ? {} : { correlationId: input.request.correlationId }), diagnostics: [diagnostic(input.code, input.message)], - events: [], + ...retainedHistory(input.record), id: input.id, input: canonical ?? renderedInput ?? {}, kind: input.route.kind as RouteInvocationKind, @@ -1173,62 +1182,66 @@ const failedInvocation = (input: { interface InvocationStreamRecord { readonly controller: AbortController; + /** The newest document a render event carried, kept outside the evictable window. */ + latestDocument?: AgentDocument; readonly listeners: Set<(message: RouteInvocationStreamMessage) => void>; + /** Kernel trace messages and the terminal `final`; never evicted. */ readonly messages: RouteInvocationStreamMessage[]; + readonly renders: RetainedRenderEvents; readonly running: RunningRouteInvocation; cancelRequested: boolean; final?: RouteInvocation; 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 truncatedMarker = deepFreeze({ type: 'truncated' }); + +/** What a run that did not complete keeps of its render history: the retained window and the newest document. */ +const retainedHistory = ( + record: InvocationStreamRecord, +): Pick => ({ + ...(record.latestDocument === undefined ? {} : { document: record.latestDocument }), + events: [...record.renders.events], + ...(record.renders.evicted === 0 ? {} : { evictedEvents: record.renders.evicted }), +}); + +/** Replay order: the truncation marker, the retained render window, then kernel trace and `final`. */ +const replayMessages = (record: InvocationStreamRecord): readonly RouteInvocationStreamMessage[] => [ + ...(record.renders.evicted === 0 ? [] : [truncatedMarker]), + ...record.renders.events.map((event) => deepFreeze({ event, type: 'render' })), + ...record.messages, +]; const cancelledInvocation = (input: { readonly context: RequestContextProvenance; readonly id: string; readonly manifest: RouteManifest; - readonly messages: readonly RouteInvocationStreamMessage[]; + readonly record: InvocationStreamRecord; 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.record), + 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[] = []; @@ -1280,7 +1293,7 @@ export class RouteInvocationService { 404, ); } - for (const message of record.messages) listener(message); + for (const message of replayMessages(record)) listener(message); if (record.final === undefined) record.listeners.add(listener); return () => record.listeners.delete(listener); } @@ -1315,22 +1328,18 @@ 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 marker = deepFreeze({ type: 'truncated' }); - record.messages.unshift(marker); - for (const listener of record.listeners) listener(marker); - } + const frozen = deepFreeze(message); + if (frozen.type === 'render') { + const { event } = frozen; + if (event.type === 'shell' || event.type === 'replace' || event.type === 'complete') record.latestDocument = event.document; + const evictedBefore = record.renders.evicted; + retainRenderEvent(record.renders, event); + if (evictedBefore === 0 && record.renders.evicted > 0) { + for (const listener of record.listeners) listener(truncatedMarker); } + } else { + record.messages.push(frozen); } - const frozen = deepFreeze(message); - record.messages.push(frozen); for (const listener of record.listeners) listener(frozen); } @@ -1409,6 +1418,7 @@ export class RouteInvocationService { controller: operationController, listeners: new Set(), messages: [], + renders: emptyRetainedRenderEvents(), running: runningInvocation, }; this.#streams.set(id, streamRecord); @@ -1549,7 +1559,7 @@ export class RouteInvocationService { context, id, manifest, - messages: streamRecord.messages, + record: streamRecord, request: { ...request, input }, route, startedAt, @@ -1572,6 +1582,7 @@ export class RouteInvocationService { : controller.signal.aborted ? 'Route invocation child stopped because the service closed.' : `${plainScript === undefined ? 'Route invocation child' : 'Script run'} failed: ${error instanceof Error ? error.message : String(error)}`, + record: streamRecord, request: { ...request, input }, route, startedAt, @@ -1585,6 +1596,8 @@ export class RouteInvocationService { const projectionStartedAt = this.#now(); const projection = invocationProjection(route, surface, rawInput, child, prepared, this.#registry); const completedAt = this.#now(); + const retained = retainRenderEvents(child.events); + const evictedEvents = (child.evictedEvents ?? 0) + retained.evicted; const canonical = route.kind === 'event-route' ? (child.input as JsonObject).canonical : undefined; @@ -1607,7 +1620,8 @@ export class RouteInvocationService { }, } : {}), - events: child.events, + events: retained.events, + ...(evictedEvents === 0 ? {} : { evictedEvents }), id, input: canonical ?? child.input, kind: route.kind as RouteInvocationKind, @@ -1636,7 +1650,7 @@ export class RouteInvocationService { context: cancellationContext, id, manifest: queued, - messages: streamRecord.messages, + record: streamRecord, request, route, startedAt, @@ -1654,6 +1668,7 @@ export class RouteInvocationService { id, manifest: queued, message: error instanceof Error ? error.message : String(error), + record: streamRecord, request, route, startedAt, 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 92f6601e9..4ff7525f9 100644 --- a/packages/agent-bundle/tests/route-invocation-dev-server.test.ts +++ b/packages/agent-bundle/tests/route-invocation-dev-server.test.ts @@ -6,7 +6,14 @@ 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 type { AgentRenderEvent } from '@agent-bundle/runtime'; + +import { + RENDER_EVENT_RETENTION, + renderEventBytes, + type RouteInvocation, + type 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'; @@ -17,6 +24,7 @@ import { createWorkbenchAssetSource } from '../src/dev/workbench-assets.ts'; import { startDevServer } from '../src/dev/workbench-server.ts'; import { createProjectFixture } from './helpers/project-fixture.ts'; import { agentBundleNodeModules } from './helpers/workspace-paths.ts'; +import { deferred } from './support/eventually.ts'; import { replaceWatchedSourceAndAwaitRebuild } from './support/watched-files.ts'; import { runNodeScript } from './support/run-node-script.ts'; @@ -44,26 +52,40 @@ const readEvent = async ( } }; -const readInvocationStream = async (response: Response): Promise[]> => { +/** Reads invocation stream messages until `done` accepts the collection; the stream is left open. */ +const readInvocationStreamUntil = ( + response: Response, + done: (messages: readonly Record[]) => boolean, +): Promise[]> => { const reader = response.body!.pipeThrough(new TextDecoderStream()).getReader(); const messages: Record[] = []; let buffered = ''; - for (;;) { - const next = await reader.read(); - if (next.done) throw new Error('Invocation stream ended before final.'); - buffered += next.value; - const frames = buffered.split('\n\n'); - buffered = frames.pop() ?? ''; - for (const frame of frames) { - const data = frame.split('\n').find((line) => line.startsWith('data: ')); - if (data === undefined) continue; - const message = JSON.parse(data.slice('data: '.length)) as Record; - messages.push(message); - if (message.type === 'final') return messages; + return (async () => { + for (;;) { + const next = await reader.read(); + if (next.done) throw new Error('Invocation stream ended before final.'); + buffered += next.value; + const frames = buffered.split('\n\n'); + buffered = frames.pop() ?? ''; + for (const frame of frames) { + const data = frame.split('\n').find((line) => line.startsWith('data: ')); + if (data === undefined) continue; + messages.push(JSON.parse(data.slice('data: '.length)) as Record); + if (done(messages)) return messages; + } } - } + })(); }; +const readInvocationStream = (response: Response): Promise[]> => + readInvocationStreamUntil(response, (messages) => messages.at(-1)?.type === 'final'); + +const renderMessages = (messages: readonly Record[]): readonly AgentRenderEvent[] => + messages.flatMap((message) => message.type === 'render' ? [message.event as AgentRenderEvent] : []); + +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: [ @@ -1227,3 +1249,151 @@ 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(RENDER_EVENT_RETENTION.maxEvents); + expect(renderBytes(invocation.events)).toBeLessThanOrEqual(RENDER_EVENT_RETENTION.maxBytes); + expect(invocation.evictedEvents).toBeGreaterThan(0); + // shell, one replace per settled boundary, then complete or the pending gate + expect(invocation.events.length + invocation.evictedEvents!).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(RENDER_EVENT_RETENTION.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.events).toHaveLength(completed.events.length); + expect(read.evictedEvents).toBe(completed.evictedEvents); + const replayed = await readInvocationStream(await fetch(`${server.url}/api/routes/invocations/${completed.id}/stream`, { headers })); + expect(replayed[0]).toEqual({ type: 'truncated' }); + expect(renderMessages(replayed)).toHaveLength(completed.events.length); + expect((replayed.at(-1)!.invocation as RouteInvocation).events).toHaveLength(completed.events.length); + + // 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 liveResponse = await fetch(`${server.url}/api/routes/invocations/${started.invocation.id}/stream`, { headers }); + const settled = deferred(); + const liveMessages = readInvocationStreamUntil(liveResponse, (messages) => { + if (renderMessages(messages).length >= boundaries + 1) settled.resolve(); + return messages.at(-1)?.type === 'final'; + }); + await settled.promise; + 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(renderMessages(reconnected)).toEqual(cancelled.events); + expect(reconnected.at(-1)).toMatchObject({ invocation: { evictedEvents: cancelled.evictedEvents, status: 'cancelled' }, type: 'final' }); + const live = await liveMessages; + expect(live.filter((message) => message.type === 'truncated')).toHaveLength(1); + expect(renderMessages(live).length).toBeGreaterThan(RENDER_EVENT_RETENTION.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..27b3fa398 --- /dev/null +++ b/packages/agent-bundle/tests/route-invocation-retention.test.ts @@ -0,0 +1,218 @@ +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 { RouteInvocationRoutes, type RouteInvocationRouteService } from '../src/dev/routes/route-invocation-routes.ts'; +import { + RENDER_EVENT_RETENTION, + renderEventBytes, + retainRenderEvents, + type RouteInvocationStreamMessage, +} from '../src/dev/routes/route-invocation-result.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'; +import { deferred } from './support/eventually.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 progress 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, events: readonly AgentRenderEvent[]): RouteInvocationChildResult => ({ + document: finalDocument, + events, + input: request.input, + mcp: {}, + renderDurationMs: 1, +}); + +const collect = (invocations: RouteInvocationService, id: string): RouteInvocationStreamMessage[] => { + const messages: RouteInvocationStreamMessage[] = []; + invocations.subscribe(id, (message) => messages.push(message)); + return messages; +}; + +const bytesOf = (events: readonly AgentRenderEvent[]): number => events.reduce((sum, event) => sum + renderEventBytes(event), 0); + +it('bounds a long stream by count in the replay, the envelope, and history while keeping outcome and correlation', async () => { + const events = stream(RENDER_EVENT_RETENTION.maxEvents * 4); + const invocations = service(async (request, _signal, _kernel, publishRender) => { + for (const event of events) publishRender(event); + return childResult(request, events); + }); + const invocation = await invocations.invoke({ correlationId: 'browser-1', input: {}, routeId: echoRoute.id }); + + expect(invocation.events).toHaveLength(RENDER_EVENT_RETENTION.maxEvents); + expect(invocation.evictedEvents).toBe(events.length - RENDER_EVENT_RETENTION.maxEvents); + expect(invocation.events[0]?.sequence).toBe(invocation.evictedEvents); + expect(invocation.events.at(-1)).toMatchObject({ sequence: events.length - 1, type: 'complete' }); + expect(invocation).toMatchObject({ + correlationId: 'browser-1', + document: finalDocument, + outcome: { kind: 'success' }, + routeId: echoRoute.id, + status: 'succeeded', + }); + expect(invocations.read(invocation.id)).toBe(invocation); + + const messages = collect(invocations, invocation.id); + expect(messages[0]).toEqual({ type: 'truncated' }); + expect(messages.filter((message) => message.type === 'render')).toHaveLength(RENDER_EVENT_RETENTION.maxEvents); + expect(messages.at(-1)).toMatchObject({ invocation: { evictedEvents: invocation.evictedEvents, id: invocation.id }, type: 'final' }); + + const second = await invocations.invoke({ input: {}, routeId: echoRoute.id }); + const third = await invocations.invoke({ input: {}, routeId: echoRoute.id }); + expect(invocations.read(invocation.id)).toBeUndefined(); + expect(() => invocations.subscribe(invocation.id, () => undefined)).toThrow(/was not found/); + expect(invocations.list().map((entry) => entry.id)).toEqual([third.id, second.id]); +}); + +it('bounds large intermediate snapshots by bytes without touching the final document', async () => { + const events = stream(64, 128 * 1024); + expect(bytesOf(events)).toBeGreaterThan(RENDER_EVENT_RETENTION.maxBytes); + const invocations = service(async (request, _signal, _kernel, publishRender) => { + for (const event of events) publishRender(event); + return childResult(request, events); + }); + const invocation = await invocations.invoke({ input: {}, routeId: echoRoute.id }); + + expect(invocation.events.length).toBeLessThan(events.length); + expect(bytesOf(invocation.events)).toBeLessThanOrEqual(RENDER_EVENT_RETENTION.maxBytes); + expect(invocation.evictedEvents).toBe(events.length - invocation.events.length); + expect(invocation.events.at(-1)).toMatchObject({ sequence: events.length - 1, type: 'complete' }); + expect(invocation.document).toEqual(finalDocument); + const replayed = collect(invocations, invocation.id); + expect(replayed[0]).toEqual({ type: 'truncated' }); + expect(replayed.filter((message) => message.type === 'render')).toHaveLength(invocation.events.length); +}); + +it('never evicts the newest event, however large', () => { + const oversized = stream(3, RENDER_EVENT_RETENTION.maxBytes); + const retained = retainRenderEvents(oversized); + expect(retained.events).toEqual([oversized[2]]); + expect(retained.evicted).toBe(2); +}); + +it('keeps the retained history on a failed run and omits eviction from summaries', async () => { + const events = stream(RENDER_EVENT_RETENTION.maxEvents + 5).slice(0, -1); + const invocations = service(async (_request, _signal, _kernel, publishRender) => { + for (const event of events) publishRender(event); + throw new Error('render exploded'); + }); + const failed = await invocations.invoke({ correlationId: 'browser-3', input: {}, routeId: echoRoute.id }); + expect(failed).toMatchObject({ correlationId: 'browser-3', evictedEvents: 4, status: 'failed' }); + expect(failed.events).toHaveLength(RENDER_EVENT_RETENTION.maxEvents); + expect(failed.document).toEqual(documentOf(`${String(events.length - 1)}:`)); + expect(invocations.list()[0]).not.toHaveProperty('evictedEvents'); + expect(collect(invocations, failed.id).at(-1)).toMatchObject({ invocation: { evictedEvents: 4 }, type: 'final' }); +}); + +it('serves a full count-bounded replay over the stream route without tripping the live-consumer queue', async () => { + const events = stream(RENDER_EVENT_RETENTION.maxEvents * 2); + const invocations = service(async (request, _signal, _kernel, publishRender) => { + for (const event of events) publishRender(event); + return childResult(request, 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(RENDER_EVENT_RETENTION.maxEvents); + expect(types.at(-1)).toBe('final'); + } finally { + server.close(); + } +}); + +it('reconnects with an explicit replay limitation and cancels with the latest document after the shell was evicted', async () => { + const rendered = deferred(); + // Shell plus replaces, never completing: the run stays cancellable. + const events = stream(RENDER_EVENT_RETENTION.maxEvents + 11).slice(0, -1); + const evicted = events.length - RENDER_EVENT_RETENTION.maxEvents; + const invocations = service((_request, signal, _kernel, publishRender) => new Promise((_resolve, reject) => { + for (const event of events) publishRender(event); + rendered.resolve(); + signal.addEventListener('abort', () => reject(signal.reason), { once: true }); + })); + const started = invocations.start({ correlationId: 'browser-2', input: {}, routeId: echoRoute.id }); + const live = collect(invocations, started.invocation.id); + await rendered.promise; + expect(live.filter((message) => message.type === 'render')).toHaveLength(events.length); + expect(live.filter((message) => message.type === 'truncated')).toHaveLength(1); + + const reconnected = collect(invocations, started.invocation.id); + expect(reconnected[0]).toEqual({ type: 'truncated' }); + expect(reconnected.filter((message) => message.type === 'render')).toHaveLength(RENDER_EVENT_RETENTION.maxEvents); + expect(reconnected.find((message) => message.type === 'render')).toMatchObject({ event: { sequence: evicted, type: 'replace' } }); + expect(reconnected.some((message) => message.type === 'final')).toBe(false); + + const cancelled = await invocations.cancel(started.invocation.id); + expect(cancelled).toMatchObject({ correlationId: 'browser-2', evictedEvents: evicted, status: 'cancelled' }); + expect(cancelled).not.toHaveProperty('outcome'); + expect(cancelled.events).toHaveLength(RENDER_EVENT_RETENTION.maxEvents); + expect(cancelled.document).toEqual(documentOf(`${String(events.length - 1)}:`)); + expect(reconnected.at(-1)).toMatchObject({ invocation: { id: started.invocation.id, status: 'cancelled' }, type: 'final' }); +}); diff --git a/packages/workbench/src/application/invocation-client.ts b/packages/workbench/src/application/invocation-client.ts index 2f6dee33f..3df1cb09d 100644 --- a/packages/workbench/src/application/invocation-client.ts +++ b/packages/workbench/src/application/invocation-client.ts @@ -157,6 +157,7 @@ const invocationSchema: z.ZodType = z.strictObject({ context: requestContextProvenanceSchema, document: agentDocumentSchema.optional(), events: z.array(agentRenderEventSchema), + evictedEvents: z.number().int().positive().optional(), projection: projectionSchema, providers: z.array(providerSchema), result: z.json().optional(), diff --git a/packages/workbench/src/application/invocation-model.ts b/packages/workbench/src/application/invocation-model.ts index 4ff670c08..345358a1c 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,9 +32,10 @@ export type InvocationState = | Readonly<{ readonly phase: 'idle' }> | Readonly<{ readonly correlationId: string; - readonly events?: readonly AgentRenderEvent[]; readonly invocationId?: string; readonly phase: 'running'; + /** The live render window, bounded like the server's (`RENDER_EVENT_RETENTION`). */ + readonly retained?: Readonly; readonly startedAt: number; }> | Readonly<{ readonly durationMs?: number; readonly invocation: RouteInvocation; readonly phase: 'succeeded' }> @@ -61,8 +65,17 @@ 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; +const retainLiveRenderEvent = ( + retained: Readonly | undefined, + event: AgentRenderEvent, +): Readonly => { + const next = retained === undefined + ? emptyRetainedRenderEvents() + : { bytes: retained.bytes, events: [...retained.events], evicted: retained.evicted }; + retainRenderEvent(next, event); + Object.freeze(next.events); + return Object.freeze(next); +}; export const reduceInvocationState = (state: InvocationState, action: InvocationAction): InvocationState => { switch (action.type) { @@ -79,7 +92,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, retained: retainLiveRenderEvent(state.retained, action.event) }) : 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..f7214a9f0 100644 --- a/packages/workbench/src/application/result-tabs.tsx +++ b/packages/workbench/src/application/result-tabs.tsx @@ -93,6 +93,9 @@ const RawDocument = ({ invocation }: { readonly invocation?: RouteInvocation }):

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

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

The {String(invocation.evictedEvents)} oldest events were evicted from retained history.

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

The stream carried no events.

:
    {invocation.events.map((event) =>
  1. @@ -207,7 +210,7 @@ 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 ? controller.state.retained?.events ?? [] : invocation?.events ?? []; const traceState = useTraceEntries(trace); const definitions: readonly ResultTabDefinition[] = [ { id: 'rendered', label: coreTabLabels.rendered, render: () => expect(reduceInvocationState(running, { type: 'reset' })).toBe(idleInvocationState); }); -it('retains only the newest 256 live render events', () => { +it('retains only the newest 256 live render events and counts the evicted ones', () => { let state = reduceInvocationState(idleInvocationState, { correlationId: 'c1', startedAt: 1_000, type: 'start' }); for (let sequence = 0; sequence < 300; sequence += 1) { state = reduceInvocationState(state, { @@ -111,9 +111,31 @@ it('retains only the newest 256 live render events', () => { expect(state).toMatchObject({ phase: 'running' }); 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.retained?.events).toHaveLength(256); + expect(state.retained?.events[0]?.sequence).toBe(44); + expect(state.retained?.events.at(-1)?.sequence).toBe(299); + expect(state.retained?.evicted).toBe(44); +}); + +it('bounds live render events by retained bytes like the server does', () => { + let state = reduceInvocationState(idleInvocationState, { correlationId: 'c1', startedAt: 1_000, type: 'start' }); + for (let sequence = 0; sequence < 24; sequence += 1) { + state = reduceInvocationState(state, { + event: { + boundaryId: 'b', + document: { root: { kind: 'text', text: 'x'.repeat(128 * 1024) }, status: 'success', version: 1 }, + sequence, + type: 'replace', + }, + type: 'render', + }); + } + + if (state.phase !== 'running') throw new Error('Expected a running invocation.'); + expect(state.retained!.events.length).toBeLessThan(24); + expect(state.retained?.bytes).toBeLessThanOrEqual(RENDER_EVENT_RETENTION.maxBytes); + expect(state.retained?.evicted).toBe(24 - state.retained!.events.length); + expect(state.retained?.events.at(-1)?.sequence).toBe(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..6c42fbed8 100644 --- a/packages/workbench/tests/route-workspace.test.ts +++ b/packages/workbench/tests/route-workspace.test.ts @@ -92,9 +92,9 @@ describe('invocation state contract', () => { type: 'render', }); expect(progressed).toMatchObject({ - events: [expect.objectContaining({ type: 'shell' }), expect.objectContaining({ type: 'progress' })], invocationId: 'inv-live', phase: 'running', + retained: { events: [expect.objectContaining({ type: 'shell' }), expect.objectContaining({ type: 'progress' })], evicted: 0 }, }); const { outcome: _outcome, ...withoutOutcome } = invocation; diff --git a/website/docs/en/reference/dev-server-http.mdx b/website/docs/en/reference/dev-server-http.mdx index 7e1f2658a..1f74de9cd 100644 --- a/website/docs/en/reference/dev-server-http.mdx +++ b/website/docs/en/reference/dev-server-http.mdx @@ -25,8 +25,17 @@ 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 history is bounded by one policy everywhere it is kept — the render child, +its reply, the stream replay, the completed envelope, history reads, and the Workbench's live +buffer: the newest 256 render events whose JSON totals at most 1 MiB, evicting oldest-first. The +newest event always survives, however large, so a completed run keeps its `complete` event; the +Agent Document, `result`, `outcome`, and correlation identifiers live outside the window and are +never truncated. An envelope whose window lost events carries `evictedEvents` (the count), and a +stream replay opens with one `truncated` event before the retained window; a cancelled or failed +run keeps its retained window and the newest document a render event carried, even after that +event was evicted. 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/reference/dev-server-http.mdx b/website/docs/zh/reference/dev-server-http.mdx index f7d383d10..432c91953 100644 --- a/website/docs/zh/reference/dev-server-http.mdx +++ b/website/docs/zh/reference/dev-server-http.mdx @@ -21,7 +21,12 @@ description: 'Workbench 调用、统一 Trace、Raw logs 与宿主钩子收据 POST 正文是一份 `RouteInvocationRequest`：包含 `routeId`，以及可选的 `input`、`args`、 `correlationId`、调用方 `requestId` 与事件夹具选项。前台会把这两个标识回显到调用上，使路由工作区与 Trace 可以关联此次运行。完成后的信封包含输入、provenance 上下文、providers、渲染事件、Agent -Document、结果、投影、诊断与计时。Trace 条目链接到这份完整快照，而不会复制它。取消后的调用带有 +Document、结果、投影、诊断与计时。Trace 条目链接到这份完整快照，而不会复制它。渲染历史在所有保存它的 +地方都遵循同一条策略——渲染子进程、其回复、流回放、完成后的信封、历史读取以及 Workbench 的实时缓冲： +只保留最新的 256 个渲染事件，且其 JSON 总量不超过 1 MiB，按最旧优先淘汰。最新的事件无论多大都始终保留， +因此完成的运行总会保留其 `complete` 事件；Agent Document、`result`、`outcome` 与关联标识位于该窗口之外， +永不截断。窗口丢失过事件的信封带有 `evictedEvents`（数量），流回放会在保留窗口之前先发出一个 `truncated` +事件；被取消或失败的运行仍保留其保留窗口，以及渲染事件带来的最新文档——即使承载该文档的事件已被淘汰。取消后的调用带有 `status: "cancelled"` 且不含 `outcome`；取消已终止的调用会返回 `AB8256`（409）。 ## 宿主会话