Compare commits

..

1 Commits

Author SHA1 Message Date
cfdaily 85e2004e69 fix(17): redesign on_failure to use Gitea management instead of Mail API
- Rewrite §5.2: on_failure 三分路设计(业务失败→PR comment @assignee / 系统失败→Gitea Issue @pangtong / 基础设施失败→toolchain task @jiangwei)
- Update §7.4: _send_mail 不参与 toolchain 失败处理
- Update §11.1: 向后兼容表新增 on_failure 路径变更说明
- Add D17-8: 设计决策记录 on_failure 走 Gitea 不走 Mail
- Update §10: 影响范围新增 toolchain_handler.py on_failure 改动行

Closes: on_failure redesign per 'toolchain 事件全生命周期在 toolchain 流程内闭环'
2026-06-13 22:52:30 +08:00
9 changed files with 299 additions and 1443 deletions
+2 -15
View File
@@ -27,7 +27,6 @@ jobs:
- name: Setup Python
run: |
python3 -m venv /tmp/ci-venv-lint
/tmp/ci-venv-lint/bin/pip install --quiet --upgrade pip
/tmp/ci-venv-lint/bin/pip install --quiet flake8
- name: Lint with flake8
@@ -43,24 +42,12 @@ jobs:
- name: Setup Python
run: |
rm -rf /tmp/ci-venv-test
python3 -m venv /tmp/ci-venv-test
/tmp/ci-venv-test/bin/pip install --quiet --upgrade pip
/tmp/ci-venv-test/bin/pip install --quiet --no-cache-dir fastapi pydantic pyyaml uvicorn requests pytest pytest-asyncio httpx
- name: Debug environment
run: |
echo "PWD=$(pwd)"
echo "PYTHONPATH=$PYTHONPATH"
python3 -c "import sys; [print(p) for p in sys.path if 'sanguo' in p.lower() or 'openclaw' in p.lower()]"
grep -c "assignee = agent_id" src/daemon/toolchain_handler.py || true
grep -c "_BUSINESS_FAIL_THRESHOLD" src/daemon/toolchain_handler.py || true
/tmp/ci-venv-test/bin/pip install --quiet fastapi pydantic pyyaml uvicorn requests pytest pytest-asyncio httpx
- name: Run tests (exclude E2E)
run: |
PYTHONPATH=$(pwd) /tmp/ci-venv-test/bin/pytest tests/ -m "not e2e" -x -q || \
(echo '=== RETRY WITH VERBOSE ===' && \
PYTHONPATH=$(pwd) /tmp/ci-venv-test/bin/pytest tests/ -m "not e2e" -x -v 2>&1 | tail -30)
/tmp/ci-venv-test/bin/pytest tests/ -m "not e2e" -x -q
# ── Job 3: CI 失败通知 ───────────────────────────────
# 使用 needs.<job>.result 直接判断,不查询 commit status API
-181
View File
@@ -1,181 +0,0 @@
# #16 知识注入设计
> 状态:设计中
> 作者:庞统
> 日期:2026-06-13
> 评审:待司马懿评审
## 一、问题
### 1.1 现状
Agent(庞统、司马懿、张飞等)在执行任务时,不主动查询已有知识库(wiki-vault)。导致:
1. **重复调研**:赵云查过的数据清洗经验,张飞又从头调研一遍
2. **重复踩坑**wiki-vault 里已有"vnpy load_bar 需要显式指定 end=None"的实践,张飞还是踩了
3. **方案质量低**:做方案时纯靠推理,不查已有的优秀实践
4. **知识 gap 无人管**:查不到相关知识时没记录,下次还是查不到
### 1.2 根因
不是没有知识库(wiki-vault 有 50+ practices 页面),也不是没有检索能力(wiki-query Skill 已存在)。
**根因是注入时机**:Agent 不知道什么时候该查、没有强制机制让 Agent 在关键决策点查。
现有设计(#11 四层架构)把知识注入放在 L3-4(被动参考层),但没有定义"什么时候触发被动参考"。
### 1.3 目标
1. Agent 在关键决策点**主动查询** wiki-vault
2. 查不到相关知识时**自动记录** knowledge gap
3. 定时任务处理 gap + 总结经验,**持续丰富** wiki-vault
4. 不增加 prompt token 负担(不自动注入知识全文,只引导查询)
## 二、调研
### 2.1 Superpowers:强制 Skill 检查(最有效)
**核心设计**session-start hook 注入铁律级指令——
> "If you think there is even a **1% chance** a skill might apply, you **ABSOLUTELY MUST** invoke the skill. This is not negotiable."
配合 **Red Flags 表**防止 Agent 自合理化跳过:
| Agent 的想法 | Red Flag 驳回 |
|---|---|
| "这个问题很简单" | 简单问题也需要查实践 |
| "我需要更多上下文" | Skill 检查在澄清问题之前 |
| "先看看代码" | Skill 告诉你怎么看代码 |
| "我记住了这个 Skill" | Skill 会更新,重新读 |
**为什么有效**:不靠 Agent "想起来",靠铁律强制。Skill 触发在任何响应之前。
### 2.2 Hermes:经验闭环 + Session Search
**经验闭环**:完成复杂任务(5+ tool calls)→ 自动创建 Skill → 下次自然触发。
**Session Search**:系统提示注入——"当用户提及过去内容时,主动搜索而非要求用户重复"。
**为什么有效**:不是"知识查询"而是"行为内化"——经验变成 SkillSkill 有 description 触发词。
### 2.3 结论
综合两个项目的优势:
| 设计点 | 来源 | 我们的做法 |
|--------|------|-----------|
| 铁律级强制 | Superpowers | L0 Hook + SOUL.md 双重注入 |
| Red Flags 反合理化 | Superpowers | 知识查询 Red Flags 表 |
| 经验内化 | Hermes | 经验→wiki-vault→下次查询 |
| 渐进式披露 | Hermes | 先查 summary,按需读全文 |
## 三、设计决策
### D16-1:三层触发机制(核心设计)
不是在引擎层面自动注入知识全文(token 浪费),而是通过三层机制让 Agent **主动查询**
| 层级 | 机制 | 内容 | 覆盖范围 |
|------|------|------|---------|
| **L0 铁律** | SOUL.md 行为引导 | "做方案前先查 wiki-vault,有 1% 相关就要查" | 所有 Agent、所有场景 |
| **L1 Skill 路由** | wiki-query Skill description | 触发词:调查、研究、分析、优秀实践、经验、怎么做X | Agent 匹配到触发词时 |
| **L2 知识 gap 闭环** | 定时任务 | 查不到→记 gap→每天处理→写入 wiki-vault | 持续改进 |
**为什么不做 PromptComposer 自动注入**
1. 自动注入 token 浪费(每次任务都注入可能不相关的知识)
2. 覆盖范围有限(只影响 moziplus 子任务 Agent
3. Agent 主动查询更精准(知道自己缺什么知识)
### D16-2:知识查询 Red Flags(防止跳过)
在 SOUL.md 中加入 Red Flags 表,和 Superpowers 一致:
| Agent 的想法 | 反驳 |
|---|---|
| "这个我以前做过" | 知识库可能已更新,查一下确认 |
| "先做再说" | 做方案前查实践比做错了返工便宜 |
| "这个领域我熟悉" | 熟悉≠知道最新实践,wiki-vault 持续更新 |
| "查知识库浪费时间" | 重复踩坑浪费的时间远大于查询时间 |
### D16-3:知识 gap 记录机制
**触发**Agent 在 wiki-vault 中查不到相关知识时。
**记录位置**`/Volumes/KnowledgeBase/wiki-vault/_meta/knowledge-gaps.md`
**格式**(已有):
```markdown
- [日期] Agent名查"主题" → 待处理
```
**已有基础设施**
- knowledge-gaps.md 已存在,有 20+ 条历史记录
- 格式已定义,处理后会标注 `→ 已建立 ✅`
### D16-4:定时任务流程(已有 cron 基础)
| 任务 | 时间 | 内容 | 状态 |
|------|------|------|------|
| wiki-daily-update | 每天 04:00 | 处理 knowledge gaps + 当天经验总结 → 写入 wiki-vault | ✅ 已有 cron,需完善 |
| pangtong-vault-sync | 每天 05:00 | 同步 wiki-vault 到 agent workspace | ✅ 已有 |
**wiki-daily-update 完善内容**
1. 读取 knowledge-gaps.md 中"待处理"条目
2. 对每个 gap:搜索 knowledge_base 是否有相关源码/文档 → 有则提炼写入 wiki-vault
3. 搜索最近一天的 jsonl 日志,提取有价值的经验
4. 新建或更新 wiki-vault 页面
5. 更新 knowledge-gaps.md(标记为"已建立 ✅"或"无KB内容,跳过"
### D16-5wiki-vault 作为索引层
**确认原则**wiki-vault 是索引层,不是详细内容存储。
- 查询先走 wiki-vaultpractices/concepts/skills
- 如果页面指向 knowledge_base 的详细内容,**必须 follow** 获取原文
- knowledge_base 的路径:`/Volumes/KnowledgeBase/knowledge_base/`
### D16-6:和 #11 四层架构的关系
| 层级 | 知识注入角色 | 本文档覆盖 |
|------|------------|-----------|
| L0 铁律层 | "做方案前先查 wiki-vault" | ✅ D16-1 L0 |
| L1 身份层 | SOUL.md 中加入查询行为引导 | ✅ D16-1 L0 + D16-2 |
| L2 引擎注入层 | 不做自动注入(Agent 主动查询更精准) | — |
| L3 被动参考层 | wiki-query Skill 按需加载 | ✅ D16-1 L1 |
| L4 检索层 | wiki-vault grep/read + knowledge_base follow | ✅ D16-5 |
**和 #11 L3-4 的关系**#11 把知识注入放在 L3 被动参考层,本文档是 L3-4 的具体实现方案。
### D16-7:和 topic6 经验沉淀的关系
| | topic6moziplus 内部) | 本文档(知识注入) |
|---|---|---|
| **范围** | moziplus 引擎 spawn 的子任务 | 所有 Agent、所有场景 |
| **经验来源** | 黑板 decisions/reviews/comments | jsonl 日志 + knowledge gaps |
| **载体** | experiences 表(SQLite | wiki-vaultMarkdown |
| **关系** | topic6 的经验最终汇入 wiki-vault | wiki-vault 是知识的 single source of truth |
**topic6 不急**:因为 jsonl 日志中的经验也是知识来源之一,定时总结时会覆盖。topic6 可以在后续作为 moziplus 内部的额外经验来源。
## 四、改动清单
### 4.1 已完成 ✅
| 改动 | 文件 | 说明 |
|------|------|------|
| TOOLS.md 更新 | workspace-pangtong/TOOLS.md | 新增「LLM Wiki 知识库」段:速查表、检索原则、铁律 |
### 4.2 待实现
| 改动 | 文件 | 说明 |
|------|------|------|
| SOUL.md 行为引导 | workspace-pangtong/SOUL.md | 新增「知识查询」行为规则 + Red Flags |
| wiki-daily-update 完善 | cron job | gap 处理 + 经验总结逻辑 |
### 4.3 不做
| 项目 | 原因 |
|------|------|
| PromptComposer 知识注入 | token 浪费,Agent 主动查询更精准 |
| experiences 表 | wiki-vault 已覆盖,不重复建设 |
| 新 Skill | wiki-query 已有,不需要新的 |
+168 -74
View File
@@ -218,14 +218,9 @@ class ToolchainContextSection:
### 3. 不要执行任何状态转换命令
- 不要手动标 working/done/review/failed,系统会自动处理
### 4. 不需要回复
### 4. 不需要回复此邮件
- 和 request 类型不同:不需要 in_reply_to 回复
- action report 就是你的完成凭证
- 不要发送 Mail(飞鸽传书),你的所有操作在 toolchain 流程内完成
### 5. 所有协作通过 Gitea 完成
- 如果遇到问题需要其他角色支持,在关联的 PR/Issue 上创建 comment @对方
- 不要使用 Mail API(飞鸽传书)发送消息
- 你的所有操作都在 toolchain 流程内,通过 Gitea 留痕
### Red Flags(如果脑海中出现以下想法,说明你错了)
@@ -258,7 +253,7 @@ class ToolchainContextSection:
| 语气级别 | 用词 | 效果 | 使用场景 |
|---------|------|------|---------|
| **强制** | "必须"、"不可跳过"、"强制要求" | Agent 无法自合理化跳过 | 步骤执行、action report 提交 |
| **禁止** | "不要"、"违反会"、"failed" | Agent 不会越界 | 状态转换、发送 Mail |
| **禁止** | "不要"、"违反会"、"failed" | Agent 不会越界 | 状态转换、回复邮件 |
| **提醒** | "⚠️" | 视觉强调 | 关键约束前缀 |
**避免的用词**"建议"、"如需"、"可以考虑"、"参考"、"推荐"——这些词在 Agent 的推理中会被解读为"可选"。
@@ -296,19 +291,6 @@ curl -s -X POST "http://localhost:8083/api/projects/_toolchain/tasks/{task_id}/o
```
```
### 需要其他角色支持时
如果在执行过程中需要其他角色协助(如缺数据、需要审批等),在关联的 PR/Issue 上创建 comment @对方:
```bash
curl -s -X POST "http://192.168.2.154:3000/api/v1/repos/{repo}/issues/{pr_number}/comments" \
-H "Authorization: token <your-token>" \
-H "Content-Type: application/json" \
-d '{"body": "@{agent-id} 需要你的支持:{描述问题}"}'
```
⚠️ 不要使用 Mail API(飞鸽传书)。所有协作通过 Gitea 留痕。
**变化**:移除了"手动标 done"的 curl 示例(done 由 verify 自动处理),替换为 action report 提交指引。
---
@@ -371,7 +353,7 @@ def verify_completion(self, task_id: str, db_path: Path) -> VerifyResult:
保留 fallback 层次是为了平滑过渡:改造初期 Agent 可能还不习惯提交 action_reportfallback 避免"改造后所有 task 都 failed"的问题。
### 5.2 on_failure 处理
### 5.2 on_failure 分路处理
**设计原则**:toolchain 事件全生命周期在 toolchain 流程内闭环,不走 Mail API。失败本身按错误类型分类,路由到不同的 Gitea 管理动作。
@@ -380,18 +362,127 @@ def verify_completion(self, task_id: str, db_path: Path) -> VerifyResult:
| 错误类型 | 例子 | 处理方式 | 管道 |
|---------|------|---------|------|
| **业务失败** | verify 不过(no action_report)、Agent 忽略步骤 | 在关联 PR/Issue 上创建 comment @原始 assignee | Gitea webhook → §25 @mention → toolchain task |
| **系统失败** | spawner crash、timeout、max_retries、业务失败连续 3 次 | 创建 Gitea Issue 指派 pangtong-fujunshi | Gitea webhook → issue_assigned → toolchain task |
| **系统失败** | spawner crash、timeout、max_retries | 创建 Gitea Issue 指派 pangtong-fujunshibody 包含错误详情 + task ID + 日志摘要 | Gitea webhook → issue_assigned → toolchain task |
| **基础设施失败** | Gitea API 不可用、网络不通 | `_send_toolchain_task` 直接创建 toolchain task 指派 jiangwei-infra | toolchain 内部直接创建 |
三条路全在 toolchain 流程内,Mail 完全不参与。
#### 防递归保护
#### 业务失败处理(PR/Issue comment @assignee
基础设施失败创建的 toolchain taskaction_type=infrastructure_failure),其 verify_completion 始终返回 `VerifyResult(True)`,不再触发 on_failure。
verify 失败时,在原始事件关联的 Gitea PR 或 Issue 上创建 comment@原始 assignee
#### 完整设计
```python
def on_failure(self, task_id, agent_id, db_path, verify):
self._mark_task_status(db_path, task_id, "failed")
# 读取事件上下文
meta = self._read_task_meta(db_path, task_id)
failure_count = meta.get("failure_count", 0) + 1
# 分类路由
if verify.reason in SYSTEM_ERROR_REASONS:
# 系统失败 → Gitea Issue 给庞统
self._create_gitea_issue_for_system_error(task_id, meta, verify)
else:
# 业务失败 → PR/Issue comment @assignee
self._create_gitea_comment_for_business_failure(task_id, meta, verify, failure_count)
```
三分路的详细伪代码、失败上限、决策依据见 §5.2.1~§5.2.3on_failure 分路处理详细设计)。
**PR/Issue comment 内容**
```
@{assignee} 任务处理失败,需要你的支持
📋 原始事件:{event_type}
❌ 失败原因:{verify.reason}
📊 失败次数:第 {failure_count} 次
请检查并处理。完成后提交 action report。
```
**为什么 @mention 而非新建 task**
- @mention 通过 §25 webhook 自然触发新 toolchain task,不多建一条路
- Gitea comment 自带完整 PR/Issue 上下文,Agent 收到时能理解全貌
- 人也能在 Gitea 上看到失败记录,天然审计
**失败上限**:同一事件 failure_count ≥ 3 时,升级为系统失败(创建 Gitea Issue 给庞统),避免无限循环。
#### 系统失败处理(Gitea Issue 给庞统)
spawner 错误(crash/timeout/max_retries)或业务失败连续 3 次,创建 Gitea Issue
```python
def _create_gitea_issue_for_system_error(self, task_id, meta, verify):
"""创建 Gitea Issue 指派庞统排查"""
repo = meta.get("context", {}).get("repo", "sanguo/sanguo_moziplus_v2")
title = f"[toolchain] 系统错误排查: {meta.get('event_type', 'unknown')} ({task_id})"
body = (
f"## 系统错误\n\n"
f"**Task ID**: {task_id}\n"
f"**事件类型**: {meta.get('event_type', 'unknown')}\n"
f"**失败原因**: {verify.reason}\n"
f"**证据**: {verify.evidence}\n"
f"**原始 assignee**: {meta.get('context', {}).get('assignee', 'unknown')}\n\n"
f"## 排查方向\n"
f"- 检查 spawner 日志\n"
f"- 确认 Agent 是否正常运行\n"
f"- 检查是否有系统性问题(API down / 配置错误 / 资源不足)\n"
)
# Gitea API 创建 Issue,指派 pangtong-fujunshi
_create_gitea_issue(repo, title, body, assignee="pangtong-fujunshi")
```
**为什么用 Gitea Issue**
- 系统问题需要排查和 trackingIssue 有状态管理(open/closed
- Issue body 包含完整错误详情,便于事后 troubleshooting
- 指派庞统后通过 issue_assigned webhook 自然触发 toolchain task
#### 基础设施失败处理(toolchain task 给姜维)
Gitea API 不可用或网络不通时,无法创建 PR comment 或 Issue。此时降级为直接创建 toolchain task
```python
def _create_infrastructure_task(self, task_id, meta, verify, error_detail):
"""基础设施失败 → toolchain task 指派姜维"""
_send_toolchain_task(
to_agent="jiangwei-infra",
title=f"[基础设施] Gitea 不可用导致 on_failure 降级: {task_id}",
description=f"尝试处理 task {task_id} 失败时 Gitea API 不可用。\n错误: {error_detail}",
event_type="infrastructure_failure",
action_type="infrastructure_failure",
steps=[
"检查 Gitea 服务状态(http://192.168.2.154:3000",
"检查网络连通性",
"Gitea 恢复后检查原始 task 状态并补处理",
"提交 action report",
],
context_data={
"original_task_id": task_id,
"original_event_type": meta.get("event_type", ""),
"gitea_error": error_detail,
},
)
```
**为什么不走 Gitea**:Gitea 本身就是问题源,调 Gitea API 创建 Issue 会再次失败。直接在 _toolchain DB 内创建 task 是最后的降级手段。
#### 三分路总结
```
on_failure
├─ 系统错误(crash/timeout/max_retries/连续3次业务失败)?
│ └─ Gitea API 可用?
│ ├─ YES → 创建 Issue @pangtong-fujunshi → webhook → toolchain task
│ └─ NO → _send_toolchain_task @jiangwei-infra(基础设施降级)
└─ 业务错误(no action_report?
└─ Gitea API 可用?
├─ YES → PR/Issue comment @assignee → §25 webhook → toolchain task
└─ NO → _send_toolchain_task @jiangwei-infra(基础设施降级)
```
所有路径都在 toolchain 流程内闭环,Mail 不参与。
### 5.3 action_report comment 格式
@@ -435,9 +526,9 @@ Agent 可能写了 action_report 但没真做。缓解机制:
| Issue 指派 → 开发者 | issue_assigned | toolchain | 6 步 | 创建分支 + 编码 + push + CI + PR + report |
| 部署失败 → 运维 | deploy_failure | toolchain | 4 步 | 查日志 + 排查 + 修+重部署 + report |
| @mention → 被@者 | mention | toolchain | 按 guidance | 按 mention 模板的 response_guidance + report |
| PR 合并 → PR 作者 | review_merged | toolchain | 0 步 | 纯通知,走 _send_toolchain_tasksteps 为空,verify 始终通过) |
| PR 合并 → PR 作者 | review_merged | **mail (inform)** | — | 纯通知,走 _mail 路径 |
**D17-2: 所有 toolchain 场景走 ToolchainHandler**
**D17-2: 除 PR 合并通知外,所有 toolchain 场景走 ToolchainHandler**
### 6.2 各场景 steps 详细定义
@@ -554,28 +645,17 @@ context:
#### PR 合并 → PR 作者
```
event_type: review_merged
action_type: review_merged
steps: [] # 无步骤,纯通知
verify: 始终通过(inform 语义,无需 action_report
context:
pr_number, repo, pr_title, pr_author, merged_by
走 _mail 路径(inform),不走 toolchain。
理由:PR 已经合并,部署已自动触发,作者无需做任何事。纯 FYI 通知。
```
**特殊处理**review_merged 的 verify_completion 始终返回 `VerifyResult(True)`。这是唯一的纯通知场景,Agent 只需阅读即可。
### 6.3 PR 合并通知为何保持 inform
### 6.3 PR 合并通知为何也走 ToolchainHandler
- PR 已经被合并到 main
- 部署已自动触发(deploy workflow
- 作者无需做任何事
**设计原则**toolchain 和 Mail 完全分离。所有工具链事件(包括纯通知)都走 ToolchainHandler + _toolchain DB
**为什么 PR 合并通知不走 _mail**
- toolchain 事件的完整生命周期应在同一个 DB(_toolchain)中追溯
- 前端展示统一(Toolchain Tab),不需要跨 _mail 和 _toolchain 两个 Tab 查看同一类事件
- Mail 只服务 Agent 间点对点通信(inform / request),不服务工具链事件
**实现差异**review_merged 的 verify_completion 始终通过(`VerifyResult(True)`),不需要 action_report。这是 ToolchainHandler 内部的语义区分,不影响 MailHandler。
**spawn 说明**review_merged 仍会触发 spawnAgent 只需阅读通知),verify auto-pass 后标 done。未来可优化为 ticker 直接 auto-done 跳过 spawn。
这是真正的"FYI"通知,设为 inform 正确。继续走 `_send_mail` 函数,不受本设计影响
---
@@ -587,7 +667,7 @@ context:
| 函数 | 用途 | project_id | task_type | DB |
|------|------|-----------|-----------|-----|
| `_send_mail` | Agent 间点对点通信(inform / request | `_mail` | `mail` | `_mail/blackboard.db` |
| `_send_mail` | 纯通知(inform)和 Agent 间通信(request | `_mail` | `mail` | `_mail/blackboard.db` |
| `_send_toolchain_task` | 工具链动作事件(需 Agent 执行步骤) | `_toolchain` | `toolchain` | `_toolchain/blackboard.db` |
```python
@@ -709,20 +789,17 @@ async def _handle_pr_opened(payload: Dict[str, Any]) -> None:
| `_handle_pull_request_review` (COMMENTED) | review_comment | `_send_toolchain_task(...)` |
| `_handle_issue_comment` (CI failure) | ci_failure | `_send_toolchain_task(...)` |
| `_handle_issues` (assigned) | issue_assigned | `_send_toolchain_task(...)` |
| `_handle_pr_closed` (merged) | review_merged | `_send_toolchain_task(...)`(纯通知,steps 为空,verify 始终通过 |
| `_handle_pr_closed` (merged) | — | `_send_mail(...)` **不变**inform 纯通知 |
| `_send_deploy_failure_mail` | deploy_failure | `_send_toolchain_task(...)` |
| `_send_mention_mails` | mention | `_send_toolchain_task(...)` |
### 7.4 _send_mail 不参与任何 toolchain 事件
### 7.4 _send_mail 不参与 toolchain 失败处理
`_send_mail` 函数只服务个场景:**Agent 间点对点通信**inform / request)。
`_send_mail` 函数只服务个场景:
1. PR 合并通知(inform 纯通知)
2. Agent 间点对点通信(inform / request
**toolchain 事件完全不经过 Mail**
- 事件投递:`_send_toolchain_task`(不是 `_send_mail`
- 失败处理:Gitea PR comment / Gitea Issue / `_send_toolchain_task`(三分路)
- PR 合并通知:`_send_toolchain_task`review_mergedsteps 为空)
Mail 和 toolchain 是两套完全独立的系统,各自有独立的 DB、Handler、PromptSection。
**toolchain on_failure 不调用 `_send_mail`**。失败处理三分路(PR comment / Gitea Issue / toolchain task)全在 toolchain 流程内闭环。
---
@@ -752,7 +829,6 @@ if handler:
meta = json.loads(must_haves) if must_haves else {}
from_agent = meta.get("from", "")
mail_type = meta.get("performative", meta.get("type", ""))
# 注:toolchain task 的 mail_type 为空(不走 MailHandler),保留字段兼容 MailHandler
# 新增:toolchain 字段提取
event_type = meta.get("event_type", "")
@@ -830,21 +906,24 @@ ticker 需要扫描 `_toolchain` 虚拟项目。当前 ticker 通过 `TaskTypeRe
| 文件 | 改动类型 | 说明 |
|------|---------|------|
| `src/daemon/toolchain_handler.py` | 修改 | ToolchainContextSection 加 steps 渲染 + action_hintToolchainApiSection 改为 action_report 指引;ToolchainConstraintsSection 加 Red Flagsverify_completion 改用 action_report |
| `src/api/toolchain_routes.py` | 修改 | 新增 `_toolchain_db_path()` + `_send_toolchain_task()`所有 handler(含 PR merged改为调用 `_send_toolchain_task` |
| `src/api/toolchain_routes.py` | 修改 | 新增 `_toolchain_db_path()` + `_send_toolchain_task()` handler 改为调用 `_send_toolchain_task`PR merged 保持 `_send_mail` |
| `src/daemon/spawner.py` | 修改 | handler 路径 PromptContext 构造时提取 `action_type``action_steps` 字段 |
| `src/daemon/prompt_composer.py` | 修改 | PromptContext 新增 `action_type``action_steps` 字段 |
| `src/blackboard/db.py` | 修改 | comments 表 CHECK 约束处理(去掉 CHECK 或加 action_report |
| `src/daemon/mail_notify.py` | 修改 | `_REASON_MAP` 新增 `no_action_report` reason |
| `src/daemon/toolchain_handler.py`on_failure | 修改 | on_failure 三分路重写(去掉 Mail API 调用,改为 Gitea API + _send_toolchain_task | +~40 行 |
### 改动量估算
| 文件 | 改动量 | 风险 |
|------|--------|------|
| `src/daemon/toolchain_handler.py` | ~120 行 | 中(核心逻辑变化 + on_failure 三分路 |
| `src/daemon/toolchain_handler.py` | ~80 行 | 中(核心逻辑变化) |
| `src/api/toolchain_routes.py` | ~120 行 | 中(新增函数 + 8 个 handler 改造) |
| `src/daemon/spawner.py` | ~8 行 | 低(纯新增字段提取) |
| `src/daemon/prompt_composer.py` | ~3 行 | 低(dataclass 新增字段) |
| `src/blackboard/db.py` | ~5 行 | 低(CHECK 约束处理) |
| **总计** | **~256 行** | |
| `src/daemon/mail_notify.py` | ~2 行 | 低(新增一行 reason map |
| **总计** | **~218 行** | |
---
@@ -857,7 +936,8 @@ ticker 需要扫描 `_toolchain` 虚拟项目。当前 ticker 通过 `TaskTypeRe
| Agent 间手动发 inform Mail | ✅ 无影响 |
| Agent 间手动发 request Mail | ✅ 无影响 |
| MailHandler 的 verify / on_failure | ✅ 无影响 |
| `_send_mail` 函数 | ✅ 保留不变(只服务 Agent 间通信,不服务任何 toolchain 事件) |
| `_send_mail` 函数 | ✅ 保留不变 |
| ToolchainHandler on_failure 改为 Gitea 管理(PR comment / Issue / toolchain task),不经过 Mail API | ✅ 无影响 |
### 11.2 _mail DB 中已有的 toolchain task
@@ -899,15 +979,13 @@ ticker 需要扫描 `_toolchain` 虚拟项目。当前 ticker 通过 `TaskTypeRe
**理由**action_report comment 机制最简单且与现有架构一致。保留 fallback 确保平滑过渡。Agent 如果写了 report 但没执行,后续事件链(CI 不会通过、Reviewer 不会收到 Review)会自然暴露问题。
### D17-2: 所有 toolchain 场景走 ToolchainHandler
### D17-2: 除 PR 合并通知外,所有 toolchain 场景走 ToolchainHandler
**决策**全部 10 种 toolchain 场景走 ToolchainHandler`_send_toolchain_task`),包括 PR 合并通知(review_merged)。Mail 不服务任何 toolchain 事件
**决策**9 种 toolchain 场景中,8 种走 ToolchainHandler`_send_toolchain_task`),`review_merged` 走 MailHandler`_send_mail` + inform
**理由**
- toolchain 事件的完整生命周期应在同一个 DB(_toolchain)中追溯
- 前端展示统一(Toolchain Tab),不需要跨 _mail 和 _toolchain 两个 Tab
- Mail 只服务 Agent 间点对点通信(inform / request),不服务工具链事件
- review_merged 的 verify_completion 始终通过,不需要 action_report
- 8 种场景都需要 Agent 执行后续动作(修代码/审查/合并/排查/响应 mention
- PR 合并是真正的 FYI,无需 Agent 行动
### D17-3: comments 表 CHECK 约束处理
@@ -954,6 +1032,21 @@ ticker 需要扫描 `_toolchain` 虚拟项目。当前 ticker 通过 `TaskTypeRe
- action_report + verify 是更可靠的完成路径
- 减少 Agent 需要执行的 API 操作(从"标 done + 提交产出"简化为"提交 action_report"
### D17-8: on_failure 走 Gitea 管理,不走 Mail
**决策**:toolchain 失败处理三分路——业务失败在 PR/Issue comment @assignee,系统失败创建 Gitea Issue @pangtong-fujunshi,基础设施失败创建 toolchain task @jiangwei-infra。三条路全在 toolchain 流程内,Mail 不参与。
**讨论的替代方案**
- A(Mail API 通知庞统):跨系统,回溯需要在 _mail 和 _toolchain 两个 DB 之间跳
- Bon_failure 直接创建新 _toolchain task):强约束语义不对,失败应该是协作求助而非又一个强制任务
- C(只记日志 + 前端告警):system comment 没人主动看,等于没人管
**理由**
- 错误分类后路由到不同处理方式,比统一通知更精准
- Gitea Issue/PR comment 是天然的管理和审计工具,人也能看到
- §25 @mention webhook 已有端到端集成,comment 创建后自然触发新 toolchain task,管道复用
- Mail 只服务 Agent 间点对点通信和 PR 合并通知,职责清晰
---
## §13. 实施计划
@@ -974,14 +1067,15 @@ ticker 需要扫描 `_toolchain` 虚拟项目。当前 ticker 通过 `TaskTypeRe
| 2b | `toolchain_handler.py` | ToolchainApiSection 改为 action_report 指引 | 低 |
| 2c | `toolchain_handler.py` | ToolchainConstraintsSection 加 Red Flags 表 | 低 |
| 2d | `toolchain_handler.py` | verify_completion 改用 action_report(保留 fallback | 中 |
| 2e | `toolchain_handler.py` | on_failure 三分路重写(业务→PR comment / 系统→Gitea Issue / 基础设施→toolchain task | |
| 2e | `toolchain_handler.py` | on_failure 保留现有逻辑(标 failed + 通知庞统 | |
### Step 3toolchain_routes 改造
| 子步骤 | 文件 | 内容 | 风险 |
|--------|------|------|------|
| 3a | `toolchain_routes.py` | 新增 `_toolchain_db_path()` + `_send_toolchain_task()` | 低 |
| 3b | `toolchain_routes.py` | 所有 handler(含 PR merged改为调用 `_send_toolchain_task` | 中 |
| 3b | `toolchain_routes.py` | 8 个 handler 改为调用 `_send_toolchain_task`PR merged 除外) | 中 |
| 3c | `mail_notify.py` | `_REASON_MAP` 新增 `no_action_report` | 极低 |
### Step 4:测试 + 验证
@@ -992,7 +1086,7 @@ ticker 需要扫描 `_toolchain` 虚拟项目。当前 ticker 通过 `TaskTypeRe
| 4c | 单元测试:_send_toolchain_task 写入 _toolchain DB |
| 4d | 集成测试:webhook → toolchain task → Agent → action_report → done |
| 4e | 回归测试:_mail 路径不受影响(inform/request 不变) |
| 4f | 回归测试:PR merged 走 _send_toolchain_taskreview_mergedverify 始终通过 |
| 4f | 回归测试:PR merged 走 _send_mailinform |
---
@@ -1000,7 +1094,7 @@ ticker 需要扫描 `_toolchain` 虚拟项目。当前 ticker 通过 `TaskTypeRe
| 风险 | 概率 | 影响 | 缓解措施 |
|------|------|------|----------|
| Agent 不提交 action_report | 中 | 高 | Prompt 强约束 + Red Flag 表 + verify 失败 → on_failure 三分路(PR comment @assignee / Gitea Issue @pangtong / toolchain task @jiangwei+ fallbackoutput/comment |
| Agent 不提交 action_report | 中 | 高 | Prompt 强约束 + Red Flag 表 + verify 失败标 failed + 通知庞统 + fallbackoutput/comment |
| Agent 提交虚假 action_report | 低 | 中 | 后续事件链自然暴露(CI 不通过、Reviewer 收不到 Review |
| Agent 混淆 toolchain 和 mail 语义 | 低 | 低 | ToolchainContextSection 明确告知"这是需要执行动作的事件" |
| _toolchain DB 未初始化 | 低 | 中 | `_toolchain_db_path()` 中调用 `init_db()` 确保目录和表存在 |
@@ -1017,7 +1111,7 @@ ticker 需要扫描 `_toolchain` 虚拟项目。当前 ticker 通过 `TaskTypeRe
§13 §15.5 已定义 6 个强约束模板,每个包含编号步骤、Gitea API 调用指令、时限要求。本设计在此基础上:
- 将步骤从模板纯文本提取到 must_hives JSON 的 `steps` 字段(结构化、可编程化)
- 通过 ToolchainHandler 的 PromptSection 强约束确保 Agent 知道必须执行
- action report(执行凭证)替代邮件回复(更适合自动化验证)
- 将"回复此 Mail 确认"改为"提交 action report"(更适合自动化验证)
### 15.2 Superpowers: Red Flags 表
@@ -1065,7 +1159,7 @@ Hermes 的验证理念:执行和验证不可分割。本设计中 verify_compl
- [ ] §3 输入约束:must_hives JSON 结构 + ToolchainContextSection 渲染增强
- [ ] §4 执行约束:Red Flags 表设计是否覆盖常见 self-rationalization 模式
- [ ] §5 输出约束:action_report verify 机制 + fallback 设计
- [ ] §6 场景 steps 定义是否完整(9 种 action 场景 + 1 种纯通知场景 review_merged
- [ ] §6 场景 steps 定义是否完整(8 种 action 场景 + 1 种 inform 场景
- [ ] §7 _send_toolchain_task 函数设计是否正确
- [ ] §8 PromptContext / spawner 改动是否和 §14 架构一致
- [ ] §9 DB 隔离是否符合 §14 原设计
+17 -284
View File
@@ -50,15 +50,7 @@ router = APIRouter(tags=["toolchain"])
_delivery_cache: Set[str] = set()
_delivery_timestamps: List[Tuple[float, str]] = []
_TTL_SECONDS = 7 * 24 * 3600
_idempotency_lock: Optional[asyncio.Lock] = None
def _get_idempotency_lock() -> asyncio.Lock:
"""懒加载 asyncio.Lock,避免模块级创建时 event loop 不存在(Python 3.9)。"""
global _idempotency_lock
if _idempotency_lock is None:
_idempotency_lock = asyncio.Lock()
return _idempotency_lock
_idempotency_lock = asyncio.Lock()
def _is_duplicate(event: str, delivery: str,
@@ -197,7 +189,6 @@ def _calc_risk_level(changed_files: List[str]) -> str:
MAIL_PROJECT_ID = "_mail"
TOOLCHAIN_PROJECT_ID = "_toolchain"
def _mail_db_path() -> Path:
@@ -209,73 +200,6 @@ def _mail_db_path() -> Path:
return db
def _toolchain_db_path() -> Path:
"""获取 Toolchain 数据库路径,确保目录和表存在。"""
root = get_data_root()
db = root / TOOLCHAIN_PROJECT_ID / "blackboard.db"
db.parent.mkdir(parents=True, exist_ok=True)
init_db(db)
return db
def _send_toolchain_task(
to_agent: str,
title: str,
description: str,
event_type: str,
action_type: str,
steps: list,
context_data: dict | None = None,
source: str = "webhook",
) -> str:
"""创建 Toolchain Task 并写入 _toolchain DB。
Args:
to_agent: 收件人 Agent ID
title: 任务标题
description: 任务描述(模板渲染后的事件信息)
event_type: 事件类型(review_result / ci_failure / ...
action_type: 动作分类(用于步骤选择和日志统计)
steps: 结构化编号步骤列表
context_data: 事件上下文数据(PR 号、仓库名等)
source: 来源标识
Returns:
创建的 Task ID
"""
if to_agent not in AGENT_IDS:
logger.warning("Unknown agent: %s, skipping toolchain task", to_agent)
return ""
task_id = f"tc-{int(datetime.now().timestamp() * 1000)}"
must_hives = json.dumps({
"event_type": event_type,
"action_type": action_type,
"steps": steps,
"context": context_data or {},
"from": "system",
"source": source,
}, ensure_ascii=False)
task = Task(
id=task_id,
title=title,
description=description,
assignee=to_agent,
assigned_by="system",
must_haves=must_hives,
task_type="toolchain",
status="pending",
)
bb = Blackboard(_toolchain_db_path())
bb.create_task(task)
logger.info(
"Toolchain task sent: %s%s [%s] action_type=%s",
title[:40], to_agent, task_id, action_type,
)
return task_id
def _send_mail(
to_agent: str,
title: str,
@@ -403,25 +327,7 @@ async def _send_mention_mails(
})
title = f"@mention ({intent_hint}): {source_type} {number_str} ({repo})"
_send_toolchain_task(
to_agent=agent_id,
title=title,
description=text,
event_type="mention",
action_type="mention",
steps=[
"按上方 mention 模板中的 response_guidance 执行",
"提交 action reportPOST http://localhost:8083/api/projects/_toolchain/tasks/<task_id>/commentscomment_type=action_report",
],
context_data={
"source_type": source_type,
"source_url": source_url,
"commenter": commenter,
"content_snippet": content[:500],
"repo": repo,
"issue_number": issue_number,
},
)
_send_mail(agent_id, title, text)
# ---------------------------------------------------------------------------
@@ -473,27 +379,7 @@ async def _handle_pr_opened(payload: Dict[str, Any]) -> None:
})
title = f"Review 请求: {pr_title} ({repo}#{pr_number})"
_send_toolchain_task(
to_agent="simayi-challenger",
title=title,
description=text,
event_type="review_request",
action_type="review_request",
steps=[
f"读取 PR diffGitea API: GET /repos/{repo}/pulls/{pr_number}.diff",
"按审查清单审查(参考 code-review Skill",
f"提交 ReviewGitea API: POST /repos/{repo}/pulls/{pr_number}/reviews)— APPROVE 或 REQUEST_CHANGES",
"提交 action reportPOST http://localhost:8083/api/projects/_toolchain/tasks/<task_id>/commentscomment_type=action_report",
],
context_data={
"pr_number": pr_number,
"repo": repo,
"pr_title": pr_title,
"pr_author": pr_author,
"branch": branch,
"risk_level": risk_level,
},
)
_send_mail("simayi-challenger", title, text)
# S3: PR body @mention 通知
pr_body = pr.get("body", "") or ""
@@ -602,25 +488,7 @@ async def _handle_pull_request_review(payload: Dict[str, Any]) -> None:
})
title = f"Review 评论: {pr_title} ({repo}#{pr_number})"
_send_toolchain_task(
to_agent=pr_author,
title=title,
description=text,
event_type="review_comment",
action_type="review_comment",
steps=[
f"查看评论(Gitea API: GET /repos/{repo}/issues/{pr_number}/comments",
"根据评论内容响应(修改代码或在 PR 上回复 comment)",
"提交 action reportPOST http://localhost:8083/api/projects/_toolchain/tasks/<task_id>/commentscomment_type=action_report",
],
context_data={
"pr_number": pr_number,
"repo": repo,
"pr_title": pr_title,
"reviewer": reviewer,
"comment_body": review_body,
},
)
_send_mail(pr_author, title, text)
# S5: Review body @mention 通知(COMMENTED 路径)
await _send_review_mentions(review_body, reviewer, pr_author, pr, repo, pr_number)
@@ -642,34 +510,7 @@ async def _handle_pull_request_review(payload: Dict[str, Any]) -> None:
})
title = f"Review {result}: {pr_title} ({repo}#{pr_number})"
if state == "APPROVED":
tc_steps = [
f"合并 PRGitea API: POST /repos/{repo}/pulls/{pr_number}/merge",
"提交 action reportPOST http://localhost:8083/api/projects/_toolchain/tasks/<task_id>/commentscomment_type=action_report",
]
else: # REQUEST_CHANGES
tc_steps = [
"按审查意见逐条修改代码",
"push 到原分支 → CI 自动跑",
"CI 通过后等重新 Review",
"提交 action reportPOST http://localhost:8083/api/projects/_toolchain/tasks/<task_id>/commentscomment_type=action_report",
]
_send_toolchain_task(
to_agent=pr_author,
title=title,
description=text,
event_type="review_result",
action_type="review_result",
steps=tc_steps,
context_data={
"pr_number": pr_number,
"repo": repo,
"pr_title": pr_title,
"result": result,
"reviewer": reviewer,
"review_body": review_body,
},
)
_send_mail(pr_author, title, text)
# S5: Review body @mention 通知(非 COMMENTED 路径)
await _send_review_mentions(review_body, reviewer, pr_author, pr, repo, pr_number)
@@ -738,31 +579,11 @@ async def _handle_pr_synchronize(payload: Dict[str, Any]) -> None:
})
title = f"PR 更新: {pr_title} ({repo}#{pr_number})"
_send_toolchain_task(
to_agent=reviewer,
title=title,
description=text,
event_type="review_updated",
action_type="review_updated",
steps=[
f"读取 PR diffGitea API: GET /repos/{repo}/pulls/{pr_number}.diff",
"重点检查上次 Review 意见的修改部分",
f"提交 ReviewGitea API: POST /repos/{repo}/pulls/{pr_number}/reviews",
"提交 action reportPOST http://localhost:8083/api/projects/_toolchain/tasks/<task_id>/commentscomment_type=action_report",
],
context_data={
"pr_number": pr_number,
"repo": repo,
"pr_title": pr_title,
"pr_author": pr_author,
"new_sha": new_sha,
"reviewer": reviewer,
},
)
_send_mail(reviewer, title, text)
def _send_deploy_failure_task(repo: str, pr_number: int, pr_title: str, reason: str) -> None:
"""CD 部署失败通知,走 ToolchainHandler。"""
def _send_deploy_failure_mail(repo: str, pr_number: int, pr_title: str, reason: str) -> None:
"""CD 部署失败通知,复用 deploy_failure 模板"""
text = render_template("deploy_failure", {
"repo": repo,
"commit_sha": f"PR #{pr_number}",
@@ -770,25 +591,7 @@ def _send_deploy_failure_task(repo: str, pr_number: int, pr_title: str, reason:
title = f"部署失败: {repo} (auto-deploy, PR #{pr_number})"
full_text = f"{text}\n\n失败原因: {reason}"
for agent_id in ("jiangwei-infra", "pangtong-fujunshi"):
_send_toolchain_task(
to_agent=agent_id,
title=title,
description=full_text,
event_type="deploy_failure",
action_type="deploy_failure",
steps=[
"检查 deploy 日志",
"排查失败原因",
"修复并重新部署",
"提交 action reportPOST http://localhost:8083/api/projects/_toolchain/tasks/<task_id>/commentscomment_type=action_report",
],
context_data={
"repo": repo,
"pr_number": pr_number,
"pr_title": pr_title,
"reason": reason,
},
)
_send_mail(agent_id, title, full_text)
async def _handle_pr_closed(payload: Dict[str, Any]) -> None:
@@ -820,21 +623,7 @@ async def _handle_pr_closed(payload: Dict[str, Any]) -> None:
})
title = f"PR 已合并: {pr_title} ({repo}#{pr_number})"
_send_toolchain_task(
to_agent=pr_author,
title=title,
description=text,
event_type="review_merged",
action_type="review_merged",
steps=[], # 纯通知,无步骤
context_data={
"pr_number": pr_number,
"repo": repo,
"pr_title": pr_title,
"pr_author": pr_author,
"merged_by": merged_by,
},
)
_send_mail(pr_author, title, text)
# 自动部署:git pull + rsync + 按需 post_deploy
try:
@@ -887,7 +676,7 @@ async def _handle_pr_closed(payload: Dict[str, Any]) -> None:
if rsync_proc.returncode != 0:
logger.error("Auto-deploy: rsync failed: %s", rsync_err.decode())
_send_deploy_failure_task(repo, pr_number, pr_title, f"rsync 失败: {rsync_err.decode()}")
_send_deploy_failure_mail(repo, pr_number, pr_title, f"rsync 失败: {rsync_err.decode()}")
return
# Step 3: 判断是否需要执行 post_deploy
@@ -942,7 +731,7 @@ async def _handle_pr_closed(payload: Dict[str, Any]) -> None:
if deploy_proc.returncode != 0:
logger.error("Auto-deploy: post_deploy failed: %s", deploy_err.decode())
_send_deploy_failure_task(repo, pr_number, pr_title, f"post_deploy 失败 ({cmd}): {deploy_err.decode()}")
_send_deploy_failure_mail(repo, pr_number, pr_title, f"post_deploy 失败 ({cmd}): {deploy_err.decode()}")
break
else:
logger.info("Auto-deploy: all post_deploy commands succeeded (files: %s)", ", ".join(file_list[:5]))
@@ -951,7 +740,7 @@ async def _handle_pr_closed(payload: Dict[str, Any]) -> None:
except asyncio.TimeoutError:
logger.error("Auto-deploy: timeout for %s", repo)
_send_deploy_failure_task(repo, pr_number, pr_title, "部署超时")
_send_deploy_failure_mail(repo, pr_number, pr_title, "部署超时")
except Exception as e:
logger.error("Auto-deploy: unexpected error: %s", e)
@@ -998,29 +787,7 @@ async def _handle_issues(payload: Dict[str, Any]) -> None:
})
title = f"Issue 指派: {issue_title} ({repo}#{issue_number})"
_send_toolchain_task(
to_agent=assignee,
title=title,
description=text,
event_type="issue_assigned",
action_type="issue_assigned",
steps=[
f"创建分支 fix/{issue_number}-{brief}",
"编码 + 写 UT",
"push → 等 CI",
f"CI 通过后创建 PRGitea API: POST /repos/{repo}/pulls",
"等 Review",
"提交 action reportPOST http://localhost:8083/api/projects/_toolchain/tasks/<task_id>/commentscomment_type=action_report",
],
context_data={
"issue_number": issue_number,
"repo": repo,
"issue_title": issue_title,
"labels": labels,
"issue_body": issue_body or "(无描述)",
"brief": brief,
},
)
_send_mail(assignee, title, text)
elif action == "opened":
if "部署失败" in issue_title:
@@ -1035,23 +802,7 @@ async def _handle_issues(payload: Dict[str, Any]) -> None:
title = f"部署失败: {repo}"
for agent_id in ("jiangwei-infra", "pangtong-fujunshi"):
_send_toolchain_task(
to_agent=agent_id,
title=title,
description=text,
event_type="deploy_failure",
action_type="deploy_failure",
steps=[
"检查 deploy 日志",
"排查失败原因",
"修复并重新部署",
"提交 action reportPOST http://localhost:8083/api/projects/_toolchain/tasks/<task_id>/commentscomment_type=action_report",
],
context_data={
"repo": repo,
"commit_sha": commit_sha or "(未知)",
},
)
_send_mail(agent_id, title, text)
# Issue body @mentionopened 时检查)
issue_body = issue.get("body", "") or ""
@@ -1118,25 +869,7 @@ async def _handle_issue_comment(payload: Dict[str, Any]) -> None:
})
title = f"CI 失败: {repo}#{issue_number}"
_send_toolchain_task(
to_agent=pr_author,
title=title,
description=text,
event_type="ci_failure",
action_type="ci_failure",
steps=[
"查看完整 CI 日志(PR 页面或 Gitea Actions 页面)",
"修复失败的测试",
"push → CI 自动重跑",
"提交 action reportPOST http://localhost:8083/api/projects/_toolchain/tasks/<task_id>/commentscomment_type=action_report",
],
context_data={
"pr_number": issue_number,
"repo": repo,
"branch": branch,
"error_summary": error_summary,
},
)
_send_mail(pr_author, title, text)
# CI 处理完不 return,继续检查 @mention
# === 路径 2:@mention 通知(新增,独立路径) ===
@@ -1227,7 +960,7 @@ async def gitea_webhook(
# 2. 幂等检查(需要在 payload 解析后,以支持内容去重)
if x_gitea_event and x_gitea_delivery:
async with _get_idempotency_lock():
async with _idempotency_lock:
if _is_duplicate(x_gitea_event, x_gitea_delivery, payload):
logger.debug(
"Duplicate webhook: %s/%s",
+1 -1
View File
@@ -293,7 +293,7 @@ _SCHEMA_STATEMENTS = [
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_id TEXT NOT NULL REFERENCES tasks(id),
author TEXT NOT NULL,
comment_type TEXT NOT NULL DEFAULT 'general',
comment_type TEXT NOT NULL DEFAULT 'general' CHECK (comment_type IN ('general','handoff','observation','review','rebuttal','rebuttal_response','debate_argument','debate_rebuttal','debate_judgment')),
body TEXT NOT NULL,
mentions TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
-2
View File
@@ -65,8 +65,6 @@ class PromptContext:
# toolchain 专用
event_type: str = "" # ci_failure / review_request / ...
event_data: Dict = field(default_factory=dict)
action_type: str = "" # 动作分类(review_result / ci_failure / ...
action_steps: list = field(default_factory=list) # 结构化编号步骤列表
# 前序产出
depends_on_outputs: Optional[List] = None
-6
View File
@@ -286,15 +286,10 @@ class AgentSpawner:
# 从 must_haves 解析 mail 元数据(from / performative
from_agent = ""
mail_type = ""
action_type = ""
action_steps = []
try:
meta = json.loads(must_haves) if must_haves else {}
from_agent = meta.get("from", "")
mail_type = meta.get("performative", meta.get("type", ""))
# toolchain 字段提取
action_type = meta.get("action_type", "")
action_steps = meta.get("steps", [])
except Exception:
pass
ctx = PromptContext(
@@ -303,7 +298,6 @@ class AgentSpawner:
agent_id=agent_id, role=spawn_type,
spawn_type=spawn_type,
from_agent=from_agent, mail_type=mail_type,
action_type=action_type, action_steps=action_steps,
)
return handler.build_prompt(ctx)
+111 -355
View File
@@ -1,16 +1,14 @@
"""toolchain_handler.py - 工具链事件 handler。
"""toolchain_handler.py 工具链事件 handler。
处理 Gitea Webhook 事件(CI 失败Review 请求Issue 指派等)
L2 引擎层强约束:输入(结构化步骤)+ 执行(Red Flags)+ 输出(action_report 验证)
处理 Gitea Webhook 事件CI 失败Review 请求Issue 指派等
"""
from __future__ import annotations
import json
import logging
import os
import urllib.request
from pathlib import Path
from typing import Dict, List
from typing import Dict
from src.daemon.base_task_handler import BaseTaskHandler, VerifyResult
from src.daemon.prompt_composer import PromptComposer, PromptContext
@@ -19,34 +17,13 @@ from src.blackboard.db import get_connection
logger = logging.getLogger("moziplus-v2.handler.toolchain")
# ---------------------------------------------------------------------------
# Gitea API 配置
# ---------------------------------------------------------------------------
_GITEA_BASE = "http://192.168.2.154:3000/api/v1"
_GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
# action_type → action_hint 映射
_ACTION_HINTS: Dict[str, str] = {
"review_result": "你收到一个 Review 结果通知,这是一个需要你执行动作的事件(不是纯通知)。",
"review_request": "你收到一个 Review 请求,这是一个需要你审查并提交 Review 的事件。",
"review_updated": "你收到一个 PR 更新通知,这是一个需要你重新审查修改部分的事件。",
"review_comment": "你收到一个 Review 评论,这是一个需要你查看并响应的事件。",
"ci_failure": "你收到一个 CI 失败通知,这是一个需要你修复失败测试的事件。",
"issue_assigned": "你收到一个 Issue 指派,这是一个需要你编码实现的事件。",
"deploy_failure": "你收到一个部署失败通知,这是一个需要你排查并修复的事件。",
"mention": "你收到一个 @mention 通知,这是一个需要你按指引响应的事件。",
"review_merged": "你收到一个 PR 合并通知。这是一条纯通知,阅读即可。",
"infrastructure_failure": "你收到一个基础设施问题报告,请排查并修复。",
}
# ---------------------------------------------------------------------------
# Toolchain PromptSections
# ---------------------------------------------------------------------------
class ToolchainContextSection:
"""事件类型 + 事件详情 + 结构化步骤 + action_hint(priority=10)"""
"""事件类型 + 事件详情priority=10"""
name: str = "toolchain_context"
priority: int = 10
@@ -55,44 +32,27 @@ class ToolchainContextSection:
event_type = context.event_type
event_data: Dict = context.event_data or {}
# Part 1: 事件信息(现有模板引擎)
if event_type in _TEMPLATE_MAP:
# 使用模板引擎渲染已知事件
variables = {k: str(v) for k, v in event_data.items()}
event_text = render_template(event_type, variables)
else:
lines = ["## 工具链事件", ""]
lines.append(f"- **事件类型**: {event_type or '未知'}")
if event_data:
lines.append("- **事件详情**:")
for key, value in event_data.items():
lines.append(f" - {key}: {value}")
lines.append("")
event_text = "\n".join(lines)
return render_template(event_type, variables)
# Part 2: 结构化编号步骤(新增,从 action_steps 渲染)
steps: List[str] = context.action_steps or []
if steps:
step_lines = ["", "### 必须执行的步骤", ""]
for i, step in enumerate(steps, 1):
step_lines.append(f"{i}. {step}")
steps_text = "\n".join(step_lines)
else:
steps_text = ""
# Part 3: action 指引(新增,按 action_type 选择)
action_hint = _ACTION_HINTS.get(
context.action_type,
"你收到一个工具链事件,这是一个需要你执行动作的事件。",
)
return f"{action_hint}\n\n{event_text}{steps_text}"
# fallback:通用事件描述
lines = ["## 工具链事件", ""]
lines.append(f"- **事件类型**: {event_type or '未知'}")
if event_data:
lines.append("- **事件详情**:")
for key, value in event_data.items():
lines.append(f" - {key}: {value}")
lines.append("")
return "\n".join(lines)
def should_include(self, context: PromptContext) -> bool:
return True
class ToolchainApiSection:
"""API 操作指令(priority=40)-- action_report 提交指引"""
"""API 操作指令priority=40),success_status=done"""
name: str = "toolchain_api"
priority: int = 40
@@ -100,48 +60,28 @@ class ToolchainApiSection:
API_HOST = "localhost:8083"
def render(self, context: PromptContext) -> str:
task_id = context.task_id
project_id = context.project_id
agent_id = context.agent_id
lines = [
"## API 操作指令",
"",
f"项目 ID: `{project_id}`",
f"任务 ID: `{task_id}`",
f"项目 ID: `{context.project_id}`",
f"任务 ID: `{context.task_id}`",
"",
"### 完成后必须提交 action report",
"",
"执行完所有步骤后,必须提交 action report:",
"### 完成后必须更新任务状态",
"完成后务必通过以下命令将任务标记为 **done**:",
"```bash",
f'curl -s -X POST "http://{self.API_HOST}/api/projects/{project_id}/tasks/{task_id}/comments" \\',
f'curl -s -X POST "http://{self.API_HOST}/api/projects/{context.project_id}/tasks/{context.task_id}/status" \\',
' -H "Content-Type: application/json" \\',
f' -d \'{{"author": "{agent_id}", "comment_type": "action_report", "body": "简要描述你执行了什么操作及结果"}}\'',
' -d \'{"status": "done"}\'',
"```",
"",
"⚠️ 不提交 action report 的任务会被标记为 failed。",
"",
"### 提交产出",
"",
"如有产出(如 review 结果、修复方案),提交到任务 outputs:",
"如有产出(如 review 结果、修复方案),提交到任务 outputs:",
"```bash",
f'curl -s -X POST "http://{self.API_HOST}/api/projects/{project_id}/tasks/{task_id}/outputs" \\',
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"}\'',
"```",
"",
"### 需要其他角色支持时",
"",
"如果在执行过程中需要其他角色协助(如缺数据、需要审批等),在关联的 PR/Issue 上创建 comment @对方:",
"```bash",
f'curl -s -X POST "{_GITEA_BASE}/repos/{{repo}}/issues/{{pr_number}}/comments" \\',
' -H "Authorization: token <your-token>" \\',
' -H "Content-Type: application/json" \\',
' -d \'{"body": "@{agent-id} 需要你的支持:{描述问题}"}\'',
"```",
"",
"⚠️ 不要使用 Mail API(飞鸽传书)。所有协作通过 Gitea 留痕。",
"",
]
return "\n".join(lines)
@@ -150,50 +90,20 @@ class ToolchainApiSection:
class ToolchainConstraintsSection:
"""硬约束 + Red Flags(priority=50)"""
"""硬约束priority=50"""
name: str = "toolchain_constraints"
priority: int = 50
def render(self, context: PromptContext) -> str:
lines = [
"## 硬约束(必须遵守)",
"## 硬约束",
"",
"⚠️ 以下是强制要求,不是建议或参考。违反任何一条都会导致任务失败。",
"",
"### 1. 必须按步骤执行",
'- 检查上方“必须执行的步骤”列表',
'- 逐条执行每个步骤,不可跳过',
'- 不要只读不做——这不是纯通知',
"",
"### 2. 必须提交 action report",
'- 执行完所有步骤后,必须提交 action report',
"- 提交方式:POST commentcomment_type='action_report'",
'- 报告内容:简要描述你执行了什么操作、结果如何',
'- ⚠️ 不提交 action report 的任务会被标记为 failed',
"",
"### 3. 不要执行任何状态转换命令",
'- 不要手动标 working/done/review/failed,系统会自动处理',
"",
"### 4. 不需要回复",
'- action report 就是你的完成凭证',
'- 不要发送 Mail(飞鸽传书),你的所有操作在 toolchain 流程内完成',
"",
"### 5. 所有协作通过 Gitea 完成",
'- 如果遇到问题需要其他角色支持,在关联的 PR/Issue 上创建 comment @对方',
'- 不要使用 Mail API(飞鸽传书)发送消息',
'- 你的所有操作都在 toolchain 流程内,通过 Gitea 留痕',
"",
"### Red Flags(如果脑海中出现以下想法,说明你错了)",
"",
'| Agent 想法 | Red Flag 驳回 |',
'|------------|--------------|',
'| “这个通知看看就行了” | ❌ 错!这是 action 指令,必须执行步骤列表中的每一项 |',
'| “我不需要做任何事” | ❌ 错!检查“必须执行的步骤”列表,每一步都要执行 |',
'| “先放着等会处理” | ❌ 错!立即执行,不要推迟 |',
'| “我已经知道了” | ❌ 知道不等于执行。执行步骤 + 提交 action report 才算完成 |',
'| “步骤太多了,选几个做就行” | ❌ 错!必须逐条执行,不可跳过 |',
'| “这个步骤不适用于当前情况” | ❌ 如果确实不适用,在 action report 中说明原因,但其他步骤必须执行 |',
"1. **必须标 done**:处理完成后必须通过 API 将任务状态更新为 `done`,否则视为未完成",
"2. **产出不能为空**:必须提交有意义的产出(output 或 comment),不能只改状态",
"3. **单一职责**:只处理本次事件相关的操作,不要越界执行无关任务",
"4. **出错即报告**:如果无法处理(如权限不足、资源不存在),在 comment 中说明原因并标 done",
"5. **不要创建新任务**:工具链事件只处理当前事件,不衍生新任务",
"",
]
return "\n".join(lines)
@@ -217,7 +127,7 @@ class ToolchainHandler(BaseTaskHandler):
return "done"
def pre_spawn(self, task_id: str, db_path: Path) -> bool:
"""auto_working:pending → working"""
"""auto_workingpending → working"""
return self._auto_mark_working(task_id, db_path)
def get_sections(self) -> list:
@@ -235,55 +145,27 @@ class ToolchainHandler(BaseTaskHandler):
return composer.compose(context)
def verify_completion(self, task_id: str, db_path: Path) -> VerifyResult:
"""检查 action report(精确验证)+ 三层 fallback"""
"""检查行动输出(output 或 comment 有实质内容)"""
try:
conn = get_connection(db_path)
try:
# 特殊处理:infrastructure_failure 始终通过(防递归)
row = conn.execute(
"SELECT must_haves FROM tasks WHERE id=?", (task_id,)
).fetchone()
if row and row["must_haves"]:
try:
meta = json.loads(row["must_haves"])
except Exception:
meta = {}
if meta.get("action_type") == "infrastructure_failure":
return VerifyResult(True, "infrastructure_passthrough",
"infrastructure_failure auto-pass")
# 特殊处理:review_merged 始终通过(纯通知)
if meta.get("action_type") == "review_merged":
return VerifyResult(True, "merged_passthrough",
"review_merged auto-pass")
# 1. 优先检查 action_report comment
report_row = conn.execute(
"SELECT id FROM comments WHERE task_id=? "
"AND comment_type='action_report' LIMIT 1",
(task_id,)
).fetchone()
if report_row:
return VerifyResult(True, "has_action_report", "action_report found")
# 2. fallback:检查 output(向后兼容)
# 检查 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}")
# 3. fallback:检查有实质内容的 comment(向后兼容)
# 检查 comment(非系统、有实质内容)
comment_count = conn.execute(
"SELECT COUNT(*) FROM comments WHERE task_id=? "
"AND author != 'system' AND LENGTH(body) >= 20",
"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",
"no action_report, no output, no valid comment")
return VerifyResult(False, "no_action", "output=0, comment=0")
finally:
conn.close()
except Exception as e:
@@ -292,217 +174,32 @@ class ToolchainHandler(BaseTaskHandler):
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)
logger.info("Toolchain %s: verify failed (%s), marked failed", task_id, verify.reason)
# 读取 must_hives 获取事件上下文 + assignee 从 tasks 表读取
meta = {}
assignee = agent_id
# 从 db 读取事件上下文
event_type = ""
event_data: Dict = {}
try:
conn = get_connection(db_path)
row = conn.execute(
"SELECT must_haves, assignee FROM tasks WHERE id=?", (task_id,)
"SELECT must_haves FROM tasks WHERE id=?", (task_id,)
).fetchone()
if row:
if row["must_haves"]:
meta = json.loads(row["must_haves"])
assignee = row["assignee"] or agent_id
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
action_type = meta.get("action_type", "")
context_data = meta.get("context", {})
# 三分路决策
route = self._classify_failure(verify)
if route == "business":
self._handle_business_failure(
task_id, agent_id, verify, action_type, context_data, assignee, db_path)
elif route == "system":
self._handle_system_failure(
task_id, agent_id, verify, action_type, context_data, db_path)
else: # infrastructure
self._handle_infrastructure_failure(
task_id, agent_id, verify, db_path)
def _classify_failure(self, verify: VerifyResult) -> str:
"""分类失败类型:business / infrastructuresystem 通过升级到达)"""
# verify_error 或 DB 不可用 → 基础设施失败
if verify.reason == "verify_error":
return "infrastructure"
# 默认:业务失败
return "business"
def _handle_business_failure(
self, task_id: str, agent_id: str, verify: VerifyResult,
action_type: str, context_data: dict, assignee: str,
db_path: Path,
) -> None:
"""业务失败 → 在关联 PR/Issue 上创建 comment @原始 assignee"""
repo = context_data.get("repo", "")
pr_number = context_data.get("pr_number") or context_data.get("issue_number", "")
if repo and pr_number:
comment_body = (
f"@{assignee or agent_id} 工具链任务执行失败\n\n"
f"任务 ID: {task_id}\n"
f"失败原因: {verify.reason}\n"
f"证据: {verify.evidence}\n\n"
f"请检查黑板任务并处理。"
)
success = self._create_gitea_comment(repo, pr_number, comment_body)
if success:
logger.info("Toolchain %s: business failure → Gitea comment on %s#%s",
task_id, repo, pr_number)
return
# Gitea API failed → escalate to system failure
logger.warning(
"Toolchain %s: Gitea comment failed, escalating to system failure",
task_id)
self._handle_system_failure(
task_id, agent_id, verify, action_type, context_data, db_path)
else:
# 没有 PR/Issue 关联 → fallback 到系统失败
logger.warning(
"Toolchain %s: no PR/Issue context for business failure, "
"escalating to system failure", task_id)
self._handle_system_failure(
task_id, agent_id, verify, action_type, context_data, db_path)
def _handle_system_failure(
self, task_id: str, agent_id: str, verify: VerifyResult,
action_type: str, context_data: dict, db_path: Path,
) -> None:
"""系统失败 → 创建 Gitea Issue @pangtong-fujunshi"""
repo = context_data.get("repo", "sanguo/sanguo_moziplus_v2")
title = f"[toolchain-handler] 工具链事件处理失败: {task_id}"
body = (
f"任务 {task_id} 验证失败\n\n"
f"事件类型: {action_type or '未知'}\n"
f"失败原因: {verify.reason}\n"
f"证据: {verify.evidence}\n\n"
f"@pangtong-fujunshi 请检查黑板任务并手动处理。"
self._notify_via_mail_api(
task_id, verify.reason, verify.evidence,
event_type, event_data,
)
# 尝试在 Gitea 创建 Issue
created = self._create_gitea_issue(repo, title, body, ["pangtong-fujunshi"])
if created:
logger.info("Toolchain %s: system failure → Gitea Issue created on %s",
task_id, repo)
else:
# Gitea API 不可用 → 基础设施失败
logger.error(
"Toolchain %s: Gitea API unavailable, escalating to infrastructure failure",
task_id)
self._handle_infrastructure_failure(
task_id, agent_id, verify, db_path)
def _handle_infrastructure_failure(
self, task_id: str, agent_id: str,
verify: VerifyResult, db_path: Path,
) -> None:
"""基础设施失败 → 直接在 _toolchain DB 创建 task @jiangwei-infra(防递归)"""
try:
from datetime import datetime
new_task_id = f"tc-{int(datetime.now().timestamp() * 1000)}"
must_hives = json.dumps({
"event_type": "infrastructure_failure",
"action_type": "infrastructure_failure",
"steps": [
"检查 Gitea 服务状态(http://192.168.2.154:3000)",
"检查网络连通性",
"恢复后提交 action report",
],
"context": {"original_task_id": task_id, "verify_reason": verify.reason},
"from": "system",
"source": "toolchain_handler_on_failure",
}, ensure_ascii=False)
conn = get_connection(db_path)
conn.execute(
"INSERT INTO tasks (id, title, description, assignee, assigned_by, "
"must_haves, task_type, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(
new_task_id,
f"[基础设施] Gitea API 不可用 - {task_id}",
f"Gitea API 不可用,原任务 {task_id} 无法通过正常路径处理。\n"
f"请检查 Gitea 服务状态和网络连通性。",
"jiangwei-infra",
"system",
must_hives,
"toolchain",
"pending",
)
)
conn.commit()
conn.close()
logger.info(
"Toolchain %s: infrastructure failure → task %s created for jiangwei-infra",
task_id, new_task_id)
except Exception as e:
logger.error(
"Toolchain %s: failed to create infrastructure_failure task: %s",
task_id, e)
# -----------------------------------------------------------------------
# Gitea API 辅助
# -----------------------------------------------------------------------
def _create_gitea_comment(
self, repo: str, pr_number: int, body: str,
) -> bool:
"""在 PR/Issue 上创建 comment。返回是否成功。"""
if not _GITEA_TOKEN:
return False
payload = json.dumps({"body": body}, ensure_ascii=False).encode("utf-8")
try:
req = urllib.request.Request(
f"{_GITEA_BASE}/repos/{repo}/issues/{pr_number}/comments",
data=payload,
headers={
"Authorization": f"token {_GITEA_TOKEN}",
"Content-Type": "application/json",
},
)
urllib.request.urlopen(req, timeout=5)
return True
except Exception as e:
logger.warning("Gitea comment failed on %s#%s: %s", repo, pr_number, e)
return False
def _create_gitea_issue(
self, repo: str, title: str, body: str,
assignees: list = None,
) -> bool:
"""创建 Gitea Issue。返回是否成功。"""
if not _GITEA_TOKEN:
return False
data = {"title": title, "body": body}
if assignees:
data["assignees"] = assignees
payload = json.dumps(data, ensure_ascii=False).encode("utf-8")
try:
req = urllib.request.Request(
f"{_GITEA_BASE}/repos/{repo}/issues",
data=payload,
headers={
"Authorization": f"token {_GITEA_TOKEN}",
"Content-Type": "application/json",
},
)
urllib.request.urlopen(req, timeout=5)
return True
except Exception as e:
logger.warning("Gitea create issue failed on %s: %s", repo, e)
return False
# -----------------------------------------------------------------------
# 兼容:保留旧方法签名(但不再被 on_failure 调用)
# -----------------------------------------------------------------------
def _build_gitea_links(self, event_type: str, event_data: dict) -> str:
"""根据事件类型构建 Gitea 链接。"""
links = []
@@ -518,4 +215,63 @@ class ToolchainHandler(BaseTaskHandler):
if "branch" in event_data and "commit" not in event_data:
links.append(f"分支: {event_data['branch']}")
return "\n".join(links) if links else "(无法提取链接,请检查黑板任务详情)"
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)
-525
View File
@@ -1,525 +0,0 @@
"""Unit tests for §17 ToolchainHandler 强约束实现."""
import json
import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
# Add project root to path
PROJECT_ROOT = Path(__file__).parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from src.daemon.prompt_composer import PromptContext, PromptComposer
from src.daemon.toolchain_handler import (
ToolchainHandler,
ToolchainContextSection,
ToolchainApiSection,
ToolchainConstraintsSection,
_ACTION_HINTS,
)
from src.daemon.base_task_handler import VerifyResult
from src.blackboard.db import init_db, get_connection
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def tmp_db():
"""Create a temporary _toolchain DB for testing."""
with tempfile.TemporaryDirectory() as d:
db_path = Path(d) / "blackboard.db"
init_db(db_path)
yield db_path
@pytest.fixture
def handler():
return ToolchainHandler()
def _insert_task(db_path, task_id, must_haves_json, status="working"):
"""Insert a task into DB for testing."""
conn = get_connection(db_path)
conn.execute(
"INSERT INTO tasks (id, title, description, assignee, assigned_by, "
"must_haves, task_type, status) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(task_id, "test", "test desc", "zhangfei-dev", "system",
must_haves_json, "toolchain", status)
)
conn.commit()
conn.close()
def _insert_comment(db_path, task_id, author, body, comment_type="general"):
"""Insert a comment into DB."""
conn = get_connection(db_path)
conn.execute(
"INSERT INTO comments (task_id, author, comment_type, body) VALUES (?, ?, ?, ?)",
(task_id, author, comment_type, body)
)
conn.commit()
conn.close()
def _insert_output(db_path, task_id, content="test output"):
"""Insert an output into DB."""
conn = get_connection(db_path)
conn.execute(
"INSERT INTO outputs (task_id, agent, output_type, title, summary) "
"VALUES (?, ?, ?, ?, ?)",
(task_id, "zhangfei-dev", "document", "test", content)
)
conn.commit()
conn.close()
# ---------------------------------------------------------------------------
# Step 1a: PromptContext new fields
# ---------------------------------------------------------------------------
class TestPromptContextFields:
def test_action_type_default(self):
ctx = PromptContext(
task_id="t1", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="a1",
)
assert ctx.action_type == ""
def test_action_steps_default(self):
ctx = PromptContext(
task_id="t1", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="a1",
)
assert ctx.action_steps == []
def test_action_type_set(self):
ctx = PromptContext(
task_id="t1", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="a1",
action_type="review_result",
)
assert ctx.action_type == "review_result"
def test_action_steps_set(self):
steps = ["step 1", "step 2"]
ctx = PromptContext(
task_id="t1", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="a1",
action_steps=steps,
)
assert ctx.action_steps == steps
# ---------------------------------------------------------------------------
# Step 2a: ToolchainContextSection steps rendering + action_hint
# ---------------------------------------------------------------------------
class TestToolchainContextSection:
def test_renders_steps(self):
ctx = PromptContext(
task_id="t1", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="a1",
event_type="review_result",
event_data={"pr_number": "42", "repo": "sanguo/test"},
action_type="review_result",
action_steps=["合并 PR", "提交 action report"],
)
section = ToolchainContextSection()
result = section.render(ctx)
assert "必须执行的步骤" in result
assert "1. 合并 PR" in result
assert "2. 提交 action report" in result
def test_renders_action_hint(self):
ctx = PromptContext(
task_id="t1", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="a1",
event_type="ci_failure",
action_type="ci_failure",
action_steps=[],
)
section = ToolchainContextSection()
result = section.render(ctx)
assert "CI 失败" in result
assert "需要你修复" in result
def test_renders_default_hint_for_unknown_action_type(self):
ctx = PromptContext(
task_id="t1", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="a1",
event_type="unknown",
action_type="unknown_type",
action_steps=[],
)
section = ToolchainContextSection()
result = section.render(ctx)
assert "需要你执行动作的事件" in result
def test_no_steps_no_steps_section(self):
ctx = PromptContext(
task_id="t1", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="a1",
event_type="review_merged",
action_type="review_merged",
action_steps=[],
)
section = ToolchainContextSection()
result = section.render(ctx)
assert "必须执行的步骤" not in result
# ---------------------------------------------------------------------------
# Step 2b: ToolchainApiSection action_report guidance
# ---------------------------------------------------------------------------
class TestToolchainApiSection:
def test_has_action_report_instruction(self):
ctx = PromptContext(
task_id="tc-123", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="zhangfei-dev",
)
section = ToolchainApiSection()
result = section.render(ctx)
assert "action_report" in result
assert "comment_type" in result
assert "tc-123" in result
def test_no_manual_done_instruction(self):
ctx = PromptContext(
task_id="tc-123", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="zhangfei-dev",
)
section = ToolchainApiSection()
result = section.render(ctx)
# Should NOT contain the old "标记为 done" instruction
assert "标记为 **done**" not in result
assert '"status": "done"' not in result
def test_has_outputs_instruction(self):
ctx = PromptContext(
task_id="tc-123", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="zhangfei-dev",
)
section = ToolchainApiSection()
result = section.render(ctx)
assert "outputs" in result
def test_has_gitea_collaboration_instruction(self):
ctx = PromptContext(
task_id="tc-123", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="zhangfei-dev",
)
section = ToolchainApiSection()
result = section.render(ctx)
assert "Gitea" in result
assert "Mail API" in result
# ---------------------------------------------------------------------------
# Step 2c: ToolchainConstraintsSection Red Flags
# ---------------------------------------------------------------------------
class TestToolchainConstraintsSection:
def test_has_red_flags_table(self):
ctx = PromptContext(
task_id="t1", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="a1",
)
section = ToolchainConstraintsSection()
result = section.render(ctx)
assert "Red Flags" in result
assert "" in result
def test_has_all_5_constraints(self):
ctx = PromptContext(
task_id="t1", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="a1",
)
section = ToolchainConstraintsSection()
result = section.render(ctx)
assert "必须按步骤执行" in result
assert "必须提交 action report" in result
assert "不要执行任何状态转换命令" in result
assert "不需要回复" in result
assert "所有协作通过 Gitea 完成" in result
def test_has_strong_language(self):
ctx = PromptContext(
task_id="t1", title="test", description="d",
must_haves="", project_id="_toolchain", agent_id="a1",
)
section = ToolchainConstraintsSection()
result = section.render(ctx)
assert "强制要求" in result
assert "不是建议" in result
# ---------------------------------------------------------------------------
# Step 2d: verify_completion tests
# ---------------------------------------------------------------------------
class TestVerifyCompletion:
def test_action_report_passes(self, handler, tmp_db):
"""action_report comment → pass"""
must_haves = json.dumps({"action_type": "review_result"})
_insert_task(tmp_db, "t1", must_haves)
_insert_comment(tmp_db, "t1", "zhangfei-dev",
"已修复 CI", comment_type="action_report")
result = handler.verify_completion("t1", tmp_db)
assert result.passed is True
assert result.reason == "has_action_report"
def test_no_action_report_fallback_output(self, handler, tmp_db):
"""No action_report but has output → pass (fallback)"""
must_haves = json.dumps({"action_type": "review_result"})
_insert_task(tmp_db, "t2", must_haves)
_insert_output(tmp_db, "t2", "review result content")
result = handler.verify_completion("t2", tmp_db)
assert result.passed is True
assert result.reason == "has_output"
def test_no_action_report_fallback_comment(self, handler, tmp_db):
"""No action_report but has substantial comment → pass (fallback)"""
must_haves = json.dumps({"action_type": "review_result"})
_insert_task(tmp_db, "t3", must_haves)
_insert_comment(tmp_db, "t3", "zhangfei-dev",
"This is a sufficiently long comment about the task.")
result = handler.verify_completion("t3", tmp_db)
assert result.passed is True
assert result.reason == "has_comment"
def test_nothing_passes(self, handler, tmp_db):
"""No action_report, no output, no comment → fail"""
must_haves = json.dumps({"action_type": "review_result"})
_insert_task(tmp_db, "t4", must_haves)
result = handler.verify_completion("t4", tmp_db)
assert result.passed is False
assert result.reason == "no_action"
def test_short_comment_fails(self, handler, tmp_db):
"""Comment < 20 chars → fail"""
must_haves = json.dumps({"action_type": "review_result"})
_insert_task(tmp_db, "t5", must_haves)
_insert_comment(tmp_db, "t5", "zhangfei-dev", "ok")
result = handler.verify_completion("t5", tmp_db)
assert result.passed is False
def test_review_merged_auto_passes(self, handler, tmp_db):
"""review_merged → always pass"""
must_haves = json.dumps({"action_type": "review_merged"})
_insert_task(tmp_db, "t6", must_haves)
result = handler.verify_completion("t6", tmp_db)
assert result.passed is True
assert result.reason == "merged_passthrough"
def test_infrastructure_failure_auto_passes(self, handler, tmp_db):
"""infrastructure_failure → always pass (anti-recursion)"""
must_haves = json.dumps({"action_type": "infrastructure_failure"})
_insert_task(tmp_db, "t7", must_haves)
result = handler.verify_completion("t7", tmp_db)
assert result.passed is True
assert result.reason == "infrastructure_passthrough"
# ---------------------------------------------------------------------------
# Step 3a: _send_toolchain_task tests
# ---------------------------------------------------------------------------
class TestSendToolchainTask:
def test_creates_task_in_toolchain_db(self):
"""_send_toolchain_task creates a task in _toolchain DB."""
from src.api.toolchain_routes import _send_toolchain_task, _toolchain_db_path
with patch("src.api.toolchain_routes.get_data_root") as mock_root:
with tempfile.TemporaryDirectory() as d:
mock_root.return_value = Path(d)
task_id = _send_toolchain_task(
to_agent="zhangfei-dev",
title="Test Task",
description="Test description",
event_type="ci_failure",
action_type="ci_failure",
steps=["Fix test", "Submit report"],
context_data={"pr_number": 42},
)
assert task_id.startswith("tc-")
# Verify task was written to _toolchain DB
db_path = _toolchain_db_path()
conn = get_connection(db_path)
row = conn.execute(
"SELECT * FROM tasks WHERE id=?", (task_id,)
).fetchone()
assert row is not None
assert row["task_type"] == "toolchain"
assert row["assignee"] == "zhangfei-dev"
# Verify must_haves JSON
meta = json.loads(row["must_haves"])
assert meta["event_type"] == "ci_failure"
assert meta["action_type"] == "ci_failure"
assert meta["steps"] == ["Fix test", "Submit report"]
assert meta["context"]["pr_number"] == 42
conn.close()
def test_unknown_agent_returns_empty(self):
"""_send_toolchain_task with unknown agent returns empty string."""
from src.api.toolchain_routes import _send_toolchain_task
task_id = _send_toolchain_task(
to_agent="unknown-agent",
title="Test",
description="desc",
event_type="test",
action_type="test",
steps=[],
)
assert task_id == ""
# ---------------------------------------------------------------------------
# Step 2e: on_failure three-way routing tests
# ---------------------------------------------------------------------------
class TestOnFailureRouting:
def test_business_failure_creates_gitea_comment(self, handler, tmp_db):
"""Business failure → Gitea PR comment @task assignee (not must_hives field)"""
# S4: must_hives does NOT contain assignee — production data doesn't have it
must_haves = json.dumps({
"action_type": "review_result",
"context": {"repo": "sanguo/test", "pr_number": 42},
"from": "system",
})
# assignee is set on the tasks table row (as production code writes it)
_insert_task(tmp_db, "t-fail", must_haves)
with patch.object(handler, "_create_gitea_comment") as mock_comment:
mock_comment.return_value = True
verify = VerifyResult(False, "no_action", "no action_report")
handler.on_failure("t-fail", "zhangfei-dev", tmp_db, verify)
mock_comment.assert_called_once()
call_args = mock_comment.call_args
assert call_args[0][0] == "sanguo/test"
assert call_args[0][1] == 42
# M2: comment body should @ the task's assignee from tasks table
comment_body = call_args[0][2]
assert "@zhangfei-dev" in comment_body
def test_infrastructure_failure_creates_task(self, handler, tmp_db):
"""Infrastructure failure → direct DB task for jiangwei-infra (no reverse dep)"""
must_haves = json.dumps({
"action_type": "review_result",
"context": {"repo": "sanguo/test", "pr_number": 42},
})
_insert_task(tmp_db, "t-infra", must_haves)
with patch.object(handler, "_create_gitea_comment") as mock_comment:
mock_comment.return_value = False # Gitea API down
with patch.object(handler, "_create_gitea_issue") as mock_issue:
mock_issue.return_value = False # Gitea API still down
verify = VerifyResult(False, "no_action", "no action_report")
handler.on_failure("t-infra", "zhangfei-dev", tmp_db, verify)
# S3: should directly INSERT into DB, not call _send_toolchain_task
# Verify a new task was created in DB for jiangwei-infra
conn = get_connection(tmp_db)
rows = conn.execute(
"SELECT * FROM tasks WHERE assignee=?",
("jiangwei-infra",)
).fetchall()
conn.close()
assert len(rows) >= 1, "No infrastructure_failure task created"
infra_task = rows[0]
assert infra_task["task_type"] == "toolchain"
meta = json.loads(infra_task["must_haves"])
assert meta["action_type"] == "infrastructure_failure"
# ---------------------------------------------------------------------------
# Regression: _mail path unaffected
# ---------------------------------------------------------------------------
class TestMailRegression:
def test_send_mail_still_exists(self):
"""_send_mail function is preserved."""
from src.api.toolchain_routes import _send_mail
assert callable(_send_mail)
def test_send_mail_not_called_by_handlers(self):
"""No toolchain handler calls _send_mail."""
import inspect
from src.api import toolchain_routes
# Get source of handler functions
source = inspect.getsource(toolchain_routes)
# _send_mail should appear only in its own definition, not in handler bodies
lines = source.split("\n")
in_handler = False
handler_send_mail_calls = []
for i, line in enumerate(lines):
if line.strip().startswith("async def _handle_") or line.strip().startswith("async def _send_mention_mails"):
in_handler = True
elif line.strip().startswith("async def ") or line.strip().startswith("def _"):
if not line.strip().startswith("async def _handle_") and not line.strip().startswith("async def _send_mention_mails"):
in_handler = False
if in_handler and "_send_mail(" in line and not line.strip().startswith("#"):
handler_send_mail_calls.append((i, line.strip()))
assert len(handler_send_mail_calls) == 0, \
f"_send_mail still called in handlers: {handler_send_mail_calls}"
# ---------------------------------------------------------------------------
# Integration: full prompt build
# ---------------------------------------------------------------------------
class TestFullPromptBuild:
def test_prompt_contains_all_sections(self, handler):
"""Full prompt has context, API, and constraints sections."""
ctx = PromptContext(
task_id="tc-test",
title="CI 失败修复",
description="Fix CI failure",
must_haves=json.dumps({
"event_type": "ci_failure",
"action_type": "ci_failure",
"steps": ["Fix test", "Push", "Submit report"],
"context": {"pr_number": 42},
}),
project_id="_toolchain",
agent_id="zhangfei-dev",
event_type="ci_failure",
event_data={"pr_number": "42", "repo": "sanguo/test"},
action_type="ci_failure",
action_steps=["Fix test", "Push", "Submit report"],
)
prompt = handler.build_prompt(ctx)
# Must have action hint
assert "CI 失败" in prompt
assert "需要你修复" in prompt
# Must have steps
assert "必须执行的步骤" in prompt
assert "1. Fix test" in prompt
# Must have API section with action_report
assert "action_report" in prompt
assert "tc-test" in prompt
# Must have constraints with Red Flags
assert "Red Flags" in prompt
assert "强制要求" in prompt