diff --git a/runs/fidelity-headroom/analyze_strong_pilot.py b/runs/fidelity-headroom/analyze_strong_pilot.py new file mode 100644 index 0000000..57e79c3 --- /dev/null +++ b/runs/fidelity-headroom/analyze_strong_pilot.py @@ -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() diff --git a/runs/fidelity-headroom/prepare_pilot_simulator.py b/runs/fidelity-headroom/prepare_pilot_simulator.py new file mode 100644 index 0000000..c3d77fb --- /dev/null +++ b/runs/fidelity-headroom/prepare_pilot_simulator.py @@ -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() diff --git a/runs/fidelity-headroom/run_pilot_simulator.py b/runs/fidelity-headroom/run_pilot_simulator.py new file mode 100644 index 0000000..d393916 --- /dev/null +++ b/runs/fidelity-headroom/run_pilot_simulator.py @@ -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() diff --git a/runs/fidelity-headroom/test_strong_pilot.py b/runs/fidelity-headroom/test_strong_pilot.py new file mode 100644 index 0000000..48f6089 --- /dev/null +++ b/runs/fidelity-headroom/test_strong_pilot.py @@ -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()