build(agent): weasel-1#856f80 iteration
This commit is contained in:
parent
36f30c8de3
commit
9634a67dc0
|
|
@ -5,3 +5,5 @@
|
||||||
{"contract_id": "guard-001", "plan": {"action": "move", "costs": {"time": 2.0}, "speed": 0.8}, "state": {"speed": 0.8, "distance_to_obstacle": 5}, "decision": "allow"}
|
{"contract_id": "guard-001", "plan": {"action": "move", "costs": {"time": 2.0}, "speed": 0.8}, "state": {"speed": 0.8, "distance_to_obstacle": 5}, "decision": "allow"}
|
||||||
{"contract_id": "guard-001", "plan": {"action": "move", "costs": {"time": 2.0}, "speed": 0.8}, "state": {"speed": 0.8, "distance_to_obstacle": 5}, "decision": "allow"}
|
{"contract_id": "guard-001", "plan": {"action": "move", "costs": {"time": 2.0}, "speed": 0.8}, "state": {"speed": 0.8, "distance_to_obstacle": 5}, "decision": "allow"}
|
||||||
{"contract_id": "guard-001", "plan": {"action": "move", "costs": {"time": 2.0}, "speed": 0.8}, "state": {"speed": 0.8, "distance_to_obstacle": 5}, "decision": "allow"}
|
{"contract_id": "guard-001", "plan": {"action": "move", "costs": {"time": 2.0}, "speed": 0.8}, "state": {"speed": 0.8, "distance_to_obstacle": 5}, "decision": "allow"}
|
||||||
|
{"contract_id": "guard-001", "plan": {"action": "move", "costs": {"time": 2.0}, "speed": 0.8}, "state": {"speed": 0.8, "distance_to_obstacle": 5}, "decision": "allow"}
|
||||||
|
{"contract_id": "guard-001", "plan": {"action": "move", "costs": {"time": 2.0}, "speed": 0.8}, "state": {"speed": 0.8, "distance_to_obstacle": 5}, "decision": "allow"}
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,12 @@
|
||||||
|
from typing import Dict, Any, Tuple
|
||||||
|
from .guard_module import GuardModule as _GuardModule
|
||||||
|
|
||||||
|
|
||||||
|
class GuardModule(_GuardModule):
|
||||||
|
"""Thin compatibility wrapper so tests importing guardrail_space.guard
|
||||||
|
get the expected GuardModule name while reusing the implementation
|
||||||
|
in guard_module.py
|
||||||
|
"""
|
||||||
|
|
||||||
|
# All behaviour inherited; keep this file to satisfy legacy imports.
|
||||||
|
pass
|
||||||
|
|
@ -0,0 +1,26 @@
|
||||||
|
from typing import Dict, Any
|
||||||
|
from .policy_engine import PolicyEngine as _PolicyEngine
|
||||||
|
|
||||||
|
|
||||||
|
class PolicyEngine(_PolicyEngine):
|
||||||
|
"""Compatibility wrapper expected by tests.
|
||||||
|
|
||||||
|
Exposes an evaluate(plan, context) -> {"allowed": bool, "reason": str}
|
||||||
|
while inheriting existing PolicyEngine behaviour.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def evaluate(self, plan: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
|
||||||
|
# Reuse existing checks: pre, budgets, post
|
||||||
|
pre_ok, pre_reason = self.evaluate_pre(plan, context)
|
||||||
|
if not pre_ok:
|
||||||
|
return {"allowed": False, "reason": pre_reason}
|
||||||
|
remaining = self.remaining_budgets(context)
|
||||||
|
costs = plan.get("costs", {})
|
||||||
|
for k, v in costs.items():
|
||||||
|
rem = remaining.get(k, 0)
|
||||||
|
if v > rem:
|
||||||
|
return {"allowed": False, "reason": f"budget-{k}-exceeded"}
|
||||||
|
post_ok, post_reason = self.evaluate_post(plan, context)
|
||||||
|
if not post_ok:
|
||||||
|
return {"allowed": False, "reason": post_reason}
|
||||||
|
return {"allowed": True, "reason": "allowed"}
|
||||||
|
|
@ -4,7 +4,11 @@ def _safe_eval(expr: str, context: Dict[str, Any]) -> bool:
|
||||||
if not expr:
|
if not expr:
|
||||||
return True
|
return True
|
||||||
allowed_globals = {"__builtins__": {}}
|
allowed_globals = {"__builtins__": {}}
|
||||||
|
# Provide both a 'state' name and individual keys so legacy expressions
|
||||||
|
# that reference state['x'] or direct keys both work.
|
||||||
local = dict(context)
|
local = dict(context)
|
||||||
|
if "state" not in local:
|
||||||
|
local = {**local, "state": dict(context)}
|
||||||
try:
|
try:
|
||||||
return bool(eval(expr, allowed_globals, local))
|
return bool(eval(expr, allowed_globals, local))
|
||||||
except Exception:
|
except Exception:
|
||||||
|
|
@ -21,13 +25,20 @@ class PolicyEngine:
|
||||||
def evaluate_pre(self, action: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, str]:
|
def evaluate_pre(self, action: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, str]:
|
||||||
if not self.contract or not self.contract.pre:
|
if not self.contract or not self.contract.pre:
|
||||||
return True, "no-precondition"
|
return True, "no-precondition"
|
||||||
ok = _safe_eval(self.contract.pre, {**context, "action": action})
|
# Ensure 'state' alias is available for legacy contract expressions.
|
||||||
|
eval_ctx = {**context, "action": action}
|
||||||
|
if "state" not in eval_ctx:
|
||||||
|
eval_ctx = {**eval_ctx, "state": dict(context)}
|
||||||
|
ok = _safe_eval(self.contract.pre, eval_ctx)
|
||||||
return (bool(ok), "precondition" if ok else "precondition-failed")
|
return (bool(ok), "precondition" if ok else "precondition-failed")
|
||||||
|
|
||||||
def evaluate_post(self, action: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, str]:
|
def evaluate_post(self, action: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, str]:
|
||||||
if not self.contract or not self.contract.post:
|
if not self.contract or not self.contract.post:
|
||||||
return True, "no-postcondition"
|
return True, "no-postcondition"
|
||||||
ok = _safe_eval(self.contract.post, {**context, "action": action})
|
eval_ctx = {**context, "action": action}
|
||||||
|
if "state" not in eval_ctx:
|
||||||
|
eval_ctx = {**eval_ctx, "state": dict(context)}
|
||||||
|
ok = _safe_eval(self.contract.post, eval_ctx)
|
||||||
return (bool(ok), "postcondition" if ok else "postcondition-failed")
|
return (bool(ok), "postcondition" if ok else "postcondition-failed")
|
||||||
|
|
||||||
def remaining_budgets(self, context: Dict[str, Any]) -> Dict[str, float]:
|
def remaining_budgets(self, context: Dict[str, Any]) -> Dict[str, float]:
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue