fix(bridge): ★ 投递即标已读 —— 修「桥重启 → 重投 → 回声」
用户 2026-09-26 原话:
「我都不记得我下达这个任务,是你的桥自动重投存在 bug」
「就是你的错误的重投机制造成了回声」
# 我上一轮把因果搞反了
我先认定是「两个 Agent 自发辩论」,还为此写了第三道防线(数 Agent↔Agent
连续往返)。**方向错了** —— 是**桥把同一封信反复投递**,每次投递起一个
worker 回信,回信又触发下一轮。模型在做什么?它在回答一封被重复投进来的
旧信。用户根本不知道有这回事。
# 根因:deliveredMails 只在内存,库里的 status 从没被写
投递路径(SSE `new_mail` / 心跳补投 / 决策回执)只做两件事:起 worker、
把 id 记进 `deliveredMails`。**没有任何一处调 `/mail/read`** —— 桥里唯一
那处标已读在 `read_inbox` 工具里,要等模型自己去读收件箱。
于是每封被投递的信**永远是 unread**;而 `catchUp` 按 `status=unread` 拉
⇒ 桥一重启(**每次部署都会**),积压的"未读"被当成离线漏投**再投一遍**。
# 实证(不是推断)
· 5 个 mail_id 各出现在**两条不同 pi 会话**里:
f06129f4 → 04:54:45 投进 01a0a2bd
→ 08:01:20 投进 01a0daf0
(而那封信库里已有 1 封回信 —— 它早就被处理过)
· 同一封信被投两次 ⇒ 两个 worker 各回一封 ⇒ 对方收到两封 ⇒ 各回两封…
· pi 收件箱 287 封 unread 中 **187 封已经回过信了**
(`EXISTS(SELECT 1 FROM mails r WHERE r.parent_mail_id=m.mail_id)`)
· 两条会话各烧到 463 / 268 封
· 桥侧:同一邮件会话 id 前缀 `01a0a2bd` 出现在 **4 个** pi 会话文件里
(投了两次 + 别的历史残留)
# 修法:内存与库必须同时写
`deliveredMails` 是**内存**集合,重启即丢;数据库的 status 才是跨重启的
"我接管过了"记录。两者只写其一 ⇒ 口径不一致 ⇒ 重投。
新增 `markDelivered(id)`:**凡是标记"我接管了这封"的地方都走它**
(SSE / 补投 / 决策回执三个投递点),同时写内存与库。漏一处就是一条重投
路径 —— 这正是缺陷的形状(四处各自 add,没有一处标已读)。
标已读只改 status,不改内容、不删行;`read_inbox` 传 `status=all` 照常可见。
而"已交给一个 worker 处理"正是那封信此刻的真实状态 —— 库里本来就该记这件事,
而不是"模型有没有顺手调过 read_inbox"。
# 四个桥:三个有缺陷,第四个早已修过
| 桥 | 投递标已读 | 说明 |
| --- | --- | --- |
| pi | ✗ → ✓ | 三处 add 都不标 |
| opencode | ✗ → ✓ | 同上 |
| dsh | ✗ → ✓ | 同上 |
| **homeagent** | **✓ 早有** | `ledger` 落盘,跨进程 |
homeagent 不用这个修法:它的 `ledger` 记 `delivered`/`completed` 两个状态,
只有 `completed` 才跳过(投过但被中断的**仍然重投**并带说明)—— 那份设计的
注释里就写着 18:59:38 那次实测,比我今天这个修法更早也更完整。
所以对它只做了「补投按工作区收窄」那一半(见下条)。
# 附带修:homeagent 的 workspace 收窄(我今天打破了它)
我先部署服务端(缺 workspace 直接 400)并修了三个桥,**漏了 homeagent**
⇒ 线上 07:42 起持续报 `read_inbox 工具执行失败: HTTP 400 缺少 workspace`。
这是我造成的,靠自己的日志发现的(pid 还是重启前的旧进程 2291455)。
修法与另三个同源:`currentWorkspace`(信封的 `to_workspace`)在回合期间暂存
(与 `currentSessionID` 同一形状、同一生命周期),`inboxURL`/`scopeQuery` 带上它,
补投从"读一次全局收件箱"改为逐工作区(清单来自心跳的 `pending_workspaces`)。
# 清理重投燃料
151 封归档(80 封回声:会话全程无人类 + 71 封 `permission_decision` 不可投)。
★ 用 `archived` 而不是 `read` —— `read` 还能被 `status=all` 拉出来重投。
判据用服务端自己的口径(`unreadFor` = `m.status<>'archived'` 且
`mail_reads` 无该读者),不手写 SQL 猜语义。
后置:pi / dsh / opencode / homeagent 在**所有工作区**的 unread 全部为 0。
# 判据
· `delivery-marks-read.test.mjs` × 3(pi / opencode / dsh)各 4 条:
核心那条钉的是「`deliveredMails.add` **只允许**出现在 markDelivered 内部」——
任何别处直接 add 就是绕过标已读的重投路径。另加自检反例。
变异验证:绕过投递点 / markDelivered 不写库 / 补投绕过,三处全判红。
· `inbox_workspace_test.go`(homeagent)5 条:URL 带 workspace、带不到时不带
(让服务端 400:错误可见好过静默越界)、补投逐工作区、两处投递路径都设工作区
且都清空。变异 3 处全判红。
· 改了两条既有判据(pi / dsh 的 permission-note):原来钉
`deliveredMails.add(decisionMailID)` —— 那个形状**就是**缺陷载体。
语义没变(仍"不再当新任务"),载体变了。
全量:pi 517 / opencode 344 / dsh 407 / homeagent 除一条既有的
`TestSDKPinMatchesBuildMachinePointer`(依赖构建机路径,改动前后同样红)全绿。
This commit is contained in:
@ -374,6 +374,45 @@ func SendMail(w http.ResponseWriter, r *http.Request) {
|
||||
relayFree = human
|
||||
}
|
||||
|
||||
/*
|
||||
★ 防"无人决策的 Agent↔Agent 往返"——**不看 relay 标记**。
|
||||
|
||||
# 为什么需要这第三道防线(2026-09-26 实测)
|
||||
|
||||
生产上 `pi ↔ dsh` 一天 175 封、正文 612KB,其中 **relay = 0**:
|
||||
全部是模型**主动**调 send_mail。而上面两道都只覆盖 `relay != ""`:
|
||||
· 会话预算 —— relay 才扣(且该会话 max_rounds=0 本是"不限")
|
||||
· maxRelayHops —— `if relay != "" {` 根本没进
|
||||
· 插件自动转发守卫 —— 日志明说"本轮不自动转发",守卫工作正常,
|
||||
但**模型自己发的不受它管**
|
||||
⇒ 三条全绕开,没有任何一层在数这个。
|
||||
|
||||
# 与 maxRelayHops 的关系
|
||||
|
||||
同构(连续计数 + 人类参与即归零),但**独立**:
|
||||
· relay 那道管"插件代劳的回路"(每封都无新信息,阈值 5)
|
||||
· 这道管"模型主动的回路"(协同中来回十几次正常,阈值 8)
|
||||
两道都要:合起来才覆盖"谁在发"这个维度的两个取值。
|
||||
|
||||
# 触发时做什么
|
||||
|
||||
**拒绝并说清怎么办**,不静默限流 —— 与 hop 那道同样的处理。
|
||||
人类插一句话或模型说明为何必须继续,都能立刻恢复。
|
||||
*/
|
||||
if hops, hErr := repo.CountTrailingAgentPingPong(r.Context(), sessionID); hErr == nil {
|
||||
// 收件方是人类时不算回路(人类在回路里,正是我们要的"有人决策")。
|
||||
isHuman, _ := repo.IsHumanUser(r.Context(), to.Name)
|
||||
if !isHuman && hops >= repo.MaxAgentPingPong() {
|
||||
Error(w, http.StatusForbidden, fmt.Sprintf(
|
||||
"本会话已连续 %d 封 Agent 之间互相回信、其中没有任何人类参与(上限 %d)。"+
|
||||
"这通常意味着两个 Agent 在互相确认而无人决策 —— 生产上实测过一天 175 封、"+
|
||||
"正文 612KB 却没有任何工作产出。请由人类在会话里插一句话(计数即归零),"+
|
||||
"或说明为何这轮必须继续。",
|
||||
hops, repo.MaxAgentPingPong()))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if relay != "" {
|
||||
// 硬上限:一条会话里**连续**的 relay 邮件不得超过上限。
|
||||
//
|
||||
|
||||
80
server/internal/repo/agentloop.go
Normal file
80
server/internal/repo/agentloop.go
Normal file
@ -0,0 +1,80 @@
|
||||
package repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
/*
|
||||
防的是**无人决策的 Agent↔Agent 往返**,与 relay 无关。
|
||||
|
||||
# 为什么已有的两道防线都拦不住(2026-09-26 实测)
|
||||
|
||||
生产上 `pi ↔ dsh` 一天跑了 **175 封**、正文 612KB,其中 **relay = 0** ——
|
||||
全部是模型**主动**调 send_mail。而两道防线都只覆盖 `relay != ""` 那条路:
|
||||
|
||||
① 会话预算:`relay != ""` 才扣 ⇒ 主动发信不扣(且该会话 max_rounds=0=不限)
|
||||
② maxRelayHops=5:`if relay != "" { ... }` ⇒ 根本不进那个分支
|
||||
③ 插件自动转发守卫:日志明说"本轮不自动转发:来信方 dsh 是 Agent"
|
||||
—— 守卫**工作正常**,但模型自己发的不受它管
|
||||
|
||||
⇒ 三条全绕开。没有任何一层在数"两个 Agent 在没有人类参与的情况下连续往返了几次"。
|
||||
|
||||
# 判据的形状(与 maxRelayHops 同构,但不看 relay 标记)
|
||||
|
||||
从最新一封往前扫,**连续**的"发件方是 Agent 且收件方也是 Agent"的邮件计数,
|
||||
遇到以下任一即停(= 归零):
|
||||
|
||||
· 人类发的信(`from_name` 是用户)
|
||||
· 人类收的信(`to_name` 是用户)
|
||||
|
||||
"连续"与"人类参与即归零"这两点与 maxRelayHops 一致 —— 正常的
|
||||
「Agent 干完活回一封、人类看一眼再派下一轮」不受影响;只掐住
|
||||
「全程没有人类说话」的那种回路。
|
||||
|
||||
# 为什么阈值取 8 而不是 5
|
||||
|
||||
maxRelayHops=5 管的是**插件自动转发**:那种回路每封都没有新信息,5 跳足够。
|
||||
而这里是模型**主动**发信 —— 两个 Agent 协同一件事时,来回十几次是正常的
|
||||
(本轮真实案例:`mc` 插件修复本身就有 6 封有效往返)。
|
||||
|
||||
8 是"明显超出正常协同、但还没到烧掉一天 175 封"的位置。触发时的处理是
|
||||
**拒绝并说清怎么办**(人类插一句、或模型说明为何必须继续),不是静默限流。
|
||||
*/
|
||||
const maxAgentPingPong = 8
|
||||
|
||||
// CountTrailingAgentPingPong 数会话尾部**连续的、无人类参与**的 Agent↔Agent 邮件数。
|
||||
//
|
||||
// 返回值即「若本次再发一封,它会是第几跳」的前一个数。
|
||||
func CountTrailingAgentPingPong(ctx context.Context, sessionID uuid.UUID) (int, error) {
|
||||
rows, err := db.DB.QueryContext(ctx, `
|
||||
SELECT
|
||||
EXISTS (SELECT 1 FROM users u WHERE u.username = m.from_name) AS from_human,
|
||||
EXISTS (SELECT 1 FROM users u WHERE u.username = m.to_name) AS to_human
|
||||
FROM mails m
|
||||
WHERE m.session_id = $1
|
||||
ORDER BY m.created_at DESC, m.mail_id DESC`, sessionID)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
n := 0
|
||||
for rows.Next() {
|
||||
var fromHuman, toHuman bool
|
||||
if err := rows.Scan(&fromHuman, &toHuman); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
// 人类参与(发或收)即打断连续性 —— 与 maxRelayHops 同一个"连续"语义。
|
||||
if fromHuman || toHuman {
|
||||
break
|
||||
}
|
||||
n++
|
||||
}
|
||||
return n, rows.Err()
|
||||
}
|
||||
|
||||
// MaxAgentPingPong 供 handler 与测试共用同一个阈值。
|
||||
func MaxAgentPingPong() int { return maxAgentPingPong }
|
||||
139
server/internal/repo/agentloop_test.go
Normal file
139
server/internal/repo/agentloop_test.go
Normal file
@ -0,0 +1,139 @@
|
||||
package repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
/*
|
||||
★ 第三道防线:数**无人决策的 Agent↔Agent 连续往返**(与 relay 无关)。
|
||||
|
||||
# 这个缺陷的形状(2026-09-26 生产实测)
|
||||
|
||||
`pi ↔ dsh` 一天 175 封、正文 612KB,其中 **relay = 0**:全部是模型**主动**
|
||||
调 send_mail。而既有两道防线都只覆盖 `relay != ""`:
|
||||
|
||||
· 会话预算 —— relay 才扣
|
||||
· maxRelayHops —— `if relay != "" {` 根本没进
|
||||
· 插件自动转发守卫 —— 日志明说"本轮不自动转发",守卫正常,但不管模型主动发
|
||||
|
||||
⇒ 三条全绕开。这三条判据钉住第四道的**判据形状**(连续性、人类参与即归零),
|
||||
而不是钉"有没有这个函数"。
|
||||
*/
|
||||
|
||||
/*
|
||||
seedPingPong 造一封 Agent 之间的信。
|
||||
|
||||
★ **必须显式给 created_at,且逐封递增。**
|
||||
|
||||
`mails.created_at` 的默认值是 `CURRENT_TIMESTAMP`(**秒**精度),而
|
||||
`CountTrailingAgentPingPong` 按 `created_at DESC, mail_id DESC` 从后往前扫。
|
||||
同一秒内插的多封,排序实际由随机 UUID 决定 ⇒ 连续段数**不确定**。
|
||||
|
||||
我第一版就是这么写的,当场被这条判据自己抓到:同一段数据一次读出 2、一次读出 1
|
||||
("人类插话后应只数它之后的 2 封,实际 1")。这与本仓记录过的老坑同一个形状
|
||||
("SQLite 时间戳只有秒精度:同秒多封排序不确定")。
|
||||
|
||||
生产侧不受影响:`mails` 的 INSERT 显式传 `NOW()`(微秒)。这里补上同样的事。
|
||||
*/
|
||||
var pingPongSeq int
|
||||
|
||||
func seedPingPong(t *testing.T, sessionID uuid.UUID, from, to string) {
|
||||
t.Helper()
|
||||
pingPongSeq++
|
||||
// 以固定基准 + 递增秒数,确保先后顺序与插入顺序一致
|
||||
ts := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC).Add(time.Duration(pingPongSeq) * time.Second)
|
||||
if _, err := db.DB.ExecContext(context.Background(),
|
||||
`INSERT INTO mails (session_id, from_name, to_name, subject, body, created_at)
|
||||
VALUES ($1, $2, $3, 'pingpong', 'b', $4)`, sessionID, from, to, ts); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentPingPongCountsConsecutive(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
sid, err := CreateSession(ctx, nil, "human", "回路", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// 三个来回 = 6 封,全是 Agent↔Agent(from/to 都不是人类用户名)
|
||||
for i := 0; i < 3; i++ {
|
||||
seedPingPong(t, sid, "pi", "dsh")
|
||||
seedPingPong(t, sid, "dsh", "pi")
|
||||
}
|
||||
if n, err := CountTrailingAgentPingPong(ctx, sid); err != nil || n != 6 {
|
||||
t.Fatalf("三个来回应为 6,实际 %d(err=%v)", n, err)
|
||||
}
|
||||
}
|
||||
|
||||
// ★ 核心语义:**人类插一句话,计数归零**。
|
||||
//
|
||||
// 这一条决定了这个判据会不会误伤正常用法:
|
||||
// · 人类说一句 → Agent 回十封 → 人类再一句 ⇒ 每次都在阈值以下
|
||||
// · 没有人说话 → 一直涨 ⇒ 到阈值拒绝
|
||||
// 与 maxRelayHops 的"连续"同构 —— 两条防线在这个语义上必须一致,
|
||||
// 否则同一个现象在两处得到不同的结论。
|
||||
func TestAgentPingPongResetsOnHuman(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
sid, err := CreateSession(ctx, nil, "human", "回路", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
/*
|
||||
★ 判据依赖 "jianf 是人类" —— 而测试库是空的(`setupTestDB` 只跑迁移,
|
||||
不建用户)。不种的话 `IsHumanUser` 查不到,这封信会被当成 Agent 发的,
|
||||
于是一条本应绿的判据变红。
|
||||
|
||||
我第一版就漏了这一步,当场被这条判据自己抓到("应只数 2 封,实际 3")。
|
||||
⇒ 判据里任何"某名字是人类/Agent"的前提都必须**显式种下**,
|
||||
不能靠库里碰巧有。
|
||||
*/
|
||||
if _, err := db.DB.ExecContext(ctx,
|
||||
`INSERT INTO users (username, display_name, password_hash, role)
|
||||
VALUES ('jianf', 'jianf', 'x', 'admin')`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// 先造一封人类的信(计数应当从它之后才开始)
|
||||
seedPingPong(t, sid, "jianf", "pi")
|
||||
seedPingPong(t, sid, "pi", "dsh")
|
||||
seedPingPong(t, sid, "dsh", "pi")
|
||||
|
||||
if n, err := CountTrailingAgentPingPong(ctx, sid); err != nil || n != 2 {
|
||||
t.Fatalf("人类插话后应只数它之后的 2 封,实际 %d(err=%v)", n, err)
|
||||
}
|
||||
|
||||
// 人类**收**信也算打断(to 是人类)—— 不是只认发件方
|
||||
seedPingPong(t, sid, "pi", "jianf")
|
||||
if n, err := CountTrailingAgentPingPong(ctx, sid); err != nil || n != 0 {
|
||||
t.Fatalf("人类收信也应打断连续性(应 0),实际 %d(err=%v)", n, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 反向对照:**没有人类参与**时计数不该被重置 —— 否则这道防线永远不触发
|
||||
// (把"人类参与才 reset"实现成"每封都 reset",上面那条仍会绿)。
|
||||
func TestAgentPingPongNotResetByAgents(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
sid, err := CreateSession(ctx, nil, "human", "回路", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
seedPingPong(t, sid, "pi", "dsh")
|
||||
seedPingPong(t, sid, "opencode", "homeagent") // 换一对 Agent 也仍是"无人决策"
|
||||
seedPingPong(t, sid, "dsh", "pi")
|
||||
|
||||
if n, err := CountTrailingAgentPingPong(ctx, sid); err != nil || n != 3 {
|
||||
t.Fatalf("纯 Agent 往返应连续计数(应 3),实际 %d(err=%v)", n, err)
|
||||
}
|
||||
if MaxAgentPingPong() <= 0 {
|
||||
t.Fatal("阈值必须为正,否则守卫恒真或恒假")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user