feat: 配额下沉到会话 + 窄屏覆盖式布局 + 工作列表卡片视图
## 配额重构：废除 Agent 终身额度
原实现在 agents 上放一个 max_rounds/used_rounds 计数器，used_rounds 单调递增、
永不重置 —— 跑满就要管理员手工重置才能再干活。那是把一次性资源模型套在长期
在线的服务上，且并行任务互相抢额度。
改为：
- 唯一被强制的预算是【会话】的往返预算（sessions.max_rounds/used_rounds），
写信时给、对话页里随时改 —— 配额的语义是「这件事值得多少个来回」，
那是任务的属性而不是 Agent 的属性
- agents.default_rounds 只作为「派给这个 Agent 的新任务」的默认值（默认 20）
- agents.used_rounds 降级为纯统计
- 新建会话速率限制（1h/20 条）堵住用 .new 开一串新会话绕过预算；
人类不受限（agentLimiterKey 返回空串即不计量）
## 窄屏适配（用户反馈「窄屏基本不可用」）
原先只有三栏并排：60(导航)+320(列表)+详情，375px 屏上详情被挤到 0。
第一版做成「一次只显示一栏」，用户纠正应当是新页面覆盖老页面并带动画，
于是重做为覆盖式：
- NarrowStack：底层列表始终挂载，详情绝对定位盖在上面。两个好处 ——
列表滚动位置与选中态天然保留；退出动画有东西可播（直接卸载再渲染另一个
组件的话，没有任何一帧能让旧页面往右滑出去）
- 因此必须区分「逻辑上是否打开」与「是否还在 DOM 里」：关闭时先播 200ms
滑出，动画结束才卸载
- 入场用双层 requestAnimationFrame：必须让浏览器至少绘制一帧「在右侧之外」
的状态，否则挂载与 translate-x-0 在同一帧内完成，transition 不触发
- 窄屏专属控件用 useIsNarrow() 条件渲染而非 md:hidden —— 后者只是视觉隐藏，
宽屏用户按 Tab 会聚焦到看不见的返回按钮
- 底部导航 + 抽屉侧栏 + env(safe-area-inset-bottom)
## 工作列表卡片视图（Phase 7.1 最后一项）
中间栏可切列表/卡片。列表答「跟谁在聊」，卡片答「在聊什么、进展如何」：
主题 + 最新一封的发件人与摘要 + 往返预算徽标。
- 两种视图共用同一份数据与同一套动作；归档确认框也共用 —— 归档是破坏性操作，
换个视图就换套确认 UI 只会让人对「自己点了什么」更没底
- 预算徽标在「不限」时不显示（对每张卡片都成立的「0/0」是纯噪声）
- 数据一次取回，不让卡片为每条会话再打一次库
## 修掉的缺陷
- GET /me/sessions 一直 500：ListSessionsFor 的 SELECT 加了预算两列却没加进
Scan，列数不匹配。联系人栏一条数据都拉不到，而错误只是「Failed to list sessions」
- GET /sessions/{id} 忘了填充附件：前端会话视图走的是这个端点，于是 Agent
回信里的附件在 UI 上完全不存在（另一个端点填了但没人调用）
- 插件曾完全没在加载：为了可测在 index.js 里 export 了辅助函数与一个 Map，
而 opencode 把入口模块的每一个导出都当成插件工厂逐个检查，多导出一个 Map
就 "Plugin export is not a function"，插件静默失效、邮件全投不进去。
逻辑挪到 lib/relay-dedup.js，并加断言钉住「入口只有 default 导出」
- 同一件事发两封邮件：模型带附件主动回信后，session.idle 又把它最后那段话
自动转了一遍（生产实测 311 与 342 字节各一封）。explicitSends 记录本轮
主动发信，自动转发据此让位；relay_key 幂等管不了这个 —— 那个键保证的是
「同一条消息不转两次」
- SQLite 时间戳只有秒精度：同秒插入的多封邮件排序不确定（实测同秒插 5 封，
顺序由随机 UUID 决定）。「会话里最早那封」（决定联系人身份）与「最后那封」
（决定最新进展）都会取错。NOW() 升到微秒 + mails 的 INSERT 显式传它
（改 schema 默认值只对新库生效，SQLite 没有 ALTER COLUMN）+ 所有
ORDER BY created_at 补 mail_id 兜底
- fillAttachments 从逐封查询改成一次 IN(...)：原来是 N+1，200 封的会话打开
要打 200 次库
- repo 层 5 处 rows.Next() 循环补 rows.Err()：没有它，读到一半连接断掉会
静默返回部分结果，UI 上表现为「邮件凭空少了几封」
- go:embed 占位页改名 placeholder.html：叫 index.html 会被 Vite 产物覆盖并
提交进去，而它引用的 assets/ 是被忽略的 —— 新克隆打开是白屏
## 回复/转发栏
- 两处都加抄送（可折叠）；原邮件带抄送时多一个「回复全部」，回填用
cc_list[].raw 而非重拼 name@path（后者会丢掉会话段）
- 会话视图每张卡片加转发入口：转发之前只存在于单封邮件视图，而人多数时间
待在会话视图里，等于功能在 UI 上找不到
- ReplyBar 的错误从 console.error 改为显示出来：预算耗尽、地址不存在、
速率限制都走这条路，之前点发送毫无反应
## 测试
- repo: 列顺序（三个 SQL 分支）、卡片字段、previewRunes 边界、时间戳亚秒精度、
批量附件查询、速率限制（80 goroutine 断言恰好 20 条通过）
- web: 窄屏布局 16 条结构性断言（覆盖而非分栏、延迟卸载、双层 rAF、
条件渲染而非 md:hidden）
- 插件: 自动转发去重 17 条（含「入口只有 default 导出」不变量）
- install.sh 把插件测试也纳入部署前门禁
This commit is contained in:
@ -48,11 +48,23 @@ func init() {
|
||||
return uuid.NewString(), nil
|
||||
})
|
||||
|
||||
// NOW() 必须非确定性：同一语句内多次调用要各自取当前时刻，
|
||||
// 且格式与 SQLite 的 CURRENT_TIMESTAMP 一致，才能统一扫进 time.Time。
|
||||
// NOW() 必须非确定性：同一语句内多次调用要各自取当前时刻。
|
||||
//
|
||||
// 精度到微秒而不是秒：SQLite 的 CURRENT_TIMESTAMP 只有秒，
|
||||
// 同一秒内插入的多封邮件排序就不确定 —— 「会话里最早那封」（决定联系人身份）
|
||||
// 与「最后那封」（决定最新进展）都会取错行。实测同秒插 5 封，
|
||||
// 按 created_at 排出来的顺序是乱的（由随机 UUID 决定）。
|
||||
//
|
||||
// 毫秒还不够：一次插入只要几十到几百微秒，循环里连插几封会落在同一毫秒。
|
||||
// 微秒是实测确认驱动能原样扫回 time.Time 的精度（纳秒也行，但没必要）。
|
||||
//
|
||||
// 格式仍是 SQLite 认得的 "YYYY-MM-DD HH:MM:SS.ffffff"，因此：
|
||||
// - 驱动能扫进 time.Time（列声明为 DATETIME 时）
|
||||
// - 与老数据（秒精度）的文本比较依然正确：前缀相同时短的排前面，
|
||||
// 而 ":31" 确实早于 ":31.767000"
|
||||
sqlite.MustRegisterScalarFunction("now", 0,
|
||||
func(*sqlite.FunctionContext, []driver.Value) (driver.Value, error) {
|
||||
return time.Now().UTC().Format("2006-01-02 15:04:05"), nil
|
||||
return time.Now().UTC().Format("2006-01-02 15:04:05.000000"), nil
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@ -76,6 +76,10 @@ var sqliteAddColumns = []struct{ table, column, ddl string }{
|
||||
// 会话级往返预算（0 = 不限）。旧库默认 0：引入预算不应该把已在进行的会话卡死。
|
||||
{"sessions", "max_rounds", "ALTER TABLE sessions ADD COLUMN max_rounds INTEGER NOT NULL DEFAULT 0"},
|
||||
{"sessions", "used_rounds", "ALTER TABLE sessions ADD COLUMN used_rounds INTEGER NOT NULL DEFAULT 0"},
|
||||
// 派给该 Agent 的新任务默认多少个来回。
|
||||
// 旧库也给 20：之前的 max_rounds 默认是 10 但那是终身额度，语义不同，
|
||||
// 不能直接搬过来当单任务预算。
|
||||
{"agents", "default_rounds", "ALTER TABLE agents ADD COLUMN default_rounds INTEGER NOT NULL DEFAULT 20"},
|
||||
}
|
||||
|
||||
// sqliteAddIndexes 是建表后才能建的索引（依赖上面补的列）。
|
||||
|
||||
@ -126,6 +126,10 @@ CREATE INDEX IF NOT EXISTS idx_mails_parent ON mails(parent_mail_id);
|
||||
-- 已存在的库补列（重复运行安全）
|
||||
ALTER TABLE mails ADD COLUMN IF NOT EXISTS cc_list JSONB NOT NULL DEFAULT '[]';
|
||||
|
||||
-- default_rounds 是【派给这个 Agent 的新任务】默认有多少个来回。
|
||||
-- 配额是任务的属性，真正的约束在 sessions.max_rounds 上；这里只提供默认值。
|
||||
ALTER TABLE agents ADD COLUMN IF NOT EXISTS default_rounds INTEGER NOT NULL DEFAULT 20;
|
||||
|
||||
-- Agent 在正文里提议改会话别名（<!-- agentmail:rename-session … -->）。
|
||||
-- 存在邮件上而非会话上：邮件是不可篡改的历史记录，「谁在哪一封里提了什么」应当留痕。
|
||||
ALTER TABLE mails ADD COLUMN IF NOT EXISTS rename_alias TEXT;
|
||||
|
||||
@ -3,6 +3,14 @@
|
||||
-- 与 init.sql（PostgreSQL）保持同一套表结构与语义，差异仅在方言：
|
||||
-- UUID → TEXT（Go 侧 uuid 或 gen_random_uuid() 注册函数生成）
|
||||
-- TIMESTAMPTZ → DATETIME（必须写 DATETIME，database/sql 才能扫进 time.Time）
|
||||
-- 默认值不用 CURRENT_TIMESTAMP：它只有【秒】精度，同一秒内插入的多行
|
||||
-- 按 created_at 排序结果不确定，
|
||||
-- 「会话里最早/最后那封邮件」都会取错行
|
||||
-- （实测同秒插 5 封，排出来的顺序是乱的）。
|
||||
-- 改用 strftime 的毫秒精度。mails 表另在 repo 层的
|
||||
-- INSERT 里显式传 NOW()（微秒精度）—— 一次插入只要
|
||||
-- 几十到几百微秒，毫秒仍可能撞车，而邮件顺序
|
||||
-- 直接决定 UI 上「最新进展」显示哪一封。
|
||||
-- JSONB → TEXT（存 JSON 字符串，用 json_each/json_extract 检索）
|
||||
-- VARCHAR(n) → TEXT（SQLite 不强制长度，长度约束由应用层负责）
|
||||
-- NOW() → 由 internal/db 注册的同名函数提供，与 PG 侧 SQL 一致
|
||||
@ -16,7 +24,7 @@ CREATE TABLE IF NOT EXISTS users (
|
||||
password_hash TEXT NOT NULL,
|
||||
role TEXT NOT NULL DEFAULT 'user',
|
||||
status TEXT NOT NULL DEFAULT 'active',
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
created_at DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f','now')),
|
||||
last_login DATETIME,
|
||||
|
||||
-- 权限边界：空数组 = 不限
|
||||
@ -28,7 +36,7 @@ CREATE TABLE IF NOT EXISTS users (
|
||||
CREATE TABLE IF NOT EXISTS user_sessions (
|
||||
token TEXT PRIMARY KEY,
|
||||
user_id TEXT NOT NULL REFERENCES users(user_id) ON DELETE CASCADE,
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
created_at DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f','now')),
|
||||
expires_at DATETIME NOT NULL,
|
||||
user_agent TEXT DEFAULT ''
|
||||
);
|
||||
@ -44,10 +52,19 @@ CREATE TABLE IF NOT EXISTS agents (
|
||||
workspaces TEXT NOT NULL DEFAULT '[]',
|
||||
platform TEXT NOT NULL DEFAULT 'pi',
|
||||
status TEXT NOT NULL DEFAULT 'offline',
|
||||
max_rounds INTEGER NOT NULL DEFAULT 10,
|
||||
-- default_rounds 是【派给这个 Agent 的新任务】默认有多少个来回。
|
||||
-- 配额是任务的属性，所以真正的约束在 sessions.max_rounds 上；
|
||||
-- 这里只提供默认值 —— 不同 Agent 能力不同，默认值分开设才合理。
|
||||
default_rounds INTEGER NOT NULL DEFAULT 20,
|
||||
|
||||
-- max_rounds / used_rounds 是历史遗留的「终身额度」。
|
||||
-- 终身额度是错的工具：跑满就得管理员手工重置才能再干活，
|
||||
-- 而 Agent 是长期在线的。现已降级为纯统计（used_rounds 只累加、不拦请求），
|
||||
-- max_rounds 保留列但不再参与判断。
|
||||
max_rounds INTEGER NOT NULL DEFAULT 0,
|
||||
used_rounds INTEGER NOT NULL DEFAULT 0,
|
||||
last_seen DATETIME,
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
created_at DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f','now'))
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS sessions (
|
||||
@ -57,8 +74,8 @@ CREATE TABLE IF NOT EXISTS sessions (
|
||||
subject TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'active',
|
||||
owner_user_id TEXT REFERENCES users(user_id),
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
created_at DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f','now')),
|
||||
updated_at DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f','now')),
|
||||
|
||||
-- 用户驳回过的改名提议。记下来才能让提示条不再反复弹同一个建议。
|
||||
rename_dismissed TEXT,
|
||||
@ -112,7 +129,7 @@ CREATE TABLE IF NOT EXISTS mails (
|
||||
permission_result TEXT,
|
||||
|
||||
status TEXT NOT NULL DEFAULT 'unread',
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
created_at DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f','now')),
|
||||
|
||||
hop_limit INTEGER DEFAULT 5,
|
||||
|
||||
@ -141,7 +158,7 @@ CREATE TABLE IF NOT EXISTS permission_requests (
|
||||
context TEXT DEFAULT '',
|
||||
result TEXT,
|
||||
decided_at DATETIME,
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
created_at DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f','now'))
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_perm_agent ON permission_requests(agent_name, result);
|
||||
@ -167,7 +184,7 @@ CREATE TABLE IF NOT EXISTS agent_keys (
|
||||
expires_at DATETIME,
|
||||
used_at DATETIME,
|
||||
created_by TEXT REFERENCES users(user_id),
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
created_at DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f','now'))
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_agent_keys_token ON agent_keys(key_token);
|
||||
@ -181,7 +198,7 @@ CREATE TABLE IF NOT EXISTS user_keys (
|
||||
key_type TEXT NOT NULL DEFAULT 'permanent',
|
||||
expires_at DATETIME,
|
||||
used_at DATETIME,
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
created_at DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f','now'))
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_user_keys_user ON user_keys(user_id);
|
||||
@ -212,7 +229,7 @@ CREATE TABLE IF NOT EXISTS relayed_mails (
|
||||
relay_key TEXT NOT NULL,
|
||||
mail_id TEXT REFERENCES mails(mail_id),
|
||||
kind TEXT NOT NULL DEFAULT '',
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
created_at DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f','now')),
|
||||
PRIMARY KEY (agent_name, relay_key)
|
||||
);
|
||||
|
||||
@ -232,7 +249,7 @@ CREATE TABLE IF NOT EXISTS attachments (
|
||||
-- sha256 既是去重依据也是磁盘路径来源，绝不用用户给的 filename 拼路径
|
||||
sha256 TEXT NOT NULL,
|
||||
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
created_at DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f','now'))
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_attachments_mail ON attachments(mail_id);
|
||||
|
||||
@ -111,18 +111,20 @@ func HeartbeatAgent(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
// 心跳回传配额：插件据此把剩余次数注入 Agent 上下文，
|
||||
// 让它在配额耗尽前主动发总结，而不是撞到 403 才发现。
|
||||
quota, qErr := repo.GetQuota(r.Context(), agentName)
|
||||
if qErr != nil {
|
||||
// 配额读不到不影响心跳本身，降级为不限额
|
||||
quota = repo.Quota{AgentName: agentName, Unlimited: true, Remaining: -1}
|
||||
// 心跳回传该 Agent 的累计统计与新任务默认预算。
|
||||
//
|
||||
// 不再回传「剩余额度」：额度属于具体任务（会话）而不属于 Agent，
|
||||
// 剩余往返随每次发信响应（budget_remaining）回传，在那里才有意义。
|
||||
stats, sErr := repo.GetAgentStats(r.Context(), agentName)
|
||||
if sErr != nil {
|
||||
// 统计读不到不影响心跳本身
|
||||
stats = repo.AgentStats{AgentName: agentName}
|
||||
}
|
||||
|
||||
JSON(w, http.StatusOK, map[string]interface{}{
|
||||
"status": "ok",
|
||||
"pending_mails": pending,
|
||||
"quota": quota,
|
||||
"stats": stats,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@ -356,12 +356,31 @@ func attachAll(w http.ResponseWriter, r *http.Request, mailID uuid.UUID, ids []u
|
||||
|
||||
// fillAttachments 给邮件填充附件列表（读取单封/线程时用）。
|
||||
// 读附件失败不该让整封邮件打不开，因此吞错只留空列表。
|
||||
//
|
||||
// 一批邮件走一次查询：逐封调 ListAttachmentsFor 是 N+1，
|
||||
// 一个 200 封的会话打开一次要打 200 次库。
|
||||
func fillAttachments(r *http.Request, mails ...*models.Mail) {
|
||||
ids := make([]uuid.UUID, 0, len(mails))
|
||||
for _, m := range mails {
|
||||
if m != nil {
|
||||
ids = append(ids, m.ID)
|
||||
}
|
||||
}
|
||||
if len(ids) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
byMail, err := repo.ListAttachmentsForMails(r.Context(), ids)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, m := range mails {
|
||||
if m == nil {
|
||||
continue
|
||||
}
|
||||
if as, err := repo.ListAttachmentsFor(r.Context(), m.ID); err == nil {
|
||||
// 没有附件的邮件保持 nil：Attachments 带 omitempty，
|
||||
// 填空切片只会给每封邮件的 JSON 加一个 "attachments":[]
|
||||
if as := byMail[m.ID]; len(as) > 0 {
|
||||
m.Attachments = as
|
||||
}
|
||||
}
|
||||
|
||||
@ -123,24 +123,27 @@ func doForward(w http.ResponseWriter, r *http.Request, mailID uuid.UUID, actor,
|
||||
subject := forwardSubject(req.Subject, src.Subject)
|
||||
|
||||
// 转发按目标地址寻址，不带 reply_to：它是一条新线索，不该并进原会话
|
||||
sessionID, _, err := resolveTarget(r, to, "", actor, subject, req.SessionAlias)
|
||||
sessionID, _, err := resolveTarget(r, to, "", actor, subject, req.SessionAlias, agentLimiterKey(isAgent, actor))
|
||||
if err != nil {
|
||||
writeErr(w, err, "Failed to resolve session")
|
||||
return
|
||||
}
|
||||
|
||||
if isAgent {
|
||||
quota, qErr := repo.ConsumeQuota(r.Context(), actor)
|
||||
if errors.Is(qErr, repo.ErrQuotaExhausted) {
|
||||
// 转发也是一次主动发信，扣【目标会话】的往返预算。
|
||||
// 扣目标而不是源：转发开启的是一条新线索，消耗的是新线索的额度。
|
||||
budget, bErr := repo.ConsumeSessionBudget(r.Context(), sessionID)
|
||||
if errors.Is(bErr, repo.ErrSessionBudgetExhausted) {
|
||||
Error(w, http.StatusForbidden, fmt.Sprintf(
|
||||
"发信配额已用尽（%d/%d）。请先向人类发送最终总结，或联系管理员重置配额。",
|
||||
quota.Used, quota.Max))
|
||||
"目标会话的往返预算已用尽（%d/%d）。请让人在对话页调高该会话的预算。",
|
||||
budget.Used, budget.Max))
|
||||
return
|
||||
}
|
||||
if qErr != nil {
|
||||
Error(w, http.StatusInternalServerError, "Failed to check quota")
|
||||
if bErr != nil {
|
||||
Error(w, http.StatusInternalServerError, "Failed to check session budget")
|
||||
return
|
||||
}
|
||||
repo.BumpSentCount(r.Context(), actor)
|
||||
} else if user != nil {
|
||||
_ = repo.SetSessionOwner(r.Context(), sessionID, user.ID)
|
||||
}
|
||||
@ -207,26 +210,35 @@ func MeForwardMail(w http.ResponseWriter, r *http.Request) {
|
||||
doForward(w, r, mailID, user.Username, "", false)
|
||||
}
|
||||
|
||||
// ---------- 配额管理（管理员） ----------
|
||||
// ---------- Agent 默认预算与统计（管理员） ----------
|
||||
|
||||
// GET /api/v1/admin/quotas
|
||||
//
|
||||
// 路径沿用 quotas（兼容已部署的前端），但语义已变：
|
||||
// 返回的是【新任务默认预算 + 累计统计】，而不是会拦请求的终身额度。
|
||||
// 真正的额度在每条会话上（GET /sessions/{id}/budget）。
|
||||
func AdminListQuotas(w http.ResponseWriter, r *http.Request) {
|
||||
quotas, err := repo.ListQuotas(r.Context())
|
||||
stats, err := repo.ListAgentStats(r.Context())
|
||||
if err != nil {
|
||||
Error(w, http.StatusInternalServerError, "Failed to list quotas")
|
||||
Error(w, http.StatusInternalServerError, "Failed to list agent stats")
|
||||
return
|
||||
}
|
||||
JSON(w, http.StatusOK, map[string]any{"quotas": quotas})
|
||||
JSON(w, http.StatusOK, map[string]any{"quotas": stats})
|
||||
}
|
||||
|
||||
type setQuotaRequest struct {
|
||||
// MaxRounds 发信配额上限；0 = 不限
|
||||
// DefaultRounds 派给该 Agent 的新任务默认多少个来回（0 = 不限）
|
||||
DefaultRounds *int `json:"default_rounds"`
|
||||
// MaxRounds 是 DefaultRounds 的旧字段名，保留兼容：
|
||||
// 已部署的前端与脚本不应该因为改名就难以察觉地失效。
|
||||
MaxRounds *int `json:"max_rounds"`
|
||||
// Reset 为 true 时把已用次数归零
|
||||
Reset bool `json:"reset"`
|
||||
}
|
||||
|
||||
// PUT /api/v1/admin/quotas/{name}
|
||||
//
|
||||
// 只能改【新任务默认预算】。不再接受 reset：
|
||||
// 累计发信数是观测数据，不拦任何请求，归零它只会销毁历史。
|
||||
// 要给某个卡住的任务加额度，去那条会话的对话页改预算。
|
||||
func AdminSetQuota(w http.ResponseWriter, r *http.Request) {
|
||||
name := strings.TrimSpace(chi.URLParam(r, "name"))
|
||||
if name == "" {
|
||||
@ -239,26 +251,23 @@ func AdminSetQuota(w http.ResponseWriter, r *http.Request) {
|
||||
Error(w, http.StatusBadRequest, "Invalid JSON")
|
||||
return
|
||||
}
|
||||
if req.MaxRounds == nil && !req.Reset {
|
||||
Error(w, http.StatusBadRequest, "需要 max_rounds 或 reset 之一")
|
||||
n := req.DefaultRounds
|
||||
if n == nil {
|
||||
n = req.MaxRounds // 兼容旧字段名
|
||||
}
|
||||
if n == nil {
|
||||
Error(w, http.StatusBadRequest, "需要 default_rounds")
|
||||
return
|
||||
}
|
||||
if *n < 0 {
|
||||
Error(w, http.StatusBadRequest, "default_rounds 不能为负")
|
||||
return
|
||||
}
|
||||
|
||||
var (
|
||||
q repo.Quota
|
||||
err error
|
||||
)
|
||||
if req.MaxRounds != nil {
|
||||
if q, err = repo.SetQuota(r.Context(), name, *req.MaxRounds); err != nil {
|
||||
Error(w, http.StatusNotFound, err.Error())
|
||||
return
|
||||
}
|
||||
st, err := repo.SetDefaultRounds(r.Context(), name, *n)
|
||||
if err != nil {
|
||||
Error(w, http.StatusNotFound, err.Error())
|
||||
return
|
||||
}
|
||||
if req.Reset {
|
||||
if q, err = repo.ResetQuota(r.Context(), name); err != nil {
|
||||
Error(w, http.StatusNotFound, err.Error())
|
||||
return
|
||||
}
|
||||
}
|
||||
JSON(w, http.StatusOK, map[string]any{"quota": q})
|
||||
JSON(w, http.StatusOK, map[string]any{"quota": st})
|
||||
}
|
||||
|
||||
@ -39,6 +39,11 @@ func errBadRequest(msg string) error { return httpError{http.StatusBadRequest, m
|
||||
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) {
|
||||
@ -126,3 +131,12 @@ func emptySlice[T any](s []T) []T {
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// agentLimiterKey 把「这是不是 Agent 发起的」翻译成速率限制的键。
|
||||
// 人类返回空串 = 不限速（手工操作的频率天然受限）。
|
||||
func agentLimiterKey(isAgent bool, actor string) string {
|
||||
if isAgent {
|
||||
return actor
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
@ -47,7 +47,11 @@ type sendMailRequest struct {
|
||||
//
|
||||
// alias 为新建会话命名（仅新建时生效），使其之后可被 name@path.<alias> 寻址。
|
||||
// reply_to 优先于地址：显式回复某封邮件时沿用该邮件的会话。
|
||||
func resolveTarget(r *http.Request, addr models.Address, replyTo, fromAgent, subject, alias string) (uuid.UUID, *uuid.UUID, error) {
|
||||
//
|
||||
// byAgent 非空时表示这是 Agent 发起的投递，新建会话要过速率限制：
|
||||
// 往返预算按会话计，Agent 用 .new 开一串会话就等于绕过预算。
|
||||
// 人类不受此限（手工点「新建邮件」的频率天然受限，加限制只会在批量派活时误伤）。
|
||||
func resolveTarget(r *http.Request, addr models.Address, replyTo, fromAgent, subject, alias string, byAgent string) (uuid.UUID, *uuid.UUID, error) {
|
||||
if replyTo != "" {
|
||||
replyID, err := uuid.Parse(replyTo)
|
||||
if err != nil {
|
||||
@ -76,10 +80,22 @@ func resolveTarget(r *http.Request, addr models.Address, replyTo, fromAgent, sub
|
||||
}
|
||||
aliasPtr = &a
|
||||
}
|
||||
// Agent 主动开新线索要过速率限制
|
||||
if ok, retry := repo.AllowNewSession(byAgent); !ok {
|
||||
return uuid.Nil, nil, errRateLimited(fmt.Sprintf(
|
||||
"新建会话过于频繁（1 小时内已开 %d 条）。请在已有会话里继续，或 %d 秒后再试。",
|
||||
repo.SessionRateLimit(), retry))
|
||||
}
|
||||
id, err := repo.CreateSession(r.Context(), aliasPtr, fromAgent, subject)
|
||||
if err != nil {
|
||||
// 建失败要把名额还回去：那次新建实际上没有发生
|
||||
repo.ReleaseNewSession(byAgent)
|
||||
}
|
||||
return id, nil, err
|
||||
|
||||
case models.SessionDefault:
|
||||
// 默认会话「从未通信则建立」也会产生新会话，但一个 name@path 只有一条，
|
||||
// 不构成暴开的手段，因此不计入速率限制。
|
||||
id, err := repo.FindOrCreateDefaultSession(r.Context(), addr.Name, addr.Path, fromAgent, subject)
|
||||
return id, nil, err
|
||||
|
||||
@ -132,7 +148,7 @@ func SendMail(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
sessionID, parentMailID, err := resolveTarget(r, to, req.ReplyTo, agentName, req.Subject, req.SessionAlias)
|
||||
sessionID, parentMailID, err := resolveTarget(r, to, req.ReplyTo, agentName, req.Subject, req.SessionAlias, agentName)
|
||||
if err != nil {
|
||||
writeErr(w, err, "Failed to resolve session")
|
||||
return
|
||||
@ -149,7 +165,6 @@ func SendMail(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
var quota repo.Quota
|
||||
var budget repo.SessionBudget
|
||||
if relay != "" {
|
||||
// 先占幂等键。重复则说明这条上游消息已经转过，
|
||||
@ -167,19 +182,19 @@ func SendMail(w http.ResponseWriter, r *http.Request) {
|
||||
Error(w, http.StatusInternalServerError, "Failed to claim relay")
|
||||
return
|
||||
}
|
||||
// 仅读快照用于回传，不扣任何一层
|
||||
quota, _ = repo.GetQuota(r.Context(), agentName)
|
||||
// 仅读快照用于回传，不扣预算
|
||||
budget, _ = repo.GetSessionBudget(r.Context(), sessionID)
|
||||
} else {
|
||||
// 两层都要过：会话预算管「这件事值得多少个来回」，
|
||||
// Agent 全局配额管「这个 Agent 总共能发多少」。
|
||||
// 先扣会话、后扣全局；全局拦下时把会话那次退回去 ——
|
||||
// 那次往返实际上没有发生，不能白掉一格。
|
||||
// 额度只看【本任务】的往返预算。
|
||||
//
|
||||
// 不再叠一层 Agent 终身额度：那种额度跑满后要管理员手工重置才能再干活，
|
||||
// 而 Agent 是长期在线的。防止 Agent 用 .new 开一串新会话绕过预算，
|
||||
// 靠的是新建会话速率限制（resolveTarget 里）。
|
||||
budget, err = repo.ConsumeSessionBudget(r.Context(), sessionID)
|
||||
if errors.Is(err, repo.ErrSessionBudgetExhausted) {
|
||||
Error(w, http.StatusForbidden, fmt.Sprintf(
|
||||
"本会话的往返预算已用尽（%d/%d）。自动转发的总结与权限询问不占预算；"+
|
||||
"若需继续主动发信，请让人在对话页调高本会话的预算。",
|
||||
"本任务的往返预算已用尽（%d/%d）。自动转发的总结与权限询问不占预算；"+
|
||||
"若需继续主动发信，请让人在对话页调高本任务的预算。",
|
||||
budget.Used, budget.Max))
|
||||
return
|
||||
}
|
||||
@ -187,21 +202,8 @@ func SendMail(w http.ResponseWriter, r *http.Request) {
|
||||
Error(w, http.StatusInternalServerError, "Failed to check session budget")
|
||||
return
|
||||
}
|
||||
|
||||
quota, err = repo.ConsumeQuota(r.Context(), agentName)
|
||||
if errors.Is(err, repo.ErrQuotaExhausted) {
|
||||
repo.RefundSessionBudget(r.Context(), sessionID)
|
||||
Error(w, http.StatusForbidden, fmt.Sprintf(
|
||||
"Agent 全局发信配额已用尽（%d/%d）。插件代劳转发的权限询问与最终总结不占配额；"+
|
||||
"若需继续主动发信请联系管理员重置配额。",
|
||||
quota.Used, quota.Max))
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
repo.RefundSessionBudget(r.Context(), sessionID)
|
||||
Error(w, http.StatusInternalServerError, "Failed to check quota")
|
||||
return
|
||||
}
|
||||
// 纯统计，不拦请求；写失败也不该让邮件发不出去
|
||||
repo.BumpSentCount(r.Context(), agentName)
|
||||
}
|
||||
|
||||
// Agent 可以在正文里提议改会话别名（<!-- agentmail:rename-session … -->）。
|
||||
@ -237,27 +239,22 @@ func SendMail(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
notifyRecipients(to, ccList, sessionID, mailID, agentName, req.Subject)
|
||||
|
||||
// 回传会话别名与剩余配额，让发件方知道后续用什么地址续谈、还能发几封
|
||||
// 回传会话别名与本任务剩余往返，让发件方知道后续用什么地址续谈、还能发几封
|
||||
resp := map[string]any{
|
||||
"mail_id": mailID.String(),
|
||||
"session_id": sessionID.String(),
|
||||
"session_alias": repo.SessionAliasOf(r.Context(), sessionID),
|
||||
}
|
||||
if !quota.Unlimited {
|
||||
resp["quota_remaining"] = quota.Remaining
|
||||
resp["quota_used"] = quota.Used
|
||||
resp["quota_max"] = quota.Max
|
||||
}
|
||||
// 会话预算是【本任务】的剩余往返，Agent 更应该看这个而不是全局配额
|
||||
// 预算属于【本任务】，不限时不回传 —— 多给一个 -1 只会让插件去判断哪个值是哨兵
|
||||
if !budget.Unlimited {
|
||||
resp["budget_remaining"] = budget.Remaining
|
||||
resp["budget_used"] = budget.Used
|
||||
resp["budget_max"] = budget.Max
|
||||
}
|
||||
if relay != "" {
|
||||
// 告知本次未扣配额，否则插件看到 quota_remaining 没变会以为数据错了
|
||||
// 告知本次未扣预算，否则插件看到 budget_remaining 没变会以为数据错了
|
||||
resp["relay"] = relay
|
||||
resp["quota_charged"] = false
|
||||
resp["budget_charged"] = false
|
||||
}
|
||||
if proposal != nil {
|
||||
// 回传规范化后的别名：Agent 提的名字可能含非法字符被改写过，
|
||||
@ -333,9 +330,11 @@ func GetInbox(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
// Agent 靠收件箱列表得知有哪些附件可下载，否则它不知道该调 attachment_id
|
||||
ptrs := make([]*models.Mail, len(mails))
|
||||
for i := range mails {
|
||||
fillAttachments(r, &mails[i])
|
||||
ptrs[i] = &mails[i]
|
||||
}
|
||||
fillAttachments(r, ptrs...)
|
||||
total, _ := repo.CountUnread(r.Context(), agentName)
|
||||
|
||||
JSON(w, http.StatusOK, map[string]interface{}{
|
||||
@ -419,3 +418,74 @@ func parseInt(s string) (int, error) {
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
type markReadRequest struct {
|
||||
// MailIDs 要标记为已读的邮件；省略/为空 = 把收件箱里全部未读标掉。
|
||||
MailIDs []string `json:"mail_ids"`
|
||||
}
|
||||
|
||||
// POST /api/v1/mail/read —— Agent 侧批量标记已读
|
||||
//
|
||||
// 为什么需要它：Agent 读完 read_inbox 后没有任何办法把邮件标掉，
|
||||
// 于是每次拉收件箱都把同一批旧邮件重新捞出来 —— 处理过的信和新来的信混在一起，
|
||||
// 模型分不清哪封该回。心跳里的未读数也永远只增不减。
|
||||
//
|
||||
// 鉴权写进 UPDATE 的 WHERE 而不是先查后改：不是发给自己的邮件根本改不动，
|
||||
// 既省一次查询，也没有「查完到改之间邮件被转走」的时间窗。
|
||||
func MarkInboxRead(w http.ResponseWriter, r *http.Request) {
|
||||
agentName := middleware.GetAgentName(r)
|
||||
if agentName == "" {
|
||||
Error(w, http.StatusUnauthorized, "Unauthorized")
|
||||
return
|
||||
}
|
||||
|
||||
var req markReadRequest
|
||||
// 允许空 body：`POST /mail/read` 不带任何内容 = 全部标掉
|
||||
if r.ContentLength > 0 {
|
||||
if err := Decode(r, &req); err != nil {
|
||||
Error(w, http.StatusBadRequest, "Invalid JSON")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// 不给 id 就把收件箱里全部未读标掉。
|
||||
// 这是 Agent 最常见的用法：一轮处理完，剩下的都不必再看。
|
||||
if len(req.MailIDs) == 0 {
|
||||
n, err := repo.MarkAllInboxReadFor(r.Context(), agentName)
|
||||
if err != nil {
|
||||
Error(w, http.StatusInternalServerError, "Failed to mark read")
|
||||
return
|
||||
}
|
||||
JSON(w, http.StatusOK, map[string]any{"status": "read", "marked": n, "scope": "all"})
|
||||
return
|
||||
}
|
||||
|
||||
const maxBatch = 200
|
||||
if len(req.MailIDs) > maxBatch {
|
||||
Error(w, http.StatusBadRequest, fmt.Sprintf("一次最多标记 %d 封", maxBatch))
|
||||
return
|
||||
}
|
||||
ids := make([]uuid.UUID, 0, len(req.MailIDs))
|
||||
for _, s := range req.MailIDs {
|
||||
id, err := uuid.Parse(strings.TrimSpace(s))
|
||||
if err != nil {
|
||||
Error(w, http.StatusBadRequest, "非法的 mail_id: "+s)
|
||||
return
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
|
||||
n, err := repo.MarkMailsReadFor(r.Context(), agentName, ids)
|
||||
if err != nil {
|
||||
Error(w, http.StatusInternalServerError, "Failed to mark read")
|
||||
return
|
||||
}
|
||||
// 不因为「有些 id 不是发给你的」而报错：那些 id 只是没被标掉。
|
||||
// 报错会让整批失败，而 Agent 通常是把上一轮列出的 id 原样传回来，
|
||||
// 其中可能混着已读的（幂等）——那不该是错误。
|
||||
JSON(w, http.StatusOK, map[string]any{
|
||||
"status": "read",
|
||||
"marked": n,
|
||||
"requested": len(ids),
|
||||
})
|
||||
}
|
||||
|
||||
@ -77,7 +77,7 @@ func MeSendMail(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
sessionID, parentMailID, err := resolveTarget(r, to, req.ReplyTo, user.Username, req.Subject, req.SessionAlias)
|
||||
sessionID, parentMailID, err := resolveTarget(r, to, req.ReplyTo, user.Username, req.Subject, req.SessionAlias, "")
|
||||
if err != nil {
|
||||
writeErr(w, err, "Failed to resolve session")
|
||||
return
|
||||
@ -85,14 +85,23 @@ func MeSendMail(w http.ResponseWriter, r *http.Request) {
|
||||
// 人类发起的会话归属于该用户
|
||||
_ = repo.SetSessionOwner(r.Context(), sessionID, user.ID)
|
||||
|
||||
// 新建会话时接受往返预算。只在新建时设：续谈已有会话若也接受这个字段，
|
||||
// 新建会话时定往返预算。只在新建时设：续谈已有会话若也接受这个字段，
|
||||
// 每封新信都会悄悄改掉对方正在遵守的预算，人却不一定意识到自己改了。
|
||||
if req.MaxRounds != nil && parentMailID == nil {
|
||||
if *req.MaxRounds < 0 {
|
||||
Error(w, http.StatusBadRequest, "max_rounds 不能为负")
|
||||
return
|
||||
//
|
||||
// 没显式给就用【收件 Agent 的默认值】。默认值挂在 Agent 上而不是全站一个数：
|
||||
// 跑测试的小工具与重构整个模块的 Agent，合理来回数差一个量级。
|
||||
if parentMailID == nil {
|
||||
rounds := 0
|
||||
if req.MaxRounds != nil {
|
||||
if *req.MaxRounds < 0 {
|
||||
Error(w, http.StatusBadRequest, "max_rounds 不能为负")
|
||||
return
|
||||
}
|
||||
rounds = *req.MaxRounds
|
||||
} else {
|
||||
rounds = repo.DefaultRoundsFor(r.Context(), to.Name)
|
||||
}
|
||||
if _, err := repo.SetSessionBudget(r.Context(), sessionID, *req.MaxRounds); err != nil {
|
||||
if _, err := repo.SetSessionBudget(r.Context(), sessionID, rounds); err != nil {
|
||||
Error(w, http.StatusInternalServerError, "Failed to set session budget")
|
||||
return
|
||||
}
|
||||
@ -154,9 +163,11 @@ func MeGetInbox(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
// 列表页要显示附件图标与下载入口
|
||||
ptrs := make([]*models.Mail, len(mails))
|
||||
for i := range mails {
|
||||
fillAttachments(r, &mails[i])
|
||||
ptrs[i] = &mails[i]
|
||||
}
|
||||
fillAttachments(r, ptrs...)
|
||||
total, _ := repo.CountUnread(r.Context(), user.Username)
|
||||
|
||||
JSON(w, http.StatusOK, map[string]interface{}{
|
||||
@ -182,9 +193,11 @@ func MeGetSent(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
mails, err := repo.ListSentBy(r.Context(), user.Username, limit)
|
||||
if err == nil {
|
||||
ptrs := make([]*models.Mail, len(mails))
|
||||
for i := range mails {
|
||||
fillAttachments(r, &mails[i])
|
||||
ptrs[i] = &mails[i]
|
||||
}
|
||||
fillAttachments(r, ptrs...)
|
||||
}
|
||||
if err != nil {
|
||||
Error(w, http.StatusInternalServerError, "Failed to list sent")
|
||||
@ -224,6 +237,11 @@ func MeGetSessions(w http.ResponseWriter, r *http.Request) {
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
MailCount int `json:"mail_count"`
|
||||
UnreadCount int `json:"unread_count"`
|
||||
// 往返预算随列表一并返回：预算是【任务】的属性，
|
||||
// 工作列表上就应当看得见哪些任务快跑满了，
|
||||
// 而不是点进去一个一个查。
|
||||
MaxRounds int `json:"max_rounds"`
|
||||
UsedRounds int `json:"used_rounds"`
|
||||
}
|
||||
|
||||
result := make([]SessionOut, 0, len(sessions))
|
||||
@ -239,6 +257,8 @@ func MeGetSessions(w http.ResponseWriter, r *http.Request) {
|
||||
UpdatedAt: s.UpdatedAt,
|
||||
MailCount: s.MailCount,
|
||||
UnreadCount: unread,
|
||||
MaxRounds: s.MaxRounds,
|
||||
UsedRounds: s.UsedRounds,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@ -5,6 +5,7 @@ import (
|
||||
"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"
|
||||
@ -52,6 +53,16 @@ func GetSession(w http.ResponseWriter, r *http.Request) {
|
||||
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,
|
||||
@ -71,9 +82,11 @@ func GetSessionMails(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
// 会话线程要展示附件，逐封填充
|
||||
ptrs := make([]*models.Mail, len(mails))
|
||||
for i := range mails {
|
||||
fillAttachments(r, &mails[i])
|
||||
ptrs[i] = &mails[i]
|
||||
}
|
||||
fillAttachments(r, ptrs...)
|
||||
JSON(w, http.StatusOK, map[string]interface{}{
|
||||
"mails": emptySlice(mails),
|
||||
})
|
||||
|
||||
@ -16,7 +16,12 @@ type Agent struct {
|
||||
Workspaces []Workspace `json:"workspaces"`
|
||||
Platform string `json:"platform"`
|
||||
Status string `json:"status"`
|
||||
MaxRounds int `json:"max_rounds"`
|
||||
// DefaultRounds 是派给该 Agent 的新任务默认多少个来回（0 = 不限）。
|
||||
// 真正的额度在每条会话上（sessions.max_rounds），这里只是默认值。
|
||||
DefaultRounds int `json:"default_rounds"`
|
||||
// UsedRounds 是累计发信数，纯统计，不拦请求。
|
||||
// 它原本是「终身额度」——那种额度跑满要人工重置才能再干活，
|
||||
// 而 Agent 是长期在线的，所以已降级为观测数据。
|
||||
UsedRounds int `json:"used_rounds"`
|
||||
LastSeen *time.Time `json:"last_seen"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
|
||||
@ -4,6 +4,8 @@ import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
@ -224,3 +226,46 @@ func AttachmentAccessible(ctx context.Context, a *models.Attachment, name string
|
||||
`, *a.MailID, name).Scan(&n)
|
||||
return n > 0, err
|
||||
}
|
||||
|
||||
// ListAttachmentsForMails 批量取多封邮件的附件，返回 mail_id → 附件列表。
|
||||
//
|
||||
// 为什么要批量：会话线程与收发件箱都是「一批邮件」，逐封调 ListAttachmentsFor
|
||||
// 就是 N+1 —— 一个 200 封的会话打开一次要打 200 次库。
|
||||
// 用 IN (...) 一次取回后在内存里分组。
|
||||
//
|
||||
// 占位符手工拼而非用数组参数：SQLite 驱动不支持 PG 的 = ANY($1)，
|
||||
// 而这里的元素是已解析的 uuid.UUID，不存在注入面。
|
||||
func ListAttachmentsForMails(ctx context.Context, mailIDs []uuid.UUID) (map[uuid.UUID][]models.Attachment, error) {
|
||||
out := map[uuid.UUID][]models.Attachment{}
|
||||
if len(mailIDs) == 0 {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
ph := make([]string, len(mailIDs))
|
||||
args := make([]any, len(mailIDs))
|
||||
for i, id := range mailIDs {
|
||||
ph[i] = fmt.Sprintf("$%d", i+1)
|
||||
args[i] = id
|
||||
}
|
||||
|
||||
rows, err := db.DB.QueryContext(ctx,
|
||||
`SELECT `+attachmentCols+` FROM attachments
|
||||
WHERE mail_id IN (`+strings.Join(ph, ",")+`)
|
||||
ORDER BY created_at`, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
a, err := scanAttachment(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if a.MailID == nil {
|
||||
continue // WHERE 已排除，只是防御
|
||||
}
|
||||
out[*a.MailID] = append(out[*a.MailID], *a)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
142
gateway/internal/repo/attachments_test.go
Normal file
142
gateway/internal/repo/attachments_test.go
Normal file
@ -0,0 +1,142 @@
|
||||
package repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// seedClock 给测试数据发严格递增的时间戳。
|
||||
//
|
||||
// 不靠挂钟：测试在一个循环里连插几封，很可能落在同一毫秒里，
|
||||
// 于是「会话里最早/最后那封」的排序由 mail_id（随机 UUID）决定 —— 结果随机。
|
||||
// 生产里两封邮件至少隔着一次模型推理，同毫秒撞车不现实；
|
||||
// 但测试必须确定，所以显式发号。
|
||||
var seedClock = time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
|
||||
|
||||
func nextSeedTime() string {
|
||||
seedClock = seedClock.Add(time.Second)
|
||||
return seedClock.Format("2006-01-02 15:04:05.000")
|
||||
}
|
||||
|
||||
// seedMailIn 在指定会话里插一封邮件，返回其 id。
|
||||
// 时间戳严格递增，因此调用顺序就是邮件的先后顺序。
|
||||
func seedMailIn(t *testing.T, sessionID uuid.UUID, from, to, subject string) uuid.UUID {
|
||||
t.Helper()
|
||||
var id uuid.UUID
|
||||
err := db.DB.QueryRowContext(context.Background(), `
|
||||
INSERT INTO mails (session_id, from_name, from_workspace, to_name, to_workspace,
|
||||
subject, body, cc_list, created_at)
|
||||
VALUES ($1, $2, '', $3, '', $4, 'body', '[]', $5)
|
||||
RETURNING mail_id
|
||||
`, sessionID, from, to, subject, nextSeedTime()).Scan(&id)
|
||||
if err != nil {
|
||||
t.Fatalf("seed mail: %v", err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func seedSessionRow(t *testing.T, alias string) uuid.UUID {
|
||||
t.Helper()
|
||||
var id uuid.UUID
|
||||
err := db.DB.QueryRowContext(context.Background(), `
|
||||
INSERT INTO sessions (from_agent, subject, session_alias)
|
||||
VALUES ('opencode', 'attach test', $1)
|
||||
RETURNING session_id
|
||||
`, alias).Scan(&id)
|
||||
if err != nil {
|
||||
t.Fatalf("seed session: %v", err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func attach(t *testing.T, mailID uuid.UUID, name, sum string) {
|
||||
t.Helper()
|
||||
a, err := CreateAttachment(context.Background(), "admin", name, "text/plain", 3, sum)
|
||||
if err != nil {
|
||||
t.Fatalf("create attachment: %v", err)
|
||||
}
|
||||
if err := AttachToMail(context.Background(), mailID, []uuid.UUID{a.ID}, "admin"); err != nil {
|
||||
t.Fatalf("attach: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ListAttachmentsForMails 存在的理由是消掉 N+1：
|
||||
// 原先每封邮件单独查一次，一个 200 封的会话打开要打 200 次库。
|
||||
func TestListAttachmentsForMails(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
sid := seedSessionRow(t, "batch-attach")
|
||||
|
||||
m1 := seedMailIn(t, sid, "admin", "opencode", "两个附件")
|
||||
m2 := seedMailIn(t, sid, "opencode", "admin", "一个附件")
|
||||
m3 := seedMailIn(t, sid, "admin", "opencode", "没有附件")
|
||||
|
||||
attach(t, m1, "a.txt", "sum-a")
|
||||
attach(t, m1, "b.txt", "sum-b")
|
||||
attach(t, m2, "c.txt", "sum-c")
|
||||
|
||||
got, err := ListAttachmentsForMails(context.Background(),
|
||||
[]uuid.UUID{m1, m2, m3})
|
||||
if err != nil {
|
||||
t.Fatalf("批量查询失败: %v", err)
|
||||
}
|
||||
|
||||
if n := len(got[m1]); n != 2 {
|
||||
t.Errorf("m1 应有 2 个附件，实际 %d", n)
|
||||
}
|
||||
if n := len(got[m2]); n != 1 {
|
||||
t.Errorf("m2 应有 1 个附件，实际 %d", n)
|
||||
}
|
||||
// 无附件的邮件不该出现在 map 里：调用方据此保持 Attachments 为 nil，
|
||||
// 这样带 omitempty 的字段不会给每封邮件的 JSON 白加一个 "attachments":[]
|
||||
if _, ok := got[m3]; ok {
|
||||
t.Errorf("m3 无附件却出现在结果里：%#v", got[m3])
|
||||
}
|
||||
|
||||
// 同一封内按 created_at 排序，顺序不能乱
|
||||
if len(got[m1]) == 2 && got[m1][0].Filename != "a.txt" {
|
||||
t.Errorf("同一封内应按上传顺序，首个是 %s", got[m1][0].Filename)
|
||||
}
|
||||
|
||||
// 每条都要带回 mail_id，否则调用方分不清是谁的
|
||||
for _, a := range got[m1] {
|
||||
if a.MailID == nil || *a.MailID != m1 {
|
||||
t.Errorf("附件 %s 的 mail_id 不对：%v", a.Filename, a.MailID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 空输入必须返回空 map 而非 nil：调用方直接索引不该 panic。
|
||||
func TestListAttachmentsForMailsEmpty(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
|
||||
for _, ids := range [][]uuid.UUID{nil, {}} {
|
||||
got, err := ListAttachmentsForMails(context.Background(), ids)
|
||||
if err != nil {
|
||||
t.Fatalf("空输入不该报错: %v", err)
|
||||
}
|
||||
if got == nil {
|
||||
t.Fatal("空输入应返回空 map 而非 nil")
|
||||
}
|
||||
if len(got) != 0 {
|
||||
t.Errorf("空输入应返回空结果，实际 %d 项", len(got))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 传入不存在的 mail_id 不该报错，只是查不到 —— 调用方可能拿着已删邮件的 id。
|
||||
func TestListAttachmentsForMailsUnknownID(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
|
||||
got, err := ListAttachmentsForMails(context.Background(),
|
||||
[]uuid.UUID{uuid.New(), uuid.New()})
|
||||
if err != nil {
|
||||
t.Fatalf("未知 id 不该报错: %v", err)
|
||||
}
|
||||
if len(got) != 0 {
|
||||
t.Errorf("未知 id 应查不到，实际 %d 项", len(got))
|
||||
}
|
||||
}
|
||||
144
gateway/internal/repo/markread_test.go
Normal file
144
gateway/internal/repo/markread_test.go
Normal file
@ -0,0 +1,144 @@
|
||||
package repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// seedMailTo 造一封给 recipient 的未读邮件，可选带抄送。
|
||||
func seedMailTo(t *testing.T, recipient string, cc string) uuid.UUID {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
sid, err := CreateSession(ctx, nil, "sender", "t")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ccJSON := "[]"
|
||||
if cc != "" {
|
||||
ccJSON = `[{"name":"` + cc + `","path":"","session":"","raw":"` + cc + `"}]`
|
||||
}
|
||||
var id uuid.UUID
|
||||
err = db.DB.QueryRowContext(ctx,
|
||||
`INSERT INTO mails (session_id, from_name, to_name, subject, body, cc_list)
|
||||
VALUES ($1, 'sender', $2, 's', 'b', $3) RETURNING mail_id`,
|
||||
sid, recipient, ccJSON).Scan(&id)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func statusOf(t *testing.T, id uuid.UUID) string {
|
||||
t.Helper()
|
||||
var s string
|
||||
if err := db.DB.QueryRowContext(context.Background(),
|
||||
`SELECT status FROM mails WHERE mail_id = $1`, id).Scan(&s); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func TestMarkMailsReadForOnlyOwnMail(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
mine := seedMailTo(t, "bot", "")
|
||||
others := seedMailTo(t, "other", "")
|
||||
|
||||
// 一次请求里混着别人的邮件：自己的标掉，别人的动不了。
|
||||
// 鉴权写在 UPDATE 的 WHERE 里，所以这不是「先查后拒」而是根本改不动。
|
||||
n, err := MarkMailsReadFor(ctx, "bot", []uuid.UUID{mine, others})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("影响行数 = %d，期望 1（只有自己那封）", n)
|
||||
}
|
||||
if statusOf(t, mine) != "read" {
|
||||
t.Fatal("自己的邮件没被标记")
|
||||
}
|
||||
if statusOf(t, others) != "unread" {
|
||||
t.Fatal("别人的邮件被标记了 —— 鉴权失效")
|
||||
}
|
||||
}
|
||||
|
||||
// 重复标记是幂等的：Agent 通常把上一轮列出的 id 原样传回来，
|
||||
// 其中混着已读的不该算错误
|
||||
func TestMarkMailsReadForIsIdempotent(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
id := seedMailTo(t, "bot", "")
|
||||
|
||||
if n, _ := MarkMailsReadFor(ctx, "bot", []uuid.UUID{id}); n != 1 {
|
||||
t.Fatalf("首次应标掉 1 封，实际 %d", n)
|
||||
}
|
||||
n, err := MarkMailsReadFor(ctx, "bot", []uuid.UUID{id})
|
||||
if err != nil {
|
||||
t.Fatalf("重复标记不该报错: %v", err)
|
||||
}
|
||||
if n != 0 {
|
||||
t.Fatalf("重复标记影响行数 = %d，期望 0", n)
|
||||
}
|
||||
}
|
||||
|
||||
// 被抄送的邮件也在收件箱里，也该能标掉
|
||||
func TestMarkMailsReadForCoversCC(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
id := seedMailTo(t, "other", "bot") // 主收件人是 other，bot 被抄送
|
||||
|
||||
n, err := MarkMailsReadFor(ctx, "bot", []uuid.UUID{id})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("被抄送的邮件应可标记，影响行数 = %d", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMarkMailsReadForEmptyList(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
// 空列表直接返回，不该拼出 `IN ()` 这种非法 SQL
|
||||
n, err := MarkMailsReadFor(context.Background(), "bot", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("空列表不该报错: %v", err)
|
||||
}
|
||||
if n != 0 {
|
||||
t.Fatalf("空列表影响行数 = %d", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMarkAllInboxReadForSkipsArchivedAndOthers(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
a := seedMailTo(t, "bot", "")
|
||||
b := seedMailTo(t, "bot", "")
|
||||
others := seedMailTo(t, "other", "")
|
||||
|
||||
// 把 b 所在会话归档：那封在收件箱里根本看不到，
|
||||
// 标掉它只会让「标记了 N 封」与用户看到的对不上
|
||||
var sid uuid.UUID
|
||||
db.DB.QueryRowContext(ctx, `SELECT session_id FROM mails WHERE mail_id = $1`, b).Scan(&sid)
|
||||
db.DB.ExecContext(ctx, `UPDATE sessions SET status = 'archived' WHERE session_id = $1`, sid)
|
||||
|
||||
n, err := MarkAllInboxReadFor(ctx, "bot")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("影响行数 = %d，期望 1（排除归档会话）", n)
|
||||
}
|
||||
if statusOf(t, a) != "read" {
|
||||
t.Fatal("活跃会话里的未读没被标掉")
|
||||
}
|
||||
if statusOf(t, b) != "unread" {
|
||||
t.Fatal("归档会话里的邮件被标掉了")
|
||||
}
|
||||
if statusOf(t, others) != "unread" {
|
||||
t.Fatal("别人的邮件被标掉了")
|
||||
}
|
||||
}
|
||||
@ -13,123 +13,144 @@ import (
|
||||
|
||||
// ---------- 配额 ----------
|
||||
//
|
||||
// 配额限制的是 Agent「主动发信」的次数，不限制收信：
|
||||
// 收信是被动的，卡住收信只会让邮件凭空消失；卡住发信才能真正阻止 Agent 无限自我循环。
|
||||
// **配额是任务的属性，不是 Agent 的属性。**
|
||||
//
|
||||
// agents.max_rounds = 0 表示不限额。used_rounds 单调递增，由管理员显式重置。
|
||||
|
||||
// ErrQuotaExhausted 表示 Agent 的发信配额已用尽。
|
||||
var ErrQuotaExhausted = errors.New("quota exhausted")
|
||||
|
||||
// Quota 是一个 Agent 的配额快照。
|
||||
type Quota struct {
|
||||
// 真正的约束在 `sessions.max_rounds`（见本文件末尾的「会话级往返预算」）：
|
||||
// 每条会话独立计数，人在派活时给、在对话页里随时调。
|
||||
//
|
||||
// `agents` 表这边只剩两样东西：
|
||||
//
|
||||
// default_rounds —— 派给这个 Agent 的**新任务**默认多少个来回。
|
||||
// 不同 Agent 能力不同（跑测试的小工具 vs 重构整个模块），
|
||||
// 默认值分开设才合理。
|
||||
//
|
||||
// used_rounds —— 纯统计，累计发信数。**不再拦任何请求。**
|
||||
// 它原本是「终身额度」：跑满就得管理员手工重置才能再干活，
|
||||
// 而 Agent 是长期在线的 —— 终身额度是错的工具。
|
||||
// 保留是因为「这个 Agent 一共发了多少信」本身有观测价值。
|
||||
//
|
||||
// 防止 Agent 用 `.new` 开一串新会话绕过预算，靠的是**新建会话速率限制**
|
||||
// （见 sessionRateLimiter），而不是终身额度。
|
||||
// AgentStats 是一个 Agent 的配额默认值与累计统计。
|
||||
//
|
||||
// 没有 Remaining / Unlimited 字段：这里不再有「剩余额度」的概念 ——
|
||||
// 额度属于会话（SessionBudget），这里只有「新任务默认多少来回」与「一共发了多少信」。
|
||||
type AgentStats struct {
|
||||
AgentName string `json:"agent_name"`
|
||||
Max int `json:"max_rounds"` // 0 = 不限
|
||||
Used int `json:"used_rounds"`
|
||||
Remaining int `json:"remaining"` // 不限时为 -1
|
||||
Unlimited bool `json:"unlimited"`
|
||||
// DefaultRounds 派给该 Agent 的新任务默认多少个来回（0 = 不限）
|
||||
DefaultRounds int `json:"default_rounds"`
|
||||
// SentTotal 累计发信数（纯统计，不拦请求）
|
||||
SentTotal int `json:"sent_total"`
|
||||
// ActiveSessions 该 Agent 参与的未归档会话数，配合默认值判断设多少合适
|
||||
ActiveSessions int `json:"active_sessions"`
|
||||
}
|
||||
|
||||
func makeQuota(name string, max, used int) Quota {
|
||||
q := Quota{AgentName: name, Max: max, Used: used, Unlimited: max <= 0}
|
||||
if q.Unlimited {
|
||||
q.Remaining = -1
|
||||
return q
|
||||
}
|
||||
if r := max - used; r > 0 {
|
||||
q.Remaining = r
|
||||
}
|
||||
return q
|
||||
}
|
||||
|
||||
// GetQuota 读取某 Agent 的配额状态。
|
||||
func GetQuota(ctx context.Context, agentName string) (Quota, error) {
|
||||
var max, used int
|
||||
err := db.DB.QueryRowContext(ctx,
|
||||
`SELECT max_rounds, used_rounds FROM agents WHERE agent_name = $1`, agentName).Scan(&max, &used)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return Quota{}, fmt.Errorf("agent %q 不存在", agentName)
|
||||
}
|
||||
if err != nil {
|
||||
return Quota{}, err
|
||||
}
|
||||
return makeQuota(agentName, max, used), nil
|
||||
}
|
||||
|
||||
// ConsumeQuota 原子地占用一次发信配额，返回占用后的快照。
|
||||
// DefaultRoundsFor 读取该 Agent 的新任务默认预算。
|
||||
//
|
||||
// 判断与自增必须在同一条 UPDATE 里完成（WHERE used_rounds < max_rounds），
|
||||
// 否则并发发信会双双通过检查再各自 +1，把配额刷穿。
|
||||
// 配额耗尽时返回 ErrQuotaExhausted，同时给出快照供调用方生成提示。
|
||||
func ConsumeQuota(ctx context.Context, agentName string) (Quota, error) {
|
||||
tag, err := db.DB.ExecContext(ctx, `
|
||||
UPDATE agents SET used_rounds = used_rounds + 1
|
||||
WHERE agent_name = $1
|
||||
AND (max_rounds <= 0 OR used_rounds < max_rounds)
|
||||
`, agentName)
|
||||
if err != nil {
|
||||
return Quota{}, err
|
||||
// Agent 不存在时返回全局兜底值而非报错：派活的人不该因为「对方还没注册」
|
||||
// 就拿不到一个合理的默认预算 —— 邮件本来就支持发给尚未上线的收件人。
|
||||
func DefaultRoundsFor(ctx context.Context, agentName string) int {
|
||||
var n int
|
||||
err := db.DB.QueryRowContext(ctx,
|
||||
`SELECT COALESCE(default_rounds, 0) FROM agents WHERE agent_name = $1`,
|
||||
agentName).Scan(&n)
|
||||
if err != nil || n < 0 {
|
||||
return fallbackDefaultRounds
|
||||
}
|
||||
n, _ := tag.RowsAffected()
|
||||
if n == 0 {
|
||||
// 要么 Agent 不存在，要么配额用尽——用快照区分
|
||||
q, qErr := GetQuota(ctx, agentName)
|
||||
if qErr != nil {
|
||||
return Quota{}, qErr
|
||||
}
|
||||
return q, ErrQuotaExhausted
|
||||
}
|
||||
return GetQuota(ctx, agentName)
|
||||
return n
|
||||
}
|
||||
|
||||
// SetQuota 设置某 Agent 的配额上限（0 = 不限）。
|
||||
func SetQuota(ctx context.Context, agentName string, max int) (Quota, error) {
|
||||
if max < 0 {
|
||||
max = 0
|
||||
// fallbackDefaultRounds 是 Agent 未注册时的兜底默认预算。
|
||||
// 与建表默认值保持一致；改这里要同时改两份 schema。
|
||||
const fallbackDefaultRounds = 20
|
||||
|
||||
// SetDefaultRounds 设置该 Agent 的新任务默认预算（0 = 不限）。
|
||||
func SetDefaultRounds(ctx context.Context, agentName string, n int) (AgentStats, error) {
|
||||
if n < 0 {
|
||||
n = 0
|
||||
}
|
||||
tag, err := db.DB.ExecContext(ctx,
|
||||
`UPDATE agents SET max_rounds = $2 WHERE agent_name = $1`, agentName, max)
|
||||
`UPDATE agents SET default_rounds = $2 WHERE agent_name = $1`, agentName, n)
|
||||
if err != nil {
|
||||
return Quota{}, err
|
||||
return AgentStats{}, err
|
||||
}
|
||||
if n, _ := tag.RowsAffected(); n == 0 {
|
||||
return Quota{}, fmt.Errorf("agent %q 不存在", agentName)
|
||||
if k, _ := tag.RowsAffected(); k == 0 {
|
||||
return AgentStats{}, fmt.Errorf("agent %q 不存在", agentName)
|
||||
}
|
||||
return GetQuota(ctx, agentName)
|
||||
return GetAgentStats(ctx, agentName)
|
||||
}
|
||||
|
||||
// ResetQuota 把已用次数归零（配额上限不变）。
|
||||
func ResetQuota(ctx context.Context, agentName string) (Quota, error) {
|
||||
tag, err := db.DB.ExecContext(ctx,
|
||||
`UPDATE agents SET used_rounds = 0 WHERE agent_name = $1`, agentName)
|
||||
// GetAgentStats 读取某 Agent 的默认预算与累计统计。
|
||||
func GetAgentStats(ctx context.Context, agentName string) (AgentStats, error) {
|
||||
var st AgentStats
|
||||
st.AgentName = agentName
|
||||
err := db.DB.QueryRowContext(ctx, `
|
||||
SELECT COALESCE(default_rounds, 0), COALESCE(used_rounds, 0)
|
||||
FROM agents WHERE agent_name = $1`, agentName,
|
||||
).Scan(&st.DefaultRounds, &st.SentTotal)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return AgentStats{}, fmt.Errorf("agent %q 不存在", agentName)
|
||||
}
|
||||
if err != nil {
|
||||
return Quota{}, err
|
||||
return AgentStats{}, err
|
||||
}
|
||||
if n, _ := tag.RowsAffected(); n == 0 {
|
||||
return Quota{}, fmt.Errorf("agent %q 不存在", agentName)
|
||||
}
|
||||
return GetQuota(ctx, agentName)
|
||||
st.ActiveSessions = countActiveSessionsFor(ctx, agentName)
|
||||
return st, nil
|
||||
}
|
||||
|
||||
// ListQuotas 列出所有 Agent 的配额（管理员视图）。
|
||||
func ListQuotas(ctx context.Context) ([]Quota, error) {
|
||||
// countActiveSessionsFor 统计该 Agent 参与的未归档会话数。
|
||||
// 查不出来返回 0：这只是个展示用的数字，不该让整个统计接口失败。
|
||||
func countActiveSessionsFor(ctx context.Context, agentName string) int {
|
||||
var n int
|
||||
err := db.DB.QueryRowContext(ctx, `
|
||||
SELECT COUNT(DISTINCT s.session_id)
|
||||
FROM sessions s
|
||||
JOIN mails m ON m.session_id = s.session_id
|
||||
WHERE s.status <> 'archived'
|
||||
AND (m.from_name = $1 OR m.to_name = $1 OR `+db.CCHas("m.cc_list", 1)+`)
|
||||
`, agentName).Scan(&n)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// BumpSentCount 累加发信统计。
|
||||
//
|
||||
// **绝不拦请求**：它是观测数据，不是额度。返回值只有 error，
|
||||
// 而且调用方应当忽略它 —— 统计写失败不该让一封已经该发出的邮件失败。
|
||||
func BumpSentCount(ctx context.Context, agentName string) {
|
||||
_, _ = db.DB.ExecContext(ctx,
|
||||
`UPDATE agents SET used_rounds = COALESCE(used_rounds, 0) + 1 WHERE agent_name = $1`,
|
||||
agentName)
|
||||
}
|
||||
|
||||
// ListAgentStats 列出所有 Agent 的默认预算与统计（管理员视图）。
|
||||
func ListAgentStats(ctx context.Context) ([]AgentStats, error) {
|
||||
rows, err := db.DB.QueryContext(ctx,
|
||||
`SELECT agent_name, max_rounds, used_rounds FROM agents ORDER BY agent_name`)
|
||||
`SELECT agent_name, COALESCE(default_rounds, 0), COALESCE(used_rounds, 0)
|
||||
FROM agents ORDER BY agent_name`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := []Quota{}
|
||||
out := []AgentStats{}
|
||||
for rows.Next() {
|
||||
var name string
|
||||
var max, used int
|
||||
if err := rows.Scan(&name, &max, &used); err != nil {
|
||||
var st AgentStats
|
||||
if err := rows.Scan(&st.AgentName, &st.DefaultRounds, &st.SentTotal); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, makeQuota(name, max, used))
|
||||
out = append(out, st)
|
||||
}
|
||||
return out, rows.Err()
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// 会话数逐个查：Agent 数量是个位数到几十，不值得为它写一个 GROUP BY 的联合查询
|
||||
for i := range out {
|
||||
out[i].ActiveSessions = countActiveSessionsFor(ctx, out[i].AgentName)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ---------- 转发 ----------
|
||||
|
||||
@ -2,10 +2,8 @@ package repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
@ -30,141 +28,95 @@ func setupTestDB(t *testing.T) {
|
||||
func seedAgent(t *testing.T, name string, max int) {
|
||||
t.Helper()
|
||||
_, err := db.DB.ExecContext(context.Background(),
|
||||
`INSERT INTO agents (agent_name, secret, platform, max_rounds) VALUES ($1, 'x', 'test', $2)`,
|
||||
`INSERT INTO agents (agent_name, secret, platform, default_rounds) VALUES ($1, 'x', 'test', $2)`,
|
||||
name, max)
|
||||
if err != nil {
|
||||
t.Fatalf("seed agent: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConsumeQuotaCountsDown(t *testing.T) {
|
||||
// default_rounds 是「派给这个 Agent 的新任务默认多少个来回」，
|
||||
// 不是会拦请求的终身额度 —— 真正的额度在 sessions.max_rounds 上。
|
||||
func TestDefaultRoundsRoundTrip(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
seedAgent(t, "bot", 3)
|
||||
seedAgent(t, "bot", 0)
|
||||
ctx := context.Background()
|
||||
|
||||
for i := 1; i <= 3; i++ {
|
||||
q, err := ConsumeQuota(ctx, "bot")
|
||||
if err != nil {
|
||||
t.Fatalf("第 %d 次占用失败: %v", i, err)
|
||||
}
|
||||
if q.Used != i || q.Remaining != 3-i {
|
||||
t.Errorf("第 %d 次: used=%d remaining=%d, want used=%d remaining=%d",
|
||||
i, q.Used, q.Remaining, i, 3-i)
|
||||
}
|
||||
st, err := SetDefaultRounds(ctx, "bot", 15)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if st.DefaultRounds != 15 {
|
||||
t.Fatalf("default_rounds = %d，期望 15", st.DefaultRounds)
|
||||
}
|
||||
if got := DefaultRoundsFor(ctx, "bot"); got != 15 {
|
||||
t.Fatalf("DefaultRoundsFor = %d，期望 15", got)
|
||||
}
|
||||
|
||||
q, err := ConsumeQuota(ctx, "bot")
|
||||
if !errors.Is(err, ErrQuotaExhausted) {
|
||||
t.Fatalf("第 4 次应耗尽，得到 err=%v", err)
|
||||
// 负数归一为 0（不限），而不是造出一个永远发不出信的默认值
|
||||
if st, err = SetDefaultRounds(ctx, "bot", -3); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// 耗尽时仍要给出快照，调用方才能在错误文案里写清 used/max
|
||||
if q.Used != 3 || q.Max != 3 {
|
||||
t.Errorf("耗尽时快照不对: %+v", q)
|
||||
if st.DefaultRounds != 0 {
|
||||
t.Fatalf("负数应归一为 0，实际 %d", st.DefaultRounds)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConsumeQuotaUnlimited(t *testing.T) {
|
||||
// 未注册的 Agent 取默认预算时给兜底值而不是报错：
|
||||
// 派活的人不该因为「对方还没上线」就拿不到一个合理默认值 ——
|
||||
// 邮件本来就支持发给尚未上线的收件人。
|
||||
func TestDefaultRoundsForUnknownAgentFallsBack(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
seedAgent(t, "free", 0) // 0 = 不限
|
||||
if got := DefaultRoundsFor(context.Background(), "ghost"); got != fallbackDefaultRounds {
|
||||
t.Fatalf("未注册 Agent 应回落到 %d，实际 %d", fallbackDefaultRounds, got)
|
||||
}
|
||||
}
|
||||
|
||||
// BumpSentCount 是纯统计：只累加，绝不拦请求，也绝不返回错误
|
||||
func TestBumpSentCountOnlyCounts(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
seedAgent(t, "bot", 0)
|
||||
ctx := context.Background()
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
q, err := ConsumeQuota(ctx, "free")
|
||||
if err != nil {
|
||||
t.Fatalf("不限额时不应失败: %v", err)
|
||||
}
|
||||
if !q.Unlimited || q.Remaining != -1 {
|
||||
t.Errorf("不限额快照不对: %+v", q)
|
||||
}
|
||||
BumpSentCount(ctx, "bot")
|
||||
}
|
||||
}
|
||||
|
||||
// 配额的核心保证：并发发信不能把额度刷穿。
|
||||
// 判断与自增若分成两步（先 SELECT 再 UPDATE），并发下两个请求会双双通过检查。
|
||||
func TestConsumeQuotaConcurrentDoesNotOverdraw(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
const limit = 10
|
||||
seedAgent(t, "racer", limit)
|
||||
ctx := context.Background()
|
||||
|
||||
const attempts = 40
|
||||
var (
|
||||
wg sync.WaitGroup
|
||||
mu sync.Mutex
|
||||
granted int
|
||||
)
|
||||
for i := 0; i < attempts; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
if _, err := ConsumeQuota(ctx, "racer"); err == nil {
|
||||
mu.Lock()
|
||||
granted++
|
||||
mu.Unlock()
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
if granted != limit {
|
||||
t.Errorf("并发 %d 次请求在上限 %d 下放行了 %d 次", attempts, limit, granted)
|
||||
}
|
||||
|
||||
q, err := GetQuota(ctx, "racer")
|
||||
st, err := GetAgentStats(ctx, "bot")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if q.Used != limit {
|
||||
t.Errorf("used_rounds = %d，应恰好等于上限 %d（未刷穿也未少记）", q.Used, limit)
|
||||
if st.SentTotal != 5 {
|
||||
t.Fatalf("累计发信 = %d，期望 5", st.SentTotal)
|
||||
}
|
||||
|
||||
// 不存在的 Agent 也不该 panic 或报错 —— 它只是没有行可更新
|
||||
BumpSentCount(ctx, "ghost")
|
||||
}
|
||||
|
||||
func TestGetAgentStatsUnknownAgent(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
if _, err := GetAgentStats(context.Background(), "ghost"); err == nil {
|
||||
t.Fatal("不存在的 Agent 应报错")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetAndResetQuota(t *testing.T) {
|
||||
func TestListAgentStats(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
seedAgent(t, "bot", 2)
|
||||
seedAgent(t, "alpha", 0)
|
||||
seedAgent(t, "beta", 0)
|
||||
ctx := context.Background()
|
||||
SetDefaultRounds(ctx, "alpha", 5)
|
||||
|
||||
if _, err := ConsumeQuota(ctx, "bot"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
q, err := SetQuota(ctx, "bot", 5)
|
||||
list, err := ListAgentStats(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// 调上限不应清掉已用次数
|
||||
if q.Max != 5 || q.Used != 1 || q.Remaining != 4 {
|
||||
t.Errorf("SetQuota 后 %+v，want max=5 used=1 remaining=4", q)
|
||||
if len(list) != 2 {
|
||||
t.Fatalf("应有 2 个 Agent，实际 %d", len(list))
|
||||
}
|
||||
|
||||
q, err = ResetQuota(ctx, "bot")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if q.Used != 0 || q.Remaining != 5 {
|
||||
t.Errorf("ResetQuota 后 %+v，want used=0 remaining=5", q)
|
||||
}
|
||||
|
||||
// 负数上限归一为 0（不限），而不是造出一个永远发不出信的 Agent
|
||||
if q, err = SetQuota(ctx, "bot", -3); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if !q.Unlimited {
|
||||
t.Errorf("负数上限应视为不限，得到 %+v", q)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQuotaUnknownAgent(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := GetQuota(ctx, "ghost"); err == nil {
|
||||
t.Error("不存在的 Agent 应报错")
|
||||
}
|
||||
if _, err := ConsumeQuota(ctx, "ghost"); err == nil {
|
||||
t.Error("不存在的 Agent 占用配额应报错")
|
||||
}
|
||||
if errors.Is(func() error { _, e := ConsumeQuota(ctx, "ghost"); return e }(), ErrQuotaExhausted) {
|
||||
t.Error("不存在的 Agent 不该被报成「配额耗尽」")
|
||||
// 按名字排序，alpha 在前
|
||||
if list[0].AgentName != "alpha" || list[0].DefaultRounds != 5 {
|
||||
t.Fatalf("alpha 的记录不对：%+v", list[0])
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -6,6 +6,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
@ -17,6 +18,8 @@ import (
|
||||
|
||||
func CreateOrUpdateAgent(ctx context.Context, name, secret, platform string, workspaces []models.Workspace) error {
|
||||
wsJSON, _ := json.Marshal(workspaces)
|
||||
// 注意 DO UPDATE 里【不】碰 default_rounds：
|
||||
// 那是管理员配的值，Agent 重启重新注册不应该把它冲回默认。
|
||||
_, err := db.DB.ExecContext(ctx, `
|
||||
INSERT INTO agents (agent_name, secret, workspaces, platform, status, last_seen)
|
||||
VALUES ($1, $2, $3, $4, 'online', NOW())
|
||||
@ -41,7 +44,10 @@ func HeartbeatAgent(ctx context.Context, agentName string) (int, error) {
|
||||
}
|
||||
|
||||
func ListAgents(ctx context.Context, statusFilter string) ([]models.Agent, error) {
|
||||
q := `SELECT agent_id, agent_name, workspaces, platform, status FROM agents`
|
||||
// 带上 default_rounds：前端补全收件人时要显示「派给它的任务默认几个来回」，
|
||||
// 否则人得先去管理员页查一遍才敢派活。
|
||||
q := `SELECT agent_id, agent_name, workspaces, platform, status,
|
||||
COALESCE(default_rounds, 0) FROM agents`
|
||||
args := []any{}
|
||||
if statusFilter != "" {
|
||||
q += ` WHERE status = $1`
|
||||
@ -59,7 +65,8 @@ func ListAgents(ctx context.Context, statusFilter string) ([]models.Agent, error
|
||||
for rows.Next() {
|
||||
var a models.Agent
|
||||
var wsJSON []byte
|
||||
if err := rows.Scan(&a.ID, &a.Name, &wsJSON, &a.Platform, &a.Status); err != nil {
|
||||
if err := rows.Scan(&a.ID, &a.Name, &wsJSON, &a.Platform, &a.Status,
|
||||
&a.DefaultRounds); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if wsJSON != nil {
|
||||
@ -177,7 +184,7 @@ func ListSessions(ctx context.Context, statusFilter string, limit int) ([]models
|
||||
}
|
||||
sessions = append(sessions, s)
|
||||
}
|
||||
return sessions, nil
|
||||
return sessions, rows.Err()
|
||||
}
|
||||
|
||||
// ---------- Mail ----------
|
||||
@ -190,9 +197,13 @@ func CreateMail(ctx context.Context, sessionID uuid.UUID, parentMailID *uuid.UUI
|
||||
ccJSON, _ := json.Marshal(ccList)
|
||||
var id uuid.UUID
|
||||
err := db.DB.QueryRowContext(ctx,
|
||||
// created_at 显式给 NOW()：SQLite 的 DEFAULT CURRENT_TIMESTAMP 只有秒精度，
|
||||
// 同秒插入的多封邮件排序不确定（「会话里最早/最后那封」都会取错行）。
|
||||
// 改 schema 的默认值只对新库生效 —— CREATE TABLE IF NOT EXISTS 不改已存在的表，
|
||||
// 而 SQLite 没有 ALTER COLUMN，因此这里显式传。
|
||||
`INSERT INTO mails (session_id, parent_mail_id, from_name, from_workspace,
|
||||
to_name, to_workspace, subject, body, cc_list)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING mail_id`,
|
||||
to_name, to_workspace, subject, body, cc_list, created_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NOW()) RETURNING mail_id`,
|
||||
sessionID, parentMailID, fromName, fromWorkspace, toName, toWorkspace, subject, body, ccJSON,
|
||||
).Scan(&id)
|
||||
return id, err
|
||||
@ -203,8 +214,8 @@ func CreatePermissionMail(ctx context.Context, sessionID uuid.UUID, fromName, to
|
||||
optsJSON, _ := json.Marshal(options)
|
||||
var id uuid.UUID
|
||||
err := db.DB.QueryRowContext(ctx,
|
||||
`INSERT INTO mails (session_id, from_name, to_name, subject, body, mail_type, permission_options)
|
||||
VALUES ($1, $2, $3, $4, $5, 'permission_request', $6) RETURNING mail_id`,
|
||||
`INSERT INTO mails (session_id, from_name, to_name, subject, body, mail_type, permission_options, created_at)
|
||||
VALUES ($1, $2, $3, $4, $5, 'permission_request', $6, NOW()) RETURNING mail_id`,
|
||||
sessionID, fromName, toUser, "权限请求: "+question, body, optsJSON,
|
||||
).Scan(&id)
|
||||
return id, err
|
||||
@ -218,8 +229,8 @@ func CreateDecisionMail(ctx context.Context, sessionID uuid.UUID, parentMailID u
|
||||
body = fmt.Sprintf("%s\n\n备注: %s", decision, note)
|
||||
}
|
||||
err := db.DB.QueryRowContext(ctx,
|
||||
`INSERT INTO mails (session_id, parent_mail_id, from_name, to_name, subject, body)
|
||||
VALUES ($1, $2, $3, $4, $5, $6) RETURNING mail_id`,
|
||||
`INSERT INTO mails (session_id, parent_mail_id, from_name, to_name, subject, body, created_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, NOW()) RETURNING mail_id`,
|
||||
sessionID, parentMailID, fromUser, toAgent, "Re: 权限请求 - "+decision, body,
|
||||
).Scan(&id)
|
||||
return id, err
|
||||
@ -284,7 +295,7 @@ func ListInbox(ctx context.Context, agentName, status string, limit int) ([]mode
|
||||
q += ` AND m.status = $2`
|
||||
args = append(args, status)
|
||||
}
|
||||
q += ` ORDER BY m.created_at DESC`
|
||||
q += ` ORDER BY m.created_at DESC, m.mail_id DESC`
|
||||
if limit > 0 {
|
||||
q += fmt.Sprintf(` LIMIT %d`, limit)
|
||||
}
|
||||
@ -323,7 +334,7 @@ func ListInbox(ctx context.Context, agentName, status string, limit int) ([]mode
|
||||
}
|
||||
mails = append(mails, m)
|
||||
}
|
||||
return mails, nil
|
||||
return mails, rows.Err()
|
||||
}
|
||||
|
||||
func CountUnread(ctx context.Context, agentName string) (int, error) {
|
||||
@ -348,7 +359,7 @@ func GetSessionMails(ctx context.Context, sessionID uuid.UUID) ([]models.Mail, e
|
||||
FROM mails m
|
||||
JOIN sessions s ON m.session_id = s.session_id
|
||||
WHERE m.session_id = $1
|
||||
ORDER BY m.created_at ASC`, sessionID)
|
||||
ORDER BY m.created_at ASC, m.mail_id ASC`, sessionID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@ -376,7 +387,7 @@ func GetSessionMails(ctx context.Context, sessionID uuid.UUID) ([]models.Mail, e
|
||||
}
|
||||
mails = append(mails, m)
|
||||
}
|
||||
return mails, nil
|
||||
return mails, rows.Err()
|
||||
}
|
||||
|
||||
// ---------- Permission ----------
|
||||
@ -678,6 +689,22 @@ type Contact struct {
|
||||
MailCount int `json:"mail_count"`
|
||||
UnreadCount int `json:"unread_count"`
|
||||
LastActivity time.Time `json:"last_activity"`
|
||||
|
||||
// Subject 是会话主题（多由 Agent 平台的模型生成的摘要）。
|
||||
// 卡片视图要靠它回答「这条线索在干什么」—— 光有 name@path.alias
|
||||
// 只能看出跟谁在聊，看不出在聊什么。
|
||||
Subject string `json:"subject"`
|
||||
|
||||
// MaxRounds/UsedRounds 是本任务的往返预算（0 = 不限）。
|
||||
// 列表上直接可见，才不用点进每条会话去查哪件事快跑满了。
|
||||
MaxRounds int `json:"max_rounds"`
|
||||
UsedRounds int `json:"used_rounds"`
|
||||
|
||||
// LastFrom/LastPreview 是最后一封邮件的发件人与正文摘要，
|
||||
// 卡片视图用它显示「最新进展」——列表视图只显示地址时，
|
||||
// 人必须逐条点开才知道哪条有新动静。
|
||||
LastFrom string `json:"last_from"`
|
||||
LastPreview string `json:"last_preview"`
|
||||
}
|
||||
|
||||
// ListContactsFor 按 (agent, path, session) 聚合出联系人清单。
|
||||
@ -723,6 +750,16 @@ func ListContactsFor(ctx context.Context, forUser string, archived bool) ([]Cont
|
||||
)`
|
||||
}
|
||||
|
||||
// 最后一封邮件用关联子查询取，不再 JOIN 一次：
|
||||
// 两个 JOIN（最早一封 + 最新一封）在 SQLite 下要写两段方言分支，
|
||||
// 而这里每个会话只多两次索引查找（idx_mails_session 已有）。
|
||||
lastMail := `(
|
||||
SELECT mail_id FROM mails
|
||||
WHERE session_id = s.session_id
|
||||
ORDER BY created_at DESC, mail_id DESC
|
||||
LIMIT 1
|
||||
)`
|
||||
|
||||
rows, err := db.DB.QueryContext(ctx, `
|
||||
SELECT s.session_id,
|
||||
COALESCE(NULLIF(m.to_name, 'human'), m.from_name) AS agent_name,
|
||||
@ -731,7 +768,12 @@ func ListContactsFor(ctx context.Context, forUser string, archived bool) ([]Cont
|
||||
s.status,
|
||||
(SELECT COUNT(*) FROM mails x WHERE x.session_id = s.session_id),
|
||||
(SELECT COUNT(*) FROM mails x WHERE x.session_id = s.session_id AND x.status = 'unread'),
|
||||
s.updated_at
|
||||
s.updated_at,
|
||||
s.subject,
|
||||
COALESCE(s.max_rounds, 0),
|
||||
COALESCE(s.used_rounds, 0),
|
||||
COALESCE((SELECT from_name FROM mails WHERE mail_id = `+lastMail+`), ''),
|
||||
COALESCE((SELECT body FROM mails WHERE mail_id = `+lastMail+`), '')
|
||||
FROM sessions s`+firstMail+`
|
||||
WHERE s.status `+op+` 'archived'`+scope+`
|
||||
ORDER BY s.updated_at DESC
|
||||
@ -745,16 +787,32 @@ func ListContactsFor(ctx context.Context, forUser string, archived bool) ([]Cont
|
||||
for rows.Next() {
|
||||
var c Contact
|
||||
if err := rows.Scan(&c.SessionID, &c.AgentName, &c.Path, &c.SessionAlias,
|
||||
&c.Status, &c.MailCount, &c.UnreadCount, &c.LastActivity); err != nil {
|
||||
&c.Status, &c.MailCount, &c.UnreadCount, &c.LastActivity,
|
||||
&c.Subject, &c.MaxRounds, &c.UsedRounds,
|
||||
&c.LastFrom, &c.LastPreview); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.Address = c.AgentName + "@" + c.Path
|
||||
if c.SessionAlias != "" {
|
||||
c.Address += "." + c.SessionAlias
|
||||
}
|
||||
// 正文只留一段摘要：卡片上放不下全文，而整个列表带全文可能几百 KB。
|
||||
// 按 rune 截断而非字节 —— 中文 3 字节/字，裸切会产生 U+FFFD。
|
||||
c.LastPreview = previewRunes(c.LastPreview, 90)
|
||||
contacts = append(contacts, c)
|
||||
}
|
||||
return contacts, nil
|
||||
return contacts, rows.Err()
|
||||
}
|
||||
|
||||
// previewRunes 按字符数截断，附省略号。
|
||||
// 不按字节切：中文一字三字节，裸切会在末尾留半个字符（渲染成 U+FFFD）。
|
||||
func previewRunes(s string, n int) string {
|
||||
s = strings.TrimSpace(s)
|
||||
rs := []rune(s)
|
||||
if len(rs) <= n {
|
||||
return s
|
||||
}
|
||||
return string(rs[:n]) + "…"
|
||||
}
|
||||
|
||||
// ArchiveSession 归档一个会话（邮箱界面不再展示，数据保留）
|
||||
@ -837,7 +895,7 @@ func SuggestSessionsFor(ctx context.Context, forUser, peerName, path string) ([]
|
||||
out = append(out, alias)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// ListSentBy 列出某发件人发出的邮件（发件箱），排除已归档会话
|
||||
@ -850,7 +908,7 @@ func ListSentBy(ctx context.Context, fromName string, limit int) ([]models.Mail,
|
||||
FROM mails m
|
||||
JOIN sessions s ON m.session_id = s.session_id
|
||||
WHERE m.from_name = $1 AND s.status <> 'archived'
|
||||
ORDER BY m.created_at DESC
|
||||
ORDER BY m.created_at DESC, m.mail_id DESC
|
||||
LIMIT $2
|
||||
`, fromName, limit)
|
||||
if err != nil {
|
||||
@ -885,7 +943,7 @@ func ListSentBy(ctx context.Context, fromName string, limit int) ([]models.Mail,
|
||||
}
|
||||
mails = append(mails, m)
|
||||
}
|
||||
return mails, nil
|
||||
return mails, rows.Err()
|
||||
}
|
||||
|
||||
// ListPendingPermissionsFor 列出待决权限请求；forUser 非空时只列发给该用户的
|
||||
@ -956,13 +1014,16 @@ func ListSessionsFor(ctx context.Context, forUser string, limit int) ([]models.S
|
||||
sessions := []models.Session{}
|
||||
for rows.Next() {
|
||||
var s models.Session
|
||||
// 列数必须与上面的 SELECT 一一对应 —— 预算两列曾经只加进了查询而没加进这里，
|
||||
// 结果 /me/sessions 整个 500，联系人栅拉不到任何数据。
|
||||
if err := rows.Scan(&s.ID, &s.Alias, &s.FromAgent, &s.Subject, &s.Status,
|
||||
&s.OwnerUserID, &s.CreatedAt, &s.UpdatedAt, &s.MailCount); err != nil {
|
||||
&s.OwnerUserID, &s.CreatedAt, &s.UpdatedAt,
|
||||
&s.MaxRounds, &s.UsedRounds, &s.MailCount); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sessions = append(sessions, s)
|
||||
}
|
||||
return sessions, nil
|
||||
return sessions, rows.Err()
|
||||
}
|
||||
|
||||
// CountUnreadInSession 统计某人在某会话内的未读数（含被抄送）
|
||||
@ -1044,3 +1105,57 @@ func DismissRenameProposal(ctx context.Context, sessionID uuid.UUID, alias strin
|
||||
alias, sessionID)
|
||||
return err
|
||||
}
|
||||
|
||||
// MarkMailsReadFor 把一批邮件标记为某收件人已读，返回实际影响的行数。
|
||||
//
|
||||
// **鉴权写进 WHERE 而不是先查后改**：`to_name = $1 OR cc 含 $1` 直接放在
|
||||
// UPDATE 条件里，于是「不是发给我的邮件」根本改不动 —— 既省掉一次查询，
|
||||
// 也没有「查完到改之间邮件被转走」的时间窗。
|
||||
//
|
||||
// 已经是 read 的不计入影响行数（`status = 'unread'` 条件），
|
||||
// 调用方据此知道这次真正标掉了几封。
|
||||
func MarkMailsReadFor(ctx context.Context, recipient string, ids []uuid.UUID) (int, error) {
|
||||
if len(ids) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// IN 子句的占位符按方言编号（SQLite 与 PG 都认 $N）。
|
||||
// 不用一条条 UPDATE：一次网络往返 + 一次事务，SQLite 单写者下差别明显。
|
||||
ph := make([]string, len(ids))
|
||||
args := make([]any, 0, len(ids)+1)
|
||||
args = append(args, recipient)
|
||||
for i, id := range ids {
|
||||
ph[i] = fmt.Sprintf("$%d", i+2)
|
||||
args = append(args, id)
|
||||
}
|
||||
|
||||
res, err := db.DB.ExecContext(ctx, `
|
||||
UPDATE mails SET status = 'read'
|
||||
WHERE mail_id IN (`+strings.Join(ph, ",")+`)
|
||||
AND status = 'unread'
|
||||
AND (to_name = $1 OR `+db.CCHas("cc_list", 1)+`)
|
||||
`, args...)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
n, _ := res.RowsAffected()
|
||||
return int(n), nil
|
||||
}
|
||||
|
||||
// MarkAllInboxReadFor 把某收件人收件箱里全部未读标为已读，返回影响行数。
|
||||
//
|
||||
// 排除已归档会话：那些邮件在收件箱里根本看不到，
|
||||
// 标掉它们只会让「标记了 N 封」这个数字与用户看到的对不上。
|
||||
func MarkAllInboxReadFor(ctx context.Context, recipient string) (int, error) {
|
||||
res, err := db.DB.ExecContext(ctx, `
|
||||
UPDATE mails SET status = 'read'
|
||||
WHERE status = 'unread'
|
||||
AND (to_name = $1 OR `+db.CCHas("cc_list", 1)+`)
|
||||
AND session_id IN (SELECT session_id FROM sessions WHERE status <> 'archived')
|
||||
`, recipient)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
n, _ := res.RowsAffected()
|
||||
return int(n), nil
|
||||
}
|
||||
|
||||
228
gateway/internal/repo/repo_test.go
Normal file
228
gateway/internal/repo/repo_test.go
Normal file
@ -0,0 +1,228 @@
|
||||
package repo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/agentmail/gateway/internal/db"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// ListSessionsFor 的 Scan 列数必须与 SELECT 一致。
|
||||
//
|
||||
// 这个测试存在的理由：预算两列（max_rounds/used_rounds）加进了 SELECT 却忘了加进
|
||||
// Scan，于是 /me/sessions 整个 500 —— 联系人栏一条数据都拉不到，
|
||||
// 而错误信息只是 "Failed to list sessions"，看不出是列数不匹配。
|
||||
// 列数错位是纯结构问题，一个最小用例就能钉住。
|
||||
func TestListSessionsForScanMatchesSelect(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
|
||||
if _, err := db.DB.ExecContext(context.Background(),
|
||||
`INSERT INTO users (username, display_name, password_hash, role)
|
||||
VALUES ('alice', 'Alice', 'x', 'user')`); err != nil {
|
||||
t.Fatalf("seed user: %v", err)
|
||||
}
|
||||
|
||||
sid := seedSessionRow(t, "list-scan")
|
||||
if _, err := db.DB.ExecContext(context.Background(),
|
||||
`UPDATE sessions SET max_rounds = 7, used_rounds = 3 WHERE session_id = $1`,
|
||||
sid); err != nil {
|
||||
t.Fatalf("set budget: %v", err)
|
||||
}
|
||||
seedMailIn(t, sid, "alice", "opencode", "hello")
|
||||
|
||||
// 无过滤（管理员 all=true 走这条）
|
||||
all, err := ListSessionsFor(context.Background(), "", 50)
|
||||
if err != nil {
|
||||
t.Fatalf("列出全部会话失败: %v", err)
|
||||
}
|
||||
if len(all) != 1 {
|
||||
t.Fatalf("应有 1 个会话，实际 %d", len(all))
|
||||
}
|
||||
// 预算两列要真的读出来，不是零值
|
||||
if all[0].MaxRounds != 7 || all[0].UsedRounds != 3 {
|
||||
t.Errorf("预算未读出：max=%d used=%d（期望 7/3）",
|
||||
all[0].MaxRounds, all[0].UsedRounds)
|
||||
}
|
||||
if all[0].MailCount != 1 {
|
||||
t.Errorf("邮件数应为 1，实际 %d —— 列顺序可能错位", all[0].MailCount)
|
||||
}
|
||||
|
||||
// 带用户过滤（普通用户走这条，SQL 分支不同，要分别验）
|
||||
mine, err := ListSessionsFor(context.Background(), "alice", 50)
|
||||
if err != nil {
|
||||
t.Fatalf("列出自己的会话失败: %v", err)
|
||||
}
|
||||
if len(mine) != 1 {
|
||||
t.Fatalf("alice 参与过该会话，应能看到，实际 %d 个", len(mine))
|
||||
}
|
||||
if mine[0].MaxRounds != 7 || mine[0].MailCount != 1 {
|
||||
t.Errorf("过滤分支的列顺序错位：%+v", mine[0])
|
||||
}
|
||||
|
||||
// 与自己无关的人看不到
|
||||
other, err := ListSessionsFor(context.Background(), "bob", 50)
|
||||
if err != nil {
|
||||
t.Fatalf("列出 bob 的会话失败: %v", err)
|
||||
}
|
||||
if len(other) != 0 {
|
||||
t.Errorf("bob 未参与该会话，不该看到，实际 %d 个", len(other))
|
||||
}
|
||||
|
||||
// 归档会话不出现在列表里
|
||||
if _, err := db.DB.ExecContext(context.Background(),
|
||||
`UPDATE sessions SET status = 'archived' WHERE session_id = $1`, sid); err != nil {
|
||||
t.Fatalf("archive: %v", err)
|
||||
}
|
||||
after, err := ListSessionsFor(context.Background(), "", 50)
|
||||
if err != nil {
|
||||
t.Fatalf("归档后列出失败: %v", err)
|
||||
}
|
||||
if len(after) != 0 {
|
||||
t.Errorf("归档会话不该出现在列表里，实际 %d 个", len(after))
|
||||
}
|
||||
}
|
||||
|
||||
// 工作列表卡片视图需要「这条线索在干什么 / 还剩几个来回 / 最新进展是什么」，
|
||||
// 这些都从 ListContactsFor 一次取回 —— 否则卡片要为每条会话再打一次库。
|
||||
func TestListContactsForCardFields(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
|
||||
if _, err := db.DB.ExecContext(context.Background(),
|
||||
`INSERT INTO users (username, display_name, password_hash, role)
|
||||
VALUES ('alice', 'Alice', 'x', 'user')`); err != nil {
|
||||
t.Fatalf("seed user: %v", err)
|
||||
}
|
||||
|
||||
sid := seedSessionRow(t, "card-fields")
|
||||
if _, err := db.DB.ExecContext(context.Background(),
|
||||
`UPDATE sessions SET subject = '缓存层选型评估', max_rounds = 5, used_rounds = 2
|
||||
WHERE session_id = $1`, sid); err != nil {
|
||||
t.Fatalf("set session: %v", err)
|
||||
}
|
||||
|
||||
// 三封：最早一封决定联系人身份，最后一封决定「最新进展」
|
||||
seedMailIn(t, sid, "alice", "opencode", "第一封")
|
||||
seedMailIn(t, sid, "opencode", "alice", "第二封")
|
||||
last := seedMailIn(t, sid, "opencode", "alice", "第三封")
|
||||
if _, err := db.DB.ExecContext(context.Background(),
|
||||
`UPDATE mails SET body = '已经跑完压测，Redis 方案在这个负载下明显更稳。'
|
||||
WHERE mail_id = $1`, last); err != nil {
|
||||
t.Fatalf("set body: %v", err)
|
||||
}
|
||||
|
||||
got, err := ListContactsFor(context.Background(), "alice", false)
|
||||
if err != nil {
|
||||
t.Fatalf("列出联系人失败: %v", err)
|
||||
}
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("应有 1 个联系人，实际 %d", len(got))
|
||||
}
|
||||
c := got[0]
|
||||
|
||||
if c.Subject != "缓存层选型评估" {
|
||||
t.Errorf("主题未带回：%q", c.Subject)
|
||||
}
|
||||
if c.MaxRounds != 5 || c.UsedRounds != 2 {
|
||||
t.Errorf("预算未带回：%d/%d（期望 5/2）", c.UsedRounds, c.MaxRounds)
|
||||
}
|
||||
// 最新进展取的是【最后】一封，不是第一封
|
||||
if c.LastFrom != "opencode" {
|
||||
t.Errorf("最新发件人应为 opencode，实际 %q", c.LastFrom)
|
||||
}
|
||||
if !strings.Contains(c.LastPreview, "Redis 方案") {
|
||||
t.Errorf("最新摘要应来自最后一封，实际 %q", c.LastPreview)
|
||||
}
|
||||
// 联系人身份仍取最早一封的对端
|
||||
if c.AgentName != "opencode" {
|
||||
t.Errorf("联系人应为 opencode，实际 %q", c.AgentName)
|
||||
}
|
||||
if c.MailCount != 3 {
|
||||
t.Errorf("邮件数应为 3，实际 %d —— 列顺序可能错位", c.MailCount)
|
||||
}
|
||||
if c.Address != "opencode@.card-fields" && !strings.HasSuffix(c.Address, ".card-fields") {
|
||||
t.Errorf("地址应带会话别名，实际 %q", c.Address)
|
||||
}
|
||||
}
|
||||
|
||||
// 摘要按字符截断，不按字节 —— 中文一字三字节，裸切会留半个字符。
|
||||
func TestPreviewRunes(t *testing.T) {
|
||||
cases := []struct {
|
||||
in string
|
||||
n int
|
||||
want string
|
||||
}{
|
||||
{"短文本", 10, "短文本"},
|
||||
{" 两边有空白 ", 10, "两边有空白"},
|
||||
{"", 5, ""},
|
||||
{"一二三四五六", 3, "一二三…"},
|
||||
{"abcdefgh", 3, "abc…"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := previewRunes(c.in, c.n); got != c.want {
|
||||
t.Errorf("previewRunes(%q, %d) = %q，期望 %q", c.in, c.n, got, c.want)
|
||||
}
|
||||
}
|
||||
|
||||
// 截断结果必须是合法 UTF-8（不含替换字符）
|
||||
long := strings.Repeat("汉字", 200)
|
||||
got := previewRunes(long, 90)
|
||||
if strings.ContainsRune(got, '\uFFFD') {
|
||||
t.Error("截断产生了 U+FFFD，说明按字节切了")
|
||||
}
|
||||
if n := len([]rune(got)); n != 91 { // 90 + 省略号
|
||||
t.Errorf("截断后应为 90 字符 + 省略号，实际 %d 字符", n)
|
||||
}
|
||||
}
|
||||
|
||||
// 时间戳精度回归：SQLite 的 CURRENT_TIMESTAMP 只有秒，同秒插入的多行排序不确定，
|
||||
// 「会话里最早那封」（决定联系人身份）与「最后那封」（决定最新进展）都会取错。
|
||||
// NOW() 现在返回毫秒精度，且 mails 的 INSERT 显式传它 —— 这两点都要钉住。
|
||||
func TestMailTimestampSubSecond(t *testing.T) {
|
||||
setupTestDB(t)
|
||||
sid := seedSessionRow(t, "ts-precision")
|
||||
|
||||
// 连续插 8 封（不显式给时间戳，走 CreateMail 里的 NOW()）
|
||||
ids := make([]uuid.UUID, 0, 8)
|
||||
for i := 0; i < 8; i++ {
|
||||
id, err := CreateMail(context.Background(), sid, nil,
|
||||
"alice", "", "opencode", "", fmt.Sprintf("第%d封", i), "body", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("创建邮件 %d 失败: %v", i, err)
|
||||
}
|
||||
ids = append(ids, id)
|
||||
}
|
||||
|
||||
// 至少要出现亚秒差异，否则说明 NOW() 又退回秒精度
|
||||
var distinct int
|
||||
if err := db.DB.QueryRowContext(context.Background(),
|
||||
`SELECT COUNT(DISTINCT created_at) FROM mails WHERE session_id = $1`,
|
||||
sid).Scan(&distinct); err != nil {
|
||||
t.Fatalf("统计不同时间戳失败: %v", err)
|
||||
}
|
||||
if distinct < 2 {
|
||||
var sample string
|
||||
db.DB.QueryRowContext(context.Background(),
|
||||
`SELECT CAST(created_at AS TEXT) FROM mails WHERE session_id = $1 LIMIT 1`,
|
||||
sid).Scan(&sample)
|
||||
t.Fatalf("8 封邮件只有 %d 个不同时间戳（样例 %q）—— NOW() 精度不足，"+
|
||||
"同秒邮件的先后顺序会由随机 UUID 决定", distinct, sample)
|
||||
}
|
||||
|
||||
// GetSessionMails 按时间升序，顺序必须与插入顺序一致
|
||||
got, err := GetSessionMails(context.Background(), sid)
|
||||
if err != nil {
|
||||
t.Fatalf("取会话邮件失败: %v", err)
|
||||
}
|
||||
if len(got) != len(ids) {
|
||||
t.Fatalf("应有 %d 封，实际 %d", len(ids), len(got))
|
||||
}
|
||||
for i, m := range got {
|
||||
if m.ID != ids[i] {
|
||||
t.Errorf("第 %d 封顺序错位：期望 %s，实际 %s（主题 %q）",
|
||||
i, ids[i], m.ID, m.Subject)
|
||||
}
|
||||
}
|
||||
}
|
||||
139
gateway/internal/repo/sessionrate.go
Normal file
139
gateway/internal/repo/sessionrate.go
Normal file
@ -0,0 +1,139 @@
|
||||
package repo
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ---------- 新建会话速率限制 ----------
|
||||
//
|
||||
// 会话往返预算（sessions.max_rounds）管住了「一条线索能来回多少次」，
|
||||
// 但 Agent 仍可以用 `name@path.new` 开一串新会话，每条都是全新预算 ——
|
||||
// 预算就被绕过了。
|
||||
//
|
||||
// 为什么用速率限制而不是「终身额度」：
|
||||
// 终身额度跑满后要管理员手工重置才能再干活，而 Agent 是长期在线的 ——
|
||||
// 那是把一次性资源的模型套在长期服务上。速率限制只压住「短时间内暴开」这个
|
||||
// 真正的滥用形态，过一个窗口自动恢复，无需人工介入。
|
||||
//
|
||||
// 为什么不禁止 Agent 主动开新会话：那会堵死 Agent 之间的主动协作
|
||||
//（A 发现问题主动找 B），而这正是这个平台存在的意义。
|
||||
|
||||
// ErrSessionRateLimited 表示该 Agent 短时间内新建会话过多。
|
||||
var ErrSessionRateLimited = errors.New("session creation rate limited")
|
||||
|
||||
const (
|
||||
// sessionRateWindow 是滑动窗口长度
|
||||
sessionRateWindow = time.Hour
|
||||
// sessionRateLimit 是窗口内允许新建的会话数。
|
||||
//
|
||||
// 取 20：正常协作里 Agent 一小时开二十条新线索已经很多了；
|
||||
// 真到了这个量级，更可能是循环而不是在干活。
|
||||
sessionRateLimit = 20
|
||||
)
|
||||
|
||||
// sessionRateLimiter 记录每个 Agent 新建会话的时间戳。
|
||||
//
|
||||
// 进程内内存计数，与登录限速（handler/ratelimit.go）同一取舍：
|
||||
// 单实例部署下够用；多实例时各自计数，等效上限变成 N 倍 ——
|
||||
// 那时应当换成数据库计数或 Redis。这个限制记在 README 的已知取舍里。
|
||||
type sessionRateLimiter struct {
|
||||
mu sync.Mutex
|
||||
marks map[string][]time.Time
|
||||
}
|
||||
|
||||
var sessionLimiter = &sessionRateLimiter{marks: make(map[string][]time.Time)}
|
||||
|
||||
// Allow 判断是否允许新建，允许则记账。
|
||||
//
|
||||
// 判断与记账在同一把锁里：分开的话并发请求会双双通过检查，把上限刷穿 ——
|
||||
// 与配额那条 UPDATE 同样的道理。
|
||||
func (l *sessionRateLimiter) Allow(name string) (bool, int) {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
|
||||
now := time.Now()
|
||||
cutoff := now.Add(-sessionRateWindow)
|
||||
|
||||
kept := l.marks[name][:0]
|
||||
for _, t := range l.marks[name] {
|
||||
if t.After(cutoff) {
|
||||
kept = append(kept, t)
|
||||
}
|
||||
}
|
||||
l.marks[name] = kept
|
||||
|
||||
if len(kept) >= sessionRateLimit {
|
||||
// 最早那条何时过期 = 何时能再开一条
|
||||
retry := int(kept[0].Add(sessionRateWindow).Sub(now).Seconds()) + 1
|
||||
if retry < 1 {
|
||||
retry = 1
|
||||
}
|
||||
return false, retry
|
||||
}
|
||||
l.marks[name] = append(l.marks[name], now)
|
||||
return true, 0
|
||||
}
|
||||
|
||||
// Release 撤销一次记账。
|
||||
//
|
||||
// 允许之后建会话失败时必须还回去，否则那次没发生的新建也占了名额。
|
||||
func (l *sessionRateLimiter) Release(name string) {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
m := l.marks[name]
|
||||
if len(m) > 0 {
|
||||
l.marks[name] = m[:len(m)-1]
|
||||
}
|
||||
}
|
||||
|
||||
// AllowNewSession 供 handler 调用：Agent 新建会话前先过速率限制。
|
||||
//
|
||||
// 第二个返回值是建议的重试等待秒数（供 Retry-After 使用）。
|
||||
// 人类用户不走这条路径 —— 人手工点「新建邮件」的频率天然受限，
|
||||
// 给它加限制只会在批量派活时误伤。
|
||||
func AllowNewSession(agentName string) (bool, int) {
|
||||
if agentName == "" {
|
||||
return true, 0
|
||||
}
|
||||
return sessionLimiter.Allow(agentName)
|
||||
}
|
||||
|
||||
// ReleaseNewSession 建会话失败后归还名额。
|
||||
func ReleaseNewSession(agentName string) {
|
||||
if agentName == "" {
|
||||
return
|
||||
}
|
||||
sessionLimiter.Release(agentName)
|
||||
}
|
||||
|
||||
// 定期清理空闲 Agent 的记录，避免 map 随 Agent 名无限增长。
|
||||
func init() {
|
||||
go func() {
|
||||
t := time.NewTicker(sessionRateWindow)
|
||||
defer t.Stop()
|
||||
for range t.C {
|
||||
sessionLimiter.mu.Lock()
|
||||
cutoff := time.Now().Add(-sessionRateWindow)
|
||||
for name, marks := range sessionLimiter.marks {
|
||||
fresh := marks[:0]
|
||||
for _, ts := range marks {
|
||||
if ts.After(cutoff) {
|
||||
fresh = append(fresh, ts)
|
||||
}
|
||||
}
|
||||
if len(fresh) == 0 {
|
||||
delete(sessionLimiter.marks, name)
|
||||
} else {
|
||||
sessionLimiter.marks[name] = fresh
|
||||
}
|
||||
}
|
||||
sessionLimiter.mu.Unlock()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// SessionRateLimit 暴露窗口内的新建上限，供错误文案使用。
|
||||
// 不导出常量本身：外部只该读它，不该改它。
|
||||
func SessionRateLimit() int { return sessionRateLimit }
|
||||
117
gateway/internal/repo/sessionrate_test.go
Normal file
117
gateway/internal/repo/sessionrate_test.go
Normal file
@ -0,0 +1,117 @@
|
||||
package repo
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 每个测试用独立的 limiter，避免相互污染（全局那个是进程级的）
|
||||
func newLimiter() *sessionRateLimiter {
|
||||
return &sessionRateLimiter{marks: make(map[string][]time.Time)}
|
||||
}
|
||||
|
||||
func TestSessionRateAllowsUpToLimit(t *testing.T) {
|
||||
l := newLimiter()
|
||||
for i := 1; i <= sessionRateLimit; i++ {
|
||||
if ok, _ := l.Allow("bot"); !ok {
|
||||
t.Fatalf("第 %d 次应放行（上限 %d）", i, sessionRateLimit)
|
||||
}
|
||||
}
|
||||
ok, retry := l.Allow("bot")
|
||||
if ok {
|
||||
t.Fatal("超过上限应拦下")
|
||||
}
|
||||
if retry < 1 {
|
||||
t.Fatalf("应给出正的重试等待秒数，实际 %d", retry)
|
||||
}
|
||||
if retry > int(sessionRateWindow.Seconds())+1 {
|
||||
t.Fatalf("重试等待 %d 秒超过了窗口长度", retry)
|
||||
}
|
||||
}
|
||||
|
||||
// 不同 Agent 各自计数，一个刷满不该影响另一个
|
||||
func TestSessionRateIsPerAgent(t *testing.T) {
|
||||
l := newLimiter()
|
||||
for i := 0; i < sessionRateLimit; i++ {
|
||||
l.Allow("busy")
|
||||
}
|
||||
if ok, _ := l.Allow("busy"); ok {
|
||||
t.Fatal("busy 应已被拦")
|
||||
}
|
||||
if ok, _ := l.Allow("idle"); !ok {
|
||||
t.Fatal("另一个 Agent 不该被牵连")
|
||||
}
|
||||
}
|
||||
|
||||
// 判断与记账必须在同一把锁里，否则并发请求双双通过检查把上限刷穿
|
||||
func TestSessionRateConcurrentDoesNotOverrun(t *testing.T) {
|
||||
l := newLimiter()
|
||||
var wg sync.WaitGroup
|
||||
var mu sync.Mutex
|
||||
passed := 0
|
||||
|
||||
for i := 0; i < sessionRateLimit*4; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
if ok, _ := l.Allow("bot"); ok {
|
||||
mu.Lock()
|
||||
passed++
|
||||
mu.Unlock()
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
if passed != sessionRateLimit {
|
||||
t.Fatalf("%d 并发下放行 %d 次，期望恰好 %d 次",
|
||||
sessionRateLimit*4, passed, sessionRateLimit)
|
||||
}
|
||||
}
|
||||
|
||||
// 建会话失败时要还名额：那次新建实际上没有发生
|
||||
func TestSessionRateRelease(t *testing.T) {
|
||||
l := newLimiter()
|
||||
for i := 0; i < sessionRateLimit; i++ {
|
||||
l.Allow("bot")
|
||||
}
|
||||
if ok, _ := l.Allow("bot"); ok {
|
||||
t.Fatal("应已刷满")
|
||||
}
|
||||
l.Release("bot")
|
||||
if ok, _ := l.Allow("bot"); !ok {
|
||||
t.Fatal("归还名额后应能再开一条")
|
||||
}
|
||||
// 空记录上 Release 不该 panic
|
||||
l2 := newLimiter()
|
||||
l2.Release("nobody")
|
||||
}
|
||||
|
||||
// 窗口滑过后自动恢复 —— 这正是它优于「终身额度」的地方：
|
||||
// 终身额度跑满要人工重置，速率限制过一个窗口自己好
|
||||
func TestSessionRateWindowSlides(t *testing.T) {
|
||||
l := newLimiter()
|
||||
old := time.Now().Add(-sessionRateWindow - time.Minute)
|
||||
marks := make([]time.Time, sessionRateLimit)
|
||||
for i := range marks {
|
||||
marks[i] = old
|
||||
}
|
||||
l.marks["bot"] = marks
|
||||
|
||||
if ok, _ := l.Allow("bot"); !ok {
|
||||
t.Fatal("窗口外的记录应被清掉，此次应放行")
|
||||
}
|
||||
if len(l.marks["bot"]) != 1 {
|
||||
t.Fatalf("过期记录未清理，剩余 %d 条", len(l.marks["bot"]))
|
||||
}
|
||||
}
|
||||
|
||||
// 人类不走限速（空 agentName）
|
||||
func TestAllowNewSessionSkipsHumans(t *testing.T) {
|
||||
for i := 0; i < sessionRateLimit*3; i++ {
|
||||
if ok, _ := AllowNewSession(""); !ok {
|
||||
t.Fatal("人类不该被限速")
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -374,7 +374,7 @@ func ListActiveUsernames(ctx context.Context) ([]string, error) {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// ---------- 会话归属 ----------
|
||||
@ -500,7 +500,7 @@ func AllWorkspaceNames(ctx context.Context) ([]string, error) {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// IsHumanUser 判断某个三维地址 name 位是否为人类用户
|
||||
|
||||
@ -15,10 +15,18 @@ var (
|
||||
once sync.Once
|
||||
)
|
||||
|
||||
// GetIndex 返回入口页。
|
||||
//
|
||||
// index.html 缺失时回退到 placeholder.html：前端产物不进版本库，
|
||||
// 新克隆里只有占位页。回退而不是报错是故意的 ——
|
||||
// 只改后端的人应当能直接 go run 起来调 API，而不必先装 node。
|
||||
func GetIndex() []byte {
|
||||
once.Do(func() {
|
||||
data, _ := fs.ReadFile(staticFS, "static/index.html")
|
||||
indexHTML = data
|
||||
if data, err := fs.ReadFile(staticFS, "static/index.html"); err == nil {
|
||||
indexHTML = data
|
||||
return
|
||||
}
|
||||
indexHTML, _ = fs.ReadFile(staticFS, "static/placeholder.html")
|
||||
})
|
||||
return indexHTML
|
||||
}
|
||||
|
||||
@ -1,13 +0,0 @@
|
||||
<!DOCTYPE html>
|
||||
<html lang="zh-CN">
|
||||
<head>
|
||||
<meta charset="UTF-8" />
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
|
||||
<title>AgentMail</title>
|
||||
<script type="module" crossorigin src="/assets/index-C3ZRvtHr.js"></script>
|
||||
<link rel="stylesheet" crossorigin href="/assets/index-Bi5AQIOl.css">
|
||||
</head>
|
||||
<body class="bg-gray-50 text-gray-900 antialiased">
|
||||
<div id="root"></div>
|
||||
</body>
|
||||
</html>
|
||||
69
gateway/internal/static/static/placeholder.html
Normal file
69
gateway/internal/static/static/placeholder.html
Normal file
@ -0,0 +1,69 @@
|
||||
<!doctype html>
|
||||
<!--
|
||||
前端未构建时的占位页。
|
||||
|
||||
为什么文件名不是 index.html：那正是 Vite 构建产物的名字。用 index.html 当占位，
|
||||
每次构建后真实产物都会盖掉它并被 git 视为改动；提交进去的 index.html 引用着
|
||||
被忽略的 assets/，新克隆打开就是白屏而不是这张提示页。
|
||||
|
||||
改叫 placeholder.html：构建不会产生这个名字，因此永不被覆盖。
|
||||
static.go 在 index.html 缺失时回退到它。
|
||||
它同时让 go:embed 有文件可嵌 —— 否则新克隆连 go build 都过不去。
|
||||
-->
|
||||
<html lang="zh-CN">
|
||||
<head>
|
||||
<meta charset="utf-8" />
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1" />
|
||||
<title>AgentMail — 前端未构建</title>
|
||||
<style>
|
||||
body {
|
||||
margin: 0;
|
||||
min-height: 100vh;
|
||||
display: flex;
|
||||
align-items: center;
|
||||
justify-content: center;
|
||||
background: #f1f5f9;
|
||||
color: #334155;
|
||||
font: 14px/1.7 system-ui, -apple-system, "Segoe UI", sans-serif;
|
||||
}
|
||||
main { max-width: 34rem; padding: 2rem; }
|
||||
h1 { font-size: 1rem; margin: 0 0 0.75rem; }
|
||||
code {
|
||||
background: #e2e8f0;
|
||||
padding: 0.1rem 0.35rem;
|
||||
border-radius: 3px;
|
||||
font-size: 0.85em;
|
||||
}
|
||||
pre {
|
||||
background: #fff;
|
||||
border: 1px solid #e2e8f0;
|
||||
border-radius: 6px;
|
||||
padding: 0.75rem 1rem;
|
||||
overflow-x: auto;
|
||||
font-size: 0.85em;
|
||||
}
|
||||
p { margin: 0 0 0.75rem; }
|
||||
.dim { color: #64748b; font-size: 0.9em; }
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<main>
|
||||
<h1>AgentMail 前端未构建</h1>
|
||||
<p>
|
||||
当前二进制里没有前端产物。后端 API 仍然可用（<code>/api/v1</code>、
|
||||
<code>/health</code>）。
|
||||
</p>
|
||||
<p>构建并嵌入前端：</p>
|
||||
<pre>sudo ./deploy/install.sh</pre>
|
||||
<p>或者手工：</p>
|
||||
<pre>cd web && npm ci && npm run build
|
||||
rm -rf ../gateway/internal/static/static/assets
|
||||
cp -r dist/. ../gateway/internal/static/static/
|
||||
cd ../gateway && go build ./cmd/server</pre>
|
||||
<p class="dim">
|
||||
构建产物不进版本库；这个占位文件的存在只是为了让
|
||||
<code>go:embed</code> 在新克隆里能编译通过。
|
||||
</p>
|
||||
</main>
|
||||
</body>
|
||||
</html>
|
||||
Reference in New Issue
Block a user