fix: S1 handler display_name + S2 import 移顶部 + W1 注释

- S1: vp_name 硬编码字典 → handler.display_name 属性
- S2: ticker/spawner 中 TaskTypeRegistry 局部 import → 移文件顶部
- W1: TaskHandler executor verify 失败不调 on_failure 加注释说明

.gitea/workflows/ci.yml

@@ -1,9 +1,10 @@
 # CI 管道 — moziplus v2.0
 #
 # 触发条件：
-#   - push（非 main 分支）
 #   - pull_request（opened, synchronize）
 #
+# 注意：只保留 pull_request 触发，避免 push + pull_request 双倍触发
+#
 # Gitea v1.23.4 限制注意：
 #   - 不支持 failure() 表达式，用 always() + shell 条件判断替代
 #   - 不支持 concurrency / continue-on-error / timeout-minutes / permissions
@@ -13,10 +14,6 @@
 name: CI
 
 on:
-  push:
-    branches:
-      - '**'
-      - '!main'
   pull_request:
     types: [opened, synchronize]
 
@@ -46,14 +43,15 @@ jobs:
       - name: Setup Python
         run: |
           python3 -m venv /tmp/ci-venv-test
-          /tmp/ci-venv-test/bin/pip install --quiet -r requirements.txt
+          /tmp/ci-venv-test/bin/pip install --quiet fastapi pydantic pyyaml uvicorn requests pytest pytest-asyncio httpx
 
       - name: Run tests (exclude E2E)
         run: |
           /tmp/ci-venv-test/bin/pytest tests/ -m "not e2e" -x -q
 
   # ── Job 3: CI 失败通知 ───────────────────────────────
-  # v1.23 不支持 failure()，用 always() + shell 检查 commit status 替代
+  # 使用 needs.<job>.result 直接判断，不查询 commit status API
+  # 根因：notify 自身的 pending status 会污染 commit status 查询结果（竞态条件）
   notify-on-failure:
     runs-on: macos-arm64
     needs: [lint, test]
@@ -62,31 +60,34 @@ jobs:
       - name: Check results and notify
         env:
           GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
+          LINT_RESULT: ${{ needs.lint.result }}
+          TEST_RESULT: ${{ needs.test.result }}
         run: |
-          # 查询当前 commit 的 status
-          STATUS=$(curl -sf \
-            -H "Authorization: token $GITEA_TOKEN" \
-            "${{ gitea.api_url }}/repos/${{ gitea.repository }}/commits/${{ gitea.sha }}/status" \
-            | python3 -c "import sys,json; print(json.load(sys.stdin).get('state',''))" 2>/dev/null || echo "")
+          echo "Lint result: $LINT_RESULT"
+          echo "Test result: $TEST_RESULT"
 
-          echo "Commit status: $STATUS"
-
-          if [ "$STATUS" != "success" ]; then
-            echo "CI failed or status unknown, sending notification..."
+          # 只有 lint 或 test 明确失败时才发通知
+          if [ "$LINT_RESULT" = "failure" ] || [ "$TEST_RESULT" = "failure" ]; then
+            echo "CI has failures, sending notification..."
 
             # 如果是 PR 事件，写评论通知
             PR_NUMBER="${{ gitea.event.pull_request.number }}"
             if [ -n "$PR_NUMBER" ]; then
+              # 构建失败摘要
+              FAILED_JOBS=""
+              [ "$LINT_RESULT" = "failure" ] && FAILED_JOBS="${FAILED_JOBS}lint "
+              [ "$TEST_RESULT" = "failure" ] && FAILED_JOBS="${FAILED_JOBS}test "
+
               curl -sf -X POST \
                 -H "Authorization: token $GITEA_TOKEN" \
                 -H "Content-Type: application/json" \
                 "${{ gitea.api_url }}/repos/${{ gitea.repository }}/issues/${PR_NUMBER}/comments" \
-                -d "{\"body\": \"[CI] 失败\\n\\n分支: ${{ gitea.ref_name }}\\n触发 commit: \`${{ gitea.sha }}\`\\n请检查 CI 日志并修复。\"}" \
+                -d "{\"body\": \"[CI] 失败\\n\\n分支: ${{ gitea.ref_name }}\\n触发 commit: \`${{ gitea.sha }}\`\\n失败 Job: ${FAILED_JOBS}\\n请检查 CI 日志并修复。\"}" \
                 || echo "Failed to post PR comment"
               echo "PR comment posted."
             else
               echo "Not a PR event, skipping PR comment."
             fi
           else
-            echo "CI passed, no notification needed."
+            echo "No explicit failures (results: lint=$LINT_RESULT, test=$TEST_RESULT), no notification needed."
           fi

.gitea/workflows/deploy.yml

@@ -23,16 +23,16 @@ jobs:
 
       - name: Setup Python
         run: |
-          python3 -m venv .venv
-          .venv/bin/pip install --quiet -r requirements.txt
+          python3 -m venv /tmp/ci-venv-deploy
+          /tmp/ci-venv-deploy/bin/pip install --quiet flake8 fastapi pydantic pyyaml uvicorn requests pytest pytest-asyncio httpx
 
       - name: Lint
         run: |
-          .venv/bin/flake8 src/ --max-line-length=120 --extend-ignore=E501
+          /tmp/ci-venv-deploy/bin/flake8 src/ --max-line-length=120 --extend-ignore=E501
 
       - name: Unit & Integration Tests
         run: |
-          .venv/bin/pytest tests/ -m "not e2e" -x -q
+          /tmp/ci-venv-deploy/bin/pytest tests/ -m "not e2e" -x -q
 
   # ── Job 2: 部署 ─────────────────────────────────────
   deploy:

docs/design/08-classify-outcome-optimization.md

@@ -110,8 +110,8 @@ TCP 握手只能检测进程端口是否监听，无法检测 Gateway **业务
 
 | 编号 | 条件 | outcome | 可恢复？ | 处理 |
 |------|------|---------|----------|------|
-| A12 | exit=0 + task_status ∈ {done, review} | completed | — | 正常完成 |
-| A13 | exit=0 + task_status ∉ {done, review} | agent_error | ❌ | 标 failed + 原因写黑板 |
+| A12 | ~~已合并到 A13 revised~~ | — | — | 见下方 A13 revised |
+| **A13 revised** | exit=0（无 JSON 输出） | completed | — | 信任进程退出码，exit=0 即正常完成。旧逻辑按 task_status 区分，非终态判 agent_error → 导致 inform Mail 永不标 done，与 dispatcher inform auto-done 形成死循环 |
 | **A14** | exit=130 (SIGINT) 或 exit=143 (SIGTERM) | interrupted | ✅ | retry |
 | **A15** | exit≠0 + stderr 含 network 关键字 | gateway_unreachable | ✅ | retry + cooldown 30s |
 | **A16** | exit≠0 + stderr 含 compact 关键字 | compact_interrupted | ✅ | retry + cooldown 60s |

docs/design/13-toolchain-and-dev-workflow.md

@@ -1,6 +1,6 @@
 # 三国团队工具链与开发流程设计
 
-> **状态**: v3.1 — P3 端到端验证通过 + 调研结论写入 + Review API 枚举值修正
+> **状态**: v3.3 — #19 上下文四层改造合并 + CI 修复 + A13 修订
 > **作者**: 庞统（副军师）🐦
 > **评审**: 司马懿（仲达）🗡️
 > **日期**: 2026-06-06
@@ -1590,7 +1590,7 @@ daemon 内部 ───────┘    │ 5. 创建 Mail    │
 | 只处理白名单内的事件类型 | 未知的忽略 + 日志 |
 | issue_comment 需判断来源 | 只处理 CI workflow 写的评论（按特定前缀匹配：`❌ **CI 失败**` 或统一后的 `[CI]` 前缀） |
 | PR 作者/审查者必须是已知 Agent | 未知的忽略 + 日志 |
-| 幂等：同一事件不重复创建 Mail | 按 `{x_gitea_event}-{x_gitea_delivery}` 去重（delivery ID 来自 `X-Gitea-Delivery` header） |
+| 幂等：同一事件不重复创建 Mail | 双重去重：① delivery UUID（`{event}-{delivery}`）标准幂等；② review 事件 payload 内容去重（`{event}:{pr_num}:{sender}:{sha256(body_or_content)[:16]}`），防御同一 review 被不同来源重复提交（2026-06-09 新增） |
 
 ---
 
@@ -2007,6 +2007,9 @@ CI workflow 已有 `notify-on-failure` job（ci.yml），当前格式：
 | 7 | 签名算法 | ✅ 已确认 | Gitea 使用 HMAC-SHA256，代码注释已补 |
 | 8 | Webhook 作用范围 | ✅ 组织级 | Gitea 组织级 webhook（Hook ID=28），覆盖 sanguo 下所有仓库，新增仓库自动覆盖 |
 | 9 | ALLOWED_HOST_LIST | ✅ 已修复 | Gitea 容器配置 `192.168.2.153, 127.0.0.1, localhost, 172.17.0.0/16, 192.168.2.0/24` |
+| 10 | Gitea review payload 格式 | ✅ 姜维调研确认（2026-06-08） | Gitea v1.23.4 review payload 只有 `type` + `content`，没有 `state`/`body`/`user`，这不是 org vs repo 差异而是 Gitea 设计。v1.24.0 格式不变。双格式兼容是防御性编码，保持现状 |
+| 11 | Spawner compact 检测窗口 | ✅ 已修复 | 窗口 300s→900s，尾部读取 50KB→1MB。实测长对话中 compact 记录被推出窗口导致漏检 |
+| 12 | inform 类型 Mail crash 误标 done | ✅ 已修复 | `_mail_auto_complete` 增加 outcome 感知，inform 用白名单（completed/claimed/no_reply）控制 done 标记。spawner crash cooldown 300s→60s |
 
 ---
 
@@ -2713,10 +2716,10 @@ Gitea v1.23.4 自带完整的 CI 管理界面：
 | # | 条件 | 状态 | 谁确认 |
 |---|------|------|--------|
 | 1 | act-runner 已注册且 label = `macos-arm64` | ✅ PM2 托管（sanguo-act-runner, id=44），崩溃自动重启 | 姜维确认 |
-| 2 | Gitea repository secrets 已配置（CI_TOKEN） | ⚠️ 需确认 | 姜维 |
+| 2 | Gitea repository secrets 已配置（CI_TOKEN） | ✅ 姜维确认（sanguo/moziplus-v2 已配 CI_TOKEN） | 姜维 |
 | 3 | Gitea 组织级 Webhook 已启用（Hook ID=28） | ✅ 已确认 | 已确认 |
-| 4 | 各 Agent 的 GITEA_TOKEN 环境变量 | ⚠️ 待分配 | 庞统协调 |
-| 5 | main 分支保护规则（Review 才能 merge） | ⚠️ 需确认 | 姜维 |
+| 4 | 各 Agent 的 GITEA_TOKEN 环境变量 | ✅ 已写入各 Agent TOOLS.md，姜维确认 token 记录存在 | 庞统+姜维 |
+| 5 | main 分支保护规则（Review 才能 merge） | ✅ 姜维已配置（moziplus-v2 + sanguo_moziplus_v2，需1个approve） | 姜维 |
 | 6 | 禁止在 daemon 运行时跑全量 E2E | ✅ 已警告司马懿 | 已确认 |
 
 > 第 5 点很关键——如果 main 分支没有保护规则，开发者可以直接 push main 跳过 Review。
@@ -2753,3 +2756,387 @@ Gitea v1.23.4 自带完整的 CI 管理界面：
 | §17.6.4 | 新增 P3 端到端验证结果（S1-S6 逐项） |
 | §17.6.4 | 新增调研发现：Review API 枚举值、PullRequestReview webhook 支持、act-runner PM2 托管 |
 | §17.10 | #1 状态更新：act-runner 已纳入 PM2 托管 |
+
+### v3.1 → v3.2 变更（工具链修复 + Mail 投递 bug 修复）
+
+| 编号 | 变更内容 |
+|------|----------|
+| §16.4 | Review handler 双格式兼容：HANDLERS 注册表同时注册 `pull_request_review` / `pull_request_approved` 等多种事件名；`_handle_pull_request_review` 兼容 repo webhook（review.state/body/user）和 org webhook（review.type/content/sender）两种 payload 格式 |
+| §16.8 #10 | Gitea v1.23.4 review payload 调研结论（姜维 2026-06-08）：Gitea v1.23.4 review payload 只有 `type` + `content`，没有 `state`/`body`/`user`，这不是 org vs repo 差异而是 Gitea 设计。v1.24.0 格式不变。双格式兼容是防御性编码，保持现状 |
+| §16.8 #11 | Spawner compact 检测窗口修复：窗口 300s→900s，尾部读取 50KB→1MB。实测长对话中 compact 记录被推出窗口导致漏检 |
+| §16.8 #12 | inform 类型 Mail crash 误标 done bug 修复：`_mail_auto_complete` 增加 outcome 感知，inform 用白名单（completed/claimed/no_reply）控制 done 标记。spawner crash cooldown 300s→60s |
+
+---
+
+### 一、问题诊断
+
+#### 1.1 E2E 真实场景测试暴露的三个断层
+
+主公在 moziplus-v2 仓库创建了 Issue #32（添加 /api/stats 端点），指派张飞。链条在第一步就断了。
+
+| 断层 | 现象 | 根因 |
+|------|------|------|
+| **Agent 不知道该做什么** | 张飞收到 Issue 指派 Mail，回复"已阅"就结束了 | Mail 模板（issue_assigned.md）5 行信息，无流程引导；spawn prompt 说"已阅即可" |
+| **Agent 去错了仓库** | 张飞去读了 sanguo_moziplus_v2 平台代码，而不是空的实验仓库 moziplus-v2 | Mail 模板没有仓库 clone URL，张飞凭习惯去了开发目录 |
+| **Agent 在 Control UI 提问** | 张飞遇到问题直接在 Control UI 问主公，没有去 Gitea Issue 评论 | 没有任何地方引导"有疑问去 Gitea Issue 评论" |
+| **Agent 不知道怎么协作** | 张飞判断任务需要澄清，但不知道该怎么请求澄清 | 没有"做不了→在 Issue 评论 / Mail 庞统"的回退路径 |
+| **跨 Agent @mention 无法通知** | 张飞在 Issue 评论 @赵云，赵云收不到通知 | issue_comment handler 只处理 [CI] 评论，@mention 被忽略 |
+
+#### 1.2 根因：工具链在四层架构中的断层
+
+| 层 | 应该有 | 实际有 | Gap |
+|---|---|---|---|
+| **L0 铁律** | — | — | 无需改动 |
+| **L1 角色** | 工具链协作行为规范（所有 Agent 共享） | 无 | AGENTS.md 没有工具链相关内容 |
+| **L2 引擎注入** | 事件上下文（仓库 clone URL、Gitea API、Issue/PR 详情） | Mail 模板只有 5 行摘要 | 缺仓库信息和流程引导 |
+| **L3 被动参考** | 技术细节（分支命名、commit 规范、PR 创建方式） | git-workflow 等 Skill 已存在但没人触发 | Agent 不知道该加载哪个 Skill |
+
+---
+
+### 二、改造方案：四层归属
+
+#### 2.1 分层原则
+
+| 层 | 放什么 | 不放什么 | 理由 |
+|---|---|---|---|
+| **L0** | 不放 | — | 工具链不是安全底线 |
+| **L1** | 协作行为规范：收到什么通知该做什么、遇到问题怎么办 | 技术细节（分支命名、commit 格式） | 行为规范是团队常识，每个 Agent 都要知道 |
+| **L2** | 事件上下文：仓库 clone URL、Gitea API URL、Issue/PR 链接、动态信息 | 固定的协作流程 | 动态信息每次不同，由 Mail 模板 + spawn 时注入 |
+| **L3** | 技术细节：git-workflow、code-review 等 Skill 全文 | — | 按需加载，Agent 知道"我要提 PR"后自己读 |
+
+#### 2.2 各层具体内容
+
+##### L1：AGENTS.md 加工具链协作行为段（所有 Agent 统一）
+
+```markdown
+## 工具链协作（Gitea）
+
+收到 Gitea 事件通知（Issue 指派、Review 请求、CI 失败等）时，按以下流程操作：
+
+### 基本流程
+- **Issue 指派** → clone 仓库 → 开分支 → 编码 → 提 PR（参考 git-workflow Skill）
+- **Review 请求** → 读 PR diff（Gitea API）→ 提交 Review（参考 code-review Skill）
+- **Review 通过** → 等 merge
+- **Review 驳回** → 看 review body → 修代码 → 重新 push
+- **CI 失败** → 看错误摘要 → 修代码 → push（自动重触发 CI）
+- **部署失败** → 查 deploy 日志 → 修复
+
+### 协作规则
+- **有疑问？** 在 Gitea Issue 下评论，不要在 Control UI 或 Mail 里问
+- **需要别人帮忙？** 在 Issue 评论中 @mention 对应 Agent（如 @zhaoyun-data）
+- **做不了？** 回复 Mail 说明原因和建议的接手人
+- **获取完整上下文** → 用 Gitea API 拉取 Issue 详情和评论，不要只看 Mail 里的快照
+
+### Gitea API 速查
+> 其中 `{owner}/{repo}` 替换为实际仓库，如 `sanguo/sanguo_moziplus_v2`
+- Issue 详情: GET /api/v1/repos/{owner}/{repo}/issues/{number}
+- Issue 评论: GET /api/v1/repos/{owner}/{repo}/issues/{number}/comments
+- PR diff: GET /api/v1/repos/{owner}/{repo}/pulls/{number}.diff
+- 提交 Review: POST /api/v1/repos/{owner}/{repo}/pulls/{number}/reviews
+```
+
+**改动范围**：6 个 Agent 的 AGENTS.md 各加一段（内容统一）。
+
+##### L2：Mail 模板精简 + 事件上下文注入
+
+**原则**：模板只放摘要 + 链接 + 仓库信息，不写固定步骤（步骤在 L1）。
+
+**issue_assigned.md** 改为：
+
+```markdown
+Issue 指派
+
+Issue: {issue_url}
+标题: {issue_title}
+标签: {labels}
+
+📋 获取完整上下文（先读再动手）：
+- Issue 详情: GET {gitea_api}/repos/{repo}/issues/{issue_number}
+- Issue 评论: GET {gitea_api}/repos/{repo}/issues/{issue_number}/comments
+
+仓库: {repo_clone_url}
+建议分支: feat/issue-{issue_number}-{brief}
+```
+
+**review_request.md** 改为：
+
+```markdown
+PR Review 请求
+
+PR: {pr_url}
+标题: {pr_title}
+作者: {pr_author}
+分支: {branch}
+风险级别: {risk_level}
+
+📋 获取完整上下文：
+- PR diff: GET {gitea_api}/repos/{repo}/pulls/{pr_number}.diff
+- PR 文件列表: GET {gitea_api}/repos/{repo}/pulls/{pr_number}/files
+```
+
+**review_result.md** 改为：
+
+```markdown
+Review {result}
+
+PR: {pr_url}
+标题: {pr_title}
+审查者: {reviewer}
+
+{review_body}
+```
+
+**ci_failure.md** 改为：
+
+```markdown
+CI 失败
+
+Issue: {issue_url}
+分支: {branch}
+
+错误摘要:
+{error_summary}
+
+📋 CI 日志: {gitea_url}/{repo}/actions
+修复后 push 会自动重触发 CI。
+```
+
+**deploy_failure.md** 改为：
+
+```markdown
+部署失败
+
+仓库: {repo}
+Commit: {commit_sha}
+
+📋 排查步骤:
+- CI 日志: {gitea_url}/{repo}/actions
+- 服务器: pm2 logs {service_name}
+```
+
+**L2 代码改动**（`toolchain_routes.py`）：
+
+1. 从 Webhook payload 的 `repository` 对象提取 `clone_url` 和 `html_url`
+2. `render_template()` 传入新变量：`gitea_api`、`gitea_url`、`repo_clone_url`
+3. 所有模板变量统一补齐
+
+##### L3：Skill 按需加载（不改 Skill 本身）
+
+git-workflow、code-review 等 Skill 保持不变。
+
+L1 的协作行为段里会引用 Skill 名称（"参考 git-workflow Skill"），Agent 收到 Mail 后根据 L1 的引导自主加载对应 Skill。
+
+**不改 Skill 路由机制**——靠 L1 的文案触发 Agent 的 Skill 路由器匹配。
+
+---
+
+### 三、新增功能：issue_comment @mention 通知
+
+#### 3.1 设计
+
+当前 `_handle_issue_comment` 只处理 `[CI]` 前缀评论。扩展为：
+
+```
+issue_comment 事件
+  ├── 含 [CI] / CI 失败 → 原有 CI 失败通知逻辑
+  └── 含 @username → 解析 @mention → Mail 通知被 @的 Agent
+```
+
+#### 3.2 实现
+
+**`toolchain_routes.py` 新增 `_handle_issue_comment_mention()`**：
+
+```python
+AGENT_IDS = {
+    "zhangfei-dev", "guanyu-dev", "zhaoyun-data",
+    "jiangwei-infra", "simayi-challenger", "pangtong-fujunshi",
+}
+
+# 前缀映射：@张飞 → zhangfei-dev
+# 中文名映射：Agent 在 Gitea Issue 评论中可能用中文名 @mention
+# 英文短名映射：Agent 可能用不带 -dev/-infra 后缀的短名
+AGENT_ALIAS = {
+    "张飞": "zhangfei-dev",
+    "关羽": "guanyu-dev",
+    "赵云": "zhaoyun-data",
+    "姜维": "jiangwei-infra",
+    "司马懿": "simayi-challenger",
+    "庞统": "pangtong-fujunshi",
+    "pangtong": "pangtong-fujunshi",
+    "simayi": "simayi-challenger",
+    "zhangfei": "zhangfei-dev",
+    "guanyu": "guanyu-dev",
+    "zhaoyun": "zhaoyun-data",
+    "jiangwei": "jiangwei-infra",
+}
+
+def extract_mentions(body: str, sender: str) -> list[str]:
+    """从评论 body 中提取 @mention 的 Agent ID"""
+    candidates = re.findall(r"@([a-zA-Z\u4e00-\u9fa5][a-zA-Z0-9\u4e00-\u9fff-]*)", body)
+    result = set()
+    for c in candidates:
+        # 精确匹配
+        if c in AGENT_IDS:
+            result.add(c)
+        # 前缀/别名匹配
+        elif c in AGENT_ALIAS:
+            result.add(AGENT_ALIAS[c])
+        else:
+            # 前缀模糊匹配：@zhangfei → zhangfei-dev
+            for aid in AGENT_IDS:
+                if aid.startswith(c):
+                    result.add(aid)
+                    break
+    # 过滤掉评论者自己
+    result.discard(sender)
+    return list(result)
+```
+
+**新增 mention 通知模板** `templates/toolchain/mention.md`：
+
+```markdown
+你在 Issue 中被 @mention
+
+Issue: {issue_url}
+评论者: {commenter}
+评论内容:
+{comment_body}
+
+📋 获取完整上下文：
+- Issue 详情: GET {gitea_api}/repos/{repo}/issues/{issue_number}
+- Issue 评论: GET {gitea_api}/repos/{repo}/issues/{issue_number}/comments
+```
+
+**改动 `_handle_issue_comment`**：
+
+```python
+async def _handle_issue_comment(payload):
+    comment = payload.get("comment", {})
+    body = comment.get("body", "")
+    sender = comment.get("user", {}).get("login", "")
+    repo = _repo_fullname(payload)
+    issue = payload.get("issue", {})
+
+    # 原有 CI 失败逻辑（不变）
+    if "[CI]" in body or "CI 失败" in body:
+        # ... 原有逻辑 ...
+
+    # 新增：@mention 通知
+    mentions = extract_mentions(body, sender)
+    if mentions:
+        issue_number = issue.get("number", 0)
+        issue_title = issue.get("title", "")
+        text = render_template("mention", {
+            "repo": repo,
+            "issue_number": str(issue_number),
+            "issue_url": issue.get("html_url", ""),
+            "commenter": sender,
+            "comment_body": body[:500],
+            "gitea_api": "http://192.168.2.154:3000/api/v1",
+        })
+        title = f"@mention: {issue_title} ({repo}#{issue_number})"
+        for agent_id in mentions:
+            _send_mail(agent_id, title, text)
+```
+
+#### 3.3 去重
+
+- 同一条评论 @多人：每人一封 Mail（不同 to，内容相同）
+- 同一事件 org webhook + repo webhook 双触发：现有 delivery UUID 去重机制覆盖
+- 同一人被 @多次：`extract_mentions` 返回 set，自动去重
+
+---
+
+### 四、Mail Spawn Prompt 改造
+
+#### 4.1 问题
+
+当前工具链 Mail 走 Mail 通道，spawn prompt 是：
+
+```
+你收到一封飞鸽传书(纯通知)。
+发件者: system
+主题: Issue 指派: xxx
+内容: [工具链模板]
+已阅即可。
+```
+
+"已阅即可"直接让 Agent 不做事。
+
+#### 4.2 方案
+
+**不改 MAIL_INFORM_TEMPLATE / MAIL_REQUEST_TEMPLATE 本身**（那是 Mail 系统通用的）。
+
+改为：**工具链 Mail 使用 `type=request`（而不是默认的 inform）**。
+
+在 `_send_mail()` 中，工具链事件创建的 Mail 默认 `performative=request`，这样 Agent 收到时走 `MAIL_REQUEST_TEMPLATE`，知道需要处理。
+
+具体改动在 `_send_mail()` 函数或其调用处：工具链路由调用 `_send_mail` 时传入 `performative="request"`。
+
+**⚠️ 验证要点**：改为 request 后，Agent spawn prompt 变为 "请处理以下请求"，需确认：
+1. Agent 不再把工具链 Mail 当纯通知忽略
+2. Agent 能正确处理「已阅型」工具链事件（如 CI 失败通知——不需要回复，但需要知道）
+3. 对已关闭 PR/Issue 的延迟通知，Agent 不会尝试去处理
+
+验证方法：部署后发一条 Issue 指派 Mail，观察 Agent 行为是否符合预期。
+
+---
+
+### 五、完整改动清单
+
+| # | 改什么 | 改动内容 | 层 | 风险 |
+|---|--------|---------|---|------|
+| 1 | 6 个 Agent 的 `AGENTS.md` | 加"工具链协作"段（内容统一） | L1 | 低（纯追加） |
+| 2 | `templates/toolchain/issue_assigned.md` | 精简 + 加仓库上下文 + Gitea API 引导 | L2 | 低 |
+| 3 | `templates/toolchain/review_request.md` | 精简 + 加 Gitea API 引导 | L2 | 低 |
+| 4 | `templates/toolchain/review_result.md` | 精简 | L2 | 低 |
+| 5 | `templates/toolchain/ci_failure.md` | 精简 + 加 CI 日志链接 | L2 | 低 |
+| 6 | `templates/toolchain/deploy_failure.md` | 精简 + 加排查步骤 | L2 | 低 |
+| 7 | **新建** `templates/toolchain/mention.md` | @mention 通知模板 | L2 | 低 |
+| 8 | `src/api/toolchain_routes.py` | 提取 clone_url/html_url 传入模板；issue_comment 增加 @mention 解析；工具链 Mail 改为 request 类型 | L2 | 中 |
+| 9 | 不改 | git-workflow 等 Skill 保持不变 | L3 | — |
+| 10 | 不改 | daemon 核心逻辑、BootstrapBuilder、Skill 路由 | — | — |
+
+---
+
+### 六、验证方案
+
+#### 6.1 单元验证
+
+| 验证点 | 方法 |
+|--------|------|
+| `extract_mentions()` 提取 `@zhangfei-dev` | unit test |
+| `extract_mentions()` 别名匹配 `@张飞` → zhangfei-dev | unit test |
+| `extract_mentions()` 前缀匹配 `@zhangfei` → zhangfei-dev | unit test |
+| `extract_mentions()` 过滤自己 | unit test |
+| 模板渲染新变量不报错 | unit test |
+
+#### 6.2 真实场景 E2E 验证
+
+重复 Issue #32 的场景：
+1. 创建 Issue 指派张飞
+2. **验证**：张飞收到的 Mail 含 clone URL + Gitea API 引导
+3. **验证**：张飞 spawn 后知道该做什么（L1 AGENTS.md 有流程引导）
+4. **验证**：张飞有疑问时去 Gitea Issue 评论（而不是 Control UI）
+5. 在 Issue 评论 @赵云
+6. **验证**：赵云收到 @mention Mail
+
+---
+
+### 七、不做的事（标记为后续）
+
+| 标记 | 描述 | 原因 |
+|------|------|------|
+| 后续-1 | Agent 离开工具链讨论后，是否有意识回到工具链 | 需要更多真实场景观察 |
+| 后续-2 | 工具链使用标准在所有 Agent 间的一致性验证 | L1 统一段落是第一步，需要 E2E 验证 |
+| 后续-3 | Mail 通道接入 BootstrapBuilder L2 注入 | 改动大，当前方案（L1 统一段落 + 模板引导）够用 |
+| 后续-4 | Skill 路由器自动触发（引擎注入） | 改动 daemon 核心，当前靠 L1 文案触发 |
+
+---
+
+### 八、变更记录
+
+| 日期 | 版本 | 变更 |
+|------|------|------|
+| 2026-06-09 | v1.0 | 初版：E2E 真实场景暴露问题 → 四层改造方案 + @mention 通知 + Mail type 改造 |
+

docs/design/18-toolchain-e2e-test.md

@@ -0,0 +1,121 @@
+# §18. 工具链端到端验证测试
+
+> 日期：2026-06-09
+> 状态：已完成 ✅
+> 目标：用真实 Webhook 触发验证整条 Mail 通知链路
+
+## 前置确认
+
+- Gitea 用户名 ↔ Agent ID 映射：完全一致（admin, guanyu-dev, jiangwei-infra, pangtong-fujunshi, simayi-challenger, zhangfei-dev, zhaoyun-data）
+- Gitea 组织级 Webhook（Hook ID=28）：姜维确认最近 5 条投递全部 is_succeed=1
+- Daemon 在线：sanguo-moziplus-v2 运行中
+- 测试仓库：sanguo/moziplus-v2
+
+## 命名规范
+
+- Issue 标题：`[E2E-TEST] xxx`
+- PR 标题：`[E2E-TEST] xxx`
+- 分支名：`test/e2e-<timestamp>`
+
+## 验证步骤
+
+| 步骤 | 操作 | 触发事件 | 预期 Mail 通知 | 验证点 |
+|------|------|----------|---------------|--------|
+| 1 | 创建 Issue `[E2E-TEST] Issue指派测试`，assignee=zhangfei-dev | issues (assigned) | zhangfei-dev 收到 "Issue 指派" Mail | Mail to/模板正确 |
+| 2 | 开分支 `test/e2e-<ts>`，创建 PR `[E2E-TEST] Review请求测试` | pull_request (opened) | simayi-challenger 收到 "Review 请求" Mail | Mail to/风险级别/文件列表 |
+| 3 | PR Review APPROVED | pull_request_review (approved) | PR 作者(pangtong-fujunshi) 收到 "Review 通过 ✓" Mail | result=通过 ✓ |
+| 4 | PR Review REQUEST_CHANGES | pull_request_review (rejected) | PR 作者收到 "Review 驳回 ✗" Mail | result=驳回 ✗ |
+| 5 | Issue 上发评论 `[CI] CI 失败 — 分支: test/e2e-xxx, 错误: build timeout` | issue_comment | Issue 作者收到 "CI 失败" Mail | 模板含分支/错误摘要 |
+| 6 | 创建标题含"部署失败"的 Issue（无指派） | issues (opened) | jiangwei-infra + pangtong-fujunshi 各收到 "部署失败" Mail | 双收件人 |
+| 7 | 关闭步骤 1 的 Issue，再发 CI 失败评论 | issue_comment (closed issue) | 不产生 Mail（负面测试） | handler 跳过 closed |
+| 8 | 重发步骤 1 Webhook（相同 delivery ID） | 重复事件 | 不产生新 Mail（幂等测试） | 返回 duplicate |
+
+## 签名校验
+
+已测试（GITEA_WEBHOOK_SECRET 已配置且生效）：
+- ✅ 正确签名：请求正常处理
+- ✅ 无签名：返回 403 `signature verification failed`
+
+## Review 意见来源
+
+- 姜维（基础设施确认 + 边界验证建议）
+- 司马懿（遗漏点补充 + 命名规范 + 风险防范）
+
+---
+
+## 执行记录
+
+> 2026-06-09 00:40~00:50 CST
+
+### 步骤 1：Issue 指派 ✅
+- 操作：创建 Issue #22 `[E2E-TEST] Issue指派测试`，assignee=zhangfei-dev
+- Mail：`mail-1780936736480`，from=system, to=zhangfei-dev, title=`Issue 指派: [E2E-TEST] Issue指派测试`
+- 模板渲染正确（含 Issue 链接、标签、描述、建议分支名）
+
+### 步骤 2：PR Review 请求 ✅
+- 操作：创建分支 `test/e2e-1780936838`，创建 PR #23
+- Mail：`mail-1780936851715`，from=system, to=simayi-challenger
+- 模板含 PR 链接、标题、作者(pangtong-fujunshi)、分支、风险级别(standard)
+- 附带：CI 失败通知 `mail-1780936876572`（CI 自动触发，符合预期）
+
+### 步骤 3：Review APPROVED ✅
+- 操作：用 simayi-challenger token 提交 APPROVED review
+- Mail：`mail-1780936968411`，from=system, to=pangtong-fujunshi, title=`Review 通过 ✓`
+- 描述含审查者(simayi-challenger)、review body
+- ⚠️ 收到 2 封重复 Mail（org webhook + repo webhook 双触发）
+
+### 步骤 4：Review REQUEST_CHANGES ✅
+- 操作：用 simayi-challenger token 提交 REQUEST_CHANGES review
+- Mail：`mail-1780936972207`，from=system, to=pangtong-fujunshi, title=`Review 驳回 ✗`
+- ⚠️ 同上，收到 2 封重复 Mail
+
+### 步骤 5：CI 失败评论 ✅
+- 操作：在 Issue #22 发评论 `[CI] CI 失败 — 分支: test/e2e-1780936838, 错误: build timeout`
+- Mail：`mail-1780936994513`，from=system, to=pangtong-fujunshi, title=`CI 失败: sanguo/moziplus-v2#22`
+- 模板含分支提取和错误摘要
+
+### 步骤 6：部署失败 Issue ✅
+- 操作：创建 Issue #24 `[E2E-TEST] 部署失败: test deploy`（无指派）
+- Mail：`mail-1780936999660` to=jiangwei-infra, `mail-1780936999684` to=pangtong-fujunshi
+- 双收件人验证通过 ✅
+
+### 步骤 7：已关闭 Issue 负面测试 ✅
+- 操作：关闭 Issue #22 后发 `[CI] CI 失败 — 应被过滤`
+- 结果：未产生新 Mail ✅（只有步骤 5 的 1 封 CI Mail，步骤 7 的评论被正确过滤）
+
+### 步骤 8：幂等测试 ✅
+- 操作：构造带正确 HMAC-SHA256 签名的 Webhook，用同一 delivery ID `test-idempotency-002` 发两次
+- 第一次：返回 `ok`，产生 Mail ✅
+- 第二次：返回 `duplicate`，无新 Mail ✅
+- 额外验证：不带签名的请求返回 403 `signature verification failed`（签名校验正常工作）
+
+---
+
+## 汇总
+
+| 步骤 | 状态 | 备注 |
+|------|------|------|
+| 1. Issue 指派 | ✅ 通过 | Mail to/模板正确 |
+| 2. PR Review 请求 | ✅ 通过 | Mail to/风险级别/文件列表正确 |
+| 3. Review APPROVED | ✅ 通过 | E2E 测试中产生 2 封 Mail（根因已查明，非平台问题） |
+| 4. Review REQUEST_CHANGES | ✅ 通过 | 同上 |
+| 5. CI 失败评论 | ✅ 通过 | 分支提取正确 |
+| 6. 部署失败 Issue | ✅ 通过 | 双收件人验证通过 |
+| 7. 已关闭 Issue 过滤 | ✅ 通过 | 负面测试通过，无新 Mail |
+| 8. 幂等测试 | ✅ 通过 | 第二次返回 duplicate，无新 Mail；签名校验正常拦截无签名请求 |
+
+## 发现的问题
+
+### Review 事件双 Mail（已修复）
+- **现象**：E2E 测试步骤 3/4 中 Review 事件产生 2 封 Mail
+- **根因**（姜维深入调查确认）：E2E 测试中庞统手动用 simayi token 提交了 Review，同时 simayi agent 收到 Review 请求 Mail 后也自主提交了 Review。是两次独立的 API 调用，**不是 Gitea bug 或平台配置问题**
+  - 姜维控制实验：一次 review API 调用只产生 1 个 hook_task
+  - Gitea 路由日志确认两次 POST 间隔 7 秒，payload 有差异（review_comments、updated_at 不同）
+  - 之前的错误分析（"Gitea webhookNotifier + actionsNotifier 双投递"）已被推翻：actionsNotifier 走 handleWorkflows() 不创建 hook_task
+- **修复**：payload 内容去重作为防御性编程保留（`_is_duplicate` 新增内容去重 key = event + pr_num + sender + sha256(body_or_content)），司马懿 APPROVED
+- **验证**：PR #27 实测只产生 1 封 Mail ✅
+
+### 根因分析教训
+- 姜维第一次分析给出了错误根因（Gitea 双 notifier），第二次深入调查后自我纠正
+- 庞统把姜维的第一次结论当事实汇报给主公，没有标注"这是姜维的调查结论，尚未独立验证"
+- **改进**：SOUL.md 新增规则——推测 vs 事实显式标注、引用他人结论时标注来源、结论被推翻时及时更正

docs/design/step5-audit-report.md

@@ -0,0 +1,74 @@
+# Step 5 双重审计报告
+
+## 摘要
+- 设计一致性检查项: 8
+- 特殊逻辑覆盖检查项: 22
+- 一致/覆盖: 24
+- **偏差/遗漏: 6（严重 3 / 轻微 3）**
+
+---
+
+## 偏差/遗漏清单
+
+| # | 维度 | 设计要求 / 旧逻辑 | 代码实际 | 严重程度 | 建议 |
+|---|------|-------------------|---------|---------|------|
+| **D1** | B1.2 pre_spawn | 旧 `_mail_on_checks_passed`: `if not _mail_auto_working(): raise RuntimeError` — pre_spawn 失败时中止 spawn | 新 `_handler_on_checks_passed`: `_handler.pre_spawn(...)` 返回值未检查，`handler_marked_working = True` 无条件执行 | **严重** | 改为 `if not _handler.pre_spawn(...): raise RuntimeError("handler_pre_spawn_failed")` |
+| **D2** | B3.1 PromptContext | 旧 `_build_mail_prompt` 从 must_haves JSON 解析 `from_agent` 和 `performative` 传入模板 | 新 `spawner._build_spawn_message` 构建 PromptContext 时缺少 `from_agent` 和 `mail_type`，均为空字符串 | **严重** | 从 `must_haves` JSON 提取 `from` 和 `performative` 填入 PromptContext |
+| **D3** | B1.3 inform outcome 白名单 | 旧 `_mail_auto_complete`: inform 类型有 outcome 白名单 `{"completed", "claimed", "no_reply"}`，不在白名单的 outcome 跳过 auto-done | 新 `MailHandler.verify_completion`: inform 始终返回 True，不检查 outcome | **轻微** | CRASH_OUTCOMES 已被基类处理。剩余异常 outcome（session_revived/api_error/fallback_timeout）极少出现，且旧逻辑不标 done 只是等 ticker 重投，最终效果差异不大。但严格对齐需要加白名单检查 |
+| **D4** | A. 设计 §6 retry 逻辑 | 设计文档要求 retry 逻辑中 `handler = TaskTypeRegistry.get_by_project(project_id); if handler: return handler.build_retry_prompt(...)` | spawner L1118-1130 重试 prompt 仍用 `is_mail = project_id == "_mail"` 硬编码 | **轻微** | 当前不影响运行（旧的 `_build_mail_prompt` 仍保留且可用），但与设计文档不一致 |
+| **D5** | B1.5 _check_reply 语义差异 | 旧 `_mail_check_reply`: `SELECT id FROM tasks WHERE id != ? AND must_haves LIKE ?` — 检查是否有其他任务的 must_haves 包含当前 task_id（即 in_reply_to 匹配） | 新 `MailHandler._check_reply`: `SELECT COUNT(*) FROM comments WHERE task_id=? AND author != 'daemon' AND comment_type != 'system'` — 检查当前任务是否有非系统 comment | **严重** | 两个查询语义完全不同。旧逻辑检查的是 **mail 表的回复任务**（通过 must_haves 中 in_reply_to 关联），新逻辑检查的是 **当前任务的 comments**。这可能导致 request 类型邮件的幻觉门控行为不同 |
+| **D6** | B1.3 标 done 重试机制 | 旧 `_mail_auto_complete`: 标 done 时外层有 `for attempt in range(3)` 循环 | 新 `BaseTaskHandler._mark_task_status`: H1 修复后已有 3 次重试 | **轻微** | ✅ 已修复，但注意旧代码标 done 和标 failed 是分开的重试循环，新代码统一走 `_mark_task_status`。行为等价 |
+
+---
+
+## 一致确认项
+
+### A. 设计一致性
+
+| # | 维度 | 检查点 | 结果 |
+|---|------|--------|------|
+| A1 | §6 dispatcher | classify_outcome 后调 handler.post_complete | ✅ on_complete 闭包替换为 handler.post_complete |
+| A2 | §6 dispatcher | on_checks_passed → handler.pre_spawn | ✅ _handler_on_checks_passed 调用 handler.pre_spawn（但返回值未检查，见 D1） |
+| A3 | §6 dispatcher | guardrail 跳过 → handler 判断 | ✅ `is_handler_task = handler is not None` |
+| A4 | §6 spawner | _build_prompt → handler.build_prompt | ✅ handler 路径调用 handler.build_prompt(ctx) |
+| A5 | §6 spawner | _build_api_section → handler 查询 | ✅ handler 存在时 success_status 从 handler.target_success_status 获取 |
+| A6 | §6 ticker | 虚拟项目扫描 → registry.virtual_projects() | ✅ 循环 `TaskTypeRegistry.virtual_projects()` |
+| A7 | §6 ticker | check_completion → handler.check_completion | ✅ 超时检查中调 `handler.check_completion(task.id, db_path)` |
+| A8 | §6 兼容期 | 设计说"兼容期保留旧逻辑" | ✅ 无 handler 的项目走旧路径（legacy_on_complete） |
+
+### B. 特殊逻辑覆盖
+
+| # | 维度 | 检查点 | 结果 |
+|---|------|--------|------|
+| B1 | 1.1 guardrail | handler 项目跳过，_general 等走 guardrail | ✅ |
+| B2 | 1.2 _mail_auto_working | `BEGIN IMMEDIATE` + status 检查 + 标 working | ✅ `_auto_mark_working` 完全一致 |
+| B3 | 1.3 request 无回复 → 标 failed + notify | ✅ MailHandler.on_failure 调 `_mark_task_status(failed)` + `notify_mail_failed` |
+| B4 | 1.4 _mail_revert_to_pending | spawn 失败回退 working → pending | ✅ Exception handler 中有 `BEGIN IMMEDIATE` + 状态检查回退 |
+| B5 | 1.6 Task review verdict 读取 | approved → done | ✅ handle_review_complete |
+| B6 | 1.6 Task review 非 approved → @mention assignee + 保持 review | ✅ H3 修复后保持 review + INSERT comment with comment_type='review' |
+| B7 | 1.6 Task executor 三信号验证 | output/comment/terminal status → review | ✅ verify_completion 完全一致 |
+| B8 | 1.7 Legacy dispatch 路径 | handler 替代 is_mail_legacy | ✅ handler_legacy 查注册表 |
+| B9 | 2.1 _transition_status assignee 清空 | handler 项目不清空 | ✅ |
+| B10 | 2.2 跳过 claimed 状态 | handler 项目跳过 claimed 直接 working | ✅ |
+| B11 | 2.3 _dispatch_reviews 跳过 | handler 项目不走 review | ✅ |
+| B12 | 2.5 startup recovery | `_general` + virtual_projects() | ✅ 不会重复扫描 |
+| B13 | 3.1 _build_api_section | handler 存在时正确获取 success_status | ✅ |
+| B14 | B4.1 TaskHandler.post_complete | 区分 executor/review 流程 | ✅ 通过读 DB status 判断 |
+| B15 | B4.2 MailHandler.post_complete | 基类统一流程 | ✅ |
+| B16 | B4.3 ToolchainHandler.post_complete | 基类统一流程 | ✅ |
+| B17 | B1.5 _check_reply 异常保守处理 | 旧: return True（保守）/ 新: return False | 见 D5 |
+| B18 | CRASH_OUTCOMES 集合 | 与旧 ROLLBACK_CURRENT_AGENT_OUTCOMES 一致 | ✅ 完全一致 |
+| B19 | B2.1 _toolchain ticker 扫描 | _toolchain 会被 ticker 扫描 | ✅ _toolchain 有 blackboard.db 时会被 tick_project 处理 |
+| B20 | B2.3 handler 项目都跳过 claimed | _toolchain 也跳过 | ✅ 所有 handler 项目统一处理 |
+
+---
+
+## 修复优先级
+
+| 优先级 | # | 修复内容 |
+|--------|---|---------|
+| **P0** | D1 | dispatcher _handler_on_checks_passed 检查 pre_spawn 返回值 |
+| **P0** | D2 | spawner PromptContext 从 must_haves 提取 from_agent 和 mail_type |
+| **P0** | D5 | MailHandler._check_reply 恢复旧查询语义（检查 must_haves 中的 in_reply_to） |
+| P1 | D3 | inform outcome 白名单（可选，影响极小） |
+| P2 | D4 | retry prompt 用 handler 路径替代硬编码 |

docs/design/step5-impact-analysis.md

@@ -0,0 +1,324 @@
+# Step 5 引擎接入 — 影响分析与逐点对照
+
+## 方法论
+
+逐行审查 dispatcher.py / spawner.py / ticker.py 中所有 `is_mail` / `_mail` / `project_id == "_mail"` 分支，
+对照 handler 实现，确认每个特殊处理的去向。
+
+---
+
+## 一、dispatcher.py（985 行）
+
+### 1.1 Guardrail 跳过（L127-129）
+
+```python
+is_mail = project_config.get("project_id") == "_mail" if project_config else False
+if self.guardrails and not is_mail:
+    violations = self.guardrails.check_task(task)
+```
+
+**特殊处理**：Mail 不做 guardrail 检查。
+
+**Handler 覆盖**：设计文档 D6 "skip_guardrail 从接口删除，guardrail 自己判断"。Step 5 改为：`if self.guardrails and handler is None`（无 handler 时走 guardrail），或者用 handler.virtual_project 判断。handler 存在时跳过 guardrail。
+
+**改动**：`is_mail` → `TaskTypeRegistry.get_by_project(project_id) is not None`
+
+---
+
+### 1.2 Mail on_checks_passed（L194-213）
+
+```python
+on_checks_passed = None
+_mail_marked_working = False
+if is_mail and db_path:
+    def _mail_on_checks_passed():
+        nonlocal _mail_marked_working
+        if not _disp._mail_auto_working(_task_id, _mail_db):
+            raise RuntimeError("mail_auto_working_failed")
+        _mail_marked_working = True
+    on_checks_passed = _mail_on_checks_passed
+```
+
+**特殊处理**：Mail spawn 前通过 on_checks_passed 回调标 working，标记成功后才 spawn，spawn 失败回退。
+
+**Handler 覆盖**：MailHandler.pre_spawn 调用 `_auto_mark_working`，和 `_mail_auto_working` 逻辑完全一致。
+
+**改动**：
+- `on_checks_passed` 改为调用 `handler.pre_spawn(task_id, db_path)`
+- `_mail_marked_working` 标记保留，用于 Exception 回退
+
+---
+
+### 1.3 Mail on_complete（L224-238）
+
+```python
+if is_mail:
+    def _mail_on_complete(aid, outcome):
+        _dispatcher._mail_auto_complete(_task_id, aid, _mail_db, _must_haves, outcome=outcome)
+    on_complete = _mail_on_complete
+```
+
+**特殊处理**：Mail on_complete 调用 `_mail_auto_complete`（含 inform/request 分支、幻觉门控、重试 3 次、失败通知）。
+
+**Handler 覆盖**：MailHandler 使用基类 post_complete 统一流程（crash→verify→mark→notify）。但现有 `_mail_auto_complete` 有几个细节差异需要注意：
+
+| 现有逻辑 | Handler 覆盖 | 差异 |
+|---------|-------------|------|
+| request 无回复 → 重试 3 次标 failed | on_failure 标 failed + notify | ⚠️ 缺少 3 次重试 |
+| inform 只在特定 outcome 标 done | verify 始终返回 True → 基类标 done | ✅ 简化了，合理 |
+| 标 done 重试 3 次 | _mark_task_status 单次 | ⚠️ 缺少重试 |
+| notify_mail_failed | on_failure 中调用 notify_mail_failed | ✅ 一致 |
+
+**⚠️ 关键发现**：现有代码标状态时有 **重试 3 次** 机制（防止 DB 锁），handler 的 `_mark_task_status` 只做一次。需要把重试逻辑补到 `_mark_task_status` 或在 handler 层加。
+
+**改动**：on_complete 改为调用 `handler.post_complete(task_id, agent_id, outcome, db_path)`
+
+---
+
+### 1.4 Task on_complete（L241-310）
+
+```python
+else:
+    def _task_on_complete(aid, outcome):
+        # #07.2: crash 回退
+        if outcome in ROLLBACK_CURRENT_AGENT_OUTCOMES and _task_db:
+            _dispatcher._rollback_current_agent(_task_db, _task_id, aid)
+        
+        if _is_review:
+            if outcome in ("completed", "session_revived"):
+                # 读 verdict → approved 标 done / 非 approved @mention assignee
+            else:
+                logger.warning("review agent outcome=%s, NOT marking done", outcome)
+        else:
+            # executor: 三信号验证 → 标 review
+            _dispatcher._task_auto_complete(_task_id, _task_db)
+```
+
+**特殊处理清单**：
+
+1. **#07.2 crash 回退**：executor 和 review 都回退 current_agent → assignee
+2. **review 分支**：outcome 必须是 "completed" 或 "session_revived" 才走 verdict 读取
+3. **review verdict 读取**：approved → done，非 approved → @mention assignee + 保持 review
+4. **review @mention**：通过 Blackboard.add_comment，comment_type="review"
+5. **executor 分支**：走 _task_auto_complete → 三信号验证 → review
+
+**Handler 覆盖**：
+- crash 回退：✅ BaseTaskHandler.post_complete 第一步
+- review verdict：⚠️ **TaskHandler.handle_review_complete 存在但未被 dispatcher 调用**。现有 dispatcher 直接在闭包里做了，不走 handler。
+- @mention：⚠️ handler 用 `conn.execute("INSERT INTO comments")` 直接插入，dispatcher 用 `Blackboard.add_comment`（会做更多处理，如 comment_type="review"）
+- executor 三信号：✅ TaskHandler.verify_completion
+
+**⚠️ 关键发现**：
+1. dispatcher 的 review @mention 用 `bb.add_comment(..., comment_type="review")`，handler 直接 INSERT 不带 comment_type。需要修复 handler。
+2. dispatcher 对 review outcome 有白名单检查（只处理 "completed"/"session_revived"），handler 的 post_complete 没有 outcome 白名单——crash 已在基类处理，其他 outcome 都会走 verify。
+3. dispatcher review 非 approved 时**保持 review 状态**，handler 的 handle_review_complete 标回 working。这是**行为差异**。
+
+**改动**：需要先修复 handler 的 review 分支，再替换 on_complete。
+
+---
+
+### 1.5 Mail spawn 失败回退（L355-358）
+
+```python
+except Exception as e:
+    if _mail_marked_working:
+        self._mail_revert_to_pending(task.id, db_path)
+```
+
+**特殊处理**：spawn 失败（subprocess 启动失败）回退 working → pending。
+
+**Handler 覆盖**：❌ handler 没有这个。这是 dispatcher 级别的异常处理，和 handler 无关。但 toolchain 也需要类似逻辑。
+
+**改动**：保留在 dispatcher 中，改为 `_mail_marked_working` → `handler_marked_working`。
+
+---
+
+### 1.6 Legacy dispatch（L584-660）
+
+```python
+is_mail_legacy = project_config.get("project_id") == "_mail"
+if is_mail_legacy:
+    if not self._mail_auto_working(task.id, db_path_legacy):
+        return error
+```
+
+**特殊处理**：legacy 路径（router=None 时触发）也有 mail 特殊处理。
+
+**Handler 覆盖**：同 1.2/1.3，用 handler 替代。
+
+**改动**：同样用 handler.pre_spawn 和 handler.post_complete 替代。
+
+---
+
+### 1.7 现有 Mail 辅助方法（L658-870）
+
+`_mail_auto_working` / `_mail_revert_to_pending` / `_mail_auto_complete` / `_mail_check_reply`
+
+**改动**：Step 5 不删这些方法（安全起见保留，标记 deprecated），只改调用方。确认稳定后再删。
+
+---
+
+## 二、spawner.py（1704 行）
+
+### 2.1 _build_prompt 中的 mail 分支（L282-284）
+
+```python
+if project_id == "_mail":
+    return self._build_mail_prompt(task_id, title, description, must_haves, agent_id)
+```
+
+**特殊处理**：Mail 用专用精简模板。
+
+**Handler 覆盖**：MailHandler.build_prompt 通过 PromptComposer 拼 3 个 section。
+
+**改动**：查注册表 → handler.build_prompt(context)。需要构建 PromptContext 传入。
+
+---
+
+### 2.2 _build_api_section（L321-325）
+
+```python
+success_status = '"done"' if project_id == "_mail" else '"review"'
+```
+
+**特殊处理**：Mail 的 success_status 是 done。
+
+**Handler 覆盖**：已由 handler 的 PromptSection 处理（TaskApiSection hardcode review，MailApiSection 不含 status 回写指令）。
+
+**改动**：如果 handler 存在，跳过 _build_api_section（handler.build_prompt 已包含）。
+
+---
+
+### 2.3 classify_outcome 中的 handler 调用
+
+spawner 在 classify_outcome 后调 on_complete(outcome)。on_complete 是 dispatcher 传入的闭包。
+
+**改动**：on_complete 闭包改为调用 handler.post_complete。spawner 本身不直接查注册表。
+
+---
+
+## 三、ticker.py（1897 行）
+
+### 3.1 虚拟项目扫描（L218-229）
+
+```python
+mail_db = Path(self.registry.root) / "_mail" / "blackboard.db"
+if mail_db.exists() and "_mail" not in active_projects:
+    pr = await self._tick_project("_mail", {...})
+```
+
+**特殊处理**：_mail 硬编码扫描。
+
+**Handler 覆盖**：TaskTypeRegistry.virtual_projects() 返回 ["_toolchain", "_mail"]。
+
+**改动**：循环 `TaskTypeRegistry.virtual_projects()` 替代硬编码。_toolchain 如果也需要 ticker 扫描就自动发现。但需确认 _toolchain 是否需要 ticker——当前 toolchain 任务创建和完成都在 toolchain_routes.py 中处理，可能不需要 ticker 扫描。
+
+---
+
+### 3.2 _transition_status 中 mail assignee 不清空（L953-960）
+
+```python
+if new_status == "pending":
+    if self._current_project_id == "_mail":
+        # Mail 的 assignee 是收件人，永不清空
+        conn.execute("UPDATE tasks SET status=?, updated_at=? WHERE id=?", ...)
+    else:
+        conn.execute("UPDATE tasks SET status=?, assignee=NULL, ...", ...)
+```
+
+**特殊处理**：Mail 重置到 pending 时不清空 assignee（assignee 是收件人）。
+
+**Handler 覆盖**：❌ handler 不管 ticker 的状态转换逻辑。这是 ticker 内部逻辑。
+
+**改动**：用 `TaskTypeRegistry.get_by_project(project_id)` 判断替代硬编码。
+
+---
+
+### 3.3 Mail 跳过 claimed 状态（L1029-1043）
+
+```python
+if project_id == "_mail":
+    conn.execute("UPDATE tasks SET current_agent=? WHERE id=?", ...)
+    # 跳过 claimed，直接 working
+```
+
+**特殊处理**：Mail 不走 claimed 中间态（已在 dispatcher 中标 working）。
+
+**Handler 覆盖**：handler.pre_spawn 的 _auto_mark_working 跳过了 claimed。
+
+**改动**：用 handler 判断替代硬编码。
+
+---
+
+### 3.4 _dispatch_reviews 跳过 mail（L1304）
+
+```python
+if project_id == "_mail":
+    return []
+```
+
+**特殊处理**：Mail 不走 review 流程。
+
+**Handler 覆盖**：MailHandler.target_success_status = "done"，不走 review。但 ticker 的 _dispatch_reviews 是看项目级。
+
+**改动**：用 handler 判断。
+
+---
+
+### 3.5 Mail 幻觉门控兜底（L1474-1492）
+
+```python
+if self._current_project_id == "_mail":
+    has_reply = self._mail_check_reply(task.id, db_path)
+    if has_reply:
+        # working → done
+```
+
+**特殊处理**：Ticker 超时检查时，如果 mail 有回复，标 done 而非 failed。
+
+**Handler 覆盖**：❌ handler 的 check_completion 只返回 bool，不做状态标记。
+
+**改动**：调用 handler.check_completion 替代 _mail_check_reply。状态标记逻辑保留在 ticker 中。
+
+---
+
+### 3.6 _mail_check_reply（L1555-1575）
+
+和 dispatcher 版本一致。
+
+**改动**：用 handler.check_completion 替代。
+
+---
+
+### 3.7 虚拟项目 init + recovery 扫描（L1625-1643）
+
+```python
+for virtual_id in ("_general", "_mail"):
+    ...
+    # _mail 项目不清空 assignee
+```
+
+**改动**：virtual_projects() + _general 硬编码。
+
+---
+
+## 四、Handler 缺陷（需在 Step 5 前修复）
+
+| # | 缺陷 | 影响 | 修复方案 |
+|---|------|------|---------|
+| H1 | BaseTaskHandler._mark_task_status 无重试 | DB 锁时标状态失败，任务卡住 | 加 3 次重试（和 dispatcher 现有行为一致） |
+| H2 | TaskHandler.handle_review_complete 中 @mention 不带 comment_type="review" | review comment 无类型标记 | INSERT 加 comment_type |
+| H3 | dispatcher review 非 approved 保持 review 状态，handler 标 working | **行为差异** | handler 改为保持 review 状态（和 dispatcher 一致） |
+| H4 | dispatcher review outcome 有白名单（"completed"/"session_revived"），handler 无 | crash 之外的异常 outcome 也会走 verify | handler 的 post_complete 已在基类处理 crash，其余 outcome 走 verify 是合理的 |
+
+**H3 最关键**——dispatcher review 非 approved 保持 review 状态（等 assignee 自己处理），handler 标 working 会触发 ticker 重新 dispatch executor，这不是预期行为。
+
+## 五、改动策略
+
+**不删旧代码，只改调用方**：
+1. dispatcher 中 is_mail → handler 判断，on_checks_passed/on_complete → handler.pre_spawn/post_complete
+2. spawner 中 _build_prompt → handler.build_prompt
+3. ticker 中虚拟项目扫描 → registry.virtual_projects()，mail 特殊判断 → handler 判断
+4. 旧方法（_mail_auto_working 等）标记 @deprecated 保留，不删
+
+**先修 handler 缺陷（H1-H3），再改引擎**。

docs/test-guide.md

@@ -11,9 +11,10 @@
 | 场景 | 命令 | 耗时 | 说明 |
 |------|------|------|------|
 | **改了某个模块** | `pytest tests/unit/test_spawner.py` | <5s | 只跑改动的模块对应的单元测试 |
-| **改了 API 层** | `pytest tests/integration/` | ~1min | 跑全部集成测试 |
-| **提交前快速验证** | `pytest -m "not e2e"` | ~2min | 不跑 E2E，验证不破坏现有功能 |
-| **部署前全量验证** | `RUN_INTEGRATION=1 pytest` | ~60min | 含 E2E，真实 Agent |
+| **改了 API 层** | `RUN_INTEGRATION=1 pytest tests/integration/` | ~1min | 跑全部集成测试 |
+| **提交前快速验证** | `pytest` | ~2min | 默认排除 integration 和 e2e |
+| **含集成测试** | `RUN_INTEGRATION=1 pytest` | ~5min | 包含 integration 测试 |
+| **部署前全量验证** | `RUN_INTEGRATION=1 pytest` | ~60min | 含 e2e，真实 Agent |
 | **只跑 E2E 场景** | `RUN_INTEGRATION=1 pytest tests/e2e/test_e2e_scenarios.py` | ~30min | 串行，一个跑完再下一个 |
 | **只跑 E2E 压力** | `RUN_INTEGRATION=1 pytest tests/e2e/test_e2e_stress.py` | ~10min | 并发测试 |
 
@@ -101,7 +102,7 @@ E2E（慢，真实 Agent）    → 验证完整链路，需要 RUN_INTEGRATION=1
 ## 关键规则
 
 1. **只有 E2E 会 spawn 真实 Agent**，单元和集成不会
-2. **不带 `RUN_INTEGRATION=1` 跑 `pytest` 是安全的**，E2E 全部 skip
+2. **直接跑 `pytest` 是安全的**，integration 和 e2e 全部被排除（需 `RUN_INTEGRATION=1` 才跑）
 3. **E2E 场景测试串行**，一个完成再下一个，失败要分析根因再继续
 4. **E2E 压力测试并行**，场景测试全通过后再跑
 5. **测试数据用 `e2e-` 前缀**，atexit 兜底清理，手动清理见上方

pyproject.toml

@@ -8,8 +8,10 @@ requires-python = ">=3.9"
 asyncio_mode = "auto"
 testpaths = ["tests"]
 markers = [
-    "integration: real agent tests (requires RUN_INTEGRATION=1)",
+    "integration: integration tests (requires RUN_INTEGRATION=1)",
+    "e2e: end-to-end tests with real daemon + Agent (requires RUN_INTEGRATION=1)",
 ]
+# Default deselection of integration/e2e handled in conftest.py pytest_collection_modifyitems
 
 [tool.pyright]
 venvPath = "."

src/api/blackboard_routes.py

@@ -5,14 +5,14 @@ from __future__ import annotations
 import json
 import os
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Optional
 
 from fastapi import APIRouter, HTTPException, Query
 
 from src.blackboard.operations import Blackboard
 from src.blackboard.models import Task, Review
 from src.blackboard.queries import Queries
-from src.blackboard.db import VALID_STATUSES, VALID_TRANSITIONS, COMMENT_TYPES, OUTPUT_TYPES
+from src.blackboard.db import VALID_STATUSES, OUTPUT_TYPES
 from src.blackboard.registry import ProjectRegistry
 
 from src.utils import get_data_root
@@ -59,7 +59,10 @@ async def list_tasks(project_id: str,
                      assignee: Optional[str] = None,
                      parent_task: Optional[str] = None):
     bb = _bb(project_id)
-    tasks = bb.list_tasks(status=status, assignee=assignee, parent_task=parent_task)
+    tasks = bb.list_tasks(
+        status=status,
+        assignee=assignee,
+        parent_task=parent_task)
     return {"tasks": [_task_to_dict(t) for t in tasks]}
 
 
@@ -79,10 +82,12 @@ async def get_task(project_id: str, task_id: str,
             result["outputs_count"] = detail.get("outputs_count", 0)
             result["review_status"] = detail.get("review_status")
             result["latest_event_detail"] = detail.get("latest_event_detail")
-        result["comments"] = [dict(c.__dict__) for c in bb.get_comments(task_id)]
+        result["comments"] = [dict(c.__dict__)
+                              for c in bb.get_comments(task_id)]
         result["outputs"] = [dict(o.__dict__) for o in bb.get_outputs(task_id)]
         result["reviews"] = [dict(r.__dict__) for r in bb.get_reviews(task_id)]
-        result["decisions"] = [dict(d.__dict__) for d in bb.get_decisions(task_id)]
+        result["decisions"] = [dict(d.__dict__)
+                               for d in bb.get_decisions(task_id)]
         result["events"] = q.task_events(task_id)
         result["experiences"] = q.task_experiences(task_id)
     return result
@@ -134,7 +139,8 @@ async def create_task(project_id: str, body: Dict[str, Any]):
         priority=body.get("priority", 5),
         assignee=assignee,
         assigned_by=body.get("assigned_by", "user"),
-        depends_on=json.dumps(body["depends_on"]) if "depends_on" in body else None,
+        depends_on=json.dumps(
+            body["depends_on"]) if "depends_on" in body else None,
         parent_task=body.get("parent_task"),
         risk_level=body.get("risk_level", "standard"),
         stage=body.get("stage"),
@@ -175,7 +181,8 @@ async def _generate_title(description: str) -> str | None:
         resp = client.chat.completions.create(
             model=model,
             messages=[
-                {"role": "system", "content": "你是一个任务标题生成器。根据用户的需求描述，生成一个简洁的中文标题（5-15字），只输出标题，不要任何其他内容。"},
+                {"role": "system",
+                 "content": "你是一个任务标题生成器。根据用户的需求描述，生成一个简洁的中文标题（5-15字），只输出标题，不要任何其他内容。"},
                 {"role": "user", "content": description[:500]},
             ],
             max_tokens=50,
@@ -187,7 +194,8 @@ async def _generate_title(description: str) -> str | None:
             return title
     except Exception as e:
         import logging
-        logging.getLogger("moziplus-v2").warning(f"Title generation failed: {e}")
+        logging.getLogger(
+            "moziplus-v2").warning(f"Title generation failed: {e}")
     return None
 
 
@@ -205,7 +213,8 @@ async def task_progress(project_id: str, task_id: str):
 async def claim_task(project_id: str, task_id: str, body: Dict[str, Any]):
     bb = _bb(project_id)
     if not bb.claim_task(task_id, body["agent"]):
-        raise HTTPException(409, "Claim failed (already claimed or wrong assignee)")
+        raise HTTPException(
+            409, "Claim failed (already claimed or wrong assignee)")
     return {"ok": True}
 
 
@@ -240,7 +249,7 @@ async def update_status(project_id: str, task_id: str, body: Dict[str, Any]):
         })
 
     if not bb.update_task_status(task_id, new_status,
-                                  agent=body.get("agent")):
+                                 agent=body.get("agent")):
         raise HTTPException(409, {
             "error": "transition_failed",
             "detail": f"Status update failed for {task_id}",
@@ -265,6 +274,7 @@ async def update_status(project_id: str, task_id: str, body: Dict[str, Any]):
 # --- @mention 自动提取（#04） ---
 _KNOWN_AGENT_IDS: list = []
 
+
 def _init_agent_ids():
     """从配置文件加载 Agent ID 列表"""
     global _KNOWN_AGENT_IDS
@@ -272,18 +282,32 @@ def _init_agent_ids():
         return
     try:
         import yaml
-        cfg_path = os.path.join(os.path.dirname(__file__), "..", "..", "config", "default.yaml")
+        cfg_path = os.path.join(
+            os.path.dirname(__file__),
+            "..",
+            "..",
+            "config",
+            "default.yaml")
         with open(cfg_path) as f:
             cfg = yaml.safe_load(f)
-        _KNOWN_AGENT_IDS = list(cfg.get("daemon", {}).get("agent_profiles", {}).keys())
+        _KNOWN_AGENT_IDS = list(
+            cfg.get(
+                "daemon",
+                {}).get(
+                "agent_profiles",
+                {}).keys())
     except Exception:
         _KNOWN_AGENT_IDS = []
 
+
 def _extract_mentions(text: str) -> list:
     """从文本中自动提取 @agent-id 格式的 mention"""
     import re
     _init_agent_ids()
-    candidates = set(re.findall(r'@([a-z][a-z0-9]*(?:-[a-z][a-z0-9]*)+)', text))
+    candidates = set(
+        re.findall(
+            r'@([a-z][a-z0-9]*(?:-[a-z][a-z0-9]*)+)',
+            text))
     return [a for a in candidates if a in _KNOWN_AGENT_IDS]
 
 
@@ -317,8 +341,8 @@ async def add_comment(project_id: str, task_id: str, body: Dict[str, Any]):
     merged_mentions = list(set(explicit_mentions + auto_mentions))
 
     cid = bb.add_comment(task_id, body["author"], comment_body,
-                          comment_type=body.get("comment_type", "general"),
-                          mentions=merged_mentions)
+                         comment_type=body.get("comment_type", "general"),
+                         mentions=merged_mentions)
     if merged_mentions:
         bb.record_mentions(cid, task_id, merged_mentions)
     # #10: SSE 通知前端黑板有新 comment
@@ -395,7 +419,8 @@ async def write_output(project_id: str, task_id: str, body: Dict[str, Any]):
         )
         os.makedirs(artifacts_dir, exist_ok=True)
         # 安全文件名
-        safe_name = "".join(c if c.isalnum() or c in "._-" else "_" for c in title)
+        safe_name = "".join(
+            c if c.isalnum() or c in "._-" else "_" for c in title)
         if not safe_name:
             safe_name = "output"
         file_path = os.path.join(artifacts_dir, safe_name)
@@ -424,8 +449,8 @@ async def get_decisions(project_id: str, task_id: str):
 async def add_decision(project_id: str, task_id: str, body: Dict[str, Any]):
     bb = _bb(project_id)
     did = bb.add_decision(task_id, body["decider"], body["decision"],
-                           body["rationale"],
-                           alternatives=body.get("alternatives"))
+                          body["rationale"],
+                          alternatives=body.get("alternatives"))
     return {"ok": True, "decision_id": did}
 
 
@@ -435,7 +460,7 @@ async def add_decision(project_id: str, task_id: str, body: Dict[str, Any]):
 async def add_observation(project_id: str, task_id: str, body: Dict[str, Any]):
     bb = _bb(project_id)
     oid = bb.add_observation(task_id, body["observer"], body["body"],
-                              severity=body.get("severity", "info"))
+                             severity=body.get("severity", "info"))
     return {"ok": True, "observation_id": oid}
 
 

src/api/checkpoint_routes.py

@@ -12,7 +12,9 @@ from typing import Optional
 from src.blackboard.operations import Blackboard
 from src.utils import get_data_root
 
-router = APIRouter(prefix="/api/projects/{project_id}/tasks/{task_id}/checkpoints", tags=["checkpoints"])
+router = APIRouter(
+    prefix="/api/projects/{project_id}/tasks/{task_id}/checkpoints",
+    tags=["checkpoints"])
 
 
 # ── 请求模型 ──
@@ -50,10 +52,12 @@ def list_checkpoints(project_id: str, task_id: str):
 
 
 @router.post("")
-def create_checkpoint(project_id: str, task_id: str, req: CreateCheckpointRequest):
+def create_checkpoint(project_id: str, task_id: str,
+                      req: CreateCheckpointRequest):
     """Agent 创建 checkpoint"""
     if req.type not in ("verify", "decision", "action"):
-        raise HTTPException(status_code=400, detail=f"Invalid checkpoint type: {req.type}")
+        raise HTTPException(status_code=400,
+                            detail=f"Invalid checkpoint type: {req.type}")
 
     bb = _bb(project_id)
     # 验证 task 存在
@@ -73,10 +77,15 @@ def create_checkpoint(project_id: str, task_id: str, req: CreateCheckpointReques
 
 
 @router.post("/{checkpoint_id}/approve")
-def approve_checkpoint(project_id: str, task_id: str, checkpoint_id: str, req: ResolveCheckpointRequest):
+def approve_checkpoint(project_id: str, task_id: str,
+                       checkpoint_id: str, req: ResolveCheckpointRequest):
     """用户通过 checkpoint → 自动推进 task 状态"""
     bb = _bb(project_id)
-    result = bb.resolve_checkpoint(checkpoint_id, "approve", req.resolved_by, req.note)
+    result = bb.resolve_checkpoint(
+        checkpoint_id,
+        "approve",
+        req.resolved_by,
+        req.note)
     if result is None:
         raise HTTPException(status_code=404, detail="Checkpoint not found")
     if "error" in result:
@@ -97,10 +106,15 @@ def approve_checkpoint(project_id: str, task_id: str, checkpoint_id: str, req: R
 
 
 @router.post("/{checkpoint_id}/reject")
-def reject_checkpoint(project_id: str, task_id: str, checkpoint_id: str, req: ResolveCheckpointRequest):
+def reject_checkpoint(project_id: str, task_id: str,
+                      checkpoint_id: str, req: ResolveCheckpointRequest):
     """用户驳回 checkpoint → task 回到 working"""
     bb = _bb(project_id)
-    result = bb.resolve_checkpoint(checkpoint_id, "reject", req.resolved_by, req.note)
+    result = bb.resolve_checkpoint(
+        checkpoint_id,
+        "reject",
+        req.resolved_by,
+        req.note)
     if result is None:
         raise HTTPException(status_code=404, detail="Checkpoint not found")
     if "error" in result:

src/api/mail_routes.py

@@ -9,7 +9,7 @@ from __future__ import annotations
 import json
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Optional
 
 from fastapi import APIRouter, HTTPException, Query
 
@@ -34,7 +34,9 @@ def _get_valid_agents() -> set:
         except Exception:
             pass
     # fallback：硬编码
-    return {"zhangfei-dev", "guanyu-dev", "zhaoyun-data", "jiangwei-infra", "pangtong-fujunshi", "simayi-challenger"}
+    return {"zhangfei-dev", "guanyu-dev", "zhaoyun-data",
+            "jiangwei-infra", "pangtong-fujunshi", "simayi-challenger"}
+
 
 router = APIRouter(prefix="/api/mail", tags=["mail"])
 
@@ -97,7 +99,10 @@ async def list_mail(
 ):
     """Mail 列表（按时间倒序）"""
     bb = _bb()
-    tasks = bb.list_tasks(status=status, assignee=to_agent, assigned_by=from_agent)
+    tasks = bb.list_tasks(
+        status=status,
+        assignee=to_agent,
+        assigned_by=from_agent)
 
     mails = []
     for t in tasks:
@@ -222,13 +227,16 @@ async def send_mail(body: Dict[str, Any]):
 
         # A8: 只有原邮件的双方能回复（严格 1 对 1）
         if from_agent not in (orig_from, orig_to):
-            raise HTTPException(400, f"只有邮件的发送者或接收者可以回复")
+            raise HTTPException(400, "只有邮件的发送者或接收者可以回复")
 
         # A6/A7: 自动纠正 to → 原邮件发件者
         to_agent = body.get("to", "").strip()
         corrected_to = orig_from  # 回复方向固定: reply → original sender
         if to_agent and to_agent != corrected_to:
-            auto_corrected = {"field": "to", "original": to_agent, "corrected": corrected_to}
+            auto_corrected = {
+                "field": "to",
+                "original": to_agent,
+                "corrected": corrected_to}
         to_agent = corrected_to
     else:
         # --- A2: to 必填（非回复场景） ---
@@ -255,7 +263,8 @@ async def send_mail(body: Dict[str, Any]):
     conversation_id = body.get("conversation_id")
     if not conversation_id and original:
         try:
-            orig_meta = json.loads(original.must_haves) if original.must_haves else {}
+            orig_meta = json.loads(
+                original.must_haves) if original.must_haves else {}
             conversation_id = orig_meta.get("conversation_id")
         except Exception:
             pass
@@ -310,10 +319,12 @@ async def delete_mail(prefix: Optional[str] = Query(None)):
     for t in tasks:
         if t.title and t.title.startswith(prefix):
             if t.status not in ("cancelled",):
-                bb.update_task_status(t.id, "cancelled", agent="mail-cleanup-api")
+                bb.update_task_status(
+                    t.id, "cancelled", agent="mail-cleanup-api")
                 deleted_ids.append(t.id)
 
-    return {"ok": True, "deleted_count": len(deleted_ids), "deleted_ids": deleted_ids}
+    return {"ok": True, "deleted_count": len(
+        deleted_ids), "deleted_ids": deleted_ids}
 
 
 @router.patch("/{mail_id}")

src/api/project_routes.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict
 
 from fastapi import APIRouter, HTTPException, Query
 
@@ -31,8 +31,10 @@ async def list_projects():
         if db_path.exists():
             try:
                 conn = sqlite3.connect(str(db_path), timeout=5)
-                total = conn.execute("SELECT COUNT(*) FROM tasks WHERE status != 'cancelled'").fetchone()[0]
-                active = conn.execute("SELECT COUNT(*) FROM tasks WHERE COALESCE(archived,0)=0").fetchone()[0]
+                total = conn.execute(
+                    "SELECT COUNT(*) FROM tasks WHERE status != 'cancelled'").fetchone()[0]
+                active = conn.execute(
+                    "SELECT COUNT(*) FROM tasks WHERE COALESCE(archived,0)=0").fetchone()[0]
                 archived = total - active
                 conn.close()
                 info['task_count'] = active
@@ -45,8 +47,10 @@ async def list_projects():
     if general_db.exists() and "_general" not in projects:
         try:
             conn = sqlite3.connect(str(general_db), timeout=5)
-            total = conn.execute("SELECT COUNT(*) FROM tasks WHERE status != 'cancelled'").fetchone()[0]
-            active = conn.execute("SELECT COUNT(*) FROM tasks WHERE COALESCE(archived,0)=0").fetchone()[0]
+            total = conn.execute(
+                "SELECT COUNT(*) FROM tasks WHERE status != 'cancelled'").fetchone()[0]
+            active = conn.execute(
+                "SELECT COUNT(*) FROM tasks WHERE COALESCE(archived,0)=0").fetchone()[0]
             conn.close()
             projects["_general"] = {
                 "id": "_general", "name": "一般任务", "description": "无项目归属的通用任务",
@@ -60,8 +64,10 @@ async def list_projects():
         if general_db_check.exists():
             try:
                 conn = sqlite3.connect(str(general_db_check), timeout=5)
-                total = conn.execute("SELECT COUNT(*) FROM tasks WHERE status != 'cancelled'").fetchone()[0]
-                active = conn.execute("SELECT COUNT(*) FROM tasks WHERE COALESCE(archived,0)=0").fetchone()[0]
+                total = conn.execute(
+                    "SELECT COUNT(*) FROM tasks WHERE status != 'cancelled'").fetchone()[0]
+                active = conn.execute(
+                    "SELECT COUNT(*) FROM tasks WHERE COALESCE(archived,0)=0").fetchone()[0]
                 conn.close()
                 projects["_general"]["task_count"] = active
                 projects["_general"]["task_count_total"] = total
@@ -76,7 +82,7 @@ async def list_projects():
 async def create_project(body: Dict[str, Any]):
     reg = _registry()
     try:
-        info = reg.create_project(
+        reg.create_project(
             body["id"], body["name"],
             agents=body.get("agents", []),
             description=body.get("description", ""),
@@ -173,7 +179,10 @@ async def move_task(project_id: str, task_id: str, body: Dict[str, Any]):
             depends_on=child.depends_on, must_haves=child.must_haves,
         )
         tgt_bb.create_task(moved_child)
-        src_bb.update_task_status(child.id, "cancelled", detail=f"Moved to {target_project}")
+        src_bb.update_task_status(
+            child.id,
+            "cancelled",
+            detail=f"Moved to {target_project}")
         moved_ids.append(child.id)
 
     # 移动主任务
@@ -186,7 +195,10 @@ async def move_task(project_id: str, task_id: str, body: Dict[str, Any]):
         depends_on=task.depends_on, must_haves=task.must_haves,
     )
     tgt_bb.create_task(moved_task)
-    src_bb.update_task_status(task_id, "cancelled", detail=f"Moved to {target_project}")
+    src_bb.update_task_status(
+        task_id,
+        "cancelled",
+        detail=f"Moved to {target_project}")
     moved_ids.insert(0, task_id)
 
     return {"ok": True, "moved_to": target_project, "moved_ids": moved_ids}

src/api/toolchain_routes.py

@@ -46,17 +46,48 @@ _TTL_SECONDS = 7 * 24 * 3600
 _idempotency_lock = asyncio.Lock()
 
 
-def _is_duplicate(event: str, delivery: str) -> bool:
-    """检查 Webhook 是否重复投递，自动清理过期条目。"""
+def _is_duplicate(event: str, delivery: str,
+                  payload: Optional[Dict[str, Any]] = None) -> bool:
+    """检查 Webhook 是否重复投递，自动清理过期条目。
+
+    双重去重策略：
+    1. delivery UUID 去重（标准幂等）
+    2. payload 内容去重（应对 Gitea v1.23.4 的 webhookNotifier + actionsNotifier
+       对同一 review 生成不同 UUID 的双投递问题）
+    """
     now = time.time()
     # 清理过期条目
-    while _delivery_timestamps and (now - _delivery_timestamps[0][0]) > _TTL_SECONDS:
+    while _delivery_timestamps and (
+            now - _delivery_timestamps[0][0]) > _TTL_SECONDS:
         _, key = _delivery_timestamps.pop(0)
         _delivery_cache.discard(key)
 
+    # 检查 delivery UUID 去重
     key = f"{event}-{delivery}"
     if key in _delivery_cache:
         return True
+
+    # 检查 payload 内容去重（review 事件：同一 PR + 同一用户 + 同一内容）
+    # 注意：Gitea webhookNotifier 用 review.body，actionsNotifier 用 review.content
+    # 所以去重 key 需要同时取两个字段，确保两种格式生成相同 key
+    if payload and "review" in event:
+        pr_num = payload.get("pull_request", {}).get("number")
+        sender = payload.get("sender", {}).get("login")
+        review = payload.get("review", {})
+        # 取 body 或 content，优先 body（webhookNotifier 格式）
+        content = review.get("body", "") or review.get("content", "")
+        content_hash = hashlib.sha256(content.encode()).hexdigest()[:16]
+        content_key = f"content:{event}:{pr_num}:{sender}:{content_hash}"
+        if content_key in _delivery_cache:
+            logger.info(
+                "Content-based duplicate detected: %s PR#%s by %s",
+                event,
+                pr_num,
+                sender)
+            return True
+        _delivery_cache.add(content_key)
+        _delivery_timestamps.append((now, content_key))
+
     _delivery_cache.add(key)
     _delivery_timestamps.append((now, key))
     return False
@@ -112,8 +143,16 @@ async def _fetch_pr_files(repo: str, pr_number: int) -> Tuple[List[str], str]:
             last_error = str(e)
             if attempt < 2:
                 await asyncio.sleep(0.5 * (attempt + 1))
-                logger.warning("Retry %d/3 fetching PR files: %s/pulls/%d", attempt + 1, repo, pr_number)
-    logger.warning("Failed to fetch PR files after 3 retries: %s/pulls/%d - %s", repo, pr_number, last_error)
+                logger.warning(
+                    "Retry %d/3 fetching PR files: %s/pulls/%d",
+                    attempt + 1,
+                    repo,
+                    pr_number)
+    logger.warning(
+        "Failed to fetch PR files after 3 retries: %s/pulls/%d - %s",
+        repo,
+        pr_number,
+        last_error)
     return [], f"获取文件列表失败（重试3次）: {last_error}"
 
 
@@ -141,7 +180,6 @@ def _calc_risk_level(changed_files: List[str]) -> str:
 # ---------------------------------------------------------------------------
 
 
-
 MAIL_PROJECT_ID = "_mail"
 
 
@@ -227,7 +265,8 @@ async def _handle_pull_request(payload: Dict[str, Any]) -> None:
 
     pr = payload.get("pull_request")
     if not pr or not isinstance(pr, dict):
-        logger.warning("pull_request event missing pull_request field, skipping")
+        logger.warning(
+            "pull_request event missing pull_request field, skipping")
         return
     repo = _repo_fullname(payload)
     pr_number = pr.get("number", 0)
@@ -241,7 +280,8 @@ async def _handle_pull_request(payload: Dict[str, Any]) -> None:
     if fetch_error:
         file_list = f"⚠️ {fetch_error}"
     else:
-        file_list = "\n".join(f"- {f}" for f in changed_files) if changed_files else "(无文件变更)"
+        file_list = "\n".join(
+            f"- {f}" for f in changed_files) if changed_files else "(无文件变更)"
 
     text = render_template("review_request", {
         "repo": repo,
@@ -258,16 +298,34 @@ async def _handle_pull_request(payload: Dict[str, Any]) -> None:
 
 
 async def _handle_pull_request_review(payload: Dict[str, Any]) -> None:
-    """处理 pull_request_review 事件：非 COMMENTED → 通知 PR 作者。"""
+    """处理 pull_request_review 事件：非 COMMENTED → 通知 PR 作者。
+
+    支持两种 payload 格式：
+    - repo webhook: review.state = "APPROVED" / "REQUEST_CHANGES"
+    - org webhook (Gitea v1.23.4): review.type = "pull_request_review_approved" / "pull_request_review_rejected"
+    """
     review = payload.get("review")
     if not review or not isinstance(review, dict):
-        logger.warning("pull_request_review event missing review field, skipping")
+        logger.warning(
+            "pull_request_review event missing review field, skipping")
         return
     pr = payload.get("pull_request")
     if not pr or not isinstance(pr, dict):
-        logger.warning("pull_request_review event missing pull_request field, skipping")
+        logger.warning(
+            "pull_request_review event missing pull_request field, skipping")
         return
+
+    # 兼容两种 payload 格式提取 state
     state = review.get("state", "")
+    if not state:
+        # org webhook 格式：review.type = "pull_request_review_approved"
+        review_type = review.get("type", "")
+        type_map = {
+            "pull_request_review_approved": "APPROVED",
+            "pull_request_review_rejected": "REQUEST_CHANGES",
+            "pull_request_review_comment": "COMMENTED",
+        }
+        state = type_map.get(review_type, "")
 
     # 只通知 APPROVED 和 REQUEST_CHANGES，跳过 COMMENTED 和其他状态
     if state == "COMMENTED":
@@ -277,8 +335,17 @@ async def _handle_pull_request_review(payload: Dict[str, Any]) -> None:
     pr_number = pr.get("number", 0)
     pr_title = pr.get("title", "")
     pr_author = pr.get("user", {}).get("login", "unknown")
-    reviewer = review.get("user", {}).get("login", "unknown")
-    review_body = review.get("body", "(无评论)")
+    # 兼容：org webhook 的 review 没有 user，从 sender 取
+    reviewer = review.get(
+        "user",
+        {}).get(
+        "login",
+        "") or payload.get(
+            "sender",
+            {}).get(
+                "login",
+        "unknown")
+    review_body = review.get("body", "") or review.get("content", "(无评论)")
 
     result_map = {"APPROVED": "通过 ✓", "REQUEST_CHANGES": "驳回 ✗"}
     if state not in result_map:
@@ -324,7 +391,8 @@ async def _handle_issues(payload: Dict[str, Any]) -> None:
             logger.debug("Issue assigned but no assignee found, skipping")
             return
 
-        labels_list = [lbl.get("name", "") for lbl in (issue.get("labels") or [])]
+        labels_list = [lbl.get("name", "")
+                       for lbl in (issue.get("labels") or [])]
         labels = ", ".join(labels_list) if labels_list else "(无标签)"
         issue_body = issue.get("body", "(无描述)")
         brief = issue_title[:20].replace(" ", "-").lower()
@@ -372,6 +440,14 @@ async def _handle_issue_comment(payload: Dict[str, Any]) -> None:
     if not issue or not isinstance(issue, dict):
         logger.warning("issue_comment event missing issue field, skipping")
         return
+
+    # 已关闭的 Issue/PR 不再发送 CI 失败通知
+    if issue.get("state") == "closed":
+        logger.debug(
+            "Skipping CI failure notification for closed issue #%s",
+            issue.get("number"))
+        return
+
     repo = _repo_fullname(payload)
     issue_number = issue.get("number", 0)
 
@@ -401,6 +477,12 @@ async def _handle_issue_comment(payload: Dict[str, Any]) -> None:
 _EVENT_HANDLERS: Dict[str, Any] = {
     "pull_request": _handle_pull_request,
     "pull_request_review": _handle_pull_request_review,
+    "pull_request_review_approved": _handle_pull_request_review,
+    "pull_request_review_rejected": _handle_pull_request_review,
+    "pull_request_review_comment": _handle_pull_request_review,
+    # Gitea v1.23.4 实际发出的 review 子事件（无 _review_ 中间段）
+    "pull_request_approved": _handle_pull_request_review,
+    "pull_request_rejected": _handle_pull_request_review,
     "issues": _handle_issues,
     "issue_comment": _handle_issue_comment,
 }
@@ -431,27 +513,32 @@ async def gitea_webhook(
     # 1. 签名验证
     if not _verify_signature(body, x_gitea_signature):
         logger.warning("Webhook signature verification failed")
-        return Response(status_code=403, content="signature verification failed")
+        return Response(status_code=403,
+                        content="signature verification failed")
 
-    # 2. 幂等检查
-    if x_gitea_event and x_gitea_delivery:
-        async with _idempotency_lock:
-            if _is_duplicate(x_gitea_event, x_gitea_delivery):
-                logger.debug("Duplicate webhook: %s/%s", x_gitea_event, x_gitea_delivery)
-                return Response(status_code=200, content="duplicate")
-
-    # 3. 解析 payload
+    # 3. 解析 payload（提前解析，用于幂等检查）
     try:
         payload = await request.json()
     except Exception:
         logger.warning("Failed to parse webhook payload")
         return Response(status_code=200, content="invalid payload")
 
+    # 2. 幂等检查（需要在 payload 解析后，以支持内容去重）
+    if x_gitea_event and x_gitea_delivery:
+        async with _idempotency_lock:
+            if _is_duplicate(x_gitea_event, x_gitea_delivery, payload):
+                logger.debug(
+                    "Duplicate webhook: %s/%s",
+                    x_gitea_event,
+                    x_gitea_delivery)
+                return Response(status_code=200, content="duplicate")
+
     # 4. 查找 handler
     handler = _EVENT_HANDLERS.get(x_gitea_event or "")
     if not handler:
         logger.debug("Unhandled event type: %s", x_gitea_event)
-        return Response(status_code=200, content=f"unhandled event: {x_gitea_event}")
+        return Response(status_code=200,
+                        content=f"unhandled event: {x_gitea_event}")
 
     # 5. 执行 handler
     try:

src/blackboard/db.py

@@ -4,7 +4,6 @@ from __future__ import annotations
 
 import sqlite3
 from pathlib import Path
-from typing import Optional
 
 
 def init_db(db_path: Path) -> None:
@@ -133,8 +132,10 @@ def _migrate_v28(conn: sqlite3.Connection) -> None:
         resolved_by TEXT,
         resolve_note TEXT
     )""")
-    conn.execute("CREATE INDEX IF NOT EXISTS idx_checkpoints_task ON checkpoints(task_id)")
-    conn.execute("CREATE INDEX IF NOT EXISTS idx_checkpoints_status ON checkpoints(status)")
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS idx_checkpoints_task ON checkpoints(task_id)")
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS idx_checkpoints_status ON checkpoints(status)")
 
     # 4. outputs 扩展字段（M3 成果物）
     _safe_add_column(conn, "outputs", "file_name", "TEXT")
@@ -189,18 +190,20 @@ TERMINAL_STATUSES = frozenset()  # v3.1: 无终态，全靠 VALID_TRANSITIONS
 MANUAL_STATUSES = frozenset({"cancelled", "paused", "reviewing"})
 
 VALID_TRANSITIONS = {
-    "pending":       {"claimed", "paused", "blocked", "cancelled"},
-    "claimed":       {"working", "paused", "pending", "cancelled"},
-    "working":       {"review", "done", "blocked", "failed", "paused", "escalated", "waiting_human", "cancelled", "pending"},  # pending: Mail spawn 失败回退
-    "paused":        {"working", "claimed", "review", "escalated", "waiting_human", "cancelled"},  # 恢复到 resumed_from 记录的状态
-    "review":        {"done", "pending", "failed", "paused", "escalated", "waiting_human", "cancelled"},
-    "blocked":       {"pending", "escalated", "cancelled"},
-    "failed":        {"pending", "escalated", "cancelled"},
-    "escalated":     {"working", "pending", "paused", "cancelled"},
+    "pending": {"claimed", "paused", "blocked", "cancelled"},
+    "claimed": {"working", "paused", "pending", "cancelled"},
+    # pending: Mail spawn 失败回退
+    "working": {"review", "done", "blocked", "failed", "paused", "escalated", "waiting_human", "cancelled", "pending"},
+    # 恢复到 resumed_from 记录的状态
+    "paused": {"working", "claimed", "review", "escalated", "waiting_human", "cancelled"},
+    "review": {"done", "pending", "failed", "paused", "escalated", "waiting_human", "cancelled"},
+    "blocked": {"pending", "escalated", "cancelled"},
+    "failed": {"pending", "escalated", "cancelled"},
+    "escalated": {"working", "pending", "paused", "cancelled"},
     "waiting_human": {"working", "done", "paused", "cancelled"},
-    "done":          {"cancelled", "reviewing"},
-    "reviewing":     {"done", "working", "cancelled"},
-    "cancelled":     {"pending"},
+    "done": {"cancelled", "reviewing"},
+    "reviewing": {"done", "working", "cancelled"},
+    "cancelled": {"pending"},
 }
 
 COMMENT_TYPES = frozenset({
@@ -224,7 +227,8 @@ EVENT_TYPES = frozenset({
 
 OUTPUT_TYPES = frozenset({"code", "document", "data", "config", "other"})
 
-REVIEW_TYPES = frozenset({"plan_review", "output_review", "guardrail", "final_review"})
+REVIEW_TYPES = frozenset(
+    {"plan_review", "output_review", "guardrail", "final_review"})
 VERDICT_TYPES = frozenset({"approved", "rejected", "needs_revision"})
 
 EXPERIENCE_SOURCES = frozenset({

src/blackboard/models.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Any, Dict, List, Optional
+from typing import Any, List, Optional
 
 
 @dataclass

src/blackboard/operations.py

@@ -11,7 +11,6 @@ from typing import Any, Dict, List, Optional
 
 from .db import (
     VALID_TRANSITIONS,
-    VALID_STATUSES,
     COMMENT_TYPES,
     EVENT_TYPES,
     OUTPUT_TYPES,
@@ -84,7 +83,8 @@ class Blackboard:
         """获取单个任务"""
         conn = self._conn()
         try:
-            row = conn.execute("SELECT * FROM tasks WHERE id=?", (task_id,)).fetchone()
+            row = conn.execute(
+                "SELECT * FROM tasks WHERE id=?", (task_id,)).fetchone()
             return Task.from_row(row) if row else None
         finally:
             conn.close()
@@ -129,7 +129,8 @@ class Blackboard:
                 updates["completed_at"] = now  # paused 也记录时间用于恢复
                 updates["resumed_from"] = old_status  # 记录暂停前状态
             elif new_status == "pending":
-                # 所有 →pending 转换都清空 assignee（与 ticker._transition_status L414 对齐）
+                # 所有 →pending 转换都清空 assignee（与 ticker._transition_status L414
+                # 对齐）
                 updates["assignee"] = None
                 updates["claimed_at"] = None
                 updates["current_agent"] = None
@@ -693,7 +694,6 @@ class Blackboard:
         finally:
             conn.close()
 
-
     # ── Checkpoint CRUD（M3） ──
 
     def create_checkpoint(
@@ -709,7 +709,8 @@ class Blackboard:
         import uuid
         # BUG-33: 校验 payload 结构必须含 version 字段
         if not isinstance(payload, dict) or "version" not in payload:
-            raise ValueError("payload must be a dict containing 'version' field")
+            raise ValueError(
+                "payload must be a dict containing 'version' field")
         cp_id = checkpoint_id or f"cp-{uuid.uuid4().hex[:8]}"
         conn = self._conn()
         try:
@@ -966,7 +967,8 @@ class Blackboard:
         finally:
             conn.close()
 
-    def get_pending_mentions(self, max_retries: int = 5) -> List[Dict[str, Any]]:
+    def get_pending_mentions(
+            self, max_retries: int = 5) -> List[Dict[str, Any]]:
         """获取所有 pending 且未超过重试上限的 mentions"""
         conn = self._conn()
         try:
@@ -1001,7 +1003,8 @@ class Blackboard:
         conn = self._conn()
         try:
             conn.execute("BEGIN IMMEDIATE")
-            conn.execute("UPDATE mention_queue SET retry_count=retry_count+1 WHERE id=?", (mention_id,))
+            conn.execute(
+                "UPDATE mention_queue SET retry_count=retry_count+1 WHERE id=?", (mention_id,))
             conn.commit()
             return True
         finally:
@@ -1012,7 +1015,8 @@ class Blackboard:
         conn = self._conn()
         try:
             conn.execute("BEGIN IMMEDIATE")
-            conn.execute("UPDATE mention_queue SET status='failed' WHERE id=?", (mention_id,))
+            conn.execute(
+                "UPDATE mention_queue SET status='failed' WHERE id=?", (mention_id,))
             conn.commit()
             return True
         finally:

src/blackboard/queries.py

@@ -132,7 +132,8 @@ class Queries:
         """任务详情聚合（含关联数据）"""
         conn = self._conn()
         try:
-            row = conn.execute("SELECT * FROM tasks WHERE id=?", (task_id,)).fetchone()
+            row = conn.execute(
+                "SELECT * FROM tasks WHERE id=?", (task_id,)).fetchone()
             if not row:
                 return None
             task = dict(row)
@@ -159,7 +160,8 @@ class Queries:
         finally:
             conn.close()
 
-    def task_events(self, task_id: str, limit: int = 50) -> List[Dict[str, Any]]:
+    def task_events(self, task_id: str,
+                    limit: int = 50) -> List[Dict[str, Any]]:
         """任务事件列表"""
         conn = self._conn()
         try:
@@ -265,7 +267,8 @@ class Queries:
                 return "review"
 
             # 有 working/claimed → working
-            if status_counts.get("working", 0) > 0 or status_counts.get("claimed", 0) > 0:
+            if status_counts.get("working", 0) > 0 or status_counts.get(
+                    "claimed", 0) > 0:
                 return "working"
 
             # 有 pending → pending
@@ -337,7 +340,8 @@ class Queries:
             # 当前活跃 stage
             active_stage = None
             for sp in stage_progress:
-                if sp["active"] > 0 or (sp["total"] > 0 and sp["done"] < sp["total"]):
+                if sp["active"] > 0 or (
+                        sp["total"] > 0 and sp["done"] < sp["total"]):
                     if not active_stage and sp["done"] < sp["total"]:
                         active_stage = sp["label"]
 

src/blackboard/registry.py

@@ -119,7 +119,8 @@ class ProjectRegistry:
         finally:
             conn.close()
 
-    def list_projects(self, status: Optional[str] = None) -> Dict[str, Dict[str, Any]]:
+    def list_projects(
+            self, status: Optional[str] = None) -> Dict[str, Dict[str, Any]]:
         """列出项目"""
         conn = self._connect()
         try:
@@ -178,7 +179,8 @@ class ProjectRegistry:
             status="deleted",
         )
 
-    def physical_delete_project(self, project_id: str) -> Optional[Dict[str, Any]]:
+    def physical_delete_project(
+            self, project_id: str) -> Optional[Dict[str, Any]]:
         """物理删除项目（删目录 + 删 registry 条目）"""
         import shutil
 
@@ -260,7 +262,8 @@ class ProjectRegistry:
     # 迁移（从 _registry.yaml）
     # ===================================================================
 
-    def discover_sanguo_projects(self, scan_dir: Optional[Path] = None) -> List[str]:
+    def discover_sanguo_projects(
+            self, scan_dir: Optional[Path] = None) -> List[str]:
         """扫描 sanguo_projects 开发目录，自动注册正式项目"""
         scan_dir = scan_dir or Path(os.environ.get(
             "SANGUO_PROJECTS_DIR",
@@ -355,4 +358,3 @@ class ProjectRegistry:
 
     def reload(self) -> None:
         """兼容旧接口（SQLite 不需要 reload cache）"""
-        pass

src/cli/blackboard.py

@@ -10,7 +10,7 @@ from typing import List, Optional
 
 from src.blackboard.operations import Blackboard
 from src.utils import get_data_root
-from src.blackboard.models import Task, Comment, Output, Decision, Observation, Review, Experience
+from src.blackboard.models import Task, Review
 from src.blackboard.queries import Queries
 from src.blackboard.registry import ProjectRegistry
 
@@ -35,7 +35,9 @@ def _get_queries(project_id: str) -> Queries:
 
 
 def build_blackboard_parser() -> argparse.ArgumentParser:
-    parser = argparse.ArgumentParser(prog="blackboard", description="Agent blackboard operations")
+    parser = argparse.ArgumentParser(
+        prog="blackboard",
+        description="Agent blackboard operations")
     sub = parser.add_subparsers(dest="command")
 
     # read
@@ -206,7 +208,11 @@ def _cmd_comment(opts) -> int:
 
 def _cmd_decide(opts) -> int:
     bb = _get_bb(opts.project)
-    did = bb.add_decision(opts.task_id, opts.decider, opts.decision, opts.rationale)
+    did = bb.add_decision(
+        opts.task_id,
+        opts.decider,
+        opts.decision,
+        opts.rationale)
     print(f"Decision recorded: {did}")
     return 0
 
@@ -251,7 +257,8 @@ def _print_tasks(tasks, as_json: bool):
 
 
 def build_admin_parser() -> argparse.ArgumentParser:
-    parser = argparse.ArgumentParser(prog="admin", description="Admin operations")
+    parser = argparse.ArgumentParser(
+        prog="admin", description="Admin operations")
     sub = parser.add_subparsers(dest="command")
 
     # project create
@@ -262,7 +269,7 @@ def build_admin_parser() -> argparse.ArgumentParser:
     p_pc.add_argument("--description", default="")
 
     # project list
-    p_pl = sub.add_parser("project-list", help="List projects")
+    sub.add_parser("project-list", help="List projects")
 
     # project archive
     p_pa = sub.add_parser("project-archive", help="Archive project")
@@ -300,7 +307,8 @@ def run_admin_cli(args: Optional[List[str]] = None) -> int:
             for pid, info in projects.items():
                 status = info.get("status", "?")
                 agents = ",".join(info.get("agents", []))
-                print(f"  {pid}  [{status}]  {info.get('name', '')}  agents: {agents}")
+                print(
+                    f"  {pid}  [{status}]  {info.get('name', '')}  agents: {agents}")
             return 0
 
         elif opts.command == "project-archive":

src/daemon/base_task_handler.py

@@ -0,0 +1,183 @@
+"""base_task_handler.py — Task type handler 基类。
+
+收敛合理的共性能力（crash rollback + verify + mark + notify），
+子类只实现差异点。
+"""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+from src.daemon.prompt_composer import PromptContext, PromptComposer, PromptSection
+from src.blackboard.db import get_connection
+
+logger = logging.getLogger("moziplus-v2.handler")
+
+
+@dataclass
+class VerifyResult:
+    """验证结果"""
+    passed: bool
+    reason: str          # "has_output" / "no_reply" / "no_signal" / ...
+    evidence: str        # "output_count=1, comment_count=0"
+    can_retry: bool = True
+    retry_count: int = 0
+
+
+class BaseTaskHandler:
+    """所有 task type handler 的基类。
+    
+    职责：L2 引擎注入层的业务逻辑——prompt 构建、完成验证、状态标记。
+    不管：进程生命周期、exit 分类、重试决策（这些归 spawner）。
+    """
+    
+    # crash 类 outcome（进程级异常，需要 rollback）
+    CRASH_OUTCOMES = frozenset({
+        "crashed", "compact_failed", "process_crash",
+        "session_stuck", "compact_hanging",
+    })
+    
+    task_type: str = ""
+    virtual_project: Optional[str] = None
+    display_name: str = ""  # 中文展示名（ticker 扫描日志用）
+    
+    # === 子类必须实现 ===
+    
+    def build_prompt(self, context: PromptContext) -> str:
+        """构建 L2 prompt（通过 PromptComposer 拼 section）。子类实现。"""
+        raise NotImplementedError
+    
+    def verify_completion(self, task_id: str, db_path: Path) -> VerifyResult:
+        """验证任务完成质量。每个 handler 自己的验证逻辑。子类实现。"""
+        raise NotImplementedError
+    
+    def target_success_status(self) -> str:
+        """验证通过后的目标状态。task='review', mail/toolchain='done'"""
+        return "review"
+    
+    def get_sections(self) -> list[PromptSection]:
+        """返回此 handler 的 prompt section 列表。子类实现。"""
+        return []
+    
+    # === 基类提供统一流程 ===
+    
+    def pre_spawn(self, task_id: str, db_path: Path) -> bool:
+        """spawn 前业务准备。默认 True。
+        mail/toolchain override 为 auto_working。"""
+        return True
+    
+    def post_complete(self, task_id: str, agent_id: str,
+                      outcome: str, db_path: Path) -> None:
+        """spawn 完成后的业务处理。统一 4 步流程：
+        1. crash 处理 → rollback current_agent
+        2. verify → 验证产出
+        3. mark → 标目标状态
+        4. notify → 失败时 on_failure
+        """
+        # 1. crash 处理（基类提供，所有 handler 继承）
+        if outcome in self.CRASH_OUTCOMES:
+            self._rollback_current_agent(db_path, task_id, agent_id)
+            return
+        
+        # 2. verify
+        result = self.verify_completion(task_id, db_path)
+        
+        # 3. mark
+        if result.passed:
+            self._mark_task_status(db_path, task_id, self.target_success_status())
+            logger.info("Task %s: verify passed (%s), marked %s",
+                        task_id, result.reason, self.target_success_status())
+        else:
+            # 4. notify
+            self.on_failure(task_id, agent_id, db_path, result)
+    
+    def on_failure(self, task_id: str, agent_id: str,
+                   db_path: Path, verify: VerifyResult) -> None:
+        """验证失败处理。默认：标 failed。子类可 override。"""
+        self._mark_task_status(db_path, task_id, "failed")
+        logger.info("Task %s: verify failed (%s), marked failed",
+                     task_id, verify.reason)
+    
+    def check_completion(self, task_id: str, db_path: Path) -> bool:
+        """ticker 级别的完成检查。默认：False。"""
+        return False
+    
+    # === 内部工具方法 ===
+    
+    def _rollback_current_agent(self, db_path: Path, task_id: str, agent_id: str) -> None:
+        """crash 后回退 current_agent → assignee，避免 exclude_current 卡死。
+        从 dispatcher._rollback_current_agent 迁移。"""
+        try:
+            conn = get_connection(db_path)
+            try:
+                conn.execute(
+                    "UPDATE tasks SET current_agent = "
+                    "(SELECT assignee FROM tasks WHERE id=?) "
+                    "WHERE id=? AND current_agent=?",
+                    (task_id, task_id, agent_id)
+                )
+                conn.commit()
+            finally:
+                conn.close()
+            logger.info("Task %s: rolled back current_agent from %s to assignee",
+                        task_id, agent_id)
+        except Exception as e:
+            logger.warning("Task %s: failed to rollback current_agent: %s",
+                           task_id, e)
+    
+    def _mark_task_status(self, db_path: Path, task_id: str, status: str) -> None:
+        """更新任务状态 + 写审计事件（带 3 次重试，防 SQLite DB 锁）。"""
+        for attempt in range(3):
+            try:
+                conn = get_connection(db_path)
+                try:
+                    conn.execute("BEGIN IMMEDIATE")
+                    old_row = conn.execute(
+                        "SELECT status FROM tasks WHERE id=?", (task_id,)
+                    ).fetchone()
+                    old_status = old_row["status"] if old_row else "unknown"
+                    conn.execute(
+                        "UPDATE tasks SET status=?, updated_at=datetime('now') WHERE id=?",
+                        (status, task_id),
+                    )
+                    conn.execute(
+                        "INSERT INTO events (task_id, agent, event_type, payload) "
+                        "VALUES (?, 'handler', 'status_change', ?)",
+                        (task_id,
+                         f'{{"from": "{old_status}", "to": "{status}", '
+                         f'"source": "{self.task_type}_handler"}}'),
+                    )
+                    conn.commit()
+                    return
+                finally:
+                    conn.close()
+            except Exception as e:
+                logger.warning("Handler: mark %s → %s attempt %d failed: %s",
+                               task_id, status, attempt + 1, e)
+        logger.error("Handler: mark %s → %s all 3 attempts failed", task_id, status)
+    
+    def _auto_mark_working(self, task_id: str, db_path: Path) -> bool:
+        """pending → working（mail/toolchain 通用）。"""
+        try:
+            conn = get_connection(db_path)
+            try:
+                conn.execute("BEGIN IMMEDIATE")
+                row = conn.execute(
+                    "SELECT status FROM tasks WHERE id=?", (task_id,)).fetchone()
+                if not row or row["status"] not in ("pending", "claimed"):
+                    logger.warning("Task %s: cannot mark working (status=%s)",
+                                   task_id, row["status"] if row else "not found")
+                    return False
+                conn.execute(
+                    "UPDATE tasks SET status='working', updated_at=datetime('now') "
+                    "WHERE id=?", (task_id,))
+                conn.commit()
+                logger.info("Task %s: auto-marked working", task_id)
+                return True
+            finally:
+                conn.close()
+        except Exception as e:
+            logger.error("Task %s: failed to mark working: %s", task_id, e)
+            return False

src/daemon/bootstrap.py

@@ -11,8 +11,7 @@ A 类 Skill 由引擎确定性注入全文，不靠 Description 触发。
 
 import logging
 import os
-from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, List
 
 logger = logging.getLogger("moziplus-v2.bootstrap")
 
@@ -28,12 +27,12 @@ class BootstrapBuilder:
     """L2 引擎注入层构建器（v2.1 四段式）"""
 
     ROLE_SKILL_MAP = {
-        "executor":          "blackboard-executor",
-        "reviewer":          "blackboard-reviewer",
-        "reviewer-simayi":   "blackboard-reviewer-simayi",
+        "executor": "blackboard-executor",
+        "reviewer": "blackboard-reviewer",
+        "reviewer-simayi": "blackboard-reviewer-simayi",
         "reviewer-pangtong": "blackboard-reviewer-pangtong",
-        "planner":           "blackboard-planner",
-        "claim":             "blackboard-claim",
+        "planner": "blackboard-planner",
+        "claim": "blackboard-claim",
     }
 
     # 默认从环境变量或配置读取，fallback 到默认路径
@@ -62,7 +61,9 @@ class BootstrapBuilder:
 
         # 段 2: 前序产出（有依赖时注入）
         if task.get("depends_on_outputs"):
-            sections.append(self._format_prior_outputs(task["depends_on_outputs"]))
+            sections.append(
+                self._format_prior_outputs(
+                    task["depends_on_outputs"]))
 
         # 段 3: 角色操作规范全文（通过 ROLE_SKILL_MAP 从 Skill 文件读取）
         skill_name = self.ROLE_SKILL_MAP.get(role)
@@ -134,7 +135,8 @@ class BootstrapBuilder:
         """格式化前序产出摘要（段 2）"""
         parts = ["## 前序产出"]
         for out in outputs:
-            parts.append(f"- [{out.get('task_id', '?')}] {out.get('summary', '无摘要')}")
+            parts.append(
+                f"- [{out.get('task_id', '?')}] {out.get('summary', '无摘要')}")
         return "\n".join(parts)
 
     def _format_constraints(self, role: str) -> str:

src/daemon/counter.py

@@ -68,20 +68,23 @@ class ActiveAgentCounter:
         self._cooldown_until.pop(agent_id, None)
         return False
 
-    def set_cooldown(self, agent_id: str, seconds: Optional[float] = None) -> None:
+    def set_cooldown(self, agent_id: str,
+                     seconds: Optional[float] = None) -> None:
         """设置冷却期（默认 120 秒）"""
         cd = seconds if seconds is not None else self._default_cooldown_seconds
         self._cooldown_until[agent_id] = time.time() + cd
         logger.info("Cooldown set for %s: %.0fs (until %.0f)",
-                     agent_id, cd, self._cooldown_until[agent_id])
+                    agent_id, cd, self._cooldown_until[agent_id])
 
-    async def can_acquire(self, agent_id: str, session_id: str = "main") -> bool:
+    async def can_acquire(self, agent_id: str,
+                          session_id: str = "main") -> bool:
         """三层检查：cooldown → global → per agent → per session key"""
         if self.is_cooling_down(agent_id):
             return False
         if self._global_active >= self._max_global:
             return False
-        if self._agent_active.get(agent_id, 0) >= self._max_concurrent_sessions:
+        if self._agent_active.get(
+                agent_id, 0) >= self._max_concurrent_sessions:
             return False
         key = self._make_key(agent_id, session_id)
         if self._active_keys.get(key, 0) >= self._max_per_session:
@@ -122,7 +125,8 @@ class ActiveAgentCounter:
                 del self._active_keys[key]
 
         if agent_id in self._agent_active:
-            self._agent_active[agent_id] = max(0, self._agent_active[agent_id] - 1)
+            self._agent_active[agent_id] = max(
+                0, self._agent_active[agent_id] - 1)
             if self._agent_active[agent_id] == 0:
                 del self._agent_active[agent_id]
 

src/daemon/dispatcher.py

@@ -14,7 +14,6 @@ from __future__ import annotations
 import json
 import logging
 import sqlite3
-from datetime import datetime
 from enum import Enum
 from pathlib import Path
 from typing import Any, Dict, List, Optional
@@ -22,7 +21,8 @@ from typing import Any, Dict, List, Optional
 from src.blackboard.models import Task
 from src.blackboard.db import get_connection
 from src.daemon.spawner import AgentBusyError
-from src.daemon.router import AgentRouter, RouteDecision
+from src.daemon.router import AgentRouter
+from src.daemon.task_type_registry import TaskTypeRegistry
 
 logger = logging.getLogger("moziplus-v2.dispatcher")
 
@@ -64,7 +64,8 @@ class Dispatcher:
         if self._legacy_mode:
             self.registered_agents = set(registered_agents or [])
             self.capability_map = capability_map or {}
-            logger.warning("Dispatcher running in legacy mode (no AgentRouter)")
+            logger.warning(
+                "Dispatcher running in legacy mode (no AgentRouter)")
 
     def decide(self, task: Task, action_type: str = "") -> Dict[str, Any]:
         """调度决策（委托给 Router）
@@ -123,17 +124,23 @@ class Dispatcher:
              "status": "dispatched"|"skipped"|"error"|"blocked", "reason": str}
         """
         # 安全红线检查（调度前拦截）
-        # Mail 是 Agent 间通信，不做 guardrail 检查
-        is_mail = project_config.get("project_id") == "_mail" if project_config else False
-        if self.guardrails and not is_mail:
+        # handler 项目（_mail/_toolchain）不做 guardrail 检查
+        handler = TaskTypeRegistry.get_by_project(
+            project_config.get("project_id", "") if project_config else "")
+        is_handler_task = handler is not None
+        if self.guardrails and not is_handler_task:
             violations = self.guardrails.check_task(task)
-            critical = [v for v in violations if v.action in ("block_and_notify", "terminate_and_escalate")]
+            critical = [
+                v for v in violations if v.action in (
+                    "block_and_notify",
+                    "terminate_and_escalate")]
             if critical:
                 v = critical[0]
                 logger.warning("Task '%s' blocked by guardrail: %s - %s",
                                task.title, v.rule_id, v.message)
                 # 写入黑板事件
-                _routing_db = Path(project_config["db_path"]) if project_config and "db_path" in project_config else self.db_path
+                _routing_db = Path(
+                    project_config["db_path"]) if project_config and "db_path" in project_config else self.db_path
                 if _routing_db:
                     self._record_routing(task, {"level": DispatchLevel.BLOCKED, "agent_id": "none",
                                                 "reason": v.message}, "blocked", v.message, _routing_db)
@@ -152,7 +159,8 @@ class Dispatcher:
         decision = self.decide(task, action_type)
         level = decision["level"]
         # 从 project_config 获取项目级 DB 路径（路由审计日志写入项目 DB）
-        _routing_db = Path(project_config["db_path"]) if project_config and "db_path" in project_config else None
+        _routing_db = Path(
+            project_config["db_path"]) if project_config and "db_path" in project_config else None
         agent_id = decision["agent_id"]
 
         # v2.7.2: counter 检查移到 spawn_full_agent 内部
@@ -160,7 +168,8 @@ class Dispatcher:
 
         # 本地执行
         if level == DispatchLevel.LOCAL:
-            self._record_routing(task, decision, "dispatched", None, _routing_db)
+            self._record_routing(
+                task, decision, "dispatched", None, _routing_db)
             return {
                 "level": level.value,
                 "agent_id": "daemon",
@@ -172,7 +181,8 @@ class Dispatcher:
         # Full Agent / Escalate spawn
         if level in (DispatchLevel.FULL_AGENT, DispatchLevel.ESCALATE):
             if not self.spawner:
-                self._record_routing(task, decision, "error", "No spawner", _routing_db)
+                self._record_routing(
+                    task, decision, "error", "No spawner", _routing_db)
                 return {
                     "level": level.value,
                     "agent_id": agent_id,
@@ -182,106 +192,73 @@ class Dispatcher:
                 }
 
             try:
-                # [v2.7.1] Mail: 标 working 移到 spawn_full_agent 内部（check 通过后、subprocess 前）
-                is_mail = project_config.get("project_id") == "_mail" if project_config else False
-                if is_mail:
-                    db_path = Path(project_config["db_path"]) if project_config and "db_path" in project_config else None
+                # [Step 5] Handler: pre_spawn + on_checks_passed 统一
+                project_id = project_config.get("project_id", "") if project_config else ""
+                handler = TaskTypeRegistry.get_by_project(project_id)
+                db_path = Path(
+                    project_config["db_path"]) if project_config and "db_path" in project_config else None
 
-                # on_checks_passed: 所有检查通过后才标 working，检查失败不标
+                # on_checks_passed: handler 项目在 check 通过后调用 handler.pre_spawn
                 on_checks_passed = None
-                _mail_marked_working = False
-                if is_mail and db_path:
+                handler_marked_working = False
+                if handler and db_path:
                     _task_id = task.id
-                    _mail_db = db_path
-                    _disp = self
-                    def _mail_on_checks_passed():
-                        nonlocal _mail_marked_working
-                        if not _disp._mail_auto_working(_task_id, _mail_db):
-                            raise RuntimeError("mail_auto_working_failed")
-                        _mail_marked_working = True
-                    on_checks_passed = _mail_on_checks_passed
+                    _handler_db = db_path
+                    _handler = handler
+
+                    def _handler_on_checks_passed():
+                        nonlocal handler_marked_working
+                        if not _handler.pre_spawn(_task_id, _handler_db):
+                            raise RuntimeError("handler_pre_spawn_failed")
+                        handler_marked_working = True
+                    on_checks_passed = _handler_on_checks_passed
 
                 # 构建 spawn message
                 message = self._build_spawn_message(task, agent_id, project_config,
-                                                           mode=decision.get("mode", ""),
-                                                           spawn_type=action_type or "executor")
+                                                    mode=decision.get(
+                                                        "mode", ""),
+                                                    spawn_type=action_type or "executor")
 
-                # v2.7.2: on_complete 只含业务逻辑，不含 counter.release
-                # counter.release 由 spawn_full_agent 内部的 wrapped_on_complete 保证
+                # [Step 5] Handler: on_complete 统一走 handler.post_complete
+                # 保留旧路径作为 fallback（无 handler 的项目）
                 on_complete = None
-                if is_mail:
+                if handler:
                     _task_id = task.id
-                    _mail_db = db_path
-                    _must_haves = task.must_haves or ""
-                    _dispatcher = self
+                    _handler_db = db_path
+                    _handler = handler
 
-                    def _mail_on_complete(aid, outcome):
-                        # 幻觉门控：检查是否有回复，自动标 done/failed
+                    def _handler_on_complete(aid, outcome):
                         try:
-                            _dispatcher._mail_auto_complete(_task_id, aid, _mail_db, _must_haves)
+                            _handler.post_complete(
+                                _task_id, aid, outcome, _handler_db)
                         except Exception as e:
-                            logger.error("Mail %s: on_complete error: %s", _task_id, e)
-                    on_complete = _mail_on_complete
+                            logger.error(
+                                "Handler %s: on_complete error: %s", _task_id, e)
+                    on_complete = _handler_on_complete
                 else:
-                    # #02: Task 路径也加 on_complete（幻觉门控）
+                    # 旧路径：无 handler 的项目（_general 等）
                     _task_id = task.id
-                    _task_db = Path(project_config["db_path"]) if project_config and "db_path" in project_config else None
+                    _task_db = db_path
                     _dispatcher = self
                     _is_review = action_type == "review"
 
-                    # #07.2: executor/review 统一 crash 回退
                     ROLLBACK_CURRENT_AGENT_OUTCOMES = frozenset({
                         "crashed", "compact_failed", "process_crash",
                         "session_stuck", "compact_hanging",
                     })
 
-                    def _task_on_complete(aid, outcome):
+                    def _legacy_on_complete(aid, outcome):
                         try:
-                            # #07.2: 统一 crash 回退——executor 和 review 都回退 current_agent
                             if outcome in ROLLBACK_CURRENT_AGENT_OUTCOMES and _task_db:
-                                _dispatcher._rollback_current_agent(_task_db, _task_id, aid)
-
-                            if _is_review:
-                                if _task_db and outcome in ("completed", "session_revived"):
-                                    # #09: 读 verdict 决定后续动作
-                                    conn = get_connection(_task_db)
-                                    try:
-                                        review = conn.execute(
-                                            "SELECT verdict FROM reviews WHERE task_id=? ORDER BY created_at DESC LIMIT 1",
-                                            (_task_id,)
-                                        ).fetchone()
-                                    finally:
-                                        conn.close()
-
-                                    if review and review["verdict"] == "approved":
-                                        _dispatcher._mark_task_status(_task_db, _task_id, "done")
-                                        logger.info("Task %s: review approved, marking done", _task_id)
-                                    else:
-                                        # 非 approved → @mention 被审 agent（assignee，非 current_agent）
-                                        verdict_str = review["verdict"] if review else "未知"
-                                        conn2 = get_connection(_task_db)
-                                        try:
-                                            task_row = conn2.execute("SELECT assignee FROM tasks WHERE id=?", (_task_id,)).fetchone()
-                                        finally:
-                                            conn2.close()
-
-                                        if task_row and task_row["assignee"]:
-                                            from src.blackboard.blackboard import Blackboard
-                                            bb = Blackboard(_task_db)
-                                            bb.add_comment(_task_id, "daemon",
-                                                f"@{task_row['assignee']} 审查结论: {verdict_str}，请查看详情并决定接受或反驳",
-                                                comment_type="review")
-                                        logger.info("Task %s: review verdict=%s, notified assignee=%s",
-                                                    _task_id, verdict_str, task_row["assignee"] if task_row else "?")
-                                        # 不标 done，保持 review 状态
-                                else:
-                                    logger.warning("Task %s: review agent %s (%s), NOT marking done", _task_id, aid, outcome)
-                            else:
-                                # executor: 三信号验证 → 标 review
-                                _dispatcher._task_auto_complete(_task_id, _task_db)
+                                _dispatcher._rollback_current_agent(
+                                    _task_db, _task_id, aid)
+                            if not _is_review:
+                                _dispatcher._task_auto_complete(
+                                    _task_id, _task_db)
                         except Exception as e:
-                            logger.error("Task %s: on_complete error: %s", _task_id, e)
-                    on_complete = _task_on_complete
+                            logger.error(
+                                "Legacy %s: on_complete error: %s", _task_id, e)
+                    on_complete = _legacy_on_complete
 
                 session_id = await self.spawner.spawn_full_agent(
                     agent_id=agent_id,
@@ -289,7 +266,8 @@ class Dispatcher:
                     task_id=task.id,
                     on_complete=on_complete,
                     use_main_session=True,  # #02: 统一投递到 main session
-                    task_db_path=Path(project_config["db_path"]) if project_config and "db_path" in project_config else None,
+                    task_db_path=Path(
+                        project_config["db_path"]) if project_config and "db_path" in project_config else None,
                     on_checks_passed=on_checks_passed,
                 )
 
@@ -312,9 +290,14 @@ class Dispatcher:
                 else:
                     log_level = logger.debug
                     detail_msg = f"Agent busy: {reason}"
-                log_level("Dispatch skipped %s for task %s: %s", agent_id, task.id, detail_msg)
+                log_level(
+                    "Dispatch skipped %s for task %s: %s",
+                    agent_id,
+                    task.id,
+                    detail_msg)
                 # on_checks_passed 未执行（check 失败在它之前），working 未标，无需回退
-                self._record_routing(task, decision, "skipped", detail_msg, _routing_db)
+                self._record_routing(
+                    task, decision, "skipped", detail_msg, _routing_db)
                 return {
                     "level": level.value,
                     "agent_id": agent_id,
@@ -324,9 +307,28 @@ class Dispatcher:
                 }
             except Exception as e:
                 # on_checks_passed 已执行但 subprocess 失败 → 回退 working → pending
-                if _mail_marked_working:
-                    self._mail_revert_to_pending(task.id, db_path)
-                self._record_routing(task, decision, "error", str(e), _routing_db)
+                if handler_marked_working and handler and db_path:
+                    # handler 项目：回退到 pending
+                    try:
+                        conn = get_connection(db_path)
+                        try:
+                            conn.execute("BEGIN IMMEDIATE")
+                            row = conn.execute(
+                                "SELECT status FROM tasks WHERE id=?", (task.id,)).fetchone()
+                            if row and row["status"] == "working":
+                                conn.execute(
+                                    "UPDATE tasks SET status='pending', updated_at=datetime('now') WHERE id=?",
+                                    (task.id,))
+                                conn.commit()
+                                logger.info(
+                                    "Task %s: reverted working → pending (spawn failed)", task.id)
+                        finally:
+                            conn.close()
+                    except Exception as revert_err:
+                        logger.error(
+                            "Task %s: failed to revert to pending: %s", task.id, revert_err)
+                self._record_routing(
+                    task, decision, "error", str(e), _routing_db)
                 return {
                     "level": level.value,
                     "agent_id": agent_id,
@@ -385,9 +387,16 @@ class Dispatcher:
     def _build_delegate_prompt(self, task: Task,
                                project_config: Optional[Dict]) -> str:
         """构建 delegate 模式的 prompt（协调员分配任务）"""
-        api_host = getattr(self.spawner, 'api_host', '127.0.0.1') if self.spawner else '127.0.0.1'
-        api_port = getattr(self.spawner, 'api_port', 8083) if self.spawner else 8083
-        project_id = project_config.get("project_id", "") if project_config else ""
+        api_host = getattr(
+            self.spawner,
+            'api_host',
+            '127.0.0.1') if self.spawner else '127.0.0.1'
+        api_port = getattr(
+            self.spawner,
+            'api_port',
+            8083) if self.spawner else 8083
+        project_id = project_config.get(
+            "project_id", "") if project_config else ""
 
         return f"""你是任务协调员。请分析以下任务，决定最合适的执行者并分配。
 
@@ -478,7 +487,8 @@ class Dispatcher:
 
     # ── Legacy 兼容（deprecated） ──
 
-    def _legacy_decide(self, task: Task, action_type: str = "") -> Dict[str, Any]:
+    def _legacy_decide(
+            self, task: Task, action_type: str = "") -> Dict[str, Any]:
         """旧版三级决策树（兼容过渡用）"""
         LOCAL_ACTIONS = frozenset({
             "L1_guardrail", "format_check",
@@ -518,7 +528,8 @@ class Dispatcher:
                 return registered[0]
         return "pangtong-fujunshi"
 
-    async def _legacy_dispatch(self, task, action_type="", project_config=None):
+    async def _legacy_dispatch(
+            self, task, action_type="", project_config=None):
         """旧版 dispatch（兼容过渡用）
 
         v2.7.2: counter acquire/release 移到 spawn_full_agent 内部。
@@ -540,16 +551,21 @@ class Dispatcher:
             try:
                 # NOTE: _legacy_dispatch 仅在 router=None 时触发，当前配置不会进入。
                 # Mail 永远走 dispatch() 主路径（on_checks_passed 方案），不走此路径。
-                # 如果未来 legacy 路径被启用，需同步 on_checks_passed 逻辑。
-                is_mail_legacy = project_config.get("project_id") == "_mail" if project_config else False
-                if is_mail_legacy:
-                    db_path_legacy = Path(project_config["db_path"]) if project_config and "db_path" in project_config else None
-                    if not db_path_legacy or not self._mail_auto_working(task.id, db_path_legacy):
+                # [Step 5] handler 统一：用注册表查 handler
+                project_id_legacy = project_config.get("project_id", "") if project_config else ""
+                handler_legacy = TaskTypeRegistry.get_by_project(project_id_legacy)
+                if handler_legacy:
+                    db_path_legacy = Path(
+                        project_config["db_path"]) if project_config and "db_path" in project_config else None
+                    if db_path_legacy:
+                        handler_legacy.pre_spawn(task.id, db_path_legacy)
+                    else:
                         return {"level": level.value, "agent_id": agent_id,
                                 "session_id": None, "status": "error",
-                                "reason": "mail_auto_working_failed"}
+                                "reason": "no db_path for handler"}
 
-                if hasattr(self.spawner, 'build_spawn_message') and project_config:
+                if hasattr(self.spawner,
+                           'build_spawn_message') and project_config:
                     retry_ctx = self._build_retry_context(task)
                     message = self.spawner.build_spawn_message(
                         task_id=task.id, title=task.title,
@@ -568,32 +584,34 @@ class Dispatcher:
 
                 # v2.7.2: on_complete 只含业务逻辑
                 on_complete_legacy = None
-                if is_mail_legacy:
+                if handler_legacy:
                     _t_id = task.id
-                    _m_db = db_path_legacy
-                    _m_mh = task.must_haves or ""
-                    _disp = self
+                    _h_db = db_path_legacy
+                    _h = handler_legacy
 
-                    def _mail_oc_legacy(aid, outcome):
+                    def _handler_oc_legacy(aid, outcome):
                         try:
-                            _disp._mail_auto_complete(_t_id, aid, _m_db, _m_mh)
+                            _h.post_complete(_t_id, aid, outcome, _h_db)
                         except Exception as e:
-                            logger.error("Mail %s: legacy on_complete error: %s", _t_id, e)
-                    on_complete_legacy = _mail_oc_legacy
+                            logger.error(
+                                "Handler %s: legacy on_complete error: %s", _t_id, e)
+                    on_complete_legacy = _handler_oc_legacy
 
                 session_id = await self.spawner.spawn_full_agent(
                     agent_id=agent_id, message=message,
                     task_id=task.id,
                     on_complete=on_complete_legacy,
                     use_main_session=True,  # #02: 统一投递到 main session
-                    task_db_path=Path(project_config["db_path"]) if project_config and "db_path" in project_config else None,
+                    task_db_path=Path(
+                        project_config["db_path"]) if project_config and "db_path" in project_config else None,
                 )
                 return {"level": level.value, "agent_id": agent_id,
                         "session_id": session_id, "status": "dispatched",
                         "reason": decision["reason"]}
             except AgentBusyError as e:
                 reason = getattr(e, 'reason', 'busy')
-                detail_msg = f"Session busy: {reason}" if reason.startswith("session_") else f"Agent busy: {reason}"
+                detail_msg = f"Session busy: {reason}" if reason.startswith(
+                    "session_") else f"Agent busy: {reason}"
                 return {"level": level.value, "agent_id": agent_id,
                         "session_id": None, "status": "skipped",
                         "reason": detail_msg}
@@ -618,9 +636,11 @@ class Dispatcher:
             conn = get_connection(db_path)
             try:
                 conn.execute("BEGIN IMMEDIATE")
-                row = conn.execute("SELECT status FROM tasks WHERE id=?", (task_id,)).fetchone()
+                row = conn.execute(
+                    "SELECT status FROM tasks WHERE id=?", (task_id,)).fetchone()
                 if not row:
-                    logger.warning("Mail %s: cannot mark working (task not found)", task_id)
+                    logger.warning(
+                        "Mail %s: cannot mark working (task not found)", task_id)
                     return False
                 if row["status"] not in ("pending", "claimed"):
                     logger.warning("Mail %s: cannot mark working (status=%s, expected pending/claimed)",
@@ -631,7 +651,10 @@ class Dispatcher:
                     (task_id,),
                 )
                 conn.commit()
-                logger.info("Mail %s: auto-marked working (system, was %s)", task_id, row["status"])
+                logger.info(
+                    "Mail %s: auto-marked working (system, was %s)",
+                    task_id,
+                    row["status"])
                 return True
             finally:
                 conn.close()
@@ -645,30 +668,40 @@ class Dispatcher:
             conn = get_connection(db_path)
             try:
                 conn.execute("BEGIN IMMEDIATE")
-                row = conn.execute("SELECT status FROM tasks WHERE id=?", (task_id,)).fetchone()
+                row = conn.execute(
+                    "SELECT status FROM tasks WHERE id=?", (task_id,)).fetchone()
                 if row and row["status"] == "working":
                     conn.execute(
                         "UPDATE tasks SET status='pending', updated_at=datetime('now') WHERE id=?",
                         (task_id,),
                     )
                     conn.commit()
-                    logger.info("Mail %s: reverted working → pending (spawn failed)", task_id)
+                    logger.info(
+                        "Mail %s: reverted working → pending (spawn failed)", task_id)
                 else:
-                    logger.debug("Mail %s: skip revert (status=%s, expected working)", task_id, row["status"] if row else "not_found")
+                    logger.debug(
+                        "Mail %s: skip revert (status=%s, expected working)",
+                        task_id,
+                        row["status"] if row else "not_found")
             finally:
                 conn.close()
         except Exception as e:
-            logger.error("Mail %s: failed to revert to pending: %s", task_id, e)
+            logger.error(
+                "Mail %s: failed to revert to pending: %s",
+                task_id,
+                e)
 
     def _mail_auto_complete(self, task_id: str, agent_id: str,
-                             db_path: Path, must_haves: str) -> None:
+                            db_path: Path, must_haves: str, outcome=None) -> None:
         """Mail 任务：on_complete 后自动标 done/failed（含幻觉门控）"""
         try:
             # 解析 performative
             performative = "request"
             try:
                 meta = json.loads(must_haves) if must_haves else {}
-                performative = meta.get("performative", meta.get("type", "request"))
+                performative = meta.get(
+                    "performative", meta.get(
+                        "type", "request"))
             except Exception:
                 pass
 
@@ -677,13 +710,15 @@ class Dispatcher:
                 has_reply = self._mail_check_reply(task_id, db_path)
                 if not has_reply:
                     # F3: 立刻标 failed（不等 ticker 30 分钟）
-                    logger.error("Mail %s: no reply found, marking failed (no_reply_found)", task_id)
+                    logger.error(
+                        "Mail %s: no reply found, marking failed (no_reply_found)", task_id)
                     for attempt in range(3):
                         try:
                             conn = get_connection(db_path)
                             try:
                                 conn.execute("BEGIN IMMEDIATE")
-                                row = conn.execute("SELECT status FROM tasks WHERE id=?", (task_id,)).fetchone()
+                                row = conn.execute(
+                                    "SELECT status FROM tasks WHERE id=?", (task_id,)).fetchone()
                                 if not row:
                                     return
                                 if row["status"] == "working":
@@ -697,19 +732,35 @@ class Dispatcher:
                                          json.dumps({"reason": "no_reply_found"}, ensure_ascii=False)),
                                     )
                                     conn.commit()
-                                    logger.info("Mail %s: marked failed (no_reply_found)", task_id)
+                                    logger.info(
+                                        "Mail %s: marked failed (no_reply_found)", task_id)
                                     # Mail 失败通知：通知发件人
                                     try:
                                         from src.daemon.mail_notify import notify_mail_failed
-                                        notify_mail_failed(db_path, task_id, "no_reply_found")
+                                        notify_mail_failed(
+                                            db_path, task_id, "no_reply_found")
                                     except Exception as ne:
-                                        logger.warning("Mail %s: failed to send no_reply_found notification: %s", task_id, ne)
+                                        logger.warning(
+                                            "Mail %s: failed to send no_reply_found notification: %s", task_id, ne)
                                 return
                             finally:
                                 conn.close()
                         except Exception as e:
-                            logger.warning("Mail %s: failed attempt %d: %s", task_id, attempt + 1, e)
-                    logger.error("Mail %s: all 3 failed attempts failed, leaving for ticker", task_id)
+                            logger.warning(
+                                "Mail %s: failed attempt %d: %s", task_id, attempt + 1, e)
+                    logger.error(
+                        "Mail %s: all 3 failed attempts failed, leaving for ticker", task_id)
+                    return
+
+            # inform 类型：只对成功 outcome 标 done，失败 outcome 留 working 等 ticker 重投
+            # Task 路径不受此 bug 影响（走 _task_auto_complete 独立逻辑）
+            if performative == "inform":
+                INFORM_DONE_OUTCOMES = {"completed", "claimed", "no_reply"}
+                if outcome not in INFORM_DONE_OUTCOMES:
+                    logger.info(
+                        "Mail %s: inform outcome=%s, skip auto-done",
+                        task_id,
+                        outcome)
                     return
 
             # 标 done（重试 3 次）
@@ -718,7 +769,8 @@ class Dispatcher:
                     conn = get_connection(db_path)
                     try:
                         conn.execute("BEGIN IMMEDIATE")
-                        row = conn.execute("SELECT status FROM tasks WHERE id=?", (task_id,)).fetchone()
+                        row = conn.execute(
+                            "SELECT status FROM tasks WHERE id=?", (task_id,)).fetchone()
                         if not row:
                             return
                         if row["status"] == "working":
@@ -733,9 +785,15 @@ class Dispatcher:
                     finally:
                         conn.close()
                 except Exception as e:
-                    logger.warning("Mail %s: done attempt %d failed: %s", task_id, attempt + 1, e)
+                    logger.warning(
+                        "Mail %s: done attempt %d failed: %s",
+                        task_id,
+                        attempt + 1,
+                        e)
             # 3 次都失败，留 working 等 ticker 超时兜底
-            logger.error("Mail %s: all 3 done attempts failed, leaving for ticker", task_id)
+            logger.error(
+                "Mail %s: all 3 done attempts failed, leaving for ticker",
+                task_id)
 
         except Exception as e:
             logger.error("Mail %s: auto-complete error: %s", task_id, e)
@@ -780,7 +838,9 @@ class Dispatcher:
                 logger.info("Task %s: verify passed, marking review", task_id)
                 self._mark_task_status(db_path, task_id, "review")
             else:
-                logger.info("Task %s: verify not passed (no signal), leaving working", task_id)
+                logger.info(
+                    "Task %s: verify not passed (no signal), leaving working",
+                    task_id)
         except Exception as e:
             logger.error("Task %s: auto-complete error: %s", task_id, e)
 
@@ -815,7 +875,8 @@ class Dispatcher:
             logger.error("Task %s: verify error: %s", task_id, e)
             return True
 
-    def _rollback_current_agent(self, db_path: Path, task_id: str, agent_id: str) -> None:
+    def _rollback_current_agent(
+            self, db_path: Path, task_id: str, agent_id: str) -> None:
         """#07.2: crash 后回退 current_agent 到 assignee，避免 exclude_current 卡死"""
         try:
             conn = get_connection(db_path)
@@ -829,11 +890,18 @@ class Dispatcher:
                 conn.commit()
             finally:
                 conn.close()
-            logger.info("Task %s: rolled back current_agent from %s to assignee", task_id, agent_id)
+            logger.info(
+                "Task %s: rolled back current_agent from %s to assignee",
+                task_id,
+                agent_id)
         except Exception as e:
-            logger.warning("Task %s: failed to rollback current_agent: %s", task_id, e)
+            logger.warning(
+                "Task %s: failed to rollback current_agent: %s",
+                task_id,
+                e)
 
-    def _mark_task_status(self, db_path: Path, task_id: str, status: str) -> None:
+    def _mark_task_status(self, db_path: Path,
+                          task_id: str, status: str) -> None:
         """更新任务状态 + 写审计事件"""
         try:
             conn = get_connection(db_path)
@@ -849,7 +917,8 @@ class Dispatcher:
                 )
                 conn.execute(
                     "INSERT INTO events (task_id, agent, event_type, payload) VALUES (?, 'dispatcher', 'status_change', ?)",
-                    (task_id, f'{{"from": "{old_status}", "to": "{status}", "source": "auto_complete"}}'),
+                    (task_id,
+                     f'{{"from": "{old_status}", "to": "{status}", "source": "auto_complete"}}'),
                 )
                 conn.commit()
             finally:
@@ -858,7 +927,7 @@ class Dispatcher:
             logger.error("Task %s: mark status error: %s", task_id, e)
 
     @staticmethod
-    def _check_crash_limit(task_id: str, db_path: pathlib.Path, limit: int = 3,
+    def _check_crash_limit(task_id: str, db_path: Path, limit: int = 3,
                            window_minutes: int = 30) -> bool:
         """v2.8.1 Fix-3c: 检查 task 最近 window_minutes 内的 crash 次数是否超限。
 

src/daemon/experience.py

@@ -14,7 +14,7 @@ import logging
 import re
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional
 
 logger = logging.getLogger("moziplus-v2.experience")
 
@@ -68,7 +68,7 @@ class Experience:
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> Experience:
         return cls(**{k: v for k, v in data.items() if k != "id"},
-                    experience_id=data.get("id"))
+                   experience_id=data.get("id"))
 
 
 class ExperienceStore:
@@ -284,7 +284,7 @@ class ExperienceDistiller:
             all_tags.append(task_type)
 
         results = self.store.search(tags=all_tags if all_tags else None,
-                                     query=query, limit=limit)
+                                    query=query, limit=limit)
 
         # 按置信度排序
         results.sort(key=lambda e: e.confidence, reverse=True)

src/daemon/guardrails.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import logging
 import re
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
@@ -38,7 +38,9 @@ class GuardrailEngine:
             data = yaml.safe_load(f)
         self.rules = data.get("rules", [])
         self.settings = data.get("settings", {"enabled": True})
-        logger.info("Loaded %d guardrail rules from %s", len(self.rules), config_path)
+        logger.info(
+            "Loaded %d guardrail rules from %s", len(
+                self.rules), config_path)
 
     def check_task(self, task: Any) -> List[GuardrailViolation]:
         """检查 Task 是否触犯安全红线（调度前调用）"""
@@ -95,7 +97,8 @@ class GuardrailEngine:
 
         return violations
 
-    def check_token_usage(self, token_count: int) -> Optional[GuardrailViolation]:
+    def check_token_usage(
+            self, token_count: int) -> Optional[GuardrailViolation]:
         """检查 Token 消耗是否超标"""
         if not self.settings.get("enabled", True):
             return None
@@ -103,7 +106,10 @@ class GuardrailEngine:
         for rule in self.rules:
             if rule["id"] != "high_token_usage":
                 continue
-            threshold = rule.get("triggers", [{}])[0].get("token_threshold", 100000)
+            threshold = rule.get(
+                "triggers", [
+                    {}])[0].get(
+                "token_threshold", 100000)
             if token_count > threshold:
                 return GuardrailViolation(
                     rule_id=rule["id"],
@@ -114,7 +120,8 @@ class GuardrailEngine:
                 )
         return None
 
-    def check_consecutive_failure(self, failure_count: int) -> Optional[GuardrailViolation]:
+    def check_consecutive_failure(
+            self, failure_count: int) -> Optional[GuardrailViolation]:
         """检查连续失败次数"""
         if not self.settings.get("enabled", True):
             return None
@@ -122,7 +129,10 @@ class GuardrailEngine:
         for rule in self.rules:
             if rule["id"] != "consecutive_failure":
                 continue
-            threshold = rule.get("triggers", [{}])[0].get("consecutive_failures", 3)
+            threshold = rule.get(
+                "triggers", [
+                    {}])[0].get(
+                "consecutive_failures", 3)
             if failure_count >= threshold:
                 return GuardrailViolation(
                     rule_id=rule["id"],

src/daemon/health.py

@@ -9,9 +9,9 @@ from __future__ import annotations
 import json
 import logging
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import Any, Dict
 
-from src.blackboard.db import get_connection, init_db
+from src.blackboard.db import get_connection
 from src.blackboard.queries import Queries
 
 logger = logging.getLogger("moziplus-v2.health")
@@ -41,7 +41,7 @@ class HealthChecker:
             {"healthy": bool, "zombie": bool, "stale_ticks": int,
              "alert_written": bool, "resolved": bool}
         """
-        db_key = str(db_path)
+        str(db_path)
         result: Dict[str, Any] = {
             "healthy": True,
             "zombie": False,
@@ -58,7 +58,8 @@ class HealthChecker:
         # 用 event count 变化判断是否有真实变更
         conn = queries._conn()
         try:
-            total_events = conn.execute("SELECT COUNT(*) FROM events").fetchone()[0]
+            conn.execute(
+                "SELECT COUNT(*) FROM events").fetchone()[0]
             non_tick_events = conn.execute(
                 "SELECT COUNT(*) FROM events WHERE event_type != 'daemon_tick' "
                 "AND event_type != 'agent_zombie_detected'"
@@ -85,7 +86,8 @@ class HealthChecker:
             self._stale_ticks[project_id] = stale
             result["stale_ticks"] = stale
 
-            if stale >= self.zombie_threshold and not self._alerted.get(project_id):
+            if stale >= self.zombie_threshold and not self._alerted.get(
+                    project_id):
                 # 写告警
                 self._write_alert(db_path, project_id, tick_num, stale)
                 self._alerted[project_id] = True
@@ -126,7 +128,10 @@ class HealthChecker:
             conn.commit()
         finally:
             conn.close()
-        logger.warning("Zombie detected: %s (stale=%d)", project_id, stale_ticks)
+        logger.warning(
+            "Zombie detected: %s (stale=%d)",
+            project_id,
+            stale_ticks)
 
     def _write_resolution(self, db_path: Path, project_id: str,
                           tick_num: int) -> None:

src/daemon/inbox.py

@@ -15,7 +15,6 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
-import os
 from pathlib import Path
 from typing import Any, Callable, Coroutine, Dict, List, Optional
 
@@ -28,7 +27,8 @@ class InboxWatcher:
     def __init__(
         self,
         inbox_path: Path,
-        process_callback: Optional[Callable[[Dict[str, Any]], Coroutine[Any, Any, None]]] = None,
+        process_callback: Optional[Callable[[
+            Dict[str, Any]], Coroutine[Any, Any, None]]] = None,
         watch_interval: float = 1.0,
     ):
         """
@@ -57,7 +57,7 @@ class InboxWatcher:
         self._running = True
         self._task = asyncio.create_task(self._loop())
         logger.info("Inbox watcher started (path=%s, interval=%.1fs)",
-                     self.inbox_path, self.watch_interval)
+                    self.inbox_path, self.watch_interval)
 
     async def stop(self) -> None:
         """停止监听"""
@@ -69,7 +69,7 @@ class InboxWatcher:
             except asyncio.CancelledError:
                 pass
         logger.info("Inbox watcher stopped (processed=%d, errors=%d)",
-                     self._total_processed, self._total_errors)
+                    self._total_processed, self._total_errors)
 
     @property
     def is_running(self) -> bool:
@@ -160,7 +160,8 @@ class InboxWatcher:
                                    line_no, type(event).__name__)
                     self._total_errors += 1
             except json.JSONDecodeError:
-                logger.warning("Inbox line %d: invalid JSON, skipping", line_no)
+                logger.warning(
+                    "Inbox line %d: invalid JSON, skipping", line_no)
                 self._total_errors += 1
 
         return events

src/daemon/mail_handler.py

@@ -0,0 +1,210 @@
+"""mail_handler.py — Mail 任务 handler。
+
+处理 Agent 间通信（飞鸽传书），含 inform 和 request 两种类型。
+"""
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+from typing import Dict, Optional
+
+from src.daemon.base_task_handler import BaseTaskHandler, VerifyResult
+from src.daemon.prompt_composer import PromptComposer, PromptContext
+from src.blackboard.db import get_connection
+
+logger = logging.getLogger("moziplus-v2.handler.mail")
+
+class MailHandler(BaseTaskHandler):
+    """Mail 任务 handler。"""
+
+    task_type = "mail"
+    virtual_project = "_mail"
+    display_name = "飞鸽传书"
+
+    def target_success_status(self) -> str:
+        return "done"
+
+    def pre_spawn(self, task_id: str, db_path: Path) -> bool:
+        """auto_working：pending → working"""
+        return self._auto_mark_working(task_id, db_path)
+
+    def build_prompt(self, context: PromptContext) -> str:
+        """通过 PromptComposer 拼装 3 个 section。"""
+        composer = PromptComposer()
+        composer.add_many(self.get_sections())
+        return composer.compose(context)
+
+    def get_sections(self) -> list:
+        return [MailContextSection(), MailApiSection(), MailConstraintsSection()]
+
+    def verify_completion(self, task_id: str, db_path: Path) -> VerifyResult:
+        """Mail 完成验证：区分 inform/request。
+
+        - inform: 始终通过（通知已阅即 done，不需要检查产出）
+        - request: 检查是否已回复
+        """
+        performative = self._parse_performative(task_id, db_path)
+
+        if performative == "inform":
+            return VerifyResult(True, "inform_auto", f"performative={performative}")
+
+        # request: 检查是否已回复
+        has_reply = self._check_reply(task_id, db_path)
+        if has_reply:
+            return VerifyResult(True, "has_reply", f"performative={performative}")
+        return VerifyResult(False, "no_reply", f"performative={performative}")
+
+    # post_complete 由基类 BaseTaskHandler 统一处理（crash→verify→mark→notify）
+    # inform: verify 始终通过 → 基类 mark done ✅
+    # request 有回复: verify 通过 → 基类 mark done ✅
+    # request 无回复: verify 失败 → 基类调 on_failure ✅
+
+    def on_failure(self, task_id: str, agent_id: str,
+                   db_path: Path, verify: VerifyResult) -> None:
+        """request 验证失败 → 标 failed + 通知发件人"""
+        self._mark_task_status(db_path, task_id, "failed")
+        logger.info("Mail %s: request verify failed (%s), marked failed",
+                     task_id, verify.reason)
+
+        # 通知发件人
+        try:
+            from src.daemon.mail_notify import notify_mail_failed
+            notify_mail_failed(db_path, task_id, "no_reply_found")
+        except Exception as e:
+            logger.warning("Mail %s: failed to send notification: %s", task_id, e)
+
+    # === 内部方法 ===
+
+    def _parse_performative(self, task_id: str, db_path: Path) -> str:
+        """解析 mail 类型（inform/request）"""
+        try:
+            conn = get_connection(db_path)
+            try:
+                row = conn.execute(
+                    "SELECT must_haves FROM tasks WHERE id=?", (task_id,)
+                ).fetchone()
+                if row and row["must_haves"]:
+                    meta = json.loads(row["must_haves"])
+                    return meta.get("performative", meta.get("type", "request"))
+            finally:
+                conn.close()
+        except Exception:
+            pass
+        return "request"
+
+    def _check_reply(self, task_id: str, db_path: Path) -> bool:
+        """检查是否已回复（查 tasks 表找 in_reply_to 回复邮件）
+        
+        从 dispatcher._mail_check_reply 迁移。
+        Mail 回复机制：创建新 task，must_haves JSON 中包含 in_reply_to = original_task_id。
+        不能查 comments 表——回复邮件是独立的 task，不是 comment。
+        """
+        try:
+            conn = get_connection(db_path)
+            try:
+                row = conn.execute(
+                    "SELECT id FROM tasks WHERE id != ? AND must_haves LIKE ? LIMIT 1",
+                    (task_id, f'%{task_id}%'),
+                ).fetchone()
+                return row is not None
+            finally:
+                conn.close()
+        except Exception as e:
+            logger.error("Mail %s: check reply error: %s", task_id, e)
+            # 查询失败时保守处理：假设有回复（避免误标 failed）
+            return True
+
+    def check_completion(self, task_id: str, db_path: Path) -> bool:
+        """ticker 级别的完成检查：检查是否已回复"""
+        return self._check_reply(task_id, db_path)
+
+
+# ===================================================================
+# Mail PromptSections
+# ===================================================================
+
+class MailContextSection:
+    """邮件上下文段 — 发件人/收件人/主题/内容，区分 inform/request。"""
+
+    name: str = "mail_context"
+    priority: int = 10
+
+    def render(self, context: PromptContext) -> str:
+        if context.mail_type == "inform":
+            return self._render_inform(context)
+        return self._render_request(context)
+
+    def should_include(self, context: PromptContext) -> bool:  # noqa: ARG002
+        return True
+
+    @staticmethod
+    def _render_inform(context: PromptContext) -> str:
+        return (
+            f"你收到一封飞鸽传书(纯通知)。\n\n"
+            f"发件者: {context.from_agent}\n"
+            f"主题: {context.title}\n"
+            f"内容: {context.description}\n\n"
+            f"已阅即可。如需回复,用 in_reply_to 回复发件者(不需要填 to)。\n"
+            f"⚠️ 不要执行任何状态转换命令。"
+        )
+
+    @staticmethod
+    def _render_request(context: PromptContext) -> str:
+        return (
+            f"你收到一封飞鸽传书,需要你处理并回复。\n\n"
+            f"发件者: {context.from_agent}\n"
+            f"主题: {context.title}\n"
+            f"内容: {context.description}\n\n"
+            f"### 如何回复发件者\n\n"
+            f'curl -s -X POST http://localhost:8083/api/mail \\\n'
+            f"  -H 'Content-Type: application/json' \\\n"
+            f'  -d \'{{"from": "{context.agent_id}", '
+            f'"in_reply_to": "{context.task_id}", '
+            f'"title": "回复: {context.title}", '
+            f'"text": "你的回复内容"}}\'\n\n'
+            f"⚠️ 不需要填 \"to\",系统自动回复给发件者。"
+        )
+
+
+class MailApiSection:
+    """Mail API 操作指令段。"""
+
+    name: str = "mail_api"
+    priority: int = 40
+
+    def render(self, context: PromptContext) -> str:
+        return (
+            f"### 如何给其他人发新邮件\n\n"
+            f'curl -s -X POST http://localhost:8083/api/mail \\\n'
+            f"  -H 'Content-Type: application/json' \\\n"
+            f'  -d \'{{"from": "{context.agent_id}", '
+            f'"to": "对方agent-id", '
+            f'"title": "标题", '
+            f'"text": "正文", '
+            f'"type": "inform"}}\'\n\n'
+            f"⚠️ to 必须是有效的 agent id\n"
+            f"⚠️ 纯通知用 type=inform,需要对方回复不填 type(默认 request)"
+        )
+
+    def should_include(self, context: PromptContext) -> bool:
+        return context.mail_type == "request"
+
+
+class MailConstraintsSection:
+    """Mail 硬约束段。"""
+
+    name: str = "mail_constraints"
+    priority: int = 50
+
+    def render(self, context: PromptContext) -> str:  # noqa: ARG002
+        return (
+            "## 硬约束\n\n"
+            "1. ⚠️ 不要执行任何状态转换命令(标 working/done/review/failed 等),系统会自动处理。\n"
+            "2. ⚠️ 不能给自己发邮件\n"
+            "3. ⚠️ 发邮件时 to 必须是有效的 agent id\n"
+            "4. ⚠️ 纯通知用 type=inform,需要对方回复不填 type(默认 request)"
+        )
+
+    def should_include(self, context: PromptContext) -> bool:  # noqa: ARG002
+        return True

src/daemon/mail_notify.py

@@ -50,7 +50,9 @@ def notify_mail_failed(db_path: Path, original_mail_id: str,
         bb = Blackboard(db_path)
         original = bb.get_task(original_mail_id)
         if not original:
-            logger.warning("notify_mail_failed: original mail %s not found", original_mail_id)
+            logger.warning(
+                "notify_mail_failed: original mail %s not found",
+                original_mail_id)
             return
 
         # 解析原邮件元数据
@@ -58,7 +60,9 @@ def notify_mail_failed(db_path: Path, original_mail_id: str,
 
         # 防递归：系统通知邮件失败不再发通知
         if meta.get("system_notify"):
-            logger.info("Mail %s: system notify mail failed, skipping recursive notification", original_mail_id)
+            logger.info(
+                "Mail %s: system notify mail failed, skipping recursive notification",
+                original_mail_id)
             return
 
         # 获取发件人（优先 assigned_by，fallback must_haves.from）
@@ -67,7 +71,9 @@ def notify_mail_failed(db_path: Path, original_mail_id: str,
         title = original.title or ""
 
         if not from_agent:
-            logger.warning("notify_mail_failed: cannot determine sender for mail %s", original_mail_id)
+            logger.warning(
+                "notify_mail_failed: cannot determine sender for mail %s",
+                original_mail_id)
             return
 
         # 发件人不是有效 Agent（如 system）→ 通知庞统代处理，不触发广播
@@ -108,7 +114,10 @@ def notify_mail_failed(db_path: Path, original_mail_id: str,
         )
         bb.create_task(notify_task)
         logger.info("Mail %s: sent failure notification to %s (original_sender=%s, reason=%s, notify_id=%s)",
-                     original_mail_id, target_agent, from_agent, reason, notify_id)
+                    original_mail_id, target_agent, from_agent, reason, notify_id)
 
     except Exception as e:
-        logger.warning("notify_mail_failed: failed to send notification for mail %s: %s", original_mail_id, e)
+        logger.warning(
+            "notify_mail_failed: failed to send notification for mail %s: %s",
+            original_mail_id,
+            e)

src/daemon/prompt_composer.py

@@ -0,0 +1,127 @@
+"""
+prompt_composer.py — PromptSection Protocol + PromptContext + PromptComposer
+
+拼装器：有序管理 prompt 段落，按优先级排序后合并为最终 prompt。
+"""
+
+import logging
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Protocol, runtime_checkable
+
+logger = logging.getLogger("moziplus-v2.prompt_composer")
+
+# ---------------------------------------------------------------------------
+# Section 优先级范围约定
+# ---------------------------------------------------------------------------
+PRIORITY_CONTEXT = 10        # 任务上下文
+PRIORITY_PRIOR = 20          # 前序信息
+PRIORITY_ROLE = 30           # 角色规范
+PRIORITY_API = 40            # API 操作指令
+PRIORITY_CONSTRAINTS = 50    # 硬约束
+PRIORITY_EXTENSION = 60      # 扩展段
+
+
+# ---------------------------------------------------------------------------
+# PromptSection Protocol
+# ---------------------------------------------------------------------------
+@runtime_checkable
+class PromptSection(Protocol):
+    """一个 prompt 段"""
+
+    name: str                     # 段名（去重用，同名覆盖）
+    priority: int                 # 排序优先级（小数字=靠前）
+
+    def render(self, context: "PromptContext") -> str:
+        """渲染此段的文本内容。返回空字符串表示不注入。"""
+        ...
+
+    def should_include(self, context: "PromptContext") -> bool:
+        """是否注入此段（默认 True，条件段可覆盖）。"""
+        ...
+
+
+# ---------------------------------------------------------------------------
+# PromptContext 数据对象
+# ---------------------------------------------------------------------------
+@dataclass
+class PromptContext:
+    """Prompt 渲染的统一上下文"""
+
+    task_id: str
+    title: str
+    description: str
+    must_haves: str
+    project_id: str
+    agent_id: str
+
+    task: Optional[Dict] = None
+    role: str = "executor"
+    spawn_type: str = "executor"
+
+    # mail 专用
+    from_agent: str = ""
+    mail_type: str = ""          # inform / request
+
+    # toolchain 专用
+    event_type: str = ""         # ci_failure / review_request / ...
+    event_data: Dict = field(default_factory=dict)
+
+    # 前序产出
+    depends_on_outputs: Optional[List] = None
+
+
+# ---------------------------------------------------------------------------
+# PromptComposer 拼装器
+# ---------------------------------------------------------------------------
+class PromptComposer:
+    """有序拼装 prompt sections"""
+
+    SEPARATOR = "\n\n---\n\n"
+    TOKEN_BUDGET_WARN = 800      # token 预算警告阈值
+    CHARS_PER_TOKEN = 3.5        # 估算比率
+
+    def __init__(self) -> None:
+        self._sections: List[PromptSection] = []
+
+    def add(self, section: PromptSection) -> None:
+        """添加一个 section（同名覆盖）"""
+        self._sections = [s for s in self._sections if s.name != section.name]
+        self._sections.append(section)
+
+    def add_many(self, sections: List[PromptSection]) -> None:
+        """批量添加"""
+        for s in sections:
+            self.add(s)
+
+    def compose(self, context: PromptContext) -> str:
+        """拼装最终 prompt
+
+        1. 过滤 should_include=False 的段
+        2. 按 priority 排序
+        3. 逐段 render
+        4. 过滤空段
+        5. 用分隔符连接
+        6. Token 预算警告（不截断）
+        """
+        active = [s for s in self._sections if s.should_include(context)]
+        active.sort(key=lambda s: s.priority)
+
+        parts = [s.render(context) for s in active]
+        parts = [p for p in parts if p.strip()]
+
+        result = self.SEPARATOR.join(parts)
+
+        # Token 估算
+        tokens = max(1, int(len(result) / self.CHARS_PER_TOKEN))
+        logger.debug(
+            "Composed prompt from %d sections, %d tokens",
+            len(parts), tokens,
+        )
+
+        if tokens > self.TOKEN_BUDGET_WARN:
+            logger.warning(
+                "Prompt exceeds %d token budget: %d tokens (task_id=%s)",
+                self.TOKEN_BUDGET_WARN, tokens, context.task_id,
+            )
+
+        return result

src/daemon/review.py

@@ -8,15 +8,12 @@ from __future__ import annotations
 
 import json
 import logging
-import re
-from datetime import datetime
 from enum import Enum
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Tuple
+from typing import Any, Callable, Dict, List, Optional
 
 from src.blackboard.models import Task
 from src.blackboard.operations import Blackboard
-from src.blackboard.queries import Queries
 
 logger = logging.getLogger("moziplus-v2.review")
 
@@ -151,12 +148,14 @@ class ReviewPipeline:
     ) -> ReviewResult:
         """Step 2: 格式合规"""
         if not outputs:
-            return ReviewResult("format", ReviewVerdict.FAIL, 0.0, "No outputs")
+            return ReviewResult(
+                "format", ReviewVerdict.FAIL, 0.0, "No outputs")
 
         issues = []
         for out in outputs:
             # output.md 必须存在且非空
-            if out.get("type") == "markdown" or out.get("path", "").endswith(".md"):
+            if out.get("type") == "markdown" or out.get(
+                    "path", "").endswith(".md"):
                 content = out.get("content", "")
                 if not content and out.get("path"):
                     try:
@@ -167,7 +166,8 @@ class ReviewPipeline:
                     issues.append(f"Output too short: {out.get('path', '?')}")
 
             # 结论 JSON 必须有效
-            if out.get("type") == "json" or out.get("path", "").endswith(".json"):
+            if out.get("type") == "json" or out.get(
+                    "path", "").endswith(".json"):
                 content = out.get("content", "")
                 if not content and out.get("path"):
                     try:
@@ -177,7 +177,8 @@ class ReviewPipeline:
                 try:
                     data = json.loads(content)
                     if not isinstance(data, dict):
-                        issues.append(f"JSON not a dict: {out.get('path', '?')}")
+                        issues.append(
+                            f"JSON not a dict: {out.get('path', '?')}")
                 except (json.JSONDecodeError, TypeError):
                     issues.append(f"Invalid JSON: {out.get('path', '?')}")
 
@@ -194,7 +195,8 @@ class ReviewPipeline:
     ) -> ReviewResult:
         """Step 3: 内容质量（自定义检查）"""
         if not outputs:
-            return ReviewResult("quality", ReviewVerdict.FAIL, 0.0, "No outputs")
+            return ReviewResult(
+                "quality", ReviewVerdict.FAIL, 0.0, "No outputs")
 
         suggestions = []
         total_score = 0.0
@@ -215,7 +217,8 @@ class ReviewPipeline:
             avg = 1.0  # 无自定义检查默认通过
 
         verdict = ReviewVerdict.PASS if avg >= 0.6 else ReviewVerdict.FAIL
-        return ReviewResult("quality", verdict, round(avg, 2), suggestions=suggestions)
+        return ReviewResult("quality", verdict, round(
+            avg, 2), suggestions=suggestions)
 
     def _determine_gate(
         self, task: Task, results: List[ReviewResult]
@@ -329,6 +332,7 @@ class RebuttalManager:
             return 0
         try:
             observations = self.bb.get_observations(task_id=task_id)
-            return sum(1 for o in observations if "Rebuttal round" in (o.body or ""))
+            return sum(
+                1 for o in observations if "Rebuttal round" in (o.body or ""))
         except Exception:
             return 0

src/daemon/router.py

@@ -107,7 +107,8 @@ class AgentRouter:
 
         # ── 快速路径 2: retry → 原执行者 ──
         if action_type == "retry":
-            current = task_info.get("current_agent") or task_info.get("assignee")
+            current = task_info.get(
+                "current_agent") or task_info.get("assignee")
             if current and current in self.agent_profiles:
                 return RouteDecision(
                     agent_id=current,
@@ -119,7 +120,8 @@ class AgentRouter:
         # ── Mode B: Agent 声明式交接 ──
         next_cap = task_info.get("next_capability")
         if next_cap and self._validate_capability(next_cap):
-            current = task_info.get("current_agent") or task_info.get("assignee")
+            current = task_info.get(
+                "current_agent") or task_info.get("assignee")
             exclude = {current} if current else set()
             matched = self._match_capability(next_cap, exclude)
             if matched:
@@ -129,7 +131,9 @@ class AgentRouter:
                     mode="agent_handoff",
                     latency_ms=int((time.monotonic() - start) * 1000),
                 )
-            logger.info("next_capability '%s' no match, delegate to coordinator", next_cap)
+            logger.info(
+                "next_capability '%s' no match, delegate to coordinator",
+                next_cap)
 
         # ── 快速路径 3: 生命周期流转查表 ──
         lifecycle = self.LIFECYCLE_CAPABILITY.get(action_type)
@@ -140,7 +144,8 @@ class AgentRouter:
             exclude_current = lifecycle.get("exclude_current", False)
             exclude = set()
             if exclude_current:
-                current = task_info.get("current_agent") or task_info.get("assignee")
+                current = task_info.get(
+                    "current_agent") or task_info.get("assignee")
                 if current:
                     exclude.add(current)
             matched = self._match_capability(cap, exclude)
@@ -154,7 +159,8 @@ class AgentRouter:
 
         # ── 快速路径 4: 有 assignee 且非生命周期流转 ──
         assignee = task_info.get("assignee")
-        if assignee and assignee in self.agent_profiles and action_type not in ("review", "escalation"):
+        if assignee and assignee in self.agent_profiles and action_type not in (
+                "review", "escalation"):
             return RouteDecision(
                 agent_id=assignee,
                 reason=f"Direct assignee: {assignee}",

src/daemon/skill_system.py

@@ -10,12 +10,11 @@ from __future__ import annotations
 
 import json
 import logging
-import re
 from dataclasses import dataclass, field
 from datetime import datetime
 from enum import Enum
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple
 
 logger = logging.getLogger("moziplus-v2.skill")
 

src/daemon/spawner.py

@@ -15,10 +15,12 @@ from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
-from src.blackboard.db import get_connection, init_db
+from src.blackboard.db import get_connection
 
 logger = logging.getLogger("moziplus-v2.spawner")
 
+from src.daemon.task_type_registry import TaskTypeRegistry
+
 
 # ── Prompt 模板 ──
 
@@ -163,9 +165,12 @@ class AgentBusyError(Exception):
 
     #07: reason 字段区分具体原因,便于 dispatcher 层区分处理。
     """
-    def __init__(self, agent_id: str, reason: str = "busy", detail: Optional[dict] = None):
+
+    def __init__(self, agent_id: str, reason: str = "busy",
+                 detail: Optional[dict] = None):
         self.agent_id = agent_id
-        self.reason = reason   # counter_blocked / session_locked / session_running / session_compacting / session_stuck
+        # counter_blocked / session_locked / session_running / session_compacting / session_stuck
+        self.reason = reason
         self.detail = detail or {}
         super().__init__(f"{agent_id}: {reason}")
 
@@ -275,13 +280,36 @@ class AgentSpawner:
                 task_id, title, description, must_haves,
                 project_id, agent_id)
 
-        # mail 任务用精简模板
-        if project_id == "_mail":
-            return self._build_mail_prompt(task_id, title, description, must_haves, agent_id)
+        # handler 路径：Task/Mail/Toolchain 用各自的 PromptSection 构建
+        handler = TaskTypeRegistry.get_by_project(project_id)
+        if handler:
+            from src.daemon.prompt_composer import PromptContext
+            # 从 must_haves 解析 mail 元数据（from / performative）
+            from_agent = ""
+            mail_type = ""
+            try:
+                meta = json.loads(must_haves) if must_haves else {}
+                from_agent = meta.get("from", "")
+                mail_type = meta.get("performative", meta.get("type", ""))
+            except Exception:
+                pass
+            ctx = PromptContext(
+                task_id=task_id, title=title, description=description or "",
+                must_haves=must_haves or "", project_id=project_id,
+                agent_id=agent_id, role=spawn_type,
+                spawn_type=spawn_type,
+                from_agent=from_agent, mail_type=mail_type,
+            )
+            return handler.build_prompt(ctx)
+
+        # 旧路径保留：_general 等非 handler 项目
 
         # 走 BootstrapBuilder 新路径
         if self.bootstrap_builder and task is not None:
-            role_map = {"executor": "executor", "review": "reviewer", "discussion": "planner"}
+            role_map = {
+                "executor": "executor",
+                "review": "reviewer",
+                "discussion": "planner"}
             role = role_map.get(spawn_type, "executor")
             bootstrap_prompt = self.bootstrap_builder.build_for_task(
                 task=task,
@@ -293,13 +321,14 @@ class AgentSpawner:
 
         # 无 BootstrapBuilder 或无 task 对象 → 最小 fallback
         # 只保留任务上下文 + API 操作指令
-        logger.warning("No BootstrapBuilder or task object, using minimal fallback")
+        logger.warning(
+            "No BootstrapBuilder or task object, using minimal fallback")
         return self._build_minimal_fallback(
             task_id, title, description, must_haves,
             project_id, agent_id)
 
     def _build_minimal_fallback(self, task_id, title, description, must_haves,
-                             project_id, agent_id):
+                                project_id, agent_id):
         """最小 fallback:只有任务上下文 + API 指令"""
         task_section = f"""## 任务
 {title}
@@ -311,10 +340,15 @@ class AgentSpawner:
         return task_section + "\n\n---\n\n" + api_section
 
     def _build_api_section(self, project_id: str, task_id: str,
-                            agent_id: str) -> str:
+                           agent_id: str) -> str:
         """构建 API 回写操作指令(BootstrapBuilder 模式下补充)"""
-        # mail 任务直接 done,不走 review
-        success_status = '"done"' if project_id == "_mail" else '"review"'
+        # handler 项目（_mail/_toolchain）的 success_status 由 PromptSection 处理
+        # 这里只处理无 handler 的项目（normal task）
+        handler = TaskTypeRegistry.get_by_project(project_id)
+        if handler:
+            success_status = '"done"' if handler.target_success_status == "done" else '"review"'
+        else:
+            success_status = '"review"'
         return f"""## 操作指令
 
 ### 状态回写
@@ -337,8 +371,8 @@ curl -X POST http://{self.api_host}:{self.api_port}/api/projects/{project_id}/ta
 """
 
     def _build_discussion_prompt(self, task_id: str, title: str,
-                                  description: str, must_haves: str,
-                                  project_id: str, agent_id: str) -> str:
+                                 description: str, must_haves: str,
+                                 project_id: str, agent_id: str) -> str:
         """构建讨论类 spawn prompt(§3.3 框架 + Boids)"""
         goal_snapshot = description or title
         constraints = must_haves or "(无特殊约束)"
@@ -368,7 +402,8 @@ curl -X POST http://{self.api_host}:{self.api_port}/api/projects/{project_id}/ta
         if not self.guardrails:
             return "无特殊限制"
         try:
-            return "、".join(r.get("name", r.get("rule_id", "")) for r in self.guardrails.rules[:6])
+            return "、".join(r.get("name", r.get("rule_id", ""))
+                            for r in self.guardrails.rules[:6])
         except Exception:
             return "无特殊限制"
 
@@ -379,9 +414,8 @@ curl -X POST http://{self.api_host}:{self.api_port}/api/projects/{project_id}/ta
             return router.agent_profiles.get(agent_id)
         return None
 
-
     def _build_mail_prompt(self, task_id: str, title: str, description: str,
-                            must_haves: str, agent_id: str) -> str:
+                           must_haves: str, agent_id: str) -> str:
         """构建 Mail 专用精简模板"""
         # 解析 must_haves 获取 from 和 performative
         from_agent = agent_id
@@ -389,7 +423,9 @@ curl -X POST http://{self.api_host}:{self.api_port}/api/projects/{project_id}/ta
         try:
             meta = json.loads(must_haves) if must_haves else {}
             from_agent = meta.get("from", agent_id)
-            performative = meta.get("performative", meta.get("type", "request"))
+            performative = meta.get(
+                "performative", meta.get(
+                    "type", "request"))
         except Exception:
             pass
 
@@ -472,7 +508,9 @@ curl -X POST http://{self.api_host}:{self.api_port}/api/projects/{project_id}/ta
                 self._revive_session(agent_id)
             elif pre_state.get("status") == "running" and not pre_state.get("lock_pid_alive"):
                 # status=running 但 lock PID 已死 → 假死,revive
-                logger.warning("Phase 0: %s status=running but lock PID dead, reviving", agent_id)
+                logger.warning(
+                    "Phase 0: %s status=running but lock PID dead, reviving",
+                    agent_id)
                 self._revive_session(agent_id)
 
         # Phase 1: Counter acquire(互斥锁)
@@ -487,12 +525,15 @@ curl -X POST http://{self.api_host}:{self.api_port}/api/projects/{project_id}/ta
         if use_main_session:
             session_state = self._check_session_state(agent_id)
             logger.info("Phase 2 session check for %s: status=%s lock_pid=%s lock_pid_alive=%s compact=%s",
-                        agent_id, session_state.get('status'), session_state.get('lock_pid'),
+                        agent_id, session_state.get(
+                            'status'), session_state.get('lock_pid'),
                         session_state.get('lock_pid_alive'), session_state.get('recent_compact'))
 
             blockers = []
-            if session_state.get("lock_pid_alive") and not session_state.get("lock_expired"):
-                blockers.append(("session_locked", session_state.get("lock_pid")))
+            if session_state.get(
+                    "lock_pid_alive") and not session_state.get("lock_expired"):
+                blockers.append(
+                    ("session_locked", session_state.get("lock_pid")))
             if session_state.get("status") == "running":
                 if session_state.get("lock_pid_alive"):
                     # 真 running:外部进程占用
@@ -515,7 +556,8 @@ curl -X POST http://{self.api_host}:{self.api_port}/api/projects/{project_id}/ta
 
             # Phase 2.5: 假死修复(status=running + lock PID 死 → revive → 重检)
             # 此场景应被 Phase 0 提前修复,这里做兜底
-            if session_state.get("status") == "running" and not session_state.get("lock_pid_alive"):
+            if session_state.get("status") == "running" and not session_state.get(
+                    "lock_pid_alive"):
                 logger.warning("Phase 2.5: %s status=running + lock dead (should be caught in Phase 0), reviving",
                                agent_id)
                 self._revive_session(agent_id)
@@ -538,7 +580,10 @@ curl -X POST http://{self.api_host}:{self.api_port}/api/projects/{project_id}/ta
                 raise
 
         if self.dry_run:
-            logger.info("[DRY RUN] Would spawn agent %s (session=%s)", agent_id, _sid_key)
+            logger.info(
+                "[DRY RUN] Would spawn agent %s (session=%s)",
+                agent_id,
+                _sid_key)
             self._register_session(_sid_key, agent_id, task_id, pid=None)
             return _sid_key
 
@@ -554,7 +599,8 @@ curl -X POST http://{self.api_host}:{self.api_port}/api/projects/{project_id}/ta
                         if asyncio.iscoroutine(result):
                             await result
                     except Exception:
-                        logger.warning("Business on_complete failed for %s", aid, exc_info=True)
+                        logger.warning(
+                            "Business on_complete failed for %s", aid, exc_info=True)
 
         cmd = [
             "openclaw", "agent",
@@ -575,7 +621,7 @@ curl -X POST http://{self.api_host}:{self.api_port}/api/projects/{project_id}/ta
                 stderr=asyncio.subprocess.PIPE,
             )
             self._register_session(session_id, agent_id, task_id, proc.pid,
-                                     broadcast_task_ids=broadcast_task_ids)
+                                   broadcast_task_ids=broadcast_task_ids)
             logger.info("Spawned agent %s (session=%s, pid=%d)",
                         agent_id, session_id, proc.pid)
 
@@ -593,7 +639,11 @@ curl -X POST http://{self.api_host}:{self.api_port}/api/projects/{project_id}/ta
             if self.counter:
                 self.counter.release(agent_id, _sid_key)
             logger.exception("Failed to spawn agent %s", agent_id)
-            self._record_attempt(task_id, agent_id, "spawn_failed", error=str(e))
+            self._record_attempt(
+                task_id,
+                agent_id,
+                "spawn_failed",
+                error=str(e))
             raise
 
     async def spawn_subagent(
@@ -609,7 +659,9 @@ curl -X POST http://{self.api_host}:{self.api_port}/api/projects/{project_id}/ta
         session_id = str(uuid.uuid4())
 
         if self.dry_run:
-            logger.info("[DRY RUN] Would spawn subagent (session=%s)", session_id)
+            logger.info(
+                "[DRY RUN] Would spawn subagent (session=%s)",
+                session_id)
             self._register_session(session_id, "subagent", task_id, pid=None)
             return session_id
 
@@ -729,10 +781,16 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                     agent_id, session_id, json_result)
 
         # 查任务实际状态
-        task_status = self._get_task_status(db_path, task_id) if task_id else None
+        task_status = self._get_task_status(
+            db_path, task_id) if task_id else None
 
         # 分类
-        cls = self._classify_outcome(exit_code, json_result, stderr_text, task_status, stdout_text)
+        cls = self._classify_outcome(
+            exit_code,
+            json_result,
+            stderr_text,
+            task_status,
+            stdout_text)
         outcome = cls["outcome"]
 
         # 更新 session 状态
@@ -761,17 +819,21 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                     agent_id, session_id, outcome, exit_code, task_status)
 
         # 广播反馈追踪(Phase 1 bug fix)
-        if task_id == "broadcast" and hasattr(self, '_ticker') and self._ticker:
+        if task_id == "broadcast" and hasattr(
+                self, '_ticker') and self._ticker:
             # 广播任务:从 session 信息取真实 task_id 列表,逐一回调 tracker
             sess_info = self._sessions.get(session_id or "main", {})
             bt_ids = sess_info.get("broadcast_task_ids") or []
             # 广播场景一律标 no_reply:Agent 只 claim 一个任务,
             # 其余任务的 tracker 不能被 claimed 清除
             for real_task_id in bt_ids:
-                self._ticker.record_broadcast_response(real_task_id, agent_id, "no_reply")
+                self._ticker.record_broadcast_response(
+                    real_task_id, agent_id, "no_reply")
         elif task_id and hasattr(self, '_ticker') and self._ticker:
-            outcome_str = "claimed" if cls.get("status") == "ok" else "no_reply"
-            self._ticker.record_broadcast_response(task_id, agent_id, outcome_str)
+            outcome_str = "claimed" if cls.get(
+                "status") == "ok" else "no_reply"
+            self._ticker.record_broadcast_response(
+                task_id, agent_id, outcome_str)
 
         if cls["should_retry"]:
             # cooldown: 新增的可恢复场景(A14/A15/A16/A8/A10)
@@ -848,13 +910,26 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
             #      A8(gateway_unreachable), A11(lock_conflict),
             #      A10(compact_failed), A12(agent_error)
             # v2.8.1 Fix-3a: crash 类 outcome 设 cooldown,给 agent session 恢复时间
-            if outcome in ("crashed", "compact_failed", "process_crash", "session_stuck",
-                           "compact_hanging", "agent_error", "compact_interrupted") and self.counter:
+            if outcome == "crashed" and self.counter:
+                self.counter.set_cooldown(agent_id, seconds=60)
+                logger.info(
+                    "Crash cooldown set for %s: 60s (outcome=%s)",
+                    agent_id,
+                    outcome)
+            elif outcome in ("compact_failed", "process_crash", "session_stuck",
+                             "compact_hanging", "agent_error", "compact_interrupted") and self.counter:
                 self.counter.set_cooldown(agent_id, seconds=300)  # 5 分钟
-                logger.info("Crash/error cooldown set for %s: 300s (outcome=%s)", agent_id, outcome)
+                logger.info(
+                    "Error cooldown set for %s: 300s (outcome=%s)",
+                    agent_id,
+                    outcome)
             # F1: 不可恢复 outcome → 立刻标 failed + 写黑板
-            if outcome in ("auth_failed", "agent_error") and db_path and task_id:
-                logger.error("Task %s: unrecoverable outcome=%s, marking failed immediately", task_id, outcome)
+            if outcome in ("auth_failed",
+                           "agent_error") and db_path and task_id:
+                logger.error(
+                    "Task %s: unrecoverable outcome=%s, marking failed immediately",
+                    task_id,
+                    outcome)
                 self._mark_task(db_path, task_id, "failed", {
                     "reason": outcome,
                     "stderr_preview": (stderr_text or "")[:500],
@@ -878,13 +953,16 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
         except Exception:
             pass
 
-        stderr_text = b"".join(stderr_chunks).decode("utf-8", errors="replace")
+        # stderr collected but not used in this handler
+        # (kept for potential future diagnostics)
+        b"".join(stderr_chunks).decode("utf-8", errors="replace")
 
         # 检查 session 状态
         state = self._check_session_state(agent_id)
 
         # B1: 假死 - 先复活,连续假死 ≥2 次再 failed
-        if state.get("status") == "running" and not state.get("lock_pid_alive", True):
+        if state.get("status") == "running" and not state.get(
+                "lock_pid_alive", True):
             # 假死计数
             stuck_count = self._stuck_counts.get(task_id, 0) + 1
             self._stuck_counts[task_id] = stuck_count
@@ -910,7 +988,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                 await self._do_on_complete_async(on_complete, agent_id, "session_revived")
             else:
                 # 复活失败 → 标 failed
-                logger.error("Agent %s revive failed, marking failed", agent_id)
+                logger.error(
+                    "Agent %s revive failed, marking failed", agent_id)
                 self._mark_task(db_path, task_id, "failed",
                                 {"reason": "revive_failed", "stuck_count": stuck_count,
                                  "diagnostics": state})
@@ -991,7 +1070,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                         "SELECT status FROM tasks WHERE id=?", (task_id,)
                     ).fetchone()
                     # Bug-6 fix: pending 不是终态
-                    if row and row["status"] in ("done", "failed", "cancelled", "review"):
+                    if row and row["status"] in (
+                            "done", "failed", "cancelled", "review"):
                         logger.info("Retry skip: task %s already %s (agent=%s)",
                                     task_id, row["status"], agent_id)
                         # on_complete = wrapped_on_complete,会 release counter
@@ -1000,7 +1080,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                 finally:
                     conn.close()
             except Exception:
-                logger.warning("Retry status check failed for %s, proceeding", task_id)
+                logger.warning(
+                    "Retry status check failed for %s, proceeding", task_id)
 
         # 直接读写 tasks 表的 retry_count
         if retry_field == "retry_count" and db_path and task_id:
@@ -1020,7 +1101,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                 finally:
                     conn.close()
             except Exception:
-                logger.exception("Failed to update retry_count for task %s", task_id)
+                logger.exception(
+                    "Failed to update retry_count for task %s", task_id)
                 count = 1
         else:
             retry_counts = self._get_retry_counts(db_path, task_id)
@@ -1104,7 +1186,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
         """
         text = stdout_text.strip()
         if not text:
-            return {"status": None, "summary": None, "fallback_used": False, "fallback_reason": None, "payloads": []}
+            return {"status": None, "summary": None, "fallback_used": False,
+                    "fallback_reason": None, "payloads": []}
         try:
             data = json.loads(text)
         except json.JSONDecodeError:
@@ -1116,7 +1199,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                 except json.JSONDecodeError:
                     continue
             else:
-                return {"status": None, "summary": None, "fallback_used": False, "fallback_reason": None, "payloads": []}
+                return {"status": None, "summary": None, "fallback_used": False,
+                        "fallback_reason": None, "payloads": []}
 
         # 从 data.result.meta.executionTrace 取 fallback 信息
         result = data.get("result", {})
@@ -1132,7 +1216,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
         }
 
     @staticmethod
-    def _get_task_status(db_path: Optional[Path], task_id: Optional[str]) -> Optional[str]:
+    def _get_task_status(
+            db_path: Optional[Path], task_id: Optional[str]) -> Optional[str]:
         """查任务实际 API 状态"""
         if not db_path or not task_id:
             return None
@@ -1149,7 +1234,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
             return None
 
     @staticmethod
-    def _get_task_info(db_path: Optional[Path], task_id: Optional[str]) -> Optional[dict]:
+    def _get_task_info(db_path: Optional[Path],
+                       task_id: Optional[str]) -> Optional[dict]:
         """查任务基本信息"""
         if not db_path or not task_id:
             return None
@@ -1157,7 +1243,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
             conn = get_connection(db_path)
             try:
                 row = conn.execute(
-                    "SELECT id, title, status FROM tasks WHERE id=?", (task_id,)
+                    "SELECT id, title, status FROM tasks WHERE id=?", (
+                        task_id,)
                 ).fetchone()
                 if not row:
                     return None
@@ -1189,7 +1276,9 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
             sessions[main_key] = main_session
             with open(sessions_path, "w") as f:
                 json.dump(sessions, f, indent=2)
-            logger.info("Revived %s: sessions.json status changed running→idle", agent_id)
+            logger.info(
+                "Revived %s: sessions.json status changed running→idle",
+                agent_id)
             # #07 O4: 同时清理残留 lock 文件
             sf = main_session.get("sessionFile", "")
             if sf:
@@ -1197,7 +1286,10 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                 if lock_path.exists():
                     try:
                         lock_path.unlink()
-                        logger.info("Cleaned stale lock for %s: %s", agent_id, lock_path.name)
+                        logger.info(
+                            "Cleaned stale lock for %s: %s",
+                            agent_id,
+                            lock_path.name)
                     except Exception:
                         pass
             return True
@@ -1206,7 +1298,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
             return False
 
     @staticmethod
-    def _check_recent_compaction_jsonl(session_file: str, window_seconds: int = 900) -> bool:
+    def _check_recent_compaction_jsonl(
+            session_file: str, window_seconds: int = 900) -> bool:
         """v2.8.2 Fix-2: 读 session jsonl 末尾,检查是否有 window_seconds 内的 compaction 记录。
 
         比 compactionCheckpoints 更可靠:Gateway 每次完成 compact 必然在 jsonl 末尾追加记录,
@@ -1216,7 +1309,7 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
         实测 50KB 在长对话中不够（compact 记录被推出窗口导致漏检）。
         正常扫描量不变：从尾部往前扫，遇到超过 15min 的 timestamp 即 break。
         """
-        if not session_file or not pathlib.Path(session_file).exists():
+        if not session_file or not Path(session_file).exists():
             return False
         try:
             from datetime import datetime, timezone
@@ -1238,7 +1331,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                     ts = obj.get("timestamp", "")
                     if ts:
                         try:
-                            ct = datetime.fromisoformat(ts.replace("Z", "+00:00"))
+                            ct = datetime.fromisoformat(
+                                ts.replace("Z", "+00:00"))
                             if (now - ct).total_seconds() < window_seconds:
                                 return True
                         except (ValueError, TypeError):
@@ -1262,7 +1356,11 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
         v2.8.1: compact 检测改用 session jsonl 末尾扫描(Fix-1),
         替代失效的 compactionCheckpoints 检测。
         """
-        result = {"status": "unknown", "lock_pid": None, "lock_pid_alive": False, "recent_compact": False}
+        result = {
+            "status": "unknown",
+            "lock_pid": None,
+            "lock_pid_alive": False,
+            "recent_compact": False}
         sessions_path = Path(os.environ.get(
             "OPENCLAW_HOME", str(Path.home() / ".openclaw")
         )) / "agents" / agent_id / "sessions" / "sessions.json"
@@ -1301,8 +1399,10 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                                 created_at_str = lock_data.get("createdAt", "")
                                 if created_at_str:
                                     from datetime import datetime as _dt, timezone as _tz
-                                    created_dt = _dt.fromisoformat(created_at_str.replace("Z", "+00:00"))
-                                    elapsed = (_dt.now(_tz.utc) - created_dt).total_seconds()
+                                    created_dt = _dt.fromisoformat(
+                                        created_at_str.replace("Z", "+00:00"))
+                                    elapsed = (_dt.now(_tz.utc) -
+                                               created_dt).total_seconds()
                                     if elapsed > 1800:  # 30 minutes
                                         result["lock_pid_alive"] = False
                                         result["lock_expired"] = True
@@ -1315,8 +1415,10 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
 
             # v2.8.1 Fix-1: compact 检测改用 session jsonl 末尾扫描
             # 只在 agent 非空闲时才扫描(减少不必要 I/O)
-            if result["status"] not in ("done", "idle", "unknown", None) and sf:
-                result["recent_compact"] = AgentSpawner._check_recent_compaction_jsonl(sf)
+            if result["status"] not in (
+                    "done", "idle", "unknown", None) and sf:
+                result["recent_compact"] = AgentSpawner._check_recent_compaction_jsonl(
+                    sf)
         except Exception:
             pass
         return result
@@ -1361,45 +1463,53 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
             # A15/A16: stderr 含 network/compact 关键字 → 可恢复
             if stderr_text:
                 stderr_lower = stderr_text.lower()
-                if any(kw in stderr_lower for kw in ["econnrefused", "etimedout", "gateway closed", "econnreset"]):
+                if any(kw in stderr_lower for kw in [
+                       "econnrefused", "etimedout", "gateway closed", "econnreset"]):
                     return {"outcome": "gateway_unreachable", "should_retry": True,
                             "retry_field": "retry_count", "cooldown_seconds": 60}
-                if any(kw in stderr_lower for kw in ["compaction-diag", "context-overflow"]):
+                if any(kw in stderr_lower for kw in [
+                       "compaction-diag", "context-overflow"]):
                     return {"outcome": "compact_interrupted", "should_retry": True,
                             "retry_field": "retry_count", "cooldown_seconds": 60}
             # A17: 真正的 crash → 保持 working,ticker 兜底
-            return {"outcome": "crashed", "should_retry": False, "original": "process_crash"}
+            return {"outcome": "crashed", "should_retry": False,
+                    "original": "process_crash"}
 
-        # stdout 为空但 exit=0:可能是正常完成但 --json 没输出
-        # 查任务状态判断
+        # A13 revised: stdout 为空但 exit=0 → 信任进程退出码，视为正常完成
+        # 实测发现 openclaw session=None + exit=0 是正常场景（inform 通知等）
+        # 旧逻辑按 task_status 区分，非终态判 agent_error → 导致 inform 邮件永不标 done
         if status is None and not stdout_text.strip() and exit_code == 0:
-            terminal_statuses = {"done", "review"}
-            if task_status in terminal_statuses:
-                return {"outcome": "completed", "should_retry": False}
-            return {"outcome": "agent_error", "should_retry": False}
+            return {"outcome": "completed", "should_retry": False}
 
         # A7-A12: status=error → 不续杯,stderr 辅助分类
         if status == "error":
             stderr_lower = stderr_text.lower()
-            if any(kw in stderr_lower for kw in ["401", "403", "unauthorized", "auth"]):
+            if any(kw in stderr_lower for kw in [
+                   "401", "403", "unauthorized", "auth"]):
                 return {"outcome": "auth_failed", "should_retry": False}
-            if any(kw in stderr_lower for kw in ["econnrefused", "etimedout", "gateway closed", "econnreset"]):
+            if any(kw in stderr_lower for kw in [
+                   "econnrefused", "etimedout", "gateway closed", "econnreset"]):
                 return {"outcome": "gateway_unreachable", "should_retry": True,
                         "retry_field": "retry_count", "cooldown_seconds": 60}
-            if any(kw in stderr_lower for kw in ["rate_limit", "500", "503", "api error"]):
+            if any(kw in stderr_lower for kw in [
+                   "rate_limit", "500", "503", "api error"]):
                 return {"outcome": "api_error", "should_retry": False}
-            if any(kw in stderr_lower for kw in ["compaction-diag", "context-overflow"]):
+            if any(kw in stderr_lower for kw in [
+                   "compaction-diag", "context-overflow"]):
                 return {"outcome": "compact_failed", "should_retry": False}
-            if any(kw in stderr_lower for kw in ["lock", "busy", "concurrent", "lane task error"]):
+            if any(kw in stderr_lower for kw in [
+                   "lock", "busy", "concurrent", "lane task error"]):
                 return {"outcome": "lock_conflict", "should_retry": True,
                         "retry_field": "retry_count", "cooldown_seconds": 60}
             return {"outcome": "agent_error", "should_retry": False}
 
         # 兜底:status 未知值
-        return {"outcome": "agent_error", "should_retry": False, "original": "unknown_status"}
+        return {"outcome": "agent_error",
+                "should_retry": False, "original": "unknown_status"}
 
     @staticmethod
-    def _get_retry_counts(db_path: Optional[Path], task_id: Optional[str]) -> dict:
+    def _get_retry_counts(
+            db_path: Optional[Path], task_id: Optional[str]) -> dict:
         """从最新 task_attempt 的 metadata 读计数器"""
         defaults = {"retry_count": 0, "connect_retry_count": 0,
                     "api_retry_count": 0, "lock_retry_count": 0,
@@ -1425,7 +1535,7 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
         return defaults
 
     def _update_retry_counts(self, db_path: Optional[Path],
-                              task_id: Optional[str], counts: dict):
+                             task_id: Optional[str], counts: dict):
         """将 retry counts 写回最新 task_attempt 的 metadata"""
         if not db_path or not task_id:
             return
@@ -1439,7 +1549,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                     (task_id,)
                 ).fetchone()
                 if row:
-                    meta = json.loads(row["metadata"]) if row["metadata"] else {}
+                    meta = json.loads(
+                        row["metadata"]) if row["metadata"] else {}
                     meta.update(counts)
                     conn.execute(
                         "UPDATE task_attempts SET metadata=? WHERE rowid=?",
@@ -1449,7 +1560,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
             finally:
                 conn.close()
         except Exception:
-            logger.exception("Failed to update retry counts for task %s", task_id)
+            logger.exception(
+                "Failed to update retry counts for task %s", task_id)
 
     def _mark_task(self, db_path: Optional[Path], task_id: Optional[str],
                    status: str, detail: Optional[dict] = None):
@@ -1467,7 +1579,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                 if detail:
                     conn.execute(
                         "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
-                        (task_id, "daemon", status, json.dumps(detail, ensure_ascii=False))
+                        (task_id, "daemon", status, json.dumps(
+                            detail, ensure_ascii=False))
                     )
                 conn.commit()
             finally:
@@ -1485,10 +1598,13 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
                         from src.blackboard.operations import Blackboard
                         bb = Blackboard(db_path)
                         cid = bb.add_comment(task_id, "daemon",
-                            f"@pangtong-fujunshi 任务执行失败: {reason}，请评估是否需要介入",
-                            comment_type="system")
+                                             f"@pangtong-fujunshi 任务执行失败: {reason}，请评估是否需要介入",
+                                             comment_type="system")
                         bb.record_mentions(cid, task_id, ["pangtong-fujunshi"])
-                        logger.info("Task %s: failure notified pangtong via comment+mention (reason=%s)", task_id, reason)
+                        logger.info(
+                            "Task %s: failure notified pangtong via comment+mention (reason=%s)",
+                            task_id,
+                            reason)
                 except Exception as e:
                     logger.warning("Task %s: failed to notify: %s", task_id, e)
         except Exception:
@@ -1517,7 +1633,10 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
             if asyncio.iscoroutine(result):
                 await result
         except Exception:
-            logger.warning("on_complete callback failed for %s", agent_id, exc_info=True)
+            logger.warning(
+                "on_complete callback failed for %s",
+                agent_id,
+                exc_info=True)
 
     def _register_session(
         self,
@@ -1595,7 +1714,8 @@ curl -X POST http://{api_host}:{api_port}/api/projects/{project_id}/tasks/{task_
     def get_session_by_agent(self, agent_id: str) -> Optional[Dict[str, Any]]:
         """v2.7.2: 根据 agent_id 获取活跃 session 信息(用于进程存活性检查)"""
         for sid, info in self._sessions.items():
-            if info.get("agent_id") == agent_id and info.get("status") == "running":
+            if info.get("agent_id") == agent_id and info.get(
+                    "status") == "running":
                 return info
         return None
 

src/daemon/sse.py

@@ -9,14 +9,11 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
-import subprocess
 import uuid
 from datetime import datetime
 from enum import Enum
-from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Set
+from typing import Any, Dict, List, Optional
 
-from src.blackboard.models import Event
 
 logger = logging.getLogger("moziplus-v2.sse")
 
@@ -52,7 +49,8 @@ class SSEEvent:
         """格式化为 SSE 协议文本"""
         lines = [f"id: {self.id}"]
         lines.append(f"event: {self.event_type}")
-        lines.append(f"data: {json.dumps(self.data, ensure_ascii=False, default=str)}")
+        lines.append(
+            f"data: {json.dumps(self.data, ensure_ascii=False, default=str)}")
         return "\n".join(lines) + "\n\n"
 
 

src/daemon/task_handler.py

@@ -0,0 +1,378 @@
+"""task_handler.py — 黑板任务 handler（task_type='task'）。
+
+标准黑板任务：三信号验证 → review 状态。
+"""
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from src.daemon.base_task_handler import BaseTaskHandler, VerifyResult
+from src.daemon.prompt_composer import PromptComposer, PromptContext
+from src.blackboard.db import get_connection
+
+logger = logging.getLogger("moziplus-v2.handler")
+
+TERMINAL_STATES = frozenset({"review", "done", "failed", "cancelled"})
+
+# ---------------------------------------------------------------------------
+# Role → Skill 映射（D8 决策：L2 只给索引+引导语，不注全文）
+# ---------------------------------------------------------------------------
+ROLE_SKILL_MAP: Dict[str, str] = {
+    "executor": "blackboard-executor",
+    "reviewer": "blackboard-reviewer",
+    "reviewer-simayi": "blackboard-reviewer-simayi",
+    "reviewer-pangtong": "blackboard-reviewer-pangtong",
+    "planner": "blackboard-planner",
+    "claim": "blackboard-claim",
+}
+
+SKILL_BASE_PATH = "/Users/chufeng/.sanguo_projects/sanguo_mozi/skills"
+
+
+# ---------------------------------------------------------------------------
+# PromptSection 实现
+# ---------------------------------------------------------------------------
+
+class TaskContextSection:
+    """段 1：任务上下文（title / desc / must_haves / status）。"""
+
+    name: str = "task_context"
+    priority: int = 10
+
+    def render(self, context: PromptContext) -> str:
+        parts = ["## 任务上下文"]
+        if context.task_id:
+            parts.append(f"任务ID: {context.task_id}")
+        if context.title:
+            parts.append(f"标题: {context.title}")
+        if context.description:
+            parts.append(f"描述: {context.description}")
+        if context.must_haves:
+            parts.append(f"必须完成: {context.must_haves}")
+        if context.task and context.task.get("status"):
+            parts.append(f"当前状态: {context.task['status']}")
+        return "\n".join(parts)
+
+    def should_include(self, context: PromptContext) -> bool:
+        return bool(context.task_id or context.title)
+
+
+class PriorOutputsSection:
+    """段 2：前序产出摘要（depends_on 非空时注入）。"""
+
+    name: str = "prior_outputs"
+    priority: int = 20
+
+    def render(self, context: PromptContext) -> str:
+        outputs = context.depends_on_outputs or []
+        parts = ["## 前序产出"]
+        for out in outputs:
+            tid = out.get("task_id", "?")
+            summary = out.get("summary", "无摘要")
+            parts.append(f"- [{tid}] {summary}")
+        return "\n".join(parts)
+
+    def should_include(self, context: PromptContext) -> bool:
+        return bool(context.depends_on_outputs)
+
+
+class RoleSkillSection:
+    """段 3：角色 Skill 索引+引导语（D8 决策：不注全文）。"""
+
+    name: str = "role_skill"
+    priority: int = 30
+
+    def render(self, context: PromptContext) -> str:
+        skill_name = ROLE_SKILL_MAP.get(context.role, "")
+        lines = [
+            "## 角色操作规范",
+            f"你的角色：{context.role}",
+        ]
+        if skill_name:
+            lines.append(f"对应 Skill：{skill_name}")
+            lines.append(
+                f"请用 read 工具读取 {SKILL_BASE_PATH}/{skill_name}/SKILL.md "
+                "获取完整操作规范。"
+            )
+        else:
+            lines.append("无对应 Skill 文件，按通用规范执行。")
+        return "\n".join(lines)
+
+    def should_include(self, context: PromptContext) -> bool:
+        return True
+
+
+class TaskApiSection:
+    """段 4：API 操作指令。"""
+
+    name: str = "task_api"
+    priority: int = 40
+
+    API_HOST = "localhost"
+    API_PORT = 8083
+
+    def render(self, context: PromptContext) -> str:
+        pid = context.project_id
+        tid = context.task_id
+        aid = context.agent_id
+        success_status = '"review"'
+        base = f"http://{self.API_HOST}:{self.API_PORT}/api/projects/{pid}/tasks/{tid}"
+        return (
+            "## 操作指令\n"
+            "### 状态回写\n"
+            f"开始工作:\n"
+            f'curl -X POST {base}/status \\\n'
+            f'  -H "Content-Type: application/json" \\\n'
+            f'  -d \'{{"status": "working", "agent": "{aid}"}}\'\n\n'
+            "### 写入产出\n"
+            f'curl -X POST {base}/outputs \\\n'
+            f'  -H "Content-Type: application/json" \\\n'
+            f"  -d '{{\"type\": \"text\", \"content\": \"<your output>\"}}'\n\n"
+            "### 完成后\n"
+            f"成功: status → {success_status} | 失败: status → \"failed\""
+        )
+
+    def should_include(self, context: PromptContext) -> bool:
+        return True
+
+
+class TaskConstraintsSection:
+    """段 5：硬约束。"""
+
+    name: str = "task_constraints"
+    priority: int = 50
+
+    def render(self, context: PromptContext) -> str:
+        constraints = ["## 硬约束"]
+        role = context.role
+        if role == "executor":
+            constraints.extend([
+                "- 完成后必须标 review",
+                "- 产出物不能为空（系统会验证）",
+                "- handoff comment ≥ 50 字符",
+            ])
+        elif role.startswith("reviewer"):
+            constraints.extend([
+                "- 审查结果必须明确 pass/fail",
+                "- 评审意见须附证据（文件:行号）",
+            ])
+        elif role == "planner":
+            constraints.extend([
+                "- 需求不清时提问，不要猜",
+                "- 子任务必须有明确的终态定义",
+            ])
+        else:
+            constraints.append("- 按规范完成 assigned 任务")
+        return "\n".join(constraints)
+
+    def should_include(self, context: PromptContext) -> bool:
+        return True
+
+
+class TaskHandler(BaseTaskHandler):
+    """黑板标准任务 handler。
+    
+    - verify: 三信号检查（output / comment / terminal status）
+    - 成功 → review
+    - 失败 → 保持 working，让 ticker 重试
+    - review 完成 → 读取 verdict，approved 则 mark done
+    """
+
+    task_type: str = "task"
+    virtual_project: Optional[str] = None
+    display_name = "黑板任务"
+
+    # === 子类实现 ===
+
+    def post_complete(self, task_id: str, agent_id: str,
+                      outcome: str, db_path: Path) -> None:
+        """Task on_complete：区分 executor 和 review。
+        
+        executor: 基类统一流程（crash → verify → mark review）
+        review: handle_review_complete（读 verdict → done/keep review）
+        """
+        # crash 处理（所有类型共用）
+        if outcome in self.CRASH_OUTCOMES:
+            self._rollback_current_agent(db_path, task_id, agent_id)
+            return
+
+        # 检查当前任务状态：如果是 review 状态 → review 完成流程
+        try:
+            conn = get_connection(db_path)
+            try:
+                row = conn.execute(
+                    "SELECT status FROM tasks WHERE id=?", (task_id,)
+                ).fetchone()
+                task_status = row["status"] if row else "unknown"
+            finally:
+                conn.close()
+        except Exception:
+            task_status = "unknown"
+
+        if task_status == "review":
+            # review 完成流程：只处理正常 outcome
+            if outcome in ("completed", "session_revived"):
+                self.handle_review_complete(task_id, db_path)
+            else:
+                logger.warning(
+                    "Task %s: review agent %s abnormal outcome=%s, keeping review",
+                    task_id, agent_id, outcome)
+        else:
+            # executor 完成流程：基类统一 verify → mark
+            result = self.verify_completion(task_id, db_path)
+            if result.passed:
+                self._mark_task_status(db_path, task_id, self.target_success_status())
+                logger.info("Task %s: verify passed (%s), marked %s",
+                            task_id, result.reason, self.target_success_status())
+            else:
+                logger.info(
+                    "Task %s: verify not passed (%s), leaving working",
+                    task_id, result.reason)
+                # NOTE: executor verify 不通过时不标 failed，留 working。
+                # 原因：Agent 可能还在产出中（幻觉门控的后续轮次），
+                # ticker 超时检查会兜底处理。不调 on_failure 避免误判。
+
+    def target_success_status(self) -> str:
+        """task 类型验证通过后进 review。"""
+        return "review"
+
+    def verify_completion(self, task_id: str, db_path: Path) -> VerifyResult:
+        """三信号验证：output / comment / terminal status。"""
+        try:
+            conn = get_connection(db_path)
+            try:
+                # 信号 1：terminal status
+                row = conn.execute(
+                    "SELECT status FROM tasks WHERE id=?", (task_id,)
+                ).fetchone()
+                if not row:
+                    return VerifyResult(False, "not_found", "task not found",
+                                        can_retry=False)
+                status = row["status"]
+                if status in TERMINAL_STATES:
+                    return VerifyResult(
+                        True, "terminal_status",
+                        f"status={status}", can_retry=False
+                    )
+
+                # 信号 2：outputs
+                output_count = conn.execute(
+                    "SELECT COUNT(*) as cnt FROM outputs WHERE task_id=?",
+                    (task_id,)
+                ).fetchone()["cnt"]
+                if output_count > 0:
+                    return VerifyResult(
+                        True, "has_output",
+                        f"output_count={output_count}"
+                    )
+
+                # 信号 3：非 system 且内容 >= 50 字的 comment
+                comment_count = conn.execute(
+                    "SELECT COUNT(*) as cnt FROM comments "
+                    "WHERE task_id=? AND author != 'system' "
+                    "AND LENGTH(content) >= 50",
+                    (task_id,)
+                ).fetchone()["cnt"]
+                if comment_count > 0:
+                    return VerifyResult(
+                        True, "has_comment",
+                        f"comment_count={comment_count}"
+                    )
+
+                # 无信号
+                return VerifyResult(
+                    False, "no_signal",
+                    f"output=0, comment=0, status={status}"
+                )
+            finally:
+                conn.close()
+        except Exception as e:
+            logger.error("Task %s: verify error: %s", task_id, e)
+            return VerifyResult(False, "verify_error", str(e))
+
+    def pre_spawn(self, task_id: str, db_path: Path) -> bool:
+        """task 类型不需要 pre_spawn 逻辑。"""
+        return True
+
+    def get_sections(self) -> list:
+        """返回 5 个 PromptSection 实例。"""
+        return [
+            TaskContextSection(),
+            PriorOutputsSection(),
+            RoleSkillSection(),
+            TaskApiSection(),
+            TaskConstraintsSection(),
+        ]
+
+    def build_prompt(self, context: PromptContext) -> str:
+        """通过 PromptComposer 拼装 prompt sections。"""
+        composer = PromptComposer()
+        composer.add_many(self.get_sections())
+        return composer.compose(context)
+
+    def on_failure(self, task_id: str, agent_id: str,
+                   db_path: Path, verify: VerifyResult) -> None:
+        """验证失败：不标 failed，保持 working 让 ticker 重试。"""
+        logger.info(
+            "Task %s: verify failed (%s, evidence=%s), leaving working for ticker retry",
+            task_id, verify.reason, verify.evidence
+        )
+
+    # === Review 流程 ===
+
+    def handle_review_complete(self, task_id: str, db_path: Path) -> None:
+        """Review 完成后处理：读取 verdict → approved 则 mark done，
+        否则 @mention assignee via blackboard comment。"""
+        try:
+            conn = get_connection(db_path)
+            try:
+                # 读取最新 review
+                review_row = conn.execute(
+                    "SELECT verdict, reviewer, comment FROM reviews "
+                    "WHERE task_id=? ORDER BY created_at DESC LIMIT 1",
+                    (task_id,)
+                ).fetchone()
+
+                if not review_row:
+                    logger.warning("Task %s: no review found", task_id)
+                    return
+
+                verdict = review_row["verdict"]
+                reviewer = review_row["reviewer"]
+                review_comment = review_row["comment"] or ""
+
+                # 获取 assignee
+                task_row = conn.execute(
+                    "SELECT assignee FROM tasks WHERE id=?", (task_id,)
+                ).fetchone()
+                if not task_row:
+                    logger.warning("Task %s: task not found for review", task_id)
+                    return
+                assignee = task_row["assignee"]
+
+                if verdict == "approved":
+                    self._mark_task_status(db_path, task_id, "done")
+                    logger.info("Task %s: review approved by %s, marked done",
+                                task_id, reviewer)
+                else:
+                    # 非 approved：通过 blackboard comment @mention assignee
+                    # 保持 review 状态，让 assignee 自行决定下一步
+                    conn.execute(
+                        "INSERT INTO comments (task_id, author, content, comment_type) "
+                        "VALUES (?, 'system', ?, 'review')",
+                        (task_id,
+                         f"@{assignee} review 未通过 (verdict={verdict}, "
+                         f"reviewer={reviewer}): {review_comment}")
+                    )
+                    conn.commit()
+                    logger.info(
+                        "Task %s: review not approved (%s by %s), "
+                        "@mentioned assignee %s, keeping review status",
+                        task_id, verdict, reviewer, assignee
+                    )
+            finally:
+                conn.close()
+        except Exception as e:
+            logger.error("Task %s: handle_review_complete error: %s", task_id, e)

src/daemon/task_type_registry.py

@@ -0,0 +1,102 @@
+"""
+task_type_registry.py — Task type handler Protocol + Registry.
+
+启动时一次性加载 handler，运行时只读。
+零依赖：不导入项目内其他模块。
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from src.daemon.prompt_composer import PromptContext
+
+logger = logging.getLogger("moziplus-v2.registry")
+
+
+# ---------------------------------------------------------------------------
+# Protocol
+# ---------------------------------------------------------------------------
+
+@runtime_checkable
+class TaskTypeHandler(Protocol):
+    """所有 task type handler 的统一接口。"""
+
+    # 属性（通过 __init__ 设置）
+    task_type: str  # 类型标识：'task' | 'mail' | 'toolchain'
+    virtual_project: Optional[str]  # 虚拟项目 ID，如 '_mail'、'_toolchain'。普通任务为 None
+
+    def build_prompt(self, context: "PromptContext") -> str:
+        """构建 Agent prompt（通过 PromptComposer 拼 section）。"""
+        ...
+
+    def pre_spawn(self, task_id: str, db_path: Path) -> bool:
+        """spawn 前业务准备。默认 True，mail/toolchain override 为 auto_working。"""
+        ...
+
+    def post_complete(
+        self,
+        task_id: str,
+        agent_id: str,
+        outcome: str,
+        db_path: Path,
+    ) -> None:
+        """spawn 完成后的业务处理。统一流程：crash→verify→mark→notify。"""
+        ...
+
+    def check_completion(self, task_id: str, db_path: Path) -> bool:
+        """ticker 级别的完成检查。"""
+        ...
+
+    def get_sections(self) -> list:
+        """返回此 handler 的 prompt section 列表。"""
+        ...
+
+
+# ---------------------------------------------------------------------------
+# Registry
+# ---------------------------------------------------------------------------
+
+class TaskTypeRegistry:
+    """Task type handler 注册表。启动时一次性加载，运行时只读。"""
+
+    _handlers: Dict[str, TaskTypeHandler] = {}
+
+    @classmethod
+    def register(cls, handler: TaskTypeHandler) -> None:
+        """注册一个 handler。启动时调用一次。"""
+        if handler.task_type in cls._handlers:
+            raise ValueError(f"Task type '{handler.task_type}' already registered")
+        cls._handlers[handler.task_type] = handler
+        vp = getattr(handler, "virtual_project", None)
+        logger.info("Registered task type handler: %s (virtual_project=%s)", handler.task_type, vp)
+
+    @classmethod
+    def get_by_project(cls, project_id: str) -> Optional[TaskTypeHandler]:
+        """通过 project_id 查找 handler（匹配 virtual_project）。"""
+        for h in cls._handlers.values():
+            if h.virtual_project == project_id:
+                return h
+        return None
+
+    @classmethod
+    def get(cls, task_type: str) -> Optional[TaskTypeHandler]:
+        """通过 task_type 标识查找 handler。"""
+        return cls._handlers.get(task_type)
+
+    @classmethod
+    def virtual_projects(cls) -> list[str]:
+        """返回所有已注册的虚拟项目 ID（ticker 自动发现用）。"""
+        return [
+            h.virtual_project
+            for h in cls._handlers.values()
+            if h.virtual_project is not None
+        ]
+
+    @classmethod
+    def clear(cls) -> None:
+        """清空注册表（仅测试用）。"""
+        cls._handlers = {}

src/daemon/ticker.py

@@ -19,9 +19,10 @@ from typing import Any, Callable, Coroutine, Dict, List, Optional
 
 from dataclasses import dataclass, field as dc_field
 
+from src.daemon.task_type_registry import TaskTypeRegistry
+
 from src.blackboard.operations import Blackboard
 from src.blackboard.db import get_connection
-from src.blackboard.models import Task
 from src.daemon.spawner import AgentBusyError
 from src.blackboard.queries import Queries
 from src.blackboard.registry import ProjectRegistry
@@ -32,9 +33,11 @@ class BroadcastRound:
     """追踪单个任务的广播状态"""
     task_id: str
     notified_agents: set = dc_field(default_factory=set)   # 已 spawn 过的 Agent
-    responded_agents: set = dc_field(default_factory=set)   # 已返回反馈的 Agent（含 NO_REPLY）
+    responded_agents: set = dc_field(
+        default_factory=set)   # 已返回反馈的 Agent（含 NO_REPLY）
     round_number: int = 0                                   # 当前第几轮（0=未开始，1=第1轮）
 
+
 logger = logging.getLogger("moziplus-v2.ticker")
 
 
@@ -46,7 +49,8 @@ class Ticker:
         registry: ProjectRegistry,
         tick_interval: float = 30.0,
         max_ticks: Optional[int] = None,
-        on_tick_complete: Optional[Callable[[], Coroutine[Any, Any, None]]] = None,
+        on_tick_complete: Optional[Callable[[],
+                                            Coroutine[Any, Any, None]]] = None,
         dispatcher: Optional[Any] = None,
         spawner: Optional[Any] = None,
         max_dispatch_per_tick: int = 3,
@@ -194,7 +198,10 @@ class Ticker:
                 pr = await self._tick_project(project_id, project_info)
                 results["projects"][project_id] = pr
             except Exception as e:
-                logger.exception("Tick %d project %s error", tick_num, project_id)
+                logger.exception(
+                    "Tick %d project %s error",
+                    tick_num,
+                    project_id)
                 results["projects"][project_id] = {"error": str(e)}
 
         # 虚拟项目 _general：不在 registry 但需要调度
@@ -210,20 +217,26 @@ class Ticker:
                 logger.exception("Tick %d _general error", tick_num)
                 results["projects"]["_general"] = {"error": str(e)}
 
-        # 虚拟项目 _mail：飞鸽传书
-        mail_db = Path(self.registry.root) / "_mail" / "blackboard.db"
-        if mail_db.exists() and "_mail" not in active_projects:
-            try:
-                pr = await self._tick_project("_mail", {
-                    "id": "_mail", "name": "飞鸽传书",
-                    "status": "active", "source": "virtual",
-                })
-                results["projects"]["_mail"] = pr
-            except Exception as e:
-                logger.exception("Tick %d _mail error", tick_num)
-                results["projects"]["_mail"] = {"error": str(e)}
+        # 虚拟项目：从注册表自动发现 + _general 硬编码
+        for vp in TaskTypeRegistry.virtual_projects():
+            vp_db = Path(self.registry.root) / vp / "blackboard.db"
+            if vp_db.exists() and vp not in active_projects:
+                try:
+                    vp_handler = TaskTypeRegistry.get_by_project(vp)
+                    vp_name = vp_handler.display_name if vp_handler and vp_handler.display_name else vp
+                    pr = await self._tick_project(vp, {
+                        "id": vp, "name": vp_name,
+                        "status": "active", "source": "virtual",
+                    })
+                    results["projects"][vp] = pr
+                except Exception as e:
+                    logger.exception("Tick %d %s error", tick_num, vp)
+                    results["projects"][vp] = {"error": str(e)}
 
-        logger.debug("Tick %d complete: %d projects", tick_num, len(active_projects))
+        logger.debug(
+            "Tick %d complete: %d projects",
+            tick_num,
+            len(active_projects))
 
         if self.on_tick_complete:
             try:
@@ -314,7 +327,8 @@ class Ticker:
         # 8. 健康检查（僵尸检测）
         if self.health_checker:
             try:
-                self.health_checker.check(project_id, db_path, self._tick_count)
+                self.health_checker.check(
+                    project_id, db_path, self._tick_count)
             except Exception as e:
                 logger.warning("HealthChecker error for %s: %s", project_id, e)
 
@@ -335,7 +349,8 @@ class Ticker:
                             task_id=t.id, task_title=t.title, task_type=t.task_type
                         )
             except Exception as e:
-                logger.warning("ExperienceDistiller error for %s: %s", project_id, e)
+                logger.warning(
+                    "ExperienceDistiller error for %s: %s", project_id, e)
 
         # 10. 扫描后状态
         result["summary_after"] = queries.task_summary()
@@ -375,7 +390,8 @@ class Ticker:
                         (computed, pid),
                     )
                     refreshed.append(pid)
-                    logger.info("Parent %s status aggregated: → %s", pid, computed)
+                    logger.info(
+                        "Parent %s status aggregated: → %s", pid, computed)
 
             if refreshed:
                 conn.commit()
@@ -391,7 +407,7 @@ class Ticker:
     MAX_ROUNDS = 5  # §4.5 防无限循环
 
     async def _check_round_complete(self, db_path: Path,
-                                     project_id: str) -> List[str]:
+                                    project_id: str) -> List[str]:
         """检测 parent task 下所有 sub task 终态 → spawn 庞统 review
 
         流程（§4.4）：
@@ -462,7 +478,7 @@ class Ticker:
                         "Round %d review spawned for parent %s (subs: %s)",
                         new_round, parent_id, summary
                     )
-            except Exception as e:
+            except Exception:
                 logger.exception("Round check error for parent %s", parent_id)
 
         return reviewed
@@ -531,9 +547,9 @@ Parent Task ID: {parent_task.id}
 """
 
     async def _spawn_pangtong_review(self, parent_task,
-                                      review_prompt: str,
-                                      project_id: str,
-                                      new_round: int = 0) -> bool:
+                                     review_prompt: str,
+                                     project_id: str,
+                                     new_round: int = 0) -> bool:
         """Spawn 庞统进行 review
 
         流程：
@@ -543,7 +559,7 @@ Parent Task ID: {parent_task.id}
         """
         try:
             agent_id = "pangtong-fujunshi"
-            session_id = f"review-{parent_task.id}-r{new_round}"
+            f"review-{parent_task.id}-r{new_round}"
 
             # 构造 on_complete 回调：解析庞统结论，更新 parent 状态
             async def _on_review_complete(aid: str, outcome: str):
@@ -555,7 +571,8 @@ Parent Task ID: {parent_task.id}
                         latest_meta = None
                         latest_time = ""
                         for sid, sess in self.spawner._sessions.items():
-                            if sess.get("agent_id") == agent_id and sess.get("meta"):
+                            if sess.get(
+                                    "agent_id") == agent_id and sess.get("meta"):
                                 t = sess.get("completed_at", "")
                                 if t > latest_time:
                                     latest_time = t
@@ -586,8 +603,10 @@ Parent Task ID: {parent_task.id}
                 self._set_parent_reviewing(parent_task.id, project_id)
                 return True
             return False
-        except Exception as e:
-            logger.exception("Failed to spawn pangtong review for %s", parent_task.id)
+        except Exception:
+            logger.exception(
+                "Failed to spawn pangtong review for %s",
+                parent_task.id)
             return False
 
     def _set_parent_reviewing(self, parent_id: str, project_id: str):
@@ -603,14 +622,14 @@ Parent Task ID: {parent_task.id}
                     (parent_id,))
                 conn.commit()
                 logger.info("Parent %s → reviewing (round review in progress)",
-                             parent_id)
+                            parent_id)
             finally:
                 conn.close()
         except Exception:
             logger.exception("Failed to set parent %s to reviewing", parent_id)
 
     def _handle_review_conclusion(self, parent_id: str, project_id: str,
-                                   review_text: str, round_num: int):
+                                  review_text: str, round_num: int):
         """解析庞统 review 结论，更新 parent 状态
 
         review_text 是庞统回复的文本（从 spawner session meta payloads 拼接）。
@@ -619,7 +638,8 @@ Parent Task ID: {parent_task.id}
         conn = get_connection(db_path)
         try:
             # 解析 GOAL_ACHIEVED
-            is_achieved = bool(review_text and "GOAL_ACHIEVED" in review_text.upper())
+            is_achieved = bool(
+                review_text and "GOAL_ACHIEVED" in review_text.upper())
 
             if is_achieved:
                 # Goal 达成 → parent 最终完成
@@ -649,7 +669,9 @@ Parent Task ID: {parent_task.id}
                     "(round %d, subs=%d)",
                     parent_id, round_num, sub_count)
         except Exception:
-            logger.exception("Failed to handle review conclusion for %s", parent_id)
+            logger.exception(
+                "Failed to handle review conclusion for %s",
+                parent_id)
             # 安全恢复：reviewing → working
             try:
                 conn.execute("BEGIN IMMEDIATE")
@@ -675,7 +697,7 @@ Parent Task ID: {parent_task.id}
     MENTION_MAX_RETRIES = 5
 
     async def _process_mentions(self, db_path: Path,
-                                 project_id: str) -> List[str]:
+                                project_id: str) -> List[str]:
         """扫描 pending mentions → spawn 被 @ 的 Agent
 
         流程（§3.4）：
@@ -687,7 +709,8 @@ Parent Task ID: {parent_task.id}
             return []
 
         bb = Blackboard(db_path)
-        mentions = bb.get_pending_mentions(max_retries=self.MENTION_MAX_RETRIES)
+        mentions = bb.get_pending_mentions(
+            max_retries=self.MENTION_MAX_RETRIES)
         if not mentions:
             return []
 
@@ -751,27 +774,32 @@ Parent Task ID: {parent_task.id}
 
                                 if new_review and new_review["verdict"] == "approved":
                                     _ticker._transition_status(
-                                        get_connection(rdb_path), _t_id, "done",
+                                        get_connection(
+                                            rdb_path), _t_id, "done",
                                         agent="daemon",
                                         detail={"reason": "rebuttal_approved"})
-                                    logger.info("Rebuttal: task %s approved after rebuttal", _t_id)
+                                    logger.info(
+                                        "Rebuttal: task %s approved after rebuttal", _t_id)
                                 else:
                                     # 仍非 approved → @mention assignee
                                     verdict_str = new_review["verdict"] if new_review else "未知"
                                     rconn2 = get_connection(rdb_path)
                                     try:
-                                        t_row = rconn2.execute("SELECT assignee FROM tasks WHERE id=?", (_t_id,)).fetchone()
+                                        t_row = rconn2.execute(
+                                            "SELECT assignee FROM tasks WHERE id=?", (_t_id,)).fetchone()
                                     finally:
                                         rconn2.close()
                                     if t_row and t_row["assignee"]:
                                         from src.blackboard.blackboard import Blackboard
                                         bb2 = Blackboard(rdb_path)
                                         bb2.add_comment(_t_id, "daemon",
-                                            f"@{t_row['assignee']} 审查结论: {verdict_str}，请查看详情并决定接受或反驳",
-                                            comment_type="review")
-                                    logger.info("Rebuttal: task %s still %s after rebuttal", _t_id, verdict_str)
+                                                        f"@{t_row['assignee']} 审查结论: {verdict_str}，请查看详情并决定接受或反驳",
+                                                        comment_type="review")
+                                    logger.info(
+                                        "Rebuttal: task %s still %s after rebuttal", _t_id, verdict_str)
                         except Exception:
-                            logger.exception("Rebuttal on_complete failed for task %s", _t_id)
+                            logger.exception(
+                                "Rebuttal on_complete failed for task %s", _t_id)
 
                     result = await self.spawner.spawn_full_agent(
                         agent_id=agent_id,
@@ -794,22 +822,30 @@ Parent Task ID: {parent_task.id}
                     for item in items:
                         bb.mark_mention_notified(item["id"])
                     processed.append(agent_id)
-                    logger.info("Mention spawn success: %s (%d mentions)", agent_id, len(items))
+                    logger.info(
+                        "Mention spawn success: %s (%d mentions)",
+                        agent_id,
+                        len(items))
                 else:
                     # spawn 返回 None（其他原因）→ 递增 retry_count
                     for item in items:
                         bb.mark_mention_retry(item["id"])
-                    logger.warning("Mention spawn failed: %s, retrying next tick", agent_id)
+                    logger.warning(
+                        "Mention spawn failed: %s, retrying next tick", agent_id)
 
             except AgentBusyError:
                 # Agent 忙，不递增 retry_count，等下次 tick 自然重试
-                logger.info("Mention spawn skipped: %s busy, will retry next tick", agent_id)
+                logger.info(
+                    "Mention spawn skipped: %s busy, will retry next tick",
+                    agent_id)
 
-            except Exception as e:
-                logger.exception("Mention processing error for agent %s", agent_id)
+            except Exception:
+                logger.exception(
+                    "Mention processing error for agent %s", agent_id)
                 for item in items:
                     try:
-                        if item.get("retry_count", 0) >= self.MENTION_MAX_RETRIES - 1:
+                        if item.get("retry_count",
+                                    0) >= self.MENTION_MAX_RETRIES - 1:
                             bb.mark_mention_failed(item["id"])
                         else:
                             bb.mark_mention_retry(item["id"])
@@ -822,8 +858,14 @@ Parent Task ID: {parent_task.id}
                               mention_lines: List[str],
                               project_id: str) -> str:
         """#03: @mention prompt（身份注入）"""
-        api_host = getattr(self.spawner, 'api_host', '127.0.0.1') if self.spawner else '127.0.0.1'
-        api_port = getattr(self.spawner, 'api_port', 8083) if self.spawner else 8083
+        api_host = getattr(
+            self.spawner,
+            'api_host',
+            '127.0.0.1') if self.spawner else '127.0.0.1'
+        api_port = getattr(
+            self.spawner,
+            'api_port',
+            8083) if self.spawner else 8083
         api_base = f"http://{api_host}:{api_port}/api"
 
         # 获取 Agent 专长
@@ -899,7 +941,8 @@ Parent Task ID: {parent_task.id}
         from datetime import datetime
 
         conn.execute("BEGIN IMMEDIATE")
-        row = conn.execute("SELECT status FROM tasks WHERE id=?", (task_id,)).fetchone()
+        row = conn.execute(
+            "SELECT status FROM tasks WHERE id=?", (task_id,)).fetchone()
         if not row:
             return False
         old_status = row["status"]
@@ -910,9 +953,10 @@ Parent Task ID: {parent_task.id}
 
         now = datetime.utcnow().isoformat()
         # 重置到 pending 时清空 assignee（避免残留导致重复路由到同一 Agent）
-        # 但 Mail 的 assignee 是收件人，永不清空
+        # handler 虚拟项目（_mail 等）的 assignee 是收件人，永不清空
         if new_status == "pending":
-            if self._current_project_id == "_mail":
+            handler = TaskTypeRegistry.get_by_project(self._current_project_id)
+            if handler:
                 conn.execute(
                     "UPDATE tasks SET status=?, updated_at=? WHERE id=?",
                     (new_status, now, task_id),
@@ -938,7 +982,8 @@ Parent Task ID: {parent_task.id}
             event_type = "daemon_tick"
         conn.execute(
             "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
-            (task_id, agent, event_type, json.dumps({"from": old_status, "to": new_status, **(detail or {})})),
+            (task_id, agent, event_type, json.dumps(
+                {"from": old_status, "to": new_status, **(detail or {})})),
         )
         conn.commit()
         return True
@@ -948,7 +993,7 @@ Parent Task ID: {parent_task.id}
     # ------------------------------------------------------------------
 
     async def _dispatch_pending(self, db_path: Path,
-                                 project_id: str) -> List[str]:
+                                project_id: str) -> List[str]:
         """扫描 pending 任务并调度
 
         v3.0: 两条路径
@@ -978,20 +1023,24 @@ Parent Task ID: {parent_task.id}
             try:
                 result = await self.dispatcher.dispatch(
                     task,
-                    project_config={"project_id": project_id, "db_path": db_path},
+                    project_config={
+                        "project_id": project_id,
+                        "db_path": db_path},
                 )
-                if result["status"] == "dispatched" and result["level"] in ("full", "escalate"):
+                if result["status"] == "dispatched" and result["level"] in (
+                        "full", "escalate"):
                     conn = get_connection(db_path)
                     try:
-                        # [v2.7.1] Mail 已在 dispatcher 中标 working，跳过 claimed
-                        if project_id == "_mail":
+                        # [Step 5] handler 项目已在 dispatcher 中标 working，跳过 claimed
+                        handler = TaskTypeRegistry.get_by_project(project_id)
+                        if handler:
                             conn.execute(
                                 "UPDATE tasks SET current_agent=? WHERE id=?",
                                 (result["agent_id"], task.id),
                             )
                             conn.commit()
                             dispatched.append(task.id)
-                            logger.info("Dispatched %s to %s (session=%s, mail auto-working)",
+                            logger.info("Dispatched %s to %s (session=%s, handler auto-working)",
                                         task.id, result["agent_id"],
                                         result.get("session_id"))
                         else:
@@ -1073,7 +1122,8 @@ Parent Task ID: {parent_task.id}
                     detail={"reason": "no_taker_after_3_broadcasts",
                             "round_number": self._broadcast_tracker.get(t.id).round_number if self._broadcast_tracker.get(t.id) else 0},
                 )
-                logger.warning("Escalated %s: no taker after 3 broadcast rounds", t.id)
+                logger.warning(
+                    "Escalated %s: no taker after 3 broadcast rounds", t.id)
                 self._broadcast_tracker.pop(t.id, None)
             finally:
                 conn.close()
@@ -1083,7 +1133,8 @@ Parent Task ID: {parent_task.id}
 
         idle_agents = self._get_idle_agents()
         if not idle_agents:
-            logger.warning("No idle agents for broadcast, skipping (capacity issue)")
+            logger.warning(
+                "No idle agents for broadcast, skipping (capacity issue)")
             return []
 
         task_ids = [t.id for t in broadcastable]
@@ -1114,7 +1165,8 @@ Parent Task ID: {parent_task.id}
 
         spawned = []
         for agent_id in idle_agents:
-            prompt = self._build_claim_prompt(agent_id, broadcastable, project_id)
+            prompt = self._build_claim_prompt(
+                agent_id, broadcastable, project_id)
             try:
                 session_id = await self.spawner.spawn_full_agent(
                     agent_id=agent_id,
@@ -1128,7 +1180,8 @@ Parent Task ID: {parent_task.id}
                     spawned.append(session_id)
                     # 记录已通知的 Agent
                     for t in broadcastable:
-                        self._broadcast_tracker[t.id].notified_agents.add(agent_id)
+                        self._broadcast_tracker[t.id].notified_agents.add(
+                            agent_id)
             except AgentBusyError:
                 logger.debug("Broadcast skip %s: busy", agent_id)
             except Exception:
@@ -1139,8 +1192,14 @@ Parent Task ID: {parent_task.id}
     def _build_claim_prompt(self, agent_id: str, tasks: list,
                             project_id: str) -> str:
         """#03: 广播认领 prompt（身份+专长注入）"""
-        api_host = getattr(self.spawner, 'api_host', '127.0.0.1') if self.spawner else '127.0.0.1'
-        api_port = getattr(self.spawner, 'api_port', 8083) if self.spawner else 8083
+        api_host = getattr(
+            self.spawner,
+            'api_host',
+            '127.0.0.1') if self.spawner else '127.0.0.1'
+        api_port = getattr(
+            self.spawner,
+            'api_port',
+            8083) if self.spawner else 8083
         api_base = f"http://{api_host}:{api_port}/api"
 
         # 获取 Agent 专长
@@ -1195,7 +1254,8 @@ Parent Task ID: {parent_task.id}
     @property
     def counter(self):
         """从 Dispatcher 获取 counter"""
-        return getattr(self.dispatcher, 'counter', None) if self.dispatcher else None
+        return getattr(self.dispatcher, 'counter',
+                       None) if self.dispatcher else None
 
     @staticmethod
     def _is_pid_alive(pid: int) -> bool:
@@ -1207,7 +1267,8 @@ Parent Task ID: {parent_task.id}
         except (ProcessLookupError, PermissionError):
             return False
 
-    def record_broadcast_response(self, task_id: str, agent_id: str, outcome: str):
+    def record_broadcast_response(
+            self, task_id: str, agent_id: str, outcome: str):
         """记录 Agent 对广播任务的反馈（Spawner 调用的公共 API）"""
         tracker = self._broadcast_tracker.get(task_id)
         if not tracker:
@@ -1228,7 +1289,8 @@ Parent Task ID: {parent_task.id}
 
     def _get_all_agent_ids(self) -> List[str]:
         """获取所有配置的 Agent ID"""
-        if self.dispatcher and hasattr(self.dispatcher, 'router') and self.dispatcher.router:
+        if self.dispatcher and hasattr(
+                self.dispatcher, 'router') and self.dispatcher.router:
             return list(self.dispatcher.router.agent_profiles.keys())
         return []
 
@@ -1237,15 +1299,17 @@ Parent Task ID: {parent_task.id}
         if not self.counter:
             return []
         # agent_profiles 在 Router 初始化时从 config 填充，是完整 Agent 列表
-        all_agents = list(self.dispatcher.router.agent_profiles.keys()) if self.dispatcher else []
+        all_agents = list(
+            self.dispatcher.router.agent_profiles.keys()) if self.dispatcher else []
         active = self.counter.active_agents
         return [aid for aid in all_agents if active.get(aid, 0) == 0]
 
     async def _dispatch_reviews(self, db_path: Path,
-                                 project_id: str) -> List[str]:
+                                project_id: str) -> List[str]:
         """扫描 review 状态任务，检查是否有产出，调度审查 Agent"""
-        # mail 任务不走 review 流程，直接跳过
-        if project_id == "_mail":
+        # handler 项目（_mail/_toolchain）不走 review 流程
+        handler = TaskTypeRegistry.get_by_project(project_id)
+        if handler:
             return []
 
         queries = Queries(db_path)
@@ -1291,7 +1355,9 @@ Parent Task ID: {parent_task.id}
                 result = await self.dispatcher.dispatch(
                     task,
                     action_type="review",
-                    project_config={"project_id": project_id, "db_path": db_path},
+                    project_config={
+                        "project_id": project_id,
+                        "db_path": db_path},
                 )
                 if result["status"] == "dispatched":
                     dispatched.append(task.id)
@@ -1344,7 +1410,7 @@ Parent Task ID: {parent_task.id}
                             )
                             reclaimed.append(task.id)
                             logger.warning("Escalated %s: no taker after %d broadcasts",
-                                            task.id, retry_count)
+                                           task.id, retry_count)
                         finally:
                             conn.close()
                     else:
@@ -1375,8 +1441,10 @@ Parent Task ID: {parent_task.id}
         working = queries.tasks_by_status("working")
         for task in working:
             # #07.2: crash_limit 统一检查（比超时更严重的信号）
-            if self.dispatcher and hasattr(self.dispatcher, '_check_crash_limit'):
-                if self.dispatcher._check_crash_limit(task.id, db_path, limit=3, window_minutes=30):
+            if self.dispatcher and hasattr(
+                    self.dispatcher, '_check_crash_limit'):
+                if self.dispatcher._check_crash_limit(
+                        task.id, db_path, limit=3, window_minutes=30):
                     conn = get_connection(db_path)
                     try:
                         self._transition_status(
@@ -1388,7 +1456,8 @@ Parent Task ID: {parent_task.id}
                     finally:
                         conn.close()
                     reclaimed.append(task.id)
-                    logger.error("Task %s: executor crash limit (3/30m), marking failed", task.id)
+                    logger.error(
+                        "Task %s: executor crash limit (3/30m), marking failed", task.id)
                     continue
 
             # #07.3 ACT-1: updated_at fallback 覆盖 mail auto-working（无 started_at/claimed_at）
@@ -1400,7 +1469,8 @@ Parent Task ID: {parent_task.id}
                 # per-task timeout: deadline 优先，否则用默认值
                 if task.deadline:
                     deadline_time = datetime.fromisoformat(task.deadline)
-                    timeout_minutes = (deadline_time - start_time).total_seconds() / 60.0
+                    timeout_minutes = (
+                        deadline_time - start_time).total_seconds() / 60.0
                     if timeout_minutes < 1:
                         timeout_minutes = self.default_task_timeout_minutes
                 else:
@@ -1408,10 +1478,9 @@ Parent Task ID: {parent_task.id}
 
                 elapsed = (now - start_time).total_seconds() / 60.0
                 if elapsed > timeout_minutes:
-                    # [v2.7.1] Mail 幻觉门控兜底：有回复 + working → done
-                    if self._current_project_id == "_mail":
-                        has_reply = self._mail_check_reply(task.id, db_path)
-                        if has_reply:
+                    # [Step 5] handler 幻觉门控兜底：check_completion 通过 + working → done
+                    handler = TaskTypeRegistry.get_by_project(self._current_project_id)
+                    if handler and handler.check_completion(task.id, db_path):
                             conn = get_connection(db_path)
                             try:
                                 ok = self._transition_status(
@@ -1423,7 +1492,7 @@ Parent Task ID: {parent_task.id}
                                 if ok:
                                     reclaimed.append(task.id)
                                     logger.info("Mail %s: ticker recheck found reply, marked done (%.1fm)",
-                                                 task.id, elapsed)
+                                                task.id, elapsed)
                             finally:
                                 conn.close()
                             continue
@@ -1440,15 +1509,17 @@ Parent Task ID: {parent_task.id}
                         if ok:
                             reclaimed.append(task.id)
                             logger.warning("Task %s timed out (working %.1fm > %.1fm)",
-                                            task.id, elapsed, timeout_minutes)
+                                           task.id, elapsed, timeout_minutes)
                     finally:
                         conn.close()
             except (ValueError, TypeError):
                 pass
 
         # v2.7.2: 进程存活性检查 — counter 占用但进程已死的兜底
-        if self.spawner and self.counter and hasattr(self.counter, "active_agents"):
-            for agent_id in list(self.counter.active_agents.keys()) if hasattr(self.counter, "active_agents") else []:
+        if self.spawner and self.counter and hasattr(
+                self.counter, "active_agents"):
+            for agent_id in list(self.counter.active_agents.keys()) if hasattr(
+                    self.counter, "active_agents") else []:
                 session_info = self.spawner.get_session_by_agent(agent_id)
                 if not session_info:
                     continue
@@ -1465,20 +1536,24 @@ Parent Task ID: {parent_task.id}
                             conn = get_connection(db_path)
                             try:
                                 current_row = conn.execute(
-                                    "SELECT status FROM tasks WHERE id=?", (task_id_check,)
+                                    "SELECT status FROM tasks WHERE id=?", (
+                                        task_id_check,)
                                 ).fetchone()
                                 if current_row and current_row["status"] == "review":
-                                    logger.info("Task %s in review, keeping status (process dead)", task_id_check)
+                                    logger.info(
+                                        "Task %s in review, keeping status (process dead)", task_id_check)
                                 else:
                                     self._transition_status(
                                         conn, task_id_check, "pending",
                                         agent="daemon",
-                                        detail={"reason": "process_dead", "pid": pid},
+                                        detail={
+                                            "reason": "process_dead", "pid": pid},
                                     )
                             finally:
                                 conn.close()
                         except Exception:
-                            logger.exception("Failed to handle process dead for task %s", task_id_check)
+                            logger.exception(
+                                "Failed to handle process dead for task %s", task_id_check)
 
         # #07.2: Fix-3b 已删除。review 超时/crash 统一由 process_dead + _check_timeouts 处理
 
@@ -1497,16 +1572,20 @@ Parent Task ID: {parent_task.id}
             finally:
                 conn.close()
         except Exception as e:
-            logger.error("Mail %s: ticker reply check error: %s", original_task_id, e)
+            logger.error(
+                "Mail %s: ticker reply check error: %s",
+                original_task_id,
+                e)
             return True  # 保守：查询失败假设有回复
 
     def _check_recent_routing(self, db_path: Path, task_id: str,
-                                action_type: str) -> bool:
+                              action_type: str) -> bool:
         """检查最近 5 分钟内是否已 dispatch 过指定类型的路由（防重复）"""
         try:
             conn = get_connection(db_path)
             try:
-                # 检查是否有 from_status=review 的 dispatched 记录（防止重复 review dispatch）
+                # 检查是否有 from_status=review 的 dispatched 记录（防止重复 review
+                # dispatch）
                 if action_type == "review":
                     row = conn.execute(
                         "SELECT COUNT(*) as cnt FROM routing_decisions "
@@ -1537,17 +1616,23 @@ Parent Task ID: {parent_task.id}
         NON_TERMINAL = {"claimed", "working", "review", "reviewing"}
 
         projects = self.registry.list_projects()
-        recovery_report = {"projects": {}, "total_recovered": 0, "total_noop": 0}
+        recovery_report = {
+            "projects": {},
+            "total_recovered": 0,
+            "total_noop": 0}
 
         # 收集所有需要扫描的项目（registry + 虚拟项目）
         project_dirs = {}
         for project_id, project_info in projects.items():
             if project_info.get("status") == "active":
-                project_dirs[project_id] = self.registry.root / project_id / "blackboard.db"
+                project_dirs[project_id] = self.registry.root / \
+                    project_id / "blackboard.db"
 
-        # 虚拟项目
-        for virtual_id in ("_general", "_mail"):
-            virtual_db = Path(self.registry.root) / virtual_id / "blackboard.db"
+        # 虚拟项目：_general + 注册表自动发现
+        virtual_ids = ["_general"] + TaskTypeRegistry.virtual_projects()
+        for virtual_id in virtual_ids:
+            virtual_db = Path(self.registry.root) / \
+                virtual_id / "blackboard.db"
             if virtual_db.exists() and virtual_id not in project_dirs:
                 project_dirs[virtual_id] = virtual_db
 
@@ -1567,25 +1652,28 @@ Parent Task ID: {parent_task.id}
             old_pid = self._current_project_id
             self._current_project_id = project_id
             try:
-                recovered, noop_count = self._recover_project(db_path, NON_TERMINAL)
+                recovered, noop_count = self._recover_project(
+                    db_path, NON_TERMINAL)
                 if recovered:
                     recovery_report["projects"][project_id] = recovered
                     recovery_report["total_recovered"] += len(recovered)
                 recovery_report["total_noop"] += noop_count
             except Exception:
-                logger.exception("Startup recovery failed for project %s", project_id)
+                logger.exception(
+                    "Startup recovery failed for project %s", project_id)
             finally:
                 self._current_project_id = old_pid
 
         if recovery_report["total_recovered"] > 0:
             logger.info("Startup recovery: %d tasks recovered across %d projects",
-                         recovery_report["total_recovered"],
-                         len(recovery_report["projects"]))
+                        recovery_report["total_recovered"],
+                        len(recovery_report["projects"]))
         elif recovery_report["total_noop"] > 0:
             logger.info("Startup recovery: %d tasks kept as-is (no recovery needed)",
-                         recovery_report["total_noop"])
+                        recovery_report["total_noop"])
         else:
-            logger.info("Startup recovery: no non-terminal tasks found, clean start")
+            logger.info(
+                "Startup recovery: no non-terminal tasks found, clean start")
 
         return recovery_report
 
@@ -1608,10 +1696,13 @@ Parent Task ID: {parent_task.id}
 
                 for task in rows:
                     try:
-                        action = self._determine_recovery_action(conn, task, status, db_path)
+                        action = self._determine_recovery_action(
+                            conn, task, status, db_path)
                         if action:
-                            self._execute_recovery(conn, task["id"], action, db_path)
-                            recovered.append({"task_id": task["id"], "from": status, "action": action})
+                            self._execute_recovery(
+                                conn, task["id"], action, db_path)
+                            recovered.append(
+                                {"task_id": task["id"], "from": status, "action": action})
                         else:
                             # 审计：保持原状的任务也记录事件
                             noop_count += 1
@@ -1622,14 +1713,15 @@ Parent Task ID: {parent_task.id}
                             )
                             conn.commit()
                     except Exception:
-                        logger.exception("Startup recovery failed for task %s", task["id"])
+                        logger.exception(
+                            "Startup recovery failed for task %s", task["id"])
         finally:
             conn.close()
 
         return recovered, noop_count
 
     def _determine_recovery_action(self, conn, task, status: str,
-                                    db_path: Path) -> Optional[str]:
+                                   db_path: Path) -> Optional[str]:
         """根据黑板线索决定恢复动作，返回 None 表示不需要干预"""
         task_id = task["id"]
 
@@ -1700,7 +1792,8 @@ Parent Task ID: {parent_task.id}
         # 无审查结论 → 保持 review，ticker 自然会 dispatch reviewer
         return None
 
-    def _execute_recovery(self, conn, task_id: str, action: str, db_path: Path):
+    def _execute_recovery(self, conn, task_id: str,
+                          action: str, db_path: Path):
         """执行恢复动作"""
         # 获取原始状态（用于审计）
         orig_row = conn.execute(
@@ -1712,17 +1805,22 @@ Parent Task ID: {parent_task.id}
             self._transition_status(
                 conn, task_id, "pending",
                 agent="daemon",
-                detail={"reason": "startup_recovery", "original_status": orig_status},
+                detail={
+                    "reason": "startup_recovery",
+                    "original_status": orig_status},
             )
             # 清空 current_agent（常规推 pending，无特定 agent 接手）
-            conn.execute("UPDATE tasks SET current_agent=NULL WHERE id=?", (task_id,))
+            conn.execute(
+                "UPDATE tasks SET current_agent=NULL WHERE id=?", (task_id,))
             conn.commit()
 
         elif action == "push_to_pending_keep_agent":
             self._transition_status(
                 conn, task_id, "pending",
                 agent="daemon",
-                detail={"reason": "startup_recovery", "original_status": orig_status},
+                detail={
+                    "reason": "startup_recovery",
+                    "original_status": orig_status},
             )
             # 保留 current_agent，让同一 agent 重新接手
             conn.commit()
@@ -1731,7 +1829,9 @@ Parent Task ID: {parent_task.id}
             self._transition_status(
                 conn, task_id, "review",
                 agent="daemon",
-                detail={"reason": "startup_recovery", "original_status": "working"},
+                detail={
+                    "reason": "startup_recovery",
+                    "original_status": "working"},
             )
             conn.commit()
 
@@ -1739,7 +1839,9 @@ Parent Task ID: {parent_task.id}
             self._transition_status(
                 conn, task_id, "done",
                 agent="daemon",
-                detail={"reason": "startup_recovery", "original_status": orig_status},
+                detail={
+                    "reason": "startup_recovery",
+                    "original_status": orig_status},
             )
             conn.commit()
 
@@ -1747,22 +1849,30 @@ Parent Task ID: {parent_task.id}
             self._transition_status(
                 conn, task_id, "failed",
                 agent="daemon",
-                detail={"reason": "startup_recovery", "original_status": orig_status},
+                detail={
+                    "reason": "startup_recovery",
+                    "original_status": orig_status},
             )
             conn.commit()
 
         # 记录恢复审计事件
         conn.execute(
             "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?, ?, ?, ?)",
-            (task_id, "daemon", "startup_recovery", json.dumps({"action": action}))
+            (task_id, "daemon", "startup_recovery",
+             json.dumps({"action": action}))
         )
         conn.commit()
 
-        logger.info("Recovery: task %s → %s (action=%s)", task_id, action, action)
+        logger.info(
+            "Recovery: task %s → %s (action=%s)",
+            task_id,
+            action,
+            action)
 
     def _find_pre_reviewing_status(self, conn, task_id: str) -> str:
         """查 events 表找到 reviewing 之前的状态（done 或 failed）"""
-        # _transition_status 写入 event_type=f"task_{new_status}"，detail 用 from/to
+        # _transition_status 写入 event_type=f"task_{new_status}"，detail 用
+        # from/to
         rows = conn.execute(
             """SELECT detail FROM events
                WHERE task_id=? AND event_type='task_reviewing'
@@ -1773,7 +1883,8 @@ Parent Task ID: {parent_task.id}
         for event in rows:
             try:
                 detail = json.loads(event["detail"])
-                # _transition_status detail 格式: {"from": old_status, "to": new_status, ...}
+                # _transition_status detail 格式: {"from": old_status, "to":
+                # new_status, ...}
                 prev = detail.get("from") or detail.get("old_status")
                 if prev in ("done", "failed"):
                     return prev

src/daemon/toolchain_handler.py

@@ -0,0 +1,277 @@
+"""toolchain_handler.py — 工具链事件 handler。
+
+处理 Gitea Webhook 事件（CI 失败、Review 请求、Issue 指派等）。
+"""
+from __future__ import annotations
+
+import json
+import logging
+import urllib.request
+from pathlib import Path
+from typing import Dict
+
+from src.daemon.base_task_handler import BaseTaskHandler, VerifyResult
+from src.daemon.prompt_composer import PromptComposer, PromptContext
+from src.daemon.toolchain_templates import render_template, _TEMPLATE_MAP
+from src.blackboard.db import get_connection
+
+logger = logging.getLogger("moziplus-v2.handler.toolchain")
+
+
+# ---------------------------------------------------------------------------
+# Toolchain PromptSections
+# ---------------------------------------------------------------------------
+
+class ToolchainContextSection:
+    """事件类型 + 事件详情（priority=10）"""
+
+    name: str = "toolchain_context"
+    priority: int = 10
+
+    def render(self, context: PromptContext) -> str:
+        event_type = context.event_type
+        event_data: Dict = context.event_data or {}
+
+        if event_type in _TEMPLATE_MAP:
+            # 使用模板引擎渲染已知事件
+            variables = {k: str(v) for k, v in event_data.items()}
+            return render_template(event_type, variables)
+
+        # fallback：通用事件描述
+        lines = [f"## 工具链事件", f""]
+        lines.append(f"- **事件类型**: {event_type or '未知'}")
+        if event_data:
+            lines.append(f"- **事件详情**:")
+            for key, value in event_data.items():
+                lines.append(f"  - {key}: {value}")
+        lines.append(f"")
+        return "\n".join(lines)
+
+    def should_include(self, context: PromptContext) -> bool:
+        return True
+
+
+class ToolchainApiSection:
+    """API 操作指令（priority=40），success_status=done"""
+
+    name: str = "toolchain_api"
+    priority: int = 40
+
+    API_HOST = "localhost:8083"
+
+    def render(self, context: PromptContext) -> str:
+        lines = [
+            "## API 操作指令",
+            "",
+            f"项目 ID: `{context.project_id}`",
+            f"任务 ID: `{context.task_id}`",
+            "",
+            "### 完成后必须更新任务状态",
+            "完成后务必通过以下命令将任务标记为 **done**：",
+            "```bash",
+            f'curl -s -X POST "http://{self.API_HOST}/api/projects/{context.project_id}/tasks/{context.task_id}/status" \\',
+            '  -H "Content-Type: application/json" \\',
+            '  -d \'{"status": "done"}\'',
+            "```",
+            "",
+            "### 提交产出",
+            "如有产出（如 review 结果、修复方案），提交到任务 outputs：",
+            "```bash",
+            f'curl -s -X POST "http://{self.API_HOST}/api/projects/{context.project_id}/tasks/{context.task_id}/outputs" \\',
+            '  -H "Content-Type: application/json" \\',
+            '  -d \'{"content": "<你的产出内容>", "type": "text"}\'',
+            "```",
+            "",
+        ]
+        return "\n".join(lines)
+
+    def should_include(self, context: PromptContext) -> bool:
+        return True
+
+
+class ToolchainConstraintsSection:
+    """硬约束（priority=50）"""
+
+    name: str = "toolchain_constraints"
+    priority: int = 50
+
+    def render(self, context: PromptContext) -> str:
+        lines = [
+            "## 硬约束",
+            "",
+            "1. **必须标 done**：处理完成后必须通过 API 将任务状态更新为 `done`，否则视为未完成",
+            "2. **产出不能为空**：必须提交有意义的产出（output 或 comment），不能只改状态",
+            "3. **单一职责**：只处理本次事件相关的操作，不要越界执行无关任务",
+            "4. **出错即报告**：如果无法处理（如权限不足、资源不存在），在 comment 中说明原因并标 done",
+            "5. **不要创建新任务**：工具链事件只处理当前事件，不衍生新任务",
+            "",
+        ]
+        return "\n".join(lines)
+
+    def should_include(self, context: PromptContext) -> bool:
+        return True
+
+
+# ---------------------------------------------------------------------------
+# ToolchainHandler
+# ---------------------------------------------------------------------------
+
+class ToolchainHandler(BaseTaskHandler):
+    """工具链事件 handler。"""
+
+    task_type = "toolchain"
+    virtual_project = "_toolchain"
+    display_name = "工具链事件"
+
+    def target_success_status(self) -> str:
+        return "done"
+
+    def pre_spawn(self, task_id: str, db_path: Path) -> bool:
+        """auto_working：pending → working"""
+        return self._auto_mark_working(task_id, db_path)
+
+    def get_sections(self) -> list:
+        """返回 3 个 Toolchain PromptSection 实例"""
+        return [
+            ToolchainContextSection(),
+            ToolchainApiSection(),
+            ToolchainConstraintsSection(),
+        ]
+
+    def build_prompt(self, context: PromptContext) -> str:
+        """通过 PromptComposer 拼装 sections 为最终 prompt"""
+        composer = PromptComposer()
+        composer.add_many(self.get_sections())
+        return composer.compose(context)
+
+    def verify_completion(self, task_id: str, db_path: Path) -> VerifyResult:
+        """检查行动输出（output 或 comment 有实质内容）"""
+        try:
+            conn = get_connection(db_path)
+            try:
+                # 检查 output
+                output_count = conn.execute(
+                    "SELECT COUNT(*) FROM outputs WHERE task_id=?", (task_id,)
+                ).fetchone()[0]
+                if output_count > 0:
+                    return VerifyResult(True, "has_output", f"output_count={output_count}")
+
+                # 检查 comment（非系统、有实质内容）
+                comment_count = conn.execute(
+                    "SELECT COUNT(*) FROM comments WHERE task_id=? "
+                    "AND author != 'system' AND LENGTH(content) >= 20",
+                    (task_id,)
+                ).fetchone()[0]
+                if comment_count > 0:
+                    return VerifyResult(True, "has_comment", f"comment_count={comment_count}")
+
+                return VerifyResult(False, "no_action", "output=0, comment=0")
+            finally:
+                conn.close()
+        except Exception as e:
+            logger.error("Toolchain %s: verify error: %s", task_id, e)
+            return VerifyResult(False, "verify_error", str(e))
+
+    def on_failure(self, task_id: str, agent_id: str,
+                   db_path: Path, verify: VerifyResult) -> None:
+        """验证失败 → 标 failed + Mail API 通知主公"""
+        self._mark_task_status(db_path, task_id, "failed")
+        logger.info("Toolchain %s: verify failed (%s), marked failed", task_id, verify.reason)
+
+        # 从 db 读取事件上下文
+        event_type = ""
+        event_data: Dict = {}
+        try:
+            conn = get_connection(db_path)
+            row = conn.execute(
+                "SELECT must_haves FROM tasks WHERE id=?", (task_id,)
+            ).fetchone()
+            if row and row["must_haves"]:
+                meta = json.loads(row["must_haves"])
+                event_type = meta.get("event_type", "")
+                raw = meta.get("event_data", "{}")
+                event_data = json.loads(raw) if isinstance(raw, str) else raw
+            conn.close()
+        except Exception:
+            pass
+
+        self._notify_via_mail_api(
+            task_id, verify.reason, verify.evidence,
+            event_type, event_data,
+        )
+
+    def _build_gitea_links(self, event_type: str, event_data: dict) -> str:
+        """根据事件类型构建 Gitea 链接。"""
+        links = []
+        repo = event_data.get("repo", "")
+        base_url = "http://192.168.2.154:3000"
+
+        if "pr_number" in event_data:
+            links.append(f"PR: {base_url}/{repo}/pulls/{event_data['pr_number']}")
+        if "issue_number" in event_data:
+            links.append(f"Issue: {base_url}/{repo}/issues/{event_data['issue_number']}")
+        if "commit" in event_data:
+            links.append(f"Commit: {base_url}/{repo}/commit/{event_data['commit']}")
+        if "branch" in event_data and "commit" not in event_data:
+            links.append(f"分支: {event_data['branch']}")
+
+        return "\n".join(links) if links else "（无法提取链接，请检查黑板任务详情）"
+
+    def _notify_via_mail_api(
+        self,
+        task_id: str,
+        reason: str,
+        evidence: str,
+        event_type: str,
+        event_data: Dict,
+    ) -> None:
+        """通过 Mail API 发送丰富的失败通知给主公。"""
+        # 构建行动指引
+        action_hint = "请检查黑板任务并手动处理。"
+        et_lower = event_type.lower()
+        if "ci" in et_lower or "deploy" in et_lower:
+            action_hint = "建议创建任务派给 jiangwei-infra 检查 CI/部署问题。"
+        elif "review" in et_lower:
+            action_hint = "建议查看 PR review 状态，必要时通知相关开发者。"
+        elif "issue" in et_lower:
+            action_hint = "建议创建任务派给对应开发者处理 Issue。"
+
+        # 构建事件详情
+        event_details = ""
+        if event_data:
+            event_details = "\n".join(
+                f"  - {k}: {v}" for k, v in event_data.items()
+            )
+
+        # 构建 Gitea 链接
+        gitea_links = self._build_gitea_links(event_type, event_data)
+
+        title = f"[toolchain-handler] 工具链事件处理失败: {task_id}"
+        text = (
+            f"任务 {task_id} 验证失败\n\n"
+            f"事件类型: {event_type or '未知'}\n"
+            f"事件详情:\n{event_details or '  (无)'}\n\n"
+            f"失败原因: {reason}\n"
+            f"证据: {evidence}\n\n"
+            f"{gitea_links}\n\n"
+            f"行动指引: {action_hint}"
+        )
+
+        payload = json.dumps({
+            "from": "daemon",
+            "to": "pangtong-fujunshi",
+            "title": title,
+            "text": text,
+            "type": "inform",
+        }, ensure_ascii=False).encode("utf-8")
+
+        try:
+            req = urllib.request.Request(
+                "http://localhost:8083/api/mail",
+                data=payload,
+                headers={"Content-Type": "application/json"},
+            )
+            urllib.request.urlopen(req, timeout=5)
+            logger.info("Toolchain %s: sent failure notification via Mail API", task_id)
+        except Exception as e:
+            logger.warning("Toolchain %s: failed to notify via Mail API: %s", task_id, e)

src/frontend/src/components/TaskModal.tsx

@@ -426,7 +426,7 @@ export default function TaskModal() {
             {/* 状态操作 */}
             <div style={{ marginBottom: 16 }}>
               <SectionLabel icon="🔄" title="状态操作" />
-              <StatusButtons status={task.status} taskId={task.id} resumedFrom={task.resumed_from} />
+              <StatusButtons status={task.status} taskId={task.id} resumedFrom={task.resumed_from ?? undefined} />
             </div>
 
             {/* v2.7: 子 Task 进度 + 列表 */}

src/frontend/src/store.ts

@@ -57,6 +57,7 @@ export interface V2Task {
   estimated_duration_minutes: number | null;
   escalated: number;
   archived: number;  // v2.8: 归档标记
+  resumed_from: string | null;  // v2.8: 续杯来源
   // API 聚合字段
   comments_count?: number;
   outputs_count?: number;

src/main.py

@@ -1,6 +1,13 @@
 """v2.6 主入口 - FastAPI + Daemon ticker 共享 asyncio event loop"""
 
 from __future__ import annotations
+from src.api.toolchain_routes import router as toolchain_router
+from src.api.mail_routes import router as mail_router
+from src.api.sse_routes import router as sse_router
+from src.api.project_routes import router as project_router
+from src.api.daemon_routes import router as daemon_router
+from src.api.checkpoint_routes import router as checkpoint_router
+from src.api.blackboard_routes import router as blackboard_router
 
 import logging
 from contextlib import asynccontextmanager
@@ -131,7 +138,8 @@ async def lifespan(app: FastAPI):
     counter = ActiveAgentCounter(
         max_global=daemon_config.get("max_global_agents", 5),
         max_per_session=daemon_config.get("max_per_session", 1),
-        max_concurrent_sessions=daemon_config.get("max_concurrent_sessions", 3),
+        max_concurrent_sessions=daemon_config.get(
+            "max_concurrent_sessions", 3),
         default_cooldown_seconds=daemon_config.get("cooldown_seconds", 120),
     )
     # BootstrapBuilder（L2 四段式引擎注入层，v2.1）
@@ -181,7 +189,10 @@ async def lifespan(app: FastAPI):
         spawner=spawner,
         counter=counter,
         db_path=default_db_path,
-        guardrails=GuardrailEngine(config_path=Path(__file__).parent.parent / "config" / "guardrails.yaml"),
+        guardrails=GuardrailEngine(
+            config_path=Path(__file__).parent.parent /
+            "config" /
+            "guardrails.yaml"),
     )
 
     # ── 集成模块 ──
@@ -191,7 +202,7 @@ async def lifespan(app: FastAPI):
     )
 
     # ExperienceDistiller（经验自动蒸馏）
-    experience_config = config.get("experience", {})
+    config.get("experience", {})
     experience_distiller = ExperienceDistiller(
         store=ExperienceStore(store_path=DATA_ROOT / "experiences.jsonl"),
     )
@@ -252,13 +263,6 @@ app.add_middleware(
 # API 路由注册
 # ---------------------------------------------------------------------------
 
-from src.api.blackboard_routes import router as blackboard_router
-from src.api.checkpoint_routes import router as checkpoint_router
-from src.api.daemon_routes import router as daemon_router
-from src.api.project_routes import router as project_router
-from src.api.sse_routes import router as sse_router
-from src.api.mail_routes import router as mail_router
-from src.api.toolchain_routes import router as toolchain_router
 
 app.include_router(blackboard_router)
 app.include_router(checkpoint_router)
@@ -268,6 +272,17 @@ app.include_router(sse_router)
 app.include_router(mail_router)
 app.include_router(toolchain_router)
 
+# ---------------------------------------------------------------------------
+# 健康检查端点
+# ---------------------------------------------------------------------------
+
+
+@app.get("/api/healthz")
+async def healthz():
+    """轻量级健康检查，无需认证"""
+    return {"status": "ok"}
+
+
 # ---------------------------------------------------------------------------
 # 兼容端点
 # ---------------------------------------------------------------------------
@@ -289,16 +304,17 @@ async def list_projects_compat():
 DIST_DIR = Path(__file__).parent / "frontend" / "dist"
 if DIST_DIR.exists():
     # v3.1: 缓存策略 - HTML 不缓存（确保新版本生效），JS/CSS 长缓存（Vite content hash 已处理）
-    import mimetypes
     _static_app = StaticFiles(directory=str(DIST_DIR), html=True)
-    
+
     class CachedStaticFiles:
         """包装 StaticFiles，添加 Cache-Control 头"""
+
         def __init__(self, app):
             self._app = app
-        
+
         async def __call__(self, scope, receive, send):
             original_send = send
+
             async def patched_send(message):
                 if message.get("type") == "http.response.start":
                     headers = dict(message.get("headers", []))
@@ -310,5 +326,5 @@ if DIST_DIR.exists():
                     message["headers"] = list(headers.items())
                 await original_send(message)
             await self._app(scope, receive, patched_send)
-    
+
     app.mount("/", CachedStaticFiles(_static_app), name="frontend")

src/utils.py

@@ -10,7 +10,6 @@ from __future__ import annotations
 
 import os
 from pathlib import Path
-from typing import Optional
 
 
 def get_data_root() -> Path:

tests/conftest.py

@@ -55,6 +55,21 @@ def client_with_isolation(isolated_data_root):
 
 # ── E2E gate ──
 
+def pytest_collection_modifyitems(config, items):
+    if not os.environ.get("RUN_INTEGRATION"):
+        skip_reason = "needs RUN_INTEGRATION=1"
+        remaining = []
+        deselected = []
+        for item in items:
+            if "integration" in item.keywords or "e2e" in item.keywords:
+                deselected.append(item)
+            else:
+                remaining.append(item)
+        if deselected:
+            config.hook.pytest_deselected(items=deselected)
+            items[:] = remaining
+
+
 skip_no_integration = pytest.mark.skipif(
     not os.environ.get("RUN_INTEGRATION"),
     reason="Set RUN_INTEGRATION=1 to run E2E tests against real daemon",

tests/e2e/test_e2e_v27.py

@@ -1,12 +1,12 @@
 import pytest
 
-pytestmark = pytest.mark.e2e
-
 skip_no_integration = pytest.mark.skipif(
     not __import__("os").environ.get("RUN_INTEGRATION"),
     reason="Set RUN_INTEGRATION=1 to run E2E tests against real daemon",
 )
 
+pytestmark = [pytest.mark.e2e, skip_no_integration]
+
 """v2.7 端到端测试 — 全链路真实环境
 
 覆盖：项目管理 → Task CRUD → SubTask → Stage进度 → 状态聚合 → 依赖链 → 超时 → Mail → 真实Agent调度

tests/unit/test_classify_outcome.py

@@ -123,7 +123,7 @@ class TestClassifyNoJsonExit0:
 
     def test_task_status_pending(self):
         result = Spawner._classify_outcome(0, {}, "", "pending", "")
-        assert result["outcome"] == "agent_error"
+        assert result["outcome"] == "completed"
         assert result["should_retry"] is False