From 7a3631b528965bc2a5f20bb66eb138a8a0a83e88 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Tue, 14 Jul 2026 15:44:37 +0800 Subject: [PATCH] Audit telemetry residual tuning premise --- ...emetry-residual-tuning-roadmap-20260714.md | 286 +++++++++++ runs/telemetry-residual/analyze_p1_state.py | 174 +++++++ runs/telemetry-residual/analyze_r0_gate.py | 292 +++++++++++ .../analyze_residual_transfer.py | 480 ++++++++++++++++++ runs/telemetry-residual/common_state.py | 400 +++++++++++++++ runs/telemetry-residual/run_frontier_state.py | 316 ++++++++++++ .../run_frontier_state_campaign.py | 169 ++++++ runs/telemetry-residual/test_common_state.py | 183 +++++++ .../test_residual_transfer.py | 67 +++ 9 files changed, 2367 insertions(+) create mode 100644 docs/telemetry-residual-tuning-roadmap-20260714.md create mode 100644 runs/telemetry-residual/analyze_p1_state.py create mode 100644 runs/telemetry-residual/analyze_r0_gate.py create mode 100644 runs/telemetry-residual/analyze_residual_transfer.py create mode 100644 runs/telemetry-residual/common_state.py create mode 100644 runs/telemetry-residual/run_frontier_state.py create mode 100644 runs/telemetry-residual/run_frontier_state_campaign.py create mode 100644 runs/telemetry-residual/test_common_state.py create mode 100644 runs/telemetry-residual/test_residual_transfer.py diff --git a/docs/telemetry-residual-tuning-roadmap-20260714.md b/docs/telemetry-residual-tuning-roadmap-20260714.md new file mode 100644 index 0000000..ba6ceab --- /dev/null +++ b/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. diff --git a/runs/telemetry-residual/analyze_p1_state.py b/runs/telemetry-residual/analyze_p1_state.py new file mode 100644 index 0000000..2c2e7b0 --- /dev/null +++ b/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() diff --git a/runs/telemetry-residual/analyze_r0_gate.py b/runs/telemetry-residual/analyze_r0_gate.py new file mode 100644 index 0000000..d9912e5 --- /dev/null +++ b/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() diff --git a/runs/telemetry-residual/analyze_residual_transfer.py b/runs/telemetry-residual/analyze_residual_transfer.py new file mode 100644 index 0000000..a07283a --- /dev/null +++ b/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() diff --git a/runs/telemetry-residual/common_state.py b/runs/telemetry-residual/common_state.py new file mode 100644 index 0000000..9850e1c --- /dev/null +++ b/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"]), + }, + } diff --git a/runs/telemetry-residual/run_frontier_state.py b/runs/telemetry-residual/run_frontier_state.py new file mode 100644 index 0000000..2fde9c8 --- /dev/null +++ b/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() diff --git a/runs/telemetry-residual/run_frontier_state_campaign.py b/runs/telemetry-residual/run_frontier_state_campaign.py new file mode 100644 index 0000000..7b9ae0e --- /dev/null +++ b/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() diff --git a/runs/telemetry-residual/test_common_state.py b/runs/telemetry-residual/test_common_state.py new file mode 100644 index 0000000..175f1c8 --- /dev/null +++ b/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() diff --git a/runs/telemetry-residual/test_residual_transfer.py b/runs/telemetry-residual/test_residual_transfer.py new file mode 100644 index 0000000..1944f6a --- /dev/null +++ b/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()