"""SDLC 项目级人类任务清单 capability。 人类待办三类来源（统一视图，见 list_my_human_todos / count_my_human_tasks）： - general 显式人类任务 → 落 pipeline_human_tasks（owner/PM 派发） - bug_acceptance bug 验收任务 → 落 pipeline_human_tasks（bug_flow 在 verified 时创建） - question 冒泡问题 → 不落此表，union 查 pipeline_agent_questions 处理权限：处理者须同机构（org_id 匹配）+ 匹配 assignee_role 或 assignee_id。 状态：status(pending/done/rejected) + qc_status(pending/passed/rejected，仅 general 走 QC)。 """ import json import logging from datetime import datetime, timedelta from sqlor.dbpools import DBPools from appPublic.uniqueID import getID DBNAME = "pipeline" logger = logging.getLogger("pipeline.human_task_capability") S_PENDING = "pending" S_DONE = "done" S_REJECTED = "rejected" QC_PENDING = "pending" QC_PASSED = "passed" QC_REJECTED = "rejected" T_GENERAL = "general" T_BUG_ACCEPT = "bug_acceptance" T_REQ_CONFIRM = "requirement_confirmation" T_DESIGN_CONFIRM = "design_confirmation" # ══════════════ 待办提供者注册表（2026-09-10 工单模块接入） ══════════════ # 平台待办是平台级机制：各部分（产线任务/冒泡问题/工单/…）产生待办， # 本模块聚合。外部模块通过 register_todo_provider 注入，不跨模块读表。 # provider 签名: async fn(user_id, roles, limit) -> [todo dict] # todo dict 须含 source（弹窗按它分流详情端点）+ title/description/created_at。 _TODO_PROVIDERS = [] def register_todo_provider(fn): """注册待办提供者（幂等：同函数不重复注册）。""" if callable(fn) and fn not in _TODO_PROVIDERS: _TODO_PROVIDERS.append(fn) logger.info("todo provider registered: %s", getattr(fn, '__module__', '?')) def _get_db(): db = DBPools() if not db.databases: from appPublic.jsonConfig import getConfig config = getConfig() if config.databases: db.databases = config.databases return db, DBNAME def _rec_to_dict(rec): if rec is None: return {} if isinstance(rec, dict): return dict(rec) try: return dict(rec) except (TypeError, ValueError): return {} async def _get_user_org(sor, user_id): """查用户 org_id。""" if not user_id: return "" recs = await sor.sqlExe( "SELECT orgid FROM users WHERE id=${u}$ LIMIT 1", {"u": user_id}) await sor.sqlExe("COMMIT", {}) return getattr(recs[0], 'orgid', '') if recs else '' async def _get_user_roles(sor, user_id): """查用户 RBAC 角色名列表（{orgtypeid}.{name} 格式）。""" roles = [] if not user_id: return roles recs = await sor.sqlExe( "SELECT r.orgtypeid, r.name FROM userrole ur JOIN role r ON ur.roleid=r.id " "WHERE ur.userid=${u}$", {"u": user_id}) await sor.sqlExe("COMMIT", {}) for r in (recs or []): o = getattr(r, 'orgtypeid', '') or '' n = getattr(r, 'name', '') or '' if o and n: roles.append(f"{o}.{n}") 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="", form_schema=None): """派发项目级人类任务。assignee_role 或 assignee_id 至少指定一个。 task_id：确认类任务绑定原引擎任务 id（用于确认后继续派发/退回重做）。 form_schema：动态表单声明（可选，与问题通道同格式）——待办详情按它渲染 上传/填写界面，人类在待办里直接完成任务（文件落项目工作空间，路径进 result_data 供 QC 与任务链消费）。 返回 (True, human_task_id) 或 (False, 错误信息) """ if not project_id: return False, "缺少 project_id" if not title or not title.strip(): return False, "缺少标题" if not assignee_role and not assignee_id: return False, "须指定 assignee_role 或 assignee_id" db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: # 项目存在性 + 状态门禁（2026-09-15 用户要求：暂停/删除的项目不再产生待办） proj = await sor.sqlExe( "SELECT id, org_id, status FROM sd_projects WHERE id=${pid}$", {"pid": project_id}) await sor.sqlExe("COMMIT", {}) if not proj: return False, "项目不存在（可能已删除，不再产生待办）" p_status = getattr(proj[0], 'status', '') or '' if p_status in ('paused', 'archived', 'cancelled', 'completed'): return False, ("项目当前状态为 %s，不再产生新待办（恢复推进后由对账器按需补发）" % p_status) if not iteration_id: it = await sor.sqlExe( "SELECT id FROM sd_iterations WHERE project_id=${pid}$ " "AND status='in_progress' ORDER BY seq_no DESC LIMIT 1", {"pid": project_id}) await sor.sqlExe("COMMIT", {}) iteration_id = getattr(it[0], 'id', '') if it else '' 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 'step_name': '', # 同上 'version': 1, 'task_type': task_type or T_GENERAL, 'assignee_role': assignee_role or '', 'assignee_id': assignee_id or '', 'form_schema': _fs_json, 'result_data': None, 'status': S_PENDING, 'submitted_by': created_by or '', 'project_id': project_id, 'iteration_id': iteration_id or '', 'bug_id': bug_id or '', '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", hid, project_id, task_type, title.strip()) 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。 general 任务 done 后创建 qc_review 的 QC 任务（复用 qc poller → qc_review_run）。 bug_acceptance 任务不走这里（走 bug_accept）。 """ 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]) if ht.get('status') != S_PENDING: return False, f"任务不在待处理状态 (当前: {ht.get('status')})" # 同机构校验 pid = ht.get('project_id', '') proj = await sor.sqlExe( "SELECT org_id FROM sd_projects WHERE id=${pid}$", {"pid": pid}) await sor.sqlExe("COMMIT", {}) proj_org = getattr(proj[0], 'org_id', '') if proj else '' user_org = await _get_user_org(sor, operator_id) if proj_org and user_org != proj_org: return False, "仅同机构用户可处理该任务" # 匹配 assignee：assignee_id 指定则须等于本人；assignee_role 指定则本人角色须含之 assignee_id = ht.get('assignee_id', '') or '' assignee_role = ht.get('assignee_role', '') or '' if assignee_id: if operator_id != assignee_id: return False, "该任务未指派给你" elif assignee_role: roles = await _get_user_roles(sor, operator_id) if assignee_role not in roles: return False, "该任务未指派给你的角色" task_type = ht.get('task_type', '') or T_GENERAL result_json = json.dumps(result_data, ensure_ascii=False, default=str) \ if isinstance(result_data, (dict, list)) else str(result_data or '') if task_type == T_BUG_ACCEPT: return False, "bug 验收任务请走 bug_accept 接口" # 2026-09-15 用户裁定（人类答复权威）：人类待办的答复不再送 agent QC。 # 原链路 complete→human_task_qc→通用QC提示词审查人类答复→LLM 判「答复不合规」 # →qc_reject→待办回 pending→用户再说「忽略」又被退——实测甘肃项目用户答复 # 「忽略任这些问题」被连退（45EbV3Nw 一个待办堆了 3 个 QC 任务）。 # 待办的本质是「请人类办事」，人类说完成/忽略 = 最终裁定，agent 审查人类是倒挂； # 与「人工 override 权威守卫须跳过」铁律一致（确定性规则，不靠 LLM 自觉）。 await sor.sqlExe( "UPDATE pipeline_human_tasks SET status='done', result_data=${rd}$, " "qc_status='passed', qc_comment='人类答复即权威（不送agent QC）', " "submitted_by=${oid}$, submitted_at=NOW() WHERE id=${hid}$", {"rd": result_json, "oid": operator_id, "hid": human_task_id}) await sor.sqlExe("COMMIT", {}) return True, human_task_id async def qc_human_task(human_task_id, passed, comment=None, operator_id=None): """QC 检查人类任务（general）。passed → qc_status=passed；否则 qc_status=rejected + status=pending 退回。""" if not human_task_id: return False, "缺少 human_task_id" db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: recs = await sor.sqlExe( "SELECT status FROM pipeline_human_tasks WHERE id=${hid}$", {"hid": human_task_id}) await sor.sqlExe("COMMIT", {}) if not recs: return False, "人类任务不存在" if passed: await sor.sqlExe( "UPDATE pipeline_human_tasks SET qc_status='passed', qc_comment=${c}$ " "WHERE id=${hid}$", {"c": comment or '', "hid": human_task_id}) else: # 退回重做：status 回 pending，qc_status=rejected await sor.sqlExe( "UPDATE pipeline_human_tasks SET status='pending', qc_status='rejected', " "qc_comment=${c}$ WHERE id=${hid}$", {"c": comment or '', "hid": human_task_id}) await sor.sqlExe("COMMIT", {}) return True, human_task_id async def list_project_human_tasks(project_id, status=None, assignee_id=None): """列出项目人类任务（显式任务 + bug 验收，不含冒泡问题）。""" db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: conds = ["project_id=${pid}$"] params = {"pid": project_id} if status: conds.append("status=${st}$") params["st"] = status if assignee_id: conds.append("assignee_id=${aid}$") params["aid"] = assignee_id where = " AND ".join(conds) recs = await sor.sqlExe( f"SELECT * FROM pipeline_human_tasks WHERE {where} ORDER BY created_at DESC", params) await sor.sqlExe("COMMIT", {}) return [_rec_to_dict(r) for r in (recs or [])] async def has_blocking_human_task(sor, project_id): """当前迭代是否有未完成的人类任务（流转门禁用）。 general：status=pending 或 status=done 但 qc_status!=passed → 阻塞 bug_acceptance：status=pending → 阻塞（done/rejected = 已验收出结果，不阻塞） """ if not project_id: return False it = await sor.sqlExe( "SELECT id FROM sd_iterations WHERE project_id=${pid}$ AND status='in_progress' " "ORDER BY seq_no DESC LIMIT 1", {"pid": project_id}) await sor.sqlExe("COMMIT", {}) if not it: return False iid = getattr(it[0], 'id', '') or '' if not iid: return False recs = await sor.sqlExe( "SELECT COUNT(*) AS c FROM pipeline_human_tasks " "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", {}) c = getattr(recs[0], 'c', 0) if recs else 0 return int(c) > 0 async def count_my_human_tasks(user_id, project_id=None): """角标 X = pending 问题(agentid=我) + 人类任务(assignee_id=我) + 人类任务(assignee_role∈我角色)。""" if not user_id: return 0 db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: # ① pending 问题 current_handler_agentid = 我 q_cond = "status='pending' AND current_handler_agentid=${u}$" q_params = {"u": user_id} if project_id: q_cond += " AND tenant_id=${pid}$" q_params["pid"] = project_id q = await sor.sqlExe( f"SELECT COUNT(*) AS c FROM pipeline_agent_questions WHERE {q_cond}", q_params) await sor.sqlExe("COMMIT", {}) # ② + ③ 人类任务（assignee_id=我 或 assignee_role∈我角色） roles = await _get_user_roles(sor, user_id) ht_cond = "status='pending' AND (assignee_id=${u}$" ht_params = {"u": user_id} if roles: ph = ",".join(f"'{r}'" for r in roles) ht_cond += f" OR assignee_role IN ({ph})" ht_cond += ")" if project_id: ht_cond += " AND project_id=${pid}$" ht_params["pid"] = project_id ht = await sor.sqlExe( f"SELECT COUNT(*) AS c FROM pipeline_human_tasks WHERE {ht_cond}", ht_params) await sor.sqlExe("COMMIT", {}) n_q = getattr(q[0], 'c', 0) if q else 0 n_ht = getattr(ht[0], 'c', 0) if ht else 0 _roles_cache = roles total = int(n_q) + int(n_ht) # 外部待办提供者（工单等，2026-09-10）：仅全局角标（project_id 空）计入—— # 工单不属于任何产线项目，项目内角标不混入。坏 provider 不拖垮计数。 if not project_id: for fn in list(_TODO_PROVIDERS): try: extra = await fn(user_id, _roles_cache, 100) total += len(extra or []) except Exception as e: logger.warning("todo provider count %s failed: %s", getattr(fn, '__module__', '?'), str(e)[:160]) return total async def list_my_human_todos(user_id, limit=100): """跨项目「我的待办」列表：union 人类任务 + 冒泡问题，按创建时间倒序。""" if not user_id: return [] db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: roles = await _get_user_roles(sor, user_id) role_ph = ",".join(f"'{r}'" for r in roles) if roles else "''" # 人类任务：assignee_id=我 或 assignee_role∈我角色，pending ht = await sor.sqlExe( f"SELECT id, project_id, iteration_id, bug_id, task_type, assignee_role, " f"assignee_id, status, qc_status, created_at, submitted_by, result_data, " f"title, description " f"FROM pipeline_human_tasks " f"WHERE status='pending' AND (assignee_id=${{u}}$ OR assignee_role IN ({role_ph})) " f"ORDER BY created_at DESC LIMIT {limit}", {"u": user_id}) await sor.sqlExe("COMMIT", {}) # 冒泡问题：current_handler_agentid=我 或 current_handler_role∈我角色，pending qs = await sor.sqlExe( f"SELECT id, tenant_id AS project_id, task_id, from_role, question, " f"problem_type, current_handler_role, current_handler_agentid, created_at " f"FROM pipeline_agent_questions " f"WHERE status='pending' AND (current_handler_agentid=${{u}}$ " f"OR current_handler_role IN ({role_ph})) " f"ORDER BY created_at DESC LIMIT {limit}", {"u": user_id}) await sor.sqlExe("COMMIT", {}) todos = [] for r in (ht or []): d = _rec_to_dict(r) d['source'] = 'human_task' todos.append(d) for r in (qs or []): d = _rec_to_dict(r) d['source'] = 'question' todos.append(d) _roles_cache = roles # 外部待办提供者（工单等，2026-09-10）：在 db context 外调用，provider 自开连接。 # 逐个 try/except——坏 provider 不拖垮整个待办列表（平台级机制必须健壮）。 for fn in list(_TODO_PROVIDERS): try: extra = await fn(user_id, _roles_cache, limit) for d in (extra or []): todos.append(d) except Exception as e: logger.warning("todo provider %s failed: %s", getattr(fn, '__module__', '?'), str(e)[:160]) todos.sort(key=lambda x: str(x.get('created_at') or ''), reverse=True) return todos[:limit] async def bug_accept(bug_id, accept, user_id, comment=None): """bug 验收：accept=True → bug closed + 验收任务 done；accept=False → bug reopen + 任务 rejected。 校验 user_id == bug.reporter_id。返回 (True, 消息) 或 (False, 错误)。 """ if not bug_id: return False, "缺少 bug_id" if not user_id: return False, "未登录" db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: brecs = await sor.sqlExe( "SELECT reporter_id, iteration_id, status FROM sd_bugs WHERE id=${bid}$", {"bid": bug_id}) await sor.sqlExe("COMMIT", {}) if not brecs: return False, "Bug 不存在" reporter_id = getattr(brecs[0], 'reporter_id', '') or '' iteration_id = getattr(brecs[0], 'iteration_id', '') or '' if reporter_id and user_id != reporter_id: return False, "仅 Bug 报告人可验收" if accept: from . import bug_capability ok, msg = await bug_capability.close_bug(bug_id, iteration_id, who='human') if not ok: return False, f"关闭失败: {msg}" # 验收任务 → done await sor.sqlExe( "UPDATE pipeline_human_tasks SET status='done', result_data=${c}$, " "submitted_by=${u}$, submitted_at=NOW() " "WHERE bug_id=${bid}$ AND task_type='bug_acceptance' AND status='pending'", {"c": json.dumps({"accept": True, "comment": comment or ''}, ensure_ascii=False), "u": user_id, "bid": bug_id}) await sor.sqlExe("COMMIT", {}) return True, "验收通过，Bug 已关闭" else: from . import bug_capability ok, msg = await bug_capability.reopen_bug(bug_id, iteration_id, who='human') if not ok: return False, f"重开失败: {msg}" await sor.sqlExe( "UPDATE pipeline_human_tasks SET status='rejected', result_data=${c}$, " "submitted_by=${u}$, submitted_at=NOW() " "WHERE bug_id=${bid}$ AND task_type='bug_acceptance' AND status='pending'", {"c": json.dumps({"accept": False, "comment": comment or ''}, ensure_ascii=False), "u": user_id, "bid": bug_id}) await sor.sqlExe("COMMIT", {}) return True, "验收不通过，Bug 已重新打开"