5.4 流式事件如何传播到界面
本页分析版本earendil-works/pi@16787ad2026-09-21本章解决什么问题：模型吐出的第一个字，要经过几次「换装」才变成你终端里闪烁的那几个字符？为什么 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:652-668
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" | "deferred">;
message: AssistantMessage;
}
| { type: "error"; reason: Extract<StopReason, "aborted" | "error">; error: AssistantMessage };AssistantMessageEvent数一数联合分支：12 种（源码事实，可用本章实践任务复核）。注意三个细节：
- 除两个终止事件外，每个事件都带
partial——截止到此刻已累积好的AssistantMessage。也就是说，消费者既可以用delta做真增量渲染，也可以直接拿partial全量重绘。后面会看到 Pi 的 TUI 选了后者。但要注意，partial是 Provider 一路原地修改的同一个活对象，不是事件发生那一刻的快照：事件还在队列里排队时，它可能已经又长了几个字。所以处理事件时读它可以，把它存下来当历史就不可靠（类型定义上方的注释与packages/ai/README.md都这样说明，见packages/ai/src/types.ts:645-650）。 contentIndex是关联键。官方文档说明：不同内容块的事件可能交错出现，消费者必须靠contentIndex把 delta 归到对应的块上，不能假设一个块的*_start/*_delta/*_end是连续的（来源：packages/ai/README.md的 Complete Event Reference 一节）。- 流必须以
done或error结束，二选一。正常情况是start→ 若干更新 →done；生成开始后出错则是start→ 若干更新 →error；如果在生成开始之前（比如组装请求时）就失败了，整条流可能只有一个error。这条协议约定写在类型定义正上方的注释里（packages/ai/src/types.ts:636-651）。done的reason除了stop、length、toolUse，还可以是deferred，它和 pi-ai 的延迟请求（deferred provider request）有关（来源：上游 CHANGELOG），本章不展开。
事件是从哪里造出来的
以 Anthropic 为例，上游是 SSE（Server-Sent Events，服务器推送事件）文本流。pi-ai 自己写了 SSE 解析器，把字节切成 event: / data: 行（packages/ai/src/api/anthropic-messages.ts:411），再把上游的 content_block_delta 翻译成统一事件：
// packages/ai/src/api/anthropic-messages.ts:674-686
} 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 取，中间靠一个先进先出队列（FifoQueue，同文件第 3 行起）加一组等待中的 resolve 函数做「推-拉桥接」。
第二层：pi-agent-core 的 AgentEvent
Agent Loop 拿到上面那串事件后，并不原样转发。它发出的是自己的一套：
// packages/agent/src/types.ts:485-500
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 system, 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 种（源码事实）。注释里的 system 值得留意：系统提示词在对话里是一条系统消息（见 5.2 第 2 棒），它进入对话时同样会有一对 message_start / message_end。请特别注意：这一层没有 text_delta，也没有 thinking_delta。第一层那 12 种事件里的 9 种「块内事件」全部被折叠进了一个 message_update，原事件作为 assistantMessageEvent 字段随行——这就是「信封」的字面含义。
转换发生在 streamAssistantResponse() 里的一个 for await 循环（它本身在 agent-loop.ts:241 被主循环调用）：
// packages/agent/src/agent-loop.ts:411-438（有省略）
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:164-180（有省略）
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:918 与 919）。也就是说扩展（Extension）看到事件的时间早于界面。扩展收到的 message_update 依然带着 assistantMessageEvent 字段（packages/coding-agent/src/core/extensions/types.ts:869-873），所以想做「逐 token 统计输出速度」这类扩展，材料是齐的。
图 5.4-1 三层事件类型的映射关系
从上往下读。左侧三条线是「一条 assistant 消息的生命周期」，右侧 OTH 那一组是循环自己的事件，它们不经过第一层。请重点看 E1 到第二层的三条箭头：12 种事件被压成 3 种，其中 9 种全部塞进 MU 这一个信封。E1 对应 packages/ai/src/types.ts:652-668，MU 对应 packages/agent/src/types.ts:495，E3 对应 packages/coding-agent/src/core/agent-session.ts:164-205。
消费端一：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:3278-3282
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:3401-3434（有省略）
case "message_update":
if (this.streamingComponent && event.message.role === "assistant") {
this.streamingMessage = event.message;
this.streamingComponent.updateContent(this.streamingMessage, true); // 用累积的消息全量重建
for (const content of this.streamingMessage.content) {
if (content.type === "toolCall") {
if (!this.pendingTools.has(content.id)) {
// …（省略：新建 ToolExecutionComponent 并登记进 pendingTools）
} else {
const component = this.pendingTools.get(content.id);
if (component) {
component.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:91 起；第二个参数 true 告诉组件这条消息仍在流式中）。
「那不是每次都全量重绘吗？」在组件层确实是。真正的增量性由下游两道闸门补回来：
ui.requestRender()不会立刻画。TuiBase.requestRender()只置一个标志位，用process.nextTick调度scheduleRender()，后者按MIN_RENDER_INTERVAL_MS = 16节流（packages/tui/src/tui.ts:952-961与986-1004，常量在tui.ts:477）。所以一秒内最多约 60 帧，几十个text_delta会被合并成一次绘制。- 真正写终端时做行级差分：
TuiMainScreen.doRender()（packages/tui/src/tui-main-screen.ts:247起）把整棵组件树渲染成string[]，与上一帧逐行比较，找出firstChanged/lastChanged（tui-main-screen.ts:362-388），只重写这个区间内的行（tui-main-screen.ts:487-490）。这是默认的 regular 模式；用--tui-mode fullscreen启动时，界面换成占满整个终端的TuiAltScreen（packages/tui/src/tui-alt-screen.ts，选择逻辑在packages/coding-agent/src/modes/interactive/tui-renderer.ts:21-48），它的绘制方式留到 6.9。
从源码结构看，这是一次明确的分工：组件层写得尽量笨（清空重建、逻辑简单不易错），性能交给「节流 + 行 diff」兜底。一种看法是这让组件作者的心智负担极低；代价是超长回复下每帧的组件重建成本随消息长度增长——不过这一点属于结构推断（尚未在源码中直接证实），仓库内没有对应的性能测量数据。
图 5.4-2 一个 text 增量从 SSE 字节到屏幕字符的旅程
从上到下按时间顺序读。关注三处「减速带」：第 3 步之后事件进入队列等消费者；第 6 步每个监听器都被 await；第 9 步 16ms 节流会把多个增量合并成一帧。对应源码：SSE 与 MAP 在 packages/ai/src/api/anthropic-messages.ts 的 411 行与 674 行，ES 在 packages/ai/src/utils/event-stream.ts:43，SAR 在 packages/agent/src/agent-loop.ts:411，AG 在 packages/agent/src/agent.ts:561，SESS 在 packages/coding-agent/src/core/agent-session.ts:894，UI 在 interactive-mode.ts:3401，SCR 在 packages/tui/src/tui-main-screen.ts:247。
消费端二：--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:111-126）。它的实现短得惊人：
// packages/coding-agent/src/modes/print-mode.ts:106-127
unsubscribe?.();
unsubscribeBackpressure?.();
unsubscribe = session.subscribe((event) => {
if (mode === "json") {
writeRawStdout(`${JSON.stringify(toJsonEvent(event))}\n`);
}
});
unsubscribeBackpressure =
mode === "json"
? session.agent.subscribe(async () => {
await waitForRawStdoutBackpressure();
})
: undefined;
};
try {
if (mode === "json") {
const header = session.sessionManager.getHeader();
if (header) {
writeRawStdout(`${JSON.stringify(header)}\n`);
}
}writeRawStdout事件并没有被原样序列化，中间过了一道 toJsonEvent：
toJsonEvent也就是说，JSON 输出里的 message_update 只剩三个字段：type、usage，以及去掉了 partial 的 assistantMessageEvent（json-event.ts:11-15）。源码注释（json-event.ts:40-45）把分工说得很清楚：message_start 给出初始消息，一路的增量把它拼起来，message_end 给出最终的权威消息。
对比两个消费端，能看清「三层事件」的价值：
| 对比项 | TUI（interactive） | --mode json（print） |
|---|---|---|
| 订阅点 | interactive-mode.ts:3279 | print-mode.ts:108 |
| 处理方式 | 大 switch，按事件类型改组件树 | 不区分类型，toJsonEvent 后 JSON.stringify 写一行 |
用不用 assistantMessageEvent | 不用，只读累积的 event.message | 只输出它（去掉 partial），所以每个 text_delta 都作为一行出现在结果里 |
| 输出体量 | 一屏字符 | 每个增量一行，行数随回复变长而增加，每行的长度只取决于这一小块增量 |
除了精简 message_update，json 模式不做任何过滤：全部事件都会出现，一次几百 token 的回复会产生几百行 message_update。这是故意的：json 模式的定位是「给程序看的单向事件流」，由调用方自己决定关心哪几种 type；需要完整消息的调用方，自己在 message_start 与 message_end 之间把增量拼起来即可。从设计上看，如果每一行都附带一份累积快照，第 N 行就要重复前 N−1 块的全部内容，输出总量会随回复长度平方增长——上游 CHANGELOG 正是以此为由去掉了累积字段。
RPC 模式（远程过程调用，Remote Procedure Call）的事件转发几乎一模一样：同样经过 toJsonEvent，同样额外订阅了底层 Agent，在每个事件之后等待 stdout 的写入排空（packages/coding-agent/src/modes/rpc/rpc-mode.ts:353-363；json 模式对应 print-mode.ts:113-118）。这样事件生产速度就被下游管道的读取速度约束住，不会把内存撑爆。RPC 协议本身留到 6.10 交互模式与 RPC 模式 讲。
-p（text 模式），那条分支只打印最后一条 assistant 消息的 text 内容（print-mode.ts:139-156）。把 json 模式的输出直接展示给最终用户，会得到几百行 JSON。 实践任务
目标：亲手确认三层事件各有多少种，并跑通两个不需要 API Key 的真实测试，观察事件顺序。全程只读，不修改任何源码。
步骤 1 · 数一数三层的分支数。在 _sources/pi 目录下执行：
# 第一层：AssistantMessageEvent 的联合分支
sed -n '652,668p' packages/ai/src/types.ts | grep -oE 'type: "[a-z_]+"' | wc -l
# 第二层：AgentEvent 的联合分支
sed -n '485,500p' packages/agent/src/types.ts | grep -oE 'type: "[a-z_]+"' | wc -l
# 第三层：AgentSessionEvent 新增或覆写的 type 名（去重后）
sed -n '164,205p' packages/coding-agent/src/core/agent-session.ts \
| grep -oE 'type: "[a-z_]+"' | sort -u | wc -l预期现象：三条命令分别输出 12、10、14。第三层数的是「去重后的新增或覆写的 type 名」，因为这一段里 agent_end（Exclude 里一次、覆写的新分支一次）与 summarization_retry_attempt_start（两种 source 各一个分支）各出现两次。
步骤 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 | 34 skipped。测试源码在 packages/agent/test/agent-loop.test.ts:121 起，它断言事件里含有 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:652-668、packages/agent/src/types.ts:485-500、packages/coding-agent/src/core/agent-session.ts:164-205、packages/ai/test/faux-provider.test.ts:382、packages/agent/test/agent-loop.test.ts:121。
--mode json 需要一个真实可用的模型才能产生事件。没有 API Key 时，它只输出一行会话头（{"type":"session",...}），随后请求就在发给模型之前被 AgentSession.prompt 的凭证校验拦下，报 No API key found for the selected model.（本书在锁定版本上实际运行所得；交互界面里对应的是真实采集 research/cli-captures/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），所以慢监听器会给整条流制造背压；json 与 RPC 模式还主动利用了这一点做 stdout 背压保护。 - 两个消费端风格截然不同：TUI 忽略
assistantMessageEvent、每次用累积的partial全量重建组件，靠 16ms 节流与行级差分拿回增量性；--mode json不过滤事件，把每个事件写成一行 JSON，只把message_update精简成不带累积快照的增量。 - 关键术语：可辨识联合（Discriminated Union）、事件信封、
contentIndex、partial（累积快照）、背压（backpressure）、行级差分渲染。 - 关键源码索引：
packages/ai/src/types.ts:652-668（第一层事件协议）packages/ai/src/api/anthropic-messages.ts:674-686（SSE 到统一事件）packages/ai/src/utils/event-stream.ts:43-58（推-拉桥接）packages/agent/src/types.ts:485-500（第二层事件）packages/agent/src/agent-loop.ts:411-438（12 种压成 3 种）packages/agent/src/agent.ts:561-608（顺序 await 派发）packages/coding-agent/src/core/agent-session.ts:164-205、894-919（第三层与转发）packages/coding-agent/src/modes/interactive/interactive-mode.ts:3278-3282、3401-3434（TUI 消费端）packages/coding-agent/src/modes/print-mode.ts:106-127、packages/coding-agent/src/modes/json-event.ts:40-61（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里却会产生多行输出。为什么两者差这么多？json 模式输出的message_update里为什么找不到完整消息？
- 一个扩展想统计「每秒输出多少 token」，它应该订阅哪一层的哪个事件、读哪个字段？为什么只用
- 下一章：5.5 Session 的创建、保存与恢复——本章只说了事件如何流向界面，还没说它们如何被写进磁盘。
- 本章尚未展开的内容：
partial-json的多级回退解析（留 6.1）、ToolExecutionComponent与工具参数增量渲染（留 6.4）、pi-tui 的差分渲染与 overlay 合成细节（留 6.9）、RPC 事件协议与背压实现（留 6.10）、agent_settled与自动重试相关事件（留 5.6）。
✅ 自测问题参考答案先自己回答，再点开对照
- 订阅第二层（或第三层转发出来的）
message_update事件，读它随行携带的assistantMessageEvent字段——里面才是第一层那 12 种事件，text_delta的delta就是「这一小块新增了多少文字」，直接累加即可。只用event.message不方便，是因为它是累积后的完整快照：要算增量得自己保存上一次的内容再做差；而且partial那份快照在源码里是同一个被反复修改的对象，存下来当「那一刻的样子」并不可靠。 - 好处是只覆写要改的那一个，其余全部自动继承。
AgentSessionEvent真正想改的只有agent_end（多带一个willRetry字段），用Exclude把它从AgentEvent里剔掉、再并上自己那份新定义，剩下 9 种一个都不用重抄。将来 pi-agent-core 给AgentEvent新增成员时，这里自动跟着涨——如果当初是把 10 种手抄一遍，就得记得回来同步，漏一种就是一个「新事件在会话层莫名消失」的 bug。 - 会变慢，而且是整条流一起慢。依据是
Agent.processEvents（packages/agent/src/agent.ts:561-608）——它是顺序 await 每个监听器的，你这个await sleep(1000)会把事件派发卡住一秒，上游的EventStream于是堆积，形成背压。这不是缺陷而是被利用的特性：json 与 RPC 模式正是靠同一个机制，在每个事件后等 stdout 排空，让事件生产速度被下游管道的读取速度约束住。 - 因为两者消费的不是同一样东西。TUI 忽略
assistantMessageEvent，每次拿累积的event.message全量重建组件，最终屏幕上就是那 2 个汉字；--mode json则不过滤事件，把每一个AgentSessionEvent经toJsonEvent转换后JSON.stringify成一行——「你好」两个字在流式过程中，text_start、若干text_delta、text_end各自对应一个message_update，前后还有message_start、message_end以及循环生命周期事件。至于完整消息：toJsonEvent（json-event.ts:48-61）专门把message_update里累积的message和partial去掉了，只留增量和usage，否则每行都重复前面的全部内容，输出会随回复长度平方增长；需要完整消息的调用方，要么自己拼增量，要么直接读message_end里那份权威消息。