diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 05dfb69..04f5bb1 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -527,15 +527,6 @@ __ROLE_SKILLS__ - create_tasks(tasks) — 批量创建并派发后续任务（tasks 是 JSON 数组，每项 {title, role, description, key?, depends_on?, parent_id?}；key 供同批任务间 depends_on 引用，depends_on 是前序 key 或任务ID 数组，空=并行/非空=串行） - list_tasks(role, state) — 列出项目现有任务（派发前先查，避免重复） - cancel_task(task_id) — 取消任务（重做/作废前必须先取消旧任务，避免两个相同任务并存） -- list_bugs(status) — 查当前迭代的 Bug 清单（可按 status 过滤） -- confirm_bug(bug_id) — 确认 Bug 有效（open→confirmed），准备派发修复 -- reject_bug(bug_id, comment) — 驳回 Bug（误报/重复/非缺陷） -- reopen_bug(bug_id) — 重新打开已关闭/驳回的 Bug - -## Bug 生命周期流转 -Bug 是独立于任务链的第二条循环（open → confirmed → fixing → fixed → verified → closed）。 -你（PM）拥有「添加流转」的能力，每次审核任何任务都要检查当前迭代 Bug 状态并推动流转—— -具体规则（何时 confirm/驳回/派发修复任务/派发复测）见 bug-confirm 技能，先 load_skill 加载全文再执行。 ## 审核流程 1. 先用工具检查代码/交付件是否实际产出（git_status / list_files / read_file） @@ -1864,23 +1855,6 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): result = await _pm_cancel_task(sor, project_id, params) elif tool == 'load_skill': result = await _load_skill_by_name(sor, project_id, 'pm', org_id, params.get('name', ''), params.get('file_path') or None) - elif tool in ('list_bugs', 'confirm_bug', 'reject_bug', 'reopen_bug'): - # Bug 生命周期流转（独立循环）：PM 拥有「添加流转」能力，通过能力工具推动 bug 状态机。 - from .capability_tools import exec_capability_tool - _iter = None - try: - from .iteration_capability import get_current_iteration - _iter = await get_current_iteration(sor, project_id) - except Exception: - _iter = None - bug_ctx = { - "project_id": project_id, - "iteration_id": (_iter.get('id', '') if _iter else ''), - "who": "agent.pm", - "agent_id": agent_id or "", - "org_id": org_id or '0', - } - result = await exec_capability_tool(tool, params, bug_ctx) else: if turn >= 3: # 硬约束：最后两轮拒绝执行探索类工具（read_file/git_status/list_files/run_shell 等）， diff --git a/pipeline_service/bug_flow.py b/pipeline_service/bug_flow.py new file mode 100644 index 0000000..112e39b --- /dev/null +++ b/pipeline_service/bug_flow.py @@ -0,0 +1,313 @@ +"""Bug 生命周期独立循环（与任务循环平行，两套独立逻辑）。 + +任务循环（init.py 的 role/pm/qc/failed poller）扫 pipeline_tasks 表驱动任务状态机； +Bug 循环（本模块 + init.py 的 bug poller）扫 sd_bugs 表驱动 bug 状态机。 + +状态机：open → confirmed → fixing → fixed → verified → closed（+ rejected） + +两套循环唯一的交互点： +- Bug 循环在 confirmed/fixed 状态时「创建任务」（修复/复测），交给任务循环执行； +- 任务执行时角色 agent 调 start_fix/fix_bug/verify_bug 反馈推进 bug 状态。 + +本模块只做「确定性流转」（无需 LLM 判断）： +- verified → closed（自动关闭） +- fixed → 派发「复测」任务给 test +- confirmed → 派发「修复 Bug」任务给 develop + +「判断性流转」（open → confirmed/rejected）由 PM 独立确认流程负责（pm_bug_confirm_run）。 + +只处理「进行中迭代（in_progress）」的 bug——completed/cancelled 迭代的 bug 是历史遗留， +不在本循环范围内（避免把已结束迭代的旧 bug 重新拉起来跑）。 +""" + +import json +import logging +import re + +from appPublic.uniqueID import getID +from sqlor.dbpools import DBPools +from . import bug_capability + +logger = logging.getLogger("pipeline.bug_flow") + +# 活跃任务状态（这些状态的任务还没跑完，同迭代同类任务存在即视为「已派发」，不重复创建） +_ACTIVE_STATES = ('submitted', 'running', 'review', 'qc_review', 'waiting') + + +def _rec_to_dict(rec): + if rec is None: + return {} + if isinstance(rec, dict): + return dict(rec) + try: + return dict(rec) + except (TypeError, ValueError): + if hasattr(rec, 'to_dict'): + try: + return rec.to_dict() + except Exception: + pass + return {} + + +async def _create_bug_task(sor, project_id, iteration_name, title, role, desc, task_kind): + """创建 bug 循环派发的任务（修复/复测）。返回任务 id。""" + tparams = {"description": desc, "pm_assigned": True, "task_kind": task_kind} + if iteration_name: + tparams["iteration_id"] = iteration_name + tid = getID() + await sor.C('pipeline_tasks', { + 'id': tid, 'tenant_id': project_id, 'pipeline_id': 'role_task', + 'owner_id': 'bug_flow', 'title': title, 'state': 'submitted', + 'role': role, 'params': json.dumps(tparams, ensure_ascii=False), + 'parent_id': None, + }) + return tid + + +async def _has_active_task(sor, project_id, iteration_name, task_kind): + """某迭代是否已有活跃的某类 bug 任务（幂等：避免重复派发）。""" + states = ",".join(f"'{s}'" for s in _ACTIVE_STATES) + recs = await sor.sqlExe( + f"SELECT COUNT(*) AS c FROM pipeline_tasks " + f"WHERE tenant_id=${{pid}}$ AND state IN ({states}) " + f"AND JSON_UNQUOTE(JSON_EXTRACT(params,'$.task_kind'))=${{tk}}$", + {"pid": project_id, "tk": task_kind}) + await sor.sqlExe("COMMIT", {}) + c = getattr(recs[0], 'c', 0) if recs else 0 + return c > 0 + + +async def _advance_bug_states(sor): + """一轮 bug 循环：确定性流转（verified→close / fixed→复测 / confirmed→修复）。""" + result = {"closed": 0, "fix_tasks": 0, "retest_tasks": 0} + + # 1) verified → closed（自动关闭，无需 LLM） + verified = await sor.sqlExe( + "SELECT id, iteration_id FROM sd_bugs WHERE status='verified' LIMIT 100", {}) + await sor.sqlExe("COMMIT", {}) + for r in (verified or []): + d = _rec_to_dict(r) + bid, iid = d.get('id', ''), d.get('iteration_id', '') + if bid and iid: + ok, _ = await bug_capability.close_bug(bid, iid, who='agent.bug_flow') + if ok: + result["closed"] += 1 + + # 2) 按「进行中迭代」分组收集 confirmed / fixed bug + confirmed_rows = await sor.sqlExe( + "SELECT b.id, b.title, b.severity, b.iteration_id, i.project_id, i.iteration_name " + "FROM sd_bugs b JOIN sd_iterations i ON b.iteration_id=i.id " + "WHERE b.status='confirmed' AND i.status='in_progress' ORDER BY b.created_at ASC", {}) + await sor.sqlExe("COMMIT", {}) + + fixed_rows = await sor.sqlExe( + "SELECT b.id, b.title, b.iteration_id, i.project_id, i.iteration_name " + "FROM sd_bugs b JOIN sd_iterations i ON b.iteration_id=i.id " + "WHERE b.status='fixed' AND i.status='in_progress' ORDER BY b.created_at ASC", {}) + await sor.sqlExe("COMMIT", {}) + + # 3) confirmed → 派发「修复 Bug」任务给 develop（按 project 聚合，幂等） + by_project = {} + for r in (confirmed_rows or []): + d = _rec_to_dict(r) + pid = d.get('project_id', '') + if not pid: + continue + by_project.setdefault(pid, {"name": d.get('iteration_name', ''), "bugs": []}) + by_project[pid]["bugs"].append(d) + + for pid, info in by_project.items(): + if await _has_active_task(sor, pid, info["name"], "bug_fix"): + continue + bugs = info["bugs"] + bug_list = "\n".join( + f"- [{b.get('id','')}] {b.get('title','')}" for b in bugs) + desc = (f"修复当前迭代的 {len(bugs)} 个已确认功能 Bug（status=confirmed）。\n" + f"逐个 classify_task 判断来源（bug_fix）→ list_bugs 定位 → start_fix → 改代码 → fix_bug 标记 fixed。\n" + f"必须检查所有模块的同类问题，不能只修报出来的那一个。\n\n" + f"Bug 清单：\n{bug_list}") + tid = await _create_bug_task( + sor, pid, info["name"], + f"修复当前迭代 {len(bugs)} 个功能 Bug（{info['name']}）", + 'agent.develop', desc, 'bug_fix') + result["fix_tasks"] += 1 + logger.info(f"bug_flow: 派发修复任务 {tid} (project={pid}, bugs={len(bugs)})") + await sor.sqlExe("COMMIT", {}) + + # 4) fixed → 派发「复测」任务给 test（按 project 聚合，幂等） + by_project2 = {} + for r in (fixed_rows or []): + d = _rec_to_dict(r) + pid = d.get('project_id', '') + if not pid: + continue + by_project2.setdefault(pid, {"name": d.get('iteration_name', ''), "bugs": []}) + by_project2[pid]["bugs"].append(d) + + for pid, info in by_project2.items(): + if await _has_active_task(sor, pid, info["name"], "retest"): + continue + bugs = info["bugs"] + bug_list = "\n".join( + f"- [{b.get('id','')}] {b.get('title','')}" for b in bugs) + desc = (f"复测已修复的 {len(bugs)} 个 Bug（status=fixed），验证修复有效后逐个 " + f"verify_bug → close_bug 关闭。\n\nBug 清单：\n{bug_list}") + tid = await _create_bug_task( + sor, pid, info["name"], + f"复测已修复 Bug（{info['name']}）", + 'agent.test', desc, 'retest') + result["retest_tasks"] += 1 + logger.info(f"bug_flow: 派发复测任务 {tid} (project={pid}, bugs={len(bugs)})") + await sor.sqlExe("COMMIT", {}) + + return result + + +async def _resolve_model_org(sor, project_id): + """解析 bug 确认用的模型 + org_id（简版：项目缺省模型 → deepseek-v4-pro）。""" + org_id = "" + model = "" + try: + recs = await sor.sqlExe( + "SELECT org_id, default_model, pipeline_id FROM sd_projects WHERE id=${pid}$", + {"pid": project_id}) + if recs: + org_id = getattr(recs[0], "org_id", "") or "" + model = getattr(recs[0], "default_model", "") or "" + pipeline_id = getattr(recs[0], "pipeline_id", "") or "" + if not model and pipeline_id: + precs = await sor.sqlExe( + "SELECT default_model FROM pipelines WHERE id=${pid}$", {"pid": pipeline_id}) + if precs: + model = getattr(precs[0], "default_model", "") or "" + except Exception: + pass + await sor.sqlExe("COMMIT", {}) + return (model or "deepseek-v4-pro"), (org_id or "0") + + +def _parse_bug_decisions(raw): + """解析 LLM 输出的 bug 确认决策（JSON 数组 [{bug_id, decision, comment}]）。 + + 宽容解析：剥离 markdown 代码块、提取首个 JSON 数组。 + """ + raw = (raw or "").strip() + if raw.startswith("```"): + raw = raw.split("\n", 1)[1].rsplit("```", 1)[0].strip() + m = re.search(r"\[.*\]", raw, re.DOTALL) + if not m: + return [] + try: + arr = json.loads(m.group(0)) + if not isinstance(arr, list): + return [] + out = [] + for d in arr: + if isinstance(d, dict) and d.get("bug_id"): + out.append(d) + return out + except (json.JSONDecodeError, ValueError): + return [] + + +async def pm_bug_confirm_run(sor, project_id): + """PM 独立确认 open bug（open → confirmed/rejected）。 + + 独立于任务审核循环：bug poller 扫到 open bug 时触发本流程，PM 读 open bug 清单 + 逐个判断「真缺陷(confirm) / 误报重复(reject)」，不依赖任务链任何环节。 + 只处理「进行中迭代」的 open bug。 + """ + open_rows = await sor.sqlExe( + "SELECT b.id, b.title, b.description, b.severity, b.reporter_type, b.iteration_id " + "FROM sd_bugs b JOIN sd_iterations i ON b.iteration_id=i.id " + "WHERE b.status='open' AND i.status='in_progress' AND i.project_id=${pid}$ " + "ORDER BY b.created_at ASC LIMIT 50", + {"pid": project_id}) + await sor.sqlExe("COMMIT", {}) + if not open_rows: + return {"confirmed": 0, "rejected": 0, "skipped": 0} + + bugs = [_rec_to_dict(r) for r in open_rows] + model, org_id = await _resolve_model_org(sor, project_id) + + bug_list = "\n".join( + f"- [{b.get('id','')}] severity={b.get('severity','')} reporter={b.get('reporter_type','')}: " + f"{b.get('title','')}\n {((b.get('description','') or '').strip())[:200]}" + for b in bugs) + + system = ( + "你是项目经理（PM），负责确认测试上报的 Bug 是否真实。\n" + "逐个判断每个 Bug：真缺陷 → confirm（确认，进入修复流程）；误报/重复/非缺陷 → reject（驳回）。\n" + "判断标准：\n" + "- confirm：功能错误、代码缺陷、接口返回错误、字段/金额/权限/RBAC 校验错、表结构错、部署失败导致的真实代码问题。\n" + "- reject：测试环境问题导致的假失败、重复上报同一问题、非缺陷（如文案/样式已符合预期）。\n" + "必须对每个 Bug 都给出 decision，不要遗漏。\n" + "只输出一个 JSON 数组，格式：[{\"bug_id\":\"...\",\"decision\":\"confirm|reject\",\"comment\":\"一句话理由\"}]" + ) + user = f"以下是当前迭代上报的 {len(bugs)} 个 open Bug，请逐个确认：\n\n{bug_list}" + + from .llm_bridge import llm_call_msgs + raw = "" + try: + raw = await llm_call_msgs( + [{"role": "system", "content": system}, {"role": "user", "content": user}], + model=model, temperature=0.2, org_id=org_id) + except Exception as e: + logger.warning(f"pm_bug_confirm_run LLM 失败 project={project_id}: {e}") + return {"confirmed": 0, "rejected": 0, "skipped": len(bugs)} + + decisions = _parse_bug_decisions(raw) + if not decisions: + logger.warning(f"pm_bug_confirm_run 解析失败 project={project_id}: {raw[:200]}") + return {"confirmed": 0, "rejected": 0, "skipped": len(bugs)} + + # bug_id → iteration_id 映射（decision 里只带 bug_id） + iter_map = {b.get('id', ''): b.get('iteration_id', '') for b in bugs} + confirmed = rejected = 0 + for d in decisions: + bid = (d.get('bug_id') or '').strip() + iid = iter_map.get(bid, '') + if not bid or not iid: + continue + decision = (d.get('decision') or '').strip().lower() + if decision == 'confirm': + ok, _ = await bug_capability.confirm_bug(bid, iid, who='agent.pm') + if ok: + confirmed += 1 + elif decision == 'reject': + ok, _ = await bug_capability.reject_bug( + bid, iid, who='agent.pm', comment=(d.get('comment') or '误报/重复/非缺陷')) + if ok: + rejected += 1 + await sor.sqlExe("COMMIT", {}) + logger.info(f"pm_bug_confirm_run project={project_id}: confirmed={confirmed} rejected={rejected}") + return {"confirmed": confirmed, "rejected": rejected, "skipped": len(bugs) - confirmed - rejected} + + +async def bug_flow_poll_once(db): + """一轮完整的 bug 循环：PM 确认 open + 确定性流转（verified/fixed/confirmed）。""" + async with db.sqlorContext("pipeline") as sor: + # 1) open → PM 独立确认（按项目分组，只对有 open bug 的项目触发） + proj_rows = await sor.sqlExe( + "SELECT DISTINCT i.project_id FROM sd_bugs b " + "JOIN sd_iterations i ON b.iteration_id=i.id " + "WHERE b.status='open' AND i.status='in_progress' LIMIT 20", {}) + await sor.sqlExe("COMMIT", {}) + confirm_sum = {"confirmed": 0, "rejected": 0, "skipped": 0} + for r in (proj_rows or []): + pid = getattr(r, 'project_id', '') + if not pid: + continue + try: + c = await pm_bug_confirm_run(sor, pid) + confirm_sum["confirmed"] += c.get("confirmed", 0) + confirm_sum["rejected"] += c.get("rejected", 0) + confirm_sum["skipped"] += c.get("skipped", 0) + except Exception as e: + logger.warning(f"bug_flow pm confirm error project={pid}: {e}") + + # 2) 确定性流转（verified→close / fixed→复测 / confirmed→修复） + adv = await _advance_bug_states(sor) + return {"confirm": confirm_sum, "advance": adv} + diff --git a/pipeline_service/init.py b/pipeline_service/init.py index 03d56d0..c58c34c 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -897,6 +897,28 @@ def load_pipeline_service(): add_startup(_failed_poller) if _run_pollers else None + # Bug 生命周期 poller: 独立的 bug 循环（与任务循环平行的第二套循环）。 + # 扫 sd_bugs 表驱动 bug 状态机（open→confirmed→fixing→fixed→verified→closed）， + # 不寄生在任务循环的任何环节（PM 审核/角色 agent 执行都不再顺手处理 bug）。 + async def _bug_poller(app): + from sqlor.dbpools import DBPools as _DBP4 + from .bug_flow import bug_flow_poll_once + bug_db = _DBP4() + + async def _bug_poll_loop(): + while True: + try: + # watchdog：每轮 bug poll 最多 120 秒（含 LLM 确认 open bug），超时跳过本轮 + await asyncio.wait_for(bug_flow_poll_once(bug_db), timeout=120) + except Exception as e: + debug(f"bug_poller error/timeout: {e}") + await asyncio.sleep(20) + + asyncio.create_task(_bug_poll_loop()) + debug("bug poller started") + + add_startup(_bug_poller) if _run_pollers else None + # 铁律：运行期不做任何 schema 变更。v2 引擎的建表/加列已全部迁到 # 部署期 scripts/create_tables.py（build.sh 会调用），此处不再启动时建表。