gahow/aituner
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add fidelity-aware verification pilot

Browse Source
This commit is contained in:
Gahow Wang
2026-07-14 12:49:53 +08:00
parent 57dd6a9fac
commit 16239bef00
18 changed files with 8165 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
docs/fidelity-aware-harness-headroom-20260714.md Normal file
View File

@@ -0,0 +1,95 @@
# Fidelity-aware harness headroom audit
Status: **PROMISING PREMISE, NO CONTRIBUTION CLAIM**.
The audit answers whether engine instrumentation has enough incremental signal
to justify a prospective experiment. It does not establish generalization.
## Simulator shortlist lower bound
On the frozen 12-cell SimFid task, the strongest calibrated SLO simulator
reading places TP2/MNS32 and TP2/MNS64 in the same first tie bucket. Real-final
evaluation of that two-cell bucket selects TP2/MNS32 and has zero real regret.
A method requiring a real calibration probe plus final verification cannot beat
two real cell evaluations on this task. Therefore “better initial selection”
is not a viable claim here; the remaining headroom is shorter real verification
inside the same shortlist.
## Five-second prefix result
The retrospective Phase-6 dataset contains 37 primary anchors across 12 cells.
Stable labels use the frozen same-placement 2-of-3 adjudication: 28 feasible and
9 infeasible. Three TP4 primary measurements disagree with their repeated
labels, so single-run feasibility is not treated as ground truth.
Using leave-one-cell-out folds, identical L2 logistic models, and a 5-second
prefix:
| Metric | Outcome-only | Instrumentation-aware | Delta |
|---|---:|---:|---:|
| Accuracy | 78.38% | 89.19% | +10.81 pp |
| Balanced accuracy | 70.63% | 81.55% | +10.92 pp |
| Brier score | 0.1297 | 0.0901 | -0.0396 |
| Correct only in this model | 0 | 4 | +4 |
| McNemar exact two-sided p | — | 0.125 | not significant |
At the frozen conservative threshold 0.95, both policies make zero false
accepts and zero false rejects on this retrospective set. Outcome-only safely
cuts 36.35% of measured primary-trial cost; instrumentation-aware safely cuts
61.10%, an additional 24.75 percentage points. Regularization sensitivity for
accuracy delta is `[0.00, +10.81]` percentage points, so the sign is
non-negative but the magnitude is not stable.
Longer prefixes do not strengthen the case monotonically. At 10 seconds,
headline accuracy is 91.89% outcome-only versus 89.19% instrumentation-aware;
at 15 seconds it is 88.89% versus 91.67%; at 20 seconds it is 86.11% versus
91.67%, but both 0.95 policies make one false reject. Five seconds is therefore
a training-selected operating point, not a test result.
## Interpretation
There is enough headroom to run a held-out pilot, but not enough evidence to
claim the harness contribution:
- the 5-second cost gap is operationally large;
- only four paired classifications differ, so significance is absent;
- all examples share one workload/SLO/engine task;
- completion timestamps are reconstructed from arrival + TTFT + TPOT rather
than recorded directly;
- three adjudication disagreements are concentrated in transient TP4 runs;
- outcome-only already recovers the simulator shortlist oracle with very few
real cells.
The next experiment must therefore freeze the 5-second model and threshold,
record exact monotonic completions, use a held-out trace, and label each anchor
with three full repetitions. The registered protocol is
`docs/fidelity-aware-harness-protocol-20260714.md`.
## Artifacts
- `runs/fidelity-headroom/analyze_existing.py`
- `runs/fidelity-headroom/metrics.json`
- `runs/fidelity-headroom/analyze_prefixes.py`
- `runs/fidelity-headroom/prefix-metrics.json`
- `runs/fidelity-headroom/test_analysis.py`
- `runs/fidelity-headroom/test_prefix_analysis.py`
## Sanity block
| Family | n | Min | Max | Distinct | Invariant/result |
|---|---:|---:|---:|---:|---|
| Real SimFid cell scores | 12 | 1.2833 | 3.2833 | 7 | Non-negative; not identical |
| Prefix examples at 5 s | 37 | 5 s | 5 s | 1 expected | All 12 cells represented |
| Adjudicated labels | 37 | 0 | 1 | 2 | 28 positive / 9 negative |
| Primary/adjudicated disagreement | 37 | 0 | 1 | 2 | 3 TP4 disagreements retained |
| Full primary elapsed time | 37 | 14.566 s | 62.064 s | 37 | Every 5 s prefix is in range |
| Outcome probability | 37 | in `[0,1]` | in `[0,1]` | >1 | Checked before metrics |
| Instrumentation probability | 37 | in `[0,1]` | in `[0,1]` | >1 | Checked before metrics |
| Layer-1 streams | 12 | 14,174 records | 58,725 records | 12 | Contiguous, zero drops |
Checked invariants: same folds/model family and cutoff; no full verdict in a
feature; prefix-only Layer-1 slicing; non-negative costs/counters; bounded
ratios/probabilities; both labels present; per-config results not identical;
tie expansion before top-k; no imputation of non-monotonic frontiers. The main
limitation is reconstructed request completion time, explicitly marked on all
37 five-second examples.

193
docs/fidelity-aware-harness-protocol-20260714.md Normal file
View File

@@ -0,0 +1,193 @@
# Fidelity-aware real-verification harness protocol
Status: **PRE-REGISTERED STAGED EVALUATION; CONTRIBUTION NOT YET ESTABLISHED**.
Date frozen: 2026-07-14 (Asia/Singapore).
## Research question and contribution bar
The harness has an independent systems contribution only if engine-internal
instrumentation improves a tuning decision beyond what is already achievable
with a simulator shortlist and external benchmark outcomes. The intended
claim is therefore deliberately stronger than “telemetry explains a run”:
> Given the same simulator ranking, the same candidate order, and the same
> short real-GPU probe, a learned instrumentation-aware verifier reaches a
> configuration with at most 5% real SLO-goodput regret using materially fewer
> H20-hours than both (a) simulator top-k followed by full real evaluation and
> (b) an outcome-only verifier given exactly the same probe.
The paper-facing gate is:
- at least 20% lower real-verification H20-hours than outcome-only calibration;
- at least 30% lower real-verification H20-hours than simulator top-k plus full
real final evaluation;
- paired 95% task-bootstrap confidence interval for the outcome-only cost
reduction strictly above zero;
- selected-configuration SLO-goodput regret at most 5% on every headline task;
- no false-safe early accept in the pilot and at most 1% in the expanded suite;
- profiling, warm-up, confirmation, instrumentation, and failed-run costs are
included rather than amortized away. An amortized profile-cost view may be
reported only as a secondary result.
If these conditions fail, instrumentation remains a debugging facility. It is
not an independent tuning-harness contribution.
## What is learned, and what is not a rule
The decision target is a stable, repeated real verdict, not a hand-authored
diagnosis such as “queue length above N means reject.” Each anchor receives
three full real repetitions and a frozen 2-of-3 feasibility label. A nested
pair of regularized models predicts that label from a fixed prefix:
- **Outcome-only input X:** configuration, offered rate, admitted/completed
progress, observed TTFT/TPOT margins, failures, and known workload lengths.
- **Instrumentation input Z:** the same X plus generic engine state: running and
waiting queues, decode-batch shape, KV usage, graph mode and padding, prefill
share, preemptions, and model-step rate.
Both models use the same L2 logistic family, train split, standardization,
regularization, cutoff, and probability threshold. The only experimental
difference is Z. The initial family is intentionally simple: a positive result
then demonstrates value in the engine signal rather than capacity in a larger
learner. A sequence model is admissible only as a later, paired ablation.
The frozen first policy uses a 5-second prefix, L2 regularization 1.0, and a
two-sided abstaining threshold of 0.95: accept at `p(feasible)>=0.95`, reject at
`p(feasible)<=0.05`, otherwise continue the exact same trial to completion.
Threshold and cutoff were selected on the historical training task and are
therefore not evidence; all claims come from subsequent held-out tasks.
## Fair baselines
| Method | Simulator | 5-second real prefix | External outcomes | Engine state | Full real continuation |
|---|---:|---:|---:|---:|---:|
| Real-only oracle | no | no | full | optional diagnostic | every candidate/anchor |
| Sim top-k + real final | yes | included in full run | full | no decision use | every shortlisted candidate/anchor |
| Outcome-only calibration | yes | yes | yes | no | only on abstention |
| Instrumentation-aware | yes | yes | yes | yes | only on abstention |
Tie buckets are expanded before top-k. `k` is selected on training tasks and
is fixed on held-out tasks; an oracle per-task k is forbidden. Outcome-only
receives all information available outside the engine, including config and
workload features. Instrumentation cannot use any record submitted after the
cutoff. The full label, confirmation votes, simulator error, and later
requests are never model features.
## Staged experiment
### R0: historical premise and headroom audit
The frozen SimFid surface has 12 cells. The strongest calibrated SLO simulator
reading has a top tie bucket `{TP2/MNS32, TP2/MNS64}`; full real evaluation of
those two cells already finds the oracle with zero regret. Consequently this
single task cannot demonstrate a selection-count advantage: any method needing
one real calibration probe and one real final verification has a lower bound of
two real cells.
The viable estimand is instead the duration and number of full real frontier
evaluations inside a fixed shortlist. Historical Phase-6 prefixes are analyzed
only as training/premise data. Their request completion times are reconstructed
from arrival, TTFT, TPOT, and token count, so they cannot support a final claim.
### P1: exact-timestamp prospective pilot
- Engine/model/hardware: patched vLLM 0.24.1.dev3, Qwen3-30B-A3B, one solo
server/client on dash0, NVIDIA H20, `TP in {1,2,4}`.
- Held-out workload: `chat_w20260312_1000`, 60-second replay after the frozen
0.1 time scale, raw input `[0,8192]`, exactly 128 output tokens.
- SLO: stepped TTFT 2/4/6 seconds, TPOT 50 ms, 95% request pass rate.
- Cells: TP1/MNS8, TP1/MNS64, TP2/MNS8, TP2/MNS64, TP4/MNS16, TP4/MNS64.
- Per cell: one attainable low offered rate near 0.85x the historical v0.24
frontier and one high rate near 1.25x. The exact threshold and selected
request hashes are frozen by a CPU preflight before launch.
- Each cell uses a fresh server, the accepted long-request warm-up, one
unmeasured full-window burn-in, then three repetitions per rate. Rate order
alternates and reverses across cells to prevent a fixed warm-state/order
confound.
- The first repetition supplies the exact prefix. All three repetitions supply
the 2-of-3 label. Every request records a monotonic completion timestamp;
Layer-1 records are cut at the same monotonic boundary.
- Placement is serialized. Co-location is forbidden because Phase 6 observed
up to 92.86 percentage-point pass-rate shifts under co-location.
- Hard cap: 3.5 H20-hours, including startup, warm-up, burn-in, all repetitions,
failures, and cleanup. Projected cap violation stops before the next cell.
P1 opens P2 only if all data invariants pass and instrumentation-aware has zero
false accept/reject, is no worse than outcome-only, and either makes at least
three additional correct early decisions or improves total valid trial-cost
reduction by at least 15 absolute percentage points. The pilot is a gate, not
paper evidence.
### P2: held-out task replication
If P1 passes, freeze the model and run at least six independent task groups:
three trace windows spanning distinct date/slot combinations and two SLO
regimes. No task used for threshold/model selection enters the headline test.
The candidate surface is the full 12-cell `TP={1,2,4} x MNS={8,16,32,64}`
surface. Splits are by complete task, never by anchor or request. A task-level
paired bootstrap (10,000 repetitions, fixed seed) estimates cost and regret
intervals. Non-monotonic or split 2-of-3 anchors remain explicit; no frontier
is imputed.
### P3: end-to-end shortlist and search replay
For each P2 task, run the same frozen simulator and tie-expanded top-k policy.
Replay the real binary/frontier search under all three verification policies:
full real, outcome-only, and instrumentation-aware. The policy consumes only
prefixes that would have been available at that decision point. Report:
- selected cell and real SLO-goodput regret;
- number of real cells, anchors, and confirmations;
- measured H20-hours and wall time;
- false accept, false reject, and abstention counts;
- profile, startup/warm-up, probe, full-continuation, confirmation, logging, and
failure cost breakdowns.
### P4: simulator-rank-error attribution
This phase distinguishes an outdated implementation/profile from a structural
simulator limitation. For each held-out task compare:
1. the original simulator/profile;
2. a version-matched re-profiled simulator;
3. a trajectory-conditioned run supplied with the realized arrival and request
length sequence;
4. outcome-only residual calibration;
5. instrumentation-aware residual calibration.
The engine trace is extended only as needed with a worker-level step UID and
CUDA-event duration, because current async submit-to-complete spans overlap and
are not GPU step time. Residuals are decomposed into operator-profile error,
scheduler/state error, and run-to-run noise. If re-profiling alone restores the
ranking, the old 30% loss was an implementation/profile defect. If exact
profiles and realized trajectories still mis-rank cells, and the residual is
systematically explained by queue/KV/graph/batch state unavailable to the
simulator, that is evidence of a structural state-abstraction gap. Correlation
alone is not called causal.
## Failure modes that reject the route
- Outcome-only matches or beats instrumentation-aware under the same cutoff.
- Instrumentation gains average accuracy but introduces false-safe decisions.
- Gains disappear under task-level rather than request/anchor-level splitting.
- Savings come only from excluding startup, warm-up, profiling, confirmations,
or failed trials.
- A different cutoff/threshold must be selected after seeing each test task.
- The simulator top-k baseline already reaches the target with equal or lower
total H20-hours.
- Exact instrumentation overhead exceeds 1% throughput or materially changes
p95/p99 latency.
- Results depend on TP4 transient/non-monotonic trials and do not replicate on
held-out tasks.
## Data sanity contract
Every analysis ends with n, min/max, distinct count, label balance, and these
invariants: non-negative counters/costs; probabilities and ratios in `[0,1]`;
per-config results not all identical; timestamps monotonic; every prefix record
at or before its cutoff; selected request ID/arrival/length hashes stable across
repetitions; exact 128-token completion or counted failure; no dropped Layer-1
records; 2-of-3 labels reproducible; no co-resident GPU process; total H20-hours
below the hard cap; final GPUs idle. A red flag is reported first and blocks
the contribution claim.

508
runs/fidelity-headroom/analyze_existing.py Normal file
View File

@@ -0,0 +1,508 @@
#!/usr/bin/env python3
"""Retrospective headroom audit for a fidelity-aware tuning harness.
This analysis intentionally separates two questions:
1. How many real cell evaluations does a simulator top-k shortlist already
need to recover the real optimum on the frozen SimFid surface?
2. On the P6 anchor ladder, do Layer-1 engine features predict the next
anchor's feasibility better than outcome-only features from the same
current anchor?
The second question is diagnostic rather than decision-bearing: it uses a
small, already-observed single-workload surface and full current-anchor
summaries. It is a premise check for a future prospective early-probe study.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import math
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable
import numpy as np
SCHEMA = "fidelity-headroom-v1"
DEFAULT_REGULARIZATION = 1.0
REGULARIZATION_SENSITIVITY = (0.1, 1.0, 10.0)
BOOTSTRAP_SEED = 20260714
BOOTSTRAP_REPLICATES = 10_000
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 numeric(values: Iterable[float | int]) -> dict[str, Any]:
array = [float(value) for value in values]
return {
"n": len(array),
"min": min(array) if array else None,
"max": max(array) if array else None,
"distinct_n": len(set(array)),
}
def score_buckets(scores: dict[str, float], tolerance: float) -> dict[str, int]:
if tolerance <= 0:
raise ValueError("score tolerance must be positive")
return {cell: math.floor(float(score) / tolerance) for cell, score in scores.items()}
def topk_curve(
real_scores: dict[str, float],
simulated_scores: dict[str, float],
tolerance: float,
) -> dict[str, Any]:
if set(real_scores) != set(simulated_scores):
raise ValueError("real and simulator score cells differ")
buckets = score_buckets(simulated_scores, tolerance)
ordered = sorted(
simulated_scores,
key=lambda cell: (-buckets[cell], -float(simulated_scores[cell]), cell),
)
real_best = max(float(value) for value in real_scores.values())
points = []
for nominal_k in range(1, len(ordered) + 1):
cutoff_bucket = buckets[ordered[nominal_k - 1]]
candidates = [cell for cell in ordered if buckets[cell] >= cutoff_bucket]
selected = max(candidates, key=lambda cell: (float(real_scores[cell]), cell))
selected_score = float(real_scores[selected])
points.append(
{
"nominal_k": nominal_k,
"expanded_k": len(candidates),
"candidates": candidates,
"selected_cell_after_real_final": selected,
"selected_real_score": selected_score,
"real_regret": 1.0 - selected_score / real_best,
}
)
minimum_k = {}
for name, threshold in (("zero", 1e-15), ("one_percent", 0.01), ("five_percent", 0.05)):
eligible = [point for point in points if point["real_regret"] <= threshold]
minimum_k[name] = (
{
"nominal_k": eligible[0]["nominal_k"],
"expanded_k": eligible[0]["expanded_k"],
}
if eligible
else None
)
return {
"real_best": real_best,
"minimum_k": minimum_k,
"points": points,
}
@dataclass(frozen=True)
class Transition:
cell: str
current_anchor: float
next_anchor: float
external: tuple[float, ...]
instrumentation: tuple[float, ...]
next_feasible: int
EXTERNAL_FEATURES = (
"log_current_rate_per_gpu",
"log_next_over_current_rate",
"log2_tp",
"log2_mns",
"current_pass_rate",
"ttft_max_over_6s",
"tpot_max_over_50ms",
"exact_output_fraction",
"early_stopped",
)
INSTRUMENTATION_FEATURES = (
"waiting_mean",
"waiting_max",
"decode_batch_mean",
"decode_batch_cv",
"kv_usage_mean",
"kv_usage_max",
"graph_none_share",
"graph_full_share",
"padding_fraction",
"prefill_token_fraction",
"model_steps_per_second",
)
def _finite(value: float | int | None) -> float:
if value is None:
return 0.0
result = float(value)
if not math.isfinite(result):
raise ValueError(f"non-finite feature: {value}")
return result
def build_transitions(phase6: dict[str, Any]) -> list[Transition]:
transitions = []
for cell, cell_result in sorted(phase6["cells"].items()):
anchors = sorted(cell_result["anchors"], key=lambda item: float(item["anchor"]))
for current, following in zip(anchors, anchors[1:]):
if following["accepted_feasible"] is None:
continue
primary = current["primary"]
next_primary = following["primary"]
layer = current["layer1"]
rate = float(primary["selection"]["offered_req_s_per_gpu"])
next_rate = float(next_primary["selection"]["offered_req_s_per_gpu"])
selected_count = int(primary["selection"]["count"])
if rate <= 0 or next_rate <= 0 or selected_count <= 0:
raise ValueError("rates and selected counts must be positive")
external = (
math.log(rate),
math.log(next_rate / rate),
math.log2(float(cell_result["tp"])),
math.log2(float(cell_result["mns"])),
float(primary["pass_rate"]),
_finite(primary["ttft_ms"]["max"]) / 6000.0,
_finite(primary["tpot_ms"]["max"]) / 50.0,
float(primary["exact_output_count"]) / selected_count,
float(bool(primary["early_stopped"])),
)
graph_shares = layer.get("graph_mode_shares", {})
prefill_tokens = _finite(layer["prefill_tokens"])
decode_tokens = _finite(layer["decode_tokens"])
instrumentation = (
_finite(layer["waiting_mean"]),
_finite(layer["waiting_max"]),
_finite(layer["decode_B_mean"]),
_finite(layer["decode_B_cv"]),
_finite(layer["kv_usage_mean"]),
_finite(layer["kv_usage_max"]),
float(graph_shares.get("NONE", 0.0)),
float(graph_shares.get("FULL", 0.0)),
_finite(layer["padding_fraction"]),
prefill_tokens / max(1.0, prefill_tokens + decode_tokens),
_finite(layer["model_steps"]) / float(primary["interval"]["elapsed_s"]),
)
transitions.append(
Transition(
cell=cell,
current_anchor=float(current["anchor"]),
next_anchor=float(following["anchor"]),
external=external,
instrumentation=instrumentation,
next_feasible=int(bool(following["accepted_feasible"])),
)
)
return transitions
def _sigmoid(values: np.ndarray) -> np.ndarray:
clipped = np.clip(values, -30.0, 30.0)
return 1.0 / (1.0 + np.exp(-clipped))
def _fit_logistic(x: np.ndarray, y: np.ndarray, regularization: float) -> np.ndarray:
weights = np.zeros(x.shape[1], dtype=np.float64)
penalty = np.eye(x.shape[1], dtype=np.float64)
penalty[0, 0] = 0.0
for _ in range(100):
probability = _sigmoid(x @ weights)
gradient = x.T @ (probability - y) / len(y)
gradient += regularization * penalty @ weights / len(y)
curvature = probability * (1.0 - probability)
hessian = (x.T * curvature) @ x / len(y)
hessian += regularization * penalty / len(y)
step = np.linalg.lstsq(hessian, gradient, rcond=None)[0]
weights -= step
if float(np.max(np.abs(step))) < 1e-9:
break
return weights
def _classification_metrics(y: np.ndarray, probability: np.ndarray) -> dict[str, Any]:
if np.any(probability < 0.0) or np.any(probability > 1.0):
raise ValueError("classification probabilities must be in [0, 1]")
prediction = probability >= 0.5
true_positive = int(np.sum(prediction & (y == 1)))
true_negative = int(np.sum(~prediction & (y == 0)))
false_positive = int(np.sum(prediction & (y == 0)))
false_negative = int(np.sum(~prediction & (y == 1)))
positive_total = true_positive + false_negative
negative_total = true_negative + false_positive
balanced = 0.5 * (
true_positive / positive_total + true_negative / negative_total
)
clipped = np.clip(probability, 1e-12, 1.0 - 1e-12)
return {
"accuracy": float(np.mean(prediction == y)),
"balanced_accuracy": float(balanced),
"brier": float(np.mean((probability - y) ** 2)),
"log_loss": float(np.mean(-(y * np.log(clipped) + (1 - y) * np.log(1 - clipped)))),
"confusion": {
"true_positive": true_positive,
"true_negative": true_negative,
"false_positive": false_positive,
"false_negative": false_negative,
},
}
def _mcnemar_exact_p(outcome_only_correct: int, instrumentation_only_correct: int) -> float:
discordant = outcome_only_correct + instrumentation_only_correct
if discordant == 0:
return 1.0
tail = sum(
math.comb(discordant, value)
for value in range(min(outcome_only_correct, instrumentation_only_correct) + 1)
) / (2**discordant)
return min(1.0, 2.0 * tail)
def grouped_predictions(
transitions: list[Transition],
*,
instrumentation_aware: bool,
regularization: float,
) -> tuple[np.ndarray, np.ndarray, list[str]]:
probabilities = []
labels = []
test_cells = []
for held_out in sorted({transition.cell for transition in transitions}):
train = [transition for transition in transitions if transition.cell != held_out]
test = [transition for transition in transitions if transition.cell == held_out]
def row(transition: Transition) -> np.ndarray:
values = transition.external
if instrumentation_aware:
values += transition.instrumentation
return np.asarray((1.0, *values), dtype=np.float64)
x_train = np.stack([row(transition) for transition in train])
x_test = np.stack([row(transition) for transition in test])
y_train = np.asarray([transition.next_feasible for transition in train], dtype=np.float64)
mean = x_train[:, 1:].mean(axis=0)
standard_deviation = x_train[:, 1:].std(axis=0)
standard_deviation[standard_deviation < 1e-8] = 1.0
x_train[:, 1:] = (x_train[:, 1:] - mean) / standard_deviation
x_test[:, 1:] = (x_test[:, 1:] - mean) / standard_deviation
weights = _fit_logistic(x_train, y_train, regularization)
probabilities.extend(_sigmoid(x_test @ weights).tolist())
labels.extend(transition.next_feasible for transition in test)
test_cells.extend(held_out for _ in test)
return (
np.asarray(labels, dtype=np.int64),
np.asarray(probabilities, dtype=np.float64),
test_cells,
)
def _group_bootstrap_delta(
y: np.ndarray,
outcome_probability: np.ndarray,
instrumentation_probability: np.ndarray,
cells: list[str],
) -> dict[str, Any]:
groups = sorted(set(cells))
indices = {group: np.asarray([i for i, cell in enumerate(cells) if cell == group]) for group in groups}
random = np.random.default_rng(BOOTSTRAP_SEED)
accuracy_deltas = []
brier_deltas = []
for _ in range(BOOTSTRAP_REPLICATES):
sampled = random.choice(groups, size=len(groups), replace=True)
selected = np.concatenate([indices[group] for group in sampled])
selected_y = y[selected]
outcome = outcome_probability[selected]
instrumentation = instrumentation_probability[selected]
accuracy_deltas.append(
float(np.mean((instrumentation >= 0.5) == selected_y))
- float(np.mean((outcome >= 0.5) == selected_y))
)
brier_deltas.append(
float(np.mean((instrumentation - selected_y) ** 2))
- float(np.mean((outcome - selected_y) ** 2))
)
return {
"semantics": "group bootstrap over cells; diagnostic confidence interval",
"replicates": BOOTSTRAP_REPLICATES,
"seed": BOOTSTRAP_SEED,
"accuracy_delta_instrumentation_minus_outcome": {
"point": float(np.mean((instrumentation_probability >= 0.5) == y))
- float(np.mean((outcome_probability >= 0.5) == y)),
"ci95": [float(x) for x in np.percentile(accuracy_deltas, [2.5, 97.5])],
},
"brier_delta_instrumentation_minus_outcome": {
"point": float(np.mean((instrumentation_probability - y) ** 2))
- float(np.mean((outcome_probability - y) ** 2)),
"ci95": [float(x) for x in np.percentile(brier_deltas, [2.5, 97.5])],
},
}
def transition_analysis(transitions: list[Transition]) -> dict[str, Any]:
sensitivity = {}
headline_payload = None
for regularization in REGULARIZATION_SENSITIVITY:
y, outcome_probability, cells = grouped_predictions(
transitions,
instrumentation_aware=False,
regularization=regularization,
)
instrumentation_y, instrumentation_probability, instrumentation_cells = grouped_predictions(
transitions,
instrumentation_aware=True,
regularization=regularization,
)
if not np.array_equal(y, instrumentation_y) or cells != instrumentation_cells:
raise AssertionError("model folds or labels differ")
outcome_correct = (outcome_probability >= 0.5) == y
instrumentation_correct = (instrumentation_probability >= 0.5) == y
payload = {
"outcome_only": _classification_metrics(y, outcome_probability),
"instrumentation_aware": _classification_metrics(y, instrumentation_probability),
"paired_correctness": {
"both_correct": int(np.sum(outcome_correct & instrumentation_correct)),
"outcome_only_correct": int(np.sum(outcome_correct & ~instrumentation_correct)),
"instrumentation_only_correct": int(np.sum(~outcome_correct & instrumentation_correct)),
"both_wrong": int(np.sum(~outcome_correct & ~instrumentation_correct)),
},
"bootstrap": _group_bootstrap_delta(
y,
outcome_probability,
instrumentation_probability,
cells,
),
}
payload["paired_correctness"]["mcnemar_exact_two_sided_p"] = _mcnemar_exact_p(
payload["paired_correctness"]["outcome_only_correct"],
payload["paired_correctness"]["instrumentation_only_correct"],
)
sensitivity[str(regularization)] = payload
if regularization == DEFAULT_REGULARIZATION:
headline_payload = payload
assert headline_payload is not None
labels = [transition.next_feasible for transition in transitions]
accuracy_deltas = [
value["instrumentation_aware"]["accuracy"] - value["outcome_only"]["accuracy"]
for value in sensitivity.values()
]
brier_deltas = [
value["instrumentation_aware"]["brier"] - value["outcome_only"]["brier"]
for value in sensitivity.values()
]
return {
"status": "RETROSPECTIVE_DIAGNOSTIC_ONLY",
"estimand": "next-anchor feasibility from the full current-anchor summary",
"split": "leave-one-cell-out",
"model": "L2 logistic regression with train-fold standardization",
"external_features": list(EXTERNAL_FEATURES),
"instrumentation_features": list(INSTRUMENTATION_FEATURES),
"headline_regularization": DEFAULT_REGULARIZATION,
"headline": headline_payload,
"regularization_sensitivity": sensitivity,
"sensitivity_summary": {
"accuracy_delta_min_max": [min(accuracy_deltas), max(accuracy_deltas)],
"brier_delta_min_max": [min(brier_deltas), max(brier_deltas)],
"incremental_signal_verdict": "NEEDS_PROSPECTIVE_EVIDENCE",
},
"label_sanity": {
**numeric(labels),
"positive": sum(labels),
"negative": len(labels) - sum(labels),
},
}
def analyze(simfid_path: Path, phase6_path: Path) -> dict[str, Any]:
simfid = json.loads(simfid_path.read_text())
phase6 = json.loads(phase6_path.read_text())
real_scores = {cell: float(score) for cell, score in simfid["real_scores"].items()}
topk = {}
for reading, payload in sorted(simfid["analyses"].items()):
tie = payload["metrics"]["tie_buckets"]["simulator"]
topk[reading] = topk_curve(
real_scores,
{cell: float(score) for cell, score in payload["simulated_scores"].items()},
float(tie["tolerance"]),
)
transitions = build_transitions(phase6)
transition_result = transition_analysis(transitions)
red_flags = []
if len(real_scores) != 12:
red_flags.append("unexpected_simfid_cell_count")
if len(transitions) == 0 or len(set(x.next_feasible for x in transitions)) != 2:
red_flags.append("transition_labels_missing_or_single_class")
if any(not math.isfinite(value) or value < 0 for value in real_scores.values()):
red_flags.append("invalid_real_score")
return {
"schema": SCHEMA,
"status": "PASS" if not red_flags else "STOP",
"scope": "retrospective single-workload premise audit; not prospective contribution evidence",
"provenance": {
"simfid_metrics": str(simfid_path.resolve()),
"simfid_sha256": sha256_file(simfid_path),
"phase6_metrics": str(phase6_path.resolve()),
"phase6_sha256": sha256_file(phase6_path),
},
"topk_headroom": topk,
"next_anchor_prediction": transition_result,
"decision": {
"current_surface_can_show_selection_contribution": False,
"reason": (
"The strongest frozen-calibrated SLO reading reaches zero real regret "
"after real evaluation of its first two-cell tie bucket. A method that "
"requires one calibration probe and one final verification cannot use "
"this single task to demonstrate fewer real cell evaluations."
),
"prospective_target": (
"Test whether internal features from a short, shared real probe reduce "
"the number or duration of full frontier evaluations relative to an "
"outcome-only model given the same probe."
),
},
"sanity": {
"real_scores": numeric(real_scores.values()),
"simulator_readings": len(topk),
"transitions": len(transitions),
"transition_cells": len({transition.cell for transition in transitions}),
"red_flags": red_flags,
"invariants": {
"same_cells_all_readings": all(
set(payload["simulated_scores"]) == set(real_scores)
for payload in simfid["analyses"].values()
),
"scores_nonnegative": all(value >= 0 for value in real_scores.values()),
"transition_features_finite": all(
all(math.isfinite(value) for value in (*item.external, *item.instrumentation))
for item in transitions
),
"probabilities_bounded": True,
},
},
}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--simfid-metrics", type=Path, required=True)
parser.add_argument("--phase6-metrics", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
result = analyze(args.simfid_metrics, args.phase6_metrics)
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n")
print(json.dumps({"status": result["status"], "output": str(args.output)}, sort_keys=True))
if __name__ == "__main__":
main()

293
runs/fidelity-headroom/analyze_pilot.py Normal file
View File

@@ -0,0 +1,293 @@
#!/usr/bin/env python3
"""Evaluate frozen outcome-only and instrumentation-aware policies on P1."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any
import numpy as np
from analyze_existing import _classification_metrics, _mcnemar_exact_p
from analyze_prefixes import (
PrefixExample,
_load_jsonl,
_prefix_features,
numeric,
policy_metrics,
predict_frozen_model,
sha256_file,
)
def result_path(run_root: Path, cell: str, level: str, replicate: int) -> Path:
return run_root / "cells" / cell / f"{level}-rep{replicate}" / "result.json"
def requests_path(run_root: Path, cell: str, level: str, replicate: int) -> Path:
return run_root / "cells" / cell / f"{level}-rep{replicate}" / "requests.jsonl"
def selection_for(
manifest: dict[str, Any], cell: str, level: str, replicate: int
) -> dict[str, Any]:
role = f"{level}{replicate}"
return manifest["cells"][cell]["targets"][level]["selections"][role]
def build_pilot_examples(
manifest: dict[str, Any], run_root: Path, cutoff_s: float
) -> tuple[list[PrefixExample], list[dict[str, Any]], list[str]]:
examples = []
details = []
red_flags = []
for cell, config in sorted(manifest["cells"].items()):
stream_path = next((run_root / "cells" / cell / "opprof").glob("*.jsonl"))
stream = _load_jsonl(stream_path, require_key="submit_mono_ns")
for level in ("low", "high"):
results = [
json.loads(result_path(run_root, cell, level, replicate).read_text())
for replicate in (1, 2, 3)
]
votes = [bool(result["feasible"]) for result in results]
adjudicated = sum(votes) >= 2
primary = results[0]
requests = _load_jsonl(requests_path(run_root, cell, level, 1))
exact_timestamps = sum(
request.get("completed_elapsed_s") is not None for request in requests
)
actual_outcomes = sum(
request.get("completed_mono_ns") is not None for request in requests
)
if exact_timestamps != actual_outcomes:
red_flags.append(f"timestamp_count_mismatch_{cell}_{level}")
expected = selection_for(manifest, cell, level, 1)
if int(primary["selection"]["count"]) != int(expected["selected_count"]):
red_flags.append(f"selection_count_mismatch_{cell}_{level}")
for result_key, manifest_key in (
("request_id_order_sha256", "request_id_order_sha256"),
("arrival_order_sha256", "arrival_order_sha256"),
("raw_length_order_sha256", "input_length_order_sha256"),
):
if primary["selection"][result_key] != expected[manifest_key]:
red_flags.append(f"selection_hash_mismatch_{cell}_{level}_{result_key}")
start_ns = int(primary["interval"]["start_mono_ns"])
end_ns = start_ns + int(cutoff_s * 1e9)
records = [
record
for record in stream
if record.get("model_executed")
and start_ns <= int(record["submit_mono_ns"]) <= end_ns
]
outcome, instrumentation, completion_source = _prefix_features(
primary=primary,
tp=int(config["tp"]),
max_num_seqs=int(config["mns"]),
requests=requests,
records=records,
cutoff_s=cutoff_s,
)
example = PrefixExample(
cell=cell,
anchor=float(primary["anchor"]),
cutoff_s=cutoff_s,
tp=int(config["tp"]),
full_elapsed_s=float(primary["interval"]["elapsed_s"]),
feasible=int(adjudicated),
primary_feasible=int(bool(primary["feasible"])),
outcome=outcome,
instrumentation=instrumentation,
completion_time_source=completion_source,
)
examples.append(example)
details.append(
{
"cell": cell,
"level": level,
"anchor_rep1": primary["anchor"],
"selected_count_rep1": primary["selection"]["count"],
"votes": votes,
"pass_rates": [result["pass_rate"] for result in results],
"adjudicated_feasible": adjudicated,
"primary_feasible": bool(primary["feasible"]),
"actual_timestamped_outcomes": actual_outcomes,
"selected_outcomes": len(requests),
"prefix_layer1_records": len(records),
"completion_time_source": completion_source,
}
)
return examples, details, red_flags
def analyze(
manifest_path: Path,
model_path: Path,
run_root: Path,
) -> dict[str, Any]:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
models = json.loads(model_path.read_text(encoding="utf-8"))
state_path = run_root / "controller-state.json"
state = json.loads(state_path.read_text(encoding="utf-8"))
cutoff_s = float(models["cutoff_s"])
threshold = float(models["accept_probability"])
examples, details, red_flags = build_pilot_examples(manifest, run_root, cutoff_s)
labels = np.asarray([example.feasible for example in examples], dtype=np.int64)
outcome_probability = predict_frozen_model(models["models"]["outcome_only"], examples)
instrumentation_probability = predict_frozen_model(
models["models"]["instrumentation_aware"], examples
)
outcome_policy = policy_metrics(
examples, labels, outcome_probability, threshold
)
instrumentation_policy = policy_metrics(
examples, labels, instrumentation_probability, threshold
)
outcome_correct = (outcome_probability >= 0.5) == labels
instrumentation_correct = (instrumentation_probability >= 0.5) == labels
paired = {
"both_correct": int(np.sum(outcome_correct & instrumentation_correct)),
"outcome_only_correct": int(np.sum(outcome_correct & ~instrumentation_correct)),
"instrumentation_only_correct": int(np.sum(~outcome_correct & instrumentation_correct)),
"both_wrong": int(np.sum(~outcome_correct & ~instrumentation_correct)),
}
paired["mcnemar_exact_two_sided_p"] = _mcnemar_exact_p(
paired["outcome_only_correct"], paired["instrumentation_only_correct"]
)
for detail, outcome_p, instrumentation_p in zip(
details, outcome_probability, instrumentation_probability
):
detail["outcome_probability_feasible"] = float(outcome_p)
detail["instrumentation_probability_feasible"] = float(instrumentation_p)
positive = int(np.sum(labels))
negative = len(labels) - positive
if state["status"] != "complete" or int(state["completed_cells"]) != 6:
red_flags.append("campaign_incomplete")
if positive < 3 or negative < 3:
red_flags.append("insufficient_label_balance")
if any(
detail["actual_timestamped_outcomes"] == 0 for detail in details
):
red_flags.append("no_exact_request_timestamps")
if float(state["gpu_hours_total"]) >= float(state["hard_cap_h20_hours"]):
red_flags.append("hard_cap_exceeded")
outcome_errors = outcome_policy["false_accept"] + outcome_policy["false_reject"]
instrumentation_errors = (
instrumentation_policy["false_accept"]
+ instrumentation_policy["false_reject"]
)
outcome_decisions = outcome_policy["early_accept"] + outcome_policy["early_reject"]
instrumentation_decisions = (
instrumentation_policy["early_accept"]
+ instrumentation_policy["early_reject"]
)
outcome_reduction = outcome_policy["valid_cost_reduction_fraction"]
instrumentation_reduction = instrumentation_policy["valid_cost_reduction_fraction"]
cost_delta = (
instrumentation_reduction - outcome_reduction
if outcome_reduction is not None and instrumentation_reduction is not None
else None
)
data_valid = not red_flags
safety_gate = instrumentation_errors == 0 and instrumentation_errors <= outcome_errors
incremental_gate = (
instrumentation_decisions - outcome_decisions >= 3
or (cost_delta is not None and cost_delta >= 0.15)
)
pilot_pass = data_valid and safety_gate and incremental_gate
return {
"schema": "fidelity-prefix-pilot-result-v1",
"status": "PILOT_PASS" if pilot_pass else "PILOT_FAIL",
"scope": "held-out single-task gate; not paper-facing contribution evidence",
"provenance": {
"manifest": str(manifest_path.resolve()),
"manifest_sha256": sha256_file(manifest_path),
"frozen_models": str(model_path.resolve()),
"frozen_models_sha256": sha256_file(model_path),
"controller_state": str(state_path.resolve()),
"controller_state_sha256": sha256_file(state_path),
},
"cutoff_s": cutoff_s,
"threshold": threshold,
"examples": details,
"outcome_only": {
"classification": _classification_metrics(labels, outcome_probability),
"policy": outcome_policy,
},
"instrumentation_aware": {
"classification": _classification_metrics(labels, instrumentation_probability),
"policy": instrumentation_policy,
},
"paired_correctness": paired,
"gate": {
"data_valid": data_valid,
"safety_gate": safety_gate,
"incremental_gate": incremental_gate,
"additional_early_decisions": instrumentation_decisions - outcome_decisions,
"valid_cost_reduction_fraction_delta": cost_delta,
"opens_expanded_p2": pilot_pass,
},
"gpu": {
"actual_h20_hours": state["gpu_hours_total"],
"hard_cap_h20_hours": state["hard_cap_h20_hours"],
},
"sanity": {
"red_flags": red_flags,
"labels": {
**numeric(labels.tolist()),
"positive": positive,
"negative": negative,
},
"full_elapsed_s": numeric(example.full_elapsed_s for example in examples),
"remaining_h20_hours": numeric(
example.remaining_h20_hours for example in examples
),
"outcome_probability": numeric(outcome_probability.tolist()),
"instrumentation_probability": numeric(
instrumentation_probability.tolist()
),
"invariants": {
"examples_12": len(examples) == 12,
"cells_6": len({example.cell for example in examples}) == 6,
"ratios_bounded": bool(
np.all((outcome_probability >= 0) & (outcome_probability <= 1))
and np.all(
(instrumentation_probability >= 0)
& (instrumentation_probability <= 1)
)
),
"costs_nonnegative": all(
example.remaining_h20_hours >= 0 for example in examples
),
"all_cell_validations": all(
all(cell["validation"]["invariants"].values())
for cell in state["cells"].values()
),
},
},
}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--manifest", type=Path, required=True)
parser.add_argument("--frozen-models", type=Path, required=True)
parser.add_argument("--run-root", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
result = analyze(args.manifest, args.frozen_models, args.run_root)
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n")
print(json.dumps({
"status": result["status"],
"gate": result["gate"],
"sanity_red_flags": result["sanity"]["red_flags"],
}, sort_keys=True))
if __name__ == "__main__":
main()

629
runs/fidelity-headroom/analyze_prefixes.py Normal file
View File

@@ -0,0 +1,629 @@
#!/usr/bin/env python3
"""Retrospective, leakage-bounded audit of short real-probe prefixes.
The outcome-only and instrumentation-aware models receive the same trial
prefix. The latter differs only by Layer-1 engine state. Existing Phase-6
request artifacts predate exact completion timestamps, so their completion
time is reconstructed from arrival + TTFT + token intervals and is explicitly
marked approximate. New artifacts use ``completed_elapsed_s`` directly.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import math
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable
import numpy as np
from analyze_existing import (
DEFAULT_REGULARIZATION,
REGULARIZATION_SENSITIVITY,
_classification_metrics,
_fit_logistic,
_group_bootstrap_delta,
_mcnemar_exact_p,
_sigmoid,
)
SCHEMA = "fidelity-prefix-v1"
DEFAULT_CUTOFFS = (5.0, 10.0, 15.0, 20.0)
POLICY_THRESHOLDS = (0.8, 0.9, 0.95)
OUTCOME_FEATURES = (
"log_offered_rate_per_gpu",
"log2_tp",
"log2_max_num_seqs",
"admitted_fraction",
"completed_over_admitted",
"completed_pass_rate",
"completed_fail_fraction_of_total",
"outstanding_over_admitted",
"ttft_max_over_slo_max",
"ttft_mean_over_slo_max",
"tpot_max_over_slo",
"tpot_mean_over_slo",
"admitted_input_tokens_mean_over_limit",
)
INSTRUMENTATION_FEATURES = (
"model_steps_per_second",
"waiting_mean",
"waiting_max",
"waiting_nonzero_share",
"running_mean",
"running_max",
"decode_batch_mean",
"decode_batch_max",
"decode_batch_cv",
"kv_usage_mean",
"kv_usage_max",
"kv_usage_end_minus_start",
"graph_none_share",
"graph_full_share",
"padding_fraction",
"prefill_token_fraction",
"preemptions",
)
@dataclass(frozen=True)
class PrefixExample:
cell: str
anchor: float
cutoff_s: float
tp: int
full_elapsed_s: float
feasible: int
primary_feasible: int
outcome: tuple[float, ...]
instrumentation: tuple[float, ...]
completion_time_source: str
@property
def remaining_h20_hours(self) -> float:
return self.tp * max(0.0, self.full_elapsed_s - self.cutoff_s) / 3600.0
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 numeric(values: Iterable[float | int]) -> dict[str, Any]:
array = [float(value) for value in values]
return {
"n": len(array),
"min": min(array) if array else None,
"max": max(array) if array else None,
"distinct_n": len(set(array)),
}
def _cv(values: list[float]) -> float:
if not values:
return 0.0
array = np.asarray(values, dtype=np.float64)
mean = float(array.mean())
return float(array.std(ddof=0) / mean) if mean else 0.0
def completion_elapsed_s(request: dict[str, Any]) -> tuple[float | None, str]:
exact = request.get("completed_elapsed_s")
if exact is not None:
value = float(exact)
if value < 0 or not math.isfinite(value):
raise ValueError(f"invalid completed_elapsed_s={exact}")
return value, "exact_monotonic"
if not request.get("success"):
return None, "unobserved_failure"
required = (
request.get("arrival_s"),
request.get("ttft_ms"),
request.get("tpot_ms"),
request.get("completion_tokens"),
)
if any(value is None for value in required):
return None, "unobserved_failure"
arrival_s, ttft_ms, tpot_ms, completion_tokens = required
value = float(arrival_s) + (
float(ttft_ms) + max(int(completion_tokens) - 1, 0) * float(tpot_ms)
) / 1000.0
if value < 0 or not math.isfinite(value):
raise ValueError(f"invalid reconstructed completion time={value}")
return value, "reconstructed_from_latency"
def _load_jsonl(path: Path, *, require_key: str | None = None) -> list[dict[str, Any]]:
records = []
with path.open(encoding="utf-8") as source:
for line in source:
item = json.loads(line)
if require_key is None or require_key in item:
records.append(item)
return records
def _anchor_directory(cell_root: Path, anchor: float) -> Path:
matches = []
for result_path in cell_root.glob("anchor-*/result.json"):
payload = json.loads(result_path.read_text(encoding="utf-8"))
if math.isclose(float(payload["anchor"]), anchor, rel_tol=0.0, abs_tol=1e-15):
matches.append(result_path.parent)
if len(matches) != 1:
raise ValueError(f"expected one primary directory for anchor {anchor}: {matches}")
return matches[0]
def _prefix_features(
*,
primary: dict[str, Any],
tp: int,
max_num_seqs: int,
requests: list[dict[str, Any]],
records: list[dict[str, Any]],
cutoff_s: float,
) -> tuple[tuple[float, ...], tuple[float, ...], str]:
admitted = [request for request in requests if float(request["arrival_s"]) <= cutoff_s]
completed = []
sources = set()
for request in requests:
completed_s, source = completion_elapsed_s(request)
if completed_s is None or completed_s > cutoff_s:
continue
completed.append(request)
sources.add(source)
if not admitted or not records:
raise ValueError("prefix has no admitted requests or Layer-1 records")
if any(request not in admitted for request in completed):
raise ValueError("completed request was not admitted inside prefix")
total = len(requests)
passed = sum(bool(request["slo_pass"]) for request in completed)
ttft = [float(request["ttft_ms"]) for request in completed if request["ttft_ms"] is not None]
tpot = [float(request["tpot_ms"]) for request in completed if request["tpot_ms"] is not None]
offered_rate = float(primary["selection"]["offered_req_s_per_gpu"])
if offered_rate <= 0 or total <= 0:
raise ValueError("offered rate and selected request count must be positive")
outcome = (
math.log(offered_rate),
math.log2(float(tp)),
math.log2(float(max_num_seqs)),
len(admitted) / total,
len(completed) / len(admitted),
passed / max(1, len(completed)),
(len(completed) - passed) / total,
(len(admitted) - len(completed)) / len(admitted),
max(ttft, default=0.0) / 6000.0,
float(np.mean(ttft)) / 6000.0 if ttft else 0.0,
max(tpot, default=0.0) / 50.0,
float(np.mean(tpot)) / 50.0 if tpot else 0.0,
float(np.mean([float(request["raw_input_tokens"]) for request in admitted])) / 8192.0,
)
waiting = [float(record["queues"]["waiting"]) for record in records]
running = [float(record["queues"]["running"]) for record in records]
decode_batch = [float(record["decode_batch_size"]) for record in records]
kv_usage = [float(record["kv"]["usage"]) for record in records]
graph_modes = [str(record["cudagraph"]["runtime_mode"]) for record in records]
bucket_tokens = sum(int(record["cudagraph"]["bucket_tokens"]) for record in records)
padding_tokens = sum(int(record["cudagraph"]["padding_tokens"]) for record in records)
prefill_tokens = sum(int(record["prefill_tokens"]) for record in records)
decode_tokens = sum(int(record["decode_tokens"]) for record in records)
instrumentation = (
len(records) / cutoff_s,
float(np.mean(waiting)),
max(waiting),
sum(value > 0 for value in waiting) / len(waiting),
float(np.mean(running)),
max(running),
float(np.mean(decode_batch)),
max(decode_batch),
_cv(decode_batch),
float(np.mean(kv_usage)),
max(kv_usage),
kv_usage[-1] - kv_usage[0],
graph_modes.count("NONE") / len(graph_modes),
graph_modes.count("FULL") / len(graph_modes),
padding_tokens / max(1, bucket_tokens),
prefill_tokens / max(1, prefill_tokens + decode_tokens),
float(sum(int(record["preemptions"]) for record in records)),
)
completion_source = "+".join(sorted(sources)) if sources else "none_completed"
return outcome, instrumentation, completion_source
def build_examples(
phase6: dict[str, Any],
raw_root: Path,
cutoff_s: float,
) -> list[PrefixExample]:
examples = []
for cell, cell_result in sorted(phase6["cells"].items()):
cell_root = raw_root / cell
stream_path = next((cell_root / "opprof").glob("*.jsonl"))
stream = _load_jsonl(stream_path, require_key="submit_mono_ns")
for anchor in cell_result["anchors"]:
primary = anchor["primary"]
full_elapsed_s = float(primary["interval"]["elapsed_s"])
if full_elapsed_s + 1e-9 < cutoff_s:
continue
anchor_value = float(primary["anchor"])
anchor_root = _anchor_directory(cell_root, anchor_value)
requests = _load_jsonl(anchor_root / "requests.jsonl")
start_ns = int(primary["interval"]["start_mono_ns"])
end_ns = start_ns + int(cutoff_s * 1e9)
records = [
record
for record in stream
if record.get("model_executed")
and start_ns <= int(record["submit_mono_ns"]) <= end_ns
]
outcome, instrumentation, source = _prefix_features(
primary=primary,
tp=int(cell_result["tp"]),
max_num_seqs=int(cell_result["mns"]),
requests=requests,
records=records,
cutoff_s=cutoff_s,
)
examples.append(
PrefixExample(
cell=cell,
anchor=anchor_value,
cutoff_s=cutoff_s,
tp=int(cell_result["tp"]),
full_elapsed_s=full_elapsed_s,
feasible=int(bool(anchor["accepted_feasible"])),
primary_feasible=int(bool(primary["feasible"])),
outcome=outcome,
instrumentation=instrumentation,
completion_time_source=source,
)
)
return examples
def grouped_predictions(
examples: list[PrefixExample],
*,
instrumentation_aware: bool,
regularization: float,
) -> tuple[np.ndarray, np.ndarray, list[str]]:
probabilities = []
labels = []
groups = []
for held_out in sorted({example.cell for example in examples}):
train = [example for example in examples if example.cell != held_out]
test = [example for example in examples if example.cell == held_out]
def row(example: PrefixExample) -> np.ndarray:
values = example.outcome
if instrumentation_aware:
values += example.instrumentation
return np.asarray((1.0, *values), dtype=np.float64)
x_train = np.stack([row(example) for example in train])
x_test = np.stack([row(example) for example in test])
y_train = np.asarray([example.feasible for example in train], dtype=np.float64)
if len(set(y_train.tolist())) != 2:
raise ValueError(f"training fold for {held_out} has a single label")
mean = x_train[:, 1:].mean(axis=0)
standard_deviation = x_train[:, 1:].std(axis=0)
standard_deviation[standard_deviation < 1e-8] = 1.0
x_train[:, 1:] = (x_train[:, 1:] - mean) / standard_deviation
x_test[:, 1:] = (x_test[:, 1:] - mean) / standard_deviation
weights = _fit_logistic(x_train, y_train, regularization)
probabilities.extend(_sigmoid(x_test @ weights).tolist())
labels.extend(example.feasible for example in test)
groups.extend(held_out for _ in test)
return (
np.asarray(labels, dtype=np.int64),
np.asarray(probabilities, dtype=np.float64),
groups,
)
def fit_frozen_model(
examples: list[PrefixExample],
*,
instrumentation_aware: bool,
regularization: float,
) -> dict[str, Any]:
def row(example: PrefixExample) -> np.ndarray:
values = example.outcome
if instrumentation_aware:
values += example.instrumentation
return np.asarray((1.0, *values), dtype=np.float64)
matrix = np.stack([row(example) for example in examples])
labels = np.asarray([example.feasible for example in examples], dtype=np.float64)
if len(set(labels.tolist())) != 2:
raise ValueError("frozen model requires both feasibility labels")
mean = matrix[:, 1:].mean(axis=0)
standard_deviation = matrix[:, 1:].std(axis=0)
standard_deviation[standard_deviation < 1e-8] = 1.0
standardized = matrix.copy()
standardized[:, 1:] = (standardized[:, 1:] - mean) / standard_deviation
weights = _fit_logistic(standardized, labels, regularization)
probabilities = _sigmoid(standardized @ weights)
names = list(OUTCOME_FEATURES)
if instrumentation_aware:
names.extend(INSTRUMENTATION_FEATURES)
return {
"instrumentation_aware": instrumentation_aware,
"regularization": regularization,
"feature_names": names,
"feature_mean": mean.tolist(),
"feature_standard_deviation": standard_deviation.tolist(),
"weights_with_intercept_first": weights.tolist(),
"training_classification": _classification_metrics(labels, probabilities),
}
def predict_frozen_model(
model: dict[str, Any],
examples: list[PrefixExample],
) -> np.ndarray:
instrumentation_aware = bool(model["instrumentation_aware"])
rows = []
for example in examples:
values = example.outcome
if instrumentation_aware:
values += example.instrumentation
rows.append((1.0, *values))
matrix = np.asarray(rows, dtype=np.float64)
mean = np.asarray(model["feature_mean"], dtype=np.float64)
standard_deviation = np.asarray(
model["feature_standard_deviation"], dtype=np.float64
)
weights = np.asarray(model["weights_with_intercept_first"], dtype=np.float64)
if matrix.shape[1] != len(weights) or matrix.shape[1] - 1 != len(mean):
raise ValueError("frozen model feature dimensions do not match examples")
matrix[:, 1:] = (matrix[:, 1:] - mean) / standard_deviation
return _sigmoid(matrix @ weights)
def policy_metrics(
examples: list[PrefixExample],
labels: np.ndarray,
probabilities: np.ndarray,
threshold: float,
) -> dict[str, Any]:
accept = probabilities >= threshold
reject = probabilities <= 1.0 - threshold
decide = accept | reject
prediction = accept.astype(np.int64)
correct = prediction == labels
remaining = np.asarray(
[example.remaining_h20_hours for example in examples], dtype=np.float64
)
full_cost = sum(example.tp * example.full_elapsed_s / 3600.0 for example in examples)
saved = float(np.sum(remaining[decide]))
correct_saved = float(np.sum(remaining[decide & correct]))
invalid_saved = float(np.sum(remaining[decide & ~correct]))
def describe(mask: np.ndarray) -> list[dict[str, Any]]:
return [
{
"cell": example.cell,
"anchor": example.anchor,
"label_feasible": bool(label),
"probability_feasible": float(probability),
"remaining_h20_hours": example.remaining_h20_hours,
}
for example, label, probability, selected in zip(
examples, labels, probabilities, mask
)
if selected
]
return {
"threshold": threshold,
"early_accept": int(np.sum(accept)),
"early_reject": int(np.sum(reject)),
"abstain_continue_full": int(np.sum(~decide)),
"false_accept": int(np.sum(accept & (labels == 0))),
"false_reject": int(np.sum(reject & (labels == 1))),
"false_accept_examples": describe(accept & (labels == 0)),
"false_reject_examples": describe(reject & (labels == 1)),
"decision_coverage": float(np.mean(decide)),
"full_trial_h20_hours": float(full_cost),
"remaining_h20_hours_at_cutoff": float(np.sum(remaining)),
"saved_h20_hours_if_decisions_used": saved,
"correctly_saved_h20_hours": correct_saved,
"invalidly_saved_h20_hours": invalid_saved,
"valid_zero_error_policy": bool(np.all(correct[decide])),
"valid_cost_reduction_fraction": (
correct_saved / full_cost if invalid_saved == 0.0 and full_cost else None
),
}
def analyze_cutoff(examples: list[PrefixExample]) -> dict[str, Any]:
sensitivity = {}
headline = None
for regularization in REGULARIZATION_SENSITIVITY:
labels, outcome_probability, groups = grouped_predictions(
examples,
instrumentation_aware=False,
regularization=regularization,
)
instrument_labels, instrument_probability, instrument_groups = grouped_predictions(
examples,
instrumentation_aware=True,
regularization=regularization,
)
if not np.array_equal(labels, instrument_labels) or groups != instrument_groups:
raise AssertionError("paired folds or labels differ")
if groups != [example.cell for example in examples]:
raise AssertionError("prediction order differs from example order")
outcome_correct = (outcome_probability >= 0.5) == labels
instrument_correct = (instrument_probability >= 0.5) == labels
result = {
"outcome_only": {
"classification": _classification_metrics(labels, outcome_probability),
"policies": [
policy_metrics(examples, labels, outcome_probability, threshold)
for threshold in POLICY_THRESHOLDS
],
},
"instrumentation_aware": {
"classification": _classification_metrics(labels, instrument_probability),
"policies": [
policy_metrics(examples, labels, instrument_probability, threshold)
for threshold in POLICY_THRESHOLDS
],
},
"paired_correctness": {
"both_correct": int(np.sum(outcome_correct & instrument_correct)),
"outcome_only_correct": int(np.sum(outcome_correct & ~instrument_correct)),
"instrumentation_only_correct": int(np.sum(~outcome_correct & instrument_correct)),
"both_wrong": int(np.sum(~outcome_correct & ~instrument_correct)),
},
"bootstrap": _group_bootstrap_delta(
labels,
outcome_probability,
instrument_probability,
groups,
),
}
paired = result["paired_correctness"]
paired["mcnemar_exact_two_sided_p"] = _mcnemar_exact_p(
paired["outcome_only_correct"], paired["instrumentation_only_correct"]
)
sensitivity[str(regularization)] = result
if regularization == DEFAULT_REGULARIZATION:
headline = result
assert headline is not None
labels = [example.feasible for example in examples]
return {
"examples": len(examples),
"cells": len({example.cell for example in examples}),
"label_sanity": {
**numeric(labels),
"positive": sum(labels),
"negative": len(labels) - sum(labels),
"primary_adjudicated_disagreements": sum(
example.feasible != example.primary_feasible for example in examples
),
},
"completion_time_sources": {
source: sum(example.completion_time_source == source for example in examples)
for source in sorted({example.completion_time_source for example in examples})
},
"headline_regularization": DEFAULT_REGULARIZATION,
"headline": headline,
"regularization_sensitivity": sensitivity,
"remaining_h20_hours": numeric(
example.remaining_h20_hours for example in examples
),
}
def analyze(
phase6_path: Path,
raw_root: Path,
cutoffs: tuple[float, ...],
) -> dict[str, Any]:
phase6 = json.loads(phase6_path.read_text(encoding="utf-8"))
by_cutoff = {}
red_flags = []
for cutoff in cutoffs:
examples = build_examples(phase6, raw_root, cutoff)
if len({example.feasible for example in examples}) != 2:
red_flags.append(f"single_label_at_{cutoff:g}s")
continue
by_cutoff[f"{cutoff:g}"] = analyze_cutoff(examples)
if len({example.cell for example in examples}) != 12:
red_flags.append(f"incomplete_cells_at_{cutoff:g}s")
if not all(
math.isfinite(value)
for example in examples
for value in (*example.outcome, *example.instrumentation)
):
red_flags.append(f"nonfinite_features_at_{cutoff:g}s")
headline_deltas = {
cutoff: {
"accuracy": (
result["headline"]["instrumentation_aware"]["classification"]["accuracy"]
- result["headline"]["outcome_only"]["classification"]["accuracy"]
),
"brier": (
result["headline"]["instrumentation_aware"]["classification"]["brier"]
- result["headline"]["outcome_only"]["classification"]["brier"]
),
}
for cutoff, result in by_cutoff.items()
}
return {
"schema": SCHEMA,
"status": "PASS" if not red_flags else "STOP",
"scope": (
"retrospective single-workload prefix diagnostic; model selection, "
"threshold choice, and contribution claims require held-out prospective tasks"
),
"estimand": (
"2-of-3 adjudicated anchor feasibility from the first primary trial's "
"identical short real prefix"
),
"split": "leave-one-configuration-cell-out",
"model": "same L2 logistic model and folds; instrumentation model appends Layer-1 features",
"outcome_features": list(OUTCOME_FEATURES),
"instrumentation_features": list(INSTRUMENTATION_FEATURES),
"provenance": {
"phase6_metrics": str(phase6_path.resolve()),
"phase6_metrics_sha256": sha256_file(phase6_path),
"raw_root": str(raw_root.resolve()),
},
"cutoffs_s": list(cutoffs),
"cutoffs": by_cutoff,
"headline_incremental_deltas": headline_deltas,
"decision": {
"contribution_established": False,
"reason": (
"This dataset contains one workload and reconstructed rather than exact request "
"completion times. Three TP4 primary trials also disagree with their 2-of-3 "
"labels. It can reject a missing-signal premise but cannot establish "
"generalization or a paper-facing cost reduction."
),
},
"sanity": {
"red_flags": red_flags,
"cutoff_count": len(by_cutoff),
"invariants": {
"cutoffs_positive": all(cutoff > 0 for cutoff in cutoffs),
"paired_same_model_family": True,
"probabilities_checked_in_unit_interval": True,
"full_trial_label_not_used_as_feature": True,
"records_strictly_prefix_sliced": True,
},
},
}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--phase6-metrics", type=Path, required=True)
parser.add_argument("--raw-root", type=Path, required=True)
parser.add_argument("--cutoffs", type=float, nargs="+", default=DEFAULT_CUTOFFS)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
result = analyze(args.phase6_metrics, args.raw_root, tuple(args.cutoffs))
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(json.dumps({"status": result["status"], "output": str(args.output)}, sort_keys=True))
if __name__ == "__main__":
main()

93
runs/fidelity-headroom/freeze_models.py Normal file
View File

@@ -0,0 +1,93 @@
#!/usr/bin/env python3
"""Freeze the training-task prefix models before prospective GPU work."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from analyze_prefixes import (
DEFAULT_REGULARIZATION,
POLICY_THRESHOLDS,
build_examples,
fit_frozen_model,
sha256_file,
)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--phase6-metrics", type=Path, required=True)
parser.add_argument("--prefix-metrics", type=Path, required=True)
parser.add_argument("--raw-root", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
cutoff_s = 5.0
threshold = 0.95
if threshold not in POLICY_THRESHOLDS:
raise AssertionError("frozen threshold is outside audited policy thresholds")
phase6 = json.loads(args.phase6_metrics.read_text(encoding="utf-8"))
examples = build_examples(phase6, args.raw_root, cutoff_s)
payload = {
"schema": "fidelity-prefix-model-v1",
"status": "FROZEN_BEFORE_PROSPECTIVE_RUN",
"cutoff_s": cutoff_s,
"accept_probability": threshold,
"reject_probability": 1.0 - threshold,
"regularization": DEFAULT_REGULARIZATION,
"label": "same-placement 2-of-3 adjudicated anchor feasibility",
"training_split_role": "historical training only; never headline test",
"training_examples": [
{
"cell": example.cell,
"anchor": example.anchor,
"label_feasible": bool(example.feasible),
"primary_feasible": bool(example.primary_feasible),
"completion_time_source": example.completion_time_source,
}
for example in examples
],
"models": {
"outcome_only": fit_frozen_model(
examples,
instrumentation_aware=False,
regularization=DEFAULT_REGULARIZATION,
),
"instrumentation_aware": fit_frozen_model(
examples,
instrumentation_aware=True,
regularization=DEFAULT_REGULARIZATION,
),
},
"provenance": {
"phase6_metrics": str(args.phase6_metrics.resolve()),
"phase6_metrics_sha256": sha256_file(args.phase6_metrics),
"prefix_metrics": str(args.prefix_metrics.resolve()),
"prefix_metrics_sha256": sha256_file(args.prefix_metrics),
"raw_root": str(args.raw_root.resolve()),
},
"sanity": {
"n": len(examples),
"positive": sum(example.feasible for example in examples),
"negative": sum(not example.feasible for example in examples),
"cells": len({example.cell for example in examples}),
"invariants": {
"n_37": len(examples) == 37,
"cells_12": len({example.cell for example in examples}) == 12,
"both_labels": len({example.feasible for example in examples}) == 2,
"cutoff_5s": cutoff_s == 5.0,
"threshold_0.95": threshold == 0.95,
},
},
}
if not all(payload["sanity"]["invariants"].values()):
raise RuntimeError(f"model freeze invariants failed: {payload['sanity']}")
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
print(json.dumps({"status": payload["status"], "output": str(args.output)}))
if __name__ == "__main__":
main()

515
runs/fidelity-headroom/frozen-models.json Normal file
View File

@@ -0,0 +1,515 @@
{
"accept_probability": 0.95,
"cutoff_s": 5.0,
"label": "same-placement 2-of-3 adjudicated anchor feasibility",
"models": {
"instrumentation_aware": {
"feature_mean": [
0.8984976998643891,
0.8378378378378378,
4.324324324324325,
0.07552086023066117,
0.6758807403968693,
0.9459459459459459,
0.0,
0.3241192596031305,
0.04468545442770093,
0.025590516908533558,
0.23873649352596996,
0.1943716628122394,
0.4321792125178198,
112.70270270270272,
0.21087752102856197,
0.918918918918919,
0.055470351361483664,
4.904239530899751,
10.162162162162161,
4.822982150502539,
10.135135135135135,
0.43557131397798415,
0.031387890158936414,
0.05804311436894179,
0.03298678556030958,
0.030437119300455177,
0.9503065396705037,
0.07127076398319926,
0.6234198543231205,
0.0
],
"feature_names": [
"log_offered_rate_per_gpu",
"log2_tp",
"log2_max_num_seqs",
"admitted_fraction",
"completed_over_admitted",
"completed_pass_rate",
"completed_fail_fraction_of_total",
"outstanding_over_admitted",
"ttft_max_over_slo_max",
"ttft_mean_over_slo_max",
"tpot_max_over_slo",
"tpot_mean_over_slo",
"admitted_input_tokens_mean_over_limit",
"model_steps_per_second",
"waiting_mean",
"waiting_max",
"waiting_nonzero_share",
"running_mean",
"running_max",
"decode_batch_mean",
"decode_batch_max",
"decode_batch_cv",
"kv_usage_mean",
"kv_usage_max",
"kv_usage_end_minus_start",
"graph_none_share",
"graph_full_share",
"padding_fraction",
"prefill_token_fraction",
"preemptions"
],
"feature_standard_deviation": [
0.2953332526155246,
0.8546696378833459,
1.1402715194448103,
0.006588255148989237,
0.2751217635728275,
0.22612433149569594,
1.0,
0.27512176357282747,
0.048292427420964075,
0.02574874589991541,
0.1635381690436309,
0.14098674719611365,
0.02516276437103069,
61.39272994412853,
0.7234949448561444,
2.198013579605131,
0.18326586413988316,
2.4542471212960844,
6.08726412391018,
2.4074006634033043,
6.067913672185017,
0.12556414020947543,
0.03256962310836033,
0.054049610008010444,
0.048321850100969746,
0.04298231458641556,
0.041906068064246155,
0.08212268757576466,
0.4089385238422411,
1.0
],
"instrumentation_aware": true,
"regularization": 1.0,
"training_classification": {
"accuracy": 0.972972972972973,
"balanced_accuracy": 0.9444444444444444,
"brier": 0.02820726479488704,
"confusion": {
"false_negative": 0,
"false_positive": 1,
"true_negative": 8,
"true_positive": 28
},
"log_loss": 0.11247563885308659
},
"weights_with_intercept_first": [
2.109507425802979,
-0.8372240489271802,
-0.2476229678897366,
0.18172257646801393,
-0.07076358054975332,
0.3035586906752765,
0.08500005412355496,
-7.754818242684634e-26,
-0.3035586906752766,
0.4014234393196892,
0.513218716194957,
-0.35161457106287,
0.10558147889556725,
0.5674345291616134,
0.15895995157114373,
-0.4274260624362057,
-0.048791959001756195,
-0.37221380985270663,
-0.35527537277290255,
0.20582736797173468,
-0.35837576944545413,
0.2342062515631318,
0.45071068059490843,
0.3326948315186803,
0.2698892549960913,
0.017868065865726347,
-0.1540209080477302,
0.3412427440368233,
0.5831011876762794,
-0.583920360300169,
0.0
]
},
"outcome_only": {
"feature_mean": [
0.8984976998643891,
0.8378378378378378,
4.324324324324325,
0.07552086023066117,
0.6758807403968693,
0.9459459459459459,
0.0,
0.3241192596031305,
0.04468545442770093,
0.025590516908533558,
0.23873649352596996,
0.1943716628122394,
0.4321792125178198
],
"feature_names": [
"log_offered_rate_per_gpu",
"log2_tp",
"log2_max_num_seqs",
"admitted_fraction",
"completed_over_admitted",
"completed_pass_rate",
"completed_fail_fraction_of_total",
"outstanding_over_admitted",
"ttft_max_over_slo_max",
"ttft_mean_over_slo_max",
"tpot_max_over_slo",
"tpot_mean_over_slo",
"admitted_input_tokens_mean_over_limit"
],
"feature_standard_deviation": [
0.2953332526155246,
0.8546696378833459,
1.1402715194448103,
0.006588255148989237,
0.2751217635728275,
0.22612433149569594,
1.0,
0.27512176357282747,
0.048292427420964075,
0.02574874589991541,
0.1635381690436309,
0.14098674719611365,
0.02516276437103069
],
"instrumentation_aware": false,
"regularization": 1.0,
"training_classification": {
"accuracy": 0.9459459459459459,
"balanced_accuracy": 0.8888888888888888,
"brier": 0.051887373873176545,
"confusion": {
"false_negative": 0,
"false_positive": 2,
"true_negative": 7,
"true_positive": 28
},
"log_loss": 0.184988719119571
},
"weights_with_intercept_first": [
1.8996338126233983,
-1.1536861934230125,
-0.3806404559018098,
0.5901136731733696,
0.022432085012851908,
0.5805554730881304,
0.25786307099613026,
-8.077935669463161e-27,
-0.5805554730881304,
-0.15413292402348447,
0.0986842306063204,
-0.5181573573074624,
0.06283513013708956,
0.911619634884147
]
}
},
"provenance": {
"phase6_metrics": "/home/gahow/phd/aituner/runs/opprof-phase6/phase6/metrics.json",
"phase6_metrics_sha256": "290ba7fcb8727291166de7e4d47afdc84e230052495c81dd087db0ace9f93a16",
"prefix_metrics": "/home/gahow/phd/aituner/runs/fidelity-headroom/prefix-metrics.json",
"prefix_metrics_sha256": "cda821bcde1ae8427507aa4f03a1c116ccc7f7b8b717f73ca587bee3670a0340",
"raw_root": "/home/gahow/phd/aituner/runs/opprof-phase6/phase6/solo-authoritative/cells"
},
"regularization": 1.0,
"reject_probability": 0.050000000000000044,
"sanity": {
"cells": 12,
"invariants": {
"both_labels": true,
"cells_12": true,
"cutoff_5s": true,
"n_37": true,
"threshold_0.95": true
},
"n": 37,
"negative": 9,
"positive": 28
},
"schema": "fidelity-prefix-model-v1",
"status": "FROZEN_BEFORE_PROSPECTIVE_RUN",
"training_examples": [
{
"anchor": 0.24609375,
"cell": "tp1_mns16",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.25,
"cell": "tp1_mns16",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.5,
"cell": "tp1_mns16",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": false,
"primary_feasible": false
},
{
"anchor": 0.2421875,
"cell": "tp1_mns32",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.24609375,
"cell": "tp1_mns32",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.25,
"cell": "tp1_mns32",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.5,
"cell": "tp1_mns32",
"completion_time_source": "none_completed",
"label_feasible": false,
"primary_feasible": false
},
{
"anchor": 0.2421875,
"cell": "tp1_mns64",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.24609375,
"cell": "tp1_mns64",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.25,
"cell": "tp1_mns64",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.5,
"cell": "tp1_mns64",
"completion_time_source": "none_completed",
"label_feasible": false,
"primary_feasible": false
},
{
"anchor": 0.21875,
"cell": "tp1_mns8",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.2265625,
"cell": "tp1_mns8",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.23046875,
"cell": "tp1_mns8",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.234375,
"cell": "tp1_mns8",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.25,
"cell": "tp1_mns8",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.5,
"cell": "tp1_mns8",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": false,
"primary_feasible": false
},
{
"anchor": 0.4921875,
"cell": "tp2_mns16",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.49609375,
"cell": "tp2_mns16",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.5,
"cell": "tp2_mns16",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.75,
"cell": "tp2_mns32",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.75390625,
"cell": "tp2_mns32",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": false,
"primary_feasible": false
},
{
"anchor": 0.5,
"cell": "tp2_mns64",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.75,
"cell": "tp2_mns64",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": false,
"primary_feasible": false
},
{
"anchor": 0.4921875,
"cell": "tp2_mns8",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.49609375,
"cell": "tp2_mns8",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": false,
"primary_feasible": false
},
{
"anchor": 0.033182214016,
"cell": "tp4_mns16",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": false
},
{
"anchor": 0.033717411016,
"cell": "tp4_mns16",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": false,
"primary_feasible": false
},
{
"anchor": 0.034252608017,
"cell": "tp4_mns16",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.033717411016,
"cell": "tp4_mns32",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": false
},
{
"anchor": 0.034252608017,
"cell": "tp4_mns32",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.033717411016,
"cell": "tp4_mns64",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": false
},
{
"anchor": 0.034252608017,
"cell": "tp4_mns64",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.016055910008,
"cell": "tp4_mns8",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.016591107009,
"cell": "tp4_mns8",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.017126304009,
"cell": "tp4_mns8",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": true,
"primary_feasible": true
},
{
"anchor": 0.034252608017,
"cell": "tp4_mns8",
"completion_time_source": "reconstructed_from_latency",
"label_feasible": false,
"primary_feasible": false
}
],
"training_split_role": "historical training only; never headline test"
}

1142
runs/fidelity-headroom/metrics.json Normal file
View File

File diff suppressed because it is too large Load Diff

436
runs/fidelity-headroom/pilot_controller.py Normal file
View File

@@ -0,0 +1,436 @@
#!/usr/bin/env python3
"""Serialized dash0 controller for the exact-timestamp prefix pilot."""
from __future__ import annotations
import argparse
import json
import os
import shlex
import subprocess
import sys
import time
from pathlib import Path
from typing import Any
HERE = Path(__file__).resolve().parent
PHASE6 = HERE.parent / "opprof-phase6"
sys.path.insert(0, str(PHASE6))
import opprof_phase6_controller as base # noqa: E402
ORDER = (
"tp1_mns8",
"tp1_mns64",
"tp2_mns8",
"tp2_mns64",
"tp4_mns16",
"tp4_mns64",
)
CELL_ESTIMATE_H20_HOURS = {1: 0.20, 2: 0.40, 4: 0.80}
SAFETY_H20_HOURS = 0.20
def atomic_json(path: Path, payload: Any) -> None:
base.atomic_json(path, payload)
def wait_all_idle(timeout_s: float = 30.0) -> None:
deadline = time.monotonic() + timeout_s
last_error: Exception | None = None
while time.monotonic() < deadline:
try:
base.assert_all_idle()
return
except RuntimeError as error:
last_error = error
time.sleep(1.0)
raise last_error or RuntimeError("GPU idle timeout")
def configure_base(args: argparse.Namespace, manifest: dict[str, Any]) -> None:
base.WORKDIR = args.run_root.parent
base.RUN_ROOT = args.run_root
base.STATE = args.run_root / "controller-state.json"
base.SOURCE = args.vllm_source
base.VENV = args.venv
base.AITUNER = args.aituner_root
base.MODEL = args.model
base.CLIENT = args.client
base.GPU_LIMIT = float(manifest["execution"]["hard_cap_h20_hours"])
base.MARKER = "fidelity-prefix-pilot-20260714"
base.CELLS = {
cell: {"tp": int(config["tp"]), "mns": int(config["mns"])}
for cell, config in manifest["cells"].items()
}
def load_state(path: Path, hard_cap: float) -> dict[str, Any]:
if path.exists():
return json.loads(path.read_text(encoding="utf-8"))
return {
"schema": "fidelity-prefix-pilot-state-v1",
"status": "initialized",
"hard_cap_h20_hours": hard_cap,
"gpu_hours_total": 0.0,
"completed_cells": 0,
"cells": {},
"failures": [],
"started_at": time.time(),
}
def save_state(path: Path, state: dict[str, Any]) -> None:
atomic_json(path, state)
def append_echo(run_root: Path, line: str) -> None:
run_root.mkdir(parents=True, exist_ok=True)
with (run_root / "launch-echo.log").open("a", encoding="utf-8") as target:
target.write(line + "\n")
print(line, flush=True)
def remaining_projection(manifest: dict[str, Any], index: int) -> float:
return sum(
CELL_ESTIMATE_H20_HOURS[int(manifest["cells"][cell]["tp"])]
for cell in ORDER[index:]
) + SAFETY_H20_HOURS
def start_server(
*,
cell: str,
index: int,
run_root: Path,
) -> dict[str, Any]:
config = base.CELLS[cell]
gpus = tuple(range(int(config["tp"])))
cell_root = run_root / "cells" / cell
cell_root.mkdir(parents=True, exist_ok=True)
port = 8900 + index
command = base.server_command(cell, gpus, port)
with (cell_root / "commands.log").open("a", encoding="utf-8") as log:
log.write(f"SERVER {shlex.join(command)}\n")
server_log = (cell_root / "server.log").open("ab", buffering=0)
environment = os.environ.copy()
environment.update(
{
"CUDA_VISIBLE_DEVICES": ",".join(map(str, gpus)),
"VLLM_OPPROF_DIR": str(cell_root / "opprof"),
"OPPROF_PHASE6_MARKER": base.MARKER,
"AITUNER_ROOT": str(base.AITUNER),
"HF_HUB_OFFLINE": "1",
"TRANSFORMERS_OFFLINE": "1",
"PYTHONUNBUFFERED": "1",
}
)
server = subprocess.Popen(
command,
cwd=base.SOURCE,
env=environment,
stdout=server_log,
stderr=subprocess.STDOUT,
start_new_session=True,
)
base.OWNED_PGIDS.add(server.pid)
return {
"cell": cell,
"gpus": gpus,
"port": port,
"dir": cell_root,
"server": server,
"server_handle": server_log,
"spawned_at": time.time(),
"results": [],
}
def selection_for(
manifest: dict[str, Any], cell: str, role: str
) -> tuple[str, dict[str, Any]]:
level = "low" if role == "burnin" or role.startswith("low") else "high"
return level, manifest["cells"][cell]["targets"][level]["selections"][role]
def client_command(
entry: dict[str, Any],
*,
role: str,
selection: dict[str, Any],
output: Path,
warmup: bool,
) -> list[str]:
config = base.CELLS[entry["cell"]]
return [
"taskset",
"-c",
base.cpu_mask(entry["gpus"]),
str(base.VENV / "bin/python"),
str(base.CLIENT),
"warmup" if warmup else "run-anchor",
"--study",
str(selection["study"]),
"--cell",
entry["cell"],
"--anchor",
str(selection["anchor"]),
"--tp",
str(config["tp"]),
"--mns",
str(config["mns"]),
"--base-url",
f"http://127.0.0.1:{entry['port']}",
"--result-dir",
str(output),
]
def run_client(
*,
entry: dict[str, Any],
role: str,
selection: dict[str, Any],
output: Path,
state: dict[str, Any],
warmup: bool = False,
) -> dict[str, Any]:
command = client_command(
entry, role=role, selection=selection, output=output, warmup=warmup
)
with (entry["dir"] / "commands.log").open("a", encoding="utf-8") as log:
log.write(f"CLIENT role={role} {shlex.join(command)}\n")
handle = (output.parent / f"{output.name}.log").open("ab", buffering=0)
environment = os.environ.copy()
environment.update({"AITUNER_ROOT": str(base.AITUNER), "PYTHONUNBUFFERED": "1"})
process = subprocess.Popen(
command,
cwd=base.WORKDIR,
env=environment,
stdout=handle,
stderr=subprocess.STDOUT,
start_new_session=True,
)
deadline = time.monotonic() + 180.0
try:
while process.poll() is None:
if time.monotonic() > deadline:
process.terminate()
raise TimeoutError(f"client timeout: {entry['cell']} {role}")
if entry["server"].poll() is not None:
raise RuntimeError(f"server exited during {entry['cell']} {role}")
base.assert_no_other_compute()
if state["gpu_hours_total"] + base.live_gpu_hours([entry]) >= base.GPU_LIMIT:
process.terminate()
raise RuntimeError("pilot H20-hour hard cap reached")
time.sleep(1.0)
finally:
handle.close()
if process.returncode:
raise RuntimeError(
f"client failed: cell={entry['cell']} role={role} rc={process.returncode}"
)
result = json.loads((output / "result.json").read_text(encoding="utf-8"))
if int(result["selection"]["count"]) != int(selection["selected_count"]):
raise RuntimeError(f"selection count mismatch: {entry['cell']} {role}")
for key in (
"request_id_order_sha256",
"arrival_order_sha256",
"raw_length_order_sha256",
):
manifest_key = (
"input_length_order_sha256" if key == "raw_length_order_sha256" else key
)
if result["selection"][key] != selection[manifest_key]:
raise RuntimeError(f"selection hash mismatch {key}: {entry['cell']} {role}")
entry["results"].append(
{"anchor": float(selection["anchor"]), "dir": str(output), "kind": result["kind"]}
)
return result
def execute_cell(
*,
index: int,
cell: str,
manifest: dict[str, Any],
run_root: Path,
state_path: Path,
state: dict[str, Any],
) -> None:
if state["cells"].get(cell, {}).get("status") == "complete":
return
projection = remaining_projection(manifest, index)
if state["gpu_hours_total"] + projection > base.GPU_LIMIT:
state["status"] = "budget_projection_stop"
state["budget_stop"] = {
"before_cell": cell,
"spent_h20_hours": state["gpu_hours_total"],
"remaining_projection_h20_hours": projection,
"hard_cap_h20_hours": base.GPU_LIMIT,
}
save_state(state_path, state)
raise RuntimeError(f"projected pilot cost exceeds hard cap before {cell}")
config = manifest["cells"][cell]
echo = (
f"PILOT_CELL_ECHO cell={cell} tp={config['tp']} mns={config['mns']} "
f"gpus=0-{int(config['tp']) - 1} workload={manifest['source']['window_id']} "
f"roles=burnin+low1/high1/low2/high2/low3/high3 "
f"spent_h20h={state['gpu_hours_total']:.6f} "
f"remaining_projection_h20h={projection:.3f} cap_h20h={base.GPU_LIMIT:.1f} "
f"manifest={run_root / 'pilot-manifest.json'}"
)
append_echo(run_root, echo)
wait_all_idle()
cell_state = {
"status": "starting",
"tp": int(config["tp"]),
"mns": int(config["mns"]),
"started_at": time.time(),
"runs": [],
}
state["status"] = "running"
state["cells"][cell] = cell_state
save_state(state_path, state)
entry = start_server(cell=cell, index=index, run_root=run_root)
failure: Exception | None = None
try:
base.wait_ready(entry)
_level, burnin = selection_for(manifest, cell, "burnin")
cell_state["status"] = "warmup"
save_state(state_path, state)
warmup = run_client(
entry=entry,
role="burnin",
selection=burnin,
output=entry["dir"] / "warmup",
state=state,
warmup=True,
)
cell_state["warmup"] = {
"exact_output_count": warmup["exact_output_count"],
"long_gt4096": warmup["selection"]["long_gt4096"],
}
cell_state["status"] = "burnin"
save_state(state_path, state)
burnin_result = run_client(
entry=entry,
role="burnin",
selection=burnin,
output=entry["dir"] / "burnin",
state=state,
)
cell_state["burnin"] = {
"pass_rate": burnin_result["pass_rate"],
"feasible": burnin_result["feasible"],
}
role_order = manifest["execution"][
"even_cell_order" if index % 2 == 0 else "odd_cell_order"
]
cell_state["status"] = "measured"
cell_state["role_order"] = role_order
save_state(state_path, state)
for role in role_order:
level, selection = selection_for(manifest, cell, role)
result = run_client(
entry=entry,
role=role,
selection=selection,
output=entry["dir"] / f"{level}-rep{role[-1]}",
state=state,
)
cell_state["runs"].append(
{
"role": role,
"level": level,
"anchor": selection["anchor"],
"selected_count": selection["selected_count"],
"pass_rate": result["pass_rate"],
"feasible": result["feasible"],
"elapsed_s": result["interval"]["elapsed_s"],
}
)
save_state(state_path, state)
cell_state["status"] = "stopping"
save_state(state_path, state)
except Exception as error: # noqa: BLE001
failure = error
finally:
try:
base.stop_entry(entry)
except Exception as error: # noqa: BLE001
failure = failure or error
time.sleep(2.0)
try:
wait_all_idle()
except Exception as error: # noqa: BLE001
failure = failure or error
cell_hours = base.live_gpu_hours([entry])
state["gpu_hours_total"] += cell_hours
cell_state["gpu_hours"] = cell_hours
if failure is not None:
cell_state["status"] = "failed"
cell_state["failure"] = repr(failure)
state["status"] = "failed"
state["failures"].append({"cell": cell, "failure": repr(failure)})
save_state(state_path, state)
raise failure
validation = base.validate_cell(entry)
cell_state["validation"] = validation
cell_state["status"] = "complete"
cell_state["completed_at"] = time.time()
state["completed_cells"] += 1
save_state(state_path, state)
def parser() -> argparse.ArgumentParser:
result = argparse.ArgumentParser()
result.add_argument("--manifest", type=Path, required=True)
result.add_argument("--run-root", type=Path, required=True)
result.add_argument("--aituner-root", type=Path, required=True)
result.add_argument("--vllm-source", type=Path, required=True)
result.add_argument("--venv", type=Path, required=True)
result.add_argument("--model", type=Path, required=True)
result.add_argument("--client", type=Path, required=True)
return result
def main() -> None:
args = parser().parse_args()
manifest = json.loads(args.manifest.read_text(encoding="utf-8"))
if manifest["status"] != "PASS":
raise RuntimeError("pilot manifest did not pass preflight")
args.run_root.mkdir(parents=True, exist_ok=True)
copied_manifest = args.run_root / "pilot-manifest.json"
if not copied_manifest.exists():
atomic_json(copied_manifest, manifest)
configure_base(args, manifest)
state_path = args.run_root / "controller-state.json"
state = load_state(state_path, base.GPU_LIMIT)
state["status"] = "running"
save_state(state_path, state)
for index, cell in enumerate(ORDER):
execute_cell(
index=index,
cell=cell,
manifest=manifest,
run_root=args.run_root,
state_path=state_path,
state=state,
)
state["status"] = "complete"
state["completed_at"] = time.time()
save_state(state_path, state)
print(json.dumps({
"status": state["status"],
"completed_cells": state["completed_cells"],
"gpu_hours_total": state["gpu_hours_total"],
}, sort_keys=True))
if __name__ == "__main__":
main()

3696
runs/fidelity-headroom/prefix-metrics.json Normal file
View File

File diff suppressed because it is too large Load Diff

351
runs/fidelity-headroom/prepare_pilot.py Normal file
View File

@@ -0,0 +1,351 @@
#!/usr/bin/env python3
"""Materialize session-disjoint pilot repeats and freeze attainable anchors.
The private outputs retain prompt text and stay on the experiment host. The
public manifest contains only aggregate counts, hashes, paths, and parameters.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import math
import os
import sys
from pathlib import Path
from typing import Any
AITUNER_ROOT = Path(os.environ.get("AITUNER_ROOT", Path(__file__).resolve().parents[2]))
sys.path.insert(0, str(AITUNER_ROOT / "src"))
from aituner.spec import load_study_spec # noqa: E402
from aituner.trace import load_trace_requests, select_requests_for_threshold # noqa: E402
ROLES = ("burnin", "low1", "high1", "low2", "high2", "low3", "high3")
CELLS = {
"tp1_mns8": {"tp": 1, "mns": 8, "frontier_req_s_gpu": 2.3833333333333333},
"tp1_mns64": {"tp": 1, "mns": 64, "frontier_req_s_gpu": 2.3833333333333333},
"tp2_mns8": {"tp": 2, "mns": 8, "frontier_req_s_gpu": 2.2416666666666667},
"tp2_mns64": {"tp": 2, "mns": 64, "frontier_req_s_gpu": 2.3},
"tp4_mns16": {"tp": 4, "mns": 16, "frontier_req_s_gpu": 2.5},
"tp4_mns64": {"tp": 4, "mns": 64, "frontier_req_s_gpu": 2.5},
}
TARGET_MULTIPLIERS = {"low": 0.85, "high": 1.25}
def atomic_json(path: Path, payload: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(path.suffix + ".tmp")
tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
os.replace(tmp, path)
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 order_hash(values: list[str]) -> str:
return hashlib.sha256("\n".join(values).encode()).hexdigest()
def resolve_source_trace(windows_path: Path, window_id: str) -> tuple[dict[str, Any], Path]:
payload = json.loads(windows_path.read_text(encoding="utf-8"))
for window in payload["windows"]:
if window["window_id"] != window_id:
continue
trace = Path(window["trace_file"])
if not trace.is_absolute():
trace = (windows_path.parent / trace).resolve()
return window, trace
raise ValueError(f"window not found: {window_id}")
def materialize_bands(
source_trace: Path,
source_window: dict[str, Any],
private_root: Path,
) -> tuple[Path, dict[str, Any]]:
traces_root = private_root / "traces"
traces_root.mkdir(parents=True, exist_ok=True)
temporary = {role: traces_root / f".{role}.jsonl.tmp" for role in ROLES}
final = {role: traces_root / f"{role}.jsonl" for role in ROLES}
handles = {role: temporary[role].open("w", encoding="utf-8") for role in ROLES}
stats = {
role: {
"rows": 0,
"sum_input_tokens": 0,
"min_timestamp": None,
"max_timestamp": None,
}
for role in ROLES
}
try:
with source_trace.open(encoding="utf-8") as source:
for line_number, line in enumerate(source):
row = json.loads(line)
value = float(row["sampling_u"])
if not 0.0 <= value <= 1.0:
raise ValueError(f"sampling_u outside [0,1] at line {line_number}")
band = min(len(ROLES) - 1, int(value * len(ROLES)))
role = ROLES[band]
remapped = value * len(ROLES) - band
row["sampling_u"] = min(remapped, math.nextafter(1.0, 0.0))
row["fidelity_pilot_band"] = role
handles[role].write(json.dumps(row, ensure_ascii=False) + "\n")
timestamp = float(row["timestamp"])
item = stats[role]
item["rows"] += 1
item["sum_input_tokens"] += int(row.get("input_length") or 0)
item["min_timestamp"] = (
timestamp if item["min_timestamp"] is None
else min(float(item["min_timestamp"]), timestamp)
)
item["max_timestamp"] = (
timestamp if item["max_timestamp"] is None
else max(float(item["max_timestamp"]), timestamp)
)
finally:
for handle in handles.values():
handle.close()
for role in ROLES:
os.replace(temporary[role], final[role])
stats[role]["sha256"] = sha256_file(final[role])
stats[role]["bytes"] = final[role].stat().st_size
windows = []
for role in ROLES:
window = dict(source_window)
window["window_id"] = f"fidelity_pilot_{role}"
window["trace_file"] = f"traces/{role}.jsonl"
window["num_requests"] = stats[role]["rows"]
window["sum_input_length"] = stats[role]["sum_input_tokens"]
window["sampling_strategy"] = "session_uniform_seven_disjoint_bands_remapped"
window["fidelity_pilot_role"] = role
windows.append(window)
private_windows = private_root / "windows.json"
atomic_json(
private_windows,
{
"schema": "fidelity-pilot-private-windows-v1",
"roles": list(ROLES),
"windows": windows,
},
)
return private_windows, stats
def write_studies(
*,
base_primary: Path,
base_tp4: Path,
private_windows: Path,
private_root: Path,
) -> dict[str, dict[str, Path]]:
bases = {
"primary": json.loads(base_primary.read_text(encoding="utf-8")),
"tp4": json.loads(base_tp4.read_text(encoding="utf-8")),
}
result: dict[str, dict[str, Path]] = {}
for role in ROLES:
result[role] = {}
for tier, base in bases.items():
payload = json.loads(json.dumps(base))
payload["study_id"] = f"fidelity-prefix-pilot-{role}-{tier}"
payload["hardware"]["host_candidates"] = ["dash0"]
payload["engine"]["engine_version"] = "0.24.1.dev3+opprof"
payload["trace"]["windows_path"] = str(private_windows)
payload["trace"]["window_id"] = f"fidelity_pilot_{role}"
path = private_root / "studies" / f"{role}-{tier}.json"
atomic_json(path, payload)
result[role][tier] = path
return result
def attainable_anchor(requests: list[Any], target_count: int) -> tuple[float, list[Any]]:
ordered = sorted(float(request.sampling_u) for request in requests)
if not ordered:
raise ValueError("no requests after study filtering")
candidate_indices = sorted({
max(0, min(len(ordered) - 1, target_count - 1)),
max(0, min(len(ordered) - 1, target_count)),
})
candidates = []
for index in candidate_indices:
anchor = ordered[index]
selected = select_requests_for_threshold(requests, threshold=anchor)
candidates.append((abs(len(selected) - target_count), len(selected), anchor, selected))
_error, _count, anchor, selected = min(candidates, key=lambda item: (item[0], item[1]))
return anchor, selected
def selected_record(selected: list[Any], *, tp: int, duration_s: float) -> dict[str, Any]:
return {
"anchor": max(float(request.sampling_u) for request in selected),
"selected_count": len(selected),
"offered_req_s": len(selected) / duration_s,
"offered_req_s_per_gpu": len(selected) / duration_s / tp,
"request_id_order_sha256": order_hash([request.row_id for request in selected]),
"arrival_order_sha256": order_hash([f"{request.arrival_s:.12f}" for request in selected]),
"input_length_order_sha256": order_hash(
[str(request.prompt_tokens_hint) for request in selected]
),
}
def build_manifest(
*,
studies: dict[str, dict[str, Path]],
private_windows: Path,
band_stats: dict[str, Any],
source_trace: Path,
source_windows: Path,
source_window_id: str,
) -> dict[str, Any]:
loaded = {}
durations = {}
for role, tiers in studies.items():
loaded[role] = {}
for tier, path in tiers.items():
study = load_study_spec(path)
window, requests = load_trace_requests(study, study_spec_path=path)
loaded[role][tier] = requests
durations[role] = float(window.window_end - window.window_start)
cells = {}
all_hashes = []
for cell, config in CELLS.items():
tp = int(config["tp"])
tier = "tp4" if tp == 4 else "primary"
targets = {}
for level, multiplier in TARGET_MULTIPLIERS.items():
target_rate = float(config["frontier_req_s_gpu"]) * multiplier
target_count = round(target_rate * durations["low1"] * tp)
roles = [role for role in ROLES if role == "burnin" or role.startswith(level)]
selections = {}
for role in roles:
anchor, selected = attainable_anchor(loaded[role][tier], target_count)
record = selected_record(selected, tp=tp, duration_s=durations[role])
record["anchor"] = anchor
record["study"] = str(studies[role][tier])
selections[role] = record
all_hashes.append(record["request_id_order_sha256"])
targets[level] = {
"multiplier": multiplier,
"target_req_s_per_gpu": target_rate,
"target_count": target_count,
"selections": selections,
}
cells[cell] = {**config, "targets": targets}
red_flags = []
for cell, config in cells.items():
for level, target in config["targets"].items():
if not target["selections"]:
red_flags.append(f"missing_{cell}_{level}")
for selection in target["selections"].values():
if selection["selected_count"] <= 0:
red_flags.append(f"empty_{cell}_{level}")
per_cell_distinct = {}
for cell, config in cells.items():
hashes = [
selection["request_id_order_sha256"]
for target in config["targets"].values()
for selection in target["selections"].values()
]
per_cell_distinct[cell] = len(hashes) == len(set(hashes))
if not per_cell_distinct[cell]:
red_flags.append(f"session_bands_overlap_{cell}")
return {
"schema": "fidelity-prefix-pilot-manifest-v1",
"status": "PASS" if not red_flags else "STOP",
"source": {
"windows": str(source_windows),
"window_id": source_window_id,
"trace": str(source_trace),
"trace_sha256": sha256_file(source_trace),
},
"private": {
"windows": str(private_windows),
"windows_sha256": sha256_file(private_windows),
"band_stats": band_stats,
"studies": {
role: {tier: str(path) for tier, path in tiers.items()}
for role, tiers in studies.items()
},
},
"roles": list(ROLES),
"cells": cells,
"execution": {
"cutoff_s": 5.0,
"replicates_per_level": 3,
"label": "2-of-3 session-disjoint repetitions",
"even_cell_order": ["low1", "high1", "high2", "low2", "low3", "high3"],
"odd_cell_order": ["high1", "low1", "low2", "high2", "high3", "low3"],
"hard_cap_h20_hours": 3.5,
},
"sanity": {
"red_flags": red_flags,
"n_cells": len(cells),
"n_roles": len(ROLES),
"selected_sets": len(all_hashes),
"distinct_selected_sets": len(set(all_hashes)),
"per_cell_selected_sets_distinct": per_cell_distinct,
"invariants": {
"cells_6": len(cells) == 6,
"roles_7": len(ROLES) == 7,
"band_rows_nonzero": all(stats["rows"] > 0 for stats in band_stats.values()),
"session_bands_disjoint_per_cell": all(per_cell_distinct.values()),
},
},
}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--source-windows", type=Path, required=True)
parser.add_argument("--source-window-id", default="chat_w20260312_1000")
parser.add_argument("--base-primary-study", type=Path, required=True)
parser.add_argument("--base-tp4-study", type=Path, required=True)
parser.add_argument("--private-root", type=Path, required=True)
parser.add_argument("--public-manifest", type=Path, required=True)
args = parser.parse_args()
source_window, source_trace = resolve_source_trace(
args.source_windows, args.source_window_id
)
private_windows, band_stats = materialize_bands(
source_trace, source_window, args.private_root
)
studies = write_studies(
base_primary=args.base_primary_study,
base_tp4=args.base_tp4_study,
private_windows=private_windows,
private_root=args.private_root,
)
manifest = build_manifest(
studies=studies,
private_windows=private_windows,
band_stats=band_stats,
source_trace=source_trace,
source_windows=args.source_windows,
source_window_id=args.source_window_id,
)
atomic_json(args.public_manifest, manifest)
print(json.dumps({
"status": manifest["status"],
"manifest": str(args.public_manifest),
"sanity": manifest["sanity"],
}, sort_keys=True))
if manifest["status"] != "PASS":
raise RuntimeError(f"pilot preflight failed: {manifest['sanity']['red_flags']}")
if __name__ == "__main__":
main()

40
runs/fidelity-headroom/test_analysis.py Normal file
View File

@@ -0,0 +1,40 @@
#!/usr/bin/env python3
from __future__ import annotations
import importlib.util
import math
import sys
from pathlib import Path
HERE = Path(__file__).resolve().parent
def load_analysis():
spec = importlib.util.spec_from_file_location("fidelity_headroom", HERE / "analyze_existing.py")
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
def main() -> None:
analysis = load_analysis()
curve = analysis.topk_curve(
{"a": 3.0, "b": 2.0, "c": 1.0},
{"a": 1.0, "b": 2.0, "c": 2.0},
2e-6,
)
assert curve["points"][0]["expanded_k"] == 2
assert curve["points"][0]["candidates"] == ["b", "c"]
assert math.isclose(curve["points"][0]["real_regret"], 1.0 / 3.0)
assert curve["points"][2]["real_regret"] == 0.0
assert curve["minimum_k"]["five_percent"] == {"nominal_k": 3, "expanded_k": 3}
assert analysis._mcnemar_exact_p(0, 1) == 1.0
assert analysis._mcnemar_exact_p(0, 5) == 0.0625
print("fidelity headroom analysis: PASS")
if __name__ == "__main__":
main()

85
runs/fidelity-headroom/test_pilot_tools.py Normal file
View File

@@ -0,0 +1,85 @@
#!/usr/bin/env python3
from __future__ import annotations
import json
import math
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import pilot_controller as controller # noqa: E402
import prepare_pilot as prepare # noqa: E402
@dataclass
class Request:
row_id: str
sampling_u: float
arrival_s: float = 0.0
prompt_tokens_hint: int = 1
def main() -> None:
requests = [
Request("a", 0.1),
Request("b", 0.2),
Request("c", 0.2),
Request("d", 0.9),
]
anchor, selected = prepare.attainable_anchor(requests, target_count=2)
assert anchor == 0.2
assert [request.row_id for request in selected] == ["a", "b", "c"]
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
source = root / "source.jsonl"
rows = []
for index, role in enumerate(prepare.ROLES):
rows.append(
{
"request_id": role,
"timestamp": float(index),
"sampling_u": (index + 0.5) / len(prepare.ROLES),
"input_length": 16 + index,
"messages": [{"role": "user", "content": role}],
}
)
source.write_text(
"".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8"
)
windows, stats = prepare.materialize_bands(
source,
{
"window_id": "source",
"trace_type": "chat",
"window_start": 0.0,
"window_end": 600.0,
},
root / "private",
)
assert windows.is_file()
assert all(stats[role]["rows"] == 1 for role in prepare.ROLES)
for role in prepare.ROLES:
row = json.loads((root / "private" / "traces" / f"{role}.jsonl").read_text())
assert row["fidelity_pilot_band"] == role
assert abs(float(row["sampling_u"]) - 0.5) < 1e-12
assert len(controller.ORDER) == 6
assert set(controller.ORDER) == set(prepare.CELLS)
assert math.isclose(
sum(
controller.CELL_ESTIMATE_H20_HOURS[int(config["tp"])]
for config in prepare.CELLS.values()
) + controller.SAFETY_H20_HOURS,
3.0,
)
print("fidelity pilot tools: PASS")
if __name__ == "__main__":
main()

72
runs/fidelity-headroom/test_prefix_analysis.py Normal file
View File

@@ -0,0 +1,72 @@
#!/usr/bin/env python3
from __future__ import annotations
import math
import sys
from pathlib import Path
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import analyze_prefixes as analysis # noqa: E402
def main() -> None:
exact, exact_source = analysis.completion_elapsed_s(
{"completed_elapsed_s": 7.25}
)
assert exact == 7.25 and exact_source == "exact_monotonic"
reconstructed, reconstructed_source = analysis.completion_elapsed_s(
{
"success": True,
"arrival_s": 2.0,
"ttft_ms": 100.0,
"tpot_ms": 10.0,
"completion_tokens": 11,
}
)
assert math.isclose(reconstructed or 0.0, 2.2)
assert reconstructed_source == "reconstructed_from_latency"
missing, missing_source = analysis.completion_elapsed_s({"success": False})
assert missing is None and missing_source == "unobserved_failure"
examples = [
analysis.PrefixExample(
cell=f"c{index}",
anchor=float(index),
cutoff_s=5.0,
tp=1,
full_elapsed_s=65.0,
feasible=label,
primary_feasible=label,
outcome=(float(index),),
instrumentation=(float(index % 2),),
completion_time_source="exact_monotonic",
)
for index, label in enumerate((0, 1, 1))
]
labels = analysis.np.asarray([0, 1, 1])
probabilities = analysis.np.asarray([0.01, 0.99, 0.60])
policy = analysis.policy_metrics(examples, labels, probabilities, 0.95)
assert policy["early_accept"] == 1
assert policy["early_reject"] == 1
assert policy["abstain_continue_full"] == 1
assert policy["false_accept"] == 0 and policy["false_reject"] == 0
assert policy["valid_zero_error_policy"]
assert policy["valid_cost_reduction_fraction"] is not None
model = analysis.fit_frozen_model(
examples,
instrumentation_aware=True,
regularization=1.0,
)
frozen_probability = analysis.predict_frozen_model(model, examples)
assert len(frozen_probability) == len(examples)
assert analysis.np.all(frozen_probability >= 0.0)
assert analysis.np.all(frozen_probability <= 1.0)
print("fidelity prefix analysis: PASS")
if __name__ == "__main__":
main()

5
runs/opprof-phase6/opprof_phase6_client.py
View File

@@ -122,6 +122,11 @@ def run_replay(args: argparse.Namespace, *, warmup: bool) -> dict[str, Any]:
"tpot_ms": outcome.tpot_ms, "tpot_ms": outcome.tpot_ms,
"completion_tokens": outcome.completion_tokens, "completion_tokens": outcome.completion_tokens,
"completion_tokens_source": outcome.completion_tokens_source, "completion_tokens_source": outcome.completion_tokens_source,
"completed_mono_ns": outcome.completed_mono_ns,
"completed_elapsed_s": (
(outcome.completed_mono_ns - interval_start_mono_ns) / 1e9
if outcome.completed_mono_ns is not None else None
),
"slo_pass": evaluation.passed, "slo_pass": evaluation.passed,
"reasons": evaluation.reasons, "reasons": evaluation.reasons,
"error": outcome.error, "error": outcome.error,

1
src/aituner/slo.py
View File

@@ -16,6 +16,7 @@ class RequestOutcome:
completion_tokens: int | None completion_tokens: int | None
error: str = "" error: str = ""
completion_tokens_source: str = "" completion_tokens_source: str = ""
completed_mono_ns: int | None = None
@dataclass(frozen=True) @dataclass(frozen=True)

4
src/aituner/worker.py
View File

@@ -127,6 +127,7 @@ def _run_one_request(
f"actual={actual_completion_tokens}" f"actual={actual_completion_tokens}"
), ),
completion_tokens_source=completion_tokens_source, completion_tokens_source=completion_tokens_source,
completed_mono_ns=time.monotonic_ns(),
) )
if actual_completion_tokens != expected_completion_tokens: if actual_completion_tokens != expected_completion_tokens:
return RequestOutcome( return RequestOutcome(
@@ -142,6 +143,7 @@ def _run_one_request(
f"actual={actual_completion_tokens}" f"actual={actual_completion_tokens}"
), ),
completion_tokens_source=completion_tokens_source, completion_tokens_source=completion_tokens_source,
completed_mono_ns=time.monotonic_ns(),
) )
return RequestOutcome( return RequestOutcome(
request_id=request.row_id, request_id=request.row_id,
@@ -151,6 +153,7 @@ def _run_one_request(
prompt_tokens=request.prompt_tokens_hint, prompt_tokens=request.prompt_tokens_hint,
completion_tokens=actual_completion_tokens or request.completion_tokens_hint, completion_tokens=actual_completion_tokens or request.completion_tokens_hint,
completion_tokens_source=completion_tokens_source, completion_tokens_source=completion_tokens_source,
completed_mono_ns=time.monotonic_ns(),
) )
except HttpClientError as exc: except HttpClientError as exc:
return RequestOutcome( return RequestOutcome(
@@ -161,6 +164,7 @@ def _run_one_request(
prompt_tokens=request.prompt_tokens_hint, prompt_tokens=request.prompt_tokens_hint,
completion_tokens=request.completion_tokens_hint, completion_tokens=request.completion_tokens_hint,
error=str(exc), error=str(exc),
completed_mono_ns=time.monotonic_ns(),
) )

12
tests/test_core_flow.py
View File

@@ -5604,15 +5604,17 @@ class CoreFlowTests(unittest.TestCase):
completion_tokens=1, completion_tokens=1,
), ),
): ):
outcome = _run_one_request( with mock.patch("aituner.worker.time.monotonic_ns", return_value=123456789):
request, outcome = _run_one_request(
base_url="http://127.0.0.1:8000", request,
timeout_s=1.0, base_url="http://127.0.0.1:8000",
) timeout_s=1.0,
)
self.assertFalse(outcome.success) self.assertFalse(outcome.success)
self.assertEqual(outcome.error, "completion_tokens_mismatch expected=2 actual=1") self.assertEqual(outcome.error, "completion_tokens_mismatch expected=2 actual=1")
self.assertEqual(outcome.completion_tokens, 1) self.assertEqual(outcome.completion_tokens, 1)
self.assertEqual(outcome.completed_mono_ns, 123456789)
def test_build_prompt_mentions_completion_tokens_override(self) -> None: def test_build_prompt_mentions_completion_tokens_override(self) -> None:
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp: