fix(dev): bound route-invocation render history by count and bytes end to end

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

@@ -0,0 +1,5 @@
+---
+'agent-bundle': patch
+---
+
+Bound route-invocation render history end to end with one policy (`RENDER_EVENT_RETENTION`, exported from `contracts/invocations`): the render child, its IPC reply, the `GET /api/routes/invocations/<id>/stream` replay, the completed `RouteInvocation` envelope returned by `POST /api/routes/invocations` and `GET /api/routes/invocations/<id>`, 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)

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,

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<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);
+  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) => {

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

@@ -30,6 +30,7 @@ import {
   ROUTE_INVOCATION_COMPILED_ROUTE_UNAVAILABLE_CODE,
   ROUTE_INVOCATION_PREPARATION_FAILURE_CODE,
 } from './route-invocation-production-error.ts';
+import { emptyRetainedRenderEvents, retainRenderEvent } from './route-invocation-result.ts';
 import type { RouteInvocationProvider, RouteInvocationTiming } from './route-invocation.ts';
 
 interface CompiledCliInvocationModule {
@@ -447,6 +448,7 @@ const renderCompiled = async (
   readonly document: AgentDocument;
   readonly durationMs: number;
   readonly events: readonly AgentRenderEvent[];
+  readonly evictedEvents: number;
   readonly observed: {
     readonly providers: readonly RouteInvocationProvider[];
     readonly timings: readonly RouteInvocationTiming[];
@@ -457,21 +459,22 @@ const renderCompiled = async (
   for (const workerPath of candidates) {
     const startedAt = performance.now();
     const session = streamFromWorker(workerPath, request, invocation, input, signal, env, trace);
-    const events: AgentRenderEvent[] = [];
+    const retained = emptyRetainedRenderEvents();
     try {
       const reader = session.events.getReader();
       for (;;) {
         const next = await reader.read();
         if (next.done) break;
-        events.push(next.value);
+        retainRenderEvent(retained, next.value);
         publishRender?.(next.value);
       }
-      const complete = events.findLast((event) => event.type === 'complete');
-      if (complete === undefined) throw new Error('Compiled route render ended without a complete event.');
+      const complete = retained.events.at(-1);
+      if (complete?.type !== 'complete') throw new Error('Compiled route render ended without a complete event.');
       return Object.freeze({
         document: complete.document,
         durationMs: performance.now() - startedAt,
-        events: Object.freeze(events),
+        events: Object.freeze(retained.events),
+        evictedEvents: retained.evicted,
         observed: {
           providers: Object.freeze([...session.observed.providers]),
           timings: Object.freeze([...session.observed.timings]),
@@ -558,6 +561,7 @@ export const renderProductionRoute = async (
     return Object.freeze({
       document: rendered.document,
       events: rendered.events,
+      evictedEvents: rendered.evictedEvents,
       ...(exitCode === undefined ? {} : { exitCode }),
       input: prepared.input,
       ...(kind === 'tool'

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

@@ -10,12 +10,62 @@ import type {
   RouteInvocationSummary,
 } from './route-invocation.ts';
 
+/**
+ * The one retention policy for an invocation's render history, applied by
+ * every keeper of it: the render child while it streams, the child's IPC
+ * reply, the live stream replay, the completed envelope (and so history reads
+ * and the final stream message), and the browser's live buffer. Past either
+ * bound the oldest event is evicted first; the newest event always survives,
+ * so a completed stream keeps its `complete` event whatever its size. The
+ * final `document`, `outcome`, and correlation identifiers live outside this
+ * window and are never evicted.
+ */
+export const RENDER_EVENT_RETENTION = Object.freeze({
+  /** JSON text of the retained events, in UTF-16 code units. */
+  maxBytes: 1024 * 1024,
+  maxEvents: 256,
+});
+
+export interface RetainedRenderEvents {
+  /** JSON text size of `events`. */
+  bytes: number;
+  /** Older events the bounds evicted — the truncation indication. */
+  evicted: number;
+  readonly events: AgentRenderEvent[];
+}
+
+export const renderEventBytes = (event: AgentRenderEvent): number => JSON.stringify(event).length;
+
+export const emptyRetainedRenderEvents = (): RetainedRenderEvents => ({ bytes: 0, events: [], evicted: 0 });
+
+/** Appends `event`, then evicts oldest-first while `RENDER_EVENT_RETENTION` is exceeded. */
+export const retainRenderEvent = (retained: RetainedRenderEvents, event: AgentRenderEvent): void => {
+  retained.events.push(event);
+  retained.bytes += renderEventBytes(event);
+  while (
+    retained.events.length > 1
+    && (retained.events.length > RENDER_EVENT_RETENTION.maxEvents || retained.bytes > RENDER_EVENT_RETENTION.maxBytes)
+  ) {
+    retained.bytes -= renderEventBytes(retained.events.shift()!);
+    retained.evicted += 1;
+  }
+};
+
+/** The retained window of an already-collected stream. */
+export const retainRenderEvents = (events: readonly AgentRenderEvent[]): RetainedRenderEvents => {
+  const retained = emptyRetainedRenderEvents();
+  for (const event of events) retainRenderEvent(retained, event);
+  return retained;
+};
+
 export interface RouteInvocation extends RouteInvocationSummary {
   readonly context: RequestContextProvenance;
   /** The final Agent Document; absent when rendering failed before a document existed. */
   readonly document?: AgentDocument;
-  /** The production `shell | progress | replace | error | complete` stream, in order. */
+  /** The retained window of the production `shell | progress | replace | error | complete` stream, in order (`RENDER_EVENT_RETENTION`). */
   readonly events: readonly AgentRenderEvent[];
+  /** Older render events `RENDER_EVENT_RETENTION` evicted from `events`; absent when none were. */
+  readonly evictedEvents?: number;
   readonly projection: RouteInvocationProjection;
   readonly providers: readonly RouteInvocationProvider[];
   /** Structured value recorded by the selected surface; its presence alone proves neither a `resultSchema` declaration nor validation. */

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

@@ -1,3 +1,4 @@
+import { Buffer } from 'node:buffer';
 import type { IncomingMessage, ServerResponse } from 'node:http';
 
 import type { ProjectEventHub } from '../events.ts';
@@ -37,6 +38,19 @@ import {
 
 const streamQueueByteLimit = 256 * 1024;
 const streamQueueEntryLimit = 128;
+
+interface PendingFrame {
+  readonly bytes: number;
+  readonly final: boolean;
+  /** Arrived after the replay snapshot was taken. */
+  readonly live: boolean;
+  readonly text: string;
+}
+
+const pendingFrame = (message: RouteInvocationStreamMessage, live: boolean): PendingFrame => {
+  const text = `event: ${message.type}\ndata: ${JSON.stringify(message)}\n\n`;
+  return { bytes: Buffer.byteLength(text, 'utf8'), final: message.type === 'final', live, text };
+};
 const invalidShape = badRequest(
   ROUTE_INVOCATION_MALFORMED_REQUEST_CODE,
   'Route invocation request has an invalid shape.',
@@ -258,6 +272,12 @@ export class RouteInvocationRoutes {
     return true;
   }
 
+  /**
+   * Replays the retained window one frame per socket drain, then delivers
+   * live messages through the bounded queue. The replay is bounded by
+   * retention, not by the queue: bursting it into the queue would destroy
+   * every reconnect whose window outgrew the queue's live-consumer limits.
+   */
   #stream(service: RouteInvocationRouteService, id: string, response: ServerResponse): void {
     let terminal = false;
     const stream = { unsubscribe: undefined as (() => void) | undefined };
@@ -266,30 +286,56 @@ export class RouteInvocationRoutes {
       stream.unsubscribe?.();
       response.end();
     };
+    const deliver = (frame: PendingFrame): void => {
+      if (writer.enqueue(frame.text) === 'overflow') response.destroy();
+      if (frame.final) terminal = true;
+      finish();
+    };
+    // The replay drains one frame per socket drain. Live frames that arrive
+    // meanwhile wait behind it, held to the live queue's own limits.
+    const pending: PendingFrame[] = [];
+    let replaying = true;
+    let subscribed = false;
+    let liveBytes = 0;
+    let liveRecords = 0;
+    const pump = (): void => {
+      while (replaying && writer.idle && !response.destroyed) {
+        const next = pending.shift();
+        if (next === undefined) {
+          replaying = false;
+          break;
+        }
+        if (next.live) {
+          liveBytes -= next.bytes;
+          liveRecords -= 1;
+        }
+        deliver(next);
+      }
+      finish();
+    };
     const writer = createBackpressuredWriter(response, {
       byteLimit: streamQueueByteLimit,
-      onIdle: finish,
+      onIdle: pump,
       recordLimit: streamQueueEntryLimit,
     });
-    const deliver = (message: RouteInvocationStreamMessage): void => {
-      const result = writer.enqueue(`event: ${message.type}\ndata: ${JSON.stringify(message)}\n\n`);
-      if (result === 'overflow') response.destroy();
-      if (message.type === 'final') terminal = true;
-      finish();
-    };
     response.once('close', () => {
       writer.markClosed();
       stream.unsubscribe?.();
     });
-    const replay: RouteInvocationStreamMessage[] = [];
-    let replaying = true;
-    stream.unsubscribe = service.subscribe(id, (message) => replaying ? replay.push(message) : deliver(message));
+    stream.unsubscribe = service.subscribe(id, (message) => {
+      const frame = pendingFrame(message, subscribed);
+      if (!replaying) return deliver(frame);
+      pending.push(frame);
+      if (!frame.live) return;
+      liveBytes += frame.bytes;
+      liveRecords += 1;
+      if (liveBytes > streamQueueByteLimit || liveRecords > streamQueueEntryLimit) response.destroy();
+    });
+    subscribed = true;
     writeKeepAliveStreamHead(response, {
       cacheControl: 'no-cache',
       contentType: 'text/event-stream; charset=utf-8',
     });
-    replaying = false;
-    for (const message of replay) deliver(message);
-    finish();
+    pump();
   }
 }