fix(human_task): 待办归属动态解析+expired_at地雷修复(2026-09-18银联事故实锤)——①新增resolve_project_owner_assignee:项目真人创建者进其我的待办,无主回退owner.superuser;flow_plan确认/修订/重做失败4处调用点改动态;②create_human_task显式写expired_at=+7d(列默认current_timestamp()→创建即过期,实测全部历史待办expired_at==created_at)

pipeline_service/flow_plan_capability.py

@@ -25,9 +25,19 @@ from appPublic.uniqueID import getID
 from .audit import record_audit
 from .human_task_capability import (
     create_human_task, _get_user_org, _get_user_roles,
+    resolve_project_owner_assignee,
 )
 
 DBNAME = "pipeline"
+
+
+async def _owner_assignee(sor, project_id):
+    """确认/介入待办归属 (role, id)：项目真人创建者优先（进其「我的待办」），
+    无主项目回退 owner.superuser。2026-09-18 银联事故：硬编码 owner.superuser
+    导致非超管机构创建者待办 0 条、确认无门。"""
+    return await resolve_project_owner_assignee(sor, project_id)
+
+
 logger = logging.getLogger("pipeline.flow_plan")
 
 # 计划状态
@@ -370,11 +380,13 @@ async def propose_flow_plan(project_id, trim_keys=None, user_requirements="",
                                   user_requirements, propose_note, version,
                                   flow_key=flow_key, pipeline_id=pid,
                                   detect_note=detect_note)
+        _ar, _ai = await _owner_assignee(sor, project_id)
         ok_ht, ht = await create_human_task(
             project_id,
             "流程裁剪确认：%s（v%d）" % (pname, version),
             desc, task_type=T_FLOW_CONFIRM,
-            assignee_role="owner.superuser", created_by=who or "agent.main_agent")
+            assignee_role=_ar, assignee_id=_ai,
+            created_by=who or "agent.main_agent")
         if ok_ht:
             await sor.sqlExe(
                 "UPDATE pipeline_flow_plans SET confirm_task_id=${h}$, updated_at=NOW() "
@@ -961,12 +973,14 @@ async def revise_flow_plan(project_id, user_requirements="", reentry_stage="",
             reentry_md += ["**确认后系统将执行：**", "", rdesc.strip(), "",
                            "> ⚠️ 重做会重置上述阶段的已有产出（见影响说明），流程从该阶段自动重新推进。"]
             desc = desc + "\n".join(reentry_md)
+        _ar2, _ai2 = await _owner_assignee(sor, project_id)
         ok_ht, ht = await create_human_task(
             project_id,
             ("流程修订确认：%s（v%d%s）" % (
                 pname, version, "，含定向重做" if reentry_stage else "")),
             desc, task_type=T_FLOW_CONFIRM,
-            assignee_role="owner.superuser", created_by=who or "agent.main_agent")
+            assignee_role=_ar2, assignee_id=_ai2,
+            created_by=who or "agent.main_agent")
         if ok_ht:
             await sor.sqlExe(
                 "UPDATE pipeline_flow_plans SET confirm_task_id=${h}$, updated_at=NOW() "
@@ -1143,12 +1157,13 @@ async def _run_stage_reentry(sor, plan, operator_id):
     if not fn:
         # 提案时已校验过钩子存在；确认时消失 = 部署窗口异常，诚实上报
         msg = "产线「%s」阶段重做钩子不可用，未执行重置——请人工处理" % pid
+        _ar3, _ai3 = await _owner_assignee(sor, project_id)
         await create_human_task(
             project_id, "流程修订已确认但阶段重做未执行：%s" % stage_key,
             msg + "\n\n修订计划已生效（v%s），但定向重做的产出重置未执行。\n"
                   "请人工按修订确认待办里的影响说明处置，或修复钩子后让平台重试。"
                   % plan.get("version"),
-            task_type="general", assignee_role="owner.superuser")
+            task_type="general", assignee_role=_ar3, assignee_id=_ai3)
         await sor.sqlExe("COMMIT", {})
         return False, msg
     try:
@@ -1161,12 +1176,13 @@ async def _run_stage_reentry(sor, plan, operator_id):
         msg = "阶段重做执行异常：%s" % str(e)[:300]
         ok = False
     if not ok:
+        _ar4, _ai4 = await _owner_assignee(sor, project_id)
         await create_human_task(
             project_id, "流程修订已确认但阶段重做执行失败：%s" % stage_key,
             (msg or "钩子返回失败") + "\n\n修订计划已生效（v%s），但阶段重置未完成，"
-                  "流程可能停在旧产出上。请人工处置后项目自动恢复推进。"
+            "流程可能停在旧产出上。请人工处置后项目自动恢复推进。"
-                  % plan.get("version"),
+            % plan.get("version"),
-            task_type="general", assignee_role="owner.superuser")
+            task_type="general", assignee_role=_ar4, assignee_id=_ai4)
         await sor.sqlExe("COMMIT", {})
     await record_audit(project_id, "pipeline_flow_plans", plan.get("id", ""),
                        "stage_reentry", to_state=stage_key, who=operator_id,

pipeline_service/human_task_capability.py

@@ -12,6 +12,7 @@
 
 import json
 import logging
+from datetime import datetime, timedelta
 
 from sqlor.dbpools import DBPools
 from appPublic.uniqueID import getID
@@ -95,6 +96,32 @@ async def _get_user_roles(sor, user_id):
     return roles
 
 
+async def resolve_project_owner_assignee(sor, project_id):
+    """项目人类待办的归属人解析（2026-09-18 银联事故实锤修复）。
+
+    返回 (assignee_role, assignee_id)：
+      - 项目有真人创建者 → ('', created_by)：待办直接进创建者「我的待办」。
+        旧实现硬编码 assignee_role='owner.superuser'，而该角色只有平台超管
+        （user-01）持有——非超管机构用户创建的项目，确认/介入待办全派给超管，
+        创建者「我的待办」里 0 条（实测 ymq 银联项目 5 条 flow_plan_confirm
+        全 assignee_role=owner.superuser、assignee_id 空）。
+      - 无真人创建者（自动立项/agent 创建）→ ('owner.superuser', '')：回退
+        平台超管兜底，保持旧行为不丢待办。
+    角色名不硬编码进业务判断——owner.superuser 仅作为「无主项目」的兜底指派，
+    与运行时角色 CRUD 不冲突（它仍是 role 表里的真实角色）。
+    """
+    if not project_id:
+        return "owner.superuser", ""
+    recs = await sor.sqlExe(
+        "SELECT created_by FROM sd_projects WHERE id=${pid}$ LIMIT 1",
+        {"pid": project_id})
+    await sor.sqlExe("COMMIT", {})
+    owner = str(getattr(recs[0], "created_by", "") or "") if recs else ""
+    if owner and not owner.startswith("agent.") and owner != "system":
+        return "", owner
+    return "owner.superuser", ""
+
+
 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, task_id="",
@@ -135,6 +162,9 @@ async def create_human_task(project_id, title, description="", task_type=T_GENER
 
         hid = getID()
         _fs_json = json.dumps(form_schema, ensure_ascii=False) if form_schema else None
+        # expired_at 显式写（2026-09-18 实锤地雷）：列 DDL 默认 current_timestamp()
+        # → 不显式写则创建即过期（实测全部历史待办 expired_at==created_at）。
+        # 人类待办默认 7 天有效；NULL 语义不可用（列 NOT NULL）。
         await sor.C('pipeline_human_tasks', {
             'id': hid,
             'task_id': task_id or '',  # 确认类任务绑定原引擎任务 id
@@ -153,6 +183,7 @@ async def create_human_task(project_id, title, description="", task_type=T_GENER
             'qc_status': QC_PENDING,
             'title': title.strip(),
             'description': description or '',
+            'expired_at': datetime.now() + timedelta(days=7),
         })
         await sor.sqlExe("COMMIT", {})
         logger.info("create_human_task: %s project=%s type=%s title=%s",