diff --git a/changelog.d/fixes/10223-deepseek-responses-sse-cjk-deltas.md b/changelog.d/fixes/10223-deepseek-responses-sse-cjk-deltas.md new file mode 100644 index 0000000000..8f3c19bb20 --- /dev/null +++ b/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)) \ No newline at end of file diff --git a/open-sse/transformer/responsesTransformer.ts b/open-sse/transformer/responsesTransformer.ts index a9c0db949e..1ef35e4aa6 100644 --- a/open-sse/transformer/responsesTransformer.ts +++ b/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); diff --git a/tests/unit/responses-transformer-cjk-split.test.ts b/tests/unit/responses-transformer-cjk-split.test.ts new file mode 100644 index 0000000000..59cf48266c --- /dev/null +++ b/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"); +}); \ No newline at end of file