fix(task-flow): QC/PM 驳回加重试上限，打破「空交付件→驳回→重做」死循环

根因：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

@@ -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}
 

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