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)
- 生产已部署
351 lines
11 KiB
Go
351 lines
11 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"
|
||
"github.com/agentmail/gateway/internal/sse"
|
||
"github.com/google/uuid"
|
||
)
|
||
|
||
// ---------- Session(均需登录,且做会话级鉴权) ----------
|
||
|
||
// requireSessionAccess 解析路径中的会话 ID 并校验当前用户有权访问
|
||
func requireSessionAccess(w http.ResponseWriter, r *http.Request) (uuid.UUID, bool) {
|
||
user := middleware.GetUser(r)
|
||
if user == nil {
|
||
Error(w, http.StatusUnauthorized, "not authenticated")
|
||
return uuid.Nil, false
|
||
}
|
||
sessionID, ok := pathUUID(w, r, "id")
|
||
if !ok {
|
||
return uuid.Nil, false
|
||
}
|
||
allowed, err := repo.UserCanAccessSession(r.Context(), user, sessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to check permission")
|
||
return uuid.Nil, false
|
||
}
|
||
if !allowed {
|
||
Error(w, http.StatusForbidden, "无权访问该会话")
|
||
return uuid.Nil, false
|
||
}
|
||
return sessionID, true
|
||
}
|
||
|
||
// GET /api/v1/sessions/{id}
|
||
func GetSession(w http.ResponseWriter, r *http.Request) {
|
||
sessionID, ok := requireSessionAccess(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
session, err := repo.GetSessionByID(r.Context(), sessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusNotFound, "Session not found")
|
||
return
|
||
}
|
||
mails, err := repo.GetSessionMails(r.Context(), sessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to get session mails")
|
||
return
|
||
}
|
||
// 会话线程要展示附件,逐封填充。
|
||
//
|
||
// 这里漏掉过:前端的会话视图走的是本端点(GET /sessions/{id}),
|
||
// 而不是下面那个 /sessions/{id}/mails —— 后者填了附件但没人调用,
|
||
// 于是 Agent 回信里的附件在 UI 上完全不存在。
|
||
ptrs := make([]*models.Mail, len(mails))
|
||
for i := range mails {
|
||
ptrs[i] = &mails[i]
|
||
}
|
||
fillAttachments(r, ptrs...)
|
||
|
||
JSON(w, http.StatusOK, map[string]interface{}{
|
||
"session": session,
|
||
"mails": emptySlice(mails),
|
||
})
|
||
}
|
||
|
||
// GET /api/v1/sessions/{id}/mails
|
||
func GetSessionMails(w http.ResponseWriter, r *http.Request) {
|
||
sessionID, ok := requireSessionAccess(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
mails, err := repo.GetSessionMails(r.Context(), sessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to get mails")
|
||
return
|
||
}
|
||
// 会话线程要展示附件,逐封填充
|
||
ptrs := make([]*models.Mail, len(mails))
|
||
for i := range mails {
|
||
ptrs[i] = &mails[i]
|
||
}
|
||
fillAttachments(r, ptrs...)
|
||
JSON(w, http.StatusOK, map[string]interface{}{
|
||
"mails": emptySlice(mails),
|
||
})
|
||
}
|
||
|
||
type updateAliasRequest struct {
|
||
Alias string `json:"alias"`
|
||
}
|
||
|
||
// PUT /api/v1/sessions/{id}/alias
|
||
//
|
||
// 会话别名负责三维寻址(name@path.<alias>),因此必须全局唯一,
|
||
// 且不能叫 "new"(那是寻址保留字)。
|
||
func UpdateSessionAlias(w http.ResponseWriter, r *http.Request) {
|
||
sessionID, ok := requireSessionAccess(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
var req updateAliasRequest
|
||
if !DecodeBody(w, r, &req) {
|
||
return
|
||
}
|
||
alias := strings.TrimSpace(req.Alias)
|
||
if alias == "" {
|
||
Error(w, http.StatusBadRequest, "Missing alias")
|
||
return
|
||
}
|
||
if err := validateSessionAlias(alias); err != nil {
|
||
writeErr(w, err, "Invalid alias")
|
||
return
|
||
}
|
||
|
||
// 被其他会话占用时报 409,而不是默默造出两个同名可寻址会话
|
||
if s, err := repo.FindSessionByAlias(r.Context(), alias); err == nil && s.ID != sessionID {
|
||
Error(w, http.StatusConflict, "会话别名 \""+alias+"\" 已被其他会话占用")
|
||
return
|
||
}
|
||
|
||
if err := repo.UpdateSessionAlias(r.Context(), sessionID, alias); err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to update alias")
|
||
return
|
||
}
|
||
JSON(w, http.StatusOK, map[string]string{
|
||
"status": "updated",
|
||
"alias": alias,
|
||
})
|
||
}
|
||
|
||
// syncSessionRequest 是 Agent 平台回传自己那侧的会话标识。
|
||
//
|
||
// 各 Agent 平台(opencode / Claude Code / DSH…)都会由模型为会话生成一个摘要标题,
|
||
// 并配一个短 slug。不在本侧另造一套命名:平台那边叫什么,本侧就叫什么。
|
||
type syncSessionRequest struct {
|
||
// Alias 是平台侧的短标识(如 opencode 的 slug "jolly-cactus"),写入本侧 session_alias 供寻址。
|
||
Alias string `json:"alias"`
|
||
// Title 是平台侧模型生成的摘要标题(如「修复登录态丢失」),写入本侧 subject 供展示。
|
||
Title string `json:"title"`
|
||
}
|
||
|
||
// POST /api/v1/sessions/{id}/sync
|
||
//
|
||
// Agent 侧端点:把平台生成的会话标题与 slug 同步到本侧。
|
||
// alias 撞名时自动追加 -2/-3 后缀(本侧别名负责寻址必须唯一,而平台 slug 不保证全局唯一),
|
||
// 因此本接口不会因撞名失败,响应里回传最终落库的别名。
|
||
func SyncSession(w http.ResponseWriter, r *http.Request) {
|
||
agentName := middleware.GetAgentName(r)
|
||
if agentName == "" {
|
||
Error(w, http.StatusUnauthorized, "Unauthorized")
|
||
return
|
||
}
|
||
sessionID, ok := pathUUID(w, r, "id")
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
// Agent 只能同步自己参与过的会话
|
||
allowed, err := repo.AgentCanAccessSession(r.Context(), agentName, sessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to check permission")
|
||
return
|
||
}
|
||
if !allowed {
|
||
Error(w, http.StatusForbidden, "无权修改未参与的会话")
|
||
return
|
||
}
|
||
|
||
var req syncSessionRequest
|
||
if !DecodeBody(w, r, &req) {
|
||
return
|
||
}
|
||
|
||
resp := map[string]string{"status": "synced"}
|
||
|
||
if title := strings.TrimSpace(req.Title); title != "" {
|
||
if err := repo.SyncSessionTitle(r.Context(), sessionID, title); err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to sync title")
|
||
return
|
||
}
|
||
resp["title"] = title
|
||
}
|
||
|
||
if alias := strings.TrimSpace(req.Alias); alias != "" {
|
||
// 平台 slug 可能带非法字符,落库前按本侧寻址规则规范化
|
||
norm := normalizeAlias(alias)
|
||
if norm == "" {
|
||
Error(w, http.StatusBadRequest, "alias 规范化后为空,无法作为寻址别名")
|
||
return
|
||
}
|
||
final, err := repo.SyncSessionAlias(r.Context(), sessionID, norm)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to sync alias")
|
||
return
|
||
}
|
||
resp["alias"] = final
|
||
}
|
||
|
||
// 让参与方前端立即看到新标题/别名
|
||
sse.Default.Broadcast("session_update", map[string]interface{}{
|
||
"session_id": sessionID.String(),
|
||
"alias": resp["alias"],
|
||
"title": resp["title"],
|
||
})
|
||
|
||
JSON(w, http.StatusOK, resp)
|
||
}
|
||
|
||
// GET /api/v1/sessions/{id}/rename-proposal
|
||
//
|
||
// 返回该会话里最新一条尚未处理的改名提议(Agent 在正文里提的)。
|
||
// 「尚未处理」= 既不是当前别名(已接受),也不在驳回记录里。
|
||
// 无提议时返回 {"proposal": null},前端据此决定要不要显示提示条。
|
||
func GetRenameProposal(w http.ResponseWriter, r *http.Request) {
|
||
sessionID, ok := requireSessionAccess(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
alias, reason, err := repo.PendingRenameProposal(r.Context(), sessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to load proposal")
|
||
return
|
||
}
|
||
if alias == "" {
|
||
JSON(w, http.StatusOK, map[string]interface{}{"proposal": nil})
|
||
return
|
||
}
|
||
JSON(w, http.StatusOK, map[string]interface{}{
|
||
"proposal": map[string]string{"alias": alias, "reason": reason},
|
||
})
|
||
}
|
||
|
||
// POST /api/v1/sessions/{id}/rename-proposal/dismiss
|
||
//
|
||
// 用户驳回当前提议。记下被驳回的别名,好让提示条不再反复弹同一个建议 ——
|
||
// 否则每次打开会话都要重新点一次「忽略」。
|
||
//
|
||
// 接受提议走已有的 PUT /sessions/{id}/alias,不另开端点:
|
||
// 那条路径已经有唯一性校验与 409 处理,复制一遍只会多一个出错的地方。
|
||
func DismissRenameProposal(w http.ResponseWriter, r *http.Request) {
|
||
sessionID, ok := requireSessionAccess(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
alias, _, err := repo.PendingRenameProposal(r.Context(), sessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to load proposal")
|
||
return
|
||
}
|
||
if alias == "" {
|
||
// 已经没有待处理提议(可能是另一个标签页刚处理过),当作成功
|
||
JSON(w, http.StatusOK, map[string]string{"status": "no_pending"})
|
||
return
|
||
}
|
||
if err := repo.DismissRenameProposal(r.Context(), sessionID, alias); err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to dismiss proposal")
|
||
return
|
||
}
|
||
JSON(w, http.StatusOK, map[string]string{
|
||
"status": "dismissed",
|
||
"dismissed": alias,
|
||
})
|
||
}
|
||
|
||
type sessionBudgetRequest struct {
|
||
// MaxRounds 是本会话的往返预算上限(0 = 不限)。
|
||
MaxRounds *int `json:"max_rounds"`
|
||
// Reset 把已用次数归零(上限不变)。可与 MaxRounds 同时给:
|
||
// 「加到 20 并从头算」是一次很自然的操作,拆成两个请求只会让前端多一次往返。
|
||
Reset bool `json:"reset"`
|
||
}
|
||
|
||
// GET /api/v1/sessions/{id}/budget
|
||
//
|
||
// 本会话的往返预算。与 Agent 全局配额是两层,都要过:
|
||
// 会话预算管「这件事值得多少个来回」,全局配额管「这个 Agent 总共能发多少」。
|
||
func GetSessionBudgetHandler(w http.ResponseWriter, r *http.Request) {
|
||
sessionID, ok := requireSessionAccess(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
b, err := repo.GetSessionBudget(r.Context(), sessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusNotFound, "Session not found")
|
||
return
|
||
}
|
||
JSON(w, http.StatusOK, b)
|
||
}
|
||
|
||
// PUT /api/v1/sessions/{id}/budget
|
||
//
|
||
// 在对话页里随时调本任务的预算 —— 这是配额最该被编辑的地方:
|
||
// 人看着往来内容才知道这件事还值不值得再来几个回合。
|
||
func UpdateSessionBudget(w http.ResponseWriter, r *http.Request) {
|
||
sessionID, ok := requireSessionAccess(w, r)
|
||
if !ok {
|
||
return
|
||
}
|
||
var req sessionBudgetRequest
|
||
if !DecodeBody(w, r, &req) {
|
||
return
|
||
}
|
||
if req.MaxRounds == nil && !req.Reset {
|
||
Error(w, http.StatusBadRequest, "需要给出 max_rounds 或 reset")
|
||
return
|
||
}
|
||
|
||
b, err := repo.GetSessionBudget(r.Context(), sessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusNotFound, "Session not found")
|
||
return
|
||
}
|
||
if req.MaxRounds != nil {
|
||
if *req.MaxRounds < 0 {
|
||
Error(w, http.StatusBadRequest, "max_rounds 不能为负")
|
||
return
|
||
}
|
||
// 允许调到低于已用次数:那表示「就到这里为止」,是人的合法意图。
|
||
// 此时剩余为 0,Agent 下次发信即被拦。
|
||
b, err = repo.SetSessionBudget(r.Context(), sessionID, *req.MaxRounds)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to set budget")
|
||
return
|
||
}
|
||
}
|
||
if req.Reset {
|
||
b, err = repo.ResetSessionBudget(r.Context(), sessionID)
|
||
if err != nil {
|
||
Error(w, http.StatusInternalServerError, "Failed to reset budget")
|
||
return
|
||
}
|
||
}
|
||
|
||
// 让会话的其他参与方(含 Agent 侧界面)立刻看到新预算
|
||
sse.Default.Broadcast("session_update", map[string]interface{}{
|
||
"session_id": sessionID.String(),
|
||
"budget_max": b.Max,
|
||
"budget_used": b.Used,
|
||
"budget_remaining": b.Remaining,
|
||
})
|
||
JSON(w, http.StatusOK, b)
|
||
}
|