pi 桥改为工作进程池 + homeagent 落盘投递账本

## pi 桥:模型工作下到子进程(并发模型重构)

主进程原来自己跑模型,而 pi 的会话装载是同步的:`SessionManager.open()` 走
`openSync` + `readSync` 循环把整个 `.jsonl` 读进内存并逐行 JSON.parse。实测本机
最大那条会话 23MB,`open` 一次**阻塞事件循环 118ms**;模型跑起来后 SDK 内部还有
更多同步工作。SSE 读循环在那期间完全停住 → 后续邮件卡在 TCP 缓冲区 → 久到
Gateway 认为连接死了 → 重连 → 重放。

上一轮我在几个调用点前加 `setImmediate` 是无效的仪式(让出一次之后同步工作照样
占满线程),已回退。这一轮把模型工作整体搬进子进程:实测同样的活在 fork 出的
子进程里跑,主进程事件循环阻塞 **0ms**。

- 新增 `src/worker.mjs`:一封邮件一个进程,跑完就退。权限询问期间的挂起只影响
  那一个 worker(原来 `await new Promise(...)` 等人决策,整座桥不再收信)。
- 新增 `src/pool.mjs`:**不同会话并发**(上限 3,每个 worker 约 140MB RSS)、
  **同一会话严格串行**(pi 假定「一文件一持有者」,两个进程同时装载同一条会话
  文件会让各自的内存索引看不见对方追加的行 → 会话树分叉)、满载排队不丢邮件、
  硬超时 SIGKILL 回收卡死进程。
- `src/index.mjs` 只剩 I/O 与调度:SSE、心跳、去重、分派。
- 选进程而不是 `worker_threads`:模型会跑 bash/write/edit,一次 OOM 不该带走
  整座桥。两者实测都能建起 AgentSession,但线程与主线程共享堆和生命周期。
- 「接管会话短暂持有」那套机制(adopted / adoptTimers / releaseAdopted + 兜底
  计时器)整个删掉 —— worker 退出**就是**释放,且普通会话与接管会话一视同仁。
- 轮次超时 60s → 10 分钟:60s 那个数字是「主进程要腾出手收下一封」的产物,
  worker 没有这个理由,等真结论更准(带工具调用的一轮跑几分钟很正常)。
- IPC 只传路径与标量(sessionFile / cwd / grants / 命名指纹)—— AgentSession
  跨不了进程边界,worker 每次从 sessionFile 重新装载。

`test/pool.test.mjs` +19 例,真 fork 子进程、用桩 worker(不装 SDK)跑毫秒级:
并发上限、同会话串行、不同会话真并发(判据是两个进程的心跳交错,不是 running
map 里有两个条目)、sessionFile/grants/命名指纹跨 worker 传递、config() 每次重取、
硬超时回收、权限决策路由、决策原文透传、worker 退出后清路由、mailDrivenIDs、
kind 透传、stop 先发 shutdown 再杀。跑过三组负向对照确认用例真能抓回归:
拆掉串行守卫 / 不传 sessionFile+grants / 硬超时不杀,对应用例分别失败。

## homeagent:投递去重必须落盘

用户报的重复投递不是上一轮那个 bug。两段提示词的措辞差异指出了来源:
SSE 那段写「你把本轮工作做完」,补投那段写「你把结论说出来就行」。

`deliveredMails` 是进程内的 map,而 homeagent 的插件跑在**子进程**里:

  1. 18:59:38 邮件落库,旧插件进程的 SSE 收到,注入第一次
  2. 同一秒 homed 被重启,那一轮被掐断(`context canceled`)
  3. 18:59:45 新进程起来,`deliveredMails` 是空的
  4. 心跳报 `pending_mails: 1`(第一轮没跑完 → read_inbox 没执行 → 仍未读)
     → catchUp 注入第二次

**不能只记「投过没有」**:那会把「重复」换成「丢件」—— 第 2 步里发件人没收到
回信,而记录说「已投过」→ 永远跳过。丢件比重复严重,重复至少人能看出来。

新增 `ledger.go`:JSONL 账本记两个状态。`completed` 才跳过;`delivered` 但未
`completed` 的仍然重投,但提示词前面插一段说明「上一轮被中断,别把同一件事做
两次」。落在 SDK 的 `Settings().DataDir()`;拿不到时退回 key 文件目录;目录不可
写时退化为纯内存(不比修复前差,也不该让插件起不来)。

- 判定与记录在同一把锁里:SSE 与 catchUp 两个 goroutine 的竞态
- 每行写完 fsync:这个文件的全部意义就是「进程死了之后还算数」
- 坏行跳过而不是报错退出(崩溃时最后一行可能写残)→ 那封退化为重投,安全
- 14 天保留期;过期过半时「临时文件 + rename」压实
- `shortID()` 替代 `id[:8]`:日志不该有能力 panic 掉投递协程

`ledger_test.go` +14 例,含两组负向对照(只记「投过」→ 丢件用例失败;不读账本
→ 跨进程用例失败)。

## 契约文档

`B-7.7`(MUST):子进程形式的插件去重必须落盘且区分「投过」与「跑完」,含事故
时序、两状态表、何时标 completed。已知取舍那节标注投递账本是唯一必须落盘的状态。
验收清单加「模型跑到一半重启宿主」一项。

## 生产验证

- pi 三封 → 三条会话:三个 worker PID 并存,回信「收到 1/2/3」各落自己线索
- pi 同一会话两封:严格串行(收到A 19:26:39 → 收到B 19:26:48,全程单 worker)
- pi 主进程事件循环阻塞 1ms(旧版单进程 open 23MB 一次就 118ms)
- homeagent 正常一封:账本 `c:false` → `c:true`,一封回信
- homeagent 处理中重启:日志「上一轮被中断,带说明重投」,**只有一封 Re:**
- homeagent 再次重启:账本 2 条 completed,不再投递,会话邮件数不变
- gateway 7 包 / web 176+26 / pi 269 / dsh 219 / opencode 201 / homeagent 14
This commit is contained in:
2026-09-04 20:33:44 +08:00
parent c297468819
commit 8e501f041e
8 changed files with 2117 additions and 704 deletions

View File

@ -349,6 +349,7 @@ SSE 只推连上之后的事件。插件重启前发来的邮件不会再推一
| B-7.4 | 按时间**正序**投(收件箱是倒序返回的) | MUST |
| B-7.5 | `mail_type !== 'normal'` 的不补投 | MUST |
| B-7.6 | 逐封投递前**再查一次**去重集合 | SHOULD |
| B-7.7 | 插件以**子进程**形式运行时,去重必须**落盘**,且区分「投过」与「跑完」 | MUST |
> **B-7.1 为什么只补一次**:每轮都补会把「模型正在处理中、尚未标已读」的邮件
> 重复投递 —— 一封邮件起两轮模型。
@ -364,6 +365,39 @@ SSE 只推连上之后的事件。插件重启前发来的邮件不会再推一
>
> **B-7.5 为什么跳过 permission**:原来的工具调用早随进程一起没了,
> 投过去模型没有可恢复的上下文。
>
> **B-7.7 为什么内存去重不够**生产事故2026-09-04
> `deliveredMails` 是进程内的,而 homeagent 的插件跑在**子进程**里 ——
> homed 重启(或插件崩溃自动重启)会换一个新进程,那个集合随之清空。时序:
>
> 1. 邮件落库,**旧**进程的 SSE 收到,注入第一次
> 2. 同一秒进程被重启,那一轮被掉断(日志:`context canceled`
> 3. **新**进程起来,`deliveredMails` 是空的
> 4. 心跳报 `pending_mails: 1`(第一轮没跑完 → `read_inbox` 没执行
> → 邮件仍未读)→ 补投注入第二次
>
> 模型上下文里因此出现两段几乎相同的指令。
>
> **为什么不能只记「投过没有」**:那会把「重复」换成「丢件」。上面第 2 步里
> 那一轮被掉断,发件人**没有**收到回信,而落盘记录说「已投过」→ 永远跳过。
> 丢件比重复严重:重复至少人能看出来,丢件是静默的。
>
> 因此记两个状态:
>
> | 状态 | 含义 | 再次收到时 |
> |---|---|---|
> | `delivered` | 注入过(可能被中断) | **仍然重投**,但提示词里说明「上一轮被中断」 |
> | `completed` | 那一轮真的跑完且回信已发出 | 跳过 |
>
> 「说明上一轮被中断」不是装饰:不说的话模型在上下文里看到两段相似指令,
> 会以为人重复交代了一遍,于是可能把同一件事做两次。
>
> 何时标 `completed`回信发出去了、模型自己回过了B-5.3 让位)、
> 或失败通知发出去了B-6—— 三者都是「发件人得到了一个交代」。
> 回信**发失败**时不标 —— 那时发件人一个字都没收到。
>
> 常驻守护进程形态的插件(如 pi 桥)不受这一条约束:它自己就是进程,
> 重启同时也会丢掉 SSE 连接,那时走的是 B-7 补拉而不是两条路径并发。
### B-8 平台权限询问 → 邮件若平台有审批环节MUST没有则 N/A
@ -1055,11 +1089,16 @@ GET /api/v1/attachments/{id}
| `mailDrivenSessions` | 快照里 `mail_driven` 全变成 `false` |
| `pendingPermissions` | 决策回来时走 `B-4.2` 的退化路径 |
| `explicitSends` | 重启后的第一轮可能重复转发 |
| `deliveredMails` | 由 `B-7`补拉去重兜住 |
| `deliveredMails` | 同一进程内的 SSE 重放靠它;**跨进程**的重复必须`B-7.7`落盘账本兜住 |
要持久化的话应当落在平台的会话元数据里,而不是插件自己的文件 ——
那样才能跟着会话一起被平台清理。
> **例外:投递账本必须落盘**`B-7.7`)。上面那些丢了只是「多开一条会话」
> 或「多转发一次」,而投递去重丢了是「同一封邮件被注入两遍」—— 模型上下文
> 里出现两段相同指令,可能把同一件事做两次。子进程形式的插件尤其如此:
> 它的重启频率由宙主进程决定,不是罕见事件。
---
## 七、验收清单 / Acceptance Checklist
@ -1169,6 +1208,12 @@ GET /api/v1/attachments/{id}
[ ] 插件在线时再发一封
→ 只收到一封回信没有重复投递B-7.3
[ ] (子进程形式的插件)发一封 → 模型跑到一半时重启宙主进程B-7.7
→ 数据库里只有一封 `Re:`(模型上下文里也只有一段有效指令)
→ 日志出现「上一轮被中断,带说明重投」
→ 账本里该 mail_id 先一行 `c:false` 后一行 `c:true`
→ 再次重启(邮件已跑完)→ **不再投递**,会话邮件数不变
```
### 7.6 权限(若平台支持)

View File

@ -0,0 +1,303 @@
package main
// 投递账本:跨进程记住「这封邮件投过没有、跑完没有」。
//
// # 为什么必须落盘
//
// `deliveredMails` 是进程内的 map重启即丢。而 homeagent 的插件是**子进程**
// homed 重启(或插件崩溃自动重启)会换一个新进程 —— 于是:
//
// 1. 18:59:38 邮件落库,旧插件进程的 SSE 收到,`InjectInputSync` 注入第一次
// 2. 同一秒 homed 被重启那一轮被掐断eventloop 报
// `process error: all 2 providers failed, last error: context canceled`
// 3. 18:59:45 新插件进程起来,`deliveredMails` 是空的
// 4. 心跳报 `pending_mails: 1`(第一轮没跑完 → read_inbox 没执行 → 邮件仍未读)
// → `catchUp` 注入第二次
//
// 模型的上下文里因此出现两段几乎相同的指令(措辞的细微差别正好指出来源:
// SSE 那段写「你把本轮工作做完」,补投那段写「你把结论说出来就行」)。
//
// # 为什么不能只记「投过没有」
//
// 那会把「重复」换成「丢件」:上面第 2 步里那一轮被掐断,发件人**没有**收到
// 回信,而落盘记录说「已投过」→ 永远跳过 → 那封邮件再也不会被处理。
// 丢件比重复严重:重复至少人能看出来,丢件是静默的。
//
// 所以记的是两个状态:
//
// - delivered = 注入过(可能被中断)
// - completed = 那一轮真的跑完并且回信发出去了
//
// 判据因此是「completed 才跳过」。delivered 但未 completed 的仍要重投,
// 但换一段提示词明确说「上一轮被中断」—— 模型看到的不再是两条重复指令。
//
// # 为什么是 JSONL 而不是 SQLite
//
// 写入是纯 append 的单行记录读取只在启动时一次。SQLite 要多一个依赖、
// 多一次 schema 迁移,换来的是这里用不到的查询能力。
// 崩溃时最坏情况是最后一行写残 —— 解析时跳过坏行即可(见 loadLedger
import (
"bufio"
"encoding/json"
"fmt"
"log"
"os"
"path/filepath"
"strings"
"sync"
"time"
)
// ledgerEntry 是账本里的一行。
//
// 字段用短名:这个文件会被追加很多行,而它没有人类读者(排查时用 jq
type ledgerEntry struct {
MailID string `json:"m"`
Completed bool `json:"c"`
At int64 `json:"t"` // Unix 秒,仅供排查与过期清理
}
// ledgerState 是一封邮件的投递状态。
type ledgerState struct {
delivered bool
completed bool
}
// deliveryLedger 是账本的内存视图 + 落盘句柄。
type deliveryLedger struct {
mu sync.Mutex
path string
state map[string]*ledgerState
// file 为 nil 表示落盘不可用(目录没权限之类)。此时退化为纯内存 ——
// 那正是修复前的行为,不比它更糟,而且不该让插件起不来。
file *os.File
}
// ledgerRetention 是账本条目的保留期。
//
// 14 天:足够覆盖「插件停了一阵子再起来」的情形,又不会让文件无限增长。
// 判据是条目时间而不是文件大小 —— 后者要读全文才能算,而这里的目的只是
// 别让一个长期运行的部署攒下几十万行。
const ledgerRetention = 14 * 24 * time.Hour
// openDeliveryLedger 打开(或新建)账本。
//
// dir 取 SDK 的 `Settings().DataDir()``<daemon data>/plugin_data/<plugin>`
// SDK 保证存在)。拿不到时退回 key 文件所在目录 —— 那个目录本来就要能写。
func openDeliveryLedger(dir string) *deliveryLedger {
l := &deliveryLedger{state: map[string]*ledgerState{}}
if strings.TrimSpace(dir) == "" {
log.Printf("[homeagent-mail-bridge] 未取到数据目录,投递账本退化为纯内存(重启后可能重复投递)")
return l
}
l.path = filepath.Join(dir, "delivered.jsonl")
// 先读旧记录,再打开追加句柄:反过来的话 O_TRUNC 之类的手误会清空历史。
kept := l.load()
if err := os.MkdirAll(dir, 0o700); err != nil {
log.Printf("[homeagent-mail-bridge] 建数据目录 %s 失败,账本退化为纯内存: %v", dir, err)
return l
}
// 过期条目多到一半以上时重写一遍,否则纯追加。
// 重写走「临时文件 + rename」原地截断时崩溃会留下一个残缺账本
// 而那比多几行过期记录严重得多。
if kept.dropped > 0 && kept.dropped*2 >= kept.total {
l.compact()
}
f, err := os.OpenFile(l.path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
if err != nil {
log.Printf("[homeagent-mail-bridge] 打开投递账本 %s 失败,退化为纯内存: %v", l.path, err)
return l
}
l.file = f
log.Printf("[homeagent-mail-bridge] 投递账本 %s%d 条有效记录)", l.path, len(l.state))
return l
}
type loadStats struct{ total, dropped int }
// load 把账本读进内存。坏行跳过而不是报错退出。
func (l *deliveryLedger) load() loadStats {
var st loadStats
f, err := os.Open(l.path)
if err != nil {
return st // 不存在是正常的(首次运行)
}
defer f.Close()
cutoff := time.Now().Add(-ledgerRetention).Unix()
sc := bufio.NewScanner(f)
// 单行很短,但给个宽松上限防止一行坏数据把扫描器卡住
sc.Buffer(make([]byte, 0, 4096), 64*1024)
for sc.Scan() {
line := strings.TrimSpace(sc.Text())
if line == "" {
continue
}
st.total++
var e ledgerEntry
if err := json.Unmarshal([]byte(line), &e); err != nil || e.MailID == "" {
// 崩溃时最后一行可能写残。跳过它 —— 那封邮件退化为「没记录」,
// 于是会被重投一次,而重投是安全的(下面 shouldDeliver 的语义)。
st.dropped++
continue
}
if e.At > 0 && e.At < cutoff {
st.dropped++
continue
}
s := l.state[e.MailID]
if s == nil {
s = &ledgerState{}
l.state[e.MailID] = s
}
s.delivered = true
// 同一封可能有两行(先 delivered 后 completedcompleted 只增不减。
if e.Completed {
s.completed = true
}
}
return st
}
// compact 把内存状态重写成一个干净的账本。
func (l *deliveryLedger) compact() {
tmp := l.path + ".tmp"
f, err := os.OpenFile(tmp, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600)
if err != nil {
return // 压实失败不影响正确性,只是文件继续变长
}
now := time.Now().Unix()
w := bufio.NewWriter(f)
for id, s := range l.state {
b, _ := json.Marshal(ledgerEntry{MailID: id, Completed: s.completed, At: now})
w.Write(b)
w.WriteByte('\n')
}
if err := w.Flush(); err != nil {
f.Close()
os.Remove(tmp)
return
}
// fsync 后再 renamerename 本身是原子的,但内容没落盘时掉电会得到空文件。
f.Sync()
f.Close()
if err := os.Rename(tmp, l.path); err != nil {
os.Remove(tmp)
}
}
// append 写一行。落盘不可用时只更新内存。
func (l *deliveryLedger) appendLine(e ledgerEntry) {
if l.file == nil {
return
}
b, err := json.Marshal(e)
if err != nil {
return
}
// 单次 Write 写完整行:多次 Write 之间崩溃会留下半行。
if _, err := l.file.Write(append(b, '\n')); err != nil {
log.Printf("[homeagent-mail-bridge] 写投递账本失败(本次去重仅内存生效): %v", err)
return
}
// 每行都 Sync这个文件的全部意义就是「进程死了之后还算数」
// 攒在页缓存里等于没写。一封邮件一次 fsync代价可以忽略。
l.file.Sync()
}
/*
claim 判定这封邮件要不要投,并在要投时记下 delivered。
返回值:
deliver = false → 已经**跑完**过,跳过
deliver = true, resumed = false → 全新的邮件
deliver = true, resumed = true → 投过但没跑完(上一轮被中断),
调用方应换一段说明中断的提示词
这里把「判定」与「记录」放在同一把锁里:分开的话 SSE 与 catchUp 两个 goroutine
可能都判定为要投(这正是修复前的竞态,只是那时连内存都没查)。
*/
func (l *deliveryLedger) claim(mailID string) (deliver, resumed bool) {
if mailID == "" {
return false, false
}
l.mu.Lock()
defer l.mu.Unlock()
s := l.state[mailID]
if s != nil && s.completed {
return false, false
}
if s != nil && s.delivered {
// 投过但没跑完。**仍然重投** —— 上一轮被中断意味着发件人还没收到回信,
// 跳过它就是静默丢件。但要让调用方知道这是重投。
//
// 不在这里加「重投次数上限」:一封邮件反复中断说明有别的问题
// (模型一直崩、进程反复重启),而那时停止重投只会让问题更难发现。
return true, true
}
if s == nil {
s = &ledgerState{}
l.state[mailID] = s
}
s.delivered = true
l.appendLine(ledgerEntry{MailID: mailID, Completed: false, At: time.Now().Unix()})
return true, false
}
// complete 标记这封邮件真的处理完了(回信已发出或已确认无需回信)。
func (l *deliveryLedger) complete(mailID string) {
if mailID == "" {
return
}
l.mu.Lock()
defer l.mu.Unlock()
s := l.state[mailID]
if s == nil {
s = &ledgerState{delivered: true}
l.state[mailID] = s
}
if s.completed {
return // 幂等:重复标记不再写盘
}
s.completed = true
l.appendLine(ledgerEntry{MailID: mailID, Completed: true, At: time.Now().Unix()})
}
// close 关闭落盘句柄。
func (l *deliveryLedger) close() {
l.mu.Lock()
defer l.mu.Unlock()
if l.file != nil {
l.file.Close()
l.file = nil
}
}
// resumeNote 是重投时插在提示词前面的说明。
//
// 必须说清「上一轮被中断」:不说的话模型在上下文里看到两段几乎相同的指令,
// 会以为人重复交代了一遍,于是可能把同一件事做两次(或者反问「你是不是发重了」)。
func resumeNote(mailID string) string {
return fmt.Sprintf(
"(这封邮件 %s 之前已经通知过你一次,但那一轮被中断了 —— 进程重启或模型出错,"+
"所以你可能在上文里看到一段几乎相同的通知。请把它当作**同一件事**继续处理,"+
"不要重复执行已经做过的操作。)\n\n", mailID)
}
// shortID 把 UUID 截成 8 位用于日志。
//
// 不直接写 `id[:8]`mail_id 理论上可能短于 8 字节(畸形事件、测试桩),
// 那会 panic 在一条日志语句上 —— 日志不该有能力弄死投递协程。
func shortID(id string) string {
if len(id) <= 8 {
return id
}
return id[:8]
}

View File

@ -0,0 +1,314 @@
package main
// 投递账本的单元测试。
//
// 核心不变量只有两条,但它们互相拉扯,必须分别钉住:
//
// 1. **跑完的不再投** —— 否则跨进程重启时同一封邮件被注入两遍
// (生产实测:模型上下文里两段几乎相同的通知)。
// 2. **投过但没跑完的仍要投** —— 否则那一轮被中断时就是静默丢件,
// 发件人永远收不到回信。这条比第 1 条更重要:重复至少人能看出来。
import (
"os"
"path/filepath"
"strings"
"testing"
"time"
)
func TestClaimFreshMail(t *testing.T) {
l := openDeliveryLedger(t.TempDir())
defer l.close()
deliver, resumed := l.claim("m1")
if !deliver {
t.Fatal("全新邮件必须投递")
}
if resumed {
t.Error("全新邮件不该标记为重投")
}
}
func TestClaimEmptyIDRejected(t *testing.T) {
l := openDeliveryLedger(t.TempDir())
defer l.close()
if deliver, _ := l.claim(""); deliver {
t.Error("空 mail_id 不该被投递(畸形事件)")
}
}
func TestCompletedMailNotRedelivered(t *testing.T) {
l := openDeliveryLedger(t.TempDir())
defer l.close()
l.claim("m1")
l.complete("m1")
if deliver, _ := l.claim("m1"); deliver {
t.Error("跑完的邮件必须跳过 —— 重投会让模型上下文里出现两段相同通知")
}
}
// 这一条是整个账本存在的理由:同进程内 claim 两次的第二次是「重投」而不是「跳过」。
func TestDeliveredButNotCompletedIsResumed(t *testing.T) {
l := openDeliveryLedger(t.TempDir())
defer l.close()
l.claim("m1") // 注入了,但那一轮被中断(没有 complete
deliver, resumed := l.claim("m1")
if !deliver {
t.Fatal("投过但没跑完的必须重投 —— 跳过它就是静默丢件,发件人收不到回信")
}
if !resumed {
t.Error("必须标记为重投,好让调用方在提示词里说明「上一轮被中断」")
}
}
// 模拟真实事故:进程 A 注入后被杀,进程 B 起来重读账本。
func TestSurvivesProcessRestart_ResumesIncomplete(t *testing.T) {
dir := t.TempDir()
a := openDeliveryLedger(dir)
a.claim("m1")
a.close() // 进程 A 死了,那一轮没跑完
b := openDeliveryLedger(dir)
defer b.close()
deliver, resumed := b.claim("m1")
if !deliver {
t.Fatal("新进程必须重投未完成的邮件")
}
if !resumed {
t.Error("新进程必须知道这是重投(上一个进程注入过一次)")
}
}
// 反过来:跑完的那封,新进程必须跳过 —— 这正是修复的目标。
func TestSurvivesProcessRestart_SkipsCompleted(t *testing.T) {
dir := t.TempDir()
a := openDeliveryLedger(dir)
a.claim("m1")
a.complete("m1")
a.close()
b := openDeliveryLedger(dir)
defer b.close()
if deliver, _ := b.claim("m1"); deliver {
t.Error("跨进程也必须跳过已跑完的邮件(这就是那次重复投递的根因)")
}
}
func TestCompleteIsIdempotent(t *testing.T) {
dir := t.TempDir()
l := openDeliveryLedger(dir)
l.claim("m1")
l.complete("m1")
sizeAfterFirst := fileSize(t, filepath.Join(dir, "delivered.jsonl"))
l.complete("m1")
l.complete("m1")
l.close()
if got := fileSize(t, filepath.Join(dir, "delivered.jsonl")); got != sizeAfterFirst {
t.Errorf("重复 complete 不该再写盘:%d → %d 字节", sizeAfterFirst, got)
}
}
// complete 一封从没 claim 过的邮件(理论上不该发生)也不能让状态错乱。
func TestCompleteWithoutClaim(t *testing.T) {
dir := t.TempDir()
a := openDeliveryLedger(dir)
a.complete("ghost")
a.close()
b := openDeliveryLedger(dir)
defer b.close()
if deliver, _ := b.claim("ghost"); deliver {
t.Error("已标记完成的邮件不该被投递,即使它没走过 claim")
}
}
// 崩溃时最后一行可能写残。坏行必须被跳过,而不是让整个账本失效。
func TestCorruptLineSkipped(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "delivered.jsonl")
good := `{"m":"m-good","c":true,"t":` + itoa(time.Now().Unix()) + "}\n"
// 半行 + 完全不是 JSON 的一行
if err := os.WriteFile(path, []byte(good+`{"m":"m-half","c":fal`+"\nnot json at all\n"), 0o600); err != nil {
t.Fatal(err)
}
l := openDeliveryLedger(dir)
defer l.close()
if deliver, _ := l.claim("m-good"); deliver {
t.Error("完好的那行必须生效")
}
// 写残的那封退化为「没记录」→ 重投。重投是安全的,跳过才危险。
deliver, resumed := l.claim("m-half")
if !deliver {
t.Error("坏行对应的邮件应当重投(宁可重复也不能丢)")
}
if resumed {
t.Error("坏行没留下任何有效记录,不该被当成重投")
}
}
func TestExpiredEntriesDropped(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "delivered.jsonl")
old := time.Now().Add(-ledgerRetention - time.Hour).Unix()
fresh := time.Now().Unix()
body := `{"m":"m-old","c":true,"t":` + itoa(old) + "}\n" +
`{"m":"m-fresh","c":true,"t":` + itoa(fresh) + "}\n"
if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
t.Fatal(err)
}
l := openDeliveryLedger(dir)
defer l.close()
if deliver, _ := l.claim("m-fresh"); deliver {
t.Error("保留期内的记录必须生效")
}
if deliver, _ := l.claim("m-old"); !deliver {
t.Error("超过保留期的记录应被丢弃(否则账本无限增长)")
}
}
// 过期条目过半时压实,且压实不能弄丢仍然有效的记录。
func TestCompactPreservesLiveEntries(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "delivered.jsonl")
old := time.Now().Add(-ledgerRetention - time.Hour).Unix()
fresh := time.Now().Unix()
var sb strings.Builder
for i := 0; i < 10; i++ {
sb.WriteString(`{"m":"old-` + itoa(int64(i)) + `","c":true,"t":` + itoa(old) + "}\n")
}
sb.WriteString(`{"m":"keep","c":true,"t":` + itoa(fresh) + "}\n")
if err := os.WriteFile(path, []byte(sb.String()), 0o600); err != nil {
t.Fatal(err)
}
before := fileSize(t, path)
l := openDeliveryLedger(dir)
l.close()
if after := fileSize(t, path); after >= before {
t.Errorf("过期条目过半时应压实:%d → %d 字节", before, after)
}
// 压实后重开,仍然认得那条有效记录
l2 := openDeliveryLedger(dir)
defer l2.close()
if deliver, _ := l2.claim("keep"); deliver {
t.Error("压实不该弄丢仍然有效的记录")
}
}
// 目录不可写时退化为纯内存,而不是让插件起不来。
func TestUnwritableDirDegradesToMemory(t *testing.T) {
l := openDeliveryLedger("")
defer l.close()
if deliver, _ := l.claim("m1"); !deliver {
t.Fatal("退化为纯内存后仍要能投递")
}
l.complete("m1")
if deliver, _ := l.claim("m1"); deliver {
t.Error("纯内存模式下同进程内的去重仍要生效")
}
}
// 并发 claim 同一封:只有一个能拿到「全新」,另一个必须是「重投」。
// 这正是 SSE 与 catchUp 两个 goroutine 的竞态。
func TestConcurrentClaimSameMail(t *testing.T) {
l := openDeliveryLedger(t.TempDir())
defer l.close()
const n = 20
type res struct{ deliver, resumed bool }
out := make(chan res, n)
start := make(chan struct{})
for i := 0; i < n; i++ {
go func() {
<-start
d, r := l.claim("hot")
out <- res{d, r}
}()
}
close(start)
fresh := 0
for i := 0; i < n; i++ {
r := <-out
if !r.deliver {
t.Error("未完成的邮件在任何一次 claim 上都该返回要投递")
}
if !r.resumed {
fresh++
}
}
if fresh != 1 {
t.Errorf("恰好一次 claim 该被判为全新,实际 %d 次 —— 判定与记录必须在同一把锁里", fresh)
}
}
func TestResumeNoteMentionsInterruption(t *testing.T) {
note := resumeNote("abc-123")
for _, want := range []string{"abc-123", "中断", "同一件事"} {
if !strings.Contains(note, want) {
t.Errorf("重投说明必须包含 %q实际%s", want, note)
}
}
}
func TestShortIDDoesNotPanicOnShortInput(t *testing.T) {
// mail_id 理论上可能短于 8 字节(畸形事件、测试桩)。
// 日志不该有能力弄死投递协程。
for _, in := range []string{"", "ab", "12345678", "123456789"} {
got := shortID(in)
if len(got) > 8 {
t.Errorf("shortID(%q) = %q超过 8 字节", in, got)
}
}
}
// ─── 辅助 ───
func fileSize(t *testing.T, path string) int64 {
t.Helper()
st, err := os.Stat(path)
if err != nil {
t.Fatalf("stat %s: %v", path, err)
}
return st.Size()
}
func itoa(n int64) string {
if n == 0 {
return "0"
}
neg := n < 0
if neg {
n = -n
}
var b []byte
for n > 0 {
b = append([]byte{byte('0' + n%10)}, b...)
n /= 10
}
if neg {
return "-" + string(b)
}
return string(b)
}

View File

@ -84,8 +84,17 @@ type Plugin struct {
// B-7.3 邮件级去重SSE 重放会重发同一批事件,没有这层去重
// 每封邮件会被注入 agent 两遍。契约 B-7.3 要求:每封只注入一次。
//
// 这只挡得住**本进程内**的重复。跨进程homed 重启、插件子进程被换)
// 靠 ledger —— 它落盘,且区分「投过」与「跑完」。
deliveredMails map[string]bool
// 跨进程投递账本(见 ledger.go
//
// 它与 deliveredMails 不是重复:后者是同一进程内 SSE 重放的快速路径,
// 前者回答的是「上一个进程有没有已经把这封跑完」。
ledger *deliveryLedger
// 单调递增的 last-seen-ID被重放的旧事件不会让它回退。
// 原来直接赋值p.lastEventID = eidGateway 重放时发旧 ID
// 于是 lastEventID 从 123 退回 116 → 下次重连又报 116 → 又重放。
@ -185,6 +194,20 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
// B-1.1 解析密钥(环境变量 → 本地文件 → 生成)
p.key, p.keyFile = resolveKey()
// 跨进程投递账本子进程被换时homed 重启 / 插件崩溃自动重启)
// 内存里的 deliveredMails 全丢,只靠它防不住重复注入。
//
// DataDir() 是 SDK 保证存在的插件专属目录;拿不到时退回 key 文件所在目录
// (那个目录本来就要能写)。
dataDir := ""
if sett := s.Settings(); sett != nil {
dataDir = sett.DataDir()
}
if strings.TrimSpace(dataDir) == "" {
dataDir = filepath.Dir(p.keyFile)
}
p.ledger = openDeliveryLedger(dataDir)
// ─── 注册工具 ───
// registerTool 包一层只为计数:日志里的工具数必须与实际注册数一致。
@ -413,7 +436,14 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
}
func (p *Plugin) Stop() error {
p.stopOnce.Do(func() { close(p.stopCh) })
p.stopOnce.Do(func() {
close(p.stopCh)
// 关账本句柄。每行写入都 Sync 过,所以不关也不丢数据 ——
// 关只是为了不把 fd 泄给下一个插件实例。
if p.ledger != nil {
p.ledger.close()
}
})
return nil
}
@ -532,8 +562,31 @@ func (p *Plugin) catchUp(pending int) {
continue
}
// 构造注入消息(与 handleNewMail 一致)
prompt := fmt.Sprintf(
// 跨进程去重:这次重启前那个进程可能已经把这封跑完了。
//
// 这才是那次事故的真正修法:死掉的那个进程已经注入过一次,
// 而 deliveredMails 随它一起消失了。ledger 落盘,能说出区别:
// - 已跑完 → 真的跳过
// - 投过但未跑完(上一轮被中断)→ 仍然重投,但带上说明
// 后一条很要紧:那一轮被中断意味着发件人没收到回信,跳过它就是静默丢件。
deliver, resumed := p.ledger.claim(m.MailID)
if !deliver {
log.Printf("[homeagent-mail-bridge] 补投跳过 %s已在之前的进程里处理完毕", shortID(m.MailID))
continue
}
if resumed {
log.Printf("[homeagent-mail-bridge] 补投 %s上一轮被中断带说明重投", shortID(m.MailID))
}
// 注入消息。与 handleNewMail 那份的差异只在一句措辞上(这里不说
// 「把本轮工作做完」)—— 那个差异正好是上次定位重复投递的线索:
// 两段提示词同时出现在上下文里,一看措辞就知道一段来自 SSE、
// 一段来自补投。
prefix := ""
if resumed {
prefix = resumeNote(m.MailID)
}
prompt := prefix + fmt.Sprintf(
"你收到一封新邮件AgentMail。\n\n"+
"发件人:%s\n主题%s\n邮件 ID%s\n身份你是 %s\n\n"+
"请先调用 read_inbox 读取完整正文,然后处理其中的请求。\n\n"+
@ -545,8 +598,9 @@ func (p *Plugin) catchUp(pending int) {
reply := p.sdk.InjectInputSync(p.name, p.name, prompt)
if reply == "" {
// B-6模型没回发一封告知
// B-6模型没回发一封告知。发出去就算处理完(理由同 handleNewMail
p.sendFailureReply(m.FromName, m.Subject, m.MailID, "模型未产生回复")
p.ledger.complete(m.MailID)
continue
}
// B-5.3:检查模型是否已经自己发过信
@ -557,11 +611,17 @@ func (p *Plugin) catchUp(pending int) {
if sent {
// 模型已经在这一轮里自己回了这封信,不再重复 relay
p.ledger.complete(m.MailID)
continue
}
// B-5.2:自动回信带 relay:"summary" —— 搬运不算模型自主发信,不扣配额
p.sendMailRelay(m.FromName, "Re: "+m.Subject, reply, m.MailID, "homeagent:"+m.MailID)
if err := p.sendMailRelay(m.FromName, "Re: "+m.Subject, reply, m.MailID, rk); err != nil {
// 回信没发出去 —— 不标完成,下次重启重试。
log.Printf("[homeagent-mail-bridge] 补投回信失败(不标完成): %v", err)
continue
}
p.ledger.complete(m.MailID)
}
}
@ -707,12 +767,22 @@ func (p *Plugin) parseSSELine(line string) {
p.deliveredMails[evt.MailID] = true
p.sseMu.Unlock()
// 跨进程去重:上一个插件子进程可能已经把这封跑完了。
// deliveredMails 只在本进程内有效homed 重启会把它清空 ——
// 实测过一次两段几乎相同的通知堆在模型上下文里(一段来自这里的
// SSE 路径,一段来自重启后的 catchUp
deliver, resumed := p.ledger.claim(evt.MailID)
if !deliver {
log.Printf("[homeagent-mail-bridge] 邮件 %s 已在之前的进程里处理完毕,跳过", shortID(evt.MailID))
return
}
// InjectInputSync 会阻塞几十秒(查日志、调工具、转发 QQ
// 而它跑在 readSSE 的读循环里 —— 循环卡住期间 SSE 事件积压在
// TCP 缓冲区,卡到超时断线重连后 Gateway 全部重放一遍。
// 把处理丢到独立 goroutineparseSSELine 立刻返回,读循环继续。
// homeagent 是单事件循环InjectInputSync 自己会排队。
go p.handleNewMail(evt)
go p.handleNewMail(evt, resumed)
}
}
@ -822,8 +892,18 @@ func (p *Plugin) handleNewMail(evt struct {
Workspace string `json:"to_workspace"`
Alias string `json:"session_alias"`
ReplyAddr string `json:"reply_address"`
}) {
prompt := fmt.Sprintf(
}, resumed bool) {
// resumed = 上一个进程注入过这封但那一轮被中断了。
//
// 必须把这件事告诉模型:不说的话它在上下文里看到两段几乎相同的指令,
// 会以为人重复交代了一遍,于是可能把同一件事做两次。
prefix := ""
if resumed {
prefix = resumeNote(evt.MailID)
log.Printf("[homeagent-mail-bridge] 邮件 %s 重投(上一轮被中断)", shortID(evt.MailID))
}
prompt := prefix + fmt.Sprintf(
"你收到一封新邮件AgentMail。\n\n"+
"发件人:%s\n主题%s\n邮件 ID%s\n身份你是 %s\n\n"+
"请先调用 read_inbox 读取完整正文,然后处理其中的请求。\n\n"+
@ -839,8 +919,12 @@ func (p *Plugin) handleNewMail(evt struct {
// B-6模型没回空 = turn/end 信号 kind=error或模型没说话
if reply == "" {
log.Printf("[homeagent-mail-bridge] 邮件 %s来自 %s%sagent 无回复,发失败通知",
evt.MailID[:8], evt.FromName, evt.Subject)
shortID(evt.MailID), evt.FromName, evt.Subject)
p.sendFailureReply(evt.FromName, evt.Subject, evt.MailID, "模型未产生回复")
// 失败通知发出去了就算**处理完**:发件人得到了一个明确的交代。
// 不标的话下次重启会把同一封再投一遍 —— 而模型上一次就没回,
// 重投只会再发一封相同的失败通知。
p.ledger.complete(evt.MailID)
return
}
@ -855,15 +939,19 @@ func (p *Plugin) handleNewMail(evt struct {
if sent {
// 模型已经在这一轮里自己回了这封信,让位
log.Printf("[homeagent-mail-bridge] 邮件 %s 模型已自行回复,跳过自动 relay", evt.MailID[:8])
log.Printf("[homeagent-mail-bridge] 邮件 %s 模型已自行回复,跳过自动 relay", shortID(evt.MailID))
p.ledger.complete(evt.MailID)
return
}
// B-5.2:自动回信带 relay:"summary" + relay_key
if err := p.sendMailRelay(evt.FromName, "Re: "+evt.Subject, reply, evt.MailID, rk); err != nil {
log.Printf("[homeagent-mail-bridge] 自动回信失败: %v", err)
// 回信没发出去 —— **不标完成**,让下次重启能重试。
// 发件人至今一个字都没收到,这时标「已完成」就是静默丢件。
log.Printf("[homeagent-mail-bridge] 自动回信失败(不标完成,下次会重试): %v", err)
} else {
log.Printf("[homeagent-mail-bridge] 已自动回信给 %s%d 字)", evt.FromName, len(reply))
p.ledger.complete(evt.MailID)
}
// 清理过期的 explicitSends 记录

View File

@ -4,18 +4,32 @@
*
* 形态是**常驻守护进程**,不是 pi 扩展。原因见 src/session-pool.mjs 顶部:
* 扩展被加载进一条已存在的会话cwd 由启动 pi 的人决定;而 B-3.1 要求每封邮件的
* to_workspace 成为会话 cwd。桥用 SDK 的 createAgentSession 按邮件起会话,
* 一个进程里并存多条不同 cwd 的会话(实测可行)。
* to_workspace 成为会话 cwd。
*
* # 进程结构
*
* 主进程**只做 I/O 与调度**SSE、心跳、去重、把邮件派给子进程。模型工作全部
* 下到 `src/worker.mjs`(一封邮件一个进程,跑完就退),由 `src/pool.mjs` 调度。
*
* 这不是为了并行度,是为了**不阻塞事件循环**。pi 的会话装载是同步的:
* `SessionManager.open()` 走 `openSync` + `readSync` 循环把整个 `.jsonl` 读进内存
* 并逐行 JSON.parse。实测本机最大那条会话 23MB`open` 一次阻塞事件循环 118ms
* 模型跑起来之后 SDK 内部还有更多同步工作。原来这些都在主线程上 —— SSE 读循环
* 在那期间完全停住,后续邮件卡在 TCP 缓冲区,久到 Gateway 认为连接死了,
* 重连又触发重放。实测同样的活在 fork 出的子进程里跑,主进程阻塞 0ms。
*
* 权限询问期间的挂起也随之只影响那一个 worker原来 `await new Promise(...)`
* 等人做决定,整座桥在那段时间不再收信。
*
* 契约实现对照docs/PLUGIN-CONTRACT.md
* B-1 启动 → main()
* B-2 心跳 → beat()30 秒
* B-3 new_mail → deliverMail()
* B-3 new_mail → pool.submit()(投递本体在 worker.mjs
* B-4 决策 → handlePermissionDecision()
* B-5 转发 → relaySummary(),挂在 agent_end 上
* B-6 失败回信 → deliverMail() 末尾的 renderFailureReport
* B-5 转发 → worker.mjs 的 relaySummary()
* B-6 失败回信 → worker.mjs 末尾的 renderFailureReport
* B-7 补拉 → catchUp()
* B-8 权限 → permissionExtension() 的 tool_call 钩子
* B-8 权限 → worker.mjs 的 permissionExtension()
* B-9 关停 → shutdown()
*/
@ -26,16 +40,11 @@ import { ModelRuntime } from '@earendil-works/pi-coding-agent';
import { GatewayClient, readLocalKey, generateLocalKey, saveConfig } from './gateway.mjs';
import { createMailTools } from './tools.mjs';
import { openSession, runTurn } from './session-pool.mjs';
import { buildMailPrompt, lastAssistantText, replySubject, relayKeyFor, describeError } from './turn.mjs';
import { planNamingSync, planWriteBack } from './naming.mjs';
import { resolveWorkspaceCwd, ensureCwd } from '../lib/workspace.js';
import { modelAttemptOrder, renderFailureReport, snapshotPiModels } from '../lib/model-scope.js';
import { createWorkerPool } from './pool.mjs';
import { describeError } from './turn.mjs';
import { snapshotPiModels } from '../lib/model-scope.js';
import { snapshotPiSessions } from '../lib/session-snapshot.js';
import { selectCatchup } from '../lib/catchup.js';
import { explicitSends, shouldSkipAutoRelay } from '../lib/relay-dedup.js';
import { adoptedSessionID, adoptMissingMessage } from '../lib/adopt.js';
import { createGrantStore, isApproval } from '../lib/permission-grants.js';
// ─── 配置 ───
@ -44,7 +53,34 @@ const AGENT_NAME = process.env.AGENTMAIL_AGENT_NAME || 'pi';
const AGENT_SECRET = process.env.AGENTMAIL_AGENT_SECRET || '';
const REPLY_PROVIDER = process.env.AGENTMAIL_REPLY_PROVIDER || '';
const REPLY_MODEL = process.env.AGENTMAIL_REPLY_MODEL || '';
const TURN_TIMEOUT_MS = Number(process.env.AGENTMAIL_TURN_TIMEOUT_MS || 60_000);
/**
* 单轮超时。
*
* 从 60 秒放宽到 10 分钟60 秒那个数字是「主进程要腾出手来收下一封邮件」的
* 产物 —— 超时按成功返回,好让 deliverMail 早点结束。worker 没有这个理由,
* 它只为这封邮件活着,等真结论更准。带工具调用的一轮跑几分钟很正常,
* 60 秒返回会让转发落在一个还没说完的结论上。
*/
const TURN_TIMEOUT_MS = Number(process.env.AGENTMAIL_TURN_TIMEOUT_MS || 600_000);
/**
* 并发上限。
*
* 每个 worker 约 140MB RSS实测并且每个都会对上游 provider 发请求。
* 3 是内存与吞吐的折中;同一条会话无论如何都是串行的(见 pool.mjs
*/
const MAX_WORKERS = Number(process.env.AGENTMAIL_MAX_WORKERS || 3);
/**
* worker 硬超时。
*
* 比轮次超时留出余量:正常情况下 worker 自己会在轮次超时后收尾退出,
* 这个数字兜的是「连收尾都没做」(进程卡死、权限等不到决策而决策事件也丢了)。
* 到点 SIGKILL —— 否则那条会话的后续邮件永远排队。
*/
const WORKER_MAX_MS = Number(process.env.AGENTMAIL_WORKER_MAX_MS || TURN_TIMEOUT_MS + 120_000);
const LOCK_FILE = join(process.env.AGENTMAIL_CONFIG_DIR || join(homedir(), '.agentmail'), 'pi-bridge.lock');
/** 日志一律 console.error它一定进 journalctl契约 9.8)。 */
@ -52,23 +88,16 @@ const log = (...args) => console.error('[pi-mail-bridge]', ...args);
// ─── 进程内状态 ───
//
// 全部在内存重启即丢 —— 这是契约第六节列明的已知取舍。
// 要持久化的话该落在 pi 的会话元数据里,而不是桥自己的文件。
// 主进程只留「调度需要的」那几样,全部在内存重启即丢契约第六节的已知取舍
// 会话映射、权限授权、命名指纹都下沉到 pool 里按邮件会话存 —— 主进程不再持有
// AgentSession 对象(那东西跨不了进程边界)。
const sessions = new Map(); // agentmail session_id -> { session, sessionManager, cwd }
const reverseMap = new Map(); // pi session id -> agentmail session_id
const mailDriven = new Set(); // pi session id
const mailContexts = new Map(); // agentmail session_id -> { replyTo, subject, mailID }
const relayedSummaries = new Map(); // pi session id -> 已转发过的 relay_key
const syncedNames = new Map(); // pi session id -> 上次提交给 Gateway 的名字
const pendingPermissions = new Map(); // relay_key -> { resolve, piSessionId }
// 人点过「一直同意」的 (会话, 工具)。作用域与清理语义见 lib/permission-grants.js。
const permissionGrants = createGrantStore();
const deliveredMails = new Set(); // 已投过的 mail_idSSE 与补拉共用B-7.3
let allowedModels = [];
let modelRuntime = null;
let client = null;
let pool = null;
let heartbeatTimer = null;
let shuttingDown = false;
@ -113,648 +142,31 @@ function releaseLock() {
} catch { /* 已经没了 */ }
}
// ─── 权限钩子B-8───
/**
* 内联 pi 扩展:把 pi 拦下的危险工具调用转成一封邮件问人。
*
* 这是 `I-1` 最直接的体现 —— 被平台真正拦下的那一次才是事实,
* 不依赖模型「记得」调 request_permission它会忘也会在不需要时乱调
*
* pi 的 `tool_call` 钩子**可以 await**C-9 实测成立:处理器里 await 300ms
* 再返回 {block:true}pi 会等),所以这里能真的等人做决定,
* 不必走「先拒一次再重试」的退化路径。
*
* @param {string} piSessionIdRef 用一个 getter 拿会话 id扩展工厂在
* createAgentSession **内部**被调用,那时 session 对象还没返回给桥。
*/
function permissionExtension(getMailContext) {
// pi 默认放行内建工具;桥只拦真正有副作用的那几个。
// read/grep/ls 之类不拦:每一步都问人会让 Agent 什么也做不成,
// 而人也会很快开始无脑点同意(那比不问更危险)。
const GUARDED = new Set(['bash', 'write', 'edit']);
return (pi) => {
pi.on('tool_call', async (event, ctx) => {
if (!GUARDED.has(event.toolName)) return;
const piSessionId = ctx?.sessionManager?.getSessionId?.() || '';
const mailSessionId = reverseMap.get(piSessionId);
// 不是邮件驱动的会话 → 让位给 pi 自己的本地 UIB-8.2)。
// 占着钩子不放会让人在 TUI 里干活时每一步都卡住等邮件。
if (!mailSessionId) return;
// 人对这条会话的这个工具点过「一直同意」→ 直接放行,不再发邮件。
// 这一步必须在 POST 之前:否则每条命令都生成一封邮件,人点过的
// 「一直同意」形同虚设(实测同一条会话被问了 15 次 bash
if (permissionGrants.isGranted(piSessionId, event.toolName)) {
return;
}
// relay_key 用 pi 给的 toolCallIdB-8.1):服务端会随决策事件回传它,
// 桥重启丢了 pendingPermissions 也能对上B-4.2)。自造随机 id 做不到。
const relayKey = `${piSessionId}:${event.toolCallId}`;
const ctxInfo = getMailContext(mailSessionId);
try {
// **不传 `to`**(这里曾经传 `ctxInfo.replyTo`,那是个死锁 bug
//
// replyTo 是来信人的名字,而来信人可能是另一个 Agent —— Agent 把任务
// 分派给自己的另一条会话时pi→pi权限邮件就发给了 `pi` 自己。
// 后果是死锁而不是报错Agent 不可能在 Web 界面上点「同意」,
// 服务端的 SendToUser 又投进一个不存在的用户通道(没有任何人被提醒),
// 于是下面那个 await 永不 resolve —— 会话永久挂死,没有超时、没有日志。
//
// 决策人交给服务端定:它按 会话 owner → 线索里最近的人类 → 无人可问则
// 409 的顺序解析,那是唯一能看到整条线索的地方。插件只有本地那点上下文,
// 猜不出「这条 Agent 链最初是谁派的活」。
await client.post('/permission/request', {
question: `是否允许执行 ${event.toolName}`,
options: ['同意', '一直同意', '拒绝'],
// 带上触发这次询问的来信B-8.4)。决策人未必是这条会话的参与者 ——
// Agent 转派出来的会话,人从没见过它,只给一句「是否允许执行 bash」
// 是无从判断的:得知道这活是谁派的、为的什么事。
context: [
describeToolCall(event),
ctxInfo?.subject ? `\n触发任务:${ctxInfo.subject}` : '',
ctxInfo?.replyTo ? `任务来自:${ctxInfo.replyTo}` : '',
].filter(Boolean).join('\n'),
session_id: mailSessionId,
relay_key: relayKey,
});
} catch (e) {
// 409 = 服务端已判定这条任务链上没有人类,永远不会有人来点头。
//
// 不能「让位给本地 UI」邮件驱动的会话没有 TUI让位之后 pi 按默认
// 策略处置,而默认策略在没有交互端时就是等 —— 又一次无声挂死。
//
// 直接 block 并把服务端的建议原文当作 reason模型从工具报错里
// 看到「这条链上没人可问,换不需要权限的方式」才能自己改道,
// 而挂死时它连重试的机会都没有。
if (e?.status === 409) {
const b = e.body || {};
const reason = [
b.error || '权限询问无法送达:这条任务链上没有人类用户',
b.detail || '',
b.suggestion || '',
].filter(Boolean).join('\n');
log(`权限询问无人可投,当场拒绝 ${relayKey}${b.error || ''}`);
return { block: true, reason };
}
// 其余失败网络抖动、Gateway 重启)是暂时的 → 让位给 pi 本地 UIB-8.2)。
// 返回 undefined 表示「这个钩子不表态」pi 会走它自己的批准流程。
log(`权限转发失败,让位给本地决策: ${describeError(e)}`);
return;
}
log(`权限询问已发出(${event.toolName}key=${relayKey}),等待决策…`);
const decision = await new Promise((resolve) => {
pendingPermissions.set(relayKey, { resolve, piSessionId });
});
// fail closedB-9.2 / N-9只有明确的同意才放行。
// 关停时 shutdown() 会用 'shutdown' 唤醒所有等待者,落到这里的 else。
if (isApproval(decision)) {
// 「一直同意」要真的记住,否则这个选项是在骗人:人点了它,
// 下一条命令照样来一封邮件。grant() 内部只认精确的 always 文本 ——
// 「同意」是单次授权,把它当 always 会放行人没看过的后续命令。
if (permissionGrants.grant(piSessionId, event.toolName, decision)) {
log(`本会话的 ${event.toolName} 已获「一直同意」,后续不再询问`);
}
log(`权限 ${relayKey} 获批(${decision}),放行 ${event.toolName}`);
return;
}
return { block: true, reason: `用户${decision === 'shutdown' ? '未及决策(桥已关停)' : `拒绝了这次 ${event.toolName} 调用`}` };
});
};
}
/** 把一次工具调用摘要成人能判断的文本B-8.4)。 */
function describeToolCall(event) {
const input = event?.input ?? {};
if (event.toolName === 'bash') {
return `命令:\n${String(input.command ?? '').slice(0, 800)}`;
}
if (event.toolName === 'write' || event.toolName === 'edit') {
return `文件:${input.file_path ?? input.path ?? '(未给出)'}`;
}
return JSON.stringify(input).slice(0, 800);
}
// ─── 会话解析B-3───
/**
* 没有可用 `to_workspace` 时的兜底目录。
*
* 与 DSH 的 `mailSessionFallback` 同构,但目录名是 `.pi`:那个函数在
* lib/ 下(三平台逐字节相同),写死了 `.dsh`,不能为 pi 改。
* 让 pi 的会话落进 `~/.dsh/` 会让人以为是 DSH 在干活。
*/
function piMailFallback(sessionKey) {
return join(homedir(), '.pi', 'mail-sessions', String(sessionKey || 'default'));
}
/**
* 接管过的 pi 会话pi session id。用完即释放见 releaseAdopted。
*
* 与 `sessions` 的区别:那里存的是桥自己起的、长期持有的会话;这里是
* 「借用一下磁盘上人家的会话」,一轮结束就还回去。
*/
const adopted = new Set();
/** 接管会话的兜底释放计时器pi session id -> Timeout。 */
const adoptTimers = new Map();
/**
* 接管会话最长持有多久。
*
* 取轮次超时的两倍:`runTurn` 60 秒就按成功返回(长任务很正常,判成失败会
* 换模型重跑一遍),但会话仍在跑。正常结束走 agent_end 提前释放,这个数字
* 只兜「事件永远不来」的底。
*/
const ADOPT_MAX_HOLD_MS = TURN_TIMEOUT_MS * 2;
/**
* 接管一条磁盘上已经存在的 pi 会话,把这封邮件投进去。
*
* # 为什么必须**短暂持有**
*
* pi 没有任何锁机制,它假定「一个文件一个持有者」。活着的 SessionManager
* 不 watch 文件外部TUI追加的行它看不见之后它自己的写入算出的 parentId
* 指向一个对方不知道的 entry —— 文件不会坏(写入是纯 append但会话树分叉。
*
* 所以这里 open → 跑一轮 → 丢弃,**不放进 sessions 长期缓存**。下一封邮件
* 再来时重新 open那一次读到的就是 TUI 期间写的全部内容。
*
* 窗口是一轮对话的时长。人正好在这期间也在 TUI 里发消息仍会分叉,但那需要
* 两边同时动手,且后果是历史看起来少一段,不是数据损坏。
*
* # 为什么不校验「TUI 是否正开着这条会话」
*
* pi 不提供这个信息(没有 lockfile、没有 pid 记录)。能做的只有猜 mtime
* 而任何阈值都是猜。与其用一个猜出来的数字拒掉合法投递,不如让窗口尽量短。
*/
async function adoptSession(platformID, data, mailTools) {
const mailSessionID = data.session_id;
const { SessionManager } = await import('@earendil-works/pi-coding-agent');
// listAll 而不是 list(cwd):桥的进程 cwd 与会话 cwd 无关。
const all = await SessionManager.listAll();
const info = all.find((e) => e?.id === platformID);
if (!info?.path) {
// 镜像是快照,可以过期:那条会话可能已经被删了。
// **不能**退回「新建一条」—— 那会让人在 TUI 里看不到这封邮件带来的对话,
// 而那正是接管的目的N-8 同理:静默改语义比报错糟)。
throw new Error(adoptMissingMessage(platformID, '磁盘上已无这个会话文件'));
}
// cwd 取会话自己的SessionInfo.cwd 来自持久化 header
// 老会话的 cwd 是空串,那种情况退回地址里的 path 位。
const { cwd } = resolveWorkspaceCwd(
info.cwd || data.to_workspace, piMailFallback(mailSessionID));
const opened = await openSession({
cwd,
modelRuntime,
customTools: mailTools,
extension: permissionExtension((id) => mailContexts.get(id)),
sessionFile: info.path,
});
for (const d of opened.diagnostics) {
log(`扩展诊断: ${d?.message ?? JSON.stringify(d)}`);
}
const piSessionId = opened.session.sessionId;
const entry = { session: opened.session, sessionManager: opened.sessionManager, cwd };
if (mailSessionID) {
sessions.set(mailSessionID, entry);
reverseMap.set(piSessionId, mailSessionID);
// 加进 mailDriven接管之后这条会话**开始**参与邮件往来,轮次结束要把
// 总结转回发件人。不加的话邮件投进去了却永远没有回音。
mailDriven.add(piSessionId);
adopted.add(piSessionId);
}
// 兜底释放:`agent_end` 不来就永远握着这个文件,而握着它的期间 TUI 那边
// 的写入对我们不可见 —— 正是要避免的分叉窗口。会话跑挂、事件丢失、
// 模型一直不结束都属于这种情形。
//
// 时长取轮次超时的两倍runTurn 自己 60 秒就按成功返回了(长任务很正常),
// 那之后会话仍在跑,正常结束时 agent_end 会照常触发并提前释放。
const safety = setTimeout(() => {
if (!adopted.has(piSessionId)) return;
log(`接管会话 ${piSessionId} 超过 ${ADOPT_MAX_HOLD_MS / 1000}s 未结束,强制释放`);
releaseAdopted(piSessionId, mailSessionID);
}, ADOPT_MAX_HOLD_MS);
if (typeof safety.unref === 'function') safety.unref();
adoptTimers.set(piSessionId, safety);
// 只挂 agent_end不挂 session_info_changed改名同步会把 Gateway 侧的别名
// 覆盖成 pi 的标题,而接管会话的别名是人从补全里选的那个 slug ——
// 改掉会让他找不到自己刚发的信。
opened.session.subscribe((event) => {
if (event?.type !== 'agent_end') return;
if (event.willRetry) return;
// 转发完再释放relaySummary 要读 sessions 里的 entry。
relaySummary(piSessionId)
.catch((e) => log(`自动转发失败: ${describeError(e)}`))
.finally(() => {
// **还在跑就不能释放。**
//
// 同一条会话可能已经排了下一封邮件runTurn 在 isStreaming 时走
// `streamingBehavior: 'followUp'`,那封信排在当轮之后。此时 dispose
// 会把排着的那一轮一起杀掉 —— 发件人只看到信发出去后再无音讯。
// 排着的那轮结束时会再触发一次 agent_end由它来释放。
if (opened.session.isStreaming) {
log(`接管会话 ${piSessionId} 仍有排队轮次,暂不释放`);
return;
}
releaseAdopted(piSessionId, mailSessionID);
});
});
log(`接管 pi 会话 ${piSessionId}cwd=${cwd},文件 ${info.path}`);
// reused: true —— 这条会话有历史,提示词不该重新自我介绍,
// 且 deliverMail 的续谈支不做模型降级(换模型要换会话,会丢掉整条上下文)。
return { ...entry, reused: true };
}
/**
* 还回一条接管来的会话dispose + 清缓存。
*
* mailDriven 不清:它同时喂给心跳快照的 mail_driven 标记,那条平台会话
* 确实已经在邮件往来里了。reverseMap 也不清 —— 留着让迟到的事件能找到线索,
* 而 relaySummary 在 entry 缺失时会自己早退。
*/
function releaseAdopted(piSessionId, mailSessionID) {
if (!adopted.has(piSessionId)) return;
adopted.delete(piSessionId);
const timer = adoptTimers.get(piSessionId);
if (timer) {
clearTimeout(timer);
adoptTimers.delete(piSessionId);
}
const entry = mailSessionID ? sessions.get(mailSessionID) : undefined;
if (entry?.session?.sessionId === piSessionId) {
sessions.delete(mailSessionID);
}
try {
entry?.session?.dispose?.();
} catch (e) {
log(`释放接管会话失败(不影响后续): ${describeError(e)}`);
}
log(`释放接管会话 ${piSessionId}(文件已交还,下一封邮件重新打开)`);
}
/**
* 找到(或建立)这封邮件该落进的 pi 会话。
*
* Gateway 已经按三维地址的 session 位做完了「复用默认 / 新建 / 具名必须存在」
* 的判定,推来的 session_id 就是判定结果 —— 桥只负责忠实映射,
* 不自己决定开不开新会话N-8404 后自动改用 .new 是禁止的)。
*/
async function resolveSession(data, mailTools) {
const mailSessionID = data.session_id;
const bound = mailSessionID ? sessions.get(mailSessionID) : undefined;
if (bound) return { ...bound, reused: true };
// 服务端说这条邮件会话**接管了平台上已经存在的那条会话**(人在 TUI 里开的
// 那种)—— 投进它而不是新建。TUI 与邮箱是同一个 Agent 的两个入口。
const adoptedID = adoptedSessionID(data);
if (adoptedID) {
return await adoptSession(adoptedID, data, mailTools);
}
// cwd 取寻址里的 path 位B-3.1)。校验走共用模块:目录不存在时**不创建**
// N-2笔误会在磁盘上落下真目录而 Agent 在里面一无所获拒绝相对路径N-3
//
// 兜底用 `~/.pi/mail-sessions/<会话>` 而不是共用模块里的 mailSessionFallback ——
// 后者写死了 `.dsh` 目录名(那是 DSH 的家pi 的会话落进去会让人以为
// DSH 在干活。lib/ 里的函数三平台逐字节相同,不能为 pi 改它。
const { cwd, grouped } = resolveWorkspaceCwd(data.to_workspace, piMailFallback(mailSessionID));
if (!grouped && data.to_workspace) {
log(`工作目录 ${data.to_workspace} 不可用,回退到 ${cwd}`);
}
ensureCwd(cwd, grouped);
const opened = await openSession({
cwd,
modelRuntime,
customTools: mailTools,
extension: permissionExtension((id) => mailContexts.get(id)),
});
for (const d of opened.diagnostics) {
log(`扩展诊断: ${d?.message ?? JSON.stringify(d)}`);
}
const piSessionId = opened.session.sessionId;
const entry = { session: opened.session, sessionManager: opened.sessionManager, cwd };
if (mailSessionID) {
sessions.set(mailSessionID, entry);
reverseMap.set(piSessionId, mailSessionID);
mailDriven.add(piSessionId);
}
// 一轮结束就转发总结B-5。挂 agent_end 而不是 message_end
// 后者在流式生成中反复触发,转出去的是半截话。
// subscribe 收的是一个普通函数AgentSessionEventListener不是 {onEvent}。
opened.session.subscribe((event) => {
if (event?.type === 'agent_end') {
// willRetry 为真表示 pi 自己要重试auto_retry这一轮还没定论 —— 不转。
if (event.willRetry) return;
relaySummary(piSessionId).catch((e) => log(`自动转发失败: ${describeError(e)}`));
}
// pi 侧改名pi-web 生成标题、人在 TUI 里 /name→ 同步给 Gateway
if (event?.type === 'session_info_changed') {
syncNaming(piSessionId, event.name).catch((e) => log(`命名同步失败: ${describeError(e)}`));
}
});
log(`新建 pi 会话 ${piSessionId}cwd=${cwd}`);
return { ...entry, reused: false };
}
// ─── 命名一致C-11 / W-7───
/**
* pi 的名字 → Gateway → 定稿别名回写进 pi。
*
* 完整推理见 src/naming.mjs 顶部。这里只是把那套决策接上 I/O。
*/
async function syncNaming(piSessionId, platformName) {
const mailSessionID = reverseMap.get(piSessionId);
if (!mailSessionID) return; // 不是邮件驱动的会话,不碰
const plan = planNamingSync({
platformName,
mailSubject: mailContexts.get(mailSessionID)?.subject,
lastSynced: syncedNames.get(piSessionId),
});
if (plan.skip) return;
// 先记下指纹再发请求:响应回来时 setSessionName 会再次触发
// session_info_changed这一步是防自激循环的关键。
syncedNames.set(piSessionId, plan.signature);
const res = await client.post(`/sessions/${mailSessionID}/sync`, {
alias: plan.alias,
title: plan.title,
});
const entry = sessions.get(mailSessionID);
const back = planWriteBack({
finalAlias: res?.alias,
currentPiName: entry?.session?.sessionName,
});
log(`命名同步 ${piSessionId}: alias=${res?.alias || '(未变)'} 来源=${plan.source}`);
if (back.write && entry?.session) {
// 顺序要紧:先更新指纹,再改名。
//
// setSessionName **同步**触发 session_info_changed实测于是本函数会在
// 这一行里被重入。指纹在改名之后才更新的话,重入那次看到的还是旧指纹,
// 于是又打一次 sync —— 每条会话两次请求,内容完全相同。
//
// 记的是「把定稿别名当作平台名字」会算出的指纹:重入那次的 platformName
// 正是 back.name来源判定成 platform算出来的就是这个值。
syncedNames.set(piSessionId, `platform:${back.name}|${back.name}`);
// 只用 setSessionName走 pi 自己的写入路径)。绝不自己拼路径写会话文件:
// 首条 assistant 消息落盘前文件还不存在pi 首次落盘用 openSync(file,"wx")
// 抢先创建会让它抛 EEXIST实测
entry.session.setSessionName(back.name);
log(`别名回写 pi${back.name}${back.reason}`);
}
}
// ─── 自动转发B-5───
async function relaySummary(piSessionId) {
const mailSessionID = reverseMap.get(piSessionId);
if (!mailSessionID) return;
// 只对邮件驱动的会话转发B-5.5):人在 pi 里正常干活时不该往邮箱灌总结
if (!mailDriven.has(piSessionId)) return;
const entry = sessions.get(mailSessionID);
if (!entry) return;
// 一轮结束是命名的自然时机C-11 / D-5
//
// 这一步不能只挂在 session_info_changed 上:桥用 SDK 起的会话**永远不会**
// 触发那个事件 —— pi 的标题生成器在 pi-web 里不在内核里SDK 路径上没有它。
// 只等事件的话别名永远是空的,于是 `name@path.<别名>` 续谈无从下手
// 实测过第一封邮件跑通了sessions.session_alias 仍是空串)。
//
// 放在转发**之前**:回信里会带上会话别名,收件人看到的第一封回信就能用它续谈。
//
// **接管来的会话跳过这一步。**
//
// 它的名字是人在 TUI 里定的,也是他从补全里选中的那个 slug。同步会双向改坏它
// 别名撞上本侧已有会话时 Gateway 加后缀(`agent-only-chain` →
// `agent-only-chain-2`),而定稿别名又会**回写进 pi 的会话文件** ——
// 于是下一次心跳上报的 slug 变成加了后缀那个,人从补全里选的名字凭空消失。
// 实测撞出来过一次。
//
// 接管会话的别名由服务端在接管时按 slug 定好AdoptPlatformSession
// 这里不需要也不应该再动它。
if (!adopted.has(piSessionId)) {
await syncNaming(piSessionId, entry.session.sessionName)
.catch((e) => log(`命名同步失败: ${describeError(e)}`));
}
// 只取 type==='text' 的块B-5.1 / N-6thinking 是思考过程,不是结论
const text = lastAssistantText(entry.session.messages);
if (!text) return; // 空文本不发空邮件B-5.4
const ctx = mailContexts.get(mailSessionID);
if (!ctx?.replyTo) return; // 不知道回给谁
// 幂等键用 pi 的会话 id + 会话树叶子 id两者都落盘重启重放也是同一个键。
const relayKey = relayKeyFor(piSessionId, entry.sessionManager.getLeafId?.());
if (relayedSummaries.get(piSessionId) === relayKey) return;
// 模型这一轮已亲手回过这条线索 → 让位B-5.3)。
// 否则收件箱里是两封说同一件事的邮件(生产实测过)。
if (shouldSkipAutoRelay(explicitSends.get(piSessionId), ctx.replyTo, ctx.mailID)) {
explicitSends.delete(piSessionId);
relayedSummaries.set(piSessionId, relayKey);
log(`本轮模型已主动回信 ${ctx.replyTo},跳过自动转发`);
return;
}
await client.post('/mail/send', {
to: ctx.replyTo,
subject: replySubject(ctx.subject),
body: text,
reply_to: ctx.mailID || '',
// relay + relay_key 走免配额通道I-2模型已经把话说完了
// 桥只是把它搬到邮件里。对搬运收费会让配额用尽时 Agent 连交代都做不了。
relay: 'summary',
relay_key: relayKey,
});
relayedSummaries.set(piSessionId, relayKey);
explicitSends.delete(piSessionId); // 一轮结束,窗口关闭
log(`已转发本轮总结给 ${ctx.replyTo}${text.length} 字)`);
}
// ─── 投递B-3 / B-6───
async function deliverMail(data, kind, mailTools) {
const { session, reused } = await resolveSession(data, mailTools);
const piSessionId = session.sessionId;
// 新一轮开始:清掉上一轮「模型主动发过信」的记录。不清的话,
// 上一轮亲手回过信会永久压掉这个会话之后所有的自动转发。
explicitSends.delete(piSessionId);
if (kind === 'mail' && data.session_id) {
// 一个会话里可能来过多封信,只留最近那封 —— 回信要落回最新的线索
mailContexts.set(data.session_id, {
replyTo: data.from_name || '',
subject: data.subject || '',
mailID: data.mail_id || '',
});
}
const prompt = buildMailPrompt({ agentName: AGENT_NAME, data, kind, reused });
// 续谈会话已经存在模型也已经定了pi 的模型在 createAgentSession 时绑定),
// 所以这一支不做模型降级。runTurn 内部按 isStreaming 分流:
// 空闲就直接起一轮,正在跑就排到当轮之后(不打断上一封邮件的工作)。
if (reused) {
const outcome = await runTurn(session, prompt, TURN_TIMEOUT_MS);
log(`续谈 ${piSessionId}mail ${data.mail_id}${outcome.queued ? ',已排队' : ''}`);
// 续谈失败不换模型重试(换模型要换会话,会丢掉整条上下文 ——
// 而上下文正是发件人指定这条会话的原因),但要让失败可见。
if (!outcome.ok) throw new Error(`续谈失败: ${outcome.error}`);
return;
}
// 按管理员划定的范围逐个尝试D-3
// 关键点:`prompt()` resolve **不代表模型跑成功了** —— 无凭证的 provider
// 会让它 reject实测 `No API key found for amazon-bedrock.`
// 而上游报错走 stopReason==='error'。判定交给 classifyTurnOutcome。
const attempts = modelAttemptOrder(allowedModels, {
provider: REPLY_PROVIDER,
model: REPLY_MODEL,
});
const failures = [];
for (const route of attempts) {
const label = route ? `${route.provider}/${route.model}` : '(平台默认)';
if (route) {
const model = modelRuntime.getModel(route.provider, route.model);
if (!model) {
// 目录里根本没有这个路由:同步就能判定,不必起一轮
failures.push({ ...route, error: `平台目录里没有 ${label}` });
log(`模型 ${label} 不存在,跳过`);
continue;
}
// 换模型要换会话pi 的模型在 createAgentSession 时绑定。
// 上一次尝试失败的会话没有任何 assistant 消息,丢掉不损失内容。
const cwd = sessions.get(data.session_id)?.cwd;
const current = sessions.get(data.session_id)?.session;
current?.dispose?.();
const retried = await openSession({
cwd,
modelRuntime,
model,
customTools: mailTools,
extension: permissionExtension((id) => mailContexts.get(id)),
});
rebind(data.session_id, current?.sessionId ?? piSessionId, retried, cwd);
const outcome = await runTurn(retried.session, prompt, TURN_TIMEOUT_MS);
if (outcome.ok) {
if (failures.length) log(`${label} 成功(前 ${failures.length} 个失败)`);
return;
}
failures.push({ ...route, error: outcome.error });
log(`模型 ${label} 失败: ${outcome.error}`);
continue;
}
const outcome = await runTurn(session, prompt, TURN_TIMEOUT_MS);
if (outcome.ok) {
if (failures.length) log(`${label} 成功(前 ${failures.length} 个失败)`);
return;
}
failures.push({ error: outcome.error });
log(`模型 ${label} 失败: ${outcome.error}`);
}
// 全部失败 → 必须回信B-6模型一次都没跑起来会话里没有任何
// assistant 消息,自动转发因此什么也不会发 —— 发件人只会看到再无音讯。
if (kind === 'mail' && data.from_name) {
try {
await client.post('/mail/send', {
to: data.from_name,
subject: `处理失败: ${data.subject || '(无主题)'}`,
body: renderFailureReport(failures, data.subject),
reply_to: data.mail_id || '',
relay: 'summary',
relay_key: `model-failure:${data.mail_id || piSessionId}`,
});
log(`已回报模型调用失败给 ${data.from_name}`);
} catch (e) {
log(`失败回报也发不出去: ${describeError(e)}`);
}
}
// 发完仍要 throwB-6.4):静默会让这次失败只存在于邮件里,日志上看不出来
throw new Error(`范围内 ${failures.length} 个模型全部失败:${failures.map(f => f.error).join(' | ')}`);
}
/** 换模型重开会话后,把三张映射表指向新会话。 */
function rebind(mailSessionID, oldPiId, opened, cwd) {
reverseMap.delete(oldPiId);
mailDriven.delete(oldPiId);
// 免批授权跟着旧的 pi 会话作废:它是人对**那次**上下文的判断,
// 换模型意味着重开一条会话、重跑一遍提示,不该继承上一条的授权。
permissionGrants.revokeSession(oldPiId);
const piSessionId = opened.session.sessionId;
// cwd 由调用方传AgentSession 上没有 cwd getter只有 sessionId /
// sessionFile / sessionName从 sessionManager.getCwd() 也行,
// 但这里本来就有那个值,多绕一层没有意义。
const entry = { session: opened.session, sessionManager: opened.sessionManager, cwd };
if (mailSessionID) {
sessions.set(mailSessionID, entry);
reverseMap.set(piSessionId, mailSessionID);
mailDriven.add(piSessionId);
}
opened.session.subscribe((event) => {
if (event?.type === 'agent_end' && !event.willRetry) {
relaySummary(piSessionId).catch((e) => log(`自动转发失败: ${describeError(e)}`));
}
if (event?.type === 'session_info_changed') {
syncNaming(piSessionId, event.name).catch((e) => log(`命名同步失败: ${describeError(e)}`));
}
});
}
// ─── 权限决策回来B-4───
async function handlePermissionDecision(data, mailTools) {
/**
* 把决策路由给发起询问的那个 worker。
*
* 找不到 worker 有两种情形都不该新开会话B-4.3
* - 桥重启了:那次工具调用早已随进程消失。但人刚刚点了「同意」——
* 什么都不做的话人以为自己批准了、Agent 却毫无反应,所以退化为把决策
* 当一封通知投进原会话B-4.2)。
* - 那条邮件会话从没被处理过:连通知都无处可投,只能记一行日志。
*/
function handlePermissionDecision(data) {
const relayKey = data.relay_key || '';
const pending = relayKey ? pendingPermissions.get(relayKey) : undefined;
if (pending) {
pendingPermissions.delete(relayKey);
pending.resolve(String(data.decision || '拒绝'));
log(`权限 ${relayKey} 决策 ${data.decision}(决策人 ${data.decided_by || '?'}`);
if (relayKey && pool.routePermission(relayKey, String(data.decision || '拒绝'))) {
log(`权限 ${relayKey} 决策 ${data.decision}(决策人 ${data.decided_by || '?'})已转交 worker`);
return;
}
// 找不到挂起项(桥重启丢了内存映射)→ 退化为把决策当一封通知投进原会话B-4.2)。
// 此时 pi 侧那次工具调用早已随进程消失,但人刚刚点了「同意」——
// 什么都不做的话人以为自己批准了、Agent 却毫无反应。
if (!data.session_id || !sessions.has(data.session_id)) {
if (!data.session_id || !pool.hasSession(data.session_id)) {
// **不得凭空新开会话**B-4.3
log(`权限决策 ${relayKey} 无对应会话,忽略`);
return;
}
log(`权限 ${relayKey} 无挂起项,退化为通知投递`);
await deliverMail(data, 'permission', mailTools);
pool.submit('permission', data);
}
// ─── 心跳B-2───
@ -769,7 +181,8 @@ async function reportSessions() {
// 用 listAll 而不是 list(cwd):桥的进程 cwd 与会话 cwd 无关,
// 按前者过滤会漏掉所有真正在干活的会话。
const all = await SessionManager.listAll();
return snapshotPiSessions(all, (id) => mailDriven.has(id));
const driven = pool.mailDrivenIDs();
return snapshotPiSessions(all, (id) => driven.has(id));
} catch (e) {
// 拉不到就**省略字段**而不是传 []N-7 / W-3
// 空数组的语义是「平台确实一条会话都没有」,会把服务端镜像抹掉。
@ -790,22 +203,26 @@ async function reportModels() {
}
}
async function catchUp(pending, mailTools) {
/**
* 补投离线期间积压的未读邮件B-7
*
* SSE 只推连上之后的事件,插件重启前发来的邮件不会再推一次。
*
* 与旧版的差别:**不再 await 每一封**。旧版串行是因为「每封都要起一轮模型,
* 并发放出去等于对上游打 N 个并发请求」—— 那个约束现在由 pool 的 maxWorkers
* 承担,而且它比串行更好:同一条会话仍然串行,不同会话可以并行。
*/
async function catchUp(pending) {
if (!pending) return;
try {
const box = await client.get('/mail/inbox?status=unread&limit=20');
const tasks = selectCatchup(box?.mails ?? box, deliveredMails);
if (!tasks.length) return;
log(`补投 ${tasks.length} 封离线期间的邮件(共 ${pending} 封未读)`);
// 串行B-7.2):每封都要起一轮模型,并发放出去等于对上游打 N 个并发请求
for (const ev of tasks) {
if (deliveredMails.has(ev.mail_id)) continue; // 逐封再查B-7.6
deliveredMails.add(ev.mail_id);
try {
await deliverMail(ev, 'mail', mailTools);
} catch (e) {
log(`补投 ${ev.mail_id} 失败: ${describeError(e)}`);
}
pool.submit('mail', ev);
}
} catch (e) {
log(`补投失败: ${describeError(e)}`);
@ -828,8 +245,8 @@ async function main() {
agentSecret: AGENT_SECRET,
});
// ModelRuntime 建一次全进程共用:它要读 auth.json / models.json 并做
// 可用性探测,每条会话建一个既慢又会重复打 provider 的探测请求
// ModelRuntime 主进程也要一个:心跳的 reportModels 用它。worker 各自再建
// 一个(跨进程传不了),代价是每个 worker 多 ~20ms实测 1123ms
//
// allowModelNetwork 保持默认的 false桥启动时不去网上拉模型目录。
// 拉了也没用 —— 上报给 Gateway 的是 getAvailable()(有凭证、真能调起来的),
@ -839,16 +256,43 @@ async function main() {
const runtimeErr = modelRuntime.getError?.();
if (runtimeErr) log(`模型运行时告警: ${runtimeErr}`);
// onReconnectconnect_to_server 换了 Gateway 地址/密钥后,旧 SSE 长连仍连着
// 旧地址(或已被旧密钥打断),必须在这里重建,模型调完工具才真正「切过去」
// reconfigure 已清空 lastEventID所以 startSSE 会以「首次连接」姿态
// (不带 Last-Event-IDN-11连上新地址 —— 拿旧序号问新 Gateway 只会
// 命中一段无关的历史。
const mailTools = createMailTools({
// 工作进程池。config() 每次派活时取一次 —— allowedModels 随心跳变,
// 取快照会让 worker 用上一轮的模型范围
pool = createWorkerPool({
log,
config: () => ({
gatewayURL: client.baseURL,
agentName: AGENT_NAME,
agentKey: client.agentKey,
agentSecret: AGENT_SECRET,
allowedModels,
replyProvider: REPLY_PROVIDER,
replyModel: REPLY_MODEL,
turnTimeoutMs: TURN_TIMEOUT_MS,
}),
// worker 里 connect_to_server 换了坐标worker 马上就退了,改在它自己身上
// 等于没改。主进程据此重建 SSE后续 worker 的 job 也会带上新坐标。
onReconfigure: (url, key) => {
if (!client.reconfigure({ url, agentKey: key })) return;
log(`Gateway 坐标已更新为 ${client.baseURL},重建 SSE`);
client.stopSSE();
client.startSSE(handleSSEEvent, log);
},
maxWorkers: MAX_WORKERS,
workerMaxMs: WORKER_MAX_MS,
});
// 主进程仍要一套邮件工具:它自己不跑模型,但 connect_to_server 的
// onReconnect 语义要在这里闭环worker 侧那套只负责回报给主进程)。
//
// 这些工具不会被任何模型调用 —— 主进程没有会话。留着是因为
// createMailTools 同时承担「校验工具 schema」的职责test/tool-schema.test.mjs
// 而工具总数是契约里核对过的数字。
createMailTools({
client, log, agentName: AGENT_NAME,
onReconnect: () => {
client.stopSSE();
client.startSSE((type, data) => handleSSEEvent(type, data, mailTools), log);
client.startSSE(handleSSEEvent, log);
},
});
@ -874,7 +318,7 @@ async function main() {
if (Array.isArray(res?.allowed_models)) allowedModels = res.allowed_models; // B-2.2
if (!caughtUp) { // B-7.1:只在首个成功心跳后补一次
caughtUp = true;
await catchUp(res?.pending_mails, mailTools);
await catchUp(res?.pending_mails);
}
} catch {
// B-2.1:心跳失败不重试不报错。真连不上时 Gateway 会把它判成离线,
@ -884,7 +328,7 @@ async function main() {
await beat(); // B-1.3:不等第一个 30 秒周期
heartbeatTimer = setInterval(beat, 30_000); // B-1.5
client.startSSE((type, data) => handleSSEEvent(type, data, mailTools), log);
client.startSSE(handleSSEEvent, log);
for (const sig of ['SIGINT', 'SIGTERM']) process.on(sig, () => shutdown(sig));
}
@ -892,13 +336,15 @@ async function main() {
/**
* SSE 事件分派。
*
* 这个函数**必须保持廉价**:它跑在读循环上。派活给 pool 是同步的fork 是
* 异步的pool.submit 只是入队),所以读循环不会因为一封邮件停下。
*
* 提成命名函数是因为 connect_to_server 换地址后要用同一个处理器重建长连 ——
* 内联箭头函数在那里拿不到,只能复制一遍,而复制出来的两份迟早会分叉。
*/
function handleSSEEvent(type, data, mailTools) {
function handleSSEEvent(type, data) {
if (type === 'permission_decision') {
handlePermissionDecision(data, mailTools).catch((e) =>
log(`权限决策处理失败: ${describeError(e)}`));
handlePermissionDecision(data);
return;
}
if (type !== 'new_mail') return;
@ -906,7 +352,7 @@ function handleSSEEvent(type, data, mailTools) {
const id = data?.mail_id;
if (!id || deliveredMails.has(id)) return; // B-3 第 1 步:去重
deliveredMails.add(id);
deliverMail(data, 'mail', mailTools).catch((e) => log(`投递 ${id} 失败: ${describeError(e)}`));
pool.submit('mail', data);
}
function shutdown(reason) {
@ -917,21 +363,17 @@ function shutdown(reason) {
if (heartbeatTimer) clearInterval(heartbeatTimer); // B-9.1
client?.stopSSE();
// B-9.2 / N-9所有未决权限询问 fail closed
// 不唤醒的话 pi 侧那些 await 永不返回,整条会话挂死;
// 而默认放行一个没人批准的危险操作,比让它失败严重得多。
for (const [key, p] of pendingPermissions) {
log(`未决权限 ${key} fail closed`);
p.resolve('shutdown');
}
pendingPermissions.clear();
// pool.stop 先给每个 worker 发 shutdown让它把未决权限询问 fail closed
// B-9.2 / N-9再给 2 秒自己退,然后 SIGKILL。
//
// 不直接杀pi 侧那些 await 不会返回,而 worker 里可能正握着会话文件。
pool?.stop();
for (const { session } of sessions.values()) {
try { session.dispose?.(); } catch { /* 关停期的报错没有价值 */ }
}
releaseLock();
// 留出 pool.stop 的宽限窗口再退:主进程先死会让子进程变成孤儿
// systemd 的 KillMode 会兜住,但那时 fail closed 已经来不及做了)。
setTimeout(() => process.exit(0), 2500).unref?.();
// B-9.3:不发「插件下线」通知邮件
process.exit(0);
}
main().catch((e) => {

View File

@ -0,0 +1,277 @@
/**
* 投递工作进程池 —— 主进程侧的调度逻辑。
*
* # 它解决的问题
*
* 桥的主进程唯一的实时职责是读 SSE。模型工作放在主进程里跑会占满事件循环
* pi 的会话装载是同步的23MB 的会话文件 `SessionManager.open` 一次阻塞
* 118ms实测模型跑起来之后 SDK 内部还有大量同步工作SSE 读循环停住,
* 后续邮件卡在 TCP 缓冲区,久到 Gateway 认为连接死了 → 重连 → 重放。
*
* 所以:**收到事件就派给一个子进程,主进程立刻回去读 SSE。**
*
* # 并发与串行的边界
*
* - **不同邮件会话并发**,上限 `maxWorkers`(默认 3。上限的理由是内存
* (每个 worker 约 140MB RSS实测和对上游 provider 的并发请求数。
* - **同一邮件会话串行**。这是正确性要求不是限流pi 没有任何锁机制,
* 它假定「一个文件一个持有者」。两个 worker 同时装载同一条会话文件,各自的
* 内存索引都看不见对方追加的行,算出的 parentId 指向对方不知道的 entry
* → 会话树分叉。串行还顺带保证了同一条线索里两封邮件的先后顺序。
*
* # 排队而不是拒绝
*
* 满载时邮件进 `queue`,有 worker 空出来就派。丢掉邮件是不可接受的:
* 发件人只会看到信发出去后再无音讯。队列无上限 —— 有上限就得决定丢哪封,
* 而任何丢弃策略都比「慢一点」糟。
*
* # 主进程持有什么
*
* 只有**路径与标量**sessionFile / cwd / piSessionId / 「一直同意」表 /
* 命名同步指纹。AgentSession 对象跨不了进程边界worker 每次从 sessionFile
* 重新装载 —— 拿到的是包含 TUI 期间写入的全部历史(这也让「短暂持有」
* 从一套需要计时器兜底的机制退化成「worker 退出就是释放」)。
*
* # IPC 协议
*
* 主进程 → worker
* `{type:'job', kind, data, session:{sessionFile,cwd}, grants, lastSyncedName, config}`
* `{type:'permission_decision', relayKey, decision}`
* `{type:'shutdown'}`
* worker → 主进程:
* `{type:'ready'}` 进程起来了,可以派活
* `{type:'log', line}` 日志(主进程加 pid 前缀)
* `{type:'session_opened', piSessionId, sessionFile, cwd, reused}`
* `{type:'permission_pending', relayKey}` 主进程记下路由表
* `{type:'permission_grant', toolName}` 「一直同意」要跨 worker 活下来
* `{type:'name_synced', signature}` 命名指纹,防下一个 worker 重复 sync
* `{type:'reconfigure', url, agentKey}` connect_to_server 换了坐标
* `{type:'done', ok, error}` 这封处理完了
*/
import { fork } from 'node:child_process';
import { fileURLToPath } from 'node:url';
const WORKER_PATH = fileURLToPath(new URL('./worker.mjs', import.meta.url));
/**
* @param {object} deps
* @param {(...a: any[]) => void} deps.log
* @param {() => object} deps.config 每次派活时取一次allowedModels 会随心跳变)
* @param {(url: string, key: string) => void} deps.onReconfigure
* @param {number} [deps.maxWorkers]
* @param {number} [deps.workerMaxMs] worker 硬超时:卡死的进程必须能被回收
* @param {string} [deps.workerPath] 只为测试存在:换成不装 pi SDK 的桩 worker
* 让调度不变量(并发上限、同会话串行、硬超时)能在毫秒级验证。
*/
export function createWorkerPool({
log, config, onReconfigure,
maxWorkers = 3, workerMaxMs = 600_000, workerPath = WORKER_PATH,
}) {
/** 正在跑的 workermailSessionKey -> {child, mailID, startedAt, timer} */
const running = new Map();
/** 等着派的活,先进先出。 */
const queue = [];
/** relay_key -> mailSessionKey把决策路由回发起询问的那个 worker。 */
const permissionRoutes = new Map();
/**
* 跨 worker 存活的会话状态mailSessionKey -> {sessionFile, cwd, piSessionId,
* grants:Set, lastSyncedName}。
*
* 这是 worker 一封一进程之后仍需在主进程留存的全部东西 —— 下一封邮件靠
* sessionFile 接着谈,靠 grants 不重复问已经「一直同意」过的工具。
*/
const sessionState = new Map();
/**
* 被模型降级换掉的旧 pi 会话 id。
*
* 仍要计入 mail_driven它们已经参与过邮件往来而磁盘上的会话文件
* 不会因为换模型而消失 —— 心跳快照仍会上报它们。
*/
const retired = new Set();
let stopped = false;
/**
* 邮件会话 id 作为串行化的键。
*
* 没有 session_id 的事件(理论上不该有)退回 mail_id那样每封各占一个
* worker不会串行 —— 但它们本来也不属于同一条会话。
*/
const keyOf = (data) => data?.session_id || `mail:${data?.mail_id || Math.random()}`;
function submit(kind, data) {
if (stopped) return;
queue.push({ kind, data, key: keyOf(data) });
pump();
}
function pump() {
if (stopped) return;
for (let i = 0; i < queue.length; i++) {
const job = queue[i];
// 同一会话已有 worker 在跑 → 跳过它,看后面有没有别的会话可以先跑。
// 不能 break那会让一条慢会话把所有别的会话都堵住正是要修的病
if (running.has(job.key)) continue;
if (running.size >= maxWorkers) return;
queue.splice(i, 1);
i--;
spawn(job);
}
}
function spawn(job) {
const state = sessionState.get(job.key) || { grants: new Set(), lastSyncedName: '' };
const child = fork(workerPath, [], {
// stdio 继承worker 里 pi SDK 自己打的东西直接进 journalctl。
// 'ipc' 必须显式列出,否则 process.send 不存在。
stdio: ['ignore', 'inherit', 'inherit', 'ipc'],
});
// 硬超时worker 卡死(模型不返回、权限等不到决策而主进程也没收到事件)
// 时必须能回收,否则那条会话的后续邮件永远排队。
const timer = setTimeout(() => {
log(`worker ${child.pid} 处理 ${job.data?.mail_id} 超过 ${workerMaxMs / 1000}s强杀`);
try { child.kill('SIGKILL'); } catch { /* 已经死了 */ }
}, workerMaxMs);
if (typeof timer.unref === 'function') timer.unref();
const entry = { child, mailID: job.data?.mail_id || '', key: job.key, startedAt: Date.now(), timer };
running.set(job.key, entry);
child.on('message', (msg) => onWorkerMessage(entry, msg));
child.on('exit', (code, signal) => {
clearTimeout(timer);
running.delete(job.key);
for (const [rk, k] of permissionRoutes) if (k === job.key) permissionRoutes.delete(rk);
if (code !== 0) {
log(`worker ${child.pid}mail ${entry.mailID})异常退出 code=${code} signal=${signal || '-'}`);
}
pump();
});
child.on('error', (e) => log(`worker ${child.pid} 出错: ${e?.message || e}`));
// 等 worker 说 ready 再派活fork 返回时子进程的 import 还没跑完,
// 此时 send 的消息会排在 IPC 队列里(能收到,但 ready 让顺序确定)。
child.once('message', function first(msg) {
if (msg?.type !== 'ready') return;
child.send({
type: 'job',
kind: job.kind,
data: job.data,
session: {
sessionFile: state.sessionFile || '',
cwd: state.cwd || '',
},
grants: [...state.grants],
lastSyncedName: state.lastSyncedName || '',
config: config(),
});
});
}
function onWorkerMessage(entry, msg) {
const state = sessionState.get(entry.key) || { grants: new Set(), lastSyncedName: '' };
switch (msg?.type) {
case 'log':
log(`[w${entry.child.pid}] ${msg.line}`);
return;
case 'session_opened':
// 一条会话可能先后用过多个 pi 会话 id模型降级会换会话
// 旧 id 仍计入 mail_driven理由见 retired 的注释。
if (state.piSessionId && state.piSessionId !== msg.piSessionId) {
retired.add(state.piSessionId);
}
state.piSessionId = msg.piSessionId;
state.sessionFile = msg.sessionFile;
state.cwd = msg.cwd;
sessionState.set(entry.key, state);
return;
case 'permission_pending':
permissionRoutes.set(msg.relayKey, entry.key);
return;
case 'permission_grant':
// 「一直同意」必须跨 worker 活着worker 一封一进程,不存的话下一封
// 邮件又问一遍,那个选项就是在骗人。
state.grants.add(msg.toolName);
sessionState.set(entry.key, state);
return;
case 'name_synced':
state.lastSyncedName = msg.signature;
sessionState.set(entry.key, state);
return;
case 'reconfigure':
onReconfigure?.(msg.url, msg.agentKey);
return;
case 'done':
if (!msg.ok) log(`投递 ${entry.mailID} 失败: ${msg.error}`);
return;
default:
return;
}
}
/**
* 把权限决策路由到发起询问的那个 worker。
*
* @returns {boolean} 有没有找到对应的 worker。找不到说明那个 worker 已经退了
* (桥重启、硬超时被杀、或者处理已经结束)—— 调用方据此走 B-4.2 的
* 降级路径(把决策当一封通知投进原会话)。
*/
function routePermission(relayKey, decision) {
const key = permissionRoutes.get(relayKey);
if (!key) return false;
const entry = running.get(key);
if (!entry) {
permissionRoutes.delete(relayKey);
return false;
}
permissionRoutes.delete(relayKey);
entry.child.send({ type: 'permission_decision', relayKey, decision });
return true;
}
/** 这条邮件会话有 worker 在跑吗B-4.2 判断降级路径用)。 */
const hasSession = (mailSessionID) => sessionState.has(mailSessionID);
/**
* 邮件驱动过的 pi 会话 id喂给心跳快照的 `mail_driven` 标记。
*
* 不随 worker 退出而清worker 退了不代表那条会话不再参与邮件往来 ——
* 下一封邮件还会接着谈,而人在补全里需要看到它带着这个标记。
* 重启丢是已知取舍(契约第六节)。
*/
const mailDrivenIDs = () => {
const out = new Set(retired);
for (const st of sessionState.values()) {
if (st.piSessionId) out.add(st.piSessionId);
}
return out;
};
function stop() {
stopped = true;
queue.length = 0;
for (const { child, timer } of running.values()) {
clearTimeout(timer);
// 先 shutdown 让 worker 把未决权限 fail closedB-9.2),再给它一点
// 时间自己退。不直接 SIGKILL那样 pi 侧的 await 不会返回,而 worker
// 里可能正握着会话文件。
try { child.send({ type: 'shutdown' }); } catch { /* 通道已断 */ }
setTimeout(() => { try { child.kill('SIGKILL'); } catch { /* 已经死了 */ } }, 2000).unref?.();
}
}
/** 观测用:现在跑着几个、排了几个。 */
const stats = () => ({
running: running.size,
queued: queue.length,
sessions: sessionState.size,
workers: [...running.values()].map((e) => ({
pid: e.child.pid, mailID: e.mailID, ageMs: Date.now() - e.startedAt,
})),
});
return { submit, routePermission, hasSession, mailDrivenIDs, stop, stats };
}

View File

@ -0,0 +1,555 @@
#!/usr/bin/env node
/**
* pi-mail-bridge 的**投递工作进程** —— 一封邮件一个,跑完就退。
*
* # 为什么必须是独立进程
*
* 桥的主进程要一直读 SSE。而 pi 的会话装载是**同步**的:
* `SessionManager.open()` 走 `openSync` + `readSync` 循环把整个 `.jsonl`
* 读进内存并逐行 JSON.parseSDK core/session-manager.js 的
* loadEntriesFromFile。实测本机最大那条会话 23MB`open` 一次
* **阻塞事件循环 118ms**;模型跑起来之后 SDK 内部还有大量同步工作。
* 全都发生在主线程上SSE 读循环在那期间完全停住 —— 后续邮件卡在 TCP
* 缓冲区里,久到 Gateway 认为连接死了,重连又触发重放。
*
* 把模型工作搬进子进程后,主进程只剩「收事件 → 去重 → 分派」,
* 实测同样的活在 fork 出的子进程里跑,主进程事件循环阻塞 0ms。
*
* # 为什么是 child_process 而不是 worker_threads
*
* 两者实测都能建起 AgentSession。选进程的理由是**隔离**
* 模型会跑 bash/write/edit一次 OOM 或 uncaughtException 不该带走
* 整座桥;而 `worker_threads` 与主线程共享堆和进程生命周期。
* 代价是每个 worker 约 140MB RSS 和 ~500ms 启动,由主进程的并发上限约束。
*
* # 一封邮件一个进程带来的简化
*
* 原来的「接管会话短暂持有」adopted / adoptTimers / releaseAdopted
* 整套机制没有了worker 退出**就是**释放,而且对普通会话和接管会话
* 一视同仁 —— 每封邮件都是「open → 跑一轮 → 还回去」。
* pi 假定「一个文件一个持有者」,主进程按会话串行分派保证了这一点。
*
* # 进程边界上传什么
*
* 主进程持有的是**路径与标量**sessionFile / cwd / piSessionId / 授权表),
* 不是 AgentSession 对象 —— 那东西跨不了进程。worker 每次从 sessionFile
* 重新装载,拿到的是包含 TUI 期间写入的全部历史。
*
* 协议见 src/pool.mjs 顶部。
*/
import { existsSync } from 'node:fs';
import { homedir } from 'node:os';
import { join } from 'node:path';
import { ModelRuntime } from '@earendil-works/pi-coding-agent';
import { GatewayClient } from './gateway.mjs';
import { createMailTools } from './tools.mjs';
import { openSession, runTurn } from './session-pool.mjs';
import { buildMailPrompt, lastAssistantText, replySubject, relayKeyFor, describeError } from './turn.mjs';
import { planNamingSync, planWriteBack } from './naming.mjs';
import { resolveWorkspaceCwd, ensureCwd } from '../lib/workspace.js';
import { modelAttemptOrder, renderFailureReport } from '../lib/model-scope.js';
import { explicitSends, shouldSkipAutoRelay } from '../lib/relay-dedup.js';
import { adoptedSessionID, adoptMissingMessage } from '../lib/adopt.js';
import { isApproval, isAlwaysDecision } from '../lib/permission-grants.js';
// ─── 与主进程的通道 ───
/** 日志一律回传主进程worker 的 stdout 会混在一起,加 pid 前缀才分得清。 */
const log = (...args) => send({ type: 'log', line: args.join(' ') });
function send(msg) {
try {
process.send?.(msg);
} catch {
/* 主进程已经走了,这条日志没有去处 */
}
}
/** relay_key -> resolve决策由主进程的 SSE 收到后路由进来。 */
const pending = new Map();
/**
* 这一轮里人点过「一直同意」的工具。
*
* 初值由主进程在 job 里给(跨 worker 持久),新增的回报给主进程。
* 不用 lib/permission-grants.js 的 store那份按 (会话, 工具) 存,
* 而 worker 只服务一条会话,一个 Set 就够,且要能整体回传。
*/
const grants = new Set();
// ─── 单封邮件的全部状态 ───
//
// worker 只处理一封邮件、只碰一条会话,所以这些原来在主进程里
// 按 sessionId 分桶的 Map 在这里都退化成单个变量。
let job = null;
let client = null;
let modelRuntime = null;
let piSessionId = '';
let mailContext = { replyTo: '', subject: '', mailID: '' };
let lastSyncedName = '';
let relayedKey = '';
let finished = false;
// ─── 权限钩子B-8───
/**
* 与主进程版本逐条对应,差别只有两处:
* - 「是不是邮件驱动的会话」不必查表worker 只为邮件而存在。
* - 等决策的 promise 由主进程通过 IPC 唤醒,而不是本进程的 SSE。
*
* 决策等待期间**只有这个 worker 停住**,主进程照常读 SSE、照常给别的会话
* 派活 —— 这正是原来最难受的一处:权限询问会让整座桥不再收信。
*/
function permissionExtension() {
const GUARDED = new Set(['bash', 'write', 'edit']);
return (pi) => {
pi.on('tool_call', async (event, ctx) => {
if (!GUARDED.has(event.toolName)) return;
const sid = ctx?.sessionManager?.getSessionId?.() || '';
// 只管自己那条会话。worker 里不该出现第二条,出现了说明有 bug ——
// 让位(返回 undefined比拦错一个安全。
if (piSessionId && sid && sid !== piSessionId) return;
// 人点过「一直同意」→ 直接放行。必须在 POST 之前:否则每条命令都发一封
// 邮件,那个选项形同虚设(生产实测同一条会话被问了 15 次 bash
if (grants.has(event.toolName)) return;
// relay_key 用 pi 给的 toolCallIdB-8.1):服务端随决策事件回传它。
const relayKey = `${sid || piSessionId}:${event.toolCallId}`;
try {
// 不传 `to`:决策人由服务端按 会话 owner → 线索里最近的人类 → 409
// 解析。插件只有本地上下文,猜不出「这条 Agent 链最初是谁派的活」。
await client.post('/permission/request', {
question: `是否允许执行 ${event.toolName}`,
options: ['同意', '一直同意', '拒绝'],
context: [
describeToolCall(event),
mailContext.subject ? `\n触发任务:${mailContext.subject}` : '',
mailContext.replyTo ? `任务来自:${mailContext.replyTo}` : '',
].filter(Boolean).join('\n'),
session_id: job.data?.session_id || '',
relay_key: relayKey,
});
} catch (e) {
// 409 = 服务端判定这条任务链上没有人类,永远不会有人来点头。
// 当场 block 并把服务端建议原文当 reason模型从工具报错里看到
// 「没人可问,换不需要权限的方式」才能自己改道,挂死时连重试机会都没有。
if (e?.status === 409) {
const b = e.body || {};
const reason = [
b.error || '权限询问无法送达:这条任务链上没有人类用户',
b.detail || '',
b.suggestion || '',
].filter(Boolean).join('\n');
log(`权限询问无人可投,当场拒绝 ${relayKey}${b.error || ''}`);
return { block: true, reason };
}
// 其余失败是暂时的 → 让位给 pi 本地决策B-8.2)。
log(`权限转发失败,让位给本地决策: ${describeError(e)}`);
return;
}
log(`权限询问已发出(${event.toolName}key=${relayKey}),等待决策…`);
send({ type: 'permission_pending', relayKey });
const decision = await new Promise((resolve) => pending.set(relayKey, resolve));
// fail closedB-9.2 / N-9只有明确同意才放行。
if (isApproval(decision)) {
// 「一直同意」要真的记住,否则这个选项在骗人。判定交给 isAlwaysDecision ——
// 「同意」是单次授权,把它当 always 会放行人没看过的后续命令。
if (isAlwaysDecision(decision)) {
grants.add(event.toolName);
send({ type: 'permission_grant', toolName: event.toolName });
log(`本会话的 ${event.toolName} 已获「一直同意」,后续不再询问`);
}
log(`权限 ${relayKey} 获批(${decision}),放行 ${event.toolName}`);
return;
}
return {
block: true,
reason: `用户${decision === 'shutdown' ? '未及决策(桥已关停)' : `拒绝了这次 ${event.toolName} 调用`}`,
};
});
};
}
/** 把一次工具调用摘要成人能判断的文本B-8.4)。 */
function describeToolCall(event) {
const input = event?.input ?? {};
if (event.toolName === 'bash') return `命令:\n${String(input.command ?? '').slice(0, 800)}`;
if (event.toolName === 'write' || event.toolName === 'edit') {
return `文件:${input.file_path ?? input.path ?? '(未给出)'}`;
}
return JSON.stringify(input).slice(0, 800);
}
// ─── 会话装载 ───
/**
* 没有可用 `to_workspace` 时的兜底目录。
*
* 与 DSH 的 `mailSessionFallback` 同构,但目录名是 `.pi`:那个函数在 lib/ 下
* (三平台逐字节相同),写死了 `.dsh`,不能为 pi 改 —— pi 的会话落进 `~/.dsh/`
* 会让人以为是 DSH 在干活。
*/
function piMailFallback(sessionKey) {
return join(homedir(), '.pi', 'mail-sessions', String(sessionKey || 'default'));
}
/**
* 找到这封邮件该落进的会话文件,装载它。
*
* 三条路,优先级从高到低:
* 1. 主进程给了 sessionFile —— 这条邮件会话之前有 worker 跑过,接着谈。
* 2. 服务端说接管了平台会话(人在 TUI 里开的那条)→ 按 platform id 找文件。
* 3. 都没有 → 新建cwd 取地址的 path 位B-3.1)。
*
* 返回 `reused` 供提示词与模型降级判断:有历史的会话不重新自我介绍,
* 也不做模型降级(换模型要换会话,会丢掉整条上下文,而上下文正是
* 发件人指定这条会话的原因)。
*/
async function loadSession(mailTools) {
const data = job.data;
const given = job.session?.sessionFile;
if (given && existsSync(given)) {
const cwd = job.session.cwd || resolveWorkspaceCwd(
data.to_workspace, piMailFallback(data.session_id)).cwd;
const opened = await openSession({
cwd, modelRuntime, customTools: mailTools,
extension: permissionExtension(), sessionFile: given,
});
return { ...opened, cwd, reused: true };
}
const adoptID = adoptedSessionID(data);
if (adoptID) {
const { SessionManager } = await import('@earendil-works/pi-coding-agent');
// listAll 而不是 list(cwd)worker 的进程 cwd 与会话 cwd 无关。
const all = await SessionManager.listAll();
const info = all.find((e) => e?.id === adoptID);
if (!info?.path) {
// 镜像是快照,可以过期。**不能**退回「新建一条」—— 那会让人在 TUI 里
// 看不到这封邮件带来的对话而那正是接管的目的N-8静默改语义比报错糟
throw new Error(adoptMissingMessage(adoptID, '磁盘上已无这个会话文件'));
}
// cwd 取会话自己的SessionInfo.cwd 来自持久化 header
// 老会话的 cwd 是空串,那种情况退回地址里的 path 位。
const { cwd } = resolveWorkspaceCwd(
info.cwd || data.to_workspace, piMailFallback(data.session_id));
const opened = await openSession({
cwd, modelRuntime, customTools: mailTools,
extension: permissionExtension(), sessionFile: info.path,
});
log(`接管 pi 会话 ${adoptID}cwd=${cwd},文件 ${info.path}`);
return { ...opened, cwd, reused: true };
}
// 目录不存在时**不创建**N-2笔误会在磁盘上落下真目录而 Agent 在
// 里面一无所获拒绝相对路径N-3
const { cwd, grouped } = resolveWorkspaceCwd(
data.to_workspace, piMailFallback(data.session_id));
if (!grouped && data.to_workspace) {
log(`工作目录 ${data.to_workspace} 不可用,回退到 ${cwd}`);
}
ensureCwd(cwd, grouped);
const opened = await openSession({
cwd, modelRuntime, customTools: mailTools, extension: permissionExtension(),
});
log(`新建 pi 会话 ${opened.session.sessionId}cwd=${cwd}`);
return { ...opened, cwd, reused: false };
}
// ─── 命名一致C-11 / W-7───
async function syncNaming(session, platformName) {
const mailSessionID = job.data?.session_id;
if (!mailSessionID) return;
const plan = planNamingSync({
platformName,
mailSubject: mailContext.subject,
lastSynced: lastSyncedName,
});
if (plan.skip) return;
// 先记指纹再发请求:响应回来时 setSessionName 会再次触发
// session_info_changed这一步是防自激循环的关键。
lastSyncedName = plan.signature;
send({ type: 'name_synced', signature: plan.signature });
const res = await client.post(`/sessions/${mailSessionID}/sync`, {
alias: plan.alias, title: plan.title,
});
const back = planWriteBack({ finalAlias: res?.alias, currentPiName: session.sessionName });
log(`命名同步 ${piSessionId}: alias=${res?.alias || '(未变)'} 来源=${plan.source}`);
if (back.write) {
// 顺序要紧先更新指纹再改名。setSessionName **同步**触发
// session_info_changed实测本函数会被重入指纹后更新的话重入那次
// 看到旧指纹,于是又打一次 sync —— 每条会话两次内容相同的请求。
lastSyncedName = `platform:${back.name}|${back.name}`;
send({ type: 'name_synced', signature: lastSyncedName });
// 只用 setSessionName走 pi 自己的写入路径)。绝不自己拼路径写会话文件:
// 首条 assistant 消息落盘前文件还不存在pi 首次落盘用 openSync(file,"wx")
// 抢先创建会让它抛 EEXIST实测
session.setSessionName(back.name);
log(`别名回写 pi${back.name}${back.reason}`);
}
}
// ─── 自动转发B-5───
/**
* 把这一轮的结论转回发件人。
*
* 与主进程版本的差别:不再挂在 `agent_end` 订阅上,而是在 `runTurn` 返回后
* **确定性地**调一次。原来必须靠事件是因为主进程的 60 秒超时会先返回、
* 会话还在跑worker 没有那个约束(它就为这封邮件活着),等真结束再转发。
*
* @param {boolean} adopted 接管会话跳过命名同步 —— 它的别名是人从补全里
* 选中的平台 slug同步会双向改坏撞名时 Gateway 加后缀,定稿别名又回写进
* pi 会话文件,于是下次心跳上报的 slug 变成带后缀那个。实测撞出来过一次。
*/
async function relaySummary(session, sessionManager, adopted) {
if (!adopted) {
await syncNaming(session, session.sessionName)
.catch((e) => log(`命名同步失败: ${describeError(e)}`));
}
// 只取 type==='text' 的块B-5.1 / N-6thinking 是思考过程,不是结论。
const text = lastAssistantText(session.messages);
if (!text) return; // 空文本不发空邮件B-5.4
if (!mailContext.replyTo) return; // 不知道回给谁
// 幂等键用 pi 会话 id + 会话树叶子 id两者都落盘重放也是同一个键。
const relayKey = relayKeyFor(piSessionId, sessionManager.getLeafId?.());
if (relayedKey === relayKey) return;
// 模型这一轮已亲手回过这条线索 → 让位B-5.3)。
// 否则收件箱里是两封说同一件事的邮件(生产实测过)。
if (shouldSkipAutoRelay(explicitSends.get(piSessionId), mailContext.replyTo, mailContext.mailID)) {
relayedKey = relayKey;
log(`本轮模型已主动回信 ${mailContext.replyTo},跳过自动转发`);
return;
}
await client.post('/mail/send', {
to: mailContext.replyTo,
subject: replySubject(mailContext.subject),
body: text,
reply_to: mailContext.mailID || '',
// relay + relay_key 走免配额通道I-2模型已经把话说完了桥只是把它
// 搬到邮件里。对搬运收费会让配额用尽时 Agent 连交代都做不了。
relay: 'summary',
relay_key: relayKey,
});
relayedKey = relayKey;
log(`已转发本轮总结给 ${mailContext.replyTo}${text.length} 字)`);
}
// ─── 主流程 ───
async function run() {
const data = job.data;
const kind = job.kind;
client = new GatewayClient({
url: job.config.gatewayURL,
agentName: job.config.agentName,
agentKey: job.config.agentKey,
agentSecret: job.config.agentSecret,
});
// ModelRuntime 每个 worker 建一次。allowModelNetwork 保持默认 false
// 上报给 Gateway 的是 getAvailable()(有凭证、真能调起来的),
// 那取决于本机 auth.json不取决于目录里有多少条。
modelRuntime = await ModelRuntime.create();
const runtimeErr = modelRuntime.getError?.();
if (runtimeErr) log(`模型运行时告警: ${runtimeErr}`);
// connect_to_server 在 worker 里换了坐标要让主进程知道worker 马上就退了,
// 改在自己身上等于没改。主进程收到后重建 SSE 并写进后续 worker 的 job。
const mailTools = createMailTools({
client, log, agentName: job.config.agentName,
onReconnect: () => send({
type: 'reconfigure', url: client.baseURL, agentKey: client.agentKey,
}),
});
const { session, sessionManager, diagnostics, cwd, reused } = await loadSession(mailTools);
for (const d of diagnostics) log(`扩展诊断: ${d?.message ?? JSON.stringify(d)}`);
piSessionId = session.sessionId;
// 「这条会话是接管来的吗」以 Gateway 给的 platform_session_id 为准,
// **不能**从 `reused && !job.session.sessionFile` 推断:第二封邮件进同一条
// 接管会话时主进程给了 sessionFile那样推断会得出 false于是命名同步跑起来
// 把人从补全里选的 slug 冲掉 —— 那正是上一轮修掉的那个 bug。
const adopted = Boolean(adoptedSessionID(data));
send({
type: 'session_opened',
piSessionId,
sessionFile: session.sessionFile || '',
cwd,
reused,
});
// pi 侧改名(人在 TUI 里 /name→ 同步给 Gateway。接管会话不同步理由见
// relaySummary 的 adopted 参数注释。
session.subscribe((event) => {
if (event?.type === 'session_info_changed' && !adopted) {
syncNaming(session, event.name).catch((e) => log(`命名同步失败: ${describeError(e)}`));
}
});
const prompt = buildMailPrompt({ agentName: job.config.agentName, data, kind, reused });
// 轮次超时给得很宽(默认 10 分钟,由主进程的 workerMaxMs 派生)。
//
// 原来是 60 秒「超时按成功返回」,因为主进程要腾出手来收下一封邮件;
// worker 没有这个理由 —— 它只为这封邮件活着,等真结论更准。带工具调用的
// 一轮跑几分钟很正常60 秒返回会让转发落在一个还没说完的结论上。
const turnTimeout = job.config.turnTimeoutMs;
if (reused) {
// 续谈不做模型降级:换模型要换会话,会丢掉整条上下文 —— 而上下文正是
// 发件人指定这条会话的原因。但失败要可见。
const outcome = await runTurn(session, prompt, turnTimeout);
log(`续谈 ${piSessionId}mail ${data.mail_id}${outcome.queued ? ',已排队' : ''}`);
if (!outcome.ok) throw new Error(`续谈失败: ${outcome.error}`);
await relaySummary(session, sessionManager, adopted);
return;
}
// 按管理员划定的范围逐个尝试D-3
// 关键点:`prompt()` resolve **不代表模型跑成功了** —— 无凭证的 provider
// 会让它 reject实测 `No API key found for amazon-bedrock.`
// 而上游报错走 stopReason==='error'。判定交给 classifyTurnOutcome。
const attempts = modelAttemptOrder(job.config.allowedModels, {
provider: job.config.replyProvider,
model: job.config.replyModel,
});
const failures = [];
let live = { session, sessionManager };
for (const route of attempts) {
const label = route ? `${route.provider}/${route.model}` : '(平台默认)';
if (route) {
const model = modelRuntime.getModel(route.provider, route.model);
if (!model) {
// 目录里根本没有这个路由:同步就能判定,不必起一轮。
failures.push({ ...route, error: `平台目录里没有 ${label}` });
log(`模型 ${label} 不存在,跳过`);
continue;
}
// 换模型要换会话pi 的模型在 createAgentSession 时绑定。
// 上一次尝试失败的会话没有任何 assistant 消息,丢掉不损失内容。
try { live.session.dispose?.(); } catch { /* 已经没了 */ }
const retried = await openSession({
cwd, modelRuntime, model, customTools: mailTools,
extension: permissionExtension(),
});
live = { session: retried.session, sessionManager: retried.sessionManager };
piSessionId = retried.session.sessionId;
send({
type: 'session_opened',
piSessionId,
sessionFile: retried.session.sessionFile || '',
cwd,
reused: false,
});
}
const outcome = await runTurn(live.session, prompt, turnTimeout);
if (outcome.ok) {
if (failures.length) log(`${label} 成功(前 ${failures.length} 个失败)`);
await relaySummary(live.session, live.sessionManager, false);
return;
}
failures.push({ ...(route || {}), error: outcome.error });
log(`模型 ${label} 失败: ${outcome.error}`);
}
// 全部失败 → 必须回信B-6模型一次都没跑起来会话里没有任何 assistant
// 消息,自动转发因此什么也不会发 —— 发件人只会看到再无音讯。
if (kind === 'mail' && data.from_name) {
try {
await client.post('/mail/send', {
to: data.from_name,
subject: `处理失败: ${data.subject || '(无主题)'}`,
body: renderFailureReport(failures, data.subject),
reply_to: data.mail_id || '',
relay: 'summary',
relay_key: `model-failure:${data.mail_id || piSessionId}`,
});
log(`已回报模型调用失败给 ${data.from_name}`);
} catch (e) {
log(`失败回报也发不出去: ${describeError(e)}`);
}
}
// 发完仍要 throwB-6.4):静默会让这次失败只存在于邮件里,日志上看不出来。
throw new Error(`范围内 ${failures.length} 个模型全部失败:${failures.map((f) => f.error).join(' | ')}`);
}
// ─── 入口 ───
process.on('message', (msg) => {
if (msg?.type === 'job') {
if (job) return; // 一个 worker 只接一封
job = msg;
mailContext = {
replyTo: msg.data?.from_name || '',
subject: msg.data?.subject || '',
mailID: msg.data?.mail_id || '',
};
lastSyncedName = msg.lastSyncedName || '';
for (const t of msg.grants || []) grants.add(t);
run()
.then(() => finish({ ok: true, error: '' }))
.catch((e) => finish({ ok: false, error: describeError(e) }));
return;
}
if (msg?.type === 'permission_decision') {
const resolve = pending.get(msg.relayKey);
if (!resolve) return;
pending.delete(msg.relayKey);
resolve(String(msg.decision || '拒绝'));
return;
}
if (msg?.type === 'shutdown') {
// fail closedB-9.2 / N-9唤醒所有未决询问让 pi 侧那些 await 返回。
// 不唤醒的话会话挂死;而默认放行一个没人批准的危险操作更糟。
for (const [key, resolve] of pending) {
log(`未决权限 ${key} fail closed`);
resolve('shutdown');
}
pending.clear();
}
});
function finish(result) {
if (finished) return;
finished = true;
send({ type: 'done', ...result });
// **一律 exit 0**`done` 已经把成败说清楚了,用退出码再说一遍会让主进程
// 把「模型全部失败」这种已处理的结果也打成「worker 异常退出」。
// 非零退出码留给真正的崩溃(没来得及发 done 的那种)。
//
// 给 IPC 一个 tick 把 done 送出去再退:未 flush 的消息会丢,而主进程靠它判成败。
// unref 不影响触发IPC 通道本身在保持事件循环存活),只是不由这个计时器兜着。
const t = setTimeout(() => process.exit(0), 50);
if (typeof t.unref === 'function') t.unref();
}
// 未捕获异常也要回报:静默退出会让主进程只看到 exit code
// 而那条邮件的失败原因就此丢失。
process.on('uncaughtException', (e) => finish({ ok: false, error: `未捕获异常: ${describeError(e)}` }));
process.on('unhandledRejection', (e) => finish({ ok: false, error: `未处理拒绝: ${describeError(e)}` }));
send({ type: 'ready' });

View File

@ -0,0 +1,389 @@
/**
* 工作进程池的调度不变量。
*
* 这些用例**真的 fork 子进程**,用一个极小的桩 worker不装 pi SDK
* 要验的是调度(并发上限、同会话串行、排队、硬超时、状态跨 worker 存活),
* 而不是模型怎么跑。用桩让每个用例在几百毫秒内完成。
*
* 桩 worker 的协议与真 worker 一致:`ready` → 收 `job` → 按 `__hold` 停一会儿
* → 发 `done`。它还把收到的 job 载荷原样回声成一行 `JOB {…}` 日志 ——
* 判据因此能落在「主进程真的把 sessionFile / grants / 命名指纹传下来了」上,
* 而不是一个间接的计数(计数在字段被丢掉时依然会给出绿色)。
*/
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
// ─── 桩 worker ───
//
// 落到临时目录而不是仓库里:它是测试脚手架,不该被 check-shared-libs 之类的
// 一致性脚本看到,也不该让人误以为是第二个真 worker。
const STUB_DIR = mkdtempSync(join(tmpdir(), 'pi-pool-test-'));
const STUB = join(STUB_DIR, 'stub-worker.mjs');
writeFileSync(STUB, `
process.on('message', (msg) => {
if (msg?.type === 'job') {
const hold = msg.data?.__hold ?? 30;
process.send({ type: 'log', line: 'JOB ' + JSON.stringify({
mailID: msg.data.mail_id,
kind: msg.kind,
sessionFile: msg.session?.sessionFile || '',
cwd: msg.session?.cwd || '',
grants: msg.grants || [],
lastSyncedName: msg.lastSyncedName || '',
turnTimeoutMs: msg.config?.turnTimeoutMs ?? null,
}) });
process.send({ type: 'session_opened', piSessionId: 'pi-' + msg.data.mail_id,
sessionFile: '/tmp/f-' + msg.data.mail_id + '.jsonl', cwd: '/tmp', reused: false });
if (msg.data?.__grant) {
process.send({ type: 'permission_grant', toolName: msg.data.__grant });
}
if (msg.data?.__name) {
process.send({ type: 'name_synced', signature: msg.data.__name });
}
if (msg.data?.__reopen) {
// 模型降级换会话:同一个 worker 里第二次 session_opened
process.send({ type: 'session_opened', piSessionId: msg.data.__reopen,
sessionFile: '/tmp/f2.jsonl', cwd: '/tmp', reused: false });
}
if (msg.data?.__pending) {
process.send({ type: 'permission_pending', relayKey: msg.data.__pending });
return; // 等决策,见下面的分支
}
// 每 40ms 报一次心跳:并发的判据必须是「两个进程真的同时在干活」,
// 而不是「running map 里有两个条目」—— fork 返回后立即就有两个条目了。
const beat = setInterval(() => process.send({ type: 'log', line: 'TICK ' + msg.data.mail_id }), 40);
if (hold < 0) return; // 永不结束,用来验硬超时
setTimeout(() => {
clearInterval(beat);
process.send({ type: 'done', ok: true, error: '' });
setTimeout(() => process.exit(0), 20);
}, hold);
return;
}
if (msg?.type === 'permission_decision') {
process.send({ type: 'log', line: 'DECISION ' + msg.relayKey + '=' + msg.decision });
process.send({ type: 'done', ok: true, error: '' });
setTimeout(() => process.exit(0), 20);
return;
}
if (msg?.type === 'shutdown') {
process.send({ type: 'log', line: 'SHUTDOWN' });
setTimeout(() => process.exit(0), 10);
}
});
process.send({ type: 'ready' });
`);
const { createWorkerPool } = await import('../src/pool.mjs');
/** 建一个用桩 worker 的池。 */
function makePool(opts = {}) {
const lines = [];
const pool = createWorkerPool({
log: (...a) => lines.push(a.join(' ')),
config: () => ({ turnTimeoutMs: 1000, ...(opts.config || {}) }),
onReconfigure: opts.onReconfigure || (() => {}),
maxWorkers: opts.maxWorkers ?? 2,
workerMaxMs: opts.workerMaxMs ?? 5000,
workerPath: opts.workerPath || STUB,
});
return { pool, lines };
}
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
/** 轮询到条件成立或超时 —— 比固定 sleep 稳。 */
async function until(fn, timeoutMs = 4000) {
const t0 = Date.now();
while (Date.now() - t0 < timeoutMs) {
if (fn()) return true;
await sleep(20);
}
return false;
}
/** 从日志行里取出桩 worker 回声的 job 载荷。 */
function jobs(lines) {
const out = [];
for (const l of lines) {
const i = l.indexOf('JOB ');
if (i === -1) continue;
try { out.push(JSON.parse(l.slice(i + 4))); } catch { /* 不是完整一行 */ }
}
return out;
}
/** 某个 mailID 的心跳出现过几次。 */
const ticks = (lines, id) => lines.filter((l) => l.includes(`TICK ${id}`)).length;
test('并发上限被遵守:第三条会话要等前面空出来', async () => {
const { pool } = makePool({ maxWorkers: 2 });
for (const id of ['a', 'b', 'c']) {
pool.submit('mail', { mail_id: id, session_id: `S-${id}`, __hold: 250 });
}
let peak = 0;
const t = setInterval(() => { peak = Math.max(peak, pool.stats().running); }, 15);
const sawQueue = await until(() => pool.stats().queued > 0, 1000);
await until(() => pool.stats().running === 0 && pool.stats().queued === 0);
clearInterval(t);
pool.stop();
assert.ok(peak <= 2, `同时跑的 worker 峰值 ${peak},不该超过 maxWorkers=2`);
assert.ok(sawQueue, '满载时第三封该进队列而不是被丢掉');
});
test('同一会话串行:两个 worker 的心跳不得重叠', async () => {
const { pool, lines } = makePool({ maxWorkers: 3 });
pool.submit('mail', { mail_id: 'm1', session_id: 'SAME', __hold: 220 });
pool.submit('mail', { mail_id: 'm2', session_id: 'SAME', __hold: 60 });
// 判据一:任一时刻只有一个 worker 在跑。
let everTwo = false;
const t = setInterval(() => { if (pool.stats().running > 1) everTwo = true; }, 10);
await until(() => jobs(lines).length === 2 && pool.stats().running === 0, 5000);
clearInterval(t);
pool.stop();
assert.equal(everTwo, false, '同一条会话不得有两个 worker 同时装载会话文件');
// 判据二m2 一次心跳都没能在 m1 结束前发出 —— m1 的心跳数应当远多于 m2。
assert.ok(ticks(lines, 'm1') >= 3, `m1 该跑满 220ms实际心跳 ${ticks(lines, 'm1')}`);
assert.equal(jobs(lines).length, 2, '两封都要被处理,不能因为串行而丢掉一封');
});
test('不同会话真并发:两个进程的心跳在同一段时间里交错', async () => {
const { pool, lines } = makePool({ maxWorkers: 3 });
pool.submit('mail', { mail_id: 'p', session_id: 'S-P', __hold: 300 });
pool.submit('mail', { mail_id: 'q', session_id: 'S-Q', __hold: 300 });
const interleaved = await until(() => ticks(lines, 'p') >= 2 && ticks(lines, 'q') >= 2, 2500);
pool.stop();
assert.ok(interleaved,
`两条不同会话应当并发,实际 p=${ticks(lines, 'p')} q=${ticks(lines, 'q')} 次心跳`);
});
test('sessionFile 与 cwd 跨 worker 传下去:第二封接着第一封的会话谈', async () => {
const { pool, lines } = makePool({ maxWorkers: 2 });
pool.submit('mail', { mail_id: 'first', session_id: 'KEEP', __hold: 30 });
await until(() => jobs(lines).length === 1 && pool.stats().running === 0);
pool.submit('mail', { mail_id: 'second', session_id: 'KEEP', __hold: 30 });
await until(() => jobs(lines).length === 2 && pool.stats().running === 0);
pool.stop();
const [j1, j2] = jobs(lines);
assert.equal(j1.sessionFile, '', '第一封时还没有会话文件');
assert.equal(j2.sessionFile, '/tmp/f-first.jsonl',
'第二封必须带上第一封开出来的会话文件,否则每封邮件都从零开始、上下文全丢');
assert.equal(j2.cwd, '/tmp', 'cwd 也要传下去');
});
test('「一直同意」与命名指纹跨 worker 存活', async () => {
const { pool, lines } = makePool({ maxWorkers: 2 });
pool.submit('mail', {
mail_id: 'g1', session_id: 'GRANT', __hold: 30,
__grant: 'bash', __name: 'platform:某名字|某名字',
});
await until(() => jobs(lines).length === 1 && pool.stats().running === 0);
pool.submit('mail', { mail_id: 'g2', session_id: 'GRANT', __hold: 30 });
await until(() => jobs(lines).length === 2 && pool.stats().running === 0);
pool.stop();
const [j1, j2] = jobs(lines);
assert.deepEqual(j1.grants, [], '第一封时还没人点过「一直同意」');
assert.deepEqual(j2.grants, ['bash'],
'「一直同意」不跨 worker 存活的话,下一封邮件又问一遍 —— 那个选项就是在骗人');
assert.equal(j2.lastSyncedName, 'platform:某名字|某名字',
'命名指纹要传下去,否则每封邮件都重新 sync 一次');
});
test('config() 每次派活时重取allowedModels 随心跳变,不能用快照', async () => {
let turnTimeoutMs = 111;
const { pool, lines } = makePool({ maxWorkers: 1, config: {} });
// makePool 的 config 是固定值,这里换成动态的
pool.stop();
const lines2 = [];
const p2 = createWorkerPool({
log: (...a) => lines2.push(a.join(' ')),
config: () => ({ turnTimeoutMs }),
onReconfigure: () => {},
maxWorkers: 1,
workerMaxMs: 5000,
workerPath: STUB,
});
p2.submit('mail', { mail_id: 'c1', session_id: 'C1', __hold: 20 });
await until(() => jobs(lines2).length === 1 && p2.stats().running === 0);
turnTimeoutMs = 222;
p2.submit('mail', { mail_id: 'c2', session_id: 'C2', __hold: 20 });
await until(() => jobs(lines2).length === 2 && p2.stats().running === 0);
p2.stop();
const [j1, j2] = jobs(lines2);
assert.equal(j1.turnTimeoutMs, 111);
assert.equal(j2.turnTimeoutMs, 222, 'config() 必须每次重取,否则 worker 用的是上一轮的模型范围');
assert.equal(lines.length >= 0, true);
});
test('硬超时回收卡死的 worker且不堵住同会话后续邮件', async () => {
const { pool, lines } = makePool({ maxWorkers: 2, workerMaxMs: 400 });
pool.submit('mail', { mail_id: 'stuck', session_id: 'STUCK', __hold: -1 });
await until(() => pool.stats().running === 1, 1500);
const freed = await until(() => pool.stats().running === 0, 3000);
assert.ok(freed, '卡死的 worker 必须被硬超时回收,否则那条会话的后续邮件永远排队');
assert.ok(lines.some((l) => l.includes('强杀')), `应记下强杀日志,实际:\n${lines.join('\n')}`);
pool.submit('mail', { mail_id: 'after', session_id: 'STUCK', __hold: 30 });
const ran = await until(() => jobs(lines).some((j) => j.mailID === 'after'), 2000);
await until(() => pool.stats().running === 0);
pool.stop();
assert.ok(ran, '硬超时后同一会话的后续邮件必须能被处理');
});
test('权限决策路由到发起询问的那个 worker', async () => {
const { pool, lines } = makePool({ maxWorkers: 2 });
pool.submit('mail', { mail_id: 'perm', session_id: 'PERM', __pending: 'rk-1' });
await until(() => pool.stats().running === 1, 1500);
await sleep(150); // 等 permission_pending 到主进程
assert.equal(pool.routePermission('rk-1', '同意'), true, '应当路由成功');
await until(() => pool.stats().running === 0, 2000);
pool.stop();
assert.ok(lines.some((l) => l.includes('DECISION rk-1=同意')),
`worker 应收到决策原文,实际:\n${lines.join('\n')}`);
});
test('决策发的是选项原文而不是归一化的 allow/deny', async () => {
const { pool, lines } = makePool({ maxWorkers: 2 });
pool.submit('mail', { mail_id: 'p2', session_id: 'P2', __pending: 'rk-2' });
await until(() => pool.stats().running === 1, 1500);
await sleep(150);
pool.routePermission('rk-2', '一直同意');
await until(() => pool.stats().running === 0, 2000);
pool.stop();
// 「同意」与「一直同意」语义不同,归一化会让后者退化成单次授权
assert.ok(lines.some((l) => l.includes('DECISION rk-2=一直同意')),
`必须原文透传,实际:\n${lines.join('\n')}`);
});
test('决策找不到 worker 时返回 false调用方据此走 B-4.2 降级)', async () => {
const { pool } = makePool();
assert.equal(pool.routePermission('never-seen', '同意'), false);
pool.stop();
});
test('worker 退出后它的权限路由被清掉,不会误投给下一个 worker', async () => {
const { pool } = makePool({ maxWorkers: 2 });
pool.submit('mail', { mail_id: 'gone', session_id: 'GONE', __pending: 'rk-gone' });
await until(() => pool.stats().running === 1, 1500);
await sleep(150);
// 不给决策,直接停掉它
pool.stop();
await until(() => pool.stats().running === 0, 4000);
assert.equal(pool.routePermission('rk-gone', '同意'), false,
'worker 已退出,路由必须返回 false 让调用方走降级路径');
});
test('mailDrivenIDs 报出所有跑过的 pi 会话,且不随 worker 退出而清', async () => {
const { pool, lines } = makePool({ maxWorkers: 2 });
pool.submit('mail', { mail_id: 'd1', session_id: 'D1', __hold: 30 });
pool.submit('mail', { mail_id: 'd2', session_id: 'D2', __hold: 30 });
await until(() => jobs(lines).length === 2 && pool.stats().running === 0);
pool.stop();
const ids = pool.mailDrivenIDs();
assert.ok(ids.has('pi-d1'), 'D1 的 pi 会话该被标记为邮件驱动');
assert.ok(ids.has('pi-d2'), 'D2 的 pi 会话该被标记为邮件驱动');
});
test('模型降级换掉的旧 pi 会话仍算邮件驱动', async () => {
const { pool, lines } = makePool({ maxWorkers: 2 });
pool.submit('mail', { mail_id: 'r', session_id: 'RETIRE', __hold: 40, __reopen: 'pi-new' });
await until(() => jobs(lines).length === 1 && pool.stats().running === 0);
pool.stop();
const ids = pool.mailDrivenIDs();
assert.ok(ids.has('pi-new'), '新会话要在');
assert.ok(ids.has('pi-r'), '被换掉的旧会话也参与过邮件往来,磁盘上的文件还在,快照该报它');
});
test('hasSession 只对跑过的邮件会话为真', async () => {
const { pool } = makePool();
assert.equal(pool.hasSession('NOPE'), false);
pool.submit('mail', { mail_id: 'h1', session_id: 'HAS', __hold: 30 });
await until(() => pool.hasSession('HAS'), 2000);
await until(() => pool.stats().running === 0);
pool.stop();
assert.equal(pool.hasSession('HAS'), true, 'worker 退出后仍该记着这条会话');
});
test('kind 透传:权限通知走 permission 而不是 mail', async () => {
const { pool, lines } = makePool({ maxWorkers: 2 });
pool.submit('permission', { mail_id: 'k1', session_id: 'K1', __hold: 20 });
await until(() => jobs(lines).length === 1 && pool.stats().running === 0);
pool.stop();
assert.equal(jobs(lines)[0].kind, 'permission',
'kind 决定 worker 用哪套提示词,传错会让模型以为收到一封新邮件');
});
test('stop 之后不再派活', async () => {
const { pool } = makePool();
pool.stop();
pool.submit('mail', { mail_id: 'late', session_id: 'LATE', __hold: 30 });
await sleep(200);
assert.equal(pool.stats().running, 0, '关停后不该再起 worker');
assert.equal(pool.stats().queued, 0, '关停后队列应为空');
});
test('stop 会先给 worker 发 shutdown让它 fail closed再杀', async () => {
const { pool, lines } = makePool({ maxWorkers: 2 });
pool.submit('mail', { mail_id: 's1', session_id: 'S1', __hold: -1 });
await until(() => pool.stats().running === 1, 1500);
pool.stop();
const gotShutdown = await until(() => lines.some((l) => l.includes('SHUTDOWN')), 2000);
assert.ok(gotShutdown,
'必须先发 shutdown直接 SIGKILL 会让 pi 侧那些等权限的 await 永不返回');
});
test('没有 session_id 的事件各占一个 key不会互相串行', async () => {
const { pool, lines } = makePool({ maxWorkers: 3 });
pool.submit('mail', { mail_id: 'n1', __hold: 300 });
pool.submit('mail', { mail_id: 'n2', __hold: 300 });
const both = await until(() => ticks(lines, 'n1') >= 2 && ticks(lines, 'n2') >= 2, 2500);
pool.stop();
assert.ok(both, '无 session_id 的两封不属于同一条会话,应能并发');
});
test('reconfigure 上报被转达给主进程', async () => {
const STUB2 = join(STUB_DIR, 'stub-reconf.mjs');
writeFileSync(STUB2, `
process.on('message', (msg) => {
if (msg?.type === 'job') {
process.send({ type: 'reconfigure', url: 'http://new:9999', agentKey: 'k2' });
process.send({ type: 'done', ok: true, error: '' });
setTimeout(() => process.exit(0), 20);
}
});
process.send({ type: 'ready' });
`);
let got = null;
const { pool } = makePool({
maxWorkers: 1,
workerPath: STUB2,
onReconfigure: (url, key) => { got = { url, key }; },
});
pool.submit('mail', { mail_id: 'r1', session_id: 'R1' });
await until(() => got !== null, 3000);
pool.stop();
assert.deepEqual(got, { url: 'http://new:9999', key: 'k2' },
'worker 里 connect_to_server 换的坐标必须回到主进程 —— worker 马上就退了,改在它自己身上等于没改');
});