sanguo_moziplus_v2/docs/research/mail-reply-research.md

Mail API 回复能力调研报告

调研人: 庞统（subagent） 日期: 2026-06-06 任务来源: 主公委派，验证 Mail API 回复驱动可行性

---

调研结论

Mail API 当前能力是否满足回复驱动设计

部分满足。Mail API 已具备 in_reply_to 关联机制、conversation_id 线程聚合、type 区分 request/inform、完整的安全防御（A1-A10），以及 daemon ticker 对 _mail 虚拟项目的 30s 扫描调度。

但缺少以下关键查询能力：

1. ❌ 无法查询某个 Mail 的所有回复（无 in_reply_to 过滤参数）
2. ❌ 无法查询"所有有新回复的 Mail"（无 find_recently_replied）
3. ❌ 无法查询"所有 type=request 且无回复的 Mail"（无 find_pending_requests）
4. ❌ Mail metadata 无法存 deadline_hours（must_haves JSON 可存但无专用字段和查询路径）
5. ❌ 无 Mail 模板功能（模板固化在 Agent 的 Skill/Prompt 中）

缺失能力清单

缺失项	影响级别	实现难度
in_reply_to 查询过滤	高	低（加一个查询参数）
find_pending_requests 查询	高	低（组合查询）
deadline_hours 存储	中	极低（must_haves JSON 已支持）
find_recently_replied 查询	中	中（需子查询）
Mail 模板系统	低	不需要（设计决策：模板在 Prompt 层）

---

Mail API 现有能力

1. 发送 Mail（POST /api/mail）

源码: mail_routes.py:L93-L194（send_mail 函数）

支持字段：

字段	必填	说明	源码行
from	✅	发件 Agent ID（A1/A9 校验）	L98
to	非回复时必填	收件 Agent ID（A2/A4 校验）	L100
title	❌	标题（fallback 空）	L178
text / description	✅	正文（A10 校验）	L121
type	❌	inform/request（默认：非回复→request，回复→inform）	L170-L172
in_reply_to	❌	回复的 Mail ID（A5 存在性校验）	L105
conversation_id	❌	会话 ID（自动继承/生成）	L160-L166
id	❌	自定义 ID（默认 mail-timestamp）	L175

2. 列表查询（GET /api/mail）

源码: mail_routes.py:L79-L91

支持过滤参数：

参数	说明	实现方式
status	Task 状态过滤	SQL WHERE（直接传 Blackboard.list_tasks）
from_agent	发件人过滤	SQL WHERE（assigned_by）
to_agent	收件人过滤	SQL WHERE（assignee）
unread	未读过滤	内存过滤（解析 must_haves JSON）
limit	数量限制（默认 50，最大 200）	Python 切片

不支持的过滤：

• ❌ in_reply_to 过滤（无法查询某个 Mail 的所有回复）
• ❌ type 过滤（must_haves JSON 内字段，无法 SQL WHERE）
• ❌ conversation_id 过滤（同上）
• ❌ 时间范围过滤（since/until）

3. 详情查询（GET /api/mail/{mail_id}）

源码: mail_routes.py:L104-L121

返回 Mail 全部字段 + comments（对话线程）。Comments 来自黑板 comments 表。

4. 更新（PATCH /api/mail/{mail_id}）

源码: mail_routes.py:L208-L224

• is_read 标记已读
• mark_executed 标记已执行（status→done）

5. 批量删除（DELETE /api/mail）

按 title 前缀匹配，status→cancelled。

6. 摘要（GET /api/mail/summary）

返回 total、unread、by_type 统计。

7. Agent 列表（GET /api/mail/agents/list）

返回参与过 Mail 的所有 Agent ID。

---

回复机制详解

in_reply_to 的实际行为

存储方式: in_reply_to 存储在 Task 的 must_haves JSON 字段中（mail_routes.py:L182）。

meta = {
    "type": mail_type,
    "performative": performative,
    "is_read": False,
    "conversation_id": conversation_id,
    "in_reply_to": in_reply_to,  # ← 存在这里
    "from": from_agent,
}
task = Task(..., must_haves=json.dumps(meta))

行为特点：

1. 简单字段关联：in_reply_to 只是一个 ID 引用，不是外键约束，没有数据库级的线程聚合。线程聚合通过 conversation_id 实现。

2. 自动推断收件人（A6/A7）：回复时 to 自动从原邮件的 assigned_by（发件者）取值。无论 Agent 传什么 to，系统都静默纠正（mail_routes.py:L113-L117）。

3. 严格 1 对 1 校验（A8）：只有原邮件的 assigned_by（发件者）和 assignee（收件者）可以回复（mail_routes.py:L119-L121）。

4. conversation_id 自动继承：回复邮件自动继承原邮件的 conversation_id（mail_routes.py:L160-L166），用于前端线程展示。

5. type 自动设置：有 in_reply_to 且未显式指定 type 时，默认 inform（防循环）（mail_routes.py:L170-L172）。

查询回复的能力

当前无法直接查询某个 Mail 的所有回复。原因：

1. list_mail 不支持 in_reply_to 查询参数
2. in_reply_to 存在 must_haves JSON 中，不是独立的 SQL 列
3. 当前唯一查询回复的方式：GET /api/mail/{mail_id} 获取详情时附带 comments（但这不是回复 Mail，而是黑板 comment）

daemon 中的回复查询（ticker.py _mail_check_reply）：

# L890 左右
row = conn.execute(
    "SELECT id FROM tasks WHERE id != ? AND must_haves LIKE ? LIMIT 1",
    (original_task_id, f'%{original_task_id}%'),
).fetchone()

这是通过 LIKE '%{task_id}%' 在 must_haves JSON 中模糊匹配实现的——脆弱但能用。in_reply_to 的值恰好包含在 must_haves JSON 中，所以 LIKE 能匹配到。

---

查询能力

支持的查询

查询场景	支持方式	限制
按 status 查询	SQL WHERE	✅ 直接支持
按 from_agent 查询	SQL WHERE（assigned_by）	✅ 直接支持
按 to_agent 查询	SQL WHERE（assignee）	✅ 直接支持
按 unread 过滤	内存过滤（解析 JSON）	⚠️ 性能差（全量加载再过滤）
按 title 模糊查询	无	❌
按 time range 查询	无	❌
单封 Mail 详情	GET /api/mail/{id}	✅ 含 comments
摘要统计	GET /api/mail/summary	✅

不支持的查询

查询场景	设计需要	当前状态	影响
查询某 Mail 的所有回复	daemon ticker 检查回复	❌ 无 in_reply_to 过滤参数	高 — daemon 靠 LIKE 模糊匹配，脆弱
查询所有 type=request 的 Mail	find_pending_requests	❌ type 在 JSON 中	高 — 无法高效找未回复的 request
查询所有有新回复的 Mail	find_recently_replied	❌ 无此查询	中 — daemon 无法发现已回复的 Mail
查询某 conversation 的所有 Mail	线程展示	❌ conversation_id 在 JSON 中	中 — 前端无法获取完整线程
按 created_at/updated_at 范围查询	时间窗口	❌ 无时间参数	低 — 可通过 limit+倒序近似

时间字段

• ✅ created_at：tasks 表有，自动填充 datetime('now')
• ✅ updated_at：tasks 表有，状态变更时自动更新
• ⚠️ 两个字段都是 Task 模型自带，Mail 复用了它们

Metadata 能力

Mail 的所有扩展 metadata 都存在 must_haves JSON 字段中：

字段	当前值	可扩展性
type	inform/request	✅ 可存任意值
performative	同 type	✅
is_read	true/false	✅
conversation_id	conv-timestamp	✅
in_reply_to	mail_id 或 null	✅
deadline_hours	未存	✅ 可直接加到 JSON 中
priority	未存	✅ 可直接加
template_id	未存	✅ 可直接加

结论：must_haves JSON 完全可以存 deadline_hours 等任意 metadata，无需改表结构。问题是查询时必须解析 JSON，无法直接 SQL WHERE。

---

Mail 模板能力

当前状态

Mail API 没有模板功能。API 只负责 CRUD，模板固化在两个地方：

1. Agent Skill / Prompt 指令中：Agent 通过 Skill（如 sanguo-mail）获得发 Mail 的 curl 模板
2. Spawner Mail Prompt 模板中（spawner.py）：
   • MAIL_INFORM_TEMPLATE：纯通知模板
   • MAIL_REQUEST_TEMPLATE：需回复模板

设计决策

这是正确的设计。模板属于业务层（Prompt），不属于 API 层。API 只负责消息的发送、存储、查询。模板放在 API 层会导致：

• API 与业务逻辑耦合
• 模板变更需要改后端代码
• Agent 自主决策能力受限

---

Daemon 中的 Mail 扫描逻辑

现有扫描

Daemon ticker 在 _tick_project 中会扫描 _mail 虚拟项目（ticker.py:L243-L252），走标准调度流程：

1. _check_timeouts：超时检测（claimed 5min, working 30min）
2. _dispatch_pending：调度 pending 任务
3. _mail_check_reply：幻觉门控（working 超时时检查是否有回复）

关键逻辑

1. Mail 自动标 working（dispatcher.py）：dispatch Mail 时系统自动标 working，Agent 不需要手动标
2. 幻觉门控（ticker.py:_mail_check_reply）：request 类型邮件 working 超时时，先检查 DB 有无回复邮件，有回复则标 done 而非 failed
3. Mail 不走 review 流程（ticker.py:_dispatch_reviews）：if project_id == "_mail": return []

缺失的扫描逻辑

需要的扫描	当前状态	用途
find_pending_requests（type=request 且无回复）	❌ 无	daemon 主动催办未回复的 request
find_recently_replied（有新回复的 Mail）	❌ 无	daemon 发现回复后驱动下一步
check_stale_flows（超时检测 + 分级处置）	❌ 无	§15.4 设计的三级超时处置

---

缺失能力清单

#	缺失项	设计需要	实现建议	影响范围
1	in_reply_to 查询过滤	daemon 检查某 Mail 是否有回复	改 API：GET /api/mail?in_reply_to={id} — list_mail 加参数，用 must_haves LIKE 或子查询	mail_routes.py
2	type 查询过滤	daemon 查找 type=request 的 Mail	改 API：GET /api/mail?type=request — 同上，must_haves LIKE 匹配	mail_routes.py
3	conversation_id 查询过滤	前端线程展示	改 API：GET /api/mail?conversation_id={cid}	mail_routes.py
4	find_pending_requests	daemon 催办无回复的 request	改 daemon：ticker 中加一个扫描逻辑，组合 type=request + 无 in_reply_to 回复	ticker.py
5	find_recently_replied	daemon 发现新回复驱动下一步	改 daemon：扫描最近 updated_at 变化且 has_reply 的 Mail	ticker.py
6	deadline_hours 存储查询	§15.4 三级超时	不需改 API：直接在 send_mail 的 must_haves JSON 中加字段；daemon 读 must_haves 解析即可	调用方 + ticker.py
7	Mail 模板系统	§15.5 模板	不需实现：模板在 Prompt 层是正确设计	无

---

建议改动

优先级 P0（回复驱动必需）

改动 1：mail_routes.py list_mail 增加 in_reply_to 和 type 过滤

位置: list_mail 函数（L79-L91）

改什么: 增加两个可选查询参数 in_reply_to 和 type，用 must_haves LIKE 过滤。

伪代码:

@router.get("")
async def list_mail(
    status, from_agent, to_agent, unread, limit,
    in_reply_to: Optional[str] = None,  # 新增
    mail_type: Optional[str] = None,    # 新增（避免与 Python type 冲突）
    conversation_id: Optional[str] = None,  # 新增
):
    # ... 现有逻辑 ...
    for t in tasks:
        meta = _mail_meta(t)
        if in_reply_to and meta.get("in_reply_to") != in_reply_to:
            continue
        if mail_type and meta.get("type") != mail_type:
            continue
        if conversation_id and meta.get("conversation_id") != conversation_id:
            continue
        # ... 其余逻辑 ...

替代方案（更高效）: 加一个 DB migration，把 in_reply_to、type、conversation_id 从 must_haves JSON 提取为独立的 SQL 列。这样可以用索引加速查询。但这会改变 tasks 表结构，影响所有项目。

推荐: 先用内存过滤（快速实现），后续按需优化为独立列。

改动 2：ticker.py 增加 Mail 回复扫描逻辑

位置: _tick_project 中 _mail 虚拟项目处理之后

改什么: 新增一个 _check_mail_replies 方法：

1. 查询所有 type=request 的 Mail
2. 对每个 request，检查是否有 in_reply_to 指向它的回复
3. 有回复 → 触发下一步（通知发件者）
4. 无回复 + 超时 → 催办

伪代码:

async def _check_mail_replies(self, db_path: Path):
    """扫描 Mail 回复状态，驱动回复驱动流程"""
    conn = get_connection(db_path)
    try:
        # 查所有 working 状态的 Mail
        rows = conn.execute(
            "SELECT id, must_haves, created_at FROM tasks WHERE task_type='mail' AND status='working'"
        ).fetchall()
        for row in rows:
            meta = json.loads(row["must_haves"]) if row["must_haves"] else {}
            mail_type = meta.get("type", "inform")
            if mail_type != "request":
                continue  # inform 不需要回复

            # 检查是否有回复
            reply = conn.execute(
                "SELECT id FROM tasks WHERE task_type='mail' AND id != ? AND must_haves LIKE ? LIMIT 1",
                (row["id"], f'%"in_reply_to": "{row["id"]}"%'),
            ).fetchone()

            if reply:
                # 有回复 → 可以标 done（如果还没标）
                pass
            else:
                # 无回复 → 检查超时
                deadline_hours = meta.get("deadline_hours", 4)
                elapsed = (datetime.utcnow() - datetime.fromisoformat(row["created_at"])).total_seconds() / 3600
                if elapsed > deadline_hours:
                    # 超时 → 催办
                    pass
    finally:
        conn.close()

优先级 P1（增强）

改动 3：deadline_hours 支持

不需改 API。在发送 Mail 时，调用方在 must_haves JSON 中加入 deadline_hours：

{
  "from": "pangtong-fujunshi",
  "to": "simayi-challenger",
  "type": "request",
  "deadline_hours": 2,
  "text": "请审查..."
}

API 已支持——must_haves 是自由 JSON，deadline_hours 会被存入 DB。daemon ticker 读取时解析 JSON 即可。

优先级 P2（远期）

改动 4：独立列优化

将 in_reply_to、type、conversation_id 从 must_haves JSON 提取为 tasks 表的独立列，建索引加速查询。需要 DB migration。

---

附录：Mail 数据模型映射

Mail 基于 Task 模型，字段映射关系：

Mail 语义	Task 字段	存储位置
id	task.id	主键
title	task.title	直接列
text / description	task.description	直接列
from	task.assigned_by	直接列
to	task.assignee	直接列
status	task.status	直接列（pending→claimed→working→done/failed）
type	must_haves.type	JSON
is_read	must_haves.is_read	JSON
in_reply_to	must_haves.in_reply_to	JSON
conversation_id	must_haves.conversation_id	JSON
performative	must_haves.performative	JSON
created_at	task.created_at	直接列
updated_at	task.updated_at	直接列
task_type	task.task_type = "mail"	直接列