perf: amortize replay buffer compaction in ToolEventBroadcaster and ReusableReadableStream

.changeset/quiet-replay-streams.md

@@ -0,0 +1,16 @@
+---
+'@openrouter/agent': minor
+---
+
+Add opt-in replay compaction and terminal response-event handling for streamed model calls.
+
+```ts
+import { callModel } from '@openrouter/agent';
+
+const result = callModel(client, {
+  model: 'openai/gpt-4o',
+  input: 'Summarize this document.',
+  // Retain only the history needed by currently attached consumers.
+  streamReplay: 'active-consumers',
+});
+```

packages/agent/README.md

@@ -127,6 +127,24 @@ What each stream emits:
 | `getItemsStream()` | all output items (messages, function calls, …) — output items **only**, no usage/response metadata |
 | `getFullResponsesStream()` | every response event, including `tool.result` / `tool.call_output` execution events, and each round's `response.completed` (with that round's usage block) |
 
+#### Replay history for stream consumers
+
+Every stream getter above can start from event zero, so by default a result
+retains its full event history for the lifetime of the call. Long generator-tool
+streams make that history expensive in a constrained runtime. When all consumers
+attach before draining, `streamReplay: 'active-consumers'` releases buffered
+events once every attached consumer has advanced past them:
+
+```typescript
+const result = callModel(client, {
+  model,
+  input,
+  // Default 'full' retains complete replay history for delayed and
+  // sequential consumers; 'active-consumers' trades that for bounded memory.
+  streamReplay: 'active-consumers',
+});
+```
+
 #### Usage across a multi-round tool loop
 
 `getResponse()` resolves to the **final** round's response, so in a

packages/agent/src/index.ts

@@ -205,6 +205,7 @@ export {
   buildNextTurnParamsContext,
   executeNextTurnParamsFunctions,
 } from './lib/next-turn-params.js';
+export type { StreamReplay } from './lib/reusable-stream.js';
 // Stop condition helpers
 export {
   finishReasonIs,

packages/agent/src/inner-loop/call-model.ts

@@ -108,6 +108,7 @@ export function callModel<
     sharedContextSchema,
     onTurnStart,
     onTurnEnd,
+    streamReplay,
     allowFinalResponse,
     strictFinalResponse,
     hooks,
@@ -189,6 +190,7 @@ export function callModel<
     sharedContextSchema,
     onTurnStart,
     onTurnEnd,
+    streamReplay,
     allowFinalResponse,
     strictFinalResponse,
     hooks: hooks !== undefined ? resolveHooks(hooks) : undefined,

packages/agent/src/lib/async-params.ts

@@ -4,6 +4,7 @@ import type { DoomLoopOption } from './doom-loop.js';
 import type { HooksManager } from './hooks-manager.js';
 import type { InlineHookConfig } from './hooks-types.js';
 import type { Item } from './item-types.js';
+import type { StreamReplay } from './reusable-stream.js';
 import type { ContextInput } from './tool-context.js';
 import type {
   ParsedToolCall,
@@ -114,6 +115,13 @@ type BaseCallModelInput<
    * Receives the turn context and the completed response for that turn
    */
   onTurnEnd?: (context: TurnContext, response: OpenResponsesResult) => void | Promise<void>;
+  /**
+   * Controls replay history retained for stream getters.
+   * `full` preserves all events for delayed and sequential consumers.
+   * `active-consumers` compacts events after every attached consumer advances.
+   * @default 'full'
+   */
+  streamReplay?: StreamReplay;
   /**
    * When the loop exits because `stopWhen` was met and the last response
    * still contained tool calls, execute those pending tool calls (so they
@@ -332,6 +340,7 @@ export async function resolveAsyncFunctions<TTools extends readonly Tool[] = rea
     'sharedContextSchema', // Client-side schema for shared context validation
     'onTurnStart', // Client-side turn start callback
     'onTurnEnd', // Client-side turn end callback
+    'streamReplay', // Client-side stream replay policy
     'allowFinalResponse', // Client-side: tunes the default toolChoice:'none' final turn when stopWhen breaks the loop
     'strictFinalResponse', // Client-side: restore throw on empty final after tool rounds
     'hooks', // Client-side hook system

packages/agent/src/lib/model-result.ts

@@ -57,6 +57,7 @@ import {
   applyNextTurnParamsToRequest,
   executeNextTurnParamsFunctions,
 } from './next-turn-params.js';
+import type { StreamReplay } from './reusable-stream.js';
 import { ReusableReadableStream } from './reusable-stream.js';
 import { isStopConditionMet } from './stop-conditions.js';
 import type { ItemInProgress, StreamableOutputItem } from './stream-transformers.js';
@@ -303,6 +304,14 @@ function isEventStream(value: unknown): value is EventStream<models.StreamEvents
   return typeof maybeStream.getReader === 'function';
 }
 
+function isTerminalResponseStreamEvent(event: models.StreamEvents): boolean {
+  return (
+    isResponseCompletedEvent(event) ||
+    isResponseFailedEvent(event) ||
+    isResponseIncompleteEvent(event)
+  );
+}
+
 /**
  * Map the server's usage block onto the hook-facing ModelCallUsage shape.
  * Returns undefined when the response carried no usage accounting.
@@ -460,6 +469,8 @@ export interface GetResponseOptions<
   onTurnStart?: (context: TurnContext) => void | Promise<void>;
   /** Callback invoked at the end of each tool execution turn */
   onTurnEnd?: (context: TurnContext, response: models.OpenResponsesResult) => void | Promise<void>;
+  /** Replay history retained for delayed and sequential stream consumers. */
+  streamReplay?: StreamReplay;
   /**
    * When the loop exits because `stopWhen` was met and the last response
    * still contained tool calls, make one more model request with no tools so
@@ -631,6 +642,9 @@ export class ModelResult<
     null;
   private initialStreamPipeStarted = false;
   private initialPipePromise: Promise<void> | null = null;
+  private initialResponse: models.OpenResponsesResult | null = null;
+  private initialResponseError: Error | null = null;
+  private readonly streamReplay: StreamReplay;
 
   // Context store for typed tool context (persists across turns)
   private contextStore: ToolContextStore | null = null;
@@ -746,6 +760,7 @@ export class ModelResult<
 
   constructor(options: GetResponseOptions<TTools, TShared>) {
     this.options = options;
+    this.streamReplay = options.streamReplay ?? 'full';
     this.hooksManager = options.hooks;
     const doomLoopConfig = resolveDoomLoopOption(options.doomLoop);
     this.doomLoopMonitor = doomLoopConfig ? new DoomLoopMonitor(doomLoopConfig) : null;
@@ -860,7 +875,7 @@ export class ModelResult<
    */
   private ensureTurnBroadcaster(): ToolEventBroadcaster<CorrelatedResponseStreamEvent<TTools>> {
     if (!this.turnBroadcaster) {
-      this.turnBroadcaster = new ToolEventBroadcaster();
+      this.turnBroadcaster = new ToolEventBroadcaster(this.streamReplay);
     }
     return this.turnBroadcaster;
   }
@@ -903,7 +918,32 @@ export class ModelResult<
         timestamp: Date.now(),
       } satisfies TurnEndEvent);
     })().catch((error) => {
-      broadcaster.complete(error instanceof Error ? error : new Error(String(error)));
+      const normalizedError = error instanceof Error ? error : new Error(String(error));
+      this.initialResponseError = normalizedError;
+      broadcaster.complete(normalizedError);
+    });
+  }
+
+  private captureInitialStreamEvent(event: models.StreamEvents): void {
+    if (isResponseCompletedEvent(event) || isResponseIncompleteEvent(event)) {
+      this.initialResponse = event.response;
+      return;
+    }
+
+    if (isResponseFailedEvent(event)) {
+      this.initialResponseError = new Error(
+        `Response failed: ${JSON.stringify(event.response.error)}`,
+      );
+    }
+  }
+
+  private setReusableStream(stream: ReadableStream<models.StreamEvents>): void {
+    this.initialResponse = null;
+    this.initialResponseError = null;
+    this.reusableStream = new ReusableReadableStream(stream, {
+      streamReplay: this.streamReplay,
+      onValue: (event) => this.captureInitialStreamEvent(event),
+      isTerminalValue: isTerminalResponseStreamEvent,
     });
   }
 
@@ -1068,7 +1108,10 @@ export class ModelResult<
     turnNumber: number,
   ): Promise<models.OpenResponsesResult> {
     if (isEventStream(value)) {
-      const stream = new ReusableReadableStream(value);
+      const stream = new ReusableReadableStream(value, {
+        streamReplay: this.streamReplay,
+        isTerminalValue: isTerminalResponseStreamEvent,
+      });
       if (this.turnBroadcaster) {
         return this.pipeAndConsumeStream(stream, turnNumber);
       }
@@ -1095,6 +1138,21 @@ export class ModelResult<
     if (this.finalResponse) {
       return this.finalResponse;
     }
+
+    const initialPipePromise = this.initialPipePromise;
+    if (initialPipePromise) {
+      await initialPipePromise;
+    }
+
+    if (this.initialResponseError) {
+      throw this.initialResponseError;
+    }
+
+    if (this.initialResponse) {
+      await this.emitPendingModelCallOnce(this.initialResponse);
+      return this.initialResponse;
+    }
+
     if (this.reusableStream) {
       const response = await consumeStreamForCompletion(this.reusableStream);
       await this.emitPendingModelCallOnce(response);
@@ -1103,6 +1161,26 @@ export class ModelResult<
     throw new Error('Neither stream nor response initialized');
   }
 
+  private extractCachedCompletion(): models.OpenResponsesResult {
+    if (this.initialResponseError) {
+      throw this.initialResponseError;
+    }
+    if (this.initialResponse) {
+      return this.initialResponse;
+    }
+    if (!this.reusableStream) {
+      throw new Error('Stream not initialized');
+    }
+    return extractCompletionFromBuffer(this.reusableStream);
+  }
+
+  private tryExtractCachedCompletion(): models.OpenResponsesResult | undefined {
+    if (this.initialResponse) {
+      return this.initialResponse;
+    }
+    return this.reusableStream ? tryExtractCompletionFromBuffer(this.reusableStream) : undefined;
+  }
+
   /**
    * Save response output to state.
    * Appends the response output to the message history and records the response ID.
@@ -2481,7 +2559,7 @@ export class ModelResult<
           // Sync backward scan of the retained buffer — not a consumer
           // replay, which would cost one microtask hop per buffered event
           // on every hook-less streaming teardown.
-          await this.emitPendingModelCallOnce(extractCompletionFromBuffer(this.reusableStream));
+          await this.emitPendingModelCallOnce(this.extractCachedCompletion());
         } else if (this.reusableStream) {
           // Consumers stop at the terminal event (streamTerminationEvents),
           // usually before the pump reads the source close that flips
@@ -2491,7 +2569,7 @@ export class ModelResult<
           // dropping the parked telemetry. Stays silent (no emit, no
           // throw) when nothing terminal was buffered — e.g. an errored
           // mid-flight stream, where no materialized response exists.
-          const buffered = tryExtractCompletionFromBuffer(this.reusableStream);
+          const buffered = this.tryExtractCachedCompletion();
           if (buffered) {
             await this.emitPendingModelCallOnce(buffered);
           }
@@ -4929,6 +5007,7 @@ export class ModelResult<
       strictFinalResponse: _sfr,
       hooks: _h,
       doomLoop: _dl,
+      streamReplay: _sr,
       signal: _sig,
       toolTimeoutMs: _ttm,
       toolConcurrency: _tc,
@@ -5540,7 +5619,7 @@ export class ModelResult<
       // Handle both streaming and non-streaming responses
       // The API may return a non-streaming response even when stream: true is requested
       if (isEventStream(apiResult.value)) {
-        this.reusableStream = new ReusableReadableStream(apiResult.value);
+        this.setReusableStream(apiResult.value);
       } else if (this.isNonStreamingResponse(apiResult.value)) {
         // API returned a complete response directly - use it as the final response
         this.finalResponse = apiResult.value;
@@ -5827,7 +5906,7 @@ export class ModelResult<
 
     // Handle both streaming and non-streaming responses
     if (isEventStream(apiResult.value)) {
-      this.reusableStream = new ReusableReadableStream(apiResult.value);
+      this.setReusableStream(apiResult.value);
     } else if (this.isNonStreamingResponse(apiResult.value)) {
       this.finalResponse = apiResult.value;
       await this.emitPendingModelCallOnce(this.finalResponse);
@@ -6837,7 +6916,7 @@ export class ModelResult<
       throw new Error('Stream not initialized');
     }
 
-    const completedResponse = await consumeStreamForCompletion(this.reusableStream);
+    const completedResponse = await this.getInitialResponse();
     await this.emitPendingModelCallOnce(completedResponse);
     return extractToolCallsFromResponse(completedResponse) as ParsedToolCall<TTools[number]>[];
   }
@@ -7143,7 +7222,7 @@ export class ModelResult<
         // reusable stream is a passive observation: it buffers events
         // without executing tools or mutating conversation state, so the
         // resume generation is counted without advancing the loop.
-        await this.emitPendingModelCallOnce(await consumeStreamForCompletion(this.reusableStream));
+        await this.emitPendingModelCallOnce(await this.getInitialResponse());
       }
     } catch (error) {
       // Intentionally swallowed — see the "never rejects" note above. The