From 6087441bded742e3bc9ec813ba6628fa640c2874 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Mon, 10 Aug 2026 03:56:00 +0900 Subject: [PATCH] fix(claude): bound streamed reasoning identity --- src/claude/outbound.ts | 33 ++++++++++++++++++--- tests/claude-outbound.test.ts | 54 +++++++++++++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 4 deletions(-) diff --git a/src/claude/outbound.ts b/src/claude/outbound.ts index c8b4d20df8..095d332d49 100644 --- a/src/claude/outbound.ts +++ b/src/claude/outbound.ts @@ -9,6 +9,7 @@ * - message_delta.usage is cumulative; message_start embeds a full message snapshot. * - errors: {type:"error", error:{type,message}}; may arrive mid-stream after HTTP 200. */ +import { createHash } from "node:crypto"; import { isTransientUpstreamStatus } from "../lib/upstream-retry"; import { isTranslatorBudgetExceededError, @@ -24,6 +25,27 @@ function isRec(v: unknown): v is Rec { return !!v && typeof v === "object" && !Array.isArray(v); } +function reasoningIdentityDigest(value: string): string { + return createHash("sha256").update(value).digest("base64url"); +} + +/** Fixed-size identity that preserves protocol boundaries without retaining upstream strings. */ +function boundedReasoningIdentity(value: unknown): string { + if (typeof value === "number") { + if (Number.isSafeInteger(value) && value >= 0) return `n${value}`; + if (Number.isFinite(value)) return `d${value}`; + return Number.isNaN(value) ? "dnan" : value > 0 ? "dinf" : "d-inf"; + } + if (typeof value === "string") { + return `s${reasoningIdentityDigest(value)}`; + } + if (value === null) return "z"; + if (typeof value === "boolean") return value ? "b1" : "b0"; + if (Array.isArray(value)) return `a${reasoningIdentityDigest(JSON.stringify(value) ?? "[]")}`; + // Preserve the prior String(record) category semantics without serializing untrusted trees. + return typeof value === "object" ? "o" : "u"; +} + function uuid(): string { return crypto.randomUUID().replace(/-/g, ""); } @@ -188,7 +210,7 @@ interface OpenBlock { argsBufBytes?: number; webSearchArgsEmitted?: boolean; callId?: string; - /** Last reasoning part identity (item + summary/content index) seen by this thinking block. */ + /** Last fixed-size reasoning identity (item + summary/content index) seen by this block. */ reasoningPartKey?: string; } @@ -364,9 +386,12 @@ export function responsesSseToAnthropicSse( // 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}`; + ? `s${boundedReasoningIdentity(data.summary_index)}` + : `c${boundedReasoningIdentity(data.content_index)}`; + // Upstream string metadata can be arbitrarily large. Hash strings into fixed-size + // components while retaining item and part equality, rather than dropping item_id and + // accidentally joining distinct malformed reasoning items. + const partKey = `${boundedReasoningIdentity(data.item_id)}:${slot}`; if (open!.reasoningPartKey !== undefined && open!.reasoningPartKey !== partKey) { emit("content_block_delta", { type: "content_block_delta", index: open!.index, diff --git a/tests/claude-outbound.test.ts b/tests/claude-outbound.test.ts index a804a1eefe..f7e36a67ab 100644 --- a/tests/claude-outbound.test.ts +++ b/tests/claude-outbound.test.ts @@ -306,6 +306,60 @@ describe("claude outbound SSE", () => { expect(msg2.content.find((b: Record) => b.type === "thinking").thinking).toBe("AB"); }); + test("huge reasoning identities stay bounded without collapsing item or part boundaries", async () => { + const hugeItemA = "a".repeat(1024 * 1024); + const hugeItemB = `${"a".repeat(1024 * 1024 - 1)}b`; + const hugePartA = "p".repeat(1024 * 1024); + const hugePartB = `${"p".repeat(1024 * 1024 - 1)}q`; + const upstream = [ + sse("response.created", { response: { id: "resp_1", status: "in_progress" } }), + sse("response.reasoning_summary_text.delta", { + item_id: hugeItemA, summary_index: hugePartA, delta: "A", + }), + sse("response.reasoning_summary_text.delta", { + item_id: hugeItemA, summary_index: hugePartA, delta: "B", + }), + sse("response.reasoning_summary_text.delta", { + item_id: hugeItemB, summary_index: hugePartA, delta: "C", + }), + sse("response.reasoning_summary_text.delta", { + item_id: hugeItemB, summary_index: hugePartB, delta: "D", + }), + sse("response.completed", { + response: { status: "completed", usage: { input_tokens: 1, output_tokens: 1 } }, + }), + ].join(""); + + const msg = await collectAnthropicMessage( + responsesSseToAnthropicSse(streamFromChunks([upstream]), "m"), + "m", + ) as Record; + expect(msg.content.find((b: Record) => b.type === "thinking").thinking) + .toBe("AB\n\nC\n\nD"); + }); + + test("malformed array reasoning identities retain distinct boundaries", async () => { + const upstream = [ + sse("response.created", { response: { id: "resp_1", status: "in_progress" } }), + sse("response.reasoning_summary_text.delta", { + item_id: [1], summary_index: [0], delta: "A", + }), + sse("response.reasoning_summary_text.delta", { + item_id: [2], summary_index: [0], delta: "B", + }), + sse("response.completed", { + response: { status: "completed", usage: { input_tokens: 1, output_tokens: 1 } }, + }), + ].join(""); + + const msg = await collectAnthropicMessage( + responsesSseToAnthropicSse(streamFromChunks([upstream]), "m"), + "m", + ) as Record; + expect(msg.content.find((b: Record) => b.type === "thinking").thinking) + .toBe("A\n\nB"); + }); + 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" } }),