From 2c35abfdab2675cb39be32825f96045b626d5c3c Mon Sep 17 00:00:00 2001 From: ymq Date: Fri, 11 Sep 2026 12:58:26 +0800 Subject: [PATCH] =?UTF-8?q?feat(service):=20=E4=BA=A7=E7=BA=BF=E6=B5=81?= =?UTF-8?q?=E7=A8=8B=E8=A3=81=E5=89=AA=E6=9C=BA=E5=88=B6=E5=B1=82=E2=80=94?= =?UTF-8?q?=E2=80=94pipeline=5Fflow=5Fplans=E8=A1=A8+flow=5Fplan=5Fcapabil?= =?UTF-8?q?ity(propose=E5=85=AD=E9=A1=B9=E7=A1=AC=E6=A0=A1=E9=AA=8C/confir?= =?UTF-8?q?m-reject=20CAS/plan=5Fstage=5Fenabled/create=5Fsub=5Ftasks=20PM?= =?UTF-8?q?=E6=8B=86=E8=A7=A3)+shared=5Fability=E8=81=9A=E5=90=88=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E4=B8=89=E5=B7=A5=E5=85=B7(show=5Fflow=5Ftemplate/pro?= =?UTF-8?q?pose=5Fflow=5Fplan/get=5Fflow=5Fplan);agent=5Floop=20=5Fget=5Fn?= =?UTF-8?q?ext=5Frole=E8=B7=B3=E6=AE=B5=E6=B3=9B=E5=8C=96(=5Fskip=5Ftrimme?= =?UTF-8?q?d=5Froles=E7=BA=BF=E6=80=A7=E9=93=BE=E8=87=AA=E5=8A=A8=E8=B7=B3?= =?UTF-8?q?=E8=BF=87=E8=A2=AB=E8=A3=81=E8=A7=92=E8=89=B2)+=5Fpm=5Fcreate?= =?UTF-8?q?=5Ftasks=20extra=5Fparams(=E7=BB=A7=E6=89=BFskip=5Fgeneric=5Fqc?= =?UTF-8?q?);capability=5Ftools=E6=B3=A8=E5=86=8Ccreate=5Fsub=5Ftasks=20sc?= =?UTF-8?q?hema;init=E5=AF=BC=E5=87=BAenv.flow=5Fplan=5F*?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- models/pipeline_flow_plans.json | 121 ++++ pipeline_service/agent_loop.py | 51 +- pipeline_service/capability_tools.py | 9 + pipeline_service/flow_plan_capability.py | 690 +++++++++++++++++++++++ pipeline_service/init.py | 12 + pipeline_service/shared_ability.py | 8 + 6 files changed, 888 insertions(+), 3 deletions(-) create mode 100644 models/pipeline_flow_plans.json create mode 100644 pipeline_service/flow_plan_capability.py diff --git a/models/pipeline_flow_plans.json b/models/pipeline_flow_plans.json new file mode 100644 index 0000000..f4338d4 --- /dev/null +++ b/models/pipeline_flow_plans.json @@ -0,0 +1,121 @@ +{ + "summary": [ + { + "name": "pipeline_flow_plans", + "title": "产线流程裁剪计划", + "primary": [ + "id" + ], + "catelog": "entity" + } + ], + "fields": [ + { + "name": "id", + "title": "主键ID", + "type": "str", + "length": 32, + "nullable": "no" + }, + { + "name": "project_id", + "title": "项目ID", + "type": "str", + "length": 32, + "nullable": "no" + }, + { + "name": "pipeline_id", + "title": "产线ID", + "type": "str", + "length": 64, + "nullable": "no" + }, + { + "name": "version", + "title": "版本(驳回重提+1)", + "type": "int", + "nullable": "no", + "default": "1" + }, + { + "name": "status", + "title": "状态 draft/pending_confirm/confirmed/rejected/superseded", + "type": "str", + "length": 20, + "nullable": "no", + "default": "draft" + }, + { + "name": "stages", + "title": "阶段裁剪JSON(全量阶段快照,每项含key/enabled/reason)", + "type": "longtext", + "nullable": "no" + }, + { + "name": "user_requirements", + "title": "用户裁剪要求原文", + "type": "text" + }, + { + "name": "propose_note", + "title": "会话agent提案说明", + "type": "text" + }, + { + "name": "confirm_task_id", + "title": "确认待办ID(pipeline_human_tasks)", + "type": "str", + "length": 32 + }, + { + "name": "confirmed_by", + "title": "确认人", + "type": "str", + "length": 64 + }, + { + "name": "confirmed_at", + "title": "确认时间", + "type": "datetime" + }, + { + "name": "reject_comment", + "title": "驳回意见(会话agent按此修订重提)", + "type": "text" + }, + { + "name": "created_by", + "title": "提案人(agent.main_agent或user_id)", + "type": "str", + "length": 64 + }, + { + "name": "created_at", + "title": "创建时间", + "type": "timestamp", + "nullable": "no" + }, + { + "name": "updated_at", + "title": "更新时间", + "type": "timestamp" + } + ], + "indexes": [ + { + "name": "idx_pfp_project", + "idxtype": "index", + "idxfields": [ + "project_id" + ] + }, + { + "name": "idx_pfp_status", + "idxtype": "index", + "idxfields": [ + "status" + ] + } + ] +} diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 6f10661..95902be 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -1432,19 +1432,60 @@ async def _get_repo_state(space_dir, project_name=''): async def _get_next_role(current_role, project_id=""): - """任务链下一角色：从能力包取。""" + """任务链下一角色：从能力包取。 + + 流程裁剪跳段（通用，2026-09-11）：项目有已确认（confirmed）的流程裁剪计划时， + 下一角色若属于被裁阶段（FlowStage.roles 命中且 enabled=False），沿 next_role 链 + 前跳到第一个保留角色——线性任务链产线（SDLC/商机）裁剪后自动跳段，无需各产线改码。 + 无计划（None）= 标准全流程，行为不变。 + """ if project_id: try: pid = await _resolve_pipeline_id(project_id) from pipeline_core import get_role_spec spec = get_role_spec(pid, current_role) if spec: - return spec.next_role # 可能是 ""（终结） + nxt = spec.next_role # 可能是 ""（终结） + if nxt: + nxt = await _skip_trimmed_roles(pid, project_id, nxt) + return nxt except Exception: pass return "" +async def _skip_trimmed_roles(pipeline_id, project_id, next_role): + """沿 next_role 链跳过被裁阶段的角色，返回第一个保留角色（或 ""）。 + + 角色 → 阶段归属：能力包 flow_stages 里 roles 含该角色即属该阶段。 + 阶段不在任何 flow_stage 声明里的角色 = 不受裁剪影响，照常保留。 + 防环：最多跳 20 次（角色链本身有限，异常声明不至于死循环）。 + """ + from pipeline_core import get_role_spec, get_flow_stages + try: + from .flow_plan_capability import plan_stage_enabled + enabled = await plan_stage_enabled(project_id) + except Exception: + return next_role + if enabled is None: + return next_role # 无确认计划 = 标准全流程 + stages = get_flow_stages(pipeline_id) or [] + role_stage = {} + for s in stages: + for r in (s.roles or []): + role_stage.setdefault(r, s.key) + role = next_role + for _ in range(20): + if not role: + return "" + stage_key = role_stage.get(role) + if not stage_key or enabled.get(stage_key, True): + return role # 无阶段归属（不受裁剪影响）或阶段保留 + spec = get_role_spec(pipeline_id, role) + role = spec.next_role if spec else "" + return "" + + async def _check_orchestration_gaps(sor, project_id, task, next_role='', next_task_id=''): """编排完备性校验：代码只**查漏**，不替 PM 决策。 @@ -2593,7 +2634,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): # ── PM 审核 ── -async def _pm_create_tasks(sor, project_id, params, parent_task_id=None): +async def _pm_create_tasks(sor, project_id, params, parent_task_id=None, extra_params=None): """PM 派发后续任务（项目计划 / 任务分配）：批量创建任务并指定角色。 params.tasks: JSON 数组 [{title, role, description, key, depends_on, parent_id}]， @@ -2602,6 +2643,8 @@ async def _pm_create_tasks(sor, project_id, params, parent_task_id=None): - key：LLM 给子任务起的短标识，供同批任务间 depends_on 相互引用； - depends_on：依赖的 key 或任务ID 数组（空 = 并行；非空 = 串行等待依赖完成后才可认领）。 兼容单任务形态：params 直接含 {title, role, description}。 + extra_params（可选，向后兼容）：dict，合入每个子任务的 params—— + 用于继承父任务的产线门禁风格（如投标 skip_generic_qc），默认 None 不影响既有调用方。 """ from appPublic.uniqueID import getID tasks = params.get('tasks') or [] @@ -2657,6 +2700,8 @@ async def _pm_create_tasks(sor, project_id, params, parent_task_id=None): cancelled.append(did) tparams = {"description": desc, "pm_assigned": True} + if extra_params: + tparams.update(extra_params) # 任务来源标记：PM 派发的任务按 title/description 判断—— # 含「修复 Bug」= bug_fix（develop 必须走 fix_bug 状态机）；否则 = new_dev（模块开发等）。 _title_desc = f"{title} {desc}" diff --git a/pipeline_service/capability_tools.py b/pipeline_service/capability_tools.py index d9abde9..bc641c0 100644 --- a/pipeline_service/capability_tools.py +++ b/pipeline_service/capability_tools.py @@ -230,6 +230,15 @@ TOOL_SCHEMAS = { "content": "SKILL.md 正文（头部 定位，四段：触发条件/问题现象/根因/处理方法）"}, "required": ["name", "content"], }, + # ── flow_plan_capability：PM 复杂度拆解（通用，各产线 PM 角色技能声明即用）── + "create_sub_tasks": { + "module": "flow_plan_capability", + "description": "把当前复杂任务拆解为子任务组派发（编排子流程自动挂进项目主流程：parent_id 挂当前任务，任务树分层展示）。先按技能里的复杂度标准判断，确需拆解才调用。tasks 每项 {title, role, description, key?, depends_on?, dep_policy?}：key 供同批互引，depends_on 空=并行/非空=等前置完成，dep_policy 可选 {mode:all|any|at_least,n}。角色必须是本产线角色（list_roles 可查）", + "params": {"tasks": "子任务 JSON 数组 [{title, role, description, key?, depends_on?, dep_policy?}]", + "project_id": "项目ID(可选,默认当前)", + "parent_task_id": "父任务ID(可选,默认当前任务)"}, + "required": ["tasks"], + }, } diff --git a/pipeline_service/flow_plan_capability.py b/pipeline_service/flow_plan_capability.py new file mode 100644 index 0000000..b6b0eba --- /dev/null +++ b/pipeline_service/flow_plan_capability.py @@ -0,0 +1,690 @@ +# -*- coding: utf-8 -*- +"""产线流程裁剪能力（通用、产线无关）。 + +让任意产线的「预制流程」可被会话 agent 按用户要求裁剪，裁剪后的流程经用户在平台待办 +里确认才生效，确认后由各产线的流转机制（对账器 / next_role 链）按裁剪计划派发，PM 推进。 + +机制分工（沿用产线既有约定）： +- 阶段定义（哪些可裁、依赖关系）由各产线在能力包 flow_stages 字段声明（pipeline-core FlowStage）； +- 裁剪合法性由本模块**代码硬校验**（不靠 LLM 自觉）：不可裁阶段拒绝、依赖断裂拒绝、 + 已开跑项目拒绝、已确认计划拒绝改； +- 用户确认走平台待办（flow_plan_confirm 人类任务）+ 专用确认端点（CAS 激活）， + **agent 无确认工具**，物理上无法代替用户确认； +- 计划状态存 pipeline_flow_plans，流转机制读 get_active_plan / plan_stage_enabled 决定派发。 + +状态机：draft → pending_confirm →（confirm）confirmed /（reject）rejected →（重提）新版本 draft + superseded = 被更新版本取代的旧 pending/draft。 +""" + +import json +import logging + +from sqlor.dbpools import DBPools +from appPublic.uniqueID import getID + +from .audit import record_audit +from .human_task_capability import ( + create_human_task, _get_user_org, _get_user_roles, +) + +DBNAME = "pipeline" +logger = logging.getLogger("pipeline.flow_plan") + +# 计划状态 +S_DRAFT = "draft" +S_PENDING = "pending_confirm" +S_CONFIRMED = "confirmed" +S_REJECTED = "rejected" +S_SUPERSEDED = "superseded" + +# 确认待办类型（todo_detail.dspy 按它渲染确认/驳回按钮） +T_FLOW_CONFIRM = "flow_plan_confirm" + +# 裁剪时机门禁：项目已有角色任务**产出结果**（approved/completed）即视为流程已开跑， +# 禁止再裁。submitted/running 不拦——启动竞态（导入招标文件后对账器 15s 内可能已派 +# 编排任务）不应堵死裁剪窗口；确认后由流转机制按计划收敛（不再派发被裁阶段， +# 角色专属产线还会取消被裁阶段在办任务）。 +_STARTED_STATES = ("approved", "completed") + + +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(rec): + if rec is None: + return {} + try: + return dict(rec) + except (TypeError, ValueError): + return {} + + +def _parse_trim_keys(raw): + """trim_keys 容错解析：JSON 数组 / 逗号分隔字符串 / 已是 list → [key,...]。""" + if not raw: + return [] + if isinstance(raw, (list, tuple)): + return [str(x).strip() for x in raw if str(x).strip()] + s = str(raw).strip() + if not s: + return [] + if s.startswith("["): + try: + v = json.loads(s) + if isinstance(v, list): + return [str(x).strip() for x in v if str(x).strip()] + except (json.JSONDecodeError, ValueError): + pass + return [x.strip() for x in s.replace("，", ",").split(",") if x.strip()] + + +def _load_stages(pipeline_id): + """从能力包取产线声明的阶段，返回 (stage_map, ordered_list)。""" + from pipeline_core import get_flow_stages + stages = get_flow_stages(pipeline_id) or [] + return {s.key: s for s in stages}, list(stages) + + +async def _resolve_pipeline(sor, project_id, pipeline_id=""): + """解析项目产线 id（优先入参，缺省从 sd_projects 取），返回 (pid, proj_dict)。""" + proj = {} + if project_id: + recs = await sor.sqlExe( + "SELECT id, name, pipeline_id, status, org_id, created_by FROM sd_projects " + "WHERE id=${p}$", {"p": project_id}) + await sor.sqlExe("COMMIT", {}) + proj = _rec(recs[0]) if recs else {} + pid = (pipeline_id or proj.get("pipeline_id") or "").strip() + return pid, proj + + +# ══════════════════════ 展示预制流程 ══════════════════════ + +async def get_flow_template(project_id="", pipeline_id=""): + """返回 (pid, [stage_dict,...])：产线预制流程阶段（含可裁性），供 show_flow_template 渲染。""" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + pid, _proj = await _resolve_pipeline(sor, project_id, pipeline_id) + from pipeline_core import flow_stage_to_dict + _map, stages = _load_stages(pid) + return pid, [flow_stage_to_dict(s) for s in stages] + + +def render_template_md(stages): + """把阶段列表渲染成 markdown 表（展示给用户/待办）。""" + if not stages: + return "（该产线未声明可裁剪流程）" + trim_label = {"no": "不可裁", "warn": "可裁(需警告)", "yes": "可自由裁剪"} + lines = ["| # | 阶段 | 作用 | 可裁性 |", "|---|---|---|---|"] + for i, s in enumerate(stages, 1): + lines.append("| %d | %s | %s | %s |" % ( + i, s.get("label") or s.get("key"), + (s.get("description") or "").replace("\n", " ")[:120], + trim_label.get(s.get("trim"), s.get("trim")))) + return "\n".join(lines) + + +# ══════════════════════ 提案（裁剪草案 + 确认待办） ══════════════════════ + +async def propose_flow_plan(project_id, trim_keys=None, user_requirements="", + propose_note="", pipeline_id="", who=None, agent_id=None): + """按用户要求生成裁剪草案，落库 pending_confirm 并发确认待办。 + + 代码硬校验（任一不过即拒绝，返回可行动原因，绝不静默降级）： + 1. 项目/产线存在，且产线声明了 flow_stages； + 2. 裁剪时机：项目流程已开跑（有角色任务进入 started 态）→ 拒绝； + 3. 已有 confirmed 计划 → 拒绝（启动时裁剪一次，确认后不可改）； + 4. trim_keys 都是已知阶段； + 5. 不裁 trim=='no' 的阶段； + 6. 裁剪不导致依赖断裂（保留阶段的 deps 不得落在被裁阶段）。 + trim=='warn' 的阶段允许裁，但 warn_note 写进确认待办（双确认）。 + """ + if not project_id: + return False, "缺少 project_id（请先切换到项目）" + trim_keys = _parse_trim_keys(trim_keys) + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + pid, proj = await _resolve_pipeline(sor, project_id, pipeline_id) + if not proj: + return False, "项目不存在: %s" % project_id + if not pid: + return False, "项目未关联产线，无法裁剪流程" + stage_map, stages = _load_stages(pid) + if not stages: + return False, ("产线「%s」未声明可裁剪流程（flow_stages 为空），" + "不支持流程裁剪" % pid) + pname = proj.get("name") or project_id + + # 校验3：已确认计划不可改 + conf = await sor.sqlExe( + "SELECT id FROM pipeline_flow_plans WHERE project_id=${p}$ AND status=${s}$ LIMIT 1", + {"p": project_id, "s": S_CONFIRMED}) + await sor.sqlExe("COMMIT", {}) + if conf: + return False, ("项目流程计划已确认（plan %s），确认后不可修改。" + "如需调整请新建项目。" % getattr(conf[0], "id", "")[:8]) + + # 校验2：裁剪时机——流程已产出结果则拒绝（启动竞态容忍见 _STARTED_STATES 注释） + started = "','".join(_STARTED_STATES) + n_started = await sor.sqlExe( + "SELECT COUNT(*) AS c FROM pipeline_tasks WHERE tenant_id=${p}$ " + "AND pipeline_id='role_task' AND state IN ('" + started + "')", + {"p": project_id}) + await sor.sqlExe("COMMIT", {}) + n_started = int(getattr(n_started[0], "c", 0) if n_started else 0) + if n_started > 0 and trim_keys: + return False, ("项目流程已开始执行（%d 个角色任务已产出结果），" + "不能再裁剪。裁剪仅在项目启动阶段有效。" % n_started) + + # 校验4：未知阶段 + unknown = [k for k in trim_keys if k not in stage_map] + if unknown: + return False, ("未知阶段 key：%s。本产线可裁阶段见 show_flow_template。" + % "、".join(unknown)) + + # 校验5：不可裁阶段 + no_trim = [k for k in trim_keys if (stage_map[k].trim or "no") == "no"] + if no_trim: + detail = "；".join("%s（%s）" % (stage_map[k].label or k, + stage_map[k].description or "流程必需") + for k in no_trim) + return False, "以下阶段不可裁剪，裁掉流程不成立：%s" % detail + + # 校验6：依赖断裂——保留阶段的 dep 落在被裁阶段 + trim_set = set(trim_keys) + broken = [] + for s in stages: + if s.key in trim_set: + continue + for d in (s.deps or []): + if d in trim_set: + _dl = stage_map[d].label if d in stage_map else d + broken.append("%s 依赖被裁的 %s" % (s.label or s.key, _dl or d)) + if broken: + return False, "裁剪导致依赖断裂：%s。请保留上游阶段或一并裁剪下游。" % "；".join(broken) + + # warn 收集（trim==warn 允许裁，但风险提示进确认待办） + warns = [stage_map[k].warn_note for k in trim_keys + if (stage_map[k].trim or "") == "warn" and stage_map[k].warn_note] + + # 阶段快照 JSON（全量，每项 enabled 标记保留/裁剪） + from pipeline_core import flow_stage_to_dict + stages_snap = [] + for s in stages: + d = flow_stage_to_dict(s) + d["enabled"] = s.key not in trim_set + stages_snap.append(d) + + # version = 历史最大 + 1；旧 pending/draft 置 superseded（其确认待办同步关闭， + # 防用户看到指向旧版本的死待办） + vmax = await sor.sqlExe( + "SELECT MAX(version) AS v FROM pipeline_flow_plans WHERE project_id=${p}$", + {"p": project_id}) + await sor.sqlExe("COMMIT", {}) + version = int(getattr(vmax[0], "v", 0) or 0) + 1 + old_pend = await sor.sqlExe( + "SELECT id, confirm_task_id FROM pipeline_flow_plans WHERE project_id=${p}$ " + "AND status IN (${a}$, ${b}$)", + {"p": project_id, "a": S_PENDING, "b": S_DRAFT}) + await sor.sqlExe("COMMIT", {}) + for _op in (old_pend or []): + _op_ht = getattr(_op, "confirm_task_id", "") or "" + if _op_ht: + await sor.sqlExe( + "UPDATE pipeline_human_tasks SET status='done', qc_status='passed', " + "qc_comment='流程裁剪方案已更新版本，旧确认待办自动关闭', submitted_at=NOW() " + "WHERE id=${h}$ AND status='pending'", {"h": _op_ht}) + await sor.sqlExe( + "UPDATE pipeline_flow_plans SET status=${ns}$, updated_at=NOW() " + "WHERE project_id=${p}$ AND status IN (${a}$, ${b}$)", + {"ns": S_SUPERSEDED, "p": project_id, "a": S_PENDING, "b": S_DRAFT}) + await sor.sqlExe("COMMIT", {}) + + plan_id = getID() + await sor.C("pipeline_flow_plans", { + "id": plan_id, "project_id": project_id, "pipeline_id": pid, + "version": version, "status": S_PENDING, + "stages": json.dumps(stages_snap, ensure_ascii=False), + "user_requirements": (user_requirements or "")[:8000], + "propose_note": (propose_note or "")[:8000], + "created_by": who or "agent.main_agent", + }) + await sor.sqlExe("COMMIT", {}) + + # 确认待办（平台待办中枢；专用确认端点 flow_plan_confirm.dspy 激活） + desc = _render_confirm_md(pname, stages_snap, trim_keys, warns, + user_requirements, propose_note, version) + 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") + if ok_ht: + await sor.sqlExe( + "UPDATE pipeline_flow_plans SET confirm_task_id=${h}$, updated_at=NOW() " + "WHERE id=${i}$", {"h": ht, "i": plan_id}) + await sor.sqlExe("COMMIT", {}) + await record_audit(project_id, "pipeline_flow_plans", plan_id, "propose", + to_state=S_PENDING, who=who, agent_id=agent_id, + detail="version=%d trim=%s" % (version, ",".join(trim_keys) or "(无裁剪)"), + sor=sor) + await sor.sqlExe("COMMIT", {}) + logger.info("flow_plan propose: project=%s plan=%s v=%d trim=%s", + project_id, plan_id, version, trim_keys) + kept = [s["label"] or s["key"] for s in stages_snap if s["enabled"]] + trimmed = [s["label"] or s["key"] for s in stages_snap if not s["enabled"]] + msg = ("已生成流程裁剪草案（plan %s，v%d）并发确认待办。\n保留阶段：%s\n裁剪阶段：%s" + % (plan_id[:8], version, "、".join(kept) or "（无）", + "、".join(trimmed) or "（无）")) + if warns: + msg += "\n⚠️ 风险提示（请在待办里向用户说明）：\n" + "\n".join("- " + w for w in warns) + msg += "\n用户需在「流程裁剪确认」待办里点击确认后流程才生效；agent 不能代替确认。" + return True, msg + + +def _render_confirm_md(pname, stages_snap, trim_keys, warns, user_req, note, version): + """确认待办正文（markdown，todo_detail 直接渲染）。""" + lines = ["## 项目「%s」流程裁剪方案（v%d）" % (pname, version), ""] + if user_req: + lines += ["**你的裁剪要求：**", "", user_req.strip(), ""] + if note: + lines += ["**助手说明：**", "", note.strip(), ""] + lines += ["### 裁剪后流程（保留 → 执行；裁剪 → 跳过）", "", + "| # | 阶段 | 处理 |", "|---|---|---|"] + for i, s in enumerate(stages_snap, 1): + mark = "✅ 保留" if s["enabled"] else "✂️ 裁剪" + lines.append("| %d | %s | %s |" % (i, s.get("label") or s.get("key"), mark)) + if warns: + lines += ["", "### ⚠️ 裁剪风险（裁掉的是质量门禁，请确认知晓）", ""] + lines += ["- " + w for w in warns] + lines += ["", "---", "", + "点击「确认并启用」后流程按上表生效，**确认后不可再改**；", + "点击「驳回」并填写意见，助手会按你的意见修订后重新提案。"] + return "\n".join(lines) + + +# ══════════════════════ 用户确认 / 驳回（专用端点调用） ══════════════════════ + +async def _check_confirm_operator(sor, plan, operator_id): + """确认权限：同机构 + （owner.superuser 角色 或 项目创建者本人）。""" + if not operator_id: + return False, "未登录" + project_id = plan.get("project_id", "") + prec = await sor.sqlExe("SELECT org_id, created_by FROM sd_projects WHERE id=${p}$", + {"p": project_id}) + await sor.sqlExe("COMMIT", {}) + proj = _rec(prec[0]) if prec else {} + proj_org = proj.get("org_id", "") or "" + user_org = await _get_user_org(sor, operator_id) + if proj_org and user_org != proj_org: + return False, "仅同机构用户可确认该项目流程" + roles = await _get_user_roles(sor, operator_id) + if "owner.superuser" in roles: + return True, "" + if operator_id == (proj.get("created_by") or ""): + return True, "" + return False, "仅项目 owner / 超级管理员可确认流程裁剪" + + +async def confirm_flow_plan(plan_id, operator_id): + """用户确认：pending_confirm → confirmed（CAS）+ 关待办 + 收敛清理被裁阶段在办任务。""" + if not plan_id: + return False, "缺少 plan_id" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT * FROM pipeline_flow_plans WHERE id=${i}$", {"i": plan_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return False, "流程计划不存在: %s" % plan_id + plan = _rec(recs[0]) + if plan.get("status") != S_PENDING: + return False, "计划不在待确认状态（当前 %s）" % plan.get("status") + ok, err = await _check_confirm_operator(sor, plan, operator_id) + if not ok: + return False, err + project_id = plan.get("project_id", "") + # CAS：仅 pending_confirm 可确认（防并发/重复） + await sor.sqlExe( + "UPDATE pipeline_flow_plans SET status=${ns}$, confirmed_by=${u}$, " + "confirmed_at=NOW(), updated_at=NOW() WHERE id=${i}$ AND status=${os}$", + {"ns": S_CONFIRMED, "u": operator_id, "i": plan_id, "os": S_PENDING}) + await sor.sqlExe("COMMIT", {}) + chk = await sor.sqlExe("SELECT status FROM pipeline_flow_plans WHERE id=${i}$", + {"i": plan_id}) + await sor.sqlExe("COMMIT", {}) + if not chk or getattr(chk[0], "status", "") != S_CONFIRMED: + return False, "确认竞争失败（计划已被处理）" + # 关确认待办 + ht = plan.get("confirm_task_id", "") or "" + if ht: + await sor.sqlExe( + "UPDATE pipeline_human_tasks SET status='done', qc_status='passed', " + "qc_comment='用户已确认流程裁剪方案', submitted_by=${u}$, submitted_at=NOW() " + "WHERE id=${h}$ AND status='pending'", {"u": operator_id, "h": ht}) + await sor.sqlExe("COMMIT", {}) + # 收敛清理：被裁阶段角色的在办任务取消（防确认后旧任务残留） + n_cancel = await _cancel_trimmed_tasks(sor, project_id, plan.get("stages")) + await record_audit(project_id, "pipeline_flow_plans", plan_id, "confirm", + from_state=S_PENDING, to_state=S_CONFIRMED, who=operator_id, + detail="user confirmed; cancelled %d trimmed-stage tasks" % n_cancel, + sor=sor) + await sor.sqlExe("COMMIT", {}) + logger.info("flow_plan confirmed: project=%s plan=%s by=%s cancelled=%d", + project_id, plan_id, operator_id, n_cancel) + return True, "流程裁剪已确认并启用，系统将按裁剪后的流程派发任务。" + + +async def reject_flow_plan(plan_id, comment, operator_id): + """用户驳回：pending_confirm → rejected + 记意见（会话 agent 据此修订重提）。""" + if not plan_id: + return False, "缺少 plan_id" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT * FROM pipeline_flow_plans WHERE id=${i}$", {"i": plan_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return False, "流程计划不存在: %s" % plan_id + plan = _rec(recs[0]) + if plan.get("status") != S_PENDING: + return False, "计划不在待确认状态（当前 %s）" % plan.get("status") + ok, err = await _check_confirm_operator(sor, plan, operator_id) + if not ok: + return False, err + project_id = plan.get("project_id", "") + await sor.sqlExe( + "UPDATE pipeline_flow_plans SET status=${ns}$, reject_comment=${c}$, " + "updated_at=NOW() WHERE id=${i}$ AND status=${os}$", + {"ns": S_REJECTED, "c": (comment or "")[:4000], "i": plan_id, "os": S_PENDING}) + await sor.sqlExe("COMMIT", {}) + ht = plan.get("confirm_task_id", "") or "" + if ht: + await sor.sqlExe( + "UPDATE pipeline_human_tasks SET status='rejected', " + "qc_comment=${c}$, submitted_by=${u}$, submitted_at=NOW() " + "WHERE id=${h}$ AND status='pending'", + {"c": (comment or "")[:2000], "u": operator_id, "h": ht}) + await sor.sqlExe("COMMIT", {}) + await record_audit(project_id, "pipeline_flow_plans", plan_id, "reject", + from_state=S_PENDING, to_state=S_REJECTED, who=operator_id, + detail=(comment or "")[:500], sor=sor) + await sor.sqlExe("COMMIT", {}) + logger.info("flow_plan rejected: project=%s plan=%s by=%s", project_id, plan_id, operator_id) + return True, "已驳回，助手将按你的意见修订流程后重新提案。" + + +async def _cancel_trimmed_tasks(sor, project_id, stages_json): + """取消「仅属于被裁阶段」角色的在办任务（收敛清理，防确认后旧任务残留）。 + + 角色可能跨阶段共用（如投标 tender_analyst 属于 4 个解析维度阶段）—— + 只裁其中一个阶段时该角色仍被保留阶段需要，**不得取消**。 + 仅当角色不属于任何保留阶段时才取消其在办任务。 + """ + try: + snap = json.loads(stages_json) if isinstance(stages_json, str) else (stages_json or []) + except (json.JSONDecodeError, ValueError): + snap = [] + trimmed_roles, kept_roles = set(), set() + for s in snap: + if not isinstance(s, dict): + continue + for r in (s.get("roles") or []): + if r: + (kept_roles if s.get("enabled", True) else trimmed_roles).add(r) + roles = trimmed_roles - kept_roles # 只裁独占角色 + if not roles: + return 0 + states = "','".join(("submitted", "running", "review", "qc_review", "waiting")) + role_in = "','".join(sorted(roles)) + recs = await sor.sqlExe( + "SELECT id FROM pipeline_tasks WHERE tenant_id=${p}$ AND pipeline_id='role_task' " + "AND role IN ('" + role_in + "') AND state IN ('" + states + "')", + {"p": project_id}) + await sor.sqlExe("COMMIT", {}) + ids = [getattr(r, "id", "") for r in (recs or []) if getattr(r, "id", "")] + if not ids: + return 0 + for tid in ids: + await sor.sqlExe( + "UPDATE pipeline_tasks SET state='cancelled', claimed_by=NULL, updated_at=NOW() " + "WHERE id=${t}$ AND state IN ('" + states + "')", {"t": tid}) + await record_audit(project_id, "pipeline_tasks", tid, "cancel", + to_state="cancelled", who="system.flow_plan", + detail="流程裁剪确认：被裁阶段在办任务收敛取消", sor=sor) + await sor.sqlExe("COMMIT", {}) + return len(ids) + + +# ══════════════════════ 流转机制读取（对账器 / next_role 跳段） ══════════════════════ + +async def get_active_plan(project_id): + """取项目已确认（confirmed）的流程计划 dict；无则返回 {}。""" + if not project_id: + return {} + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT * FROM pipeline_flow_plans WHERE project_id=${p}$ AND status=${s}$ " + "ORDER BY version DESC LIMIT 1", {"p": project_id, "s": S_CONFIRMED}) + await sor.sqlExe("COMMIT", {}) + return _rec(recs[0]) if recs else {} + + +async def plan_stage_enabled(project_id): + """返回 {stage_key: bool} —— confirmed 计划的阶段启用表；无计划返回 None（用标准全流程）。 + + 流转机制（对账器/next_role）据此判定某阶段是否执行：None=不裁剪（标准全流程）， + dict 且 dict.get(key, True)=False 表示该阶段被裁。 + """ + plan = await get_active_plan(project_id) + if not plan: + return None + try: + snap = json.loads(plan.get("stages") or "[]") + except (json.JSONDecodeError, ValueError): + snap = [] + return {s.get("key"): bool(s.get("enabled", True)) for s in snap if isinstance(s, dict)} + + +async def stage_enabled(project_id, stage_key, default=True): + """单个阶段是否启用（无计划→default，通常 True=标准全流程都执行）。""" + tbl = await plan_stage_enabled(project_id) + if tbl is None: + return default + return tbl.get(stage_key, default) + + +async def get_latest_plan(project_id): + """取项目最新一条计划（任意状态，供 get_flow_plan 工具展示进度）。""" + if not project_id: + return {} + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT * FROM pipeline_flow_plans WHERE project_id=${p}$ " + "ORDER BY version DESC, created_at DESC LIMIT 1", {"p": project_id}) + await sor.sqlExe("COMMIT", {}) + return _rec(recs[0]) if recs else {} + + +# ══════════════════════ PM 复杂度拆解：create_sub_tasks（通用角色工具） ══════════════════════ + +async def create_sub_tasks(project_id="", tasks=None, parent_task_id="", task_id="", + who=None, agent_id=None): + """PM 把复杂任务拆解为子任务组并挂进项目主流程（通用，各产线 PM 角色可用）。 + + 复杂度判断归 PM（技能里的判断标准）；本工具是机制执行件，复用引擎 + _pm_create_tasks 的完整语义： + - parent_id 挂父任务（任务树分层展示，子流程自动进主流程视图）； + - key/depends_on 同批互引 + dep_policy 启动策略（all/any/at_least）； + - 依赖解析不完整 → 任务置 waiting + 冒泡，绝不静默降级为并行； + - 同标题活跃任务自动取消（重做保护）； + - 派发后跑编排查漏（G1/G2/G3），缺口当场返回给 PM。 + 角色守卫：子任务角色必须属于当前产线角色集（防跨产线角色名幻觉建死任务）。 + """ + if not project_id: + return False, "缺少 project_id" + if isinstance(tasks, str): + try: + tasks = json.loads(tasks) + except (json.JSONDecodeError, ValueError): + return False, "tasks 必须是 JSON 数组字符串" + if not tasks or not isinstance(tasks, list): + return False, "tasks 必须是子任务 JSON 数组（每项 {title, role, description, key?, depends_on?}）" + + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + # 产线角色守卫（同 _h_create_task 语义）：查项目产线，角色必须在其角色集 + prec = await sor.sqlExe( + "SELECT pipeline_id FROM sd_projects WHERE id=${p}$", {"p": project_id}) + await sor.sqlExe("COMMIT", {}) + pid = getattr(prec[0], "pipeline_id", "") if prec else "" + if pid: + from pipeline_core import get_role_spec, list_roles as _list_roles + if _list_roles(pid): + bad = [] + for t in tasks: + if not isinstance(t, dict): + continue + r = str(t.get("role") or "").strip() + if not r: + bad.append("(缺role)「%s」" % str(t.get("title") or "")[:30]) + continue + spec = get_role_spec(pid, r) + if not spec: + bad.append("「%s」" % r) + else: + t["role"] = spec.name # 归一到注册表规范名 + if bad: + avail = "、".join(s.name for s in _list_roles(pid)) + return False, ("角色不属于产线「%s」：%s。可用角色：%s" + % (pid, "、".join(bad), avail)) + + from .agent_loop import _pm_create_tasks + parent = (parent_task_id or task_id or "").strip() + # 继承父任务的产线门禁风格：投标等产线角色任务带 skip_generic_qc（产线自带 + # 质量门禁，豁免引擎通用 QC）——子任务必须继承，否则拆解出的任务误走通用门禁。 + extra_params = None + if parent: + prec2 = await sor.sqlExe( + "SELECT params FROM pipeline_tasks WHERE id=${t}$ AND tenant_id=${p}$", + {"t": parent, "p": project_id}) + await sor.sqlExe("COMMIT", {}) + if prec2: + try: + pp = json.loads(getattr(prec2[0], "params", "") or "{}") + except (json.JSONDecodeError, TypeError): + pp = {} + if isinstance(pp, dict) and pp.get("skip_generic_qc"): + extra_params = {"skip_generic_qc": True} + result = await _pm_create_tasks( + sor, project_id, {"tasks": tasks, "parent_id": parent}, + parent_task_id=parent or None, extra_params=extra_params) + await sor.sqlExe("COMMIT", {}) + await record_audit(project_id, "pipeline_tasks", parent or project_id, + "create_sub_tasks", who=who, agent_id=agent_id, + detail="PM 拆解 %d 个子任务 parent=%s" % (len(tasks), parent or "(无)"), + sor=sor) + await sor.sqlExe("COMMIT", {}) + if str(result).startswith("FAIL"): + return False, str(result) + return True, str(result) + + +# ══════════════════════ 会话工具（FLOW_PLAN_TOOLS / HANDLERS，shared_ability 聚合） ══════════════════════ + +try: + from pipeline_core import ToolDefinition +except Exception: # pragma: no cover - core 不可用时降级（工具不注册） + ToolDefinition = None + +FLOW_PLAN_TOOLS = [] +if ToolDefinition is not None: + FLOW_PLAN_TOOLS = [ + ToolDefinition( + name="show_flow_template", + description="查看当前项目所属产线的预制流程（各阶段作用 + 可裁性：不可裁/可裁需警告/可自由裁剪）。用户提出裁剪流程前先用它了解可裁项", + parameters={}, required=[], category="project"), + ToolDefinition( + name="propose_flow_plan", + description=("按用户要求裁剪项目流程，生成裁剪草案并发「流程裁剪确认」待办给用户。" + "裁剪合法性由系统硬校验（不可裁阶段/依赖断裂/已开跑会被拒绝并说明原因）。" + "草案需用户在待办里确认才生效——你没有确认权，禁止声称已确认。" + "trim_keys 传要裁掉的阶段 key 数组（来自 show_flow_template）"), + parameters={"trim_keys": "要裁剪的阶段key数组(如[\"analysis_cost_benefit\"])，空数组=不裁剪仅走确认", + "user_requirements": "用户裁剪要求原文(必填，写进确认待办)", + "propose_note": "你对裁剪方案的说明/风险提示(可选)"}, + required=["user_requirements"], category="project"), + ToolDefinition( + name="get_flow_plan", + description="查看当前项目流程裁剪计划的状态（草案/待确认/已确认/已驳回+驳回意见）", + parameters={}, required=[], category="project"), + ] + + +async def _h_show_flow_template(sor, params, ctx): + pid = (ctx.get("pipeline_id") or "").strip() + project_id = ctx.get("project_id") or "" + pid2, stages = await get_flow_template(project_id=project_id, pipeline_id=pid) + if not stages: + return ("当前产线（%s）未声明可裁剪流程，按标准全流程执行，无需裁剪。" + % (pid2 or "未知")) + return ("产线「%s」预制流程（共 %d 阶段）：\n%s\n\n" + "用户要裁剪时，把要裁掉的阶段 key 传给 propose_flow_plan。" + % (pid2 or "?", len(stages), render_template_md(stages))) + + +async def _h_propose_flow_plan(sor, params, ctx): + project_id = ctx.get("project_id") or "" + if not project_id: + return "ERROR: 请先切换到项目（用 switch_project 或在项目列表选择）" + ok, msg = await propose_flow_plan( + project_id, + trim_keys=params.get("trim_keys"), + user_requirements=params.get("user_requirements", "") or "", + propose_note=params.get("propose_note", "") or "", + pipeline_id=ctx.get("pipeline_id", "") or "", + who="agent.main_agent", + agent_id=ctx.get("user_id", "") or "") + return ("OK: " if ok else "ERROR: ") + msg + + +async def _h_get_flow_plan(sor, params, ctx): + project_id = ctx.get("project_id") or "" + if not project_id: + return "ERROR: 请先切换到项目" + plan = await get_latest_plan(project_id) + if not plan: + return "当前项目尚无流程裁剪计划（按产线标准全流程执行）。用 show_flow_template 查看可裁项。" + st_label = {S_DRAFT: "草案", S_PENDING: "待用户确认", S_CONFIRMED: "已确认生效", + S_REJECTED: "已驳回", S_SUPERSEDED: "已被新版本取代"} + try: + snap = json.loads(plan.get("stages") or "[]") + except (json.JSONDecodeError, ValueError): + snap = [] + kept = [str(s.get("label") or s.get("key") or "?") for s in snap if isinstance(s, dict) and s.get("enabled")] + trimmed = [str(s.get("label") or s.get("key") or "?") for s in snap if isinstance(s, dict) and not s.get("enabled")] + lines = ["流程裁剪计划 v%s（%s）" % (plan.get("version"), st_label.get(plan.get("status"), plan.get("status"))), + "保留：%s" % ("、".join(kept) or "（无）"), + "裁剪：%s" % ("、".join(trimmed) or "（无）")] + if plan.get("status") == S_REJECTED and plan.get("reject_comment"): + lines.append("驳回意见：%s" % plan.get("reject_comment")) + lines.append("→ 请按驳回意见修订后用 propose_flow_plan 重新提案（version 自动 +1）。") + if plan.get("status") == S_PENDING: + lines.append("→ 已发确认待办，等用户在「流程裁剪确认」待办里确认；你不能代替确认。") + return "\n".join(lines) + + +FLOW_PLAN_HANDLERS = { + "show_flow_template": _h_show_flow_template, + "propose_flow_plan": _h_propose_flow_plan, + "get_flow_plan": _h_get_flow_plan, +} diff --git a/pipeline_service/init.py b/pipeline_service/init.py index 031a70d..ff693c7 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -531,6 +531,18 @@ def load_pipeline_service(): env.suggest_kb_ingest = suggest_kb_ingest env.decide_kb_ingest = decide_kb_ingest + # 产线流程裁剪（提案/确认/驳回/查询，确认端点 flow_plan_confirm.dspy 调用） + from .flow_plan_capability import ( + propose_flow_plan, confirm_flow_plan, reject_flow_plan, + get_active_plan, get_latest_plan, get_flow_template, + ) + env.flow_plan_propose = propose_flow_plan + env.flow_plan_confirm = confirm_flow_plan + env.flow_plan_reject = reject_flow_plan + env.flow_plan_active = get_active_plan + env.flow_plan_latest = get_latest_plan + env.flow_template = get_flow_template + # Register default handler register_default_handler() diff --git a/pipeline_service/shared_ability.py b/pipeline_service/shared_ability.py index b77fa13..dfb8c5f 100644 --- a/pipeline_service/shared_ability.py +++ b/pipeline_service/shared_ability.py @@ -55,6 +55,14 @@ def _build(): missing = names - set(SHARED_HANDLERS) if missing: logger.info("shared_ability: 以下通用工具未找到 handler（跳过）: %s", sorted(missing)) + # 流程裁剪工具（通用机制，产线声明 flow_stages 即可用；未声明的产线工具诚实降级） + try: + from .flow_plan_capability import FLOW_PLAN_TOOLS, FLOW_PLAN_HANDLERS + _seen = {getattr(t, "name", "") for t in SHARED_TOOLS} + SHARED_TOOLS += [t for t in FLOW_PLAN_TOOLS if getattr(t, "name", "") not in _seen] + SHARED_HANDLERS.update(FLOW_PLAN_HANDLERS) + except Exception as e: + logger.warning("shared_ability: 流程裁剪工具不可用（跳过）: %s", str(e)[:160]) logger.info("shared_ability: %d tools / %d handlers 可复用", len(SHARED_TOOLS), len(SHARED_HANDLERS))