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'); + }); +});