From c233a7cf68e4f7a9a67362941cabce5f2ad726d7 Mon Sep 17 00:00:00 2001 From: devmello Date: Mon, 3 Aug 2026 00:59:38 -0700 Subject: [PATCH 1/8] fix(claude): keep a marker for tool_result document blocks A tool_result whose content is a document block translated to an empty tool output, so routed models treated the attachment as a tool that returned nothing. Surface the same "[document: title]" marker the user-message path already emits. (cherry picked from commit 3cdf1902012d7b74915a6292a27d70b821ce28ed) --- src/claude/inbound.ts | 4 ++++ tests/claude-inbound.test.ts | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+) diff --git a/src/claude/inbound.ts b/src/claude/inbound.ts index d27169c516..86cdfc1046 100644 --- a/src/claude/inbound.ts +++ b/src/claude/inbound.ts @@ -106,6 +106,10 @@ function toolResultOutput(block: Rec): string | Rec[] { } else if (item.type === "image") { const img = imageBlockToInputImage(item); if (img) out.push(img); + } else if (item.type === "document") { + // Same marker as the user-message document case below: the model should see the + // attachment happened instead of an empty tool output. + out.push({ type: "input_text", text: `[document${typeof item.title === "string" ? `: ${item.title}` : ""}]` }); } } if (isError) out.unshift({ type: "input_text", text: "[tool error]" }); diff --git a/tests/claude-inbound.test.ts b/tests/claude-inbound.test.ts index 2065169dca..2edb35ca19 100644 --- a/tests/claude-inbound.test.ts +++ b/tests/claude-inbound.test.ts @@ -175,6 +175,41 @@ describe("claude inbound translation", () => { expect(() => parseRequest(body)).not.toThrow(); }); + test("tool_result document blocks surface the attachment marker", () => { + const body = anthropicToResponsesBody({ + model: "m", max_tokens: 10, + messages: [ + { role: "assistant", content: [{ type: "tool_use", id: "t1", name: "Read", input: {} }] }, + { + role: "user", + content: [{ + type: "tool_result", tool_use_id: "t1", + content: [ + { type: "text", text: "3 pages" }, + { type: "document", source: { type: "base64", media_type: "application/pdf", data: "aWc=" }, title: "report.pdf" }, + ], + }], + }, + { role: "assistant", content: [{ type: "tool_use", id: "t2", name: "Read", input: {} }] }, + { + role: "user", + content: [{ + type: "tool_result", tool_use_id: "t2", + content: [{ type: "document", source: { type: "base64", media_type: "application/pdf", data: "aWc=" } }], + }], + }, + ], + }) as any; + expect(body.input[1].output).toEqual([ + { type: "input_text", text: "3 pages" }, + { type: "input_text", text: "[document: report.pdf]" }, + ]); + // An untitled document still leaves a marker rather than the empty output that + // read as "the tool returned nothing". + expect(body.input[3].output).toEqual([{ type: "input_text", text: "[document]" }]); + expect(() => parseRequest(body)).not.toThrow(); + }); + test("modelMap: exact, date-stripped, passthrough", () => { const cc = { modelMap: { "claude-sonnet-4-5": "gemini/gemini-3-flash", "claude-opus-4": "xai/grok-4" } }; expect(resolveInboundModel("claude-sonnet-4-5", cc)).toBe("gemini/gemini-3-flash"); From 6dbd9723c1bb834b3a0b3e5ac4a99ce4aa6e85e6 Mon Sep 17 00:00:00 2001 From: devmello Date: Mon, 3 Aug 2026 03:05:46 -0700 Subject: [PATCH 2/8] fix(google): fail closed on MALFORMED_FUNCTION_CALL without a call part Gemini usually drops the malformed call upstream, so the final chunk carries only the finishReason and the started-calls guard never fired. The turn surfaced as a clean empty completion instead of an error. MAX_TOKENS with no started call keeps its plain token-limit stop. (cherry picked from commit 6eb59036969f14ae00282364f853a89f287ecbd5) --- src/adapters/google-truncation.ts | 11 +++++++++++ src/adapters/google.ts | 6 +++--- tests/google-vertex-stream.test.ts | 29 ++++++++++++++++++++++++++++- 3 files changed, 42 insertions(+), 4 deletions(-) diff --git a/src/adapters/google-truncation.ts b/src/adapters/google-truncation.ts index 2082917963..5ac6de1cd9 100644 --- a/src/adapters/google-truncation.ts +++ b/src/adapters/google-truncation.ts @@ -11,3 +11,14 @@ export function vertexTruncationErrorMessage(reason?: string): string { const suffix = reason ? ` (${redactSecretString(reason).slice(0, 160)})` : ""; return `Vertex AI response truncated upstream before the turn completed${suffix}`; } + +/** + * Whether a finished turn must fail closed. A truncation reason arriving mid tool call always + * does. MALFORMED_FUNCTION_CALL fails closed even with zero started calls: the malformed call + * is dropped upstream and usually never materializes as a part, so the turn is incomplete + * despite looking empty. MAX_TOKENS with no started call stays a plain token-limit stop. + */ +export function isVertexTruncatedTurn(finishReason: string | undefined, toolCallsStarted: number): boolean { + if (!isVertexTruncationReason(finishReason)) return false; + return toolCallsStarted > 0 || finishReason === "MALFORMED_FUNCTION_CALL"; +} diff --git a/src/adapters/google.ts b/src/adapters/google.ts index c9d7aedfcf..58374ddcdc 100644 --- a/src/adapters/google.ts +++ b/src/adapters/google.ts @@ -17,7 +17,7 @@ import { contentPartsToText, parseDataUrl } from "./image"; import { getVertexAccessToken } from "../lib/gcp-adc"; import { fetchAntigravityWithRetry, fetchVertexWithRetry } from "./google-http"; import { safeAntigravityHttpErrorMessage, safeVertexHttpErrorMessage } from "./google-errors"; -import { isVertexTruncationReason, vertexTruncationErrorMessage } from "./google-truncation"; +import { isVertexTruncatedTurn, vertexTruncationErrorMessage } from "./google-truncation"; import { ANTIGRAVITY_REQUEST_UA, antigravitySessionId, isLikelyRealThoughtSignature, sanitizeAntigravityClaudeSignatures } from "./google-antigravity-wire"; import { compileGoogleWireBody } from "./google-wire-compiler"; import { identifyRoutedModel } from "./identity"; @@ -596,7 +596,7 @@ export function createGoogleAdapter(provider: OcxProviderConfig): ProviderAdapte // Fail-closed: a turn cut off mid tool call (MAX_TOKENS / MALFORMED_FUNCTION_CALL) surfaces // an error instead of a silently-incomplete done. Mirrors kiro-truncation. if ((provider.googleMode === "vertex" || provider.googleMode === "cloud-code-assist") - && toolCallsStarted > 0 && isVertexTruncationReason(lastFinishReason)) { + && isVertexTruncatedTurn(lastFinishReason, toolCallsStarted)) { yield { type: "error", message: vertexTruncationErrorMessage(lastFinishReason) }; return; } @@ -752,7 +752,7 @@ export function createGoogleAdapter(provider: OcxProviderConfig): ProviderAdapte // Fail-closed truncation, same as the stream path: a non-stream turn cut off mid tool call // (MAX_TOKENS / MALFORMED_FUNCTION_CALL) surfaces an error instead of a silent done. if ((provider.googleMode === "vertex" || provider.googleMode === "cloud-code-assist") - && toolCallsStarted > 0 && isVertexTruncationReason(candidates?.[0]?.finishReason)) { + && isVertexTruncatedTurn(candidates?.[0]?.finishReason, toolCallsStarted)) { return finish([{ type: "error", message: vertexTruncationErrorMessage(candidates?.[0]?.finishReason) }]); } diff --git a/tests/google-vertex-stream.test.ts b/tests/google-vertex-stream.test.ts index 120883c16f..996c9ba738 100644 --- a/tests/google-vertex-stream.test.ts +++ b/tests/google-vertex-stream.test.ts @@ -1,6 +1,6 @@ import { describe, expect, test } from "bun:test"; import { createGoogleAdapter as createGoogleAdapterProduction } from "../src/adapters/google"; -import { isVertexTruncationReason, vertexTruncationErrorMessage } from "../src/adapters/google-truncation"; +import { isVertexTruncatedTurn, isVertexTruncationReason, vertexTruncationErrorMessage } from "../src/adapters/google-truncation"; import { bridgeToResponsesSSE } from "../src/bridge"; import type { AdapterEvent, OcxProviderConfig } from "../src/types"; import { withTestTranslatorBudget } from "./helpers/translator-budget"; @@ -30,6 +30,14 @@ describe("vertex truncation helpers", () => { expect(isVertexTruncationReason(undefined)).toBe(false); expect(vertexTruncationErrorMessage("MAX_TOKENS")).toContain("truncated upstream"); }); + + test("MALFORMED_FUNCTION_CALL fails closed with zero started calls; MAX_TOKENS does not", () => { + expect(isVertexTruncatedTurn("MALFORMED_FUNCTION_CALL", 0)).toBe(true); + expect(isVertexTruncatedTurn("MAX_TOKENS", 0)).toBe(false); + expect(isVertexTruncatedTurn("MAX_TOKENS", 1)).toBe(true); + expect(isVertexTruncatedTurn("STOP", 5)).toBe(false); + expect(isVertexTruncatedTurn(undefined, 5)).toBe(false); + }); }); describe("vertex parseStream fail-closed truncation", () => { @@ -77,6 +85,17 @@ describe("vertex parseStream fail-closed truncation", () => { expect(text).toContain('"incomplete_details":{"reason":"max_output_tokens"}'); }); + test("MALFORMED_FUNCTION_CALL with NO emitted call part yields a terminal error, not done", async () => { + // The malformed call is dropped upstream, so the final chunk usually carries only the + // finishReason. Without the guard this surfaced as a clean empty completion. + const events = await collect(vertexProvider, [ + { candidates: [{ finishReason: "MALFORMED_FUNCTION_CALL" }], usageMetadata: { promptTokenCount: 5, candidatesTokenCount: 0 } }, + ]); + const last = events[events.length - 1]; + expect(last.type).toBe("error"); + expect(events.some(e => e.type === "done")).toBe(false); + }); + test("usage-only final chunk (no candidates) is not dropped", async () => { const events = await collect(vertexProvider, [ { candidates: [{ content: { parts: [{ text: "hi" }] } }] }, @@ -98,6 +117,14 @@ describe("vertex parseResponse fail-closed truncation (non-streaming)", () => { expect(events.some(e => e.type === "done")).toBe(false); }); + test("MALFORMED_FUNCTION_CALL with no call part yields a terminal error, not done", async () => { + const adapter = createGoogleAdapter(vertexProvider); + const body = JSON.stringify({ candidates: [{ finishReason: "MALFORMED_FUNCTION_CALL" }], usageMetadata: { promptTokenCount: 5, candidatesTokenCount: 0 } }); + const events = await adapter.parseResponse!(new Response(body, { status: 200 })); + expect(events[events.length - 1].type).toBe("error"); + expect(events.some(e => e.type === "done")).toBe(false); + }); + test("clean STOP non-stream response yields done", async () => { const adapter = createGoogleAdapter(vertexProvider); const body = JSON.stringify({ candidates: [{ content: { parts: [{ text: "ok" }] }, finishReason: "STOP" }], usageMetadata: { promptTokenCount: 2, candidatesTokenCount: 1 } }); From 528cfc12d2a045743be261661ccc5f1f33508ecf Mon Sep 17 00:00:00 2001 From: devmello Date: Mon, 3 Aug 2026 03:31:51 -0700 Subject: [PATCH 3/8] fix(claude): separate reasoning summary parts in streamed thinking The streaming translator folded every summary part into one thinking block with no separator, so multi-part summaries rendered as run-on text. The JSON path already joins parts with a blank line; the stream now emits the same separator at part boundaries. (cherry picked from commit b23cc1f8f5c0ef3f367ca73d7668709f1989c0e7) --- src/claude/outbound.ts | 17 ++++++++++++ tests/claude-outbound.test.ts | 52 +++++++++++++++++++++++++++++++++++ 2 files changed, 69 insertions(+) diff --git a/src/claude/outbound.ts b/src/claude/outbound.ts index 0a9c81b2a7..e025762f85 100644 --- a/src/claude/outbound.ts +++ b/src/claude/outbound.ts @@ -187,6 +187,8 @@ interface OpenBlock { argsBufBytes?: number; webSearchArgsEmitted?: boolean; callId?: string; + /** Last reasoning part identity (item + summary/content index) seen by this thinking block. */ + reasoningPartKey?: string; } /** Streaming: Responses SSE bytes -> Anthropic Messages SSE bytes. */ @@ -353,6 +355,21 @@ export function responsesSseToAnthropicSse( case "response.reasoning_text.delta": { if (typeof data.delta !== "string" || data.delta.length === 0) break; ensureBlock("thinking"); + // The JSON path joins reasoning summary/content parts with "\n\n" + // (responsesJsonToAnthropicMessage); mirror that at part and item boundaries + // so multi-part summaries do not glue into one run-on paragraph. Frames + // without part indices produce a constant key and never get a separator. + const slot = eventName === "response.reasoning_summary_text.delta" + ? `s${String(data.summary_index)}` + : `c${String(data.content_index)}`; + const partKey = `${String(data.item_id)}:${slot}`; + if (open!.reasoningPartKey !== undefined && open!.reasoningPartKey !== partKey) { + emit("content_block_delta", { + type: "content_block_delta", index: open!.index, + delta: { type: "thinking_delta", thinking: "\n\n" }, + }); + } + open!.reasoningPartKey = partKey; emit("content_block_delta", { type: "content_block_delta", index: open!.index, delta: { type: "thinking_delta", thinking: data.delta }, diff --git a/tests/claude-outbound.test.ts b/tests/claude-outbound.test.ts index ab997823be..8651f779d1 100644 --- a/tests/claude-outbound.test.ts +++ b/tests/claude-outbound.test.ts @@ -207,6 +207,58 @@ describe("claude outbound SSE", () => { expect(startIndexes).toEqual([0, 1, 2]); }); + test("multi-part reasoning summaries keep the JSON path's part separator", async () => { + const upstream = [ + sse("response.created", { response: { id: "resp_1", status: "in_progress" } }), + sse("response.output_item.added", { output_index: 0, item: { type: "reasoning", id: "rs_1" } }), + sse("response.reasoning_summary_part.added", { item_id: "rs_1", output_index: 0, summary_index: 0, part: { type: "summary_text", text: "" } }), + sse("response.reasoning_summary_text.delta", { item_id: "rs_1", output_index: 0, summary_index: 0, delta: "**A**\n\nOne." }), + sse("response.reasoning_summary_part.added", { item_id: "rs_1", output_index: 0, summary_index: 1, part: { type: "summary_text", text: "" } }), + sse("response.reasoning_summary_text.delta", { item_id: "rs_1", output_index: 0, summary_index: 1, delta: "**B**\n\nTwo." }), + sse("response.output_item.done", { output_index: 0, item: { type: "reasoning", id: "rs_1" } }), + sse("response.output_item.added", { output_index: 1, item: { type: "reasoning", id: "rs_2" } }), + sse("response.reasoning_summary_text.delta", { item_id: "rs_2", output_index: 1, summary_index: 0, delta: "Three." }), + sse("response.completed", { response: { status: "completed", usage: { input_tokens: 1, output_tokens: 1 } } }), + ].join(""); + const msg = await collectAnthropicMessage(responsesSseToAnthropicSse(streamFrom(upstream), "m"), "m") as Record; + // Parts within an item are separated; a new reasoning item opens its own block. + const thinkingBlocks = msg.content.filter((b: Record) => b.type === "thinking"); + expect(thinkingBlocks.map((b: Record) => b.thinking)).toEqual([ + "**A**\n\nOne.\n\n**B**\n\nTwo.", + "Three.", + ]); + + // Parity: the non-streaming translator joins the same summary parts identically. + const json = responsesJsonToAnthropicMessage({ + id: "resp_1", + status: "completed", + output: [{ type: "reasoning", id: "rs_1", summary: [{ type: "summary_text", text: "**A**\n\nOne." }, { type: "summary_text", text: "**B**\n\nTwo." }] }], + usage: { input_tokens: 1, output_tokens: 1 }, + }, "m") as Record; + const jsonThinking = json.content.find((b: Record) => b.type === "thinking"); + expect(jsonThinking.thinking).toBe("**A**\n\nOne.\n\n**B**\n\nTwo."); + }); + + test("same-part deltas and index-free reasoning frames never get a separator", async () => { + const samePart = [ + sse("response.created", { response: { id: "resp_1", status: "in_progress" } }), + sse("response.reasoning_summary_text.delta", { item_id: "rs_1", output_index: 0, summary_index: 0, delta: "Hel" }), + sse("response.reasoning_summary_text.delta", { item_id: "rs_1", output_index: 0, summary_index: 0, delta: "lo" }), + sse("response.completed", { response: { status: "completed", usage: { input_tokens: 1, output_tokens: 1 } } }), + ].join(""); + const msg1 = await collectAnthropicMessage(responsesSseToAnthropicSse(streamFrom(samePart), "m"), "m") as Record; + expect(msg1.content.find((b: Record) => b.type === "thinking").thinking).toBe("Hello"); + + const indexFree = [ + sse("response.created", { response: { id: "resp_1", status: "in_progress" } }), + sse("response.reasoning_text.delta", { delta: "A" }), + sse("response.reasoning_text.delta", { delta: "B" }), + sse("response.completed", { response: { status: "completed", usage: { input_tokens: 1, output_tokens: 1 } } }), + ].join(""); + const msg2 = await collectAnthropicMessage(responsesSseToAnthropicSse(streamFrom(indexFree), "m"), "m") as Record; + expect(msg2.content.find((b: Record) => b.type === "thinking").thinking).toBe("AB"); + }); + test("data-only Responses frames infer event names from payload types", async () => { const upstream = [ dataOnlySse({ type: "response.created", response: { id: "resp_data_only", status: "in_progress" } }), From 0bb2820621b670739d349ede0a8d9fba0c76bfd3 Mon Sep 17 00:00:00 2001 From: devmello Date: Mon, 3 Aug 2026 04:02:45 -0700 Subject: [PATCH 4/8] fix(compact): report upstream usage for native compact turns The native branch buffers the upstream compact JSON and returns it without inspecting the body, so the request log row lands with no usage. Lift usage and response metadata from the buffered body the same way the routed branch gets it through handleResponses. (cherry picked from commit 2ae3b5fc8de4942b6056eac1fe489845de107404) --- src/server/responses/compact.ts | 4 ++++ tests/responses-compaction-routing.test.ts | 28 ++++++++++++++++++++++ 2 files changed, 32 insertions(+) diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 505c5c3979..87dc4fe8f3 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -491,6 +491,10 @@ export async function handleResponsesCompact( // Always record the real upstream status: a local buffering failure after a // 200 upstream response must not soft-avoid a healthy account or rotate a thread. recordCompactPoolOutcome(outcomeCtx, upstream.status, { retryAfter, resetAt }); + // Lift usage and response metadata from the buffered upstream JSON into the + // request log; the routed branch gets the same through handleResponses. The + // synthetic buffer errors are not upstream bodies and stay uninspected. + if (buffered.ok) inspectResponseLogJson(logCtx, await buffered.clone().text()); return buffered; } diff --git a/tests/responses-compaction-routing.test.ts b/tests/responses-compaction-routing.test.ts index 346aed2d51..b1d3505103 100644 --- a/tests/responses-compaction-routing.test.ts +++ b/tests/responses-compaction-routing.test.ts @@ -23,6 +23,7 @@ import { resolveCodexAuthContext, } from "../src/codex/auth-context"; import { supportsNativeResponsesCompactEndpoint } from "../src/providers/openai-tiers"; +import type { RequestLogContext } from "../src/server/request-log"; import type { OcxConfig, OcxProviderConfig } from "../src/types"; const originalFetch = globalThis.fetch; @@ -164,6 +165,33 @@ describe("supportsNativeResponsesCompactEndpoint (#422)", () => { }); }); +describe("native compact usage reporting", () => { + test("the buffered upstream body fills the request log usage and stays intact for the client", async () => { + const config = { + defaultProvider: "openai-apikey", + providers: { + "openai-apikey": { + adapter: "openai-responses", + baseUrl: "https://api.openai.com/v1", + authMode: "key", + apiKey: "sk-test", + }, + }, + } as unknown as OcxConfig; + globalThis.fetch = (async () => jsonResponse(completedPayload("native summary"))) as typeof fetch; + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponsesCompact( + compactionRequest(baseCompactionBody({ model: "openai-apikey/gpt-5.5" })), + config, + logCtx, + ); + expect(response.status).toBe(200); + const body = await response.json() as { usage?: Record }; + expect(body.usage).toMatchObject({ input_tokens: 10, output_tokens: 5, total_tokens: 15 }); + expect(logCtx.usage).toMatchObject({ inputTokens: 10, outputTokens: 5, totalTokens: 15 }); + }); +}); + describe("native Codex pool compaction", () => { test("keeps a Spark reset cooldown separate from a Terra compact request (#590)", async () => { const testDir = mkdtempSync(join(tmpdir(), "ocx-compact-scope-")); From 02ca79a37cf005be1a4e9a1b1f6e6126948b7509 Mon Sep 17 00:00:00 2001 From: WangHongxiao Date: Mon, 3 Aug 2026 17:20:35 +0800 Subject: [PATCH 5/8] fix(responses): close passthrough streams at terminal events (cherry picked from commit 88d60c3a0ae048b849a29ec2a9930e216f97aa27) --- src/lib/bun-stream-caps.ts | 3 +- src/server/index.ts | 9 +- src/server/relay-eager.ts | 56 ++++++++--- src/server/relay.ts | 126 ++++++++++++++++++++++-- src/server/responses/core.ts | 21 ++-- structure/04_transports-and-sidecars.md | 20 ++-- tests/passthrough-abort.test.ts | 9 +- tests/relay-eager.test.ts | 85 ++++++++++++++-- tests/sse-failed-tail.test.ts | 55 +++++++++++ 9 files changed, 325 insertions(+), 59 deletions(-) diff --git a/src/lib/bun-stream-caps.ts b/src/lib/bun-stream-caps.ts index 188f10b51c..3b5265e706 100644 --- a/src/lib/bun-stream-caps.ts +++ b/src/lib/bun-stream-caps.ts @@ -6,7 +6,8 @@ * PR #32120, merged 2026-06-21). No RELEASED Bun version is proven to carry * that fix yet, so `MIN_FIXED_BUN_VERSION` is null: every runtime is * "known-bad" until a bundle-bump commit sets it. Windows no-rewrite traffic - * follows this runtime/config decision. Darwin no-rewrite traffic stays on tee + * follows this runtime/config decision, preserving the explicit legacy-tee + * safety pin. Darwin no-rewrite traffic stays on tee * for `auto` regardless of runtime capability and reaches eager relay only via * explicit `streamMode: "eager-relay"` opt-in (see * devlog/_plan/260731_macos_rss_retention/100_darwin_eager_optin.md). diff --git a/src/server/index.ts b/src/server/index.ts index dc56c01eab..1ef97c64ae 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -251,18 +251,15 @@ function attachLiveSidebandUpstream(ws: ServerWebSocket): void { // if (isEventStream && upstreamResponse.body) { // const repairConfig = route.provider.responsesItemIdRepair; // const needsClientRewrite = imageGenCallAliases.size > 0 -// #314 gated shape: win32 no-rewrite traffic follows runtime/config policy; darwin no-rewrite -// traffic requires explicit config-eager opt-in (`auto` always stays tee on darwin). Default OFF -// on the bundled known-bad runtime; policy lives in 260731_macos_rss_retention phase 100. +// #314 gated shape: win32 always uses the terminal-aware eager relay so a keep-alive +// upstream cannot hold Codex open after response.completed; darwin no-rewrite traffic +// requires explicit config-eager opt-in (`auto` always stays tee on darwin). // selectEagerPath(process.platform, needsClientRewrite, config.streamMode ?? "auto") // relaySseEagerBounded(upstreamResponse.body, turnAc, // new Response(eagerBody, // Default shape (tee + background inspection): // upstreamResponse.body.tee() // const repairedBody = hasResponsesItemIdRepair(repairConfig) -// process.platform === "win32" -// && !needsClientRewrite -// ? nativeBody // relaySseWithFailedTail(repairedBody, upstream) // new Response(clientBody // markNativePassthroughSseResponse diff --git a/src/server/relay-eager.ts b/src/server/relay-eager.ts index ee8ac88e7e..5c1241f22d 100644 --- a/src/server/relay-eager.ts +++ b/src/server/relay-eager.ts @@ -24,7 +24,7 @@ * up to the drain window. */ -import { buildFailedTailPayload } from "./relay"; +import { buildFailedTailPayload, createSseTerminalOutputBoundary } from "./relay"; import { nextSseBlock, replaceSseDataPayload, @@ -92,6 +92,7 @@ export function relaySseEagerBounded( const now = opts?.now ?? Date.now; const reader = body.getReader(); + const terminalBoundary = createSseTerminalOutputBoundary(); const rewrite = hooks.rewritePayload; const rewriteDecoder = rewrite ? new TextDecoder() : null; const rewriteEncoder = rewrite ? new TextEncoder() : null; @@ -161,6 +162,7 @@ export function relaySseEagerBounded( let queuedBytes = 0; let cancelled = false; let done = false; + const terminalSentinel = new TextEncoder().encode("data: [DONE]\n\n"); // Pause gate: resolved by client pull, client cancel, or upstream abort so a // paused producer ALWAYS resumes (audit blocker 2 — no deadlock; onDone and // turn unregistration stay reachable, drainAndShutdown never hangs). @@ -216,12 +218,17 @@ export function relaySseEagerBounded( if (upstream.signal.aborted) break; if (upstreamDone) { hooks.finishInspection(); + const boundedTail = terminalBoundary.finish(); if (rewrite) { - const tail = flushRewriteTail(); + const rewritten = rewriteOutbound(boundedTail); + const tail = joinUint8Arrays(rewritten, flushRewriteTail()); if (tail.byteLength > 0 && !cancelled) { queuedBytes += tail.byteLength; try { controllerRef?.enqueue(tail); } catch { /* client already gone */ } } + } else if (boundedTail.byteLength > 0 && !cancelled) { + queuedBytes += boundedTail.byteLength; + try { controllerRef?.enqueue(boundedTail); } catch { /* client already gone */ } } if (!hooks.sawTerminal() && !cancelled && !upstream.signal.aborted) { syntheticKind = "incomplete"; @@ -237,17 +244,30 @@ export function relaySseEagerBounded( } continue; } - const outbound = rewrite ? rewriteOutbound(value) : value; - if (outbound.byteLength === 0) continue; - queuedBytes += outbound.byteLength; - try { - controllerRef?.enqueue(outbound); - } catch { - // Controller already torn down (client went away without cancel()). - cancelled = true; - drainDeadline = now() + drainMs; - armDrainTimer(); - continue; + const terminalBounded = terminalBoundary.feed(value); + const outbound = rewrite ? rewriteOutbound(terminalBounded) : terminalBounded; + if (outbound.byteLength > 0) { + queuedBytes += outbound.byteLength; + try { + controllerRef?.enqueue(outbound); + } catch { + // Controller already torn down (client went away without cancel()). + cancelled = true; + drainDeadline = now() + drainMs; + armDrainTimer(); + continue; + } + } + if (terminalBoundary.terminalSeen()) { + // The Responses terminal event ends the turn even when a compatible + // gateway keeps its HTTP connection alive. Add the conventional + // sentinel and stop the single-reader relay at that protocol boundary. + if (!terminalBoundary.doneSeen()) { + queuedBytes += terminalSentinel.byteLength; + try { controllerRef?.enqueue(terminalSentinel); } catch { /* client already gone */ } + } + reader.cancel("Responses terminal event received").catch(() => {}); + break; } while (queuedBytes > maxQueueBytes && !cancelled && !upstream.signal.aborted) { await paused(); @@ -279,6 +299,7 @@ export function relaySseEagerBounded( try { rewriteBudget.releaseRetained(frameBufferBytes, { kind: "live_transient" }); } catch { /* teardown must not throw */ } frameBufferBytes = 0; } + terminalBoundary.dispose(); if (syntheticKind) hooks.onSynthetic(syntheticKind); if (cancelled && !hooks.sawTerminal()) { hooks.onClientCancel(); @@ -318,3 +339,12 @@ export function relaySseEagerBounded( }, }); } + +function joinUint8Arrays(first: Uint8Array, second: Uint8Array): Uint8Array { + if (first.byteLength === 0) return second; + if (second.byteLength === 0) return first; + const joined = new Uint8Array(first.byteLength + second.byteLength); + joined.set(first); + joined.set(second, first.byteLength); + return joined; +} diff --git a/src/server/relay.ts b/src/server/relay.ts index 06d1fec18b..3b5aae2366 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -95,6 +95,79 @@ export function buildFailedTailPayload(err: unknown): string { }); } +export type SseTerminalOutputBoundary = { + feed(chunk: Uint8Array): Uint8Array; + finish(): Uint8Array; + terminalSeen(): boolean; + doneSeen(): boolean; + dispose(): void; +}; + +/** + * Frame-aware client output boundary shared by both native Responses relays. + * It buffers only the current incomplete SSE block, forwards complete blocks + * through the first Responses terminal, and drops every later block/byte. + */ +export function createSseTerminalOutputBoundary(): SseTerminalOutputBoundary { + let decoder: TextDecoder | null = new TextDecoder(); + const encoder = new TextEncoder(); + let buffer = ""; + let terminal = false; + let done = false; + let disposed = false; + + const process = (flush: boolean): Uint8Array => { + if (disposed || terminal) return new Uint8Array(0); + let output = ""; + let responsesTerminal = false; + for (;;) { + const next = nextSseBlock(buffer); + if (!next) break; + buffer = next.rest; + const payload = sseDataPayload(next.block); + if (!responsesTerminal) output += next.block + next.delimiter; + if (payload === "[DONE]") { + done = true; + if (responsesTerminal) output += next.block + next.delimiter; + continue; + } + if (!responsesTerminal && payload && terminalStatusFromSsePayload(payload)) { + responsesTerminal = true; + } + } + if (responsesTerminal) { + terminal = true; + buffer = ""; + } + if (flush && !terminal && buffer.length > 0) { + output += buffer; + buffer = ""; + } + return encoder.encode(output); + }; + + return { + feed(chunk) { + if (disposed || terminal) return new Uint8Array(0); + buffer += decoder!.decode(chunk, { stream: true }); + return process(false); + }, + finish() { + if (disposed || terminal) return new Uint8Array(0); + buffer += decoder!.decode(); + return process(true); + }, + terminalSeen: () => terminal, + doneSeen: () => done, + dispose() { + if (disposed) return; + disposed = true; + decoder = null; + buffer = ""; + }, + }; +} + /** * Relay a passthrough SSE body like relayWithAbort, but convert a MID-STREAM failure (upstream * reset after headers) into a clean terminal: any partial block is closed off, then a synthetic @@ -110,18 +183,57 @@ export function relaySseWithFailedTail( ): ReadableStream { const reader = body.getReader(); const encoder = new TextEncoder(); + const terminalBoundary = createSseTerminalOutputBoundary(); + let closed = false; + const relayChunk = ( + controller: ReadableStreamDefaultController, + value: Uint8Array, + ): "terminal" | "output" | "buffered" => { + const outbound = terminalBoundary.feed(value); + if (outbound.byteLength > 0) controller.enqueue(outbound); + if (!terminalBoundary.terminalSeen()) return outbound.byteLength > 0 ? "output" : "buffered"; + + // A Responses terminal frame is the protocol boundary. Some compatible + // gateways leave the HTTP connection open after response.completed, which + // otherwise leaves Codex waiting forever even though the model turn is done. + // Preserve through the terminal block only, add the conventional sentinel + // when there was no real [DONE] data event, then stop reading upstream. + if (!terminalBoundary.doneSeen()) { + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + } + closed = true; + controller.close(); + const reason = "Responses terminal event received"; + // Notify the tee inspection branch as well. It has already received the + // same terminal-bearing upstream chunk, so its bounded drain records the + // real terminal and then releases the turn/upstream keep-alive connection. + onClientGone?.(reason); + reader.cancel(reason).catch(() => {}); + terminalBoundary.dispose(); + return "terminal"; + }; return new ReadableStream({ async pull(controller) { try { - const { done, value } = await reader.read(); - if (done) { - controller.close(); - return; + for (;;) { + const { done, value } = await reader.read(); + if (done) { + const tail = terminalBoundary.finish(); + if (tail.byteLength > 0) controller.enqueue(tail); + terminalBoundary.dispose(); + controller.close(); + return; + } + const result = relayChunk(controller, value); + if (result !== "buffered") return; } - controller.enqueue(value); } catch (err) { + const partial = terminalBoundary.finish(); + terminalBoundary.dispose(); + if (closed) return; const payload = buildFailedTailPayload(err); try { + if (partial.byteLength > 0) controller.enqueue(partial); // Leading blank line terminates a partial SSE block so the failed frame parses cleanly. controller.enqueue(encoder.encode(`\n\nevent: response.failed\ndata: ${payload}\n\ndata: [DONE]\n\n`)); controller.close(); @@ -130,6 +242,7 @@ export function relaySseWithFailedTail( } }, cancel(reason) { + terminalBoundary.dispose(); if (onClientGone) onClientGone(reason); else upstream.abort(reason); reader.cancel(reason).catch(() => {}); @@ -137,11 +250,12 @@ export function relaySseWithFailedTail( }); } -export function nextSseBlock(buffer: string): { block: string; rest: string } | null { +export function nextSseBlock(buffer: string): { block: string; delimiter: string; rest: string } | null { const match = buffer.match(/\r?\n\r?\n/); if (!match || match.index === undefined) return null; return { block: buffer.slice(0, match.index), + delimiter: match[0], rest: buffer.slice(match.index + match[0].length), }; } diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 2271b04fd0..8975a65c4d 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1903,7 +1903,9 @@ async function handleResponsesInner( inspectChunk: chunk => inspector.feed(chunk), finishInspection: () => inspector.finish(), disposeInspection: () => inspector.dispose(), - sawTerminal: () => inspector.reported(), + // Stream lifetime follows the protocol terminal even when this request + // has no outcome callback configured (reported() would stay false). + sawTerminal: () => inspector.terminalSeen(), ...(win32EagerRewrite ? { rewritePayload: composeSsePayloadRewrites(...payloadRewrites) } : {}), @@ -1921,9 +1923,10 @@ async function handleResponsesInner( onClientCancel: () => options.onNativePassthroughCancel?.(), onDone: () => unregisterTurn(turnAc), }, win32EagerRewrite ? { rewriteBudget: translatorBudget } : undefined); - // selectEagerPath admits only no-rewrite traffic on both eligible platforms; - // win32 rewrite traffic reaches this relay too, but with the payload rewrite - // applied inline — never via an image/item-id JS pull wrapper (#32111, #864). + // When selected, this relay closes response.completed even if upstream + // keeps the connection alive. Windows rewrite traffic applies its + // payload transform inline — never via the Bun#32111-unsafe + // tee()+JS-pull chain (#864). if (!headers.has("content-type")) headers.set("content-type", "text/event-stream"); return markEagerRelaySseResponse( markNativePassthroughSseResponse(new Response(eagerBody, { @@ -1990,15 +1993,13 @@ async function handleResponsesInner( ); } if (!headers.has("content-type")) headers.set("content-type", "text/event-stream"); - // win32 must keep the pure native relay (Bun#32111 JS-sink segfault); elsewhere a JS pull - // relay is established practice (relayWithAbort, relaySseWithHeartbeat) and lets a - // mid-stream reset end with a clean response.failed terminal instead of a raw socket error. + // Windows was handled by the eager terminal-aware branch above. Remaining + // tee traffic can use the JS relay to close on a protocol terminal and to + // convert a mid-stream reset into a clean response.failed event. const rewrittenBody = payloadRewrites.length > 0 ? relaySseWithPayloadRewrite(nativeBody, composeSsePayloadRewrites(...payloadRewrites), translatorBudget) : nativeBody; - const clientBody = process.platform === "win32" && !needsClientRewrite - ? nativeBody - : relaySseWithFailedTail(rewrittenBody, upstream, reason => clientGone.abort(reason)); + const clientBody = relaySseWithFailedTail(rewrittenBody, upstream, reason => clientGone.abort(reason)); return markNativePassthroughSseResponse(new Response(clientBody, { status: upstreamResponse.status, headers, diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 78f2cc90f1..ca161b65bb 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -38,19 +38,19 @@ to GUI static serving. Native passthrough SSE has TWO shapes, selected per request in `src/server/responses/core.ts`: -- **Default: tee + background inspection.** `upstreamResponse.body.tee()` sends - branch[0] to the client (pure native relay on win32 without any client-facing - rewrite — the Bun#32111 crash workaround; a JS relay elsewhere) while branch[1] is +- **Default outside Windows: tee + background inspection.** `upstreamResponse.body.tee()` sends + branch[0] through a terminal-aware client relay while branch[1] is drained eagerly by `consumeForInspection`/`consumeForResponseLogMetadata` for terminal-outcome recording, quota, the passthrough continuation cache, and request logs. This remains the default shape on bundled Bun 1.3.14. -- **Gated: eager bounded relay** (`src/server/relay-eager.ts`). win32 and darwin - no-client-rewrite traffic only (neither image-gen aliases nor item-id repair), - selected by `selectEagerPath` in `src/lib/bun-stream-caps.ts`. Windows `auto` - becomes eager only on runtimes proven to carry the Bun#32111 fix - (`MIN_FIXED_BUN_VERSION`, null until a bundle bump), while explicit - `streamMode: "eager-relay"` opts in today. Darwin is explicit-only: `auto` - stays tee even after a future threshold bump. One eager reader + byte-bounded +- **Terminal-aware eager bounded relay** (`src/server/relay-eager.ts`). Windows + uses this single-reader shape for rewrite traffic and for no-rewrite traffic + selected by `selectEagerPath` in `src/lib/bun-stream-caps.ts`; the latter keeps + `legacy-tee` and known-bad-runtime `auto` on tee as documented. When selected, + `response.completed` closes the client stream even if upstream keeps HTTP/SSE + alive. Darwin uses it for no-client-rewrite traffic only (neither image-gen + aliases nor item-id repair) and is explicit-only: `auto` stays tee even after + a future threshold bump. One eager reader + byte-bounded client queue + post-cancel bounded discard-drain replaces the tee and goes directly to the response without a JS rewrite wrapper, preserving the full inspection side-effect set (shared `createSseInspector` factory in `relay.ts`) diff --git a/tests/passthrough-abort.test.ts b/tests/passthrough-abort.test.ts index 28592fb0d7..d702452cb7 100644 --- a/tests/passthrough-abort.test.ts +++ b/tests/passthrough-abort.test.ts @@ -47,15 +47,14 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => { ); expect(sseBranch).toContain("upstreamResponse.body.tee()"); - // win32 must receive the tee'd body untouched when no client rewrite is required — no JS pull - // wrapper on the default path (Bun#32111 segfault). + // Windows no-rewrite traffic must honor the stream-mode/runtime gate so + // legacy-tee remains a safety escape hatch for Bun#32111. expect(sseBranch).toContain("const repairConfig = route.provider.responsesItemIdRepair;"); expect(sseBranch).toContain("const needsClientRewrite = imageGenCallAliases.size > 0"); expect(sseBranch).toContain("new Response(eagerBody"); expect(sseBranch).toContain("const rewrittenBody = payloadRewrites.length > 0"); - expect(sseBranch).toContain('process.platform === "win32"'); - expect(sseBranch).toContain("&& !needsClientRewrite"); - expect(sseBranch).toContain("? nativeBody"); + expect(sseBranch).toContain("eagerPath?.useEagerRelay || win32EagerRewrite"); + expect(sseBranch).not.toContain("win32TerminalRelay"); // #864: win32 traffic that DOES need a client rewrite takes the eager single // reader with the payload rewrite applied inline — never the tee()+JS-pull // chain that loses the terminal block on Windows (Bun#32111). diff --git a/tests/relay-eager.test.ts b/tests/relay-eager.test.ts index a66b759d15..9840ead4b0 100644 --- a/tests/relay-eager.test.ts +++ b/tests/relay-eager.test.ts @@ -176,8 +176,10 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => { expect(text).toContain("RESTORED"); expect(text).not.toContain("image_gen__gen"); expect(text).toContain("response.completed"); - // A partial trailing block reaches the client verbatim at EOF. - expect(text).toContain("trailing-partial"); + // The protocol terminal ends the client stream; bytes produced after it + // belong to the gateway's retained connection and must not hold Codex open. + expect(text).not.toContain("trailing-partial"); + expect(text.endsWith("data: [DONE]\n\n")).toBe(true); }); test("identity rewrite preserves framing byte-for-byte", async () => { @@ -198,11 +200,37 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => { up.close(); const text = await reading; - expect(text).toBe(new TextDecoder().decode(joinBytes([first, enc.encode(second)]))); + expect(text).toBe( + new TextDecoder().decode(joinBytes([first, enc.encode(second)])) + "data: [DONE]\n\n", + ); // The rewrite actually ran — this is what makes the test red pre-fix. expect(rewriteCalls).toBeGreaterThan(0); }); + test("drops coalesced post-terminal frames and detects only a real DONE event", async () => { + for (const realDone of [false, true]) { + const up = controlledUpstream(); + const { hooks } = makeHooks(); + const relayed = relaySseEagerBounded(up.stream, new AbortController(), hooks); + const reading = readAll(relayed); + const completed = JSON.stringify({ + type: "response.completed", + response: { status: "completed", note: "data: [DONE]" }, + }); + up.push(enc.encode( + `event: response.completed\ndata: ${completed}\n\n` + + (realDone ? "data: [DONE]\n\n" : "") + + `data: {"type":"response.output_text.delta","delta":"must not leak"}\n\n`, + )); + up.close(); + + const text = await reading; + expect(text).not.toContain("must not leak"); + expect(countOccurrences(text, "\ndata: [DONE]\n\n")).toBe(1); + expect(text.endsWith("data: [DONE]\n\n")).toBe(true); + } + }); + test("unchanged multi-data-line events keep their original framing", async () => { const up = controlledUpstream(); const { hooks } = makeHooks(); @@ -238,7 +266,7 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => { expect(text).toBe("data: �"); }); - test("retained rewrite-budget bytes are released on upstream abort", async () => { + test("terminal framing keeps partial blocks out of the rewrite budget", async () => { const budget = createTranslatorBudget(); const up = controlledUpstream(); const ac = new AbortController(); @@ -248,13 +276,15 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => { up.push(enc.encode(`data: {"type":"unterminated"`)); await settle(); - expect(budget.snapshot().currentBytes).toBeGreaterThan(0); + // The shared terminal boundary now owns incomplete SSE framing, so the + // downstream rewrite stage never retains an unterminated block. + expect(budget.snapshot().currentBytes).toBe(0); ac.abort(new Error("test abort")); await settle(); expect(budget.snapshot().currentBytes).toBe(0); }); - test("blocks without a data field pass through untouched", async () => { + test("blocks without a data field pass through untouched before the terminal", async () => { const up = controlledUpstream(); const { hooks } = makeHooks(); let rewriteCalls = 0; @@ -265,8 +295,8 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => { const relayed = relaySseEagerBounded(up.stream, new AbortController(), hooks); const reading = readAll(relayed); - up.push(enc.encode(`event: response.completed\ndata: ${COMPLETED}\n\n`)); up.push(enc.encode(`: keepalive comment\n\n`)); + up.push(enc.encode(`event: response.completed\ndata: ${COMPLETED}\n\n`)); up.close(); const text = await reading; @@ -362,7 +392,7 @@ describe("relaySseEagerBounded — side-effect parity", () => { const clientBytes = await readAllBytes(relayed); await settle(); - expect(clientBytes).toEqual(joinBytes(frames)); + expect(clientBytes).toEqual(joinBytes([...frames, enc.encode("data: [DONE]\n\n")])); const wireText = new TextDecoder().decode(clientBytes); expect(wireText).not.toContain('"output":'); expect(rec.completed).toHaveLength(1); @@ -470,6 +500,45 @@ describe("relaySseEagerBounded — #44 cancel semantics", () => { expect(rec.dones).toBe(1); }); + test("post-cancel terminal ends metadata-only drain without waiting for timeout", async () => { + const inspector = createSseInspector({}); + const up = controlledUpstream(); + const rec = { cancels: 0, dones: 0, synthetics: [] as string[] }; + let resolveDone!: () => void; + const relayDone = new Promise(resolve => { resolveDone = resolve; }); + const relayed = relaySseEagerBounded(up.stream, new AbortController(), { + inspectChunk: chunk => inspector.feed(chunk), + finishInspection: () => inspector.finish(), + disposeInspection: () => inspector.dispose(), + // Mirrors the no-onTerminal wiring in responses/core.ts. + sawTerminal: () => inspector.terminalSeen(), + onSynthetic: kind => { rec.synthetics.push(kind); }, + onClientCancel: () => { rec.cancels += 1; }, + onDone: () => { rec.dones += 1; resolveDone(); }, + }, { postCancelDrainMs: 5_000 }); + const reader = relayed.getReader(); + up.push(sse(DELTA)); + await settle(5); + await reader.cancel(); + + // Keep upstream open after delivering the terminal. The protocol terminal, + // not EOF or the five-second drain timer, must finish the relay lifecycle. + up.push(sse(COMPLETED)); + await Promise.race([ + relayDone, + new Promise((_, reject) => setTimeout( + () => reject(new Error("metadata-only terminal drain waited for timeout")), + 200, + )), + ]); + + expect(inspector.reported()).toBe(false); + expect(inspector.terminalSeen()).toBe(true); + expect(rec.cancels).toBe(0); + expect(rec.synthetics).toEqual([]); + expect(rec.dones).toBe(1); + }); + test("(d) post-cancel drain timeout → onClientCancel fired, upstream aborted", async () => { const { hooks, rec } = makeHooks(); const up = controlledUpstream(); diff --git a/tests/sse-failed-tail.test.ts b/tests/sse-failed-tail.test.ts index 1faddcca11..70876dafdb 100644 --- a/tests/sse-failed-tail.test.ts +++ b/tests/sse-failed-tail.test.ts @@ -59,6 +59,61 @@ describe("relaySseWithFailedTail", () => { expect(upstream.signal.aborted).toBe(false); }); + test("closes at response.completed when the upstream keeps its SSE connection open", async () => { + const upstream = new AbortController(); + let sourceCancelled = false; + let sentTerminal = false; + const src = new ReadableStream({ + pull(controller) { + if (!sentTerminal) { + sentTerminal = true; + controller.enqueue(encoder.encode( + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}\n\n', + )); + } + // Deliberately never close: several Responses-compatible gateways keep + // this connection alive after the protocol terminal event. + }, + cancel() { sourceCancelled = true; }, + }); + + const out = await Promise.race([ + drain(relaySseWithFailedTail(src, upstream)), + new Promise((_, reject) => setTimeout(() => reject(new Error("relay did not close at terminal")), 200)), + ]); + + expect(out).toContain("response.completed"); + expect(out.endsWith("data: [DONE]\n\n")).toBe(true); + expect(sourceCancelled).toBe(true); + expect(upstream.signal.aborted).toBe(false); + }); + + test("drops frames coalesced after the terminal block", async () => { + const upstream = new AbortController(); + const src = sourceStream([ + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}\n\n' + + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"must not leak"}\n\n', + ]); + + const out = await drain(relaySseWithFailedTail(src, upstream)); + + expect(out).toContain("response.completed"); + expect(out).not.toContain("must not leak"); + expect(out.endsWith("data: [DONE]\n\n")).toBe(true); + }); + + test("recognizes only a real DONE data event", async () => { + const ordinaryText = 'data: {"type":"response.completed","response":{"status":"completed","note":"data: [DONE]"}}\n\n'; + const withRealDone = ordinaryText + "data: [DONE]\n\n"; + + const ordinaryOut = await drain(relaySseWithFailedTail(sourceStream([ordinaryText]), new AbortController())); + const realOut = await drain(relaySseWithFailedTail(sourceStream([withRealDone]), new AbortController())); + + expect(ordinaryOut.endsWith("data: [DONE]\n\n")).toBe(true); + expect(ordinaryOut.split("\ndata: [DONE]\n\n").length - 1).toBe(1); + expect(realOut).toBe(withRealDone); + }); + test("mid-stream error keeps prior bytes and appends a clean failed terminal", async () => { const upstream = new AbortController(); const src = sourceStream(['data: {"type":"response.output_text.delta","delta":"hel', ""], { failAfter: true }); From a4c78b270c9d42b3ec929ea306c24bd1a0d9292d Mon Sep 17 00:00:00 2001 From: Mushikingh <164845020+mushikingh@users.noreply.github.com> Date: Mon, 3 Aug 2026 16:33:26 +0200 Subject: [PATCH 6/8] fix(kiro): round-trip the redactedContent reasoning blob MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Kiro never returns plaintext reasoning for its Sol-family models. Its `reasoningContentEvent` carries a KMS-encrypted `redactedContent` blob, and `gpt-5.6-sol`'s `additionalModelRequestFieldsSchema` accepts only `reasoning.effort` — there is no display or summary opt-in. Kiro's own CLI replays that blob on the matching `assistantResponseMessage.reasoningContent` to preserve model reasoning across turns. The adapter read only `reasoningContentEvent.text`, which is absent on this wire, so the blob was dropped and never replayed. Every turn therefore restarted without the previous turn's reasoning. - kiro-events: parse `redactedContent`; add the previously unhandled `contextUsageEvent` (Kiro reports context pressure there, not in `metadataEvent`, which carries only `stopReason`). - Carry the blob through the existing `ocxr1:` envelope as `krc` on an envelope-only reasoning item, so it round-trips while staying invisible in the app — the same contract the hidden-thinking path already uses. - Pair it backwards: Kiro emits the event at the END of a turn, after content and tool calls, so a krc-only item belongs to the assistant turn that already closed. Folding it forward would attach turn N's blob to turn N+1. With no assistant turn to own it, the blob is dropped rather than mis-paired. - Replay it on `assistantResponseMessage.reasoningContent`. Verified against kiro-cli 2.14.1 and 2.16.0 request/response captures. (cherry picked from commit cbb5d21ec2d333f7cc1acd9ac78aca95ea997116) --- src/adapters/kiro-events.ts | 18 +++++++++++- src/adapters/kiro.ts | 23 +++++++++++---- src/bridge.ts | 24 +++++++++++++++ src/responses/parser.ts | 12 ++++++++ src/responses/reasoning-envelope.ts | 10 ++++++- src/types.ts | 9 ++++++ structure/04_transports-and-sidecars.md | 24 +++++++++++++++ tests/anthropic-thinking-signature.test.ts | 33 +++++++++++++++++++++ tests/kiro-adapter.test.ts | 26 +++++++++++++++++ tests/kiro-stream.test.ts | 34 ++++++++++++++++++++++ 10 files changed, 206 insertions(+), 7 deletions(-) diff --git a/src/adapters/kiro-events.ts b/src/adapters/kiro-events.ts index 0def314cc6..2e55369e65 100644 --- a/src/adapters/kiro-events.ts +++ b/src/adapters/kiro-events.ts @@ -3,7 +3,8 @@ import { kiroTruncationReason } from "./kiro-truncation"; export type ParsedKiroEvent = | { type: "content"; data?: string; modelId?: string } - | { type: "reasoning"; data?: string } + | { type: "reasoning"; data?: string; redactedContent?: string } + | { type: "context_usage"; contextUsagePercentage: number } | { type: "tool"; name?: string; toolUseId?: string; input?: string; stop?: boolean } | { type: "truncation"; data: string } | { type: "metadata"; usage?: OcxUsage; contextUsagePercentage?: number; stopReason?: string } @@ -17,6 +18,9 @@ const KNOWN_EVENT_TYPES = new Set([ "toolUseEvent", "messageMetadataEvent", "metadataEvent", + // Kiro reports context pressure in its OWN event type, not inside metadataEvent (verified + // against kiro-cli 2.14.1 and 2.16.0: metadataEvent carries only `stopReason`). + "contextUsageEvent", "invalidStateEvent", "error", ]); @@ -114,11 +118,16 @@ export function parseKiroEvent(eventType: string, payload: Uint8Array): ParsedKi : {}), }; case "reasoningContentEvent": + // `text` is plaintext reasoning; `redactedContent` is the encrypted blob the model actually + // returns for Sol-family models. Both may be absent on a bare event. return { type: "reasoning", ...(optionalString(eventType, parsed, "text") !== undefined ? { data: optionalString(eventType, parsed, "text") } : {}), + ...(optionalString(eventType, parsed, "redactedContent") !== undefined + ? { redactedContent: optionalString(eventType, parsed, "redactedContent") } + : {}), }; case "toolUseEvent": return { @@ -161,6 +170,13 @@ export function parseKiroEvent(eventType: string, payload: Uint8Array): ParsedKi ...(stopReason !== undefined ? { stopReason } : {}), }; } + case "contextUsageEvent": { + const contextUsagePercentage = parsed.contextUsagePercentage; + if (typeof contextUsagePercentage !== "number" || !Number.isFinite(contextUsagePercentage)) { + return malformed(eventType, "contextUsagePercentage must be a finite number"); + } + return { type: "context_usage", contextUsagePercentage }; + } case "invalidStateEvent": return { type: "invalid_state", message: optionalString(eventType, parsed, "message") }; case "error": diff --git a/src/adapters/kiro.ts b/src/adapters/kiro.ts index 3e2440045b..2ea1bc84df 100644 --- a/src/adapters/kiro.ts +++ b/src/adapters/kiro.ts @@ -99,7 +99,11 @@ interface KiroUserInputMessage { } interface KiroHistoryEntry { userInputMessage?: KiroUserInputMessage; - assistantResponseMessage?: { content: string; toolUses?: KiroToolUse[] }; + assistantResponseMessage?: { + content: string; + toolUses?: KiroToolUse[]; + reasoningContent?: { redactedContent: string }; + }; } function kiroToolWireNames(tools: readonly unknown[]): string[] { @@ -326,7 +330,7 @@ function validateKiroCapabilities(parsed: OcxParsedRequest): void { type KiroTurn = | { kind: "user"; content: string; images: KiroImage[]; toolResults: KiroToolResult[] } - | { kind: "assistant"; content: string; toolUses: KiroToolUse[] }; + | { kind: "assistant"; content: string; toolUses: KiroToolUse[]; redactedReasoning?: string }; function appendTurnText(target: string, next: string): string { if (!next) return target; @@ -471,13 +475,15 @@ export function buildKiroPayload( turns.push({ kind: "user", content, images: [...images], toolResults: [...toolResults] }); } }; - const pushAssistant = (content: string, toolUses: KiroToolUse[]): void => { + const pushAssistant = (content: string, toolUses: KiroToolUse[], redactedReasoning?: string): void => { const last = turns.at(-1); if (last?.kind === "assistant") { last.content = appendTurnText(last.content, content); last.toolUses.push(...toolUses); + // Merged turns keep the newest blob: it covers the reasoning up to the merged turn's end. + if (redactedReasoning) last.redactedReasoning = redactedReasoning; } else { - turns.push({ kind: "assistant", content, toolUses: [...toolUses] }); + turns.push({ kind: "assistant", content, toolUses: [...toolUses], ...(redactedReasoning ? { redactedReasoning } : {}) }); } }; @@ -507,7 +513,7 @@ export function buildKiroPayload( const hasReasoning = aMsg.content.some(part => part.type === "thinking" && part.thinking.trim()); if (hasReasoning) continue; } - pushAssistant(text, toolUses); + pushAssistant(text, toolUses, aMsg.kiroRedactedReasoning); } else if (msg.role === "toolResult") { const tr = msg as OcxToolResultMessage; if (tr.containsEncryptedContent) { @@ -561,6 +567,7 @@ export function buildKiroPayload( assistantResponseMessage: { content: turn.content, ...(turn.toolUses.length > 0 ? { toolUses: turn.toolUses } : {}), + ...(turn.redactedReasoning ? { reasoningContent: { redactedContent: turn.redactedReasoning } } : {}), }, } : { @@ -1183,6 +1190,12 @@ async function* parseKiroAttemptEvents( if (ev.data) { yield* emitRetained(stage({ type: "reasoning_raw_delta", text: ev.data })); } + if (ev.redactedContent) { + yield* emitRetained(stage({ type: "kiro_redacted_reasoning", data: ev.redactedContent })); + } + break; + case "context_usage": + if (ev.contextUsagePercentage > 0) contextUsagePercentage = ev.contextUsagePercentage; break; case "tool": { for (const contentEvent of thinking.flush()) { diff --git a/src/bridge.ts b/src/bridge.ts index 21fa851183..bb625f58bb 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -869,6 +869,21 @@ export function bridgeToResponsesSSE( pendingRedacted.push(event.data); break; } + case "kiro_redacted_reasoning": { + // Opaque and unrenderable: emit an envelope-only reasoning item (no summary, no text + // deltas) so it stays invisible in the app while still round-tripping. Kiro sends it + // once the turn's content and tool calls are done, so nothing open is disturbed. + const encrypted = encodeReasoningEnvelope({ krc: event.data }); + const reservation = budget?.reserveTransient(bytesOf(encrypted), { kind: "reasoning" }); + const itemId = `rs_${uuid()}`; + const item = { type: "reasoning", id: itemId, summary: [] as never[], encrypted_content: encrypted }; + emit("response.output_item.added", { output_index: outputIndex, item }); + emit("response.output_item.done", { output_index: outputIndex, item }); + reservation?.commitRetained(); + retainFinishedItem(item as OutputItem, bytesOf(encrypted), "reasoning"); + outputIndex++; + break; + } case "reasoning_raw_delta": { if (options?.hideThinkingSummary) { ({ value: hiddenRawReasoningText, bytes: hiddenRawReasoningBytes } = appendString( @@ -1528,6 +1543,15 @@ function buildResponseJSONWithBudget( } batchRedacted.push(e.data); break; + case "kiro_redacted_reasoning": + // Envelope-only reasoning item, same contract as the streaming path. + { + const encrypted = encodeReasoningEnvelope({ krc: e.data }); + pushOutput({ + type: "reasoning", id: `rs_${uuid()}`, summary: [], encrypted_content: encrypted, + }, bytesOf(encrypted), "reasoning"); + } + break; case "reasoning_raw_delta": if (currentText) flushText("commentary"); if (currentSummaryReasoning) flushSummaryReasoning(); diff --git a/src/responses/parser.ts b/src/responses/parser.ts index 339fd0002c..cdfb1ff337 100644 --- a/src/responses/parser.ts +++ b/src/responses/parser.ts @@ -416,6 +416,18 @@ export function parseRequest(body: unknown): OcxParsedRequest { : null; const thinkingText = envelope?.txt || text; + // Kiro reasoning round-trip: a krc-only item carries nothing renderable — it is provider + // state for the assistant turn that ALREADY closed, because Kiro emits its + // reasoningContentEvent at the END of a turn (after content AND tool calls, verified + // against kiro-cli 2.14.1/2.16.0). Folding it into the FOLLOWING turn like ordinary + // reasoning would attach turn N's blob to turn N+1, so attach it backwards instead. With + // no assistant turn to own it the blob is dropped rather than mis-paired. + if (envelope?.krc && thinkingText.length === 0) { + const previous = messages[messages.length - 1]; + if (previous?.role === "assistant") previous.kiroRedactedReasoning = envelope.krc; + continue; + } + // Native/non-ocxr1 encrypted-only reasoning is opaque here. Do not create a detached // assistant turn or invent replayable plaintext/signatures from the encrypted payload. if (thinkingText.length > 0) { diff --git a/src/responses/reasoning-envelope.ts b/src/responses/reasoning-envelope.ts index 18c6bc6394..1735f775fb 100644 --- a/src/responses/reasoning-envelope.ts +++ b/src/responses/reasoning-envelope.ts @@ -24,6 +24,12 @@ export interface ReasoningEnvelope { * so replay needs it even though the visible summary was suppressed. */ txt?: string; + /** + * Kiro `reasoningContentEvent.redactedContent`: a KMS-encrypted reasoning blob that is opaque to + * the proxy. Kiro's own CLI replays it on the matching `assistantResponseMessage` to preserve + * model reasoning across turns, so it round-trips here the same way a signature does. + */ + krc?: string; } export function encodeReasoningEnvelope(envelope: ReasoningEnvelope): string { @@ -45,7 +51,9 @@ export function decodeReasoningEnvelope(encryptedContent: string): ReasoningEnve } const txt = (parsed as { txt?: unknown }).txt; if (typeof txt === "string" && txt.length > 0) envelope.txt = txt; - return envelope.sig || envelope.red || envelope.txt ? envelope : null; + const krc = (parsed as { krc?: unknown }).krc; + if (typeof krc === "string" && krc.length > 0) envelope.krc = krc; + return envelope.sig || envelope.red || envelope.txt || envelope.krc ? envelope : null; } catch { return null; } diff --git a/src/types.ts b/src/types.ts index 0bbdc7e416..faceb99681 100644 --- a/src/types.ts +++ b/src/types.ts @@ -85,6 +85,12 @@ export interface OcxAssistantMessage { phase?: OcxMessagePhase; model?: string; timestamp: number; + /** + * Kiro `reasoningContent.redactedContent` for THIS assistant turn — an opaque encrypted blob + * Kiro replays to preserve model reasoning across turns. Provider-specific and unrenderable, so + * it rides the message rather than a content part: any other adapter simply ignores it. + */ + kiroRedactedReasoning?: string; } export interface OcxDeveloperMessage { @@ -254,6 +260,9 @@ export type AdapterEvent = // opaque redacted_thinking blocks. Both must be replayed verbatim or tool-use turns 400. | { type: "thinking_signature"; signature: string } | { type: "redacted_thinking"; data: string } + // Kiro reasoning round-trip: the encrypted `redactedContent` blob for the CURRENT assistant turn. + // Never rendered — it only rides the reasoning item's envelope so the next request can replay it. + | { type: "kiro_redacted_reasoning"; data: string } | { type: "reasoning_raw_delta"; text: string } | { type: "tool_call_start"; id: string; name: string } | { type: "tool_call_delta"; arguments: string } diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index ca161b65bb..1d12a3287e 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -399,6 +399,30 @@ Grounded in the open-sourced official client (xai-org/grok-build); unit + eviden `fetchWithHeaderTimeout` takes an executor so provider fetch wrappers stay inside the timeout race. +## Kiro reasoning round-trip (`redactedContent`) + +Kiro never returns plaintext reasoning for its Sol-family models: `reasoningContentEvent` carries a +KMS-encrypted `redactedContent` blob, and `gpt-5.6-sol`'s `additionalModelRequestFieldsSchema` +(`ListAvailableModels`) accepts only `reasoning.effort` — there is no display/summary opt-in. Kiro's +own CLI replays that blob on the matching `assistantResponseMessage.reasoningContent` to preserve +model reasoning across turns; dropping it makes every turn restart without the previous turn's +reasoning. Verified against kiro-cli 2.14.1 and 2.16.0. + +- The blob rides the existing `ocxr1:` envelope as `krc` (`src/responses/reasoning-envelope.ts`) on + an envelope-only reasoning item — `summary: []`, no text deltas — so it stays invisible in the + Codex app while round-tripping, exactly like the hidden-thinking path. +- **Pairing is backwards.** Kiro emits `reasoningContentEvent` at the END of an assistant turn, + after content AND tool calls. A `krc`-only item therefore belongs to the turn that already + closed, so the parser attaches it to the PRECEDING assistant message rather than folding it into + the following turn like ordinary reasoning (`src/responses/parser.ts`). With no assistant turn to + own it, the blob is dropped rather than mis-paired. +- The blob lives on `OcxAssistantMessage.kiroRedactedReasoning`, not on a thinking content part, so + no other adapter replays provider-private state if the conversation switches providers. + +Kiro also reports context pressure in its own `contextUsageEvent`; `metadataEvent` carries only +`stopReason`. Spend arrives in `meteringEvent` as **credits, not tokens** — there is no `tokenUsage` +on this wire at all, which is why Kiro usage stays estimated. + ## Parallel tool calls (default-on for chat providers) The openai-chat adapter buffers ALL streamed `tool_calls` deltas (keyed by `index`, falling back to diff --git a/tests/anthropic-thinking-signature.test.ts b/tests/anthropic-thinking-signature.test.ts index 14180680a4..e1227e33ab 100644 --- a/tests/anthropic-thinking-signature.test.ts +++ b/tests/anthropic-thinking-signature.test.ts @@ -166,6 +166,39 @@ describe("parser ocxr1 decode + anthropic replay", () => { expect(thinking?.redacted).toEqual(["RED1"]); }); + // Kiro emits its reasoningContentEvent at the END of an assistant turn (after content AND tool + // calls), so a krc-only envelope belongs to the turn BEFORE it. Folding it forward like ordinary + // reasoning would attach turn N's blob to turn N+1 and hand Kiro a mismatched blob. + test("krc-only reasoning attaches to the preceding assistant turn", async () => { + const parsed = parseRequest({ + model: "kiro/gpt-5.6-sol", + input: [ + { type: "message", role: "user", content: [{ type: "input_text", text: "hi" }] }, + { type: "message", role: "assistant", content: [{ type: "output_text", text: "first" }] }, + { type: "reasoning", id: "rs_1", summary: [], encrypted_content: encodeReasoningEnvelope({ krc: "BLOB1" }) }, + { type: "message", role: "user", content: [{ type: "input_text", text: "more" }] }, + { type: "message", role: "assistant", content: [{ type: "output_text", text: "second" }] }, + ], + }); + const assistants = parsed.context.messages.filter(m => m.role === "assistant"); + expect(assistants).toHaveLength(2); + expect((assistants[0] as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBe("BLOB1"); + expect((assistants[1] as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBeUndefined(); + }); + + test("krc-only reasoning with no preceding assistant turn is dropped, not mis-paired", async () => { + const parsed = parseRequest({ + model: "kiro/gpt-5.6-sol", + input: [ + { type: "reasoning", id: "rs_1", summary: [], encrypted_content: encodeReasoningEnvelope({ krc: "ORPHAN" }) }, + { type: "message", role: "user", content: [{ type: "input_text", text: "hi" }] }, + { type: "message", role: "assistant", content: [{ type: "output_text", text: "answer" }] }, + ], + }); + const assistant = parsed.context.messages.find(m => m.role === "assistant"); + expect((assistant as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBeUndefined(); + }); + test("hidden signed text (txt) is restored as the thinking body", async () => { const encrypted = encodeReasoningEnvelope({ sig: "RealSig1234567890==", txt: "the hidden signed text" }); const parsed = parseRequest({ diff --git a/tests/kiro-adapter.test.ts b/tests/kiro-adapter.test.ts index 2288567eb4..2cc49fb382 100644 --- a/tests/kiro-adapter.test.ts +++ b/tests/kiro-adapter.test.ts @@ -268,6 +268,32 @@ describe("kiro adapter — buildRequest", () => { expect(results[0].status).toBe("success"); }); + // Kiro's own client replays the encrypted reasoning blob on the assistant turn it belongs to; + // dropping it makes every turn start without the previous turn's reasoning. + test("assistant history replays the Kiro redacted reasoning blob", async () => { + const messages = [ + { role: "user", content: "think" }, + { role: "assistant", content: [{ type: "text", text: "answer" }], kiroRedactedReasoning: "LktUUn5+blob" }, + { role: "user", content: "again" }, + ]; + const { body } = await createKiroAdapter(provider).buildRequest(parsedWith(messages)); + const arm = JSON.parse(body).conversationState.history + .find((h: { assistantResponseMessage?: unknown }) => h.assistantResponseMessage)?.assistantResponseMessage; + expect(arm.reasoningContent).toEqual({ redactedContent: "LktUUn5+blob" }); + }); + + test("assistant history omits reasoningContent when no blob was captured", async () => { + const messages = [ + { role: "user", content: "think" }, + { role: "assistant", content: [{ type: "text", text: "answer" }] }, + { role: "user", content: "again" }, + ]; + const { body } = await createKiroAdapter(provider).buildRequest(parsedWith(messages)); + const arm = JSON.parse(body).conversationState.history + .find((h: { assistantResponseMessage?: unknown }) => h.assistantResponseMessage)?.assistantResponseMessage; + expect(arm).not.toHaveProperty("reasoningContent"); + }); + test("empty tool output is normalized to a non-empty Kiro result block", async () => { const messages = [ { role: "user", content: "run it" }, diff --git a/tests/kiro-stream.test.ts b/tests/kiro-stream.test.ts index 4de16927f2..b2a92bdf17 100644 --- a/tests/kiro-stream.test.ts +++ b/tests/kiro-stream.test.ts @@ -1306,6 +1306,40 @@ describe("kiro adapter — parseStream", () => { ]); }); + // Kiro's Sol-family models never return plaintext reasoning: reasoningContentEvent carries an + // encrypted `redactedContent` blob (verified against kiro-cli 2.14.1 and 2.16.0), which the + // official client replays on the matching assistantResponseMessage to preserve reasoning across + // turns. Reading only `text` dropped it entirely. + test("reasoningContentEvent redactedContent is captured for round-trip", async () => { + const events = await collectAdapterEvents(createKiroAdapter(provider).parseStream(new Response(streamOf( + eventFrame({ content: "visible answer" }), + eventFrame({ redactedContent: "LktUUn5+encrypted" }, "reasoningContentEvent"), + )))); + expect(events).toEqual([ + { type: "text_delta", text: "visible answer" }, + { type: "kiro_redacted_reasoning", data: "LktUUn5+encrypted" }, + expect.objectContaining({ type: "done", endTurn: true }), + ]); + }); + + test("reasoningContentEvent carrying both text and redactedContent emits both", async () => { + const events = await collectAdapterEvents(createKiroAdapter(provider).parseStream(new Response(streamOf( + eventFrame({ text: "plain", redactedContent: "blob" }, "reasoningContentEvent"), + )))); + expect(events).toEqual([ + { type: "reasoning_raw_delta", text: "plain" }, + { type: "kiro_redacted_reasoning", data: "blob" }, + expect.objectContaining({ type: "done" }), + ]); + }); + + // Kiro reports context pressure in its own event type; metadataEvent carries only stopReason, so + // reading contextUsagePercentage from metadataEvent alone never saw a value. + test("contextUsageEvent supplies the absolute context usage percentage", () => { + const parsed = parseKiroEvent("contextUsageEvent", enc.encode(JSON.stringify({ contextUsagePercentage: 42.5 }))); + expect(parsed).toEqual({ type: "context_usage", contextUsagePercentage: 42.5 }); + }); + test("thinking tags split across chunks are parsed as reasoning", async () => { const frames = [ eventFrame({ content: " Date: Mon, 3 Aug 2026 16:50:43 +0200 Subject: [PATCH 7/8] fix(kiro): defer the reasoning envelope until the turn's items close MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit caught a real defect that made the round-trip a no-op in the streaming path. Kiro sends its reasoning blob at the END of a turn, while the assistant message is still open. Emitting the envelope-only item on arrival was wrong twice over: - `outputIndex` only advances when an item CLOSES, so the open message and the envelope item were emitted under the same output index. - The envelope landed BEFORE the assistant message, and the parser pairs a krc-only item backwards, so it found no preceding assistant turn and dropped the blob as orphaned — silently defeating the fix. Both paths now stash the blob and flush it after every open item has closed: after the closes in the streaming `done` case, and after the trailing flushes in the batch path. Message phase inference is untouched, so a Kiro final answer is still classified `final_answer` rather than being force-closed as commentary. The batch path also released `bytesOf(encrypted)` through `pushOutput` without ever retaining it. It now charges the blob when stashing and lets `pushOutput` release that retained allocation, so the translator budget balances. Adds tests/kiro-reasoning-roundtrip.test.ts, which bridges adapter events and re-parses the emitted items the way Codex replays history — the end-to-end coverage the original tests lacked. Three of its five cases fail against the previous commit. Scope: verified that gpt-5.6-terra and gpt-5.6-luna return `redactedContent` exactly like gpt-5.6-sol, so the whole GPT-5.6 family was affected. Handling keys off the wire field, not the model id. (cherry picked from commit bd13d4887e15f9f9a88afc155054231d199e66de) --- src/adapters/kiro-events.ts | 5 +- src/bridge.ts | 65 ++++++++---- structure/04_transports-and-sidecars.md | 20 ++-- tests/kiro-reasoning-roundtrip.test.ts | 130 ++++++++++++++++++++++++ 4 files changed, 195 insertions(+), 25 deletions(-) create mode 100644 tests/kiro-reasoning-roundtrip.test.ts diff --git a/src/adapters/kiro-events.ts b/src/adapters/kiro-events.ts index 2e55369e65..24e8f23370 100644 --- a/src/adapters/kiro-events.ts +++ b/src/adapters/kiro-events.ts @@ -118,8 +118,9 @@ export function parseKiroEvent(eventType: string, payload: Uint8Array): ParsedKi : {}), }; case "reasoningContentEvent": - // `text` is plaintext reasoning; `redactedContent` is the encrypted blob the model actually - // returns for Sol-family models. Both may be absent on a bare event. + // `text` is plaintext reasoning; `redactedContent` is the encrypted blob the GPT-5.6 family + // (sol/terra/luna) actually returns — they never send `text`. Keyed off the wire field, not + // the model id. Both may be absent on a bare event. return { type: "reasoning", ...(optionalString(eventType, parsed, "text") !== undefined diff --git a/src/bridge.ts b/src/bridge.ts index bb625f58bb..cc121e2e5d 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -442,6 +442,29 @@ export function bridgeToResponsesSSE( retainFinishedItem(item as OutputItem, bytesOf(encrypted), "reasoning"); outputIndex++; }; + // Kiro reasoning round-trip. Kiro sends its encrypted blob at the END of a turn, while the + // assistant message is still open, so this CANNOT emit on arrival: the open message still + // owns `outputIndex` (it only advances on close), and an item emitted here would both reuse + // that index and land BEFORE the message — where the parser's backwards pairing drops it as + // orphaned. Stash it and flush after `done` has closed every open item instead. + let pendingKiroRedacted: string | undefined; + let pendingKiroRedactedBytes = 0; + const flushKiroRedactedReasoning = () => { + if (!pendingKiroRedacted) return; + const previousBytes = pendingKiroRedactedBytes; + const encrypted = encodeReasoningEnvelope({ krc: pendingKiroRedacted }); + const reservation = budget?.reserveTransient(bytesOf(encrypted), { kind: "reasoning" }); + pendingKiroRedacted = undefined; + pendingKiroRedactedBytes = 0; + reservation?.commitRetained(); + budget?.releaseRetained(previousBytes, { kind: "reasoning" }); + const itemId = `rs_${uuid()}`; + const item = { type: "reasoning", id: itemId, summary: [] as never[], encrypted_content: encrypted }; + emit("response.output_item.added", { output_index: outputIndex, item }); + emit("response.output_item.done", { output_index: outputIndex, item }); + retainFinishedItem(item as OutputItem, bytesOf(encrypted), "reasoning"); + outputIndex++; + }; // Full assistant text of a compaction turn (across message boundaries) — becomes the // synthetic compaction item's payload on done. let compactionText = ""; @@ -870,18 +893,9 @@ export function bridgeToResponsesSSE( break; } case "kiro_redacted_reasoning": { - // Opaque and unrenderable: emit an envelope-only reasoning item (no summary, no text - // deltas) so it stays invisible in the app while still round-tripping. Kiro sends it - // once the turn's content and tool calls are done, so nothing open is disturbed. - const encrypted = encodeReasoningEnvelope({ krc: event.data }); - const reservation = budget?.reserveTransient(bytesOf(encrypted), { kind: "reasoning" }); - const itemId = `rs_${uuid()}`; - const item = { type: "reasoning", id: itemId, summary: [] as never[], encrypted_content: encrypted }; - emit("response.output_item.added", { output_index: outputIndex, item }); - emit("response.output_item.done", { output_index: outputIndex, item }); - reservation?.commitRetained(); - retainFinishedItem(item as OutputItem, bytesOf(encrypted), "reasoning"); - outputIndex++; + // Stash only — see flushKiroRedactedReasoning. One blob per turn, so last wins. + pendingKiroRedactedBytes = replaceRetainedString(pendingKiroRedactedBytes, event.data, "reasoning"); + pendingKiroRedacted = event.data; break; } case "reasoning_raw_delta": { @@ -1054,6 +1068,9 @@ export function bridgeToResponsesSSE( // Redacted-only turns (or hidden thinking without a trailing signature event) still // need their envelope-only reasoning item so the blocks replay next turn. flushHiddenReasoningEnvelope(); + // After every close above, so the blob lands AFTER the assistant message it belongs + // to and the parser's backwards pairing finds it. + flushKiroRedactedReasoning(); if (options?.compaction) { // Exactly one compaction item per turn; codex-rs takes the first and fatals on 0. const item = { @@ -1376,6 +1393,10 @@ function buildResponseJSONWithBudget( let batchSignatureBytes = 0; let batchRedacted: string[] = []; let batchRedactedBytes = 0; + // Kiro reasoning blob, held until after the trailing flushes so it lands AFTER the assistant + // message (see the streaming path). Retained because it outlives releaseTranslatedEvent. + let batchKiroRedacted: string | undefined; + let batchKiroRedactedBytes = 0; let currentToolCallId = ""; let currentToolCallName = ""; let currentToolCallArgs = ""; @@ -1544,13 +1565,14 @@ function buildResponseJSONWithBudget( batchRedacted.push(e.data); break; case "kiro_redacted_reasoning": - // Envelope-only reasoning item, same contract as the streaming path. + // Stash only — pushed after the trailing flushes. One blob per turn, so last wins. { - const encrypted = encodeReasoningEnvelope({ krc: e.data }); - pushOutput({ - type: "reasoning", id: `rs_${uuid()}`, summary: [], encrypted_content: encrypted, - }, bytesOf(encrypted), "reasoning"); + const dataBytes = bytesOf(e.data); + budget?.chargeRetained(dataBytes, { kind: "reasoning" }); + if (batchKiroRedactedBytes > 0) budget?.releaseRetained(batchKiroRedactedBytes, { kind: "reasoning" }); + batchKiroRedactedBytes = dataBytes; } + batchKiroRedacted = e.data; break; case "reasoning_raw_delta": if (currentText) flushText("commentary"); @@ -1650,6 +1672,15 @@ function buildResponseJSONWithBudget( flushRawReasoning(); // Open tool call on a failed/incomplete turn must not land as status:"completed". if (currentToolCallId) flushToolCall(errorEvent || incompleteEvent ? "incomplete" : "completed"); + if (batchKiroRedacted) { + // pushOutput reserves the item itself and releases the retained raw blob it replaces. + pushOutput({ + type: "reasoning", id: `rs_${uuid()}`, summary: [], + encrypted_content: encodeReasoningEnvelope({ krc: batchKiroRedacted }), + }, batchKiroRedactedBytes, "reasoning"); + batchKiroRedacted = undefined; + batchKiroRedactedBytes = 0; + } // A truncated turn must never be installed as replacement history: emit the // compaction item only when the turn actually completed (#422). if ( diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 1d12a3287e..2875f98c83 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -401,12 +401,20 @@ Grounded in the open-sourced official client (xai-org/grok-build); unit + eviden ## Kiro reasoning round-trip (`redactedContent`) -Kiro never returns plaintext reasoning for its Sol-family models: `reasoningContentEvent` carries a -KMS-encrypted `redactedContent` blob, and `gpt-5.6-sol`'s `additionalModelRequestFieldsSchema` -(`ListAvailableModels`) accepts only `reasoning.effort` — there is no display/summary opt-in. Kiro's -own CLI replays that blob on the matching `assistantResponseMessage.reasoningContent` to preserve -model reasoning across turns; dropping it makes every turn restart without the previous turn's -reasoning. Verified against kiro-cli 2.14.1 and 2.16.0. +Kiro never returns plaintext reasoning for its **GPT-5.6 family** (`gpt-5.6-sol`, `-terra`, +`-luna`): `reasoningContentEvent` carries a KMS-encrypted `redactedContent` blob, never `text`. +Their `additionalModelRequestFieldsSchema` (`ListAvailableModels`) accepts only `reasoning.effort` +with `additionalProperties: false` — there is no display/summary opt-in, so this is the only +reasoning these models can return. Kiro's own CLI replays the blob on the matching +`assistantResponseMessage.reasoningContent` to preserve model reasoning across turns; dropping it +makes every turn restart without the previous turn's reasoning. Verified on kiro-cli 2.14.1 and +2.16.0, all three models. + +The Claude 4.6+/5 entries advertise a different, richer contract (`thinking.type` adaptive/disabled, +`thinking.display` summarized/omitted, `output_config.effort`, `max_tokens`) and are not covered by +that measurement; older Claude, deepseek, minimax, glm, and qwen entries advertise no additional +fields at all. The handling below keys off the wire field, not the model id, so any model that +sends `redactedContent` round-trips. - The blob rides the existing `ocxr1:` envelope as `krc` (`src/responses/reasoning-envelope.ts`) on an envelope-only reasoning item — `summary: []`, no text deltas — so it stays invisible in the diff --git a/tests/kiro-reasoning-roundtrip.test.ts b/tests/kiro-reasoning-roundtrip.test.ts new file mode 100644 index 0000000000..5ce62cf050 --- /dev/null +++ b/tests/kiro-reasoning-roundtrip.test.ts @@ -0,0 +1,130 @@ +import { describe, expect, test } from "bun:test"; +import { bridgeToResponsesSSE, buildResponseJSON } from "../src/bridge"; +import { parseRequest } from "../src/responses/parser"; +import { decodeReasoningEnvelope } from "../src/responses/reasoning-envelope"; +import type { AdapterEvent } from "../src/types"; +import { createTranslatorBudget } from "../src/lib/translator-budget"; + +const BLOB = "LktUUn5+ZXlKbGJtTnllWEIwYVc5dVVtVm5hVzl1SWpvaQ=="; + +async function* replay(events: AdapterEvent[]): AsyncGenerator { + for (const event of events) yield event; +} + +async function drain(stream: ReadableStream): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let out = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + out += decoder.decode(value, { stream: true }); + } + return out; +} + +/** Output items in emission order, as Codex reconstructs them from the SSE stream. */ +function doneItems(sse: string): Record[] { + const items: Record[] = []; + for (const line of sse.split("\n")) { + if (!line.startsWith("data: ")) continue; + try { + const json = JSON.parse(line.slice(6)) as { type?: string; item?: Record; output_index?: number }; + if (json.type === "response.output_item.done" && json.item) { + items.push({ ...json.item, __index: json.output_index }); + } + } catch { /* partial frame */ } + } + return items; +} + +/** Feed emitted items back as Responses input, the way Codex replays history next turn. */ +function reparse(items: Record[]) { + return parseRequest({ + model: "kiro/gpt-5.6-sol", + input: [ + { type: "message", role: "user", content: [{ type: "input_text", text: "hi" }] }, + ...items.map(({ __index, status, ...item }) => item), + ], + }); +} + +// Kiro emits its reasoning blob at the END of a turn, while the assistant message is still open. +// Emitting the envelope item on arrival reused the open message's output_index AND placed the blob +// before the message, where the parser's backwards pairing drops it as orphaned — silently +// defeating the round-trip. Both paths must defer it until the message has closed. +describe("kiro redacted-reasoning round-trip (bridge → parse)", () => { + const events: AdapterEvent[] = [ + { type: "text_delta", text: "the answer" }, + { type: "kiro_redacted_reasoning", data: BLOB }, + { type: "done", usage: { inputTokens: 1, outputTokens: 2 }, endTurn: true }, + ]; + + test("SSE: the blob lands after the assistant message, on its own output index", async () => { + const items = doneItems(await drain(bridgeToResponsesSSE(replay(events), "kiro/gpt-5.6-sol"))); + const types = items.map(i => i.type); + expect(types).toEqual(["message", "reasoning"]); + + const indexes = items.map(i => i.__index); + expect(new Set(indexes).size).toBe(indexes.length); // no output_index collision + + const envelope = decodeReasoningEnvelope(items[1].encrypted_content as string); + expect(envelope?.krc).toBe(BLOB); + expect(items[1].summary).toEqual([]); + }); + + test("SSE: replayed history attaches the blob to the assistant turn that produced it", async () => { + const items = doneItems(await drain(bridgeToResponsesSSE(replay(events), "kiro/gpt-5.6-sol"))); + const assistant = reparse(items).context.messages.find(m => m.role === "assistant"); + expect((assistant as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBe(BLOB); + }); + + test("batch: the blob lands after the assistant message and survives replay", () => { + const response = buildResponseJSON( + [ + { type: "text_delta", text: "the answer" }, + { type: "kiro_redacted_reasoning", data: BLOB }, + { type: "done", usage: { inputTokens: 1, outputTokens: 2 }, endTurn: true }, + ], + "kiro/gpt-5.6-sol", + ); + const output = response.output as Record[]; + expect(output.map(i => i.type)).toEqual(["message", "reasoning"]); + + const assistant = reparse(output).context.messages.find(m => m.role === "assistant"); + expect((assistant as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBe(BLOB); + }); + + test("batch: the retained blob is released, leaving no translator-budget drift", () => { + const budget = createTranslatorBudget(); + buildResponseJSON( + [ + { type: "text_delta", text: "the answer" }, + { type: "kiro_redacted_reasoning", data: BLOB }, + { type: "done", usage: { inputTokens: 1, outputTokens: 2 }, endTurn: true }, + ], + "kiro/gpt-5.6-sol", + undefined, + undefined, + { budget }, + ); + // Only the finalized output items stay retained; the raw blob must not be double-counted or + // released without ever having been charged. + expect(budget.snapshot().currentBytes).toBeGreaterThanOrEqual(0); + expect(budget.snapshot().overflows).toBe(0); + }); + + test("a turn ending in a tool call still pairs the blob with that assistant turn", async () => { + const items = doneItems(await drain(bridgeToResponsesSSE(replay([ + { type: "tool_call_start", id: "call_1", name: "bash" }, + { type: "tool_call_delta", arguments: "{\"command\":\"ls\"}" }, + { type: "tool_call_end" }, + { type: "kiro_redacted_reasoning", data: BLOB }, + { type: "done", usage: { inputTokens: 1, outputTokens: 1 }, endTurn: false }, + ]), "kiro/gpt-5.6-sol"))); + expect(items.map(i => i.type)).toEqual(["function_call", "reasoning"]); + + const assistant = reparse(items).context.messages.find(m => m.role === "assistant"); + expect((assistant as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBe(BLOB); + }); +}); From 9664abdd187eca4456807042b3e531a8cf9dea27 Mon Sep 17 00:00:00 2001 From: Mushikingh <164845020+mushikingh@users.noreply.github.com> Date: Mon, 3 Aug 2026 17:03:32 +0200 Subject: [PATCH 8/8] test(kiro): make the budget regression actually observe the retained blob MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The batch budget assertion was vacuous twice over. It passed the budget in the wrong argument position, so `buildResponseJSON` built its own internal budget and the snapshot under test was never written to (`highWaterBytes: 0`). Even wired correctly, asserting `currentBytes >= 0` could not distinguish a leaked raw blob from the finalized items that legitimately stay retained. It now uses a 4 KB blob and asserts `currentBytes` equals EXACTLY the finalized output items' bytes — a still-retained raw blob shows up as ~4 KB of excess, and releasing bytes that were never charged shows up as a shortfall. A separate `highWaterBytes` assertion proves the blob was charged while held rather than never accounted for. Four of the file's five cases now fail against cbb5d21e. Also corrects the metadataEvent description: every capture put the context percentage in `contextUsageEvent`, but the parser still accepts a finite `contextUsagePercentage` (and `tokenUsage`) from `metadataEvent` as a fallback, so documenting it as impossible was wrong. (cherry picked from commit 8610630ecdc0069f1470b9dde1c9188cca743e59) --- src/adapters/kiro-events.ts | 5 +++-- structure/04_transports-and-sidecars.md | 13 +++++++++--- tests/kiro-reasoning-roundtrip.test.ts | 27 ++++++++++++++++--------- 3 files changed, 30 insertions(+), 15 deletions(-) diff --git a/src/adapters/kiro-events.ts b/src/adapters/kiro-events.ts index 24e8f23370..eab61ea511 100644 --- a/src/adapters/kiro-events.ts +++ b/src/adapters/kiro-events.ts @@ -18,8 +18,9 @@ const KNOWN_EVENT_TYPES = new Set([ "toolUseEvent", "messageMetadataEvent", "metadataEvent", - // Kiro reports context pressure in its OWN event type, not inside metadataEvent (verified - // against kiro-cli 2.14.1 and 2.16.0: metadataEvent carries only `stopReason`). + // Authoritative context pressure. Every capture (kiro-cli 2.14.1 and 2.16.0) put the percentage + // HERE and left `metadataEvent` carrying only `stopReason`; metadataEvent's own + // contextUsagePercentage stays supported as a fallback rather than being dropped. "contextUsageEvent", "invalidStateEvent", "error", diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 2875f98c83..00b36e03ef 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -427,9 +427,16 @@ sends `redactedContent` round-trips. - The blob lives on `OcxAssistantMessage.kiroRedactedReasoning`, not on a thinking content part, so no other adapter replays provider-private state if the conversation switches providers. -Kiro also reports context pressure in its own `contextUsageEvent`; `metadataEvent` carries only -`stopReason`. Spend arrives in `meteringEvent` as **credits, not tokens** — there is no `tokenUsage` -on this wire at all, which is why Kiro usage stays estimated. +Kiro reports context pressure in its own `contextUsageEvent`, which is the authoritative source. On +every capture taken (2.14.1 and 2.16.0) `metadataEvent` carried only `stopReason` — which is why +reading the percentage from `metadataEvent` alone never saw a value — but the parser still accepts a +finite `contextUsagePercentage` (and a `tokenUsage` block) there as a fallback, so a value parsed +from `metadataEvent` is legitimate rather than impossible. Both feed the same field, and any +positive value overwrites an earlier one. + +Spend arrives in `meteringEvent` as **credits, not tokens**. No captured response carried +`tokenUsage` on any event, which is why Kiro usage stays estimated; `meteringEvent` is currently +ignored because a credit is not a token count. ## Parallel tool calls (default-on for chat providers) diff --git a/tests/kiro-reasoning-roundtrip.test.ts b/tests/kiro-reasoning-roundtrip.test.ts index 5ce62cf050..a38121d59b 100644 --- a/tests/kiro-reasoning-roundtrip.test.ts +++ b/tests/kiro-reasoning-roundtrip.test.ts @@ -95,23 +95,30 @@ describe("kiro redacted-reasoning round-trip (bridge → parse)", () => { expect((assistant as { kiroRedactedReasoning?: string }).kiroRedactedReasoning).toBe(BLOB); }); - test("batch: the retained blob is released, leaving no translator-budget drift", () => { + test("batch: the raw blob is retained then released, leaving only the finalized items", () => { + // A big blob makes the accounting unambiguous: the raw string is an allocation distinct from + // the finalized item that embeds its base64. + const bigBlob = "X".repeat(4000); const budget = createTranslatorBudget(); - buildResponseJSON( + const response = buildResponseJSON( [ { type: "text_delta", text: "the answer" }, - { type: "kiro_redacted_reasoning", data: BLOB }, + { type: "kiro_redacted_reasoning", data: bigBlob }, { type: "done", usage: { inputTokens: 1, outputTokens: 2 }, endTurn: true }, ], "kiro/gpt-5.6-sol", - undefined, - undefined, - { budget }, + { translatorBudget: budget }, ); - // Only the finalized output items stay retained; the raw blob must not be double-counted or - // released without ever having been charged. - expect(budget.snapshot().currentBytes).toBeGreaterThanOrEqual(0); - expect(budget.snapshot().overflows).toBe(0); + const items = response.output as Record[]; + const finalizedBytes = items.reduce((sum, item) => sum + Buffer.byteLength(JSON.stringify(item)), 0); + const { currentBytes, highWaterBytes, overflows } = budget.snapshot(); + + // EXACTLY the finalized output items remain retained. A raw blob still held would show up as + // ~4000 extra bytes here; releasing bytes that were never charged would show up as a shortfall. + expect(currentBytes).toBe(finalizedBytes); + // ...and it really was charged while held, rather than never accounted for at all. + expect(highWaterBytes).toBeGreaterThanOrEqual(finalizedBytes + bigBlob.length); + expect(overflows).toBe(0); }); test("a turn ending in a tool call still pairs the blob with that assistant turn", async () => {