build(agent): jabba#56a767 iteration

AGENTS.md

@@ -14,3 +14,6 @@ The goal is to provide a production-ready scaffold that can be progressively ext
 - Additional adapters for Gazebo/ROS integration and offline reconciliation scaffolding will be implemented progressively.
 
 If you add features, update this document to reflect schema changes and testing commands.
+
+Testing
+- Run `./test.sh` to execute unit tests and build a sdist/wheel with `python3 -m build`.

README.md

@@ -1,13 +1,13 @@
-# GuardRail.Space MVP
+# GuardRail.Space — Safety Contracts DSL (partial)
 
-A lightweight, open-source safety framework to govern onboard AGI planners for space robotics.
+This repository contains a minimal Python package implementing a lightweight SafetyContract DSL and a tiny runtime policy engine suitable for use as the basis of GuardRail.Space. It is intentionally small: a focused, well-tested chunk that implements SafetyContracts, basic budget checks, and a policy engine that can veto or rewrite proposed actions.
 
-- Safety Contract DSL: define pre/post conditions, budgets, collision/warning rules, and trust policy.
+What is included in this change
-- Runtime policy engine: veto or adjust proposed actions.
+- A small Python package `guardrail_space` with:
-- Shadow planner: risk-aware alternative planning running in parallel.
+  - `contract.py`: SafetyContract dataclasses and checks
-- Offline-first design: deterministic reconciliation and auditable logs.
+  - `engine.py`: a minimal policy engine that can accept, veto, or rewrite actions
-- Privacy-by-design: policy data stays on-device; aggregation only where allowed.
+- Unit tests under `tests/` exercising contract evaluation and engine behavior
-- Lightweight verification hooks and adapters for CatOpt-style interoperability.
+- Packaging metadata (`pyproject.toml`, `setup.cfg`) so `python3 -m build` works
-- Gazebo/ROS integration path and HIL validation plan (skeletons provided).
+- `test.sh` which runs the test-suite and builds a sdist/wheel
 
-This repository provides a minimal production-ready scaffold to be extended with adapters and a full test harness.
+This is a focused increment toward the full GuardRail.Space vision (WASM runner, ROS shims, fuzz harness, etc.). Next steps can extend the DSL, add serialization (YAML/DSL), shadow runner integration, and the situation-summary artifacts.

guard_logs.jsonl

@@ -8,3 +8,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"}

guardrail_space/__init__.py

@@ -1,12 +1,2 @@
-"""GuardRail.Space MVP: Verifiable Safety Contracts for Onboard AGI-Driven Systems
+"""GuardRail.Space minimal package"""
-
+__version__ = "0.1.0"
-This package provides a minimal, production-ready skeleton for a Safety Contract DSL,
-runtime policy engine, and a guard module with a shadow planner. It is designed as a
-foundational core for further integration with Gazebo/ROS and CatOpt-style adapters.
-"""
-
-from .contract import SafetyContract
-from .guard import GuardModule
-from .shadow_planner import ShadowPlanner
-
-__all__ = ["SafetyContract", "GuardModule", "ShadowPlanner"]

guardrail_space/contract.py

@@ -1,61 +1,216 @@
-from __future__ import annotations
-
 from dataclasses import dataclass, field
-from typing import List, Dict, Any
+from typing import Any, Dict, Optional, List, Tuple
 
 
-def safe_eval(expr: str, context: Dict[str, Any]) -> bool:
+def safe_eval(expr: Optional[str], context: Dict[str, Any]) -> bool:
-    # Extremely small, sandboxed evaluator for MVP.
+    """Safely evaluate a simple boolean expression string against context.
-    allowed_builtins = {"abs": abs, "min": min, "max": max, "sum": sum, "len": len}
+
+    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, {"__builtins__": allowed_builtins}, context))
+        return bool(eval(expr, allowed_globals, local))
     except Exception:
-        # If evaluation fails, be conservative and treat as not satisfied
         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:
-    contract_id: str
+    """Flexible SafetyContract compatible with legacy and dataclass-style construction.
-    pre_conditions: List[str] = field(default_factory=list)
-    post_conditions: List[str] = field(default_factory=list)
-    budgets: Dict[str, float] = field(default_factory=dict)  # e.g., {"time": 10.0, "energy": 100.0}
-    collision_rules: List[str] = field(default_factory=list)
-    trust_policy: Dict[str, str] = field(default_factory=dict)
 
-    def to_dict(self) -> Dict[str, Any]:
+    Legacy tests construct using keyword names like `contract_id`, `pre_conditions` (list of
-        return {
+    string expressions), `budgets` as a dict, and `collision_rules`. Newer tests may create
-            "contract_id": self.contract_id,
+    Precondition/Postcondition dataclasses and pass them via `preconditions`/`postconditions`.
-            "pre_conditions": self.pre_conditions,
-            "post_conditions": self.post_conditions,
-            "budgets": self.budgets,
-            "collision_rules": self.collision_rules,
-            "trust_policy": self.trust_policy,
-        }
 
-    @staticmethod
+    This class accepts either form and exposes both evaluation helpers and the simpler
-    def from_dict(data: Dict[str, Any]) -> "SafetyContract":
+    check_* helpers used by the minimal engine in this package.
-        return SafetyContract(
+    """
-            contract_id=data.get("contract_id", "unnamed-contract"),
-            pre_conditions=data.get("pre_conditions", []),
-            post_conditions=data.get("post_conditions", []),
-            budgets=data.get("budgets", {}),
-            collision_rules=data.get("collision_rules", []),
-            trust_policy=data.get("trust_policy", {}),
-        )
 
-    def evaluate_pre(self, state: Dict[str, Any]) -> bool:
+    def __init__(
-        # Evaluate all pre-conditions in the given state/context
+        self,
-        # Some DSLs reference a "state" object; support that by packaging the current state under 'state'
+        id: Optional[str] = None,
-        local_context = {"state": state}
+        contract_id: Optional[str] = None,
-        for expr in self.pre_conditions:
+        pre_conditions: Optional[List[str]] = None,
-            if not safe_eval(expr, local_context):
+        post_conditions: Optional[List[str]] = None,
-                return False
+        pre: Optional[str] = None,
-        return True
+        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")
 
-    def evaluate_post(self, state: Dict[str, Any]) -> bool:
+        # Support both string-expression pre/post (legacy) and dataclass-style
-        local_context = {"state": state}
+        if pre is not None:
-        for expr in self.post_conditions:
+            self.pre = pre
-            if not safe_eval(expr, local_context):
+        elif pre_conditions is not None:
-                return False
+            if isinstance(pre_conditions, list):
-        return True
+                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)

guardrail_space/engine.py

@@ -0,0 +1,38 @@
+from typing import Dict, Any, Tuple
+from .contract import SafetyContract
+
+
+def evaluate_action(contract: SafetyContract, state: Dict[str, Any], action: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]:
+    """Evaluate a proposed action against a SafetyContract.
+
+    Returns a tuple (verdict, payload) where verdict is one of:
+    - 'accept' (payload is possibly rewritten action)
+    - 'veto' (payload contains reason)
+
+    This minimal engine performs precondition and budget checks and may
+    rewrite actions to a safe alternative when possible.
+    """
+    ok, reason = contract.check_preconditions(state, action)
+    if not ok:
+        return "veto", {"reason": reason}
+
+    ok, reason = contract.check_budgets(state, action)
+    if not ok:
+        # attempt a simple rewrite: scale down costs/speeds if present
+        if "speed" in action:
+            safe_action = dict(action)
+            safe_action["speed"] = action["speed"] * 0.5
+            # Re-check budgets assuming cost scales with speed linearly
+            costs = dict(action.get("costs", {}))
+            for k in costs:
+                costs[k] = costs[k] * 0.5
+            safe_action["costs"] = costs
+            ok2, reason2 = contract.check_budgets(state, safe_action)
+            if ok2:
+                contract.charge_budgets(safe_action)
+                return "accept", safe_action
+        return "veto", {"reason": reason}
+
+    # Accept and charge budgets
+    contract.charge_budgets(action)
+    return "accept", action

guardrail_space/policy.py

@@ -29,7 +29,9 @@ class PolicyEngine:
         # 3) Collision/warning rules
         for expr in self.contract.collision_rules:
             if expr:
-                if safe_eval(expr, {**state, **plan}):
+                # collision_rules are expressions that should be True when safe.
+                # If the expression evaluates to False we treat this as a violation.
+                if not safe_eval(expr, {**state, **plan}):
                     return {"allowed": False, "reason": "collision_rule_violation"}
 
         return {"allowed": True}

pyproject.toml

@@ -1,16 +1,3 @@
 [build-system]
-requires = ["setuptools", "wheel"]
+requires = ["setuptools>=61.0","wheel"]
 build-backend = "setuptools.build_meta"
-
-[tool.poetry]  # optional, kept for readability if Poetry is used later
-name = "guardrail_space"
-version = "0.1.0"
-
-[project]
-name = "guardrail_space"
-version = "0.1.0"
-description = "MVP: verifiable safety contracts and offline safety monitor for onboard AGI-driven space robotics"
-readme = "README.md"
-license = {text = "MIT"}
-requires-python = ">=3.8"
-dependencies = []

setup.cfg

@@ -0,0 +1,13 @@
+[metadata]
+name = guardrail-space
+version = 0.1.0
+description = GuardRail.Space — verifiable safety contracts and runtime guard for onboard robotics
+readme = README.md
+long_description_content_type = text/markdown
+author = GuardRail Contributors
+license = MIT
+
+[options]
+packages = find:
+include_package_data = True
+python_requires = >=3.8

test.sh

@@ -1,5 +1,10 @@
 #!/usr/bin/env bash
 set -euo pipefail
-echo "Running unit tests for GuardRail.Space MVP..."
+
-python3 -m unittest discover -s tests -p 'test_*.py'
+echo "Running unit tests..."
-echo "All tests passed."
+pytest -q
+
+echo "Building package (sdist + wheel)..."
+python3 -m build
+
+echo "All done."

tests/test_contract.py

@@ -1,15 +1,58 @@
-import unittest
+import os
-from guardrail_space.contract import SafetyContract
+import sys
+import pytest
+
+# Ensure top-level package directory takes precedence over any on-src compatibility
+ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+if ROOT not in sys.path:
+    sys.path.insert(0, ROOT)
+
+from guardrail_space.contract import SafetyContract, Precondition, Budget
+from guardrail_space.engine import evaluate_action
 
 
-class TestSafetyContract(unittest.TestCase):
+def test_precondition_veto():
-    def test_pre_post_eval(self):
+    contract = SafetyContract(
-        c = SafetyContract(
+        id="c1",
-            contract_id="test-001",
+        preconditions=[Precondition(key="state.battery", op=">=", value=10)],
-            pre_conditions=["state['speed'] <= 1.0"],
+        budgets={"energy": Budget(name="energy", limit=100.0)},
-            post_conditions=["state['completed'] == True"],
+    )
-            budgets={"time": 10.0, "energy": 100.0},
+
-            collision_rules=["state['distance_to_obstacle'] >= 0"],
+    state = {"battery": 5}
-        )
+    action = {"name": "drive", "costs": {"energy": 5}, "speed": 1.0}
-        self.assertTrue(c.evaluate_pre({"speed": 0.5, "distance_to_obstacle": 5}))
+
-        self.assertFalse(c.evaluate_pre({"speed": 2.0, "distance_to_obstacle": 5}))
+    verdict, payload = evaluate_action(contract, state, action)
+    assert verdict == "veto"
+    assert "precondition failed" in payload["reason"]
+
+
+def test_budget_rewrite_to_safe_action():
+    contract = SafetyContract(
+        id="c2",
+        preconditions=[Precondition(key="state.battery", op=">=", value=1)],
+        budgets={"energy": Budget(name="energy", limit=2.0)},
+    )
+
+    state = {"battery": 50}
+    action = {"name": "drive", "costs": {"energy": 4.0}, "speed": 2.0}
+
+    verdict, payload = evaluate_action(contract, state, action)
+    # With rewrite, engine should accept a scaled-down action
+    assert verdict == "accept"
+    assert payload["speed"] == pytest.approx(1.0)
+    assert payload["costs"]["energy"] == pytest.approx(2.0)
+
+
+def test_accept_action_and_charge_budget():
+    contract = SafetyContract(
+        id="c3",
+        preconditions=[Precondition(key="state.battery", op=">=", value=1)],
+        budgets={"energy": Budget(name="energy", limit=10.0)},
+    )
+
+    state = {"battery": 50}
+    action = {"name": "drive", "costs": {"energy": 3.0}, "speed": 1.0}
+
+    verdict, payload = evaluate_action(contract, state, action)
+    assert verdict == "accept"
+    assert contract.budgets["energy"].used == pytest.approx(3.0)