From 57461af5cf66cbe935a03bdc8183113c89155f57 Mon Sep 17 00:00:00 2001 From: Paco Cartones <253313177+pacocartones@users.noreply.github.com> Date: Tue, 25 Aug 2026 18:11:58 +0200 Subject: [PATCH] fix(duckduckgo): preserve streamed chunk boundaries (#11528) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Validated in a combined 10-PR batch worktree off release/v3.8.51 tip. - Focused test: tests/unit/duckduckgo-stream-chunks.test.ts — 2/2 pass - typecheck:core, file-size, changelog-integrity, complexity, cognitive-complexity gates — all OK - Full-repo lint: 503 pre-existing problems confirmed identical on the pure release/v3.8.51 tip — unrelated to this diff ⚠️ base-red inherited: #11449 Thanks for keeping the streaming decoder/line-buffer intact across DuckDuckGo transport chunk boundaries. --- .../duckduckgo-stream-chunk-boundaries.md | 1 + open-sse/executors/duckduckgo-web.ts | 59 +++++++++++-------- tests/unit/duckduckgo-stream-chunks.test.ts | 58 ++++++++++++++++++ 3 files changed, 93 insertions(+), 25 deletions(-) create mode 100644 changelog.d/fixes/duckduckgo-stream-chunk-boundaries.md create mode 100644 tests/unit/duckduckgo-stream-chunks.test.ts diff --git a/changelog.d/fixes/duckduckgo-stream-chunk-boundaries.md b/changelog.d/fixes/duckduckgo-stream-chunk-boundaries.md new file mode 100644 index 0000000000..cc79e0ed77 --- /dev/null +++ b/changelog.d/fixes/duckduckgo-stream-chunk-boundaries.md @@ -0,0 +1 @@ +- Fixed DuckDuckGo streaming responses losing JSON lines and UTF-8 characters split across network chunks. diff --git a/open-sse/executors/duckduckgo-web.ts b/open-sse/executors/duckduckgo-web.ts index 3b066d3c0f..e112e4c116 100644 --- a/open-sse/executors/duckduckgo-web.ts +++ b/open-sse/executors/duckduckgo-web.ts @@ -815,33 +815,42 @@ export class DuckDuckGoWebExecutor extends BaseExecutor { }); } - const transformStream = new TransformStream({ - async transform(chunk, controller) { - const text = new TextDecoder().decode(chunk); - const lines = text.split("\n"); + const decoder = new TextDecoder(); + const encoder = new TextEncoder(); + let pendingLine = ""; - for (const line of lines) { - if (!line.trim()) continue; - if (line === "[DONE]") { - controller.enqueue(new TextEncoder().encode("data: [DONE]\n\n")); - continue; - } + const enqueueLine = (line: string, controller: TransformStreamDefaultController) => { + const normalizedLine = line.endsWith("\r") ? line.slice(0, -1) : line; + if (!normalizedLine.trim()) return; + if (normalizedLine === "[DONE]") { + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + return; + } - const data = parseDuckDuckGoDataLine(line); - const content = extractDuckDuckGoContent(data); - if (content) { - const openaiFormat = { - choices: [ - { - delta: { content }, - index: 0, - }, - ], - }; - const encoded = new TextEncoder().encode(`data: ${JSON.stringify(openaiFormat)}\n\n`); - controller.enqueue(encoded); - } - } + const data = parseDuckDuckGoDataLine(normalizedLine); + const content = extractDuckDuckGoContent(data); + if (content) { + const openaiFormat = { + choices: [ + { + delta: { content }, + index: 0, + }, + ], + }; + controller.enqueue(encoder.encode(`data: ${JSON.stringify(openaiFormat)}\n\n`)); + } + }; + + const transformStream = new TransformStream({ + transform(chunk, controller) { + const lines = `${pendingLine}${decoder.decode(chunk, { stream: true })}`.split("\n"); + pendingLine = lines.pop() ?? ""; + for (const line of lines) enqueueLine(line, controller); + }, + flush(controller) { + pendingLine += decoder.decode(); + if (pendingLine) enqueueLine(pendingLine, controller); }, }); diff --git a/tests/unit/duckduckgo-stream-chunks.test.ts b/tests/unit/duckduckgo-stream-chunks.test.ts new file mode 100644 index 0000000000..4a80f58929 --- /dev/null +++ b/tests/unit/duckduckgo-stream-chunks.test.ts @@ -0,0 +1,58 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; + +import { DuckDuckGoWebExecutor } from "../../open-sse/executors/duckduckgo-web.ts"; + +type DuckDuckGoResponseProcessor = { + processResponse( + response: Response, + streaming: boolean, + hasTools: boolean, + requestedTools: unknown[] + ): Promise; +}; + +async function transformChunks(chunks: Uint8Array[]): Promise { + const body = new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(chunk); + controller.close(); + }, + }); + const executor = new DuckDuckGoWebExecutor() as unknown as DuckDuckGoResponseProcessor; + const response = await executor.processResponse( + new Response(body, { headers: { "Content-Type": "text/event-stream" } }), + true, + false, + [] + ); + return response.text(); +} + +describe("DuckDuckGo streaming chunk boundaries", () => { + it("preserves a data line split across transport chunks", async () => { + const encoder = new TextEncoder(); + const output = await transformChunks([ + encoder.encode('data: {"message":"hel'), + encoder.encode('lo"}\n[DONE]\n'), + ]); + + assert.equal( + output, + 'data: {"choices":[{"delta":{"content":"hello"},"index":0}]}\n\n' + "data: [DONE]\n\n" + ); + }); + + it("preserves a multi-byte UTF-8 character split across transport chunks", async () => { + const encoded = new TextEncoder().encode('data: {"message":"café ☕"}'); + const coffeeStart = encoded.indexOf(0xe2); + assert.notEqual(coffeeStart, -1, "fixture must contain the three-byte coffee character"); + + const output = await transformChunks([ + encoded.slice(0, coffeeStart + 1), + encoded.slice(coffeeStart + 1), + ]); + + assert.equal(output, 'data: {"choices":[{"delta":{"content":"café ☕"},"index":0}]}\n\n'); + }); +});