1354 lines
70 KiB
Python
1354 lines
70 KiB
Python
# -*- 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,
|
||
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")
|
||
|
||
# 计划状态
|
||
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, flow_key=""):
|
||
"""从能力包取产线声明的阶段，返回 (stage_map, ordered_list)。
|
||
|
||
flow_key 非空时取多流程模板（FlowTemplate）对应阶段；空 = 单流程 flow_stages。
|
||
"""
|
||
from pipeline_core import get_flow_stages
|
||
stages = get_flow_stages(pipeline_id, flow_key=flow_key or "") 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="", flow_key=""):
|
||
"""返回 (pid, [stage_dict,...])：产线预制流程阶段（含可裁性），供 show_flow_template 渲染。
|
||
|
||
flow_key 空时：项目已有确认/待确认计划则取计划的 base_flow_key（展示实际将走的流程）；
|
||
多流程产线无计划时展示全部模板概览由调用方处理。
|
||
"""
|
||
db, dbname = _get_db()
|
||
async with db.sqlorContext(dbname) as sor:
|
||
pid, _proj = await _resolve_pipeline(sor, project_id, pipeline_id)
|
||
fk = (flow_key or "").strip()
|
||
if not fk and project_id:
|
||
plan = await _latest_plan_rec(sor, project_id)
|
||
fk = (plan.get("base_flow_key") or "").strip()
|
||
from pipeline_core import flow_stage_to_dict
|
||
_map, stages = _load_stages(pid, fk)
|
||
return pid, [flow_stage_to_dict(s) for s in stages]
|
||
|
||
|
||
async def list_flow_templates(pipeline_id=""):
|
||
"""列出产线的多流程模板 [(key, name, description, n_stages)]；单流程产线返回 []。"""
|
||
from pipeline_core import get_flow_templates
|
||
return [(t.key, t.name or t.key, t.description or "", len(t.stages or []))
|
||
for t in get_flow_templates(pipeline_id)]
|
||
|
||
|
||
# ══════════════════════ 文件自动判流（多流程机制，2026-09-14 批2） ══════════════════════
|
||
|
||
async def detect_flow_template(pipeline_id, file_name="", file_text="",
|
||
project_id="", org_id=""):
|
||
"""LLM 按各模板 detect_criteria 给导入文件判流（语义判断归 LLM，2026-09-11 用户铁律）。
|
||
|
||
返回 (flow_key, detect_note)：
|
||
- 单流程产线（模板 <2）→ ("", "")，调用方走现状（不判流不确认）；
|
||
- 判流成功 → (模板key, 识别结论文本：置信度+判据摘录，进确认待办给用户核对)；
|
||
- LLM 失败/无法解析 → ("", 错误说明)：调用方**不得**默认选流程硬跑，
|
||
应发人工介入待办（判流是流程选择的语义判断，禁词匹配兜底、禁静默降级）。
|
||
错判由 flow_plan_confirm 确认门兜住：用户在待办里看到识别结论，驳回附意见重提。
|
||
"""
|
||
from pipeline_core import get_flow_templates
|
||
tpls = get_flow_templates(pipeline_id)
|
||
if len(tpls) < 2:
|
||
return "", ""
|
||
snippet = (file_text or "").strip()[:6000]
|
||
if not snippet and not file_name:
|
||
return "", "判流失败：文件内容为空"
|
||
tpl_lines = []
|
||
for t in tpls:
|
||
tpl_lines.append("- key=%s 名称=%s\n 适用判据：%s" % (
|
||
t.key, t.name or t.key, (t.detect_criteria or t.description or "（未写判据）")))
|
||
prompt = (
|
||
"你是产线流程路由器。根据输入文件的特征，判断它应该走哪条预制流程。\n\n"
|
||
"【候选流程】\n%s\n\n"
|
||
"【文件名】%s\n\n【文件正文（截断）】\n%s\n\n"
|
||
"严格只输出一个 JSON 对象：\n"
|
||
"{\"flow_key\": \"候选之一的key\", \"confidence\": \"high|medium|low\",\n"
|
||
" \"evidence\": \"判据摘录：文件里哪些特征（原文引用≤3处）支持这个判断\"}\n"
|
||
"无法归入任何候选流程时 flow_key 输出空字符串并在 evidence 说明原因。"
|
||
% ("\n".join(tpl_lines), file_name or "（无）", snippet or "（无正文）")
|
||
)
|
||
try:
|
||
from pipeline_service.llm_bridge import llm_call
|
||
raw = await llm_call(prompt, purpose="utility", org_id=org_id or "",
|
||
project_id=project_id or "", 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]
|
||
fk = str(data.get("flow_key") or "").strip()
|
||
conf = str(data.get("confidence") or "").strip()
|
||
ev = str(data.get("evidence") or "").strip()
|
||
valid = {t.key for t in tpls}
|
||
if fk and fk not in valid:
|
||
return "", "判流返回未知流程 key：%s（候选：%s）" % (fk, "、".join(sorted(valid)))
|
||
if not fk:
|
||
return "", "判流无法归类：%s" % (ev or "（LLM 未说明原因）")
|
||
note = "识别为「%s」流程（置信度 %s）。\n判据：%s" % (
|
||
fk, conf or "?", ev or "（未提供判据摘录）")
|
||
if conf == "low":
|
||
note += "\n⚠️ 识别置信度低，请重点核对——不对就驳回并说明应走哪条流程。"
|
||
return fk, note
|
||
|
||
|
||
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="", flow_key="",
|
||
detect_note="", who=None, agent_id=None):
|
||
"""按用户要求生成裁剪草案，落库 pending_confirm 并发确认待办。
|
||
|
||
flow_key（多流程机制，2026-09-14 批2）：指定基准流程模板 key。
|
||
- 空 = 单流程产线（用 flow_stages 声明），或多流程产线沿用已有计划的模板；
|
||
- 非空 = 按该模板的阶段链出草案（base_flow_key 落库，流转机制按它走分支）。
|
||
detect_note：判流依据说明（LLM 自动判流时写识别结论+判据摘录，进确认待办给用户核对）。
|
||
|
||
代码硬校验（任一不过即拒绝，返回可行动原因，绝不静默降级）：
|
||
1. 项目/产线存在，且产线声明了 flow_stages（或 flow_key 对应模板）；
|
||
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)
|
||
flow_key = (flow_key or "").strip()
|
||
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, "项目未关联产线，无法裁剪流程"
|
||
# flow_key 校验：多流程产线必须是已声明的模板 key
|
||
if flow_key:
|
||
from pipeline_core import get_flow_template as _get_ftpl
|
||
if not _get_ftpl(pid, flow_key):
|
||
tpls = await list_flow_templates(pid)
|
||
avail = "、".join(k for k, _n, _d, _c in tpls) or "（该产线未声明多流程模板）"
|
||
return False, ("未知流程模板 key：%s。可用流程：%s" % (flow_key, avail))
|
||
else:
|
||
# 未指定：已有计划则沿用其 base_flow_key（驳回重提场景）
|
||
old_plan = await _latest_plan_rec(sor, project_id)
|
||
flow_key = (old_plan.get("base_flow_key") or "").strip()
|
||
stage_map, stages = _load_stages(pid, flow_key)
|
||
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）。运行中调整流程请改用 "
|
||
"revise_flow_plan 提出修订（可改裁剪、可从任一阶段按用户意见"
|
||
"定向重做），修订经用户确认后生效。"
|
||
% 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,
|
||
"base_flow_key": flow_key,
|
||
"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,
|
||
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=_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() "
|
||
"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 flow_key=%s trim=%s" % (
|
||
version, flow_key or "(单流程)",
|
||
",".join(trim_keys) or "(无裁剪)"),
|
||
sor=sor)
|
||
await sor.sqlExe("COMMIT", {})
|
||
logger.info("flow_plan propose: project=%s plan=%s v=%d flow_key=%s trim=%s",
|
||
project_id, plan_id, version, flow_key, 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 flow_key:
|
||
msg = "基准流程：%s\n%s" % (flow_key, msg)
|
||
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,
|
||
flow_key="", pipeline_id="", detect_note=""):
|
||
"""确认待办正文（markdown，todo_detail 直接渲染）。"""
|
||
lines = ["## 项目「%s」流程裁剪方案（v%d）" % (pname, version), ""]
|
||
if flow_key:
|
||
# 多流程产线：展示判定的流程模板名 + 判流依据（用户核对识别结论）
|
||
try:
|
||
from pipeline_core import get_flow_template as _gft
|
||
tpl = _gft(pipeline_id, flow_key) if pipeline_id else None
|
||
tname = (tpl.name if tpl else "") or flow_key
|
||
except Exception:
|
||
tname = flow_key
|
||
lines += ["**将走的流程：** %s（`%s`）" % (tname, flow_key), ""]
|
||
if detect_note:
|
||
lines += ["### 文件识别结论（自动判流）", "", detect_note.strip(), "",
|
||
"> 识别不对？点「驳回」并说明应走哪条流程，助手会修订重提。", ""]
|
||
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 += ["", "---", "",
|
||
"点击「确认并启用」后流程按上表生效；生效后可随时再提修订（revise_flow_plan，",
|
||
"含从任一阶段定向重做）；点击「驳回」并填写意见，助手会按你的意见修订后重新提案。"]
|
||
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", {})
|
||
# 修订版换版（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; 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)
|
||
# 定向重做（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):
|
||
"""用户驳回：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:
|
||
return await _latest_plan_rec(sor, project_id)
|
||
|
||
|
||
async def _latest_plan_rec(sor, project_id):
|
||
"""取项目最新一条计划（已开 sor 上下文版，内部复用）。"""
|
||
if not project_id:
|
||
return {}
|
||
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 {}
|
||
|
||
|
||
async def get_plan_flow_key(project_id):
|
||
"""项目已确认计划的 base_flow_key（'' = 无计划/单流程）。流转机制按它选流程分支。"""
|
||
plan = await get_active_plan(project_id)
|
||
return (plan.get("base_flow_key") or "").strip() if plan 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)
|
||
|
||
|
||
|
||
|
||
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)
|
||
_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=_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() "
|
||
"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
|
||
_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=_ar3, assignee_id=_ai3)
|
||
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:
|
||
_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"),
|
||
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,
|
||
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:
|
||
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": "你对裁剪方案的说明/风险提示(可选)",
|
||
"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="查看当前项目流程裁剪计划的状态（草案/待确认/已确认/已驳回+驳回意见）",
|
||
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 ""
|
||
tpls = await list_flow_templates(pid)
|
||
head = ""
|
||
if len(tpls) >= 2:
|
||
head = ("本产线是多流程产线，可选流程模板（propose_flow_plan 用 flow_key 指定）：\n"
|
||
+ "\n".join("- %s（%s）：%s，%d 阶段" % (k, n, d[:80], c)
|
||
for k, n, d, c in tpls) + "\n\n")
|
||
pid2, stages = await get_flow_template(project_id=project_id, pipeline_id=pid)
|
||
if not stages:
|
||
return head + ("当前产线（%s）未声明可裁剪流程，按标准全流程执行，无需裁剪。"
|
||
% (pid2 or "未知"))
|
||
return head + ("产线「%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 "",
|
||
flow_key=params.get("flow_key", "") 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("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）。")
|
||
if plan.get("status") == S_PENDING:
|
||
lines.append("→ 已发确认待办，等用户在「流程裁剪确认」待办里确认；你不能代替确认。")
|
||
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,
|
||
}
|