跳到主要内容

fetch + ReadableStream 实战：从 a2ui-full 拆一条 AG-UI SSE 全链路

· 阅读需 62 分钟
sumshare
sumshare
coder

浏览器里那种「AI 一边想、界面一边长出来」的效果，落到代码其实只有两件事：

  • 服务端把一串事件写进一条不肯结束的 HTTP 响应；
  • 客户端从 response.body 里一块一块读出来，边读边解析、边渲染。

听起来像是一个下午就能写完的活。但真正把它接到 Agent 上，坑会集中在几个边界上：一次网络分片不等于一个事件、响应头发出之后状态码就改不了、连接关闭不等于任务完成、在错误的位置 throw 会让客户端永远等下去。这些都不是 API 会不会用的问题，而是「边界由谁负责」的问题。

这篇拿一个真实项目开刀：bigfe/encode_lesson/a2ui-full。它是一个 A2UI Playground——用户在网页里描述界面，Koa 服务端调用 LLM 生成 A2UI 协议，浏览器通过 fetch + ReadableStream 接收，边收边把界面渲染到右侧预览区。服务端用 @ag-ui/core 与 @ag-ui/encoder 组织事件，客户端手写 SSE 分帧，协议载荷则是 A2UI 的 JSONL。

我会按四条线拆：

  1. @ag-ui 是什么，它和 A2UI、MCP、A2A 各管哪一层；
  2. Node 端推 SSE 的完整步骤，从设置响应头到错误、取消、心跳；
  3. 客户端 fetch + ReadableStream 的完整步骤，从 getReader() 到分帧、解析、收尾；
  4. 这条链路里最值钱的十几个坑，其中几个是我在本机复现出来的，不是「听说」。

文中的实测脚本与官方文档原文都留档在 source/a2ui-sse-lab/。SSE 的基础格式、EventSource 与 fetch 的取舍，站内已经有两篇拆过——《SSE 是什么》与《EventSource 是什么，为什么 Agent 流式接收最后都换成了 fetch + ReadableStream》——这里不重复背景，只在需要的地方给出链接。

0. 先把结论摆出来​

如果你只想带走五句话：

  1. 服务端的最小骨架是：async generator 产出事件 → 编码成 data: JSON\n\n → Readable.from() → 交给 Koa 的 ctx.body。真正要写的不是这四行，而是围绕它的错误、取消、心跳与降级。
  2. 客户端的最小骨架是：fetch 拿到 Response → response.body.getReader() → TextDecoder({ stream: true }) → 字符串缓冲区 → 按 \n\n 切帧 → 解析 data: 行。缺任何一环都会在某个边界上烂掉。
  3. AG-UI 是事件信封，A2UI 是信封里的 payload。这个项目把 A2UI 的 JSONL 文本碎片塞进 AG-UI 的 CUSTOM 事件，既满足了 AG-UI「一帧一个完整事件」的约定，又让 A2UI 可以按任意字节边界切片。
  4. 最危险的不是「流没开始」，而是「流已经开始之后出错」。响应头一旦发出，HTTP 状态码就冻结了；此时唯一正确的报错方式是在流里发一个 RUN_ERROR，而不是 throw。
  5. 这个项目的 LLM 调用本身不是流式的。它先一次性拿到完整回答，再把 JSONL 按 4–48 个字符随机切片、间隔 20–120ms 推给客户端，制造「打字机」效果。真正的 token 级流式在另一个路由 /api/chat 里。

0.1 一张图看清链路​

0.2 先认识这几个文件​

后面会反复回到它们：

路径职责
server/a2ui-playground-server/src/routes/agent.ts/api/agent 路由：校验、选模式、设响应头、把事件流交给 Koa
server/a2ui-playground-server/src/agent/llmA2uiAgentStream.tsLLM 路径：组 messages、调模型、解析、合并、切片、发事件
server/a2ui-playground-server/src/agent/mockStream.tsMock 路径：随机挑一个场景，走同样的切片逻辑
server/a2ui-playground-server/src/routes/chat.ts/api/chat：手写 SSE 的最简版本，真·token 流式
web/a2ui-playground/src/utils/agent-sse.ts客户端核心：fetch 响应 → SSE 事件 → 回调
web/a2ui-playground/src/hooks/use-agent-chat.ts把 SSE 事件接到聊天状态、parser、UI 上
packages/a2ui-core/src/parser/index.tsJSONL 缓冲区与增量解析：任意碎片进来，能拼成消息就先渲染
source/a2ui-sse-lab/本文的复现脚本、输出日志、官方原文留档

1. 先认识 @ag-ui：Agent 和前端之间的「事件语法」​

在讲具体代码之前，得先把这个项目里反复出现的 RUN_STARTED、CUSTOM、RUN_FINISHED 是从哪来的讲清楚。它们不是 A2UI 的东西，而是 AG-UI 的东西。

1.1 它解决的问题：请求 / 响应装不下 Agent​

传统前端后端的模型是：客户端发一个请求，服务端返回一份数据，客户端渲染，交互结束。Agent 不是这样：

  • 它可能跑很久，中间不断产出中间结果；
  • 它可能同时产出文本、工具调用、状态更新、UI 指令；
  • 它可能中途需要人确认，也可能被用户取消；
  • 它的输出是不确定的，前端不能假设「一次请求一定对应一次完整回答」。

AG-UI 官方的定位是一句话：

【官方原文】"AG-UI is an open, lightweight, event-based protocol that standardizes how AI agents connect to user-facing applications." —— AG-UI Overview（抓取日期 2026-10-22，curl，HTTP 200）

关键词是 event-based：它不规定你用哪种传输，而是先规定「Agent 运行过程中会发出哪些类型的事件」。传输可以是 SSE、WebSocket、webhook，甚至是你自己的队列。

1.2 它和 A2UI、MCP、A2A 不是竞争关系​

这是最容易混的地方。官方在 Overview 里直接放了一张分层表：

层协议管什么
Agent ↔ User InteractionAG-UIAgent 怎么把运行过程、消息、状态、UI 意图实时送给前端
Agent ↔ Tools & DataMCPAgent 怎么安全地接工具、接数据源
Agent ↔ AgentA2A多个 Agent 之间怎么协作、分工

而 A2UI 又是另一件事。官方在同一页特意解释了这两个名字很像的协议：

【官方原文】"A2UI is a generative UI specification - allowing agents to deliver UI widgets, where AG-UI is the Agent↔User Interaction protocol - which connects an agentic frontend to any agentic backend." —— 同上

翻成大白话：

  • A2UI 管「UI 长什么样」：Agent 用 JSON 描述「这里放一个 Column，里面一个 Text、一个 Button」，客户端用自己的组件库渲染。它不关心这份 JSON 是怎么送到浏览器的。
  • AG-UI 管「Agent 和前端怎么说话」：它定义 run 的开始、结束、文本增量、工具调用、状态同步、自定义事件。它不关心 payload 里是不是 UI。
  • 两者可以叠加：AG-UI 负责把事件送到前端，A2UI 的 JSON 作为事件 payload 的一种。

A2UI 官方文档也把 AG-UI 列为可用传输之一：

【官方原文】"A2UI is transport-agnostic: use any method that can send JSON." —— A2UI Transports（抓取日期 2026-10-22，curl，HTTP 200）

同一页的「Available Transports」表里，AG-UI 的状态是 Stable，用途写的是 “Full stack React, Vue, Angular applications (CopilotKit)”；SSE 单独列出的状态则是 Proposed。

这里要避免一个误读：「SSE 是 Proposed」不等于「A2UI 不能走 SSE」。那张表说的是「A2UI 官方有没有为这种传输写专门章节」。A2UI 的正文明确说 transport-agnostic，只要能把 JSON 送到就行。这个项目走的是 AG-UI 的 HTTP+SSE 绑定，A2UI 消息作为 CUSTOM 事件的 value 传输——这仍然落在「任何能送 JSON 的传输」的范围内。

1.3 两个包：@ag-ui/core 与 @ag-ui/encoder​

项目里装的是 @ag-ui/core@0.0.48 与 @ag-ui/encoder@0.0.48。它们的分工很清楚：

包提供什么在本项目里怎么用
@ag-ui/core消息/状态/工具/RunAgentInput 的类型与 zod schema，以及 EventType 枚举RunAgentInputSchema.safeParse() 校验请求；EventType.RUN_STARTED 等构造事件
@ag-ui/encoderEventEncoder：按 Accept 协商 SSE / protobuf，负责把事件编码成帧encoder.encodeSSE(event) 把事件变成 data: {...}\n\n
@ag-ui/protoprotobuf 定义，供 binary 传输使用本项目不用，但 encoder 会依赖它

EventEncoder 的实现非常短，核心就一行：

encodeSSE(event) {
  return `data: ${JSON.stringify(event)}\n\n`;
}

也就是说，AG-UI 的 SSE 帧格式没有任何魔法：一个 data: 行，一段 JSON，两个换行。你完全可以手写。这个项目用 encoder，主要是为了拿类型和未来切换 protobuf 的可能性。

1.4 一次 run 的输入：RunAgentInput​

客户端 POST 的 body 不是一个随便的 JSON，而是一个 RunAgentInput。它的顶层字段（以 0.0.48 的 schema 为准）是：

字段类型作用
threadIdstring会话线程 id，多轮对话靠它关联
runIdstring本次运行 id，同一次 run 的所有事件都带它
messagesMessage[]历史消息；role 可以是 user / assistant / system / developer / tool / activity / reasoning
toolsTool[]客户端声明的工具（name / description / parameters）
contextContext[]额外的上下文片段（value / description）
stateany共享状态；本项目传空对象
forwardedPropsany协议不解释的透传字段，本项目把 A2UI 画布快照塞在这里
parentRunIdstring?分支 / 时间旅行的父 run

messages 里的内容既可以是纯字符串，也可以是 AG-UI 的多模态数组：

[
  { "type": "text", "text": "把这个卡片改成深色" },
  { "type": "binary", "mimeType": "image/png", "data": "<base64>" }
]

服务端再把 binary 转成 OpenAI 的 image_url。这段转换在 agUiContentToOpenAi.ts 里，和传输层无关，但它是「AG-UI 消息模型 → 具体模型 API」的一个典型适配层。

forwardedProps 是这个项目多轮修改的关键：

const forwardedProps = hasSurfaceInStore
  ? { a2uiCurrentProtocol: buildA2uiProtocolSnapshot(storeState) }
  : {};

浏览器把当前画布反推成一份 A2UI 协议快照，塞进 forwardedProps.a2uiCurrentProtocol。服务端拿到模型输出后，先把增量合并进这份快照，再整体下发。从传输层看，它只是 POST body 里的一个普通字段；从产品层看，它让「多轮改 UI」不必依赖聊天历史里的长 JSON。

1.5 事件模型：33 个 EventType，先记住六类​

@ag-ui/core@0.0.48 的 EventType 枚举有 33 个成员（含 5 个已废弃的 THINKING_*）。不用背，按类别记：

类别代表事件用途
生命周期RUN_STARTED、RUN_FINISHED、RUN_ERROR、STEP_STARTED、STEP_FINISHED一次 run 的边界与阶段
文本消息TEXT_MESSAGE_START、TEXT_MESSAGE_CONTENT、TEXT_MESSAGE_END、TEXT_MESSAGE_CHUNK逐 token / 逐段文本
工具调用TOOL_CALL_START、TOOL_CALL_ARGS、TOOL_CALL_END、TOOL_CALL_RESULT、TOOL_CALL_CHUNK工具调用的流式参数与结果
状态同步STATE_SNAPSHOT、STATE_DELTA、MESSAGES_SNAPSHOT快照 + 增量
活动与推理ACTIVITY_SNAPSHOT、ACTIVITY_DELTA、REASONING_*长任务进度、推理过程
特殊RAW、CUSTOM透传外部事件、应用自定义事件

其中只有 RUN_STARTED 和 RUN_FINISHED / RUN_ERROR 是强制的：一次 run 必须有开始，并且必须以「完成」或「错误」之一结束。官方 Events 文档写得很明确：

"The RunStarted and either RunFinished or RunError events are mandatory, forming the boundaries of an agent run." —— AG-UI Events（抓取日期 2026-10-22，curl，HTTP 200）

这句话直接决定了客户端怎么写：没收到终态事件，就不能把这次流当成成功。

1.6 为什么这个项目用 CUSTOM 装 A2UI​

AG-UI 的 CUSTOM 事件只有两个字段：

字段含义
name应用自定义的事件名
value应用自定义的载荷

官方对它的定位是「应用自定义语义的扩展机制」：

【官方原文】"Custom events are explicitly part of the protocol but with application-defined semantics. The name property identifies the specific custom event type, while the value property contains the associated data." —— AG-UI Events · Custom（抓取日期 2026-10-22，curl，HTTP 200）

本项目定义了三个：

namevalue用途
a2ui.jsonl.chunkstring一段 A2UI JSONL 文本碎片，可能切在 JSON 中间
a2ui.llm.rawstring模型完整原文，仅调试面板使用
a2ui.messageobject兼容旧 mock：一条 CUSTOM 携带一个完整的单键 A2UI 消息

这个设计很值得抄。它把两个层次分开了：

  • AG-UI 层：每一帧仍然是一个完整的 AG-UI 事件，满足「一帧一个事件」的约定；
  • A2UI 层：value 内部是一段可以任意切片的 JSONL 文本流，客户端自己缓冲、自己拼行。

如果反过来，把 A2UI 的半截 JSON 直接当成一个 AG-UI 事件发出去，就违反了 AG-UI 的帧契约；而如果把整条 A2UI 消息塞进一个 CUSTOM，又失去了渐进渲染。用 CUSTOM 的 value 装碎片，是这两者之间的折中。

1.7 传输绑定：HTTP + SSE 的几条硬约定​

AG-UI 的规范里，HTTP + SSE 是「default binding」。它规定得很细，几条和本文直接相关：

【官方原文】"The client sends POST to the agent endpoint. The body is the RunAgentInput as a single JSON object, UTF-8 encoded, with Content-Type: application/json."

【官方原文】"The client sends Accept: text/event-stream (adding the protobuf media type when it can consume that binding too)."

【官方原文】"Each SSE event’s data payload is exactly one protocol event as a JSON object — never more than one, never a fragment."

【官方原文】"A producer MUST frame the stream with LF (\n) line endings."

【官方原文】"A failure after the stream opens travels in-stream, as RUN_ERROR. The HTTP status is already sent and cannot change; a consumer MUST NOT infer success from 200 alone."

【官方原文】"A connection that drops without a terminal event is a truncated run."

【官方原文】"The binding has no stream resumption: SSE’s Last-Event-ID mechanism is not used, and a broken stream cannot be re-entered." —— AG-UI · HTTP + Server-Sent Events（抓取日期 2026-10-22，curl，HTTP 200）

这几条几乎就是本文后半部分的目录：

  • POST 而不是 GET → EventSource 用不了；
  • 一帧一个完整事件 → 碎片要放在 value 里，不能直接当帧；
  • LF 分帧 → 客户端别只认 \r\n，也别假设 \n\n 一定来自网络分片；
  • 200 不等于成功 → 必须等 RUN_FINISHED；
  • 没有断点续传 → 断了就是新 run，别指望 Last-Event-ID。

2. 把镜头拉近：a2ui-full 的端到端链路​

2.1 一次请求到底发生了什么​

先补一句 A2UI 的最小背景，免得后面的 JSONL 看起来像天书：A2UI v0.8 的「服务端 → 客户端」消息只有四类——beginRendering、surfaceUpdate、dataModelUpdate、deleteSurface；每条消息是一个独立 JSON 对象，流式传输时通常写成 JSONL（一行一条 JSON）。本项目只实现单 surface 和一部分组件，协议细节可以看站内的 A2UI 笔记系列。

把时序展开：

实际抓一条 ?mock=1 的流（完整 4932 字节留档在 source/a2ui-sse-lab/project-mock-stream.log），首尾是这样：

data: {"type":"RUN_STARTED","threadId":"t1","runId":"r1","input":{...}}

data: {"type":"CUSTOM","name":"a2ui.jsonl.chunk","value":"{\"beginRendering\":{\"surfaceId\":\""}

data: {"type":"CUSTOM","name":"a2ui.jsonl.chunk","value":"surface-001\",\"root\":\"card-"}

data: {"type":"CUSTOM","name":"a2ui.jsonl.chunk","value":"container\"}}\n{\"surfaceUpdate\":{\"surf"}

...

data: {"type":"RUN_FINISHED","threadId":"t1","runId":"r1","result":{"mock":"card-demo","a2uiMessageCount":2,"protocolChunkCount":44,"jsonlBytes":1126,"random":true,"streamMode":"jsonl-chunks"}}

两个数字值得注意：1126 字节的 JSONL 被切成了 44 个 chunk，平均每片约 25 个字符；value 里能看到 \n 和半截的 "card- / "surf——这正是「碎片可以切在任何位置」的直观证据。

2.2 一个容易被忽略的事实：LLM 本身不是流式的​

llmA2uiAgentStream.ts 里调用的是：

const completion = await client.chat.completions.create(requestParams);

没有 stream: true。模型一次性返回完整回答，服务端再把它切成小片。所以你在页面上看到的「打字机」，本质上是服务端人为制造的节奏，不是模型逐 token 到达。

这样做有两个现实原因：

  1. A2UI 的 JSON 必须先完整解析、合并、拆分，才能保证顺序和合法性。半截 JSON 直接往下发，客户端解析器要处理的边界会多得多。
  2. 渐进渲染的收益主要来自「组件逐个出现」，而不是「字符逐个出现」。把已经合法的 JSONL 切片，客户端每收到一个完整组件就能先画出来，视觉效果已经够了。

代价也很明确：

  • 首字节延迟包含完整 LLM 调用时间。工具循环最多 6 轮，可能几十秒。
  • 这不是真正的 token 级流式；如果模型输出很长，用户等待首屏的时间不会因为切片而缩短。
  • 切片是随机的，调试时同一段协议每次的到达节奏都不同。

项目里另一个路由 /api/chat 展示了真正的流式：chat.completions.create({ stream: true })，for await 拿到每个 delta，立刻 yield 成 SSE 帧。两条路由可以对照着看。

2.3 为什么切片大小是 4–48 个字符​

代码里：

const PROTO_CHUNK_CHARS_MIN = 4;
const PROTO_CHUNK_CHARS_MAX = 48;
const CHUNK_GAP_MS_MIN = 20;
const CHUNK_GAP_MS_MAX = 120;

这个范围不是协议要求，是产品调参：

  • 太小（比如 1 个字符）：事件数量爆炸，每个 CUSTOM 都要 JSON.stringify 一次，网络和客户端回调的开销都上去了；
  • 太大（比如整条 JSONL）：一次到达，失去了渐进渲染；
  • 4–48 个字符、20–120ms：肉眼看起来像「在写」，但又不至于把事件循环打满。

更重要的是，切片边界可以落在任何地方：JSON 字符串中间、转义字符中间、甚至 UTF-16 代理对中间。客户端的 JSONLBuffer 必须能接受这种输入。这一点后面单独讲。

2.4 同一个项目里的两条 SSE 路由​

项目里其实有两条 SSE 路径，正好代表两个抽象层级：

/api/agent/api/chat
协议AG-UI 事件自定义 { type, text }
事件类型RUN_STARTED / CUSTOM / RUN_FINISHEDstart / delta / done / error
编码EventEncoder.encodeSSE()手写模板字符串
LLM非流式，事后切片stream: true，逐 delta 转发
客户端consumeAgentSseconsumeChatSse
用途A2UI 协议生成纯模型对话 / 对照实验

/api/chat 的服务端核心只有十几行：

async function* sseLines(): AsyncGenerator<string> {
  yield `data: ${JSON.stringify({ type: 'start' })}\n\n`;
  try {
    const streamResp = await openai.chat.completions.create({
      model,
      messages: openAiMessages,
      stream: true
    });
    for await (const part of streamResp) {
      const delta = part.choices[0]?.delta?.content ?? '';
      if (delta) {
        yield `data: ${JSON.stringify({ type: 'delta', text: delta })}\n\n`;
      }
    }
    yield `data: ${JSON.stringify({ type: 'done' })}\n\n`;
  } catch (e) {
    yield `data: ${JSON.stringify({ type: 'error', message: String(e) })}\n\n`;
  }
}

ctx.body = Readable.from(sseLines(), { objectMode: false });

它没有 AG-UI 的类型系统，也没有 CUSTOM 的扩展点，但把「开流 → 逐段写 → 终态 → 错误」这条骨架演示得最干净。如果你想先跑通一个 SSE 端点，建议从这个形状开始，再逐步加协议。

一个小的工程问题：客户端的 consumeAgentSse 和 consumeChatSse 是两份几乎相同的分帧代码。传输层的「按 \n\n 切帧 + 解析 data:」应该抽成一个通用函数，业务层只做事件分派。本文 §4.9 给了一个可以复用的版本。

3. Node 端推 SSE：完整步骤​

下面按「从零写一个 SSE 端点」的顺序走。每一步都对应项目里的真实代码，最后给一个可以直接跑的 Koa 版本。

步骤 0：先定事件契约，再写代码​

在碰 res.write 之前，先回答四个问题：

  1. 一次 run 的开始事件是什么？ 这里用 RUN_STARTED。
  2. 中间载荷用什么事件？ 这里用 CUSTOM，name 是 a2ui.jsonl.chunk。
  3. 正常结束发什么？ RUN_FINISHED。
  4. 出错发什么？ RUN_ERROR，而且只能有一个终态。

把它写成一张契约表：

时机事件是否必须
流打开后立刻RUN_STARTED是
有 A2UI 碎片时CUSTOM / a2ui.jsonl.chunk0..N
调试信息CUSTOM / a2ui.llm.raw0..1
正常结束RUN_FINISHED与 RUN_ERROR 二选一
失败RUN_ERROR与 RUN_FINISHED 二选一

约定三条铁律：

  • RUN_STARTED 之前不要发业务事件；
  • 终态之后不要再发任何事件；
  • 不要让生成器「无声地结束」。没有终态的流，在客户端看来就是截断。

步骤 1：校验输入，决定模式​

项目里的做法：

const normalized = normalizeRunAgentBody(ctx.request.body);
const parsed = RunAgentInputSchema.safeParse(normalized);
if (!parsed.success) {
  ctx.status = 400;
  ctx.body = {
    error: 'Invalid RunAgentInput',
    details: parsed.error.flatten()
  };
  return;
}

这一步发生在流打开之前，所以还可以用正常的 HTTP 状态码。校验失败就 400，不要先 200 text/event-stream 再在流里报「参数错了」——那会让客户端误以为 run 已经开始。

模式选择也在这一步：

const client = getOpenAiCompatibleClient();
const useMock = useMockA2uiAgent(ctx);
const mode = client && !useMock ? 'llm' : 'mock';

没有 API Key、或者 AGENT_USE_MOCK=1、或者 URL 带 ?mock=1，就走 mock。这个降级让项目在没有密钥时也能演示完整链路，是很实用的工程决策。

步骤 2：设置响应头​

这是 SSE 端点最容易漏的一步。项目里是：

ctx.set('Content-Type', 'text/event-stream');
ctx.set('Cache-Control', 'no-cache');
ctx.set('Connection', 'keep-alive');
ctx.set('X-Accel-Buffering', 'no');

逐个解释：

响应头值为什么
Content-Typetext/event-stream客户端 / 代理识别 SSE 的依据；最好带 ; charset=utf-8
Cache-Controlno-cache不让浏览器 / 代理缓存这条长响应；更稳妥的写法是 no-cache, no-transform
Connectionkeep-aliveHTTP/1.1 下明确保持连接；HTTP/2 里这是被禁止的 hop-by-hop 头，写了也会被忽略
X-Accel-Bufferingno告诉 nginx 不要缓冲这条响应；只对 nginx 生效，但成本极低
Content-Length不要设长度未知；设了会把流变成定长响应，必须等够字节才发
Content-Encoding可设 identity避免某些中间层强行压缩；更根本的做法是让压缩中间件跳过 text/event-stream

关于压缩：SSE 是长连接、小帧、低延迟；gzip 之类的压缩中间件往往会攒一批再压，把流式退化成一次性返回——站内《EventSource 是什么，为什么 Agent 流式接收最后都换成了 fetch + ReadableStream》里实测过，开 gzip 后 50 次分片被压成了 1 次。如果你的栈里有 compression / koa-compress，要么把 text/event-stream 排除，要么确认它会逐帧 flush。这个项目没挂压缩中间件，所以没这个问题。

关于 CORS：如果前端和 API 不同源，fetch 读 SSE 仍然要过 CORS。项目用 koa-cors，默认 origin: '*'，本机玩没问题；生产环境要收窄，并且注意带 cookie 时不能用 *。

还有一个顺序问题：先设 ctx.status = 200，再 flush 头。如果你在 ctx.status 还是默认 404 的时候调用 ctx.res.flushHeaders()，客户端会先收到一个 404 响应头，然后正文照样开始流——这是一个很难查的错位。本项目没有手动 flush，靠第一个事件触发发头，所以没踩到；但如果你要「先发头、稍后发首帧」，顺序必须写对。

步骤 3：用 async generator 产出事件​

项目里所有事件源都是 AsyncGenerator<BaseEvent>：

async function* agentEvents(): AsyncGenerator<BaseEvent> {
  const core = mode === 'llm'
    ? llmA2uiAgentEventStream(input, client!)
    : mockAgentEventStream(input, loadA2uiMockJson);
  yield* traceEvents(core);
}

为什么用 generator，而不是「在路由里直接 res.write」？

  • 事件生产和传输解耦：生成器只负责「按顺序产出事件」，不关心 SSE、不关心 Koa、不关心 socket；
  • 天然背压：消费端不 next()，生成器就停在 yield，不会无限生产；
  • 错误可以统一包在生成器内部：这一点在步骤 6 会展开；
  • 可测试：你可以 for await 一个生成器，把事件收集成数组，不需要起 HTTP 服务。

项目里还有一个 traceEvents 包装器，只做计数和日志：

async function* traceEvents(source: AsyncGenerator<BaseEvent>) {
  const counts: Record<string, number> = {};
  let n = 0;
  for await (const ev of source) {
    n += 1;
    const t = ev.type ?? 'unknown';
    counts[t] = (counts[t] ?? 0) + 1;
    yield ev;
  }
  a2uiAgentInfo('stream generator finished', { eventCount: n, counts });
}

这个模式很值得保留：在事件源和编码器之间插一层「只观察、不修改」的包装，你就能在不污染业务代码的情况下拿到每类事件的数量、耗时和终态。

步骤 4：把事件编码成 SSE 帧​

AG-UI 的做法：

const encoder = new EventEncoder({ accept: 'text/event-stream' });

async function* eventStrings() {
  for await (const event of agentEvents()) {
    yield encoder.encodeSSE(event as BaseEvent);
  }
}

encodeSSE 的产出就是：

data: {"type":"RUN_STARTED","threadId":"...","runId":"..."}

data: {"type":"CUSTOM","name":"a2ui.jsonl.chunk","value":"{\"beginRen"}

每一帧以两个换行结束。JSON.stringify 会把字符串里的换行转义成 \n，所以事件 JSON 本身不会破坏帧边界；你唯一要保证的是帧与帧之间真的有 \n\n。

这里有一个项目里特意绕开的坑，值得单独说。

为什么 EventEncoder 被硬编码成 text/event-stream​

EventEncoder 会根据构造参数里的 accept 判断客户端能不能收 protobuf：

constructor(params) {
  this.acceptsProtobuf = params?.accept
    ? this.isProtobufAccepted(params.accept)
    : false;
}

getContentType() {
  if (this.acceptsProtobuf) return AGUI_MEDIA_TYPE; // application/vnd.ag-ui.event+proto
  return 'text/event-stream';
}

AG-UI 客户端经常会带：

Accept: application/vnd.ag-ui.event+proto, text/event-stream

如果你把 ctx.headers.accept 原样交给 EventEncoder，它会认为客户端接受 protobuf，getContentType() 返回 application/vnd.ag-ui.event+proto。但项目里实际调用的是 encodeSSE()——正文是文本，响应头却说自己是 protobuf。客户端按二进制解析，直接烂掉。

项目的修法是：

// 忽略客户端对 protobuf 的协商，本接口固定为 SSE（data 行内 JSON）
const encoder = new EventEncoder({ accept: 'text/event-stream' });

这不是「不支持 protobuf」的缺陷，而是一个明确的取舍：要么完整支持两种编码（getContentType() 配 encodeBinary()），要么固定一种并显式声明。最怕的是「头说 A、正文是 B」。

顺带一提：在 @ag-ui/encoder@0.0.48 里，encode() 永远返回字符串，encodeBinary() 才会根据协商结果返回 Uint8Array。如果你用 getContentType() 做协商，就必须配套用 encodeBinary()，不能继续用 encode() / encodeSSE()。

步骤 5：把字符串流交给 HTTP 响应​

Koa 的写法只有三行：

ctx.status = 200;
ctx.body = Readable.from(eventStrings(), { objectMode: false });

Readable.from() 把 async iterable 适配成 Node 的可读流，Koa 的 respond 看到 body instanceof Stream 就 body.pipe(res)。

几个细节：

  • objectMode: false：我们 yield 的是字符串 / Buffer，不是对象。对象模式也能跑，但语义上这是字节流，关掉更准确。
  • 不要设 Content-Length：长度未知，Node 会自动用 chunked。
  • 背压由 pipe 负责：客户端读得慢，res.write() 返回 false，pipe 暂停源流，Readable.from 就不会继续拉生成器。这就是为什么不要把生成器换成「一次性 JSON.stringify 整个数组再 res.end」。
  • 不要混用 ctx.body 和 ctx.res.write()：Koa 会在中间件结束后统一 respond，你手动写过的字节会和它再写一遍，轻则重复，重则报 write after end。

如果你不用 Koa，裸 Node 的等价写法是：

res.writeHead(200, {
  'Content-Type': 'text/event-stream; charset=utf-8',
  'Cache-Control': 'no-cache, no-transform',
  Connection: 'keep-alive',
  'X-Accel-Buffering': 'no'
});

for await (const frame of eventStrings()) {
  if (!res.write(frame)) {
    await new Promise((resolve) => res.once('drain', resolve));
  }
}
res.end();

注意 res.write() 的返回值：忽略它，内存会在慢客户端上越积越多。pipe 帮你做了这件事，手写循环就得自己做。

步骤 6：错误处理，分三个阶段​

SSE 的错误处理不能笼统写一个 try/catch，要按「响应头有没有发出去」分三段。

阶段 A：响应头之前​

参数错、鉴权失败、没有 API Key——这些都还能用 HTTP 状态码：

ctx.status = 400;
ctx.body = { error: 'Invalid RunAgentInput', details: ... };

或者：

ctx.status = 503;
ctx.body = { error: 'LLM 未配置：请设置 OPENAI_API_KEY' };

客户端 fetch 拿到非 2xx，读 JSON 错误体，不要把它当 SSE 解析。

阶段 B：响应头之后，但生成器还能控制​

一旦 200 text/event-stream 发出去，状态码就冻结了。此时唯一的报错方式是在流里发 RUN_ERROR，然后正常结束流：

try {
  // ... 生产事件
} catch (e) {
  yield { type: EventType.RUN_ERROR, message: `LLM 调用失败: ${msg}` };
  return;
}

项目在 LLM 调用失败、模型输出无法解析、输出里没有 A2UI 键这三种情况下都是这么做的。客户端收到 RUN_ERROR 后，把错误文案写进气泡，结束 loading。

阶段 C：生成器直接 throw​

这是最危险的情况。很多人会想：「生成器抛错，流自然就结束了吧？」——在 Koa 2.16.4 + Node 24 下，不是。

我写了一个最小复现（source/a2ui-sse-lab/koa-hang.mjs）：服务端先发两条事件，然后 throw new Error('boom-after-headers')。客户端的 reader.read() 结果是这样：

status 200 at 48 ms
read# 1 at 49 "data: {\"n\":0}\n\n"
read# 2 at 243 "data: {\"n\":1}\n\n"

  Error: boom-after-headers
    at gen (...)

read threw TimeoutError The operation was aborted due to timeout at 3002 reads 2

注意最后一行：客户端不是收到 done，也不是收到错误，而是一直挂着，直到它自己的 3 秒超时把连接掐掉。

原因在 Koa 的实现里：

  • ctx.body = stream 时，Koa 给流挂了 stream.once('error', err => ctx.onerror(err))；
  • ctx.onerror() 一进来就检查 this.headerSent || !this.writable；
  • 头已经发了，headerSent 为真，onerror 直接 return，既不销毁 res，也不 res.end()；
  • Node 的 readable.pipe(writable) 在源流出错时不会自动结束目标流。

于是源流进入了错误态，目标 res 却还开着，客户端永远等不到结束帧。

这条实测结论的价值不在于「Koa 有 bug」，而在于它给出了一个硬规则：

在 SSE 生成器里，yield 过第一个事件之后，不要再让异常冒泡出去。 所有可预期的失败都要在生成器内部捕获，转成 RUN_ERROR；所有清理放在 finally。

一个更稳的写法：

async function* eventStream() {
  let terminal = false;
  try {
    yield { type: EventType.RUN_STARTED, threadId, runId };
    // ... 业务事件
    terminal = true;
    yield { type: EventType.RUN_FINISHED, threadId, runId, result };
  } catch (e) {
    if (!terminal) {
      terminal = true;
      yield { type: EventType.RUN_ERROR, message: toMessage(e) };
    }
  } finally {
    // 关连接、清定时器、取消上游请求
  }
}

如果连 yield 本身都失败（比如客户端已经断开），那说明流已经没人读了；此时 finally 里做清理即可，不要再试图写。

阶段 D：客户端已经断开​

客户端 abort() 后，fetch 的 reader.read() 会抛 AbortError。服务端这边，Koa 在响应结束时用 onFinish 销毁 ctx.body 对应的流；Readable.from 的销毁会调用 async iterator 的 return()，于是生成器的 finally 会执行。

我用裸 Node 验证了「服务端能感知客户端断开」这件事（fetch-lab.mjs 的 /abort）：

{ path: '/abort', headerAt: 2, abortError: 'AbortError', buf: 'data: {"i":0}\n\ndata: {"i":1}\n\ndata: {"i":2}\n\n' }
[server] res close early -> abortFinally = true

但要注意：生成器的 finally 能执行，不代表上游的 LLM 请求会被取消。项目里的 client.chat.completions.create() 没有传 signal，所以用户点「停止」之后，服务端可能还在等模型返回。正确做法是把 AbortController 一路传到 OpenAI SDK：

const completion = await client.chat.completions.create(params, {
  signal: inputSignal
});

步骤 7：取消与清理​

服务端要清理的东西通常有四类：

资源例子清理位置
定时器心跳、切片间隔的 setTimeoutfinally
上游请求LLM、数据库、工具调用finally + 传 signal
队列 / 缓冲区你自己实现的异步队列finally
日志 / 指标记录本次 run 的终态finally

判断客户端是否断开，Koa 里可以监听响应的 close：

ctx.res.on('close', () => {
  // 正常结束时也会触发；用 writableEnded 区分「写完了」和「被掐断」
  if (!ctx.res.writableEnded) {
    // 请求方提前关闭了连接
  }
});

不建议用 req.on('close') 来判断：req 的 close 语义在不同 Node 版本 / 请求方法下不够稳定（它可能表示「请求体读完了」，而不是「连接断了」）；res 的 close 才直接对应响应生命周期的结束。更稳的是依赖流的 finally，因为它同时覆盖「正常结束」和「异常销毁」。两者可以都做：监听 close 设置一个 aborted 标志，生成器在每次 yield 前检查。

步骤 8：心跳与空闲超时​

SSE 是长连接，但中间层不一定有耐心。常见的空闲超时：

  • nginx proxy_read_timeout 默认 60s；
  • 云负载均衡 / API 网关常见 30s–120s；
  • 浏览器 / 移动网络也会掐掉长时间无数据的连接。

心跳就是一个注释帧：

: ping

它以一个冒号开头，没有 data: 行。SSE 规范里注释行会被忽略，项目的客户端也会因为它没有 data: 行而跳过——所以心跳不会污染业务解析。

什么时候需要心跳？

  • 两次业务事件之间的间隔可能超过 30s；
  • 尤其注意：这个项目在 RUN_STARTED 之后要等完整的 LLM 调用（可能几十秒）才发第一个 chunk。如果中间层 30s 就断，用户会看到「流莫名结束」。

项目目前没有实现心跳。它靠 RUN_STARTED 立即发出，保证首字节很快；但 LLM 调用期间的长空档仍然没有保活。生产环境建议加一个 15–30s 的 : ping\n\n。

实现上有两种做法：

  1. 手动写 res：setInterval(() => res.write(': ping\n\n'), 15000)，在 finally 里 clearInterval。简单，但要自己处理背压和「写已结束」。
  2. 把心跳并进事件流：用一个异步队列 / Promise.race，让生成器在等待业务事件时也能 yield 心跳。复杂一点，但和 Readable.from 的背压模型一致。

无论哪种，心跳都必须是独立的一帧，不能插在某个 data: 帧中间。

步骤 9：留一个非 SSE 的降级出口​

项目支持 ?sse=0（或 false / no / off / json）：

if (!useSse) {
  const events: BaseEvent[] = [];
  for await (const event of agentEvents()) {
    events.push(event);
  }
  ctx.set('Content-Type', 'application/json; charset=utf-8');
  ctx.body = { events };
  return;
}

用途：

  • 用 curl / Postman 调试时，不用处理流；
  • 写集成测试时，直接断言事件数组；
  • 某些不支持流式的客户端可以退化成一次性响应。

注意：它复用同一个事件源。这是 generator 设计的额外收益——同一套业务逻辑，既能流式发，也能收集成数组。

步骤 10：可观测性​

至少记录这些：

指标为什么
threadId / runId把一次请求的所有日志串起来
mode（llm / mock）排查「为什么没调模型」
事件总数、每类事件数判断是「没产出」还是「产出了但客户端没收到」
首个事件耗时、总耗时区分 LLM 慢和网络慢
终态（RUN_FINISHED / RUN_ERROR / 无终态）无终态是最容易被忽略的故障
发送字节数判断切片是否过碎

项目的 traceEvents 已经统计了事件数；a2uiAgentLog.ts 还会把 system prompt、模型原文、请求体落盘。后者对调试很有用，但要注意：默认把完整 prompt 和用户输入写进日志，在生产环境是隐私风险。项目用环境变量提供了关闭开关，这是对的。

一个可直接跑的 Koa SSE 端点​

把上面的步骤合起来，去掉 A2UI 业务，骨架大概是这样：

import Koa from 'koa';
import Router from 'koa-router';
import { Readable } from 'node:stream';

const app = new Koa();
const router = new Router();

// 省略：app.use(bodyParser())，否则 ctx.request.body 是 undefined

router.post('/api/agent', async (ctx) => {
  // 1. 流开始前：还能用 HTTP 状态码
  const input = ctx.request.body;
  if (!input?.threadId || !input?.runId) {
    ctx.status = 400;
    ctx.body = { error: 'threadId and runId are required' };
    return;
  }

  // 2. 响应头
  ctx.set('Content-Type', 'text/event-stream; charset=utf-8');
  ctx.set('Cache-Control', 'no-cache, no-transform');
  ctx.set('Connection', 'keep-alive');
  ctx.set('X-Accel-Buffering', 'no');
  ctx.status = 200;

  // 3. 事件源：只产出事件，不碰 res
  async function* events() {
    let terminal = false;
    try {
      yield { type: 'RUN_STARTED', threadId: input.threadId, runId: input.runId };

      for (let i = 0; i < 5; i++) {
        yield {
          type: 'CUSTOM',
          name: 'demo.chunk',
          value: `第 ${i} 段`
        };
        await new Promise((r) => setTimeout(r, 200));
      }

      terminal = true;
      yield { type: 'RUN_FINISHED', threadId: input.threadId, runId: input.runId };
    } catch (e) {
      if (!terminal) {
        terminal = true;
        yield {
          type: 'RUN_ERROR',
          message: e instanceof Error ? e.message : String(e)
        };
      }
    }
  }

  // 4. 编码 + 交给 Koa
  async function* frames() {
    for await (const ev of events()) {
      yield `data: ${JSON.stringify(ev)}\n\n`;
    }
  }

  ctx.body = Readable.from(frames(), { objectMode: false });
});

app.use(router.routes()).use(router.allowedMethods());
app.listen(3847, () => console.log('http://localhost:3847'));

这段代码没有心跳、没有上游取消、没有 protobuf；但它把「事件生产 / 编码 / 传输 / 错误 / 终态」五件事分开了。剩下的都是在这五个位置上加东西。

4. 客户端 fetch + ReadableStream：完整步骤​

步骤 1：为什么不是 EventSource​

站内《EventSource 是什么，为什么 Agent 流式接收最后都换成了 fetch + ReadableStream》已经把规范依据拆透了，这里只留结论：

需求EventSourcefetch + ReadableStream
POST + JSON body不支持（只能 GET）支持
自定义请求头（如 Authorization）不支持支持
看 HTTP 状态码不支持支持
主动取消close()，但语义有限AbortController
断线自动重连内置自己写
断点续传内置 Last-Event-ID自己写

AG-UI 的 HTTP+SSE 绑定要求 POST RunAgentInput，所以 EventSource 从第一步就不成立。保留 SSE 的帧格式，换掉读它的那只手，这就是 fetch + ReadableStream 的位置。

步骤 2：fetch 的五个关键参数​

项目里：

const res = await fetch(`/api/agent?t=${Date.now()}`, {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    Accept: 'text/event-stream'
  },
  signal: abortRef.current.signal,
  body: JSON.stringify({
    threadId: threadIdRef.current,
    runId: `run-${Date.now()}`,
    state: {},
    messages: chatMessagesToAgentApiPayload(history, { ... }),
    tools: [],
    context: [],
    forwardedProps
  })
});

逐个说：

参数作用注意
method: 'POST'携带 RunAgentInputSSE 本身不限制方法，但 AG-UI 绑定要求 POST
Content-Type: application/json告诉服务端 body 是 JSON少了它，Koa 的 bodyparser 不会解析
Accept: text/event-stream声明期望流式响应服务端应据此选 SSE；本项目固定 SSE
signal取消请求abort() 会让 reader.read() 抛 AbortError
body: JSON.stringify(...)请求体不要在这里放二进制；图片走 base64 字段

?t=${Date.now()} 是缓存破坏；POST 通常不会被缓存，但加上无害。真正要防缓存的是响应头，不是 URL。

步骤 3：先看 status 和 content-type，再决定怎么读​

fetch 在响应头到达时就 resolve，不是在 body 结束时。所以：

if (!res.ok) {
  const raw = await res.text();
  // 尝试解析 JSON 错误体
  return;
}

const ct = res.headers.get('content-type') || '';
if (!ct.includes('text/event-stream') && !ct.includes('event-stream')) {
  // 服务端退化成 JSON 了
  const raw = await res.text();
  // 按 JSON 处理
  return;
}

这段检查很值钱：

  • 服务端可能返回 400 / 503，body 是 JSON，不是 SSE；
  • 服务端可能忽略 Accept，直接返回一次性 JSON；
  • 代理可能把 Content-Type 改掉（少见，但要能诊断）。

项目里两条路径都做了这个判断：/api/agent 和 /api/chat。如果 content-type 不是流，就把整个 body 读出来，按 data: 行或 JSON 解析，并给用户一个「已按非标准响应处理」的提示。

步骤 4：getReader() + TextDecoder({ stream: true })​

const body = res.body;
if (!body) throw new Error('响应无 body');

const reader = body.getReader();
const decoder = new TextDecoder();
let buf = '';

三个点：

  1. res.body 可能为 null（比如 HEAD 请求、或者响应已经被消费）。先判空。
  2. 一个 ReadableStream 同一时间只能有一个 reader。不要在别处再 getReader()；读完之前也不要 releaseLock()。
  3. TextDecoder 必须配合 { stream: true }。

第三点单独说，因为它是「看起来能跑、偶尔烂一个字」的典型。

网络分片是按字节切的。一个 UTF-8 汉字通常 3 个字节，完全可能被切在两个分片之间。TextDecoder 默认的 decode(value) 会把不完整的尾字节当成非法序列，输出 U+FFFD（替换字符，显示为 �）；带了 { stream: true }，它会把不完整的尾字节缓存起来，等下一片到了再一起解码。

我用 fetch-lab.mjs 复现了：

-- stream:true  --
{ path: '/utf8', buf: 'data: {"text":"你好"}\n\n' }

-- stream:false --
{ path: '/utf8', buf: 'data: {"text":"��好"}\n\n' }

服务端故意把「你」的 3 个字节切成 2 + 1。正确解码还原出「你好」；默认解码得到两个 �。

流结束时，严格的写法还应该补一次 decoder.decode()，把可能残留的尾字节 flush 出来。项目的 consumeAgentSse 没有做，因为服务端保证每条事件都以完整 UTF-8 字符结束；但通用客户端应该补。

步骤 5：按空行分帧​

SSE 的帧边界是空行，不是网络分片。一个分片可能包含：

  • 半个事件；
  • 一个完整事件；
  • 十几个完整事件。

所以必须有一个跨分片的字符串缓冲区：

buf += decoder.decode(value, { stream: true });

let sep: number;
while ((sep = buf.indexOf('\n\n')) !== -1) {
  const block = buf.slice(0, sep);
  buf = buf.slice(sep + 2);
  // 处理 block
}

这段逻辑看着简单，边界不少：

情况处理
buf 里没有 \n\n继续 read()
一个分片里有多个 \n\n内层 while 全部切出来
流结束时 buf 还有残余按规范丢弃（残帧不是完整事件）
服务端用 \r\n 分帧indexOf('\n\n') 找不到，见下面的坑
心跳注释帧 : ping\n\n切出来，但没有 data: 行，跳过
空行本身block 为空，跳过

第 4 条是这个项目客户端的一个真实缺陷。代码注释里写了「若代理改写成 \r\n，\r 会留在行尾，由 slice(5).trim() 去掉」——但那只处理了 data: 行尾的 \r，帧分隔符本身 \r\n\r\n 里并没有连续的 \n\n，indexOf('\n\n') 会一直返回 -1，整个流一个事件都切不出来。

验证很简单：

'\r\n\r\n'.indexOf('\n\n') // -1

AG-UI 规范要求 producer 用 LF，所以这个项目自己的服务端不会触发；但只要有中间层改写行尾，客户端就会静默卡住。通用写法是先做一次归一化：

buf = buf.replace(/\r\n/g, '\n');

或者在切帧时同时找两种分隔符。代价很小，收益是少一个「线上偶发卡死」。

步骤 6：解析 data: 行与 JSON​

一个 SSE 事件块由若干 field: value 行组成：

event: message
id: 42
data: {"type":"CUSTOM","name":"a2ui.jsonl.chunk","value":"..."}
: comment

项目只关心 data:，而且是第一条：

const line = block.split('\n').find((l) => l.startsWith('data:'));
if (!line) continue;

const json = line.slice(5).trim();
if (!json) continue;

let ev;
try {
  ev = JSON.parse(json);
} catch {
  continue;
}

这里有三处简化，通用客户端要知道：

  1. 多行 data:：SSE 规范允许多个 data: 行，客户端应该用 \n 把它们拼起来。项目只取第一行。它的服务端每个事件只写一行 data:，所以没问题；换一个服务端就可能丢内容。
  2. event: / id: / retry:：项目全部忽略。AG-UI 的绑定规定 consumer 可以忽略除 data 外的字段，所以这是合规的。但如果你要支持 EventSource 式的命名事件或续传，就得解析它们。
  3. JSON 解析失败静默跳过：这是有意的取舍——宁可少渲染一条，也不让整个页面崩。但必须记录指标，否则你会不知道「为什么少了一块 UI」。

步骤 7：把事件分派给业务​

项目在 consumeAgentSse 里做的是「通用 SSE 解析 + AG-UI 事件分派」，不碰 A2UI：

if (t === 'CUSTOM') {
  const cname = ev.name;
  if (cname === A2UI_LLM_RAW_NAME && typeof ev.value === 'string') {
    options?.onLlmRaw?.(ev.value);
  }
  const w = a2uiCustomEventToWriteString(ev);
  if (w !== null) onWrite(w);
}

if (t === 'RUN_FINISHED') {
  finishedSummary = `Agent 完成：${JSON.stringify(ev.result)}`;
}

if (t === 'RUN_ERROR') {
  runError = typeof ev.message === 'string' ? ev.message : 'RUN_ERROR';
}

a2uiCustomEventToWriteString 是这一层的核心适配器：

export function a2uiCustomEventToWriteString(ev) {
  if (ev.type !== 'CUSTOM') return null;

  if (ev.name === 'a2ui.jsonl.chunk' && typeof ev.value === 'string') {
    return ev.value; // 原样透传，不能补 \n
  }

  if (ev.name === 'a2ui.message' && ev.value != null) {
    return `${JSON.stringify(ev.value)}\n`; // 对象要序列化 + 补行终止符
  }

  return null;
}

两条分支不对称，原因值得记住：

  • a2ui.jsonl.chunk 的 value 是已经序列化好的 JSONL 文本碎片，换行符在字符串内部，服务端已经按协议切好片。再加 \n 会凭空插入空行，污染调试副本，也可能把恰好切在行中间的碎片改坏。
  • a2ui.message 的 value 是对象，必须 JSON.stringify 才能变成文本；而 JSON.stringify 只产出 {...}，不带任何行终止符，所以要补 \n。

一句话：序列化负责变成文本，换行负责分帧，两者不能混。JSONL 的行终止符不是排版，是记录边界。

步骤 8：流结束后的收尾​

consumeAgentSse 返回，只代表网络流结束了。此时业务层还有两件事没做：

  1. 解析器缓冲里可能还有未闭合的行 / 未闭合的组件；
  2. 渲染节流窗口里可能还有一次 treeBuild 没执行。

所以项目的调用方在 await consumeAgentSse(...) 之后必须：

a2uiParser.endStream();
a2uiParser.flushPendingRender();
setStoreState(createdStore.getState());

这三行的顺序不能乱：

  • endStream() 把缓冲区里剩下的内容做最后一次解析（如果能解析）；
  • flushPendingRender() 立即执行被 400ms 节流压住的那次渲染；
  • 最后把 store 快照同步给 React state，让调试面板和预览区拿到最终状态。

漏掉 flushPendingRender() 的典型症状是：最后一个组件要等到下一次用户操作才出现。漏掉 endStream() 的典型症状是：最后一条没有换行结尾的消息永远不渲染。

步骤 9：取消、超时与截断​

客户端有三种「没正常结束」：

情况表现应该怎么处理
用户主动取消reader.read() 抛 AbortError显示「已取消」，保留已收到的部分
网络断开 / 服务端崩溃read() 抛 TypeError / terminated显示「连接中断」，标记为不完整
服务端正常关闭但没发终态read() 返回 done: true按截断处理，不要当成功

项目对第一种有明确处理：

catch (e) {
  if ((e as Error).name === 'AbortError') {
    // 保留 agentJsonlAccumRef / agentLlmRawRef，气泡显示「已取消」
    return;
  }
  // 其他错误显示「客户端错误：...」
}

对第三种，它有一个兜底文案：

const assistantText = runError ?? finishedSummary ?? '流结束（未收到 RUN_FINISHED）。';

但从产品语义上，这还应该被标成 不完整，而不是普通完成。AG-UI 规范说得很直接：

"A connection that drops without a terminal event is a truncated run."

一个更严谨的客户端会记录：

const complete = Boolean(finishedSummary) && !runError;

然后把 complete 传给 UI：不完整的流可以保留已渲染的界面，但要提示用户「本次生成未完成，可重试」。

超时也建议分两层：

  • 首字节超时：fetch 发出后 N 秒还没收到响应头，直接放弃；
  • 空闲超时：两个分片之间超过 N 秒没有数据，放弃。

AbortSignal.timeout() 只能设总时长，不适合长流；空闲超时需要每次 read() 用 Promise.race 包一个定时器。很多线上「流卡住」不是断了，而是某一端在等一个永远不会来的字节。

一个可复用的通用 SSE 读取器​

把项目里的实现抽掉 A2UI 业务，骨架如下：

export async function consumeSse(
  response: Response,
  onEvent: (ev: unknown) => void,
  options: { signal?: AbortSignal } = {}
): Promise<{ sawTerminal: boolean; error?: string }> {
  const body = response.body;
  if (!body) throw new Error('响应无 body');

  const reader = body.getReader();
  const decoder = new TextDecoder();
  let buf = '';
  let sawTerminal = false;
  let error: string | undefined;

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    if (!value) continue;

    // 兼容被中间层改写成 CRLF 的情况。
    // 注意先拼接、再归一化：否则切在 \r 和 \n 之间的分片会漏掉。
    buf += decoder.decode(value, { stream: true });
    buf = buf.replace(/\r\n/g, '\n');

    let sep: number;
    while ((sep = buf.indexOf('\n\n')) !== -1) {
      const block = buf.slice(0, sep);
      buf = buf.slice(sep + 2);

      // 多行 data: 按规范用 \n 拼接
      const dataLines = block
        .split('\n')
        .filter((l) => l.startsWith('data:'))
        .map((l) => l.slice(5).replace(/^ /, ''));

      if (dataLines.length === 0) continue;

      const json = dataLines.join('\n').trim();
      if (!json) continue;

      let ev: any;
      try {
        ev = JSON.parse(json);
      } catch {
        continue; // 单条坏事件不影响整条流
      }

      if (ev?.type === 'RUN_FINISHED') sawTerminal = true;
      if (ev?.type === 'RUN_ERROR') {
        sawTerminal = true;
        error = typeof ev.message === 'string' ? ev.message : 'RUN_ERROR';
      }

      onEvent(ev);
    }
  }

  // 流结束：残余 buf 按规范丢弃
  return { sawTerminal, error };
}

调用方：

const res = await fetch('/api/agent', { method: 'POST', ... });
if (!res.ok) throw new Error(await res.text());

const { sawTerminal, error } = await consumeSse(res, (ev) => {
  // 分派
});

if (!sawTerminal) {
  // 截断：不要当成功
}

这段代码不依赖 A2UI，也不依赖 AG-UI，可以放进任何 SSE 客户端。

5. 这条链路里最值钱的坑​

下面这些坑按「分帧 / 传输 / 生命周期」分组。每条尽量给：现象 → 原因 → 做法。

5.1 分帧与编码​

坑 1：把网络分片当成事件​

现象：本地跑得好好的，线上偶尔 JSON 解析失败、UI 少一块。

原因：reader.read() 返回的是网络分片，和 SSE 事件边界没有任何对应关系。一个分片可能是半个事件，也可能是十个事件。

做法：永远维护字符串缓冲区，按 \n\n 切；切不出来的留在 buf 里等下一片。不要对每个 value 直接 JSON.parse。

坑 2：TextDecoder 不带 stream: true​

现象：中文偶尔变成 �，英文和数字永远正常。

原因：一个 UTF-8 汉字 3 个字节，被网络分片切在中间；默认 decode 会把不完整的尾字节判为非法。

做法：decoder.decode(value, { stream: true })；流结束时再 decoder.decode() flush 一次。

这条我在本机复现过：同一个被切成 2 + 1 的「你」，stream: true 得到「你好」，默认得到 ��好。

坑 3：按 UTF-16 下标切片，切开了代理对​

现象：服务端按字符数切片，某些 emoji 在日志里变成半个字符；客户端最终 JSON 却又能解析。

原因：jsonl.slice(pos, pos + size) 按 UTF-16 code unit 切，一个 emoji 可能是两个 code unit（代理对）。切在中间会得到两个「半个代理」。

做法：不要假设每个 chunk 都是合法的独立字符串 / JSON。客户端要把 chunk 当作不透明字节流，先拼进缓冲区再解析。本项目正是这么做的：a2ui.jsonl.chunk 的 value 原样写入 JSONLBuffer，不单独 JSON.parse。如果服务端要在日志里打印每个 chunk，先做一次 Buffer.from(chunk).toString('utf8') 也救不了半个代理——最好只打印长度或转义后的形式。

坑 4：\n\n 与 \r\n\r\n​

现象：某个代理 / 网关后面，客户端一个事件都收不到，连接却一直开着。

原因：客户端只找 \n\n，而中间层把行尾改成了 CRLF，\r\n\r\n 里没有连续的 \n\n。

做法：入缓冲区后先归一化 \r\n 为 \n，或者切帧时同时匹配两种分隔符。AG-UI 规范要求 producer 用 LF，但 consumer 最好宽容一点。

坑 5：给 JSONL 碎片补 \n​

现象：调试面板里出现空行；偶尔某个组件不渲染。

原因：a2ui.jsonl.chunk 的 value 已经是带换行的 JSONL 文本，再补 \n 会插入空行；空行不是合法 JSON 行，解析器要么跳过，要么污染缓冲区。

做法：区分两种 CUSTOM：

  • a2ui.jsonl.chunk（字符串碎片）→ 原样透传；
  • a2ui.message（对象）→ JSON.stringify 后补 \n。

坑 6：EventEncoder 协商成 protobuf，正文却写 SSE​

现象：某些 AG-UI 客户端（会带 Accept: application/vnd.ag-ui.event+proto）收到乱码 / 解析失败；curl 却正常。

原因：getContentType() 根据 Accept 返回了 protobuf 媒体类型，但代码调用的是 encodeSSE()，正文是文本。

做法：二选一：

  • 固定 SSE：new EventEncoder({ accept: 'text/event-stream' })，并且响应头写 text/event-stream；
  • 完整协商：getContentType() 配 encodeBinary()，用 Buffer 写响应。

最忌讳「头说 A、正文是 B」。

5.2 传输与部署​

坑 7：响应头要等第一个 chunk 才发​

现象：服务端明明已经进入路由，客户端 fetch 却迟迟不 resolve；监控上「首字节时间」等于「第一个业务事件时间」。

原因：Node 的 HTTP 响应头通常在第一次 write() 时才真正发出。如果生成器先等 1.5 秒才 yield 第一个事件，fetch 就要等 1.5 秒才 resolve。

我实测（fetch-lab.mjs）：

不 flush：headerAt 1546ms，firstByteAt 1547ms
先 flush：headerAt 4ms，firstByteAt 1551ms

做法：

  • 流打开后立刻发一个 RUN_STARTED（本项目就是这么做的）；
  • 或者先 ctx.status = 200，再 ctx.res.flushHeaders()，让客户端先拿到响应头。

注意顺序：先设 status，再 flush。反了会把默认 404 发出去。

坑 8：代理缓冲 / 压缩把流攒成一次性​

现象：本地逐字出现，部署到 nginx / 网关后「转圈很久，然后整段出现」。

原因：中间层开了响应缓冲或压缩，攒够一批才往下发。

做法：

  • X-Accel-Buffering: no（nginx）；
  • Cache-Control: no-cache, no-transform；
  • 压缩中间件排除 text/event-stream，或确认它逐帧 flush；
  • 检查网关的缓冲配置和空闲超时。

坑 9：HTTP/1.1 的 6 连接上限​

现象：开多个标签页 / 多个组件同时连 SSE，后面的请求一直 pending。

原因：浏览器对同一 origin 的 HTTP/1.1 并发连接通常限制在 6 条；每条 SSE 长期占用一条。

做法：生产用 HTTP/2（多路复用）；或者合并流、限制同时打开的流数量。这个项目是本地 Playground，不受影响，但做产品要想。

坑 10：客户端断开后，上游还在跑​

现象：用户点了「停止」，界面不更新了，但服务端日志还在打 LLM 请求 / 工具调用。

原因：客户端的 AbortController 只取消了浏览器到服务端的连接；服务端到 LLM 的请求是另一条链路，没有传 signal。

做法：把取消信号一路传下去：

await client.chat.completions.create(params, { signal });

并在生成器的 finally 里清理定时器、队列、数据库连接。

5.3 生命周期与错误​

坑 11：头发出后 throw，客户端永久挂起​

现象：流前面几条正常，然后客户端一直转圈，既不结束也不报错；服务端日志里有异常。

原因：Koa 的 onerror 在 headerSent 时直接返回，不结束 res；pipe 也不会因为源流出错而结束目标流。

做法：生成器内部捕获所有异常，转成 RUN_ERROR；yield 过第一个事件之后，不要让异常冒泡。清理放 finally。

这是本文最值得记住的一条，因为它不是「概率问题」，而是一旦发生就必然挂到超时。

坑 12：把 200 当成功​

现象：HTTP 200，客户端显示「完成」，但其实只收到半截 UI。

原因：SSE 的响应头在第一个字节就发出去了；后续失败只能通过流内事件表达。fetch 的 res.ok 只代表「流打开了」，不代表「run 成功了」。

做法：以 RUN_FINISHED / RUN_ERROR 为准；没有终态就是截断。UI 上要区分「完成」「失败」「不完整」三种状态。

坑 13：断线后自动重连，重复执行 run​

现象：网络抖动后，同一个问题被回答了两次；或者 UI 出现两套组件。

原因：EventSource 会自动重连；即使你用 fetch，也可能在业务层写了「失败重试」。AG-UI 的 HTTP+SSE 绑定没有断点续传：

"The binding has no stream resumption... Re-running is a new run with a new runId."

做法：重连 = 新的 run；不要复用 runId 假装续传。如果业务需要幂等，用 threadId + 客户端生成的请求 id 去重。

坑 14：心跳帧被业务解析器误判​

现象：加了心跳后，客户端偶尔报「JSON 解析失败」。

原因：心跳写成了 data: ping\n\n，而不是注释帧 : ping\n\n。

做法：心跳用冒号开头的注释行；客户端切到没有 data: 的块就跳过。

坑 15：慢客户端把内存打爆​

现象：服务端内存随连接数上涨；GC 频繁。

原因：生成器产出速度远大于网络发送速度，而代码忽略了 res.write() 的返回值 / 没用 pipe 的背压。

做法：用 Readable.from() + pipe，或者手写 if (!res.write(frame)) await once(res, 'drain')。同时限制单条流的最大缓冲、最大事件数、最大时长。

坑 16：调试副本无限增长​

现象：长时间流式会话后，页面变卡，内存上涨。

原因：客户端把每个 chunk 都累加到 agentJsonlAccumRef，为了「View A2UI JSON」调试面板保留全文。协议越大，副本越大。

做法：调试副本设上限（比如保留前 1MB + 后 1MB），或者只在调试开关打开时累积。生产环境不要默认保留完整模型原文。

5.4 调试 SSE 的四个手法​

SSE 的调试体验比普通接口差，因为它「不结束」。下面几个手法可以少走弯路。

① curl -N 看原始帧

curl -N -X POST http://localhost:3847/api/agent \
  -H 'Content-Type: application/json' \
  -H 'Accept: text/event-stream' \
  -d '{"threadId":"t1","runId":"r1","messages":[],"tools":[],"context":[]}'

-N 关掉 curl 的输出缓冲，你能看到帧到达的真实节奏。如果所有帧「唰」地一起出现，优先怀疑代理 / 压缩，而不是客户端。

② 浏览器的 Network → EventStream

Chrome DevTools 的 Network 面板对 text/event-stream 有专门的 EventStream 视图，能看到每个事件的到达时间和 data。它比 console.log 更接近线上，因为经过了真实的浏览器网络栈。

③ 服务端计数

在 traceEvents 里按类型计数，结束时打一条摘要：

eventCount: 87, counts: { RUN_STARTED: 1, CUSTOM: 85, RUN_FINISHED: 1 }

如果客户端说「没收到」，而服务端计数正常，问题在传输；如果服务端计数里没有 RUN_FINISHED，问题在生成器。

④ 非流式降级对照

同一个事件源，用 ?sse=0 收成数组：

curl -s -X POST 'http://localhost:3847/api/agent?sse=0' ...

如果 JSON 里事件齐全、SSE 却缺，说明分帧 / 传输层有问题；如果 JSON 里也缺，说明业务事件源有问题。把「业务」和「传输」分开验证，比盯着浏览器一行行看快得多。

6. 最小可复现：不用 Koa 也能跑​

如果你想亲手验证「分片不等于事件」「UTF-8 被切半」「错误发生在头之后」，可以不装任何依赖，用 Node 内置的 http 写一个 60 行的服务端。

6.1 服务端​

import http from 'node:http';

const server = http.createServer(async (req, res) => {
  if (req.url !== '/sse') {
    res.writeHead(404).end();
    return;
  }

  res.writeHead(200, {
    'Content-Type': 'text/event-stream; charset=utf-8',
    'Cache-Control': 'no-cache, no-transform',
    Connection: 'keep-alive',
    'X-Accel-Buffering': 'no'
  });

  let closed = false;
  res.on('close', () => {
    if (!res.writableEnded) {
      closed = true;
      console.log('[server] client closed early');
    }
  });

  try {
    for (let i = 0; i < 5; i++) {
      if (closed) break;

      const ev = {
        type: i === 0 ? 'RUN_STARTED' : 'CUSTOM',
        name: i === 0 ? undefined : 'demo.chunk',
        value: i === 0 ? undefined : `第 ${i} 段：你好`
      };

      const frame = `data: ${JSON.stringify(ev)}\n\n`;

      if (!res.write(frame)) {
        await new Promise((resolve) => res.once('drain', resolve));
      }

      await new Promise((r) => setTimeout(r, 300));
    }

    if (!closed) {
      res.write(`data: ${JSON.stringify({ type: 'RUN_FINISHED' })}\n\n`);
    }
  } catch (e) {
    // 注意：头已经发了，不能再改状态码
    if (!closed) {
      res.write(`data: ${JSON.stringify({
        type: 'RUN_ERROR',
        message: e instanceof Error ? e.message : String(e)
      })}\n\n`);
    }
  } finally {
    res.end();
  }
});

server.listen(3906, () => console.log('http://localhost:3906/sse'));

6.2 客户端​

const res = await fetch('http://localhost:3906/sse', {
  headers: { Accept: 'text/event-stream' }
});

const reader = res.body.getReader();
const decoder = new TextDecoder();
let buf = '';

while (true) {
  const { done, value } = await reader.read();
  if (done) break;

  buf += decoder.decode(value, { stream: true });

  let sep;
  while ((sep = buf.indexOf('\n\n')) !== -1) {
    const block = buf.slice(0, sep);
    buf = buf.slice(sep + 2);

    const line = block.split('\n').find((l) => l.startsWith('data:'));
    if (!line) continue;

    const json = line.slice(5).trim();
    if (!json) continue;

    try {
      console.log(JSON.parse(json));
    } catch {
      console.warn('bad event', json);
    }
  }
}

6.3 你可以自己改的几个实验​

改动观察什么
把服务端 res.write(frame) 改成一次写半帧客户端不会立刻解析，等下一片补全
把 res.writeHead 里的 Content-Type 改成 application/json客户端仍能收，但你的检查逻辑应该走降级
在第一个 yield 后 throw对比 Koa 版本：裸 Node 里 res.end() 在 finally 会结束；Koa 里不会
去掉 { stream: true }中文偶发 �
客户端 450ms abort()服务端 res 的 close 事件触发（writableEnded 为 false）

7. 为什么 A2UI 能「边收边渲染」​

传输层只负责把字节送到。真正让界面「长出来」的，是 packages/a2ui-core/src/parser/index.ts 里的 JSONLBuffer。

7.1 正常路径：按行切​

JSONLBuffer.write() 把新数据追加到内部字符串，然后：

const lines = trimmedBuffer.split('\n');
if (lines.length > 1) {
  // 除最后一行外，都当作完整行立即解析
  for (let i = 0; i < lines.length - 1; i++) {
    this.processLine(lines[i]);
  }
  // 最后一行留作「可能不完整」的缓冲
  this.buffer = lines[lines.length - 1];
}

这就是 JSONL 的标准处理：换行是记录边界。服务端保证每条 A2UI 消息以 \n 结尾，客户端就能逐行解析。

7.2 加速路径：半条 surfaceUpdate 也能先渲染组件​

如果缓冲区里只有一行，而且还没闭合，它会先试 isCompleteJSON()（花括号计数 + 字符串状态扫描）；如果还不完整，但包含 "surfaceUpdate":，就进入 extractSurfaceIdAndComponents()：

  1. 正则取出 surfaceId；
  2. 从 "components":[ 之后做大括号匹配，跳过字符串和转义；
  3. 每提取到一个完整的 component，就立刻回调一次 surfaceUpdate，只带这一个组件；
  4. 把这个组件从缓冲区里移除，继续找下一个。

效果是：整条 surfaceUpdate 还没收完，已经完整的组件可以先画出来。这是这个项目「渐进渲染」观感的关键。

代价也很明显：

  • 只覆盖 surfaceUpdate，其他消息必须等完整 JSON；
  • 大括号匹配是手写的，边界情况（字符串里的 {、转义、嵌套）靠状态机兜；
  • JSON.parse 失败被静默忽略（parser/index.ts:278），语法错误会导致该组件被跳过，但不会中断流；
  • 没有 schema 校验，结构错了也能进渲染层。

这些是「教学原型」的合理取舍，不是通用 SDK 的做法。真要做生产客户端，应该用 schema 校验 + 增量 JSON 解析器，而不是手写括号匹配。

7.3 渲染节流：400ms​

即使传输层每 20–120ms 到一个 chunk，也不应该每来一个 chunk 就重建一次 React 树。项目的 parser 用 400ms 节流：

  • 距上次渲染不足 400ms：只安排一次延迟渲染；
  • 流结束或本地数据更新：调用 flushPendingRender() 立即刷新。

这个数字是体感调参：太小会抖，太大会显得卡。它和传输层的切片节奏是两套独立的节流，不要混为一谈。传输层决定「数据多快到达」，渲染层决定「界面多快更新」。

7.4 一个需要知道的协议顺序问题​

服务端 splitCombinedA2uiMessage() 拆分的顺序是：

['beginRendering', 'surfaceUpdate', 'dataModelUpdate', 'deleteSurface']

这个顺序是为这个客户端量身定做的：它的 parser 收到 beginRendering 时会清空 hydrateNodes，然后等后续 surfaceUpdate 建树。所以必须先发 beginRendering。

但 A2UI v0.8 规范的要求是反过来的：客户端应该先缓冲 surfaceUpdate / dataModelUpdate，等 beginRendering 到达后再首次渲染。项目自己的 CHECK.md 把这一点列为 P0：如果接一个按规范顺序（组件先、beginRendering 后）的服务端，这个客户端会崩。

也就是说：传输层没问题，协议时序是项目自己的耦合。如果你要复用这套 SSE 代码，别把「先 beginRendering」当成 A2UI 的通用规则。

8. 生产清单​

把上面的内容压成一张表，部署前逐项过：

服务端​

检查项说明
路由先校验再开流参数错用 4xx JSON，不要先 200 再报错
Content-Type: text/event-stream带 charset
Cache-Control: no-cache, no-transform防缓存 / 防转换
X-Accel-Buffering: nonginx 不缓冲
不设 Content-Length让 Node 用 chunked
压缩中间件排除 SSE或确认逐帧 flush
首个事件立即发RUN_STARTED 或 flushHeaders()
事件契约有终态RUN_FINISHED / RUN_ERROR 二选一
生成器内部捕获异常头发出后不要 throw
客户端断开能清理finally + 上游 signal
心跳 15–30s注释帧 : ping
背压Readable.from / drain
非 SSE 降级?sse=0 返回事件数组
指标事件数、首字节、总耗时、终态
日志脱敏不要默认记录完整 prompt / 用户输入

客户端​

检查项说明
检查 res.ok非 2xx 读 JSON 错误体
检查 content-type不是流就走降级
res.body 判空可能为 null
TextDecoder({ stream: true })防止多字节字符被切
跨分片缓冲区按 \n\n 切帧
兼容 CRLF归一化或双分隔符
多行 data:按规范用 \n 拼接
JSON 解析失败不中断但要计数 / 打日志
区分三种结束完成 / 失败 / 截断
AbortController用户取消要能立刻生效
首字节 / 空闲超时防止永久挂起
流结束收尾endStream() + flushPendingRender()
调试副本限长防止内存无限增长
渲染节流不要每个 chunk 重建一次树

9. 参考资料与实测环境​

9.1 一手来源（抓取日期 2026-10-22，curl，HTTP 200）​

资料URL用途
AG-UI Overviewdocs.ag-ui.com/introductionAG-UI 定位、与 A2UI/MCP/A2A 的分层
AG-UI Eventsdocs.ag-ui.com/concepts/events事件类别、CUSTOM、生命周期强制事件
AG-UI HTTP + SSEdocs.ag-ui.com/spec/1.0/basic/transports/http-ssePOST、一帧一事件、LF、错误语义、无续传
AG-UI Architecturedocs.ag-ui.com/concepts/architecture传输无关、HttpAgent
A2UI Transportsa2ui.org/concepts/transportstransport-agnostic、AG-UI 状态
A2UI Data Flowa2ui.org/concepts/data-flowJSONL 消息格式

原文留档在 source/a2ui-sse-lab/official/，逐字引用与抓取命令见 source/a2ui-sse-lab/official-sources.md。

9.2 项目与实测​

  • 项目：/Users/aster/projj/gitlab.sumshare.cn/10022/bigfe/encode_lesson/a2ui-full
  • 代码基线：8afaa38（2026-10-22）
  • 实测环境：Node v24.14.1、macOS；Koa 2.16.4（取自项目 pnpm store）
  • 脚本与日志：source/a2ui-sse-lab/
    • fetch-lab.mjs / fetch-lab.log：响应头时机、UTF-8 切半、abort
    • koa-hang.mjs / koa-hang.log：头发出后 throw 导致客户端挂起
    • README.md：跑法与结果表

9.3 局限​

  • 全部实测在本机回环、单进程完成，没有经过 nginx / CDN / 真实浏览器；代理缓冲、连接上限、空闲超时这些结论来自实践共识，不是本文实测。
  • Koa 挂起结论只覆盖 koa@2.16.4 + Node v24.14.1 + ctx.body = Readable.from(asyncGenerator) 这一组合；不同 Koa 版本、不同中间件可能有不同行为。
  • 「不 flush 时 fetch 要等首字节」是 Node HTTP/1.1 服务端 + undici fetch 的行为；不同运行时可能不同。
  • 项目使用 A2UI v0.8（官方当前生产版本是 v0.9.1），协议版本的细节不适用于 v0.9；但本文讨论的传输层、SSE 分帧、错误语义与版本无关。
  • 引用的 AG-UI 文档是官网当前页面（含 1.0 规范的 HTTP+SSE 绑定页），项目安装的是 @ag-ui/core@0.0.48。两者在本文涉及的事件类型、CUSTOM 语义、SSE 帧格式上一致，但版本号并非严格对应；引用时已注明页面与抓取日期。
  • 项目自身的 CHECK.md 已指出多 surface、beginRendering 时序、模板子树、标准 userAction 等协议层问题；本文只讨论传输，不重复评估。

10. 最后一句​

把这条链路拆完，会发现真正难的从来不是「怎么 fetch」「怎么 getReader」——这些 API 半小时就能学会。难的是边界的所有权：

  • 分片边界归客户端缓冲区；
  • 帧边界归 SSE 的 \n\n；
  • 事件边界归 AG-UI 的事件；
  • 消息边界归 A2UI 的 JSONL 行；
  • 生命周期边界归 RUN_STARTED / RUN_FINISHED / RUN_ERROR。

每一层都只做自己那一小段，并且不假设下一层会替自己兜底。这条链路就稳了。