feat(sdlc): requirement/design 审核通过后加 owner 人工确认门禁

- pm_review_run: requirement/design approved 后不再直接派发下一角色任务，而是发确认任务给 owner
- human_task_capability: 新增 confirm_stage_gate(确认通过→继续派发下一角色; 不确定→退回原角色重做)
- create_human_task 加 task_id 参数绑定原引擎任务
- has_blocking_human_task 纳入确认任务 pending 阻塞
- init.py 注册 confirm_stage_gate

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:

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", {})

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