Add static-policy oracle gap experiment

scripts/oracle_gap/__init__.py

@@ -0,0 +1 @@
+"""Oracle-gap experiment helpers."""

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()

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()

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()