Pi Messenger · Crew internals
主 Pi 进程里的 Messenger Extension 是流程导演
阅读正文 ↗点击知识块看内容，点击连线看依据。没有关系的块也保留。
正在读取关系，知识块已可查看。
给熟悉 Java、但不熟悉 Node.js 插件和事件循环的读者。本文不从“多 Agent 框架”这种大词开始， 而是从操作系统进程、函数调用、文件、stdout 和模型回合逐段追踪。
**先给答案：**Messenger Extension 不是单独进程。它是加载在 Pi 主进程里的代码模块， 与主 Pi 使用同一个 PID。只有调用 spawn("pi", ...) 后，Planner、Worker、Reviewer 才是新的子进程。
**不会。**安装只让 Pi 在启动时发现并加载这份 TypeScript 模块。
操作系统进程 1
主 Pi CLI · PID 1000
Pi Runtime
主 Session + Agent Loop
主模型的调用控制
:::card Messenger Extension 实例 M0 仍然是 PID 1000 :::
执行 Crew 后才出现
子 Pi · PID 1001/1002/…
Planner / Worker / Reviewer Session
自己的模型调用
:::card 自己的 Extension 实例 M1/M2/… 不同 PID，不共享对象 :::
**Java 类比：**主 Pi 加载 Extension，类似一个 JVM 加载 JAR 并创建 Spring Bean， 不是启动另一个 Spring Boot 服务。Worker 子 Pi 则类似 ProcessBuilder(...).start() 启动了另一套 JVM；它也加载同一个 JAR，但里面的对象是另一份。
| 东西 | 是否新进程 | 内存是否与主 Pi 共享 |
|---|---|---|
| 主 Pi 内的 Messenger Extension | 否 | 是，同一 Node.js 进程 |
| Planner Pi | 是 | 否 |
| Worker Pi | 是 | 否 |
| Reviewer Pi | 是 | 否 |
| 各子 Pi 内的 Messenger Extension | 不额外创建进程 | 只与自己所在的子 Pi 共享 |
Planner
读 PRD、扫描代码、生成任务图。输出以后进程退出。
它是编剧，不负责一直推进。
Worker / Reviewer
Worker 做一个任务，Reviewer 审一个结果，做完退出。
它们是演员和验收员。
主进程 Extension
保存状态、算 Ready Tasks、限制并发、spawn 子 Pi、判断完成或阻塞。
它是确定性的流程导演。
但当前实现不是 Extension 自己执行一个永久的 while 循环。每一波之间还要借助主模型：
主进程 Messenger Extension 主 Pi 模型
│ │
│ 判断还有 Ready Tasks │
│ │
│ pi.sendMessage("继续下一波", │
│ triggerTurn: true) │
├─────────────────────────────────────────►│
│ │ 新模型回合
│ │ 调用 pi_messenger(work)
│◄─────────────────────────────────────────┤
│ spawn 下一批 Worker │
因此它是双层导演：Extension 负责确定性状态机和调度；主模型负责根据导演注入的指令， 再次调用 work Tool。若只能选一个调度核心，就是主 Pi 进程里的 Messenger Extension。
把这五条分开，整个系统就不会再混乱。
:::card 通道 1 · 启动参数
主 Extension → 子 Pi
通过 spawn("pi", args) 的命令行参数和 Prompt，把任务第一次交给 Planner/Worker/Reviewer。
:::
:::card 通道 2 · stdout JSONL 子 Pi → 主 Extension 子 Pi 的工具调用、Token、文本和错误通过 stdout JSONL 流回主进程，用于实时监控和进程结果。 :::
:::card 通道 3 · 任务文件
所有进程 ↔ 项目状态库
plan.json、tasks/task-N.json、进度文件保存业务状态。Worker 调用 task.done 才会把任务正式改为 done。
:::
:::card 通道 4 · 邮箱文件
Agent ↔ Agent
registry/*.json 用于发现，inbox/<name>/*.json 用于通信，fs.watch 负责通知收件方。
:::
:::card 通道 5 · 本地 API
Extension → 自己所属的 Pi Session
pi.sendMessage(..., { triggerTurn: true }) 把消息注入本 Session，并要求 Pi 发起新模型回合。
:::
**最容易犯的错误：**把五条通道都叫“Agent 通信”。实际上，stdout 是子进程监控， task JSON 是业务状态，inbox 才是 Agent 邮件，而 pi.sendMessage 只是一个进程内部的 Session API。
1
用户要求开始 Crew
主模型调用 pi_messenger({ action: "plan" })。这个 Tool 由主进程里的 Extension 执行。
2
Extension 启动 Planner 子 Pi
spawn("pi", [
"--mode", "json",
"--no-session",
"--extension", EXTENSION_DIR,
"-p",
plannerPrompt
])Planner 是新 PID；它的任务从启动参数进入，不是先通过 Messenger 邮箱发送。
3
Planner 通过 stdout 返回规划输出
主 Extension 异步读取 JSONL；Planner 退出后，解析最终文本，创建任务文件和依赖图。
PRD→ Planner 输出→ plan.json + task-N.json
4
Extension 触发主模型开始第一波
Plan Tool 返回后，主模型回合结束，触发 agent_end。Extension 发现 pending auto-work，调用本地的 pi.sendMessage + triggerTurn。
5
主模型在新回合调用 work
主模型调用 pi_messenger({ action: "work", autonomous: true })。Extension 读取任务文件，计算当前 Ready Tasks。
6
Extension 并行 spawn Worker
它建立 queue/running 列表，最多启动配置允许的并发数。每个 Worker 是新的 Pi 进程，并加载自己的 Extension 实例。
7
Worker 执行并写业务状态
Worker join、读取任务、task.start、预留文件、修改代码、测试、提交，然后调用 task.done。它的 Extension 将任务 JSON 改成 done。
8
主 Extension 等待 Worker 进程结束
spawnAgents() 返回 Promise；所有 Worker close 后，Extension 同时检查退出码和任务 JSON，不能只相信一句“完成了”。
9
Extension spawn Reviewer
Reviewer 根据 base commit、Git diff 和任务规范输出 SHIP、NEEDS_WORK 或 MAJOR_RETHINK。
10
Extension 决定结束还是下一波
全部 done 就完成；剩余任务全 blocked 就停止；仍有 Ready Tasks 就记录下一波，通过主 Session 再次触发 work。
先读图例：【本地进程】 才是操作系统里的 PID； 【进程内组件】 与外层进程共用 PID；【远程服务】 和 【磁盘资源】 都不是本机新增进程。Session 也是内存里的会话状态，不是进程。
【远程服务】 ┌──────【本地 OS 进程 PID 1000：主 Pi】──────┐ 【磁盘资源】 ┌─【本地 OS 子进程 PID 1001+】─┐ 【远程服务】
用户 主模型 API │ Pi Runtime / 主 Session Extension M0 │ plan/tasks │ 子 Pi Runtime / 子 Session │ 子模型 API
│ │ │ │ │ │ │ │ │ │ │
│ 开始 Crew │ │ │ │ │ │ │ │ │ │
├────────────────────────────────►│ │ │ │ │ │ │ │ │
│ │◄─模型请求──│─────────────┤ │ │ │ │ │ │ │
│ │─plan Call─►│─────────────┤ │ │ │ │ │ │ │
│ │ │ │ 调用 plan Tool │ │ │ │ │ │ │
│ │ │ ├───────────────────►│ │ │ │ │ │ │
│ │ │ │ │ spawn Planner ────────────────►│ 新 PID │ │ │
│ │ │ │ │ │ │ │ │──模型请求───────►│ │
│ │ │ │ │ │ │ │ │◄─模型响应────────│ │
│ │ │ │ │◄════════ stdout JSONL ════════│ │ │ │
│ │ │ │ │ 写计划 ────────►│ │ │ │ │
│ │ │ │◄── Plan Tool 结果 ─┤ │ │ │ 进程退出 │ │ │
│ │◄─后续请求──│─────────────┤ │ │ │ └────────────────────────────────┘ │
│ │─回合结束──►│─────────────┤ agent_end │ │ │ │
│ │ │ ├───────────────────►│ 判断应开始第一波│ │ │
│ │ │ │◄─ sendMessage + │ │ │ │
│ │ │ │ triggerTurn ─────┤ │ │ │
│ │◄─新模型请求│─────────────┤ │ │ │ │
│ │─work Call─►│─────────────┤──调用 work Tool────►│ 读 Ready ◄──────┤ │
│ │ │ │ │ spawn Worker ────────────────►│ 新 PID │ │ │
│ │ │ │ │◄════════ stdout JSONL ════════│ │ │ │
│ │ │ │ │ │◄─task.done────│ │ │ │
│ │ │ │ │◄════════ close ═══════════════│ │ │ │
│ │ │ │ │ spawn Reviewer ──────────────►│ 新 PID │ │ │
│ │ │ │ │◄════ verdict + close ═════════│ │ │ │
│ │ │ │◄── Wave Tool 结果 ─┤ 更新状态►│ └────────────────────────────────┘ │
│ │ │ │ │ │ │
│ │ │ │ agent_end 后仍有 Ready Tasks？ │
│ │ │ │◄─ Extension sendMessage + triggerTurn：启动新的模型请求，再调用 work │
│ │ └────────────────────────────────────────────┘ │
│ │ ……重复，直到全部 done 或 blocked…… │
图中“主 Pi 模型”和“子 Pi 模型”是经网络访问的模型服务。真正持有 Session、执行 Tool、加载 Extension、 调用 spawn 的，是本地 Pi Runtime 进程。
Worker 有两条返回通道，职责不同。
:::card 运行层返回 stdout JSONL + process close 告诉主 Extension：Worker 调了什么 Tool、消耗多少 Token、最终文本是什么、进程是否正常退出。 **它解决的是：**进程运行和监控。 :::
:::card 业务层返回 task.done / task.block Worker 通过自己进程里的 Messenger Extension 修改共享 task JSON，写入状态、摘要、commit 和测试证据。 **它解决的是：**任务到底完成没有。 :::
主 Extension 的判定近似：
if (worker.exitCode === 0 && task.status === "done") {
succeeded.push(task.id);
} else if (task.status === "blocked") {
blocked.push(task.id);
} else {
// 只说“完成”但没调用 task.done，不算业务完成
task.status = "todo";
failed.push(task.id);
}**因此：**stdout 中出现“已经完成”只是自然语言；task-N.json.status = done 才是 Crew 状态机认可的业务事实。
假设严格依赖模式下有三个任务：
task-1 · 无依赖 task-3 · 无依赖 task-2 · 依赖 task-1
第一波执行 task-1 和 task-3。结束后，Extension 重新读取任务文件：
task-1 = done→ task-2 的依赖已满足→ task-2 = ready
然后发生“两把钥匙点火”：
钥匙一：Extension 决策
确定还有 Ready Task，检查最大波次数、审批、阻塞和重复推进保护。
钥匙二：主模型调用 Tool
Extension 注入“继续 Wave 2”，触发模型回合；主模型再调用一次 work。
Wave 1 的 work Tool 返回
│
▼
主模型完成当前回复
│
▼
主 Pi 发出 agent_end 事件
│
▼
Extension 检查 autonomousState + Ready Tasks
│
▼
pi.sendMessage("继续 Wave 2", triggerTurn=true)
│
▼
主模型新回合调用 work
│
▼
Extension spawn Wave 2 Worker
这不是新 Session，也不是新主 Pi 进程；它是同一个主 Session 里的下一次模型回合。
Worker-1 和 Worker-2 都是独立进程，彼此不能调用对方的 JavaScript 函数。它们复用基础 Messenger 邮箱机制：
Worker-1 Pi 共享 Messenger 目录 Worker-2 Pi
PID 2001 PID 2002
│ │
│ join │ join
├── registry/Worker-1.json ───►│◄──── registry/Worker-2.json ───────────────┤
│ │ │
│ send(to=Worker-2) │ │
├── inbox/Worker-2/msg.json ──►│ │
│ │──── fs.watch 事件 ────────────────────────►│
│ │ │ 读取 JSON
│ │ │ pi.sendMessage
│ │ │ triggerTurn
│ │ │ Worker-2 模型处理
Worker 还可以预留文件。预留信息保存在注册状态中，其他 Worker 调用 edit/write 前，自己的 Extension 会在 tool_call Hook 中检查冲突，并可返回：
{
block: true,
reason: "该文件已被另一个 Worker 预留"
}所以文件预留不是模型之间口头承诺，而是每个 Worker 进程里的 Extension 在 Tool 调用边界执行硬拦截。
Worker 退出但没 task.done
stdout 正常、exitCode 也可能是 0，但任务仍是 in_progress。Extension 会把它当作没有完成，并重置或失败处理。
Reviewer 给 NEEDS_WORK
已完成任务被重置为 todo，Reviewer 反馈保存下来，后续 Worker 带着反馈重试。
主 Pi 退出
主 Extension 也随之消失，不存在独立守护进程继续导演。Session shutdown 会尝试停止它启动的 Worker。
停止 Worker
先向 Worker inbox 写入 shutdown 消息，给它时间释放文件；超时后再发送 SIGTERM，仍不退出才 SIGKILL。
主模型没有按提示调用 work
自动推进依赖模型响应“继续下一波”消息。实现中有重复推进保护，持续没有进展会停止，而不是无限循环。
默认依赖是 advisory
默认配置不会把所有依赖都作为硬调度门槛。需要严格波次依赖时，应明确配置 dependencies 为 strict。
| Pi Messenger 概念 | 接近的 Java 概念 | 注意 |
|---|---|---|
| 主 Pi 进程 | 一个 JVM 应用 | 拥有主 Session 和 Agent Loop |
| Messenger Extension | 同 JVM 内的插件 / Spring Bean | 不是另一个服务 |
| spawn 一个 Worker | ProcessBuilder.start() | 不是线程，是新 OS 进程 |
| spawnAgents 并发控制 | 有上限的任务执行池 | 池中运行的是子进程 Promise |
| stdout JSONL | 读取 Process.getInputStream() | 用于运行事件和输出 |
| task-N.json | 持久化任务表 | 业务状态的权威来源 |
| registry | 简化的服务注册表 | 用 PID 检查存活 |
| inbox JSON | 本机文件队列 | 不是 HTTP/RPC |
| fs.watch | WatchService 回调 | 操作系统通知目录变化 |
| pi.sendMessage | currentSession.addMessage() | 只操作自己所属的 Session |
| triggerTurn | 提交 runModelTurn 任务 | 本质是让 Pi 发起新的 LLM API 调用 |
| agent_end Hook | 生命周期事件监听器 | 用于检查是否需要下一波 |
class CrewCoordinatorExtension {
TaskStore taskStore;
ProcessLauncher processLauncher;
PiRuntimeApi currentPi;
WaveResult work() {
List<Task> ready = taskStore.findReadyTasks();
List<ProcessResult> workers = processLauncher.runParallel(ready);
verifyTaskFiles(workers);
runReviewers();
return decideCompletedBlockedOrContinue();
}
void onMainAgentEnd() {
if (hasReadyTasks()) {
currentPi.injectIntoCurrentSession("请调用 work 开始下一波");
currentPi.triggerModelTurn();
}
}
}这是帮助理解职责的 Java 伪代码，不是 Pi Messenger 的真实 Java 实现。
Planner 负责编排任务，Worker 负责执行，Reviewer 负责验收，任务 JSON 是场记板； 主 Pi 进程里的 Messenger Extension 是流程导演，主模型负责按照导演注入的消息反复调用下一次 work。
而所有“通信”不能混成一件事：
任务首次下发 = spawn 参数里的 Prompt
运行过程返回 = stdout JSONL
业务完成状态 = task.done 写 task JSON
Agent 相互通信 = registry + inbox JSON + fs.watch
推进主模型下一轮 = 本进程 pi.sendMessage + triggerTurn
原文:https://resource.vyibc.com/pi-messenger-crew-communication.html