356 lines
16 KiB
Python
356 lines
16 KiB
Python
"""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 → 分流（agent 报告 → closed 自动关闭；human 报告 → 创建 bug 验收任务给报告人）
|
||
- 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 _create_bug_acceptance_task(sor, bug_id, iteration_id, reporter_id):
|
||
"""为 human 报告的 verified bug 创建验收任务（幂等：已有 pending 验收任务则不重复建）。"""
|
||
recs = await sor.sqlExe(
|
||
"SELECT i.project_id, i.iteration_name, b.title FROM sd_bugs b "
|
||
"JOIN sd_iterations i ON b.iteration_id=i.id WHERE b.id=${bid}$",
|
||
{"bid": bug_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
return None
|
||
pid = getattr(recs[0], 'project_id', '') or ''
|
||
title = getattr(recs[0], 'title', '') or ''
|
||
# 幂等：已有 pending 验收任务则不重复建
|
||
ex = await sor.sqlExe(
|
||
"SELECT COUNT(*) AS c FROM pipeline_human_tasks "
|
||
"WHERE bug_id=${bid}$ AND task_type='bug_acceptance' AND status='pending'",
|
||
{"bid": bug_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if ex and getattr(ex[0], 'c', 0) > 0:
|
||
return None
|
||
hid = getID()
|
||
await sor.C('pipeline_human_tasks', {
|
||
'id': hid, 'task_id': '', 'step_name': '', 'version': 1,
|
||
'task_type': 'bug_acceptance',
|
||
'assignee_role': '', 'assignee_id': reporter_id or '',
|
||
'form_schema': None, 'result_data': None,
|
||
'status': 'pending',
|
||
'project_id': pid, 'iteration_id': iteration_id, 'bug_id': bug_id,
|
||
'qc_status': 'pending',
|
||
'title': f"Bug 验收：{title}",
|
||
'description': "您报告的 Bug 已修复并测试通过，请验收（通过则关闭，不通过则重新打开）。",
|
||
})
|
||
logger.info(f"bug_flow: 创建 bug 验收任务 {hid} (bug={bug_id}, reporter={reporter_id})")
|
||
return hid
|
||
|
||
|
||
async def _advance_bug_states(sor):
|
||
"""一轮 bug 循环：确定性流转（verified→close/验收 / fixed→复测 / confirmed→修复）。"""
|
||
result = {"closed": 0, "accept_tasks": 0, "fix_tasks": 0, "retest_tasks": 0}
|
||
|
||
# 1) verified → 分流：agent 报告自动 close；human 报告创建 bug 验收任务（等报告人验收）
|
||
verified = await sor.sqlExe(
|
||
"SELECT id, iteration_id, reporter_type, reporter_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 not bid or not iid:
|
||
continue
|
||
if (d.get('reporter_type', '') or '') == 'human':
|
||
# human 报告 → 创建验收任务（幂等），bug 停在 verified 等报告人验收
|
||
await _create_bug_acceptance_task(sor, bid, iid, d.get('reporter_id', '') or '')
|
||
result["accept_tasks"] += 1
|
||
else:
|
||
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}
|
||
|