feat: Bug 生命周期独立循环（与任务循环平行的第二套逻辑）
用户定：任务和 bug 做成真正独立的两个事情，机制一样（poller+状态机）、 两套独立逻辑。之前把 bug 动作嵌在 PM 审核任务里（在任务里加 bug 动作）是错的。 改动： 1. 撤销 pm_review_run 里的 bug 动作（PM_SYSTEM_PROMPT bug 段 + tool_call bug 路由） 2. 新增 bug_flow.py：独立 bug 循环模块 - _advance_bug_states：确定性流转（verified→close / fixed→派发复测 / confirmed→派发修复） - pm_bug_confirm_run：PM 独立确认 open bug（open→confirmed/rejected，LLM 判断） - bug_flow_poll_once：一轮完整 bug 循环入口 3. init.py 新增 _bug_poller（扫 sd_bugs 驱动状态机，与 role/pm/qc/failed 任务 poller 平行） 只处理进行中迭代（in_progress）的 bug。两循环唯一交互点：bug 循环在 confirmed/fixed 时创建任务（修复/复测）交给任务循环执行，任务执行时角色 agent 调 start_fix/fix_bug/verify_bug 反馈推进 bug 状态。
This commit is contained in:
ymq
parent
8672068f39
commit
434ab76d88
@ -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 等），
|
||||
|
||||
313
pipeline_service/bug_flow.py
Normal file
313
pipeline_service/bug_flow.py
Normal file
@ -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}
|
||||
|
||||
@ -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 会调用），此处不再启动时建表。
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user