Files
MailUI4Agents/gateway/internal/handler/helpers.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

216 lines
7.1 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 (
"encoding/json"
"errors"
"io"
"net/http"
"reflect"
"strconv"
"strings"
"unicode/utf8"
"github.com/agentmail/gateway/internal/repo"
)
// JSON 写入 JSON 响应
func JSON(w http.ResponseWriter, status int, data interface{}) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
json.NewEncoder(w).Encode(data)
}
// Error 写入错误响应
func Error(w http.ResponseWriter, status int, msg string) {
JSON(w, status, map[string]string{"error": msg})
}
// Decode 从请求体解析 JSON
func Decode(r *http.Request, v interface{}) error {
return json.NewDecoder(r.Body).Decode(v)
}
// DecodeBody 解析请求体,失败时直接写 400 并返回 false。
//
// 与直接用 Decode 的区别是错误信息**指向具体字段**。原先 22 处调用点
// 一律回一句固定的 "Invalid JSON",客户端只知道「有问题」却不知道哪里有问题 ——
// 实测踩过一次:`workspaces` 要的是 `[{name, path}]`,传字符串数组得到的
// 就是那句固定文案,只能靠翻服务端结构体才发现。第三方客户端没有这个条件。
func DecodeBody(w http.ResponseWriter, r *http.Request, v interface{}) bool {
if err := Decode(r, v); err != nil {
Error(w, http.StatusBadRequest, decodeErrMsg(err))
return false
}
return true
}
// decodeErrMsg 把 json 解码错误翻成一句能照着改的话。
//
// 刻意不回显 json 包的原文:它带 Go 的类型名(如 models.Workspace
// 那是本侧的实现细节,对调用方没有意义,也不该出现在公开 API 的响应里。
func decodeErrMsg(err error) string {
if errors.Is(err, io.EOF) {
return "请求体为空"
}
// 截断的 JSON 走的不是 SyntaxError 而是 ErrUnexpectedEOF ——
// 不单独处理的话会落到最后那句笼统的兜底文案里
if errors.Is(err, io.ErrUnexpectedEOF) {
return "JSON 语法错误:请求体在解析完成前就结束了(可能被截断)"
}
var typeErr *json.UnmarshalTypeError
if errors.As(err, &typeErr) {
if typeErr.Field != "" {
return "字段 \"" + typeErr.Field + "\" 类型不对:期望 " +
jsonKindName(typeErr.Type) + ",收到 " + typeErr.Value
}
return "请求体类型不对:期望 " + jsonKindName(typeErr.Type) + ",收到 " + typeErr.Value
}
var syntaxErr *json.SyntaxError
if errors.As(err, &syntaxErr) {
return "JSON 语法错误(第 " + strconv.FormatInt(syntaxErr.Offset, 10) + " 字节处)"
}
return "请求体不是合法 JSON"
}
// jsonKindName 把 Go 类型说成 JSON 的说法。
// 调用方写的是 JSON用 []models.Workspace 去解释它要的是什么毫无帮助。
func jsonKindName(t reflect.Type) string {
if t == nil {
return "未知类型"
}
switch t.Kind() {
case reflect.Slice, reflect.Array:
return jsonKindName(t.Elem()) + " 数组"
case reflect.Map, reflect.Struct:
return "object"
case reflect.String:
return "string"
case reflect.Bool:
return "boolean"
case reflect.Ptr:
return jsonKindName(t.Elem())
default:
if k := t.Kind(); k >= reflect.Int && k <= reflect.Float64 {
return "number"
}
return t.Kind().String()
}
}
// httpError 携带 HTTP 状态码的错误
type httpError struct {
status int
msg string
}
func (e httpError) Error() string { return e.msg }
func errBadRequest(msg string) error { return httpError{http.StatusBadRequest, msg} }
func errNotFound(msg string) error { return httpError{http.StatusNotFound, msg} }
func errConflict(msg string) error { return httpError{http.StatusConflict, msg} }
// errRateLimited 用于新建会话速率限制。用 429 而不是 403
// 前者表示「稍后再来」,后者表示「你没这个权限」——语义完全不同,
// 客户端据此决定是重试还是放弃。
func errRateLimited(msg string) error { return httpError{http.StatusTooManyRequests, msg} }
// writeKeyErr 把 repo 层的密钥错误映射成 HTTP 响应。
// 「已使用 / 已过期」与「无效」分开报,便于运维判断是重签还是查配置。
func writeKeyErr(w http.ResponseWriter, err error) {
switch {
case errors.Is(err, repo.ErrKeyUsed):
Error(w, http.StatusUnauthorized, "密钥已使用(一次性密钥只能用一次)")
case errors.Is(err, repo.ErrKeyExpired):
Error(w, http.StatusUnauthorized, "密钥已过期")
case errors.Is(err, repo.ErrKeyNotFound):
Error(w, http.StatusUnauthorized, "密钥无效")
case errors.Is(err, repo.ErrKeyTypeInvalid):
Error(w, http.StatusBadRequest, "密钥类型非法,应为 permanent / one_time / timed")
case errors.Is(err, repo.ErrKeyNeedsExpiry):
Error(w, http.StatusBadRequest, "timed 密钥必须给出正的 expires_hours")
case errors.Is(err, repo.ErrKeyTooShort):
Error(w, http.StatusBadRequest, "密钥太短(至少 32 位)")
case errors.Is(err, repo.ErrKeyTokenTaken):
Error(w, http.StatusConflict, "该密钥已登记过")
default:
Error(w, http.StatusInternalServerError, "密钥操作失败")
}
}
// validateSessionAlias 校验会话别名是否可安全出现在三维地址 name@path.<alias> 的末段。
// "new" 是寻址保留字;含 . 会让 path/session 切分歧义;含 @ 与空白同理。
func validateSessionAlias(alias string) error {
if alias == "new" {
return errBadRequest(`会话别名不可为 "new":该词已作为寻址保留字`)
}
if strings.ContainsAny(alias, ". \t/@") {
return errBadRequest("会话别名不可含 . 空白 / 或 @(会与三维地址解析冲突)")
}
return nil
}
// normalizeAlias 把 Agent 平台侧的 slug/标题改写为合法的寻址别名。
//
// 平台侧命名不一定遵守本侧的寻址约束(可能含 . / @ 空白),直接入库会让
// name@path.session 切分歧义,因此非法字符统一换成 -,并压缩连续的 -。
// 保留字 "new" 加前缀避开;全部不可用时返回空串交由调用方报错。
func normalizeAlias(s string) string {
var b strings.Builder
lastDash := false
for _, r := range s {
switch {
case r == '.' || r == '/' || r == '@' || r == ' ' || r == '\t' || r == '\n' || r == '\r':
if !lastDash && b.Len() > 0 {
b.WriteByte('-')
lastDash = true
}
default:
b.WriteRune(r)
lastDash = false
}
}
out := strings.Trim(b.String(), "-")
if out == "new" {
return "session-new"
}
// VARCHAR(128) 上限,按字节截断时不能切坏多字节字符
const maxBytes = 128
if len(out) > maxBytes {
cut := out[:maxBytes]
for len(cut) > 0 && !utf8.ValidString(cut) {
cut = cut[:len(cut)-1]
}
out = strings.Trim(cut, "-")
}
return out
}
// writeErr 将 httpError 按其状态码写出,其余错误统一 500 + fallback 文案
func writeErr(w http.ResponseWriter, err error, fallback string) {
if he, ok := err.(httpError); ok {
Error(w, he.status, he.msg)
return
}
Error(w, http.StatusInternalServerError, fallback)
}
// emptySlice 把 nil slice 转为空 JSON 数组 []
func emptySlice[T any](s []T) []T {
if s == nil {
return []T{}
}
return s
}
// agentLimiterKey 把「这是不是 Agent 发起的」翻译成速率限制的键。
// 人类返回空串 = 不限速(手工操作的频率天然受限)。
func agentLimiterKey(isAgent bool, actor string) string {
if isAgent {
return actor
}
return ""
}