fix: relay 死循环防护 + DSH 工作区注册修复 + homeagent 工具集补齐 + 三插件 connect_to_server
## relay 死循环防护（两道防线） ### 主防线：免配额只给发往人类的 relay（handler/mail.go） 原设计：relay 走免配额通道（harness 搬运不该算模型自主发信）。 问题：收件方是另一个同样会自动转发的 Agent 时，整个回路里没有任何 一处在计数——生产上跑出过 41 封（会话 f3d824ce），间隔从 15 分钟 缩到 5 秒，且用了 37 封才烧掉 4/20 预算。 改为：repo.IsHumanUser(to.Name) 判定。Agent→Agent 的 relay 照样扣预算。 顺带修次序问题：原来是「先占幂等键再扣预算」，预算耗尽时幂等键 已被占用，加了额度也无法重发。现在预算失败会 ReleaseRelay 还回去。 ### 兜底：hop_limit 列接通（repo/relayhops.go） schema 里早有 hop_limit INT DEFAULT 5，从未有代码读它。 CountTrailingRelayHops 从最新邮件往前扫，遇到第一封非 relay 邮件即停（中间有一封自主发信或人类插话就归零）。 5 测试：空会话 / 只数 relay / 自主发信打断归零 / 达到上限 / 按会话独立 ## DSH 工作区注册修复 问题：上一轮加的 workspaceRegistry.create(cwd) 用了兜底值 cwd（来自 resolveWorkspaceCwd，可能是 ~/.dsh/mail-sessions/mail-<uuid>）， 而不是会话 header 里的真实 cwd。两者不一致时 attachSession 拒绝， 且 create 已先执行，每封邮件都往注册表里塞一条空的垃圾 workspace。 修复：读 handle.agent.session.header.cwd —— create 路径下是 meta.cwd， resume 路径下是持久化 header 里那个。 ## homeagent 插件：11 工具齐平 opencode tools.go 新增：read_mail / forward_mail / suggest_address / list_contacts / session_participants / read_thread / connect_to_server + handleConnectToServer（注册到 Gateway 前先用候选坐标试注册， 成功才写回 p.gwURL/p.key，失败不破坏原配置） 关键修：Plugin.name（插件名，homed 注册用）与 Plugin.agentName （AgentMail 身份，Gateway 密钥绑定用）是两个命名空间。 它们混淆会导致 403：「该密钥已绑定到 Agent 'homeagent'，不能用于 注册 'homeagent-mail-bridge'」。现已分开，并在 systemd drop-in 里显式设 AGENTMAIL_AGENT_NAME=homeagent。 ## 三插件补齐 connect_to_server 之前只有 opencode 有。后果：Gateway 换地址或密钥需要重新登记时， opencode 里的模型能自己修好，其他平台只能干等环境变量被人改。 DSH 版：从 GatewayClient 内部调 register()，成功后写回 client.baseURL 与 client.agentKey 当场生效。 pi 版：新导出 KEY_FILE / saveLocalKey（从 gateway.mjs），connect 工具直接用。 # 测试 relayhops_test.go 5 例 opencode 172 / dsh 188 / pi 214 全绿 check-shared-libs.sh 三方同源（rename-proposal 已纳入校验）
This commit is contained in:
@ -186,7 +186,44 @@ func SendMail(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
var budget repo.SessionBudget
|
||||
// relayFree 表示本次 relay 走免配额通道。
|
||||
//
|
||||
// **免配额只给发往人类的 relay。**
|
||||
//
|
||||
// 豁免的理由是「harness 把平台原生的权限询问与最终总结搬进邮件，
|
||||
// 不该算模型的自主发信」—— 而那是**假定收件方是人**写的。
|
||||
// 收件方是另一个同样会自动转发的 Agent 时，双方都不在做决定，
|
||||
// 整个回路里没有任何一处在计数 —— 生产上跑出过 41 封且间隔从
|
||||
// 15 分钟缩到 5 秒的无穷循环（会话 f3d824ce）。
|
||||
//
|
||||
// 因此 Agent→Agent 的 relay 照样扣会话预算，max_rounds 就能截断它。
|
||||
relayFree := false
|
||||
if relay != "" {
|
||||
human, hErr := repo.IsHumanUser(r.Context(), to.Name)
|
||||
if hErr != nil {
|
||||
Error(w, http.StatusInternalServerError, "Failed to resolve recipient")
|
||||
return
|
||||
}
|
||||
relayFree = human
|
||||
}
|
||||
|
||||
if relay != "" {
|
||||
// 硬上限：一条会话里**连续**的 relay 邮件不得超过上限。
|
||||
//
|
||||
// 与预算无关的第二道防线：预算给得大（比如 200）时，两个 Agent 仍能
|
||||
// 烧掉 200 个来回；而故障报告这类**必须**走 relay 的邮件也需要受约束。
|
||||
//
|
||||
// 「连续」是关键：中间只要有一封自主发信或人类插话，计数就归零。
|
||||
hops, hopErr := repo.CountTrailingRelayHops(r.Context(), sessionID)
|
||||
if hopErr == nil && hops >= repo.MaxRelayHops() {
|
||||
Error(w, http.StatusForbidden, fmt.Sprintf(
|
||||
"本会话已连续 %d 封自动转发（上限 %d）。这通常意味着两个 Agent 在互相"+
|
||||
"唤醒而无人决策。若确实需要继续，请由模型主动调 send_mail（不带 relay），"+
|
||||
"或由人类在会话里插一句话。",
|
||||
hops, repo.MaxRelayHops()))
|
||||
return
|
||||
}
|
||||
|
||||
// 先占幂等键。重复则说明这条上游消息已经转过，
|
||||
// 这是插件重试 / SSE 重放的正常结果，不是故障 —— 幂等地返回成功。
|
||||
if cErr := repo.ClaimRelay(r.Context(), agentName, relayKey, relay); cErr != nil {
|
||||
@ -202,7 +239,10 @@ func SendMail(w http.ResponseWriter, r *http.Request) {
|
||||
Error(w, http.StatusInternalServerError, "Failed to claim relay")
|
||||
return
|
||||
}
|
||||
// 仅读快照用于回传，不扣预算
|
||||
}
|
||||
|
||||
if relayFree {
|
||||
// 只读快照用于回传，不扣预算
|
||||
budget, _ = repo.GetSessionBudget(r.Context(), sessionID)
|
||||
} else {
|
||||
// 额度只看【本任务】的往返预算。
|
||||
@ -212,6 +252,11 @@ func SendMail(w http.ResponseWriter, r *http.Request) {
|
||||
// 靠的是新建会话速率限制（resolveTarget 里）。
|
||||
budget, err = repo.ConsumeSessionBudget(r.Context(), sessionID)
|
||||
if errors.Is(err, repo.ErrSessionBudgetExhausted) {
|
||||
// 预算耗尽时要把幂等键还回去：否则那条上游消息永远转不出来了，
|
||||
// 之后管理员加了额度也无法重发。
|
||||
if relay != "" {
|
||||
_ = repo.ReleaseRelay(r.Context(), agentName, relayKey)
|
||||
}
|
||||
Error(w, http.StatusForbidden, fmt.Sprintf(
|
||||
"本任务的往返预算已用尽（%d/%d）。自动转发的总结与权限询问不占预算；"+
|
||||
"若需继续主动发信，请让人在对话页调高本任务的预算。",
|
||||
|
||||
83
gateway/internal/repo/relayhops.go
Normal file
83
gateway/internal/repo/relayhops.go
Normal file
@ -0,0 +1,83 @@
|
||||
package repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// 连续 relay 跳数限制 —— 防止两个 Agent 靠自动转发互相唤醒到无穷。
|
||||
//
|
||||
// # 这是什么问题
|
||||
//
|
||||
// 每个插件都在「一轮结束时把模型最后那段话自动发回去」（契约 B-5）。
|
||||
// 当收件方也是一个装了同类插件的 Agent 时，这封信唤醒对方 → 对方跑一轮 →
|
||||
// 对方也自动回一封 → 循环。**双方都没有「决定继续」，因为双方都不在做决定** ——
|
||||
// 发信是插件代劳的。
|
||||
//
|
||||
// 生产上真实发生过：会话 f3d824ce（dsh 与 opencode 联调 llmsproxy）共 41 封，
|
||||
// 最后一封人类意图的邮件之后，**每一封都是 relay:summary**，
|
||||
// 间隔从 15 分钟一路缩到 5 秒，内容已无新增信息。
|
||||
//
|
||||
// # 为什么 relay_key 拦不住
|
||||
//
|
||||
// 它是幂等键，职责是「同一条上游消息不重复转发」，这一点它做对了。
|
||||
// 但每一轮都是**真正不同**的新消息：opencode 侧是 assistant message id
|
||||
// （msg_0655bbf6…、msg_0656cc7fc…），dsh 侧是事件计数（…:12220、…:13347）。
|
||||
// 每次 ClaimRelay 都合法通过。
|
||||
//
|
||||
// # 为什么需要两道防线
|
||||
//
|
||||
// 主防线是「免配额只给发往人类的 relay」（见 handler.SendMail）：
|
||||
// Agent→Agent 的自动转发转而消耗会话预算，max_rounds 会截断它。
|
||||
//
|
||||
// 但那还不够：预算给得大（比如 200）时，两个 Agent 仍能烧掉 200 个来回；
|
||||
// 而故障报告这类**必须**走 relay 的邮件也需要受约束。因此这里再加一道
|
||||
// 与预算无关的硬上限：一条会话里**连续**的 relay 邮件不得超过 maxRelayHops。
|
||||
//
|
||||
// 「连续」是关键：只要中间有一封自主发信（模型真的决定说什么）或人类插话，
|
||||
// 计数就归零。这让正常的「模型回一封、插件补一封总结」不受影响，
|
||||
// 只掐住「全程无人决策」的那种回路。
|
||||
//
|
||||
// hop_limit 列早就在 schema 里（DEFAULT 5）却从没有人读它 —— 它显然
|
||||
// 就是为这件事准备的。这里把它接上，取同一个默认值。
|
||||
const maxRelayHops = 5
|
||||
|
||||
// CountTrailingRelayHops 数会话尾部**连续**的 relay 邮件数。
|
||||
//
|
||||
// 从最新一封往前扫，遇到第一封非 relay 邮件即停。返回值即「若本次再发一封
|
||||
// relay，它会是第几跳」的前一个数。
|
||||
//
|
||||
// 判据用 relayed_mails 的存在性而不是 mails 上的某个标记：
|
||||
// relay 身份本来就记在那张表里，在 mails 上再存一份等于给同一事实留两个答案。
|
||||
func CountTrailingRelayHops(ctx context.Context, sessionID uuid.UUID) (int, error) {
|
||||
rows, err := db.DB.QueryContext(ctx, `
|
||||
SELECT CASE WHEN r.mail_id IS NULL THEN 0 ELSE 1 END AS is_relay
|
||||
FROM mails m
|
||||
LEFT JOIN relayed_mails r ON r.mail_id = m.mail_id
|
||||
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()
|
||||
|
||||
hops := 0
|
||||
for rows.Next() {
|
||||
var isRelay int
|
||||
if err := rows.Scan(&isRelay); err != nil {
|
||||
return hops, err
|
||||
}
|
||||
if isRelay == 0 {
|
||||
// 遇到一封自主发信/人类邮件：链条到此为止
|
||||
break
|
||||
}
|
||||
hops++
|
||||
}
|
||||
return hops, rows.Err()
|
||||
}
|
||||
|
||||
// MaxRelayHops 暴露上限供错误文案与测试使用。
|
||||
func MaxRelayHops() int { return maxRelayHops }
|
||||
126
gateway/internal/repo/relayhops_test.go
Normal file
126
gateway/internal/repo/relayhops_test.go
Normal file
@ -0,0 +1,126 @@
|
||||
package repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// 连续 relay 跳数上限守的是一个真实事故：会话 f3d824ce（dsh 与 opencode 联调
|
||||
// llmsproxy）共 41 封，最后一封人类意图的邮件之后每一封都是 relay:summary，
|
||||
// 间隔从 15 分钟一路缩到 5 秒。双方都没有「决定继续」，因为双方都不在做决定 ——
|
||||
// 发信是插件代劳的，而免配额通道让整个回路里没有任何一处在计数。
|
||||
|
||||
func TestTrailingRelayHopsEmptySession(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
sid, _ := CreateSession(ctx, nil, "admin", "空会话", "/tmp/ws")
|
||||
n, err := CountTrailingRelayHops(ctx, sid)
|
||||
if err != nil {
|
||||
t.Fatalf("空会话应正常返回: %v", err)
|
||||
}
|
||||
if n != 0 {
|
||||
t.Fatalf("空会话跳数应为 0，实为 %d", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTrailingRelayHopsCountsOnlyRelay(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
sid, _ := CreateSession(ctx, nil, "admin", "混合", "/tmp/ws")
|
||||
// 人类发一封（非 relay）
|
||||
mustMail(t, sid, "admin", "", "dsh", "/tmp/ws", nil)
|
||||
// 插件连续转发三封
|
||||
for i := 0; i < 3; i++ {
|
||||
seedRelayMail(t, ctx, sid, "dsh", "opencode")
|
||||
}
|
||||
|
||||
n, err := CountTrailingRelayHops(ctx, sid)
|
||||
if err != nil {
|
||||
t.Fatalf("数跳数: %v", err)
|
||||
}
|
||||
if n != 3 {
|
||||
t.Fatalf("尾部连续 relay 应为 3，实为 %d", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTrailingRelayHopsResetsOnAutonomousSend(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
// 「连续」是这条规则的关键：中间只要有一封自主发信（模型真的决定说什么）
|
||||
// 或人类插话，计数就归零。否则正常的「模型回一封、插件补一封总结」
|
||||
// 会被误判成回路。
|
||||
sid, _ := CreateSession(ctx, nil, "admin", "打断", "/tmp/ws")
|
||||
for i := 0; i < 4; i++ {
|
||||
seedRelayMail(t, ctx, sid, "dsh", "opencode")
|
||||
}
|
||||
// 模型亲手发了一封 —— 链条到此为止
|
||||
mustMail(t, sid, "dsh", "", "opencode", "/tmp/ws", nil)
|
||||
seedRelayMail(t, ctx, sid, "opencode", "dsh")
|
||||
|
||||
n, err := CountTrailingRelayHops(ctx, sid)
|
||||
if err != nil {
|
||||
t.Fatalf("数跳数: %v", err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("自主发信之后只剩 1 跳，实为 %d", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTrailingRelayHopsReachesLimit(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
// 攒到上限：此时 handler 应当拒绝下一封 relay。
|
||||
sid, _ := CreateSession(ctx, nil, "admin", "到顶", "/tmp/ws")
|
||||
mustMail(t, sid, "admin", "", "dsh", "/tmp/ws", nil)
|
||||
for i := 0; i < MaxRelayHops(); i++ {
|
||||
seedRelayMail(t, ctx, sid, "dsh", "opencode")
|
||||
}
|
||||
|
||||
n, _ := CountTrailingRelayHops(ctx, sid)
|
||||
if n < MaxRelayHops() {
|
||||
t.Fatalf("应达到上限 %d，实为 %d", MaxRelayHops(), n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTrailingRelayHopsIsPerSession(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
// 一条会话的回路不该影响另一条：两个 Agent 在 A 会话里刷爆了，
|
||||
// B 会话的正常自动转发仍应放行。
|
||||
a, _ := CreateSession(ctx, nil, "admin", "A", "/tmp/a")
|
||||
b, _ := CreateSession(ctx, nil, "admin", "B", "/tmp/b")
|
||||
for i := 0; i < 5; i++ {
|
||||
seedRelayMail(t, ctx, a, "dsh", "opencode")
|
||||
}
|
||||
seedRelayMail(t, ctx, b, "dsh", "admin")
|
||||
|
||||
na, _ := CountTrailingRelayHops(ctx, a)
|
||||
nb, _ := CountTrailingRelayHops(ctx, b)
|
||||
if na != 5 || nb != 1 {
|
||||
t.Fatalf("跳数应按会话独立计：A=%d（期望 5）B=%d（期望 1）", na, nb)
|
||||
}
|
||||
}
|
||||
|
||||
// seedRelayMail 建一封走 relay 通道的邮件（同时占幂等键并关联 mail_id），
|
||||
// 复刻 handler.SendMail 的真实写入顺序。
|
||||
func seedRelayMail(t *testing.T, ctx context.Context, sid uuid.UUID, from, to string) {
|
||||
t.Helper()
|
||||
key := "relay-" + uuid.NewString()
|
||||
if err := ClaimRelay(ctx, from, key, "summary"); err != nil {
|
||||
t.Fatalf("占幂等键: %v", err)
|
||||
}
|
||||
mid, err := CreateMail(ctx, sid, nil, from, "", to, "", "Re: 主题", "正文", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("建邮件: %v", err)
|
||||
}
|
||||
if err := BindRelayMail(ctx, from, key, mid); err != nil {
|
||||
t.Fatalf("关联 relay: %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user