Audit telemetry residual tuning premise

docs/telemetry-residual-tuning-roadmap-20260714.md

@@ -0,0 +1,286 @@
+# Telemetry-conditioned residual tuning roadmap
+
+Status: **R0 COMPLETE / FAILED; R1 AND R2 CLOSED FOR THIS MODEL**.
+
+Date: 2026-07-14 (Asia/Singapore).
+
+## Research question and claim boundary
+
+The question is whether a small number of real engine observations can correct
+a simulator's task-specific error over **unmeasured configurations**, and
+whether that correction reduces the real-GPU cost of finding a high
+SLO-goodput serving configuration.
+
+The intended headline claim, if the evidence supports it, is:
+
+> An engine-state-conditioned residual model turns a simulator prediction into
+> a task-specific posterior over unmeasured serving configurations, allowing a
+> sequential tuner to reach near-oracle SLO-goodput with materially fewer
+> H20-hours than simulator-only and outcome-only tuning.
+
+Classification accuracy, simulator-error diagnosis, and telemetry overhead are
+supporting evidence.  None is an end-to-end tuning contribution by itself.
+
+The following method is closed and will not be revived under another name:
+per-candidate five-second accept/reject as the headline contribution.  The P1
+result showed only 1.426% cost reduction in the frozen `k=2` workflow.
+
+## Two models, one evaluation
+
+Both branches use the same legal candidate set, real measurements, task split,
+cost accounting, and acquisition function.
+
+### Simulator-residual branch (primary)
+
+For measured anchor `c_t` and unmeasured candidate `c'`:
+
+```text
+y_hat(c') = y_real(c_t)
+          + [y_sim(c') - y_sim(c_t)]
+          + f(state_real(c_t) - state_sim(c_t), c' - c_t, workload, SLO)
+```
+
+The simulator delta is the prior.  The learned model may correct it only with
+training-supported state/config transitions; uncertainty or distribution shift
+must shrink the correction back toward the simulator prior.
+
+### Telemetry-only branch (mandatory)
+
+```text
+y_hat(c') = y_real(c_t)
+          + g(state_real(c_t), c' - c_t, workload, SLO)
+```
+
+This branch tests whether the simulator is actually necessary.  It does not
+use a hand-authored bottleneck-to-knob rule.
+
+### Search policy
+
+Legal configurations are enumerated independently of telemetry.  A generic
+cost-aware acquisition rule ranks candidates from predicted improvement,
+uncertainty, and measured H20 cost.  The current production harness's
+bottleneck scores, topology-first ordering, and hand-set relief constants are
+not consumed by either branch.  The validator may enforce legality,
+full-config no-repeat, failure accounting, and resource caps only.
+
+## Hypotheses
+
+| ID | Hypothesis | Direct test | Failure meaning |
+|---|---|---|---|
+| H0 | Existing artifacts can express a common, direct-measurement state without heuristic labels. | Engine/simulator extractor coverage and invariants. | Route is not currently implementable. |
+| H1 | Simulator errors are predictable from engine/simulator state discrepancy at measured anchors. | Task-held-out pairwise inversion correction and new-inversion rate. | Telemetry is diagnostic but cannot correct the surface. |
+| H2 | Telemetry alone predicts useful config transitions beyond outcome-only history. | Telemetry-only versus real-outcome-only sequential replay. | Direct telemetry-guided tuning has no independent value. |
+| H3 | Residual correction changes actual tuning decisions and cost. | H20-hours to 95% oracle and regret AUC against the strongest safe baseline. | No system contribution even if H1/H2 prediction metrics improve. |
+
+## Common-state contract
+
+Only directly observed or exactly reconstructed quantities are admitted.
+
+| Quantity | vLLM Layer-1 | Frontier | R0 status |
+|---|---|---|---|
+| Scheduled requests / batch size | Per scheduler step | Existing per-batch metric, disabled in P1 output | Common after CPU replay |
+| Scheduled prefill/decode tokens | Per scheduler step | Existing per-batch metrics | Common after CPU replay |
+| Scheduler/batch rate | Monotonic step timestamps | Batch count / simulated duration | Common after CPU replay |
+| Waiting queue area | Time-weighted queue gauge | Sum of request waiting times | Common aggregate |
+| Running request area | Time-weighted running gauge | Sum of E2E minus waiting time | Common aggregate, semantics audited |
+| Preemption count | Per step | Per request | Common |
+| KV usage/headroom | Exact blocks and ratio | Not in committed output | Engine-only until exact reconstruction exists |
+| CUDA graph mode/padding | Exact per step | Not modeled | Engine-only omitted-mechanism signal |
+| Request TTFT/TPOT/pass rate | Exact real outcomes | Exact simulated request metrics | Common outcome, not state |
+
+Unavailable fields remain null.  They cannot be imputed from a human
+`prefill/decode/queueing` label.
+
+Frontier already contains the required detailed batch and timestamped
+stage-batch ledger output.  P1 disabled it for artifact size.  R0 replays the
+same immutable fixtures with the existing output flags enabled; it does not
+change the simulator model or calibration.
+
+## Data separation
+
+- Phase 6 / `chat_w20260311_1000`: development only.
+- P1 / `chat_w20260312_1000`: development only.
+- R1 / `chat_w20260313_1000`: new development surface.
+- R2: trace windows not used for feature, model, threshold, candidate-space,
+  cutoff, or acquisition decisions.
+- Splits are by complete workload/SLO task.  Anchor- or pair-level random
+  splits are prohibited.
+- Sequential-policy seeds measure algorithmic variability; they are not
+  counted as independent system tasks.
+
+The two existing development tasks have an important limitation: the now-
+available SLO-gated simulator reading already retains the real oracle at its
+top rank/tie.  They therefore cannot establish a positive end-to-end ranking
+claim.  They are used for plumbing, known false-feasible cases, and negative
+evidence.  R1 must be run as an unbiased complete surface, not selected after
+observing simulator success or failure.
+
+## Step-by-step roadmap
+
+### R0.1 — Inventory and roadmap
+
+Deliverables:
+
+- this roadmap;
+- rolling untracked `ONGOING.md`;
+- exact engine/simulator field and artifact inventory.
+
+Gate: every claimed input has an authoritative file path and provenance.
+
+### R0.2 — Common-state plumbing
+
+Deliverables:
+
+- `runs/telemetry-residual/common_state.py`;
+- synthetic correctness tests;
+- one exact P1 Frontier replay with individual batch metrics and the full
+  stage-batch ledger enabled;
+- paired engine/simulator state summary for the same fixture.
+
+Gate:
+
+- replay request count and SLO scorer exactly agree with the committed replay;
+- batch/ledger outputs are non-empty;
+- all counters are non-negative, ratios bounded, times monotonic;
+- no GPU is visible to Frontier;
+- output volume is practical before expanding to twelve replays.
+
+### R0.3 — Development residual/headroom audit
+
+Use all frozen P1 primary fixtures and corresponding engine intervals.  Produce:
+
+- common-state residuals per anchor;
+- simulator-error labels and continuous SLO/goodput residuals;
+- ordered source/target diagnostic that removes both config identities from
+  both roles in every training fold;
+- oracle upper bound for cross-candidate correction;
+- explicit comparison with simulator+outcome and telemetry-only features.
+
+R0 is a feasibility gate, not headline evidence.  Proceed to R1 only if:
+
+1. state features are collected with the measured source anchor, vary across
+   cells, and are available before any target config is evaluated;
+2. at least one known simulator error has a state discrepancy not exposed by
+   the matched external prefix outcome;
+3. a prior-preserving model can correct development errors without introducing
+   a larger number of new errors under regularization sensitivity;
+4. an oracle cross-candidate correction has at least 15% sequential tuning-cost
+   headroom under full startup/warm-up accounting.
+
+### R0 result and decision
+
+R0 completed without a data-validity red flag, but failed condition 3.  The
+decision is **STOP_BEFORE_R1**; no H20 job was launched for this route.
+
+- All 12 detailed Frontier CPU replays exactly reproduced their committed SLO
+  scorers.  Runtime was 23.943--54.786 seconds per replay, detailed artifacts
+  were 4.12--13.53 MB, CUDA visibility was empty, and there were zero failures.
+- The paired surface contains 12 real/sim anchors, two known simulator
+  false-feasible anchors, and 120 legal cross-config ordered transitions.  A
+  fold removes both the source and target TP/MNS identity from source and
+  target roles; the two offered-load anchors remain part of the same task.
+- Raw Frontier feasibility is 83.33% on the repeated transition view.  The
+  structurally correct hybrid model uses
+  `r_target = r_source + delta_r`; the direct model uses
+  `y_target = y_source + delta_y` and never reads simulator fields.
+- Direct telemetry is not robust relative to real-outcome-only: its accuracy
+  delta over L2 `{0.1,1,10,100}` is `{-0.83,+1.67,0,-4.17}` percentage points,
+  and its best absolute accuracy is 54.17%, below the raw simulator's 83.33%.
+- Hybrid telemetry raises classification accuracy over the corresponding
+  simulator+outcome transition regression by 1.67--4.17 percentage points,
+  but worsens pass-rate RMSE by 0.141--0.201 and MAE by 0.084--0.125.  Its full
+  correction reaches only 46.67--53.33% absolute accuracy.
+- Across 24 nonzero `(L2, raw-simulator-prior weight)` combinations, no model
+  both corrects an existing simulator error without more new errors and avoids
+  worsening RMSE/MAE.  Whenever a correction fixes at least one error, it
+  corrupts at least 11 previously correct transitions.
+- A perfect correction could skip the frozen simulator rank-2 real final and
+  save 0.043469 H20-hours: 15.45% of the prospective online `k=2` cost, or
+  14.40% when the prior failed launch is charged.  On this development task the
+  simulator top-1 already is the real oracle with zero regret, so headroom
+  versus the observed-safe top-1 baseline is 0%.
+
+The result does not prove that engine telemetry is useless.  It shows that the
+current one-task anchor-transition evidence cannot support either a safe
+simulator-residual tuner or a simulator-free telemetry tuner.  A larger model
+or an R1 run would add capacity/data after a failed gate and is therefore not
+authorized under this roadmap.
+
+### R1 — New development surface
+
+Status: **NOT LAUNCHED; CLOSED BY R0**.
+
+Frozen starting setup:
+
+- host: dash0, eight NVIDIA H20 GPUs;
+- cells run solo; no co-location for SLO verdicts;
+- patched vLLM 0.24.1.dev3, Qwen3-30B-A3B BF16;
+- trace: `chat_w20260313_1000`;
+- output tokens: exactly 128;
+- SLO: stepped TTFT 2/4/6 seconds, TPOT 50 ms, pass rate at least 0.95;
+- config surface: TP `{1,2,4}` × MNS `{8,16,32,64}`;
+- hard campaign cap: 4 H20-hours.
+
+The load ladder, repetitions, randomized order, exact commands, expected wall
+time, and artifact paths are frozen only after R0.  A resolved echo is required
+before launch.
+
+R1 passes only if a frozen sequential replay shows at least 15% E2E H20-hour
+headroom over the strongest safe baseline with final regret at most 5%.  R1 is
+development evidence and cannot be reported as the held-out result.
+
+### R2 — Held-out sequential tuning
+
+Status: **NOT LAUNCHED; CLOSED BY R0**.
+
+Required baselines:
+
+1. random search;
+2. real-outcome-only Bayesian/sequential search;
+3. Frontier ranking plus real top-k final;
+4. simulator plus real-outcome residual;
+5. telemetry-only transition tuner;
+6. simulator plus telemetry residual tuner;
+7. complete real surface as oracle, not as a cost competitor.
+
+Primary metric: end-to-end H20-hours to first reach 95% of the real full-surface
+SLO-goodput oracle.  Secondary metrics are cost-normalized regret AUC, final
+regret at fixed budgets, oracle false-prune, wall time, and per-task regressions.
+
+The route is successful only if the winning telemetry method reduces the
+primary cost by at least 20% versus the strongest safe baseline and ends within
+5% regret on every headline task.  If hybrid beats telemetry-only by at least
+10%, simulator residual correction is the primary method.  If telemetry-only
+is within 5% or better, the simulator dependency is removed.  If neither clears
+the contribution bar, the route is closed and telemetry remains a diagnostic
+facility only.
+
+## Cost discipline
+
+- R0 simulator work is CPU-only and must set empty CUDA visibility.
+- R1 cannot exceed 4 H20-hours.
+- R2 receives no budget until R1 passes.
+- Startup, warm-up, burn-in, failed launches, real probes, continuation, and
+  final validation are charged.  Benchmark-only annotation repeats are
+  reported separately and cannot disappear from campaign accounting.
+
+## Final R0 sanity block
+
+| Data | n | Min | Max | Distinct | Checked invariant |
+|---|---:|---:|---:|---:|---|
+| Phase 6 cells | 12 | TP1/MNS8 | TP4/MNS64 | 12 | Surface not identical; solo SLO tier authoritative |
+| Phase 6 Layer-1 primary steps | 37 streams | 343 | 12,103 | 37 | Contiguous; zero drops |
+| P1 primary anchors | 12 | infeasible | feasible | 2 labels | 7 feasible / 5 infeasible |
+| P1 Frontier runtime | 12 | 24.093 s | 54.575 s | 12 | CPU-only; zero failures |
+| Detailed Frontier replay runtime | 12 | 23.943 s | 54.786 s | 12 | Exact committed scorers; CUDA hidden |
+| Detailed artifact bytes | 12 | 4,123,724 | 13,527,776 | 12 | Non-negative; practical CPU replay size |
+| Cross-config transitions | 120 | real pass 0.1067 | real pass 1.0 | 6 outcomes | Both endpoint config identities held out |
+| State residual vectors | 12 | 16 fields | 16 fields | 12 vectors | Finite; no missing common field |
+| R0 E2E cost values | 4 | 0.237914 | 0.301935 H20-h | 4 | Non-negative; `k=1/2`, online/conservative |
+
+Checked invariants: non-negative counts and costs; pass rates in `[0,1]`;
+simulator results not all identical; exact request count/hash agreement; Layer-1
+step continuity and zero drops; no co-resident SLO measurements; no calibration
+or evaluation split reuse for a future headline claim.  No current red flag
+invalidates R0 plumbing.  The R0 tuning gate itself failed because safe
+prior-preserving correction was absent.

runs/telemetry-residual/analyze_p1_state.py

@@ -0,0 +1,174 @@
+#!/usr/bin/env python3
+"""Pair P1 engine intervals with detailed Frontier state summaries."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import sys
+from pathlib import Path
+from typing import Any
+
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
+
+from common_state import load_jsonl, numeric, residual, summarize_engine  # noqa: E402
+
+
+def atomic_json(path: Path, payload: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    temporary.replace(path)
+
+
+def adjudicated_labels(path: Path) -> dict[tuple[str, str], bool]:
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    return {
+        (row["cell"], row["level"]): bool(row["adjudicated_feasible"])
+        for row in payload["examples"]
+    }
+
+
+def state_runs(root: Path) -> list[Path]:
+    candidates = list(root.glob("*/result.json"))
+    if (root / "result.json").is_file():
+        candidates.append(root / "result.json")
+    return sorted(
+        path
+        for path in candidates
+        if json.loads(path.read_text(encoding="utf-8")).get("schema")
+        == "telemetry-residual-frontier-state-result-v1"
+    )
+
+
+def execute(args: argparse.Namespace) -> dict[str, Any]:
+    labels = adjudicated_labels(args.pilot_metrics)
+    examples = []
+    red_flags = []
+    for state_result_path in state_runs(args.sim_state_root):
+        run_root = state_result_path.parent
+        manifest = json.loads((run_root / "run_manifest.json").read_text(encoding="utf-8"))
+        cell = manifest["entry"]["cell"]
+        role = manifest["entry"]["role"]
+        level = "low" if role.startswith("low") else "high"
+        real_run = args.real_root / "cells" / cell / f"{level}-rep1"
+        real_result = json.loads((real_run / "result.json").read_text(encoding="utf-8"))
+        stream_paths = sorted((args.real_root / "cells" / cell / "opprof").glob("*.jsonl"))
+        if len(stream_paths) != 1:
+            raise ValueError(f"expected one engine stream for {cell}, found {len(stream_paths)}")
+        engine = summarize_engine(
+            load_jsonl(stream_paths[0]),
+            start_ns=int(real_result["interval"]["start_mono_ns"]),
+            end_ns=int(real_result["interval"]["end_mono_ns"]),
+            request_count=int(real_result["selection"]["count"]),
+        )
+        simulator = json.loads((run_root / "common-state.json").read_text(encoding="utf-8"))
+        scorer = json.loads((run_root / "scorer_output.json").read_text(encoding="utf-8"))
+        if engine["interval"]["request_count"] != simulator["interval"]["request_count"]:
+            red_flags.append(f"request_count_mismatch:{cell}:{role}")
+        difference = residual(engine, simulator)
+        if any(not math.isfinite(float(value)) for value in difference["values"].values()):
+            red_flags.append(f"nonfinite_residual:{cell}:{role}")
+        real_feasible = labels[(cell, level)]
+        sim_feasible = bool(scorer["slo"]["feasible"])
+        examples.append(
+            {
+                "cell": cell,
+                "role": role,
+                "level": level,
+                "tp": int(cell.split("_")[0][2:]),
+                "mns": int(cell.split("_")[1][3:]),
+                "request_count": engine["interval"]["request_count"],
+                "real_feasible": real_feasible,
+                "sim_feasible": sim_feasible,
+                "simulator_error": sim_feasible != real_feasible,
+                "simulator_false_feasible": sim_feasible and not real_feasible,
+                "real_pass_rate_rep1": float(real_result["pass_rate"]),
+                "offered_req_s": float(real_result["selection"]["offered_req_s"]),
+                "offered_req_s_per_gpu": float(
+                    real_result["selection"]["offered_req_s_per_gpu"]
+                ),
+                "sim_pass_rate": float(scorer["slo"]["pass_rate"]),
+                "pass_rate_residual": float(real_result["pass_rate"])
+                - float(scorer["slo"]["pass_rate"]),
+                "engine": engine,
+                "simulator": simulator,
+                "state_residual": difference,
+                "paths": {
+                    "real_result": str((real_run / "result.json").resolve()),
+                    "engine_stream": str(stream_paths[0].resolve()),
+                    "simulator_result": str(state_result_path.resolve()),
+                },
+            }
+        )
+    if not examples:
+        red_flags.append("no_state_examples")
+    request_counts = [int(row["request_count"]) for row in examples]
+    pass_rate_residuals = [float(row["pass_rate_residual"]) for row in examples]
+    result = {
+        "schema": "telemetry-residual-p1-state-pairs-v1",
+        "status": "PASS" if not red_flags else "STOP",
+        "scope": "P1 development plumbing; not held-out contribution evidence",
+        "examples": examples,
+        "red_flags": red_flags,
+        "sanity": {
+            "n": len(examples),
+            "request_count": numeric(request_counts) if request_counts else None,
+            "pass_rate_residual": numeric(pass_rate_residuals)
+            if pass_rate_residuals
+            else None,
+            "simulator_errors": sum(bool(row["simulator_error"]) for row in examples),
+            "invariants": {
+                "request_counts_match": not any(
+                    flag.startswith("request_count_mismatch") for flag in red_flags
+                ),
+                "finite_residuals": not any(
+                    flag.startswith("nonfinite_residual") for flag in red_flags
+                ),
+                "ratios_bounded": all(
+                    0.0 <= row["real_pass_rate_rep1"] <= 1.0
+                    and 0.0 <= row["sim_pass_rate"] <= 1.0
+                    for row in examples
+                ),
+                "per_config_not_identical": len(set(pass_rate_residuals)) > 1
+                if len(pass_rate_residuals) > 1
+                else None,
+            },
+        },
+    }
+    atomic_json(args.output, result)
+    if result["status"] != "PASS":
+        raise RuntimeError(red_flags)
+    return result
+
+
+def parser() -> argparse.ArgumentParser:
+    result = argparse.ArgumentParser()
+    result.add_argument("--real-root", type=Path, required=True)
+    result.add_argument("--sim-state-root", type=Path, required=True)
+    result.add_argument("--pilot-metrics", type=Path, required=True)
+    result.add_argument("--output", type=Path, required=True)
+    return result
+
+
+def main() -> None:
+    result = execute(parser().parse_args())
+    print(
+        json.dumps(
+            {
+                "status": result["status"],
+                "examples": len(result["examples"]),
+                "simulator_errors": result["sanity"]["simulator_errors"],
+                "sanity": result["sanity"],
+                "red_flags": result["red_flags"],
+            },
+            sort_keys=True,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

runs/telemetry-residual/analyze_r0_gate.py

@@ -0,0 +1,292 @@
+#!/usr/bin/env python3
+"""Make the registered R0 go/no-go decision from development artifacts."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+from pathlib import Path
+from typing import Any
+
+
+def atomic_json(path: Path, payload: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    temporary.replace(path)
+
+
+def load_pass(path: Path, name: str) -> dict[str, Any]:
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if payload.get("status") != "PASS":
+        raise RuntimeError(f"{name} is not a valid PASS artifact: {path}")
+    return payload
+
+
+def reduction(reference: float, candidate: float) -> float:
+    if reference <= 0.0 or candidate < 0.0 or not math.isfinite(candidate):
+        raise ValueError("costs must be finite and non-negative with positive reference")
+    return 1.0 - candidate / reference
+
+
+def execute(args: argparse.Namespace) -> dict[str, Any]:
+    paired = load_pass(args.paired_state, "paired state")
+    transfer = load_pass(args.transfer, "transfer diagnostic")
+    e2e = load_pass(args.pilot_e2e, "P1 E2E replay")
+    red_flags = []
+
+    if len(paired.get("examples", [])) != 12:
+        red_flags.append("paired_state_not_12_anchors")
+    if transfer.get("sanity", {}).get("transitions") != 120:
+        red_flags.append("transfer_not_120_cross_config_transitions")
+    if transfer.get("red_flags"):
+        red_flags.append("transfer_has_red_flags")
+
+    examples = paired["examples"]
+    state_available = all(
+        row["state_residual"]["coverage"]["missing"] == 0
+        and row["state_residual"]["coverage"]["available"] > 0
+        for row in examples
+    )
+    state_vectors = {
+        tuple(sorted(row["state_residual"]["values"].items())) for row in examples
+    }
+    state_varies = len(state_vectors) > 1
+    error_examples = [row for row in examples if row["simulator_error"]]
+    simulator_errors = len(error_examples)
+    error_state_discrepancy = any(
+        any(
+            abs(float(value)) > 1e-12
+            for value in row["state_residual"]["values"].values()
+        )
+        for row in error_examples
+    )
+
+    simulator = transfer["simulator"]
+    prior_safe = []
+    direct_sensitivity = []
+    hybrid_incremental = []
+    for regularization, detail in transfer["regularization_sensitivity"].items():
+        for weight, models in detail["hybrid"]["prior_shrinkage"].items():
+            if float(weight) == 0.0:
+                continue
+            telemetry = models["raw_simulator_prior"][
+                "sim_plus_outcome_plus_telemetry"
+            ]
+            decision_safe = (
+                telemetry["simulator_errors_corrected"] >= 1
+                and telemetry["simulator_errors_corrected"]
+                >= telemetry["simulator_correct_corrupted"]
+            )
+            continuous_safe = (
+                telemetry["rmse"] <= simulator["rmse"] + 1e-12
+                and telemetry["mae"] <= simulator["mae"] + 1e-12
+            )
+            if decision_safe and continuous_safe:
+                prior_safe.append(
+                    {
+                        "regularization": float(regularization),
+                        "prior_weight": float(weight),
+                        "simulator_errors_corrected": telemetry[
+                            "simulator_errors_corrected"
+                        ],
+                        "simulator_correct_corrupted": telemetry[
+                            "simulator_correct_corrupted"
+                        ],
+                        "rmse": telemetry["rmse"],
+                        "mae": telemetry["mae"],
+                    }
+                )
+
+        direct = detail["direct"]
+        direct_cmp = direct["comparison"]
+        direct_sensitivity.append(
+            {
+                "regularization": float(regularization),
+                "accuracy_delta": direct_cmp["delta_telemetry_minus_baseline"][
+                    "feasibility_accuracy"
+                ],
+                "rmse_delta": direct_cmp["delta_telemetry_minus_baseline"]["rmse"],
+                "mae_delta": direct_cmp["delta_telemetry_minus_baseline"]["mae"],
+                "errors_corrected": direct_cmp["baseline_errors_corrected"],
+                "correct_corrupted": direct_cmp["baseline_correct_corrupted"],
+                "telemetry_accuracy": direct["telemetry_only"][
+                    "feasibility_accuracy"
+                ],
+            }
+        )
+        hybrid_cmp = detail["hybrid"]["comparison"]
+        hybrid_incremental.append(
+            {
+                "regularization": float(regularization),
+                "accuracy_delta": hybrid_cmp["delta_telemetry_minus_baseline"][
+                    "feasibility_accuracy"
+                ],
+                "rmse_delta": hybrid_cmp["delta_telemetry_minus_baseline"]["rmse"],
+                "mae_delta": hybrid_cmp["delta_telemetry_minus_baseline"]["mae"],
+                "errors_corrected": hybrid_cmp["baseline_errors_corrected"],
+                "correct_corrupted": hybrid_cmp["baseline_correct_corrupted"],
+            }
+        )
+
+    k1 = e2e["by_k"]["1"]["sim_top_k_plus_real_final"]
+    k2 = e2e["by_k"]["2"]["sim_top_k_plus_real_final"]
+    headroom = {
+        "interpretation": (
+            "oracle correction stops after the simulator top-1 real final instead "
+            "of evaluating the frozen safety top-2"
+        ),
+        "online": {
+            "reference_k2_h20_hours": k2["online_h20_hours"],
+            "oracle_k1_h20_hours": k1["online_h20_hours"],
+            "absolute_h20_hours": k2["online_h20_hours"] - k1["online_h20_hours"],
+            "fraction": reduction(k2["online_h20_hours"], k1["online_h20_hours"]),
+        },
+        "with_prior_failure": {
+            "reference_k2_h20_hours": k2["conservative_h20_hours_with_prior_failure"],
+            "oracle_k1_h20_hours": k1["conservative_h20_hours_with_prior_failure"],
+            "absolute_h20_hours": k2["conservative_h20_hours_with_prior_failure"]
+            - k1["conservative_h20_hours_with_prior_failure"],
+            "fraction": reduction(
+                k2["conservative_h20_hours_with_prior_failure"],
+                k1["conservative_h20_hours_with_prior_failure"],
+            ),
+        },
+        "versus_observed_safe_k1_fraction": 0.0,
+        "k1_zero_regret": k1["real_regret"] == 0.0,
+        "k2_zero_regret": k2["real_regret"] == 0.0,
+    }
+
+    condition_1 = state_available and state_varies
+    condition_2 = simulator_errors >= 1 and error_state_discrepancy
+    condition_3 = bool(prior_safe)
+    condition_4 = headroom["online"]["fraction"] >= 0.15
+    conditions = {
+        "state_available_and_varies": condition_1,
+        "known_simulator_error_has_state_discrepancy": condition_2,
+        "prior_preserving_safe_correction_exists": condition_3,
+        "oracle_online_headroom_at_least_15pct": condition_4,
+    }
+    gate_pass = not red_flags and all(conditions.values())
+    direct_incremental = all(
+        row["accuracy_delta"] >= -1e-12
+        and row["errors_corrected"] >= row["correct_corrupted"]
+        for row in direct_sensitivity
+    )
+    result = {
+        "schema": "telemetry-residual-r0-gate-v1",
+        "status": "STOP" if red_flags else "PASS",
+        "scope": "P1 development premise/headroom audit; not headline evidence",
+        "decision": "PROCEED_TO_R1" if gate_pass else "STOP_BEFORE_R1",
+        "r0_gate_pass": gate_pass,
+        "conditions": conditions,
+        "route_findings": {
+            "hybrid_prior_safe_candidates": prior_safe,
+            "hybrid_incremental_regularization": hybrid_incremental,
+            "direct_incremental_regularization": direct_sensitivity,
+            "direct_incremental_decision_signal_all_lambdas": direct_incremental,
+            "direct_best_absolute_accuracy": max(
+                row["telemetry_accuracy"] for row in direct_sensitivity
+            ),
+            "raw_simulator_accuracy": simulator["feasibility_accuracy"],
+        },
+        "headroom": headroom,
+        "red_flags": red_flags,
+        "sanity": {
+            "anchors": {
+                "n": len(examples),
+                "min": min(row["real_pass_rate_rep1"] for row in examples),
+                "max": max(row["real_pass_rate_rep1"] for row in examples),
+                "distinct_n": len(
+                    {row["real_pass_rate_rep1"] for row in examples}
+                ),
+            },
+            "state_vectors": {
+                "n": len(examples),
+                "min": min(len(row["state_residual"]["values"]) for row in examples),
+                "max": max(len(row["state_residual"]["values"]) for row in examples),
+                "distinct_n": len(state_vectors),
+            },
+            "costs_h20_hours": {
+                "n": 4,
+                "min": min(
+                    k1["online_h20_hours"],
+                    k2["online_h20_hours"],
+                    k1["conservative_h20_hours_with_prior_failure"],
+                    k2["conservative_h20_hours_with_prior_failure"],
+                ),
+                "max": max(
+                    k1["online_h20_hours"],
+                    k2["online_h20_hours"],
+                    k1["conservative_h20_hours_with_prior_failure"],
+                    k2["conservative_h20_hours_with_prior_failure"],
+                ),
+                "distinct_n": len(
+                    {
+                        k1["online_h20_hours"],
+                        k2["online_h20_hours"],
+                        k1["conservative_h20_hours_with_prior_failure"],
+                        k2["conservative_h20_hours_with_prior_failure"],
+                    }
+                ),
+            },
+            "invariants": {
+                "no_data_red_flags": not red_flags,
+                "state_nonempty_and_varied": condition_1,
+                "pass_rates_bounded": all(
+                    0.0 <= row["real_pass_rate_rep1"] <= 1.0
+                    and 0.0 <= row["sim_pass_rate"] <= 1.0
+                    for row in examples
+                ),
+                "costs_nonnegative": all(
+                    value >= 0.0
+                    for value in (
+                        k1["online_h20_hours"],
+                        k2["online_h20_hours"],
+                        k1["conservative_h20_hours_with_prior_failure"],
+                        k2["conservative_h20_hours_with_prior_failure"],
+                    )
+                ),
+                "per_config_not_identical": len(
+                    {row["real_pass_rate_rep1"] for row in examples}
+                )
+                > 1,
+            },
+        },
+    }
+    atomic_json(args.output, result)
+    if result["status"] != "PASS":
+        raise RuntimeError(red_flags)
+    return result
+
+
+def parser() -> argparse.ArgumentParser:
+    result = argparse.ArgumentParser()
+    result.add_argument("--paired-state", type=Path, required=True)
+    result.add_argument("--transfer", type=Path, required=True)
+    result.add_argument("--pilot-e2e", type=Path, required=True)
+    result.add_argument("--output", type=Path, required=True)
+    return result
+
+
+def main() -> None:
+    result = execute(parser().parse_args())
+    print(
+        json.dumps(
+            {
+                "status": result["status"],
+                "decision": result["decision"],
+                "r0_gate_pass": result["r0_gate_pass"],
+                "conditions": result["conditions"],
+                "headroom": result["headroom"],
+                "sanity": result["sanity"],
+                "red_flags": result["red_flags"],
+            },
+            sort_keys=True,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

runs/telemetry-residual/analyze_residual_transfer.py

@@ -0,0 +1,480 @@
+#!/usr/bin/env python3
+"""Development-only cross-config telemetry transfer diagnostic for P1.
+
+Each example asks whether state observed at one source anchor helps predict the
+pass rate at a different target config.  The hybrid branch predicts the
+real-minus-simulator residual; the direct branch never reads simulator state or
+outcomes.  Folds exclude both the source and target config identities.  The two
+offered-load anchors belong to one trace/SLO task, so this is a premise check
+rather than generalization evidence.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+from pathlib import Path
+from typing import Any, Sequence
+
+import numpy as np
+
+
+REGULARIZATION = (0.1, 1.0, 10.0, 100.0)
+PRIOR_SHRINKAGE = (0.0, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0)
+
+
+def atomic_json(path: Path, payload: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    temporary.replace(path)
+
+
+def finite(value: Any, *, name: str) -> float:
+    result = float(value)
+    if not math.isfinite(result):
+        raise ValueError(f"non-finite feature {name}={value!r}")
+    return result
+
+
+def flatten_residual_state(example: dict[str, Any]) -> tuple[float, ...]:
+    residual = example["state_residual"]["values"]
+    values = [finite(residual[name], name=name) for name in sorted(residual)]
+    engine_only = example["engine"]["engine_only"]
+    values.extend(
+        finite(engine_only[name], name=name)
+        for name in sorted(engine_only)
+    )
+    return tuple(values)
+
+
+def flatten_engine_state(example: dict[str, Any]) -> tuple[float, ...]:
+    values = []
+    for name in sorted(example["engine"]["common"]):
+        value = example["engine"]["common"][name]
+        if isinstance(value, dict):
+            values.extend(
+                finite(value[statistic], name=f"{name}.{statistic}")
+                for statistic in ("mean", "max", "cv")
+            )
+        elif value is not None:
+            values.append(finite(value, name=name))
+    engine_only = example["engine"]["engine_only"]
+    values.extend(
+        finite(engine_only[name], name=name)
+        for name in sorted(engine_only)
+    )
+    return tuple(values)
+
+
+def config_transition_features(
+    source: dict[str, Any], target: dict[str, Any]
+) -> tuple[float, ...]:
+    source_rate = finite(
+        source["offered_req_s_per_gpu"], name="source_offered_req_s_per_gpu"
+    )
+    target_rate = finite(
+        target["offered_req_s_per_gpu"], name="target_offered_req_s_per_gpu"
+    )
+    return (
+        math.log2(float(source["tp"])),
+        math.log2(float(source["mns"])),
+        math.log2(float(target["tp"])),
+        math.log2(float(target["mns"])),
+        math.log2(float(target["tp"]) / float(source["tp"])),
+        math.log2(float(target["mns"]) / float(source["mns"])),
+        math.log2(source_rate),
+        math.log2(target_rate),
+        math.log2(target_rate / source_rate),
+    )
+
+
+def hybrid_base_features(
+    source: dict[str, Any], target: dict[str, Any]
+) -> tuple[float, ...]:
+    return (
+        finite(source["pass_rate_residual"], name="source_pass_rate_residual"),
+        finite(source["sim_pass_rate"], name="source_sim_pass_rate"),
+        finite(target["sim_pass_rate"], name="target_sim_pass_rate"),
+    ) + config_transition_features(source, target)
+
+
+def direct_base_features(
+    source: dict[str, Any], target: dict[str, Any]
+) -> tuple[float, ...]:
+    return (
+        finite(source["real_pass_rate_rep1"], name="source_real_pass_rate"),
+    ) + config_transition_features(source, target)
+
+
+def transitions(examples: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
+    rows = []
+    for source in examples:
+        residual_state = flatten_residual_state(source)
+        engine_state = flatten_engine_state(source)
+        for target in examples:
+            # Low/high are offered-load anchors inside one workload/SLO task.
+            # Cross-load transitions are legal; same-cell transitions are
+            # excluded because R0 asks about transfer to a new configuration.
+            if source["cell"] == target["cell"]:
+                continue
+            hybrid_base = hybrid_base_features(source, target)
+            direct_base = direct_base_features(source, target)
+            rows.append(
+                {
+                    "source_cell": source["cell"],
+                    "source_level": source["level"],
+                    "target_cell": target["cell"],
+                    "target_level": target["level"],
+                    "hybrid_base": hybrid_base,
+                    "hybrid_telemetry": hybrid_base + residual_state,
+                    "direct_base": direct_base,
+                    "direct_telemetry": direct_base + engine_state,
+                    "target_residual": finite(
+                        target["pass_rate_residual"], name="target_residual"
+                    ),
+                    "target_residual_delta": finite(
+                        target["pass_rate_residual"], name="target_residual"
+                    )
+                    - finite(source["pass_rate_residual"], name="source_residual"),
+                    "target_real_pass_rate": finite(
+                        target["real_pass_rate_rep1"], name="target_real_pass_rate"
+                    ),
+                    "target_real_pass_rate_delta": finite(
+                        target["real_pass_rate_rep1"], name="target_real_pass_rate"
+                    )
+                    - finite(
+                        source["real_pass_rate_rep1"], name="source_real_pass_rate"
+                    ),
+                    "source_residual": finite(
+                        source["pass_rate_residual"], name="source_residual"
+                    ),
+                    "source_real_pass_rate": finite(
+                        source["real_pass_rate_rep1"], name="source_real_pass_rate"
+                    ),
+                    "target_sim_pass_rate": finite(
+                        target["sim_pass_rate"], name="target_sim_pass_rate"
+                    ),
+                    "target_real_feasible": bool(target["real_feasible"]),
+                    "target_sim_feasible": bool(target["sim_feasible"]),
+                }
+            )
+    return rows
+
+
+def fit_predict(
+    train_x: np.ndarray,
+    train_y: np.ndarray,
+    test_x: np.ndarray,
+    regularization: float,
+) -> np.ndarray:
+    mean = train_x.mean(axis=0)
+    std = train_x.std(axis=0)
+    std[std == 0.0] = 1.0
+    train = (train_x - mean) / std
+    test = (test_x - mean) / std
+    train = np.column_stack((np.ones(len(train)), train))
+    test = np.column_stack((np.ones(len(test)), test))
+    penalty = np.eye(train.shape[1], dtype=np.float64)
+    penalty[0, 0] = 0.0
+    weights = np.linalg.lstsq(
+        train.T @ train + regularization * penalty,
+        train.T @ train_y,
+        rcond=None,
+    )[0]
+    return test @ weights
+
+
+def grouped_predictions(
+    rows: Sequence[dict[str, Any]],
+    *,
+    feature_name: str,
+    target_name: str,
+    regularization: float,
+) -> np.ndarray:
+    predictions = np.zeros(len(rows), dtype=np.float64)
+    groups = sorted({(row["source_cell"], row["target_cell"]) for row in rows})
+    for source_cell, target_cell in groups:
+        held_out_cells = {source_cell, target_cell}
+        test_indexes = [
+            index
+            for index, row in enumerate(rows)
+            if row["source_cell"] == source_cell and row["target_cell"] == target_cell
+        ]
+        train_indexes = [
+            index
+            for index, row in enumerate(rows)
+            if row["source_cell"] not in held_out_cells
+            and row["target_cell"] not in held_out_cells
+        ]
+        if not test_indexes or not train_indexes:
+            raise ValueError(f"empty grouped fold for {source_cell}->{target_cell}")
+        train_x = np.asarray([rows[index][feature_name] for index in train_indexes])
+        train_y = np.asarray([rows[index][target_name] for index in train_indexes])
+        test_x = np.asarray([rows[index][feature_name] for index in test_indexes])
+        predictions[test_indexes] = fit_predict(
+            train_x, train_y, test_x, regularization
+        )
+    return predictions
+
+
+def metrics(rows: Sequence[dict[str, Any]], predicted_pass: np.ndarray) -> dict[str, Any]:
+    truth = np.asarray(
+        [row["target_real_pass_rate"] for row in rows], dtype=np.float64
+    )
+    real_feasible = np.asarray(
+        [row["target_real_feasible"] for row in rows], dtype=bool
+    )
+    sim_feasible = np.asarray(
+        [row["target_sim_feasible"] for row in rows], dtype=bool
+    )
+    clipped_pass = np.clip(predicted_pass, 0.0, 1.0)
+    predicted_feasible = clipped_pass >= 0.95
+    baseline_correct = sim_feasible == real_feasible
+    model_correct = predicted_feasible == real_feasible
+    return {
+        "rmse": float(np.sqrt(np.mean((predicted_pass - truth) ** 2))),
+        "mae": float(np.mean(np.abs(predicted_pass - truth))),
+        "feasibility_accuracy": float(np.mean(model_correct)),
+        "false_feasible": int(np.sum(predicted_feasible & ~real_feasible)),
+        "false_infeasible": int(np.sum(~predicted_feasible & real_feasible)),
+        "simulator_errors_corrected": int(np.sum(~baseline_correct & model_correct)),
+        "simulator_correct_corrupted": int(np.sum(baseline_correct & ~model_correct)),
+        "predicted_pass_rate": {
+            "n": len(clipped_pass),
+            "min": float(clipped_pass.min()),
+            "max": float(clipped_pass.max()),
+            "distinct_n": len(set(float(value) for value in clipped_pass)),
+        },
+    }
+
+
+def compare(
+    rows: Sequence[dict[str, Any]],
+    baseline_prediction: np.ndarray,
+    telemetry_prediction: np.ndarray,
+    baseline: dict[str, Any],
+    telemetry: dict[str, Any],
+) -> dict[str, Any]:
+    truth = np.asarray([row["target_real_feasible"] for row in rows], dtype=bool)
+    baseline_feasible = np.clip(baseline_prediction, 0.0, 1.0) >= 0.95
+    telemetry_feasible = np.clip(telemetry_prediction, 0.0, 1.0) >= 0.95
+    baseline_correct = baseline_feasible == truth
+    telemetry_correct = telemetry_feasible == truth
+    return {
+        "delta_telemetry_minus_baseline": {
+            "rmse": telemetry["rmse"] - baseline["rmse"],
+            "mae": telemetry["mae"] - baseline["mae"],
+            "feasibility_accuracy": telemetry["feasibility_accuracy"]
+            - baseline["feasibility_accuracy"],
+        },
+        "baseline_errors_corrected": int(
+            np.sum(~baseline_correct & telemetry_correct)
+        ),
+        "baseline_correct_corrupted": int(
+            np.sum(baseline_correct & ~telemetry_correct)
+        ),
+    }
+
+
+def execute(args: argparse.Namespace) -> dict[str, Any]:
+    paired = json.loads(args.paired_state.read_text(encoding="utf-8"))
+    if paired.get("status") != "PASS" or len(paired["examples"]) != 12:
+        raise RuntimeError("paired P1 state evidence is incomplete")
+    rows = transitions(paired["examples"])
+    residual_truth = np.asarray(
+        [row["target_residual"] for row in rows], dtype=np.float64
+    )
+    pass_truth = np.asarray(
+        [row["target_real_pass_rate"] for row in rows], dtype=np.float64
+    )
+    sim_pass = np.asarray(
+        [row["target_sim_pass_rate"] for row in rows], dtype=np.float64
+    )
+    simulator = metrics(rows, sim_pass)
+    sensitivity = {}
+    for regularization in REGULARIZATION:
+        hybrid_base_delta = grouped_predictions(
+            rows,
+            feature_name="hybrid_base",
+            target_name="target_residual_delta",
+            regularization=regularization,
+        )
+        hybrid_telemetry_delta = grouped_predictions(
+            rows,
+            feature_name="hybrid_telemetry",
+            target_name="target_residual_delta",
+            regularization=regularization,
+        )
+        source_residual = np.asarray(
+            [row["source_residual"] for row in rows], dtype=np.float64
+        )
+        hybrid_base_correction = source_residual + hybrid_base_delta
+        hybrid_telemetry_correction = source_residual + hybrid_telemetry_delta
+        hybrid_base_prediction = sim_pass + hybrid_base_correction
+        hybrid_telemetry_prediction = sim_pass + hybrid_telemetry_correction
+        hybrid_base = metrics(rows, hybrid_base_prediction)
+        hybrid_telemetry = metrics(rows, hybrid_telemetry_prediction)
+        prior_shrinkage = {}
+        for weight in PRIOR_SHRINKAGE:
+            prior_shrinkage[str(weight)] = {
+                "raw_simulator_prior": {
+                    "sim_plus_outcome": metrics(
+                        rows, sim_pass + weight * hybrid_base_correction
+                    ),
+                    "sim_plus_outcome_plus_telemetry": metrics(
+                        rows, sim_pass + weight * hybrid_telemetry_correction
+                    ),
+                },
+                "anchor_offset_prior": {
+                    "sim_plus_outcome": metrics(
+                        rows,
+                        sim_pass + source_residual + weight * hybrid_base_delta,
+                    ),
+                    "sim_plus_outcome_plus_telemetry": metrics(
+                        rows,
+                        sim_pass + source_residual + weight * hybrid_telemetry_delta,
+                    ),
+                },
+            }
+
+        direct_base_prediction = grouped_predictions(
+            rows,
+            feature_name="direct_base",
+            target_name="target_real_pass_rate_delta",
+            regularization=regularization,
+        )
+        direct_telemetry_prediction = grouped_predictions(
+            rows,
+            feature_name="direct_telemetry",
+            target_name="target_real_pass_rate_delta",
+            regularization=regularization,
+        )
+        source_real_pass = np.asarray(
+            [row["source_real_pass_rate"] for row in rows], dtype=np.float64
+        )
+        direct_base_prediction = source_real_pass + direct_base_prediction
+        direct_telemetry_prediction = source_real_pass + direct_telemetry_prediction
+        direct_base = metrics(rows, direct_base_prediction)
+        direct_telemetry = metrics(rows, direct_telemetry_prediction)
+        sensitivity[str(regularization)] = {
+            "hybrid": {
+                "sim_plus_outcome": hybrid_base,
+                "sim_plus_outcome_plus_telemetry": hybrid_telemetry,
+                "comparison": compare(
+                    rows,
+                    hybrid_base_prediction,
+                    hybrid_telemetry_prediction,
+                    hybrid_base,
+                    hybrid_telemetry,
+                ),
+                "prior_shrinkage": prior_shrinkage,
+            },
+            "direct": {
+                "real_outcome_only": direct_base,
+                "telemetry_only": direct_telemetry,
+                "comparison": compare(
+                    rows,
+                    direct_base_prediction,
+                    direct_telemetry_prediction,
+                    direct_base,
+                    direct_telemetry,
+                ),
+            },
+        }
+    red_flags = []
+    if any(not math.isfinite(value) for value in residual_truth):
+        red_flags.append("nonfinite_residual_target")
+    if any(not math.isfinite(value) for value in pass_truth):
+        red_flags.append("nonfinite_pass_rate_target")
+    result = {
+        "schema": "telemetry-residual-cross-config-diagnostic-v1",
+        "status": "PASS" if not red_flags else "STOP",
+        "scope": (
+            "single P1 trace/SLO-task development diagnostic; ordered transitions "
+            "are not independent tasks and cannot support a generalization claim"
+        ),
+        "split": (
+            "hold out both ordered source config and target config identities; "
+            "each fold contains both offered-load anchors"
+        ),
+        "features": {
+            "sim_plus_outcome": len(rows[0]["hybrid_base"]),
+            "sim_plus_outcome_plus_telemetry": len(rows[0]["hybrid_telemetry"]),
+            "real_outcome_only": len(rows[0]["direct_base"]),
+            "telemetry_only": len(rows[0]["direct_telemetry"]),
+        },
+        "simulator": simulator,
+        "regularization_sensitivity": sensitivity,
+        "red_flags": red_flags,
+        "sanity": {
+            "transitions": len(rows),
+            "load_levels": len(
+                {row["source_level"] for row in rows}
+                | {row["target_level"] for row in rows}
+            ),
+            "cross_load_transitions": sum(
+                row["source_level"] != row["target_level"] for row in rows
+            ),
+            "ordered_cell_pairs": len(
+                {(row["source_cell"], row["target_cell"]) for row in rows}
+            ),
+            "target_residual": {
+                "n": len(residual_truth),
+                "min": float(residual_truth.min()),
+                "max": float(residual_truth.max()),
+                "distinct_n": len(set(float(value) for value in residual_truth)),
+            },
+            "target_real_pass_rate": {
+                "n": len(pass_truth),
+                "min": float(pass_truth.min()),
+                "max": float(pass_truth.max()),
+                "distinct_n": len(set(float(value) for value in pass_truth)),
+            },
+            "invariants": {
+                "finite_targets": not red_flags,
+                "ratios_bounded": all(
+                    0.0 <= row["target_sim_pass_rate"] <= 1.0 for row in rows
+                ),
+                "source_differs_from_target": all(
+                    row["source_cell"] != row["target_cell"] for row in rows
+                ),
+                "per_config_not_identical": len(
+                    set(float(value) for value in pass_truth)
+                )
+                > 1,
+            },
+        },
+    }
+    atomic_json(args.output, result)
+    if result["status"] != "PASS":
+        raise RuntimeError(red_flags)
+    return result
+
+
+def parser() -> argparse.ArgumentParser:
+    result = argparse.ArgumentParser()
+    result.add_argument("--paired-state", type=Path, required=True)
+    result.add_argument("--output", type=Path, required=True)
+    return result
+
+
+def main() -> None:
+    result = execute(parser().parse_args())
+    print(
+        json.dumps(
+            {
+                "status": result["status"],
+                "transitions": result["sanity"]["transitions"],
+                "sensitivity": result["regularization_sensitivity"],
+                "sanity": result["sanity"],
+                "red_flags": result["red_flags"],
+            },
+            sort_keys=True,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

runs/telemetry-residual/common_state.py

@@ -0,0 +1,400 @@
+#!/usr/bin/env python3
+"""Extract directly measurable engine/simulator state into one schema.
+
+The schema deliberately keeps common, engine-only, and simulator-only fields
+separate.  Missing simulator mechanisms must remain missing; they must not be
+filled by a heuristic bottleneck label.
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import math
+from pathlib import Path
+from statistics import fmean
+from typing import Any, Iterable, Mapping, Sequence
+
+
+SCHEMA = "telemetry-common-state-v1"
+
+
+def numeric(values: Iterable[float | int]) -> dict[str, Any]:
+    finite = [float(value) for value in values]
+    if not finite:
+        raise ValueError("numeric summary requires at least one value")
+    if any(not math.isfinite(value) for value in finite):
+        raise ValueError("numeric summary received a non-finite value")
+    mean = fmean(finite)
+    variance = fmean((value - mean) ** 2 for value in finite)
+    return {
+        "n": len(finite),
+        "min": min(finite),
+        "max": max(finite),
+        "mean": mean,
+        "cv": math.sqrt(variance) / abs(mean) if mean else 0.0,
+        "distinct_n": len(set(finite)),
+    }
+
+
+def load_jsonl(path: Path) -> list[dict[str, Any]]:
+    rows = []
+    with path.open(encoding="utf-8") as source:
+        for line_number, line in enumerate(source, start=1):
+            if not line.strip():
+                continue
+            row = json.loads(line)
+            if not isinstance(row, dict):
+                raise ValueError(f"{path}:{line_number}: expected a JSON object")
+            rows.append(row)
+    if not rows:
+        raise ValueError(f"{path}: no JSONL records")
+    return rows
+
+
+def _as_number(value: Any, *, name: str) -> float:
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise ValueError(f"{name} must be numeric, got {value!r}")
+    result = float(value)
+    if not math.isfinite(result):
+        raise ValueError(f"{name} must be finite, got {value!r}")
+    return result
+
+
+def _time_weighted_mean(
+    records: Sequence[Mapping[str, Any]],
+    *,
+    start_ns: int,
+    end_ns: int,
+    value,
+) -> float:
+    if end_ns <= start_ns:
+        raise ValueError("time-weighted interval must be positive")
+    selected = [
+        record
+        for record in records
+        if start_ns <= int(record["submit_mono_ns"]) <= end_ns
+    ]
+    if not selected:
+        raise ValueError("time-weighted interval contains no records")
+    selected.sort(key=lambda record: int(record["submit_mono_ns"]))
+    cursor = start_ns
+    total = 0.0
+    current = _as_number(value(selected[0]), name="time-weighted value")
+    for record in selected[1:]:
+        timestamp = int(record["submit_mono_ns"])
+        if timestamp < cursor:
+            raise ValueError("telemetry timestamps are not monotonic")
+        total += current * (timestamp - cursor)
+        cursor = timestamp
+        current = _as_number(value(record), name="time-weighted value")
+    total += current * (end_ns - cursor)
+    return total / (end_ns - start_ns)
+
+
+def summarize_engine(
+    records: Sequence[Mapping[str, Any]],
+    *,
+    start_ns: int,
+    end_ns: int,
+    request_count: int,
+) -> dict[str, Any]:
+    """Summarize a measured engine interval from Layer-1 records."""
+
+    if request_count <= 0:
+        raise ValueError("request_count must be positive")
+    layer1 = [record for record in records if "step_index" in record]
+    if not layer1:
+        raise ValueError("engine stream has no Layer-1 records")
+    step_indexes = [int(record["step_index"]) for record in layer1]
+    if len(step_indexes) != len(set(step_indexes)):
+        raise ValueError("engine Layer-1 step indexes are not unique")
+    if step_indexes != sorted(step_indexes):
+        raise ValueError("engine Layer-1 step indexes are not ordered")
+    if any(int(record.get("dropped_records_before", 0)) != 0 for record in layer1):
+        raise ValueError("engine Layer-1 stream reports dropped records")
+
+    interval = [
+        record
+        for record in layer1
+        if start_ns <= int(record["submit_mono_ns"]) <= end_ns
+    ]
+    if not interval:
+        raise ValueError("engine interval has no Layer-1 records")
+    executed = [record for record in interval if bool(record["model_executed"])]
+    if not executed:
+        raise ValueError("engine interval has no executed model steps")
+    duration_s = (end_ns - start_ns) / 1e9
+    if duration_s <= 0:
+        raise ValueError("engine interval duration must be positive")
+
+    batch_sizes = [int(record["scheduled_requests"]) for record in executed]
+    prefill_tokens = [int(record["prefill_tokens"]) for record in executed]
+    decode_tokens = [int(record["decode_tokens"]) for record in executed]
+    batch_tokens = [
+        prefill + decode
+        for prefill, decode in zip(prefill_tokens, decode_tokens, strict=True)
+    ]
+    decode_batches = [int(record["decode_batch_size"]) for record in executed]
+    if any(value < 0 for value in batch_sizes + batch_tokens + decode_batches):
+        raise ValueError("engine batch counters must be non-negative")
+    if any(
+        int(record["prefill_tokens"]) + int(record["decode_tokens"])
+        <= 0
+        for record in executed
+    ):
+        raise ValueError("executed engine step has no scheduled tokens")
+
+    waiting_mean = _time_weighted_mean(
+        interval,
+        start_ns=start_ns,
+        end_ns=end_ns,
+        value=lambda record: record["queues"]["waiting"],
+    )
+    running_mean = _time_weighted_mean(
+        interval,
+        start_ns=start_ns,
+        end_ns=end_ns,
+        value=lambda record: record["queues"]["running"],
+    )
+    kv_mean = _time_weighted_mean(
+        interval,
+        start_ns=start_ns,
+        end_ns=end_ns,
+        value=lambda record: record["kv"]["usage"],
+    )
+    kv_values = [float(record["kv"]["usage"]) for record in interval]
+    if any(not 0.0 <= value <= 1.0 for value in kv_values):
+        raise ValueError("engine KV usage must be in [0, 1]")
+    total_prefill = sum(prefill_tokens)
+    total_decode = sum(decode_tokens)
+    graph_modes = [str(record["cudagraph"]["runtime_mode"]) for record in executed]
+    bucket_tokens = sum(int(record["cudagraph"]["bucket_tokens"]) for record in executed)
+    padding_tokens = sum(int(record["cudagraph"]["padding_tokens"]) for record in executed)
+
+    common = {
+        "scheduler_steps_per_s": len(executed) / duration_s,
+        "batch_size": numeric(batch_sizes),
+        "batch_tokens": numeric(batch_tokens),
+        "decode_batch_size": numeric(decode_batches),
+        "prefill_token_fraction": total_prefill / (total_prefill + total_decode),
+        "queue_waiting_mean": waiting_mean,
+        "queue_running_mean": running_mean,
+        "queue_waiting_time_per_request_ms": waiting_mean * duration_s * 1000.0 / request_count,
+        "queue_running_time_per_request_ms": running_mean * duration_s * 1000.0 / request_count,
+        "preemptions": sum(int(record["preemptions"]) for record in executed),
+    }
+    result = {
+        "schema": SCHEMA,
+        "source": "engine_layer1",
+        "interval": {
+            "start_ns": start_ns,
+            "end_ns": end_ns,
+            "duration_s": duration_s,
+            "request_count": request_count,
+        },
+        "common": common,
+        "engine_only": {
+            "kv_usage_mean": kv_mean,
+            "kv_usage_max": max(kv_values),
+            "kv_usage_end_minus_start": kv_values[-1] - kv_values[0],
+            "graph_none_share": graph_modes.count("NONE") / len(graph_modes),
+            "graph_full_share": graph_modes.count("FULL") / len(graph_modes),
+            "graph_padding_fraction": padding_tokens / max(1, bucket_tokens),
+        },
+        "simulator_only": {},
+        "sanity": {
+            "records": len(interval),
+            "executed_steps": len(executed),
+            "step_index_min": min(int(record["step_index"]) for record in interval),
+            "step_index_max": max(int(record["step_index"]) for record in interval),
+            "invariants": {
+                "positive_duration": duration_s > 0,
+                "positive_request_count": request_count > 0,
+                "zero_drops": True,
+                "nonnegative_counters": True,
+                "kv_bounded": True,
+                "batch_values_not_all_identical": any(
+                    summary["distinct_n"] > 1
+                    for summary in (
+                        common["batch_size"],
+                        common["batch_tokens"],
+                        common["decode_batch_size"],
+                    )
+                ),
+            },
+        },
+    }
+    return result
+
+
+def _csv_rows(path: Path) -> list[dict[str, str]]:
+    with path.open(encoding="utf-8", newline="") as source:
+        rows = list(csv.DictReader(source))
+    if not rows:
+        raise ValueError(f"{path}: CSV contains no rows")
+    return rows
+
+
+def _column(rows: Sequence[Mapping[str, str]], name: str) -> list[float]:
+    if name not in rows[0]:
+        raise ValueError(f"CSV is missing required column {name!r}")
+    values = []
+    for row in rows:
+        text = row.get(name, "")
+        if text == "":
+            raise ValueError(f"CSV column {name!r} contains an empty value")
+        values.append(_as_number(float(text), name=name))
+    return values
+
+
+def summarize_frontier(
+    *,
+    system_metrics_path: Path,
+    request_metrics_path: Path,
+    batch_metrics_path: Path | None = None,
+    ledger_path: Path | None = None,
+) -> dict[str, Any]:
+    """Summarize a Frontier run, retaining unavailable state as null."""
+
+    system = json.loads(system_metrics_path.read_text(encoding="utf-8"))
+    throughput = system["throughput_metrics"]
+    duration_s = _as_number(
+        throughput["total_duration_seconds"], name="total_duration_seconds"
+    )
+    if duration_s <= 0:
+        raise ValueError("Frontier duration must be positive")
+    request_rows = _csv_rows(request_metrics_path)
+    waiting_ms = _column(request_rows, "request_waiting_time_total")
+    e2e_ms = _column(request_rows, "request_e2e_time")
+    running_ms = [max(0.0, e2e - waiting) for e2e, waiting in zip(e2e_ms, waiting_ms, strict=True)]
+    duration_ms = duration_s * 1000.0
+    request_count = len(request_rows)
+
+    common: dict[str, Any] = {
+        "scheduler_steps_per_s": None,
+        "batch_size": None,
+        "batch_tokens": None,
+        "decode_batch_size": None,
+        "prefill_token_fraction": None,
+        "queue_waiting_mean": sum(waiting_ms) / duration_ms,
+        "queue_running_mean": sum(running_ms) / duration_ms,
+        "queue_waiting_time_per_request_ms": fmean(waiting_ms),
+        "queue_running_time_per_request_ms": fmean(running_ms),
+        "preemptions": sum(
+            int(float(row.get("request_total_preemption_count") or 0))
+            for row in request_rows
+        ),
+    }
+    batch_rows: list[dict[str, str]] = []
+    if batch_metrics_path is not None:
+        batch_rows = _csv_rows(batch_metrics_path)
+        batch_sizes = _column(batch_rows, "batch_size")
+        batch_tokens = _column(batch_rows, "batch_num_tokens")
+        prefill_tokens = _column(batch_rows, "batch_num_prefill_tokens")
+        decode_tokens = _column(batch_rows, "batch_num_decode_tokens")
+        if any(value < 0 for value in batch_sizes + batch_tokens + prefill_tokens + decode_tokens):
+            raise ValueError("Frontier batch counters must be non-negative")
+        common.update(
+            {
+                "scheduler_steps_per_s": len(batch_rows) / duration_s,
+                "batch_size": numeric(batch_sizes),
+                "batch_tokens": numeric(batch_tokens),
+                "decode_batch_size": numeric(decode_tokens),
+                "prefill_token_fraction": sum(prefill_tokens)
+                / max(1.0, sum(prefill_tokens) + sum(decode_tokens)),
+            }
+        )
+
+    ledger_rows: list[dict[str, Any]] = []
+    if ledger_path is not None:
+        ledger_rows = load_jsonl(ledger_path)
+        for row in ledger_rows:
+            start = _as_number(row["stage_start_ts"], name="stage_start_ts")
+            end = _as_number(row["stage_end_ts"], name="stage_end_ts")
+            if end < start:
+                raise ValueError("Frontier ledger has a negative stage duration")
+
+    batch_distinct = (
+        max(
+            summary["distinct_n"]
+            for summary in (
+                common["batch_size"],
+                common["batch_tokens"],
+                common["decode_batch_size"],
+            )
+        )
+        if batch_rows
+        else None
+    )
+    return {
+        "schema": SCHEMA,
+        "source": "frontier",
+        "interval": {
+            "duration_s": duration_s,
+            "request_count": request_count,
+        },
+        "common": common,
+        "engine_only": {
+            "kv_usage_mean": None,
+            "kv_usage_max": None,
+            "kv_usage_end_minus_start": None,
+            "graph_none_share": None,
+            "graph_full_share": None,
+            "graph_padding_fraction": None,
+        },
+        "simulator_only": {
+            "request_waiting_time_ms": numeric(waiting_ms),
+            "request_running_time_ms": numeric(running_ms),
+            "ledger_rows": len(ledger_rows) if ledger_path is not None else None,
+        },
+        "sanity": {
+            "request_rows": request_count,
+            "batch_rows": len(batch_rows),
+            "ledger_rows": len(ledger_rows),
+            "invariants": {
+                "positive_duration": duration_s > 0,
+                "positive_request_count": request_count > 0,
+                "nonnegative_counters": True,
+                "request_values_not_all_identical": max(
+                    numeric(waiting_ms)["distinct_n"],
+                    numeric(running_ms)["distinct_n"],
+                )
+                > 1,
+                "batch_values_not_all_identical": (
+                    batch_distinct > 1 if batch_distinct is not None else None
+                ),
+            },
+        },
+    }
+
+
+def residual(real: Mapping[str, Any], simulated: Mapping[str, Any]) -> dict[str, Any]:
+    if real.get("schema") != SCHEMA or simulated.get("schema") != SCHEMA:
+        raise ValueError("residual inputs must use the common-state schema")
+    values = {}
+    missing = []
+    for name, real_value in real["common"].items():
+        sim_value = simulated["common"].get(name)
+        if isinstance(real_value, dict):
+            if not isinstance(sim_value, dict):
+                missing.append(name)
+                continue
+            for statistic in ("mean", "max", "cv"):
+                key = f"{name}.{statistic}"
+                values[key] = float(real_value[statistic]) - float(sim_value[statistic])
+            continue
+        if real_value is None or sim_value is None:
+            missing.append(name)
+            continue
+        values[name] = float(real_value) - float(sim_value)
+    return {
+        "schema": "telemetry-state-residual-v1",
+        "values": values,
+        "missing_common_fields": sorted(missing),
+        "coverage": {
+            "available": len(values),
+            "missing": len(missing),
+            "common_field_count": len(real["common"]),
+        },
+    }

runs/telemetry-residual/run_frontier_state.py

@@ -0,0 +1,316 @@
+#!/usr/bin/env python3
+"""Replay frozen Frontier fixtures with state-detail outputs enabled.
+
+This runner does not modify the frozen fixture/config inputs.  It reuses the
+audited SimFid command and only turns on existing Frontier batch/ledger output
+flags in a separate result root.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import importlib.util
+import json
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+from typing import Any
+
+
+HERE = Path(__file__).resolve().parent
+AITUNER_ROOT = HERE.parents[1]
+sys.path.insert(0, str(HERE))
+
+from common_state import numeric, summarize_frontier  # noqa: E402, F401
+
+
+def load_module(name: str, path: Path):
+    module_root = str(path.parent.resolve())
+    if module_root not in sys.path:
+        sys.path.insert(0, module_root)
+    spec = importlib.util.spec_from_file_location(name, path)
+    if spec is None or spec.loader is None:
+        raise ImportError(path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1 << 20), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def git_capture(root: Path, *arguments: str) -> str:
+    return subprocess.run(
+        ["git", "-C", str(root), *arguments],
+        check=True,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    ).stdout
+
+
+def atomic_json(path: Path, payload: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    os.replace(temporary, path)
+
+
+def enable_state_outputs(command: list[str]) -> list[str]:
+    result = list(command)
+    disabled_ledger = "--no-metrics_config_store_frontier_stage_batch_ledger"
+    enabled_ledger = "--metrics_config_store_frontier_stage_batch_ledger"
+    enabled_batches = "--metrics_config_keep_individual_batch_metrics"
+    disabled_batches = "--no-metrics_config_keep_individual_batch_metrics"
+    if result.count(disabled_ledger) != 1 or enabled_ledger in result:
+        raise ValueError("base command must explicitly disable one full Frontier ledger")
+    result[result.index(disabled_ledger)] = enabled_ledger
+    if disabled_batches in result:
+        if result.count(disabled_batches) != 1:
+            raise ValueError("duplicate disabled individual-batch flag")
+        result[result.index(disabled_batches)] = enabled_batches
+    elif enabled_batches not in result:
+        result.append(enabled_batches)
+    if result.count(enabled_ledger) != 1 or result.count(enabled_batches) != 1:
+        raise ValueError("state-detail flags were not enabled exactly once")
+    return result
+
+
+def find_state_metrics(run_root: Path) -> dict[str, Path]:
+    patterns = {
+        "system": "frontier_metrics/**/system_metrics.json",
+        "requests": "frontier_metrics/**/request_metrics.csv",
+        "batches": "frontier_metrics/**/monolithic_batch_metrics.csv",
+        "ledger": "frontier_metrics/**/frontier_stage_batch_ledger.jsonl",
+    }
+    result = {}
+    for name, pattern in patterns.items():
+        matches = sorted(run_root.glob(pattern))
+        if len(matches) != 1:
+            raise ValueError(
+                f"{run_root}: expected exactly one {name} artifact, found {len(matches)}"
+            )
+        result[name] = matches[0]
+    if len({path.parent for path in result.values()}) != 1:
+        raise ValueError("Frontier state artifacts are not colocated")
+    return result
+
+
+def execute(args: argparse.Namespace) -> dict[str, Any]:
+    prepared = json.loads(args.prepared_manifest.read_text(encoding="utf-8"))
+    if prepared.get("status") != "PASS":
+        raise RuntimeError("prepared simulator manifest did not pass")
+    matches = [
+        entry
+        for entry in prepared["entries"]
+        if entry["cell"] == args.cell and entry["role"] == args.role
+    ]
+    if len(matches) != 1:
+        raise ValueError(
+            f"expected one prepared entry for {args.cell}/{args.role}, found {len(matches)}"
+        )
+    entry = matches[0]
+    if args.output.exists() and any(args.output.iterdir()):
+        raise FileExistsError(f"non-empty output already exists: {args.output}")
+    args.output.mkdir(parents=True, exist_ok=True)
+
+    driver = load_module(
+        "telemetry_residual_execution_driver",
+        args.replayserve_root / "runs/simfid_s2rb/results/execution_driver.py",
+    )
+    config_path = Path(entry["config"])
+    config = json.loads(config_path.read_text(encoding="utf-8"))
+    fixture_manifest_path = Path(entry["fixture_manifest"])
+    fixture = json.loads(fixture_manifest_path.read_text(encoding="utf-8"))
+    trace_path = Path(entry["frontier_csv"])
+    sidecar_path = Path(entry["sidecar"])
+    metrics_root = args.output / "frontier_metrics"
+    run_id = f"telemetry_residual_r0_{args.cell}_{args.role}"
+    knobs = config["frontier"]["knobs"]
+    base_command = driver.build_command(
+        trace_path=trace_path,
+        metrics_root=metrics_root,
+        run_id=run_id,
+        knobs=knobs,
+    )
+    driver.audit_command(base_command, knobs)
+    command = enable_state_outputs(base_command)
+    row = {
+        "hook_path": config["calibration"]["hook_path"],
+        "applied_a_tp": config["calibration"]["a_tp"],
+        "sidecar_path": str(sidecar_path),
+        "request_count": int(fixture["request_count"]),
+        "tensor_parallel_size": int(fixture["tensor_parallel_size"]),
+    }
+    environment = driver.environment_for(row)
+    if environment.get("CUDA_VISIBLE_DEVICES") != "":
+        raise ValueError("Frontier state replay must hide CUDA devices")
+    manifest = {
+        "schema": "telemetry-residual-frontier-state-run-v1",
+        "entry": {
+            "cell": args.cell,
+            "role": args.role,
+            "anchor": entry["anchor"],
+            "request_count": entry["selected_count"],
+        },
+        "inputs": {
+            "prepared_manifest": str(args.prepared_manifest.resolve()),
+            "prepared_manifest_sha256": sha256_file(args.prepared_manifest),
+            "config": str(config_path.resolve()),
+            "config_sha256": sha256_file(config_path),
+            "fixture_manifest": str(fixture_manifest_path.resolve()),
+            "fixture_manifest_sha256": sha256_file(fixture_manifest_path),
+            "frontier_csv": str(trace_path.resolve()),
+            "frontier_csv_sha256": sha256_file(trace_path),
+            "sidecar": str(sidecar_path.resolve()),
+            "sidecar_sha256": sha256_file(sidecar_path),
+        },
+        "frontier": {
+            "root": str(args.frontier_root.resolve()),
+            "git_head": git_capture(args.frontier_root, "rev-parse", "HEAD").strip(),
+            "git_status_short": git_capture(args.frontier_root, "status", "--short"),
+        },
+        "runner": {
+            "script": str(Path(__file__).resolve()),
+            "script_sha256": sha256_file(Path(__file__).resolve()),
+            "aituner_git_head": git_capture(AITUNER_ROOT, "rev-parse", "HEAD").strip(),
+            "aituner_git_status_short": git_capture(AITUNER_ROOT, "status", "--short"),
+        },
+        "environment": {
+            key: environment[key]
+            for key in (
+                "PYTHONPATH",
+                "FRONTIER_EXECUTION_TIME_SCALE",
+                "CUDA_VISIBLE_DEVICES",
+                "NVIDIA_VISIBLE_DEVICES",
+                "FRONTIER_LOG_LEVEL",
+            )
+        },
+        "command": command,
+        "state_outputs": {
+            "individual_batch_metrics": True,
+            "full_stage_batch_ledger": True,
+        },
+        "contains_prompt_text": False,
+    }
+    atomic_json(args.output / "run_manifest.json", manifest)
+    start = time.monotonic()
+    with (args.output / "stdout.log").open("w", encoding="utf-8") as stdout, (
+        args.output / "stderr.log"
+    ).open("w", encoding="utf-8") as stderr:
+        try:
+            process = subprocess.run(
+                command,
+                cwd=args.frontier_root,
+                env=environment,
+                stdout=stdout,
+                stderr=stderr,
+                timeout=args.timeout_s,
+            )
+            return_code = int(process.returncode)
+        except subprocess.TimeoutExpired:
+            return_code = 124
+    runtime_s = time.monotonic() - start
+    if return_code != 0:
+        failure = {
+            "status": "STOP",
+            "return_code": return_code,
+            "runtime_s": runtime_s,
+        }
+        atomic_json(args.output / "failure.json", failure)
+        raise RuntimeError(f"Frontier state replay failed: {failure}")
+
+    paths = find_state_metrics(args.output)
+    summary = summarize_frontier(
+        system_metrics_path=paths["system"],
+        request_metrics_path=paths["requests"],
+        batch_metrics_path=paths["batches"],
+        ledger_path=paths["ledger"],
+    )
+    atomic_json(args.output / "common-state.json", summary)
+    scorer = driver.score_trial(row, paths["system"], paths["requests"])
+    atomic_json(args.output / "scorer_output.json", scorer)
+    sizes = {name: path.stat().st_size for name, path in paths.items()}
+    red_flags = []
+    if summary["interval"]["request_count"] != int(entry["selected_count"]):
+        red_flags.append("request_count_mismatch")
+    if summary["sanity"]["batch_rows"] <= 0:
+        red_flags.append("no_batch_rows")
+    if summary["sanity"]["ledger_rows"] <= 0:
+        red_flags.append("no_ledger_rows")
+    result = {
+        "schema": "telemetry-residual-frontier-state-result-v1",
+        "status": "PASS" if not red_flags else "STOP",
+        "runtime_s": runtime_s,
+        "return_code": return_code,
+        "paths": {name: str(path.resolve()) for name, path in paths.items()},
+        "bytes": sizes,
+        "common_state": str((args.output / "common-state.json").resolve()),
+        "red_flags": red_flags,
+        "sanity": {
+            "request_rows": summary["sanity"]["request_rows"],
+            "batch_rows": summary["sanity"]["batch_rows"],
+            "ledger_rows": summary["sanity"]["ledger_rows"],
+            "artifact_bytes": {
+                "n": len(sizes),
+                "min": min(sizes.values()),
+                "max": max(sizes.values()),
+                "distinct_n": len(set(sizes.values())),
+            },
+            "invariants": {
+                "zero_failures": return_code == 0,
+                "gpu_visibility_disabled": True,
+                "request_count_match": not red_flags,
+                "nonnegative_file_sizes": all(size >= 0 for size in sizes.values()),
+                "state_values_not_all_identical": summary["sanity"]["invariants"][
+                    "batch_values_not_all_identical"
+                ],
+            },
+        },
+    }
+    atomic_json(args.output / "result.json", result)
+    if result["status"] != "PASS":
+        raise RuntimeError(red_flags)
+    return result
+
+
+def parser() -> argparse.ArgumentParser:
+    result = argparse.ArgumentParser()
+    result.add_argument("--prepared-manifest", type=Path, required=True)
+    result.add_argument("--output", type=Path, required=True)
+    result.add_argument("--replayserve-root", type=Path, required=True)
+    result.add_argument("--frontier-root", type=Path, required=True)
+    result.add_argument("--cell", required=True)
+    result.add_argument("--role", choices=("low1", "high1"), required=True)
+    result.add_argument("--timeout-s", type=float, default=300.0)
+    return result
+
+
+def main() -> None:
+    result = execute(parser().parse_args())
+    print(
+        json.dumps(
+            {
+                "status": result["status"],
+                "runtime_s": result["runtime_s"],
+                "request_rows": result["sanity"]["request_rows"],
+                "batch_rows": result["sanity"]["batch_rows"],
+                "ledger_rows": result["sanity"]["ledger_rows"],
+                "red_flags": result["red_flags"],
+            },
+            sort_keys=True,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

runs/telemetry-residual/run_frontier_state_campaign.py

@@ -0,0 +1,169 @@
+#!/usr/bin/env python3
+"""Run the frozen 12-fixture P1 Frontier state-detail campaign, CPU only."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import time
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
+
+import run_frontier_state as state_runner  # noqa: E402
+
+
+def atomic_json(path: Path, payload: Any) -> None:
+    state_runner.atomic_json(path, payload)
+
+
+def scorer_without_runtime(value: dict[str, Any]) -> dict[str, Any]:
+    result = dict(value)
+    result.pop("runtime_s", None)
+    return result
+
+
+def execute(args: argparse.Namespace) -> dict[str, Any]:
+    prepared = json.loads(args.prepared_manifest.read_text(encoding="utf-8"))
+    committed = json.loads(args.committed_results.read_text(encoding="utf-8"))
+    if prepared.get("status") != "PASS" or committed.get("status") != "PASS":
+        raise RuntimeError("prepared or committed simulator evidence did not pass")
+    committed_by_key = {
+        (row["cell"], row["role"]): scorer_without_runtime(row["scorer"])
+        for row in committed["results"]
+    }
+    entries = prepared["entries"]
+    if len(entries) != 12 or len(committed_by_key) != 12:
+        raise ValueError("P1 state campaign requires exactly 12 fixtures")
+    args.output.mkdir(parents=True, exist_ok=True)
+    results = []
+    campaign_start = time.monotonic()
+    for index, entry in enumerate(entries, start=1):
+        key = (entry["cell"], entry["role"])
+        output = args.output / f"{entry['cell']}_{entry['role']}"
+        result_path = output / "result.json"
+        if args.resume and result_path.is_file():
+            result = json.loads(result_path.read_text(encoding="utf-8"))
+            if result.get("status") != "PASS":
+                raise RuntimeError(f"cannot resume failed state replay: {result_path}")
+            resumed = True
+        else:
+            if output.exists() and any(output.iterdir()):
+                raise FileExistsError(f"non-empty state replay output: {output}")
+            print(
+                f"RUN {index:02d}/12 {entry['cell']}/{entry['role']}",
+                flush=True,
+            )
+            result = state_runner.execute(
+                SimpleNamespace(
+                    prepared_manifest=args.prepared_manifest,
+                    output=output,
+                    replayserve_root=args.replayserve_root,
+                    frontier_root=args.frontier_root,
+                    cell=entry["cell"],
+                    role=entry["role"],
+                    timeout_s=args.timeout_s,
+                )
+            )
+            resumed = False
+        observed_scorer = json.loads(
+            (output / "scorer_output.json").read_text(encoding="utf-8")
+        )
+        exact_scorer_match = scorer_without_runtime(observed_scorer) == committed_by_key[key]
+        if not exact_scorer_match:
+            raise ValueError(f"state-output replay changed the committed scorer: {key}")
+        results.append(
+            {
+                "cell": entry["cell"],
+                "role": entry["role"],
+                "runtime_s": result["runtime_s"],
+                "request_rows": result["sanity"]["request_rows"],
+                "batch_rows": result["sanity"]["batch_rows"],
+                "ledger_rows": result["sanity"]["ledger_rows"],
+                "artifact_bytes": sum(result["bytes"].values()),
+                "exact_committed_scorer_match": exact_scorer_match,
+                "resumed": resumed,
+                "result": str(result_path.resolve()),
+            }
+        )
+        print(
+            f"DONE {index:02d}/12 {entry['cell']}/{entry['role']} "
+            f"runtime={result['runtime_s']:.3f}s batches={result['sanity']['batch_rows']}",
+            flush=True,
+        )
+    runtimes = [float(row["runtime_s"]) for row in results]
+    batches = [int(row["batch_rows"]) for row in results]
+    bytes_values = [int(row["artifact_bytes"]) for row in results]
+    red_flags = []
+    if len(results) != 12:
+        red_flags.append("runs_not_12")
+    if not all(row["exact_committed_scorer_match"] for row in results):
+        red_flags.append("committed_scorer_mismatch")
+    if any(value <= 0 for value in batches):
+        red_flags.append("empty_batch_output")
+    result = {
+        "schema": "telemetry-residual-frontier-state-campaign-v1",
+        "status": "PASS" if not red_flags else "STOP",
+        "prepared_manifest": str(args.prepared_manifest.resolve()),
+        "committed_results": str(args.committed_results.resolve()),
+        "campaign_elapsed_s": time.monotonic() - campaign_start,
+        "results": results,
+        "red_flags": red_flags,
+        "sanity": {
+            "n": len(results),
+            "runtime_s": state_runner.numeric(runtimes),
+            "batch_rows": state_runner.numeric(batches),
+            "artifact_bytes": state_runner.numeric(bytes_values),
+            "invariants": {
+                "runs_12": len(results) == 12,
+                "zero_failures": not red_flags,
+                "exact_committed_scorers": all(
+                    row["exact_committed_scorer_match"] for row in results
+                ),
+                "nonnegative_counts": all(value > 0 for value in batches),
+                "per_config_not_identical": len(set(batches)) > 1,
+                "gpu_visibility_disabled": True,
+            },
+        },
+    }
+    atomic_json(args.output / "campaign-metrics.json", result)
+    if result["status"] != "PASS":
+        raise RuntimeError(red_flags)
+    return result
+
+
+def parser() -> argparse.ArgumentParser:
+    result = argparse.ArgumentParser()
+    result.add_argument("--prepared-manifest", type=Path, required=True)
+    result.add_argument("--committed-results", type=Path, required=True)
+    result.add_argument("--output", type=Path, required=True)
+    result.add_argument("--replayserve-root", type=Path, required=True)
+    result.add_argument("--frontier-root", type=Path, required=True)
+    result.add_argument("--timeout-s", type=float, default=300.0)
+    result.add_argument("--resume", action="store_true")
+    return result
+
+
+def main() -> None:
+    result = execute(parser().parse_args())
+    print(
+        json.dumps(
+            {
+                "status": result["status"],
+                "runs": len(result["results"]),
+                "elapsed_s": result["campaign_elapsed_s"],
+                "sanity": result["sanity"],
+                "red_flags": result["red_flags"],
+            },
+            sort_keys=True,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

runs/telemetry-residual/test_common_state.py

@@ -0,0 +1,183 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import csv
+import json
+import math
+import tempfile
+from pathlib import Path
+
+from common_state import numeric, residual, summarize_engine, summarize_frontier
+from run_frontier_state import enable_state_outputs
+
+
+def engine_record(
+    step: int,
+    timestamp_ns: int,
+    *,
+    batch: int,
+    prefill: int,
+    decode: int,
+    waiting: int,
+    running: int,
+    kv: float,
+) -> dict[str, object]:
+    return {
+        "step_index": step,
+        "submit_mono_ns": timestamp_ns,
+        "model_executed": True,
+        "scheduled_requests": batch,
+        "decode_batch_size": decode,
+        "prefill_tokens": prefill,
+        "decode_tokens": decode,
+        "preemptions": 0,
+        "queues": {"waiting": waiting, "running": running},
+        "kv": {"usage": kv},
+        "cudagraph": {
+            "runtime_mode": "NONE" if step == 0 else "FULL",
+            "bucket_tokens": prefill + decode + 1,
+            "padding_tokens": 1,
+        },
+        "dropped_records_before": 0,
+    }
+
+
+def main() -> None:
+    assert numeric((-2.0, -1.0))["cv"] >= 0.0
+    command = [
+        "python3",
+        "-m",
+        "frontier.main",
+        "--no-metrics_config_store_frontier_stage_batch_ledger",
+    ]
+    state_command = enable_state_outputs(command)
+    assert "--metrics_config_store_frontier_stage_batch_ledger" in state_command
+    assert "--metrics_config_keep_individual_batch_metrics" in state_command
+    assert "--no-metrics_config_store_frontier_stage_batch_ledger" not in state_command
+
+    engine = summarize_engine(
+        [
+            engine_record(
+                0,
+                0,
+                batch=2,
+                prefill=6,
+                decode=0,
+                waiting=1,
+                running=2,
+                kv=0.1,
+            ),
+            engine_record(
+                1,
+                1_000_000_000,
+                batch=4,
+                prefill=0,
+                decode=4,
+                waiting=3,
+                running=4,
+                kv=0.3,
+            ),
+        ],
+        start_ns=0,
+        end_ns=2_000_000_000,
+        request_count=2,
+    )
+    assert math.isclose(engine["common"]["queue_waiting_mean"], 2.0)
+    assert math.isclose(engine["common"]["queue_running_mean"], 3.0)
+    assert math.isclose(
+        engine["common"]["queue_waiting_time_per_request_ms"], 2000.0
+    )
+    assert engine["common"]["batch_size"]["distinct_n"] == 2
+    assert math.isclose(engine["common"]["prefill_token_fraction"], 0.6)
+
+    with tempfile.TemporaryDirectory() as temporary:
+        root = Path(temporary)
+        system = root / "system.json"
+        requests = root / "requests.csv"
+        batches = root / "batches.csv"
+        ledger = root / "ledger.jsonl"
+        system.write_text(
+            json.dumps(
+                {
+                    "throughput_metrics": {
+                        "total_duration_seconds": 2.0,
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        with requests.open("w", encoding="utf-8", newline="") as output:
+            writer = csv.DictWriter(
+                output,
+                fieldnames=[
+                    "request_e2e_time",
+                    "request_waiting_time_total",
+                    "request_total_preemption_count",
+                ],
+            )
+            writer.writeheader()
+            writer.writerow(
+                {
+                    "request_e2e_time": 2000,
+                    "request_waiting_time_total": 1000,
+                    "request_total_preemption_count": 0,
+                }
+            )
+            writer.writerow(
+                {
+                    "request_e2e_time": 3000,
+                    "request_waiting_time_total": 3000,
+                    "request_total_preemption_count": 0,
+                }
+            )
+        with batches.open("w", encoding="utf-8", newline="") as output:
+            writer = csv.DictWriter(
+                output,
+                fieldnames=[
+                    "batch_size",
+                    "batch_num_tokens",
+                    "batch_num_prefill_tokens",
+                    "batch_num_decode_tokens",
+                ],
+            )
+            writer.writeheader()
+            writer.writerow(
+                {
+                    "batch_size": 2,
+                    "batch_num_tokens": 6,
+                    "batch_num_prefill_tokens": 6,
+                    "batch_num_decode_tokens": 0,
+                }
+            )
+            writer.writerow(
+                {
+                    "batch_size": 4,
+                    "batch_num_tokens": 4,
+                    "batch_num_prefill_tokens": 0,
+                    "batch_num_decode_tokens": 4,
+                }
+            )
+        ledger.write_text(
+            json.dumps({"stage_start_ts": 0.0, "stage_end_ts": 1.0}) + "\n"
+            + json.dumps({"stage_start_ts": 1.0, "stage_end_ts": 2.0})
+            + "\n",
+            encoding="utf-8",
+        )
+        simulator = summarize_frontier(
+            system_metrics_path=system,
+            request_metrics_path=requests,
+            batch_metrics_path=batches,
+            ledger_path=ledger,
+        )
+    assert math.isclose(simulator["common"]["queue_waiting_mean"], 2.0)
+    assert math.isclose(simulator["common"]["queue_running_mean"], 0.5)
+    assert simulator["common"]["batch_size"]["distinct_n"] == 2
+    difference = residual(engine, simulator)
+    assert difference["coverage"]["missing"] == 0
+    assert difference["coverage"]["available"] == 16
+    assert math.isclose(difference["values"]["queue_waiting_mean"], 0.0)
+    print("telemetry common state: PASS")
+
+
+if __name__ == "__main__":
+    main()

runs/telemetry-residual/test_residual_transfer.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+"""Small structural tests for the cross-config transfer diagnostic."""
+
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+
+
+HERE = Path(__file__).resolve().parent
+SPEC = importlib.util.spec_from_file_location(
+    "analyze_residual_transfer", HERE / "analyze_residual_transfer.py"
+)
+assert SPEC is not None and SPEC.loader is not None
+MODULE = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(MODULE)
+
+
+def example(cell: str, level: str, value: float) -> dict:
+    tp_text, mns_text = cell.split("_")
+    return {
+        "cell": cell,
+        "level": level,
+        "tp": int(tp_text[2:]),
+        "mns": int(mns_text[3:]),
+        "pass_rate_residual": value - 0.5,
+        "real_pass_rate_rep1": value,
+        "sim_pass_rate": 0.5,
+        "offered_req_s_per_gpu": 1.0 if level == "low" else 2.0,
+        "real_feasible": value >= 0.95,
+        "sim_feasible": False,
+        "state_residual": {"values": {"batch.mean": value}},
+        "engine": {
+            "common": {"batch": {"mean": value, "max": value, "cv": 0.0}},
+            "engine_only": {"kv": value},
+        },
+    }
+
+
+def main() -> None:
+    cells = ("tp1_mns8", "tp1_mns16", "tp2_mns8", "tp2_mns16")
+    examples = [
+        example(cell, level, 0.1 + index / 10.0)
+        for level in ("low", "high")
+        for index, cell in enumerate(cells)
+    ]
+    rows = MODULE.transitions(examples)
+    assert len(rows) == 48
+    assert any(row["source_level"] != row["target_level"] for row in rows)
+    assert all(row["source_cell"] != row["target_cell"] for row in rows)
+    assert len(rows[0]["hybrid_telemetry"]) > len(rows[0]["hybrid_base"])
+    assert len(rows[0]["direct_telemetry"]) > len(rows[0]["direct_base"])
+    for feature, target in (
+        ("hybrid_base", "target_residual_delta"),
+        ("hybrid_telemetry", "target_residual_delta"),
+        ("direct_base", "target_real_pass_rate_delta"),
+        ("direct_telemetry", "target_real_pass_rate_delta"),
+    ):
+        prediction = MODULE.grouped_predictions(
+            rows, feature_name=feature, target_name=target, regularization=1.0
+        )
+        assert len(prediction) == len(rows)
+    print("telemetry residual transfer: PASS")
+
+
+if __name__ == "__main__":
+    main()