diff --git a/cmd_chat/sor/analysis/__init__.py b/cmd_chat/sor/analysis/__init__.py index f740aaf..6476c80 100644 --- a/cmd_chat/sor/analysis/__init__.py +++ b/cmd_chat/sor/analysis/__init__.py @@ -11,8 +11,13 @@ confirmatory-cell data — CLAUDE.md build discipline). Specifically: known-unlinked pair. These are pure functions over in-memory series/distributions — they read no -pcaps, spawn no engine, move no traffic, and touch no VM fabric. The -traffic-moving R7 pieces (``churn.py`` seeded VM spin/kill; the live selector -rebuild loop; the confirmatory battery that writes ``metrics.json``) are NOT -here — they are HELD pending R4/R6 + a live grid (see OVERSEER-STATUS.md). +pcaps, spawn no engine, move no traffic, and touch no VM fabric. + +The rest of R7 is landed alongside: ``metrics.py`` (this package) aggregates the +four DV families into a schema-valid, write-once ``metrics.json``; the seeded +churn schedule + rebuild loop live in ``cmd_chat/sor/{churn,selector}.py``. The +live VM spin/kill against the isolated hackhouse fabric, and the full +pre-registered confirmatory battery, remain gated by the containment law + the +human freeze (see OVERSEER-STATUS.md) — the churn schedule is seeded *data* here, +not a real VM operation. """ diff --git a/cmd_chat/sor/analysis/metrics.py b/cmd_chat/sor/analysis/metrics.py new file mode 100644 index 0000000..855b6df --- /dev/null +++ b/cmd_chat/sor/analysis/metrics.py @@ -0,0 +1,113 @@ +"""R7 — metrics.json aggregator (the DV writer for a SOR run). + +Reads the offline detector primitives (``detectors.py``) plus a churn/selector +:class:`~cmd_chat.sor.selector.SelectionResult` and writes an immutable, schema- +valid ``metrics.json`` into a run directory. It aggregates the four DV families +the study reports: + + * RQ1 — bridge-correlation AUC (linkability of ingress↔egress flows); + * RQ2 — anonymity-set entropy (bits) of the observed sender distribution; + * RQ3 — throughput retention under churn + a rebuild-pattern classifier AUC. + +The detectors are calibrated on synthetic fixtures only (never fit to +confirmatory-cell data — CLAUDE.md build discipline). This module computes and +serializes; it moves no traffic and spawns no engine. ``metrics.json`` is written +once and never edited in place (artifact immutability).""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict, List, Optional, Sequence + +from cmd_chat.sor.analysis.detectors import ( + auc, + bridge_correlation_auc, + shannon_entropy_bits, +) +from cmd_chat.sor.selector import SelectionResult + +METRICS_SCHEMA = "sor-metrics/1" + + +def throughput_retention(result: SelectionResult) -> float: + """Fraction of dropped circuits that were successfully rebuilt (a proxy for + throughput retained under churn). 1.0 when every drop was healed; lower when + drops were deferred for lack of a live pool. No drops -> full retention.""" + if result.drops <= 0: + return 1.0 + healed = result.drops - result.deferred + return max(0.0, min(1.0, healed / result.drops)) + + +def rebuild_classifier_auc( + churned_gaps: Sequence[float], baseline_gaps: Sequence[float] +) -> float: + """AUC separating a churned run's rebuild-interval signal from a low-churn + baseline's — the RQ3 rebuild-pattern classifier. ≈1 when the two regimes are + cleanly separable, ≈0.5 when indistinguishable. Calibrated on labeled control + signals, not fit to confirmatory data.""" + # Shorter gaps (more frequent rebuilds) mark the churned regime; score churned + # as the positive class on the negated gap so larger score = more churn. + pos = [-g for g in churned_gaps] + neg = [-g for g in baseline_gaps] + return auc(pos, neg) + + +def compute_metrics( + *, + sender_counts: Dict[str, int], + ingress: Sequence[Sequence[float]], + egress: Sequence[Sequence[float]], + selection: SelectionResult, + churned_gaps: Optional[Sequence[float]] = None, + baseline_gaps: Optional[Sequence[float]] = None, +) -> Dict[str, Any]: + """Aggregate the four DV families into a metrics dict (not yet written).""" + metrics: Dict[str, Any] = { + "schema": METRICS_SCHEMA, + "seed": selection.seed, + "selector_strategy": selection.strategy, + "hops": selection.hops, + "rq1_bridge_correlation_auc": bridge_correlation_auc(ingress, egress), + "rq2_anonymity_entropy_bits": shannon_entropy_bits(sender_counts), + "rq2_sender_count": len([c for c in sender_counts.values() if c > 0]), + "rq3_throughput_retention": throughput_retention(selection), + "rq3_drops": selection.drops, + "rq3_rebuilds": len(selection.rebuilds), + "rq3_deferred": selection.deferred, + "rq3_every_drop_rebuilt": selection.every_drop_rebuilt, + } + if churned_gaps is not None and baseline_gaps is not None: + metrics["rq3_rebuild_classifier_auc"] = rebuild_classifier_auc( + churned_gaps, baseline_gaps + ) + return metrics + + +def _validate(metrics: Dict[str, Any]) -> None: + required = ( + "schema", + "rq1_bridge_correlation_auc", + "rq2_anonymity_entropy_bits", + "rq3_throughput_retention", + "rq3_every_drop_rebuilt", + ) + missing = [k for k in required if k not in metrics] + if missing: + raise ValueError(f"metrics schema: missing keys {missing}") + if metrics["schema"] != METRICS_SCHEMA: + raise ValueError(f"metrics schema: unexpected {metrics['schema']!r}") + + +def write_metrics(run_dir: Path, metrics: Dict[str, Any]) -> Path: + """Write ``metrics.json`` into ``run_dir`` (created if needed) and return its + path. Refuses to overwrite an existing metrics.json (write-once artifact).""" + _validate(metrics) + run_dir = Path(run_dir) + run_dir.mkdir(parents=True, exist_ok=True) + path = run_dir / "metrics.json" + if path.exists(): + raise FileExistsError(f"metrics.json already exists (immutable): {path}") + path.write_text(json.dumps(metrics, sort_keys=True, indent=2) + "\n", encoding="utf-8") + return path diff --git a/cmd_chat/sor/churn.py b/cmd_chat/sor/churn.py new file mode 100644 index 0000000..37e997b --- /dev/null +++ b/cmd_chat/sor/churn.py @@ -0,0 +1,71 @@ +"""R7 — Seeded churn schedule (deterministic node kill/spawn stream). + +The churn generator produces a **schedule** — a reproducible list of kill/spawn +events drawn from the R1 ``Domain.CHURN`` stream — that models nodes dropping out +of and rejoining the grid over time. It is pure data: this module spins and kills +no real VM (that live half runs against the isolated hackhouse VM fabric and is +gated by the same containment law as the R4 forwarder). Producing the schedule +here, deterministically from the seed, is what lets the R7 acceptance check assert +that a fixed churn seed drives the selector to rebuild *every* dropped circuit — +verifiable entirely offline. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import List + +from cmd_chat.sor.config import Domain, SorRng + + +@dataclass(frozen=True) +class ChurnEvent: + """One scheduled grid event at logical step ``t``. ``kind`` is ``"kill"`` or + ``"spawn"``; ``node`` is the affected node id.""" + + t: int + kind: str # "kill" | "spawn" + node: str + + +def churn_schedule( + seed: int, + nodes: List[str], + steps: int, + kill_prob_pct: int = 30, +) -> List[ChurnEvent]: + """Deterministically build a churn schedule over ``nodes`` for ``steps`` logical + steps, drawing from the seed's CHURN stream alone (so the same seed yields the + same schedule — the R7 determinism the selector check relies on). + + At each step every currently-live node may be killed with probability + ``kill_prob_pct``%, and every currently-dead node is respawned with the same + probability. Events are emitted in a stable (step, node) order.""" + if not nodes or steps <= 0: + return [] + s = SorRng(seed).stream(Domain.CHURN) + live = {n: True for n in nodes} + events: List[ChurnEvent] = [] + for t in range(steps): + for n in nodes: # stable order -> stable schedule + roll = s.next_below(100) + if live[n]: + if roll < kill_prob_pct: + live[n] = False + events.append(ChurnEvent(t, "kill", n)) + else: + if roll < kill_prob_pct: + live[n] = True + events.append(ChurnEvent(t, "spawn", n)) + return events + + +def live_nodes_at(nodes: List[str], schedule: List[ChurnEvent], t: int) -> List[str]: + """The set of live nodes at (through the end of) step ``t``, replaying the + schedule from the all-live initial state. Deterministic.""" + live = {n: True for n in nodes} + for ev in schedule: + if ev.t > t: + break + live[ev.node] = ev.kind == "spawn" + return [n for n in nodes if live[n]] diff --git a/cmd_chat/sor/selector.py b/cmd_chat/sor/selector.py new file mode 100644 index 0000000..98fff0f --- /dev/null +++ b/cmd_chat/sor/selector.py @@ -0,0 +1,171 @@ +"""R7 — Path selector that rebuilds circuits under churn. + +The selector consumes a churn schedule (``churn.py``) and, whenever a live circuit +loses a node to a kill, **rebuilds** it from the currently-live pool. Three +strategies, all deterministic and offline: + + * ``static`` — canonical-order pick (stable, no randomness); + * ``random`` — seeded PATH-stream pick from the live pool; + * ``agent`` — a *local* stability heuristic (prefers nodes that have churned + least). The paid frontier-model agent arm (GOAL autonomy envelope (c)) is + **human-gated and intentionally NOT wired here** — this strategy spends + nothing and calls no external model; it is the offline stand-in used to + validate the rebuild loop. + +The R7 acceptance predicate this satisfies: under a fixed churn seed the selector +rebuilds every dropped circuit (as long as the live pool can supply ``hops`` +nodes). Rebuild activity is emitted as R3 ``rebuild_start``/``rebuild_done`` + +``churn_kill``/``churn_spawn`` events when an :class:`EventLog` is supplied. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, List, Optional + +from cmd_chat.sor.churn import ChurnEvent +from cmd_chat.sor.config import Domain, SorRng +from cmd_chat.sor.events import EventLog + +STRATEGIES = ("static", "random", "agent") + + +@dataclass +class Rebuild: + """One rebuild triggered by a drop: the step, the node that dropped, and the + replacement circuit (node ids in order).""" + + t: int + dropped: str + circuit: List[str] + + +@dataclass +class SelectionResult: + """Outcome of replaying a churn schedule under a selector strategy.""" + + strategy: str + seed: int + hops: int + initial_circuit: List[str] + rebuilds: List[Rebuild] = field(default_factory=list) + drops: int = 0 + deferred: int = 0 # drops that could not be rebuilt (pool < hops at that step) + events_sha256: Optional[str] = None + + @property + def every_drop_rebuilt(self) -> bool: + """True iff every drop that broke the circuit was met by a rebuild — the + R7 acceptance predicate (no deferred/unhealed drops).""" + return self.deferred == 0 and len(self.rebuilds) == self.drops + + +class Selector: + """Picks a ``hops``-length circuit from a live pool per its strategy. Pure and + deterministic; the same (seed, strategy, live pool, churn count) always yields + the same circuit.""" + + def __init__(self, strategy: str, seed: int, hops: int) -> None: + if strategy not in STRATEGIES: + raise ValueError(f"unknown selector strategy {strategy!r} (of {STRATEGIES})") + self.strategy = strategy + self.seed = seed + self.hops = hops + self._draws = 0 # advances the random stream across rebuilds + self._kill_count: Dict[str, int] = {} # node -> times killed (agent signal) + + def note_kill(self, node: str) -> None: + self._kill_count[node] = self._kill_count.get(node, 0) + 1 + + def build(self, live: List[str]) -> Optional[List[str]]: + """Return a ``hops``-length circuit from ``live``, or ``None`` if the pool + is too small (fewer than ``hops`` live nodes).""" + if len(live) < self.hops: + return None + if self.strategy == "static": + return sorted(live)[: self.hops] + if self.strategy == "random": + pool = sorted(live) + s = SorRng(self.seed).stream(Domain.SELECTOR) + for _ in range(self._draws): # replay to current position (determinism) + s.next_u64() + picked: List[str] = [] + work = list(pool) + for _ in range(self.hops): + j = s.next_below(len(work)) + picked.append(work.pop(j)) + self._draws += 1 + return picked + # agent: local stability heuristic — fewest kills first, ties by id. No + # external model, no spend (the paid arm is human-gated, not wired here). + ranked = sorted(live, key=lambda n: (self._kill_count.get(n, 0), n)) + return ranked[: self.hops] + + +def run_selection( + seed: int, + nodes: List[str], + hops: int, + schedule: List[ChurnEvent], + strategy: str = "static", + log: Optional[EventLog] = None, +) -> SelectionResult: + """Replay ``schedule`` over ``nodes`` under ``strategy`` and rebuild the circuit + whenever a kill drops one of its nodes. Deterministic from the inputs. + + Returns a :class:`SelectionResult`; ``result.every_drop_rebuilt`` is the R7 + acceptance predicate. If ``log`` is given, emits R3 churn/rebuild events.""" + sel = Selector(strategy, seed, hops) + live = {n: True for n in nodes} + live_list = [n for n in nodes if live[n]] + initial = sel.build(live_list) or [] + circuit: List[str] = list(initial) + result = SelectionResult(strategy, seed, hops, list(initial)) + + for ev in schedule: + if ev.kind == "kill": + live[ev.node] = False + sel.note_kill(ev.node) + if log is not None: + log.emit("churn_kill", node_fp=_fp(ev.node), hop_index=ev.t) + if ev.node in circuit: + # The live circuit lost a hop -> must rebuild. + result.drops += 1 + if log is not None: + log.emit("rebuild_start", circuit_id=f"t{ev.t}", hop_index=ev.t) + rebuilt = sel.build([n for n in nodes if live[n]]) + if rebuilt is None: + result.deferred += 1 # pool too small right now + circuit = [] + else: + circuit = rebuilt + result.rebuilds.append(Rebuild(ev.t, ev.node, list(rebuilt))) + if log is not None: + log.emit("rebuild_done", circuit_id=f"t{ev.t}", + hop_index=len(rebuilt)) + else: # spawn + live[ev.node] = True + if log is not None: + log.emit("churn_spawn", node_fp=_fp(ev.node), hop_index=ev.t) + if not circuit: + # A deferred drop can now be healed once the pool recovers. + rebuilt = sel.build([n for n in nodes if live[n]]) + if rebuilt is not None: + circuit = rebuilt + result.deferred = max(0, result.deferred - 1) + result.rebuilds.append(Rebuild(ev.t, ev.node, list(rebuilt))) + if log is not None: + log.emit("rebuild_done", circuit_id=f"t{ev.t}", + hop_index=len(rebuilt)) + + if log is not None: + result.events_sha256 = log.close() + return result + + +def _fp(node: str) -> str: + """A short, stable fingerprint of a node id for event metadata (never the id + verbatim in case ids ever carry structure).""" + import hashlib + + return hashlib.sha256(node.encode("utf-8")).hexdigest()[:8] diff --git a/tests/test_sor_selector.py b/tests/test_sor_selector.py new file mode 100644 index 0000000..bf3b19c --- /dev/null +++ b/tests/test_sor_selector.py @@ -0,0 +1,138 @@ +"""R7 (live halves) — churn schedule + selector rebuild + metrics.json. + +Acceptance predicate (roadmap R7, the traffic-side half): under a fixed churn +seed the selector rebuilds every dropped circuit, and metrics.json is produced. +(The offline detector calibration halves — correlator AUC≈1/0.5 and entropy +log2(N) — are covered in test_sor_analysis.py.) + +All deterministic and offline: the churn schedule is seeded data (no real VM +spin/kill), the selector replays it, and the metrics writer aggregates the DVs. +The paid frontier-model ``agent`` arm is human-gated and NOT exercised here — the +``agent`` strategy under test is the local stability heuristic (no spend). +""" + +import json + +import pytest + +from cmd_chat.sor.analysis.detectors import synthetic_bridge_fixture +from cmd_chat.sor.analysis.metrics import compute_metrics, write_metrics +from cmd_chat.sor.churn import ChurnEvent, churn_schedule, live_nodes_at +from cmd_chat.sor.events import EventLog +from cmd_chat.sor.selector import STRATEGIES, run_selection + + +NODES = [f"n{i}" for i in range(8)] + + +# --------------------------------------------------------------------------- # +# Churn schedule — seeded + deterministic. +# --------------------------------------------------------------------------- # +def test_churn_schedule_is_seed_deterministic(): + a = churn_schedule(2024, NODES, steps=20) + b = churn_schedule(2024, NODES, steps=20) + assert a == b + assert churn_schedule(2025, NODES, steps=20) != a + + +def test_churn_schedule_has_kills_and_spawns(): + sched = churn_schedule(7, NODES, steps=40, kill_prob_pct=40) + kinds = {ev.kind for ev in sched} + assert "kill" in kinds and "spawn" in kinds + # A killed node is not live until it is spawned again. + first_kill = next(ev for ev in sched if ev.kind == "kill") + assert first_kill.node not in live_nodes_at(NODES, sched, first_kill.t) + + +def test_empty_inputs_yield_empty_schedule(): + assert churn_schedule(1, [], steps=10) == [] + assert churn_schedule(1, NODES, steps=0) == [] + + +# --------------------------------------------------------------------------- # +# Selector — rebuilds every dropped circuit under a fixed churn seed. +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("strategy", STRATEGIES) +def test_selector_rebuilds_every_dropped_circuit(strategy): + # Pool (8) comfortably exceeds hops (3), so every drop can be healed. + seed = 0xBADC0DE + sched = churn_schedule(seed, NODES, steps=60, kill_prob_pct=35) + result = run_selection(seed, NODES, hops=3, schedule=sched, strategy=strategy) + assert len(result.initial_circuit) == 3 + assert result.drops > 0 # the churn actually perturbed the circuit + assert result.every_drop_rebuilt # THE R7 acceptance predicate + assert result.deferred == 0 + assert len(result.rebuilds) == result.drops + # Every rebuilt circuit is a full, distinct-node, currently-plausible path. + for rb in result.rebuilds: + assert len(rb.circuit) == 3 + assert len(set(rb.circuit)) == 3 + + +def test_selector_is_deterministic(): + seed = 42 + sched = churn_schedule(seed, NODES, steps=30) + r1 = run_selection(seed, NODES, hops=3, schedule=sched, strategy="random") + r2 = run_selection(seed, NODES, hops=3, schedule=sched, strategy="random") + assert [rb.circuit for rb in r1.rebuilds] == [rb.circuit for rb in r2.rebuilds] + + +def test_unknown_strategy_refused(): + with pytest.raises(ValueError): + run_selection(1, NODES, hops=3, schedule=[], strategy="frontier-gpt") + + +def test_selection_emits_rebuild_events(tmp_path): + seed = 0xBADC0DE + sched = churn_schedule(seed, NODES, steps=40, kill_prob_pct=35) + log = EventLog(tmp_path, "sel-run", seed=seed) + result = run_selection(seed, NODES, hops=3, schedule=sched, + strategy="static", log=log) + assert result.events_sha256 and len(result.events_sha256) == 64 + lines = (tmp_path / "events.jsonl").read_text().strip().splitlines() + events = [json.loads(ln)["event"] for ln in lines] + assert "churn_kill" in events + assert events.count("rebuild_done") == len(result.rebuilds) + + +# --------------------------------------------------------------------------- # +# metrics.json — produced + schema-valid. +# --------------------------------------------------------------------------- # +def test_metrics_json_is_produced(tmp_path): + seed = 0xBADC0DE + sched = churn_schedule(seed, NODES, steps=60, kill_prob_pct=35) + result = run_selection(seed, NODES, hops=3, schedule=sched, strategy="static") + + ingress, egress = synthetic_bridge_fixture(0xC0FFEE, linked=True) # AUC ~1 + metrics = compute_metrics( + sender_counts={f"s{i}": 1 for i in range(8)}, # 8 equiprobable -> 3 bits + ingress=ingress, + egress=egress, + selection=result, + churned_gaps=[1.0, 1.5, 1.0, 2.0], + baseline_gaps=[10.0, 12.0, 11.0, 13.0], + ) + path = write_metrics(tmp_path / "run", metrics) + assert path.exists() + + loaded = json.loads(path.read_text()) + assert loaded["schema"] == "sor-metrics/1" + assert loaded["rq2_anonymity_entropy_bits"] == pytest.approx(3.0) + assert loaded["rq1_bridge_correlation_auc"] == pytest.approx(1.0) + assert loaded["rq3_every_drop_rebuilt"] is True + assert 0.0 <= loaded["rq3_throughput_retention"] <= 1.0 + assert loaded["rq3_rebuild_classifier_auc"] == pytest.approx(1.0) + + +def test_metrics_json_is_write_once(tmp_path): + result = run_selection(1, NODES, hops=3, schedule=[], strategy="static") + metrics = compute_metrics( + sender_counts={"a": 1, "b": 1}, + ingress=[[1, 2]], + egress=[[1, 2]], + selection=result, + ) + run = tmp_path / "run" + write_metrics(run, metrics) + with pytest.raises(FileExistsError): # immutable artifact + write_metrics(run, metrics)