Files
MailUI4Agents/gateway/internal/handler/agents.go
JianFeeeee 9e5c557cdf feat: 跨主机 Agent 验证 + 离线邮件补投 + 400 指向具体字段
7.8「跨主机 Agent 发现」原计划(Gateway + Registry 拆分、etcd/Consul 注册)
取消,改为验证现有协议已经够用。验证过程暴露两个真实缺陷,一并修掉。

## 为什么不做注册中心

它要解决「Gateway 怎么找到 Agent」,而这个问题在本架构里不存在:
连接方向是单向的 —— Agent 主动连 Gateway,Gateway 从不外呼。
远端 Agent 只需要一个公网 URL 加一把密钥,被叫方自己会打进来。
注册中心要解决的「被叫方在哪」根本没出现过。

同一个理由此前已经决定了平台会话同步走插件上报而不是 Gateway 拉取。

## 验证方式:一个纯标准库脚本

`deploy/remote-agent-demo.py` 在另一台主机(192.168.2.106)上跑,
不装 AgentMail 的任何代码。注册 / 心跳(带模型目录)/ SSE 长连 /
收件箱 / 标记已读 / 发信全通,Gateway 侧 status=online 且 last_seen 随心跳推进。
完整一轮往返跑通:admin 发给 remotebot@/tmp/remotebot-ws,脚本回信入库。

「协议层面已支持」的含义就是这个:跨主机不需要新组件,只需要三个环境变量。

## 缺陷一:SSE 只推连上之后的事件,没人补拉积压

写那个脚本时第一版只挂了 SSE,启动前发的邮件永远不会被处理。
查了才发现**两个正式插件也有这个洞** —— 原以为它们做了补拉,实际没有。
后果比明确的失败更难排查:邮件躺在收件箱里,而发件人以为 Agent 收到了。

新增共用模块 `lib/catchup.js`,两插件在首个成功心跳后补投一次。五条约束
都对应一种具体的坏行为:

- 只在**首个**心跳后补 —— 每轮都补会把「模型正在处理中、尚未标已读」的
  邮件重复投递
- 串行、一次最多 5 封 —— 每封都要起一轮模型,并发放出去等于对上游打 N 个
  并发请求,且最后几封要等前面全部跑完
- 与 SSE 共用 deliveredMails 去重 —— 心跳与 SSE 建连之间有个窗口,
  那期间到的邮件两条路都会到
- 按时间**正序**投(收件箱倒序返回)—— 倒着塞进去同一会话的上下文是乱的
- permission 类不补投 —— 原来的工具调用早随进程没了,没有可恢复的上下文

端到端两平台各验一次:停插件 → 发信 → 启插件 → 日志「补投 1 封离线期间的
邮件」→ 回信入库;随后在线再发一封确认只回一次。

## 缺陷二:400 只说 "Invalid JSON",不说是哪个字段

脚本把 `workspaces` 传成字符串数组(它要 `[{name, path}]`),
得到的只是一句固定文案,只能靠翻服务端结构体才能发现。
两个官方插件都传 `workspaces: []`,所以这个洞一直没暴露;
第三方客户端没有「翻服务端源码」这个条件。

新增 `handler.DecodeBody`,22 处 `Decode` + 固定文案的调用点全部换过去:

    {"error": "字段 \"workspaces\" 类型不对:期望 object,收到 string"}
    {"error": "JSON 语法错误(第 8 字节处)"}
    {"error": "请求体为空"}

刻意不回显 encoding/json 的原文 —— 它带 Go 类型名(models.Workspace),
那是本侧的实现细节,不该出现在公开 API 的响应里。期望类型用 JSON 的说法。
截断的 JSON 走 io.ErrUnexpectedEOF 而不是 json.SyntaxError,单独一条分支,
否则会落到笼统的兜底文案里(写测试时才发现)。

## 验证

- Go:13 个新测试(decode_test.go 含「不得泄漏 Go 类型名」断言)
- 插件:两侧各 10 个补投测试,共 200 个
- 共用模块同源校验通过(catchup 已纳入 check-shared-libs.sh)
- 生产已部署
2026-09-02 22:47:31 +08:00

211 lines
7.0 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package handler
import (
"net/http"
"github.com/agentmail/gateway/internal/middleware"
"github.com/agentmail/gateway/internal/models"
"github.com/agentmail/gateway/internal/repo"
)
// ---------- Agent ----------
type registerRequest struct {
Name string `json:"name"`
Secret string `json:"secret"`
Workspaces []models.Workspace `json:"workspaces"`
Platform string `json:"platform"`
}
// heartbeatRequest 是心跳可选带的上报体。
//
// 字段全可省:旧插件发空心跳,不能因为新增了上报就把它们报错。
type heartbeatRequest struct {
// PlatformSessions 是平台侧当前的会话快照(按最近活跃排序)。
//
// 为什么让插件上报而不是 Gateway 反向拉取:当前架构是单向的
// Agent 持密钥主动连 GatewayGateway 从不外呼)。反向拉取需要 Gateway
// 保存各平台的地址与凭证,那是另一套信任模型。
//
// nil 与空数组语义不同nil = 本次不上报(保留现有镜像),
// 空数组 = 平台侧确实一条会话都没有(清空镜像)。
// 拿不到会话列表的插件应当省略该字段,而不是传空数组把镜像抹掉。
PlatformSessions []repo.PlatformSession `json:"platform_sessions"`
// Models 是平台当前看得见的模型目录,供配置页勾选。
//
// 随心跳上报而不是只在注册时上报:模型清单会在运行中变
// (换 provider 配置、上游上下线、换了 API key。只在注册时报一次的话
// 目录会静静变陈,而管理员在配置页上看到的是上次重启时的快照 ——
// 选中一个平台已经调不到的模型,失败要到真发邮件时才暴露。
//
// 与 PlatformSessions 同一约定nil = 本次不上报(保留现有目录),
// 空数组 = 平台确实一个模型都拿不到。拿不到目录时必须省略:
// 清空目录会让配置页变成空白,管理员以为该平台没有任何可用模型。
Models []repo.CatalogModel `json:"models"`
}
// POST /api/v1/agent/register
//
// 两种认证方式:
// 1. Authorization: Bearer <agent_key_token> —— 密钥认证(推荐)。
// 密钥未绑定时用本请求的 name 落定;已绑定时 name 必须与之一致,
// 否则等于拿别人的密钥冒充新身份。
// 2. body 里带 secret —— 旧方式,兼容保留。
func RegisterAgent(w http.ResponseWriter, r *http.Request) {
var req registerRequest
if !DecodeBody(w, r, &req) {
return
}
if req.Name == "" {
Error(w, http.StatusBadRequest, "Missing name")
return
}
keyToken := middleware.BearerToken(r)
if keyToken == "" && req.Secret == "" {
Error(w, http.StatusBadRequest, "需要 Authorization: Bearer <密钥> 或 body 里的 secret")
return
}
if keyToken != "" {
bound, err := repo.VerifyAgentKey(r.Context(), keyToken)
if err != nil {
writeKeyErr(w, err)
return
}
if bound != "" && bound != req.Name {
Error(w, http.StatusForbidden,
"该密钥已绑定到 Agent \""+bound+"\",不能用于注册 \""+req.Name+"\"")
return
}
}
if req.Platform == "" {
req.Platform = "pi"
}
// 三维地址的 name 位与人类用户名共用命名空间,不得重名
if ok, err := repo.AgentNameAvailable(r.Context(), req.Name); err != nil {
Error(w, http.StatusInternalServerError, "Failed to validate agent name")
return
} else if !ok {
Error(w, http.StatusConflict, "该名称已被人类用户占用")
return
}
if req.Name == "human" {
Error(w, http.StatusBadRequest, "human 是保留别名,不能作为 Agent 名")
return
}
// 密钥认证时不需要 secret但 agents.secret 非空约束仍在;
// 存密钥本身作占位,旧的 name/secret 路径不受影响。
secret := req.Secret
if secret == "" {
secret = keyToken
}
if err := repo.CreateOrUpdateAgent(r.Context(), req.Name, secret, req.Platform, req.Workspaces); err != nil {
Error(w, http.StatusInternalServerError, "Failed to register agent")
return
}
// 待绑定密钥在首次注册成功后落定到该 Agent
if keyToken != "" {
if err := repo.ClaimAgentKey(r.Context(), keyToken, req.Name); err != nil {
Error(w, http.StatusInternalServerError, "Failed to bind key")
return
}
}
JSON(w, http.StatusOK, map[string]string{
"status": "registered",
"agent_name": req.Name,
})
}
// POST /api/v1/agent/heartbeat
func HeartbeatAgent(w http.ResponseWriter, r *http.Request) {
agentName := middleware.GetAgentName(r)
if agentName == "" {
Error(w, http.StatusUnauthorized, "Unauthorized")
return
}
pending, err := repo.HeartbeatAgent(r.Context(), agentName)
if err != nil {
Error(w, http.StatusInternalServerError, "Failed to heartbeat")
return
}
// 可选的平台会话快照。解不开就当作没带:心跳的主职责是「我还活着」,
// 不该因为上报体格式不对就把 Agent 判成离线。
var req heartbeatRequest
if r.ContentLength > 0 {
_ = Decode(r, &req)
}
syncedSessions := -1 // -1 = 本次未上报
if req.PlatformSessions != nil {
if err := repo.ReplacePlatformSessions(r.Context(), agentName, req.PlatformSessions); err != nil {
// 镜像写失败只影响候选补全,不影响投递,因此不报错
syncedSessions = -1
} else {
syncedSessions = len(req.PlatformSessions)
}
}
// 模型目录同理:写失败只让配置页看到的目录陈一轮,下一次心跳会补上。
syncedModels := -1
if req.Models != nil {
if err := repo.ReplaceModelCatalog(r.Context(), agentName, req.Models); err == nil {
syncedModels = len(req.Models)
}
}
// 心跳回传该 Agent 的累计统计与新任务默认预算。
//
// 不再回传「剩余额度」:额度属于具体任务(会话)而不属于 Agent
// 剩余往返随每次发信响应budget_remaining回传在那里才有意义。
stats, sErr := repo.GetAgentStats(r.Context(), agentName)
if sErr != nil {
// 统计读不到不影响心跳本身
stats = repo.AgentStats{AgentName: agentName}
}
resp := map[string]interface{}{
"status": "ok",
"pending_mails": pending,
"stats": stats,
}
if syncedSessions >= 0 {
resp["platform_sessions_synced"] = syncedSessions
}
if syncedModels >= 0 {
resp["models_synced"] = syncedModels
}
// 回传当前生效的模型范围,插件无需另起一个请求去读。
//
// 随心跳回传而不是让插件自己轮询:管理员在配置页改了范围后,
// 插件最多一个心跳周期30 秒)就能看到新值,不需要重启。
if allowed, aErr := repo.ListAllowedModels(r.Context(), agentName); aErr == nil {
resp["allowed_models"] = allowed
resp["models_unrestricted"] = len(allowed) == 0
}
JSON(w, http.StatusOK, resp)
}
// GET /api/v1/agents
func ListAgents(w http.ResponseWriter, r *http.Request) {
statusFilter := r.URL.Query().Get("status")
agents, err := repo.ListAgents(r.Context(), statusFilter)
if err != nil {
Error(w, http.StatusInternalServerError, "Failed to list agents")
return
}
JSON(w, http.StatusOK, map[string]interface{}{
"agents": emptySlice(agents),
})
}