5.4 流式事件如何传播到界面
本页分析版本earendil-works/pi@c13ffe12026-07-30本章解决什么问题：模型吐出的第一个字，要经过几次「换装」才变成你终端里闪烁的那几个字符？为什么 Pi 需要三套事件类型而不是一套？ 前置知识：2.6 异步迭代器与 for await、3.3 流式输出、5.2 一次普通请求的完整路径。 学习目标：读完后你能 ① 说出 Pi 三层事件体系各自的名字、种类数与所属 package；② 指出文本增量藏在哪一层的哪个字段里；③ 画出一条
text_delta从 SSE 字节到屏幕字符的完整调用链；④ 解释 TUI 与--mode json这两个消费端为什么消费同一份事件却长得完全不同；⑤ 自己用测试验证事件顺序。
建立直觉：三个信封，而不是一根管子
初学者容易把流式输出（Streaming）想象成一根笔直的水管：模型那边流出字符，终端这边显示字符。但 Pi 的实际结构更像三层信封——同一份内容被拆开、重新装进更大的信封、再交给下一层：
- 最里层由
@earendil-works/pi-ai定义，只关心「一条 assistant 消息是怎么一个字一个字长出来的」。它不知道什么是工具执行，也不知道什么是会话（Session）。 - 中间层由
@earendil-works/pi-agent-core定义，只关心「一次 Agent Loop（Agent 循环）里发生了哪些事」——开始、一个 turn、一条消息、一次工具执行。它把里层的整封事件原样塞进自己的一个字段带走。 - 最外层由
@earendil-works/pi-coding-agent定义，在中间层基础上补上「会话」才有的事情：上下文压缩（Context Compaction）、自动重试、队列变化、会话改名。
第一层：pi-ai 的 AssistantMessageEvent
这是最贴近网络协议的一层。它是一个可辨识联合（Discriminated Union），用 type 字段区分分支：
// packages/ai/src/types.ts:501-513
export type AssistantMessageEvent =
| { type: "start"; partial: AssistantMessage }
| { type: "text_start"; contentIndex: number; partial: AssistantMessage }
| { type: "text_delta"; contentIndex: number; delta: string; partial: AssistantMessage }
| { type: "text_end"; contentIndex: number; content: string; partial: AssistantMessage }
| { type: "thinking_start"; contentIndex: number; partial: AssistantMessage }
| { type: "thinking_delta"; contentIndex: number; delta: string; partial: AssistantMessage }
| { type: "thinking_end"; contentIndex: number; content: string; partial: AssistantMessage }
| { type: "toolcall_start"; contentIndex: number; partial: AssistantMessage }
| { type: "toolcall_delta"; contentIndex: number; delta: string; partial: AssistantMessage }
| { type: "toolcall_end"; contentIndex: number; toolCall: ToolCall; partial: AssistantMessage }
| { type: "done"; reason: Extract<StopReason, "stop" | "length" | "toolUse">; message: AssistantMessage }
| { type: "error"; reason: Extract<StopReason, "aborted" | "error">; error: AssistantMessage };AssistantMessageEvent数一数联合分支：12 种（源码事实，可用本章实践任务复核）。注意三个细节：
- 每个事件都带
partial——一份「截止到此刻已累积好」的完整AssistantMessage。也就是说，消费者既可以用delta做真增量渲染，也可以直接拿partial全量重绘。后面会看到 Pi 的 TUI 选了后者。 contentIndex是关联键。官方文档说明：不同内容块的事件可能交错出现，消费者必须靠contentIndex把 delta 归到对应的块上，不能假设一个块的*_start/*_delta/*_end是连续的（来源：packages/ai/README.md的 Complete Event Reference 一节）。- 流必须以
done或error结束，二选一。这条协议约定写在类型定义正上方的注释里（packages/ai/src/types.ts:493-500）。
事件是从哪里造出来的
以 Anthropic 为例，上游是 SSE（Server-Sent Events，服务器推送事件）文本流。pi-ai 自己写了 SSE 解析器，把字节切成 event: / data: 行（packages/ai/src/api/anthropic-messages.ts:387），再把上游的 content_block_delta 翻译成统一事件：
// packages/ai/src/api/anthropic-messages.ts:629-641
} else if (event.type === "content_block_delta") {
if (event.delta.type === "text_delta") {
const index = blocks.findIndex((b) => b.index === event.index);
const block = blocks[index];
if (block && block.type === "text") {
block.text += event.delta.text; // 先累积进 partial
stream.push({
type: "text_delta",
contentIndex: index,
delta: event.delta.text,
partial: output,
});
}
}
// …（省略：thinking_delta / input_json_delta / signature_delta 三个平行分支）content_block_delta同一个 if / else if 链里，input_json_delta 分支处理的是工具参数的增量。工具参数以 JSON 字符串分片到达，中途一定是残缺的（比如只到 {"path": "src/ind）。pi-ai 用 parseStreamingJson() 做「尽力而为」的解析，保证 partial 里的 arguments 永远是个可用对象（packages/ai/src/utils/json-parse.ts:104-124，内部依赖 partial-json 库）。这一点本章点到为止，6.1 pi-ai：统一的模型接口 会展开讲它的三级回退策略。
stream.push(...) 里的 stream 是 AssistantMessageEventStream，它继承自一个 60 行出头的通用 EventStream：生产者 push，消费者用 for await 取，中间靠一个队列加一组等待中的 resolve 函数做「推-拉桥接」。
第二层：pi-agent-core 的 AgentEvent
Agent Loop 拿到上面那串事件后，并不原样转发。它发出的是自己的一套：
// packages/agent/src/types.ts:422-437
export type AgentEvent =
// Agent lifecycle
| { type: "agent_start" }
| { type: "agent_end"; messages: AgentMessage[] }
// Turn lifecycle - a turn is one assistant response + any tool calls/results
| { type: "turn_start" }
| { type: "turn_end"; message: AgentMessage; toolResults: ToolResultMessage[] }
// Message lifecycle - emitted for user, assistant, and toolResult messages
| { type: "message_start"; message: AgentMessage }
// Only emitted for assistant messages during streaming
| { type: "message_update"; message: AgentMessage; assistantMessageEvent: AssistantMessageEvent }
| { type: "message_end"; message: AgentMessage }
// Tool execution lifecycle
| { type: "tool_execution_start"; toolCallId: string; toolName: string; args: any }
| { type: "tool_execution_update"; toolCallId: string; toolName: string; args: any; partialResult: any }
| { type: "tool_execution_end"; toolCallId: string; toolName: string; result: any; isError: boolean };AgentEvent10 种（源码事实）。请特别注意：这一层没有 text_delta，也没有 thinking_delta。第一层那 12 种事件里的 9 种「块内事件」全部被折叠进了一个 message_update，原事件作为 assistantMessageEvent 字段随行——这就是「信封」的字面含义。
转换发生在 streamAssistantResponse() 里的一个 for await 循环（它本身在 agent-loop.ts:193 被主循环调用）：
// packages/agent/src/agent-loop.ts:317-344（有省略）
for await (const event of response) {
switch (event.type) {
case "start":
partialMessage = event.partial;
context.messages.push(partialMessage);
addedPartial = true;
await emit({ type: "message_start", message: { ...partialMessage } });
break;
case "text_start": case "text_delta": case "text_end":
case "thinking_start": case "thinking_delta": case "thinking_end":
case "toolcall_start": case "toolcall_delta": case "toolcall_end":
if (partialMessage) {
partialMessage = event.partial;
context.messages[context.messages.length - 1] = partialMessage;
await emit({ type: "message_update", assistantMessageEvent: event, message: { ...partialMessage } });
}
break;
// …（省略：done / error 分支，取 response.result() 后 emit message_end 并 return）
}
}message_update从源码结构看，这个映射表非常整齐：start → message_start，9 种块内事件 → message_update，done / error → message_end。循环本身只需要知道「一条消息开始了 / 在变 / 结束了」，块级细节它一律不解释，转手交给上层。
还有一个容易被忽略的点：每个 emit 都被 await 了。emit 的最终去处是 Agent.processEvents()，它先归约内部状态，再按注册顺序逐个 await 监听器：
processEvents这意味着事件传播是有背压（backpressure）的：如果某个监听器是异步的且执行得慢，Agent Loop 会真的被它拖住。官方 README 把这一性质说成「assistant 的 message_end 可以当作工具执行前的屏障」（来源：packages/agent/README.md）。代价也很直接——写一个卡顿的界面监听器，会拖慢模型响应的消费速度。
第三层：coding-agent 的 AgentSessionEvent
再往外一层，AgentSession 定义了自己的事件类型。它的写法很值得一读：
// packages/coding-agent/src/core/agent-session.ts:139-155（有省略）
export type AgentSessionEvent =
| Exclude<AgentEvent, { type: "agent_end" }> // 继承其余 9 种
| { type: "agent_end"; messages: AgentMessage[]; willRetry: boolean } // 覆写：多一个 willRetry
| { type: "agent_settled" }
| { type: "queue_update"; steering: readonly string[]; followUp: readonly string[] }
| { type: "compaction_start"; reason: "manual" | "threshold" | "overflow" }
| { type: "entry_appended"; entry: SessionEntry }
| { type: "session_info_changed"; name: string | undefined }
| { type: "thinking_level_changed"; level: ThinkingLevel }
// …（省略：compaction_end、auto_retry_*、summarization_retry_*、bash_execution_update）AgentSessionEventExclude<AgentEvent, { type: "agent_end" }> 这个写法本身就是一堂类型课：它从中间层的联合里挖掉 agent_end 分支，再补上一个多了 willRetry 字段的同名分支。于是订阅者拿到的 agent_end 会多告诉你一件事——这次结束之后还会不会自动重试。
按 type 字面量去重后，这一层新增或覆写了 14 个 type 名（源码事实，数法见实践任务），加上从中间层继承的 9 个，订阅者一共可能收到 23 种 type。
转发点只有一处：
_handleAgentEvent顺序值得记住：先扩展、后监听者（agent-session.ts:619 与 622）。也就是说扩展（Extension）看到事件的时间早于界面。扩展收到的 message_update 依然带着 assistantMessageEvent 字段（packages/coding-agent/src/core/extensions/types.ts:748-753），所以想做「逐 token 统计输出速度」这类扩展，材料是齐的。
图 5.4-1 三层事件类型的映射关系
从上往下读。左侧三条线是「一条 assistant 消息的生命周期」，右侧 OTH 那一组是循环自己的事件，它们不经过第一层。请重点看 E1 到第二层的三条箭头：12 种事件被压成 3 种，其中 9 种全部塞进 MU 这一个信封。E1 对应 packages/ai/src/types.ts:501-513，MU 对应 packages/agent/src/types.ts:432，E3 对应 packages/coding-agent/src/core/agent-session.ts:139-181。
消费端一：TUI 怎么把事件变成字符
交互模式（TUI，Terminal User Interface，终端界面）对会话事件的订阅只有一处（可用 grep -n "\.subscribe(" packages/coding-agent/src/modes/interactive/interactive-mode.ts 复核，只有 1 行命中）：
// packages/coding-agent/src/modes/interactive/interactive-mode.ts:2875-2879
private subscribeToAgent(): void {
this.unsubscribe = this.session.subscribe(async (event) => {
await this.handleEvent(event);
});
}handleEvent 是一个大 switch，把事件翻译成对组件树的增删改。message_update 分支是本章的主角：
// packages/coding-agent/src/modes/interactive/interactive-mode.ts:2960-2993（有省略）
case "message_update":
if (this.streamingComponent && event.message.role === "assistant") {
this.streamingMessage = event.message;
this.streamingComponent.updateContent(this.streamingMessage); // 用累积的 partial 全量重建
for (const content of this.streamingMessage.content) {
if (content.type === "toolCall") {
if (!this.pendingTools.has(content.id)) {
// …（省略：新建 ToolExecutionComponent 并登记进 pendingTools）
} else {
this.pendingTools.get(content.id)?.updateArgs(content.arguments);
}
}
}
this.ui.requestRender();
}
break;message_update这里有一个反直觉但很关键的事实：TUI 根本没有使用 assistantMessageEvent。它每次都拿累积好的 event.message，让 AssistantMessageComponent.updateContent() 清空子组件容器再整体重建（packages/coding-agent/src/modes/interactive/components/assistant-message.ts:83 起）。
「那不是每次都全量重绘吗？」在组件层确实是。真正的增量性由下游两道闸门补回来：
ui.requestRender()不会立刻画。TuiBase.requestRender()只置一个标志位，用process.nextTick调度，并按MIN_RENDER_INTERVAL_MS = 16节流（packages/tui/src/tui.ts:730-751，常量在tui.ts:321）。所以一秒内最多约 60 帧，几十个text_delta会被合并成一次绘制。- 真正写终端时做行级差分：
TuiMainScreen.doRender()把整棵组件树渲染成string[]，与上一帧逐行比较，找出firstChanged/lastChanged，只重写这个区间内的行（packages/tui/src/TuiMainScreen.ts:146-173与260-286）。
从源码结构看，这是一次明确的分工：组件层写得尽量笨（清空重建、逻辑简单不易错），性能交给「节流 + 行 diff」兜底。一种看法是这让组件作者的心智负担极低；代价是超长回复下每帧的组件重建成本随消息长度增长——不过这一点属于结构推断（尚未在源码中直接证实），仓库内没有对应的性能测量数据。
图 5.4-2 一个 text 增量从 SSE 字节到屏幕字符的旅程
从上到下按时间顺序读。关注三处「减速带」：第 3 步之后事件进入队列等消费者；第 6 步每个监听器都被 await；第 9 步 16ms 节流会把多个增量合并成一帧。对应源码：SSE 与 MAP 在 packages/ai/src/api/anthropic-messages.ts 的 387 行与 629 行，ES 在 packages/ai/src/utils/event-stream.ts:21，SAR 在 packages/agent/src/agent-loop.ts:317，AG 在 packages/agent/src/agent.ts:529，SESS 在 packages/coding-agent/src/core/agent-session.ts:595，UI 在 interactive-mode.ts:2960，SCR 在 packages/tui/src/TuiMainScreen.ts:146。
消费端二：--mode json 的逐行输出
同一份事件流，换一个消费者，界面就完全不同。真实采集的帮助文本里写着这个开关（真实采集，research/cli-captures/pi-help.txt 第 22 行）：
--mode <mode> Output mode: text (default), json, or rpc--mode json 走的是 print 模式的 JSON 输出分支（模式判定见 packages/coding-agent/src/main.ts:109-124）。它的实现短得惊人：
// packages/coding-agent/src/modes/print-mode.ts:104-117
unsubscribe?.();
unsubscribe = session.subscribe((event) => {
if (mode === "json") {
writeRawStdout(`${JSON.stringify(event)}\n`);
}
});
};
try {
if (mode === "json") {
const header = session.sessionManager.getHeader();
if (header) {
writeRawStdout(`${JSON.stringify(header)}\n`);
}
}writeRawStdout对比两个消费端，能看清「三层事件」的价值：
| 对比项 | TUI（interactive） | --mode json（print） |
|---|---|---|
| 订阅点 | interactive-mode.ts:2876 | print-mode.ts:104 |
| 处理方式 | 大 switch，按事件类型改组件树 | 不区分类型，JSON.stringify 后写一行 |
用不用 assistantMessageEvent | 不用，只读累积的 event.message | 原样输出，所以每个 text_delta 都出现在结果里 |
| 输出体量 | 一屏字符 | 每个增量一行，回复越长行数越多 |
因为不做任何过滤，--mode json 的输出里会包含全部 message_update 事件，而每个 message_update 又带着一份完整的 partial 快照——一次几百 token 的回复会产生几百行、每行都比上一行长一点的 JSON。这是故意的：json 模式的定位是「给程序看的单向事件流」，由调用方自己决定关心哪几种 type。
RPC 模式（远程过程调用，Remote Procedure Call）的事件转发几乎一模一样，只多了一道背压保护：它额外订阅了底层 Agent，在每个事件之后等待 stdout 的写入排空（packages/coding-agent/src/modes/rpc/rpc-mode.ts:354-362）。这样事件生产速度就被下游管道的读取速度约束住，不会把内存撑爆。RPC 协议本身留到 6.10 交互模式与 RPC 模式 讲。
-p（text 模式），那条分支只打印最后一条 assistant 消息的 text 内容（print-mode.ts:129-146）。把 json 模式的输出直接展示给最终用户，会得到几百行 JSON。 实践任务
目标：亲手确认三层事件各有多少种，并跑通两个不需要 API Key 的真实测试，观察事件顺序。全程只读，不修改任何源码。
步骤 1 · 数一数三层的分支数。在 _sources/pi 目录下执行：
# 第一层：AssistantMessageEvent 的联合分支
sed -n '501,513p' packages/ai/src/types.ts | grep -oE 'type: "[a-z_]+"' | wc -l
# 第二层：AgentEvent 的联合分支
sed -n '422,437p' packages/agent/src/types.ts | grep -oE 'type: "[a-z_]+"' | wc -l
# 第三层：AgentSessionEvent 新增或覆写的 type 名（去重后）
sed -n '139,181p' packages/coding-agent/src/core/agent-session.ts \
| grep -oE 'type: "[a-z_]+"' | sort -u | wc -l预期现象：三条命令分别输出 12、10、14。第三层数的是「去重后的新增 type 名」，因为源码里 auto_retry_end 与 summarization_retry_attempt_start 各出现两次。
步骤 2 · 看第一层事件的真实顺序。pi-ai 内置了不需要任何 Key 的 faux provider，它会真的把消息切成小块逐个发事件。仓库里已有一个测试断言了完整顺序：
cd packages/ai
../../node_modules/.bin/vitest --run test/faux-provider.test.ts \
-t "streams an exact event order for fixed-size chunks" --reporter=verbose预期现象：该用例通过，汇总行显示 Tests 1 passed | 22 skipped，其余用例被 -t 过滤跳过。打开 packages/ai/test/faux-provider.test.ts:382 看它断言的数组：一条「先 thinking、再 text、最后 toolcall」的消息产生的 11 个事件依次是 start → thinking_start → thinking_delta → thinking_end → text_start → text_delta → text_end → toolcall_start → toolcall_delta → toolcall_end → done。
步骤 3 · 看第二层事件的真实顺序：
cd ../agent
../../node_modules/.bin/vitest --run test/agent-loop.test.ts \
-t "should emit events with AgentMessage types" --reporter=verbose预期现象：汇总行显示 Tests 1 passed | 20 skipped。测试源码在 packages/agent/test/agent-loop.test.ts:118 起，它断言事件里含有 agent_start / turn_start / message_start / message_end / turn_end / agent_end——正好是第二层的循环生命周期。
如何判断成功：你能回答「为什么步骤 2 的那 11 个事件，在步骤 3 的断言里一个都看不到」。答案是它们本该被折叠进 message_update；而该测试的假流函数直接推了一个 done，中间没有任何块内事件，所以连 message_update 都没产生。
常见错误：① 直接敲 vitest 会提示找不到命令——本仓库没有全局安装，必须用 node_modules/.bin/vitest 这个相对路径；② 忘记 cd 到对应 package 目录，vitest 会找不到该包的配置；③ 数第三层时用 grep -c 'type:' 而不是 grep -oE ... | sort -u，会因为一行里含多个字段、以及重复分支而数错。
对应源码位置：packages/ai/src/types.ts:501-513、packages/agent/src/types.ts:422-437、packages/coding-agent/src/core/agent-session.ts:139-181、packages/ai/test/faux-provider.test.ts:382、packages/agent/test/agent-loop.test.ts:118。
--mode json 需要一个真实可用的模型才能产生事件，没有 API Key 时会在启动阶段就提示没有可用模型（可参考真实采集 research/cli-captures/pi-list-models-claude.txt 与 pi-tui-main.txt 里的 No models available 警告）。上面两个测试走的是内置 faux provider 与假流函数，完全离线，因此更适合作为第一次观察。 本章小结
- Pi 的流式事件分三层：pi-ai 的
AssistantMessageEvent（12 种）→ pi-agent-core 的AgentEvent（10 种）→ coding-agent 的AgentSessionEvent（继承 9 种 + 覆写agent_end+ 新增 13 个 type 名，订阅者可见 23 种type）。 - 文本增量只存在于第一层。第二层把 9 种块内事件统一折叠成
message_update，原事件放在assistantMessageEvent字段里随行——这是全章最容易找不到的一步。 - 事件派发是顺序 await 的（
Agent.processEvents），所以慢监听器会给整条流制造背压；RPC 模式还主动利用了这一点做 stdout 背压保护。 - 两个消费端风格截然不同：TUI 忽略
assistantMessageEvent、每次用累积的partial全量重建组件，靠 16ms 节流与行级差分拿回增量性；--mode json不做任何过滤，把每个事件原样写成一行 JSON。 - 关键术语：可辨识联合（Discriminated Union）、事件信封、
contentIndex、partial（累积快照）、背压（backpressure）、行级差分渲染。 - 关键源码索引：
packages/ai/src/types.ts:501-513（第一层事件协议）packages/ai/src/api/anthropic-messages.ts:629-641（SSE 到统一事件）packages/ai/src/utils/event-stream.ts:21-36（推-拉桥接）packages/agent/src/types.ts:422-437（第二层事件）packages/agent/src/agent-loop.ts:317-344（12 种压成 3 种）packages/agent/src/agent.ts:529-576（顺序 await 派发）packages/coding-agent/src/core/agent-session.ts:139-181、595-622（第三层与转发）packages/coding-agent/src/modes/interactive/interactive-mode.ts:2875-2879、2960-2993（TUI 消费端）packages/coding-agent/src/modes/print-mode.ts:104-117（json 消费端）- 测试：
packages/ai/test/faux-provider.test.ts、packages/agent/test/agent-loop.test.ts、packages/ai/test/anthropic-sse-parsing.test.ts、packages/coding-agent/test/print-mode.test.ts
- 自测问题：
- 一个扩展想统计「每秒输出多少 token」，它应该订阅哪一层的哪个事件、读哪个字段？为什么只用
event.message不够方便？ AgentSessionEvent用Exclude<AgentEvent, { type: "agent_end" }>而不是直接写AgentEvent，得到了什么好处？- 如果你在 TUI 的事件监听器里写了一个
await sleep(1000)，模型响应的消费会不会变慢？依据是哪一行源码？ - 同一句「你好」，在 TUI 里只显示 2 个汉字，在
--mode json里却会产生多行输出。为什么两者差这么多？
- 一个扩展想统计「每秒输出多少 token」，它应该订阅哪一层的哪个事件、读哪个字段？为什么只用
- 下一章：5.5 Session 的创建、保存与恢复——本章只说了事件如何流向界面，还没说它们如何被写进磁盘。
- 本章尚未展开的内容：
partial-json的多级回退解析（留 6.1）、ToolExecutionComponent与工具参数增量渲染（留 6.4）、pi-tui 的差分渲染与 overlay 合成细节（留 6.9）、RPC 事件协议与背压实现（留 6.10）、agent_settled与自动重试相关事件（留 5.6）。