OmniRoute DuckDuckGo 流式响应修复：如何在网络分块边界上安全重组 JSON 行与 UTF-8 字符

导读

本文以 changelog.d/fixes/duckduckgo-stream-chunk-boundaries.md 所记录的修复为线索，剖析 OmniRoute 中 duckduckgo-web（Duck.ai）免费流式执行器如何彻底解决两类经典流式传输缺陷：JSON 数据行被网络分块劈断而丢失，以及多字节 UTF-8 字符被分块劈断而产生乱码（U+FFFD）。读完你将理解 TextDecoder 的 stream: true 语义、基于 pendingLine 的行缓冲重组原理、对应单元测试的构造方法，以及这套"分块边界安全"范式在仓库内同类 Web 执行器上的通用价值。

一、修复背景：duckduckgo-web 免费流式网关

OmniRoute 的 duckduckgo-web（别名 ddgw，见 open-sse/executors/index.ts）是一个 authType: "none" 的免费无密钥 Provider，注册在 open-sse/config/providers/registry/duckduckgo-web/index.ts。它通过逆向 Duck.ai 的 /duckchat/v1 接口，把聊天请求匿名转发到 https://duck.ai，再把上游的 SSE/NDJSON 流翻译成 OpenAI 兼容格式。注册表里维护的免费模型阵容包括 gpt-5.4-mini、gpt-5.6-luna、claude-haiku-4-5、mistral-small-2603、tinfoil/gpt-oss-120b、tinfoil/gemma4-31b。

执行器本身（open-sse/executors/duckduckgo-web.ts）已经相当健壮：VQD 挑战令牌获取与求解、匿名会话 Cookie 维护、指纹轮换会话池、模型目录热刷新、熔断器、以及 418 ERR_BN_LIMIT / 429 等上游限流语义的精确映射。而本次 changelog 修复聚焦在最末端的一环——流式响应体的重组与翻译：上游返回的 data: 前缀 JSON 行与裸 [DONE] 哨兵，经 TCP/HTTP 分块传输后落在不同的网络包上时，执行器必须原样把它们拼回完整的行，再进行逐行翻译，否则就会出现"丢内容"和"乱码"。

二、缺陷表象与根因

HTTP 流式响应在网络上会被切分成大小不一的字节块（chunk），这些块的切分点是任意字节偏移，与消息自身的语义边界（换行符、多字节字符边界）毫无关系。changelog 描述的症状可归为两类经典故障：

场景一：JSON 数据行被网络分块劈断而丢失

一条完整的 SSE 数据行 data: {"message":"hello"}\n 可能被网络切成 data: {"message":"hel 与 lo"}\n 两段到达。若解析器"来一块、切一块"，对每个网络 chunk 单独执行 split("\n")，则：

• 第一块末尾的半个行没有换行符，会被当成一个"完整行"提前解析 → JSON.parse 必然失败或被丢弃；
• 或者解析器只取每块中含 \n 的完整行、把末尾不完整的残行直接扔掉 → lo"} 连同换行符一起丢失，最终表现为输出 token 缺失、[DONE] 永远等不到、下游流提前悬挂或截断。

这正是 changelog 所称 "losing JSON lines split across network chunks"。

场景二：多字节 UTF-8 字符被网络分块劈断而乱码

中文、emoji 等在 UTF-8 下由 2~4 字节编码（例如 ☕ 占 3 字节：E2 98 95）。网络分块可能恰好把一个字符的字节序列切成两半。若对每个 chunk 用独立新建的 TextDecoder（或未启用流式语义）解码，末尾的半截字符在 fatal: false 默认语义下会被直接替换为 U+FFFD（�），后续到达的剩余字节又会另起炉灶解码，产生第二个乱码——"café ☕"可能变成"café � ��"。仓库中同类缺陷在 Ollama 转换链上留下过一致描述：11921-ollama-multibyte-utf8-boundary.md 明确指出 Ollama NDJSON 转换曾把跨流块的 CJK/emoji 损坏成 U+FFFD。

三、修复实现：流式边界安全重组（源码剖析）

本次修复落在 open-sse/executors/duckduckgo-web.ts 中 processResponse 的流式分支（streaming === true，约 L875-L983）。修复后代码以三件套保证跨 chunk 语义完整性：单一长生命周期 TextDecoder + { stream: true } 增量解码 + pendingLine 行缓冲器。

1. 单一 TextDecoder 与 stream: true

const decoder = new TextDecoder();
const encoder = new TextEncoder();
let pendingLine = "";

解码器在整个流生命周期内只创建一次，并在每次 decode(chunk, { stream: true }) 调用时明确声明"这是分块字节流中的一段"。启用 stream: true 后，TextDecoder 会把每次解码末尾遇到的不完整多字节序列暂存于内部状态，等待下一块字节补齐，而不会立刻输出 U+FFFD。流结束时再通过不带参数的 decoder.decode() 冲刷掉内部残留（见下文 flush）。这正是解决"UTF-8 字符被劈断"的标准做法。

2. pendingLine 行缓冲与 flush

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);
  },
});

逐行拆解其正确性：

• transform 先把上一次遗留下来的 pendingLine（不完整的残行）与本次增量解码得到的文本拼接，再统一 split("\n")；
• split 结果的最后一段必然没有换行符（"半行"），于是 lines.pop() 将其存回 pendingLine，等待下一块拼接——跨块劈断的 JSON 行由此被原样重组，不会丢失半个 token；
• 中间所有带换行符的完整行立即交给 enqueueLine 处理，实现边收边发、低首字节延迟；
• flush 在流收尾时先用 decoder.decode() 冲刷解码器内部残留的多字节尾字节，再处理最后一段 pendingLine，保证流末没有内容被吞掉。

3. 逐行归一化与 OpenAI SSE 输出

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(normalizedLine);
  const content = extractDuckDuckGoContent(data);
  if (content) {
    const openaiFormat = {
      choices: [{ delta: { content }, index: 0 }],
    };
    controller.enqueue(encoder.encode(`data: ${JSON.stringify(openaiFormat)}\n\n`));
  }
};

重组出的每一行还要经过三层归一化，才能变成 OpenAI 兼容 SSE：

1. CRLF 归一化：\r\n 结尾的行先裁掉 \r，兼容上游混合换行风格；
2. 空行与哨兵处理：空白行直接跳过；裸 [DONE] 被翻译成标准 SSE 终止帧 data: [DONE]\n\n（原样保留 [DONE] 字样以符合 OpenAI SSE 约定，测试断言中可清晰看到这一映射）；
3. JSON 抽取与协议翻译：parseDuckDuckGoDataLine（L181-L189）仅接受 data: 前缀行，容错地 JSON.parse；extractDuckDuckGoContent（L171-L179）同时兼容 content 与 message 两种字段形态；抽取到文本后用 {choices:[{delta:{content},index:0}]} 包装并以 data: {json}\n\n 的 SSE 帧格式发出，转换后的响应 Content-Type 保持 text/event-stream。

非流式路径（streaming === false，L943-L981）没有分块问题，仍走"整段文本按行聚合后一次性返回 JSON"的逻辑，并通过 buildToolAwareResult 支持工具调用场景，两种路径互不干扰。

四、测试验证：精确构造分块边界

仓库为该修复配套了针对性极强的单元测试 tests/unit/duckduckgo-stream-chunks.test.ts。测试用 ReadableStream 把构造好的字节数组逐块入队，模拟真实网络中任意切分的到达时序，再调用 processResponse(response, true, false, []) 对结果整体取文本做断言：

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();
}

两个用例逐一对应 changelog 中列举的两类故障：

用例 1：跨块劈断的 JSON 行不丢失

const output = await transformChunks([
  encoder.encode('data: {"message":"hel'),
  encoder.encode('lo"}\n[DONE]\n'),
]);
// assert output ===
// 'data: {"choices":[{"delta":{"content":"hello"},"index":0}]}\n\n' + 'data: [DONE]\n\n'

data: {"message":"hel 与 lo"}\n 拆成两块投喂：内容字段被劈成两半，[DONE] 哨兵与换行符也在第二块才到齐。只有正确实现 pendingLine 缓冲与 flush 语义，最终才能重组出 hello 并正常收尾；任何"逐块独立 split"或"残行丢弃"的实现都会在此用例上失败。

用例 2：跨块劈断的多字节 UTF-8 字符不乱码

const encoded = new TextEncoder().encode('data: {"message":"café ☕"}');
const coffeeStart = encoded.indexOf(0xe2); // ☕ 三字节编码 E2 98 95 的首字节
const output = await transformChunks([
  encoded.slice(0, coffeeStart + 1), // 在 ☕ 三字节的中间切开
  encoded.slice(coffeeStart + 1),
]);
// assert output === 'data: {"choices":[{"delta":{"content":"café ☕"},"index":0}]}\n\n'

测试用 encoded.indexOf(0xe2) 精确定位 ☕ 三字节序列（E2 98 95）的起点，并故意在三字节正中间断开：第一块只含 E2 一个字节。若解码器不支持流式续传，第一块末尾会产生 U+FFFD，断言必然失败；修复后 café ☕ 得以逐字还原。该用例可结合仓库 vitest 配置（vitest.config.ts）运行验证。

五、在 OmniRoute 中的工程上下文

1. 从注册表到执行器的调用链

这条流式路径处于完整的 provider 管道末端：请求经调度进入 open-sse/executors/index.ts 按 duckduckgo-web / ddgw 懒加载 DuckDuckGoWebExecutor → execute()（L457-L711）完成 VQD 获取、模型目录校验与角色归一化后，向 CHAT_URL 发起流式 POST → 上游响应交给 processResponse 的流式分支翻译。由于该 Provider 无需任何凭据（注册表中 authType: "none"），它同时被归类为免费 Provider，在组合路由（combo）与免费回落链路中作为兜底模型出场——因此这条流式重组路径的质量直接影响免费档用户的长文本与多语种体验。

2. 同一缺陷族的系统性治理

仓库对"字节流跨块语义"问题有系统性的测试覆盖意识，本次修复并非孤立样本：

• 同样的 { stream: true } 增量解码模式被广泛用于其它 SSE 类执行器（如 grok-web、kimi、moonshot、copilot-web 等大量执行器文件均出现 stream: true 解码调用）；
• 同类 UTF-8/流边界问题在 Ollama 转换路径上也被单独记录修复（changelog.d/fixes/11921-ollama-multibyte-utf8-boundary.md）；
• 测试层另有针对 CJK 跨块、token 边界、SSE 解析器的专门用例，例如 tests/unit/responses-transformer-cjk-split.test.ts、tests/unit/stream-markdown-token-boundary.test.ts 与 tests/unit/correctness/sse-parser.property.test.ts，共同构成对传输层边界 bug 的回归防线。

六、可复用的通用范式与排查清单

从本次修复可提炼出适用于所有 SSE/NDJSON 流翻译器的硬性规则：

关注点	错误做法	正确范式（本次修复采用）
解码器生命周期	每个 chunk 新建 TextDecoder	全程一个解码器
多字节字符	直接 decode(chunk)	decode(chunk, { stream: true })，收尾 decode()
行重组	逐 chunk split("\n") 或丢弃残行	pendingLine 拼接残行后再 split，残段存回缓冲
收尾	忽略流末缓冲	flush 中先冲刷解码器再处理最后残行
换行风格	假定仅 \n	先裁 \r 再判空行
哨兵	特殊处理丢到主协议里	[DONE] 独立映射为标准终止帧

排查这类"偶发丢字/乱码/流悬挂"时，可按以下顺序复核：确认解码器是否跨 chunk 复用并启用 stream: true；确认 pendingLine 是否在每块 transform 之前拼接、之后回存；确认流关闭时 flush 是否冲刷了解码器尾字节与最后的残行；最后用"在任意字节偏移断开 + 断言重组后整文本"的用例（参考 tests/unit/duckduckgo-stream-chunks.test.ts 的构造手法）把竞态固化为回归测试。只要解析器把"字节流的切分边界"当作与协议无关的物理事实来对待，SSE 翻译层就不会因为上游或网络的分块习惯而丢内容、乱码或截断。