From 34e1f4c144641c2b2fcc308a2b776de09f747a24 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Mon, 13 Jul 2026 20:25:44 +0800 Subject: [PATCH] Add static-policy oracle gap experiment --- docs/opprof/oracle-gap-protocol.md | 159 ++++++ scripts/oracle_gap/__init__.py | 1 + scripts/oracle_gap/analyze.py | 366 +++++++++++++ scripts/oracle_gap/phase3_upper_bound.py | 91 +++ scripts/oracle_gap/run_frontier.py | 671 +++++++++++++++++++++++ tests/test_oracle_gap.py | 143 +++++ 6 files changed, 1431 insertions(+) create mode 100644 docs/opprof/oracle-gap-protocol.md create mode 100644 scripts/oracle_gap/__init__.py create mode 100644 scripts/oracle_gap/analyze.py create mode 100644 scripts/oracle_gap/phase3_upper_bound.py create mode 100644 scripts/oracle_gap/run_frontier.py create mode 100644 tests/test_oracle_gap.py diff --git a/docs/opprof/oracle-gap-protocol.md b/docs/opprof/oracle-gap-protocol.md new file mode 100644 index 0000000..634d4a9 --- /dev/null +++ b/docs/opprof/oracle-gap-protocol.md @@ -0,0 +1,159 @@ +# Static-policy oracle-gap protocol + +Status: **FROZEN BEFORE NEW GPU EXECUTION**. + +Date frozen: 2026-07-13 (Asia/Singapore). Existing Phase-3 measurements were +inspected only to choose the workload pair and rate brackets. They are +exploratory calibration data, not primary observations in this protocol. + +## Question and decision gate + +The candidate motivation is: + +> A single global static batching policy leaves at least 10% end-to-end +> SLO-goodput on the table when serving temporally heterogeneous phases; a +> phase-aware runtime policy can recover that gap without changing hardware, +> model, precision, or tensor-parallel topology. + +This experiment tests a necessary condition in the existing TP1 policy space +`{C00,C10,C01,C11}`. The optimistic oracle knows the phase and switches with +zero delay, zero state-transfer cost, and no prediction error. If even this +oracle cannot beat the best one-config-for-all-phases policy by 10%, an online +controller over these MNS/MBT choices cannot do so either. + +The primary gate uses a conservative capacity bracket: + +- `L[p,c]`: highest offered rate accepted as SLO-feasible for phase `p` and + config `c`; +- `U[p,c]`: lowest higher offered rate accepted as SLO-infeasible; +- oracle upper bound at phase-time weights `w`: + `sum_p w[p] * max_c U[p,c]`; +- best-static lower bound: + `max_c sum_p w[p] * L[p,c]`. + +We scan every P01/P06 time mixture, including pure endpoints. The current +motivation is **REFUTED** if the maximum conservative ratio +`oracle_upper / static_lower - 1` is below 10%. It is **NOT ESTABLISHED** if the +bound crosses 10% but the observed point estimate does not. A positive result +requires a point-estimate gap of at least 10% and then a separately +pre-registered interleaved-trace validation; this frontier experiment alone +cannot establish a positive E2E contribution. + +The conclusion is scoped to the measured MNS/MBT policy family and the chosen +strongest-conflict phase pair. It does not rule out new scheduling mechanisms, +KV-state policies, topology changes, or other workload phases. + +## Fixed system boundary + +| Item | Frozen value | +|---|---| +| Host | `dash0`, one run at a time on physical GPU0 | +| GPU | NVIDIA H20; no other GPU process anywhere on the host | +| Model | `/home/admin/cpfs/wjh/models/Qwen/Qwen3-30B-A3B`, BF16 | +| Runtime | `/tmp/wjh-opprof-phase2-dash0-20260711/.venv`, vLLM `0.24.1.dev3+g668cfb7e2` | +| vLLM source | `/home/admin/cpfs/wjh/opprof-phase2-dash0-20260711/vllm-v0.24.0` | +| Topology | TP1, one server, no data/pipeline parallelism | +| Fixed mechanisms | chunked prefill on; prefix caching on | +| Client | Phase-5 timestamp/fixed-rate wrapper over the Phase-3 exact-token client | +| Seeds | workload `20260712`; trial token-domain seed derived only from phase/rate/repetition, never config | + +SLO co-location results in Phase 6 showed pass-rate flips despite small +throughput deltas. Therefore unused H20s remain idle: parallel placement is not +authoritative for this experiment. + +## Workloads and policies + +The pair is chosen before new measurements because Phase 3 showed the strongest +opposing static preference: + +- **P01:** input `U[128,512]`, output exactly 64 tokens, deterministic steady + arrivals. C10 lost 24.27% saturation throughput relative to C00. +- **P06:** 50/50 input mixture `U[128,512]`/`U[4096,8192]`, output exactly 512 + tokens, deterministic bursts of eight. C10 gained 3.37% over C00. + +Both reuse the immutable 32,768-row Phase-3 manifests. For every trial a +derived manifest preserves request order, lengths, outputs, and arrival class, +but applies a trial-specific token-seed offset. The same derived manifest is +used for all four configs. This prevents prefix-cache carry-over when a hot +server executes several anchors without changing the logical workload. + +| Config | Effective MNS | Effective MBT | Extra flags | +|---|---:|---:|---| +| C00 | 1024 | 8192 | none | +| C10 | 64 | 8192 | `--max-num-seqs 64` | +| C01 | 1024 | 2048 | `--max-num-batched-tokens 2048` | +| C11 | 64 | 2048 | both flags | + +Startup logs must confirm these values. A default drift is a stop condition. + +## Load grid, order, and repetitions + +Primary grids: + +- P01: `{26,28,30,32,34,36}` requests/s; execution order + `32,26,36,28,34,30`. +- P06: `{1.4,1.5,1.6,1.7,1.8,1.9,2.0}` requests/s; execution order + `1.7,1.4,2.0,1.5,1.9,1.6,1.8`. + +Every primary anchor runs once. For each phase/config, the highest primary +feasible anchor and its next higher primary anchor are then run two more times, +giving three trials at both sides of the boundary. If all primary anchors are +feasible, extend upward in the fixed order P01 `38,40,42` or P06 `2.1,2.2,2.3`. +If all are infeasible, extend downward in the fixed order P01 `24,22,20` or P06 +`1.3,1.2,1.1`. Stop extending at the first bracket. + +One primary server is launched per config in order `C11,C00,C01,C10`. +Confirmation servers are fresh and launch in reverse order +`C10,C01,C00,C11`; their boundary anchors run high-to-low. This balances +machine-time drift and makes confirmation independent of the primary server's +cache/compiler state. + +Timelines: + +- P01: 60 s warm-up + 60 s clean measurement; drain cap 120 s. +- P06: 60 s warm-up + 120 s clean measurement; drain cap 240 s. +- no Kineto profiling; exact greedy output with `ignore_eos`; maximum client + concurrency 256. + +A trial is SLO-feasible when at least 95% of requests admitted during the clean +interval eventually finish successfully and individually satisfy both: + +- TTFT <= 2 s for input <= 4,096 tokens; <= 4 s for input <= 32,768; <= 6 s + otherwise; +- TPOT <= 50 ms, computed as `(completion - first_token)/(output_tokens - 1)`. + +SLO-goodput is the number of those passing clean-admission requests divided by +clean seconds. Client schedule lag must stay <=1 s and achieved clean offered +rate must be within 5% of target. Failure of either condition makes the anchor +infeasible; its admitted-only latency is not used to rescue it. + +At a repeated boundary, feasibility is the majority of three trial verdicts. +All accepted anchor verdicts must be monotone in offered rate. A persistent +non-monotone result after the registered repeats is a red flag and stops the +oracle-gap inference. + +## Validity and stopping rules + +Before every server launch record host, GPU, driver, clocks, runtime package +versions, git/source hashes, manifest hashes, exact commands, and process +contamination. Stop on another GPU process, request/output mismatch, manifest +drift, server crash, non-finite latency, ratio outside `[0,1]`, negative +counter, or discontinuous/non-monotone accepted frontier. + +The controller is detached and resumable. It kills only process groups it +created, checks zero GPU memory after every server, never overwrites a complete +trial, and writes state atomically. The hard budget is 6 H20-hours; expected +cost is 3.0--4.0 H20-hours and approximately the same wall time because runs +are serialized. + +## Required report + +The report includes every trial's target/achieved rate, clean cohort size, +pass count/rate, SLO-goodput, TTFT/TPOT percentiles, schedule lag, failure +reasons, accepted frontier brackets, per-phase oracle choices, best static +choice, equal-time gap, worst-mixture conservative gap, and GPU-hours. + +The final statistics section ends with a data-sanity block containing `n`, +min/max, distinct-value counts, and checks for non-negative counters, ratios in +`[0,1]`, non-identical per-config results, exact output work, monotone +frontiers, and continuous rate brackets. diff --git a/scripts/oracle_gap/__init__.py b/scripts/oracle_gap/__init__.py new file mode 100644 index 0000000..dfae6c4 --- /dev/null +++ b/scripts/oracle_gap/__init__.py @@ -0,0 +1 @@ +"""Oracle-gap experiment helpers.""" diff --git a/scripts/oracle_gap/analyze.py b/scripts/oracle_gap/analyze.py new file mode 100644 index 0000000..f83f9c9 --- /dev/null +++ b/scripts/oracle_gap/analyze.py @@ -0,0 +1,366 @@ +#!/usr/bin/env python3 +"""Score fixed-rate request logs and summarize static-vs-oracle frontiers.""" + +from __future__ import annotations + +import argparse +import json +import math +import os +from collections import Counter, defaultdict +from pathlib import Path +from typing import Any, Iterable + + +TARGET_PASS_RATE = 0.95 +TPOT_LIMIT_MS = 50.0 +PHASES = ("P01", "P06") +CONFIGS = ("C00", "C10", "C01", "C11") + + +def atomic_json(path: Path, value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f"{path.name}.tmp.{os.getpid()}") + temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n") + os.replace(temporary, path) + + +def numeric(values: Iterable[float | int | None]) -> dict[str, Any]: + materialized = list(values) + finite = [ + float(value) + for value in materialized + if value is not None and math.isfinite(float(value)) + ] + return { + "n": len(materialized), + "finite_n": len(finite), + "missing_n": len(materialized) - len(finite), + "min": min(finite) if finite else None, + "max": max(finite) if finite else None, + "distinct_n": len(set(finite)), + } + + +def percentile(values: Iterable[float], quantile: float) -> float | None: + ordered = sorted(float(value) for value in values) + if not ordered: + return None + index = max(0, min(len(ordered) - 1, math.ceil(quantile * len(ordered)) - 1)) + return ordered[index] + + +def ttft_limit_ms(input_tokens: int) -> float: + if input_tokens <= 4096: + return 2000.0 + if input_tokens <= 32768: + return 4000.0 + return 6000.0 + + +def score_trial( + request_path: Path, + result_path: Path, + *, + phase: str, + config: str, + target_rate: float, + repetition: int, + role: str, +) -> dict[str, Any]: + result = json.loads(result_path.read_text()) + rows = [json.loads(line) for line in request_path.read_text().splitlines() if line] + clean_start = float(result["warmup_seconds"]) + clean_seconds = float(result["clean_segment_seconds"]) * int( + result["num_clean_segments"] + ) + clean_end = clean_start + clean_seconds + cohort = [row for row in rows if clean_start <= float(row["admitted_s"]) < clean_end] + + ttft_values: list[float] = [] + tpot_values: list[float] = [] + lag_values: list[float] = [] + reasons: Counter[str] = Counter() + passes = 0 + exact_outputs = 0 + for row in cohort: + lag_ms = (float(row["admitted_s"]) - float(row["scheduled_s"])) * 1000.0 + lag_values.append(lag_ms) + request_reasons: list[str] = [] + if not bool(row["success"]): + request_reasons.append(str(row.get("error_kind") or "request_failed")) + first = row.get("first_token_s") + if first is None: + request_reasons.append("ttft_missing") + ttft = None + else: + ttft = (float(first) - float(row["admitted_s"])) * 1000.0 + ttft_values.append(ttft) + if ttft > ttft_limit_ms(int(row["input_tokens"])): + request_reasons.append("ttft_slo") + actual = row.get("actual_output_tokens") + requested = int(row["requested_output_tokens"]) + if actual == requested: + exact_outputs += 1 + if first is None or actual is None or int(actual) <= 1: + request_reasons.append("tpot_missing") + tpot = None + else: + tpot = ( + (float(row["completed_s"]) - float(first)) + * 1000.0 + / (int(actual) - 1) + ) + tpot_values.append(tpot) + if tpot > TPOT_LIMIT_MS: + request_reasons.append("tpot_slo") + if request_reasons: + reasons.update(set(request_reasons)) + else: + passes += 1 + + achieved_rate = len(cohort) / clean_seconds if clean_seconds else 0.0 + pass_rate = passes / len(cohort) if cohort else 0.0 + max_lag_ms = max(lag_values, default=math.inf) + offered_rate_valid = ( + target_rate > 0 and abs(achieved_rate / target_rate - 1.0) <= 0.05 + ) + schedule_valid = bool(cohort) and max_lag_ms <= 1000.0 + raw_feasible = pass_rate >= TARGET_PASS_RATE + feasible = raw_feasible and offered_rate_valid and schedule_valid + invariants = { + "cohort_nonempty": bool(cohort), + "clean_duration_positive": clean_seconds > 0, + "timestamps_nondecreasing": all( + float(row["scheduled_s"]) <= float(row["admitted_s"]) + <= float(row["completed_s"]) + for row in cohort + ), + "exact_output_or_failed": all( + (not bool(row["success"])) + or row.get("actual_output_tokens") == row.get("requested_output_tokens") + for row in cohort + ), + "latencies_nonnegative": all(value >= 0 for value in ttft_values + tpot_values), + "pass_rate_in_0_1": 0.0 <= pass_rate <= 1.0, + "goodput_nonnegative": passes >= 0, + } + if not all(invariants.values()): + raise RuntimeError(f"trial data invariant failed: {invariants}") + return { + "schema": 1, + "phase": phase, + "config": config, + "target_rate_rps": target_rate, + "repetition": repetition, + "role": role, + "clean_start_s": clean_start, + "clean_end_s": clean_end, + "clean_seconds": clean_seconds, + "cohort_n": len(cohort), + "pass_n": passes, + "pass_rate": pass_rate, + "slo_goodput_rps": passes / clean_seconds, + "achieved_offered_rps": achieved_rate, + "offered_rate_valid": offered_rate_valid, + "schedule_valid": schedule_valid, + "raw_slo_feasible": raw_feasible, + "feasible": feasible, + "exact_output_n": exact_outputs, + "failure_reasons": dict(sorted(reasons.items())), + "ttft_ms": { + **numeric(ttft_values), + "p50": percentile(ttft_values, 0.50), + "p95": percentile(ttft_values, 0.95), + "p99": percentile(ttft_values, 0.99), + }, + "tpot_ms": { + **numeric(tpot_values), + "p50": percentile(tpot_values, 0.50), + "p95": percentile(tpot_values, 0.95), + "p99": percentile(tpot_values, 0.99), + }, + "schedule_lag_ms": { + **numeric(lag_values), + "p95": percentile(lag_values, 0.95), + "p99": percentile(lag_values, 0.99), + }, + "invariants": invariants, + } + + +def accepted_rate(rows: list[dict[str, Any]]) -> dict[str, Any]: + verdicts = [bool(row["feasible"]) for row in rows] + pass_n = sum(int(row["pass_n"]) for row in rows) + cohort_n = sum(int(row["cohort_n"]) for row in rows) + return { + "trials": len(rows), + "trial_feasible": verdicts, + "accepted_feasible": sum(verdicts) > len(verdicts) / 2, + "pooled_pass_n": pass_n, + "pooled_cohort_n": cohort_n, + "pooled_pass_rate": pass_n / cohort_n if cohort_n else 0.0, + "median_goodput_rps": sorted(float(row["slo_goodput_rps"]) for row in rows)[ + len(rows) // 2 + ], + } + + +def frontier_for_cell(rows: list[dict[str, Any]]) -> dict[str, Any]: + by_rate: dict[float, list[dict[str, Any]]] = defaultdict(list) + for row in rows: + by_rate[float(row["target_rate_rps"])].append(row) + rates = [] + for rate in sorted(by_rate): + rates.append({"rate_rps": rate, **accepted_rate(by_rate[rate])}) + verdicts = [bool(row["accepted_feasible"]) for row in rates] + # Once a failure appears, no higher anchor may pass. + monotone = not any( + (not verdicts[i]) and any(verdicts[i + 1 :]) for i in range(len(verdicts)) + ) + feasible_rates = [row["rate_rps"] for row in rates if row["accepted_feasible"]] + infeasible_rates = [row["rate_rps"] for row in rates if not row["accepted_feasible"]] + lower = max(feasible_rates, default=None) + upper_candidates = [rate for rate in infeasible_rates if lower is None or rate > lower] + upper = min(upper_candidates, default=None) + boundary_repeated = bool( + lower is not None + and upper is not None + and next(row for row in rates if row["rate_rps"] == lower)["trials"] >= 3 + and next(row for row in rates if row["rate_rps"] == upper)["trials"] >= 3 + ) + bracketed = lower is not None and upper is not None and monotone and boundary_repeated + return { + "rates": rates, + "lower_feasible_rps": lower, + "upper_infeasible_rps": upper, + "bracketed": bracketed, + "boundary_repeated": boundary_repeated, + "monotone": monotone, + } + + +def gap_at_weight( + lower: dict[str, dict[str, float]], + upper: dict[str, dict[str, float]], + p01_weight: float, +) -> dict[str, Any]: + weights = {"P01": p01_weight, "P06": 1.0 - p01_weight} + oracle = sum(weights[p] * max(upper[p].values()) for p in PHASES) + static_values = { + config: sum(weights[p] * lower[p][config] for p in PHASES) + for config in CONFIGS + } + best_config = max(static_values, key=static_values.get) + static = static_values[best_config] + return { + "p01_weight": p01_weight, + "oracle_upper_rps": oracle, + "static_lower_rps": static, + "best_static_config": best_config, + "gap": oracle / static - 1.0, + } + + +def summarize_trials(rows: list[dict[str, Any]]) -> dict[str, Any]: + grouped: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list) + for row in rows: + grouped[(str(row["phase"]), str(row["config"]))].append(row) + expected = {(phase, config) for phase in PHASES for config in CONFIGS} + if set(grouped) != expected: + raise RuntimeError(f"cell coverage mismatch: {sorted(set(grouped) ^ expected)}") + frontiers = { + phase: { + config: frontier_for_cell(grouped[(phase, config)]) + for config in CONFIGS + } + for phase in PHASES + } + all_bracketed = all( + frontiers[p][c]["bracketed"] for p in PHASES for c in CONFIGS + ) + all_monotone = all( + frontiers[p][c]["monotone"] for p in PHASES for c in CONFIGS + ) + lower = { + p: {c: float(frontiers[p][c]["lower_feasible_rps"]) for c in CONFIGS} + for p in PHASES + } if all_bracketed else {} + upper = { + p: {c: float(frontiers[p][c]["upper_infeasible_rps"]) for c in CONFIGS} + for p in PHASES + } if all_bracketed else {} + scan = [gap_at_weight(lower, upper, step / 10000) for step in range(10001)] if all_bracketed else [] + worst = max(scan, key=lambda row: row["gap"]) if scan else None + equal = gap_at_weight(lower, upper, 0.5) if all_bracketed else None + distinct_by_cell = { + f"{p}-{c}": len( + {round(float(row["slo_goodput_rps"]), 12) for row in grouped[(p, c)]} + ) + for p in PHASES for c in CONFIGS + } + sanity = { + "trial_count": numeric([row["cohort_n"] for row in rows]), + "target_rates": numeric([row["target_rate_rps"] for row in rows]), + "pass_rates": numeric([row["pass_rate"] for row in rows]), + "goodput_rps": numeric([row["slo_goodput_rps"] for row in rows]), + "distinct_goodput_by_cell": distinct_by_cell, + "invariants": { + "all_counters_nonnegative": all( + int(row["cohort_n"]) >= 0 and int(row["pass_n"]) >= 0 for row in rows + ), + "all_ratios_in_0_1": all(0 <= float(row["pass_rate"]) <= 1 for row in rows), + "all_trial_invariants": all(all(row["invariants"].values()) for row in rows), + "all_cells_bracketed": all_bracketed, + "all_frontiers_monotone": all_monotone, + "per_cell_results_not_all_identical": all(value > 1 for value in distinct_by_cell.values()), + "weight_scan_continuous": len(scan) in (0, 10001), + }, + } + verdict = "INCONCLUSIVE" + if all(sanity["invariants"].values()) and worst is not None: + verdict = "REFUTED" if float(worst["gap"]) < 0.10 else "NOT_ESTABLISHED" + return { + "schema": 1, + "verdict": verdict, + "threshold": 0.10, + "frontiers": frontiers, + "equal_time_conservative": equal, + "worst_mixture_conservative": worst, + "sanity": sanity, + } + + +def main() -> None: + parser = argparse.ArgumentParser() + sub = parser.add_subparsers(dest="command", required=True) + score = sub.add_parser("score") + score.add_argument("--requests", required=True) + score.add_argument("--result", required=True) + score.add_argument("--phase", choices=PHASES, required=True) + score.add_argument("--config", choices=CONFIGS, required=True) + score.add_argument("--target-rate", type=float, required=True) + score.add_argument("--repetition", type=int, required=True) + score.add_argument("--role", required=True) + score.add_argument("--out", required=True) + summary = sub.add_parser("summarize") + summary.add_argument("--trial-glob", required=True) + summary.add_argument("--out", required=True) + args = parser.parse_args() + if args.command == "score": + value = score_trial( + Path(args.requests), Path(args.result), phase=args.phase, + config=args.config, target_rate=args.target_rate, + repetition=args.repetition, role=args.role, + ) + else: + import glob + + paths = [Path(path) for path in sorted(glob.glob(args.trial_glob, recursive=True))] + value = summarize_trials([json.loads(path.read_text()) for path in paths]) + atomic_json(Path(args.out), value) + print(json.dumps(value, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/scripts/oracle_gap/phase3_upper_bound.py b/scripts/oracle_gap/phase3_upper_bound.py new file mode 100644 index 0000000..5dfd6f3 --- /dev/null +++ b/scripts/oracle_gap/phase3_upper_bound.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python3 +"""Retrospective point-estimate oracle bound from accepted Phase-3 cells.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any + +from analyze import CONFIGS, PHASES, atomic_json, numeric + + +def build(metrics: dict[str, Any]) -> dict[str, Any]: + tables: dict[str, dict[str, dict[str, float]]] = {} + for load in ("saturation", "moderate"): + table: dict[str, dict[str, float]] = {} + for phase in PHASES: + row = {} + for config in CONFIGS: + run = metrics.get("runs", {}).get(f"{phase}-{config}-{load}") + if run is not None: + row[config] = float(run["clean"]["completed_throughput_rps"]) + table[phase] = row + tables[load] = table + + analyses = {} + all_values = [] + for load, table in tables.items(): + complete = all(set(row) == set(CONFIGS) for row in table.values()) + if not complete: + raise RuntimeError(f"missing sentinel throughput cell for {load}: {table}") + per_phase = {} + c00_regrets = [] + for phase, row in table.items(): + oracle_config = max(row, key=row.get) + oracle = row[oracle_config] + regret = oracle / row["C00"] - 1.0 + c00_regrets.append(regret) + per_phase[phase] = { + "throughput_rps": row, + "oracle_config": oracle_config, + "oracle_rps": oracle, + "c00_regret": regret, + } + all_values.extend(row.values()) + equal_oracle = sum(item["oracle_rps"] for item in per_phase.values()) / len(PHASES) + static = { + config: sum(table[p][config] for p in PHASES) / len(PHASES) + for config in CONFIGS + } + analyses[load] = { + "per_phase": per_phase, + "universal_c00_point_bound": max(c00_regrets), + "equal_time_oracle_rps": equal_oracle, + "equal_time_best_static_config": max(static, key=static.get), + "equal_time_best_static_rps": max(static.values()), + "equal_time_gap": equal_oracle / max(static.values()) - 1.0, + "interpretation": ( + "valid additive point-estimate precheck" + if load == "saturation" + else "diagnostic only: each config used its own offered rate" + ), + } + return { + "schema": 1, + "scope": list(PHASES), + "analyses": analyses, + "sanity": { + "throughput_rps": numeric(all_values), + "invariants": { + "all_nonnegative": all(value >= 0 for value in all_values), + "all_cells_present": len(all_values) == 2 * len(PHASES) * len(CONFIGS), + "per_config_not_identical": len(set(all_values)) > len(CONFIGS), + }, + }, + } + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--metrics", required=True) + parser.add_argument("--out", required=True) + args = parser.parse_args() + result = build(json.loads(Path(args.metrics).read_text())) + atomic_json(Path(args.out), result) + print(json.dumps(result, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/scripts/oracle_gap/run_frontier.py b/scripts/oracle_gap/run_frontier.py new file mode 100644 index 0000000..af01c65 --- /dev/null +++ b/scripts/oracle_gap/run_frontier.py @@ -0,0 +1,671 @@ +#!/usr/bin/env python3 +"""Detached, resumable solo-H20 controller for the oracle-gap frontier.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shlex +import shutil +import signal +import subprocess +import time +import urllib.request +from pathlib import Path +from typing import Any + +from analyze import CONFIGS, PHASES, atomic_json, score_trial, summarize_trials + + +SCHEMA = 1 +REMOTE_ROOT = Path("/home/admin/cpfs/wjh/oracle-gap-20260713") +RUN_ROOT = REMOTE_ROOT / "runs" +STATE = RUN_ROOT / "controller-state.json" +PRIVATE = Path("/home/admin/cpfs/wjh/opprof-phase3-private/manifests") +MODEL = Path("/home/admin/cpfs/wjh/models/Qwen/Qwen3-30B-A3B") +SOURCE = Path("/home/admin/cpfs/wjh/opprof-phase2-dash0-20260711/vllm-v0.24.0") +VENV = Path("/tmp/wjh-opprof-phase2-dash0-20260711/.venv") +REPO = Path(os.environ.get("AITUNER_ORACLE_REPO", Path(__file__).resolve().parents[2])) +P5_CLIENT = REPO / "runs/opprof-phase5/opprof_phase5_client.py" +P3_CLIENT_DIR = REPO / "runs/opprof-phase3/provenance" +GPU = 0 +CPU_MASK = "0-19" +PORT = 8820 +GPU_HOUR_LIMIT = 6.0 +PRIMARY_ORDER = ("C11", "C00", "C01", "C10") +CONFIRM_ORDER = tuple(reversed(PRIMARY_ORDER)) +CONFIG_DETAILS = { + "C00": {"mns": 1024, "mbt": 8192, "flags": []}, + "C10": {"mns": 64, "mbt": 8192, "flags": ["--max-num-seqs", "64"]}, + "C01": { + "mns": 1024, + "mbt": 2048, + "flags": ["--max-num-batched-tokens", "2048"], + }, + "C11": { + "mns": 64, + "mbt": 2048, + "flags": [ + "--max-num-seqs", "64", "--max-num-batched-tokens", "2048" + ], + }, +} +BASE_RATES = { + "P01": (32.0, 26.0, 36.0, 28.0, 34.0, 30.0), + "P06": (1.7, 1.4, 2.0, 1.5, 1.9, 1.6, 1.8), +} +UP_EXTENSIONS = {"P01": (38.0, 40.0, 42.0), "P06": (2.1, 2.2, 2.3)} +DOWN_EXTENSIONS = {"P01": (24.0, 22.0, 20.0), "P06": (1.3, 1.2, 1.1)} +TIMELINE = { + "P01": {"warmup": 60, "clean": 60, "drain": 120}, + "P06": {"warmup": 60, "clean": 120, "drain": 240}, +} + + +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 run_text(command: list[str], *, check: bool = True) -> str: + result = subprocess.run( + command, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT + ) + if check and result.returncode: + raise RuntimeError( + f"command failed ({result.returncode}): {shlex.join(command)}\n{result.stdout}" + ) + return result.stdout + + +def load_state(resume: bool) -> dict[str, Any]: + if STATE.exists(): + if not resume: + raise RuntimeError(f"state exists; use --resume: {STATE}") + return json.loads(STATE.read_text()) + return { + "schema": SCHEMA, + "status": "created", + "created_at": time.time(), + "gpu_hours": 0.0, + "completed_trials": [], + "stages": {}, + "fingerprint": {}, + "owned_pgids": [], + } + + +def save_state(state: dict[str, Any]) -> None: + state["updated_at"] = time.time() + state["controller_pid"] = os.getpid() + atomic_json(STATE, state) + + +def compute_apps() -> list[dict[str, Any]]: + output = run_text( + [ + "nvidia-smi", + "--query-compute-apps=gpu_uuid,pid,process_name,used_memory", + "--format=csv,noheader,nounits", + ], + check=False, + ) + rows = [] + for line in output.splitlines(): + parts = [part.strip() for part in line.split(",", 3)] + if len(parts) == 4 and parts[1].isdigit(): + rows.append( + { + "gpu_uuid": parts[0], + "pid": int(parts[1]), + "process_name": parts[2], + "used_memory_mib": int(parts[3].split()[0]), + } + ) + return rows + + +def gpu_snapshot() -> dict[str, Any]: + query = run_text( + [ + "nvidia-smi", + "--query-gpu=index,name,uuid,memory.used,utilization.gpu,clocks.sm,clocks.mem,power.draw", + "--format=csv,noheader,nounits", + ] + ) + return { + "time": time.time(), + "gpus": query.splitlines(), + "compute_apps": compute_apps(), + "loadavg": list(os.getloadavg()), + } + + +def assert_idle() -> None: + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + if not compute_apps(): + return + time.sleep(1) + raise RuntimeError(f"GPU host is not idle: {compute_apps()}") + + +def descendants(root_pid: int) -> set[int]: + output = run_text(["ps", "-e", "-o", "pid=,ppid="], check=False) + children: dict[int, list[int]] = {} + for line in output.splitlines(): + parts = line.split() + if len(parts) == 2: + pid, parent = map(int, parts) + children.setdefault(parent, []).append(pid) + result = {root_pid} + pending = [root_pid] + while pending: + for child in children.get(pending.pop(), []): + if child not in result: + result.add(child) + pending.append(child) + return result + + +def assert_only_server_apps(server_pid: int) -> None: + allowed = descendants(server_pid) + unexpected = [row for row in compute_apps() if int(row["pid"]) not in allowed] + if unexpected: + raise RuntimeError(f"unexpected GPU process during run: {unexpected}") + + +def wait_idle_after_stop() -> None: + deadline = time.monotonic() + 60 + while time.monotonic() < deadline: + output = run_text( + [ + "nvidia-smi", "--query-gpu=index,memory.used", + "--format=csv,noheader,nounits", + ] + ) + memory = [int(line.split(",")[1].strip()) for line in output.splitlines()] + if not compute_apps() and all(value == 0 for value in memory): + return + time.sleep(1) + raise RuntimeError("GPU memory/processes did not return to zero") + + +def fingerprint() -> dict[str, Any]: + manifests = {} + for phase in PHASES: + path = PRIVATE / f"{phase}.jsonl" + summary = json.loads(path.with_suffix(path.suffix + ".summary.json").read_text()) + if int(summary["rows"]) != 32768 or summary["sha256"] != sha256_file(path): + raise RuntimeError(f"manifest mismatch: {phase}") + manifests[phase] = {"path": str(path), "sha256": summary["sha256"]} + return { + "controller_sha256": sha256_file(Path(__file__).resolve()), + "analyzer_sha256": sha256_file(Path(__file__).with_name("analyze.py")), + "p5_client_sha256": sha256_file(P5_CLIENT), + "p3_client_sha256": sha256_file(P3_CLIENT_DIR / "opprof_phase3_client.py"), + "repo_commit": run_text(["git", "-C", str(REPO), "rev-parse", "HEAD"]).strip(), + "repo_tree": run_text(["git", "-C", str(REPO), "rev-parse", "HEAD^{tree}"]).strip(), + "vllm_commit": run_text(["git", "-C", str(SOURCE), "rev-parse", "HEAD"]).strip(), + "model": str(MODEL), + "manifests": manifests, + "runtime": run_text( + [str(VENV / "bin/python"), "-c", "import torch,vllm; print(torch.__version__,torch.version.cuda,vllm.__version__)"] + ).strip(), + "driver": run_text(["nvidia-smi", "--query-gpu=driver_version", "--format=csv,noheader"]).splitlines()[0], + "config_details": CONFIG_DETAILS, + "base_rates": {key: list(value) for key, value in BASE_RATES.items()}, + } + + +def ensure_provenance(current: dict[str, Any]) -> None: + destination = RUN_ROOT / "provenance" + destination.mkdir(parents=True, exist_ok=True) + for source in ( + Path(__file__).resolve(), + Path(__file__).with_name("analyze.py"), + P5_CLIENT, + P3_CLIENT_DIR / "opprof_phase3_client.py", + REPO / "docs/opprof/oracle-gap-protocol.md", + ): + target = destination / source.name + if target.exists() and sha256_file(target) != sha256_file(source): + raise RuntimeError(f"provenance file changed: {target}") + if not target.exists(): + shutil.copy2(source, target) + atomic_json(destination / "fingerprint.json", current) + atomic_json(destination / "host-before.json", gpu_snapshot()) + (destination / "nvidia-smi-q.txt").write_text(run_text(["nvidia-smi", "-q"])) + + +def rate_label(rate: float) -> str: + return f"{rate:.3f}".rstrip("0").rstrip(".") + + +def trial_key(phase: str, config: str, rate: float, repetition: int) -> str: + return f"{phase}-{config}-r{rate_label(rate)}-rep{repetition}" + + +def derived_manifest(phase: str, rate: float, repetition: int) -> Path: + label = f"{phase}-r{rate_label(rate)}-rep{repetition}" + output = RUN_ROOT / "manifests" / f"{label}.jsonl" + summary_path = output.with_suffix(output.suffix + ".summary.json") + source = PRIVATE / f"{phase}.jsonl" + domain = int.from_bytes( + hashlib.sha256(f"oracle-gap:{label}".encode()).digest()[:4], "big" + ) + if output.exists() and summary_path.exists(): + summary = json.loads(summary_path.read_text()) + if summary["sha256"] == sha256_file(output) and summary["source_sha256"] == sha256_file(source): + return output + raise RuntimeError(f"derived manifest mismatch: {output}") + output.parent.mkdir(parents=True, exist_ok=True) + temporary = output.with_name(f"{output.name}.tmp.{os.getpid()}") + rows = 0 + input_sum = 0 + output_sum = 0 + with source.open() as src, temporary.open("w") as dst: + for line in src: + row = json.loads(line) + row["token_seed"] = (int(row.get("token_seed", 0)) + domain * 1000003) & ((1 << 63) - 1) + row["token_domain"] = domain + dst.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n") + rows += 1 + input_sum += int(row["input_tokens"]) + output_sum += int(row["output_tokens"]) + dst.flush() + os.fsync(dst.fileno()) + os.replace(temporary, output) + summary = { + "schema": 1, + "phase": phase, + "rate": rate, + "repetition": repetition, + "rows": rows, + "input_token_sum": input_sum, + "output_token_sum": output_sum, + "token_domain": domain, + "source_sha256": sha256_file(source), + "sha256": sha256_file(output), + "invariants": {"rows_32768": rows == 32768, "positive_work": input_sum > 0 and output_sum > 0}, + } + if not all(summary["invariants"].values()): + raise RuntimeError(f"derived manifest invalid: {summary}") + atomic_json(summary_path, summary) + return output + + +def server_command(config: str) -> list[str]: + return [ + "taskset", "-c", CPU_MASK, str(VENV / "bin/vllm"), "serve", str(MODEL), + "--host", "127.0.0.1", "--port", str(PORT), + "--tensor-parallel-size", "1", "--enable-chunked-prefill", + "--enable-prefix-caching", "--shutdown-timeout", "120", + *CONFIG_DETAILS[config]["flags"], + ] + + +def wait_ready(process: subprocess.Popen[Any], timeout: float = 300) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError(f"server exited before ready: {process.returncode}") + try: + with urllib.request.urlopen(f"http://127.0.0.1:{PORT}/health", timeout=1) as response: + if response.status == 200: + return + except Exception: + pass + time.sleep(1) + raise TimeoutError("server readiness timeout") + + +def validate_startup(log_path: Path, config: str) -> None: + log = log_path.read_text(errors="replace") + details = CONFIG_DETAILS[config] + invariants = { + "triton_moe": "Using TRITON Unquantized MoE backend" in log, + "tp1": "tensor_parallel_size=1" in log, + "mbt": ( + "Chunked prefill is enabled with max_num_batched_tokens=8192" in log + if details["mbt"] == 8192 + else "'max_num_batched_tokens': 2048" in log + ), + "mns": details["mns"] == 1024 or "'max_num_seqs': 64" in log, + } + if not all(invariants.values()): + raise RuntimeError(f"server startup invariants failed {config}: {invariants}") + + +def start_server(config: str, stage: str, state: dict[str, Any]) -> dict[str, Any]: + assert_idle() + directory = RUN_ROOT / "servers" / stage + directory.mkdir(parents=True, exist_ok=True) + command = server_command(config) + echo = ( + f"RUN_ECHO stage={stage} host=dash0 gpu=0 cpus={CPU_MASK} config={config} " + f"model={MODEL} source={SOURCE} manifests={PRIVATE}/P01,P06.jsonl " + f"output={RUN_ROOT} expected_server_plus_trials=20-45min budget_cap={GPU_HOUR_LIMIT}H20h" + ) + with (RUN_ROOT / "launch-echo.log").open("a") as handle: + handle.write(echo + "\n") + print(echo, flush=True) + (directory / "command.txt").write_text(shlex.join(command) + "\n") + atomic_json(directory / "gpu-before.json", gpu_snapshot()) + handle = (directory / "server.log").open("ab", buffering=0) + environment = os.environ.copy() + environment.update( + { + "CUDA_VISIBLE_DEVICES": str(GPU), + "VLLM_OPPROF_DIR": str(directory / "opprof"), + "HF_HUB_OFFLINE": "1", + "TRANSFORMERS_OFFLINE": "1", + "PYTHONUNBUFFERED": "1", + "AITUNER_ORACLE_GAP_MARKER": stage, + } + ) + process = subprocess.Popen( + command, cwd=SOURCE, env=environment, stdout=handle, + stderr=subprocess.STDOUT, start_new_session=True, + ) + state["owned_pgids"] = [process.pid] + save_state(state) + wait_ready(process) + validate_startup(directory / "server.log", config) + return { + "config": config, "stage": stage, "dir": directory, "process": process, + "handle": handle, "started_at": time.time(), + } + + +def stop_server(entry: dict[str, Any], state: dict[str, Any]) -> None: + process = entry["process"] + if process.poll() is None: + try: + os.kill(process.pid, signal.SIGINT) + except ProcessLookupError: + pass + try: + process.wait(timeout=150) + except subprocess.TimeoutExpired: + for signum, timeout in ((signal.SIGTERM, 10), (signal.SIGKILL, 30)): + if process.poll() is not None: + break + try: + os.killpg(process.pid, signum) + except ProcessLookupError: + pass + try: + process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + continue + if process.poll() is None: + raise TimeoutError(f"server process group did not stop: {entry['stage']}") + entry["handle"].close() + elapsed = time.time() - float(entry["started_at"]) + state["gpu_hours"] = float(state["gpu_hours"]) + elapsed / 3600.0 + state["owned_pgids"] = [] + atomic_json(entry["dir"] / "gpu-after.json", gpu_snapshot()) + log = (entry["dir"] / "server.log").read_text(errors="replace") + if "mode=drain timeout=120s" not in log: + raise RuntimeError(f"server did not use drain shutdown: {entry['stage']}") + wait_idle_after_stop() + save_state(state) + + +def client_command( + phase: str, rate: float, repetition: int, output: Path +) -> list[str]: + timeline = TIMELINE[phase] + return [ + "taskset", "-c", CPU_MASK, str(VENV / "bin/python"), str(P5_CLIENT), "run", + "--manifest", str(derived_manifest(phase, rate, repetition)), + "--base-url", f"http://127.0.0.1:{PORT}", "--model", str(MODEL), + "--load-point", "moderate", "--fixed-request-rate", str(rate), + "--max-concurrency", "256", "--ignore-eos", "--temperature", "0", + "--warmup-seconds", str(timeline["warmup"]), + "--clean-segment-seconds", str(timeline["clean"]), + "--num-clean-segments", "1", "--post-clean-seconds", "0", + "--drain-timeout-seconds", str(timeline["drain"]), + "--workload-seed", "20260712", "--server-seed", "20260712", + "--result-dir", str(output / "client"), + ] + + +def run_trial( + state: dict[str, Any], server: dict[str, Any], phase: str, rate: float, + repetition: int, role: str, +) -> dict[str, Any]: + config = str(server["config"]) + key = trial_key(phase, config, rate, repetition) + output = RUN_ROOT / "trials" / f"{phase}-{config}" / f"rate-{rate_label(rate)}" / f"rep-{repetition}" + score_path = output / "score.json" + if key in state["completed_trials"]: + if not score_path.exists(): + raise RuntimeError(f"completed state lacks score: {key}") + return json.loads(score_path.read_text()) + output.mkdir(parents=True, exist_ok=True) + command = client_command(phase, rate, repetition, output) + (output / "command.txt").write_text(shlex.join(command) + "\n") + if server["process"].poll() is not None: + raise RuntimeError(f"server exited before trial: {key}") + assert_only_server_apps(int(server["process"].pid)) + environment = os.environ.copy() + environment["PYTHONPATH"] = str(P3_CLIENT_DIR) + os.pathsep + environment.get("PYTHONPATH", "") + started = time.time() + with (output / "client.log").open("ab", buffering=0) as handle: + result = subprocess.run( + command, cwd=REPO, env=environment, stdout=handle, + stderr=subprocess.STDOUT, timeout=900, + ) + assert_only_server_apps(int(server["process"].pid)) + if server["process"].poll() is not None: + raise RuntimeError(f"server exited during trial: {key}") + client_result = output / "client/result.json" + client_sanity = output / "client/sanity.json" + if not client_result.exists() or not client_sanity.exists(): + raise RuntimeError(f"client produced no result: {key} rc={result.returncode}") + sanity = json.loads(client_sanity.read_text())["invariants"] + failed = [name for name, passed in sanity.items() if not passed] + allowed = {"moderate_offered_within_5pct"} + if result.returncode != 0 and not failed: + raise RuntimeError(f"client failed without sanity marker: {key}") + if set(failed) - allowed: + raise RuntimeError(f"client validity failure {key}: {failed}") + score = score_trial( + output / "client/requests.jsonl", client_result, phase=phase, + config=config, target_rate=rate, repetition=repetition, role=role, + ) + score["wall_seconds"] = time.time() - started + score["client_returncode"] = result.returncode + score["client_failed_invariants"] = failed + score["manifest_sha256"] = sha256_file(derived_manifest(phase, rate, repetition)) + atomic_json(score_path, score) + state["completed_trials"].append(key) + state["last_trial"] = { + "key": key, "feasible": score["feasible"], "pass_rate": score["pass_rate"], + "goodput": score["slo_goodput_rps"], "completed_at": time.time(), + } + save_state(state) + print( + f"TRIAL {key} feasible={score['feasible']} pass={score['pass_rate']:.6f} " + f"goodput={score['slo_goodput_rps']:.6f} lagmax={score['schedule_lag_ms']['max']:.3f}", + flush=True, + ) + return score + + +def primary_bracket(rows: list[dict[str, Any]]) -> tuple[float, float] | None: + verdict = {float(row["target_rate_rps"]): bool(row["feasible"]) for row in rows} + ordered = sorted(verdict) + passes = [rate for rate in ordered if verdict[rate]] + failures = [rate for rate in ordered if not verdict[rate]] + if not passes or not failures: + return None + lower = max(passes) + upper = min((rate for rate in failures if rate > lower), default=None) + if upper is None or any(verdict[rate] for rate in ordered if rate > upper): + raise RuntimeError(f"non-monotone primary frontier: {verdict}") + return lower, upper + + +def run_primary_config(state: dict[str, Any], config: str) -> None: + stage = f"primary-{config}" + if state["stages"].get(stage, {}).get("status") == "complete": + return + state["stages"][stage] = {"status": "starting", "started_at": time.time()} + save_state(state) + server = start_server(config, stage, state) + failure = None + try: + for phase in PHASES: + phase_rows = [run_trial(state, server, phase, rate, 0, "primary") for rate in BASE_RATES[phase]] + bracket = primary_bracket(phase_rows) + if bracket is None: + extension = UP_EXTENSIONS[phase] if all(row["feasible"] for row in phase_rows) else DOWN_EXTENSIONS[phase] + for rate in extension: + phase_rows.append(run_trial(state, server, phase, rate, 0, "primary-extension")) + bracket = primary_bracket(phase_rows) + if bracket is not None: + break + if bracket is None: + raise RuntimeError(f"failed to bracket {phase}-{config}") + state["stages"][stage].setdefault("brackets", {})[phase] = list(bracket) + save_state(state) + except Exception as error: + failure = error + finally: + try: + stop_server(server, state) + except Exception as error: + failure = failure or error + if failure is not None: + state["stages"][stage]["status"] = "failed" + state["stages"][stage]["failure"] = repr(failure) + state["status"] = "failed" + save_state(state) + raise failure + state["stages"][stage]["status"] = "complete" + state["stages"][stage]["completed_at"] = time.time() + save_state(state) + + +def run_confirm_config(state: dict[str, Any], config: str) -> None: + stage = f"confirm-{config}" + if state["stages"].get(stage, {}).get("status") == "complete": + return + brackets = state["stages"][f"primary-{config}"]["brackets"] + state["stages"][stage] = {"status": "starting", "started_at": time.time(), "brackets": brackets} + save_state(state) + server = start_server(config, stage, state) + failure = None + try: + for phase in PHASES: + lower, upper = (float(value) for value in brackets[phase]) + for rate in (upper, lower): + for repetition in (1, 2): + run_trial(state, server, phase, rate, repetition, "boundary-confirmation") + except Exception as error: + failure = error + finally: + try: + stop_server(server, state) + except Exception as error: + failure = failure or error + if failure is not None: + state["stages"][stage]["status"] = "failed" + state["stages"][stage]["failure"] = repr(failure) + state["status"] = "failed" + save_state(state) + raise failure + state["stages"][stage]["status"] = "complete" + state["stages"][stage]["completed_at"] = time.time() + save_state(state) + + +def cleanup_recorded(state: dict[str, Any]) -> None: + for pgid in state.get("owned_pgids", []): + try: + os.killpg(int(pgid), signal.SIGKILL) + except ProcessLookupError: + pass + state["owned_pgids"] = [] + save_state(state) + wait_idle_after_stop() + + +def execute(resume: bool) -> None: + RUN_ROOT.mkdir(parents=True, exist_ok=True) + state = load_state(resume) + if resume and state.get("owned_pgids"): + cleanup_recorded(state) + assert_idle() + current = fingerprint() + if state["fingerprint"] and state["fingerprint"] != current: + raise RuntimeError("resume fingerprint changed") + state["fingerprint"] = current + state["status"] = "running" + save_state(state) + ensure_provenance(current) + for config in PRIMARY_ORDER: + if float(state["gpu_hours"]) >= GPU_HOUR_LIMIT: + raise RuntimeError("GPU-hour hard cap reached before primary completion") + run_primary_config(state, config) + for config in CONFIRM_ORDER: + if float(state["gpu_hours"]) >= GPU_HOUR_LIMIT: + raise RuntimeError("GPU-hour hard cap reached before confirmations") + run_confirm_config(state, config) + score_paths = sorted((RUN_ROOT / "trials").glob("**/score.json")) + summary = summarize_trials([json.loads(path.read_text()) for path in score_paths]) + summary["gpu_hours"] = state["gpu_hours"] + summary["trial_files"] = len(score_paths) + atomic_json(RUN_ROOT / "metrics.json", summary) + state["status"] = "complete" + state["completed_at"] = time.time() + state["verdict"] = summary["verdict"] + save_state(state) + print(json.dumps({"status": "complete", "verdict": summary["verdict"], "gpu_hours": state["gpu_hours"]}, sort_keys=True)) + + +def plan() -> dict[str, Any]: + primary = sum(len(BASE_RATES[phase]) for phase in PHASES) * len(CONFIGS) + confirmations = 2 * 2 * len(PHASES) * len(CONFIGS) + return { + "schema": 1, + "placement": "serialized solo GPU0", + "primary_order": list(PRIMARY_ORDER), + "confirm_order": list(CONFIRM_ORDER), + "primary_trials_without_extensions": primary, + "confirmation_trials": confirmations, + "expected_total_trials": primary + confirmations, + "expected_h20_hours": "3.0-4.0", + "hard_cap_h20_hours": GPU_HOUR_LIMIT, + "rates": {key: list(value) for key, value in BASE_RATES.items()}, + "timelines": TIMELINE, + "output": str(RUN_ROOT), + } + + +def main() -> None: + parser = argparse.ArgumentParser() + sub = parser.add_subparsers(dest="command", required=True) + run = sub.add_parser("run") + run.add_argument("--resume", action="store_true") + sub.add_parser("plan") + sub.add_parser("status") + args = parser.parse_args() + if args.command == "run": + execute(args.resume) + elif args.command == "plan": + print(json.dumps(plan(), indent=2, sort_keys=True)) + else: + print(STATE.read_text() if STATE.exists() else json.dumps({"status": "not-started"})) + + +if __name__ == "__main__": + main() diff --git a/tests/test_oracle_gap.py b/tests/test_oracle_gap.py new file mode 100644 index 0000000..ee44e7d --- /dev/null +++ b/tests/test_oracle_gap.py @@ -0,0 +1,143 @@ +from __future__ import annotations + +import json +import sys +from pathlib import Path + + +ORACLE_GAP = Path(__file__).resolve().parents[1] / "scripts/oracle_gap" +sys.path.insert(0, str(ORACLE_GAP)) + +from analyze import score_trial, summarize_trials # noqa: E402 + + +def _request( + request_id: str, + admitted: float, + first: float, + completed: float, + *, + input_tokens: int = 512, + output_tokens: int = 64, +) -> dict: + return { + "request_id": request_id, + "scheduled_s": admitted - 0.001, + "admitted_s": admitted, + "first_token_s": first, + "completed_s": completed, + "input_tokens": input_tokens, + "requested_output_tokens": output_tokens, + "actual_output_tokens": output_tokens, + "success": True, + "error_kind": None, + } + + +def test_score_trial_uses_clean_admission_cohort_and_both_slos(tmp_path: Path) -> None: + result_path = tmp_path / "result.json" + result_path.write_text( + json.dumps( + { + "warmup_seconds": 10, + "clean_segment_seconds": 20, + "num_clean_segments": 1, + } + ) + ) + rows = [ + _request("warmup", 9.0, 9.1, 10.0), + _request("pass", 11.0, 11.1, 13.0), + # (16.0 - 12.1) / 63 = 61.9 ms: TPOT miss. + _request("tpot-fail", 12.0, 12.1, 16.0), + _request("after-clean", 31.0, 31.1, 32.0), + ] + request_path = tmp_path / "requests.jsonl" + request_path.write_text("".join(json.dumps(row) + "\n" for row in rows)) + + scored = score_trial( + request_path, + result_path, + phase="P01", + config="C00", + target_rate=0.1, + repetition=0, + role="test", + ) + + assert scored["cohort_n"] == 2 + assert scored["pass_n"] == 1 + assert scored["pass_rate"] == 0.5 + assert scored["slo_goodput_rps"] == 0.05 + assert scored["failure_reasons"] == {"tpot_slo": 1} + assert scored["offered_rate_valid"] + assert not scored["feasible"] + + +def _frontier_row( + phase: str, + config: str, + rate: float, + feasible: bool, +) -> dict: + cohort = 100 + passes = 99 if feasible else 80 + return { + "phase": phase, + "config": config, + "target_rate_rps": rate, + "repetition": 0, + "role": "test", + "cohort_n": cohort, + "pass_n": passes, + "pass_rate": passes / cohort, + "slo_goodput_rps": rate * passes / cohort, + "feasible": feasible, + "invariants": {"test": True}, + } + + +def test_conservative_oracle_bound_can_refute_ten_percent_gate() -> None: + brackets = { + "P01": { + "C00": (30.0, 32.0), + "C10": (28.0, 30.0), + "C01": (29.0, 31.0), + "C11": (27.0, 29.0), + }, + "P06": { + "C00": (1.6, 1.7), + "C10": (1.7, 1.8), + "C01": (1.55, 1.65), + "C11": (1.6, 1.7), + }, + } + rows = [] + for phase, configs in brackets.items(): + for config, (lower, upper) in configs.items(): + for _ in range(3): + rows.append(_frontier_row(phase, config, lower, True)) + rows.append(_frontier_row(phase, config, upper, False)) + + summary = summarize_trials(rows) + + assert summary["verdict"] == "REFUTED" + assert summary["worst_mixture_conservative"]["gap"] < 0.10 + assert summary["sanity"]["invariants"]["all_cells_bracketed"] + assert summary["sanity"]["invariants"]["all_frontiers_monotone"] + + +def test_nonmonotone_frontier_blocks_inference() -> None: + rows = [] + for phase in ("P01", "P06"): + for config in ("C00", "C10", "C01", "C11"): + for _ in range(3): + rows.append(_frontier_row(phase, config, 1.0, True)) + rows.append(_frontier_row(phase, config, 2.0, False)) + for _ in range(3): + rows.append(_frontier_row("P01", "C00", 3.0, True)) + + summary = summarize_trials(rows) + + assert summary["verdict"] == "INCONCLUSIVE" + assert not summary["sanity"]["invariants"]["all_frontiers_monotone"]