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>
This commit is contained in:
leetcrypt
@@ -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.
|
||||
"""
|
||||
|
||||
@@ -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
|
||||
@@ -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]]
|
||||
@@ -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]
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user