SSE断线重连：协议机制与OpenAI简历流详解

断流不只是一个问题。是该恢复、重启还是放弃，取决于服务端提供了什么能力；客户端猜错了，要么产生重复 token，要么为同一次补全付两份钱。

三种重连场景

带 id 的纯 Server-Sent Events。SSE 协议内置了恢复机制。服务器在每个事件上发送 id: 字段，符合标准的客户端会记住它，重连时客户端通过 Last-Event-ID 请求头把它发回去，这样服务器就能从断点继续。retry: 字段让服务器可以建议以毫秒为单位的重连延迟。如果你自己控制服务器——比如在模型前架设自己的 API——就应该实现这个机制，测试要验证请求头带着正确的值发出。

支持显式恢复的 Provider 流。OpenAI 的 Responses API 后台模式提供了这个能力：创建响应时将 background 和 stream 都设为 true，跟踪每个事件上携带的 sequence_number，如果连接断开就重连，并把看到的最后一个值作为 starting_after 查询参数传回去。文档明确指出，只有用流式传输创建的后台响应才能启动新流，数据保留大约十分钟。参见 OpenAI 后台模式指南。

完全无法恢复。普通的同步流式调用就是这种情况。如果套接字中断，生成就消失了。唯一的办法是重新发起整个请求——再次付费，得到不同的文本——或者直接失败。如果假装可以恢复，就会产生重复 token 的 bug。

记录下你的每个流式调用点属于三种情况中的哪一种。这个领域一半的重连 bug 都是客户端用第一种情况的行为去对接第三种情况的服务端。

需要断言的不变式

不要对事件的顺序做断言，也不要对文本内容做断言，要断言的是等价性：

// Given the same deterministic upstream, the assembled result must not
// depend on whether the transport was interrupted.
expect(assembledWithDrop).toEqual(assembledWithoutDrop);

这一个断言能捕获所有的重复和所有的缺口，因为两者都会改变最终拼接出的字符串。在此基础上，三个更窄的断言也值得显式写出来，因为它们各有不同的诱因：

没有重复片段。经典失败场景是在 drop 发生在事件中途时从最后一个完成的事件恢复，重放一段已经追加过的片段。要用一个可区分的 delta fixture 来做断言——不要用重复的词——这样重复就能看出来。

恰好一个终端事件到达消费者。重连如果重放了流的结尾，可能会让完成回调执行两次，而应用如果在完成时保存结果就会写入两条记录。

重试次数有上限。对拒绝连接的服务端发起无限重连循环会把故障放大。要断言重试次数以及延迟在增长。

模拟事件中途的断流

重要的细节是，真实的断流发生在一行的中间，而不是整齐地位于两个事件之间。如果测试只在事件边界上切断，就无法触碰到 buffer 处理代码——而 bug 就藏在那里。让伪造的服务器把切断点作为参数，在多个切断点运行测试，包括在一个 delta 的 JSON 内部以及紧跟在 data: 前缀之后，并使用一个小的本地 HTTP 服务器而不是 SDK 层面的 mock——目的是用真实的解析器处理真实被截断的字节流。

import { createServer } from "node:http";

// Serves a fixed SSE script, then destroys the socket after N bytes.
export function sseServerThatDropsAfter(bytes: number, script: string) {
  return createServer((req, res) => {
    res.writeHead(200, {
      "content-type": "text/event-stream",
      "cache-control": "no-cache",
      connection: "keep-alive",
    });
    const resumeFrom = req.headers["last-event-id"];
    const body = resumeFrom ? script.slice(Number(resumeFrom)) : script;

    if (resumeFrom) {
      res.end(body);              // second connection: serve the remainder
      return;
    }
    res.write(body.slice(0, bytes));
    res.socket?.destroy();        // no end event, no close frame: a real drop
  });
}

销毁 socket 很重要。调用 res.end() 是干净的关闭，大多数客户端会把干净的关闭视为流的结束而不是失败——所以那样构建的测试无法断言任何关于重连的东西。"流结束了"和"流停止了"之间的区别，正是客户端必须做对的事情。

序列号与恢复

对于支持显式恢复的端点，要测试的客户端状态就是一个数字：它最后成功处理的序列值。两个断言覆盖它。第一，序列号只在事件已经应用到拼接输出之后才前进，而不是在收到时就前进——否则两者之间的崩溃会永久丢失一个事件。第二，重连请求实际携带了它；用 mock 捕获发出的查询字符串并断言 starting_after 等于期望值，是最便宜的测试，能捕获请求构建器悄悄丢掉这个参数的情况。

还要测试过期路径。如果可恢复窗口已过，重连不会成功，客户端需要一个有定义的错误行为而不是循环。要断言它抛出一个调用方可以处理的类型化错误。

无法恢复时

对于第三种情况，诚实的设计是不重连。有两件事值得测试。第一，部分输出作为部分内容暴露给调用方，带有标志，而不是假装完整地返回——这里最坏的结果是一段没有任何标记的截断回答，然后被存储、总结和引用。第二，如果你确实要重新发起，重新发起从用户角度看是幂等的——不会对内部预算双重扣费，也不会重新执行第一次尝试已经执行过的副作用。

如果重连的重要性足以值得正确实现，常见的答案是把生成完全移出请求路径：异步运行任务，在 delta 到达时持久化它们，让客户端轮询或订阅持久化的记录。这把一个不可恢复的流转换成了一个可恢复的读取，测试也变成了普通的数据库测试。代价是延迟和复杂度，所以只对长任务值得这样做——对于短任务，重启确实比这套机制更划算。