fix: M1 git rm 误提交的安装目录文件 + S1 docstring 修正 + S2 去掉 CHECK 约束（司马懿 Review #111）
This commit is contained in:
cfdaily
@@ -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' CHECK (comment_type IN ('general','handoff','observation','review','rebuttal','rebuttal_response','debate_argument','debate_rebuttal','debate_judgment','action_report')),
|
||||
comment_type TEXT NOT NULL DEFAULT 'general',
|
||||
body TEXT NOT NULL,
|
||||
mentions TEXT,
|
||||
created_at TEXT NOT NULL DEFAULT (datetime('now'))
|
||||
|
||||
@@ -330,7 +330,7 @@ class ToolchainHandler(BaseTaskHandler):
|
||||
task_id, agent_id, verify, db_path)
|
||||
|
||||
def _classify_failure(self, verify: VerifyResult) -> str:
|
||||
"""分类失败类型:business / system / infrastructure"""
|
||||
"""分类失败类型:business / infrastructure（system 通过升级到达）"""
|
||||
# verify_error 或 DB 不可用 → 基础设施失败
|
||||
if verify.reason == "verify_error":
|
||||
return "infrastructure"
|
||||
|
||||
@@ -1,169 +0,0 @@
|
||||
"""@mention 解析工具模块。供所有 toolchain handler 复用。"""
|
||||
|
||||
import re
|
||||
import logging
|
||||
from typing import List, Set
|
||||
|
||||
from src.config.agents import AGENT_IDS
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Gitea API 基地址常量（避免硬编码）
|
||||
GITEA_API_BASE = "http://192.168.2.154:3000/api/v1"
|
||||
GITEA_WEB_BASE = "http://192.168.2.154:3000"
|
||||
|
||||
# Agent 别名映射
|
||||
# 规则：
|
||||
# 1. 中文名（如"张飞"）→ 完整 Agent ID
|
||||
# 2. 英文短名（如"zhangfei"）→ 完整 Agent ID
|
||||
# 3. 前缀模糊匹配需唯一匹配（见 extract_mentions 假设 A2）
|
||||
AGENT_ALIAS: dict[str, str] = {
|
||||
# 中文名
|
||||
"张飞": "zhangfei-dev",
|
||||
"关羽": "guanyu-dev",
|
||||
"赵云": "zhaoyun-data",
|
||||
"姜维": "jiangwei-infra",
|
||||
"司马懿": "simayi-challenger",
|
||||
"庞统": "pangtong-fujunshi",
|
||||
# 字+号（常见写法）
|
||||
"翼德": "zhangfei-dev",
|
||||
"云长": "guanyu-dev",
|
||||
"子龙": "zhaoyun-data",
|
||||
"伯约": "jiangwei-infra",
|
||||
"仲达": "simayi-challenger",
|
||||
"士元": "pangtong-fujunshi",
|
||||
# 英文短名
|
||||
"zhangfei": "zhangfei-dev",
|
||||
"guanyu": "guanyu-dev",
|
||||
"zhaoyun": "zhaoyun-data",
|
||||
"jiangwei": "jiangwei-infra",
|
||||
"simayi": "simayi-challenger",
|
||||
"pangtong": "pangtong-fujunshi",
|
||||
}
|
||||
|
||||
# 正则：匹配 @后面跟着的合法 Agent 名（英文字母/中文/数字/连字符）
|
||||
_MENTION_PATTERN = re.compile(r"@([a-zA-Z\u4e00-\u9fa5][a-zA-Z0-9\u4e00-\u9fff-]*)")
|
||||
|
||||
|
||||
def extract_mentions(body: str, sender: str) -> list[str]:
|
||||
"""从文本中提取 @mention 的 Agent ID 列表。
|
||||
|
||||
Args:
|
||||
body: 评论文本
|
||||
sender: 评论者 Gitea 用户名（用于排除自己 @自己）
|
||||
|
||||
Returns:
|
||||
去重后的 Agent ID 列表
|
||||
|
||||
匹配优先级：精确 > 别名 > 前缀模糊（需唯一匹配，多候选则跳过）
|
||||
"""
|
||||
candidates = _MENTION_PATTERN.findall(body)
|
||||
result: Set[str] = set()
|
||||
|
||||
for c in candidates:
|
||||
# 1. 精确匹配（@zhangfei-dev）
|
||||
if c in AGENT_IDS:
|
||||
result.add(c)
|
||||
# 2. 别名匹配（@张飞、@zhangfei）
|
||||
elif c in AGENT_ALIAS:
|
||||
result.add(AGENT_ALIAS[c])
|
||||
else:
|
||||
# 3. 前缀模糊匹配（@zhangf → zhangfei-dev）
|
||||
# 假设 A2：多个候选时不匹配，只 log warning
|
||||
matches = [aid for aid in AGENT_IDS if aid.startswith(c)]
|
||||
if len(matches) == 1:
|
||||
result.add(matches[0])
|
||||
elif len(matches) > 1:
|
||||
logger.warning(
|
||||
"Prefix '%s' matched %d agents (%s), skipping ambiguous mention",
|
||||
c, len(matches), matches)
|
||||
|
||||
# 排除自己 @自己（假设 A1：Gitea login = Agent ID）
|
||||
result.discard(sender)
|
||||
return list(result)
|
||||
|
||||
|
||||
def should_suppress_mention(
|
||||
mentioned_agent: str,
|
||||
auto_notify_targets: List[str],
|
||||
) -> bool:
|
||||
"""判断 @mention 通知是否应被抑制（因为自动流转已通知同一人）。
|
||||
|
||||
Args:
|
||||
mentioned_agent: 被 @的 Agent ID
|
||||
auto_notify_targets: 本次事件自动流转已通知的目标列表
|
||||
|
||||
Returns:
|
||||
True 表示应抑制（不发 @mention Mail）
|
||||
"""
|
||||
return mentioned_agent in auto_notify_targets
|
||||
|
||||
|
||||
def infer_intent(body: str) -> str:
|
||||
"""从 @mention 内容推断意图。
|
||||
|
||||
Returns:
|
||||
"help" | "notify" | "collaborate" | "assign"
|
||||
"""
|
||||
# 分配子任务关键词
|
||||
assign_keywords = ["交给", "分配", "负责", "认领", "做一下", "帮忙做", "implement"]
|
||||
if any(kw in body for kw in assign_keywords):
|
||||
return "assign"
|
||||
|
||||
# 求助关键词（注意："帮忙"已由 assign_keywords 的"帮忙做"覆盖，"请帮忙"由 collab_keywords 覆盖）
|
||||
help_keywords = ["怎么", "如何", "？", "?", "什么", "哪个", "能否"]
|
||||
if any(kw in body for kw in help_keywords):
|
||||
return "help"
|
||||
|
||||
# 协作请求关键词
|
||||
collab_keywords = ["请帮忙", "请协助", "请澄清", "请review", "请审查", "评估"]
|
||||
if any(kw in body for kw in collab_keywords):
|
||||
return "collaborate"
|
||||
|
||||
# 默认为通知关注
|
||||
return "notify"
|
||||
|
||||
|
||||
def _build_response_guidance(
|
||||
intent: str,
|
||||
gitea_api: str,
|
||||
repo: str,
|
||||
issue_number: int,
|
||||
commenter: str,
|
||||
) -> str:
|
||||
"""根据意图类型生成响应指引文本。"""
|
||||
if intent == "help":
|
||||
return (
|
||||
f"这是一条求助，请到 Gitea 评论回复：\n"
|
||||
f"1. 获取评论上下文（上方 API）\n"
|
||||
f"2. 组织回答\n"
|
||||
f"3. 在 Gitea 评论回复: POST {gitea_api}/repos/{repo}/issues/{issue_number}/comments\n"
|
||||
f' Body: {{"body": "你的回答内容"}}'
|
||||
)
|
||||
elif intent == "notify":
|
||||
return (
|
||||
f"这是一条通知，请查看并知晓。如有意见，可到 Gitea 评论：\n"
|
||||
f"- 查看 Issue/PR 详情（上方 API）\n"
|
||||
f"- 如有意见，评论回复: POST {gitea_api}/repos/{repo}/issues/{issue_number}/comments"
|
||||
)
|
||||
elif intent == "collaborate":
|
||||
return (
|
||||
f"这是一条协作请求，请评估后回复（评论或 Mail）：\n"
|
||||
f"1. 获取详情（上方 API）\n"
|
||||
f"2. 评估可行性\n"
|
||||
f"3a. 评论回复: POST {gitea_api}/repos/{repo}/issues/{issue_number}/comments\n"
|
||||
f' Body: {{"body": "你的回复"}}\n'
|
||||
f"3b. 或通过 Mail 回复评论者: {commenter}"
|
||||
)
|
||||
elif intent == "assign":
|
||||
return (
|
||||
f"这是一条任务分配，请认领并执行：\n"
|
||||
f"1. 获取 Issue 详情（上方 API）\n"
|
||||
f"2. 评估可行性\n"
|
||||
f"3. 认领 Issue: POST {gitea_api}/repos/{repo}/issues/{issue_number}/assignees\n"
|
||||
f' Body: {{"assignees": ["{{your_agent_id}}"]}}\n'
|
||||
f"4. 执行任务\n"
|
||||
f"5. 完成后更新 Issue 状态: PATCH {gitea_api}/repos/{repo}/issues/{issue_number}\n"
|
||||
f' Body: {{"state": "closed"}}'
|
||||
)
|
||||
return "请查看详情（上方 API）并按需回复。"
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,129 +0,0 @@
|
||||
"""
|
||||
prompt_composer.py — PromptSection Protocol + PromptContext + PromptComposer
|
||||
|
||||
拼装器：有序管理 prompt 段落，按优先级排序后合并为最终 prompt。
|
||||
"""
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Dict, List, Optional, Protocol, runtime_checkable
|
||||
|
||||
logger = logging.getLogger("moziplus-v2.prompt_composer")
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Section 优先级范围约定
|
||||
# ---------------------------------------------------------------------------
|
||||
PRIORITY_CONTEXT = 10 # 任务上下文
|
||||
PRIORITY_PRIOR = 20 # 前序信息
|
||||
PRIORITY_ROLE = 30 # 角色规范
|
||||
PRIORITY_API = 40 # API 操作指令
|
||||
PRIORITY_CONSTRAINTS = 50 # 硬约束
|
||||
PRIORITY_EXTENSION = 60 # 扩展段
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PromptSection Protocol
|
||||
# ---------------------------------------------------------------------------
|
||||
@runtime_checkable
|
||||
class PromptSection(Protocol):
|
||||
"""一个 prompt 段"""
|
||||
|
||||
name: str # 段名（去重用，同名覆盖）
|
||||
priority: int # 排序优先级（小数字=靠前）
|
||||
|
||||
def render(self, context: "PromptContext") -> str:
|
||||
"""渲染此段的文本内容。返回空字符串表示不注入。"""
|
||||
...
|
||||
|
||||
def should_include(self, context: "PromptContext") -> bool:
|
||||
"""是否注入此段（默认 True，条件段可覆盖）。"""
|
||||
...
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PromptContext 数据对象
|
||||
# ---------------------------------------------------------------------------
|
||||
@dataclass
|
||||
class PromptContext:
|
||||
"""Prompt 渲染的统一上下文"""
|
||||
|
||||
task_id: str
|
||||
title: str
|
||||
description: str
|
||||
must_haves: str
|
||||
project_id: str
|
||||
agent_id: str
|
||||
|
||||
task: Optional[Dict] = None
|
||||
role: str = "executor"
|
||||
spawn_type: str = "executor"
|
||||
|
||||
# mail 专用
|
||||
from_agent: str = ""
|
||||
mail_type: str = "" # inform / request
|
||||
|
||||
# toolchain 专用
|
||||
event_type: str = "" # ci_failure / review_request / ...
|
||||
event_data: Dict = field(default_factory=dict)
|
||||
action_type: str = "" # 动作分类（review_result / ci_failure / ...）
|
||||
action_steps: list = field(default_factory=list) # 结构化编号步骤列表
|
||||
|
||||
# 前序产出
|
||||
depends_on_outputs: Optional[List] = None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PromptComposer 拼装器
|
||||
# ---------------------------------------------------------------------------
|
||||
class PromptComposer:
|
||||
"""有序拼装 prompt sections"""
|
||||
|
||||
SEPARATOR = "\n\n---\n\n"
|
||||
TOKEN_BUDGET_WARN = 800 # token 预算警告阈值
|
||||
CHARS_PER_TOKEN = 3.5 # 估算比率
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._sections: List[PromptSection] = []
|
||||
|
||||
def add(self, section: PromptSection) -> None:
|
||||
"""添加一个 section（同名覆盖）"""
|
||||
self._sections = [s for s in self._sections if s.name != section.name]
|
||||
self._sections.append(section)
|
||||
|
||||
def add_many(self, sections: List[PromptSection]) -> None:
|
||||
"""批量添加"""
|
||||
for s in sections:
|
||||
self.add(s)
|
||||
|
||||
def compose(self, context: PromptContext) -> str:
|
||||
"""拼装最终 prompt
|
||||
|
||||
1. 过滤 should_include=False 的段
|
||||
2. 按 priority 排序
|
||||
3. 逐段 render
|
||||
4. 过滤空段
|
||||
5. 用分隔符连接
|
||||
6. Token 预算警告（不截断）
|
||||
"""
|
||||
active = [s for s in self._sections if s.should_include(context)]
|
||||
active.sort(key=lambda s: s.priority)
|
||||
|
||||
parts = [s.render(context) for s in active]
|
||||
parts = [p for p in parts if p.strip()]
|
||||
|
||||
result = self.SEPARATOR.join(parts)
|
||||
|
||||
# Token 估算
|
||||
tokens = max(1, int(len(result) / self.CHARS_PER_TOKEN))
|
||||
logger.debug(
|
||||
"Composed prompt from %d sections, %d tokens",
|
||||
len(parts), tokens,
|
||||
)
|
||||
|
||||
if tokens > self.TOKEN_BUDGET_WARN:
|
||||
logger.warning(
|
||||
"Prompt exceeds %d token budget: %d tokens (task_id=%s)",
|
||||
self.TOKEN_BUDGET_WARN, tokens, context.task_id,
|
||||
)
|
||||
|
||||
return result
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,512 +0,0 @@
|
||||
"""toolchain_handler.py - 工具链事件 handler。
|
||||
|
||||
处理 Gitea Webhook 事件(CI 失败、Review 请求、Issue 指派等)。
|
||||
L2 引擎层强约束:输入(结构化步骤)+ 执行(Red Flags)+ 输出(action_report 验证)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Dict, List
|
||||
|
||||
from src.daemon.base_task_handler import BaseTaskHandler, VerifyResult
|
||||
from src.daemon.prompt_composer import PromptComposer, PromptContext
|
||||
from src.daemon.toolchain_templates import render_template, _TEMPLATE_MAP
|
||||
from src.blackboard.db import get_connection
|
||||
|
||||
logger = logging.getLogger("moziplus-v2.handler.toolchain")
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Gitea API 配置
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_GITEA_BASE = "http://192.168.2.154:3000/api/v1"
|
||||
_GITEA_TOKEN = "a6d596b826f4bfeaf983ef4d25ac25dab95bbc4e"
|
||||
|
||||
# 业务失败连续次数阈值,超过则升级为系统失败
|
||||
_BUSINESS_FAIL_THRESHOLD = 3
|
||||
|
||||
# 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)"""
|
||||
|
||||
name: str = "toolchain_context"
|
||||
priority: int = 10
|
||||
|
||||
def render(self, context: PromptContext) -> str:
|
||||
event_type = context.event_type
|
||||
event_data: Dict = context.event_data or {}
|
||||
|
||||
# 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)
|
||||
|
||||
# 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}"
|
||||
|
||||
def should_include(self, context: PromptContext) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
class ToolchainApiSection:
|
||||
"""API 操作指令(priority=40)-- action_report 提交指引"""
|
||||
|
||||
name: str = "toolchain_api"
|
||||
priority: int = 40
|
||||
|
||||
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}`",
|
||||
"",
|
||||
"### 完成后必须提交 action report",
|
||||
"",
|
||||
"执行完所有步骤后,必须提交 action report:",
|
||||
"```bash",
|
||||
f'curl -s -X POST "http://{self.API_HOST}/api/projects/{project_id}/tasks/{task_id}/comments" \\',
|
||||
' -H "Content-Type: application/json" \\',
|
||||
f' -d \'{{"author": "{agent_id}", "comment_type": "action_report", "body": "简要描述你执行了什么操作及结果"}}\'',
|
||||
"```",
|
||||
"",
|
||||
"⚠️ 不提交 action report 的任务会被标记为 failed。",
|
||||
"",
|
||||
"### 提交产出",
|
||||
"",
|
||||
"如有产出(如 review 结果、修复方案),提交到任务 outputs:",
|
||||
"```bash",
|
||||
f'curl -s -X POST "http://{self.API_HOST}/api/projects/{project_id}/tasks/{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" \\',
|
||||
f' -H "Authorization: token <your-token>" \\',
|
||||
' -H "Content-Type: application/json" \\',
|
||||
' -d \'{"body": "@{agent-id} 需要你的支持:{描述问题}"}\'',
|
||||
"```",
|
||||
"",
|
||||
"⚠️ 不要使用 Mail API(飞鸽传书)。所有协作通过 Gitea 留痕。",
|
||||
"",
|
||||
]
|
||||
return "\n".join(lines)
|
||||
|
||||
def should_include(self, context: PromptContext) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
class ToolchainConstraintsSection:
|
||||
"""硬约束 + Red Flags(priority=50)"""
|
||||
|
||||
name: str = "toolchain_constraints"
|
||||
priority: int = 50
|
||||
|
||||
def render(self, context: PromptContext) -> str:
|
||||
lines = [
|
||||
"## 硬约束（必须遵守）",
|
||||
"",
|
||||
"⚠️ 以下是强制要求，不是建议或参考。违反任何一条都会导致任务失败。",
|
||||
"",
|
||||
"### 1. 必须按步骤执行",
|
||||
'- 检查上方“必须执行的步骤”列表',
|
||||
'- 逐条执行每个步骤，不可跳过',
|
||||
'- 不要只读不做——这不是纯通知',
|
||||
"",
|
||||
"### 2. 必须提交 action report",
|
||||
'- 执行完所有步骤后，必须提交 action report',
|
||||
"- 提交方式：POST comment（comment_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 中说明原因，但其他步骤必须执行 |',
|
||||
"",
|
||||
]
|
||||
return "\n".join(lines)
|
||||
|
||||
def should_include(self, context: PromptContext) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ToolchainHandler
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class ToolchainHandler(BaseTaskHandler):
|
||||
"""工具链事件 handler。"""
|
||||
|
||||
task_type = "toolchain"
|
||||
virtual_project = "_toolchain"
|
||||
display_name = "工具链事件"
|
||||
|
||||
def target_success_status(self) -> str:
|
||||
return "done"
|
||||
|
||||
def pre_spawn(self, task_id: str, db_path: Path) -> bool:
|
||||
"""auto_working:pending → working"""
|
||||
return self._auto_mark_working(task_id, db_path)
|
||||
|
||||
def get_sections(self) -> list:
|
||||
"""返回 3 个 Toolchain PromptSection 实例"""
|
||||
return [
|
||||
ToolchainContextSection(),
|
||||
ToolchainApiSection(),
|
||||
ToolchainConstraintsSection(),
|
||||
]
|
||||
|
||||
def build_prompt(self, context: PromptContext) -> str:
|
||||
"""通过 PromptComposer 拼装 sections 为最终 prompt"""
|
||||
composer = PromptComposer()
|
||||
composer.add_many(self.get_sections())
|
||||
return composer.compose(context)
|
||||
|
||||
def verify_completion(self, task_id: str, db_path: Path) -> VerifyResult:
|
||||
"""检查 action report(精确验证)+ 三层 fallback"""
|
||||
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_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_count = conn.execute(
|
||||
"SELECT COUNT(*) FROM comments WHERE task_id=? "
|
||||
"AND author != 'system' AND LENGTH(body) >= 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")
|
||||
finally:
|
||||
conn.close()
|
||||
except Exception as e:
|
||||
logger.error("Toolchain %s: verify error: %s", task_id, e)
|
||||
return VerifyResult(False, "verify_error", str(e))
|
||||
|
||||
def on_failure(self, task_id: str, agent_id: str,
|
||||
db_path: Path, verify: VerifyResult) -> None:
|
||||
"""验证失败 → 三分路处理(业务/系统/基础设施)"""
|
||||
self._mark_task_status(db_path, task_id, "failed")
|
||||
logger.info("Toolchain %s: verify failed (%s), marked failed",
|
||||
task_id, verify.reason)
|
||||
|
||||
# 读取 must_haves 获取事件上下文
|
||||
meta = {}
|
||||
try:
|
||||
conn = get_connection(db_path)
|
||||
row = conn.execute(
|
||||
"SELECT must_haves FROM tasks WHERE id=?", (task_id,)
|
||||
).fetchone()
|
||||
if row and row["must_haves"]:
|
||||
meta = json.loads(row["must_haves"])
|
||||
# 统计该 task 的业务失败次数
|
||||
fail_count = conn.execute(
|
||||
"SELECT COUNT(*) FROM events WHERE task_id=? "
|
||||
"AND event_type='status_change' AND payload LIKE '%failed%'",
|
||||
(task_id,)
|
||||
).fetchone()[0]
|
||||
conn.close()
|
||||
except Exception:
|
||||
fail_count = 0
|
||||
|
||||
action_type = meta.get("action_type", "")
|
||||
context_data = meta.get("context", {})
|
||||
assignee = meta.get("assignee", "") or meta.get("from", "")
|
||||
|
||||
# 三分路决策
|
||||
route = self._classify_failure(verify, fail_count)
|
||||
|
||||
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, fail_count: int) -> str:
|
||||
"""分类失败类型:business / system / infrastructure"""
|
||||
# verify_error 或 DB 不可用 → 基础设施失败
|
||||
if verify.reason == "verify_error":
|
||||
return "infrastructure"
|
||||
# 连续业务失败超过阈值 → 升级为系统失败
|
||||
if fail_count >= _BUSINESS_FAIL_THRESHOLD:
|
||||
return "system"
|
||||
# 默认:业务失败
|
||||
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 请检查黑板任务并手动处理。"
|
||||
)
|
||||
|
||||
# 尝试在 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:
|
||||
"""基础设施失败 → _send_toolchain_task @jiangwei-infra(防递归)"""
|
||||
# 直接在 _toolchain DB 创建 task(不走 Gitea webhook)
|
||||
try:
|
||||
from src.api.toolchain_routes import _send_toolchain_task
|
||||
_send_toolchain_task(
|
||||
to_agent="jiangwei-infra",
|
||||
title=f"[基础设施] Gitea API 不可用 - {task_id}",
|
||||
description=(
|
||||
f"Gitea API 不可用,原任务 {task_id} 无法通过正常路径处理。\n"
|
||||
f"请检查 Gitea 服务状态和网络连通性。"
|
||||
),
|
||||
event_type="infrastructure_failure",
|
||||
action_type="infrastructure_failure",
|
||||
steps=[
|
||||
"检查 Gitea 服务状态(http://192.168.2.154:3000)",
|
||||
"检查网络连通性",
|
||||
"恢复后提交 action report",
|
||||
],
|
||||
context_data={"original_task_id": task_id, "verify_reason": verify.reason},
|
||||
source="toolchain_handler",
|
||||
)
|
||||
logger.info("Toolchain %s: infrastructure failure → task created for jiangwei-infra",
|
||||
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。返回是否成功。"""
|
||||
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。返回是否成功。"""
|
||||
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 = []
|
||||
repo = event_data.get("repo", "")
|
||||
base_url = "http://192.168.2.154:3000"
|
||||
|
||||
if "pr_number" in event_data:
|
||||
links.append(f"PR: {base_url}/{repo}/pulls/{event_data['pr_number']}")
|
||||
if "issue_number" in event_data:
|
||||
links.append(f"Issue: {base_url}/{repo}/issues/{event_data['issue_number']}")
|
||||
if "commit" in event_data:
|
||||
links.append(f"Commit: {base_url}/{repo}/commit/{event_data['commit']}")
|
||||
if "branch" in event_data and "commit" not in event_data:
|
||||
links.append(f"分支: {event_data['branch']}")
|
||||
|
||||
return "\n".join(links) if links else "(无法提取链接,请检查黑板任务详情)"
|
||||
@@ -1,89 +0,0 @@
|
||||
"""工具链事件模板引擎（Toolchain Event Hub）
|
||||
|
||||
加载 templates/toolchain/ 下的 Markdown 模板，提供 {variable} 占位符渲染。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
TEMPLATES_DIR = Path(__file__).parent.parent.parent / "templates" / "toolchain"
|
||||
|
||||
# 模板文件名映射
|
||||
_TEMPLATE_MAP: Dict[str, str] = {
|
||||
"review_request": "review_request.md",
|
||||
"review_result": "review_result.md",
|
||||
"issue_assigned": "issue_assigned.md",
|
||||
"ci_failure": "ci_failure.md",
|
||||
"deploy_failure": "deploy_failure.md",
|
||||
"review_updated": "review_updated.md",
|
||||
"review_comment": "review_comment.md",
|
||||
"review_merged": "review_merged.md",
|
||||
"mention": "mention.md",
|
||||
}
|
||||
|
||||
# 模板缓存
|
||||
_template_cache: Dict[str, str] = {}
|
||||
|
||||
|
||||
def _load_template(name: str) -> str:
|
||||
"""加载并缓存模板文件内容。
|
||||
|
||||
Args:
|
||||
name: 模板名称（不含 .md 后缀）
|
||||
|
||||
Returns:
|
||||
模板文本内容
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: 模板文件不存在
|
||||
"""
|
||||
if name in _template_cache:
|
||||
return _template_cache[name]
|
||||
|
||||
filename = _TEMPLATE_MAP.get(name)
|
||||
if not filename:
|
||||
raise ValueError(f"Unknown template: {name}")
|
||||
|
||||
path = TEMPLATES_DIR / filename
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(f"Template not found: {path}")
|
||||
|
||||
content = path.read_text(encoding="utf-8")
|
||||
_template_cache[name] = content
|
||||
logger.debug("Loaded template: %s (%d bytes)", name, len(content))
|
||||
return content
|
||||
|
||||
|
||||
def _escape_braces(value: str) -> str:
|
||||
"""转义花括号防止 format_map 报错"""
|
||||
return str(value).replace("{", "{{").replace("}", "}}")
|
||||
|
||||
|
||||
def render_template(name: str, variables: Dict[str, str]) -> str:
|
||||
"""渲染模板，将 {variable} 占位符替换为实际值。
|
||||
|
||||
使用 defaultdict(str) 确保未提供的变量替换为空字符串而非报错。
|
||||
|
||||
Args:
|
||||
name: 模板名称
|
||||
variables: 变量字典
|
||||
|
||||
Returns:
|
||||
渲染后的文本
|
||||
"""
|
||||
template_text = _load_template(name)
|
||||
# 先对所有变量值转义花括号，防止 format_map 报错
|
||||
escaped_vars = {k: _escape_braces(v) for k, v in variables.items()}
|
||||
safe_vars: Dict[str, str] = defaultdict(str, escaped_vars)
|
||||
return template_text.format_map(safe_vars)
|
||||
|
||||
|
||||
def clear_cache() -> None:
|
||||
"""清空模板缓存（用于测试或热更新）"""
|
||||
_template_cache.clear()
|
||||
@@ -1,16 +0,0 @@
|
||||
{mention_type}通知
|
||||
|
||||
来源: {source_type} {source_url}
|
||||
评论者: {commenter}
|
||||
意图: {intent_hint}
|
||||
内容:
|
||||
{content_snippet}
|
||||
|
||||
📋 获取完整上下文：
|
||||
1. 查看{source_type}详情: GET {gitea_api}/repos/{repo}/{source_detail_api_path}
|
||||
2. 查看评论列表: GET {gitea_api}/repos/{repo}/{source_comments_api_path}
|
||||
|
||||
📌 响应指引：
|
||||
{response_guidance}
|
||||
|
||||
完成后按指引操作。
|
||||
@@ -1,129 +0,0 @@
|
||||
"""mention_utils 单元测试 — §25.7 覆盖。"""
|
||||
|
||||
import pytest
|
||||
|
||||
from src.api.mention_utils import (
|
||||
extract_mentions,
|
||||
should_suppress_mention,
|
||||
infer_intent,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# extract_mentions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestExtractMentions:
|
||||
"""测试 @mention 提取逻辑。"""
|
||||
|
||||
def test_exact_match(self):
|
||||
"""@zhangfei-dev 精确匹配。"""
|
||||
assert extract_mentions("@zhangfei-dev 请看一下", "someone") == ["zhangfei-dev"]
|
||||
|
||||
def test_chinese_alias(self):
|
||||
"""@张飞 中文别名匹配。"""
|
||||
assert extract_mentions("@张飞 帮忙看看", "someone") == ["zhangfei-dev"]
|
||||
|
||||
def test_english_short_name(self):
|
||||
"""@zhangfei 英文短名匹配。"""
|
||||
assert extract_mentions("@zhangfei 快来", "someone") == ["zhangfei-dev"]
|
||||
|
||||
def test_prefix_unique(self):
|
||||
"""@zhangf 前缀唯一匹配。"""
|
||||
assert extract_mentions("@zhangf 来一下", "someone") == ["zhangfei-dev"]
|
||||
|
||||
def test_prefix_ambiguous_no_match(self):
|
||||
"""@z 前缀模糊，多个候选，不匹配。"""
|
||||
assert extract_mentions("@z 看看", "someone") == []
|
||||
|
||||
def test_dedup_same_person(self):
|
||||
"""@张飞 @zhangfei-dev 同时出现去重。"""
|
||||
result = extract_mentions("@张飞 @zhangfei-dev 来一下", "someone")
|
||||
assert result == ["zhangfei-dev"]
|
||||
|
||||
def test_exclude_self(self):
|
||||
"""@zhangfei-dev 排除自己（sender=zhangfei-dev）。"""
|
||||
assert extract_mentions("@zhangfei-dev 自己说", "zhangfei-dev") == []
|
||||
|
||||
def test_unknown_person(self):
|
||||
"""@unknown 不匹配任何 Agent。"""
|
||||
assert extract_mentions("@unknown 你好", "someone") == []
|
||||
|
||||
def test_multiple_mentions(self):
|
||||
"""多个 @mention 返回多个 Agent。"""
|
||||
result = set(extract_mentions("@张飞 @关羽 来讨论", "someone"))
|
||||
assert result == {"zhangfei-dev", "guanyu-dev"}
|
||||
|
||||
def test_mention_with_hyphen_in_middle(self):
|
||||
"""@mention 后面紧跟标点也能识别。"""
|
||||
result = extract_mentions("@赵云，请看下", "someone")
|
||||
assert result == ["zhaoyun-data"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# should_suppress_mention
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestShouldSuppressMention:
|
||||
"""测试 @mention 通知抑制逻辑。"""
|
||||
|
||||
def test_suppress_when_in_list(self):
|
||||
"""被提及者在自动通知列表中 → 抑制。"""
|
||||
assert should_suppress_mention("zhangfei-dev", ["zhangfei-dev", "guanyu-dev"]) is True
|
||||
|
||||
def test_not_suppress_when_not_in_list(self):
|
||||
"""被提及者不在自动通知列表中 → 不抑制。"""
|
||||
assert should_suppress_mention("zhangfei-dev", ["guanyu-dev"]) is False
|
||||
|
||||
def test_suppress_empty_list(self):
|
||||
"""自动通知列表为空 → 不抑制。"""
|
||||
assert should_suppress_mention("zhangfei-dev", []) is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# infer_intent
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestInferIntent:
|
||||
"""测试意图推断逻辑。
|
||||
|
||||
优先级：assign → collaborate → help → notify（默认）
|
||||
"""
|
||||
|
||||
def test_help_question_mark(self):
|
||||
"""疑问句 → help。"""
|
||||
assert infer_intent("@赵云 数据格式是什么？") == "help"
|
||||
|
||||
def test_notify_plain_mention(self):
|
||||
"""纯通知（无关键词） → notify。"""
|
||||
assert infer_intent("@关羽 这个 PR 涉及风控变更") == "notify"
|
||||
|
||||
def test_collaborate_please_help(self):
|
||||
"""'请帮忙' → collaborate（NOT help!）。"""
|
||||
assert infer_intent("@庞统 请帮忙澄清需求") == "collaborate"
|
||||
|
||||
def test_assign_keywords(self):
|
||||
"""'交给你' → assign。"""
|
||||
assert infer_intent("@张飞 前端部分交给你") == "assign"
|
||||
|
||||
def test_help_how_to(self):
|
||||
"""'如何' → help。"""
|
||||
assert infer_intent("@姜维 如何部署这个服务") == "help"
|
||||
|
||||
def test_collaborate_please_review(self):
|
||||
"""'请review' → collaborate。"""
|
||||
assert infer_intent("@司马懿 请review 这个方案") == "collaborate"
|
||||
|
||||
def test_notify_default(self):
|
||||
"""无任何关键词 → notify。"""
|
||||
assert infer_intent("@赵云 已更新数据") == "notify"
|
||||
|
||||
def test_assign_takes_priority_over_help(self):
|
||||
"""assign 关键词优先于 help 关键词。"""
|
||||
# "交给" in body → assign, even though "？" also present
|
||||
assert infer_intent("@张飞 这个模块交给你，有问题？") == "assign"
|
||||
|
||||
def test_collaborate_takes_priority_over_help(self):
|
||||
"""collaborate 关键词优先于 help 关键词。"""
|
||||
# "请帮忙" in body → collaborate, even though "？" absent
|
||||
assert infer_intent("@赵云 请帮忙看看数据") == "collaborate"
|
||||
Reference in New Issue
Block a user