fix(types): narrow Claude stream deltas (#9990)

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) {

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<string, unknown>)
+      : {};
+  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" };
+}

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<Uint8Array>({
+    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);
+});