diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index fe495ea..2bff03f 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -2036,6 +2036,42 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): logger.info(f"模块级 develop approved，跳过 deploy_test（部署以应用为单位）: {task_id}") return {"status": "approved", "task_id": task_id, "comment": comment, "next_role": "", "skip_next": "module_not_deployed_independently"} + # 人工确认节点：requirement/design 审核通过后，不直接派发下一角色任务， + # 而是发「需求/设计确认」人工任务给项目 owner。owner 确认通过 → confirm_stage_gate + # 继续派发下一角色；不确定 → 退回原角色重做（带修改意见）。 + if task_role in ('agent.requirement', 'agent.design'): + _own = await sor.sqlExe( + "SELECT created_by FROM sd_projects WHERE id=${pid}$", {"pid": project_id}) + await sor.sqlExe("COMMIT", {}) + owner_id = getattr(_own[0], 'created_by', '') if _own else '' + if not owner_id: + owner_id = 'user-01' # 历史项目 NULL 回填 admin + confirm_type = 'requirement_confirmation' if task_role == 'agent.requirement' else 'design_confirmation' + confirm_label = '需求' if task_role == 'agent.requirement' else '设计' + iter_id = '' + if task_iter: + _it2 = await sor.sqlExe( + "SELECT id FROM sd_iterations WHERE project_id=${pid}$ AND iteration_name=${n}$", + {"pid": project_id, "n": task_iter}) + iter_id = getattr(_it2[0], 'id', '') if _it2 else '' + from .human_task_capability import create_human_task + ok, hid = await create_human_task( + project_id, + f"{confirm_label}确认：{title}", + f"请项目 owner 确认「{confirm_label}」是否满足要求。\n\n" + f"· 确认通过 → 继续后续任务（{next_role}）。\n" + f"· 不确定 → 请填写修改意见，将退回 {task_role} 重做。\n\n" + f"原任务：{title}\nPM 审核意见：{comment or '通过'}", + task_type=confirm_type, + assignee_id=owner_id, + iteration_id=iter_id, + created_by='agent.pm', + task_id=task_id, + ) + logger.info(f"人工确认节点：{task_role} approved → 创建 {confirm_type} " + f"给 owner={owner_id}, human_task={'OK' if ok else 'FAIL:' + str(hid)}") + return {"status": "approved", "task_id": task_id, "next_role": next_role, + "human_confirm": hid if ok else "", "comment": comment} next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment) return {"status": "approved", "task_id": task_id, "next_task_id": next_tid, "next_role": next_role, "comment": comment} else: diff --git a/pipeline_service/human_task_capability.py b/pipeline_service/human_task_capability.py index e1a4b6e..339079f 100644 --- a/pipeline_service/human_task_capability.py +++ b/pipeline_service/human_task_capability.py @@ -29,6 +29,8 @@ QC_REJECTED = "rejected" T_GENERAL = "general" T_BUG_ACCEPT = "bug_acceptance" +T_REQ_CONFIRM = "requirement_confirmation" +T_DESIGN_CONFIRM = "design_confirmation" def _get_db(): @@ -81,9 +83,10 @@ async def _get_user_roles(sor, user_id): async def create_human_task(project_id, title, description="", task_type=T_GENERAL, assignee_role=None, assignee_id=None, - iteration_id=None, bug_id=None, created_by=None): + iteration_id=None, bug_id=None, created_by=None, task_id=""): """派发项目级人类任务。assignee_role 或 assignee_id 至少指定一个。 + task_id：确认类任务绑定原引擎任务 id（用于确认后继续派发/退回重做）。 返回 (True, human_task_id) 或 (False, 错误信息)。 """ if not project_id: @@ -111,7 +114,7 @@ async def create_human_task(project_id, title, description="", task_type=T_GENER hid = getID() await sor.C('pipeline_human_tasks', { 'id': hid, - 'task_id': '', # 项目级人类任务不绑定引擎任务 + 'task_id': task_id or '', # 确认类任务绑定原引擎任务 id 'step_name': '', # 同上 'version': 1, 'task_type': task_type or T_GENERAL, @@ -134,6 +137,81 @@ async def create_human_task(project_id, title, description="", task_type=T_GENER return True, hid +async def confirm_stage_gate(human_task_id, confirmed, comment, operator_id): + """需求/设计人工确认节点（owner 确认门禁）。 + + confirmed=True → 确认通过：任务 done + 创建下一角色任务（继续任务链）。 + confirmed=False → 不确定：任务 rejected + 退回原角色重做（带修改意见）。 + + 确认任务由 pm_review_run 在 requirement/design 审核通过后创建， + task_id 字段存原引擎任务 id（用于此处继续/退回）。仅 owner 可确认。 + """ + if not human_task_id: + return False, "缺少 human_task_id" + if not operator_id: + return False, "未登录" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT * FROM pipeline_human_tasks WHERE id=${hid}$", {"hid": human_task_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return False, "确认任务不存在" + ht = _rec_to_dict(recs[0]) + task_type = ht.get('task_type', '') or '' + if task_type not in (T_REQ_CONFIRM, T_DESIGN_CONFIRM): + return False, "该任务不是需求/设计确认任务" + if ht.get('status') != S_PENDING: + return False, f"任务不在待确认状态 (当前: {ht.get('status')})" + + # owner 校验：确认任务 assignee_id=owner，须等于本人 + assignee_id = ht.get('assignee_id', '') or '' + if assignee_id and operator_id != assignee_id: + return False, "仅项目 owner 可确认" + + pid = ht.get('project_id', '') + original_task_id = ht.get('task_id', '') or '' + result_json = json.dumps({"confirmed": bool(confirmed), "comment": comment or ''}, + ensure_ascii=False) + + if confirmed: + await sor.sqlExe( + "UPDATE pipeline_human_tasks SET status='done', result_data=${rd}$, " + "submitted_by=${oid}$, submitted_at=NOW() WHERE id=${hid}$", + {"rd": result_json, "oid": operator_id, "hid": human_task_id}) + await sor.sqlExe("COMMIT", {}) + if original_task_id: + from .agent_loop import _create_next_task, _get_next_role + trecs = await sor.R('pipeline_tasks', {'id': original_task_id}) + if trecs: + t = trecs[0] + nrole = await _get_next_role(getattr(t, 'role', ''), pid) + if nrole: + next_tid, next_title = await _create_next_task( + sor, pid, t, nrole, comment or "owner 确认通过") + await sor.sqlExe("COMMIT", {}) + return True, f"已确认，继续派发 {nrole}" + return True, "已确认" + else: + await sor.sqlExe( + "UPDATE pipeline_human_tasks SET status='rejected', result_data=${rd}$, " + "submitted_by=${oid}$, submitted_at=NOW() WHERE id=${hid}$", + {"rd": result_json, "oid": operator_id, "hid": human_task_id}) + await sor.sqlExe("COMMIT", {}) + if original_task_id: + from .agent_loop import _rollback_task_chain + trecs = await sor.R('pipeline_tasks', {'id': original_task_id}) + if trecs: + t = trecs[0] + role = getattr(t, 'role', '') + r = await _rollback_task_chain( + sor, pid, original_task_id, role, + comment or "owner 确认不通过，退回重做") + await sor.sqlExe("COMMIT", {}) + return True, f"已退回重做: {r.get('new_task_id', '')}" + return True, "已退回重做" + + async def complete_human_task(human_task_id, result_data, operator_id=None): """处理者提交完成（done）。校验同机构 + 匹配 assignee。 @@ -277,6 +355,7 @@ async def has_blocking_human_task(sor, project_id): "WHERE iteration_id=${iid}$ AND (" " (task_type='general' AND (status='pending' OR (status='done' AND qc_status != 'passed'))) " " OR (task_type='bug_acceptance' AND status='pending')" + " OR (task_type IN ('requirement_confirmation','design_confirmation') AND status='pending')" ")", {"iid": iid}) await sor.sqlExe("COMMIT", {}) diff --git a/pipeline_service/init.py b/pipeline_service/init.py index 7614045..6bff45c 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -511,12 +511,13 @@ def load_pipeline_service(): # SDLC 项目级人类任务清单 + owner 校验 from .human_task_capability import ( - create_human_task, complete_human_task, qc_human_task, + create_human_task, complete_human_task, qc_human_task, confirm_stage_gate, list_project_human_tasks, count_my_human_tasks, list_my_human_todos, bug_accept) from .project_capability import check_project_owner, check_task_owner, check_tenant_owner env.create_human_task = create_human_task env.complete_human_task = complete_human_task env.qc_human_task = qc_human_task + env.confirm_stage_gate = confirm_stage_gate env.list_project_human_tasks = list_project_human_tasks env.count_my_human_tasks = count_my_human_tasks env.list_my_human_todos = list_my_human_todos