R7/RQ3: executor-wiring stone — offline selector DVs + calibration gate
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 <noreply@anthropic.com>
This commit is contained in:
leetcrypt
@@ -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)
|
||||
@@ -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)."""
|
||||
|
||||
+211
-2
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user