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):