diff --git a/AGENTS.md b/AGENTS.md index 5b12fee..8dd9be1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,10 +20,14 @@ New modules added in this iteration: 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. - - src/guardrail_space/capsule.py: Poetic Situation Capsule generator that - emits a tiny 128-512B human+machine readable capsule (round,plan_hash,PoV, - energy_gap,verdict,human_summary). Tests in tests/test_capsule.py cover - deterministic generation and size constraints. +- src/guardrail_space/capsule.py: Poetic Situation Capsule generator that + emits a tiny 128-512B human+machine readable capsule (round,plan_hash,PoV, + energy_gap,verdict,human_summary). Tests in tests/test_capsule.py cover + deterministic generation and size constraints. + - src/guardrail_space/collision_oracle.py: deterministic pairwise plan-collision + oracle useful for multi-agent custody-window checks and small sidecar + conflict verdicts. Tests in tests/test_collision_oracle.py exercise basic + conflict and compatibility scenarios. 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 84579a3..1969e0b 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ This repository contains a minimal Python package implementing a lightweight Saf 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 +- `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 @@ -15,3 +15,7 @@ Added in this iteration: - `src/guardrail_space/capsule.py`: Poetic Situation Capsule generator for tiny (128-512B) human+machine readable summaries useful in low-bandwidth operator triage. See tests/test_capsule.py for deterministic behavior and size checks. + - `src/guardrail_space/collision_oracle.py`: deterministic pairwise plan-collision + oracle useful for multi-agent custody-window checks and small sidecar + conflict verdicts. See tests/test_collision_oracle.py for examples and + determinism. diff --git a/src/guardrail_space/collision_oracle.py b/src/guardrail_space/collision_oracle.py new file mode 100644 index 0000000..e043fa3 --- /dev/null +++ b/src/guardrail_space/collision_oracle.py @@ -0,0 +1,105 @@ +"""Deterministic Plan-Collision Oracle + +This small, dependency-free module provides a lightweight sidecar-style +collision checker suitable for offline/fleet custody windows. It accepts +simple plan deltas (trajectory lists of (time, x, y)) and reports whether +two or more plans are compatible given a distance threshold and time +tolerance. The implementation is intentionally small and deterministic so +it can be used in CI and merkle-anchored counterexample bundles. +""" +from typing import List, Dict, Tuple, Any, Optional +import math + + +def _interp_position(trajectory: List[Tuple[float, float, float]], t: float) -> Optional[Tuple[float, float]]: + """Linearly interpolate (x,y) at time t from a trajectory list [(t,x,y),...]. + Returns None if t is outside the trajectory time bounds. + """ + if not trajectory: + return None + # ensure sorted + traj = sorted(trajectory, key=lambda p: p[0]) + if t < traj[0][0] or t > traj[-1][0]: + return None + # exact match + for tt, x, y in traj: + if abs(tt - t) < 1e-12: + return (x, y) + # find enclosing interval + for i in range(len(traj) - 1): + t0, x0, y0 = traj[i] + t1, x1, y1 = traj[i + 1] + if t0 <= t <= t1: + if t1 == t0: + return (x0, y0) + alpha = (t - t0) / (t1 - t0) + return (x0 + alpha * (x1 - x0), y0 + alpha * (y1 - y0)) + return None + + +def _distance(a: Tuple[float, float], b: Tuple[float, float]) -> float: + dx = a[0] - b[0] + dy = a[1] - b[1] + return math.hypot(dx, dy) + + +def check_pair_conflict( + plan_a: Dict[str, Any], + plan_b: Dict[str, Any], + distance_threshold: float = 0.5, + time_tolerance: float = 0.05, +) -> Dict[str, Any]: + """Check two plans for a collision. + + Plans are dicts with keys: + - id: identifier + - trajectory: List of (time, x, y) + + Returns a verdict dict: + {"verdict": "compatible"|"conflict", "conflicts": [ ... ]} + + Each conflict entry contains {time, pos_a, pos_b, distance} for the + earliest detected conflict(s). Deterministic: search times derived + from both trajectories' sampled times. + """ + traj_a = plan_a.get("trajectory", []) + traj_b = plan_b.get("trajectory", []) + + # collect candidate times to check: all timestamps from both trajectories + times = sorted(set([float(t) for t, *_ in traj_a] + [float(t) for t, *_ in traj_b])) + conflicts = [] + + for t in times: + # also check nearby times within tolerance to account for asynchronous samples + check_times = [t - time_tolerance, t, t + time_tolerance] + for ct in check_times: + pa = _interp_position(traj_a, ct) + pb = _interp_position(traj_b, ct) + if pa is None or pb is None: + continue + d = _distance(pa, pb) + if d <= distance_threshold: + conflicts.append({ + "time": ct, + "pos_a": (round(pa[0], 6), round(pa[1], 6)), + "pos_b": (round(pb[0], 6), round(pb[1], 6)), + "distance": round(d, 6), + }) + if conflicts: + break + + verdict = "conflict" if conflicts else "compatible" + return {"verdict": verdict, "conflicts": conflicts} + + +def aggregate_check(plans: List[Dict[str, Any]], distance_threshold: float = 0.5, time_tolerance: float = 0.05) -> Dict[Tuple[str, str], Dict[str, Any]]: + """Run pairwise checks across a list of plans and return mapping of pair->verdict.""" + out = {} + n = len(plans) + for i in range(n): + for j in range(i + 1, n): + a = plans[i] + b = plans[j] + key = (a.get("id", f"{i}"), b.get("id", f"{j}")) + out[key] = check_pair_conflict(a, b, distance_threshold=distance_threshold, time_tolerance=time_tolerance) + return out diff --git a/tests/test_collision_oracle.py b/tests/test_collision_oracle.py new file mode 100644 index 0000000..c65bb21 --- /dev/null +++ b/tests/test_collision_oracle.py @@ -0,0 +1,32 @@ +from guardrail_space.collision_oracle import check_pair_conflict, aggregate_check + + +def _make_line_plan(pid: str, start: float, x0: float, y0: float, x1: float, y1: float) -> dict: + # simple two-point trajectory from start time to start+1.0 + return {"id": pid, "trajectory": [(start, x0, y0), (start + 1.0, x1, y1)]} + + +def test_direct_collision_detected(): + a = _make_line_plan("A", 0.0, 0.0, 0.0, 1.0, 0.0) + b = _make_line_plan("B", 0.0, 0.0, 0.0, -1.0, 0.0) + res = check_pair_conflict(a, b, distance_threshold=0.1) + assert res["verdict"] == "conflict" + assert len(res["conflicts"]) >= 1 + + +def test_spatially_separated_are_compatible(): + a = _make_line_plan("A", 0.0, 0.0, 0.0, 1.0, 0.0) + b = _make_line_plan("B", 0.0, 10.0, 0.0, 11.0, 0.0) + res = check_pair_conflict(a, b, distance_threshold=0.5) + assert res["verdict"] == "compatible" + + +def test_aggregate_pairwise(): + p1 = _make_line_plan("p1", 0.0, 0.0, 0.0, 1.0, 0.0) + p2 = _make_line_plan("p2", 0.0, 0.0, 0.0, -1.0, 0.0) + p3 = _make_line_plan("p3", 0.0, 10.0, 0.0, 11.0, 0.0) + out = aggregate_check([p1, p2, p3], distance_threshold=0.2) + # p1 vs p2 conflict, others compatible + assert out[("p1", "p2")]["verdict"] == "conflict" + assert out[("p1", "p3")]["verdict"] == "compatible" + assert out[("p2", "p3")]["verdict"] == "compatible"