diff --git a/AGENTS.md b/AGENTS.md index 3ebbe6d..89502b3 100644 --- a/AGENTS.md +++ b/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`. diff --git a/README.md b/README.md index f96a291..448c22c 100644 --- a/README.md +++ b/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. -- Runtime policy engine: veto or adjust proposed actions. -- Shadow planner: risk-aware alternative planning running in parallel. -- Offline-first design: deterministic reconciliation and auditable logs. -- Privacy-by-design: policy data stays on-device; aggregation only where allowed. -- Lightweight verification hooks and adapters for CatOpt-style interoperability. -- Gazebo/ROS integration path and HIL validation plan (skeletons provided). +What is included in this change +- A small Python package `guardrail_space` with: + - `contract.py`: SafetyContract dataclasses and checks + - `engine.py`: a minimal policy engine that can accept, veto, or rewrite actions +- Unit tests under `tests/` exercising contract evaluation and engine behavior +- Packaging metadata (`pyproject.toml`, `setup.cfg`) so `python3 -m build` works +- `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. diff --git a/guard_logs.jsonl b/guard_logs.jsonl index a9e763e..06c88fc 100644 --- a/guard_logs.jsonl +++ b/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"} diff --git a/guardrail_space/__init__.py b/guardrail_space/__init__.py index 36694be..b8edb5f 100644 --- a/guardrail_space/__init__.py +++ b/guardrail_space/__init__.py @@ -1,12 +1,2 @@ -"""GuardRail.Space MVP: Verifiable Safety Contracts for Onboard AGI-Driven Systems - -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 minimal package""" +__version__ = "0.1.0" diff --git a/guardrail_space/contract.py b/guardrail_space/contract.py index 2c3f74d..4a227c5 100644 --- a/guardrail_space/contract.py +++ b/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: - # Extremely small, sandboxed evaluator for MVP. - allowed_builtins = {"abs": abs, "min": min, "max": max, "sum": sum, "len": len} +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, {"__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 - 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) + """Flexible SafetyContract compatible with legacy and dataclass-style construction. - def to_dict(self) -> Dict[str, Any]: - return { - "contract_id": self.contract_id, - "pre_conditions": self.pre_conditions, - "post_conditions": self.post_conditions, - "budgets": self.budgets, - "collision_rules": self.collision_rules, - "trust_policy": self.trust_policy, - } + 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`. - @staticmethod - def from_dict(data: Dict[str, Any]) -> "SafetyContract": - 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", {}), - ) + This class accepts either form and exposes both evaluation helpers and the simpler + check_* helpers used by the minimal engine in this package. + """ - def evaluate_pre(self, state: Dict[str, Any]) -> bool: - # Evaluate all pre-conditions in the given state/context - # Some DSLs reference a "state" object; support that by packaging the current state under 'state' - local_context = {"state": state} - for expr in self.pre_conditions: - if not safe_eval(expr, local_context): - return False - return True + 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") - def evaluate_post(self, state: Dict[str, Any]) -> bool: - local_context = {"state": state} - for expr in self.post_conditions: - if not safe_eval(expr, local_context): - return False - return True + # 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) diff --git a/guardrail_space/engine.py b/guardrail_space/engine.py new file mode 100644 index 0000000..71f418d --- /dev/null +++ b/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 diff --git a/guardrail_space/policy.py b/guardrail_space/policy.py index 8fd579b..a955e3c 100644 --- a/guardrail_space/policy.py +++ b/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} diff --git a/pyproject.toml b/pyproject.toml index 6992801..5818a2c 100644 --- a/pyproject.toml +++ b/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 = [] diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..41c630c --- /dev/null +++ b/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 diff --git a/test.sh b/test.sh index 5a37851..46508ae 100644 --- a/test.sh +++ b/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 "All tests passed." + +echo "Running unit tests..." +pytest -q + +echo "Building package (sdist + wheel)..." +python3 -m build + +echo "All done." diff --git a/tests/test_contract.py b/tests/test_contract.py index b64fa40..63e8baa 100644 --- a/tests/test_contract.py +++ b/tests/test_contract.py @@ -1,15 +1,58 @@ -import unittest -from guardrail_space.contract import SafetyContract +import os +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_pre_post_eval(self): - c = SafetyContract( - contract_id="test-001", - pre_conditions=["state['speed'] <= 1.0"], - post_conditions=["state['completed'] == True"], - budgets={"time": 10.0, "energy": 100.0}, - collision_rules=["state['distance_to_obstacle'] >= 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})) +def test_precondition_veto(): + contract = SafetyContract( + id="c1", + preconditions=[Precondition(key="state.battery", op=">=", value=10)], + budgets={"energy": Budget(name="energy", limit=100.0)}, + ) + + state = {"battery": 5} + action = {"name": "drive", "costs": {"energy": 5}, "speed": 1.0} + + 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)