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)
- 生产已部署
222 lines
6.0 KiB
Go
222 lines
6.0 KiB
Go
package handler
|
||
|
||
import (
|
||
"net/http"
|
||
|
||
"github.com/agentmail/gateway/internal/middleware"
|
||
"github.com/agentmail/gateway/internal/models"
|
||
"github.com/agentmail/gateway/internal/repo"
|
||
"github.com/agentmail/gateway/internal/sse"
|
||
"github.com/google/uuid"
|
||
)
|
||
|
||
// ---------- Contacts(左侧联系人界面,按登录用户隔离) ----------
|
||
|
||
// GET /api/v1/contacts?archived=false
|
||
func ListContacts(w http.ResponseWriter, r *http.Request) {
|
||
user := middleware.GetUser(r)
|
||
if user == nil {
|
||
Error(w, http.StatusUnauthorized, "not authenticated")
|
||
return
|
||
}
|
||
archived := r.URL.Query().Get("archived") == "true"
|
||
|
||
// 管理员可用 ?all=true 查看全部
|
||
scope := user.Username
|
||
if user.IsAdmin() && r.URL.Query().Get("all") == "true" {
|
||
scope = ""
|
||
}
|
||
|
||
contacts, err := repo.ListContactsFor(r.Context(), scope, archived)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to list contacts")
|
||
return
|
||
}
|
||
JSON(w, http.StatusOK, map[string]interface{}{
|
||
"contacts": emptySlice(contacts),
|
||
})
|
||
}
|
||
|
||
type archiveRequest struct {
|
||
Address string `json:"address"`
|
||
SessionID string `json:"session_id"`
|
||
}
|
||
|
||
// POST /api/v1/contacts/archive
|
||
// 归档指定 name@path.session:Agent 侧会话归档 + 邮箱界面移除
|
||
func ArchiveContact(w http.ResponseWriter, r *http.Request) {
|
||
user := middleware.GetUser(r)
|
||
if user == nil {
|
||
Error(w, http.StatusUnauthorized, "not authenticated")
|
||
return
|
||
}
|
||
|
||
var req archiveRequest
|
||
if !DecodeBody(w, r, &req) {
|
||
return
|
||
}
|
||
|
||
var sessionID uuid.UUID
|
||
switch {
|
||
case req.SessionID != "":
|
||
id, err := uuid.Parse(req.SessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusBadRequest, "Invalid session_id")
|
||
return
|
||
}
|
||
sessionID = id
|
||
|
||
case req.Address != "":
|
||
addr, err := models.ParseAddress(req.Address)
|
||
if err != nil {
|
||
Error(w, http.StatusBadRequest, "Invalid address: "+err.Error())
|
||
return
|
||
}
|
||
id, err := repo.FindSessionByAddress(r.Context(), addr.Name, addr.Path, addr.Session)
|
||
if err != nil {
|
||
Error(w, http.StatusNotFound, "No session matches "+req.Address)
|
||
return
|
||
}
|
||
sessionID = id
|
||
|
||
default:
|
||
Error(w, http.StatusBadRequest, "Provide address or session_id")
|
||
return
|
||
}
|
||
|
||
// 鉴权:只能归档自己参与的会话(管理员不限)
|
||
allowed, err := repo.UserCanAccessSession(r.Context(), user, sessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to check permission")
|
||
return
|
||
}
|
||
if !allowed {
|
||
Error(w, http.StatusForbidden, "无权归档他人的会话")
|
||
return
|
||
}
|
||
|
||
session, err := repo.GetSessionByID(r.Context(), sessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusNotFound, "Session not found")
|
||
return
|
||
}
|
||
mails, _ := repo.GetSessionMails(r.Context(), sessionID)
|
||
|
||
if err := repo.ArchiveSession(r.Context(), sessionID); err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to archive session")
|
||
return
|
||
}
|
||
|
||
alias := ""
|
||
if session.Alias != nil {
|
||
alias = *session.Alias
|
||
}
|
||
payload := map[string]interface{}{
|
||
"session_id": sessionID.String(),
|
||
"session_alias": alias,
|
||
"archived_by": user.Username,
|
||
}
|
||
|
||
// 通知会话内所有参与方(Agent 与人类),各自归档/移除
|
||
notified := map[string]bool{}
|
||
for _, m := range mails {
|
||
names := append([]string{m.FromName, m.ToName}, ccNames(m.CCList)...)
|
||
for _, name := range names {
|
||
if name == "" || notified[name] {
|
||
continue
|
||
}
|
||
notified[name] = true
|
||
sse.Default.SendToRecipient(name, "session_archived", payload)
|
||
}
|
||
}
|
||
if !notified[user.Username] {
|
||
sse.Default.SendToUser(user.Username, "session_archived", payload)
|
||
}
|
||
|
||
JSON(w, http.StatusOK, map[string]interface{}{
|
||
"status": "archived",
|
||
"session_id": sessionID.String(),
|
||
"session_alias": alias,
|
||
})
|
||
}
|
||
|
||
// GET /api/v1/contacts/suggest?name=xxx&path=yyy
|
||
// 三段式补全:无 name 给 Agent+人类用户名;有 name 给工作区;两者都有给会话别名(含 new)
|
||
func SuggestAddress(w http.ResponseWriter, r *http.Request) {
|
||
user := middleware.GetUser(r)
|
||
if user == nil {
|
||
Error(w, http.StatusUnauthorized, "not authenticated")
|
||
return
|
||
}
|
||
|
||
name := r.URL.Query().Get("name")
|
||
path := r.URL.Query().Get("path")
|
||
|
||
if name == "" {
|
||
agents, err := repo.ListAgents(r.Context(), "")
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to list agents")
|
||
return
|
||
}
|
||
users, _ := repo.ListActiveUsernames(r.Context())
|
||
|
||
names := make([]string, 0, len(agents)+len(users))
|
||
for _, a := range agents {
|
||
names = append(names, a.Name)
|
||
}
|
||
for _, u := range users {
|
||
if u == user.Username {
|
||
continue // 不建议给自己发信
|
||
}
|
||
names = append(names, u)
|
||
}
|
||
JSON(w, http.StatusOK, map[string]interface{}{
|
||
"kind": "name",
|
||
"suggestions": emptySlice(names),
|
||
})
|
||
return
|
||
}
|
||
|
||
if path == "" {
|
||
// 人类用户没有工作区,直接给空列表(前端会继续走 session 段)
|
||
paths, _ := repo.SuggestPaths(r.Context(), name)
|
||
JSON(w, http.StatusOK, map[string]interface{}{
|
||
"kind": "path",
|
||
"suggestions": emptySlice(paths),
|
||
})
|
||
return
|
||
}
|
||
|
||
sessions, err := repo.SuggestSessionCandidates(r.Context(), user.Username, name, path)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to suggest sessions")
|
||
return
|
||
}
|
||
// suggestions 保留纯字符串形式:已部署的前端与第三方客户端只认这一个字段。
|
||
// 带标题与来源的完整形式另放 candidates,两个字段同序。
|
||
aliases := make([]string, 0, len(sessions)+1)
|
||
for _, c := range sessions {
|
||
aliases = append(aliases, c.Alias)
|
||
}
|
||
// new 总是可选且永远在最后:它不是一条已存在的会话,
|
||
// 排在前面会让人在想续谈时随手回车开出一条新线索。
|
||
aliases = append(aliases, "new")
|
||
sessions = append(sessions, repo.SessionCandidate{Alias: "new", Source: "new", Title: "新建会话"})
|
||
|
||
JSON(w, http.StatusOK, map[string]interface{}{
|
||
"kind": "session",
|
||
"suggestions": aliases,
|
||
"candidates": sessions,
|
||
})
|
||
}
|
||
|
||
func ccNames(list []models.Address) []string {
|
||
out := make([]string, 0, len(list))
|
||
for _, a := range list {
|
||
if a.Name != "" {
|
||
out = append(out, a.Name)
|
||
}
|
||
}
|
||
return out
|
||
}
|