52 lines
2.0 KiB
Python
52 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
from .decision_schema import DecisionGateResult, ModeCard, SemanticDecision
|
|
|
|
|
|
class SemanticDecisionGatekeeper:
|
|
@staticmethod
|
|
def _next_plan_status(*, accepted: bool, decision: SemanticDecision) -> str:
|
|
if not accepted:
|
|
return "active"
|
|
if decision.mode_decision == "ready_confirm":
|
|
return "awaiting_execute_confirm"
|
|
if decision.mode_decision == "request_execute":
|
|
return "execution_pending"
|
|
if decision.mode_decision == "answer_only":
|
|
return "idle"
|
|
return "active"
|
|
|
|
@classmethod
|
|
def evaluate(
|
|
cls,
|
|
*,
|
|
mode_card: ModeCard,
|
|
decision: SemanticDecision,
|
|
) -> DecisionGateResult:
|
|
blocked_reasons: list[str] = []
|
|
allowed_actions: list[str] = []
|
|
allowed_set = {str(item).strip() for item in mode_card.allowed_actions if str(item or "").strip()}
|
|
forbidden_set = {str(item).strip() for item in mode_card.forbidden_actions if str(item or "").strip()}
|
|
|
|
for action in decision.requested_actions:
|
|
normalized_action = str(action or "").strip()
|
|
if not normalized_action:
|
|
continue
|
|
if normalized_action in forbidden_set:
|
|
blocked_reasons.append(f"forbidden_action:{normalized_action}")
|
|
continue
|
|
if allowed_set and normalized_action not in allowed_set:
|
|
blocked_reasons.append(f"unapproved_action:{normalized_action}")
|
|
continue
|
|
allowed_actions.append(normalized_action)
|
|
|
|
accepted = len(blocked_reasons) == 0
|
|
return DecisionGateResult(
|
|
accepted=accepted,
|
|
blocked_reasons=blocked_reasons,
|
|
allowed_actions=allowed_actions,
|
|
next_plan_status=cls._next_plan_status(accepted=accepted, decision=decision),
|
|
needs_confirmation=bool(decision.requires_confirmation),
|
|
mode_state_after=decision.mode_decision,
|
|
)
|