fix(advisor): bound retained stream and replay state

scripts/test-layout/layout.json

@@ -3,6 +3,7 @@
   "root": "tests",
   "explicitShards": ["layout-clients.json"],
   "explicit": { "service-log.test.ts": "service", "messages-native-empty-inline.test.ts": "claude-integration",
+    "advisor-retention.test.ts": "advisor",
     "owner-registry-isolation.test.ts": "config",
     "codex-history-pending-wal.test.ts": "codex-integration", "responses-anthropic-billing-send-budget.test.ts": "responses", "anthropic-billing-ownership.test.ts": "adapters/anthropic", "server-anthropic-billing.test.ts": "server", "server-anthropic-billing-boundaries.test.ts": "server", "router-anthropic-billing-failover.test.ts": "routing", "anthropic-billing-refusal.test.ts": "adapters/anthropic", "messages-request-id-headers.test.ts": "claude-integration", "bounded-file-read.test.ts": "lib", "responses-undeclared-tool-guard-inbound.test.ts": "responses", "explicit-empty-tool-catalog.test.ts": "responses", "explicit-empty-tool-catalog-history.test.ts": "responses",
     "messages-request-id-endpoint.test.ts": "claude-integration",

src/server/responses/adapter-continuation.ts

@@ -739,6 +739,7 @@ export function createAdapterContinuations(
     return parsed._advisorGuard
       ? parsed._advisorGuard({
           parsed,
+          translatorBudget,
           firstEvents: terminalGuardedRetry,
           continuation: fetchTerminalGuardContinuation,
         })

src/server/responses/adapter-delivery.ts

@@ -127,6 +127,7 @@ export async function deliverAdapterResponse(
     const advisorStream = parsed._advisorGuard
       ? parsed._advisorGuard({
           parsed,
+          translatorBudget,
           firstEvents: eventStream,
           continuation: fetchTerminalGuardContinuation,
         })
@@ -266,6 +267,7 @@ export async function deliverAdapterResponse(
         const advisorEvents: AdapterEvent[] = [];
         for await (const event of parsed._advisorGuard({
           parsed,
+          translatorBudget,
           firstEvents: (async function* () { yield* guardedEvents; })(),
           continuation: fetchTerminalGuardContinuation,
         })) advisorEvents.push(event);

src/server/responses/advisor-retention.ts

@@ -0,0 +1,117 @@
+import { jsonUtf8Bytes } from "../../lib/json-byte-size";
+import {
+  TRANSLATOR_MAX_TURN_BYTES,
+  TranslatorBudgetExceededError,
+  type TranslatorBudget,
+} from "../../lib/translator-budget";
+import type { AdapterEvent, OcxMessage, OcxUsage } from "../../types";
+import { mergeUsage } from "./terminal-guard";
+
+export const MAX_ADVISOR_RETAINED_EVENTS = 4_096;
+export const MAX_ADVISOR_RETAINED_CALLS = 1_024;
+
+/** Independent ownership: never borrow the bridge's event or tool-call leases. */
+export class AdvisorRetention {
+  private legBytes = 0;
+  private historyBytes = 0;
+  private copyBytes = 0;
+  private usageBytes = 0;
+  private events = 0;
+  private calls = 0;
+
+  constructor(private readonly shared?: TranslatorBudget) {}
+
+  private charge(bytes: number, kind: "retained_collectors" | "request_copies"): void {
+    if (bytes > TRANSLATOR_MAX_TURN_BYTES - this.legBytes - this.historyBytes - this.copyBytes - this.usageBytes) {
+      throw new TranslatorBudgetExceededError(kind, TRANSLATOR_MAX_TURN_BYTES);
+    }
+    this.shared?.chargeRetained(bytes, { kind });
+  }
+
+  /** Usage can alias a terminal payload. Keep its own owner after the raw leg ends. */
+  usage(first: OcxUsage | undefined, second: OcxUsage | undefined): OcxUsage | undefined {
+    if (!second) return first;
+    // Two-input merge materializes a new object. Reserve before merging, including
+    // a conservative bound for concatenation if a provider supplied malformed scalars.
+    const bytes = first
+      ? (jsonUtf8Bytes(first) + jsonUtf8Bytes(second)) * 2 + 256
+      : jsonUtf8Bytes(second) + 1;
+    this.charge(bytes, "request_copies");
+    this.copyBytes += bytes;
+    let retained = 0;
+    try {
+      const merged = mergeUsage(first, second);
+      const measured = jsonUtf8Bytes(merged, bytes) + 1;
+      if (measured > bytes) throw new TranslatorBudgetExceededError("request_copies", TRANSLATOR_MAX_TURN_BYTES);
+      retained = measured;
+      const previous = this.usageBytes;
+      this.usageBytes = retained;
+      this.shared?.releaseRetained(previous, { kind: "request_copies" });
+      return merged;
+    } finally {
+      this.copyBytes -= bytes;
+      this.shared?.releaseRetained(bytes - retained, { kind: "request_copies" });
+    }
+  }
+
+  /**
+   * `counted: false` is only for non-empty text/thinking deltas: the leg's 16 KiB character
+   * caps already bound how many of those can be held, and an ordinary long answer streams
+   * thousands of them. They still consume bytes.
+   */
+  retain(event: AdapterEvent, call = false, counted = true): void {
+    if ((counted && this.events >= MAX_ADVISOR_RETAINED_EVENTS) || (call && this.calls >= MAX_ADVISOR_RETAINED_CALLS)) {
+      throw new TranslatorBudgetExceededError("retained_collectors", TRANSLATOR_MAX_TURN_BYTES);
+    }
+    const bytes = jsonUtf8Bytes(event) + 1;
+    this.charge(bytes, "retained_collectors");
+    this.legBytes += bytes;
+    if (counted) this.events++;
+    if (call) this.calls++;
+  }
+
+  /** Hold a bounded materialization while its raw sources are still retained. */
+  copy<T>(bytes: number, build: () => T): T {
+    this.charge(bytes, "request_copies");
+    this.copyBytes += bytes;
+    try { return build(); }
+    finally {
+      this.copyBytes -= bytes;
+      this.shared?.releaseRetained(bytes, { kind: "request_copies" });
+    }
+  }
+
+  /** Reserve before constructing replay JSON; transfer the reservation to history. */
+  history<T extends OcxMessage | undefined>(bytes: number, build: () => T): T {
+    this.charge(bytes, "request_copies");
+    this.copyBytes += bytes;
+    let retained = 0;
+    try {
+      const message = build();
+      const measured = message ? jsonUtf8Bytes(message, bytes) + 1 : 0;
+      if (measured > bytes) throw new TranslatorBudgetExceededError("request_copies", TRANSLATOR_MAX_TURN_BYTES);
+      retained = measured;
+      this.historyBytes += retained;
+      return message;
+    } finally {
+      this.copyBytes -= bytes;
+      this.shared?.releaseRetained(bytes - retained, { kind: "request_copies" });
+    }
+  }
+
+  /** Event envelopes dominate the replay content; allow new message/call framing too. */
+  replayReservation(): number { return this.legBytes + (this.calls + 1) * 256; }
+
+  releaseLeg(): void {
+    this.shared?.releaseRetained(this.legBytes, { kind: "retained_collectors" });
+    this.legBytes = this.events = this.calls = 0;
+  }
+
+  release(): void {
+    this.releaseLeg();
+    this.shared?.releaseRetained(this.historyBytes, { kind: "request_copies" });
+    this.historyBytes = 0;
+    this.shared?.releaseRetained(this.usageBytes, { kind: "request_copies" });
+    this.usageBytes = 0;
+  }
+}

src/server/responses/advisor-slot.ts

@@ -28,7 +28,9 @@ import type {
   OcxToolResultMessage,
   OcxUsage,
 } from "../../types";
-import { mergeUsage } from "./terminal-guard";
+import { jsonStringPartsUtf8Bytes, jsonUtf8Bytes } from "../../lib/json-byte-size";
+import { isTranslatorBudgetExceededError, TRANSLATOR_MAX_CALL_ARGUMENT_BYTES, TranslatorBudgetExceededError, type TranslatorBudget } from "../../lib/translator-budget";
+import { AdvisorRetention } from "./advisor-retention";
 
 export const ADVISOR_TOOL_NAME = "advisor";
 
@@ -78,12 +80,19 @@ export interface AdvisorPlan {
 interface HeldAdvisorCall {
   id: string;
   name: string;
-  argsBuf: string;
+  args: string[];
+  argumentBytes: number;
+  question?: string;
   closed: boolean;
   providerMetadata?: import("../../types").OcxProviderOpaqueToolCallMetadata;
 }
 
-function parseAdvisorArgs(argsBuf: string): { question?: string } {
+function parseAdvisorArgs(call: HeldAdvisorCall, retention: AdvisorRetention): { question?: string } {
+  // Reserve both the joined input and parsed JSON before either is materialized.
+  return retention.copy(call.argumentBytes * 2 + 4, () => parseArgs(call.args.join("")));
+}
+
+function parseArgs(argsBuf: string): { question?: string } {
   if (argsBuf.trim() === "") return {};
   try {
     const parsed: unknown = JSON.parse(argsBuf);
@@ -99,6 +108,7 @@ function assistantMessageFromLeg(
   events: readonly AdapterEvent[],
   held: readonly HeldAdvisorCall[],
   timestamp: number,
+  retention: AdvisorRetention,
 ): OcxAssistantMessage | undefined {
   let text = "";
   let thinking = "";
@@ -119,11 +129,13 @@ function assistantMessageFromLeg(
   if (text) content.push({ type: "text", text });
   for (const call of held) {
     if (!call.closed) continue;
+    const args = parseAdvisorArgs(call, retention);
+    call.question = args.question;
     content.push({
       type: "toolCall",
       id: call.id,
       name: call.name,
-      arguments: parseAdvisorArgs(call.argsBuf),
+      arguments: args,
       ...(call.providerMetadata ? { providerMetadata: call.providerMetadata } : {}),
     });
   }
@@ -150,6 +162,8 @@ export interface AdvisorGuardOptions {
   parsed: OcxParsedRequest;
   plan: AdvisorPlan;
   firstEvents: AsyncIterable<AdapterEvent>;
+  /** Independent guard ownership within the optional shared request budget. */
+  translatorBudget?: TranslatorBudget;
   /** One bounded worker continuation re-dispatch (same machinery as the terminal guard). */
   continuation: (parsed: OcxParsedRequest) => AsyncIterable<AdapterEvent> | Promise<AsyncIterable<AdapterEvent>>;
 }
@@ -172,7 +186,8 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable<OcxParsedRequ
     let parsed = options.parsed;
     let accumulatedUsage: OcxUsage | undefined;
     let source: AsyncIterable<AdapterEvent> = options.firstEvents;
-
+    const retention = new AdvisorRetention(options.translatorBudget);
+    try {
     while (true) {
       const held: HeldAdvisorCall[] = [];
       const legEvents: AdapterEvent[] = [];
@@ -189,15 +204,33 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable<OcxParsedRequ
       for await (const event of source) {
         if (event.type === "tool_call_start") {
           if (pending) { held.push(pending); pending = null; }
-          pending = { id: event.id, name: event.name, argsBuf: "", closed: false, ...(event.providerMetadata ? { providerMetadata: event.providerMetadata } : {}) };
           holdingCurrent = event.name === ADVISOR_TOOL_NAME;
-          if (!holdingCurrent) hasRealToolCall = true;
-          else continue;
+          if (!holdingCurrent) {
+            hasRealToolCall = true;
+            // No continuation can use this leg. Real-tool arguments are never copied.
+            held.length = legEvents.length = 0;
+            retention.releaseLeg();
+          } else {
+            if (!hasRealToolCall) {
+              retention.retain(event, true);
+              pending = { id: event.id, name: event.name, args: [], argumentBytes: 0, closed: false, ...(event.providerMetadata ? { providerMetadata: event.providerMetadata } : {}) };
+            }
+            continue;
+          }
         } else if (event.type === "tool_call_delta") {
-          if (pending) pending.argsBuf += event.arguments;
+          if (pending) {
+            const bytes = jsonStringPartsUtf8Bytes([event.arguments], TRANSLATOR_MAX_CALL_ARGUMENT_BYTES + 2) - 2;
+            if (bytes > TRANSLATOR_MAX_CALL_ARGUMENT_BYTES - pending.argumentBytes) {
+              throw new TranslatorBudgetExceededError("tool_args", TRANSLATOR_MAX_CALL_ARGUMENT_BYTES);
+            }
+            retention.retain(event);
+            pending.argumentBytes += bytes;
+            pending.args.push(event.arguments);
+          }
           if (holdingCurrent) continue;
         } else if (event.type === "tool_call_end") {
           if (pending) {
+            retention.retain(event);
             pending.closed = true;
             if (pending.name === ADVISOR_TOOL_NAME) held.push(pending);
             pending = null;
@@ -207,29 +240,31 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable<OcxParsedRequ
             continue;
           }
         } else if (event.type === "done") {
+          if (!hasRealToolCall) retention.retain(event);
           if (pending) { held.push(pending); pending = null; }
           terminalEvent = event;
           terminalConsumed = true;
           break;
         } else if (event.type === "incomplete" || event.type === "error") {
           // A broken leg cannot safely anchor a continuation: surface the terminal as-is.
           if (pending) { pending = null; }
-          const usage = mergeUsage(accumulatedUsage, event.usage);
-          yield usage ? { ...event, usage } : event;
+          accumulatedUsage = retention.usage(accumulatedUsage, event.usage);
+          yield accumulatedUsage ? { ...event, usage: accumulatedUsage } : event;
           return;
         } else {
           // Retain (bounded) the leg's visible content so the rebuilt assistant message is complete.
           if (event.type === "text_delta") {
             legTextChars += event.text.length;
-            if (legTextChars <= MAX_LEG_TEXT_CHARS) legEvents.push(event);
+            // Empty deltas add nothing to the rebuilt message; non-empty ones are bounded by the cap.
+            if (!hasRealToolCall && event.text !== "" && legTextChars <= MAX_LEG_TEXT_CHARS) { retention.retain(event, false, false); legEvents.push(event); }
           } else if (event.type === "thinking_delta") {
             legThinkingChars += event.thinking.length;
-            if (legThinkingChars <= MAX_LEG_THINKING_CHARS) legEvents.push(event);
+            if (!hasRealToolCall && event.thinking !== "" && legThinkingChars <= MAX_LEG_THINKING_CHARS) { retention.retain(event, false, false); legEvents.push(event); }
           } else if (
             event.type === "thinking_signature"
             || event.type === "redacted_thinking"
           ) {
-            legEvents.push(event);
+            if (!hasRealToolCall) { retention.retain(event); legEvents.push(event); }
           }
         }
         yield event;
@@ -245,14 +280,14 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable<OcxParsedRequ
       if (!shouldIntercept) {
         // Plain leg: surface the terminal with merged usage from any earlier intercepted legs.
         if (terminalEvent) {
-          const usage = mergeUsage(accumulatedUsage, terminalEvent.usage);
-          yield usage ? { ...terminalEvent, usage } : terminalEvent;
+          accumulatedUsage = retention.usage(accumulatedUsage, terminalEvent.usage);
+          yield accumulatedUsage ? { ...terminalEvent, usage: accumulatedUsage } : terminalEvent;
         }
         return;
       }
 
       // Consume this leg's done — the continuation replaces it.
-      accumulatedUsage = mergeUsage(accumulatedUsage, terminalEvent?.usage);
+      accumulatedUsage = retention.usage(accumulatedUsage, terminalEvent?.usage);
       if (continuations >= MAX_ADVISOR_CONTINUATIONS_PER_REQUEST || finalContinuationUsed) {
         yield {
           type: "error",
@@ -269,22 +304,22 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable<OcxParsedRequ
       continuations += 1;
       finalContinuationUsed ||= isFinalContinuation;
       const timestamp = Date.now();
-      const assistant = assistantMessageFromLeg(legEvents, advisorCalls, timestamp);
+      const assistant = retention.history(retention.replayReservation(), () => assistantMessageFromLeg(legEvents, advisorCalls, timestamp, retention));
       const messages: OcxMessage[] = [...parsed.context.messages];
       if (assistant) messages.push(assistant);
 
       for (const call of advisorCalls) {
         if (consultations >= maxConsultations) {
           // Runtime-owned limit text: not advice, and never a genuine advice wrapper.
-          messages.push(advisorToolResult(call, {
+          const outcome = {
             ok: false,
             isError: true,
             content: plan.formatUnavailable("limit", "consultation limit reached for this request"),
-          }, timestamp));
+          };
+          messages.push(retention.history(jsonUtf8Bytes(outcome) + jsonUtf8Bytes(call.id) + 256, () => advisorToolResult(call, outcome, timestamp))!);
           continue;
         }
         consultations += 1;
-        const args = parseAdvisorArgs(call.argsBuf);
         // Later calls in the same leg must see what came before them: the rebuilt worker turn
         // and every earlier advisor result in this leg's accumulated messages.
         const consultParsed: OcxParsedRequest = {
@@ -293,8 +328,9 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable<OcxParsedRequ
         };
         let outcome: AdvisorConsultOutcome;
         try {
-          outcome = await plan.consult(consultParsed, "manual", args.question);
+          outcome = await plan.consult(consultParsed, "manual", call.question);
         } catch (error) {
+          if (isTranslatorBudgetExceededError(error)) throw error;
           // Runtime-owned failure text only: the implementation neutralizes untrusted exception
           // text so it can never forge a genuine advice wrapper.
           outcome = {
@@ -303,7 +339,7 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable<OcxParsedRequ
             content: plan.formatUnavailable("manual", error instanceof Error ? error.message : String(error)),
           };
         }
-        messages.push(advisorToolResult(call, outcome, timestamp));
+        messages.push(retention.history(jsonUtf8Bytes(outcome) + jsonUtf8Bytes(call.id) + 256, () => advisorToolResult(call, outcome, timestamp))!);
       }
 
       const exhausted = consultations >= maxConsultations;
@@ -320,10 +356,16 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable<OcxParsedRequ
           : parsed.options,
       };
       parsed = nextParsed;
+      // All replay owners were admitted while the raw leg was still live.
+      held.length = legEvents.length = advisorCalls.length = 0;
+      pending = null;
+      terminalEvent = undefined;
+      retention.releaseLeg();
       yield { type: "assistant_boundary" };
       try {
         source = await options.continuation(parsed);
       } catch (error) {
+        if (isTranslatorBudgetExceededError(error)) throw error;
         yield {
           type: "error",
           message: error instanceof Error ? error.message : String(error),
@@ -332,5 +374,10 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable<OcxParsedRequ
         return;
       }
     }
+    } catch (error) {
+      if (!isTranslatorBudgetExceededError(error)) throw error;
+      yield { type: "error", status: 502, errorType: "upstream_error", code: "translation_buffer_limit",
+        message: "upstream translation buffer exceeded the safe limit", ...(accumulatedUsage ? { usage: accumulatedUsage } : {}) };
+    } finally { retention.release(); }
   };
 }