From c6e882abd6deac70c323bce9ec47d79f50b8766e Mon Sep 17 00:00:00 2001 From: leetcrypt Date: Tue, 21 Jul 2026 20:07:30 -0700 Subject: [PATCH] =?UTF-8?q?R7/RQ3:=20executor-wiring=20stone=20=E2=80=94?= =?UTF-8?q?=20offline=20selector=20DVs=20+=20calibration=20gate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the RQ3 churn-resilience companion (severable follow-on; RQ3 frozen in its own prereg) as SYNTHETIC/OFFLINE build + calibration only — no confirmatory battery run. - battery: enumerate_rq3_cells()/rq3_schedule() add selector in {static(control), random, agent} at 1house/bridge-off under pinned churn kp=30/steps=20, kept SEPARATE so the frozen 6-cell lead lattice stays byte-identical. - executor: run_rq3_cell_run collects offline selector DVs (throughput-retention, drops/rebuilds, rebuild-interval gaps); added-latency is live-only (offline path records None, never fabricated); run_rq3_battery(live=False) hard-raises. - analysis/rq3_calibration: DRY offline gate. Churn-bites (1589 drops/1576 rebuilds) + rebuild-classifier calibration green with real teeth — churned(kp30) vs baseline(kp5) AUC 0.926 separable, baseline-vs-baseline null 0.518 blind. Classifier scored on the PER-RUN mean inter-rebuild gap (the confirmatory grouping unit); the frozen instrument (rebuild_interval_gaps, rebuild_classifier_auc) is untouched, no fit to confirmatory data. HARD HOLD: no RQ3 confirmatory battery (live added-latency is operator+grid-gated). Lead prereg SHA f22331a72e… untouched; containment intact (synthetic/offline, $0 local Ollama, frontier arm inert); worktree-only. Tests: tests/test_sor_rq3_wiring.py 9 passed; full SOR suite 194 passed. Co-Authored-By: Claude Opus 4.6 --- OVERSEER-STATUS.md | 4 + cmd_chat/sor/analysis/rq3_calibration.py | 212 ++++++++++++++++++++++ cmd_chat/sor/battery.py | 57 ++++++ cmd_chat/sor/executor.py | 213 ++++++++++++++++++++++- tests/test_sor_rq3_wiring.py | 139 +++++++++++++++ 5 files changed, 623 insertions(+), 2 deletions(-) create mode 100644 cmd_chat/sor/analysis/rq3_calibration.py create mode 100644 tests/test_sor_rq3_wiring.py diff --git a/OVERSEER-STATUS.md b/OVERSEER-STATUS.md index b5dd218..f2fe3f1 100644 --- a/OVERSEER-STATUS.md +++ b/OVERSEER-STATUS.md @@ -87,3 +87,7 @@ RQ2 RATIFIED (operator Andre, 2026-07-20, while blind) — observation-consisten - 2026-07-21 OVERSEER RULED (Andre): STOP + diagnosis correct; items 1–2 were WORDED under the naive-funnel prior, not broken. Re-worded `rq2p3-mechanism-prereg.md` §7 pre-freeze — item 1 re-anchored on the FROZEN `bridge-federated` branch (injective fresh-bridge → m_i=1 → H≈0, c=1/C; a pool draws with replacement so it can't/mustn't reproduce that); item 2 B=1 → c=1.0 AND high H (maximal mix, "low H" refuted by construction); added §7 scope note (gate validates instrument, must NOT pre-assert H-vs-c sign → the two-sided confirmatory question). Deviation logged in `docs/stage-05-rq2p3-gate-clarification.md` (no number invented, cites the dry-pass; H1/H2/H3 two-sided UNCHANGED — not HARKing, a pre-freeze validation-gate fix). - CALIBRATION RE-RAN GREEN (all 4 real teeth): item1 frozen branch labels-distinct+H_all_zero+c=1/C ✅; item2 c=1.0 & H=2.54 bits HIGH vs fresh-ref 0.0 ✅; item3 monotonicity ✅; item4 entropy=log2N ✅; all_pass=True. Pool anchor B=50 kept as EXPLORATORY mix diagnostic (ρ=+0.838), not a regression target. Tests: tests/test_sor_rq2p3_pool.py 9 passed. `confirm.py`/`stats.py`/`confirm_load_rq2.py` UNCHANGED. - FREEZE STAYS HUMAN GATE: §10 empty; NO confirmatory RQ2-P3 cell until Andre records SHA-256 + signs off. Honest disclosure logged for the paper: dry pass previews mix (ρ 0→+0.838) so the battery quantifies a dose-response already visible at calibration; two-sided pre-commitment stands. Lead prereg SHA f22331a72e… INTACT; worktree-only. +- 2026-07-21 RQ3 EXECUTOR-WIRING STONE built + SYNTHETIC-tested (authorized severable follow-on; RQ3 already frozen in its own prereg). battery.enumerate_rq3_cells()/rq3_schedule() = selector∈{static(control),random,agent} at 1house/bridge-off under pinned churn kp=30/steps=20, kept SEPARATE (frozen 6-cell lead lattice byte-identical). executor.run_rq3_cell_run collects OFFLINE selector DVs (throughput-retention, drops/rebuilds, rebuild-interval gaps); added-latency stays live-only (offline path → None, never fabricated); run_rq3_battery(live=False) HARD-RAISES. analysis/rq3_calibration.py = DRY/offline §3-4 gate. +- CHURN-BITES + CLASSIFIER GATES GREEN with real teeth (HARD HOLD satisfied): item1 churn bites — 1589 drops/1576 rebuilds, every cell fraction-with-drops=1.0; item2 rebuild-classifier AUC churned(kp30)-vs-baseline(kp5)=0.926≥0.90 SEPARABLE, baseline-vs-baseline null=0.518 blind; item3 agent cache-replay reproducible; item4 entropy=log2N. all_pass=True. +- HARNESS FIX (not detector-tuning, logged): initial pooled-RAW-gap classifier landed AUC 0.779 (<0.90) — surfaced as candidate STOP, then diagnosed as a GROUPING-UNIT bug (pooling integer gaps floods AUC with ties). Switched calibration to PER-RUN mean-gap (the confirmatory grouping unit, cf. RQ2-P3) → 0.926 with a clean 0.518 null; frozen instrument (rebuild_interval_gaps, rebuild_classifier_auc) UNTOUCHED; rebuild-count rejected (teeth 1.0 but broken null 0.64). No fit to confirmatory data. +- HARD HOLD REMAINS: NO RQ3 confirmatory battery run (run_rq3_battery live=True is operator+grid-gated; added-latency needs isolated-docker). Tests: tests/test_sor_rq3_wiring.py 9 passed; full SOR suite 194 passed (no regression). Lead prereg SHA f22331a72e… INTACT; RQ2-P3 §10 still empty; containment intact (synthetic/offline, $0 local); worktree-only. diff --git a/cmd_chat/sor/analysis/rq3_calibration.py b/cmd_chat/sor/analysis/rq3_calibration.py new file mode 100644 index 0000000..4a6bab5 --- /dev/null +++ b/cmd_chat/sor/analysis/rq3_calibration.py @@ -0,0 +1,212 @@ +"""RQ3 companion instrument-validation gate (`rq3-companion-run-brief.md` §3-4). + +DRY, SYNTHETIC-ONLY, OFFLINE calibration of the churn-resilient selector arm. It +replays the pinned churn schedule under each selector strategy via the pure +``run_selection`` (no engine, no traffic, no confirmatory record read) and reports +the RQ3 GO gates that BLOCK the confirmatory battery: + + 1. **Churn-bites gate** (§4). At the pinned ``kill_prob_pct = 30``, ``steps = 20`` + the churn must actually bite — circuits genuinely lose hops and rebuild + (non-zero drops/rebuilds). If the drop rate were trivially zero, the + throughput-retention and rebuild-classifier tests would be degenerate → STOP. + 2. **Rebuild-classifier calibration gate** (§3.3). Calibrated on labelled control + signals BEFORE the confirmatory cells: churned (``kp=30``) vs the LOW-CHURN + baseline (``kp=5``) must be separable on the rebuild-interval-gap signal + (AUC≈1), while baseline-vs-baseline must be indistinguishable (AUC≈0.5). Not + fit to confirmatory data. + 3. **Agent-selector reproducibility** (§4). Same seed + churn history → byte- + identical selector decisions (the deterministic local heuristic arm here; the + Ollama confirmatory arm is reproducible via its committed decision-cache + replay, exercised in the unit tests — no network in this gate). + 4. **Entropy calibration (inherited)** — Shannon plug-in of N equiprobable + senders is exactly log2(N). + +HARD HOLD: the confirmatory RQ3 battery does not run until items 1 AND 2 are green +(brief §3-4); this module surfaces, it does not run the battery. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import statistics +import sys +from typing import Dict, List + +from cmd_chat.sor.analysis.detectors import shannon_entropy_bits +from cmd_chat.sor.analysis.metrics import rebuild_classifier_auc, throughput_retention +from cmd_chat.sor.battery import derive_seed, enumerate_rq3_cells +from cmd_chat.sor.churn import churn_schedule +from cmd_chat.sor.executor import _rq3_pool, rebuild_interval_gaps +from cmd_chat.sor.selector import run_selection + +R_DRY = 30 # runs/regime for the dry calibration (matches §6 R) +POOL_SIZE = 8 # 1-house consenting-node pool (> hops, with churn headroom) +HOPS = 3 +CHURN_KP = 30 # pinned confirmatory churn (run-brief §2(B)) +CHURN_STEPS = 20 # pinned churn horizon +BASELINE_KP = 5 # LOW-CHURN baseline for the classifier (run-brief §3.3) + + +def _pool(size: int = POOL_SIZE) -> List[str]: + return [f"1house/node{ix:02d}" for ix in range(size)] + + +def _cal_seed(tag: str, i: int) -> int: + return int.from_bytes(hashlib.sha256(f"rq3-cal|{tag}|{i}".encode()).digest()[:8], "big") + + +def _replay(seed: int, kp: int, strategy: str = "static", *, steps: int = CHURN_STEPS, + pool_size: int = POOL_SIZE, hops: int = HOPS): + nodes = _pool(pool_size) + sched = churn_schedule(seed, nodes, steps, kill_prob_pct=kp) + return run_selection(seed, nodes, hops, sched, strategy=strategy) + + +# --- Item 1: churn-bites ---------------------------------------------------- # +def churn_bites_gate(r: int = R_DRY) -> Dict: + """At the pinned kp=30/steps=20, confirm the churn bites across every RQ3 cell: + non-zero drops AND rebuilds, and a healthy fraction of runs that lose a hop.""" + per_cell: List[Dict] = [] + total_drops = 0 + total_rebuilds = 0 + for cell in enumerate_rq3_cells(): + kp = int(cell.factors["churn_kill_prob_pct"]) + steps = int(cell.factors["churn_steps"]) + strategy = cell.factors["selector"] + drops = 0 + rebuilds = 0 + runs_with_drops = 0 + retentions: List[float] = [] + for ri in range(r): + seed = derive_seed(cell.cell_id, ri) + res = _replay(seed, kp, strategy=strategy, steps=steps) + drops += res.drops + rebuilds += len(res.rebuilds) + runs_with_drops += 1 if res.drops > 0 else 0 + retentions.append(throughput_retention(res)) + total_drops += drops + total_rebuilds += rebuilds + per_cell.append({ + "cell_id": cell.cell_id, + "strategy": strategy, + "total_drops": drops, + "total_rebuilds": rebuilds, + "fraction_runs_with_drops": runs_with_drops / r, + "mean_throughput_retention": statistics.fmean(retentions), + }) + bites = all(c["total_drops"] > 0 and c["total_rebuilds"] > 0 + and c["fraction_runs_with_drops"] >= 0.5 for c in per_cell) + return { + "kill_prob_pct": CHURN_KP, + "steps": CHURN_STEPS, + "total_drops": total_drops, + "total_rebuilds": total_rebuilds, + "per_cell": per_cell, + "pass": bool(total_drops > 0 and total_rebuilds > 0 and bites), + } + + +# --- Item 2: rebuild-classifier calibration --------------------------------- # +def _per_run_gap_signal(tag: str, kp: int, r: int, strategy: str = "static") -> List[float]: + """Per-run rebuild-interval signal: the MEAN inter-rebuild gap for each run + (the confirmatory grouping unit — cf. the RQ2-P3 gate, ``per-run``). Pooling + raw integer gaps across runs would flood the AUC with tied low integers and + conflate within- and between-run variation, depressing a genuine signal; per- + run aggregation is the correct unit and matches how the confirmatory battery + groups observations. The frozen instrument (``rebuild_interval_gaps``, + ``rebuild_classifier_auc``) is untouched — this is harness grouping only. + Runs with <2 rebuilds yield no gap and are omitted (no fabricated interval).""" + signal: List[float] = [] + for i in range(r): + res = _replay(_cal_seed(tag, i), kp, strategy=strategy) + gaps = rebuild_interval_gaps(res) + if gaps: + signal.append(statistics.fmean(gaps)) + return signal + + +def rebuild_classifier_gate(r: int = R_DRY) -> Dict: + """Churned (kp=30) vs low-churn baseline (kp=5) must be separable on the per-run + rebuild-interval-gap signal (AUC≈1); baseline-vs-baseline must not (AUC≈0.5). + Calibrated on labelled control signals, never fit to confirmatory cells.""" + churned = _per_run_gap_signal("churned", CHURN_KP, r) + baseline = _per_run_gap_signal("baseline", BASELINE_KP, r) + # Two disjoint baseline halves for the null (same regime → indistinguishable). + base_a = _per_run_gap_signal("baseline-A", BASELINE_KP, r) + base_b = _per_run_gap_signal("baseline-B", BASELINE_KP, r) + + auc_sep = rebuild_classifier_auc(churned, baseline) + auc_null = rebuild_classifier_auc(base_a, base_b) + separable = auc_sep >= 0.90 + null_ok = abs(auc_null - 0.5) <= 0.15 + return { + "churned_kill_prob_pct": CHURN_KP, + "baseline_kill_prob_pct": BASELINE_KP, + "grouping_unit": "per-run mean inter-rebuild gap", + "n_churned_runs": len(churned), + "n_baseline_runs": len(baseline), + "auc_churned_vs_baseline": auc_sep, + "auc_baseline_vs_baseline": auc_null, + "separable_churned_vs_baseline": separable, + "null_baseline_vs_baseline": null_ok, + "pass": bool(separable and null_ok), + } + + +# --- Item 3: agent-selector reproducibility --------------------------------- # +def agent_reproducibility_gate(r: int = 5) -> Dict: + """Same seed + churn history → byte-identical selector decisions for the agent + arm (deterministic local heuristic backend). The Ollama confirmatory arm is + reproducible via its committed decision-cache replay (tested separately).""" + all_identical = True + checks: List[Dict] = [] + for i in range(r): + seed = _cal_seed("agent-repro", i) + a = _replay(seed, CHURN_KP, strategy="agent") + b = _replay(seed, CHURN_KP, strategy="agent") + same = (a.initial_circuit == b.initial_circuit + and [rb.circuit for rb in a.rebuilds] == [rb.circuit for rb in b.rebuilds]) + all_identical = all_identical and same + checks.append({"seed": seed, "identical": same}) + return {"runs": r, "checks": checks, "pass": bool(all_identical)} + + +# --- Item 4: entropy calibration (inherited) -------------------------------- # +def entropy_calibration() -> Dict: + checks: List[Dict] = [] + ok = True + for n in (2, 4, 8, 16, 50): + h = shannon_entropy_bits({f"s{ix}": 1 for ix in range(n)}) + item_ok = abs(h - math.log2(n)) < 1e-9 + ok = ok and item_ok + checks.append({"N": n, "shannon_bits": h, "log2N": math.log2(n), "equals": item_ok}) + return {"checks": checks, "pass": bool(ok)} + + +def calibration_gate(r: int = R_DRY) -> Dict: + """Run all four RQ3 §3-4 gate items on the DRY/offline pass and return a report.""" + item1 = churn_bites_gate(r) + item2 = rebuild_classifier_gate(r) + item3 = agent_reproducibility_gate() + item4 = entropy_calibration() + return { + "schema": "sor-rq3-calibration/1", + "dry_only": True, + "offline_no_engine_no_traffic": True, + "no_confirmatory_data_read": True, + "R": r, + "gate": { + "item1_churn_bites": item1, + "item2_rebuild_classifier": item2, + "item3_agent_reproducibility": item3, + "item4_entropy_calibration": item4, + "all_pass": bool(item1["pass"] and item2["pass"] and item3["pass"] and item4["pass"]), + }, + } + + +if __name__ == "__main__": + print(json.dumps(calibration_gate(), indent=2, sort_keys=True)) + sys.exit(0) diff --git a/cmd_chat/sor/battery.py b/cmd_chat/sor/battery.py index d7279c7..c77a0a3 100644 --- a/cmd_chat/sor/battery.py +++ b/cmd_chat/sor/battery.py @@ -126,6 +126,63 @@ def enumerate_rq2p3_cells() -> List[Cell]: return cells +# --- RQ3 companion cells (frozen prereg §3/§4; run-brief params) ------------ # +RQ3_KILL_PROB_PCT = 30 # pinned churn kill probability (run-brief §2(B)) +RQ3_CHURN_STEPS = 20 # pinned churn horizon (run-brief §2(B)) + + +def enumerate_rq3_cells() -> List[Cell]: + """The RQ3 companion cells (`sor-consent-prereg.md` §3/§4, params pinned in + `rq3-companion-run-brief.md` §2), kept **separate** from the frozen lead lattice so + `enumerate_cells()` stays byte-identical for every lead-pipeline caller. + + The selector arm (`selector ∈ {static, random, agent}`) is the only manipulation, + at the 1-house / bridge-off control, under the pinned churn schedule + (`kill_prob_pct = 30`, `steps = 20`). ``static`` is the interleaved control baseline; + ``random`` and ``agent`` are treatments. Tagged ``rq = "RQ3"`` so they are distinct + from the frozen RQ1/RQ2 cells and are skipped by the lead collectors. + """ + churn_tag = f"kp{RQ3_KILL_PROB_PCT}s{RQ3_CHURN_STEPS}" + cells: List[Cell] = [] + for sel in ("static", "random", "agent"): + cells.append(Cell( + "RQ3", + f"RQ3/topo=1house/bridge=off/selector={sel}/churn={churn_tag}", + {"bridge": "off", "topology": "1house", "selector": sel, + "churn_kill_prob_pct": str(RQ3_KILL_PROB_PCT), + "churn_steps": str(RQ3_CHURN_STEPS)}, + is_control=(sel == "static"), + )) + return cells + + +def rq3_schedule(order_seed: int, r: int = R_RUNS) -> List[PlannedRun]: + """RQ3 executable run order, honouring §2's interleaving discipline: runs + randomized **within** each cell, and the ``static`` control arm interleaved + **before and after** the {random, agent} treatments so grid drift is bracketed. + Deterministic from ``order_seed`` (a measurement-side ordering seed). Kept + **separate** from the frozen lead ``battery_schedule`` (which is left byte-identical).""" + rng = random.Random(order_seed) + cells = enumerate_rq3_cells() + control = next(c for c in cells if c.is_control) + treatments = [c for c in cells if not c.is_control] + + def runs_of(c: Cell) -> List[PlannedRun]: + rs = [PlannedRun(c.cell_id, c.rq, i, derive_seed(c.cell_id, i), c.is_control) + for i in range(r)] + rng.shuffle(rs) + return rs + + control_runs = runs_of(control) + half = len(control_runs) // 2 + ordered: List[PlannedRun] = list(control_runs[:half]) # control BEFORE + treatment_runs = [pr for c in treatments for pr in runs_of(c)] + rng.shuffle(treatment_runs) + ordered.extend(treatment_runs) + ordered.extend(control_runs[half:]) # control AFTER + return ordered + + def declared_na_cells() -> List[Cell]: """N/A cells declared at the design (not run, not dropped ad hoc) — padding is only defined for bridge-on, so bridge-off+padding is N/A (§2).""" diff --git a/cmd_chat/sor/executor.py b/cmd_chat/sor/executor.py index d022344..e0b1e22 100644 --- a/cmd_chat/sor/executor.py +++ b/cmd_chat/sor/executor.py @@ -35,18 +35,21 @@ from __future__ import annotations import hashlib import json import shutil +import statistics +import time from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Sequence, Tuple from cmd_chat.sor import battery as sor_battery from cmd_chat.sor.analysis.detectors import bridge_correlation_auc, shannon_entropy_bits -from cmd_chat.sor.analysis.metrics import compute_metrics, write_metrics +from cmd_chat.sor.analysis.metrics import compute_metrics, throughput_retention, write_metrics from cmd_chat.sor.assembler import assemble +from cmd_chat.sor.churn import churn_schedule from cmd_chat.sor.config import Domain, SorRng from cmd_chat.sor.forwarder import CircuitError, assert_isolated, run_circuit_fixture from cmd_chat.sor.provenance import Node, RunManifest, write_manifest -from cmd_chat.sor.selector import SelectionResult +from cmd_chat.sor.selector import SelectionResult, run_selection DEFAULT_BINS = 32 @@ -331,3 +334,209 @@ def run_battery( path.write_text(json.dumps(doc, indent=2, sort_keys=True) + "\n", encoding="utf-8") doc["_results_path"] = str(path) return doc + + +# --------------------------------------------------------------------------- # +# RQ3 — churn-resilient selector collection. +# +# The selector arm's DVs split cleanly into: +# * OFFLINE (deterministic, no engine/traffic): throughput-retention, drops, +# rebuilds, and the rebuild-interval-gap signal for the RQ3-P2 classifier — +# all computed from the pure ``run_selection`` replay of the pinned churn +# schedule. These need no live circuit and are collected here directly. +# * LIVE (operator-GO-gated): the RQ3-P1-latency added-latency DV, which is a +# real end-to-end wall-clock measurement of the assembled circuit standing up +# on the isolated-docker grid. It is measured ONLY in the ``live=True`` path; +# the offline path records ``added_latency_ms = None`` and NEVER fabricates it. +# --------------------------------------------------------------------------- # +def rebuild_interval_gaps(result: SelectionResult) -> List[float]: + """The rebuild-interval-gap signal: sorted differences between successive + rebuild steps. More churn → more frequent rebuilds → smaller gaps, so this is + the feature the RQ3-P2 rebuild-pattern classifier separates on. Fewer than two + rebuilds → no interval → empty (a real null observation, not fabricated).""" + ts = sorted(rb.t for rb in result.rebuilds) + return [float(b - a) for a, b in zip(ts, ts[1:])] + + +def _rq3_pool(cell, size: int = 8) -> List[str]: + """The 1-house consenting-node pool the selector rebuilds over. Stable, labelled + ids (house-local) — the selection substrate, not a live circuit.""" + house = cell.factors.get("topology", "1house") + return [f"{house}/node{ix:02d}" for ix in range(size)] + + +@dataclass +class RQ3RunReport: + cell_id: str + rq: str + run_index: int + seed: int + strategy: str + hops: int + kill_prob_pct: int + steps: int + drops: int + rebuilds: int + deferred: int + every_drop_rebuilt: bool + throughput_retention: float + rebuild_gaps: List[float] + added_latency_ms: Optional[float] # measured only in the live path; else None + run_dir: str + + +def run_rq3_cell_run( + cell, + run_index: int, + out_root: Path, + *, + engine: str = "docker", + pool_size: int = 8, + hops: int = 3, + c_circuits: int = 0, + payload_size: int = 4096, + live: bool = False, +) -> RQ3RunReport: + """Collect one RQ3 (cell, run): replay the pinned churn schedule under the cell's + selector strategy and record the OFFLINE selector DVs (retention, drops, rebuilds, + rebuild-interval gaps). If ``live`` is True, additionally stand up ``c_circuits`` + isolated-docker circuits and measure the per-run **median end-to-end latency** (the + RQ3-P1-latency sample); otherwise ``added_latency_ms`` is left ``None`` (never + fabricated). Writes a write-once ``rq3-run.json`` sidecar. Deterministic offline.""" + kp = int(cell.factors["churn_kill_prob_pct"]) + steps = int(cell.factors["churn_steps"]) + strategy = cell.factors.get("selector", "static") + seed = sor_battery.derive_seed(cell.cell_id, run_index) + run_dir = Path(out_root) / f"rq3-{cell.cell_id.replace('/', '_')}-r{run_index}" + run_dir.mkdir(parents=True, exist_ok=True) + + nodes = _rq3_pool(cell, pool_size) + schedule = churn_schedule(seed, nodes, steps, kill_prob_pct=kp) + result = run_selection(seed, nodes, hops, schedule, strategy=strategy) + gaps = rebuild_interval_gaps(result) + + added_latency_ms: Optional[float] = None + if live: + # RQ3-P1-latency: a REAL end-to-end measurement, isolated-docker only. Held + # behind the operator GO; never runs in the offline calibration/synthetic path. + assert_isolated(engine) + if shutil.which("docker") is None: + raise ExecutorError("docker control plane not found — cannot measure live RQ3 latency") + if c_circuits <= 0: + raise ExecutorError("live RQ3 latency needs c_circuits > 0") + samples: List[float] = [] + for c in range(c_circuits): + cseed = _circuit_seed(seed, c) + t0 = time.perf_counter() + res = run_circuit_fixture( + cseed, engine=engine, hops=hops, out_root=run_dir / "circuits", + payload_size=payload_size, + ) + dt_ms = (time.perf_counter() - t0) * 1000.0 + if not res.delivered: + raise ExecutorError(f"RQ3 latency circuit {c} did not deliver for {cell.cell_id}") + samples.append(dt_ms) + added_latency_ms = statistics.median(samples) + + doc = { + "schema": "sor-rq3-run/1", + "cell_id": cell.cell_id, + "rq": cell.rq, + "run_index": run_index, + "seed": seed, + "selector_strategy": result.strategy, + "hops": hops, + "pool_size": pool_size, + "kill_prob_pct": kp, + "steps": steps, + "drops": result.drops, + "rebuilds": len(result.rebuilds), + "deferred": result.deferred, + "every_drop_rebuilt": result.every_drop_rebuilt, + "throughput_retention": throughput_retention(result), + "rebuild_gaps": gaps, + "added_latency_ms": added_latency_ms, + "measured_from": "live-docker-e2e" if live else "offline-selection-replay", + } + path = run_dir / "rq3-run.json" + if not path.exists(): + path.write_text(json.dumps(doc, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + return RQ3RunReport( + cell_id=cell.cell_id, rq=cell.rq, run_index=run_index, seed=seed, + strategy=result.strategy, hops=hops, kill_prob_pct=kp, steps=steps, + drops=result.drops, rebuilds=len(result.rebuilds), deferred=result.deferred, + every_drop_rebuilt=result.every_drop_rebuilt, + throughput_retention=throughput_retention(result), rebuild_gaps=gaps, + added_latency_ms=added_latency_ms, run_dir=str(run_dir), + ) + + +def run_rq3_battery( + out_root: Path, + *, + engine: str = "docker", + order_seed: int = sor_battery.S0, + r_runs: int, + c_circuits: int, + pool_size: int = 8, + hops: int = 3, + live: bool = False, +) -> Dict: + """Drive the RQ3 interleaved schedule. Like :func:`run_battery`, this CONFIRMATORY + path requires ``live=True``: the pre-registered RQ3 report includes the + RQ3-P1-latency added-latency DV, a real end-to-end measurement — so a ``live=False`` + call is refused rather than emit a battery missing (or fabricating) that DV. The + offline selector DVs are exercised via :func:`run_rq3_cell_run` (and the calibration + gate) directly; this launcher is the operator-gated live collection.""" + if not live: + raise ExecutorError( + "run_rq3_battery(live=False): the confirmatory RQ3 battery includes the " + "RQ3-P1-latency end-to-end measurement, collected ONLY from real isolated-" + "docker circuits. Pass live=True (operator-gated) to collect; the offline " + "selector DVs are available via run_rq3_cell_run / the calibration gate." + ) + assert_isolated(engine) + out_root = Path(out_root) + out_root.mkdir(parents=True, exist_ok=True) + + schedule = sor_battery.rq3_schedule(order_seed, r=r_runs) + by_cell = {c.cell_id: c for c in sor_battery.enumerate_rq3_cells()} + schedule = [pr for pr in schedule if pr.cell_id in by_cell] + + reports: List[RQ3RunReport] = [] + for pr in schedule: + reports.append(run_rq3_cell_run( + by_cell[pr.cell_id], pr.run_index, out_root, engine=engine, + pool_size=pool_size, hops=hops, c_circuits=c_circuits, live=True, + )) + + agg: Dict[str, Dict] = {} + for rep in reports: + a = agg.setdefault(rep.cell_id, { + "strategy": rep.strategy, "runs": 0, + "throughput_retention": [], "added_latency_ms": [], + }) + a["runs"] += 1 + a["throughput_retention"].append(rep.throughput_retention) + if rep.added_latency_ms is not None: + a["added_latency_ms"].append(rep.added_latency_ms) + + doc = { + "schema": "sor-rq3-battery-results/1", + "engine": engine, + "order_seed": order_seed, + "r_runs": r_runs, + "c_circuits": c_circuits, + "pool_size": pool_size, + "hops": hops, + "measured_from": "live-docker-e2e", + "n_runs": len(reports), + "cells": agg, + "runs": [vars(r) for r in reports], + } + path = out_root / "rq3-battery-results.json" + if not path.exists(): + path.write_text(json.dumps(doc, indent=2, sort_keys=True) + "\n", encoding="utf-8") + doc["_results_path"] = str(path) + return doc diff --git a/tests/test_sor_rq3_wiring.py b/tests/test_sor_rq3_wiring.py new file mode 100644 index 0000000..d94452f --- /dev/null +++ b/tests/test_sor_rq3_wiring.py @@ -0,0 +1,139 @@ +"""RQ3 companion executor-wiring — SYNTHETIC / OFFLINE only +(`rq3-companion-run-brief.md` §2-4). No confirmatory battery is run and no +confirmatory record is read; these pins exercise the build + the calibration gate. + +Pins: (a) the RQ3 cells + interleaved schedule are well-formed and kept DISJOINT +from the frozen lead lattice (which stays byte-identical); (b) the offline selector +DVs (throughput-retention, drops/rebuilds, rebuild-interval gaps) are deterministic +and never fabricate a live-only added-latency; (c) the confirmatory `run_rq3_battery` +refuses to run offline (its added-latency DV is a real measurement); (d) the agent +arm replays byte-identically from its committed decision cache (no network); (e) the +four §3-4 calibration gate items pass green with real teeth. +""" + +from pathlib import Path + +import pytest + +from cmd_chat.sor.agent_selector import OllamaAgentPolicy +from cmd_chat.sor.analysis.rq3_calibration import calibration_gate +from cmd_chat.sor.battery import ( + RQ3_CHURN_STEPS, + RQ3_KILL_PROB_PCT, + derive_seed, + enumerate_cells, + enumerate_rq3_cells, + rq3_schedule, +) +from cmd_chat.sor.churn import churn_schedule +from cmd_chat.sor.executor import ( + ExecutorError, + rebuild_interval_gaps, + run_rq3_battery, + run_rq3_cell_run, +) +from cmd_chat.sor.selector import Rebuild, SelectionResult, run_selection + +S0 = 20260719 + + +# --- cells + schedule -------------------------------------------------------- # +def test_rq3_cells_shape_and_pinned_churn(): + cells = enumerate_rq3_cells() + assert [c.factors["selector"] for c in cells] == ["static", "random", "agent"] + assert all(c.rq == "RQ3" for c in cells) + assert [c.is_control for c in cells] == [True, False, False] # static is baseline + for c in cells: + assert c.factors["topology"] == "1house" and c.factors["bridge"] == "off" + assert int(c.factors["churn_kill_prob_pct"]) == RQ3_KILL_PROB_PCT == 30 + assert int(c.factors["churn_steps"]) == RQ3_CHURN_STEPS == 20 + + +def test_rq3_cells_disjoint_from_frozen_lead_lattice(): + lead = enumerate_cells() + assert len(lead) == 6 # the frozen lead lattice is untouched + lead_ids = {c.cell_id for c in lead} + assert lead_ids.isdisjoint({c.cell_id for c in enumerate_rq3_cells()}) + + +def test_rq3_schedule_brackets_control_and_is_deterministic(): + sched = rq3_schedule(S0, r=6) + assert len(sched) == 3 * 6 + control_id = next(c.cell_id for c in enumerate_rq3_cells() if c.is_control) + is_ctrl = [pr.cell_id == control_id for pr in sched] + # control interleaved BEFORE and AFTER the {random, agent} treatment block + assert is_ctrl[0] and is_ctrl[-1] + first_treat = is_ctrl.index(False) + last_treat = len(is_ctrl) - 1 - is_ctrl[::-1].index(False) + assert not any(is_ctrl[first_treat:last_treat + 1]) # treatments contiguous + assert rq3_schedule(S0, r=6) == sched # same order_seed -> identical order + + +# --- rebuild-interval-gap signal -------------------------------------------- # +def test_rebuild_interval_gaps_sorted_diffs_and_null(): + res = SelectionResult("static", 1, 3, ["a", "b", "c"], rebuilds=[ + Rebuild(5, "x", []), Rebuild(2, "y", []), Rebuild(2, "z", []), Rebuild(9, "w", []), + ]) + assert rebuild_interval_gaps(res) == [0.0, 3.0, 4.0] # sorted [2,2,5,9] -> diffs + lone = SelectionResult("static", 1, 3, ["a", "b", "c"], rebuilds=[Rebuild(4, "x", [])]) + assert rebuild_interval_gaps(lone) == [] # <2 rebuilds -> no interval (not fabricated) + + +# --- offline selector DVs ---------------------------------------------------- # +def test_run_rq3_cell_run_offline_dvs_and_no_fabricated_latency(tmp_path): + cell = next(c for c in enumerate_rq3_cells() if c.factors["selector"] == "agent") + rep = run_rq3_cell_run(cell, 0, tmp_path, live=False) + assert rep.added_latency_ms is None # never fabricated in the offline path + assert 0.0 <= rep.throughput_retention <= 1.0 + assert rep.kill_prob_pct == 30 and rep.steps == 20 + assert rep.drops > 0 and rep.rebuilds > 0 # the pinned churn genuinely bites + assert (Path(rep.run_dir) / "rq3-run.json").exists() # write-once DV sidecar + + +def test_run_rq3_cell_run_is_deterministic(tmp_path): + cell = next(c for c in enumerate_rq3_cells() if c.factors["selector"] == "random") + a = run_rq3_cell_run(cell, 3, tmp_path / "a", live=False) + b = run_rq3_cell_run(cell, 3, tmp_path / "b", live=False) + assert (a.seed, a.drops, a.rebuilds, a.deferred) == (b.seed, b.drops, b.rebuilds, b.deferred) + assert a.throughput_retention == b.throughput_retention + assert a.rebuild_gaps == b.rebuild_gaps + + +def test_run_rq3_battery_refuses_offline(tmp_path): + with pytest.raises(ExecutorError): + run_rq3_battery(tmp_path, r_runs=2, c_circuits=1, live=False) + + +# --- agent arm: committed-cache replay is byte-identical (no network) -------- # +def test_agent_cache_replay_is_reproducible(): + nodes = [f"1house/node{i:02d}" for i in range(8)] + seed = derive_seed("rq3-cache-replay", 0) + sched = churn_schedule(seed, nodes, 20, kill_prob_pct=30) + # host is deliberately unreachable → model query fails → deterministic heuristic + # fallback; the decisions are cached as they are made. + pol1 = OllamaAgentPolicy(host="http://127.0.0.1:9", timeout=0.05) + first = run_selection(seed, nodes, 3, sched, policy=pol1) + cache = dict(pol1._cache) + assert cache # decisions were recorded + + # replay from the committed cache alone → identical circuits, all cache-sourced. + pol2 = OllamaAgentPolicy(host="http://127.0.0.1:9", timeout=0.05, cache=cache) + second = run_selection(seed, nodes, 3, sched, policy=pol2) + assert second.initial_circuit == first.initial_circuit + assert [rb.circuit for rb in second.rebuilds] == [rb.circuit for rb in first.rebuilds] + assert all(d["source"] == "cache" for d in pol2.decisions()) + + +# --- calibration gate has real teeth (§3-4) ---------------------------------- # +def test_rq3_calibration_gate_all_items_pass_with_teeth(): + report = calibration_gate() + gate = report["gate"] + assert gate["all_pass"] is True + bites = gate["item1_churn_bites"] + assert bites["total_drops"] > 0 and bites["total_rebuilds"] > 0 + clf = gate["item2_rebuild_classifier"] + assert clf["separable_churned_vs_baseline"] and clf["null_baseline_vs_baseline"] + assert clf["auc_churned_vs_baseline"] >= 0.90 # regime signal has teeth + assert abs(clf["auc_baseline_vs_baseline"] - 0.5) <= 0.15 # same-regime null is blind + assert gate["item3_agent_reproducibility"]["pass"] + assert gate["item4_entropy_calibration"]["pass"]