fix(duckduckgo): preserve streamed chunk boundaries (#11528)

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.

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.

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

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<Response>;
+};
+
+async function transformChunks(chunks: Uint8Array[]): Promise<string> {
+  const body = new ReadableStream<Uint8Array>({
+    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');
+  });
+});