R7 (live halves): seeded churn + rebuild selector + metrics.json

churn.py emits a seed-deterministic kill/spawn schedule (pure data — no real VM
spin/kill; the live fabric half stays gated by the containment law). selector.py
consumes the schedule and rebuilds a circuit whenever a kill drops one of its
hops, across static | random | agent strategies; the paid frontier-model agent
arm (GOAL envelope (c)) is human-gated and NOT wired — the agent strategy here is
a local stability heuristic that spends nothing. analysis/metrics.py aggregates
the four DV families (RQ1 correlation AUC, RQ2 entropy bits, RQ3 throughput
retention + rebuild-classifier AUC) into a schema-valid, write-once metrics.json.
Acceptance check green: under a fixed churn seed the selector rebuilds every
dropped circuit (every_drop_rebuilt, all strategies) and metrics.json is
produced. Python R7 selector suite 11 passed; full SOR suite 92 passed.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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
+pcaps, spawn no engine, move no traffic, and touch no VM fabric.
-traffic-moving R7 pieces (``churn.py`` seeded VM spin/kill; the live selector
+
-rebuild loop; the confirmatory battery that writes ``metrics.json``) are NOT
+The rest of R7 is landed alongside: ``metrics.py`` (this package) aggregates the
-here — they are HELD pending R4/R6 + a live grid (see OVERSEER-STATUS.md).
+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.
 """

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

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]]

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]

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)