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)
- 生产已部署
177 lines
4.8 KiB
Go
177 lines
4.8 KiB
Go
package handler
|
||
|
||
import (
|
||
"net/http"
|
||
"strings"
|
||
|
||
"github.com/agentmail/gateway/internal/middleware"
|
||
"github.com/agentmail/gateway/internal/models"
|
||
"github.com/agentmail/gateway/internal/repo"
|
||
)
|
||
|
||
// ---------- 密钥管理 ----------
|
||
//
|
||
// 两套接口,权限边界不同:
|
||
// /admin/agent-keys —— 管理员签发 Agent 接入密钥
|
||
// /me/keys —— 用户自助签发客户端连接密钥(不能注册 Agent)
|
||
//
|
||
// 密钥全文只在创建响应里出现一次,列表接口只给前 8 位 hint。
|
||
|
||
type createKeyRequest struct {
|
||
// AgentName 仅 Agent 密钥使用;留空表示「待绑定」,首次注册时按注册请求的 name 落定
|
||
AgentName string `json:"agent_name"`
|
||
// Label 人类可读备注(如「我的笔记本」「CI 机器」)
|
||
Label string `json:"label"`
|
||
// KeyType permanent / one_time / timed
|
||
KeyType string `json:"key_type"`
|
||
// ExpiresHours 仅 timed 使用,必须为正
|
||
ExpiresHours int `json:"expires_hours"`
|
||
// KeyToken 仅 Agent 密钥使用:登记一把客户端已在本地生成的密钥。
|
||
// 插件首次安装时自己生成密钥并打印出来,管理员把它填到这里完成登记,
|
||
// 密钥全文因此不需要从服务器往客户端传。留空则由服务器生成。
|
||
KeyToken string `json:"key_token"`
|
||
}
|
||
|
||
// normalizeKeyType 默认给 permanent,避免调用方漏填时落到非法值
|
||
func normalizeKeyType(t string) string {
|
||
t = strings.TrimSpace(t)
|
||
if t == "" {
|
||
return models.KeyPermanent
|
||
}
|
||
return t
|
||
}
|
||
|
||
// POST /api/v1/admin/agent-keys
|
||
func CreateAgentKey(w http.ResponseWriter, r *http.Request) {
|
||
admin := middleware.GetUser(r)
|
||
if admin == nil {
|
||
Error(w, http.StatusUnauthorized, "not authenticated")
|
||
return
|
||
}
|
||
|
||
var req createKeyRequest
|
||
if !DecodeBody(w, r, &req) {
|
||
return
|
||
}
|
||
|
||
key, err := repo.CreateAgentKey(r.Context(),
|
||
strings.TrimSpace(req.AgentName), normalizeKeyType(req.KeyType),
|
||
strings.TrimSpace(req.Label), req.ExpiresHours, admin.ID,
|
||
strings.TrimSpace(req.KeyToken))
|
||
if err != nil {
|
||
writeKeyErr(w, err)
|
||
return
|
||
}
|
||
|
||
// 唯一一次回传全文
|
||
JSON(w, http.StatusOK, map[string]any{"key": key})
|
||
}
|
||
|
||
// GET /api/v1/admin/agent-keys?agent_name=xxx
|
||
func ListAgentKeys(w http.ResponseWriter, r *http.Request) {
|
||
keys, err := repo.ListAgentKeys(r.Context(), r.URL.Query().Get("agent_name"))
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to list keys")
|
||
return
|
||
}
|
||
JSON(w, http.StatusOK, map[string]any{"keys": keys})
|
||
}
|
||
|
||
// DELETE /api/v1/admin/agent-keys/{id}
|
||
func DeleteAgentKey(w http.ResponseWriter, r *http.Request) {
|
||
id, ok := pathUUID(w, r, "id")
|
||
if !ok {
|
||
return
|
||
}
|
||
if err := repo.DeleteAgentKey(r.Context(), id); err != nil {
|
||
writeKeyErr(w, err)
|
||
return
|
||
}
|
||
JSON(w, http.StatusOK, map[string]string{"status": "deleted"})
|
||
}
|
||
|
||
type bindKeyRequest struct {
|
||
AgentName string `json:"agent_name"`
|
||
}
|
||
|
||
// POST /api/v1/admin/agent-keys/{id}/bind
|
||
func BindAgentKey(w http.ResponseWriter, r *http.Request) {
|
||
id, ok := pathUUID(w, r, "id")
|
||
if !ok {
|
||
return
|
||
}
|
||
var req bindKeyRequest
|
||
if !DecodeBody(w, r, &req) {
|
||
return
|
||
}
|
||
name := strings.TrimSpace(req.AgentName)
|
||
if name == "" {
|
||
Error(w, http.StatusBadRequest, "Missing agent_name")
|
||
return
|
||
}
|
||
if err := repo.BindAgentKey(r.Context(), id, name); err != nil {
|
||
writeKeyErr(w, err)
|
||
return
|
||
}
|
||
JSON(w, http.StatusOK, map[string]string{"status": "bound", "agent_name": name})
|
||
}
|
||
|
||
// ---------- 用户连接密钥 ----------
|
||
|
||
// POST /api/v1/me/keys
|
||
func CreateMyKey(w http.ResponseWriter, r *http.Request) {
|
||
user := middleware.GetUser(r)
|
||
if user == nil {
|
||
Error(w, http.StatusUnauthorized, "not authenticated")
|
||
return
|
||
}
|
||
|
||
var req createKeyRequest
|
||
if !DecodeBody(w, r, &req) {
|
||
return
|
||
}
|
||
|
||
key, err := repo.CreateUserKey(r.Context(), user.ID,
|
||
strings.TrimSpace(req.Label), normalizeKeyType(req.KeyType), req.ExpiresHours)
|
||
if err != nil {
|
||
writeKeyErr(w, err)
|
||
return
|
||
}
|
||
JSON(w, http.StatusOK, map[string]any{"key": key})
|
||
}
|
||
|
||
// GET /api/v1/me/keys
|
||
func ListMyKeys(w http.ResponseWriter, r *http.Request) {
|
||
user := middleware.GetUser(r)
|
||
if user == nil {
|
||
Error(w, http.StatusUnauthorized, "not authenticated")
|
||
return
|
||
}
|
||
keys, err := repo.ListUserKeys(r.Context(), user.ID)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to list keys")
|
||
return
|
||
}
|
||
JSON(w, http.StatusOK, map[string]any{"keys": keys})
|
||
}
|
||
|
||
// DELETE /api/v1/me/keys/{id}
|
||
func DeleteMyKey(w http.ResponseWriter, r *http.Request) {
|
||
user := middleware.GetUser(r)
|
||
if user == nil {
|
||
Error(w, http.StatusUnauthorized, "not authenticated")
|
||
return
|
||
}
|
||
id, ok := pathUUID(w, r, "id")
|
||
if !ok {
|
||
return
|
||
}
|
||
// repo 层带 user_id 条件,删不到就是不属于自己或不存在,统一 404
|
||
if err := repo.DeleteUserKey(r.Context(), user.ID, id); err != nil {
|
||
writeKeyErr(w, err)
|
||
return
|
||
}
|
||
JSON(w, http.StatusOK, map[string]string{"status": "deleted"})
|
||
}
|
||
|