fix(responses): keep stream-aware TextDecoder across SSE transform chunks (#10223)

changelog.d/fixes/10223-deepseek-responses-sse-cjk-deltas.md

@@ -0,0 +1 @@
+- **fix(responses):** repair corrupted SSE deltas for non-ASCII streams by keeping a single stream-aware `TextDecoder` (`{ stream: true }`) across `transform()` calls instead of recreating it per chunk and decoding without the `stream` flag. When a multi-byte UTF-8 character (CJK/emoji) was split across two TCP chunks — common in Chinese streaming text — the per-chunk decoder truncated it to `U+FFFD`, corrupting every delta while the rebuilt `*.done` snapshot stayed internally identical ([#10223](https://github.com/diegosouzapw/OmniRoute/issues/10223))

open-sse/transformer/responsesTransformer.ts

@@ -231,6 +231,11 @@ export function createResponsesApiTransformStream(
   };
 
   const encoder = new TextEncoder();
+  // #10223: a stream:false TextDecoder recreated per transform() chunk has no
+  // cross-call state, so a multi-byte UTF-8 character (CJK/emoji) split across
+  // two TCP chunks got truncated to U+FFFD, corrupting the deltas. A single
+  // persistent decoder with { stream: true } carries pending bytes between chunks.
+  const decoder = new TextDecoder();
   const nextSeq = () => ++state.seq;
 
   // Normalize output_index to a non-negative integer (replaces fragile parseInt calls)
@@ -577,7 +582,7 @@ export function createResponsesApiTransformStream(
         (state.keepaliveTimer as { unref?: () => void })?.unref?.();
       },
       transform(chunk, controller) {
-        const text = new TextDecoder().decode(chunk);
+        const text = decoder.decode(chunk, { stream: true });
         logger?.logInput(text.trim());
         state.buffer += text;
 
@@ -887,6 +892,11 @@ export function createResponsesApiTransformStream(
       },
 
       flush(controller) {
+        // #10223: stream-end flush — drain any bytes the persistent decoder is
+        // still holding. With { stream:true } complete multi-byte chars are
+        // emitted within transform(), so normally there is nothing left; this
+        // only releases a terminating truncated byte and frees the decoder.
+        state.buffer += decoder.decode();
         // Clear keepalive timer
         if (state.keepaliveTimer) {
           clearInterval(state.keepaliveTimer);

tests/unit/responses-transformer-cjk-split.test.ts

@@ -0,0 +1,111 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+
+// Regression guard for #10223 — DeepSeek /v1/responses corrupted SSE deltas.
+//
+// ROOT CAUSE (open-sse/transformer/responsesTransformer.ts:580): the transform()
+// handler created a brand-new `new TextDecoder()` on every chunk and decoded it
+// WITHOUT `{ stream: true }`. A stream:false decoder has no cross-call state, so
+// whenever a multi-byte UTF-8 character (CJK: 3 bytes, emoji: 4) is split across
+// two TCP chunks — the normal case in Chinese streaming text (the reporter's
+// scenario), the trailing partial bytes are replaced with U+FFFD and the deltas
+// accumulate garbage.
+//
+// This test feeds a CJK text split at a byte boundary INSIDE a multi-byte
+// character and asserts a round-trip against the source text — NOT the
+// `join(deltas) === done` invariant, which cannot catch this bug because done is
+// rebuilt from the same corrupted buffer as the deltas.
+
+const { createResponsesApiTransformStream } = await import(
+  "../../open-sse/transformer/responsesTransformer.ts"
+);
+
+const encoder = new TextEncoder();
+const decoder = new TextDecoder();
+
+function concatBytes(parts) {
+  const total = parts.reduce((sum, part) => sum + part.length, 0);
+  const out = new Uint8Array(total);
+  let offset = 0;
+  for (const part of parts) {
+    out.set(part, offset);
+    offset += part.length;
+  }
+  return out;
+}
+
+function parseSseOutput(output) {
+  return output
+    .trim()
+    .split("\n\n")
+    .map((entry) => {
+      const lines = entry.split("\n");
+      const eventLine = lines.find((line) => line.startsWith("event: "));
+      const dataLine = lines.find((line) => line.startsWith("data: "));
+      return {
+        event: eventLine ? eventLine.slice("event: ".length) : null,
+        data: dataLine ? dataLine.slice("data: ".length) : null,
+      };
+    })
+    .filter((e) => e.event !== null || e.data !== null);
+}
+
+async function runRawBytes(byteChunks, options = {}) {
+  const stream = createResponsesApiTransformStream(null, 3000, options);
+  const writer = stream.writable.getWriter();
+  const reader = stream.readable.getReader();
+
+  const raw = [];
+  const readerTask = (async () => {
+    while (true) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      if (value) raw.push(value);
+    }
+  })();
+
+  for (const chunk of byteChunks) {
+    await writer.write(chunk);
+  }
+  await writer.close();
+  await readerTask;
+
+  return decoder.decode(concatBytes(raw));
+}
+
+test("responses transform preserves multi-byte UTF-8 text split across byte chunks (#10223)", async () => {
+  const source = "REASONIX_中文测试_DEEPSEEK_OK";
+
+  const frame = (data) =>
+    encoder.encode(`data: ${JSON.stringify(data)}\n\n`);
+
+  const deltaChunk = frame({
+    choices: [{ index: 0, delta: { content: source } }],
+  });
+  const finishChunk = frame({
+    choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
+  });
+  const full = concatBytes([deltaChunk, finishChunk]);
+
+  // Split mid-byte inside the first 3-byte CJK character "中".
+  const contentPrefix = encoder.encode(
+    'data: {"choices":[{"index":0,"delta":{"content":"'
+  ).length;
+  const boundary = contentPrefix + encoder.encode("REASONIX_").length + 1;
+  const chunkA = full.slice(0, boundary);
+  const chunkB = full.slice(boundary);
+
+  const output = await runRawBytes([chunkA, chunkB]);
+
+  const events = parseSseOutput(output);
+  const deltas = events
+    .filter((e) => e.event === "response.output_text.delta")
+    .map((e) => JSON.parse(e.data).delta);
+
+  const doneEvent = events.find((e) => e.event === "response.output_text.done");
+  const doneText = JSON.parse(doneEvent.data).text;
+
+  // Round-trip against the SOURCE text — the invariant the old test missed.
+  assert.equal(deltas.join(""), source, "joined deltas should round-trip to the source text");
+  assert.equal(doneText, source, "done snapshot should round-trip to the source text");
+});