diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index bf0b0a7..63d331f 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -623,6 +623,66 @@ async def _get_workspace_dir(sor, project_id): return _resolve_workspace(ws) +def _parse_skill_frontmatter(content): + """解析技能 SKILL.md 的 frontmatter（--- 之间），返回 dict。无 frontmatter 返回 {}。""" + content = (content or "").strip() + if not content.startswith("---"): + return {} + end = content.find("---", 3) + if end == -1: + return {} + fm = {} + for line in content[3:end].strip().split("\n"): + line = line.strip() + if ":" in line: + k, _, v = line.partition(":") + k = k.strip() + v = v.strip().strip('"').strip("'") + if v.startswith("[") and v.endswith("]"): + v = [x.strip().strip('"').strip("'") for x in v[1:-1].split(",") if x.strip()] + fm[k] = v + return fm + + +def _collect_module_skills(workspace_dir): + """扫描项目 workspace 的 repos/*/skill/SKILL.md，收集模块技能（模块怎么用）。 + + 模块仓库自带 skill/SKILL.md（架构/数据模型/挂载函数/坑位），但这些不进 skill_loader + 的静态文件树，导致项目角色不知道模块怎么用。这里运行时扫描，作为「项目模块」scope + 注入技能目录 + 支持 load_skill 加载全文。返回 [{name, description, content, path}]。 + """ + modules = [] + repos_dir = os.path.join(workspace_dir, 'repos') + if not os.path.isdir(repos_dir): + return modules + for entry in sorted(os.listdir(repos_dir)): + skill_file = os.path.join(repos_dir, entry, 'skill', 'SKILL.md') + if not os.path.isfile(skill_file): + continue + try: + with open(skill_file, 'r', encoding='utf-8') as f: + content = f.read() + except Exception: + continue + fm = _parse_skill_frontmatter(content) + name = (fm.get('name') or entry).strip() + description = (fm.get('description') or '').strip() + if not description: + for line in content.split("\n"): + line = line.strip() + if line and not line.startswith("#") and not line.startswith("---"): + description = line[:200] + break + modules.append({ + "name": name, + "description": description, + "content": content, + "path": skill_file, + "repo": entry, + }) + return modules + + async def _build_qna_section(sor, task_id, role, agent_id=None): from .communication import get_task_qa qa = await get_task_qa(task_id, role=role, agentid=agent_id) @@ -780,6 +840,9 @@ async def _create_next_task(sor, project_id, task, next_role, pm_comment=''): new_title = f"{title}（{next_role}阶段）" new_params = {**params, 'previous_role': _normalize_role(getattr(task, 'role', '')), 'previous_task_id': getattr(task, 'id', ''), 'pm_comment': pm_comment} + # 任务来源标记：系统自动派生（design→develop→deploy_test→test）都是「新开发」链， + # 不含 bug 修复语义。develop 角色据此判断是否要走 fix_bug 状态机。 + new_params['task_kind'] = 'new_dev' # 清除 pm_assigned：它是「PM 按模块清单派发」的标记，只作用于当前任务； # 下一角色任务是系统自动创建（非 PM 派发），若继承会污染——例如 PM 派发的 design 任务 # 带 pm_assigned=True，design approved 后自动创建的 develop 任务继承了它，被 1785 行的 @@ -859,6 +922,8 @@ async def _rollback_task_chain(sor, project_id, task_id, rollback_role, comment) except (json.JSONDecodeError, TypeError): tp = {} new_params = {**tp, 'rollback_from': task_id, 'rollback_comment': comment} + # 任务来源标记：回退重做 = 修复导致上游失败的缺陷（质检重做），develop 据此必须走 bug 状态机。 + new_params['task_kind'] = 'rework' # 清 pm_assigned：回退重做任务是系统重建的任务（非 PM 按模块清单派发），继承 target_task 的 # pm_assigned 会污染——若 target 是应用级 develop（历史脏数据带 pm_assigned=True），approve 后被 # 「模块级 develop → 跳过 deploy_test」误判，任务链断在 approved。与 _create_next_task 的 pop 对齐。 @@ -1088,8 +1153,6 @@ async def _build_role_skills_block(sor, project_id, role, org_id=""): loader = get_skill_loader(skills_dir) merged = loader.get_merged(pipeline_id=pid, role=role, project_id=project_id, org_id=org_id or '0') - if not merged: - return "" skills = sorted(merged.values(), key=lambda s: (-int(getattr(s, 'essential', False)), -SCOPE_PRIORITY.get(getattr(s, 'scope', ''), 0))) @@ -1097,6 +1160,17 @@ async def _build_role_skills_block(sor, project_id, role, org_id=""): for s in skills[:60]: tag = SCOPE_TAG.get(getattr(s, 'scope', ''), getattr(s, 'scope', '')) lines.append(f"- [{tag}] {s.name}: {(s.description or '')[:100]}") + # 项目模块技能：workspace repos/*/skill/SKILL.md（模块怎么用——架构/数据模型/挂载函数/坑位）， + # 运行时扫描注入，让项目角色知道引用了哪些模块、每个模块怎么挂载（load_xxx 入口/库名/坑）。 + try: + ws = await _get_workspace_dir(sor, project_id) + module_skills = _collect_module_skills(ws) + if module_skills: + lines.append("\n## 项目模块技能（本项目引用的业务模块，load_skill 加载全文了解模块怎么用）") + for m in module_skills: + lines.append(f"- [项目模块] {m['name']}: {(m['description'] or '')[:100]}") + except Exception as e: + logger.warning(f"module skills collect failed: {e}") return "\n".join(lines) except Exception as e: logger.warning(f"role skills load failed: {e}") @@ -1154,6 +1228,14 @@ async def _load_skill_by_name(sor, project_id, role, org_id, name, file_path=Non merged = loader.get_merged(pipeline_id=pid, role=role, project_id=project_id, org_id=org_id or '0') if name not in merged: + # 项目模块技能（workspace repos/*/skill/SKILL.md）不在 skill_loader 静态树里，运行时补查 + try: + ws = await _get_workspace_dir(sor, project_id) + for m in _collect_module_skills(ws): + if m['name'] == name or m['repo'] == name: + return f"## [项目模块] {m['name']}\n{m['description']}\n\n{m['content']}" + except Exception: + pass names = ", ".join(sorted(merged.keys())) or "（无可用技能）" return f"FAIL: 技能 '{name}' 不存在。可用技能: {names}" skill = merged[name] @@ -1237,6 +1319,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): "iteration_id": _iter.get('id', '') if _iter else '', "who": role, "agent_id": agent_id, + "task_id": task_id, } system = (AGENT_SYSTEM_PROMPT @@ -1483,6 +1566,13 @@ async def _pm_create_tasks(sor, project_id, params, parent_task_id=None): cancelled.append(did) tparams = {"description": desc, "pm_assigned": True} + # 任务来源标记：PM 派发的任务按 title/description 判断—— + # 含「修复 Bug」= bug_fix（develop 必须走 fix_bug 状态机）；否则 = new_dev（模块开发等）。 + _title_desc = f"{title} {desc}" + if '修复 Bug' in _title_desc or '修复Bug' in _title_desc or '修复bug' in _title_desc: + tparams['task_kind'] = 'bug_fix' + else: + tparams['task_kind'] = 'new_dev' if iteration_name: tparams["iteration_id"] = iteration_name tid = getID() diff --git a/pipeline_service/bug_capability.py b/pipeline_service/bug_capability.py index aab5e25..c5e5f9d 100644 --- a/pipeline_service/bug_capability.py +++ b/pipeline_service/bug_capability.py @@ -11,6 +11,7 @@ scope 约定：sd_bugs 无 project_id 列，用 iteration_id 做范围校验（C """ import logging +import json from datetime import datetime from sqlor.dbpools import DBPools from appPublic.uniqueID import getID @@ -223,3 +224,46 @@ def _rec_to_dict(rec): except Exception: pass return {} + + +async def classify_task(task_id, iteration_id="", who=None, agent_id=None): + """判断任务来源：new_dev(新开发) / bug_fix(修复 Bug) / rework(质检重做)。 + + 优先读任务 params.task_kind 标记（三条创建路径已打）；老任务无标记则按 + params.rollback_from / title 兜底推断。develop 据此决定是否走 fix_bug 状态机： + - bug_fix / rework：任务开始 start_fix、完成 fix_bug + - new_dev：纯开发，不走 bug 状态机（除非过程中自己 report 了新 bug） + + 返回 (ok, {"task_kind": ..., "reason": ..., "title": ...} 或错误消息)。 + """ + if not task_id: + return False, "缺少 task_id" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT id, title, params FROM pipeline_tasks WHERE id=${tid}$", {"tid": task_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return False, f"任务不存在: {task_id}" + rec = recs[0] + title = getattr(rec, 'title', '') or '' + params_str = getattr(rec, 'params', '{}') or '{}' + try: + params = json.loads(params_str) if isinstance(params_str, str) else (params_str or {}) + except (json.JSONDecodeError, TypeError): + params = {} + + kind = (params.get('task_kind') or '').strip() + reason = "params.task_kind 标记" + if not kind: + # 老任务兜底：rollback_from 存在 → rework；title 含「修复 Bug」→ bug_fix；否则 new_dev + if params.get('rollback_from'): + kind = 'rework' + reason = "params.rollback_from 存在（回退重做）" + elif ('修复 Bug' in title) or ('修复Bug' in title) or ('修复bug' in title): + kind = 'bug_fix' + reason = "title 含「修复 Bug」" + else: + kind = 'new_dev' + reason = "无 bug 相关标记，默认新开发" + return True, {"task_id": task_id, "task_kind": kind, "title": title, "reason": reason} diff --git a/pipeline_service/capability_tools.py b/pipeline_service/capability_tools.py index fffdffe..e8340c2 100644 --- a/pipeline_service/capability_tools.py +++ b/pipeline_service/capability_tools.py @@ -139,6 +139,12 @@ TOOL_SCHEMAS = { "params": {"status": "状态(可选)", "severity": "严重度(可选)"}, "required": [], }, + "classify_task": { + "module": "bug_capability", + "description": "判断任务来源：new_dev(新开发)/bug_fix(修复Bug)/rework(质检重做)。develop 接任务后先调用，据此决定是否走 start_fix→fix_bug 状态机", + "params": {"task_id": "任务ID(可选，默认当前任务)"}, + "required": [], + }, "start_fix": { "module": "bug_capability", "description": "开始修复 Bug：open/confirmed → fixing（记录处理人）。回退重做/修复 Bug 任务里修完代码必须调用，open 状态也可直接修复（PM review_rollback 回退时不走 confirm_bug）", @@ -269,6 +275,7 @@ async def exec_capability_tool(tool_name, params, ctx): p.setdefault("iteration_id", ctx.get("iteration_id", "")) p.setdefault("who", ctx.get("who", "")) p.setdefault("agent_id", ctx.get("agent_id", "")) + p.setdefault("task_id", ctx.get("task_id", "")) try: mod = importlib.import_module(f"pipeline_service.{schema['module']}")