From e1a4b0caa27f8a6a6e01b1ccebe91cfcdc21b1cd Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 6 Sep 2026 20:18:50 +0000 Subject: [PATCH 1/6] fix(dev): bound route-invocation render history end to end --- .../681-bounded-invocation-render-history.md | 5 + .../dev/routes/route-invocation-production.ts | 2 + .../src/dev/routes/route-invocation-result.ts | 9 +- .../dev/routes/route-invocation-service.ts | 13 +-- .../tests/route-invocation-retention.test.ts | 99 +++++++++++++++++++ website/docs/en/reference/dev-server-http.mdx | 8 +- website/docs/zh/reference/dev-server-http.mdx | 6 +- 7 files changed, 131 insertions(+), 11 deletions(-) create mode 100644 .changeset/681-bounded-invocation-render-history.md create mode 100644 packages/agent-bundle/tests/route-invocation-retention.test.ts diff --git a/.changeset/681-bounded-invocation-render-history.md b/.changeset/681-bounded-invocation-render-history.md new file mode 100644 index 000000000..8a22c33d7 --- /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: the render child, the `GET /api/routes/invocations//stream` replay, and the completed `RouteInvocation` envelope (`events`, as returned by `POST /api/routes/invocations` and `GET /api/routes/invocations/`) each keep only the newest 256 render events, evicting oldest-first while preserving the final `complete` event, `document`, `outcome`, and correlation identifiers. (#PR) 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..2a7e3d125 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 { MAX_RETAINED_RENDER_EVENTS } from './route-invocation-result.ts'; import type { RouteInvocationProvider, RouteInvocationTiming } from './route-invocation.ts'; interface CompiledCliInvocationModule { @@ -463,6 +464,7 @@ const renderCompiled = async ( for (;;) { const next = await reader.read(); if (next.done) break; + if (events.length === MAX_RETAINED_RENDER_EVENTS) events.shift(); events.push(next.value); publishRender?.(next.value); } 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..09522f5d5 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-result.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-result.ts @@ -10,11 +10,18 @@ import type { RouteInvocationSummary, } from './route-invocation.ts'; +/** + * Render events one invocation retains — in the producer while it streams, in + * the live stream replay, and in the completed envelope. Older events are + * evicted oldest-first; the final `complete` event and `document` survive. + */ +export const MAX_RETAINED_RENDER_EVENTS = 256; + 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 newest `MAX_RETAINED_RENDER_EVENTS` of the production `shell | progress | replace | error | complete` stream, in order. */ readonly events: readonly AgentRenderEvent[]; readonly projection: RouteInvocationProjection; readonly providers: readonly RouteInvocationProvider[]; 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..553f39406 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,11 @@ import { isProductionRouteInvocationCode, ProductionRouteInvocationError, } from './route-invocation-production-error.ts'; -import type { - RouteInvocation, - RouteInvocationStart, - RouteInvocationStreamMessage, +import { + MAX_RETAINED_RENDER_EVENTS, + type RouteInvocation, + type RouteInvocationStart, + type RouteInvocationStreamMessage, } from './route-invocation-result.ts'; import { nativeEventRequestContext } from './route-invocation.ts'; import type { @@ -1318,7 +1319,7 @@ export class RouteInvocationService { if (message.type === 'render') { const renderCount = record.messages.reduce((count, retained) => count + (retained.type === 'render' ? 1 : 0), 0); - if (renderCount === 256) { + if (renderCount === MAX_RETAINED_RENDER_EVENTS) { 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'); @@ -1607,7 +1608,7 @@ export class RouteInvocationService { }, } : {}), - events: child.events, + events: child.events.slice(-MAX_RETAINED_RENDER_EVENTS), id, input: canonical ?? child.input, kind: route.kind as RouteInvocationKind, 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..a6c93bd01 --- /dev/null +++ b/packages/agent-bundle/tests/route-invocation-retention.test.ts @@ -0,0 +1,99 @@ +import { expect, it } from '@rstest/core'; + +import type { AgentRenderEvent } from '@agent-bundle/runtime'; + +import { MAX_RETAINED_RENDER_EVENTS, type RouteInvocationStreamMessage } from '../src/dev/routes/route-invocation-result.ts'; +import { + RouteInvocationService, + type RouteInvocationChildRequest, + type RouteInvocationChildResult, +} 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 document = { root: { kind: 'text', text: 'ok' }, status: 'success', version: 1 } as const; +const streamLength = MAX_RETAINED_RENDER_EVENTS * 4; + +const longRender = async ( + request: RouteInvocationChildRequest, + _signal: AbortSignal, + _publishKernelEvent: unknown, + publishRender: (event: AgentRenderEvent) => void, +): Promise => { + const events: AgentRenderEvent[] = []; + for (let sequence = 0; sequence < streamLength; sequence += 1) { + const event: AgentRenderEvent = sequence === streamLength - 1 + ? { document, sequence, type: 'complete' } + : { document, sequence, type: 'shell' }; + events.push(event); + publishRender(event); + } + return { document, events, input: request.input, mcp: {}, renderDurationMs: 1 }; +}; + +const service = (historyLimit: number): 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: longRender, +}); + +it('bounds a long render stream in the live replay and the completed envelope, keeping the outcome and correlation', async () => { + const invocations = service(2); + const started = invocations.start({ correlationId: 'browser-1', input: {}, routeId: echoRoute.id }); + const invocation = await started.result; + const messages: RouteInvocationStreamMessage[] = []; + invocations.subscribe(invocation.id, (message) => messages.push(message)); + + expect(invocation.events).toHaveLength(MAX_RETAINED_RENDER_EVENTS); + expect(invocation.events[0]?.sequence).toBe(streamLength - MAX_RETAINED_RENDER_EVENTS); + expect(invocation.events.at(-1)).toMatchObject({ sequence: streamLength - 1, type: 'complete' }); + expect(invocation).toMatchObject({ + correlationId: 'browser-1', + document, + outcome: { kind: 'success' }, + routeId: echoRoute.id, + status: 'succeeded', + }); + expect(invocations.read(invocation.id)?.events).toHaveLength(MAX_RETAINED_RENDER_EVENTS); + + expect(messages.filter((message) => message.type === 'render')).toHaveLength(MAX_RETAINED_RENDER_EVENTS); + expect(messages.filter((message) => message.type === 'truncated')).toHaveLength(1); + const final = messages.at(-1); + expect(final?.type).toBe('final'); + if (final?.type !== 'final') throw new Error('The stream did not end with the final envelope.'); + expect(final.invocation.events).toHaveLength(MAX_RETAINED_RENDER_EVENTS); + expect(final.invocation.id).toBe(invocation.id); + + 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]); + expect(invocations.read(second.id)?.events).toHaveLength(MAX_RETAINED_RENDER_EVENTS); +}); diff --git a/website/docs/en/reference/dev-server-http.mdx b/website/docs/en/reference/dev-server-http.mdx index 7e1f2658a..661d55ff5 100644 --- a/website/docs/en/reference/dev-server-http.mdx +++ b/website/docs/en/reference/dev-server-http.mdx @@ -25,8 +25,12 @@ 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: the stream replay, the completed envelope, and the +render child each keep only the newest 256 render events, evicting oldest-first; the stream +reports the eviction with one `truncated` event, and the final `complete` event, Agent Document, +`outcome`, and correlation identifiers always survive. 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..5b2026e17 100644 --- a/website/docs/zh/reference/dev-server-http.mdx +++ b/website/docs/zh/reference/dev-server-http.mdx @@ -21,8 +21,10 @@ 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 个渲染事件，按最旧优先淘汰；流会以一个 +`truncated` 事件报告这次淘汰，而最终的 `complete` 事件、Agent Document、`outcome` 与关联标识始终保留。 +取消后的调用带有 `status: "cancelled"` 且不含 `outcome`；取消已终止的调用会返回 `AB8256`（409）。 ## 宿主会话 From 30e0eed83b53b48bafe5520c2dd00889227fa754 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 6 Sep 2026 20:51:34 +0000 Subject: [PATCH 2/6] fix(dev): one render-history retention policy with count and byte bounds Producer, IPC reply, live replay, completed envelope, history, and the Workbench buffer share RENDER_EVENT_RETENTION (256 events, 1 MiB JSON). Envelopes report evictedEvents; cancellation keeps the latest document after eviction; SSE replay is paced by drain instead of the live queue. --- .../681-bounded-invocation-render-history.md | 2 +- .../agent-bundle/src/contracts/invocations.ts | 11 +- .../src/dev/routes/route-invocation-child.ts | 14 +- .../dev/routes/route-invocation-production.ts | 16 +- .../src/dev/routes/route-invocation-result.ts | 53 ++++- .../src/dev/routes/route-invocation-routes.ts | 44 +++- .../dev/routes/route-invocation-service.ts | 76 ++++--- .../tests/route-invocation-dev-server.test.ts | 200 ++++++++++++++++-- .../tests/route-invocation-retention.test.ts | 153 ++++++++++---- .../src/application/invocation-client.ts | 1 + .../src/application/invocation-model.ts | 31 ++- .../workbench/src/application/result-tabs.tsx | 5 +- .../workbench/tests/invocation-model.test.ts | 32 ++- .../workbench/tests/route-workspace.test.ts | 2 +- website/docs/en/reference/dev-server-http.mdx | 16 +- website/docs/zh/reference/dev-server-http.mdx | 11 +- 16 files changed, 518 insertions(+), 149 deletions(-) diff --git a/.changeset/681-bounded-invocation-render-history.md b/.changeset/681-bounded-invocation-render-history.md index 8a22c33d7..2241692f5 100644 --- a/.changeset/681-bounded-invocation-render-history.md +++ b/.changeset/681-bounded-invocation-render-history.md @@ -2,4 +2,4 @@ 'agent-bundle': patch --- -Bound route-invocation render history end to end: the render child, the `GET /api/routes/invocations//stream` replay, and the completed `RouteInvocation` envelope (`events`, as returned by `POST /api/routes/invocations` and `GET /api/routes/invocations/`) each keep only the newest 256 render events, evicting oldest-first while preserving the final `complete` event, `document`, `outcome`, and correlation identifiers. (#PR) +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`; a cancelled run keeps 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. (#PR) 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 2a7e3d125..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,7 +30,7 @@ import { ROUTE_INVOCATION_COMPILED_ROUTE_UNAVAILABLE_CODE, ROUTE_INVOCATION_PREPARATION_FAILURE_CODE, } from './route-invocation-production-error.ts'; -import { MAX_RETAINED_RENDER_EVENTS } from './route-invocation-result.ts'; +import { emptyRetainedRenderEvents, retainRenderEvent } from './route-invocation-result.ts'; import type { RouteInvocationProvider, RouteInvocationTiming } from './route-invocation.ts'; interface CompiledCliInvocationModule { @@ -448,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[]; @@ -458,22 +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; - if (events.length === MAX_RETAINED_RENDER_EVENTS) events.shift(); - 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]), @@ -560,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 09522f5d5..9be0c9af6 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-result.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-result.ts @@ -11,18 +11,61 @@ import type { } from './route-invocation.ts'; /** - * Render events one invocation retains — in the producer while it streams, in - * the live stream replay, and in the completed envelope. Older events are - * evicted oldest-first; the final `complete` event and `document` survive. + * 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 MAX_RETAINED_RENDER_EVENTS = 256; +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 newest `MAX_RETAINED_RENDER_EVENTS` of 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..f6de1caaf 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-routes.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-routes.ts @@ -258,6 +258,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 +272,46 @@ export class RouteInvocationRoutes { stream.unsubscribe?.(); response.end(); }; - const writer = createBackpressuredWriter(response, { - byteLimit: streamQueueByteLimit, - onIdle: finish, - 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(); }; + const replay: RouteInvocationStreamMessage[] = []; + let replaying = true; + let liveWhileReplaying = 0; + const pump = (): void => { + while (replaying && writer.idle && !response.destroyed) { + const next = replay.shift(); + if (next === undefined) { + replaying = false; + break; + } + deliver(next); + } + finish(); + }; + const writer = createBackpressuredWriter(response, { + byteLimit: streamQueueByteLimit, + onIdle: pump, + recordLimit: streamQueueEntryLimit, + }); 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) => { + if (!replaying) return deliver(message); + replay.push(message); + liveWhileReplaying += 1; + if (liveWhileReplaying > streamQueueEntryLimit) response.destroy(); + }); + liveWhileReplaying = 0; 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 553f39406..99abd9c57 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-service.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-service.ts @@ -48,7 +48,10 @@ import { ProductionRouteInvocationError, } from './route-invocation-production-error.ts'; import { - MAX_RETAINED_RENDER_EVENTS, + emptyRetainedRenderEvents, + retainRenderEvent, + retainRenderEvents, + type RetainedRenderEvents, type RouteInvocation, type RouteInvocationStart, type RouteInvocationStreamMessage, @@ -154,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 @@ -1174,47 +1180,47 @@ 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' }); + +/** 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); + const { latestDocument: document, renders } = input.record; return deepFreeze({ completedAt: input.completedAt.toISOString(), context: input.context, ...(input.request.correlationId === undefined ? {} : { correlationId: input.request.correlationId }), diagnostics: [], ...(document === undefined ? {} : { document }), - events, + events: [...renders.events], + ...(renders.evicted === 0 ? {} : { evictedEvents: renders.evicted }), id: input.id, input: input.request.input ?? {}, kind: input.route.kind as RouteInvocationKind, @@ -1281,7 +1287,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); } @@ -1316,22 +1322,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 === MAX_RETAINED_RENDER_EVENTS) { - 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); } @@ -1410,6 +1412,7 @@ export class RouteInvocationService { controller: operationController, listeners: new Set(), messages: [], + renders: emptyRetainedRenderEvents(), running: runningInvocation, }; this.#streams.set(id, streamRecord); @@ -1550,7 +1553,7 @@ export class RouteInvocationService { context, id, manifest, - messages: streamRecord.messages, + record: streamRecord, request: { ...request, input }, route, startedAt, @@ -1586,6 +1589,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; @@ -1608,7 +1613,8 @@ export class RouteInvocationService { }, } : {}), - events: child.events.slice(-MAX_RETAINED_RENDER_EVENTS), + events: retained.events, + ...(evictedEvents === 0 ? {} : { evictedEvents }), id, input: canonical ?? child.input, kind: route.kind as RouteInvocationKind, @@ -1637,7 +1643,7 @@ export class RouteInvocationService { context: cancellationContext, id, manifest: queued, - messages: streamRecord.messages, + 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 index a6c93bd01..ddb484b62 100644 --- a/packages/agent-bundle/tests/route-invocation-retention.test.ts +++ b/packages/agent-bundle/tests/route-invocation-retention.test.ts @@ -1,14 +1,21 @@ import { expect, it } from '@rstest/core'; -import type { AgentRenderEvent } from '@agent-bundle/runtime'; +import type { AgentDocument, AgentRenderEvent } from '@agent-bundle/runtime'; -import { MAX_RETAINED_RENDER_EVENTS, type RouteInvocationStreamMessage } from '../src/dev/routes/route-invocation-result.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: [], @@ -29,27 +36,21 @@ const manifest: RouteManifest = { sourceRevision: 'revision', }; -const document = { root: { kind: 'text', text: 'ok' }, status: 'success', version: 1 } as const; -const streamLength = MAX_RETAINED_RENDER_EVENTS * 4; - -const longRender = async ( - request: RouteInvocationChildRequest, - _signal: AbortSignal, - _publishKernelEvent: unknown, - publishRender: (event: AgentRenderEvent) => void, -): Promise => { - const events: AgentRenderEvent[] = []; - for (let sequence = 0; sequence < streamLength; sequence += 1) { - const event: AgentRenderEvent = sequence === streamLength - 1 - ? { document, sequence, type: 'complete' } - : { document, sequence, type: 'shell' }; - events.push(event); - publishRender(event); - } - return { document, events, input: request.input, mcp: {}, renderDurationMs: 1 }; -}; +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 = (historyLimit: number): RouteInvocationService => new RouteInvocationService({ +const service = ( + renderChild: NonNullable, + historyLimit = 2, +): RouteInvocationService => new RouteInvocationService({ historyLimit, manifest: { manifest: () => manifest }, prepared: async () => ({ @@ -60,40 +61,110 @@ const service = (historyLimit: number): RouteInvocationService => new RouteInvoc }, release: () => undefined, }), - renderChild: longRender, + renderChild, +}); + +const childResult = (request: RouteInvocationChildRequest, events: readonly AgentRenderEvent[]): RouteInvocationChildResult => ({ + document: finalDocument, + events, + input: request.input, + mcp: {}, + renderDurationMs: 1, }); -it('bounds a long render stream in the live replay and the completed envelope, keeping the outcome and correlation', async () => { - const invocations = service(2); - const started = invocations.start({ correlationId: 'browser-1', input: {}, routeId: echoRoute.id }); - const invocation = await started.result; +const collect = (invocations: RouteInvocationService, id: string): RouteInvocationStreamMessage[] => { const messages: RouteInvocationStreamMessage[] = []; - invocations.subscribe(invocation.id, (message) => messages.push(message)); + invocations.subscribe(id, (message) => messages.push(message)); + return messages; +}; + +const bytesOf = (events: readonly AgentRenderEvent[]): number => events.reduce((sum, event) => sum + renderEventBytes(event), 0); - expect(invocation.events).toHaveLength(MAX_RETAINED_RENDER_EVENTS); - expect(invocation.events[0]?.sequence).toBe(streamLength - MAX_RETAINED_RENDER_EVENTS); - expect(invocation.events.at(-1)).toMatchObject({ sequence: streamLength - 1, type: 'complete' }); +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, + document: finalDocument, outcome: { kind: 'success' }, routeId: echoRoute.id, status: 'succeeded', }); - expect(invocations.read(invocation.id)?.events).toHaveLength(MAX_RETAINED_RENDER_EVENTS); + expect(invocations.read(invocation.id)).toBe(invocation); - expect(messages.filter((message) => message.type === 'render')).toHaveLength(MAX_RETAINED_RENDER_EVENTS); - expect(messages.filter((message) => message.type === 'truncated')).toHaveLength(1); - const final = messages.at(-1); - expect(final?.type).toBe('final'); - if (final?.type !== 'final') throw new Error('The stream did not end with the final envelope.'); - expect(final.invocation.events).toHaveLength(MAX_RETAINED_RENDER_EVENTS); - expect(final.invocation.id).toBe(invocation.id); + 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]); - expect(invocations.read(second.id)?.events).toHaveLength(MAX_RETAINED_RENDER_EVENTS); +}); + +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('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 661d55ff5..f732ec084 100644 --- a/website/docs/en/reference/dev-server-http.mdx +++ b/website/docs/en/reference/dev-server-http.mdx @@ -25,12 +25,16 @@ 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. Render history is bounded: the stream replay, the completed envelope, and the -render child each keep only the newest 256 render events, evicting oldest-first; the stream -reports the eviction with one `truncated` event, and the final `complete` event, Agent Document, -`outcome`, and correlation identifiers always survive. 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 run keeps +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 5b2026e17..52c2dd1e5 100644 --- a/website/docs/zh/reference/dev-server-http.mdx +++ b/website/docs/zh/reference/dev-server-http.mdx @@ -21,10 +21,13 @@ description: 'Workbench 调用、统一 Trace、Raw logs 与宿主钩子收据 POST 正文是一份 `RouteInvocationRequest`：包含 `routeId`，以及可选的 `input`、`args`、 `correlationId`、调用方 `requestId` 与事件夹具选项。前台会把这两个标识回显到调用上，使路由工作区与 Trace 可以关联此次运行。完成后的信封包含输入、provenance 上下文、providers、渲染事件、Agent -Document、结果、投影、诊断与计时。Trace 条目链接到这份完整快照，而不会复制它。渲染历史是有界的： -流回放、完成后的信封与渲染子进程各自只保留最新的 256 个渲染事件，按最旧优先淘汰；流会以一个 -`truncated` 事件报告这次淘汰，而最终的 `complete` 事件、Agent Document、`outcome` 与关联标识始终保留。 -取消后的调用带有 `status: "cancelled"` 且不含 `outcome`；取消已终止的调用会返回 `AB8256`（409）。 +Document、结果、投影、诊断与计时。Trace 条目链接到这份完整快照，而不会复制它。渲染历史在所有保存它的 +地方都遵循同一条策略——渲染子进程、其回复、流回放、完成后的信封、历史读取以及 Workbench 的实时缓冲： +只保留最新的 256 个渲染事件，且其 JSON 总量不超过 1 MiB，按最旧优先淘汰。最新的事件无论多大都始终保留， +因此完成的运行总会保留其 `complete` 事件；Agent Document、`result`、`outcome` 与关联标识位于该窗口之外， +永不截断。窗口丢失过事件的信封带有 `evictedEvents`（数量），流回放会在保留窗口之前先发出一个 `truncated` +事件；被取消的运行即使在承载文档的事件被淘汰后，仍保留渲染事件带来的最新文档。取消后的调用带有 +`status: "cancelled"` 且不含 `outcome`；取消已终止的调用会返回 `AB8256`（409）。 ## 宿主会话 From 544a30423b59472aa0391db429c76c1e28f9654a Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 6 Sep 2026 20:56:14 +0000 Subject: [PATCH 3/6] chore: reference #699 in changeset --- .changeset/681-bounded-invocation-render-history.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/681-bounded-invocation-render-history.md b/.changeset/681-bounded-invocation-render-history.md index 2241692f5..065404bf6 100644 --- a/.changeset/681-bounded-invocation-render-history.md +++ b/.changeset/681-bounded-invocation-render-history.md @@ -2,4 +2,4 @@ '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`; a cancelled run keeps 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. (#PR) +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`; a cancelled run keeps 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) From 7eff211bab2349968c8f31a6c5c1c7a55075f27d Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 6 Sep 2026 21:15:21 +0000 Subject: [PATCH 4/6] fix(dev): keep retained history on failed runs, omit evictedEvents from summaries, count only live arrivals during replay --- .../src/dev/routes/route-invocation-routes.ts | 6 +- .../dev/routes/route-invocation-service.ts | 60 +++++++++++-------- .../tests/route-invocation-retention.test.ts | 48 +++++++++++++++ 3 files changed, 87 insertions(+), 27 deletions(-) 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 f6de1caaf..9850b75b6 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-routes.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-routes.ts @@ -280,6 +280,9 @@ export class RouteInvocationRoutes { }; const replay: RouteInvocationStreamMessage[] = []; let replaying = true; + let subscribed = false; + // Live messages that arrive while the replay is still draining are bounded + // like the live queue; the replay itself is bounded by retention. let liveWhileReplaying = 0; const pump = (): void => { while (replaying && writer.idle && !response.destroyed) { @@ -304,10 +307,11 @@ export class RouteInvocationRoutes { stream.unsubscribe = service.subscribe(id, (message) => { if (!replaying) return deliver(message); replay.push(message); + if (!subscribed) return; liveWhileReplaying += 1; if (liveWhileReplaying > streamQueueEntryLimit) response.destroy(); }); - liveWhileReplaying = 0; + subscribed = true; writeKeepAliveStreamHead(response, { cacheControl: 'no-cache', contentType: 'text/event-stream; charset=utf-8', 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 99abd9c57..9ddf6b362 100644 --- a/packages/agent-bundle/src/dev/routes/route-invocation-service.ts +++ b/packages/agent-bundle/src/dev/routes/route-invocation-service.ts @@ -298,6 +298,7 @@ export const invocationSummary = (invocation: RouteInvocation): RouteInvocationS context: _context, document: _document, events: _events, + evictedEvents: _evictedEvents, projection: _projection, providers: _providers, result: _result, @@ -1147,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; @@ -1161,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, @@ -1194,6 +1196,15 @@ interface InvocationStreamRecord { 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]), @@ -1211,31 +1222,26 @@ const cancelledInvocation = (input: { readonly startedAt: Date; readonly surface: RouteInvocationSurface; readonly completedAt: Date; -}): RouteInvocation => { - const { latestDocument: document, renders } = input.record; - return deepFreeze({ - completedAt: input.completedAt.toISOString(), - context: input.context, - ...(input.request.correlationId === undefined ? {} : { correlationId: input.request.correlationId }), - diagnostics: [], - ...(document === undefined ? {} : { document }), - events: [...renders.events], - ...(renders.evicted === 0 ? {} : { evictedEvents: renders.evicted }), - 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[] = []; @@ -1576,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, @@ -1661,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-retention.test.ts b/packages/agent-bundle/tests/route-invocation-retention.test.ts index ddb484b62..27b3fa398 100644 --- a/packages/agent-bundle/tests/route-invocation-retention.test.ts +++ b/packages/agent-bundle/tests/route-invocation-retention.test.ts @@ -1,7 +1,11 @@ +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, @@ -139,6 +143,50 @@ it('never evicts the newest event, however large', () => { 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. From 109416e2f5fb0f7d808384765a9601d4a4e0b5f7 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 6 Sep 2026 21:15:37 +0000 Subject: [PATCH 5/6] docs: failed runs keep retained render history --- .changeset/681-bounded-invocation-render-history.md | 2 +- website/docs/en/reference/dev-server-http.mdx | 5 +++-- website/docs/zh/reference/dev-server-http.mdx | 2 +- 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/.changeset/681-bounded-invocation-render-history.md b/.changeset/681-bounded-invocation-render-history.md index 065404bf6..ce1c6e50b 100644 --- a/.changeset/681-bounded-invocation-render-history.md +++ b/.changeset/681-bounded-invocation-render-history.md @@ -2,4 +2,4 @@ '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`; a cancelled run keeps 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) +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/website/docs/en/reference/dev-server-http.mdx b/website/docs/en/reference/dev-server-http.mdx index f732ec084..1f74de9cd 100644 --- a/website/docs/en/reference/dev-server-http.mdx +++ b/website/docs/en/reference/dev-server-http.mdx @@ -31,8 +31,9 @@ buffer: the newest 256 render events whose JSON totals at most 1 MiB, evicting o 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 run keeps -the newest document a render event carried even after that event was evicted. A cancelled +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). diff --git a/website/docs/zh/reference/dev-server-http.mdx b/website/docs/zh/reference/dev-server-http.mdx index 52c2dd1e5..432c91953 100644 --- a/website/docs/zh/reference/dev-server-http.mdx +++ b/website/docs/zh/reference/dev-server-http.mdx @@ -26,7 +26,7 @@ Document、结果、投影、诊断与计时。Trace 条目链接到这份完整 只保留最新的 256 个渲染事件，且其 JSON 总量不超过 1 MiB，按最旧优先淘汰。最新的事件无论多大都始终保留， 因此完成的运行总会保留其 `complete` 事件；Agent Document、`result`、`outcome` 与关联标识位于该窗口之外， 永不截断。窗口丢失过事件的信封带有 `evictedEvents`（数量），流回放会在保留窗口之前先发出一个 `truncated` -事件；被取消的运行即使在承载文档的事件被淘汰后，仍保留渲染事件带来的最新文档。取消后的调用带有 +事件；被取消或失败的运行仍保留其保留窗口，以及渲染事件带来的最新文档——即使承载该文档的事件已被淘汰。取消后的调用带有 `status: "cancelled"` 且不含 `outcome`；取消已终止的调用会返回 `AB8256`（409）。 ## 宿主会话 From a7f3b53078cd7036f86c3ec1be48c51c2df49179 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sun, 6 Sep 2026 21:22:44 +0000 Subject: [PATCH 6/6] fix(dev): bound the live backlog behind a stream replay by current records and bytes --- .../src/dev/routes/route-invocation-routes.ts | 48 +++++++++++++------ 1 file changed, 34 insertions(+), 14 deletions(-) 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 9850b75b6..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.', @@ -272,25 +286,29 @@ export class RouteInvocationRoutes { stream.unsubscribe?.(); response.end(); }; - 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; + const deliver = (frame: PendingFrame): void => { + if (writer.enqueue(frame.text) === 'overflow') response.destroy(); + if (frame.final) terminal = true; finish(); }; - const replay: RouteInvocationStreamMessage[] = []; + // 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; - // Live messages that arrive while the replay is still draining are bounded - // like the live queue; the replay itself is bounded by retention. - let liveWhileReplaying = 0; + let liveBytes = 0; + let liveRecords = 0; const pump = (): void => { while (replaying && writer.idle && !response.destroyed) { - const next = replay.shift(); + const next = pending.shift(); if (next === undefined) { replaying = false; break; } + if (next.live) { + liveBytes -= next.bytes; + liveRecords -= 1; + } deliver(next); } finish(); @@ -305,11 +323,13 @@ export class RouteInvocationRoutes { stream.unsubscribe?.(); }); stream.unsubscribe = service.subscribe(id, (message) => { - if (!replaying) return deliver(message); - replay.push(message); - if (!subscribed) return; - liveWhileReplaying += 1; - if (liveWhileReplaying > streamQueueEntryLimit) response.destroy(); + 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, {