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)
- 生产已部署
131 lines
4.4 KiB
Go
131 lines
4.4 KiB
Go
package handler
|
||
|
||
import (
|
||
"net/http"
|
||
"strings"
|
||
|
||
"github.com/agentmail/gateway/internal/middleware"
|
||
"github.com/agentmail/gateway/internal/repo"
|
||
"github.com/go-chi/chi/v5"
|
||
)
|
||
|
||
// ---------- 邮件场景下的可用模型 ----------
|
||
//
|
||
// GET /agent/models/allowed 读取被允许的模型(Agent 凭证)
|
||
// GET /admin/agents/{name}/models 管理员读目录 + 已选
|
||
// PUT /admin/agents/{name}/models 管理员保存选择与优先级
|
||
//
|
||
// **目录上报走心跳**(见 agents.go 的 heartbeatRequest.Models),不另设端点:
|
||
// 模型清单会在运行中变(换 provider 配置、上游上下线、换 API key),
|
||
// 心跳本来就是 30 秒一次的现成通道。另设一个 POST 等于给「目录是谁写的」
|
||
// 这个问题留两个答案,排查时要同时看两处。
|
||
//
|
||
// 生效的模型范围同样随心跳响应回传(allowed_models),因此插件通常不需要调
|
||
// 下面这个 GET —— 它是给非插件的第三方客户端(没有心跳循环)与排查用的。
|
||
|
||
// GET /api/v1/agent/models/allowed —— 插件读取被允许的模型
|
||
//
|
||
// 返回按优先级排序的列表。空列表表示**不限定**,插件应回退到平台自己的默认模型
|
||
// —— 与「一个都不许用」不同,后者等于让 Agent 彻底哑掉,不该是一次误配的后果。
|
||
func GetAllowedModels(w http.ResponseWriter, r *http.Request) {
|
||
agentName := middleware.GetAgentName(r)
|
||
if agentName == "" {
|
||
Error(w, http.StatusUnauthorized, "Unauthorized")
|
||
return
|
||
}
|
||
models, err := repo.ListAllowedModels(r.Context(), agentName)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to list allowed models")
|
||
return
|
||
}
|
||
JSON(w, http.StatusOK, map[string]any{
|
||
"models": models,
|
||
// unrestricted 明确表达「没配 = 不限」,省得插件自己去判断空数组的含义
|
||
"unrestricted": len(models) == 0,
|
||
})
|
||
}
|
||
|
||
// GET /api/v1/admin/agents/{name}/models —— 管理员读目录(带已选标记)
|
||
func AdminListAgentModels(w http.ResponseWriter, r *http.Request) {
|
||
name := strings.TrimSpace(chi.URLParam(r, "name"))
|
||
if name == "" {
|
||
Error(w, http.StatusBadRequest, "Missing agent name")
|
||
return
|
||
}
|
||
catalog, err := repo.ListModelCatalog(r.Context(), name)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to list model catalog")
|
||
return
|
||
}
|
||
// 已选但已不在目录里的模型要单独给出来:平台可能临时下线了某个模型,
|
||
// 界面上不显示的话管理员会以为自己没选过它,而它其实还在被插件尝试。
|
||
stale, err := repo.ListStaleAllowedModels(r.Context(), name)
|
||
if err != nil {
|
||
stale = []repo.ModelRef{}
|
||
}
|
||
JSON(w, http.StatusOK, map[string]any{
|
||
"agent_name": name,
|
||
"catalog": catalog,
|
||
"stale": stale,
|
||
})
|
||
}
|
||
|
||
// PUT /api/v1/admin/agents/{name}/models —— 管理员保存选择
|
||
//
|
||
// 入参顺序即优先级(rank)。插件按这个顺序逐个尝试,全部失败才回一封失败邮件。
|
||
func AdminSetAgentModels(w http.ResponseWriter, r *http.Request) {
|
||
name := strings.TrimSpace(chi.URLParam(r, "name"))
|
||
if name == "" {
|
||
Error(w, http.StatusBadRequest, "Missing agent name")
|
||
return
|
||
}
|
||
|
||
var req struct {
|
||
Models []repo.ModelRef `json:"models"`
|
||
}
|
||
if !DecodeBody(w, r, &req) {
|
||
return
|
||
}
|
||
if len(req.Models) > maxAllowedModels {
|
||
Error(w, http.StatusBadRequest,
|
||
"选定的模型过多(上限 "+itoa(maxAllowedModels)+" 个)")
|
||
return
|
||
}
|
||
|
||
if err := repo.SetAllowedModels(r.Context(), name, req.Models); err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to save allowed models")
|
||
return
|
||
}
|
||
// 回传保存后的实际结果而不是回显入参:repo 层会跳过重复项与空字段,
|
||
// 回显入参会让前端以为那些也存下来了。
|
||
saved, err := repo.ListAllowedModels(r.Context(), name)
|
||
if err != nil {
|
||
saved = []repo.ModelRef{}
|
||
}
|
||
JSON(w, http.StatusOK, map[string]any{
|
||
"status": "saved",
|
||
"models": saved,
|
||
})
|
||
}
|
||
|
||
// maxAllowedModels 限制管理员能选多少个模型。
|
||
//
|
||
// 降级尝试是串行的:选 50 个意味着最坏情况下一封邮件要等 50 次模型调用超时。
|
||
// 十个已经足够表达「主力 + 几个备选」。
|
||
const maxAllowedModels = 10
|
||
|
||
// itoa 避免为一个数字引入 strconv 导入(本文件只此一处用到)。
|
||
func itoa(n int) string {
|
||
if n == 0 {
|
||
return "0"
|
||
}
|
||
var b [20]byte
|
||
i := len(b)
|
||
for n > 0 {
|
||
i--
|
||
b[i] = byte('0' + n%10)
|
||
n /= 10
|
||
}
|
||
return string(b[i:])
|
||
}
|