# ============================================================ # Agent 自动编排 v1(moduleId: core-plan-orchestration, 可重生 ✅) # plan.md §3.1 + 矩阵 110 行剩余项:plan 节点驱动 workflow 分支 # P2 动作出确认卡时创建 L2 Plan 节点(DRAFT,记录 action/params/confirmId); # 批准/驳回时流转状态(APPROVED/FAILED),Plan 历史与门禁决策对齐且可审计。 # 编排尽力而为:任何失败仅记日志,绝不阻断门禁主流程。 # ============================================================ from __future__ import annotations import json import logging from typing import Any logger = logging.getLogger("aps.plan.orchestration") def stage_plan_node( *, session_id: str, action: str, params: dict[str, Any], confirm_id: str, power: str = "P2", actor: str = "planner", ) -> str | None: """P2 动作出卡时创建 L2 Plan 节点(DRAFT)。 Returns: plan_id(编排成功)或 None(尽力而为失败)。 """ try: from server.agent_core.plan_runtime import get_plan_store, canonical_inputs_hash plan_store = get_plan_store() inputs = {"action": action, "confirmId": confirm_id, "power": power} # 自动补建父链:L0 意图(按 action 幂等)→ L1 策略 → L2 决策节点 l0_id = _ensure_l0_node(plan_store, action) l1_id = _ensure_l1_node(plan_store, l0_id, action) node = plan_store.create( layer="L2", parent_id=l1_id, inputs=inputs, payload={ "kind": "gate-decision", "action": action, "params": _stable_params(params), "confirmId": confirm_id, "power": power, "paramsHash": canonical_inputs_hash(params), }, status="DRAFT", created_by="USER", ) logger.info("plan orchestration staged %s plan=%s confirm=%s", action, node.planId, confirm_id) return node.planId except Exception: logger.exception("plan orchestration stage failed for %s", action) return None def _ensure_l0_node(plan_store: Any, action: str) -> str: """幂等创建/复用 L0 意图节点(按 action 键控)。""" from server.agent_core.plan_runtime import PlanNotFoundError l0_id = f"intent-{_slug(action)}" try: plan_store.latest(l0_id) return l0_id except PlanNotFoundError: node = plan_store.create( plan_id=l0_id, layer="L0", parent_id=None, inputs={"action": action}, payload={"kind": "intent", "action": action}, status="APPROVED", created_by="USER", ) return node.planId def _ensure_l1_node(plan_store: Any, l0_id: str, action: str) -> str: """幂等创建/复用 L1 策略节点。""" from server.agent_core.plan_runtime import PlanNotFoundError l1_id = f"strategy-{_slug(action)}" try: plan_store.latest(l1_id) return l1_id except PlanNotFoundError: node = plan_store.create( plan_id=l1_id, layer="L1", parent_id=l0_id, inputs={"intent": l0_id}, payload={"kind": "strategy", "action": action}, status="APPROVED", created_by="USER", ) return node.planId def _slug(action: str) -> str: import re return re.sub(r"[^A-Za-z0-9_]", "-", action)[:60] or "gate" def decide_plan_node( *, confirm_id: str, approve: bool, note: str | None = None, actor: str = "planner", ) -> str | None: """批准/驳回时流转对应 L2 Plan 节点状态(APPROVED/FAILED)。 按 confirmId 匹配最近 staged 节点(payload.confirmId);找不到则静默跳过。 Returns: plan_id(已流转)或 None。 """ try: from server.agent_core.plan_runtime import get_plan_store plan_store = get_plan_store() # 查找 payload.confirmId == confirm_id 的 L2 节点(仅 DRAFT) for plan_id in _list_plan_ids(plan_store): try: node = plan_store.latest(plan_id) except Exception: continue if node.layer != "L2" or node.status != "DRAFT": continue if (node.payload or {}).get("confirmId") != confirm_id: continue new_status = "APPROVED" if approve else "FAILED" plan_store.transition_status( plan_id, new_status=new_status, created_by="USER", note=note or ("批准" if approve else "驳回"), ) logger.info("plan orchestration %s plan=%s confirm=%s", new_status, plan_id, confirm_id) return plan_id return None except Exception: logger.exception("plan orchestration decide failed confirm=%s", confirm_id) return None def _stable_params(params: dict[str, Any]) -> dict[str, Any]: """参数稳定投影(去除非 JSON 值,排序键),供 Plan payload 审计。""" try: return json.loads(json.dumps(params, ensure_ascii=False, sort_keys=True, default=str)) except Exception: return {"_unserializable": True} def _list_plan_ids(plan_store: Any) -> list[str]: """读取所有 plan id(PlanStore 内部 _plans;尽力而为)。""" try: return list(getattr(plan_store, "_plans", {}) or {}) except Exception: return []