feat: PM 审核新增 review_rollback 决策——测试 Bug 后按问题内容回退到任一阶段，回退点之后的关联任务全部作废并重做

pipeline_service/agent_loop.py

@@ -462,6 +462,7 @@ __REPO_STATE__
 批准：{"action":"review_approve","comment":"审核意见","next_task_title":"下阶段标题","next_task_description":"描述"}
 驳回：{"action":"review_reject","comment":"原因","questions":"修改要求"}
 完成：{"action":"review_complete","comment":"总结"}
+回退：{"action":"review_rollback","rollback_role":"develop","comment":"回退原因"}
 
 审核标准：
 - requirement：需求是否清晰完整可量化（含部署环境需求是否明确）
@@ -469,7 +470,11 @@ __REPO_STATE__
 - develop：必须用 git_status/read_file 检查代码是否实际写入仓库
 - deploy_test：测试环境部署配置完整、服务可访问
 - test：测试覆盖充分、发现问题记录完整
-- deploy_prod：生产部署配置完整、可一键部署、有回滚方案"""
+- deploy_prod：生产部署配置完整、可一键部署、有回滚方案
+
+## 回退规则（test 阶段）
+- 测试提交 Bug 后，零散 Bug → 走 Bug 闭环（report_bug→fix_bug→verify_bug），不打断任务链。
+- 系统性缺陷（Bug 多/严重、设计有缺陷、需求理解错、部署有问题）→ review_rollback，rollback_role 指定回退目标阶段（develop/design/requirement/deploy_test），回退点之后的关联任务将全部作废。"""
 
 
 QC_SYSTEM_PROMPT = """你是质量控制工程师（QC）。对交付件做合规检查和质量检查，不合规直接退回重做。
@@ -668,6 +673,81 @@ def _task_iteration_name(task):
         return ''
 
 
+async def _rollback_task_chain(sor, project_id, task_id, rollback_role, comment):
+    """回退：追溯任务链（previous_task_id），作废回退目标及之后的任务，创建回退目标的新任务。
+
+    回退点之后的关联任务全部作废（state=cancelled），回退目标阶段重新做。
+    """
+    from appPublic.uniqueID import getID
+    # 追溯任务链（previous_task_id 往前），得到 [最早 ... 当前]
+    chain = []
+    cur_id = task_id
+    visited = set()
+    while cur_id and cur_id not in visited:
+        visited.add(cur_id)
+        recs = await sor.R('pipeline_tasks', {'id': cur_id})
+        if not recs:
+            break
+        t = recs[0]
+        chain.append(t)
+        try:
+            p = json.loads(getattr(t, 'params', '{}') or '{}')
+        except (json.JSONDecodeError, TypeError):
+            p = {}
+        cur_id = p.get('previous_task_id') or ''
+    chain.reverse()
+
+    # 定位回退目标在链上的位置
+    target_idx = -1
+    for i, t in enumerate(chain):
+        if _normalize_role(getattr(t, 'role', '')) == rollback_role:
+            target_idx = i
+            break
+    if target_idx < 0:
+        return {"status": "error", "task_id": task_id,
+                "error": f"任务链上找不到回退目标角色 {rollback_role}，可用："
+                         + "、".join(_normalize_role(getattr(t, 'role', '')) for t in chain)}
+
+    target_task = chain[target_idx]
+    prev_task = chain[target_idx - 1] if target_idx > 0 else None
+
+    # 作废回退目标及之后的任务（含回退目标本身）
+    cancelled = []
+    for t in chain[target_idx:]:
+        tid = getattr(t, 'id', '')
+        await sor.sqlExe(
+            "UPDATE pipeline_tasks SET state='cancelled', claimed_by=NULL, updated_at=NOW() WHERE id=${tid}$",
+            {"tid": tid})
+        cancelled.append(tid)
+
+    # 创建回退目标的新任务（继承回退目标任务 params，记录回退信息，previous 指向前一任务）
+    try:
+        tp = json.loads(getattr(target_task, 'params', '{}') or '{}')
+    except (json.JSONDecodeError, TypeError):
+        tp = {}
+    new_params = {**tp, 'rollback_from': task_id, 'rollback_comment': comment}
+    if prev_task:
+        new_params['previous_task_id'] = getattr(prev_task, 'id', '')
+        new_params['previous_role'] = _normalize_role(getattr(prev_task, 'role', ''))
+    else:
+        new_params.pop('previous_task_id', None)
+        new_params.pop('previous_role', None)
+
+    new_tid = getID()
+    new_title = f"{getattr(target_task, 'title', '') or '任务'}（回退重做）"
+    await sor.C('pipeline_tasks', {
+        'id': new_tid, 'tenant_id': project_id, 'pipeline_id': 'role_task',
+        'owner_id': 'pm', 'title': new_title,
+        'params': json.dumps(new_params, ensure_ascii=False),
+        'role': rollback_role, 'state': 'submitted', 'claimed_by': None,
+    })
+    await sor.sqlExe("COMMIT", {})
+    logger.info(f"_rollback_task_chain: task={task_id} -> rollback {rollback_role}, "
+                f"cancelled={len(cancelled)} 任务, new_task={new_tid}")
+    return {"status": "rollback", "task_id": task_id, "rollback_role": rollback_role,
+            "cancelled": cancelled, "new_task_id": new_tid, "comment": comment}
+
+
 # ── Agent 工具执行 ──
 
 def _parse_agent_action(raw):
@@ -1144,7 +1224,7 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
                 msgs.append({"role": "user", "content":
                     "已检查足够信息。现在必须立即收尾："
                     "若本任务审批通过后需拆分为多个后续任务，现在用 create_tasks 一次性派发；"
-                    "随后必须立即输出 review_approve / review_reject / review_complete 三者之一，"
+                    "随后必须立即输出 review_approve / review_reject / review_complete / review_rollback 之一，"
                     "禁止再调用其它工具。"})
             try:
                 raw = await llm_call_msgs(msgs, model=model_name, temperature=0.3, org_id=org_id)
@@ -1165,6 +1245,9 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
             elif act.get('action') == 'review_complete':
                 decision = {'status': 'completed', 'comment': act.get('comment', '')}
                 break
+            elif act.get('action') == 'review_rollback':
+                decision = {'status': 'rollback', 'rollback_role': act.get('rollback_role', ''), 'comment': act.get('comment', '')}
+                break
             elif act.get('action') == 'tool_call':
                 tool = act.get('tool', '')
                 params = act.get('params', {})
@@ -1240,6 +1323,28 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
             await reject_task(task_id, project_id, who="agent.pm", agent_id=agent_id, comment=comment)
             return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q}
 
+        elif status == 'rollback':
+            rollback_role = _normalize_role((decision.get('rollback_role') or '').strip())
+            if not rollback_role:
+                # 未指定回退目标 → 视为驳回重审
+                from .task_capability import reject_task
+                await reject_task(task_id, project_id, who="agent.pm", agent_id=agent_id,
+                                  comment=(comment or '回退目标未指定'))
+                return {"status": "rejected", "task_id": task_id,
+                        "comment": "回退目标未指定，已驳回重审"}
+            # 记 PM 审核交付件（回退决策）
+            from appPublic.uniqueID import getID as _getID
+            pm_did = _getID()
+            await sor.C("pipeline_deliverables", {
+                "id": pm_did, "project_id": project_id, "task_id": task_id,
+                "deliverable_type": "pm_review", "title": f"PM审核(回退)：{title}",
+                "content": json.dumps(decision, ensure_ascii=False),
+                "file_path": os.path.join(workspace_dir, 'deliverables', 'pm', f"{task_id}_rollback.md"),
+                "quality_score": 0, "review_status": "rejected", "created_by": agent_id or "pm",
+            })
+            # 回退：作废回退点及之后的任务，创建回退目标的新任务
+            return await _rollback_task_chain(sor, project_id, task_id, rollback_role, comment)
+
         else:
             # review_complete（项目完成/无下一阶段）：同步把交付件标记 approved，
             # 避免任务 state=completed 但交付件 review_status 仍 pending 的状态不一致。