diff --git a/src/claude/outbound.ts b/src/claude/outbound.ts index ac06afac2d..0b76e80d87 100644 --- a/src/claude/outbound.ts +++ b/src/claude/outbound.ts @@ -215,8 +215,6 @@ interface OpenBlock { callId?: string; /** Last fixed-size reasoning identity (item + summary/content index) seen by this block. */ reasoningPartKey?: string; - thinkingBuf?: string; - thinkingBufBytes?: number; reasoningSig?: string; } @@ -260,12 +258,6 @@ export function responsesSseToAnthropicSse( const bytes = queuedLiveFrameBytes.shift(); if (bytes !== undefined) translatorBudget.releaseRetained(bytes, { kind: "live_transient" }); }; - const releaseThinkingBuffer = (block: OpenBlock | null | undefined) => { - if (block?.kind !== "thinking") return; - translatorBudget.releaseRetained(block.thinkingBufBytes ?? 0, { kind: "reasoning" }); - block.thinkingBufBytes = 0; - }; - return new ReadableStream({ start(controller) { const emit = (name: string, data: Rec) => { @@ -308,14 +300,15 @@ export function responsesSseToAnthropicSse( open.webSearchArgsEmitted = true; } if (open.kind === "thinking") { - const signature = open.reasoningSig ?? encodeReasoningEnvelope({ txt: open.thinkingBuf ?? "" }); + // Streamed thinking has already been delivered to the client. Keep the fallback + // signature bounded instead of retaining and re-encoding the entire untrusted block. + const signature = open.reasoningSig ?? encodeReasoningEnvelope({ txt: "" }); emit("content_block_delta", { type: "content_block_delta", index: open.index, delta: { type: "signature_delta", signature }, }); } emit("content_block_stop", { type: "content_block_stop", index: open.index }); - releaseThinkingBuffer(open); if (open.callId) translatorBudget.closeCall(open.callId); open = null; }; @@ -328,7 +321,7 @@ export function responsesSseToAnthropicSse( ? { type: "text", text: "" } : { type: "thinking", thinking: "", signature: "" }; emit("content_block_start", { type: "content_block_start", index, content_block: contentBlock }); - open = { kind, index, thinkingBuf: "", thinkingBufBytes: 0 }; + open = { kind, index }; }; const finish = (stopReason: string, usage: unknown) => { if (terminated) return; @@ -355,7 +348,6 @@ export function responsesSseToAnthropicSse( if (terminated && (code !== "translation_buffer_limit" || terminalDelivered)) return; terminated = true; if (code === "translation_buffer_limit") { - releaseThinkingBuffer(open); if (open?.callId) translatorBudget.closeCall(open.callId); open = null; terminalDelivered = true; @@ -421,21 +413,6 @@ export function responsesSseToAnthropicSse( const partKey = `${boundedReasoningIdentity(data.item_id)}:${slot}`; const needsPartSeparator = active.reasoningPartKey !== undefined && active.reasoningPartKey !== partKey; - const appended = `${needsPartSeparator ? "\n\n" : ""}${data.delta}`; - const previous = active.thinkingBuf ?? ""; - const previousBytes = active.thinkingBufBytes ?? 0; - const nextBytes = appendedUtf8Bytes(previous, previousBytes, appended); - const scope = { kind: "reasoning" } as const; - const reservation = translatorBudget.reserveTransient(nextBytes, scope); - try { - active.thinkingBuf = previous + appended; - active.thinkingBufBytes = nextBytes; - reservation.commitRetained(); - translatorBudget.releaseRetained(previousBytes, scope); - } catch (error) { - reservation.release(); - throw error; - } if (needsPartSeparator) { emit("content_block_delta", { type: "content_block_delta", index: active.index, @@ -762,7 +739,6 @@ export function responsesSseToAnthropicSse( fail(413, "upstream translation buffer exceeded the safe limit", false, "translation_buffer_limit"); } else fail(500, err instanceof Error ? err.message : String(err)); } finally { - releaseThinkingBuffer(open); translatorBudget.releaseRetained(bufferBytes, { kind: "live_transient" }); if (pingTimer !== undefined) clearInterval(pingTimer); reader.releaseLock(); @@ -776,7 +752,6 @@ export function responsesSseToAnthropicSse( cancel(reason) { cancelled = true; while (queuedLiveFrameBytes.length > 0) releaseDeliveredFrame(); - releaseThinkingBuffer(open); if (open?.callId) translatorBudget.closeCall(open.callId); if (pingTimer !== undefined) clearInterval(pingTimer); return reader?.cancel(reason); diff --git a/tests/claude-integration/claude-outbound.test.ts b/tests/claude-integration/claude-outbound.test.ts index 67380bb44a..c89ed367d3 100644 --- a/tests/claude-integration/claude-outbound.test.ts +++ b/tests/claude-integration/claude-outbound.test.ts @@ -275,8 +275,7 @@ describe("claude outbound SSE", () => { "**A**\n\nOne.\n\n**B**\n\nTwo.", "Three.", ]); - expect(decodeReasoningEnvelope(thinkingBlocks[0].signature)?.txt) - .toBe("**A**\n\nOne.\n\n**B**\n\nTwo."); + expect(decodeReasoningEnvelope(thinkingBlocks[0].signature)).toEqual({ txt: "" }); // Parity: the non-streaming translator joins the same summary parts identically. const json = responsesJsonToAnthropicMessage({ @@ -289,10 +288,9 @@ describe("claude outbound SSE", () => { expect(jsonThinking.thinking).toBe("**A**\n\nOne.\n\n**B**\n\nTwo."); }); - test("reasoning fallback buffering is bounded and releases its retained budget", async () => { + test("streamed reasoning does not accumulate retained fallback text", async () => { const budget = createTestTranslatorBudget({ maxTurnBytes: 8 * 1024 }); let reasoningCommitted = 0; - let reasoningReleased = 0; const trackedBudget: TranslatorBudget = { openCall: id => budget.openCall(id), closeCall: id => budget.closeCall(id), @@ -310,10 +308,7 @@ describe("claude outbound SSE", () => { budget.chargeRetained(bytes, scope); if (scope.kind === "reasoning") reasoningCommitted += bytes; }, - releaseRetained(bytes, scope) { - budget.releaseRetained(bytes, scope); - if (scope.kind === "reasoning") reasoningReleased += bytes; - }, + releaseRetained: (bytes, scope) => budget.releaseRetained(bytes, scope), observeAcceptedRequestCopy: bytes => budget.observeAcceptedRequestCopy(bytes), observeExternallyCapped: (kind, bytes) => budget.observeExternallyCapped(kind, bytes), snapshot: () => budget.snapshot(), @@ -326,115 +321,22 @@ describe("claude outbound SSE", () => { content_index: 0, delta: `${index}:` + "x".repeat(512), })), + sse("response.completed", { response: { status: "completed", usage: {} } }), ]; const events = await collectEvents(responsesSseToAnthropicSse( streamFromChunks(frames), "m", - { translatorBudget: trackedBudget }, + { translatorBudget: trackedBudget, pingIntervalMs: 0 }, )); - expect(events.at(-1)).toMatchObject({ - name: "error", - data: { error: { type: "request_too_large", code: "translation_buffer_limit" } }, - }); - expect(budget.snapshot().overflows).toBe(1); - expect(reasoningCommitted).toBeGreaterThan(0); - expect(reasoningReleased).toBe(reasoningCommitted); + expect(events.at(-1)?.name).toBe("message_stop"); + expect(budget.snapshot().overflows).toBe(0); + expect(reasoningCommitted).toBe(0); + const signature = events.find(event => event.data.delta?.type === "signature_delta")?.data.delta.signature; + expect(signature.length).toBeLessThan(256); + expect(decodeReasoningEnvelope(signature)).toEqual({ txt: "" }); }); - for (const terminal of ["eof", "failed", "completed", "incomplete"] as const) { - for (const buffered of [false, true]) { - test(`closure-only reasoning overflow: ${terminal}, ${buffered ? "collector" : "stream"}`, async () => { - // All small deltas fit, including replacement reservations. Closing needs - // the retained 32 KiB text PLUS its base64 signature frame. Capture the - // generated stream before collection: concurrent collector retention can - // exceed a shared budget during ingestion instead of exercising closure. - // Collection below reuses this SAME budget, without resetting it. - const budget = createTestTranslatorBudget({ maxTurnBytes: 70 * 1024 }); - let reasoningBytes = 0; - let maxReasoningBytes = 0; - let reasoningBytesAtOverflow = -1; - const trackedBudget: TranslatorBudget = { - openCall: id => budget.openCall(id), - closeCall: id => budget.closeCall(id), - reserveTransient(bytes, scope) { - let reservation: ReturnType; - try { reservation = budget.reserveTransient(bytes, scope); } - catch (error) { reasoningBytesAtOverflow = reasoningBytes; throw error; } - return { - commitRetained() { - reservation.commitRetained(); - if (scope.kind === "reasoning") { - reasoningBytes += bytes; - maxReasoningBytes = Math.max(maxReasoningBytes, reasoningBytes); - } - }, - release: () => reservation.release(), - }; - }, - chargeRetained: (bytes, scope) => budget.chargeRetained(bytes, scope), - releaseRetained(bytes, scope) { - if (scope.kind === "reasoning") reasoningBytes -= bytes; - budget.releaseRetained(bytes, scope); - }, - observeAcceptedRequestCopy: bytes => budget.observeAcceptedRequestCopy(bytes), - observeExternallyCapped: (kind, bytes) => budget.observeExternallyCapped(kind, bytes), - snapshot: () => budget.snapshot(), - dispose: () => budget.dispose(), - }; - const text = "x".repeat(32 * 1024); - const frames = Array.from({ length: 128 }, () => sse("response.reasoning_text.delta", { - item_id: "rs_closure", content_index: 0, delta: text.slice(0, 256), - })); - if (terminal !== "eof") { - frames.push(sse(`response.${terminal}`, { response: terminal === "failed" - ? { error: { message: "upstream failure", status: 502 } } - : terminal === "incomplete" - ? { status: "incomplete", incomplete_details: { reason: "max_output_tokens" }, usage: {} } - : { status: "completed", usage: {} } })); - // Neither a repeated completion nor a later failure may add a terminal. - frames.push(sse("response.completed", { response: { status: "completed", usage: {} } })); - frames.push(sse("response.failed", { response: { error: { message: "late failure" } } })); - } - const stream = responsesSseToAnthropicSse(streamFromChunks(frames), "m", { - translatorBudget: trackedBudget, pingIntervalMs: 0, - }); - const captured = buffered ? await new Response(stream).text() : undefined; - const capturedFrames = captured?.split("\n\n").filter(Boolean).map(frame => `${frame}\n\n`); - const events = await collectEvents(capturedFrames ? streamFromChunks(capturedFrames) : stream); - const deltas = events.filter(event => event.data.delta?.type === "thinking_delta"); - expect(deltas.map(event => event.data.delta.thinking).join("")).toBe(text); - expect(events.filter(event => event.name === "error")).toHaveLength(1); - expect(events.at(-1)).toMatchObject({ name: "error", data: { type: "error", error: { - type: "request_too_large", code: "translation_buffer_limit", - } } }); - expect(JSON.stringify(events.at(-1)).length).toBeLessThan(1024); - expect(events.some(event => event.name === "message_stop" || event.name === "message_delta" || event.name === "content_block_stop")).toBe(false); - expect(events.some(event => event.data.delta?.type === "signature_delta")).toBe(false); - if (capturedFrames) { - expect(capturedFrames.join("")).toBe(captured); - expect(reasoningBytesAtOverflow).toBe(text.length); - expect(reasoningBytes).toBe(0); - expect(budget.snapshot().overflows).toBe(1); - // Feed the actual generated frames, without inventing an error event or - // collecting one huge chunk that introduces a different buffer limit. - const message = await collectAnthropicMessage(streamFromChunks(capturedFrames), "m", trackedBudget); - expect(message).toMatchObject({ type: "error", error: { - type: "request_too_large", code: "translation_buffer_limit", - } }); - expect(message).not.toHaveProperty("content"); - expect(message).not.toHaveProperty("stop_reason"); - } - // These prove failure happened after all text was retained, not while - // ingesting a delta, and the error path released the thinking reservation. - expect(reasoningBytesAtOverflow).toBe(text.length); - expect(maxReasoningBytes).toBeGreaterThanOrEqual(text.length); - expect(reasoningBytes).toBe(0); - expect(budget.snapshot().overflows).toBe(1); - }); - } - } - 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" } }),