---
title: "TaskTypeRegistry + Handler 架构重构"
created: 2026-06-10
version: v3.0
---

# §1 现状分析（v3.0 更新说明：§1-§13 保留原样，新增 §14-§18，更新 §3/§5/§7）

# §1 现状分析

moziplus v2 的任务调度系统当前通过 `if/else` 硬编码区分两种 task type：普通任务（task）和邮件（mail）。分支逻辑散落在 dispatcher、spawner、ticker 三个核心模块中，新增 task type 需要同时改动三处。

### dispatcher.py 中的分支

- **`is_mail` 判断**：`project_id == "_mail"` 作为唯一判据
- **mail 分支**：跳过 guardrail 检查；有专属的 `on_checks_passed` 回调（`_mail_auto_working`）和 `on_complete` 回调（`_mail_auto_complete`，含幻觉门控 + 失败通知）
- **非 mail 分支**：走正常 guardrail 检查 → `on_checks_passed` → `on_complete` 全链路

### spawner.py 中的分支

- **`_build_prompt`**：`_mail` 走 `_build_mail_prompt()`，其余走 BootstrapBuilder L0-L3 全链
- **`_build_api_section`**：`_mail` 的 `success_status = "done"`，其余 `success_status = "review"`
- **retry 逻辑**：mail 用 `MAIL_RETRY_PROMPT`，task 用标准 `RETRY_PROMPT`
- **`classify_outcome` 完成处理**：mail 走幻觉门控（hallucination gate）

### ticker.py 中的分支

- **虚拟项目扫描**：`_general` 和 `_mail` 两个虚拟项目各自硬编码扫描逻辑
- **mail 专属逻辑**：`_mail_check_reply` 判断邮件回复是否闭环

**问题总结**：新增第三种 task type（toolchain）如果继续硬编码，需要在三个文件中各加一套 if/else，且未来每种新类型都重复这个模式，维护成本线性增长。

---

# §2 三种 Task Type 行为差异表

| 维度 | task（普通任务） | mail | toolchain |
|------|---------|------|-----------|
| 存储 | 项目 DB（`projects/{pid}/blackboard.db`） | `_mail` DB（`_mail/blackboard.db`） | `_toolchain` DB |
| 状态流转 | pending→claimed→working→review→done | 跳过 claimed，auto-working→auto-done | auto-working→auto-done |
| prompt 构建 | BootstrapBuilder L0-L3 | MAIL_INFORM / MAIL_REQUEST 精简模板 | TOOLCHAIN 模板 + 事件上下文 |
| guardrail | 正常检查 | 跳过 | 跳过 |
| 完成标准 | 产出物 + review | 回复邮件 / inform done | Agent 处理完毕自动 done（无需回复）。闭环由 classify_outcome 判定：Agent 输出含实质行动（代码修复/分析结论）→ done；输出为空/幻觉 → failed。CI/Deploy 类事件最终闭环在 Gitea 侧（Agent push 修复后 CI 自动重跑），mozi 不跟踪 Gitea 侧最终结果。 |
| on_complete | classify_outcome → 状态机 | 幻觉门控 + 失败通知 | auto-done + 可选 escalate |
| 路由 | Router 四条快速路径 + 广播认领 | 直接路由到收件人 | 直接路由到事件相关 agent |
| retry | 标准 `RETRY_PROMPT` | `MAIL_RETRY_PROMPT` | 标准（或专用） |
| 前端展示 | 任务看板 Tab | 飞鸽传书 Tab | 待定 |

---

# §3 Handler 接口定义

定义 Python Protocol，所有 task type handler 必须满足此接口：

```python
from typing import Protocol, Optional, Dict, Any, runtime_checkable
from pathlib import Path


@runtime_checkable
class TaskTypeHandler(Protocol):
    """所有 task type handler 的统一接口。"""

    # 属性通过 __init__ 参数设置，Protocol 不强制 property
    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 列表。
        
        返回有序的 PromptSection 列表，由 PromptComposer 统一拼装。
        build_prompt 可使用此列表自动拼装，也可 override 做自定义（向后兼容）。
        """
        ...
```

### 基类提供统一流程（§16 详细定义）

```python
# 基类收敛共性能力，子类只实现差异点
@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:
    """详见 §16"""
    CRASH_OUTCOMES = frozenset({
        "crashed", "compact_failed", "process_crash",
        "session_stuck", "compact_hanging",
    })
    
    def post_complete(self, task_id, agent_id, outcome, db_path):
        """统一 4 步流程：crash → verify → mark → notify"""
        ...
    
    def verify_completion(self, task_id, db_path) -> VerifyResult:
        """子类必须实现"""
        ...
    
    def _rollback_current_agent(self, db_path, task_id, agent_id):
        """基类提供"""
        ...
    
    def on_failure(self, task_id, agent_id, db_path, verify):
        """子类可 override"""
        ...
```

**设计原则**：

- 每个方法在现有代码中都有明确的对应实现点，不存在"悬空"抽象
- `pre_spawn` 返回 Optional[callable]，普通任务返回 None 即可，不需要回调
- `spawner` 和 `dispatcher` 参数用 `Any` 类型，避免循环导入；handler 只调用已知方法

**兼容期过渡策略**：

引擎中 handler 查询优先，无 handler 时 fallback 到现有逻辑。具体流程：
1. `TaskTypeRegistry.get_by_project(project_id)` 查到 handler → 走 handler 路径
2. 未查到 → 走现有 if/else 分支（兼容期）
3. 三种 handler 全部稳定后，Step 5 删除旧分支，统一走 handler
4. 兼容期内新旧路径互斥——同一个 project_id 只会走其中一条，不存在重复执行

---

# §4 TaskTypeRegistry 注册表

```python
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

    @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
        ]
```

**使用方式**（daemon 启动时）：

```python
from task_handler import TaskHandler
from mail_handler import MailHandler
from toolchain_handler import ToolchainHandler

TaskTypeRegistry.register(TaskHandler())
TaskTypeRegistry.register(MailHandler())
TaskTypeRegistry.register(ToolchainHandler())
```

---

# §5 三个 Handler 的实现边界

> 三个 handler 都继承 BaseTaskHandler（§16），共性由基类提供，差异收敛到以下方法：
> - `verify_completion`：各自的验证逻辑
> - `target_success_status`：done 或 review
> - `on_failure`：各自的失败处理
> - `pre_spawn`：各自的前置准备

## TaskHandler（普通任务）

将现有 default（非 mail）分支封装为 handler，**不替代 BootstrapBuilder**。

| 方法 | 实现 |
|------|------|
| `task_type` | `"task"` |
| `virtual_project` | `None` |
| `build_prompt` | 调用 `BootstrapBuilder`（透传参数） |
| `build_api_section` | 现有 default 逻辑，`success_status = "review"` |
| `skip_guardrail` | `False` |
| `pre_spawn` | `None`（不需要回调） |
| `post_complete` | `classify_outcome` → 状态机 |
| `check_completion` | `False`（由状态机管理） |
| `build_retry_prompt` | 标准 `RETRY_PROMPT` |

**改动量**：~60 行，从 spawner/dispatcher 的 default 分支包一层。

## MailHandler

将分散在 dispatcher / spawner / ticker 三处的 mail 逻辑集中到一个 handler。

| 方法 | 实现 |
|------|------|
| `task_type` | `"mail"` |
| `virtual_project` | `"_mail"` |
| `build_prompt` | 复用 `_build_mail_prompt` / `MAIL_INFORM_TEMPLATE` / `MAIL_REQUEST_TEMPLATE` |
| `build_api_section` | `success_status = "done"` |
| `skip_guardrail` | `True` |
| `pre_spawn` | auto_working 回调（从 dispatcher `_mail_on_checks_passed` 搬入） |
| `post_complete` | 幻觉门控 + auto_done + 失败通知（从 dispatcher `_mail_auto_complete` 搬入） |
| `check_completion` | `_mail_check_reply`（从 ticker 搬入） |
| `build_retry_prompt` | `MAIL_RETRY_PROMPT` |

**改动量**：~150 行，三处逻辑集中。

## ToolchainHandler

全新 handler，处理 Gitea Webhook 事件通知。

| 方法 | 实现 |
|------|------|
| `task_type` | `"toolchain"` |
| `virtual_project` | `"_toolchain"` |
| `build_prompt` | 新建 `TOOLCHAIN_TEMPLATE` + 事件上下文注入 |
| `build_api_section` | `success_status = "done"` |
| `skip_guardrail` | `True` |
| `pre_spawn` | auto_working 回调 |
| `post_complete` | auto-done。escalate 触发条件：Agent 输出包含"需要人工介入"标记（如 block/无法修复/权限不足），或连续 retry 失败达到 max_retries（默认 2 次）。escalate 方式：通过 Mail 通知项目负责人（pangtong-fujunshi） |
| `check_completion` | `False`（由 post_complete 中 classify_outcome 判定，不需要额外的完成检查） |
| `build_retry_prompt` | 标准 `RETRY_PROMPT`（或后续定制） |

**改动量**：~100 行，全新代码。

---

# §6 引擎改动点（一次性）

引擎代码改动是受控的、一次性的。改完后新增 task type 不再触碰引擎。

## dispatcher.py（~20 行改动）

```python
# dispatch() 中，替换现有的 is_mail 分支
handler = TaskTypeRegistry.get_by_project(project_id)

if handler:
    if handler.skip_guardrail(project_id):
        # 跳过 guardrail，直接 spawn
        ...
    on_passed = handler.pre_spawn(task_id, db_path, self)
    # spawn 后由 on_passed 回调驱动后续
    ...
    handler.post_complete(task_id, agent_id, outcome, db_path, must_haves, self)
else:
    # 兼容期：保留现有 default 逻辑
    ...
```

## spawner.py（~15 行改动）

```python
# _build_prompt() 中
handler = TaskTypeRegistry.get_by_project(project_id)
if handler:
    return handler.build_prompt(
        task_id, title, description, must_haves,
        project_id, agent_id, task, spawn_type, self
    )
# else: 现有 BootstrapBuilder 逻辑（兼容期）

# _build_api_section() 中
handler = TaskTypeRegistry.get_by_project(project_id)
if handler:
    return handler.build_api_section(project_id, task_id, agent_id)

# retry 逻辑中
handler = TaskTypeRegistry.get_by_project(project_id)
if handler:
    return handler.build_retry_prompt(
        task_id, agent_id, retry_count, max_retries,
        retry_field, task_info, self
    )
```

**classify_outcome 完成处理**：
- spawner.py 第 1102 行 `is_mail` 分支在 `classify_outcome` 后触发幻觉门控
- 迁移到 MailHandler.post_complete 后，此处的 `is_mail` 判断改为查注册表：
  ```python
  handler = TaskTypeRegistry.get_by_project(project_id)
  if handler:
      handler.post_complete(task_id, agent_id, outcome, db_path, must_haves, self)
  else:
      # 兼容期：现有 classify_outcome 逻辑
      ...
  ```

## ticker.py（~10 行改动）

```python
# 虚拟项目扫描：从注册表自动发现
for vp in TaskTypeRegistry.virtual_projects():
    handler = TaskTypeRegistry.get_by_project(vp)
    if handler and handler.check_completion:
        # 调用 handler.check_completion 检查
        ...

# 保留 _general 硬编码作为 fallback（非 task type 机制）
```

---

# §7 实施顺序

按风险从低到高排列，每步完成后跑 `pytest -m "not e2e"` 全量回归测试。

### Step 1：注册表 + PromptComposer + BaseTaskHandler 基础设施

- 新建 `src/daemon/task_type_registry.py`：`TaskTypeHandler` Protocol + `TaskTypeRegistry`
- 新建 `src/daemon/prompt_composer.py`：`PromptSection` Protocol + `PromptContext` + `PromptComposer`
- 新建 `src/daemon/base_task_handler.py`：`BaseTaskHandler` 基类（VerifyResult + post_complete 统一流程 + _rollback_current_agent）[将在 Step 2 PR 实现]
- 编写单元测试验证：注册/查询、section 排序/去重/条件过滤、基类 post_complete 流程
- **风险**：极低，纯新增文件，不改动现有代码

### Step 2：TaskHandler（继承 BaseTaskHandler）

- 新建 `src/daemon/task_handler.py`，继承 `BaseTaskHandler`
- 实现 5 个 section（TaskContext / PriorOutputs / RoleSkill / TaskApi / TaskConstraints）
- `build_prompt` 内部用 PromptComposer 拼装
- 实现 `verify_completion`（三信号检查）和 review 分支
- 注册到 TaskTypeRegistry
- 运行全量回归测试，验证普通任务路径不变
- **风险**：低，现有 BootstrapBuilder 逻辑包一层

### Step 3：ToolchainHandler（继承 BaseTaskHandler）

- 新建 `src/daemon/toolchain_handler.py`，继承 `BaseTaskHandler`
- 实现 3 个 section（ToolchainContext / ToolchainApi / ToolchainConstraints）
- 实现 `verify_completion`（行动输出检查）和 `on_failure`（通知主公）
- 注册到 TaskTypeRegistry
- **风险**：低，全新代码

### Step 4：MailHandler（继承 BaseTaskHandler，含 crash rollback 修复）

- 新建 `src/daemon/mail_handler.py`，继承 `BaseTaskHandler`
- 实现 3 个 section（MailContext / MailApi / MailConstraints）
- 从 dispatcher / spawner / ticker 三处迁移 mail 逻辑
- 实现 `verify_completion`（回复检查 + inform/request 区分）
- **补上 crash rollback**（当前缺失，是 bug）
- 注册到 TaskTypeRegistry
- 重点回归测试 mail 路径：发送、回复、重试、幻觉门控
- **风险**：中

### Step 5：引擎清理

- 删除 dispatcher / spawner / ticker 中的旧 if/else 分支
- 删除或保留 BootstrapBuilder（作为 TaskContextSection 的内部实现）
- 全量回归测试 + 手工验证
- **风险**：中，删除代码需确保无遗漏

---

# §8 未来新增 Task Type 的步骤

以新增 `cron`（定时任务）为例：

1. **新建 handler 文件**：`src/daemon/cron_handler.py`
2. **实现 `TaskTypeHandler` 接口**：填入每个方法的具体逻辑
3. **注册**：在 daemon 启动时 `TaskTypeRegistry.register(CronHandler())`
4. **完事**。dispatcher / spawner / ticker 不改一行。

可选：
- 如需专用 API，在 `api/` 下新增路由
- 如需前端展示，在对应 Tab 里加渲染模板

---

# §9 前后端接口联动方案

### 后端 API 设计原则

- 每个 task type 可以有自己专用的 API（如 `/api/mail`、`/api/toolchain`）
- 也可以用统一 API 查询：`GET /api/tasks?type=mail|toolchain|task`
- 新增 task type 时，API 层自由选择：复用统一接口或新增专用接口
- 统一数据协议：`{id, type, status, title, from, to, metadata}`

### 前端展示原则

- 前端 Tab 和后端 task type 解耦——一个 Tab 可以展示多种 type，一种 type 也可以出现在多个 Tab
- 新增 task type 不需要新增前端组件，只需在对应 Tab 里加渲染模板
- 前端通过 `type` 字段区分渲染逻辑

### 本次范围

**前端不动**，后端架构先行。ToolchainHandler 的前端展示方案在后续迭代中确定。

---

# §10 风险评估

| 风险 | 概率 | 影响 | 缓解措施 |
|------|------|------|----------|
| MailHandler 迁移遗漏逻辑 | 中 | 高 | 逐步迁移，每步对比迁移前后全量测试；保留旧代码注释期 |
| 注册表查询性能 | 低 | 低 | dict 查询 O(1)，handler 数量个位数，忽略不计 |
| ticker 自动发现虚拟项目引入 bug | 低 | 中 | 保留 `_general` 硬编码作为 fallback；`virtual_projects()` 仅返回显式注册的虚拟项目 |
| 并发安全 | 低 | 高 | 注册表启动时一次性加载，运行时只读，无线程安全问题 |
| Handler 接口设计不足 | 低 | 中 | Protocol 可后续扩展方法，默认实现提供合理 fallback |
| 引擎清理删除过早 | 中 | 中 | Step 5 放在最后，确认所有 handler 稳定后再删旧代码 |

---

# §11 Prompt 构建现状分析

当前系统的 prompt 构建有四条独立路径，散落在三个文件中：

| 路径 | 位置 | 结构 | 服务对象 |
|------|------|------|----------|
| BootstrapBuilder 4 段 | `bootstrap.py` | 任务上下文 → 前序产出 → 角色规范(Skill全文) → 硬约束 | 普通任务 |
| Mail inform 模板 | `spawner.py` MAIL_INFORM_TEMPLATE | from/to/title/text 纯文本 | mail 通知 |
| Mail request 模板 | `spawner.py` MAIL_REQUEST_TEMPLATE | from/to/title/text + API 指令 | mail 请求 |
| Toolchain 模板 | `toolchain_templates.py` + `templates/toolchain/*.md` | 5 个 md 文件占位符渲染 | 工具链事件 |

此外 `_build_api_section` 在 spawner 中为所有路径拼 API 操作指令，但 success_status 不同（mail="done"，其他="review"）。

**问题**：

1. **四条路径完全独立**，共性逻辑（API 指令、约束注入）重复实现
2. **新增 task type 需要理解所有路径**才能加入
3. **没有统一的 token 预算管理**
4. **段的顺序硬编码**在各自的拼装逻辑中

---

# §12 PromptSection 模式

基于知识库优秀实践（Hermes 10层有序注入、Microsoft 三层中间件），引入 PromptSection 模式。

> 五层架构定义、L1-L4 去重规则、层间引导详见 **§14**。

priority 范围与 L2 注入组件的对应关系：

| priority | L2 组件 | 说明 |
|----------|---------|------|
| 10-19 | ③ 任务上下文 | 做什么 |
| 20-29 | ④ 前序信息 | 之前做了什么 |
| 30-39 | ① 操作规范 | 怎么做（Skill 索引或全文，handler 决定） |
| 40-49 | ① 操作规范（API 部分） | 怎么回写 |
| 50-59 | ⑤ 约束 + Guardrail | 不能做什么 |
| 60-69 | ⑥⑦ 审查协议/经验 | 扩展 |

## 核心思想

把 prompt 拆成有序的 section 列表，handler 声明自己需要哪些 section，PromptComposer 统一拼装。

## PromptSection 协议

```python
class PromptSection(Protocol):
    """一个 prompt 段"""
    name: str                    # 段名（去重用，同名覆盖）
    priority: int                # 排序优先级（小数字=靠前）

    def render(self, context: PromptContext) -> str:
        """渲染此段的文本内容

        Args:
            context: 包含 task_id, title, description, must_haves,
                     project_id, agent_id, task, role 等信息的上下文对象
        Returns:
            此段的文本内容（可以为空字符串表示不注入）
        """
        ...

    def should_include(self, context: PromptContext) -> bool:
        """是否注入此段（默认 True，条件段可覆盖）

        例如：前序产出段只在有 depends_on 时注入
        """
        return True
```

## PromptContext 数据对象

```python
@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 拼装器

```python
class PromptComposer:
    """有序拼装 prompt sections"""

    def __init__(self):
        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. 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()]  # 过滤空段
        return "\n\n---\n\n".join(parts)
```

## Section 优先级约定

| 优先级范围 | 用途 | 示例 |
|------------|------|------|
| 10-19 | 任务上下文 | 任务标题/描述、Mail 内容、Toolchain 事件 |
| 20-29 | 前序信息 | depends_on 产出、handoff comment |
| 30-39 | 角色规范 | Skill 全文注入、工具链行为指引 |
| 40-49 | API 操作指令 | 状态回写、curl 示例 |
| 50-59 | 硬约束 | 安全红线、禁止行为 |
| 60-69 | 扩展段 | 保留给未来使用 |

---

# §13 三个 Handler 的 Section 注册

每个 handler 通过 `get_sections()` 声明自己需要的 section 列表。

## TaskHandler sections

```python
def get_sections(self) -> list[PromptSection]:
    return [
        TaskContextSection(priority=10),       # BootstrapBuilder 段 1
        PriorOutputsSection(priority=20),      # BootstrapBuilder 段 2（有 depends_on 时）
        RoleSkillSection(priority=30),         # BootstrapBuilder 段 3（Skill 全文）
        TaskApiSection(priority=40),           # API 操作指令，success_status="review"
        TaskConstraintsSection(priority=50),   # 硬约束
    ]
```

| Section | 来源 | 共性/个性 |
|---------|------|------------|
| TaskContextSection | BootstrapBuilder 段 1 | 个性：title/desc/must_haves 格式 |
| PriorOutputsSection | BootstrapBuilder 段 2 | 个性：只有 task 有 depends_on |
| RoleSkillSection | BootstrapBuilder 段 3 | 个性：只有 task 读 Skill 全文 |
| TaskApiSection | spawner `_build_api_section` | **共性基础 + 个性参数**（success_status） |
| TaskConstraintsSection | BootstrapBuilder 段 4 | 个性：每种 task 约束不同 |

## MailHandler sections

```python
def get_sections(self) -> list[PromptSection]:
    return [
        MailContextSection(priority=10),       # from/to/title/text，区分 inform/request
        MailApiSection(priority=40),           # API 操作指令，success_status="done"
        MailConstraintsSection(priority=50),   # 硬约束（禁止状态转换命令等）
    ]
```

| Section | 来源 | 共性/个性 |
|---------|------|------------|
| MailContextSection | MAIL_INFORM_TEMPLATE / MAIL_REQUEST_TEMPLATE | 个性：邮件格式 |
| MailApiSection | spawner `_build_api_section` 变体 | **共性基础 + 个性参数**（success_status="done"，含 Mail API 指令） |
| MailConstraintsSection | 模板中的 ⚠️ 约束 | 个性 |

## ToolchainHandler sections

```python
def get_sections(self) -> list[PromptSection]:
    return [
        ToolchainContextSection(priority=10),  # 事件类型 + 事件详情
        ToolchainApiSection(priority=40),      # API 操作指令，success_status="done"
        ToolchainConstraintsSection(priority=50),  # 硬约束
    ]
```

| Section | 来源 | 共性/个性 |
|---------|------|------------|
| ToolchainContextSection | toolchain_templates.py + md 文件 | 个性：事件格式 |
| ToolchainApiSection | spawner `_build_api_section` 变体 | **共性基础 + 个性参数** |
| ToolchainConstraintsSection | 新增 | 个性 |

## Section 复用分析

| Section | task | mail | toolchain | 是否可复用 |
|---------|------|------|-----------|-------------|
| *ContextSection | ✅ | ✅ | ✅ | ❌ 格式完全不同，各自实现 |
| *ApiSection | ✅ | ✅ | ✅ | ⚠️ 基础框架可复用（BaseApiSection），success_status 参数化 |
| *ConstraintsSection | ✅ | ✅ | ✅ | ❌ 约束内容不同，各自实现 |
| PriorOutputsSection | ✅ | ❌ | ❌ | 仅 task |
| RoleSkillSection | ✅ | ❌ | ❌ | 仅 task |

**结论**：ApiSection 可以抽一个 BaseApiSection（curl 模板 + success_status 参数），其余 section 各自实现。

---

## 附录：文件结构预览

```
src/daemon/
├── task_type_registry.py    # §3 + §4：Protocol + Registry
├── prompt_composer.py        # §12 PromptSection + PromptContext + PromptComposer
├── base_task_handler.py      # §16 BaseTaskHandler 基类
├── task_handler.py           # §13 TaskHandler（继承 BaseTaskHandler）+ 5 sections
├── mail_handler.py           # §13 MailHandler（继承 BaseTaskHandler）+ 3 sections
├── toolchain_handler.py      # §13 ToolchainHandler（继承 BaseTaskHandler）+ 3 sections
├── dispatcher.py             # §6 改动
├── spawner.py                # §6 改动
├── ticker.py                 # §6 改动
├── bootstrap.py              # 保留，TaskContextSection 内部调用
└── toolchain_templates.py    # 保留，ToolchainContextSection 内部调用
```

---

# §14 上下文五层架构统一

五层定义（统一设计语言）：

| 层 | 名称 | 机制 | 内容示例 | token |
|---|------|------|---------|-------|
| L0 | 铁律层 | Hook 每轮强制注入 | GATE 铁律、Delegation 铁律 | ~500 |
| L1 | 角色层 | Workspace 自动注入 | SOUL.md、AGENTS.md、TOOLS.md、MEMORY.md | ~2000 |
| L2 | 引擎注入层 | PromptComposer 按 handler 拼装 | 任务上下文、前序产出、角色规范、API 指令、约束 | ~1500 |
| L3 | 被动参考层 | Skills 索引注入，Agent 按需 read 全文 | OpenClaw 42 Skills + moziplus SkillRegistry | 按需 |
| L4 | 检索层 | Agent 运行时主动检索 | wiki 知识库、NAS 文档、Web 搜索 | 按需 |

> **注**：表中 token 数为估算值。`prompt_composer.py` 中 `TOKEN_BUDGET_WARN=800` 是警告阈值（超出时打日志但不截断），与表中 ~1500 估算值不矛盾——800 对应单个 handler 的典型 L2 注入量，1500 对应最复杂场景（TaskHandler 全量 5 sections）的估算上限。

**PromptComposer 是 L2 层的拼装机制**。

## L1-L4 去重规则

当前 L1 和 L2 存在重叠（Agent 身份两处注入、API 操作指令两处注入、状态流转规则两处注入）。重构后：

| 信息 | 唯一归属 | 其他层怎么处理 |
|------|---------|--------------|
| Agent 身份 | L1 | L2 删除 `_inject_agent_identity` |
| 团队协作规则 | L1 | L2 不重复 |
| API 操作方法 | L2（任务级精简版） | L1 保留黑板概述，L2 只给本次任务的 curl |
| Skill 全文 | L3（Agent 按需 read） | L2 只给索引+引导语，不注入全文 |
| 状态流转规则 | L1（完整版） | L2 只给 success_status（done/review） |
| 安全红线 | L0 | L2 不重复 |
| 任务上下文 | L2 | L1 不涉及 |

## 层间引导

每层只做自己的事，通过层间引导语串联：

- L2 prompt 末尾追加引导语：
  - “需要详细操作规范？用 `read` 读取对应 Skill 文件”（引导到 L3）
  - “需要更多知识？查看 wiki 知识库或 Web 搜索”（引导到 L4）

---

# §15 Spawner/Handler 职责边界

## Spawner 职责（进程管理层）

| 职责 | 说明 |
|------|------|
| 进程启动/监控 | spawn subprocess、monitor stdout/stderr |
| 进程退出分类 | `_classify_outcome`（A0-A17 全在 spawner） |
| 重试决策 | `should_retry` + `_do_retry` + cooldown |
| counter 管理 | acquire/release/cooldown |
| attempt 记录 | `_record_attempt` |

## Handler 职责（业务调度层）

| 职责 | 说明 |
|------|------|
| prompt 构建 | 通过 PromptComposer 拼 section |
| pre_spawn 业务准备 | auto_working 等 |
| crash 回退 | rollback current_agent |
| 完成验证 | verify_completion |
| 状态标记 | mark success/failed |
| 失败通知 | notify_failure |

## 关键边界

1. **Spawner 不做业务逻辑**：`_build_mail_prompt` 和 `_build_api_section` 迁移到 handler 后，spawner 不再构建 prompt
2. **Handler 不碰进程管理**：handler 不做 exit 分类、不做 retry 决策、不管 counter
3. **状态标记不冲突**：spawner 的 `_mark_task` 处理进程级异常（crash/auth_failed/api_error → failed），handler 的 `mark_task_status` 处理业务级完成（done/review/failed）。两者操作不同 outcome 场景，互斥不重复
4. **on_complete 是桥梁**：spawner 完成进程级处理后调 `on_complete(outcome)`，handler 收到 outcome 做业务级处理

---

# §16 BaseTaskHandler 基类设计

## 设计原则

基类收敛**合理的共性能力**，不是现有代码的归类总结。参考：
- Hermes: "Keep calling tools until complete AND verified"
- Quality Gate: 三阶段门控（机械→语义→共识）
- Edict: stalled→retry→escalate 升级策略
- OpenAI Agents SDK: Input/Output Guardrail

## 基类定义

```python
@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]
    
    def build_prompt(self, context: PromptContext) -> str:
        """构建 L2 prompt（通过 PromptComposer 拼 section）"""
        ...
    
    def verify_completion(self, task_id: str, db_path: Path) -> VerifyResult:
        """验证任务完成质量。每个 handler 自己的验证逻辑。"""
        ...
    
    def target_success_status(self) -> str:
        """验证通过后的目标状态。task='review', mail/toolchain='done'"""
        return "review"
    
    # === 基类提供统一流程 ===
    
    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 → 失败时通知
        
        spawner 已完成进程级处理（exit 分类、重试、counter release）。
        这里只做业务级处理。
        """
        # 1. crash 处理（基类提供，所有 handler 继承）
        if outcome in self.CRASH_OUTCOMES:
            self._rollback_current_agent(db_path, task_id, agent_id)
            return  # crash 不进 verify，不标状态
        
        # 2. verify
        result = self.verify_completion(task_id, db_path)
        
        # 3. mark
        if result.passed:
            mark_task_status(db_path, task_id, self.target_success_status())
        else:
            # 4. notify（on_failure 内部处理）
            self.on_failure(task_id, agent_id, db_path, result)
    
    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 迁移。"""
        ...
    
    def on_failure(self, task_id: str, agent_id: str,
                   db_path: Path, verify: VerifyResult) -> None:
        """验证失败处理。默认：标 failed。
        子类可 override 加通知等。"""
        mark_task_status(db_path, task_id, "failed")
    
    def check_completion(self, task_id: str, db_path: Path) -> bool:
        """ticker 级别的完成检查。默认：False。"""
        return False
```

## 为什么删掉了这些方法

| 删除的方法 | 原因 |
|-----------|------|
| `skip_guardrail` | guardrail 是系统级安全层，不该由 handler 开关。guardrail 规则自己判断 project_id 是否跳过 |
| `crash_rollback`（独立方法） | 合并到 post_complete 第一步，不需要独立方法 |
| `handle_failure` / `notify_failure`（独立方法） | 合并为 `on_failure`，子类 override 一个方法即可 |
| `build_retry_prompt` | retry 是 spawner 层的职责，handler 不管重试 |

## 为什么 verify_completion 是每个 handler 必须实现的

参考 Hermes 的 "Keep calling tools until complete AND verified"——验证不是可选的，是完成流程的核心环节。每个 handler 的验证逻辑不同（task 看三信号、mail 看回复、toolchain 看行动输出），但**必须验证**这个要求是共性的。

---

# §17 三个 Handler 的完整执行流程

## 统一流程骨架

```
ticker 扫描 → dispatcher.decide → 路由到 agent
  │
  ▼
handler.pre_spawn(task_id, db_path)
  │  task: return True（无准备）
  │  mail/toolchain: auto_working（pending → working）
  ▼
spawner.spawn_full_agent()
  ├── counter acquire
  ├── handler.build_prompt(context)  ← L2 prompt 拼装
  ├── subprocess 启动 Agent 进程
  ├── monitor
  │
  ▼ （Agent 进程退出）
spawner._handle_exit()
  ├── _classify_outcome → outcome
  ├── should_retry=True → _do_retry（spawner 自己处理，不调 handler）
  └── should_retry=False → on_complete(outcome)
  │
  ▼
handler.post_complete(task_id, agent_id, outcome, db_path)
  ├── 1. crash? → rollback current_agent → return
  ├── 2. verify_completion → VerifyResult
  ├── 3. passed? → mark target_success_status()
  └── 4. failed? → on_failure()
```

## TaskHandler 执行流程

| 阶段 | 动作 | 代码来源 |
|------|------|----------|
| pre_spawn | return True | — |
| build_prompt | PromptComposer 拼 5 个 section | BootstrapBuilder |
| post_complete | 见下方 | dispatcher._task_on_complete |
| 1. crash | rollback current_agent | dispatcher._rollback_current_agent |
| 2. verify | 三信号检查（output_count > 0 OR comment_count > 0 OR status 已终态） | dispatcher._task_verify_completion |
| 3. passed → mark | "review" | dispatcher._task_auto_complete |
| 3. failed → on_failure | 留 working（等 ticker 重投） | 当前行为保持 |

**Task 特殊逻辑**：review 阶段的 on_complete 需要读 verdict → approved 标 done / 非 approved @mention assignee。这是 TaskHandler 的 review 分支，不走 verify 流程。

## MailHandler 执行流程

| 阶段 | 动作 | 代码来源 |
|------|------|----------|
| pre_spawn | auto_working（pending → working） | dispatcher._mail_auto_working |
| build_prompt | PromptComposer 拼 3 个 section | spawner._build_mail_prompt |
| post_complete | 见下方 | dispatcher._mail_on_complete |
| 1. crash | rollback current_agent（**补上**） | 新增 |
| 2. verify | 区分 inform/request：request 检查是否回复，inform 检查 outcome | dispatcher._mail_auto_complete |
| 3. passed → mark | "done" | dispatcher._mail_auto_complete |
| 3. failed → on_failure | mark "failed" + Mail 通知发件人 | dispatcher._mail_auto_complete |

**Mail 修复项**：当前 mail crash 时不做 rollback current_agent，可能导致 exclude_current 卡死。重构后补上。

## ToolchainHandler 执行流程

| 阶段 | 动作 | 代码来源 |
|------|------|----------|
| pre_spawn | auto_working（pending → working） | 新增 |
| build_prompt | PromptComposer 拼 3 个 section | toolchain_templates.py |
| post_complete | 见下方 | 新增 |
| 1. crash | rollback current_agent | 新增 |
| 2. verify | 检查行动输出（output 或 comment 有实质内容） | 新增 |
| 3. passed → mark | "done" | 新增 |
| 3. failed → on_failure | mark "failed" + Mail 通知主公 | 新增 |

## 三个 handler 差异收敛表

| 差异点 | TaskHandler | MailHandler | ToolchainHandler |
|--------|------------|-------------|------------------|
| pre_spawn | 无 | auto_working | auto_working |
| sections 数量 | 5 | 3 | 3 |
| verify 逻辑 | 三信号检查 | 回复检查 + inform/request 区分 | 行动输出检查 |
| target_success_status | review | done | done |
| on_failure | 留 working | 标 failed + 通知发件人 | 标 failed + 通知主公 |
| review 分支 | 有（读 verdict） | 无 | 无 |

---

# §18 设计决策记录

本节记录设计过程中的关键讨论和决策，便于未来回顾。

## D1: 方案A（独立 task type）vs 方案B（mail 内子分支）

**决策**：方案A，独立 task type。

**理由**：
- toolchain 和 mail 的行为差异越来越大
- 方案A 数据隔离、生命周期独立、未来演进互不影响
- 改动量和方案B差不多，但架构语义更好
- 主公明确表示"不想修修补补"

## D2: 设计一步到位，实现分步

**决策**：PromptSection 模式 + BaseTaskHandler 基类 + 五层架构统一都在设计文档中完整定义，但实施按 5 步渐进。

**理由**：避免设计时偷懒、实现时痛苦。设计完整后实施每步有清晰目标。

## D3: 三种 handler 不是“子集”关系

**讨论**：最初认为 MailHandler/ToolchainHandler 是 TaskHandler 的子集。

**结论**：三种 handler 走相同的流程骨架（Protocol 定义），但每一步的实现各自不同。TaskHandler 的 prompt 最复杂（5 sections），但 MailHandler 有独特的幻觉门控和回复检查。差异是真实的，不是简单的“全”和“子集”。

## D4: 幻觉门控和 verify 应该所有 handler 都有

**发现**：当前只有 mail 有幻觉门控、只有 task 有三信号验证。实际这是所有 handler 都应该有的核心能力。

**决策**：verify_completion 成为 BaseTaskHandler 的抽象方法，所有 handler 必须实现。

## D5: crash_rollback 放在 handler 基类

**讨论**：crash 处理分散在 spawner（进程级 cooldown）和 dispatcher（业务级 rollback current_agent）。

**结论**：
- spawner 管进程级：cooldown、counter release
- handler 管业务级：rollback current_agent
- 放在 BaseTaskHandler.post_complete 第一步，所有 handler 都继承，不遗漏
- 当前 mail 缺少 crash rollback，是 bug，重构后补上

## D6: skip_guardrail 从 handler 接口删除

**理由**：guardrail 是系统级安全层，不该由 handler 开关。guardrail 规则自己判断 project_id 是否跳过。handler 不需要知道 guardrail 的存在。

## D7: spawner 的 prompt 构建迁移到 handler

**讨论**：当前 `_build_mail_prompt` 和 `_build_api_section` 在 spawner 中，按职责应该归 handler。

**结论**：handler 的 build_prompt 通过 PromptComposer 拼 section，spawner 只负责传递 prompt 给 subprocess。spawner 不再做任何 prompt 构建逻辑。

## D8: L2 Skill 段最小化

**讨论**：当前 BootstrapBuilder 段3 注入 Skill 全文（~800 token），重复了 L3 层的职责。

**结论**：L2 的 RoleSkillSection 改为注入索引+引导语（~100 token），引导 Agent 用 `read` 去读 Skill 全文（L3 层）。遵循 Hermes 的渐进式 Skill 加载模式。

---
