"""Pure workflow transition evaluation; state I/O remains outside this module.""" from __future__ import annotations from typing import Any, Callable def structural_next( current: str, plan_stages: list[str], transitions: list[dict[str, Any]], blocked_from: str | None = None, ) -> list[str]: destinations: set[str] = set() if current in plan_stages: index = plan_stages.index(current) if index + 1 < len(plan_stages): destinations.add(plan_stages[index + 1]) for transition in transitions: source = transition.get("from") destination = transition.get("to") if source == current: if destination == "": if blocked_from: destinations.add(blocked_from) elif destination: destinations.add(destination) elif source == "*" and current not in ("blocked", "closed") and destination: destinations.add(destination) destinations.discard("") destinations.discard("*") return sorted(destinations) def actor_allowed( transition: dict[str, Any], actor: str | None, role_registry: dict[str, Any], ) -> tuple[bool, str | None]: allowed = transition.get("allowed-by") or [] if not str(actor or "").strip(): return False, "전이 주체(--actor) 미지정 — 누가 전이하는지 명시해야 한다(권한/감사)." if not allowed: return True, None normalized = str(actor).strip() if normalized not in role_registry: return False, f"actor '{normalized}'는 role registry에 없는 역할이다" if isinstance(allowed, dict): concrete = allowed.get("executor") or allowed.get("concrete-roles") or [] else: concrete = [value for value in allowed if isinstance(value, str) and not value.endswith("-role-agent")] if normalized in concrete: return True, None if not concrete: return False, f"allowed-by가 placeholder만 포함해 runtime 권한을 결정할 수 없다: {allowed}" return False, f"actor '{actor}' 는 transition executor {concrete} 에 없다(권한 없음)." def evaluate_transition( transition: dict[str, Any], *, current: str, destination: str, plan: str, plan_stages: set[str], blocked_from: str | None, facts: dict[str, Any], condition_evaluator: Callable[[Any, dict[str, Any]], tuple[bool, str | None]], actor: str | None = None, role_registry: dict[str, Any] | None = None, ) -> list[str]: universal = (destination == "blocked" or current == "blocked" or destination == current or transition.get("from") == "*") if plan_stages and not universal and destination not in plan_stages: return [f"plan '{plan}' 시퀀스에 없는 stage 전이 금지: {current} -> {destination} " f"(plan stages: {sorted(plan_stages)})"] if actor: permitted, reason = actor_allowed(transition, actor, role_registry or {}) if not permitted: return [reason or "transition actor 권한 없음"] if current == "blocked" and blocked_from and destination != blocked_from: return [f"blocked 재개 대상은 {blocked_from} 여야 합니다(요청: {destination})"] reasons: list[str] = [] for condition in transition.get("required-conditions") or []: passed, reason = condition_evaluator(condition, facts) if not passed: reasons.append(reason or f"조건 미충족: {condition}") for condition in transition.get("forbidden-if") or []: passed, _reason = condition_evaluator(condition, facts) if passed: reasons.append(f"금지조건 충족: {condition}") return reasons