diff --git a/AGENTS.md b/AGENTS.md index 89502b3..694d2d4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -15,5 +15,11 @@ The goal is to provide a production-ready scaffold that can be progressively ext If you add features, update this document to reflect schema changes and testing commands. +New modules added in this iteration: +- src/guardrail_space/belief.py: a compact particle-filter sketch that emits + probability-of-violation (PoV) and an entropy summary. Includes serialize() + for compact fingerprints. Unit tests in tests/test_belief.py exercise + deterministic behavior and compact serialization. + Testing - Run `./test.sh` to execute unit tests and build a sdist/wheel with `python3 -m build`. diff --git a/guard_logs.jsonl b/guard_logs.jsonl index 06c88fc..c522303 100644 --- a/guard_logs.jsonl +++ b/guard_logs.jsonl @@ -10,3 +10,4 @@ {"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/belief.py b/guardrail_space/belief.py new file mode 100644 index 0000000..a50093e --- /dev/null +++ b/guardrail_space/belief.py @@ -0,0 +1,139 @@ +import math +import struct +import hashlib +import random +from typing import List, Tuple + + +class BeliefSketch: + """A tiny particle-filter based belief sketch for a single scalar variable. + + Purpose: maintain a compact, mission-tuned belief for a scalar safety variable + (e.g., battery, obstacle distance) and emit a probability-of-violation (PoV) + and an entropy summary. The implementation is intentionally small and + deterministic when a seed is provided so CI/edge tests are reproducible. + + This is not a full particle filter for production use, but a compact + lightweight sketch suitable for the GuardRail.Space MVP and unit tests. + """ + + def __init__( + self, + name: str, + num_particles: int = 64, + prior_mean: float = 0.0, + prior_std: float = 1.0, + seed: int = None, + ): + self.name = name + self.num_particles = max(8, int(num_particles)) + self.rng = random.Random(seed) + # initialize particles from a normal prior + self.particles: List[float] = [ + self.rng.gauss(prior_mean, prior_std) for _ in range(self.num_particles) + ] + # equal weights (implicit) + + def update(self, observation: float, obs_std: float = 1.0) -> None: + """Update belief with a scalar observation using likelihood weighting + followed by multinomial resampling and small jitter to avoid particle + degeneracy. Deterministic when seed provided at construction. + """ + if obs_std <= 0: + obs_std = 1e-6 + # quick check: if observation is far outside current particle spread, + # reinitialize particles around the observation so the sketch remains + # responsive to strong, previously-unseen signals. + n = len(self.particles) + mean0 = sum(self.particles) / n if n else 0.0 + var0 = sum((p - mean0) ** 2 for p in self.particles) / n if n else 0.0 + std0 = math.sqrt(max(var0, 1e-12)) + if abs(observation - mean0) > max(3.0 * std0, 3.0 * obs_std): + self.particles = [self.rng.gauss(observation, max(1e-6, obs_std * 0.5)) for _ in range(self.num_particles)] + return + + # compute weights proportional to Gaussian likelihood + weights = [] + two_var = 2.0 * (obs_std ** 2) + for p in self.particles: + # gaussian likelihood unnormalized + w = math.exp(-((observation - p) ** 2) / two_var) + weights.append(w) + total = sum(weights) + # If likelihoods underflow to (near) zero, treat the observation as + # dominant and reinitialize particles around the observation. This + # keeps the sketch responsive to strong signals and avoids a degenerate + # uniform-resample when all weights are effectively zero. + if total <= 1e-12: + self.particles = [self.rng.gauss(observation, max(1e-6, obs_std * 0.5)) for _ in range(self.num_particles)] + return + probs = [w / total for w in weights] + + # multinomial resampling + cumulative = [] + c = 0.0 + for p in probs: + c += p + cumulative.append(c) + + new_particles = [] + for _ in range(self.num_particles): + u = self.rng.random() + # find first cumulative >= u + for idx, c in enumerate(cumulative): + if u <= c: + new_particles.append(self.particles[idx]) + break + else: + new_particles.append(self.particles[-1]) + + # jitter with small gaussian noise proportional to obs_std + jitter_scale = max(1e-3, obs_std * 0.01) + self.particles = [p + self.rng.gauss(0.0, jitter_scale) for p in new_particles] + + def pov(self, threshold: float, operator: str = ">") -> float: + """Return probability-of-violation (PoV) that the scalar meets the + violation predicate defined by (operator, threshold). Supported + operators: '>' (default), '<', '>=', '<='. + """ + if operator == ">": + count = sum(1 for p in self.particles if p > threshold) + elif operator == "<": + count = sum(1 for p in self.particles if p < threshold) + elif operator == ">=": + count = sum(1 for p in self.particles if p >= threshold) + elif operator == "<=": + count = sum(1 for p in self.particles if p <= threshold) + else: + raise ValueError(f"unsupported operator: {operator}") + return float(count) / float(len(self.particles)) + + def entropy(self) -> float: + """Return a simple continuous-entropy proxy (Gaussian entropy using + empirical variance). This is cheap and stable for small sketches. + """ + n = len(self.particles) + if n == 0: + return 0.0 + mean = sum(self.particles) / n + var = sum((p - mean) ** 2 for p in self.particles) / n + var = max(var, 1e-12) + # differential entropy of Gaussian: 0.5*ln(2*pi*e*var) + return 0.5 * math.log(2 * math.pi * math.e * var) + + def summarize(self, threshold: float, operator: str = ">") -> dict: + """Return a compact summary dictionary with PoV and entropy. + """ + return {"name": self.name, "pov": self.pov(threshold, operator), "entropy": self.entropy()} + + def serialize(self) -> bytes: + """Produce a tiny deterministic byte fingerprint of the sketch. Format: + 8 bytes mean (double) + 8 bytes variance (double) + 8 bytes crc64-like + truncated using sha256. This keeps the sketch small (<64 bytes). + """ + n = len(self.particles) + mean = sum(self.particles) / n if n else 0.0 + var = sum((p - mean) ** 2 for p in self.particles) / n if n else 0.0 + packed = struct.pack("!dd", mean, var) + digest = hashlib.sha256(packed).digest()[:16] + return packed + digest diff --git a/src/guardrail_space/belief.py b/src/guardrail_space/belief.py new file mode 100644 index 0000000..a50093e --- /dev/null +++ b/src/guardrail_space/belief.py @@ -0,0 +1,139 @@ +import math +import struct +import hashlib +import random +from typing import List, Tuple + + +class BeliefSketch: + """A tiny particle-filter based belief sketch for a single scalar variable. + + Purpose: maintain a compact, mission-tuned belief for a scalar safety variable + (e.g., battery, obstacle distance) and emit a probability-of-violation (PoV) + and an entropy summary. The implementation is intentionally small and + deterministic when a seed is provided so CI/edge tests are reproducible. + + This is not a full particle filter for production use, but a compact + lightweight sketch suitable for the GuardRail.Space MVP and unit tests. + """ + + def __init__( + self, + name: str, + num_particles: int = 64, + prior_mean: float = 0.0, + prior_std: float = 1.0, + seed: int = None, + ): + self.name = name + self.num_particles = max(8, int(num_particles)) + self.rng = random.Random(seed) + # initialize particles from a normal prior + self.particles: List[float] = [ + self.rng.gauss(prior_mean, prior_std) for _ in range(self.num_particles) + ] + # equal weights (implicit) + + def update(self, observation: float, obs_std: float = 1.0) -> None: + """Update belief with a scalar observation using likelihood weighting + followed by multinomial resampling and small jitter to avoid particle + degeneracy. Deterministic when seed provided at construction. + """ + if obs_std <= 0: + obs_std = 1e-6 + # quick check: if observation is far outside current particle spread, + # reinitialize particles around the observation so the sketch remains + # responsive to strong, previously-unseen signals. + n = len(self.particles) + mean0 = sum(self.particles) / n if n else 0.0 + var0 = sum((p - mean0) ** 2 for p in self.particles) / n if n else 0.0 + std0 = math.sqrt(max(var0, 1e-12)) + if abs(observation - mean0) > max(3.0 * std0, 3.0 * obs_std): + self.particles = [self.rng.gauss(observation, max(1e-6, obs_std * 0.5)) for _ in range(self.num_particles)] + return + + # compute weights proportional to Gaussian likelihood + weights = [] + two_var = 2.0 * (obs_std ** 2) + for p in self.particles: + # gaussian likelihood unnormalized + w = math.exp(-((observation - p) ** 2) / two_var) + weights.append(w) + total = sum(weights) + # If likelihoods underflow to (near) zero, treat the observation as + # dominant and reinitialize particles around the observation. This + # keeps the sketch responsive to strong signals and avoids a degenerate + # uniform-resample when all weights are effectively zero. + if total <= 1e-12: + self.particles = [self.rng.gauss(observation, max(1e-6, obs_std * 0.5)) for _ in range(self.num_particles)] + return + probs = [w / total for w in weights] + + # multinomial resampling + cumulative = [] + c = 0.0 + for p in probs: + c += p + cumulative.append(c) + + new_particles = [] + for _ in range(self.num_particles): + u = self.rng.random() + # find first cumulative >= u + for idx, c in enumerate(cumulative): + if u <= c: + new_particles.append(self.particles[idx]) + break + else: + new_particles.append(self.particles[-1]) + + # jitter with small gaussian noise proportional to obs_std + jitter_scale = max(1e-3, obs_std * 0.01) + self.particles = [p + self.rng.gauss(0.0, jitter_scale) for p in new_particles] + + def pov(self, threshold: float, operator: str = ">") -> float: + """Return probability-of-violation (PoV) that the scalar meets the + violation predicate defined by (operator, threshold). Supported + operators: '>' (default), '<', '>=', '<='. + """ + if operator == ">": + count = sum(1 for p in self.particles if p > threshold) + elif operator == "<": + count = sum(1 for p in self.particles if p < threshold) + elif operator == ">=": + count = sum(1 for p in self.particles if p >= threshold) + elif operator == "<=": + count = sum(1 for p in self.particles if p <= threshold) + else: + raise ValueError(f"unsupported operator: {operator}") + return float(count) / float(len(self.particles)) + + def entropy(self) -> float: + """Return a simple continuous-entropy proxy (Gaussian entropy using + empirical variance). This is cheap and stable for small sketches. + """ + n = len(self.particles) + if n == 0: + return 0.0 + mean = sum(self.particles) / n + var = sum((p - mean) ** 2 for p in self.particles) / n + var = max(var, 1e-12) + # differential entropy of Gaussian: 0.5*ln(2*pi*e*var) + return 0.5 * math.log(2 * math.pi * math.e * var) + + def summarize(self, threshold: float, operator: str = ">") -> dict: + """Return a compact summary dictionary with PoV and entropy. + """ + return {"name": self.name, "pov": self.pov(threshold, operator), "entropy": self.entropy()} + + def serialize(self) -> bytes: + """Produce a tiny deterministic byte fingerprint of the sketch. Format: + 8 bytes mean (double) + 8 bytes variance (double) + 8 bytes crc64-like + truncated using sha256. This keeps the sketch small (<64 bytes). + """ + n = len(self.particles) + mean = sum(self.particles) / n if n else 0.0 + var = sum((p - mean) ** 2 for p in self.particles) / n if n else 0.0 + packed = struct.pack("!dd", mean, var) + digest = hashlib.sha256(packed).digest()[:16] + return packed + digest diff --git a/src/guardrail_space/contract.py b/src/guardrail_space/contract.py index 30ccbdb..4a227c5 100644 --- a/src/guardrail_space/contract.py +++ b/src/guardrail_space/contract.py @@ -1,76 +1,216 @@ -from typing import Dict, Any, Optional, List -from .policy_engine import PolicyEngine +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: - """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. + """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, - 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, + 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, ): - # 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 + # canonical id + self.id = id or contract_id or name or kwargs.get("id") - # Pre/post can be provided as lists or as a single string. + # Support both string-expression pre/post (legacy) and dataclass-style if pre is not None: self.pre = pre - else: - if pre_conditions is None: - self.pre = None + elif pre_conditions is not None: + if isinstance(pre_conditions, list): + self.pre = " and ".join(pre_conditions) else: - if isinstance(pre_conditions, list): - self.pre = " and ".join(pre_conditions) - else: - self.pre = pre_conditions + self.pre = pre_conditions + else: + self.pre = None if post is not None: self.post = post - else: - if post_conditions is None: - self.post = None + elif post_conditions is not None: + if isinstance(post_conditions, list): + self.post = " and ".join(post_conditions) else: - if isinstance(post_conditions, list): - self.post = " and ".join(post_conditions) - else: - self.post = post_conditions + 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.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. - ok, _reason = engine.evaluate_pre({"action": None}, context) - # Legacy API expected a boolean result. - return bool(ok) + # --- 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, str): - engine = PolicyEngine(self) - ok, _reason = engine.evaluate_post(action, context) - return bool(ok) + 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)) - 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, - } + # --- 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/src/guardrail_space/engine.py b/src/guardrail_space/engine.py new file mode 100644 index 0000000..71f418d --- /dev/null +++ b/src/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/tests/test_belief.py b/tests/test_belief.py new file mode 100644 index 0000000..54d6d37 --- /dev/null +++ b/tests/test_belief.py @@ -0,0 +1,36 @@ +from guardrail_space.belief import BeliefSketch + + +def test_belief_pov_and_entropy_deterministic(): + # deterministic seed so CI results are reproducible + sketch = BeliefSketch("obstacle_dist", num_particles=128, prior_mean=0.0, prior_std=1.0, seed=42) + + # an observation far above the prior should push mass upward + sketch.update(observation=5.0, obs_std=0.5) + + pov = sketch.pov(threshold=3.0, operator=">") + ent = sketch.entropy() + + # After a strong observation at 5.0 most particles should be >3.0 + assert pov > 0.8 + + # entropy should be a finite positive number + assert isinstance(ent, float) + assert ent > 0.0 + + +def test_serialize_small_and_deterministic(): + s1 = BeliefSketch("battery", num_particles=32, prior_mean=10.0, prior_std=2.0, seed=7) + s2 = BeliefSketch("battery", num_particles=32, prior_mean=10.0, prior_std=2.0, seed=7) + + s1.update(9.5, obs_std=0.5) + s2.update(9.5, obs_std=0.5) + + b1 = s1.serialize() + b2 = s2.serialize() + + # deterministic with same seed and ops + assert b1 == b2 + + # compact + assert len(b1) <= 40