[!info] related notes

• 前置: SSE
• 输出模型: LLM 输出四层模型
• 后端配合: 多服务 SSE 管道
• 上层应用: AI Chat UI
• 缓存集成: TanStack Query 与 SSE 流式数据集成
• 恢复: Durable SSE Recovery with after_seq
• 取消: Cancellation

前端 SSE 消费设计

核心问题

AI 对话需要前端逐步消费文本和结构化事件，同时正确处理：

byte chunk boundaries
UTF-8 decoding
SSE frame boundaries
runtime validation
event dispatch
transport abort
recovery cursor

“能逐字显示”只是最低要求；production client 还要保证 malformed/replayed/disconnected stream 不破坏 UI state。

为什么有时用 fetch 而不是 EventSource

维度	EventSource	fetch + ReadableStream
方法	GET	支持 POST 等方法
自定义 Header	受限	可自定义，如 Authorization
SSE 解析	浏览器处理	应用手动处理
原生重连	有 EventSource 重连语义	应用自行实现
Reader/Abort 控制	较少	更灵活

AI 对话经常需要 POST body、认证 Header 和更细的 reader/recovery 控制，所以 fetch + ReadableStream 很常见。

BodySense 当前 Consultation 就采用这一形态，而不是原生 EventSource。

字节块不等于事件边界

一次 reader.read() 得到的 Uint8Array 可能：

• 包含多个 SSE frame；
• 只包含半行；
• 在一个 UTF-8 多字节字符中间切开。

因此需要增量 decoder + buffer：

const decoder = new TextDecoder();
let buffer = '';

buffer += decoder.decode(value, { stream: true });
const lines = buffer.split('\n');
buffer = lines.pop() ?? '';

{ stream: true } 让 decoder 保留末尾不完整的多字节字符，避免中文等字符跨 chunk 时被损坏。

流结束后还要处理剩余 buffer。

SSE 行与业务 Event 是两层协议

典型 frame：

event: message.text.delta
data: {"version":1,"seq":12,"type":"message.text.delta",...}

消费路径应理解成：

bytes
→ UTF-8 text
→ SSE line/frame parsing
→ JSON.parse
→ runtime StreamEvent validation
→ handler/reducer

JSON.parse 只证明 JSON 语法正确，不证明 payload 满足业务 contract。TypeScript 类型也不能在运行时自动验证网络输入。

见 BodySense StreamEvent 可信边界。

Event Dispatch

解析得到可信事件后，再按 event.type 分发：

message.text.delta
→ append current text projection

tool.call / tool.result
→ update tool projection

state.interaction.required
→ enter interrupted/pending interaction

run.cancelled / run.failed / stream.done
→ project lifecycle/terminal semantics

最好让 Live SSE 与 Durable Replay 进入同一套 handler/reducer，这样 transport 来源不会改变事件语义。

Seq / Cursor

如果 stream event 带 run-local sequence：

(run_id, seq)

客户端可以维护当前 run 已成功接受的 high-water mark：

lastAcceptedSeq

断线后请求：

after_seq=lastAcceptedSeq

从 durable event log catch up。

注意 seq 的 namespace 必须明确；同一个 Conversation 的新 Run 可以从 seq=1 重新开始，所以不能脱离 run_id 做全局去重。

AbortController 只负责 Transport/Client Async 边界

const controller = new AbortController();
const response = await fetch(url, {
  method: 'POST',
  signal: controller.signal,
});

controller.abort();

这可以取消浏览器当前 fetch/reader 工作，但：

AbortController.abort()
≠ server Run cancelled

如果服务端 Run 是 durable/detached 的，浏览器断线后它应该能够继续执行并由客户端之后恢复。

用户真正点击“停止生成”时，应走显式 server CancelRun 业务命令，再由服务端形成 durable run.cancelled。见 Cancellation。

Disconnect 后不要把 EOF 当 Done

reader.done = true

或者 reader 抛出 network error，只能说明 transport 已结束/失败。

它不自动等于：

run.completed
run.failed
run.cancelled

若没有看到 explicit durable terminal fact，客户端应该尝试 recovery/catch-up。见 Durable SSE Recovery with after_seq。

BodySense 当前对应

POST Consultation
→ fetch + ReadableStream
→ TextDecoder(stream=true)
→ SSE line parser
→ parseStreamEvent runtime validation
→ handlers
→ ActiveTurn reducer

当前主要代码：

apps/web/src/features/consultation/hooks/useSSEProcessor.ts
apps/web/src/features/consultation/hooks/useAssistantChatRuntime.ts
apps/web/src/features/consultation/runtime/durableRunRecovery.ts

其中 parser 跟踪 maxSeq，recovery 用 after_seq 获取 durable events，再通过同一 handler/reducer 投影。

常见错误

• 假设一个 reader.read() 就对应一个完整事件。
• TextDecoder 不使用 stream mode，导致跨 chunk UTF-8 字符损坏。
• JSON 能 parse 就直接 as StreamEvent，缺少运行时 contract validation。
• Live 和 Replay 各写一套事件处理逻辑。
• 只看 seq，不看 run_id namespace。
• 把 reader EOF/network error 当业务完成。
• 把 AbortController.abort() 当作 durable Run cancellation。
• 流结束时丢弃剩余 buffer。

最短记忆

ReadableStream chunk ≠ SSE frame
JSON parse ≠ runtime contract validation
Abort fetch ≠ cancel Run
EOF ≠ business terminal
(run_id, seq) + after_seq → durable recovery

学习检查

不看正文回答：

1. ReadableStream chunk、UTF-8 字符、SSE frame、业务 Event 为什么是四种不同边界？
2. 为什么 TextDecoder 在跨 chunk 场景要保持 streaming state？
3. JSON 能成功 parse 为什么还不代表业务事件可信？
4. AbortController.abort() 与 durable Run cancellation 为什么是两个不同动作？
5. EOF / network disconnect 为什么不能被当成业务完成？
6. (run_id, seq) 与 after_seq 分别解决什么恢复问题？
7. 为什么 live 和 replay 应尽量复用同一套 event reducer？
8. 给一个“中途断网后恢复 AI 输出”的场景，从 transport 到 durable event log 画出恢复路径。