From 3ffe39e885cb9d6c75b1c5416f57feed75e6cfce Mon Sep 17 00:00:00 2001 From: ymq Date: Wed, 19 Aug 2026 18:15:39 +0800 Subject: [PATCH] =?UTF-8?q?fix(task-flow):=20QC/PM=20=E9=A9=B3=E5=9B=9E?= =?UTF-8?q?=E5=8A=A0=E9=87=8D=E8=AF=95=E4=B8=8A=E9=99=90=EF=BC=8C=E6=89=93?= =?UTF-8?q?=E7=A0=B4=E3=80=8C=E7=A9=BA=E4=BA=A4=E4=BB=98=E4=BB=B6=E2=86=92?= =?UTF-8?q?=E9=A9=B3=E5=9B=9E=E2=86=92=E9=87=8D=E5=81=9A=E3=80=8D=E6=AD=BB?= =?UTF-8?q?=E5=BE=AA=E7=8E=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因：develop agent 产出空交付件 → QC 驳回(qc_reject)直接 qc_review→submitted， 不递增 retry_count、不检查 task_max_retry 上限 → 无限重做循环；3 个 running 任务 占满 max_concurrent_agents=3 全局名额 → agent poller avail=0 不再派发 6 个 submitted 任务 → 任务整体饿死，只有手动 start_agents(绕过并发限制)才走一步。 修复： 1. task_capability.py 新增 _reject_with_retry_limit：退回时 retry_count+1，超 task_max_retry(默认3)则转 failed（释放并发名额，交 failed poller 报故障给人工）。 qc_reject_task / reject_task 统一走它。 2. agent_loop.py qc_review_run 驳回时感知超限转 failed，跳过建 qc_reject 问题 （failed poller 会报 fault_report），返回 failed 而非 rejected。 --- pipeline_service/agent_loop.py | 8 +++-- pipeline_service/task_capability.py | 52 +++++++++++++++++++++++++---- 2 files changed, 52 insertions(+), 8 deletions(-) diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 05fdcb2..a3b82f1 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -1674,6 +1674,12 @@ async def qc_review_run(project_id, agent_id=None, model_name=None): return {"status": "approved", "task_id": task_id, "comment": decision.get('comment', '')} else: comment = decision.get('comment', '') + from .task_capability import qc_reject_task + ok, msg = await qc_reject_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment=comment) + if not ok and "failed" in msg: + # 重复退回达上限 → 已转 failed，不再建 qc_reject 问题（failed poller 会报 fault_report 给人工） + logger.info(f"qc_review_run reject->failed: task={task_id} {msg}") + return {"status": "failed", "task_id": task_id, "comment": msg} from .communication import raise_problem rejection_q = decision.get("questions") or comment or "交付件不合规" await raise_problem("qc_reject", rejection_q, "agent.qc", @@ -1681,8 +1687,6 @@ async def qc_review_run(project_id, agent_id=None, model_name=None): first_handler_role=task_role, context={"qc_comment": comment, "deliverable_type": deliverable_type}, suspend_task=False) - from .task_capability import qc_reject_task - await qc_reject_task(task_id, project_id, who="agent.qc", agent_id=agent_id, comment=comment) logger.info(f"qc_review_run reject: task={task_id} -> submitted") return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q} diff --git a/pipeline_service/task_capability.py b/pipeline_service/task_capability.py index aa1ea03..68eea6c 100644 --- a/pipeline_service/task_capability.py +++ b/pipeline_service/task_capability.py @@ -117,12 +117,52 @@ async def qc_approve_task(task_id, tenant_id, who=None, agent_id=None, comment=N who=who, agent_id=agent_id, detail=comment) -async def qc_reject_task(task_id, tenant_id, who=None, agent_id=None, comment=None): - """QC 退回：qc_review → submitted（清 claimed_by，被退角色重新认领重做）。""" - return await _transition(task_id, tenant_id, S_QC_REVIEW, S_SUBMITTED, 'qc_reject', +async def _reject_with_retry_limit(task_id, tenant_id, from_state, action, + who=None, agent_id=None, comment=None): + """退回重做 + 重试上限：retry_count+1，超 task_max_retry 则转 failed（打破死循环）。 + + 未超限：from_state → submitted（正常退回重做）。 + 超限：from_state → failed（释放并发名额，交 failed poller 报故障给人工）。 + 返回 (ok, message)。 + """ + if not task_id or not tenant_id: + return False, "缺少 task_id 或 tenant_id" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + # 重复计数 +1（CAS：仅当任务仍在 from_state） + await sor.sqlExe( + "UPDATE pipeline_tasks SET retry_count=retry_count+1 " + "WHERE id=${tid}$ AND tenant_id=${tn}$ AND state=${from}$", + {"tid": task_id, "tn": tenant_id, "from": from_state}) + recs = await sor.sqlExe( + "SELECT retry_count FROM pipeline_tasks WHERE id=${tid}$", {"tid": task_id}) + await sor.sqlExe("COMMIT", {}) + rc = int(getattr(recs[0], 'retry_count', 0) or 0) if recs else 0 + from .workspace import get_max_task_retry + max_retry = await get_max_task_retry(sor) + if rc >= max_retry: + err = f"{action} 退回 {rc} 次已达上限({max_retry})：{comment or '交付件不合规'}" + await sor.sqlExe( + "UPDATE pipeline_tasks SET state='failed', claimed_by=NULL, " + "last_error=${err}$, updated_at=NOW() " + "WHERE id=${tid}$ AND state=${from}$", + {"tid": task_id, "from": from_state, "err": err}) + await record_audit(tenant_id, 'pipeline_tasks', task_id, action, + from_state=from_state, to_state=S_FAILED, + who=who, agent_id=agent_id, detail=err, sor=sor) + logger.info("reject retry limit hit: task=%s rc=%s -> failed", task_id, rc) + return False, f"退回 {rc} 次已达上限({max_retry})，任务已标记 failed，等待人工介入" + # 未超限：正常退回 submitted + return await _transition(task_id, tenant_id, from_state, S_SUBMITTED, action, who=who, agent_id=agent_id, detail=comment) +async def qc_reject_task(task_id, tenant_id, who=None, agent_id=None, comment=None): + """QC 退回：qc_review → submitted（重复计数+1，超上限转 failed 打破死循环）。""" + return await _reject_with_retry_limit(task_id, tenant_id, S_QC_REVIEW, 'qc_reject', + who=who, agent_id=agent_id, comment=comment) + + async def approve_task(task_id, tenant_id, who=None, agent_id=None, comment=None): """审核通过：review → approved。""" return await _transition(task_id, tenant_id, S_REVIEW, S_APPROVED, 'approve', @@ -130,9 +170,9 @@ async def approve_task(task_id, tenant_id, who=None, agent_id=None, comment=None async def reject_task(task_id, tenant_id, who=None, agent_id=None, comment=None): - """审核退回：review → submitted（清 claimed_by，被退角色重新认领）。""" - return await _transition(task_id, tenant_id, S_REVIEW, S_SUBMITTED, 'reject', - who=who, agent_id=agent_id, detail=comment) + """审核退回：review → submitted（重复计数+1，超上限转 failed 打破死循环）。""" + return await _reject_with_retry_limit(task_id, tenant_id, S_REVIEW, 'reject', + who=who, agent_id=agent_id, comment=comment) async def complete_task(task_id, tenant_id, who=None, agent_id=None):