From a00ba32430b9b52d7f84ae88a6482c8a768dfa0c Mon Sep 17 00:00:00 2001 From: ymq Date: Tue, 18 Aug 2026 15:38:55 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20PM=20=E5=AE=A1=E6=A0=B8=E6=96=B0?= =?UTF-8?q?=E5=A2=9E=20review=5Frollback=20=E5=86=B3=E7=AD=96=E2=80=94?= =?UTF-8?q?=E2=80=94=E6=B5=8B=E8=AF=95=20Bug=20=E5=90=8E=E6=8C=89=E9=97=AE?= =?UTF-8?q?=E9=A2=98=E5=86=85=E5=AE=B9=E5=9B=9E=E9=80=80=E5=88=B0=E4=BB=BB?= =?UTF-8?q?=E4=B8=80=E9=98=B6=E6=AE=B5=EF=BC=8C=E5=9B=9E=E9=80=80=E7=82=B9?= =?UTF-8?q?=E4=B9=8B=E5=90=8E=E7=9A=84=E5=85=B3=E8=81=94=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E5=85=A8=E9=83=A8=E4=BD=9C=E5=BA=9F=E5=B9=B6=E9=87=8D=E5=81=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_service/agent_loop.py | 109 ++++++++++++++++++++++++++++++++- 1 file changed, 107 insertions(+), 2 deletions(-) diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 7131a16..5c3d145 100644 --- a/pipeline_service/agent_loop.py +++ b/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 的状态不一致。