From d19eaae18c2cb544b2bb67d9bccea1d746560d76 Mon Sep 17 00:00:00 2001 From: pavelhov Date: Fri, 23 Oct 2026 20:38:56 -0400 Subject: [PATCH 1/2] fix(compaction): budget Opus summaries and stream Anthropic maintenance Keep verified Opus 5.5 compaction at medium effort with a 64000-token total unless the caller supplies a limit. Adaptive thinking has no hard thinking budget, so the old 40192 total could be exhausted before the handoff summary completed. Stream routed Anthropic v1 summarization internally while preserving the external JSON history, shared memory bounds, cancellation, and incomplete-summary rejection. --- .../content/docs/reference/architecture.md | 11 + src/adapters/anthropic.ts | 43 +++- src/server/responses/compact.ts | 137 +++++++---- structure/04_transports-and-sidecars.md | 13 + tests/anthropic-reasoning.test.ts | 77 +++++- tests/responses-compaction-routing.test.ts | 227 ++++++++++++++++++ tests/server-auth.test.ts | 23 +- 7 files changed, 467 insertions(+), 64 deletions(-) diff --git a/docs-site/src/content/docs/reference/architecture.md b/docs-site/src/content/docs/reference/architecture.md index 29d3168a1d6..ad9316b3e34 100644 --- a/docs-site/src/content/docs/reference/architecture.md +++ b/docs-site/src/content/docs/reference/architecture.md @@ -112,6 +112,17 @@ Anthropic SSE comments and `ping` events count as upstream activity, including w thinking during context compaction. Proxy-generated keep-alives do not reset the stall deadline; an upstream connection that stops sending activity still times out. +For Opus 5.5 through the canonical Anthropic API, routed context compaction uses at most +`medium` reasoning effort and defaults to a 64,000-token total output limit when the caller +omits one. Adaptive thinking and summary text share that limit; effort does not reserve a +fixed number of tokens for the summary. Explicit output limits and lower reasoning efforts +are preserved. This maintenance policy applies to both compaction endpoints; ordinary work +keeps its requested effort and existing defaults. A truncated summary still fails instead +of replacing the conversation history, and the proxy does not automatically retry it. +The older compact endpoint streams routed Anthropic generation internally, then returns one +JSON response to its caller. This lets upstream pings keep a long summary alive without +extending the connection-header timeout or removing the silence watchdog. + Tool calls are disambiguated into three Responses item types using the namespace map, the freeform set, and the tool-search set captured by the parser — so MCP namespaces, `apply_patch`-style freeform tools, and client-executed `tool_search` all round-trip. A `buildResponseJSON()` variant produces a diff --git a/src/adapters/anthropic.ts b/src/adapters/anthropic.ts index f585f3f95f7..5514b8f79fc 100644 --- a/src/adapters/anthropic.ts +++ b/src/adapters/anthropic.ts @@ -40,9 +40,11 @@ function toAnthropicContentPart(p: CodexCommanderContentPart): unknown { const DEFAULT_MAX_TOKENS = 8192; /** Safe ceiling for `max_tokens` (thinking + visible output) across current Claude 4.x models. */ const REASONING_MAX_TOKENS_CEILING = 32_000; -/** Adaptive-thinking ceiling: max effort budget (32k) + OUTPUT_HEADROOM (8k). - * Must exceed REASONING_MAX_TOKENS_CEILING so effort=max actually preserves visible-output room. */ +/** Ordinary adaptive-thinking sizing: heuristic max effort allowance (32k) + 8192. + * Adaptive effort is soft guidance, so this total does not reserve a hard visible-output portion. */ const ADAPTIVE_THINKING_CEILING = 40_192; +/** Bounded thinking + summary total for native Opus 5.5 maintenance compactions. */ +const OPUS_5_5_COMPACTION_MAX_TOKENS = 64_000; /** Anthropic's documented minimum `thinking.budget_tokens`. */ const MIN_THINKING_BUDGET = 1024; /** Visible-output room added above the thinking budget when sizing `max_tokens`. */ @@ -852,11 +854,28 @@ export function createAnthropicAdapter(provider: CodexCommanderProviderConfig, c enforceAnthropicImageLimits(messages); const tools = toolsToAnthropicFormat(parsed, toolNames); + // Compaction is a handoff summary, not another maximum-effort agent turn. Opus 5.5 + // always thinks; adaptive effort has no fixed thinking budget and can consume the + // ordinary 40192-token total before finishing its summary. Keep this policy local + // to the known model and native endpoint: compatible gateways may have lower limits. + const version = parsed._compactionRequest === true ? claudeFamilyVersion(parsed.modelId) : undefined; + const opusCompaction = version?.family === "opus" && version.major === 5 && version.minor === 5 + && usesNativeAnthropicEndpoint(provider); + const requestedEffort = parsed.options.reasoning; + const reasoningEffort = opusCompaction + && requestedEffort !== "none" && requestedEffort !== "minimal" && requestedEffort !== "low" + ? "medium" + : requestedEffort; + const maxOutputTokens = parsed.options.maxOutputTokens + ?? (opusCompaction ? OPUS_5_5_COMPACTION_MAX_TOKENS : undefined); + // Do not mutate parsed.options: requested-effort diagnostics describe the client. + // Anthropic has no reasoningLog field for output_config.effort in the shared contract. + const body: Record = { model: parsed.modelId, messages, stream: parsed.stream, - max_tokens: parsed.options.maxOutputTokens ?? DEFAULT_MAX_TOKENS, + max_tokens: maxOutputTokens ?? DEFAULT_MAX_TOKENS, }; if (isOAuth) { // Claude OAuth (Pro/Max) requires the first system block to be the Claude Code identity. @@ -880,21 +899,21 @@ export function createAnthropicAdapter(provider: CodexCommanderProviderConfig, c // anyway, and thinking shares the caller's `max_tokens` — which truncates a small-budget // request before it can emit its stop sequence (#545). Say "disabled" out loud where the // model both defaults to thinking and accepts being told not to. - if (parsed.options.reasoning === "none" && supportsExplicitThinkingDisable(parsed.modelId)) { + if (reasoningEffort === "none" && supportsExplicitThinkingDisable(parsed.modelId)) { body.thinking = { type: "disabled" }; - } else if (typeof parsed.options.reasoning === "string" && parsed.options.reasoning !== "none") { + } else if (typeof reasoningEffort === "string" && reasoningEffort !== "none") { if (usesAdaptiveThinking(parsed.modelId)) { // Adaptive-thinking models replace the token budget with an effort knob and reject // `thinking.type: "enabled"` outright. `max_tokens` still caps thinking plus visible // output, so high effort needs the same total-token headroom as budget thinking or a // default 8192-token request can spend everything on thought and return empty text. body.thinking = { type: "adaptive" }; - body.output_config = { effort: adaptiveEffort(parsed.options.reasoning) }; - const explicitMaxOut = parsed.options.maxOutputTokens; - const wantBudget = reasoningBudget(parsed.options.reasoning); + body.output_config = { effort: adaptiveEffort(reasoningEffort) }; + const explicitMaxOut = maxOutputTokens; + const wantBudget = reasoningBudget(reasoningEffort); const floor = wantBudget + OUTPUT_HEADROOM; - // Preserve explicit caller limits as-is; for omitted limits use the adaptive ceiling - // so effort=max (budget=32k) still leaves OUTPUT_HEADROOM tokens for visible output. + // Preserve caller limits and the scoped compaction total. Ordinary omitted limits + // use heuristic sizing; adaptive thinking does not enforce wantBudget separately. body.max_tokens = explicitMaxOut !== undefined ? explicitMaxOut : Math.min(ADAPTIVE_THINKING_CEILING, Math.max(DEFAULT_MAX_TOKENS, floor)); @@ -903,8 +922,8 @@ export function createAnthropicAdapter(provider: CodexCommanderProviderConfig, c // visible output) and budget_tokens >= 1024. Codex sends the SAME value for both, which // 400s ("max_tokens must be greater than thinking.budget_tokens"). Size them so max_tokens // always exceeds the budget within a model-safe ceiling, reserving room for visible output. - const maxOut = parsed.options.maxOutputTokens ?? DEFAULT_MAX_TOKENS; - const wantBudget = reasoningBudget(parsed.options.reasoning); + const maxOut = maxOutputTokens ?? DEFAULT_MAX_TOKENS; + const wantBudget = reasoningBudget(reasoningEffort); const maxTokens = Math.min(REASONING_MAX_TOKENS_CEILING, Math.max(maxOut, wantBudget + OUTPUT_HEADROOM)); const budget = Math.max(MIN_THINKING_BUDGET, Math.min(wantBudget, maxTokens - OUTPUT_FLOOR)); body.max_tokens = maxTokens; diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 434aa135af9..2205b0d48ca 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -2,6 +2,8 @@ import { resolveCodexTaskIdentity } from "../../codex/task-identity"; import { rememberNativeArtifacts } from "../../codex/native-ownership"; import { nativeClientMetadata, nativeCompatibilityPolicy, nativeRequestOwner, nativeReplayScope, nativeTurnId, NATIVE_RESPONSES_HEADERS } from "../../responses/native-policy"; import { UpstreamSendBudget } from "../../lib/upstream-send-budget"; +import { decodeServerSentEvents } from "../../lib/sse-decoder"; +import { createTranslatorBudget, isTranslatorBudgetExceededError, type TranslatorBudget } from "../../lib/translator-budget"; import { initializeRequestDispatch, requestDispatchContext } from "../request-log"; import { cleanupResponseDispatch, responseDispatch, observeDispatch } from "../../usage/dispatch-http"; import type { Server } from "bun"; @@ -122,6 +124,37 @@ import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, safeOriginLabel } export const COMPACT_RESPONSE_MAX_BYTES = 32 * 1024 * 1024; +type CompactTurnResponse = { output?: unknown[]; status?: unknown; error?: unknown }; + +/** Consume the bounded internal Responses stream; the external compact protocol stays JSON. */ +async function readCompactTurnStream(response: Response, signal: AbortSignal, budget: TranslatorBudget): Promise { + if (!response.body) throw new Error("missing compaction stream"); + for await (const record of decodeServerSentEvents(response.body, { signal, translatorBudget: budget })) { + if (signal.aborted) throw signal.reason; + if (record.data === "[DONE]") break; + const reservation = budget.reserveTransient(Buffer.byteLength(record.data), { kind: "retained_collectors" }); + try { + const frame = JSON.parse(record.data) as { type?: string; response?: CompactTurnResponse; error?: unknown }; + const type = frame.type ?? record.event; + if (type === "response.completed" || type === "response.incomplete" || type === "response.failed") { + if (!frame.response || typeof frame.response !== "object" || Array.isArray(frame.response)) { + throw new Error("invalid compaction terminal"); + } + // The parsed terminal remains live until v1 history is materialized below. + reservation.commitRetained(); + return frame.response; + } + if (type === "error") { + reservation.commitRetained(); + return { error: frame.error ?? "compaction stream failed" }; + } + } finally { + reservation.release(); + } + } + throw new Error("compaction stream ended without a terminal response"); +} + export function compactResponseTooLargeError(): Response { return new Response(JSON.stringify({ error: { @@ -583,9 +616,13 @@ export async function handleResponsesCompact( // ROUTED model: run the v2 synthetic-compaction turn internally (appends COMPACT_PROMPT, no // tools) and decode the resulting ccx1 envelope into plain v1 replacement-history items. const inputItems = Array.isArray(raw.input) ? (raw.input as unknown[]) : []; + // Anthropic's nonstreaming headers can arrive only after all thinking and summary + // generation, exceeding the connection deadline. Use the normal streaming bridge so + // upstream headers arrive promptly and genuine pings keep its stall watchdog alive. + const streamInternally = route.provider.adapter === "anthropic"; const internalBody = { ...raw, - stream: false, + stream: streamInternally, input: [...inputItems, { type: "compaction_trigger" }], }; const internalHeaders = new Headers({ "content-type": "application/json" }); @@ -598,48 +635,64 @@ export async function handleResponsesCompact( headers: internalHeaders, body: JSON.stringify(internalBody), }); - const response = await handleResponses(internalReq, config, logCtx, { abortSignal: req.signal, turnAdmissionLease }); - if (!response.ok) return response; - let json: { output?: unknown[]; status?: unknown; error?: unknown }; + const budget = streamInternally ? createTranslatorBudget() : undefined; try { - json = await response.json() as { output?: unknown[]; status?: unknown; error?: unknown }; - } catch { - return formatErrorResponse(502, "server_error", "compaction turn returned a non-JSON response"); - } - // The internal turn answers 200 even when it failed or was truncated, so the body - // has to be inspected. Reporting a failure beats installing "(no summary - // available)" as replacement history and silently losing the conversation (#422). - if (json.error) { - const message = typeof json.error === "string" - ? json.error - : (json.error as { message?: unknown })?.message; - return formatErrorResponse(502, "upstream_error", typeof message === "string" ? message : "compaction turn failed"); - } - if (json.status !== "completed") { - return formatErrorResponse( - 502, - "upstream_error", - `compaction turn did not complete (status: ${String(json.status ?? "unknown")})`, - ); - } - const compactionItems = (json.output ?? []).filter( - (item): item is { type: string; encrypted_content?: string } => - !!item && typeof item === "object" && (item as { type?: string }).type === "compaction", - ); - if (compactionItems.length !== 1) { - return formatErrorResponse( - 502, - "invalid_response_error", - `compaction turn produced ${compactionItems.length} compaction items, expected exactly 1`, + const response = await handleResponses(internalReq, config, logCtx, { + abortSignal: req.signal, turnAdmissionLease, ...(budget ? { translatorBudget: budget } : {}), + }); + if (!response.ok) return response; + let json: CompactTurnResponse; + try { + json = budget + ? await readCompactTurnStream(response, req.signal, budget) + : await response.json() as CompactTurnResponse; + } catch (error) { + if (req.signal.aborted) return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); + return formatErrorResponse(502, "server_error", isTranslatorBudgetExceededError(error) + ? "compaction turn exceeded the safe translation buffer limit" + : budget ? "compaction turn returned an invalid or incomplete response" + : "compaction turn returned a non-JSON response"); + } + if (req.signal.aborted) return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); + // The internal turn answers 200 even when it failed or was truncated, so the body + // has to be inspected. Reporting a failure beats installing "(no summary + // available)" as replacement history and silently losing the conversation (#422). + if (json.error) { + const message = typeof json.error === "string" + ? json.error + : (json.error as { message?: unknown })?.message; + return formatErrorResponse(502, "upstream_error", typeof message === "string" ? message : "compaction turn failed"); + } + if (json.status !== "completed") { + return formatErrorResponse( + 502, + "upstream_error", + `compaction turn did not complete (status: ${String(json.status ?? "unknown")})`, + ); + } + const compactionItems = (json.output ?? []).filter( + (item): item is { type: string; encrypted_content?: string } => + !!item && typeof item === "object" && (item as { type?: string }).type === "compaction", ); + if (compactionItems.length !== 1) { + return formatErrorResponse( + 502, + "invalid_response_error", + `compaction turn produced ${compactionItems.length} compaction items, expected exactly 1`, + ); + } + const encrypted = compactionItems[0]!.encrypted_content; + const decoded = typeof encrypted === "string" ? decodeCompactionSummary(encrypted) : null; + // An empty `ccx1:` envelope decodes to "" rather than null, so length is what matters. + if (decoded === null || decoded.trim().length === 0) { + return formatErrorResponse(502, "invalid_response_error", "compaction turn produced an empty summary"); + } + const summary = decoded; + const output = buildCompactV1Output(extractCompactUserMessages(inputItems), summary); + return new Response(JSON.stringify({ output }), { headers: { "Content-Type": "application/json" } }); + } finally { + // The compact handler owns the shared budget until its internal stream and + // decoder have both released their readers, including abort/error paths. + budget?.dispose(); } - const encrypted = compactionItems[0]!.encrypted_content; - const decoded = typeof encrypted === "string" ? decodeCompactionSummary(encrypted) : null; - // An empty `ccx1:` envelope decodes to "" rather than null, so length is what matters. - if (decoded === null || decoded.trim().length === 0) { - return formatErrorResponse(502, "invalid_response_error", "compaction turn produced an empty summary"); - } - const summary = decoded; - const output = buildCompactV1Output(extractCompactUserMessages(inputItems), summary); - return new Response(JSON.stringify({ output }), { headers: { "Content-Type": "application/json" } }); } diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index d053961a13c..aeb47978768 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -33,6 +33,19 @@ within their route; neither route falls through to the other. See and before the `/v1/*` guard. Unknown `/v1/*` paths return JSON 404 errors instead of falling through to GUI static serving. +The Anthropic adapter applies a maintenance-only budget to fresh routed compaction requests +for the verified Opus 5.5 model on the canonical Anthropic API: effort is capped at `medium` +without raising lower efforts, and an omitted output limit becomes 64,000 total tokens. +Explicit limits remain authoritative. Adaptive thinking has no fixed `budget_tokens` +reservation, so the ordinary effort-to-output heuristic cannot guarantee room for summary +text. Both v1 compact and v2 `compaction_trigger` requests carry `_compactionRequest` to this +boundary; replayed summaries do not activate it. Other models, custom destinations, and ordinary +work retain their existing behavior. Token-limit stops remain incomplete and install no partial +compaction item; there is no automatic retry or model substitution. +Routed Anthropic v1 compact generation streams internally and consumes the bounded Responses +terminal through the SSE decoder before building the v1 JSON replacement history. This preserves +the header deadline, ping-aware stall watchdog, and client cancellation for long summaries. + ### Responses reasoning effort boundary `openai-responses` preserves the caller's Responses request shape and the upstream response shape; diff --git a/tests/anthropic-reasoning.test.ts b/tests/anthropic-reasoning.test.ts index fa55ae1245d..92c999973d3 100644 --- a/tests/anthropic-reasoning.test.ts +++ b/tests/anthropic-reasoning.test.ts @@ -25,6 +25,75 @@ async function bodyOf(p: CodexCommanderParsedRequest): Promise { + test.each([ + "claude-opus-5-5", + "anthropic/claude-opus-5-5", + "claude-opus-5-5-20260922", + "claude-opus-5-5[1m]", + "claude-opus-5-5/vendor-suffix", + ])("Opus 5.5 compaction %s uses medium effort and a bounded default total", async modelId => { + const request = parsed("max", {}, modelId); + request._compactionRequest = true; + const b = await bodyOf(request); + expect(b.max_tokens).toBe(64_000); + expect(b.thinking).toEqual({ type: "adaptive" }); + expect(b.output_config).toEqual({ effort: "medium" }); + expect(request.options.reasoning).toBe("max"); + expect(request.options.maxOutputTokens).toBeUndefined(); + }); + + test.each([ + [undefined, "medium"], ["medium", "medium"], ["high", "medium"], + ["xhigh", "medium"], ["max", "medium"], ["ultra", "medium"], + ["low", "low"], ["minimal", "low"], ["none", undefined], + ])("Opus 5.5 compaction keeps effort %s at most medium", async (effort, expected) => { + const request = parsed(effort, {}, "claude-opus-5-5"); + request._compactionRequest = true; + const b = await bodyOf(request); + expect(b.max_tokens).toBe(64_000); + expect(b.output_config).toEqual(expected ? { effort: expected } : undefined); + expect(b.thinking).toEqual(expected ? { type: "adaptive" } : undefined); + expect(request.options.reasoning).toBe(effort); + }); + + test.each([4096, 96_000])("Opus 5.5 compaction preserves explicit total %s", async maxOutputTokens => { + const request = parsed("max", { maxOutputTokens }, "claude-opus-5-5"); + request._compactionRequest = true; + const b = await bodyOf(request); + expect(b.max_tokens).toBe(maxOutputTokens); + expect(b.output_config).toEqual({ effort: "medium" }); + expect(request.options).toEqual({ reasoning: "max", maxOutputTokens }); + }); + + test.each([ + ["claude-opus-5-4", 40_192], ["claude-opus-5-6", 40_192], + ["claude-opus-6-5", 40_192], ["claude-opus-5-50", 40_192], + ["claude-sonnet-5-5", 40_192], ["claude-fable-5", 40_192], + ["custom-model", 32_000], ["claude-opus-4-6", 32_000], + ])("compaction policy leaves other model %s unchanged", async (modelId, expectedMax) => { + const request = parsed("max", {}, modelId); + request._compactionRequest = true; + const b = await bodyOf(request); + expect(b.max_tokens).toBe(expectedMax); + if (expectedMax === 40_192) expect(b.output_config).toEqual({ effort: "max" }); + else expect(b.thinking).toMatchObject({ type: "enabled" }); + }); + + test("ordinary Opus 5.5 max effort keeps its existing default", async () => { + const b = await bodyOf(parsed("max", {}, "claude-opus-5-5")); + expect(b.max_tokens).toBe(40_192); + expect(b.output_config).toEqual({ effort: "max" }); + }); + + test("custom Anthropic-compatible destinations keep their existing Opus 5.5 limits", async () => { + const request = parsed("max", {}, "claude-opus-5-5"); + request._compactionRequest = true; + const built = await createAnthropicAdapter({ ...provider, baseUrl: "https://gateway.example" }).buildRequest(request); + const b = JSON.parse(String(built.body)); + expect(b.max_tokens).toBe(40_192); + expect(b.output_config).toEqual({ effort: "max" }); + }); + test("reasoning 'none' does NOT enable thinking and preserves temperature/top_p", async () => { const b = await bodyOf(parsed("none", { temperature: 0.3, topP: 0.9 })); expect(b.thinking).toBeUndefined(); @@ -69,10 +138,10 @@ describe("anthropic extended-thinking gate", () => { expect(b.output_config).toEqual({ effort: "low" }); }); - test("adaptive-thinking model resizes max_tokens for high effort (issue #246)", async () => { + test("adaptive-thinking model uses the ordinary max-effort total (issue #246)", async () => { const b = await bodyOf(parsed("max", {}, "claude-fable-5")); - // Exact regression: effort=max budget is 32000; adaptive ceiling adds OUTPUT_HEADROOM (8192) - // so max_tokens = 40192, genuinely above the reasoning budget at full effort. + // Ordinary sizing uses a 32000-token allowance plus 8192, not a hard thinking + // budget: adaptive thinking can consume any portion of this shared total. expect(b.max_tokens as number).toBe(40_192); expect(b.thinking).toEqual({ type: "adaptive" }); expect(b.output_config).toEqual({ effort: "max" }); @@ -102,7 +171,7 @@ describe("anthropic extended-thinking gate", () => { ["xhigh", 32_768], ["max", 40_192], ["ultra", 40_192], // ultra mirrors the codex-rs boundary and clamps to max on the wire. - ])("adaptive-thinking %s effort reserves visible-output headroom", async (effort, expected) => { + ])("adaptive-thinking %s effort uses its ordinary total-token default", async (effort, expected) => { const b = await bodyOf(parsed(effort, {}, "claude-fable-5")); expect(b.max_tokens).toBe(expected); }); diff --git a/tests/responses-compaction-routing.test.ts b/tests/responses-compaction-routing.test.ts index d13589c631b..a03b9a90a3d 100644 --- a/tests/responses-compaction-routing.test.ts +++ b/tests/responses-compaction-routing.test.ts @@ -27,6 +27,8 @@ import { supportsNativeResponsesCompactEndpoint } from "../src/providers/openai- import type { RequestLogContext } from "../src/server/request-log"; import { acquireNativeMainProfileDrain, tryAdmitTurn } from "../src/server/lifecycle"; import type { CodexCommanderConfig, CodexCommanderProviderConfig } from "../src/types"; +import { decodeCompactionSummary, encodeCompactionSummary } from "../src/responses/compaction"; +import { translatorAggregateCurrentBytesForTests, translatorLiveBudgetCountForTests } from "../src/lib/translator-budget"; const originalFetch = globalThis.fetch; @@ -137,6 +139,231 @@ function sseResponse(events: Array>): Response { }); } +describe("Opus 5.5 compaction output policy through the real Anthropic adapter", () => { + const config = keyProviderConfig({ adapter: "anthropic", baseUrl: "https://api.anthropic.com" }); + type WireScalars = { model: unknown; maxTokens: unknown; thinking: unknown; effort: unknown; stream: unknown; tools: boolean }; + function stubAnthropic(stopReason = "end_turn") { + const sent: WireScalars[] = []; + globalThis.fetch = (async (_url: unknown, init?: RequestInit) => { + const body = JSON.parse(String(init?.body ?? "{}")); + // Retain only synthetic wire-policy scalars, never request history or headers. + sent.push({ model: body.model, maxTokens: body.max_tokens, thinking: body.thinking?.type, + effort: body.output_config?.effort, stream: body.stream, tools: body.tools !== undefined }); + // A compact synthetic workload needs 45000 total generated tokens; no large + // generated text is needed to expose the former 40192-token truncation. + const outputTokens = Math.min(body.max_tokens, 45_000); + const effectiveStop = body.max_tokens < 45_000 ? "max_tokens" : stopReason; + if (!body.stream) return jsonResponse({ + content: [{ type: "text", text: "Progress and next steps." }], + stop_reason: effectiveStop, usage: { input_tokens: 20, output_tokens: outputTokens }, + }); + return sseResponse([ + { type: "message_start", message: { usage: { input_tokens: 20, output_tokens: 0 } } }, + { type: "content_block_start", index: 0, content_block: { type: "thinking", thinking: "" } }, + { type: "ping" }, + { type: "content_block_stop", index: 0 }, + { type: "content_block_start", index: 1, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 1, delta: { type: "text_delta", text: "Progress and next steps." } }, + { type: "content_block_stop", index: 1 }, + { type: "message_delta", delta: { stop_reason: effectiveStop }, usage: { output_tokens: outputTokens } }, + { type: "message_stop" }, + ]); + }) as typeof fetch; + return sent; + } + function body(extra: Record = {}) { + return baseCompactionBody({ model: "gw/claude-opus-5-5", reasoning: { effort: "max" }, stream: true, ...extra }); + } + function terminal(text: string) { + const frames = text.split("\n\n").flatMap(frame => { + const data = frame.split("\n").find(line => line.startsWith("data: "))?.slice(6); + return data && data !== "[DONE]" ? [JSON.parse(data)] : []; + }); + const terminals = frames.filter(frame => frame.type === "response.completed" || frame.type === "response.incomplete"); + expect(terminals).toHaveLength(1); + return terminals[0].response as { status: string; output: Array<{ type: string; encrypted_content: string }>; incomplete_details?: { reason: string } }; + } + + test.each(["v1", "v2"])("%s compaction gets one complete summary at medium with a bounded total", async version => { + const sent = stubAnthropic(); + const logCtx: RequestLogContext = { model: "", provider: "" }; + if (version === "v1") { + const requestBody = body(); + (requestBody.input as Array<{ type: string }>).pop(); + const response = await handleResponsesCompact(compactionRequest(requestBody), config, logCtx); + expect(response.status).toBe(200); + const json = await response.json() as { output: Array<{ type: string; content: Array<{ text: string }> }> }; + expect(json.output.at(-1)?.content[0]?.text).toContain("Progress and next steps."); + expect(json.output.every(item => item.type === "message")).toBe(true); + } else { + const response = await handleResponses(compactionRequest(body()), config, logCtx); + const result = terminal(await response.text()); + expect(result.status).toBe("completed"); + expect(result.output).toHaveLength(1); + expect(result.output[0]!.type).toBe("compaction"); + expect(decodeCompactionSummary(result.output[0]!.encrypted_content)).toBe("Progress and next steps."); + } + expect(sent).toEqual([{ model: "claude-opus-5-5", maxTokens: 64_000, thinking: "adaptive", + effort: "medium", stream: true, tools: false }]); + expect(logCtx.usage?.outputTokens).toBe(45_000); + }); + + test.each([4096, 96_000])("both compaction endpoints preserve caller total %s", async maxOutputTokens => { + const sent = stubAnthropic(); + const requestBody = body({ max_output_tokens: maxOutputTokens }); + await (await handleResponses(compactionRequest(requestBody), config, { model: "", provider: "" })).text(); + await (await handleResponsesCompact(compactionRequest(requestBody), config, { model: "", provider: "" })).text(); + expect(sent).toHaveLength(2); + expect(sent.map(item => [item.maxTokens, item.effort])).toEqual([[maxOutputTokens, "medium"], [maxOutputTokens, "medium"]]); + }); + + test.each(["compaction", "context_compaction"])("stored %s history does not start the maintenance policy", async type => { + const sent = stubAnthropic(); + await (await handleResponses(compactionRequest(body({ + input: [{ type, encrypted_content: encodeCompactionSummary("Earlier progress.") }, + { type: "message", role: "user", content: "Continue." }], + })), config, { model: "", provider: "" })).text(); + expect(sent).toHaveLength(1); + expect(sent[0]).toMatchObject({ maxTokens: 40_192, effort: "max" }); + }); + + test.each(["v1", "v2"])("%s max_tokens keeps partial summaries out of replacement history without retrying", async version => { + const sent = stubAnthropic("max_tokens"); + if (version === "v1") { + const response = await handleResponsesCompact(compactionRequest(body()), config, { model: "", provider: "" }); + expect(response.status).toBe(502); + const json = await response.json() as { error: { message: string }; output?: unknown }; + expect(json.error.message).toContain("did not complete"); + expect(json.output).toBeUndefined(); + } else { + const response = await handleResponses(compactionRequest(body()), config, { model: "", provider: "" }); + const result = terminal(await response.text()); + expect(result.status).toBe("incomplete"); + expect(result.incomplete_details?.reason).toBe("max_output_tokens"); + expect(result.output.some(item => item.type === "compaction")).toBe(false); + } + expect(sent).toHaveLength(1); + }); + + function delayedAnthropic(end: "complete" | "eof" | "error" | "wait") { + let sends = 0; + let cancellations = 0; + let pingCount = 0; + let upstreamBody: ReadableStream | undefined; + let fetched!: () => void; + const fetchStarted = new Promise(resolve => { fetched = resolve; }); + globalThis.fetch = (async (_url: unknown, init?: RequestInit) => { + sends++; + const stream = JSON.parse(String(init?.body ?? "{}")).stream === true; + fetched(); + if (!stream) { + // A nonstreaming provider cannot return headers until generation is done. + await new Promise((resolve, reject) => { + const timer = setTimeout(resolve, 90); + init?.signal?.addEventListener("abort", () => { clearTimeout(timer); reject(init.signal?.reason); }, { once: true }); + }); + return jsonResponse({ content: [{ type: "text", text: "Delayed summary." }], stop_reason: "end_turn" }); + } + const encoder = new TextEncoder(); + let pingTimer: ReturnType; + let finishTimer: ReturnType | undefined; + let stopped = false; + let controller: ReadableStreamDefaultController; + const clear = () => { + clearInterval(pingTimer); + clearTimeout(finishTimer); + init?.signal?.removeEventListener("abort", onAbort); + }; + const cancel = () => { + if (stopped) return; + stopped = true; + cancellations++; + clear(); + }; + // Match fetch semantics: aborting after headers rejects a locked body read too. + const onAbort = () => { cancel(); controller.error(init?.signal?.reason); }; + upstreamBody = new ReadableStream({ + start(source) { + controller = source; + init?.signal?.addEventListener("abort", onAbort, { once: true }); + pingTimer = setInterval(() => { + pingCount++; + controller.enqueue(encoder.encode('event: ping\ndata: {"type":"ping"}\n\n')); + }, 10); + if (end !== "wait") finishTimer = setTimeout(() => { + clear(); + stopped = true; + controller.enqueue(encoder.encode('event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Delayed summary."}}\n\n')); + if (end === "complete") controller.enqueue(encoder.encode('event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"input_tokens":20,"output_tokens":45000}}\n\nevent: message_stop\ndata: {"type":"message_stop"}\n\n')); + if (end === "error") controller.enqueue(encoder.encode('event: error\ndata: {"type":"error","error":{"type":"overloaded_error","message":"Synthetic upstream failure"}}\n\n')); + controller.close(); + }, 90); + }, + cancel, + }); + return new Response(upstreamBody, { headers: { "content-type": "text/event-stream" } }); + }) as typeof fetch; + return { fetchStarted, sends: () => sends, cancellations: () => cancellations, + pingCount: () => pingCount, body: () => upstreamBody }; + } + + test("v1 consumes Anthropic SSE beyond the header deadline and returns one JSON history", async () => { + const liveBudgets = translatorLiveBudgetCountForTests(); + const retainedBytes = translatorAggregateCurrentBytesForTests(); + const upstream = delayedAnthropic("complete"); + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponsesCompact(compactionRequest(body()), { ...config, connectTimeoutMs: 30 }, logCtx); + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toContain("application/json"); + const json = await response.json() as { output: Array<{ type: string; content: Array<{ text: string }> }> }; + expect(json.output.at(-1)?.content[0]?.text).toContain("Delayed summary."); + expect(json.output.every(item => item.type === "message")).toBe(true); + expect(logCtx.usage?.outputTokens).toBe(45_000); + expect(upstream.sends()).toBe(1); + expect(upstream.pingCount()).toBeGreaterThan(0); + expect(upstream.body()?.locked).toBe(false); + expect(translatorLiveBudgetCountForTests()).toBe(liveBudgets); + expect(translatorAggregateCurrentBytesForTests()).toBe(retainedBytes); + }); + + test.each(["eof", "error"] as const)("v1 Anthropic SSE %s cannot install a partial summary", async end => { + const liveBudgets = translatorLiveBudgetCountForTests(); + const retainedBytes = translatorAggregateCurrentBytesForTests(); + const upstream = delayedAnthropic(end); + const response = await handleResponsesCompact(compactionRequest(body()), { ...config, connectTimeoutMs: 30 }, { model: "", provider: "" }); + expect(response.status).toBe(502); + const json = await response.json() as { error?: unknown; output?: unknown }; + expect(json.error).toBeDefined(); + expect(json.output).toBeUndefined(); + expect(upstream.sends()).toBe(1); + expect(upstream.body()?.locked).toBe(false); + expect(translatorLiveBudgetCountForTests()).toBe(liveBudgets); + expect(translatorAggregateCurrentBytesForTests()).toBe(retainedBytes); + }); + + test("v1 client abort cancels the Anthropic SSE reader and returns no history", async () => { + const liveBudgets = translatorLiveBudgetCountForTests(); + const retainedBytes = translatorAggregateCurrentBytesForTests(); + const upstream = delayedAnthropic("wait"); + const abort = new AbortController(); + const pending = handleResponsesCompact(compactionRequest(body(), abort.signal), { ...config, connectTimeoutMs: 30 }, { model: "", provider: "" }); + await upstream.fetchStarted; + await new Promise(resolve => setTimeout(resolve, 0)); + abort.abort(); + const response = await pending; + // The shared bridge closes its adapter generator asynchronously on cancellation. + await new Promise(resolve => setTimeout(resolve, 0)); + expect(response.status).toBe(499); + const json = await response.json() as { output?: unknown }; + expect(json.output).toBeUndefined(); + expect(upstream.sends()).toBe(1); + expect(upstream.cancellations()).toBe(1); + expect(upstream.body()?.locked).toBe(false); + expect(translatorLiveBudgetCountForTests()).toBe(liveBudgets); + expect(translatorAggregateCurrentBytesForTests()).toBe(retainedBytes); + }); +}); + describe("supportsNativeResponsesCompactEndpoint (#422)", () => { const canonicalForward = { adapter: "openai-responses", diff --git a/tests/server-auth.test.ts b/tests/server-auth.test.ts index 6392e2cbc84..f1f49d9a8e4 100644 --- a/tests/server-auth.test.ts +++ b/tests/server-auth.test.ts @@ -1056,12 +1056,23 @@ describe("server local API auth", () => { const upstream = Bun.serve({ port: 0, - fetch() { - // Anthropic non-stream response carrying the summarizer's text. - return Response.json({ - content: [{ type: "text", text: "compact summary body" }], - usage: { input_tokens: 10, output_tokens: 5 }, - }); + async fetch(request) { + const body = await request.json() as { stream?: boolean }; + expect(body.stream).toBe(true); + return new Response([ + 'event: message_start\n', + 'data: {"type":"message_start","message":{"usage":{"input_tokens":10,"output_tokens":0}}}\n\n', + 'event: content_block_start\n', + 'data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}\n\n', + 'event: content_block_delta\n', + 'data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"compact summary body"}}\n\n', + 'event: content_block_stop\n', + 'data: {"type":"content_block_stop","index":0}\n\n', + 'event: message_delta\n', + 'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}\n\n', + 'event: message_stop\n', + 'data: {"type":"message_stop"}\n\n', + ].join(""), { headers: { "content-type": "text/event-stream" } }); }, }); saveConfig({ From 0d547408fc711bee7588ef090278bc52885588fb Mon Sep 17 00:00:00 2001 From: pavelhov Date: Fri, 23 Oct 2026 20:38:56 -0400 Subject: [PATCH 2/2] chore(release): prepare 0.1.24 --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index 01024581494..0fcf1649f68 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "codexcommander", - "version": "0.1.23", + "version": "0.1.24", "private": true, "description": "CodexCommander — universal provider proxy for OpenAI Codex & Claude Code — use any LLM with Codex CLI/App/SDK and Claude Code", "type": "module",