fix(workbench): bound producer / completed-envelope render history end to end

.changeset/681-bound-invocation-render-history.md

@@ -0,0 +1,5 @@
+---
+'agent-bundle': patch
+---
+
+Bound every retained copy of a Workbench route invocation's render stream under one render-history window — at most 256 events and 2 MiB serialized, always keeping the newest event and the newest `shell`/`replace`/`complete` event, which alone may exceed 2 MiB — applied alike to the live `GET /api/routes/invocations/<id>/stream` replay, the completed and cancelled envelopes, `GET /api/routes/invocations/<id>`, the terminal `final` message, the Workbench's live view, and the envelopes the Workbench synthesizes from runtime runs. Stream replay is paced by socket drain instead of queued, so a reconnect can take a whole retained window through a backpressured socket. The render child no longer returns the whole event stream over IPC and the compiled-route producer keeps only the `complete` document. Envelopes whose events were evicted carry a new optional `retention` field (`producedEvents`, `evictedEvents`, `evictedBytes`, `retainedBytes`), cancelled and failed envelopes keep the retained window and its latest document, compiled routes' progress reports now reach the Workbench as `progress` render events, and `document`, `result`, `outcome`, and `correlationId` are never truncated. (#715)

packages/agent-bundle/src/contracts/invocations.ts

@@ -1,7 +1,8 @@
 /**
  * Browser-consumable contract surface for dev-server route invocations — the
- * one execution path behind the Workbench route workspace. Type-only: routes
- * render on the server through the production runtime.
+ * one execution path behind the Workbench route workspace. Types, plus the
+ * render-history retention policy the browser's live window shares with the
+ * server; routes render on the server through the production runtime.
  */
 export type {
   RouteInvocationCliProjection,
@@ -29,4 +30,15 @@ export type {
   RouteInvocationStreamMessage,
   RunningRouteInvocationResponse,
 } from '../dev/routes/route-invocation-result.ts';
+export {
+  emptyRetainedRenderEvents,
+  renderRetention,
+  retainedLatestDocument,
+  retainedRenderEvents,
+  retainRenderEvent,
+  routeInvocationRenderHistoryLimits,
+  type RetainedRenderEvents,
+  type RouteInvocationRenderHistoryLimits,
+  type RouteInvocationRenderRetention,
+} from '../dev/routes/route-invocation-render-history.ts';
 export type { EventTraceEvent } from '../events/trace.ts';

packages/agent-bundle/src/dev/routes/route-invocation-child.ts

@@ -68,9 +68,9 @@ const forwardEventTrace = (event: EventTraceEvent): void => {
   process.send?.({ event, type: 'trace' } satisfies RouteInvocationChildResponse);
 };
 
-const forwardRenderEvent = (event: AgentRenderEvent): void => {
-  process.send?.({ event, type: 'render' } satisfies RouteInvocationChildResponse);
-};
+/** Awaited per event so the IPC channel, not an in-child queue, paces a fast producer. */
+const forwardRenderEvent = (event: AgentRenderEvent): Promise<void> =>
+  respond({ event, type: 'render' });
 
 /**
  * The exit code a generated executable would set for this unit render. There
@@ -134,10 +134,12 @@ const renderUnitRoute = async (request: RouteInvocationChildRequest): Promise<Ro
     trace?.failure('render', error);
     throw error;
   }
+  // The harness drains the stream before returning, so the unit surface's
+  // events reach the service after the render rather than live.
+  for (const event of rendered.events) await forwardRenderEvent(event);
   const exitCode = unitRenderExitCode(request, rendered.document, rendered.result ?? rendered.document.value);
   return Object.freeze({
     document: rendered.document,
-    events: rendered.events,
     ...(exitCode === undefined ? {} : { exitCode }),
     input,
     ...(request.manifest.routes[request.routeId]?.kind === 'tool'
@@ -152,15 +154,10 @@ const renderUnitRoute = async (request: RouteInvocationChildRequest): Promise<Ro
   });
 };
 
-const render = async (request: RouteInvocationChildRequest): Promise<RouteInvocationChildResult> => {
-  const result = request.surface.kind === 'unit-render'
-    ? await renderUnitRoute(request)
-    : await renderProductionRoute(request, forwardEventTrace, forwardRenderEvent);
-  if (request.surface.kind === 'unit-render') {
-    for (const event of result.events) forwardRenderEvent(event);
-  }
-  return result;
-};
+const render = (request: RouteInvocationChildRequest): Promise<RouteInvocationChildResult> =>
+  request.surface.kind === 'unit-render'
+    ? renderUnitRoute(request)
+    : renderProductionRoute(request, forwardEventTrace, forwardRenderEvent);
 
 process.once('message', (request: RouteInvocationChildRequest) => {
   const disposeTraceObserver = installEventTraceObserver(forwardEventTrace);

packages/agent-bundle/src/dev/routes/route-invocation-production.ts

@@ -10,6 +10,9 @@ import {
   createAgentRenderDispatcher,
   documentToCallToolResult,
   type AgentDocument,
+  type AgentProgressReporter,
+  type AgentProgressUpdate,
+  type AgentRenderDispatch,
   type AgentRenderEvent,
   type AgentRenderInvocation,
 } from '@agent-bundle/runtime';
@@ -287,6 +290,7 @@ const streamFromWorker = (
     readonly abort: () => void;
     readonly controller: ReadableStreamDefaultController<Uint8Array>;
     readonly dispatchSignal: AbortSignal;
+    readonly progress: AgentProgressReporter | undefined;
   }>();
   const failAll = (error: Error): void => {
     for (const [id, entry] of pending) {
@@ -302,7 +306,18 @@ const streamFromWorker = (
   worker.on('message', (message: WorkerMessage) => {
     const entry = pending.get(message.id);
     if (entry === undefined) return;
-    if (message.type === 'progress') return;
+    if (message.type === 'progress') {
+      // The route's reported progress becomes `progress` render events through
+      // the dispatcher's reporter, as the generated CLI session forwards it.
+      Promise.resolve()
+        .then(() => entry.progress?.report(message.update as AgentProgressUpdate))
+        .catch((error: unknown) => {
+          pending.delete(message.id);
+          entry.dispatchSignal.removeEventListener('abort', entry.abort);
+          entry.controller.error(error);
+        });
+      return;
+    }
     if (message.type === 'observed-providers-start') {
       trace?.providersStart();
       return;
@@ -372,10 +387,7 @@ const streamFromWorker = (
     entry.controller.error(new Error(message.message ?? 'Compiled route worker failed.'));
   });
   const host = Object.freeze({
-    execute: async (dispatch: Readonly<{
-      readonly invocation: AgentRenderInvocation;
-      readonly signal: AbortSignal;
-    }>): Promise<ReadableStream<Uint8Array>> => {
+    execute: async (dispatch: AgentRenderDispatch): Promise<ReadableStream<Uint8Array>> => {
       const id = ++sequence;
       let controller!: ReadableStreamDefaultController<Uint8Array>;
       const cancelRender = (): void => {
@@ -396,7 +408,7 @@ const streamFromWorker = (
         },
         start: (opened) => { controller = opened; },
       });
-      pending.set(id, { abort, controller, dispatchSignal: dispatch.signal });
+      pending.set(id, { abort, controller, dispatchSignal: dispatch.signal, progress: dispatch.progress });
       dispatch.signal.addEventListener('abort', abort, { once: true });
       worker.postMessage({
         actor: request.context.actor,
@@ -448,17 +460,21 @@ const missingRouteWorkerError = (error: unknown): boolean =>
     || error.message.includes('Generated rendered route must default-export')
   );
 
+/**
+ * Drives one compiled worker's render stream. Each event is handed to
+ * `publishRender` as it arrives and then dropped; only the `complete` event's
+ * document is kept, so the producer holds one document, not the stream.
+ */
 const renderCompiled = async (
   request: ProductionRequest,
   input: JsonValue,
   signal: AbortSignal,
   env: NodeJS.ProcessEnv,
   trace?: EventTracer,
-  publishRender?: (event: AgentRenderEvent) => void,
+  publishRender?: (event: AgentRenderEvent) => Promise<void> | void,
 ): Promise<Readonly<{
   readonly document: AgentDocument;
   readonly durationMs: number;
-  readonly events: readonly AgentRenderEvent[];
   readonly observed: {
     readonly providers: readonly RouteInvocationProvider[];
     readonly timings: readonly RouteInvocationTiming[];
@@ -469,21 +485,19 @@ 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[] = [];
+    let document: AgentDocument | undefined;
     try {
       const reader = session.events.getReader();
       for (;;) {
         const next = await reader.read();
         if (next.done) break;
-        events.push(next.value);
-        publishRender?.(next.value);
+        if (next.value.type === 'complete') document = next.value.document;
+        await 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.');
+      if (document === undefined) throw new Error('Compiled route render ended without a complete event.');
       return Object.freeze({
-        document: complete.document,
+        document,
         durationMs: performance.now() - startedAt,
-        events: Object.freeze(events),
         observed: {
           providers: Object.freeze([...session.observed.providers]),
           timings: Object.freeze([...session.observed.timings]),
@@ -504,7 +518,7 @@ const renderCompiled = async (
 export const renderProductionRoute = async (
   request: RouteInvocationChildRequest,
   publishTrace?: EventTraceObserver,
-  publishRender?: (event: AgentRenderEvent) => void,
+  publishRender?: (event: AgentRenderEvent) => Promise<void> | void,
 ): Promise<RouteInvocationChildResult> => {
   if (request.artifactEpoch === undefined || request.artifactRoot === undefined) {
     throw new ProductionRouteInvocationError(
@@ -539,7 +553,6 @@ export const renderProductionRoute = async (
     const value = prepared.preflight.gate as JsonValue;
     return Object.freeze({
       document: completeDocument(value),
-      events: Object.freeze([]),
       input: prepared.input,
       result: value,
       trace: Object.freeze(traceEvents),
@@ -569,7 +582,6 @@ export const renderProductionRoute = async (
         : undefined;
     return Object.freeze({
       document: rendered.document,
-      events: rendered.events,
       ...(exitCode === undefined ? {} : { exitCode }),
       input: prepared.input,
       ...(kind === 'tool'

packages/agent-bundle/src/dev/routes/route-invocation-render-history.ts

@@ -0,0 +1,143 @@
+/**
+ * The one retention policy for a route invocation's render-event history.
+ *
+ * Every place a render stream is retained — the service's live replay buffer,
+ * the completed envelope (`events`), `GET /api/routes/invocations/<id>`, the
+ * terminal `final` stream message, and the Workbench's live window — applies
+ * this window, so no reader can rehydrate what another evicted. The window is
+ * bounded by event count and by serialized bytes; the newest event and the
+ * newest document-bearing event (`shell`, `replace`, or `complete`) are never
+ * evicted, so a reader always folds to a coherent latest document. Everything
+ * older is disposable intermediate history. The final Agent Document itself
+ * is retained separately on the envelope (`document`) and is never truncated
+ * here: the runtime bounds it (`maxDocumentBytes`) before it reaches us.
+ *
+ * Browser-safe: no Node imports. The Workbench reaches it through
+ * `contracts/invocations.ts`.
+ */
+import type { AgentDocument, AgentRenderEvent } from '@agent-bundle/runtime';
+
+import { deepFreeze } from '../../core/freeze.ts';
+
+export interface RouteInvocationRenderHistoryLimits {
+  /** Serialized UTF-8 bytes of the retained events, summed. */
+  readonly maxBytes: number;
+  readonly maxEvents: number;
+}
+
+/**
+ * 256 events and 2 MiB. The window exceeds `maxBytes` only by what its two
+ * pinned events need; the runtime caps one event at `maxEventBytes`
+ * (1 MiB + 1 KiB), so a runtime-produced window never holds more than
+ * 2 MiB + 2 KiB.
+ */
+export const routeInvocationRenderHistoryLimits: RouteInvocationRenderHistoryLimits = Object.freeze({
+  maxBytes: 2 * 1024 * 1024,
+  maxEvents: 256,
+});
+
+/** What the policy evicted from a completed run; present on the envelope only when something was. */
+export interface RouteInvocationRenderRetention {
+  /** Serialized bytes of the evicted events. */
+  readonly evictedBytes: number;
+  /** Render events evicted, oldest first; `events` holds the remaining `producedEvents - evictedEvents`. */
+  readonly evictedEvents: number;
+  /** Every render event the run published, retained or not. */
+  readonly producedEvents: number;
+  /** Serialized bytes of the retained `events`. */
+  readonly retainedBytes: number;
+}
+
+interface RetainedRenderEvent {
+  readonly bytes: number;
+  readonly event: AgentRenderEvent;
+}
+
+/** An immutable retained window; `retainRenderEvent` derives the next one. */
+export interface RetainedRenderEvents {
+  readonly entries: readonly RetainedRenderEvent[];
+  readonly evictedBytes: number;
+  readonly evictedEvents: number;
+  /** The newest `shell`, `replace`, or `complete` entry, pinned against eviction. */
+  readonly latestDocument?: RetainedRenderEvent;
+  readonly producedEvents: number;
+  readonly retainedBytes: number;
+}
+
+export const emptyRetainedRenderEvents: RetainedRenderEvents = Object.freeze({
+  entries: Object.freeze([]),
+  evictedBytes: 0,
+  evictedEvents: 0,
+  producedEvents: 0,
+  retainedBytes: 0,
+});
+
+const encoder = new TextEncoder();
+
+export const renderEventBytes = (event: AgentRenderEvent): number =>
+  encoder.encode(JSON.stringify(event)).byteLength;
+
+const bearsDocument = (event: AgentRenderEvent): boolean =>
+  event.type === 'shell' || event.type === 'replace' || event.type === 'complete';
+
+/**
+ * Appends `event` and evicts the oldest disposable entries until the window
+ * fits both bounds again. The newest entry and the pinned document entry are
+ * never evicted, so a window can exceed `maxBytes` only when those two alone
+ * do. Returns the next window and the evicted events, oldest first.
+ */
+export const retainRenderEvent = (
+  retained: RetainedRenderEvents,
+  event: AgentRenderEvent,
+  limits: RouteInvocationRenderHistoryLimits = routeInvocationRenderHistoryLimits,
+): Readonly<{ readonly evicted: readonly AgentRenderEvent[]; readonly retained: RetainedRenderEvents }> => {
+  const entry: RetainedRenderEvent = Object.freeze({ bytes: renderEventBytes(event), event });
+  const latestDocument = bearsDocument(event) ? entry : retained.latestDocument;
+  const entries = [...retained.entries, entry];
+  const evicted: AgentRenderEvent[] = [];
+  let retainedBytes = retained.retainedBytes + entry.bytes;
+  let evictedBytes = retained.evictedBytes;
+  let index = 0;
+  while ((entries.length > limits.maxEvents || retainedBytes > limits.maxBytes) && index < entries.length - 1) {
+    const candidate = entries[index]!;
+    if (candidate === latestDocument) {
+      index += 1;
+      continue;
+    }
+    entries.splice(index, 1);
+    retainedBytes -= candidate.bytes;
+    evictedBytes += candidate.bytes;
+    evicted.push(candidate.event);
+  }
+  return Object.freeze({
+    evicted: Object.freeze(evicted),
+    retained: Object.freeze({
+      entries: Object.freeze(entries),
+      evictedBytes,
+      evictedEvents: retained.evictedEvents + evicted.length,
+      ...(latestDocument === undefined ? {} : { latestDocument }),
+      producedEvents: retained.producedEvents + 1,
+      retainedBytes,
+    }),
+  });
+};
+
+export const retainedRenderEvents = (retained: RetainedRenderEvents): readonly AgentRenderEvent[] =>
+  Object.freeze(retained.entries.map((entry) => entry.event));
+
+/** The document a reader folds to after eviction: the pinned newest `shell`, `replace`, or `complete`. */
+export const retainedLatestDocument = (retained: RetainedRenderEvents): AgentDocument | undefined => {
+  const event = retained.latestDocument?.event;
+  return event === undefined || event.type === 'progress' || event.type === 'error' ? undefined : event.document;
+};
+
+/** The envelope's truncation indication; `undefined` while nothing has been evicted. */
+export const renderRetention = (retained: RetainedRenderEvents): RouteInvocationRenderRetention | undefined =>
+  retained.evictedEvents === 0
+    ? undefined
+    : deepFreeze({
+        evictedBytes: retained.evictedBytes,
+        evictedEvents: retained.evictedEvents,
+        producedEvents: retained.producedEvents,
+        retainedBytes: retained.retainedBytes,
+      });

packages/agent-bundle/src/dev/routes/route-invocation-result.ts

@@ -3,6 +3,7 @@ import type { AgentDocument, AgentRenderEvent } from '@agent-bundle/runtime';
 import type { JsonValue } from '../../core/strict-json.ts';
 import type { RequestContextProvenance } from '../../contracts/request-provenance.ts';
 import type { EventTraceEvent } from '../../events/trace.ts';
+import type { RouteInvocationRenderRetention } from './route-invocation-render-history.ts';
 import type {
   RunningRouteInvocation,
   RouteInvocationProjection,
@@ -12,14 +13,25 @@ import type {
 
 export interface RouteInvocation extends RouteInvocationSummary {
   readonly context: RequestContextProvenance;
-  /** The final Agent Document; absent when rendering failed before a document existed. */
+  /**
+   * The final Agent Document of a `succeeded` run. A `cancelled` or `failed`
+   * run carries the latest document its retained stream reached, absent when
+   * none did. Never truncated by `retention`.
+   */
   readonly document?: AgentDocument;
-  /** The production `shell | progress | replace | error | complete` stream, in order. */
+  /**
+   * The production `shell | progress | replace | error | complete` stream, in
+   * order, as retained by the render-history window
+   * (`routeInvocationRenderHistoryLimits`): the newest events plus the newest
+   * document-bearing event. Complete unless `retention` is present.
+   */
   readonly events: readonly AgentRenderEvent[];
   readonly projection: RouteInvocationProjection;
   readonly providers: readonly RouteInvocationProvider[];
   /** Structured value recorded by the selected surface; its presence alone proves neither a `resultSchema` declaration nor validation. */
   readonly result?: JsonValue;
+  /** Present when the render-history window evicted events; the one truthful account of what `events` no longer holds. */
+  readonly retention?: RouteInvocationRenderRetention;
   /** Event-kernel phase events emitted by a compiled preflight execution. */
   readonly trace?: readonly EventTraceEvent[];
 }