build(agent): jabba#56a767 iteration

AGENTS.md

@@ -24,6 +24,10 @@ New modules added in this iteration:
   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`.

README.md

@@ -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.

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

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"