Agent 实现方法论 From Scratch

断线恢复与后台运行 v03：事件序号与重连续传

断线恢复与后台运行 v03：事件序号与重连续传

第 15 章我们完成了产品 v02：Agent Core 与传输层分离，主循环发出正式的流式事件协议，HTTP/SSE 服务按会话把事件流推给任意多个客户端。但 v02 有一个没解决的现实问题：连接会断。

用户不会永远在线：地铁里关掉页面、笔记本合盖、Wi-Fi 漂移、公司网络把长连接掐掉。v02 的 GET /events 是「先回放全部历史、再实时推送」：断线后重连（reconnect），服务端把从第一条开始的所有事件再发一遍。功能上没错，但有两个尴尬：重连要浪费带宽重收一遍看过的内容；更糟的是，如果断线发生在 Agent 还没跑完的时候，客户端重连后看到的只是「半截历史」。它不知道该从哪里接上。

本章做产品 v03：给每条事件盖一个会话内单调递增的事件序号（seq），客户端断线时记住自己看到了几号，重连时带着 since=seq 回来，服务端只补发它没看到的部分。同时把「Agent 的运行」和「客户端的连接」彻底解耦：断开连接后 Agent 照常跑完（事件落存储、与有没有订阅者无关），重连只是「从断点续读」。这就是 TP15「断线恢复（resume）与后台运行」。

本章目标

读完本章并做完配套练习后，你应该能够：

概念与动机：断线丢事件会怎样

先演一遍「没有序号」的断线

产品 v02 的模型是「存储 → 回放 → 实时」：事件全部落进会话缓冲，新订阅者从第一条开始回放，之后实时跟进。现在演一遍用户断线：

  1. 用户问「查上海的天气，顺便看下 UTC 时间」，POST /chat 立即返回，Agent 在后台跑；
  2. 用户的浏览器连上 /events，收到 tool_call_start、tool_result（4 条，callId 配对）……然后地铁进隧道，连接断了；
  3. Agent 在服务器上毫不知情，继续跑完：text_delta、done 两条事件落进缓冲；
  4. 出隧道，浏览器自动重连 /events，服务端从第一条开始全量回放这 6 条事件。浏览器收到重复的 4 条，还得自己判断「哪些我见过、哪些是新的」。

功能上没错，但三个问题很实在：

问题一：重连 = 全量重放。 会话越长，重连的代价越大。一个跑了一小时的 Agent 会话可能有几千条事件，断一次线全收一遍，纯浪费。

问题二：客户端无法表达「我从哪里断的」。 它只能收到什么算什么，自己数自己看了几条。但「数到第几条」是客户端本地状态，服务器不知道，双方没法对齐。

问题三：断线后「Agent 是否还在跑」对客户端是黑盒。 重连后它拿到一堆历史，但不知道这是不是全部、Agent 是否已经结束、有没有可能再等来新事件。它没有一把「尺子」去量服务器现在跑到哪了。

解法：给事件流一个「可寻址的位置」

三个问题指向同一个需求：每条事件都要有一个在会话内唯一的、可比较的位置，就像书的页码、日志的行号、数据库的自增主键。客户端说「我从第 4 条之后接着读」，服务器就知道该发什么。这个位置就是 seq（sequence，序号）。

这个思路不是我们发明的。消息系统（Kafka 的 offset、RocketMQ 的 queue offset）、数据库（自增 ID、LSN）、版本控制（commit hash 的祖先关系）都用同一个抽象：给有序日志里的每条记录一个单调递增的序号，消费方就能精确地说出自己消费到哪了。ch10 的 eventId 已经是同一个思想（全局自增）；本章把它细化成会话内的 seq，并把它接到重连续传协议上。

业界参照：OpenAI Agents SDK 的 Results & state 里，一次运行的 RunResult 可以序列化成可恢复的 RunState 快照（to_state()），审批中断后从快照恢复运行；其「继续/恢复对话」一节（previous_response_id 续链、RunState resume）就是「运行状态可以被打断、被精确恢复」的官方实践。另一个参照是 learn-claude-code（s13 后台任务）：它把「慢操作丢到后台线程执行、完成后注入通知」作为长时运行的核心机制。我们的「Agent 在后台跑、事件落存储」是同一思想的传输层版本。注意它的范围说明明确把「会话生命周期控制（resume/fork）」列为有意省略。这正是本章的差异化所在：我们用事件序号把 resume 做进了协议层，而不是产品外壳。

事件序号与存储

seq 的语义：三条规则

seq 是会话内的概念，三条规则定死：

  1. 会话内单调递增：同一会话的第 N+1 条事件，seq 严格大于第 N 条，客户端可以放心地把「最后看到的 seq」当作进度指针（resume cursor）；
  2. 跨会话独立：不同会话各自从 1 开始计数，s1 的 seq 1 和 s2 的 seq 1 互不相干，隔离是会话层（sessionId）的责任；
  3. 跨多轮运行连续：同一会话的第 2 轮对话不会重新从 1 数起。seq 是「会话」的序号，不是「单次运行」的序号。

第三条值得展开。ch15 的 loop 每次 runAgent 都会重新喂完整历史、跑完即弃（core 无状态）。如果序号按「单次运行」分配，第二轮又从 1 开始，客户端就无法区分「两轮的 1 号事件」。所以序号计数器必须活在 loop 之外、按会话持有。本章的主循环由此多了一个注入项：

// lib/loop.mjs — ch16 升级：emit 给每条事件盖会话内序号
const emit = (event) =>
  onEvent?.({ sessionId, seq: seq ? seq.next(sessionId) : undefined, ...event });

seq（这里是参数名）是一个分配器（allocator），由调用方（服务器）创建、跨运行共享。loop 依然无状态。它只是「拿到分配器就用」，分配器本身是注入的依赖。不传分配器，事件就没有 seq 字段（CLI 就是如此：终端不需要恢复，没必要多花字段）。

// lib/events.mjs — 序号分配器：每会话一个计数器
export function createSeqAllocator() {
  const lastSeqs = new Map(); // sessionId -> last issued seq
  return {
    next(sessionId) {
      const seq = (lastSeqs.get(sessionId) ?? 0) + 1;
      lastSeqs.set(sessionId, seq);
      return seq;
    },
    peek(sessionId) {
      return lastSeqs.get(sessionId) ?? 0; // 只读，不推进
    },
  };
}

next 每次给一个会话发下一个号；peek 用于回答「这个会话现在到几号了」（POST /chat 返回 lastSeq 就靠它，准确的说是存储层的 lastSeq）。三个规则对应三行代码：?? 0 让新会话从 1 开始（跨会话独立）、+ 1 保证递增（会话内单调）、分配器活在 loop 外（跨运行连续）。

事件序号状态图

一次完整演示（两轮对话 + 一次断线）的会话状态长这样：

flowchart TD
  subgraph store["会话 s1 的事件存储（append 按 seq 递增落库）"]
    E1["seq 1 · tool_call_start<br/>(call_time)"]
    E2["seq 2 · tool_result<br/>(get_time → 时间串)"]
    E3["seq 3 · tool_call_start<br/>(call_weather)"]
    E4["seq 4 · tool_result<br/>(get_weather → 天气串)"]
    E5["seq 5 · text_delta<br/>(最终答案)"]
    E6["seq 6 · done<br/>(usage)"]
  end

  subgraph client["客户端视角"]
    C1["客户端 A：连上 /events，读到 seq 1–4"]
    D1["💥 断线：客户端 A 消失"]
    C2["客户端 B：带 since=4 重连<br/>只补 seq 5、6"]
    C3["客户端 C（对照）：since=0 全量回放<br/>seq 1–6 一条不丢"]
  end

  E1 --> E2 --> E3 --> E4
  E4 -->|"断线窗口：Agent 仍在后台跑"| E5 --> E6
  C1 --> D1
  D1 --> C2
  store -. "补发 seq > 4" .-> C2
  store -. "全量回放 seq > 0" .-> C3

要点：序号是存储的属性，不是连接的属性。客户端 A 断线不影响 seq 5、6 的生成与落库；重连的客户端 B 和从未断线过的客户端 C 看到的是同一条、以 seq 对齐的事件序列，断线与否只影响「客户端从哪个 seq 开始读」，不影响「存储里有什么」。

事件存储：append / getSince / connect

ch15 的会话缓冲只有「追加 + 全量回放 + 订阅」三件事。ch16 的新模块 lib/event-store.mjs 把它升级成带序号的断点续传存储，对外承诺三条保证：

  1. 无订阅者也能写入：append(sessionId, event) 不要求任何人在听，事件照常落库、照常编号。这是「Agent 后台继续跑」的存储侧前提；
  2. 每条入库事件必有 seq：客户端可以精确说出「我从哪里断的」；
  3. getSince(sessionId, since) 只返回 seq > since 的增量：不重发、不丢。
// lib/event-store.mjs — 核心切片
export function createEventStore() {
  const sessions = new Map(); // sessionId -> { events: [], listeners: Set<fn> }

  return {
    append(sessionId, event) {
      const s = bucket(sessionId);
      s.events.push(event);              // 顺序追加（事件已由 loop 盖好 seq）
      for (const listener of s.listeners) listener(event);
      return event;
    },

    getSince(sessionId, since) {
      // 断点续传原语：只给 seq > since 的事件
      return (sessions.get(sessionId)?.events ?? []).filter((e) => e.seq > since);
    },

    lastSeq(sessionId) {
      const s = sessions.get(sessionId);
      return s?.events.length ? s.events[s.events.length - 1].seq ?? 0 : 0;
    },

    connect(sessionId, { since = 0, onEvent }) {
      // 重连：先订阅、再补发（同一同步 tick 内完成，见「竞态」坑）
      const unsubscribe = subscribe(sessionId, onEvent);
      for (const event of getSince(sessionId, since)) onEvent(event);
      return unsubscribe;
    },
    // …events / subscribe / listSessions…
  };
}

这个存储是纯内存数据结构：没有 node:* 依赖、没有定时器，浏览器沙箱里也能跑，它就是练习的判题对象。接 ch10 的持久层思路：把 sessions 换成 SQLite 的 event 表（ch10 的 eventId 正是 seq 的持久化化身），append 变成 INSERT、getSince 变成 WHERE session_id=? AND seq>?，协议侧一行都不用改。

重连续传协议

客户端如何表达「我从哪里断的」

协议只在 wire 上加了一个查询参数和一个响应字段：

位置变化语义
GET /events?sessionId=s1&since=N新增 since 参数只回放 seq > N 的事件；缺省/非法值视为 0（全量回放，与 ch15 行为一致）
POST /chat 响应新增 lastSeq 字段本次提交时刻，该会话事件流的终点。客户端订阅时从这里开始，只收这轮新事件

客户端断线重连的标准流程：

  1. 正常订阅期间，客户端记录最后一条完整收到的 seq（比如 4）；
  2. 断线。Agent 在服务器后台继续跑，事件照常落库（seq 5、6……）；
  3. 重连（reconnect）：GET /events?since=4，服务端回放 seq > 4 的增量（可能为空：事件还没生成的话，就只等实时推送），之后实时跟进；
  4. done/error 到达，流关闭。

缺省 since=0 的意义是向后兼容：老客户端（ch15 写的）不带 since，服务端全量回放，它们的行为和以前一模一样。增量回放是 on top of 全量回放的，不是替换。

为什么不直接用 SSE 的 Last-Event-ID

细心的读者会问：SSE 规范自带 id: 字段和 Last-Event-ID 请求头，浏览器 EventSource 断线自动带上它重连，为什么还要自己发明 since？

两个原因：

  1. 我们不只在浏览器里消费。协议要同时服务 EventSource、fetch 流、curl、未来的移动端。Last-Event-ID 是 SSE 传输层的机制，非 SSE 客户端拿不到；显式 since 参数是应用层协议，任何 HTTP 客户端都能用（详情见 MDN · Using server-sent events 对 Last-Event-ID 的说明）。ch15 定的规矩，data.type 是事实来源、帧格式只是小抄，在这里再次兑现：协议语义走应用层，传输层机制只作锦上添花。
  2. 断线恢复的粒度是我们自己的。Last-Event-ID 与「连接」绑定，重连由浏览器隐式发起；since 与「会话进度」绑定，重连时机由产品决定（重试策略、指数退避、多客户端同时恢复同一条流）。我们把进度指针显式化，恢复逻辑就完全可控了。

断线重连时序图

sequenceDiagram
  participant U as 用户
  participant C as 客户端
  participant SRV as lib/server.mjs
  participant ST as event-store
  participant LOOP as lib/loop.mjs
  participant LLM as mock LLM

  U->>C: 提问
  C->>SRV: POST /chat { sessionId, text }
  SRV->>SRV: 追加 user 消息；记录 lastSeq（此时为 0）
  SRV->>LOOP: startRun（后台，fire-and-forget）
  SRV-->>C: 200 { sessionId, status, lastSeq: 0 }
  C->>SRV: GET /events?since=0
  SRV->>ST: 订阅 + 全量回放（此刻为空）

  LOOP->>LLM: callLlm（第一轮）
  LLM-->>LOOP: tool_calls ×2
  LOOP->>ST: append tool_call_start (seq 1) / tool_result (seq 2) / ×2
  ST-->>C: 实时推送 seq 1–4
  C->>U: 渲染工具中间态
  C--xSRV: 💥 断线（连接断开，客户端消失）

  Note over LOOP,ST: 客户端断线不影响 Agent：第二轮继续跑
  LOOP->>LLM: callLlm（第二轮）
  LLM-->>LOOP: 最终答案
  LOOP->>ST: append text_delta (seq 5) / done (seq 6)（无订阅者，照常入库）

  C->>SRV: GET /events?sessionId=s1&since=4（重连）
  SRV->>ST: 订阅 + 只回放 seq > 4
  ST-->>C: 增量 text_delta (seq 5) / done (seq 6)
  SRV-->>C: done 到达 → 关闭流
  C->>U: 补齐剩余回答，恢复界面

时序图里最值得盯住的是中间那条注释：断线发生在客户端，Agent 的世界里什么都没有发生。它继续调模型、继续发事件、继续落库，直到 done。重连的客户端只是「从第 4 条之后接着读」，像读一本自己折过角的书。

后台运行与中断/抢占

「后台运行」的存储侧真相

「断开后 Agent 继续跑」听起来玄，落到代码上就是一句话：append 不要求有订阅者。v02 的缓冲就已经如此（事件先落缓冲、再分发），v03 只是把这件事的意义点明：存储是运行的落点，连接只是读取进度的窗口。只要「Agent 的产物（事件）」和「客户端的消费（订阅）」在存储层解耦，后台运行就是免费的：POST /chat 一发，run 与任何连接无关地跑向完成；客户端随时可以来、走、再来，进度指针（seq）永远在存储里等着。

这对产品形态是质变：服务端才是 Agent 的主场。用户关掉页面，任务照常完成；重新打开页面，带上 since 把进度补齐。类比的成熟产品是 Claude Code 的 --resume / -c 会话恢复（恢复上次会话接着干）与 learn-claude-code s13 的「慢操作丢后台、agent 继续思考」。我们给「恢复」加上了协议层的精确定位能力（seq）。

中断与抢占：本章做到哪一步

TP15 除了「断线恢复」，还点名了「中断/抢占」。本章把这件事拆成三个层次，说清边界：

  1. 已有（v02 就有）：并发保护。同一会话正在运行时再 POST /chat，返回 409 session busy，保证同一会话的事件流不会被两轮运行交织。这是最弱的「串行化」：不打断，只拒绝插队。
  2. 本章新增：中断后不丢上下文。seq 让「客户端主动放弃一条流」（关页面、切走）变得无害。它可以随时带 since 回来，把没看完的部分读完。放弃连接 ≠ 放弃任务。
  3. 本章不做：真正的抢占/取消。要取消一个正在跑的 Agent（不等它跑完），需要取消令牌从 HTTP 层贯穿 loop、provider 直到正在进行的网络请求（AbortController 全链路），还需要定义「取消后已发事件怎么办、回滚还是保留」。这涉及权限层（谁有资格取消）与前端（打断按钮）的设计，是 ch17 Web 前端与 ch18 权限层的地盘。本章的贡献是给它打好地基：有了 seq，取消后的重连补发是免费的。被取消的 run 留下的部分事件（seq 有洞）不会破坏协议，因为客户端只依赖单调、不依赖连续。

手写实现：把序号装进产品

一、主循环盖序号：lib/loop.mjs

改动只有一行（emit）+ 一个参数（seq），见上文。要点再强调一遍：分配器注入、按会话持号、loop 无状态。

二、序号分配器：lib/events.mjs

createSeqAllocator（next/peek），上文已贴。它是纯逻辑、浏览器可测，练习 stage 1 的判题对象。

三、事件存储：lib/event-store.mjs（新增）

上文已贴核心切片。完整接口：append / events / getSince / lastSeq / subscribe / connect / listSessions。其中 listSessions 把 v02 的 GET /sessions 索引升级为带 lastSeq：

listSessions() {
  return [...sessions.entries()].map(([sessionId, s]) => ({
    sessionId,
    eventCount: s.events.length,
    lastSeq: s.events.length ? s.events[s.events.length - 1].seq ?? 0 : 0,
    lastEvent: s.events[s.events.length - 1]?.type ?? null,
  }));
}

四、服务端：lib/server.mjs

三个端点各一处变化：

POST /chat 在启动 run 之前读取 lastSeq 并返回，客户端据此只订阅本轮新事件：

session.messages.push({ role: 'user', content: text });
const lastSeq = store.lastSeq(sessionId); // 本会话事件流当前终点
startRun(session, sessionId);
res.writeHead(200, { 'content-type': 'application/json' });
res.end(JSON.stringify({ sessionId, status: 'started', lastSeq }));

startRun 把 loop 的 onEvent 接到存储上（事件已带 seq），并在 runAgent 参数里注入分配器：

function startRun(session, sessionId) {
  session.status = 'running';
  runAgent(session.messages, {
    callLlm, tools, sessionId,
    maxIterations: config.maxIterations,
    seq: allocator,                            // 每条事件盖会话内下一个序号
    onEvent: (event) => store.append(sessionId, event),
  })
    .then((result) => { session.messages = result.messages; session.status = 'done'; })
    .catch(() => { session.status = 'error'; });
}

GET /events 读 since 参数，把 ch15 的「全量回放」换成「getSince 增量回放」，顺序仍是先订阅、再回放、同一同步 tick（防竞态，见坑三）：

function handleEvents(req, res) {
  const url = new URL(req.url, `http://${req.headers.host ?? 'localhost'}`);
  const sessionId = url.searchParams.get('sessionId') || 'default';
  const since = Number(url.searchParams.get('since')) || 0; // 缺省/非法 -> 0 = 全量

  res.writeHead(200, { 'content-type': 'text/event-stream', 'cache-control': 'no-cache', connection: 'keep-alive' });

  let done = false;
  const close = () => {
    if (done) return;
    done = true;
    clearInterval(keepAlive);
    unsubscribe();
    res.end();
  };
  const unsubscribe = store.subscribe(sessionId, (event) => {
    if (done) return;
    res.write(encodeSse(event));
    if (event.type === 'done' || event.type === 'error') close();
  });
  const keepAlive = setInterval(() => res.write(': keep-alive\n\n'), KEEPALIVE_MS);
  req.on('close', close);

  for (const event of store.getSince(sessionId, since)) {  // 只回放 seq > since
    if (done) return;
    res.write(encodeSse(event));
    if (event.type === 'done' || event.type === 'error') { close(); return; }
  }
}

五、跑起来看断线重连

演示脚本 demo.mjs 完整演一遍 v03 的故事：起 mock（带 600ms 延时，让 run 活过第一个连接）→ POST /chat → 客户端 A 读到 seq 4 后主动断开 → Agent 后台跑完 → 客户端 B 带 since=4 重连只收两条增量 → 客户端 C since=0 全量回放验证一条不丢：

=== product v03 demo: 事件序号 + 断线重连 + 后台运行 ===

product server listening on http://localhost:3220

--- session s1: POST /chat { text: "帮我查一下上海的天气，顺便告诉我当前的 UTC 时间。" } ---
→ {"sessionId":"s1","status":"started","lastSeq":0}  (lastSeq = 当前事件流的终点，订阅从这里开始)

--- client A: GET /events?sessionId=s1&since=0 (SSE) ---
[1] event: tool_call_start (seq 1)
    data: {"sessionId":"s1","seq":1,"type":"tool_call_start","callId":"call_time","name":"get_time","args":{}}
[2] event: tool_result (seq 2)
    data: {"sessionId":"s1","seq":2,"type":"tool_result","callId":"call_time","ok":true,"result":"2026-08-11T13:36:54.916Z"}
[3] event: tool_call_start (seq 3)
    data: {"sessionId":"s1","seq":3,"type":"tool_call_start","callId":"call_weather","name":"get_weather","args":{"location":"Shanghai"}}
[4] event: tool_result (seq 4)
    data: {"sessionId":"s1","seq":4,"type":"tool_result","callId":"call_weather","ok":true,"result":"上海：晴，27°C，微风，紫外线中等"}
! client A 断线（模拟网络中断）：读到 seq 4 后主动断开，不等 run 结束

Agent 仍在后台跑：loop 不知道连接断了，事件照常写入 store（seq 继续递增）……

--- client B（重连，since=4）: GET /events?sessionId=s1&since=4 (SSE) ---
[1] event: text_delta (seq 5)
    data: {"sessionId":"s1","seq":5,"type":"text_delta","delta":"上海今天晴，27°C，微风，适合出门，记得防晒。"}
[2] event: done (seq 6)
    data: {"sessionId":"s1","seq":6,"type":"done","usage":{"inputTokens":42,"outputTokens":24}}
← client B（重连，since=4） stream closed (2 events received)

--- client C（完整回放，since=0）: GET /events?sessionId=s1&since=0 (SSE) ---
[1] event: tool_call_start (seq 1)
    data: {"sessionId":"s1","seq":1,"type":"tool_call_start","callId":"call_time","name":"get_time","args":{}}
[2] event: tool_result (seq 2)
    data: {"sessionId":"s1","seq":2,"type":"tool_result","callId":"call_time","ok":true,"result":"2026-08-11T13:36:54.916Z"}
[3] event: tool_call_start (seq 3)
    data: {"sessionId":"s1","seq":3,"type":"tool_call_start","callId":"call_weather","name":"get_weather","args":{"location":"Shanghai"}}
[4] event: tool_result (seq 4)
    data: {"sessionId":"s1","seq":4,"type":"tool_result","callId":"call_weather","ok":true,"result":"上海：晴，27°C，微风，紫外线中等"}
[5] event: text_delta (seq 5)
    data: {"sessionId":"s1","seq":5,"type":"text_delta","delta":"上海今天晴，27°C，微风，适合出门，记得防晒。"}
[6] event: done (seq 6)
    data: {"sessionId":"s1","seq":6,"type":"done","usage":{"inputTokens":42,"outputTokens":24}}
← client C（完整回放，since=0） stream closed (6 events received)

--- GET /sessions ---
→ {
  "sessions": [
    {
      "sessionId": "s1",
      "eventCount": 6,
      "lastSeq": 6,
      "lastEvent": "done"
    }
  ]
}

demo done.

对比 client B 与 client C 的输出，v03 的成果一目了然：B 只花了 2 条事件的带宽就恢复了完整上下文；C 证明整条事件流仍在存储里、随时可全量重建。（输出中 tool_result 的时间戳每次运行不同，其余完全确定。）

常见坑与失败模式

坑一：把「单调」当「连续」，跳号就恐慌。 seq 的契约是单调递增，不是连续无洞。一轮运行被中途取消、客户端发到一半被 409 拒掉、未来多进程场景下某条写入失败重试，都会留下「洞」（seq 1、2、3、7、8……）。客户端（和重连协议）必须只依赖「seq > since 就补发、seq 大的更新」，绝不能断言「我上一跳是 4，这次必然是 5」。ch10 里 eventId 也是同理（全局自增也可能有洞）。判断标准：把「重连时数清楚少了几条」当作客户端 bug 的根源来排查。

坑二：重复投递是协议特性，幂等是客户端责任。 网络层会重试、SSE 可能重推、客户端自己也可能把 since 传错（多记了一号，漏一条；少记一号，重一条）。所以服务端允许重复投递（since 只是尽力精确的请求，不是事务），客户端必须对事件幂等：前端按 seq 去重（同一 seq 只处理一次）、done 只在正确的状态机状态下生效。这正是 ch17 Web 前端 reducer 要做的功课：tool_result 按 callId 匹配、done 有状态守卫，都是「重复到达无害」的落实。协议层只保证「单调、有序、可寻址」，不保证「恰好一次」。恰好一次需要幂等消费，那是消费方的事。

坑三：订阅与回放之间的竞态缝隙。 ch15 的规则原样继承：先订阅、再回放，且两步在同一同步 tick 内完成。JS 单线程保证两步之间插不进任何 append（追加只发生在后续任务里），所以「订阅时刻恰好有一事件入库」既不丢（订阅先挂上了）也不重（回放按 seq 严格去重）。反过来做，先回放再订阅，或两步之间夹一个 await，就会丢或重。本章练习的 connect 正是按这个顺序实现的，product 的 GET /events 也是。

坑四：序号分配与写入不同步（多进程隐患）。 本章是单进程内存实现：next 与 push 之间没有其他代码能插队，天然安全。但一旦换多进程或多实例（未来落库），「先取号、后写入」会变成两步，两个进程可能取到同一个号。ch10 的解法在落库时才是完整的：SQLite 自增主键/事务让「取号」与「落库」原子化，或者把取号下推到存储层（INSERT ... RETURNING id）。换存储时，序号的生成位置（loop 注入 vs 存储层）是第一个要重新审视的决策。这也是练习里把序号分配收敛进存储层的原因：对外行为不变，内部所有权可以挪。

坑五：把 seq 当全局唯一。 seq 只在会话内有意义，跨会话比较两个 seq 没有意义（两个会话都有 1、2、3）。识别一条事件需要 (sessionId, seq) 二元组（ch10 里是 (session_id, event_id)）。GET /events 的路径参数恰好同时带这两个键，但日志、审计、前端去重用的 key 必须是二元组。只记 seq 的日志在混会话排查时毫无头绪。

坑六：断线恢复只做了「读」，没做「写」。 本章恢复的是事件流（读侧）。会话的消息历史（写侧）还在 server.mjs 的内存 Map 里。服务器重启，会话上下文和事件全部蒸发，since 就无从谈起（存储空了）。真正的「服务重启后恢复会话」需要 ch10 的持久层接管写侧。本章的边界是「连接断了恢复」，不是「进程死了恢复」。后者是下一层工作（TP9 持久化 + 后续章节落库），设计上 getSince/append 接口已经为它留好了位置。

小结

下一章（ch17）我们做 Web 前端 v04：浏览器消费这套 SSE 事件流，把 text_delta 渲染成打字机、把 tool_call_start/tool_result 渲染成工具卡片。它会用到本章最扎实的成果：断线重连后，前端把存储里的事件重放回来，界面就恢复了；而前端对重复事件（重连导致的重放）的幂等处理，正是坑二说的「客户端责任」。先把本章练习做完：亲手实现序号分配器、断点续传存储和后台运行语义这三层。

延伸阅读

完成阅读，去做练习 →