ymq ae213b2f91 fix(零消耗): paused项目两个tokens漏点
- failed_poller→handle_failed_task: paused项目跳过LLM决策(原每轮白烧)
- bug_poller→pm_bug_confirm_run: JOIN sd_projects排除paused(原只看迭代状态)
三主poller(role/pm/qc)已排除paused;retrospective由pm_poller触发无漏点
2026-09-02 11:07:39 +08:00

442 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""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 报告自动 closehuman 报告创建 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)} 个已确认功能 Bugstatus=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)} 个 Bugstatus=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 []
_BLOCKING_HELP_PAT = re.compile(
r"待明确|待确认|待补齐|未提供|未配置|缺少环境|环境信息|无法\s*ssh|无法连接|连接不上|"
r"凭据|credentials|placeholder|\bTBD\b|需(要)?(人|PM|运维)(补|确认|提供)|"
r"占位|前置条件不具备|权限不足无法",
re.IGNORECASE)
def _is_blocking_help(bug):
"""判定 bug 是否为「阻塞求助」而非「代码缺陷」。
阻塞求助 = 缺少外部输入(环境信息/凭据/配置/权限)导致无法执行,只有人能解决。
这类必须转 need_info 冒泡给人绝不能当「非缺陷」reject——否则求助通道被掐死
2026-08-25 人事项目7 实测deploy_test 两次 report_bug 上报「env/test.json 待确认无法部署」,
被 PM 确认流程在 9 秒内以「属测试环境问题非代码缺陷」自动驳回PM 收不到求助,
转而回退 requirement 重做,而 requirement 同样拿不到环境信息 → 死循环。
判定用确定性规则(特征词),不交给 LLM是求助还是缺陷有明确特征属于机制该管的部分。
"""
text = f"{bug.get('title','')} {bug.get('description','') or ''}"
return bool(_BLOCKING_HELP_PAT.search(text))
async def _escalate_blocking_bugs(sor, project_id, help_bugs):
"""把「阻塞求助」类 bug 转成 need_info 冒泡给人bug 保持 open 等人补信息。
幂等:同一 bug 已有 pending 的 need_info 冒泡则跳过,避免 poller 每 20 秒刷一条。
"""
from .communication import raise_problem
escalated = 0
for b in help_bugs:
bid = b.get('id', '')
if not bid:
continue
# 幂等:该 bug 已冒泡且未答结 → 跳过
dup = await sor.sqlExe(
"SELECT id FROM pipeline_agent_questions "
"WHERE tenant_id=${pid}$ AND status='pending' AND problem_type='need_info' "
"AND question LIKE ${kw}$ LIMIT 1",
{"pid": project_id, "kw": f"%{bid}%"})
await sor.sqlExe("COMMIT", {})
if dup:
continue
question = (
f"【阻塞求助需人工补充信息】Bug {bid}\n"
f"标题:{b.get('title','')}\n"
f"详情:{(b.get('description','') or '').strip()[:600]}\n\n"
f"该问题属于外部输入缺口(环境信息/凭据/配置/权限未提供),"
f"agent 无法自行解决,也不是代码缺陷。请补充所需信息后回答本问题,"
f"流程将自动继续Bug 保持 open 直至信息补齐。"
)
try:
await raise_problem(
"need_info", question, from_role="agent.pm",
from_agentid="bug-flow", tenant_id=project_id,
task_id=b.get('task_id') or '',
first_handler_role="agent.main_agent",
suspend_task=bool(b.get('task_id')))
escalated += 1
logger.info(f"bug_flow 阻塞求助已冒泡给人 bug={bid} project={project_id}")
except Exception as e:
logger.warning(f"bug_flow 冒泡失败 bug={bid}: {e}")
return escalated
async def pm_bug_confirm_run(sor, project_id):
"""PM 独立确认 open bugopen → confirmed/rejected
独立于任务审核循环bug poller 扫到 open bug 时触发本流程PM 读 open bug 清单
逐个判断「真缺陷(confirm) / 误报重复(reject)」,不依赖任务链任何环节。
只处理「进行中迭代」的 open bug。
前置确定性分流:命中「阻塞求助」特征的 bug 不进 LLM 判断,直接转 need_info 冒泡给人,
bug 保持 open 等人补信息(不 reject、不 confirm
"""
open_rows = await sor.sqlExe(
"SELECT b.id, b.title, b.description, b.severity, b.reporter_type, b.iteration_id, b.task_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, "escalated": 0}
all_bugs = [_rec_to_dict(r) for r in open_rows]
# ── 前置分流:阻塞求助 → need_info 冒泡给人,不进 LLM ──
help_bugs = [b for b in all_bugs if _is_blocking_help(b)]
bugs = [b for b in all_bugs if not _is_blocking_help(b)]
escalated = 0
if help_bugs:
escalated = await _escalate_blocking_bugs(sor, project_id, help_bugs)
if not bugs:
return {"confirmed": 0, "rejected": 0, "skipped": 0, "escalated": escalated}
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"
"🔴 铁律:**缺少外部输入(环境信息/凭据/配置/权限未提供)导致无法执行,绝不是可 reject 的「非缺陷」**——"
"那是需要人补信息的阻塞求助,必须保留待处理,不得驳回。此类已由系统前置分流,你收到的列表里不应再有;"
"若仍见到,一律判 confirm 而非 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 的项目触发)
# paused 项目零消耗2026-09-02暂停 = 完全停摆open bug 不触发
# PM LLM 确认(每轮白烧 tokens恢复推进后下一轮自动继续确认。
proj_rows = await sor.sqlExe(
"SELECT DISTINCT i.project_id FROM sd_bugs b "
"JOIN sd_iterations i ON b.iteration_id=i.id "
"JOIN sd_projects p ON p.id=i.project_id AND p.status != 'paused' "
"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}