pi coding-agent JSON 事件流模式详解：用 `--mode json` 将 pi 会话接入自定义工具与 UI

pi（coding-agent）除了默认的交互式 TUI 与纯文本打印模式外，还提供了一个面向程序化集成的 JSON 事件流模式：通过 pi --mode json "Your prompt" 启动，pi 会把整个会话期间发生的所有会话事件以 JSON Lines（每行一个 JSON 对象）的形式输出到 stdout，供外部工具、自定义 UI 或 Agent 编排器消费。本文基于仓库文档 json.md 展开，并结合 print-mode.ts、json-event.ts 等源码，完整讲解该模式的事件协议、消息类型、输出格式与流式组装原理。

启用 JSON 事件流模式

基本用法：

pi --mode json "Your prompt"

--mode 参数支持三种取值，在 args.ts 中解析校验：

• text（默认）：只输出最终回复文本，对应 pi -p "prompt" 的单次打印行为；
• json：把所有会话事件以 JSON Lines 输出到 stdout；
• rpc：基于同一套 JSON 事件协议的交互式 RPC 模式（参见 rpc.md）。

从源码结构看，json 模式与默认 -p 打印模式共用同一个执行入口 runPrintMode()（print-mode.ts）。其工作方式为：

1. 以 session.subscribe() 订阅会话事件总线，每收到一个事件就调用 toJsonEvent() 转换，再用 writeRawStdout() 写出一行 JSON（print-mode.ts）；
2. 通过 session.agent.subscribe() 挂接 waitForRawStdoutBackpressure()，在管道下游消费缓慢时等待 stdout 背压释放，避免事件丢失或进程阻塞（print-mode.ts）；
3. 依次处理 initialMessage 与追加的 messages 后退出，finally 中清理订阅并 flushRawStdout()，保证最后一行事件不丢失（print-mode.ts）。

这意味着 JSON 模式是**单次运行（single-shot）**的：发完 prompt、事件流结束、进程退出，非常适合 CI 脚本、批处理和被其他进程以子进程方式拉起。

输出格式：会话头 + 事件行

stdout 上的每一行都是一个独立的 JSON 对象。第一行是会话头（session header）：

{"type":"session","version":3,"id":"uuid","timestamp":"...","cwd":"/path"}

该头来自会话管理器写入 JSONL 会话文件的头部条目，由 sessionManager.getHeader() 提供（print-mode.ts）。其结构定义在 session-manager.ts：

export const CURRENT_SESSION_VERSION = 3;

export interface SessionHeader {
	type: "session";
	version?: number; // v1 sessions don't have this
	id: string;
	timestamp: string;
	cwd: string;
	parentSession?: string;
}

其中 id 是会话 UUID，cwd 是运行目录，parentSession 在分支/派生会话时可选出现。消费者可以用会话头做流式解析的起点校验，也可据此关联 pi 落盘的 JSONL 会话文件。

紧随其后的是按发生顺序输出的事件行，一个典型的最小流如下：

{"type":"agent_start"}
{"type":"turn_start"}
{"type":"message_start","message":{"role":"assistant","content":[],...}}
{"type":"message_update","usage":{...},"assistantMessageEvent":{"type":"text_delta","contentIndex":0,"delta":"Hello"}}
{"type":"message_end","message":{...}}
{"type":"turn_end","message":{...},"toolResults":[]}
{"type":"agent_end","messages":[...]}

事件协议：JsonAgentSessionEvent

JSON 模式与 RPC 模式共用的线上事件类型是 JsonAgentSessionEvent，定义于 json-event.ts。它与 agent-session.ts 中的 AgentSessionEvent 基本一致，唯一差别是：流式 message_update 事件会省略累计快照字段。类型约束如下：

type WithoutPartial<T> = T extends { partial: unknown } ? Omit<T, "partial"> : T;

type JsonAssistantMessageEvent<T> = T extends { type: "toolcall_start"; partial: unknown }
  ? WithoutPartial<T> & { id: string; toolName: string }
  : WithoutPartial<T>;

type JsonAgentSessionEvent =
  | Exclude<AgentSessionEvent, { type: "message_update" }>
  | {
      type: "message_update";
      usage: Usage;
      assistantMessageEvent: JsonAssistantMessageEvent<AssistantMessageEvent>;
    };

为什么 message_update 只保留增量

转换函数 toJsonEvent() 的注释直接说明了设计动机（json-event.ts）：

Remove cumulative assistant snapshots from streaming wire events. message_start provides the initial message, deltas build it, and message_end provides the final authoritative message.

即流式输出遵循“起点—增量—终点”三段式：

• message_start 给出初始消息骨架；
• 一系列 message_update 只携带 delta / contentIndex 增量，用于实时拼接文本、思考内容或工具调用参数；
• message_end 携带最终权威消息。

这样每行事件尺寸保持与消息长度近似线性（而非每帧都重复全量文本），避免长回复时管道流量按 O(n²) 膨胀。同时源码保留了两类“尺寸恒定”的冗余信息：

1. message_update 顶层的 usage 字段：最新的 provider 累计 token 用量。注意部分 provider 只在请求完成时才上报用量，中途该值可能保持全零；
2. toolcall_start 事件额外补充常规模块的 id 与 toolName 字段（从 partial.content[contentIndex] 中提取，json-event.ts），使消费者无需维护完整 partial 也能立即识别是哪个工具调用开始了。

会话级扩展事件

AgentSessionEvent 在核心 AgentEvent 之上扩展了会话层事件（agent-session.ts），JSON 模式下全部可见，包括：

事件	说明
agent_settled	代理运行完全落定
queue_update	每次变更时输出完整的 pending steering 与 follow-up 队列
compaction_start / compaction_end	覆盖手动与自动压缩；reason 为 manual / threshold / overflow
entry_appended	会话条目追加
session_info_changed / thinking_level_changed	会话名、思考级别变化
auto_retry_start / auto_retry_end	自动重试的开始与结束
summarization_retry_scheduled 等	分支摘要/压缩摘要的重试调度事件
bash_execution_update	! 命令执行 bash 的增量输出

基础代理事件

基础事件来自 packages/agent/src/types.ts 的 AgentEvent，分为三层生命周期：

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 };

三层嵌套关系值得注意：一次运行（agent）包含若干轮（turn，一轮 = 一次助手回复 + 其触发的工具调用与结果），每轮内又包含多条消息（message）与工具执行（tool_execution）生命周期。工具执行的 toolCallId 与 toolcall_start 消息事件中的 id 相互对应，可在 UI 中把“参数流式生成”和“实际执行”关联起来。

事件中出现哪些消息类型

流中的 message 字段来自两套消息定义：

基础消息（packages/ai/src/types.ts）：

• UserMessage
• AssistantMessage
• ToolResultMessage

coding-agent 扩展消息（messages.ts），通过 TypeScript 声明合并注入 CustomAgentMessages，因此也会出现在 message_start/message_end 的 message 字段中：

• BashExecutionMessage（L29）：! 前缀命令的执行记录，含 command、output、exitCode、truncated、fullOutputPath 等；
• CustomMessage（L46）：扩展通过 sendMessage() 注入的自定义消息，含 customType、display、details；
• BranchSummaryMessage（L55）：会话树分支导航时生成的摘要；
• CompactionSummaryMessage（L62）：上下文压缩摘要，含 summary 与 tokensBefore。

实战集成示例

官方文档给出的最小示例——过滤出所有消息结束事件：

pi --mode json "List files" 2>/dev/null | jq -c 'select(.type == "message_end")'

基于该协议，常见的消费姿势包括：

提取最终回复文本（message_end 是最终权威消息）：

pi --mode json "Summarize this repo" 2>/dev/null \
  | jq -rs '[.[] | select(.type == "message_end") | .message | select(.role == "assistant") | .content[] | select(.type == "text") | .text] | join("\n")'

实时拼接流式文本（利用 contentIndex + delta）：

pi --mode json "Write a poem" 2>/dev/null \
  | jq -c 'select(.type == "message_update" and .assistantMessageEvent.type == "text_delta") | .assistantMessageEvent.delta'

观察工具执行：tool_execution_start / tool_execution_end 分别携带 args 与 result / isError，可以据此构建执行面板或失败告警。

由于 message_update 已去掉累计快照，长回复场景下管道数据量可控；若需要完整的最终状态，直接消费 message_end 即可，无需自行从增量重建。

与 RPC 模式的关系

从源码结构看，JSON 模式与 RPC 模式共享同一套线上事件协议：rpc-client.ts 直接复用 JsonAgentSessionEvent 类型，RPC 客户端的 collectEvents() 等接口以该类型为返回结构。换言之，--mode json 可以视为“只出不进”的单向事件流形态，而 RPC 模式 是在同一事件协议之上增加双向控制（prompt 注入、命令下发等）；理解了本文的 JSON 事件协议，阅读 RPC 协议的成本会显著降低。

小结与源码索引

• 文档主体：packages/coding-agent/docs/json.md
• 模式解析（--mode 取值校验）：packages/coding-agent/src/cli/args.ts
• JSON 流执行与 stdout 背压处理：packages/coding-agent/src/modes/print-mode.ts
• 事件转换与 JsonAgentSessionEvent 协议：packages/coding-agent/src/modes/json-event.ts
• 会话头 SessionHeader 与会话版本：packages/coding-agent/src/core/session-manager.ts
• 会话级事件 AgentSessionEvent：packages/coding-agent/src/core/agent-session.ts
• 基础事件 AgentEvent：packages/agent/src/types.ts
• 扩展消息类型：packages/coding-agent/src/core/messages.ts

掌握这套 JSON 事件流协议后，你可以把 pi 当作一个可观察的“黑盒引擎”：外部编排器只需解析 stdout 上的 JSONL 行，即可实时渲染思考/文本/工具调用、统计 token 用量、监听压缩与重试，最终从 message_end 取得权威结果——这是将 pi 集成进自建 UI 或自动化流水线时最稳定的接入面。