diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 22640c7d5b2..1f6cc3ff446 100644 --- a/scripts/test-layout/layout.json +++ b/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", diff --git a/src/server/responses/adapter-continuation.ts b/src/server/responses/adapter-continuation.ts index 8bef1f9a7c4..7ce56410fbc 100644 --- a/src/server/responses/adapter-continuation.ts +++ b/src/server/responses/adapter-continuation.ts @@ -739,6 +739,7 @@ export function createAdapterContinuations( return parsed._advisorGuard ? parsed._advisorGuard({ parsed, + translatorBudget, firstEvents: terminalGuardedRetry, continuation: fetchTerminalGuardContinuation, }) diff --git a/src/server/responses/adapter-delivery.ts b/src/server/responses/adapter-delivery.ts index e39ffde29f9..8bd992c2be9 100644 --- a/src/server/responses/adapter-delivery.ts +++ b/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); diff --git a/src/server/responses/advisor-retention.ts b/src/server/responses/advisor-retention.ts new file mode 100644 index 00000000000..8d8dbd6e2ca --- /dev/null +++ b/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(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(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; + } +} diff --git a/src/server/responses/advisor-slot.ts b/src/server/responses/advisor-slot.ts index b10bf9f1167..917c6b2715d 100644 --- a/src/server/responses/advisor-slot.ts +++ b/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; + /** 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 | Promise>; } @@ -172,7 +186,8 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable = 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 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,6 +240,7 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable= MAX_ADVISOR_CONTINUATIONS_PER_REQUEST || finalContinuationUsed) { yield { type: "error", @@ -269,22 +304,22 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable 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 advisorToolResult(call, outcome, timestamp))!); } const exhausted = consultations >= maxConsultations; @@ -320,10 +356,16 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable; /** One bounded worker continuation re-dispatch (same machinery as the terminal guard). */ continuation: (parsed: OcxParsedRequest) => AsyncIterable | Promise>; diff --git a/structure/advisor.md b/structure/advisor.md index ad252f377ec..41e0aba6a8b 100644 --- a/structure/advisor.md +++ b/structure/advisor.md @@ -23,6 +23,23 @@ Management loads `advisor-routes.ts` only for the Advisor namespace. The guard i `_advisorGuard`. `tests/advisor/advisor-core-boundary.test.ts` enforces the transitive load-time boundary from router, lifecycle, Responses core and management API, including direct dynamic imports. +`src/server/responses/advisor-retention.ts` bounds guard-owned retention independently of provider +parser leases: 32 MiB across raw legs, reconstruction copies and appended replay history, 2 MiB +per Advisor call's JSON-encoded argument fragments, 4,096 retained events and 1,024 calls per leg. +Non-empty text/thinking deltas consume bytes but no event slots, because the 16 KiB per-leg +character caps already bound them; empty ones are not retained. Other empty fragments count; +signatures, redacted blocks and opaque tool metadata consume bytes. Real +tool arguments pass through without a retained copy and end replay retention for that leg. +Reconstruction reserves space before joining or parsing admitted sources; raw and replay owners +overlap until the replacement is admitted. Appended history remains charged through the guard's +final exit. Accumulated usage has a separate retained owner across legs; replacement admits the +new usage while raw terminal data and prior usage remain charged. An optional request translator +budget receives separate local charges, never shared +event/call leases or disposal. All exits, including iterator close, release only the guard's own +charges. Refusals use content-free 502 `upstream_error` / `translation_buffer_limit` events. +`tests/advisor/advisor-retention.test.ts` covers aggregate/count admission, representations, +multi-call pairing, overlap and lifecycle release. + ## Execution paths - Translated (non-passthrough, non-run-turn) fetch path: full support — synthetic tool injection, diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 6bcf5538f8e..c269f997f87 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -15,6 +15,11 @@ the translator budget, which is why so many documents link here rather than rest ## Request-copy accounting +The optional Advisor protocol guard follows the [Advisor retention contract](../advisor.md#optional-subsystem-boundary). +Its raw leg, reconstruction and appended replay history use independent local counters, also +charged to the request translator budget when supplied by delivery or empty-completion retry. +These charges do not borrow the bridge's WeakMap event leases or the parser's call leases. + Native Ollama late-output attribution uses `src/lib/json-byte-size.ts` to count the parts as one JSON string before constructing its caption. The exact count includes one pair of quotes, escaping, UTF-8 and surrogate pairs spanning parts, without a diff --git a/tests/advisor/advisor-responses-wiring.test.ts b/tests/advisor/advisor-responses-wiring.test.ts index 572c9d1ef20..197dbcd1eea 100644 --- a/tests/advisor/advisor-responses-wiring.test.ts +++ b/tests/advisor/advisor-responses-wiring.test.ts @@ -11,6 +11,8 @@ */ import { afterEach, describe, expect, test } from "bun:test"; import { activateAdvisor } from "../../src/lib/advisor-activation"; +import { setAdvisorPlanFactory } from "../../src/server/responses/advisor-plan-slot"; +import type { TranslatorBudget } from "../../src/lib/translator-budget"; import { handleResponses } from "../../src/server/responses/core"; import { handleChatCompletions } from "../../src/server/chat-completions"; import { collectSse } from "../helpers/responses-conformance"; @@ -145,6 +147,37 @@ function workerRequest(input: unknown, threadId?: string, stream = true, tools?: } describe("advisor responses wiring (end-to-end)", () => { + test.each([true, false])("delivery and empty-completion retry pass the same request budget (stream=%s)", async stream => { + releaseSpendHome = acquireOwnedSpendHome(); + let sends = 0; + const seen: (TranslatorBudget | undefined)[] = []; + const workerFetch = (async () => { + const text = sends++ === 0 ? "" : "retry answer"; + return stream ? sse(plainFrames(text)) : chatCompletion(text); + }) as typeof fetch; + const config = advisorConfig(undefined, workerFetch); + config.emptyCompletionRetry = true; + setAdvisorPlanFactory(config, async () => ({ + tool: { name: "advisor", description: "fixture", parameters: {}, advisor: true }, + preflightInject: async () => false, + attachGuard: parsed => { + parsed._advisorGuard = async function* (options) { + seen.push(options.translatorBudget); + yield* options.firstEvents; + }; + }, + })); + try { + const response = await handleResponses(workerRequest("budget wiring", undefined, stream), config, { model: "", provider: "" }); + const body = await response.text(); + expect(body).toContain("retry answer"); + expect(sends).toBe(2); + expect(seen).toHaveLength(2); + expect(seen[0]).toBeDefined(); + expect(seen[1]).toBe(seen[0]); + } finally { setAdvisorPlanFactory(config, null); } + }); + for (const delivery of ["buffered", "bridge", "direct"] as const) { for (const retry of [false, true]) for (const catalog of ["empty", "omitted", "partial"] as const) { for (const finalTool of [false, true]) { diff --git a/tests/advisor/advisor-retention.test.ts b/tests/advisor/advisor-retention.test.ts new file mode 100644 index 00000000000..5fec8f56863 --- /dev/null +++ b/tests/advisor/advisor-retention.test.ts @@ -0,0 +1,359 @@ +import { expect, test } from "bun:test"; +import { createAdvisorGuard } from "../../src/server/responses/advisor-slot"; +import { MAX_ADVISOR_RETAINED_CALLS, MAX_ADVISOR_RETAINED_EVENTS } from "../../src/server/responses/advisor-retention"; +import { createTranslatorBudget, retainTranslatedEvent, releaseTranslatedEvent, TRANSLATOR_MAX_CALL_ARGUMENT_BYTES, type TranslatorBudget } from "../../src/lib/translator-budget"; +import { parseRequest } from "../../src/responses/parser"; +import { usageFromOpenAIChat } from "../../src/adapters/openai-chat/response-events"; +import type { AdapterEvent, OcxParsedRequest, OcxProviderOpaqueToolCallMetadata } from "../../src/types"; + +const done: AdapterEvent = { type: "done" }; +const overflow = { type: "error", status: 502, errorType: "upstream_error", code: "translation_buffer_limit" }; +const parsed = () => parseRequest({ model: "worker", input: "fixture", stream: true }); +const stream = (events: AdapterEvent[]) => (async function* () { yield* events; })(); +function* call(id: string, args = "{}", metadata?: OcxProviderOpaqueToolCallMetadata): Generator { + yield { type: "tool_call_start", id, name: "advisor", ...(metadata ? { providerMetadata: metadata } : {}) }; + yield { type: "tool_call_delta", arguments: args }; + yield { type: "tool_call_end" }; +} +async function run(firstEvents: AsyncIterable, budget?: TranslatorBudget, continuation?: (next: OcxParsedRequest) => AsyncIterable) { + const requests: OcxParsedRequest[] = []; + const questions: (string | undefined)[] = []; + const guard = createAdvisorGuard({ + consult: async (_next, _reason, question) => { questions.push(question); return { ok: true, isError: false, content: "advice" }; }, + formatUnavailable: () => "limit result", + }); + const events: AdapterEvent[] = []; + for await (const event of guard({ parsed: parsed(), firstEvents, translatorBudget: budget, continuation: next => { + requests.push(next); + return continuation?.(next) ?? stream([{ type: "text_delta", text: "answer" }, done]); + } })) events.push(event); + return { events, requests, questions }; +} + +test("parser call release cannot free the Advisor's retained argument copies", async () => { + const budget = createTranslatorBudget({ maxTurnBytes: 2_048 }); + let closed = 0; + const first = (async function* () { + for (let i = 0; i < 8; i++) { + const id = `c${i}`; + budget.openCall(id); + try { + yield { type: "tool_call_start", id, name: "advisor" } as AdapterEvent; + budget.chargeRetained(500, { kind: "tool_args", callId: id }); + yield { type: "tool_call_delta", arguments: " ".repeat(500) } as AdapterEvent; + } finally { budget.closeCall(id); closed++; } + yield { type: "tool_call_end" } as AdapterEvent; + } + yield done; + })(); + try { + const result = await run(first, budget); + expect(result.events.at(-1)).toMatchObject(overflow); + expect(result.questions).toHaveLength(0); + expect(closed).toBeGreaterThan(1); + expect(budget.snapshot().currentBytes).toBe(0); + expect(budget.snapshot().activeCalls).toBe(0); + expect(budget.snapshot().highWaterBytes).toBeLessThanOrEqual(2_048); + } finally { budget.dispose(); } +}); + +test("legitimate multi-call batches preserve questions and every excess call's paired result", async () => { + const budget = createTranslatorBudget(); + budget.chargeRetained(37, { kind: "retained_collectors" }); + budget.chargeRetained(41, { kind: "request_copies" }); + try { + const result = await run(stream([ + { type: "thinking_delta", thinking: "reason" }, { type: "thinking_signature", signature: "signature" }, + { type: "redacted_thinking", data: "opaque" }, { type: "text_delta", text: "before" }, + ...Array.from({ length: 5 }, (_, i) => [...call(`c${i}`, JSON.stringify({ question: `질문 ${i} 😀` }))]).flat(), done, + ]), budget, next => { + expect(budget.snapshot().currentBytes).toBeGreaterThan(78); + expect(next.context.messages.filter(m => m.role === "toolResult")).toHaveLength(5); + return stream([done]); + }); + expect(result.questions).toEqual(["질문 0 😀", "질문 1 😀", "질문 2 😀"]); + expect(result.requests).toHaveLength(1); + const assistant = result.requests[0]!.context.messages.find(m => m.role === "assistant"); + expect(assistant?.content).toContainEqual({ type: "thinking", thinking: "reason", signature: "signature", redacted: ["opaque"] }); + expect(result.requests[0]!.context.messages.filter(m => m.role === "toolResult").slice(-2).map(m => m.content)).toEqual(["limit result", "limit result"]); + expect(budget.snapshot().currentBytes).toBe(78); + } finally { budget.dispose(); } +}); + +test.each(["thinking_signature", "redacted_thinking", "tool_call_delta"] as const)("empty %s fragments have count admission without a shared budget", async type => { + const first = (async function* () { + if (type === "tool_call_delta") yield { type: "tool_call_start", id: "c", name: "advisor" } as AdapterEvent; + for (let i = 0; i <= MAX_ADVISOR_RETAINED_EVENTS; i++) { + yield ({ type, ...(type === "thinking_signature" ? { signature: "" } : type === "redacted_thinking" ? { data: "" } : { arguments: "" }) }) as AdapterEvent; + } + yield done; + })(); + expect((await run(first)).events.at(-1)).toMatchObject(overflow); +}); + +test("ordinary token-sized text and thinking deltas never exhaust the event count", async () => { + const deltas = (n: number): AdapterEvent[] => Array.from({ length: n }, (_, i) => + (i % 2 ? { type: "text_delta", text: "abcd" } : { type: "thinking_delta", thinking: "abcd" }) as AdapterEvent); + const empty = Array.from({ length: MAX_ADVISOR_RETAINED_EVENTS + 1 }, (_, i) => + (i % 2 ? { type: "text_delta", text: "" } : { type: "thinking_delta", thinking: "" }) as AdapterEvent); + // A plain answer passes through untouched. + const plain = await run(stream([...deltas(MAX_ADVISOR_RETAINED_EVENTS * 2), ...empty, done])); + expect(plain.events.at(-1)).toEqual(done); + expect(plain.requests).toHaveLength(0); + // A long leg that ends in an advisor call still gets its consultation and continuation. + const advised = await run(stream([...deltas(MAX_ADVISOR_RETAINED_EVENTS * 2), ...call("c", '{"question":"q"}'), done])); + expect(advised.questions).toEqual(["q"]); + expect(advised.requests).toHaveLength(1); + expect(advised.events.some(event => event.type === "error")).toBe(false); +}); + +test("closed calls have a separate count limit, including empty arguments", async () => { + const events = Array.from({ length: MAX_ADVISOR_RETAINED_CALLS + 1 }, (_, i) => [ + { type: "tool_call_start", id: String(i), name: "advisor" }, { type: "tool_call_end" }, + ] as AdapterEvent[]).flat(); + expect((await run(stream([...events, done]))).events.at(-1)).toMatchObject(overflow); +}); + +test.each(["arguments", "signature", "redacted", "metadata"])("%s bytes are admitted before retention", async representation => { + const budget = createTranslatorBudget({ maxTurnBytes: 512 }); + const payload = "😀\\\"".repeat(180); + const metadata = { fixture: payload } as unknown as OcxProviderOpaqueToolCallMetadata; + const first: AdapterEvent[] = representation === "arguments" ? [...call("c", payload)] + : representation === "signature" ? [{ type: "thinking_signature", signature: payload }] + : representation === "redacted" ? [{ type: "redacted_thinking", data: payload }] + : [...call("c", "{}", metadata)]; + try { + const result = await run(stream([...first, done]), budget); + expect(result.events.at(-1)).toMatchObject(overflow); + expect(result.questions).toHaveLength(0); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { budget.dispose(); } +}); + +test("per-call admission is independent of a permissive shared budget", async () => { + const first = stream([ + { type: "tool_call_start", id: "c", name: "advisor" }, + { type: "tool_call_delta", arguments: " ".repeat(TRANSLATOR_MAX_CALL_ARGUMENT_BYTES / 2) }, + { type: "tool_call_delta", arguments: " ".repeat(TRANSLATOR_MAX_CALL_ARGUMENT_BYTES / 2) }, + { type: "tool_call_delta", arguments: "x" }, done, + ]); + expect((await run(first)).events.at(-1)).toMatchObject(overflow); +}); + +test.each([1, 2])("an exact-limit argument in %s fragment(s) is accepted", async fragments => { + const first = (async function* () { + yield { type: "tool_call_start", id: "c", name: "advisor" } as AdapterEvent; + for (let i = 0; i < fragments; i++) { + yield { type: "tool_call_delta", arguments: " ".repeat(TRANSLATOR_MAX_CALL_ARGUMENT_BYTES / fragments) } as AdapterEvent; + } + yield { type: "tool_call_end" } as AdapterEvent; + yield done; + })(); + const result = await run(first); + expect(result.events.at(-1)).toEqual(done); + expect(result.questions).toHaveLength(1); +}); + +test("a single argument fragment one byte over the limit is rejected", async () => { + const result = await run(stream([...call("c", " ".repeat(TRANSLATOR_MAX_CALL_ARGUMENT_BYTES + 1)), done])); + expect(result.events.at(-1)).toMatchObject(overflow); + expect(result.questions).toHaveLength(0); +}); + +test("aggregate retention is bounded independently of parser call release or shared budget", async () => { + const first = (async function* () { + for (let i = 0; i < 34; i++) yield* call(String(i), " ".repeat(1024 * 1024)); + yield done; + })(); + const result = await run(first); + expect(result.events.at(-1)).toMatchObject(overflow); + expect(result.questions).toHaveLength(0); +}); + +test("real tool arguments pass through without retained copies, and discard the replay leg", async () => { + const budget = createTranslatorBudget({ maxTurnBytes: 256 }); + const real: AdapterEvent[] = [ + { type: "tool_call_start", id: "real", name: "shell" }, + { type: "tool_call_delta", arguments: "x".repeat(4_096) }, { type: "tool_call_end" }, done, + ]; + try { + const result = await run(stream([{ type: "text_delta", text: "visible" }, ...real]), budget); + expect(result.events.slice(1)).toEqual(real); + expect(result.requests).toHaveLength(0); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { budget.dispose(); } +}); + +test("replay construction charges overlapping raw and new representations", async () => { + const budget = createTranslatorBudget({ maxTurnBytes: 1_000 }); + try { + const result = await run(stream([...call("c", JSON.stringify({ question: "x".repeat(300) })), done]), budget); + expect(result.events.at(-1)).toMatchObject(overflow); + expect(result.questions).toHaveLength(0); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { budget.dispose(); } +}); + +test("a bridge lease on the same event remains owned and releasable by the bridge", async () => { + const budget = createTranslatorBudget(); + const event: AdapterEvent = { type: "text_delta", text: "shared event" }; + retainTranslatedEvent(event, budget); + const initial = budget.snapshot().currentBytes; + try { + expect((await run(stream([event, done]), budget)).events).toEqual([event, done]); + expect(budget.snapshot().currentBytes).toBe(initial); + releaseTranslatedEvent(event, budget); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { budget.dispose(); } +}); + +test("replay history stays charged while a later leg fails admission", async () => { + const budget = createTranslatorBudget({ maxTurnBytes: 2_048 }); + let beforeSecondLeg = 0; + try { + const result = await run(stream([...call("c"), done]), budget, () => { + beforeSecondLeg = budget.snapshot().currentBytes; + return stream([{ type: "redacted_thinking", data: "x".repeat(2_000) }, done]); + }); + expect(beforeSecondLeg).toBeGreaterThan(0); + expect(result.events.at(-1)).toMatchObject(overflow); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { budget.dispose(); } +}); + +test.each(["cached_tokens", "reasoning_tokens"])("retained OpenAI Chat %s usage prevents later argument over-admission", async field => { + for (const withUsage of [false, true]) { + const budget = createTranslatorBudget({ maxTurnBytes: 6_000 }); + const payload = "u".repeat(1_200); + const usage = withUsage ? usageFromOpenAIChat({ + [field === "cached_tokens" ? "prompt_tokens_details" : "completion_tokens_details"]: { [field]: payload }, + }) : undefined; + let admitted = false; + let retainedAtContinuation = 0; + try { + const result = await run(stream([...call("first"), { type: "done", usage }]), budget, () => { + retainedAtContinuation = budget.snapshot().currentBytes; + return (async function* () { + yield { type: "tool_call_start", id: "second", name: "advisor" } as AdapterEvent; + yield { type: "tool_call_delta", arguments: " ".repeat(4_600) } as AdapterEvent; + admitted = true; + yield { type: "tool_call_end" } as AdapterEvent; + // A real call ends replay retention so the control does not need a second reconstruction. + yield { type: "tool_call_start", id: "real", name: "shell" } as AdapterEvent; + yield { type: "tool_call_end" } as AdapterEvent; + yield done; + })(); + }); + expect(admitted).toBe(!withUsage); + if (withUsage) { + expect(retainedAtContinuation).toBeGreaterThan(payload.length); + expect(result.events.at(-1)).toMatchObject(overflow); + } else expect(result.events.at(-1)?.type).toBe("done"); + expect(result.requests).toHaveLength(1); + expect(budget.snapshot().currentBytes).toBe(0); + expect(budget.snapshot().highWaterBytes).toBeLessThanOrEqual(6_000); + } finally { budget.dispose(); } + } +}); + +test.each(["done", "error", "incomplete"] as const)("normal numeric usage is merged and released on final %s", async terminal => { + const budget = createTranslatorBudget(); + const first = usageFromOpenAIChat({ prompt_tokens: 2, completion_tokens: 3, prompt_tokens_details: { cached_tokens: 1 } }); + const second = usageFromOpenAIChat({ prompt_tokens: 5, completion_tokens: 7, completion_tokens_details: { reasoning_tokens: 4 } }); + try { + const result = await run(stream([...call("c"), { type: "done", usage: first }]), budget, + () => stream([{ type: terminal, usage: second } as AdapterEvent])); + expect(result.events.at(-1)).toMatchObject({ type: terminal, usage: { + inputTokens: 7, outputTokens: 10, totalTokens: 17, cachedInputTokens: 1, reasoningOutputTokens: 4, + } }); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { budget.dispose(); } +}); + +test("absent later usage preserves the admitted usage reference until iterator close", async () => { + const budget = createTranslatorBudget(); + const usage = usageFromOpenAIChat({ prompt_tokens_details: { cached_tokens: "x".repeat(400) } }); + const guard = createAdvisorGuard({ consult: async () => ({ ok: true, isError: false, content: "advice" }), formatUnavailable: () => "limit" }); + const iterator = guard({ parsed: parsed(), translatorBudget: budget, + firstEvents: stream([...call("c"), { type: "done", usage }]), continuation: () => stream([done]), + }); + try { + expect((await iterator.next()).value).toEqual({ type: "assistant_boundary" }); + const terminal = (await iterator.next()).value; + expect(terminal).toMatchObject({ type: "done", usage }); + expect(terminal && "usage" in terminal ? terminal.usage : undefined).toBe(usage); + expect(budget.snapshot().currentBytes).toBeGreaterThan(400); + await iterator.return(); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { await iterator.return(); budget.dispose(); } +}); + +test("a usage merge refusal keeps the previous owner for one content-free budget error", async () => { + const budget = createTranslatorBudget({ maxTurnBytes: 6_000 }); + const first = usageFromOpenAIChat({ prompt_tokens_details: { cached_tokens: "x".repeat(1_200) } }); + const second = usageFromOpenAIChat({ completion_tokens_details: { reasoning_tokens: "y".repeat(2_000) } }); + try { + const result = await run(stream([...call("c"), { type: "done", usage: first }]), budget, + () => stream([{ type: "incomplete", usage: second } as AdapterEvent])); + expect(result.events.filter(event => event.type === "error")).toHaveLength(1); + expect(result.events.at(-1)).toMatchObject(overflow); + const error = result.events.at(-1); + expect(error && "usage" in error ? error.usage : undefined).toBe(first); + expect(error && "message" in error ? error.message : undefined).toBe("upstream translation buffer exceeded the safe limit"); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { budget.dispose(); } +}); + +test("a successful usage replacement releases the previous large owner before the next leg", async () => { + const budget = createTranslatorBudget({ maxTurnBytes: 6_000 }); + const first = usageFromOpenAIChat({ prompt_tokens_details: { cached_tokens: "x".repeat(1_200) } }); + const second = usageFromOpenAIChat({ prompt_tokens: 3, completion_tokens: 4 }); + let legs = 0; + try { + const result = await run(stream([...call("first"), { type: "done", usage: first }]), budget, () => { + if (++legs === 1) { + expect(budget.snapshot().currentBytes).toBeGreaterThan(1_200); + return stream([...call("second"), { type: "done", usage: second }]); + } + expect(budget.snapshot().currentBytes).toBeLessThan(1_200); + return stream([done]); + }); + expect(legs).toBe(2); + expect(result.events.at(-1)).toEqual({ type: "done", usage: { inputTokens: 3, outputTokens: 4, totalTokens: 7 } }); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { budget.dispose(); } +}); + +test.each(["error", "incomplete", "throw", "abort", "close", "boundary", "continuation-throw"])("owned reservations release on %s", async exit => { + const budget = createTranslatorBudget(); + budget.chargeRetained(17, { kind: "retained_collectors" }); + let sourceClosed = false; + let dispatched = false; + const first = (async function* () { + try { + if (exit === "boundary" || exit === "continuation-throw") { yield* call("c"); yield done; return; } + yield { type: "text_delta", text: "held" } as AdapterEvent; + if (exit === "throw") throw new Error("fixture source"); + if (exit === "abort") throw new DOMException("fixture abort", "AbortError"); + yield { type: exit === "incomplete" ? "incomplete" : "error", message: "fixture" } as AdapterEvent; + } finally { sourceClosed = true; } + })(); + const guard = createAdvisorGuard({ consult: async () => ({ ok: true, isError: false, content: "advice" }), formatUnavailable: () => "limit" }); + const iterator = guard({ parsed: parsed(), firstEvents: first, translatorBudget: budget, continuation: () => { + dispatched = true; + throw new Error("fixture continuation"); + } }); + try { + if (exit === "close") { await iterator.next(); await iterator.return(); } + else if (exit === "boundary") { + expect((await iterator.next()).value).toEqual({ type: "assistant_boundary" }); + expect(budget.snapshot().currentBytes).toBeGreaterThan(17); + await iterator.return(); + expect(dispatched).toBe(false); + } else if (exit === "throw") { await expect(async () => { for await (const _event of iterator) {} }).toThrow("fixture source"); } + else if (exit === "abort") { await expect(async () => { for await (const _event of iterator) {} }).toThrow("fixture abort"); } + else { for await (const _event of iterator) {} } + expect(sourceClosed).toBe(true); + expect(budget.snapshot().currentBytes).toBe(17); + } finally { await iterator.return(); budget.dispose(); } +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index e7b826a030f..6b0324a3570 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1,4 +1,5 @@ { + "advisor-retention.test.ts": "advisor", "messages-native-empty-inline.test.ts": "claude-integration", "explicit-empty-tool-catalog.test.ts": "responses", "explicit-empty-tool-catalog-history.test.ts": "responses",