PiAgent 07｜事件管道：运行时如何把事实交给外部#

区分 session.subscribe() 与 pi.on()，从事件时序建立 UI、日志、指标和扩展策略的边界

Agent 运行时会产生大量变化：文本逐段增长、工具开始和结束、turn 切换、队列变化、压缩、重试和最终收束。如果每个模块都互相调用，新的 UI 或观测需求就会侵入核心 Loop。事件系统把“发生了什么”和“谁关心”分开。

1. 两条事件管道#

session.subscribe()：外部观察#

SDK 使用者通过 session.subscribe(listener) 观察 AgentSession 事件，适合：

• 增量渲染文本和 thinking。
• 把工具状态投影到 Web 或 TUI。
• 统计轮次、耗时、token 和失败率。
• 把需要恢复的消息写入业务数据库。

订阅者原则上是观察者：它读取事件并生成自己的投影，不应该悄悄修改核心状态。

pi.on()：扩展策略#

扩展通过 pi.on(eventName, callback) 订阅更靠近运行时的事件。某些扩展事件可以等待、修改或阻断行为，适合：

• 在工具执行前做权限判断。
• 在 Agent 启动前注入策略。
• 在压缩前提供自定义摘要或取消压缩。
• 注册命令、工具和扩展自己的生命周期。

两者看到的事件可能相近，但责任不同：subscribe 是产品外部的事实流，pi.on 是运行时内部的控制点。

2. 事件名表达的是时间点#

一条典型的生命周期可以表示为：

agent_start
  → turn_start
  → message_start / message_update / message_end
  → tool_execution_start / update / end
  → turn_end
  → agent_end
  → agent_settled

压缩、自动重试和队列事件会插入这条主线。事件不是“打印日志的另一种写法”，而是外部系统理解 Agent 当前状态的时钟。

3. 常用事件与用途#

agent_end 和 agent_settled 不应混用。自动重试时可能出现多个 agent_end；agent_settled 表达整个 Session 运行完成。单个 HTTP 请求也可以直接等待 prompt() 的 Promise 作为终局信号。

4. 文本事件有两层类型#

session.subscribe((event) => {
  if (event.type !== "message_update") return;

  switch (event.assistantMessageEvent.type) {
    case "text_delta":
      process.stdout.write(event.assistantMessageEvent.delta);
      break;
    case "thinking_delta":
      recordThinking(event.assistantMessageEvent.delta);
      break;
  }
});

外层 message_update 表示 assistant 消息发生变化，内层类型决定变化内容。不要直接假设 event.delta 存在；正确字段在 assistantMessageEvent.delta，并且只有增量类型才适合拼接。

5. 工具事件如何配对#

工具卡片依靠 toolCallId 关联开始和结束：

tool_execution_start
  { toolCallId: "A", toolName: "query_data", args: {...} }

tool_execution_update
  { toolCallId: "A", ... }

tool_execution_end
  { toolCallId: "A", result: {...}, isError: false }

前端应该按 ID 更新同一张卡片，而不是按事件数组下标匹配。并行工具调用时，事件顺序可能交错；ID 才是稳定关联键。

6. 事件投影：不要把内部协议原样暴露#

Web 服务通常需要把内部事件翻译成较小的公共协议：

function translateEvent(event: AgentSessionEvent) {
  if (event.type === "message_update") {
    const update = event.assistantMessageEvent;
    if (update.type === "text_delta") {
      return { type: "text", data: { delta: update.delta } };
    }
  }

  if (event.type === "tool_execution_start") {
    return {
      type: "tool_start",
      data: { id: event.toolCallId, name: event.toolName },
    };
  }

  return null;
}

翻译层负责隐藏内部对象、删除敏感参数、限制单条消息大小和稳定公共版本。内部事件名称可以随 SDK 版本变化，业务前端不应该直接依赖所有底层字段。

7. 订阅回调的同步性与背压#

事件订阅通常是“发生即通知”，不应该在回调里做长时间阻塞工作。更稳妥的方式是：

Agent 事件
  → 轻量同步投影
  → 有界队列
  → 日志/数据库/网络发送

如果每次 message_update 都同步写数据库或调用远程服务，慢消费者会拖累 UI，甚至拖住整个 Agent。文本增量可以合并后批量写入；工具开始/结束和 settled 事件则应优先保留。

8. 用事件做可观测性#

一次运行至少应记录：

• trace/session/run ID。
• 模型 Provider、模型 ID、thinking level。
• 每个 turn 的开始和结束时间。
• 工具名、调用 ID、耗时、结果类别和错误码。
• 重试次数、压缩前后 token 估算。
• 最终状态：成功、失败、取消或超时。

不要把完整 API Key、未脱敏工具参数或用户私密数据写入公共日志。事件是事实源，不代表所有字段都适合进入观测系统。

常见误区

• 用 agent_end 作为 HTTP 请求唯一终局，忽略重试。
• 只订阅文本，不记录工具开始/结束，导致用户不知道 Agent 在等待什么。
• 用事件顺序而不是 toolCallId 匹配工具结果。
• 把 pi.on() 当成普通 UI 监听器，在回调里做不可控副作用。
• 把底层事件对象直接发送给浏览器，造成版本和安全耦合。

小结

事件系统至少承担三件事：给外部观察者提供事实流，给扩展提供策略插入点，给 UI 和观测系统提供统一时钟。session.subscribe() 更像公共读接口，pi.on() 更像运行时控制面；事件投影层则负责把内部细节变成稳定、脱敏、可消费的产品协议。

源码定位

• 核心 Agent 事件：packages/agent/src/types.ts
• Session 事件扩展：packages/coding-agent/src/agent-session.ts
• 扩展事件与钩子：packages/coding-agent/src/extensions/types.ts
• 工具生命周期：packages/agent/src/agent-loop.ts