rpi-acp 源码解读：以 Zed 为例，一个 ACP Client 如何启动并驱动 rpi Agent

2026年10月14日21 分钟8,805 字#rpi #Rust #ACP #Zed #源码分析 #Agent

在 rpi-package 的 rpi-acp 中，ACP 指的是 Agent Client Protocol。这个 package 的实际用途很具体：让 Zed 这样的 ACP Client 通过标准输入输出启动 rpi，并把 rpi 当作一个可以发送 prompt、接收流式更新和处理权限请求的外部 Agent。

本文以 Zed 作为具体客户端，因为仓库提供了完整的 Zed 注册、配置和排错路径；但 rpi-acp 并不是 Zed 专用实现。只要其他客户端支持 ACP v1、JSON-RPC 2.0 和 stdio，并能处理本文介绍的会话与更新方法，同样可以通过 rpi-acp-bridge 启动 rpi。

本文对应 rpi-packages 中的 rpi-acp 0.1.3，主要源码如下：

packages/rpi-acp/
├── src/lib.rs       插件生命周期、ownsStdio、启动服务
├── src/main.rs      独立 rpi-acp-bridge 二进制
├── src/protocol.rs  ACP v1 JSON-RPC 服务与请求分发
├── src/session.rs   每个 ACP session 对应的 rpi RPC 子进程
├── src/history.rs   session 历史保存与回放
└── src/zed.rs       Zed settings.json 注册

源码仓库：https://github.com/pi-rust/rpi-package/tree/master/packages/rpi-acp

先把 ACP 协议本身讲清楚

如果只从 rpi-acp 的函数名开始读，最容易把 ACP 误解成“Zed 启动一个进程，再把几行 JSON 转发给它”。这不够准确。rpi-acp 是 ACP 的一个 Agent 实现，Zed 是 ACP 的 Client；两者之间有一套规定角色、生命周期和消息方向的协议。

ACP（Agent Client Protocol）要解决的问题是：编辑器或其他客户端，如何用统一方式启动一个 Agent、创建会话、发送用户请求、展示 Agent 的流式进度，并在 Agent 要执行有风险操作时把决定交还给用户。它不是模型 API，也不是 rpi 内部的 RPC 协议。后者只出现在 rpi-acp 与 rpi --mode rpc 子进程之间。

1. ACP 的两端是谁？

角色	在本文中的实例	主要职责
Client	Zed	启动 Agent，提供界面、工作区和权限决策；发送 `initialize`、`session/new`、`session/prompt`
Agent	`rpi-acp`	声明能力，管理 session，驱动模型和工具；发送 `session/update`，必要时请求权限
Agent 的内部运行时	`rpi --mode rpc`	rpi 自己的模型、工具和历史逻辑，不是 ACP 对端

因此“Zed 驱动 rpi”不是 Zed 直接调用 rpi 的模型循环，而是：Zed 调用 ACP 方法，rpi-acp 把这些方法翻译成 rpi 的内部 RPC，再把内部事件翻译回 ACP 更新。

2. 协议、传输和消息格式不是一回事

ACP v1 的消息使用 JSON-RPC 2.0。本文这个实现使用 stdio 作为传输：Client 启动 Agent 后，Agent 从 stdin 读消息、向 stdout 写消息，日志必须写 stderr。stdio 是“消息怎么抵达进程”，JSON-RPC 是“消息长什么样、如何配对”，ACP 才是“有哪些方法、它们在什么时机调用、字段意味着什么”。

JSON-RPC 有三类消息：

请求（request）：有 id，需要对方返回同一个 id 的 result 或 error；
通知（notification）：没有 id，只发事件，不等待响应；
响应（response）：带 id，对应此前的请求。

例如，Zed 初始化 rpi 的请求和响应是一个配对，而 Agent 的进度更新是没有 id 的通知：

// Client → Agent：请求
{"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolVersion":1,"clientCapabilities":{},"clientInfo":{"name":"zed","version":"..."}}}

// Agent → Client：响应，id 必须对应 0
{"jsonrpc":"2.0","id":0,"result":{"protocolVersion":1,"agentCapabilities":{"loadSession":true},"agentInfo":{"name":"rpi","version":"..."}}}

// Agent → Client：通知，没有 id，也没有响应
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"...","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"..."}}}}

在 stdio 实现中通常是一行一个 JSON-RPC 消息，但“按行读写”不是 ACP 的业务语义。真正重要的是：请求用 id 配对，通知用于异步事件；两者可以交错出现。

3. 一次 ACP 会话的完整生命周期

ACP 不是“发一个 prompt，等一个字符串”的单次接口，而是一个有状态的 session 协议：

Client 启动 Agent 进程
  → initialize：协商 protocolVersion 和双方能力
  → session/new：创建会话，得到 sessionId
  → session/prompt：发送一轮用户输入
       ↕ session/update：文本、思考、计划、工具状态等异步通知
       ↕ session/request_permission：Agent 反向请求用户授权
       ↕ session/cancel：Client 可取消当前轮
  ← session/prompt 响应：stopReason，表示这一轮真正结束
  → 下一次 session/prompt：沿用同一个 session 上下文

sessionId 是这条对话的句柄，不是某次 prompt 的请求 id。一次 session 可以有多轮 prompt；一次 prompt 也可能经历多次模型请求和多次工具调用。session/prompt 的响应应理解为“这一轮完成了”，而不是流式文本本身。流式内容通过此前的 session/update 通知抵达。

4. initialize 是能力协商，不只是握手

Client 在 initialize 中告诉 Agent 自己能提供什么，例如是否支持文件读写、终端、权限请求或其他 UI 能力；Agent 返回自己支持什么，例如是否支持 session/load、图片/音频/嵌入资源，以及可用的认证方式。

能力没有声明，就不能假定支持。比如 Agent 返回 loadSession: false 或省略该字段，Client 就不应该调用 session/load。同理，rpi-acp 当前返回 image: false、audio: false、embeddedContext: true，所以它不能把 ACP 的所有 prompt 内容块都当作 rpi 输入。

版本号和能力的作用不同：protocolVersion 解决不兼容的协议版本；capability 解决同一版本中某个可选功能是否存在。源码中的 initialize_result() 是判断 rpi 实际能力的第一入口，不能只看 README 的旧描述。

5. prompt turn 为什么要有 update 和 stopReason 两条路径？

一轮 prompt 可能是这样的：

Client 发送 session/prompt，内容是一个或多个 content block；
Agent 让模型判断下一步；
Agent 通过 session/update 报告计划、文本增量或工具调用；
若工具需要用户决定，Agent 发 session/request_permission，等待 Client 的响应；
工具执行结果继续作为 update 展示，并回到模型循环；
没有更多工具调用时，Agent 返回 session/prompt 的响应，例如 stopReason: "end_turn"。

所以这不是简单的“请求—响应”串行接口：session/prompt 请求尚未返回时，Agent 可以主动发很多通知，也可以主动请求 Client。JSON-RPC 的双向性在这里很重要——stdout 上不只有 Agent 的响应，Agent 还可能提出一个带 id 的反向请求。

最小的结束响应如下：

{"jsonrpc":"2.0","id":2,"result":{"stopReason":"end_turn"}}

常见的停止原因还包括 cancelled、max_tokens、max_turn_requests 和 refusal。Client 发 session/cancel 是通知，不另等一个取消响应；Agent 应尽快停止当前工作，并仍然完成原来的 session/prompt，以 cancelled 表示这一轮是正常取消，而不是把取消伪装成普通错误。

6. 工具到底由谁执行？

ACP 的 tool_call 和 tool_call_update 是 Agent 向 Client 报告状态 的协议对象，不等于 Client 替 Agent 执行了工具。通常是 Agent 决定调用什么工具并执行；Client 负责把进度呈现出来，并在需要时通过 session/request_permission 询问用户。

ACP 也定义了 Client 侧的能力，例如 Agent 可以请求 Client 读文件、写文件或管理终端，但这取决于 initialize 中的能力声明。当前 rpi-acp 走的是另一条路径：它让 rpi 子进程使用自己的 read、write、edit、bash 工具，并把这些动作映射成 ACP 的工具更新。因此不能看到 ACP 的 fs/*、terminal/* 方法，就推断本实现一定会调用它们。

下面再回到 rpi-acp 源码，看这套协议如何落地。

进程关系：ACP Client 启动 rpi，本文以 Zed 为例

在这篇文章里，ACP 的通信双方是 客户端和 rpi，其中用 Zed 作为具体例子。Zed 负责启动进程、发送 prompt、展示更新；rpi 负责运行自己的模型、工具和会话逻辑。

flowchart LR
    Z[ACP Client<br/>本文以 Zed 为例] -->|JSON-RPC 2.0<br/>stdio| A[rpi --acp<br/>ACP Server]
    A -->|每个 Session 一个子进程| R[rpi --mode rpc]
    R --> M[模型与工具循环]
    M -->|事件| R
    R -->|session/update<br/>stopReason| A
    A -->|JSON-RPC 响应/通知| Z

图：rpi-acp 的实际进程关系。Zed 是示例客户端，其他兼容 ACP v1 的客户端也可以位于同一位置。图中省略了权限请求和历史文件；它们会沿同一条 ACP 连接返回客户端。

rpi-acp 让 rpi 扮演 ACP 协议中的 Agent 端。它通过一条 JSON-RPC stdio 通道接收客户端请求，再把 rpi 子进程产生的状态和结果转回客户端。

第一关：为什么必须声明 `ownsStdio`？

普通 rpi 进程可能会读取 stdin、显示 TUI 或运行其他模式。如果 ACP Client 通过管道启动 rpi --acp，而宿主仍然先把 stdin 消费掉，ACP Server 就永远收不到 initialize。

因此 src/lib.rs 在插件注册阶段读取 --acp，并声明：

{"ownsStdio": true}

这一步发生得非常早：主机还没有决定是否读取管道输入时，插件已经告诉它“这一进程的 stdin/stdout 属于我”。宿主随后：

不再执行普通的 piped stdin drain；
进入 headless Rpc lifecycle，而不是 TUI/print 模式；
等待 ACP stdio owner 报告结束或 Ctrl+C。

rpi-acp 的 README 把这称为“claims the process stdio”。没有这个 ABI 能力，单纯实现 JSON-RPC 解析也无法把 ACP 嵌进 rpi 进程。

插件还处理 SessionStart 和 SessionShutdown：

SessionStart
  ├── --install-zed / --print-zed：执行一次性命令
  └── --acp：启动 ACP server 线程

SessionShutdown
  ├── 发送 stop 信号
  └── join ACP server 线程

SERVER_STARTED 原子变量防止 reload 场景启动两个 Server。

两种前端：扩展模式和 bridge 模式

这个 package 同时提供两种运行方式。

扩展模式：`rpi --acp`

这是一个 cdylib 扩展。rpi 主程序负责加载插件，插件通过 ownsStdio 接管管道，再启动 protocol::run。

{
  "agent_servers": {
    "rpi": {
      "type": "custom",
      "command": "rpi",
      "args": ["--acp"]
    }
  }
}

独立模式：`rpi-acp-bridge`

src/main.rs 编译出独立二进制。它自己拥有 stdin/stdout，不需要宿主支持 ownsStdio：

{
  "agent_servers": {
    "rpi": {
      "type": "custom",
      "command": "rpi-acp-bridge",
      "args": []
    }
  }
}

ACP Client（以 Zed 为例）
  ├── 启动 rpi --acp 或 rpi-acp-bridge
  ├── 发送 initialize / session/new / session/prompt
  ├── 接收 session/update
  └── 响应 session/request_permission

Zed 的配置只是一个具体例子。其他客户端需要提供等价的外部 Agent 配置，并把 command 指向 rpi-acp-bridge 或 rpi --acp；其中独立的 rpi-acp-bridge 不依赖 rpi 宿主的 ownsStdio 扩展能力。

bridge 查找真正要启动的 rpi 可执行文件的顺序是：

RPI_BIN；
当前进程本身是 rpi 时使用自身；
bridge 同目录下的 rpi 或 rpi.exe；
PATH 中的 rpi。

这两个入口最后都调用同一套 rpi_acp::protocol::run 和 session 实现，区别只在谁负责拥有标准输入输出。

`protocol.rs`：一条 JSON-RPC 行协议服务

src/protocol.rs 明确规定：ACP v1 使用 JSON-RPC 2.0，每行一条消息：

stdin  ← ACP Client（以 Zed 为例）发来的请求或通知
stdout → rpi 返回响应、更新和反向请求
stderr → 诊断日志

Writer 用一个异步互斥锁保护 stdout：每条消息序列化成 JSON，追加换行并 flush。这样请求分发线程、prompt 任务和权限请求不会并发写坏同一条输出流。

Writer 还维护一个 pending 表，用于处理反向请求：

rpi → ACP Client：session/request_permission (id = 7)
ACP Client → rpi：{"id": 7, "result": ...}
                 │
                 └── Writer::resolve 找到等待中的 oneshot

连接关闭时，Writer::close 会清空 pending 请求，让等待权限结果的任务结束，而不是永久挂住。

initialize：代码实际宣告了哪些能力？

initialize_result() 返回的能力包括：

{
  "protocolVersion": 1,
  "agentCapabilities": {
    "loadSession": true,
    "sessionCapabilities": {"resume": {}},
    "promptCapabilities": {
      "image": false,
      "audio": false,
      "embeddedContext": true
    }
  },
  "agentInfo": {
    "name": "rpi",
    "title": "rpi"
  },
  "authMethods": []
}

这段返回值直接决定客户端会怎样使用 rpi：

只接受 ACP v1；
支持加载和恢复会话；
接受文本、resource link 和嵌入式 resource；
不接受图片和音频 prompt；
不宣告认证方法。

这里有一个源码阅读时值得注意的细节：README 的“Capabilities and limitations”段落仍写着 loadSession 不宣告、session/load 和 session/resume 返回错误，但当前 src/protocol.rs 的 initialize_result() 已宣告 loadSession: true，并且分发器确实处理这两个方法。以当前代码为准，应以源码行为而不是过时的 README 描述判断能力。

`run`：请求分发器怎样工作？

protocol::run 创建三个核心对象：

let writer = Arc::new(Writer::new());
let sessions: Arc<Mutex<HashMap<String, Arc<Session>>>> = ...;
let mut pending: Option<String> = ...;

它循环读取 stdin：

优先消费 acp_stdin() 探测时可能已经读到的第一行；
等待下一行 JSON；
读取 method 和 id；
根据 (id, method) 分发；
响应写回 stdout。

支持的主要客户端请求是：

initialize
session/new
session/prompt
session/load
session/resume

支持的通知包括：

session/cancel

客户端返回的、没有 method 的 JSON-RPC 消息，则用于响应 rpi 发出的反向请求，例如权限确认。

未知方法返回 JSON-RPC -32601。stdin EOF 时，服务会关闭 Writer，杀掉所有 Session 子进程，并把停止原因写入 stderr。

`session/new`：真正的工作进程在哪里？

收到 session/new 后，代码读取 params.cwd，默认是 .，再调用 open_session。它先检查：

cwd 必须是绝对路径；
cwd 必须是已经存在的目录。

然后 Session::spawn 启动：

rpi --mode rpc
  --session-dir <ACP session directory>
  --session-id <session id>
  --permission-gate

子进程的 current directory 就是 ACP 客户端传入的 workspace cwd。stdin/stdout/stderr 都被管道接管：

ACP Server
  ├── 写 prompt → child stdin
  ├── 读事件   ← child stdout
  └── 读日志   ← child stderr

子进程启动后，父进程发送 get_state，等待带有相同 id 的 RPC response，确认后端确实打开了预期 session。确认失败或超时，Session 会被杀掉，不会把一个未初始化的子进程放进 sessions 表。

整个 session/new 的启动流程可以画成：

flowchart TD
    N[Zed: session/new<br/>携带 cwd] --> C{cwd 是绝对路径<br/>且目录存在？}
    C -->|否| E[返回 ACP 错误]
    C -->|是| S[Session::spawn]
    S --> P[启动 rpi --mode rpc<br/>设置 cwd/session 参数]
    P --> G[发送 get_state]
    G --> Q{收到匹配 id 的<br/>成功 response？}
    Q -->|否| K[杀掉子进程并返回错误]
    Q -->|是| OK[保存 Session<br/>返回 sessionId]

图：session/new 不只是创建一个 UUID，还要确认后端 rpi session 已经真正启动。

一次 `session/prompt` 的调用链

这是整个 package 最重要的一条链：

sequenceDiagram
    participant Z as Zed
    participant A as rpi-acp
    participant R as rpi --mode rpc
    participant M as 模型与工具循环

    Z->>A: session/prompt
    A->>A: 订阅 broadcast 事件
    A->>R: {type: "prompt", content: ...}
    R->>M: 执行模型与工具循环
    M-->>R: text/thinking/tool/plan 事件
    R-->>A: JSONL 事件
    A-->>Z: session/update
    M-->>R: response completed/aborted/failed
    R-->>A: final response
    A-->>Z: stopReason

这条调用链也可以用源码函数来定位：

Zed: session/prompt
  → protocol.rs 找到 Session
    → session.prompt
      → 订阅 child event broadcast
      → child stdin 写入 {type:"prompt", content:...}
      → 读取 rpi --mode rpc 事件
      → 转成 ACP session/update
      → child response 完成
    → 返回 {stopReason: ...}

图中 session/update 是流式过程，stopReason 是本轮 prompt 的最终响应；两者不是同一条消息。

输入：ACP content block 转成单文本

代码支持三种输入块：

Text
ResourceLink
Resource

图片、音频等其他 block 会落入 ContentBlock::Other，随后被忽略。prompt_text 会把文本、资源链接和嵌入资源转换成一个文本 prompt，因为 rpi --mode rpc 当前接收的是单个文本提示，而不是完整多模态消息。

这也是一个实际限制：ACP 能表达的输入比当前 rpi backend 能消费的输入更丰富，package 只实现了其中一部分。

中间态：`broadcast` 保证事件不丢

每个 Session 启动一个 stdout reader，把子进程每行 JSON 解析后发送到 broadcast::Sender<Value>。

prompt() 会先订阅，再向子进程发送 prompt：

let mut rx = self.events.subscribe();
self.send(json!({"id": id, "type": "prompt", "content": content}))
    .await?;

先订阅再发送是关键。如果先发送再订阅，子进程可能快速返回第一条事件，导致流式内容丢失。

prompt_lock 保证一个 ACP session 同时只能有一个 prompt；如果上一个 prompt 还在运行，新的请求直接返回 a prompt is already running。

输出：rpi RPC 事件映射为 ACP update

代码中的映射关系是：

rpi RPC 事件	ACP 更新
`message_update` / `text_delta`	`agent_message_chunk`
`message_update` / `thinking_delta`	`agent_thought_chunk`
工具开始执行	`tool_call`，状态 `in_progress`
`edit` / `write` 开始	`tool_call`，附带 ACP `diff` 内容
工具执行更新	`tool_call_update`
工具执行结束	`tool_call_update`，状态 `completed` 或 `failed`
`plan_mode_complete`	`plan`

这些更新通过：

{
  "jsonrpc": "2.0",
  "method": "session/update",
  "params": {
    "sessionId": "...",
    "update": {"sessionUpdate": "..."}
  }
}

发送给 Zed。代码还处理一个实际问题：后端的文本事件可能是累计快照，也可能是增量 delta，最终 response 里还可能有 finalText。StreamText 会比较累计文本和最终文本，在结束前补发缺失内容，并避免重复发送已经交给客户端的片段。

因此这里不是简单的“收到一行就转发一行”，而是一个带文本修复的事件适配器。

完成、失败和取消如何传递？

当子进程返回最终 response：

outcome: completed → ACP stopReason: end_turn；
outcome: aborted → stopReason: cancelled；
输出达到长度限制 → stopReason: max_tokens；
后端错误 → JSON-RPC error。

客户端发送 session/cancel 时，Session::abort 会：

增加 cancellation generation；
向子进程发送：

{"type":"abort"}

取消尚未完成的权限 dialog；
继续排空事件，直到后端返回终态。

这说明“取消”不是直接杀掉整个 session 子进程。正常取消只中止当前 prompt，session 本身仍可以继续接收下一次 prompt。

权限确认：反向请求的完整路径

子进程如果通过 permission gate 触发 UI dialog，session.rs 会识别 ui_dialog_request，再调用 Writer 向 Zed 发出：

session/request_permission

Zed 返回用户选择后，Writer 的 pending request 表找到对应 oneshot，Session 再把选择转成：

{"type":"ui_dialog_response", ...}

发送给子进程。

默认情况下，子进程会带 --permission-gate 启动；设置：

RPI_ACP_NO_PERMISSION_GATE=1

可以关闭这一行为。当前只有带 purpose: "permission" 的 confirm / selector 对话会映射为 ACP 权限请求，普通问题没有通用问答 UI，会被取消。

这条路径体现了 ACP Server 的特殊性：它不仅向客户端推送状态，也可能在 Agent 执行工具时反过来向客户端请求用户决定。

Session 历史：磁盘保存，但恢复仍有条件

新 ACP session 使用稳定 UUID，并把原生 rpi JSONL 历史保存到：

~/.rpi/agent/acp/sessions

RPI_ACP_SESSION_DIR 可以覆盖目录。创建 child 时，代码传入：

--session-dir <directory>
--session-id <id>

session/load 和 session/resume 的区别是：

session/load：恢复后回放当前 branch 的用户消息、助手文本/思考、工具调用和结果；
session/resume：只恢复 backend context，不重新回放客户端已经拥有的 transcript。

如果找不到 session，代码返回错误，而不是悄悄创建一个空会话。这避免用户以为自己恢复了历史，实际上得到的是一段全新的对话。

以 Zed 为例：注册配置不是手工 JSON 拼接

src/zed.rs 负责查找和修改 Zed 的 settings.json。这是仓库提供的一个客户端集成示例；其他 ACP Client 不需要使用这套配置文件，只需要能够启动对应命令并连接其 stdin/stdout。代码和 README 特别强调它把文件视为 JSONC：允许注释和 trailing comma，因此不会用普通 JSON parser 重新格式化整个文件。

注册动作有三个入口：

rpi --install-zed
rpi --print-zed
rpi-acp-bridge --install-zed

实现特点包括：

幂等：已经存在就不重复写；
原子写入；
第一次修改时保留 settings.json.rpi-bak；
保留原有注释和格式；
--print-zed 只打印，不写文件；
RPI_ACP_NO_AUTO_REGISTER_ZED=1 禁止 --acp 启动时自动注册。

这里的自动注册发生在 rpi --acp 启动时，但如果注册失败，代码不会让 ACP Server 因为编辑器配置写入失败而完全无法启动，只会写诊断日志。

实际限制：不要把它描述成完整 ACP 实现

阅读代码后，当前 package 的边界很明确：

只实现 ACP v1；
prompt 输入支持 text、resource_link、embedded resource；
不处理 image/audio 输入；
不使用 ACP Client 的 fs/* 和 terminal/*，而是使用 rpi 自己的 read、write、edit、bash；
只发送 message、thought、tool call、tool update 和 plan 更新；
prompt template / slash command 不作为 available_commands_update 宣告；
普通用户问答没有通用 UI，只处理权限确认；
没有认证方法。

本文中的 ACP 实现是 ACP v1 + JSON-RPC 2.0 + stdio，对接对象是 Zed 等 Agent Client；它的能力边界应以 rpi-acp 当前源码为准。

如何运行和排错？

安装：

rpi install rpi-acp

或在 rpi-packages 仓库中：

task install

独立 bridge：

cargo build --release -p rpi-acp --bin rpi-acp-bridge

手工探测初始化时，stdout 必须只有 JSON-RPC 响应，日志全部走 stderr。rpi-acp 不能直接在普通终端里裸跑，因为它需要 ACP Client 通过管道发送请求；终端启动会快速失败并提示使用 rpi --install-zed。

排错可以设置：

RPI_ACP_DEBUG=1

它会在 stderr 记录：

[rpi-acp] <- initialize 0
[rpi-acp] <- session/new 1

Zed 报 “incoming transport closed” 时，先查看 Zed 日志中的 agent stderr，再看 [rpi-acp] server stopping (...) 和每个 session 记录的 child executable 路径。源码把这些诊断放在 stderr，正是为了不污染 stdout 的 ACP JSON-RPC 流。