From c8b44fe4235ce89188b4f448c2a5c97e60673ee2 Mon Sep 17 00:00:00 2001 From: ymq Date: Fri, 18 Sep 2026 00:05:42 +0800 Subject: [PATCH] =?UTF-8?q?feat(flow=5Fplan):=20=E6=B5=81=E7=A8=8B?= =?UTF-8?q?=E4=BF=AE=E8=AE=A2/=E5=AE=9A=E5=90=91=E9=87=8D=E5=81=9A?= =?UTF-8?q?=E6=9C=BA=E5=88=B6(2026-09-17=E7=94=A8=E6=88=B7=E8=A3=81?= =?UTF-8?q?=E5=AE=9A:=E7=94=A8=E6=88=B7=E6=9C=89=E6=9D=83=E8=B0=83?= =?UTF-8?q?=E6=95=B4=E6=B5=81=E7=A8=8B,=E5=B9=B6=E8=A6=81=E6=B1=82?= =?UTF-8?q?=E4=BB=8E=E6=B5=81=E7=A8=8B=E4=BB=BB=E4=BD=95=E4=B8=80=E6=AD=A5?= =?UTF-8?q?=E6=8C=89=E6=84=8F=E8=A7=81=E9=87=8D=E5=81=9A,=E6=88=96?= =?UTF-8?q?=E5=8F=AA=E7=BB=99=E6=84=8F=E8=A7=81=E7=94=B1agent=E5=88=A4?= =?UTF-8?q?=E6=B3=A8=E5=85=A5=E7=82=B9)=E2=80=94=E2=80=94revise=5Fflow=5Fp?= =?UTF-8?q?lan:=E5=B7=B2confirmed=E8=AE=A1=E5=88=92=E6=8F=90=E4=BF=AE?= =?UTF-8?q?=E8=AE=A2=E7=89=88(v+1=20pending=5Fconfirm,=E7=A1=AE=E8=AE=A4?= =?UTF-8?q?=E5=89=8D=E6=97=A7=E8=AE=A1=E5=88=92=E6=8C=81=E7=BB=AD=E7=94=9F?= =?UTF-8?q?=E6=95=88,=E7=A1=AE=E8=AE=A4=E6=97=B6CAS=E6=8D=A2=E7=89=88?= =?UTF-8?q?=E6=97=A7=E7=89=88superseded+=E5=9B=9E=E8=B0=83=E4=BA=A7?= =?UTF-8?q?=E7=BA=BFstage=5Freentry=E9=92=A9=E5=AD=90);detect=5Freentry=5F?= =?UTF-8?q?stage:LLM=E6=8C=89=E6=84=8F=E8=A7=81+=E9=98=B6=E6=AE=B5?= =?UTF-8?q?=E5=A3=B0=E6=98=8E+=E4=BA=A7=E7=BA=BF=E8=BF=9B=E5=BA=A6?= =?UTF-8?q?=E6=91=98=E8=A6=81=E5=88=A4=E6=B3=A8=E5=85=A5=E7=82=B9(?= =?UTF-8?q?=E5=AF=B9=E9=BD=90detect=5Fflow=5Ftemplate=E5=88=A4=E6=B5=81?= =?UTF-8?q?=E5=85=88=E4=BE=8B,evidence=E8=BF=9B=E7=A1=AE=E8=AE=A4=E5=BE=85?= =?UTF-8?q?=E5=8A=9E=E7=94=A8=E6=88=B7=E6=A0=B8=E5=AF=B9,=E9=94=99?= =?UTF-8?q?=E5=88=A4=E7=A1=AE=E8=AE=A4=E9=97=A8=E5=85=9C=E4=BD=8F,?= =?UTF-8?q?=E5=88=A4=E4=B8=8D=E5=87=BA=E8=AF=9A=E5=AE=9E=E6=8A=A5=E9=94=99?= =?UTF-8?q?=E7=A6=81=E9=BB=98=E8=AE=A4);=5Faffected=5Fstage=5Fkeys:?= =?UTF-8?q?=E6=B3=A8=E5=85=A5=E9=98=B6=E6=AE=B5+deps=E4=BE=9D=E8=B5=96?= =?UTF-8?q?=E9=97=AD=E5=8C=85=E4=B8=8B=E6=B8=B8;=E9=92=A9=E5=AD=90?= =?UTF-8?q?=E5=A4=B1=E8=B4=A5=E4=B8=8D=E5=9B=9E=E6=BB=9A=E7=A1=AE=E8=AE=A4?= =?UTF-8?q?,=E8=AF=9A=E5=AE=9E=E4=B8=8A=E6=8A=A5+=E6=8A=9B=E4=BA=BA?= =?UTF-8?q?=E5=B7=A5=E5=BE=85=E5=8A=9E;=E6=96=B0=E4=BC=9A=E8=AF=9D?= =?UTF-8?q?=E5=B7=A5=E5=85=B7revise=5Fflow=5Fplan/detect=5Freentry=5Fstage?= =?UTF-8?q?=E8=BF=9BFLOW=5FPLAN=5FTOOLS(=E4=B8=89=E4=BA=A7=E7=BA=BF?= =?UTF-8?q?=E9=9B=B6=E6=94=B9=E5=8A=A8=E7=BB=A7=E6=89=BF);propose=E5=B7=B2?= =?UTF-8?q?=E7=A1=AE=E8=AE=A4=E6=97=B6=E9=94=99=E8=AF=AF=E6=8C=87=E5=BC=95?= =?UTF-8?q?=E6=94=B9=E6=8C=87revise;models=E5=8A=A0reentry=5Fstage/reentry?= =?UTF-8?q?=5Fdetect=5Fnote=E5=88=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- models/pipeline_flow_plans.json | 15 +- pipeline_service/flow_plan_capability.py | 514 ++++++++++++++++++++++- 2 files changed, 522 insertions(+), 7 deletions(-) diff --git a/models/pipeline_flow_plans.json b/models/pipeline_flow_plans.json index 91c170f..79fba58 100644 --- a/models/pipeline_flow_plans.json +++ b/models/pipeline_flow_plans.json @@ -87,6 +87,19 @@ "title": "确认时间", "type": "datetime" }, + { + "name": "reentry_stage", + "title": "定向重做注入阶段key(流程修订机制,2026-09-17;空=仅裁剪修订不重做)", + "type": "str", + "length": 64, + "nullable": "yes", + "default": "" + }, + { + "name": "reentry_detect_note", + "title": "注入点判定依据(LLM判流evidence/agent说明,进确认待办给用户核对)", + "type": "text" + }, { "name": "reject_comment", "title": "驳回意见(会话agent按此修订重提)", @@ -126,4 +139,4 @@ ] } ] -} +} \ No newline at end of file diff --git a/pipeline_service/flow_plan_capability.py b/pipeline_service/flow_plan_capability.py index 61ce025..14a04c5 100644 --- a/pipeline_service/flow_plan_capability.py +++ b/pipeline_service/flow_plan_capability.py @@ -272,8 +272,10 @@ async def propose_flow_plan(project_id, trim_keys=None, user_requirements="", {"p": project_id, "s": S_CONFIRMED}) await sor.sqlExe("COMMIT", {}) if conf: - return False, ("项目流程计划已确认（plan %s），确认后不可修改。" - "如需调整请新建项目。" % getattr(conf[0], "id", "")[:8]) + return False, ("项目流程计划已确认（plan %s）。运行中调整流程请改用 " + "revise_flow_plan 提出修订（可改裁剪、可从任一阶段按用户意见" + "定向重做），修订经用户确认后生效。" + % getattr(conf[0], "id", "")[:8]) # 校验2：裁剪时机——流程已产出结果则拒绝（启动竞态容忍见 _STARTED_STATES 注释） started = "','".join(_STARTED_STATES) @@ -429,8 +431,8 @@ def _render_confirm_md(pname, stages_snap, trim_keys, warns, user_req, note, ver lines += ["", "### ⚠️ 裁剪风险（裁掉的是质量门禁，请确认知晓）", ""] lines += ["- " + w for w in warns] lines += ["", "---", "", - "点击「确认并启用」后流程按上表生效，**确认后不可再改**；", - "点击「驳回」并填写意见，助手会按你的意见修订后重新提案。"] + "点击「确认并启用」后流程按上表生效；生效后可随时再提修订（revise_flow_plan，", + "含从任一阶段定向重做）；点击「驳回」并填写意见，助手会按你的意见修订后重新提案。"] return "\n".join(lines) @@ -494,16 +496,37 @@ async def confirm_flow_plan(plan_id, operator_id): "qc_comment='用户已确认流程裁剪方案', submitted_by=${u}$, submitted_at=NOW() " "WHERE id=${h}$ AND status='pending'", {"u": operator_id, "h": ht}) await sor.sqlExe("COMMIT", {}) + # 修订版换版（2026-09-17 流程修订机制）：旧 confirmed 版本置 superseded + # ——同一项目同时只允许一个 confirmed 计划（plan_stage_enabled/get_active_plan + # 按 version DESC 取最新，但显式换版语义更清晰，审计可追溯）。 + await sor.sqlExe( + "UPDATE pipeline_flow_plans SET status=${ns}$, updated_at=NOW() " + "WHERE project_id=${p}$ AND status=${cs}$ AND id<>${i}$", + {"ns": S_SUPERSEDED, "p": project_id, "cs": S_CONFIRMED, "i": plan_id}) + 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, + detail="user confirmed; old confirmed superseded; " + "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, "流程裁剪已确认并启用，系统将按裁剪后的流程派发任务。" + # 定向重做（reentry_stage 非空）：调产线钩子执行阶段重置。放在计划生效后—— + # 重置后产线流转机制按新计划+被重置的状态自动重新推进。 + reentry_msg = "" + if (plan.get("reentry_stage") or "").strip(): + ok_r, msg_r = await _run_stage_reentry(sor, plan, operator_id) + if ok_r: + reentry_msg = "\n定向重做已执行：%s" % (msg_r or "阶段重置完成，流程将自动重新推进") + else: + reentry_msg = "\n⚠️ 定向重做执行失败：%s（已抛人工待办）" % (msg_r or "未知错误") + _is_rev = int(plan.get("version", 1) or 1) > 1 or (plan.get("reentry_stage") or "").strip() + _base = ("流程修订已确认并启用，系统将按新计划派发任务。" if _is_rev + else "流程裁剪已确认并启用，系统将按裁剪后的流程派发任务。") + return True, _base + reentry_msg async def reject_flow_plan(plan_id, comment, operator_id): @@ -736,6 +759,423 @@ async def create_sub_tasks(project_id="", tasks=None, parent_task_id="", task_id return True, str(result) + + +async def _affected_stage_keys(stages_snap, stage_key): + """注入阶段 + 其依赖闭包下游阶段 key 集合（按快照 deps 反向传播）。""" + keys = {s.get("key") for s in stages_snap if isinstance(s, dict)} + if stage_key not in keys: + return set() + affected = {stage_key} + changed = True + while changed: + changed = False + for s in stages_snap: + if not isinstance(s, dict): + continue + k = s.get("key") + if k in affected: + continue + if any(d in affected for d in (s.get("deps") or [])): + affected.add(k) + changed = True + return affected + + +async def _render_reentry_desc(pipeline_id, project_id, stage_key, affected, stages_snap, sor): + """重做影响说明：产线钩子优先，无钩子给通用文案。""" + label_map = {(s.get("key") or ""): (s.get("label") or s.get("key") or "") + for s in stages_snap if isinstance(s, dict)} + try: + from pipeline_core import get_stage_reentry_describe + fn = get_stage_reentry_describe(pipeline_id) + if fn: + txt = await fn(sor, project_id, stage_key, sorted(affected)) + if txt: + return txt + except Exception as e: + logger.warning("stage_reentry_describe hook failed: %s", str(e)[:160]) + lines = ["从阶段「%s」起重做，将影响以下阶段（含其下游）：" % label_map.get(stage_key, stage_key), ""] + for k in stages_snap: + if isinstance(k, dict) and k.get("key") in affected: + lines.append("- %s" % (k.get("label") or k.get("key"))) + return "\n".join(lines) + + +async def revise_flow_plan(project_id, user_requirements="", reentry_stage="", + trim_keys=None, propose_note="", reentry_detect_note="", + pipeline_id="", who=None, agent_id=None): + """流程修订提案：对已 confirmed 计划提出修订版（改裁剪 和/或 定向重做）。 + + 与 propose_flow_plan 的分工： + - propose_flow_plan：项目未开跑时的首次裁剪提案（confirmed 计划存在时拒绝）； + - revise_flow_plan：已有 confirmed 计划后的修订提案（无 confirmed 计划时拒绝， + 让调用方走 propose）。 + + reentry_stage 空 = 仅调整裁剪；非空 = 定向重做（确认后产线钩子重置该阶段及 + 依赖下游的产出/状态，流转机制自动从该阶段重新推进）。 + + 校验（代码硬校验，不靠 LLM 自觉）： + 1. 存在 confirmed 计划（修订对象）； + 2. 无 pending_confirm 计划在途（一次只允许一个修订在途；旧 pending 自动 superseded + 沿用 propose 的版本语义）； + 3. trim_keys 合法性同 propose（未知阶段/不可裁/依赖断裂拒绝）； + 4. reentry_stage 必须是声明阶段且在新计划中保留（enabled）——被裁阶段无产出可重做； + 5. 产线未注册 stage_reentry 钩子且 reentry_stage 非空 → 拒绝（诚实降级：该产线 + 不支持定向重做，说明原因，不静默忽略用户要求）。 + """ + if not project_id: + return False, "缺少 project_id（请先切换到项目）" + user_requirements = (user_requirements or "").strip() + if not user_requirements: + return False, "缺少 user_requirements（用户的修订要求原文，必填——进确认待办）" + reentry_stage = (reentry_stage or "").strip() + 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, "项目未关联产线，无法修订流程" + pname = proj.get("name") or project_id + + old_plan = await _latest_plan_rec(sor, project_id) + if not old_plan: + return False, "项目尚无流程计划——首次裁剪请用 propose_flow_plan" + flow_key = (old_plan.get("base_flow_key") or "").strip() + if old_plan.get("status") == S_PENDING: + return False, ("已有修订版待用户确认（plan %s v%s）——等用户处理完再提，" + "或让用户驳回后按意见重提" % ( + (old_plan.get("id") or "")[:8], old_plan.get("version"))) + if old_plan.get("status") != S_CONFIRMED: + # rejected/superseded/draft：走 propose 重提（版本语义不变） + return False, ("最新计划状态为 %s（非已确认）——修订仅针对已生效计划，" + "请用 propose_flow_plan 重新提案" % old_plan.get("status")) + + stage_map, stages = _load_stages(pid, flow_key) + if not stages: + return False, "产线「%s」未声明可裁剪流程，不支持修订" % pid + + # 校验3：trim 合法性（与 propose 同规则） + unknown = [k for k in trim_keys if k not in stage_map] + if unknown: + return False, "未知阶段 key：%s。本产线阶段见 show_flow_template。" % "、".join(unknown) + 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 + 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: + broken.append("%s 依赖被裁的 %s" % (s.label or s.key, + stage_map[d].label or d)) + if broken: + return False, "裁剪导致依赖断裂：%s。请保留上游阶段或一并裁剪下游。" % "；".join(broken) + + # 校验4/5：reentry 合法性 + reentry_note = (reentry_detect_note or "").strip() + if reentry_stage: + if reentry_stage not in stage_map: + return False, ("未知重做阶段 key：%s。可注入阶段见 show_flow_template。" + % reentry_stage) + if reentry_stage in trim_set: + return False, ("重做阶段「%s」同时被裁剪，矛盾——被裁阶段不会执行，" + "无法从它重做。请保留该阶段。" + % (stage_map[reentry_stage].label or reentry_stage)) + try: + from pipeline_core import get_stage_reentry_handler + has_hook = get_stage_reentry_handler(pid) is not None + except Exception: + has_hook = False + if not has_hook: + return False, ("产线「%s」未注册阶段重做能力（stage_reentry 钩子），" + "暂不支持定向重做；可联系平台扩展该产线。" % pid) + + warns = [stage_map[k].warn_note for k in trim_keys + if (stage_map[k].trim or "") == "warn" and stage_map[k].warn_note] + + 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) + + # 版本号 +1；旧 pending/draft 置 superseded（同 propose 语义） + 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, + "base_flow_key": flow_key, + "version": version, "status": S_PENDING, + "stages": json.dumps(stages_snap, ensure_ascii=False), + "user_requirements": user_requirements[:8000], + "propose_note": (propose_note or "")[:8000], + "reentry_stage": reentry_stage, + "reentry_detect_note": reentry_note[:8000], + "created_by": who or "agent.main_agent", + }) + await sor.sqlExe("COMMIT", {}) + + # 确认待办正文：裁剪对照表 + 重做影响预告（破坏性动作必须在确认前可见） + desc = _render_confirm_md(pname, stages_snap, trim_keys, warns, + user_requirements, propose_note, version, + flow_key=flow_key, pipeline_id=pid) + if reentry_stage: + affected = await _affected_stage_keys(stages_snap, reentry_stage) + rdesc = await _render_reentry_desc(pid, project_id, reentry_stage, + affected, stages_snap, sor) + label = stage_map[reentry_stage].label or reentry_stage + reentry_md = ["", "### 🔁 定向重做（从「%s」阶段起）" % label, ""] + if reentry_note: + reentry_md += ["**注入点判定依据：**", "", reentry_note.strip(), ""] + reentry_md += ["**确认后系统将执行：**", "", rdesc.strip(), "", + "> ⚠️ 重做会重置上述阶段的已有产出（见影响说明），流程从该阶段自动重新推进。"] + desc = desc + "\n".join(reentry_md) + 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") + 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, "revise_propose", + to_state=S_PENDING, who=who, agent_id=agent_id, + detail="version=%d reentry=%s trim=%s" % ( + version, reentry_stage or "(无)", + ",".join(trim_keys) or "(无裁剪)"), + sor=sor) + await sor.sqlExe("COMMIT", {}) + logger.info("flow_plan revise: project=%s plan=%s v=%d reentry=%s trim=%s", + project_id, plan_id, version, reentry_stage, 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 reentry_stage: + msg += "\n定向重做注入点：%s（%s）" % ( + stage_map[reentry_stage].label or reentry_stage, reentry_stage) + if warns: + msg += "\n⚠️ 风险提示（请在待办里向用户说明）：\n" + "\n".join("- " + w for w in warns) + msg += ("\n修订确认前旧计划持续生效；用户需在「流程修订确认」待办里确认后修订才生效" + "（重做重置在确认时执行）；agent 不能代替确认。") + return True, msg + + +async def detect_reentry_stage(project_id, user_comment="", pipeline_id=""): + """LLM 判定定向重做注入点（用户只给意见时，语义判断归 LLM，2026-09-17 用户裁定）。 + + 返回 (stage_key, detect_note)： + - 判定成功 → (阶段key, 判定依据文本：置信度+evidence，进确认待办给用户核对)； + - 产线未声明流程/LLM 失败/无法归类 → ("", 错误说明)：调用方不得默认选阶段硬跑， + 应把失败原因如实告知用户（禁词匹配兜底、禁静默降级——detect_flow_template 同款纪律）。 + 错判由 flow_plan_confirm 确认门兜住：用户在修订确认待办里看到注入点+判据， + 驳回附意见重提。 + """ + if not project_id: + return "", "缺少 project_id" + user_comment = (user_comment or "").strip() + if not user_comment: + return "", "缺少用户意见原文（user_comment）" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + pid, proj = await _resolve_pipeline(sor, project_id, pipeline_id) + if not pid: + return "", "项目未关联产线" + old_plan = await _latest_plan_rec(sor, project_id) + flow_key = (old_plan.get("base_flow_key") or "").strip() if old_plan else "" + stage_map, stages = _load_stages(pid, flow_key) + if not stages: + return "", "产线「%s」未声明流程阶段，无法判定注入点" % pid + # 进度摘要：产线钩子优先，无钩子给通用任务状态摘要 + summary = "" + try: + from pipeline_core import get_progress_summary_handler + fn = get_progress_summary_handler(pid) + if fn: + summary = await fn(sor, project_id) or "" + except Exception as e: + logger.warning("progress_summary hook failed: %s", str(e)[:160]) + if not summary: + summary = await _generic_progress_summary(sor, project_id) + + stage_lines = [] + for i, s in enumerate(stages, 1): + stage_lines.append("%d. key=%s 名称=%s\n 作用：%s" % ( + i, s.key, s.label or s.key, (s.description or "（未写说明）")[:200])) + prompt = ( + "你是产线流程调度器。用户要求对进行中的项目做修改/重做，请判定应从哪个流程阶段" + "注入重做（该阶段及其下游会按用户意见重新执行）。\n\n" + "【流程阶段（按执行顺序）】\n%s\n\n" + "【项目当前进度】\n%s\n\n" + "【用户意见原文】\n%s\n\n" + "判定原则：选择「能覆盖用户意见涉及的最小上游阶段」——从更早阶段重做代价更大，" + "从更晚阶段重做可能漏改；用户意见明确指向某阶段产物时直接选该阶段。\n" + "严格只输出一个 JSON 对象：\n" + "{\"stage_key\": \"候选之一的key\", \"confidence\": \"high|medium|low\",\n" + " \"evidence\": \"判定依据：用户意见哪句话+进度哪个事实支持从该阶段注入（≤120字）\"}\n" + "无法判定时 stage_key 输出空字符串并在 evidence 说明缺什么信息。" + % ("\n".join(stage_lines), (summary or "（无进度信息）")[:4000], user_comment[:3000]) + ) + try: + from pipeline_service.llm_bridge import llm_call + raw = await llm_call(prompt, purpose="utility", + org_id=(proj.get("org_id") or "") if proj else "", + project_id=project_id, timeout=45) + except Exception as e: + return "", "注入点判定 LLM 调用失败：%s" % str(e)[:200] + txt = (raw or "").strip() + if txt.startswith("```"): + parts = txt.split("\n", 1) + txt = parts[1] if len(parts) > 1 else txt + if "```" in txt: + txt = txt.rsplit("```", 1)[0] + txt = txt.strip() + try: + if not txt.startswith("{") and "{" in txt: + txt = txt[txt.find("{"):txt.rfind("}") + 1] + data = json.loads(txt) + except (json.JSONDecodeError, ValueError): + return "", "注入点判定结果无法解析：%s" % (raw or "")[:200] + sk = str(data.get("stage_key") or "").strip() + conf = str(data.get("confidence") or "").strip() + ev = str(data.get("evidence") or "").strip() + valid = {s.key for s in stages} + if sk and sk not in valid: + return "", "判定返回未知阶段 key：%s（候选：%s）" % (sk, "、".join(sorted(valid))) + if not sk: + return "", "无法判定注入点：%s" % (ev or "（LLM 未说明原因）") + label = stage_map[sk].label or sk + note = "判定注入点：「%s」（%s，置信度 %s）。\n依据：%s" % ( + label, sk, conf or "?", ev or "（未提供依据）") + if conf == "low": + note += "\n⚠️ 判定置信度低，请在确认待办里重点核对——不对就驳回并说明应从哪个阶段重做。" + return sk, note + + +async def _generic_progress_summary(sor, project_id): + """通用进度摘要（产线无 progress_summary 钩子时的兜底）：任务状态 + 阶段产出计数。""" + try: + recs = await sor.sqlExe( + "SELECT role, state, COUNT(*) AS c FROM pipeline_tasks WHERE tenant_id=${p}$ " + "AND pipeline_id='role_task' GROUP BY role, state ORDER BY role", + {"p": project_id}) + await sor.sqlExe("COMMIT", {}) + lines = ["角色任务状态分布："] + for r in (recs or []): + lines.append("- %s %s: %s" % (getattr(r, "role", "?"), + getattr(r, "state", "?"), getattr(r, "c", 0))) + ht = await sor.sqlExe( + "SELECT task_type, title, status FROM pipeline_human_tasks WHERE project_id=${p}$ " + "AND status='pending' ORDER BY created_at DESC LIMIT 10", {"p": project_id}) + await sor.sqlExe("COMMIT", {}) + if ht: + lines.append("在办人类待办：") + for r in ht: + lines.append("- [%s] %s" % (getattr(r, "task_type", "?"), + str(getattr(r, "title", ""))[:80])) + proj = await sor.sqlExe( + "SELECT status FROM sd_projects WHERE id=${p}$", {"p": project_id}) + await sor.sqlExe("COMMIT", {}) + if proj: + lines.append("项目状态：%s" % getattr(proj[0], "status", "?")) + return "\n".join(lines) + except Exception as e: + logger.warning("generic progress summary failed: %s", str(e)[:160]) + return "" + + +async def _run_stage_reentry(sor, plan, operator_id): + """确认修订版时执行定向重做（产线钩子）。返回 (ok, msg)。 + + 钩子失败不回滚计划确认（计划换版是用户裁决，已生效）——诚实上报 + 抛人工待办， + 人工可按影响说明手工重置或让 agent 修复（不静默吞错）。 + """ + project_id = plan.get("project_id", "") + pid = plan.get("pipeline_id", "") + stage_key = (plan.get("reentry_stage") or "").strip() + if not stage_key: + return True, "" + try: + stages_snap = json.loads(plan.get("stages") or "[]") + except (json.JSONDecodeError, ValueError): + stages_snap = [] + affected = await _affected_stage_keys(stages_snap, stage_key) + try: + from pipeline_core import get_stage_reentry_handler + fn = get_stage_reentry_handler(pid) + except Exception: + fn = None + if not fn: + # 提案时已校验过钩子存在；确认时消失 = 部署窗口异常，诚实上报 + msg = "产线「%s」阶段重做钩子不可用，未执行重置——请人工处理" % pid + 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") + await sor.sqlExe("COMMIT", {}) + return False, msg + try: + ok, msg = await fn(sor, project_id, stage_key, sorted(affected), + plan.get("user_requirements") or "", operator_id) + await sor.sqlExe("COMMIT", {}) + except Exception as e: + logger.exception("stage_reentry hook crashed: project=%s stage=%s", + project_id, stage_key) + msg = "阶段重做执行异常：%s" % str(e)[:300] + ok = False + if not ok: + await create_human_task( + project_id, "流程修订已确认但阶段重做执行失败：%s" % stage_key, + (msg or "钩子返回失败") + "\n\n修订计划已生效（v%s），但阶段重置未完成，" + "流程可能停在旧产出上。请人工处置后项目自动恢复推进。" + % plan.get("version"), + task_type="general", assignee_role="owner.superuser") + await sor.sqlExe("COMMIT", {}) + await record_audit(project_id, "pipeline_flow_plans", plan.get("id", ""), + "stage_reentry", to_state=stage_key, who=operator_id, + detail=("ok" if ok else "failed") + ": " + (msg or "")[:400], + sor=sor) + await sor.sqlExe("COMMIT", {}) + return ok, msg + + # ══════════════════════ 会话工具（FLOW_PLAN_TOOLS / HANDLERS，shared_ability 聚合） ══════════════════════ try: @@ -761,6 +1201,29 @@ if ToolDefinition is not None: "propose_note": "你对裁剪方案的说明/风险提示(可选)", "flow_key": "多流程产线指定流程模板key(可选,来自show_flow_template;单流程产线留空)"}, required=["user_requirements"], category="project"), + ToolDefinition( + name="revise_flow_plan", + description=("对已确认生效的项目流程提出修订（用户有权随时调整流程）：可改裁剪(trim_keys)，" + "和/或要求从任一阶段按用户意见定向重做(reentry_stage)。生成修订草案(v+1)并发" + "「流程修订确认」待办——确认前旧计划持续生效，用户确认时才换版并执行阶段重置" + "（重置影响会在待办里预告）。用户只给意见没指明阶段时：先调 detect_reentry_stage " + "判定注入点（LLM 语义判定），把返回的 stage_key 和 detect_note 原样传入本工具" + "（note 进待办给用户核对，判定失败如实转告用户，禁止自己猜阶段）。" + "修订合法性由系统硬校验；你没有确认权，禁止声称已确认。"), + parameters={"user_requirements": "用户修订要求原文(必填，写进确认待办)", + "reentry_stage": "定向重做注入阶段key(可选；空=仅调整裁剪不重做；来自show_flow_template或detect_reentry_stage)", + "trim_keys": "要裁剪的阶段key数组(可选；修订后的完整裁剪集，沿用现状传空)", + "propose_note": "你对修订方案的说明/风险提示(可选)", + "reentry_detect_note": "注入点判定依据(detect_reentry_stage返回的note原样传入，进待办给用户核对)"}, + required=["user_requirements"], category="project"), + ToolDefinition( + name="detect_reentry_stage", + description=("用户只给修改意见、没指明从哪个阶段重做时调用：LLM按意见+流程阶段声明+项目进度" + "判定定向重做注入点（语义判断归LLM，禁词匹配）。返回stage_key+判定依据note——" + "原样传给revise_flow_plan（note进确认待办由用户核对，错判由确认门兜住）。" + "判定失败(空key)时如实把失败原因转告用户，禁止自己猜阶段硬跑。"), + parameters={"user_comment": "用户意见原文(必填)"}, + required=["user_comment"], category="project"), ToolDefinition( name="get_flow_plan", description="查看当前项目流程裁剪计划的状态（草案/待确认/已确认/已驳回+驳回意见）", @@ -820,6 +1283,10 @@ async def _h_get_flow_plan(sor, params, ctx): 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("reentry_stage") or "").strip(): + lines.append("定向重做注入点：%s" % plan.get("reentry_stage")) + if plan.get("reentry_detect_note"): + lines.append("判定依据：%s" % str(plan.get("reentry_detect_note"))[:300]) 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）。") @@ -828,8 +1295,43 @@ async def _h_get_flow_plan(sor, params, ctx): return "\n".join(lines) +async def _h_revise_flow_plan(sor, params, ctx): + project_id = ctx.get("project_id") or "" + if not project_id: + return "ERROR: 请先切换到项目（用 switch_project 或在项目列表选择）" + ok, msg = await revise_flow_plan( + project_id, + user_requirements=params.get("user_requirements", "") or "", + reentry_stage=params.get("reentry_stage", "") or "", + trim_keys=params.get("trim_keys"), + propose_note=params.get("propose_note", "") or "", + reentry_detect_note=params.get("reentry_detect_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_detect_reentry_stage(sor, params, ctx): + project_id = ctx.get("project_id") or "" + if not project_id: + return "ERROR: 请先切换到项目" + comment = (params.get("user_comment") or "").strip() + if not comment: + return "ERROR: 缺少用户意见原文（user_comment）" + stage_key, note = await detect_reentry_stage( + project_id, comment, pipeline_id=ctx.get("pipeline_id", "") or "") + if not stage_key: + return "ERROR: 注入点判定失败——%s\n请把失败原因如实转告用户（禁止自己猜阶段）。" % note + return ("OK: stage_key=%s\n%s\n\n下一步：调 revise_flow_plan(reentry_stage=%s, " + "reentry_detect_note=上面的判定依据原文, user_requirements=用户意见原文)。" + % (stage_key, note, stage_key)) + + FLOW_PLAN_HANDLERS = { "show_flow_template": _h_show_flow_template, "propose_flow_plan": _h_propose_flow_plan, + "revise_flow_plan": _h_revise_flow_plan, + "detect_reentry_stage": _h_detect_reentry_stage, "get_flow_plan": _h_get_flow_plan, }