From 36f30c8de32043fc6885380ec261f3125e0c6b59 Mon Sep 17 00:00:00 2001 From: agent-7e3bbc424e07835b Date: Sun, 19 Apr 2026 22:49:20 +0200 Subject: [PATCH] build(agent): new-agents-2#7e3bbc iteration --- guard_logs.jsonl | 2 + src/guardrail_space/contract.py | 73 +++++++++++++++++++++++++++ src/guardrail_space/guard_module.py | 26 +++++++++- src/guardrail_space/shadow_planner.py | 11 ++++ 4 files changed, 110 insertions(+), 2 deletions(-) create mode 100644 src/guardrail_space/contract.py create mode 100644 src/guardrail_space/shadow_planner.py diff --git a/guard_logs.jsonl b/guard_logs.jsonl index 7a66b5e..ca7b826 100644 --- a/guard_logs.jsonl +++ b/guard_logs.jsonl @@ -3,3 +3,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"} diff --git a/src/guardrail_space/contract.py b/src/guardrail_space/contract.py new file mode 100644 index 0000000..3d5f6c3 --- /dev/null +++ b/src/guardrail_space/contract.py @@ -0,0 +1,73 @@ +from typing import Dict, Any, Optional, List +from .policy_engine import PolicyEngine + + +class SafetyContract: + """Compatibility wrapper for tests expecting a contract API with + pre_conditions, post_conditions, and contract_id/name. + This class normalizes inputs to the internal engine which expects + string expressions for pre/post and a budgets dict. + """ + + def __init__( + self, + contract_id: Optional[str] = None, + pre_conditions: Optional[List[str]] = None, + post_conditions: Optional[List[str]] = None, + budgets: Optional[Dict[str, float]] = None, + collision_rules: Optional[List[Any]] = None, + trust_policy: Optional[Dict[str, Any]] = None, + name: Optional[str] = None, + pre: Optional[str] = None, + post: Optional[str] = None, + **kwargs, + ): + # Normalize naming to legacy fields used by tests and engine + self.contract_id = contract_id or name or (name if name is not None else None) + self.name = self.contract_id + + # Pre/post can be provided as lists or as a single string. + if pre is not None: + self.pre = pre + else: + if pre_conditions is None: + self.pre = None + else: + if isinstance(pre_conditions, list): + self.pre = " and ".join(pre_conditions) + else: + self.pre = pre_conditions + + if post is not None: + self.post = post + else: + if post_conditions is None: + self.post = None + else: + if isinstance(post_conditions, list): + self.post = " and ".join(post_conditions) + else: + self.post = post_conditions + + self.budgets = budgets or {} + self.collision_rules = collision_rules or [] + self.trust_policy = trust_policy or {} + + def evaluate_pre(self, context: Dict[str, Any]) -> (bool, str): + engine = PolicyEngine(self) + # Pass a neutral action payload; tests use only context. + return engine.evaluate_pre({"action": None}, context) + + def evaluate_post(self, action: Dict[str, Any], context: Dict[str, Any]) -> (bool, str): + engine = PolicyEngine(self) + return engine.evaluate_post(action, context) + + def to_dict(self) -> Dict[str, Any]: + return { + "name": self.name, + "pre": self.pre, + "post": self.post, + "budgets": self.budgets, + "collision_rules": self.collision_rules, + "trust_policy": self.trust_policy, + } diff --git a/src/guardrail_space/guard_module.py b/src/guardrail_space/guard_module.py index ac3c286..0ed2d35 100644 --- a/src/guardrail_space/guard_module.py +++ b/src/guardrail_space/guard_module.py @@ -2,22 +2,26 @@ import threading from typing import Dict, Any, Tuple from .contracts import SafetyContract from .policy_engine import PolicyEngine +from .shadow_planner import ShadowPlanner class GuardModule: - def __init__(self, contract: SafetyContract): + def __init__(self, contract: SafetyContract, shadow_planner: ShadowPlanner = None): self.contract = contract + # Backward compatibility: if contract is a legacy object with .pre etc, engine handles it. self.engine = PolicyEngine(contract) self.lock = threading.Lock() self.shadow_delta: Dict[str, Any] = {} + self.shadow_planner = shadow_planner or ShadowPlanner() + # Backwards-compatible method (not used by tests that rely on evaluate_plan). def decide(self, action: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, Dict[str, Any]]: with self.lock: pre_ok, _ = self.engine.evaluate_pre(action, context) if not pre_ok: return False, {"reason": "precondition-failed", "fallback": self._fallback(action, context)} remaining = self.engine.remaining_budgets(context) - cost = action.get("cost", {}) + cost = action.get("costs", {}) for k, v in cost.items(): rem = remaining.get(k, 0) if v > rem: @@ -27,6 +31,23 @@ class GuardModule: return False, {"reason": "postcondition-failed", "fallback": self._fallback(action, context)} return True, action + def evaluate_plan(self, plan: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]: + # Public API expected by tests. Returns a dict with a 'decision' key. + with self.lock: + pre_ok, _ = self.engine.evaluate_pre(plan, context) + if not pre_ok: + return {"decision": "veto", "reason": "precondition-failed"} + remaining = self.engine.remaining_budgets(context) + costs = plan.get("costs", {}) + for k, v in costs.items(): + rem = remaining.get(k, 0) + if v > rem: + return {"decision": "veto", "reason": f"budget-{k}-exceeded"} + post_ok, _ = self.engine.evaluate_post(plan, context) + if not post_ok: + return {"decision": "veto", "reason": "postcondition-failed"} + return {"decision": "allow", "plan": plan} + def _fallback(self, action: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]: safe_action = {"name": "halt", "reason": "fallback", "context": context} thread = threading.Thread(target=self._run_shadow_planner, args=(action, context)) @@ -38,3 +59,4 @@ class GuardModule: delta = {"adjustment": {"reduce_cost": True}, "base_action": action} with self.lock: self.shadow_delta = delta + # In a full MVP this would trigger the shadow planner to compute a safe delta. diff --git a/src/guardrail_space/shadow_planner.py b/src/guardrail_space/shadow_planner.py new file mode 100644 index 0000000..313d8a4 --- /dev/null +++ b/src/guardrail_space/shadow_planner.py @@ -0,0 +1,11 @@ +class ShadowPlanner: + """Minimal stub for a shadow planner used by GuardModule. + In a full MVP this would run parallel planning and return a safe delta. + Here, it remains a no-op placeholder to satisfy interfaces in tests.""" + + def __init__(self): + pass + + def propose(self, action: dict, context: dict) -> dict: + # In a full implementation, compute a safe delta. Here we just return a no-op delta. + return {"delta": None}