feat(flow_plan): 流程修订/定向重做机制(2026-09-17用户裁定:用户有权调整流程,并要求从流程任何一步按意见重做,或只给意见由agent判注入点)——revise_flow_plan:已confirmed计划提修订版(v+1 pending_confirm,确认前旧计划持续生效,确认时CAS换版旧版superseded+回调产线stage_reentry钩子);detect_reentry_stage:LLM按意见+阶段声明+产线进度摘要判注入点(对齐detect_flow_template判流先例,evidence进确认待办用户核对,错判确认门兜住,判不出诚实报错禁默认);_affected_stage_keys:注入阶段+deps依赖闭包下游;钩子失败不回滚确认,诚实上报+抛人工待办;新会话工具revise_flow_plan/detect_reentry_stage进FLOW_PLAN_TOOLS(三产线零改动继承);propose已确认时错误指引改指revise;models加reentry_stage/reentry_detect_note列
This commit is contained in:
ymq
parent
12f8384d49
commit
c8b44fe423
@ -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 @@
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
@ -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,
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user