Add simulator-aware fidelity pilot audit

This commit is contained in:
2026-07-14 13:28:16 +08:00
parent 23142aa359
commit a3b25f4a92
4 changed files with 1096 additions and 0 deletions

View File

@@ -0,0 +1,398 @@
#!/usr/bin/env python3
"""Exploratory P1 audit against the strengthened simulator-aware baseline.
P1 was already running when the strong baseline was added, so this script is
not paper-facing prospective evidence. It trains only on the historical
Phase-6 task and evaluates the exact P1 primary probes. Both nested models
receive identical Frontier predictions; engine telemetry is the sole feature
difference.
"""
from __future__ import annotations
import argparse
import json
import math
from pathlib import Path
from typing import Any
import numpy as np
from analyze_existing import (
DEFAULT_REGULARIZATION,
REGULARIZATION_SENSITIVITY,
_classification_metrics,
_fit_logistic,
_mcnemar_exact_p,
_sigmoid,
)
from analyze_pilot import build_pilot_examples
from analyze_prefixes import (
INSTRUMENTATION_FEATURES,
OUTCOME_FEATURES,
PrefixExample,
build_examples,
numeric,
policy_metrics,
sha256_file,
)
from analyze_strong_baseline import (
SIMULATOR_FEATURES,
load_simulator_features,
simulator_row,
)
def load_pilot_simulator(
path: Path,
) -> tuple[dict[tuple[str, str], tuple[float, ...]], list[str]]:
payload = json.loads(path.read_text(encoding="utf-8"))
red_flags = []
if payload.get("status") != "PASS":
red_flags.append("pilot_simulator_not_pass")
features: dict[tuple[str, str], tuple[float, ...]] = {}
for item in payload.get("results", []):
key = (str(item["cell"]), str(item["role"]))
if key in features:
red_flags.append(f"duplicate_pilot_simulator_{key[0]}_{key[1]}")
continue
scorer = item["scorer"]
throughput = float(scorer["throughput_requests_per_second_per_gpu"])
pass_rate = float(scorer["slo"]["pass_rate"])
if throughput <= 0:
red_flags.append(f"nonpositive_pilot_simulator_throughput_{key[0]}_{key[1]}")
if not 0.0 <= pass_rate <= 1.0:
red_flags.append(f"pilot_simulator_ratio_out_of_range_{key[0]}_{key[1]}")
features[key] = (
math.log(throughput),
pass_rate,
float(bool(scorer["slo"]["feasible"])),
)
if len(features) != 12:
red_flags.append("pilot_simulator_entries_not_12")
return features, red_flags
def fit_model(
examples: list[PrefixExample],
simulator: list[tuple[float, ...]],
*,
instrumentation_aware: bool,
regularization: float,
) -> dict[str, Any]:
rows = []
for example, simulator_features in zip(examples, simulator):
values = example.outcome + simulator_features
if instrumentation_aware:
values += example.instrumentation
rows.append((1.0, *values))
matrix = np.asarray(rows, dtype=np.float64)
labels = np.asarray([example.feasible for example in examples], dtype=np.float64)
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)
return {
"instrumentation_aware": instrumentation_aware,
"regularization": regularization,
"feature_mean": mean,
"feature_standard_deviation": standard_deviation,
"weights": weights,
}
def predict_model(
model: dict[str, Any],
examples: list[PrefixExample],
simulator: list[tuple[float, ...]],
) -> np.ndarray:
rows = []
for example, simulator_features in zip(examples, simulator):
values = example.outcome + simulator_features
if model["instrumentation_aware"]:
values += example.instrumentation
rows.append((1.0, *values))
matrix = np.asarray(rows, dtype=np.float64)
matrix[:, 1:] = (
matrix[:, 1:] - model["feature_mean"]
) / model["feature_standard_deviation"]
return _sigmoid(matrix @ model["weights"])
def comparison(
training_examples: list[PrefixExample],
training_simulator: list[tuple[float, ...]],
pilot_examples: list[PrefixExample],
pilot_simulator: list[tuple[float, ...]],
regularization: float,
) -> dict[str, Any]:
labels = np.asarray([example.feasible for example in pilot_examples], dtype=np.int64)
baseline_model = fit_model(
training_examples,
training_simulator,
instrumentation_aware=False,
regularization=regularization,
)
instrument_model = fit_model(
training_examples,
training_simulator,
instrumentation_aware=True,
regularization=regularization,
)
baseline_probability = predict_model(
baseline_model, pilot_examples, pilot_simulator
)
instrument_probability = predict_model(
instrument_model, pilot_examples, pilot_simulator
)
baseline_correct = (baseline_probability >= 0.5) == labels
instrument_correct = (instrument_probability >= 0.5) == labels
paired = {
"both_correct": int(np.sum(baseline_correct & instrument_correct)),
"sim_outcome_only_correct": int(
np.sum(baseline_correct & ~instrument_correct)
),
"instrumentation_only_correct": int(
np.sum(~baseline_correct & instrument_correct)
),
"both_wrong": int(np.sum(~baseline_correct & ~instrument_correct)),
}
paired["mcnemar_exact_two_sided_p"] = _mcnemar_exact_p(
paired["sim_outcome_only_correct"], paired["instrumentation_only_correct"]
)
return {
"sim_plus_outcome": {
"classification": _classification_metrics(labels, baseline_probability),
"policy_0p95": policy_metrics(
pilot_examples, labels, baseline_probability, 0.95
),
"probability": baseline_probability.tolist(),
},
"sim_plus_outcome_plus_instrumentation": {
"classification": _classification_metrics(labels, instrument_probability),
"policy_0p95": policy_metrics(
pilot_examples, labels, instrument_probability, 0.95
),
"probability": instrument_probability.tolist(),
},
"paired_correctness": paired,
}
def analyze(
phase6_path: Path,
phase6_raw_root: Path,
training_simulator_root: Path,
pilot_manifest_path: Path,
pilot_run_root: Path,
pilot_simulator_path: Path,
) -> dict[str, Any]:
phase6 = json.loads(phase6_path.read_text(encoding="utf-8"))
pilot_manifest = json.loads(pilot_manifest_path.read_text(encoding="utf-8"))
training_examples = build_examples(phase6, phase6_raw_root, 5.0)
training_simulator_map, training_simulator_sha256 = load_simulator_features(
training_simulator_root
)
training_simulator = [
simulator_row(example, training_simulator_map)
for example in training_examples
]
pilot_examples, pilot_details, red_flags = build_pilot_examples(
pilot_manifest, pilot_run_root, 5.0
)
pilot_simulator_map, simulator_red_flags = load_pilot_simulator(
pilot_simulator_path
)
red_flags.extend(simulator_red_flags)
pilot_simulator = []
for example, detail in zip(pilot_examples, pilot_details):
role = f"{detail['level']}1"
key = (example.cell, role)
if key not in pilot_simulator_map:
red_flags.append(f"missing_pilot_simulator_{example.cell}_{role}")
pilot_simulator.append((0.0, 0.0, 0.0))
else:
pilot_simulator.append(pilot_simulator_map[key])
sensitivity = {}
if not red_flags:
for regularization in REGULARIZATION_SENSITIVITY:
sensitivity[str(regularization)] = comparison(
training_examples,
training_simulator,
pilot_examples,
pilot_simulator,
regularization,
)
headline = sensitivity.get(str(DEFAULT_REGULARIZATION))
labels = [example.feasible for example in pilot_examples]
simulator_pass_rates = [row[1] for row in pilot_simulator]
simulator_labels = [int(row[2]) for row in pilot_simulator]
if len(training_examples) != 37:
red_flags.append("training_examples_not_37")
if len(pilot_examples) != 12:
red_flags.append("pilot_examples_not_12")
if len(set(labels)) != 2:
red_flags.append("pilot_single_label")
if len(set(simulator_pass_rates)) <= 1:
red_flags.append("pilot_simulator_results_identical")
if headline is None:
decision = {
"strong_incremental_gate": False,
"reason": "analysis red flag prevented nested comparison",
}
else:
baseline_policy = headline["sim_plus_outcome"]["policy_0p95"]
instrument_policy = headline[
"sim_plus_outcome_plus_instrumentation"
]["policy_0p95"]
baseline_errors = baseline_policy["false_accept"] + baseline_policy["false_reject"]
instrument_errors = (
instrument_policy["false_accept"] + instrument_policy["false_reject"]
)
baseline_reduction = baseline_policy["valid_cost_reduction_fraction"]
instrument_reduction = instrument_policy["valid_cost_reduction_fraction"]
reduction_delta = (
instrument_reduction - baseline_reduction
if baseline_reduction is not None and instrument_reduction is not None
else None
)
per_lambda_safe_and_better = []
for item in sensitivity.values():
baseline = item["sim_plus_outcome"]["policy_0p95"]
instrument = item["sim_plus_outcome_plus_instrumentation"]["policy_0p95"]
base_errors = baseline["false_accept"] + baseline["false_reject"]
inst_errors = instrument["false_accept"] + instrument["false_reject"]
base_reduction = baseline["valid_cost_reduction_fraction"]
inst_reduction = instrument["valid_cost_reduction_fraction"]
per_lambda_safe_and_better.append(
inst_errors == 0
and inst_errors <= base_errors
and base_reduction is not None
and inst_reduction is not None
and inst_reduction > base_reduction
)
decision = {
"strong_incremental_gate": bool(
not red_flags
and instrument_errors == 0
and instrument_errors <= baseline_errors
and reduction_delta is not None
and reduction_delta >= 0.15
),
"regularization_robust": all(per_lambda_safe_and_better),
"valid_cost_reduction_fraction_delta": reduction_delta,
"scope": "exploratory task; may choose P2 design but cannot establish contribution",
}
return {
"schema": "fidelity-strong-pilot-v1",
"status": "PASS" if not red_flags else "STOP",
"scope": (
"post-amendment exploratory P1 audit; strong model was not frozen before "
"partial P1 outcomes, so this is not prospective contribution evidence"
),
"features": {
"shared_outcome": list(OUTCOME_FEATURES),
"shared_simulator": list(SIMULATOR_FEATURES),
"instrumentation_only": list(INSTRUMENTATION_FEATURES),
},
"headline_regularization": DEFAULT_REGULARIZATION,
"headline": headline,
"regularization_sensitivity": sensitivity,
"simulator_only": {
"classification": _classification_metrics(
np.asarray(labels, dtype=np.int64),
np.asarray(simulator_labels, dtype=np.float64),
)
if labels
else None,
"predicted_feasible": simulator_labels,
},
"pilot_examples": [
{
**detail,
"sim_completed_throughput_per_gpu": math.exp(simulator[0]),
"sim_slo_pass_rate": simulator[1],
"sim_slo_feasible": bool(simulator[2]),
}
for detail, simulator in zip(pilot_details, pilot_simulator)
],
"decision": decision,
"provenance": {
"phase6_metrics": str(phase6_path.resolve()),
"phase6_metrics_sha256": sha256_file(phase6_path),
"phase6_raw_root": str(phase6_raw_root.resolve()),
"training_simulator_root": str(training_simulator_root.resolve()),
"training_simulator_manifest_scorer_set_sha256": training_simulator_sha256,
"pilot_manifest": str(pilot_manifest_path.resolve()),
"pilot_manifest_sha256": sha256_file(pilot_manifest_path),
"pilot_run_root": str(pilot_run_root.resolve()),
"pilot_simulator": str(pilot_simulator_path.resolve()),
"pilot_simulator_sha256": sha256_file(pilot_simulator_path),
},
"sanity": {
"red_flags": red_flags,
"training_examples": numeric([1 for _ in training_examples]),
"pilot_labels": {
**numeric(labels),
"positive": sum(labels),
"negative": len(labels) - sum(labels),
},
"pilot_simulator_pass_rate": numeric(simulator_pass_rates),
"invariants": {
"training_examples_37": len(training_examples) == 37,
"pilot_examples_12": len(pilot_examples) == 12,
"pilot_cells_6": len({example.cell for example in pilot_examples}) == 6,
"pilot_both_labels": len(set(labels)) == 2,
"simulator_ratios_bounded": all(
0.0 <= value <= 1.0 for value in simulator_pass_rates
),
"per_config_not_all_identical": len(set(simulator_pass_rates)) > 1,
"all_prefixes_exact_monotonic": all(
example.completion_time_source in {"exact_monotonic", "none_completed"}
for example in pilot_examples
),
},
},
}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--phase6-metrics", type=Path, required=True)
parser.add_argument("--phase6-raw-root", type=Path, required=True)
parser.add_argument("--training-simulator-root", type=Path, required=True)
parser.add_argument("--pilot-manifest", type=Path, required=True)
parser.add_argument("--pilot-run-root", type=Path, required=True)
parser.add_argument("--pilot-simulator", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
result = analyze(
args.phase6_metrics,
args.phase6_raw_root,
args.training_simulator_root,
args.pilot_manifest,
args.pilot_run_root,
args.pilot_simulator,
)
args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n")
print(
json.dumps(
{
"status": result["status"],
"red_flags": result["sanity"]["red_flags"],
"decision": result["decision"],
},
sort_keys=True,
)
)
if result["status"] != "PASS":
raise RuntimeError(result["sanity"]["red_flags"])
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,331 @@
#!/usr/bin/env python3
"""Prepare exact Frontier fixtures for the P1 primary low/high probes.
Prompt-bearing band traces remain under ``--private-root``. The emitted
fixtures and public manifest contain token IDs, block IDs, hashes, and
aggregate metadata, but no prompt text.
"""
from __future__ import annotations
import argparse
import hashlib
import importlib.util
import json
import math
import subprocess
import sys
from pathlib import Path
from typing import Any
from transformers import AutoTokenizer
HERE = Path(__file__).resolve().parent
AITUNER_ROOT = HERE.parents[1]
sys.path.insert(0, str(HERE))
import prepare_pilot as pilot # noqa: E402
PRIMARY_ROLES = ("low1", "high1")
def load_module(path: Path):
spec = importlib.util.spec_from_file_location("simfid_s2rb_prepare", path)
if spec is None or spec.loader is None:
raise ImportError(path)
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
def order_hash(values: list[str]) -> str:
return hashlib.sha256("\n".join(values).encode()).hexdigest()
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as source:
for chunk in iter(lambda: source.read(1 << 20), b""):
digest.update(chunk)
return digest.hexdigest()
def git_capture(root: Path, *arguments: str) -> str:
return subprocess.run(
["git", "-C", str(root), *arguments],
check=True,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
).stdout
def raw_rows(path: Path) -> dict[int, dict[str, Any]]:
result = {}
with path.open(encoding="utf-8") as source:
for index, line in enumerate(source):
if line.strip():
result[index] = json.loads(line)
return result
def selected_hashes(
selected: list[Any], rows: dict[int, dict[str, Any]]
) -> dict[str, str]:
identifiers = []
arrivals = []
lengths = []
for item in selected:
row = rows[item.row_index]
identifiers.append(str(row.get("request_id") or row.get("id") or item.row_index))
arrivals.append(f"{float(item.timestamp) * 0.1:.12f}")
lengths.append(str(int(item.input_length)))
return {
"request_id_order_sha256": order_hash(identifiers),
"arrival_order_sha256": order_hash(arrivals),
"input_length_order_sha256": order_hash(lengths),
}
def kv_blocks(raw_root: Path, cell: str) -> int:
stream = next((raw_root / cell / "opprof").glob("*.jsonl"))
with stream.open(encoding="utf-8") as source:
for line in source:
record = json.loads(line)
if "step_index" in record:
return int(record["kv"]["total_blocks"])
raise ValueError(f"no Layer-1 record for {cell}")
def source_window(windows_path: Path, window_id: str) -> tuple[dict[str, Any], Path]:
return pilot.resolve_source_trace(windows_path, window_id)
def prepare(args: argparse.Namespace) -> dict[str, Any]:
simulator = load_module(args.replayserve_root / "tools/simfid_s2rb_prepare.py")
manifest = json.loads(args.pilot_manifest.read_text(encoding="utf-8"))
window, trace = source_window(args.source_windows, args.source_window_id)
if args.band_root is not None:
role_paths = {
role: (args.band_root / f"{role}.jsonl").resolve()
for role in PRIMARY_ROLES
}
band_stats = {
role: manifest["private"]["band_stats"][role]
for role in PRIMARY_ROLES
}
for role, path in role_paths.items():
if sha256_file(path) != band_stats[role]["sha256"]:
raise ValueError(f"pre-materialized band hash mismatch: {role}")
private_windows = None
else:
private_windows, all_band_stats = pilot.materialize_bands(
trace, window, args.private_root
)
private_payload = json.loads(private_windows.read_text(encoding="utf-8"))
role_paths = {
item["fidelity_pilot_role"]: (
private_windows.parent / item["trace_file"]
).resolve()
for item in private_payload["windows"]
}
band_stats = {
role: all_band_stats[role]
for role in PRIMARY_ROLES
}
tokenizer = AutoTokenizer.from_pretrained(
args.tokenizer, local_files_only=True, use_fast=True
)
fixture_root = args.output / "fixtures"
config_root = args.output / "configs"
fixture_root.mkdir(parents=True, exist_ok=True)
config_root.mkdir(parents=True, exist_ok=True)
entries = []
red_flags = []
for role in PRIMARY_ROLES:
trace_path = role_paths[role]
retained, trace_stats = simulator.scan_trace(trace_path)
rows = raw_rows(trace_path)
primary_pool = [retained[(index * len(retained)) // 512] for index in range(512)]
selections: dict[str, list[Any]] = {}
selected_union: set[int] = set()
for cell, cell_manifest in sorted(manifest["cells"].items()):
level = "low" if role.startswith("low") else "high"
expected = cell_manifest["targets"][level]["selections"][role]
pool = retained if int(cell_manifest["tp"]) == 4 else primary_pool
selected = [item for item in pool if item.sampling_u <= float(expected["anchor"])]
selections[cell] = selected
selected_union.update(item.row_index for item in selected)
hashes = selected_hashes(selected, rows)
if len(selected) != int(expected["selected_count"]):
red_flags.append(f"selection_count_{cell}_{role}")
for key, value in hashes.items():
if value != expected[key]:
red_flags.append(f"selection_hash_{cell}_{role}_{key}")
token_gates, selected_records, block_stats = simulator.tokenize_and_hash(
trace=trace_path,
tokenizer=tokenizer,
retained=retained,
selected_union=selected_union,
)
if any(gate["status"] != "pass" for gate in token_gates.values()):
red_flags.append(f"token_gate_{role}")
for cell, selected in selections.items():
cell_manifest = manifest["cells"][cell]
level = "low" if role.startswith("low") else "high"
expected = cell_manifest["targets"][level]["selections"][role]
fixture_id = f"fidelity_p1_{cell}_{role}"
cell_record = {
"cell_id": cell,
"tensor_parallel_size": int(cell_manifest["tp"]),
"max_num_seqs": int(cell_manifest["mns"]),
"store_role": "companion" if int(cell_manifest["tp"]) == 4 else "primary",
"kv_capacity": {
"block_size_tokens": 16,
"num_blocks": kv_blocks(args.phase6_raw_root, cell),
},
}
probe = {
"probe_index": 0 if role == "low1" else 1,
"sampling_u": float(expected["anchor"]),
}
fixture = simulator.create_fixture(
fixture_root=fixture_root,
fixture_id=fixture_id,
cell=cell_record,
probe=probe,
row_indexes=[item.row_index for item in selected],
meta_by_index={item.row_index: item for item in retained},
selected_records=selected_records,
)
config_path = config_root / f"{fixture_id}.json"
config = simulator.build_config(
path=config_path,
cell=cell_record,
mode="frozen-calibrated",
fixture_ids=[fixture_id],
frontier_root=args.frontier_root,
cache_dir=args.cache_dir,
)
entries.append(
{
"cell": cell,
"role": role,
"level": level,
"anchor": expected["anchor"],
"selected_count": len(selected),
"fixture_id": fixture_id,
"fixture_manifest": str(
(fixture_root / fixture_id / "fixture_manifest.json").resolve()
),
"frontier_csv": fixture["frontier_csv"]["path"],
"sidecar": fixture["sidecar_jsonl"]["path"],
"config": str(config_path.resolve()),
"calibration_scale": config["calibration"]["a_tp"],
}
)
if block_stats["selected_union_records"] != len(selected_union):
red_flags.append(f"selected_union_{role}")
if trace_stats["retained_inclusive_0_8192"] < 512:
red_flags.append(f"retained_too_small_{role}")
selected_counts = [int(entry["selected_count"]) for entry in entries]
calibration = [float(entry["calibration_scale"]) for entry in entries]
result = {
"schema": "fidelity-p1-frontier-prepared-v1",
"status": "PASS" if not red_flags else "STOP",
"source": {
"pilot_manifest": str(args.pilot_manifest.resolve()),
"source_windows": str(args.source_windows.resolve()),
"source_window_id": args.source_window_id,
"source_trace": str(trace.resolve()),
"private_windows": (
str(private_windows.resolve()) if private_windows is not None else None
),
"pre_materialized_band_root": (
str(args.band_root.resolve()) if args.band_root is not None else None
),
"band_stats": band_stats,
},
"simulator": {
"replayserve_root": str(args.replayserve_root.resolve()),
"frontier_root": str(args.frontier_root.resolve()),
"tokenizer": str(args.tokenizer.resolve()),
"mode": "frozen-calibrated",
},
"generator": {
"script": str(Path(__file__).resolve()),
"script_sha256": sha256_file(Path(__file__).resolve()),
"aituner_git_head": git_capture(AITUNER_ROOT, "rev-parse", "HEAD").strip(),
"aituner_git_status_short": git_capture(AITUNER_ROOT, "status", "--short"),
},
"entries": entries,
"sanity": {
"red_flags": red_flags,
"n": len(entries),
"selected_count": {
"n": len(selected_counts),
"min": min(selected_counts),
"max": max(selected_counts),
"distinct_n": len(set(selected_counts)),
},
"calibration_scale": {
"n": len(calibration),
"min": min(calibration),
"max": max(calibration),
"distinct_n": len(set(calibration)),
},
"invariants": {
"entries_12": len(entries) == 12,
"roles_2": {entry["role"] for entry in entries} == set(PRIMARY_ROLES),
"cells_6": len({entry["cell"] for entry in entries}) == 6,
"selected_nonnegative": all(value > 0 for value in selected_counts),
"per_config_not_identical": len(set(selected_counts)) > 1,
},
},
}
args.public_manifest.parent.mkdir(parents=True, exist_ok=True)
args.public_manifest.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n")
return result
def parser() -> argparse.ArgumentParser:
result = argparse.ArgumentParser()
result.add_argument("--pilot-manifest", type=Path, required=True)
result.add_argument("--source-windows", type=Path, required=True)
result.add_argument("--source-window-id", required=True)
result.add_argument("--private-root", type=Path, required=True)
result.add_argument("--band-root", type=Path)
result.add_argument("--output", type=Path, required=True)
result.add_argument("--public-manifest", type=Path, required=True)
result.add_argument("--phase6-raw-root", type=Path, required=True)
result.add_argument("--replayserve-root", type=Path, required=True)
result.add_argument("--frontier-root", type=Path, required=True)
result.add_argument("--cache-dir", type=Path, required=True)
result.add_argument("--tokenizer", type=Path, required=True)
return result
def main() -> None:
result = prepare(parser().parse_args())
print(
json.dumps(
{
"status": result["status"],
"entries": len(result["entries"]),
"red_flags": result["sanity"]["red_flags"],
},
sort_keys=True,
)
)
if result["status"] != "PASS":
raise RuntimeError(result["sanity"]["red_flags"])
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,292 @@
#!/usr/bin/env python3
"""Run and score the 12 frozen Frontier P1 primary probes, CPU only."""
from __future__ import annotations
import argparse
import hashlib
import importlib.util
import json
import os
import subprocess
import sys
import time
from pathlib import Path
from typing import Any
def load_module(name: str, path: Path):
spec = importlib.util.spec_from_file_location(name, path)
if spec is None or spec.loader is None:
raise ImportError(path)
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as source:
for chunk in iter(lambda: source.read(1 << 20), b""):
digest.update(chunk)
return digest.hexdigest()
def atomic_json(path: Path, payload: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_suffix(path.suffix + ".tmp")
temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
os.replace(temporary, path)
def git_capture(root: Path, *arguments: str) -> str:
return subprocess.run(
["git", "-C", str(root), *arguments],
check=True,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
).stdout
def execute(args: argparse.Namespace) -> dict[str, Any]:
prepared = json.loads(args.prepared_manifest.read_text(encoding="utf-8"))
if prepared["status"] != "PASS":
raise RuntimeError("prepared simulator manifest did not pass")
driver = load_module(
"simfid_execution_driver",
args.replayserve_root
/ "runs/simfid_s2rb/results/execution_driver.py",
)
head = git_capture(args.frontier_root, "rev-parse", "HEAD").strip()
status_short = git_capture(args.frontier_root, "status", "--short")
aituner_root = Path(__file__).resolve().parents[2]
aituner_head = git_capture(aituner_root, "rev-parse", "HEAD").strip()
aituner_status_short = git_capture(aituner_root, "status", "--short")
results = []
failures = []
gpu_visibility_disabled = True
for sequence, entry in enumerate(prepared["entries"]):
run_root = args.output / f"{sequence:02d}_{entry['fixture_id']}"
scorer_path = run_root / "scorer_output.json"
if scorer_path.is_file() and args.resume:
scorer = json.loads(scorer_path.read_text(encoding="utf-8"))
results.append({**entry, "sequence": sequence, "scorer": scorer, "resumed": True})
continue
run_root.mkdir(parents=True, exist_ok=True)
config_path = Path(entry["config"])
config = json.loads(config_path.read_text(encoding="utf-8"))
fixture_manifest_path = Path(entry["fixture_manifest"])
fixture = json.loads(fixture_manifest_path.read_text(encoding="utf-8"))
trace_path = Path(entry["frontier_csv"])
sidecar_path = Path(entry["sidecar"])
metrics_root = run_root / "frontier_metrics"
run_id = f"fidelity_p1_frontier_{sequence:02d}_{entry['cell']}_{entry['role']}"
knobs = config["frontier"]["knobs"]
command = driver.build_command(
trace_path=trace_path,
metrics_root=metrics_root,
run_id=run_id,
knobs=knobs,
)
driver.audit_command(command, knobs)
row = {
"hook_path": config["calibration"]["hook_path"],
"applied_a_tp": config["calibration"]["a_tp"],
"sidecar_path": str(sidecar_path),
"request_count": int(fixture["request_count"]),
"tensor_parallel_size": int(fixture["tensor_parallel_size"]),
}
environment = driver.environment_for(row)
gpu_visibility_disabled = gpu_visibility_disabled and (
environment.get("CUDA_VISIBLE_DEVICES") == ""
and environment.get("NVIDIA_VISIBLE_DEVICES") == "void"
)
run_manifest = {
"schema": "fidelity-p1-frontier-run-v1",
"sequence": sequence,
"cell": entry["cell"],
"role": entry["role"],
"anchor": entry["anchor"],
"request_count": entry["selected_count"],
"frontier": {
"root": str(args.frontier_root.resolve()),
"git_head": head,
"git_status_short": status_short,
},
"runner": {
"script": str(Path(__file__).resolve()),
"script_sha256": sha256_file(Path(__file__).resolve()),
"aituner_git_head": aituner_head,
"aituner_git_status_short": aituner_status_short,
},
"inputs": {
"config": str(config_path),
"config_sha256": sha256_file(config_path),
"fixture_manifest": str(fixture_manifest_path),
"fixture_manifest_sha256": sha256_file(fixture_manifest_path),
"frontier_csv": str(trace_path),
"frontier_csv_sha256": sha256_file(trace_path),
"sidecar": str(sidecar_path),
"sidecar_sha256": sha256_file(sidecar_path),
},
"environment": {
key: environment[key]
for key in (
"PYTHONPATH",
"FRONTIER_EXECUTION_TIME_SCALE",
"CUDA_VISIBLE_DEVICES",
"NVIDIA_VISIBLE_DEVICES",
"FRONTIER_LOG_LEVEL",
)
},
"command": command,
"contains_prompt_text": False,
}
atomic_json(run_root / "run_manifest.json", run_manifest)
start = time.time()
with (run_root / "stdout.log").open("w", encoding="utf-8") as stdout, (
run_root / "stderr.log"
).open("w", encoding="utf-8") as stderr:
try:
process = subprocess.run(
command,
cwd=args.frontier_root,
env=environment,
stdout=stdout,
stderr=stderr,
timeout=args.timeout_s,
)
return_code = int(process.returncode)
except subprocess.TimeoutExpired:
return_code = 124
runtime = time.time() - start
if return_code != 0:
failure = {
"sequence": sequence,
"cell": entry["cell"],
"role": entry["role"],
"return_code": return_code,
"runtime_s": runtime,
}
failures.append(failure)
atomic_json(run_root / "failure.json", failure)
break
system_path, request_path = driver.find_metrics(run_root)
scorer = driver.score_trial(row, system_path, request_path)
scorer["runtime_s"] = runtime
atomic_json(scorer_path, scorer)
results.append({**entry, "sequence": sequence, "scorer": scorer, "resumed": False})
print(
json.dumps(
{
"sequence": sequence,
"cell": entry["cell"],
"role": entry["role"],
"runtime_s": runtime,
"sim_pass_rate": scorer["slo"]["pass_rate"],
"sim_feasible": scorer["slo"]["feasible"],
},
sort_keys=True,
),
flush=True,
)
pass_rates = [float(item["scorer"]["slo"]["pass_rate"]) for item in results]
throughputs = [
float(item["scorer"]["throughput_requests_per_second_per_gpu"])
for item in results
]
runtimes = [float(item["scorer"]["runtime_s"]) for item in results]
red_flags = []
if failures:
red_flags.append("frontier_run_failure")
if len(results) != 12:
red_flags.append("runs_not_12")
if any(not 0.0 <= value <= 1.0 for value in pass_rates):
red_flags.append("pass_rate_out_of_range")
if any(value <= 0 for value in throughputs):
red_flags.append("nonpositive_throughput")
result = {
"schema": "fidelity-p1-frontier-result-v1",
"status": "PASS" if not red_flags else "STOP",
"prepared_manifest": str(args.prepared_manifest.resolve()),
"prepared_manifest_sha256": sha256_file(args.prepared_manifest),
"frontier": {
"root": str(args.frontier_root.resolve()),
"git_head": head,
"git_status_short": status_short,
},
"runner": {
"script": str(Path(__file__).resolve()),
"script_sha256": sha256_file(Path(__file__).resolve()),
"aituner_git_head": aituner_head,
"aituner_git_status_short": aituner_status_short,
},
"results": results,
"failures": failures,
"sanity": {
"red_flags": red_flags,
"n": len(results),
"pass_rate": {
"n": len(pass_rates),
"min": min(pass_rates) if pass_rates else None,
"max": max(pass_rates) if pass_rates else None,
"distinct_n": len(set(pass_rates)),
},
"throughput_per_gpu": {
"n": len(throughputs),
"min": min(throughputs) if throughputs else None,
"max": max(throughputs) if throughputs else None,
"distinct_n": len(set(throughputs)),
},
"runtime_s": {
"n": len(runtimes),
"min": min(runtimes) if runtimes else None,
"max": max(runtimes) if runtimes else None,
"distinct_n": len(set(runtimes)),
},
"invariants": {
"runs_12": len(results) == 12,
"zero_failures": not failures,
"ratios_bounded": all(0.0 <= value <= 1.0 for value in pass_rates),
"nonnegative_metrics": all(value > 0 for value in throughputs),
"per_config_not_identical": len(set(pass_rates)) > 1,
"gpu_visibility_disabled": gpu_visibility_disabled,
},
},
}
atomic_json(args.output / "metrics.json", result)
return result
def parser() -> argparse.ArgumentParser:
result = argparse.ArgumentParser()
result.add_argument("--prepared-manifest", type=Path, required=True)
result.add_argument("--output", type=Path, required=True)
result.add_argument("--replayserve-root", type=Path, required=True)
result.add_argument("--frontier-root", type=Path, required=True)
result.add_argument("--timeout-s", type=float, default=900.0)
result.add_argument("--resume", action="store_true")
return result
def main() -> None:
result = execute(parser().parse_args())
print(
json.dumps(
{
"status": result["status"],
"runs": len(result["results"]),
"red_flags": result["sanity"]["red_flags"],
},
sort_keys=True,
)
)
if result["status"] != "PASS":
raise RuntimeError(result["sanity"]["red_flags"])
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,75 @@
#!/usr/bin/env python3
from __future__ import annotations
import json
import tempfile
from pathlib import Path
import numpy as np
from analyze_prefixes import PrefixExample
from analyze_strong_pilot import (
fit_model,
load_pilot_simulator,
predict_model,
)
def example(index: int) -> PrefixExample:
label = int(index >= 4)
return PrefixExample(
cell=f"cell-{index // 2}",
anchor=float(index),
cutoff_s=5.0,
tp=1,
full_elapsed_s=10.0,
feasible=label,
primary_feasible=label,
outcome=tuple(float(index + offset) for offset in range(13)),
instrumentation=tuple(float(index * offset + 1) for offset in range(17)),
completion_time_source="exact_monotonic",
)
def main() -> None:
examples = [example(index) for index in range(8)]
simulator = [(float(index), index / 10.0, float(index >= 4)) for index in range(8)]
for instrumentation_aware in (False, True):
model = fit_model(
examples,
simulator,
instrumentation_aware=instrumentation_aware,
regularization=1.0,
)
probability = predict_model(model, examples, simulator)
assert probability.shape == (8,)
assert np.all((probability >= 0.0) & (probability <= 1.0))
payload = {
"status": "PASS",
"results": [
{
"cell": f"cell-{index // 2}",
"role": "low1" if index % 2 == 0 else "high1",
"scorer": {
"throughput_requests_per_second_per_gpu": 1.0 + index,
"slo": {
"pass_rate": index / 12.0,
"feasible": index % 2 == 0,
},
},
}
for index in range(12)
],
}
with tempfile.TemporaryDirectory() as temporary:
path = Path(temporary) / "metrics.json"
path.write_text(json.dumps(payload), encoding="utf-8")
features, red_flags = load_pilot_simulator(path)
assert len(features) == 12
assert red_flags == []
print("fidelity strong pilot: PASS")
if __name__ == "__main__":
main()