from dataclasses import dataclass, field from typing import Any, Dict, Optional, List, Tuple def safe_eval(expr: Optional[str], context: Dict[str, Any]) -> bool: """Safely evaluate a simple boolean expression string against context. This is intentionally tiny: no builtins are exposed and evaluation failures return False. Legacy contract expressions expect `state` to be available. """ if not expr: return True allowed_globals = {"__builtins__": {}} local = dict(context or {}) if "state" not in local: local = {**local, "state": dict(context or {})} try: return bool(eval(expr, allowed_globals, local)) except Exception: return False @dataclass class Precondition: """A simple precondition that checks a predicate on the state and action. The predicate is a callable(state, action) -> bool. For this MVP we accept a dict-based predicate described by keys to compare (simple operators). """ key: str op: str value: Any def check(self, state: Dict[str, Any], action: Dict[str, Any]) -> bool: # Supports keys like 'state.battery' or 'action.energy' target = state if self.key.startswith("state.") or not self.key.startswith("action.") else action if self.key.startswith("state."): k = self.key.split("state.", 1)[1] val = state.get(k) elif self.key.startswith("action."): k = self.key.split("action.", 1)[1] val = action.get(k) else: val = state.get(self.key) if self.op == ">=": return val >= self.value if self.op == "<=": return val <= self.value if self.op == ">": return val > self.value if self.op == "<": return val < self.value if self.op == "==": return val == self.value if self.op == "!=": return val != self.value return False @dataclass class Postcondition: key: str op: str value: Any def check(self, state: Dict[str, Any], action: Dict[str, Any], next_state: Dict[str, Any]) -> bool: # Similar simple check against next_state val = next_state.get(self.key) if self.op == ">=": return val >= self.value if self.op == "<=": return val <= self.value if self.op == "==": return val == self.value return False @dataclass class Budget: name: str limit: float used: float = 0.0 def available(self) -> float: return self.limit - self.used def consume(self, amount: float) -> bool: if amount <= self.available(): self.used += amount return True return False class SafetyContract: """Flexible SafetyContract compatible with legacy and dataclass-style construction. Legacy tests construct using keyword names like `contract_id`, `pre_conditions` (list of string expressions), `budgets` as a dict, and `collision_rules`. Newer tests may create Precondition/Postcondition dataclasses and pass them via `preconditions`/`postconditions`. This class accepts either form and exposes both evaluation helpers and the simpler check_* helpers used by the minimal engine in this package. """ def __init__( self, id: Optional[str] = None, contract_id: Optional[str] = None, pre_conditions: Optional[List[str]] = None, post_conditions: Optional[List[str]] = None, pre: Optional[str] = None, post: Optional[str] = None, preconditions: Optional[List[Precondition]] = None, postconditions: Optional[List[Postcondition]] = None, budgets: Optional[Dict[str, Any]] = None, collision_rules: Optional[List[str]] = None, trust_policy: Optional[Dict[str, Any]] = None, name: Optional[str] = None, **kwargs, ): # canonical id self.id = id or contract_id or name or kwargs.get("id") # Support both string-expression pre/post (legacy) and dataclass-style if pre is not None: self.pre = pre elif pre_conditions is not None: if isinstance(pre_conditions, list): self.pre = " and ".join(pre_conditions) else: self.pre = pre_conditions else: self.pre = None if post is not None: self.post = post elif post_conditions is not None: if isinstance(post_conditions, list): self.post = " and ".join(post_conditions) else: self.post = post_conditions else: self.post = None # dataclass-style pre/postconditions if provided self.preconditions = preconditions or [] self.postconditions = postconditions or [] # budgets may be passed as simple name->limit dict (legacy) or Budget objects # Maintain both a legacy numeric mapping (self.budgets) for compatibility and # an internal mapping of Budget objects (self._budget_objs) for charging. if budgets is None: self.budgets = {} self._budget_objs = {} else: # If the caller passed Budget objects, preserve that mapping on self.budgets if isinstance(budgets, dict) and all(isinstance(v, Budget) for v in budgets.values()): self.budgets = dict(budgets) self._budget_objs = dict(budgets) else: numeric: Dict[str, float] = {} objs: Dict[str, Budget] = {} if isinstance(budgets, dict): for k, v in budgets.items(): if isinstance(v, Budget): objs[k] = v numeric[k] = v.limit else: try: numeric[k] = float(v) objs[k] = Budget(name=k, limit=float(v), used=0.0) except Exception: # ignore invalid entries pass self.budgets = numeric self._budget_objs = objs # Expose legacy 'contract_id' for compatibility with other modules self.contract_id = self.id self.collision_rules = collision_rules or [] self.trust_policy = trust_policy or {} # --- Compatibility helpers expected by policy/guard modules --- def evaluate_pre(self, context: Dict[str, Any]) -> bool: # context is typically the state dict; safe_eval will use 'state' alias return bool(safe_eval(self.pre, context)) def evaluate_post(self, action: Dict[str, Any], context: Dict[str, Any]) -> bool: eval_ctx = {**context, "action": action} return bool(safe_eval(self.post, eval_ctx)) # --- Newer-style helpers --- def check_preconditions(self, state: Dict[str, Any], action: Dict[str, Any]) -> Tuple[bool, Optional[str]]: for p in self.preconditions: if not p.check(state, action): return False, f"precondition failed: {p.key} {p.op} {p.value}" return True, None def check_budgets(self, state: Dict[str, Any], action: Dict[str, Any]) -> Tuple[bool, Optional[str]]: costs = action.get("costs", {}) for name, amt in costs.items(): b = self._budget_objs.get(name) if b is None: return False, f"unknown budget: {name}" if amt > b.available(): return False, f"budget exceeded: {name} wants {amt} available {b.available()}" return True, None def charge_budgets(self, action: Dict[str, Any]): costs = action.get("costs", {}) for name, amt in costs.items(): b = self._budget_objs.get(name) if b is not None: b.consume(amt)