From 2b5253da795410e73b3c50c8cc6e45fb3ad6c31d Mon Sep 17 00:00:00 2001 From: backryun Date: Tue, 11 Aug 2026 20:50:10 +0900 Subject: [PATCH] fix(types): narrow Claude stream deltas (#9990) --- open-sse/utils/stream.ts | 17 ++--- open-sse/utils/streamClaudeDelta.ts | 29 +++++++++ .../unit/stream-claude-delta-contract.test.ts | 65 +++++++++++++++++++ 3 files changed, 98 insertions(+), 13 deletions(-) create mode 100644 open-sse/utils/streamClaudeDelta.ts create mode 100644 tests/unit/stream-claude-delta-contract.test.ts diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index ecd2b51272..8bc363017a 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -78,6 +78,7 @@ import { restoreOpenAIToolNames, } from "../translator/helpers/toolCallHelper.ts"; import { normalizeFinalOpenAIStreamChunk } from "./openAIStreamChunk.ts"; +import { collectClaudeDelta } from "./streamClaudeDelta.ts"; /** * Race a response body read against a timeout. @@ -2005,18 +2006,8 @@ export function createSSEStream(options: StreamOptions = {}) { // Do this before translation so we capture content regardless of translator output shape // Claude format - if (parsed.delta?.text) { - const t = parsed.delta.text; - totalContentLength += t.length; - if (state?.accumulatedContent !== undefined && typeof t === "string") - state.accumulatedContent = appendBoundedText(state.accumulatedContent, t); - } - if (parsed.delta?.thinking) { - const t = parsed.delta.thinking; - totalContentLength += t.length; - if (state?.accumulatedReasoning !== undefined && typeof t === "string") - state.accumulatedReasoning = appendBoundedText(state.accumulatedReasoning, t); - } + const claudeDelta = collectClaudeDelta(parsed.delta, state); + totalContentLength += claudeDelta.contentLength; // OpenAI format if (parsed.choices?.[0]?.delta?.content) { @@ -2098,7 +2089,7 @@ export function createSSEStream(options: StreamOptions = {}) { } const translateHasContent = - typeof parsed.delta?.text === "string" || + claudeDelta.hasText || typeof parsed.choices?.[0]?.delta?.content === "string" || Boolean(getAnyReasoningValue(parsed.choices?.[0]?.delta)); if (translateHasContent && !contentAfterToolSeen) { diff --git a/open-sse/utils/streamClaudeDelta.ts b/open-sse/utils/streamClaudeDelta.ts new file mode 100644 index 0000000000..62eb918e47 --- /dev/null +++ b/open-sse/utils/streamClaudeDelta.ts @@ -0,0 +1,29 @@ +import { appendBoundedText } from "./streamHelpers.ts"; + +type ClaudeDeltaState = { + accumulatedContent?: string; + accumulatedReasoning?: string; +}; + +export function collectClaudeDelta(delta: unknown, state?: ClaudeDeltaState) { + const record = + delta && typeof delta === "object" && !Array.isArray(delta) + ? (delta as Record) + : {}; + const text = record.text; + const thinking = record.thinking; + let contentLength = 0; + + if (typeof text === "string" && text) { + contentLength += text.length; + if (state?.accumulatedContent !== undefined) + state.accumulatedContent = appendBoundedText(state.accumulatedContent, text); + } + if (typeof thinking === "string" && thinking) { + contentLength += thinking.length; + if (state?.accumulatedReasoning !== undefined) + state.accumulatedReasoning = appendBoundedText(state.accumulatedReasoning, thinking); + } + + return { contentLength, hasText: typeof text === "string" }; +} diff --git a/tests/unit/stream-claude-delta-contract.test.ts b/tests/unit/stream-claude-delta-contract.test.ts new file mode 100644 index 0000000000..a807d7267b --- /dev/null +++ b/tests/unit/stream-claude-delta-contract.test.ts @@ -0,0 +1,65 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const testDataDir = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-stream-delta-contract-")); +process.env.DATA_DIR = testDataDir; + +const { createSSEStream } = await import("../../open-sse/utils/stream.ts"); +const { FORMATS } = await import("../../open-sse/translator/formats.ts"); +const core = await import("../../src/lib/db/core.ts"); + +test.after(() => { + core.resetDbInstance(); + fs.rmSync(testDataDir, { recursive: true, force: true }); +}); + +test("createSSEStream ignores non-string Claude deltas before estimating usage", async () => { + let onCompletePayload: unknown = null; + const source = new ReadableStream({ + start(controller) { + const encoder = new TextEncoder(); + for (const payload of [ + { + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: { malformed: true }, thinking: 7 }, + }, + { + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: "Valid answer" }, + }, + { type: "message_stop" }, + ]) { + controller.enqueue(encoder.encode(`data: ${JSON.stringify(payload)}\n\n`)); + } + controller.close(); + }, + }); + + await new Response( + source.pipeThrough( + createSSEStream({ + mode: "translate", + targetFormat: FORMATS.CLAUDE, + sourceFormat: FORMATS.OPENAI, + provider: "claude", + model: "claude-opus-4-6", + body: { messages: [{ role: "user", content: "hello" }] }, + onComplete(payload) { + onCompletePayload = payload; + }, + }) + ) + ).text(); + + const completed = onCompletePayload as { + responseBody: { choices: Array<{ message: { content: string } }> }; + usage: { total_tokens: number }; + }; + assert.equal(completed.responseBody.choices[0].message.content, "Valid answer"); + assert.equal(Number.isFinite(completed.usage.total_tokens), true); +});