diff --git a/docs/action-aware-constraint-pilot-v0-protocol-20260714.md b/docs/action-aware-constraint-pilot-v0-protocol-20260714.md new file mode 100644 index 0000000..062685d --- /dev/null +++ b/docs/action-aware-constraint-pilot-v0-protocol-20260714.md @@ -0,0 +1,179 @@ +# Action-aware constraint pilot v0 protocol + +Status: **FROZEN BEFORE NEW GPU RUNS**. + +Date: 2026-07-14 (Asia/Singapore). + +## Headline question + +Can telemetry from one complete initial-config benchmark identify which of two +competing knob families should be changed, before either target configuration +is evaluated? + +This pilot tests a narrow prerequisite, not an end-to-end tuner claim. It +uses fields already present in the per-step OpProf stream to reconstruct exact +zero-slack conditions for `max_num_seqs` (MNS) and +`max_num_batched_tokens` (MBBT). No new vLLM instrumentation is justified +unless those action-conditioned conditions predict crossed real-system +intervention responses. + +## Hypothesis + +I believe config-normalized scheduler constraints provide a stronger tuning +signal than an aggregate queue symptom because the same waiting queue can be +blocked by different admission limits. + +I will verify it by holding model, hardware, TP, request bands, arrival times, +and offered load fixed while constructing two source configurations with +different binding constraints. From each source run alone, the larger +exclusive binding fraction predicts the action family. Both candidate +actions are then measured on the same requests for the full 300-second replay. + +## Frozen platform and workload + +- Host: `dash0`, solo placement on GPUs 0-3, four NVIDIA H20 GPUs. +- Model: Qwen3-30B-A3B BF16. +- Engine: patched vLLM `0.24.1.dev3+opprof`, TP=4. +- Workload: the three disjoint `mid` bands from + `chat_w20260312_1000`, 2.125 requests/s/GPU, 300-second arrival window, + exactly 128 output tokens. +- SLO: the unchanged study TTFT/TPOT thresholds and 0.95 pass-rate target. +- Every config starts one fresh server, performs the accepted 16-request + warm-up and the existing burn-in, then runs all three disjoint measured + bands in its frozen order. +- SLO early stopping is disabled. A measured run must drain all selected + requests and finish within the 450-second client deadline. + +## Frozen configuration and action matrix + +| ID | MNS | MBBT | Role | +|---|---:|---:|---| +| `b_base` | 64 | 256 | token-budget-bound source; operational gate runs first | +| `a_base` | 16 | 8192 | MNS-bound source | +| `shared` | 64 | 8192 | MNS action from A; MBBT action from B | +| `b_mns` | 128 | 256 | competing MNS action from B | +| `a_mbbt` | 16 | 16384 | competing MBBT action from A | + +The two decisions are therefore: + +```text +Regime A: a_base -> {shared (increase MNS), a_mbbt (increase MBBT)} +Regime B: b_base -> {b_mns (increase MNS), shared (increase MBBT)} +``` + +The candidate magnitudes are intentionally large in this feasibility pilot so +that a missing crossed response is not explained by an imperceptibly small +intervention. This does not establish that these are production step sizes. + +Frozen config order is `b_base`, `a_base`, `shared`, `b_mns`, `a_mbbt`. +Frozen repetition orders are respectively `123`, `231`, `312`, `132`, and +`213`, reducing band/time alignment without reusing a server across configs. + +## Pre-action signal + +For each source run, let `waiting` include the normal and deferred waiting +queues, and let `scheduled_tokens = prefill_tokens + decode_tokens`. + +```text +mns_exclusive = waiting > 0 + and running == configured MNS + and scheduled_tokens < configured MBBT + +mbbt_exclusive = waiting > 0 + and scheduled_tokens == configured MBBT + and running < configured MNS + +both = waiting > 0 + and running == configured MNS + and scheduled_tokens == configured MBBT +``` + +Each score is the fraction of all scheduler records in the measured interval +that satisfies the condition. The predicted action is the family with the +larger exclusive fraction. This uses no target telemetry or target outcome. +KV usage and preemptions are reported as possible alternative constraints but +are not silently reassigned to either score. + +These conditions reproduce two scheduler loop boundaries, but they are still +a Level-0 proxy: they do not expose the exact request rejected at the boundary +or run a shadow schedule. The pilot explicitly tests whether that additional +engine patch is warranted. + +## Outcomes and baselines + +Primary intervention outcome: + +```text +SLO-goodput = full-run SLO pass count / 300-second arrival window +``` + +Also report pass rate, TTFT p50/p95/p99, TPOT p50/p95/p99, drain elapsed time, +KV usage, preemptions, queue area, and CUDA-graph padding. + +Required decision baselines: + +1. always choose the MNS family; +2. always choose the MBBT family; +3. queue-pressure-only, which has no candidate-specific score and therefore + must use one frozen family for both regimes; +4. the pre-action exclusive-binding prediction. + +This is a mechanism ablation. It does not compare against a trained black-box +tuner because two regimes are not a valid training surface. + +## Gates and failure meanings + +Data validity requires 15 uncensored measured runs, exact request/arrival/input +hashes across each repetition, full request accounting, one continuous OpProf +stream per config, zero dropped records, monotonic timestamps and step indices, +nonnegative counters, bounded ratios, clean GPU placement, and config values in +the result matching the server command. + +The crossed-response gate passes only if, in all three repetitions: + +- the MNS target has higher SLO-goodput than the MBBT target in Regime A; +- the MBBT target has higher SLO-goodput than the MNS target in Regime B; +- each winning target exceeds its competing target by at least 10% of the + source SLO-goodput. A source with zero goodput makes the run invalid for + this relative gate rather than changing the denominator. + +The binding gate passes only if, in both regimes: + +- the predicted family matches the measured winning family in all three + repetitions; +- the median winning-family exclusive fraction is at least 0.10; +- it is at least 5x the median competing-family exclusive fraction; +- the direction is unchanged under cumulative 25%, 50%, 75%, and 100% + checkpoints after the 25% checkpoint. + +Decision meanings: + +- `STOP_WORKLOAD_NOT_CROSSED`: candidate outcomes do not have different + winners; the experiment cannot test action selection. +- `STOP_BINDING_NOT_PREDICTIVE`: outcomes cross but source-only constraint + scores do not select them; do not implement shadow scheduling from this + hypothesis. +- `STOP_NO_NEW_INSTRUMENTATION_NEEDED`: the signal works but every required + field was already present; keep it as an analysis/tuner feature and do not + claim a new engine-instrumentation contribution. +- `OPEN_EXACT_ATTRIBUTION_ABLATION`: the signal works but unresolved/both/KV + cases are material enough that exact rejection reasons could change a + decision. Only this result authorizes a minimal vLLM attribution patch. + +Ambiguity is material only when, in either regime, the median +`both + waiting_unresolved` fraction is at least the median absolute gap +between the two exclusive fractions, or when any source run records a +preemption or median source KV maximum is at least 0.90. Otherwise all fields +needed for the observed decision were already present and the result is +`STOP_NO_NEW_INSTRUMENTATION_NEEDED`. + +No result from this development pilot is a paper-level E2E tuning claim. + +## Cost and stopping discipline + +- Hard cap: 8.0 H20-hours, including failed sessions. +- Expected: 6.0-7.2 H20-hours and 90-110 minutes wall time. +- `b_base` runs first. If its first measured band cannot drain by 450 seconds, + the controller stops before any comparative analysis; MBBT=256 is then an + operationally invalid source, not negative evidence. +- Any data red flag stops analysis before computing a tuning conclusion. diff --git a/runs/action-aware-v0/action_aware_client.py b/runs/action-aware-v0/action_aware_client.py new file mode 100644 index 0000000..58fbe27 --- /dev/null +++ b/runs/action-aware-v0/action_aware_client.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""Add explicit MBBT/config provenance to the accepted Phase-6 replay client.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + + +PHASE6 = Path(__file__).resolve().parents[1] / "opprof-phase6" +sys.path.insert(0, str(PHASE6)) + +import opprof_phase6_client as base # noqa: E402 + + +def parser() -> argparse.ArgumentParser: + result = argparse.ArgumentParser() + result.add_argument("command", choices=("warmup", "run-anchor")) + result.add_argument("--study", required=True) + result.add_argument("--cell", required=True) + result.add_argument("--anchor", type=float, required=True) + result.add_argument("--tp", type=int, required=True) + result.add_argument("--mns", type=int, required=True) + result.add_argument("--mbbt", type=int, required=True) + result.add_argument("--base-url", required=True) + result.add_argument("--result-dir", required=True) + result.add_argument("--disable-slo-early-stop", action="store_true") + return result + + +def main() -> None: + args = parser().parse_args() + result = base.run_replay(args, warmup=args.command == "warmup") + result.update( + { + "schema": "action-aware-pilot-result-v0", + "config_id": args.cell, + "mbbt": args.mbbt, + } + ) + base.atomic_json(Path(args.result_dir) / "result.json", result) + print( + json.dumps( + { + key: result[key] + for key in ( + "config_id", + "mns", + "mbbt", + "kind", + "pass_rate", + "feasible", + ) + }, + sort_keys=True, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/runs/action-aware-v0/analyze_pilot.py b/runs/action-aware-v0/analyze_pilot.py new file mode 100644 index 0000000..5eafbd2 --- /dev/null +++ b/runs/action-aware-v0/analyze_pilot.py @@ -0,0 +1,584 @@ +#!/usr/bin/env python3 +"""Audit source-only constraint signals against crossed real interventions.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import os +import statistics +import sys +from pathlib import Path +from typing import Any, Iterable, Mapping + + +HERE = Path(__file__).resolve().parent +COMMON_STATE = HERE.parent / "telemetry-residual" +sys.path.insert(0, str(COMMON_STATE)) + +from common_state import summarize_engine # noqa: E402 + + +SCHEMA = "action-aware-constraint-pilot-audit-v0" + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def atomic_json(path: Path, payload: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text( + json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + os.replace(temporary, path) + + +def numeric(values: Iterable[float]) -> dict[str, Any]: + finite = [float(value) for value in values] + if not finite: + raise ValueError("numeric summary requires values") + if any(not math.isfinite(value) for value in finite): + raise ValueError("numeric summary received non-finite values") + return { + "n": len(finite), + "min": min(finite), + "max": max(finite), + "distinct_n": len(set(finite)), + } + + +def quantile(values: Iterable[float], probability: float) -> float: + ordered = sorted(float(value) for value in values) + if not ordered: + raise ValueError("quantile requires values") + position = probability * (len(ordered) - 1) + lower = math.floor(position) + upper = math.ceil(position) + if lower == upper: + return ordered[lower] + weight = position - lower + return ordered[lower] * (1.0 - weight) + ordered[upper] * weight + + +def load_jsonl(path: Path) -> list[dict[str, Any]]: + records = [] + with path.open(encoding="utf-8") as source: + for line_number, line in enumerate(source, 1): + try: + records.append(json.loads(line)) + except json.JSONDecodeError as error: + raise ValueError(f"{path}:{line_number}: invalid JSON") from error + return records + + +def binding_summary( + records: list[Mapping[str, Any]], *, mns: int, mbbt: int +) -> dict[str, Any]: + if not records: + raise ValueError("binding summary requires scheduler records") + counts = { + "mns_exclusive": 0, + "mbbt_exclusive": 0, + "both": 0, + "waiting_unresolved": 0, + "waiting": 0, + } + running_utilization = [] + token_utilization = [] + kv_usage = [] + preemptions = 0 + for record in records: + waiting = int(record["queues"]["waiting"]) + int( + record["queues"]["deferred"] + ) + running = int(record["queues"]["running"]) + scheduled_tokens = int(record["prefill_tokens"]) + int( + record["decode_tokens"] + ) + if running > mns: + raise ValueError("running requests exceed configured MNS") + if scheduled_tokens > mbbt: + raise ValueError("scheduled tokens exceed configured MBBT") + mns_hit = waiting > 0 and running == mns + mbbt_hit = waiting > 0 and scheduled_tokens == mbbt + if waiting > 0: + counts["waiting"] += 1 + if mns_hit and mbbt_hit: + counts["both"] += 1 + elif mns_hit: + counts["mns_exclusive"] += 1 + elif mbbt_hit: + counts["mbbt_exclusive"] += 1 + else: + counts["waiting_unresolved"] += 1 + running_utilization.append(running / mns) + token_utilization.append(scheduled_tokens / mbbt) + kv_usage.append(float(record["kv"]["usage"])) + preemptions += int(record["preemptions"]) + count = len(records) + return { + "records": count, + **{f"{name}_count": value for name, value in counts.items()}, + **{f"{name}_fraction": value / count for name, value in counts.items()}, + "running_utilization_mean": statistics.fmean(running_utilization), + "running_utilization_max": max(running_utilization), + "token_utilization_mean": statistics.fmean(token_utilization), + "token_utilization_max": max(token_utilization), + "kv_usage_mean": statistics.fmean(kv_usage), + "kv_usage_max": max(kv_usage), + "preemptions": preemptions, + } + + +def request_summary(path: Path, expected_count: int) -> dict[str, Any]: + rows = load_jsonl(path) + if len(rows) != expected_count: + raise ValueError(f"request row count mismatch: {path}") + ttft = [float(row["ttft_ms"]) for row in rows if row["ttft_ms"] is not None] + tpot = [float(row["tpot_ms"]) for row in rows if row["tpot_ms"] is not None] + if not ttft or not tpot: + raise ValueError(f"missing request latency values: {path}") + return { + "ttft_ms": {f"p{int(p * 100)}": quantile(ttft, p) for p in (0.5, 0.95, 0.99)}, + "tpot_ms": {f"p{int(p * 100)}": quantile(tpot, p) for p in (0.5, 0.95, 0.99)}, + } + + +def load_stream(session_root: Path) -> tuple[list[dict[str, Any]], dict[str, Any]]: + streams = sorted((session_root / "opprof").glob("*.jsonl")) + sidecars = sorted((session_root / "opprof").glob("*.jsonl.footer.json")) + if len(streams) != 1 or len(sidecars) != 1: + raise ValueError(f"expected one OpProf stream and sidecar: {session_root}") + decoded = load_jsonl(streams[0]) + records = [row for row in decoded if "step_index" in row] + footers = [row for row in decoded if row.get("record_type") == "footer"] + sidecar = json.loads(sidecars[0].read_text(encoding="utf-8")) + indexes = [int(row["step_index"]) for row in records] + invariants = { + "one_footer_last": len(footers) == 1 and decoded[-1] is footers[0], + "sidecar_final": sidecar.get("final") is True, + "zero_drops": sidecar.get("dropped_records") == 0, + "written_matches_records": sidecar.get("written_records") == len(records), + "contiguous_step_indexes": indexes == list(range(len(indexes))), + "monotonic_timestamps": all( + int(right["submit_mono_ns"]) >= int(left["submit_mono_ns"]) + for left, right in zip(records, records[1:], strict=False) + ), + } + return records, { + "stream": str(streams[0]), + "stream_sha256": sha256_file(streams[0]), + "records": len(records), + "invariants": invariants, + } + + +def analyze_run( + *, + run_root: Path, + config: Mapping[str, Any], + repetition: int, + expected: Mapping[str, Any], + stream_records: list[Mapping[str, Any]], + duration_s: float, + phase_fractions: list[float], +) -> dict[str, Any]: + result_root = run_root / "sessions" / str(config["id"]) / f"rep{repetition}" + result_path = result_root / "result.json" + result = json.loads(result_path.read_text(encoding="utf-8")) + selection = result["selection"] + invariants = { + "result_schema": result.get("schema") == "action-aware-pilot-result-v0", + "config_id": result.get("config_id") == config["id"], + "tp": int(result.get("tp", -1)) == 4, + "mns": int(result.get("mns", -1)) == int(config["mns"]), + "mbbt": int(result.get("mbbt", -1)) == int(config["mbbt"]), + "uncensored": not bool(result.get("early_stopped", True)), + "slo_early_stop_disabled": result.get("slo_early_stop_disabled") is True, + "selection_count": int(selection["count"]) == int(expected["selected_count"]), + "request_accounting": int(result["observed_count"]) + == int(expected["selected_count"]), + "request_hash": selection["request_id_order_sha256"] + == expected["request_id_order_sha256"], + "arrival_hash": selection["arrival_order_sha256"] + == expected["arrival_order_sha256"], + "length_hash": selection["raw_length_order_sha256"] + == expected["input_length_order_sha256"], + } + start_ns = int(result["interval"]["start_mono_ns"]) + arrival_end_ns = start_ns + round(duration_s * 1e9) + full_records = [ + record + for record in stream_records + if start_ns <= int(record["submit_mono_ns"]) <= arrival_end_ns + ] + if not full_records: + raise ValueError(f"no telemetry records in measured window: {result_path}") + gaps = [ + (int(right["submit_mono_ns"]) - int(left["submit_mono_ns"])) / 1e9 + for left, right in zip(full_records, full_records[1:], strict=False) + ] + coverage = { + "start_gap_s": (int(full_records[0]["submit_mono_ns"]) - start_ns) / 1e9, + "end_gap_s": (arrival_end_ns - int(full_records[-1]["submit_mono_ns"])) / 1e9, + "max_internal_gap_s": max(gaps, default=0.0), + } + invariants["telemetry_coverage"] = all( + 0.0 <= value <= 1.0 for value in coverage.values() + ) + binding = binding_summary( + full_records, mns=int(config["mns"]), mbbt=int(config["mbbt"]) + ) + phases = {} + for fraction in phase_fractions: + phase_end = start_ns + round(duration_s * fraction * 1e9) + phase_records = [ + record + for record in full_records + if int(record["submit_mono_ns"]) <= phase_end + ] + phases[f"{fraction:.2f}"] = binding_summary( + phase_records, mns=int(config["mns"]), mbbt=int(config["mbbt"]) + ) + state = summarize_engine( + full_records, + start_ns=start_ns, + end_ns=arrival_end_ns, + request_count=int(result["observed_count"]), + ) + latency = request_summary( + result_root / "requests.jsonl", int(result["observed_count"]) + ) + return { + "config_id": config["id"], + "mns": int(config["mns"]), + "mbbt": int(config["mbbt"]), + "repetition": repetition, + "result_path": str(result_path), + "result_sha256": sha256_file(result_path), + "selection": { + "count": int(selection["count"]), + "request_id_order_sha256": selection["request_id_order_sha256"], + "arrival_order_sha256": selection["arrival_order_sha256"], + "raw_length_order_sha256": selection["raw_length_order_sha256"], + }, + "outcome": { + "pass_rate": float(result["pass_rate"]), + "feasible": bool(result["feasible"]), + "slo_pass_count": int(result["slo_pass_count"]), + "slo_goodput_req_s": int(result["slo_pass_count"]) / duration_s, + "elapsed_s": float(result["interval"]["elapsed_s"]), + **latency, + }, + "binding": binding, + "phases": phases, + "state": state, + "coverage": coverage, + "invariants": invariants, + } + + +def median(values: Iterable[float]) -> float: + return float(statistics.median(float(value) for value in values)) + + +def evaluate_decisions( + runs: list[Mapping[str, Any]], manifest: Mapping[str, Any] +) -> dict[str, Any]: + by_key = { + (str(run["config_id"]), int(run["repetition"])): run for run in runs + } + repetitions = sorted(int(key) for key in manifest["repetitions"]) + regime_results = {} + all_predictions = [] + crossed_pass = True + binding_pass = True + material_ambiguity = False + for regime_name, regime in manifest["regimes"].items(): + rows = [] + source_runs = [] + for repetition in repetitions: + source = by_key[(str(regime["source"]), repetition)] + mns_target = by_key[(str(regime["actions"]["mns"]), repetition)] + mbbt_target = by_key[(str(regime["actions"]["mbbt"]), repetition)] + source_runs.append(source) + source_goodput = float(source["outcome"]["slo_goodput_req_s"]) + mns_goodput = float(mns_target["outcome"]["slo_goodput_req_s"]) + mbbt_goodput = float(mbbt_target["outcome"]["slo_goodput_req_s"]) + observed = ( + "mns" + if mns_goodput > mbbt_goodput + else "mbbt" + if mbbt_goodput > mns_goodput + else "tie" + ) + mns_score = float(source["binding"]["mns_exclusive_fraction"]) + mbbt_score = float(source["binding"]["mbbt_exclusive_fraction"]) + predicted = ( + "mns" + if mns_score > mbbt_score + else "mbbt" + if mbbt_score > mns_score + else "tie" + ) + phase_predictions = {} + for phase, summary in source["phases"].items(): + left = float(summary["mns_exclusive_fraction"]) + right = float(summary["mbbt_exclusive_fraction"]) + phase_predictions[phase] = ( + "mns" if left > right else "mbbt" if right > left else "tie" + ) + margin = ( + abs(mns_goodput - mbbt_goodput) / source_goodput + if source_goodput > 0 + else None + ) + row = { + "repetition": repetition, + "source_goodput_req_s": source_goodput, + "mns_target_goodput_req_s": mns_goodput, + "mbbt_target_goodput_req_s": mbbt_goodput, + "observed_winner": observed, + "predicted_winner": predicted, + "prediction_correct": predicted == observed, + "relative_winner_margin_over_source": margin, + "mns_exclusive_fraction": mns_score, + "mbbt_exclusive_fraction": mbbt_score, + "phase_predictions": phase_predictions, + "phase_stable": all(value == predicted for value in phase_predictions.values()), + } + rows.append(row) + all_predictions.append(row) + + expected_winner = "mns" if regime_name == "A" else "mbbt" + minimum_margin = float(manifest["gates"]["minimum_relative_winner_margin"]) + regime_crossed = all( + row["observed_winner"] == expected_winner + and row["relative_winner_margin_over_source"] is not None + and row["relative_winner_margin_over_source"] >= minimum_margin + for row in rows + ) + crossed_pass &= regime_crossed + winning_key = f"{expected_winner}_exclusive_fraction" + losing_key = ( + "mbbt_exclusive_fraction" if expected_winner == "mns" else "mns_exclusive_fraction" + ) + winning_median = median(row[winning_key] for row in rows) + losing_median = median(row[losing_key] for row in rows) + ratio_pass = winning_median >= float( + manifest["gates"]["minimum_exclusive_ratio"] + ) * losing_median + regime_binding = ( + all(row["prediction_correct"] and row["phase_stable"] for row in rows) + and winning_median + >= float(manifest["gates"]["minimum_exclusive_fraction"]) + and ratio_pass + ) + binding_pass &= regime_binding + ambiguity_median = median( + float(run["binding"]["both_fraction"]) + + float(run["binding"]["waiting_unresolved_fraction"]) + for run in source_runs + ) + score_gap_median = median( + abs( + float(run["binding"]["mns_exclusive_fraction"]) + - float(run["binding"]["mbbt_exclusive_fraction"]) + ) + for run in source_runs + ) + kv_max_median = median( + float(run["binding"]["kv_usage_max"]) for run in source_runs + ) + any_preemption = any( + int(run["binding"]["preemptions"]) > 0 for run in source_runs + ) + regime_material = ( + ambiguity_median >= score_gap_median + or kv_max_median >= float(manifest["gates"]["material_kv_usage"]) + or any_preemption + ) + material_ambiguity |= regime_material + regime_results[regime_name] = { + "source": regime["source"], + "actions": regime["actions"], + "expected_winner": expected_winner, + "crossed_response_pass": regime_crossed, + "binding_pass": regime_binding, + "winning_exclusive_median": winning_median, + "losing_exclusive_median": losing_median, + "exclusive_ratio_pass": ratio_pass, + "ambiguity_median": ambiguity_median, + "exclusive_gap_median": score_gap_median, + "kv_usage_max_median": kv_max_median, + "any_preemption": any_preemption, + "material_ambiguity": regime_material, + "repetitions": rows, + } + + if not crossed_pass: + decision = "STOP_WORKLOAD_NOT_CROSSED" + elif not binding_pass: + decision = "STOP_BINDING_NOT_PREDICTIVE" + elif material_ambiguity: + decision = "OPEN_EXACT_ATTRIBUTION_ABLATION" + else: + decision = "STOP_NO_NEW_INSTRUMENTATION_NEEDED" + correct = sum(int(row["prediction_correct"]) for row in all_predictions) + return { + "decision": decision, + "crossed_response_pass": crossed_pass, + "binding_pass": binding_pass, + "material_ambiguity": material_ambiguity, + "regimes": regime_results, + "baselines": { + "always_mns_correct": sum( + int(row["observed_winner"] == "mns") for row in all_predictions + ), + "always_mbbt_correct": sum( + int(row["observed_winner"] == "mbbt") for row in all_predictions + ), + "binding_correct": correct, + "decision_count": len(all_predictions), + }, + } + + +def analyze(run_root: Path, manifest_path: Path) -> dict[str, Any]: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + if manifest.get("schema") != "action-aware-constraint-pilot-manifest-v0": + raise ValueError("unexpected manifest schema") + duration_s = float(manifest["engine"]["duration_s"]) + phase_fractions = [float(value) for value in manifest["gates"]["phase_fractions"]] + runs = [] + stream_audits = [] + for config in manifest["configs"]: + session_root = run_root / "sessions" / str(config["id"]) + stream_records, stream_audit = load_stream(session_root) + stream_audit["config_id"] = config["id"] + stream_audits.append(stream_audit) + for repetition in sorted(int(key) for key in manifest["repetitions"]): + runs.append( + analyze_run( + run_root=run_root, + config=config, + repetition=repetition, + expected=manifest["repetitions"][str(repetition)]["selection"], + stream_records=stream_records, + duration_s=duration_s, + phase_fractions=phase_fractions, + ) + ) + invariants = { + "fifteen_runs": len(runs) == 15, + "five_streams": len(stream_audits) == 5, + "all_run_invariants": all( + all(bool(value) for value in run["invariants"].values()) for run in runs + ), + "all_stream_invariants": all( + all(bool(value) for value in stream["invariants"].values()) + for stream in stream_audits + ), + "nonnegative_counters": all( + all( + float(run["binding"][key]) >= 0 + for key in ( + "mns_exclusive_count", + "mbbt_exclusive_count", + "both_count", + "waiting_unresolved_count", + "preemptions", + ) + ) + for run in runs + ), + "ratios_bounded": all( + all( + 0.0 <= float(run["binding"][key]) <= 1.0 + for key in ( + "mns_exclusive_fraction", + "mbbt_exclusive_fraction", + "both_fraction", + "waiting_unresolved_fraction", + "kv_usage_mean", + "kv_usage_max", + ) + ) + for run in runs + ), + "per_config_results_not_all_identical": len( + {float(run["outcome"]["pass_rate"]) for run in runs} + ) + > 1, + } + red_flags = [name for name, passed in invariants.items() if not passed] + decisions = ( + evaluate_decisions(runs, manifest) + if not red_flags + else { + "decision": "STOP_DATA_INVALID", + "crossed_response_pass": False, + "binding_pass": False, + "material_ambiguity": False, + "regimes": {}, + "baselines": {}, + } + ) + payload = { + "schema": SCHEMA, + "decision": decisions["decision"], + "manifest": str(manifest_path), + "manifest_sha256": sha256_file(manifest_path), + "run_root": str(run_root), + "runs": runs, + "streams": stream_audits, + "decision_audit": decisions, + "sanity": { + "runs": len(runs), + "pass_rate": numeric(run["outcome"]["pass_rate"] for run in runs), + "slo_goodput_req_s": numeric( + run["outcome"]["slo_goodput_req_s"] for run in runs + ), + "telemetry_records_per_run": numeric( + run["binding"]["records"] for run in runs + ), + "mns_values": numeric(run["mns"] for run in runs), + "mbbt_values": numeric(run["mbbt"] for run in runs), + "invariants": invariants, + "red_flags": red_flags, + }, + } + return payload + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--run-root", type=Path, required=True) + parser.add_argument("--manifest", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + payload = analyze(args.run_root, args.manifest) + atomic_json(args.output, payload) + print( + json.dumps( + { + "decision": payload["decision"], + "sanity": payload["sanity"], + "decision_audit": payload["decision_audit"], + }, + indent=2, + sort_keys=True, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/runs/action-aware-v0/pilot-manifest.json b/runs/action-aware-v0/pilot-manifest.json new file mode 100644 index 0000000..170c02f --- /dev/null +++ b/runs/action-aware-v0/pilot-manifest.json @@ -0,0 +1,223 @@ +{ + "budget": { + "expected_h20_hours": [ + 6.0, + 7.2 + ], + "expected_wall_minutes": [ + 90, + 110 + ], + "hard_cap_h20_hours": 8.0, + "safety_h20_hours": 0.25, + "session_estimate_h20_hours": 1.35 + }, + "burnin": { + "anchor": 0.18919793755240089, + "arrival_order_sha256": "6c0ac4cb9a30ef501eeeacc8e6cc631c345e976db5ccf530ea5a1ec706d62a24", + "input_length_order_sha256": "7939cc20e1a00d1031d27d71508789f38decbbbb6ea59a1df18b2ec342fd2ef8", + "offered_req_s": 8.5, + "offered_req_s_per_gpu": 2.125, + "request_id_order_sha256": "84f4809acbc8acd3b1d14dfa357134a1dc0b9287341624b33f598dafeef54dc7", + "selected_count": 510, + "study": "/home/admin/cpfs/wjh/fidelity-prefix-pilot-20260714/private/studies/burnin-tp4.json", + "study_sha256": "5d6c2098042909a863efd3112818fbee9bafe96f22898ac98b66846dbe1fef0f" + }, + "configs": [ + { + "id": "b_base", + "mbbt": 256, + "mns": 64, + "repetition_order": [ + 1, + 2, + 3 + ] + }, + { + "id": "a_base", + "mbbt": 8192, + "mns": 16, + "repetition_order": [ + 2, + 3, + 1 + ] + }, + { + "id": "shared", + "mbbt": 8192, + "mns": 64, + "repetition_order": [ + 3, + 1, + 2 + ] + }, + { + "id": "b_mns", + "mbbt": 256, + "mns": 128, + "repetition_order": [ + 1, + 3, + 2 + ] + }, + { + "id": "a_mbbt", + "mbbt": 16384, + "mns": 16, + "repetition_order": [ + 2, + 1, + 3 + ] + } + ], + "engine": { + "client_timeout_s": 450.0, + "disable_slo_early_stop": true, + "duration_s": 300.0, + "tp": 4 + }, + "gates": { + "material_kv_usage": 0.9, + "minimum_exclusive_fraction": 0.1, + "minimum_exclusive_ratio": 5.0, + "minimum_relative_winner_margin": 0.1, + "phase_fractions": [ + 0.25, + 0.5, + 0.75, + 1.0 + ] + }, + "regimes": { + "A": { + "actions": { + "mbbt": "a_mbbt", + "mns": "shared" + }, + "source": "a_base" + }, + "B": { + "actions": { + "mbbt": "shared", + "mns": "b_mns" + }, + "source": "b_base" + } + }, + "repetitions": { + "1": { + "merged_trace": { + "bytes": 337429767, + "path": "/home/admin/cpfs/wjh/intervention-response-v3-20260714/private/traces/rep1.jsonl", + "request_id_scheme": "sha256(source_sha256:line_number:original_id)", + "rows": 9420, + "sha256": "68983266aa0e66aa589562f7c08edbd966f9ba4405e20c105adb43777d2dfbf5", + "source_sha256": [ + "b242d1d9086df3accab57b4c92445d5edd581e12f47e12cea227aa63964c6930", + "d23b549f7b69af3647308677bbf76f818a3c226a1c98f9a9f93f09ceee46be87" + ], + "sources": [ + "/home/admin/cpfs/wjh/fidelity-prefix-pilot-20260714/private/traces/low1.jsonl", + "/home/admin/cpfs/wjh/fidelity-prefix-pilot-20260714/private/traces/high1.jsonl" + ] + }, + "selection": { + "anchor": 0.48686986110831465, + "arrival_order_sha256": "c2ad99986ce558da5901a9c5ec0a00bd69f198c981d8779235f2773a5c87f1c0", + "input_length_order_sha256": "9442bfebdc3fab5062dc1f4d688dc28c02afe3fd806c56dd8159f0ac7e6d0b94", + "offered_req_s": 8.5, + "offered_req_s_per_gpu": 2.125, + "request_id_order_sha256": "0bb61dbc9c26875e991d0d4f984134910d37463e5063f86ee960cf4f8aafb771", + "selected_count": 2550, + "target_count": 2550, + "target_req_s_per_gpu": 2.125 + }, + "study": "/home/admin/cpfs/wjh/intervention-response-v3-20260714/private/studies/rep1-tp4.json", + "study_sha256": "ecfff96e33d458eb1e3b9a6d24386f00cc6f1b19ff926e2ec6320b3f671a7ae3" + }, + "2": { + "merged_trace": { + "bytes": 337509330, + "path": "/home/admin/cpfs/wjh/intervention-response-v3-20260714/private/traces/rep2.jsonl", + "request_id_scheme": "sha256(source_sha256:line_number:original_id)", + "rows": 9457, + "sha256": "f38e8938f6a481fc6725b71b21aa04ff7eaf79783cdfd6e41aa2f074156f00c2", + "source_sha256": [ + "4cbb0baac082bd54af562ce2f39104c5c23b4671672da365a67b1e8c146adf9f", + "bb0bcd2564a88000f435f12feb21c7c902eafc9ea5fe916adfe9d1eae47f3f9a" + ], + "sources": [ + "/home/admin/cpfs/wjh/fidelity-prefix-pilot-20260714/private/traces/low2.jsonl", + "/home/admin/cpfs/wjh/fidelity-prefix-pilot-20260714/private/traces/high2.jsonl" + ] + }, + "selection": { + "anchor": 0.4825698948735577, + "arrival_order_sha256": "b9fc12cf3f86bc8a79bee65296e65aa2b8bf2aeca46b2887094c669adcbb9a00", + "input_length_order_sha256": "d8d4bd6fc8ba852a45605b673b6b3e4f33b58f459e69f2a032d226ee175b074e", + "offered_req_s": 8.5, + "offered_req_s_per_gpu": 2.125, + "request_id_order_sha256": "56a0616b6b54abafd37875c7cb25f8639afef2706ccc55dfbe568f45859ea382", + "selected_count": 2550, + "target_count": 2550, + "target_req_s_per_gpu": 2.125 + }, + "study": "/home/admin/cpfs/wjh/intervention-response-v3-20260714/private/studies/rep2-tp4.json", + "study_sha256": "d92a576db031db24bb58f354ea725d7f7567cb76699d387117ac5a6c9317bbb9" + }, + "3": { + "merged_trace": { + "bytes": 337450256, + "path": "/home/admin/cpfs/wjh/intervention-response-v3-20260714/private/traces/rep3.jsonl", + "request_id_scheme": "sha256(source_sha256:line_number:original_id)", + "rows": 9431, + "sha256": "3094084b0bb20cc02eecf465091a5c919b4e5b112f704cdc36a563d1efdcee46", + "source_sha256": [ + "1f7ececb142f9a363d2d1ca25eb7b8488b2cc319a51b55faa384f2a3d51f2142", + "6f326234791e1cff4ff866bface0d097d0d6e3844eebb1c97653d8e9c35e9397" + ], + "sources": [ + "/home/admin/cpfs/wjh/fidelity-prefix-pilot-20260714/private/traces/low3.jsonl", + "/home/admin/cpfs/wjh/fidelity-prefix-pilot-20260714/private/traces/high3.jsonl" + ] + }, + "selection": { + "anchor": 0.48664343020532463, + "arrival_order_sha256": "efce7339e22d3618cb4d55e6b55bfddb2c563c18faba2a992d5829c13e3f55e9", + "input_length_order_sha256": "0792b05fff6729fbd92ab2bb4cb6d31bea7799e232ad42772936bc06efbafb54", + "offered_req_s": 8.5, + "offered_req_s_per_gpu": 2.125, + "request_id_order_sha256": "2a2fabe2c4cf176aeb7e0d32fb8e7dbb1f27429a2e7a0cd18d7d186f23096f19", + "selected_count": 2550, + "target_count": 2550, + "target_req_s_per_gpu": 2.125 + }, + "study": "/home/admin/cpfs/wjh/intervention-response-v3-20260714/private/studies/rep3-tp4.json", + "study_sha256": "fb8ffe256dace32f4ca8a8d49b662d98c3b69b94ecc8fa826e43068b238884ab" + } + }, + "sanity": { + "invariants": { + "all_repetition_orders_are_permutations": true, + "five_unique_configs": true, + "same_load_all_repetitions": true, + "shared_endpoint_reused_by_both_regimes": true, + "three_disjoint_repetitions": true + }, + "red_flags": [] + }, + "schema": "action-aware-constraint-pilot-manifest-v0", + "source": { + "base_manifest": "/home/gahow/phd/aituner/runs/intervention-response-v2/pilot-manifest-v3.json", + "base_manifest_sha256": "273db1181dcc9d6b64439650d0642ebe553b12e6aa9adebfbe3758a7977e5611", + "source_trace": "/home/admin/cpfs/wjh/aituner/aituner/trace_windows/traces/chat_w20260312_1000.jsonl", + "source_trace_sha256": "875ba869775deb78086477919f03b322da14e2673c7d070e26528c4190912757", + "window_id": "chat_w20260312_1000" + }, + "status": "PASS" +} diff --git a/runs/action-aware-v0/pilot_controller.py b/runs/action-aware-v0/pilot_controller.py new file mode 100644 index 0000000..8e8b94f --- /dev/null +++ b/runs/action-aware-v0/pilot_controller.py @@ -0,0 +1,595 @@ +#!/usr/bin/env python3 +"""Serialized controller for the crossed-constraint action-aware pilot.""" + +from __future__ import annotations + +import argparse +import json +import os +import shlex +import signal +import subprocess +import sys +import time +from pathlib import Path +from typing import Any, Mapping + + +HERE = Path(__file__).resolve().parent +PHASE6 = HERE.parent / "opprof-phase6" +sys.path.insert(0, str(PHASE6)) + +import opprof_phase6_controller as base # noqa: E402 + + +SCHEMA = "action-aware-constraint-pilot-state-v0" + + +def atomic_json(path: Path, payload: Any) -> None: + base.atomic_json(path, payload) + + +def wait_all_idle(timeout_s: float = 30.0) -> None: + deadline = time.monotonic() + timeout_s + last_error: Exception | None = None + while time.monotonic() < deadline: + try: + base.assert_all_idle() + return + except RuntimeError as error: + last_error = error + time.sleep(1.0) + raise last_error or RuntimeError("GPU idle timeout") + + +def configure(args: argparse.Namespace, manifest: Mapping[str, Any]) -> None: + base.WORKDIR = args.run_root.parent + base.RUN_ROOT = args.run_root + base.STATE = args.run_root / "controller-state.json" + base.SOURCE = args.vllm_source + base.VENV = args.venv + base.AITUNER = args.aituner_root + base.MODEL = args.model + base.CLIENT = args.client + base.GPU_LIMIT = float(manifest["budget"]["hard_cap_h20_hours"]) + base.MARKER = "action-aware-constraint-pilot-v0" + + +def validate_inputs(args: argparse.Namespace, manifest: Mapping[str, Any]) -> None: + if manifest.get("schema") != "action-aware-constraint-pilot-manifest-v0": + raise RuntimeError("unexpected action-aware manifest schema") + if manifest.get("status") != "PASS": + raise RuntimeError("action-aware manifest did not pass preflight") + red_flags = manifest.get("sanity", {}).get("red_flags", []) + if red_flags: + raise RuntimeError(f"manifest red flags: {red_flags}") + + required = { + "manifest": args.manifest, + "aituner_root": args.aituner_root, + "vllm_source": args.vllm_source, + "venv_python": args.venv / "bin/python", + "venv_vllm": args.venv / "bin/vllm", + "model": args.model, + "client": args.client, + "burnin_study": Path(manifest["burnin"]["study"]), + } + for repetition, item in manifest["repetitions"].items(): + required[f"rep{repetition}_study"] = Path(item["study"]) + required[f"rep{repetition}_trace"] = Path(item["merged_trace"]["path"]) + missing = {name: str(path) for name, path in required.items() if not path.exists()} + if missing: + raise RuntimeError(f"action-aware input paths missing: {missing}") + + +def config_map(manifest: Mapping[str, Any]) -> dict[str, dict[str, Any]]: + return {str(item["id"]): dict(item) for item in manifest["configs"]} + + +def server_command( + config: Mapping[str, Any], *, gpus: tuple[int, ...], port: int +) -> list[str]: + return [ + "taskset", + "-c", + base.cpu_mask(gpus), + str(base.VENV / "bin/vllm"), + "serve", + str(base.MODEL), + "--host", + "127.0.0.1", + "--port", + str(port), + "--served-model-name", + "qwen3-30b-a3b-community", + "--max-num-batched-tokens", + str(config["mbbt"]), + "--max-num-seqs", + str(config["mns"]), + "--tensor-parallel-size", + "4", + "--shutdown-timeout", + "120", + ] + + +def client_command( + entry: Mapping[str, Any], + config: Mapping[str, Any], + *, + study: str, + anchor: float, + output: Path, + warmup: bool, +) -> list[str]: + command = [ + "taskset", + "-c", + base.cpu_mask(entry["gpus"]), + str(base.VENV / "bin/python"), + str(base.CLIENT), + "warmup" if warmup else "run-anchor", + "--study", + study, + "--cell", + str(config["id"]), + "--anchor", + str(anchor), + "--tp", + "4", + "--mns", + str(config["mns"]), + "--mbbt", + str(config["mbbt"]), + "--base-url", + f"http://127.0.0.1:{entry['port']}", + "--result-dir", + str(output), + "--disable-slo-early-stop", + ] + return command + + +def remaining_projection( + manifest: Mapping[str, Any], *, completed_sessions: int +) -> float: + remaining = len(manifest["configs"]) - completed_sessions + return ( + remaining * float(manifest["budget"]["session_estimate_h20_hours"]) + + float(manifest["budget"]["safety_h20_hours"]) + ) + + +def dry_run_plan( + args: argparse.Namespace, manifest: Mapping[str, Any] +) -> dict[str, Any]: + sessions = [] + for index, config in enumerate(manifest["configs"]): + entry = {"gpus": (0, 1, 2, 3), "port": 9050 + index} + session_root = args.run_root / "sessions" / str(config["id"]) + first_repetition = str(config["repetition_order"][0]) + first = manifest["repetitions"][first_repetition] + commands = { + "server": server_command(config, gpus=entry["gpus"], port=entry["port"]), + "warmup": client_command( + entry, + config, + study=first["study"], + anchor=float(first["selection"]["anchor"]), + output=session_root / "warmup", + warmup=True, + ), + "burnin": client_command( + entry, + config, + study=manifest["burnin"]["study"], + anchor=float(manifest["burnin"]["anchor"]), + output=session_root / "burnin", + warmup=False, + ), + } + for repetition in config["repetition_order"]: + item = manifest["repetitions"][str(repetition)] + commands[f"rep{repetition}"] = client_command( + entry, + config, + study=item["study"], + anchor=float(item["selection"]["anchor"]), + output=session_root / f"rep{repetition}", + warmup=False, + ) + sessions.append( + { + "config": config["id"], + "mns": config["mns"], + "mbbt": config["mbbt"], + "port": entry["port"], + "repetition_order": config["repetition_order"], + "commands": { + role: shlex.join(command) for role, command in commands.items() + }, + } + ) + return { + "schema": "action-aware-constraint-pilot-dry-run-v0", + "status": "PASS", + "manifest": str(args.manifest), + "run_root": str(args.run_root), + "projected_h20_hours": remaining_projection( + manifest, completed_sessions=0 + ), + "hard_cap_h20_hours": manifest["budget"]["hard_cap_h20_hours"], + "sessions": sessions, + } + + +def load_state(path: Path, hard_cap: float) -> dict[str, Any]: + if path.exists(): + return json.loads(path.read_text(encoding="utf-8")) + return { + "schema": SCHEMA, + "status": "initialized", + "hard_cap_h20_hours": hard_cap, + "gpu_hours_total": 0.0, + "completed_sessions": 0, + "sessions": {}, + "failures": [], + "started_at": time.time(), + } + + +def append_echo(run_root: Path, line: str) -> None: + run_root.mkdir(parents=True, exist_ok=True) + with (run_root / "launch-echo.log").open("a", encoding="utf-8") as target: + target.write(line + "\n") + print(line, flush=True) + + +def start_server( + *, + args: argparse.Namespace, + config: Mapping[str, Any], + index: int, +) -> dict[str, Any]: + gpus = (0, 1, 2, 3) + session_root = args.run_root / "sessions" / str(config["id"]) + session_root.mkdir(parents=True, exist_ok=True) + port = 9050 + index + command = server_command(config, gpus=gpus, port=port) + with (session_root / "commands.log").open("a", encoding="utf-8") as log: + log.write(f"SERVER {shlex.join(command)}\n") + server_log = (session_root / "server.log").open("ab", buffering=0) + environment = os.environ.copy() + environment.update( + { + "CUDA_VISIBLE_DEVICES": "0,1,2,3", + "VLLM_OPPROF_DIR": str(session_root / "opprof"), + "OPPROF_PHASE6_MARKER": base.MARKER, + "AITUNER_ROOT": str(base.AITUNER), + "HF_HUB_OFFLINE": "1", + "TRANSFORMERS_OFFLINE": "1", + "PYTHONUNBUFFERED": "1", + } + ) + server = subprocess.Popen( + command, + cwd=base.SOURCE, + env=environment, + stdout=server_log, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + base.OWNED_PGIDS.add(server.pid) + return { + "cell": str(config["id"]), + "gpus": gpus, + "port": port, + "dir": session_root, + "server": server, + "server_handle": server_log, + "spawned_at": time.time(), + "results": [], + } + + +def validate_result( + result: Mapping[str, Any], + *, + config: Mapping[str, Any], + selection: Mapping[str, Any], + role: str, + warmup: bool, +) -> None: + if result.get("schema") != "action-aware-pilot-result-v0": + raise RuntimeError(f"unexpected result schema: {role}") + if result.get("config_id") != config["id"]: + raise RuntimeError(f"config id mismatch: {role}") + if int(result["tp"]) != 4: + raise RuntimeError(f"TP mismatch: {role}") + if int(result["mns"]) != int(config["mns"]): + raise RuntimeError(f"MNS mismatch: {role}") + if int(result["mbbt"]) != int(config["mbbt"]): + raise RuntimeError(f"MBBT mismatch: {role}") + if result.get("slo_early_stop_disabled") is not True: + raise RuntimeError(f"SLO early stop was not disabled: {role}") + if warmup: + if result["kind"] != "warmup" or int(result["selection"]["count"]) != 16: + raise RuntimeError(f"invalid warmup: {role}") + return + if bool(result["early_stopped"]): + raise RuntimeError(f"uncensored run early-stopped: {role}") + if int(result["selection"]["count"]) != int(selection["selected_count"]): + raise RuntimeError(f"selection count mismatch: {role}") + if int(result["observed_count"]) != int(selection["selected_count"]): + raise RuntimeError(f"request accounting mismatch: {role}") + for result_key, selection_key in ( + ("request_id_order_sha256", "request_id_order_sha256"), + ("arrival_order_sha256", "arrival_order_sha256"), + ("raw_length_order_sha256", "input_length_order_sha256"), + ): + if result["selection"][result_key] != selection[selection_key]: + raise RuntimeError(f"selection hash mismatch {result_key}: {role}") + + +def run_client( + *, + entry: dict[str, Any], + config: Mapping[str, Any], + role: str, + study: str, + selection: Mapping[str, Any], + output: Path, + state: Mapping[str, Any], + timeout_s: float, + warmup: bool = False, +) -> dict[str, Any]: + command = client_command( + entry, + config, + study=study, + anchor=float(selection["anchor"]), + output=output, + warmup=warmup, + ) + with (entry["dir"] / "commands.log").open("a", encoding="utf-8") as log: + log.write(f"CLIENT role={role} {shlex.join(command)}\n") + handle = (output.parent / f"{output.name}.log").open("ab", buffering=0) + environment = os.environ.copy() + environment.update({"AITUNER_ROOT": str(base.AITUNER), "PYTHONUNBUFFERED": "1"}) + process = subprocess.Popen( + command, + cwd=base.WORKDIR, + env=environment, + stdout=handle, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + deadline = time.monotonic() + timeout_s + try: + while process.poll() is None: + if time.monotonic() > deadline: + raise TimeoutError(f"client timeout: {config['id']} {role}") + if entry["server"].poll() is not None: + raise RuntimeError(f"server exited during {config['id']} {role}") + base.assert_no_other_compute() + if state["gpu_hours_total"] + base.live_gpu_hours([entry]) >= base.GPU_LIMIT: + raise RuntimeError("action-aware pilot H20-hour hard cap reached") + time.sleep(1.0) + except Exception: + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + pass + try: + process.wait(timeout=10.0) + except subprocess.TimeoutExpired: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + process.wait(timeout=10.0) + raise + finally: + handle.close() + if process.returncode: + raise RuntimeError( + f"client failed: config={config['id']} role={role} rc={process.returncode}" + ) + result = json.loads((output / "result.json").read_text(encoding="utf-8")) + validate_result( + result, + config=config, + selection=selection, + role=role, + warmup=warmup, + ) + entry["results"].append( + {"anchor": float(selection["anchor"]), "dir": str(output), "kind": result["kind"]} + ) + return result + + +def execute_session( + *, + args: argparse.Namespace, + manifest: Mapping[str, Any], + config: Mapping[str, Any], + index: int, + state: dict[str, Any], + state_path: Path, +) -> None: + name = str(config["id"]) + if state["sessions"].get(name, {}).get("status") == "complete": + return + projection = remaining_projection( + manifest, completed_sessions=int(state["completed_sessions"]) + ) + if float(state["gpu_hours_total"]) + projection > base.GPU_LIMIT: + raise RuntimeError(f"projected cost exceeds cap before {name}") + echo = ( + f"ACTION_AWARE_SESSION_ECHO host=dash0 config={name} tp=4 " + f"mns={config['mns']} mbbt={config['mbbt']} gpus=0-3 " + f"workload={manifest['source']['window_id']} load_per_gpu=2.125 " + f"duration_s=300 repetitions={','.join(map(str, config['repetition_order']))} " + f"source={args.manifest} output={args.run_root / 'sessions' / name} " + f"spent_h20h={state['gpu_hours_total']:.6f} " + f"remaining_projection_h20h={projection:.3f} cap_h20h={base.GPU_LIMIT:.1f}" + ) + append_echo(args.run_root, echo) + wait_all_idle() + session_state = { + "status": "starting", + "mns": int(config["mns"]), + "mbbt": int(config["mbbt"]), + "repetition_order": list(config["repetition_order"]), + "started_at": time.time(), + "runs": [], + } + state["status"] = "running" + state["sessions"][name] = session_state + atomic_json(state_path, state) + entry = start_server(args=args, config=config, index=index) + failure: Exception | None = None + try: + base.wait_ready(entry) + first = manifest["repetitions"][str(config["repetition_order"][0])] + session_state["status"] = "warmup" + atomic_json(state_path, state) + run_client( + entry=entry, + config=config, + role="warmup", + study=first["study"], + selection=first["selection"], + output=entry["dir"] / "warmup", + state=state, + timeout_s=180.0, + warmup=True, + ) + session_state["status"] = "burnin" + atomic_json(state_path, state) + burnin = manifest["burnin"] + run_client( + entry=entry, + config=config, + role="burnin", + study=burnin["study"], + selection=burnin, + output=entry["dir"] / "burnin", + state=state, + timeout_s=float(manifest["engine"]["client_timeout_s"]), + ) + session_state["status"] = "measured" + atomic_json(state_path, state) + for repetition in config["repetition_order"]: + item = manifest["repetitions"][str(repetition)] + role = f"rep{repetition}" + result = run_client( + entry=entry, + config=config, + role=role, + study=item["study"], + selection=item["selection"], + output=entry["dir"] / role, + state=state, + timeout_s=float(manifest["engine"]["client_timeout_s"]), + ) + session_state["runs"].append( + { + "repetition": int(repetition), + "pass_rate": result["pass_rate"], + "feasible": result["feasible"], + "slo_pass_count": result["slo_pass_count"], + "elapsed_s": result["interval"]["elapsed_s"], + } + ) + atomic_json(state_path, state) + session_state["status"] = "stopping" + atomic_json(state_path, state) + except Exception as error: # noqa: BLE001 + failure = error + finally: + try: + base.stop_entry(entry) + except Exception as error: # noqa: BLE001 + failure = failure or error + time.sleep(2.0) + try: + wait_all_idle() + except Exception as error: # noqa: BLE001 + failure = failure or error + + session_hours = base.live_gpu_hours([entry]) + state["gpu_hours_total"] += session_hours + session_state["gpu_hours"] = session_hours + if failure is not None: + session_state["status"] = "failed" + session_state["failure"] = repr(failure) + state["status"] = "failed" + state["failures"].append({"session": name, "failure": repr(failure)}) + atomic_json(state_path, state) + raise failure + validation = base.validate_cell(entry) + session_state["validation"] = validation + session_state["status"] = "complete" + session_state["completed_at"] = time.time() + state["completed_sessions"] += 1 + atomic_json(state_path, state) + + +def parser() -> argparse.ArgumentParser: + result = argparse.ArgumentParser() + result.add_argument("--manifest", type=Path, required=True) + result.add_argument("--run-root", type=Path, required=True) + result.add_argument("--aituner-root", type=Path, required=True) + result.add_argument("--vllm-source", type=Path, required=True) + result.add_argument("--venv", type=Path, required=True) + result.add_argument("--model", type=Path, required=True) + result.add_argument("--client", type=Path, required=True) + result.add_argument("--dry-run", action="store_true") + return result + + +def main() -> None: + args = parser().parse_args() + manifest = json.loads(args.manifest.read_text(encoding="utf-8")) + validate_inputs(args, manifest) + configure(args, manifest) + if args.dry_run: + print(json.dumps(dry_run_plan(args, manifest), indent=2, sort_keys=True)) + return + args.run_root.mkdir(parents=True, exist_ok=True) + copied_manifest = args.run_root / "pilot-manifest.json" + if not copied_manifest.exists(): + atomic_json(copied_manifest, manifest) + state_path = args.run_root / "controller-state.json" + state = load_state(state_path, base.GPU_LIMIT) + state["status"] = "running" + atomic_json(state_path, state) + for index, config in enumerate(manifest["configs"]): + execute_session( + args=args, + manifest=manifest, + config=config, + index=index, + state=state, + state_path=state_path, + ) + state["status"] = "complete" + state["completed_at"] = time.time() + atomic_json(state_path, state) + wait_all_idle() + print( + json.dumps( + { + "status": state["status"], + "completed_sessions": state["completed_sessions"], + "gpu_hours_total": state["gpu_hours_total"], + }, + sort_keys=True, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/runs/action-aware-v0/prepare_pilot.py b/runs/action-aware-v0/prepare_pilot.py new file mode 100644 index 0000000..51e9fd1 --- /dev/null +++ b/runs/action-aware-v0/prepare_pilot.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +"""Freeze the crossed-constraint action-aware development pilot.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +from pathlib import Path +from typing import Any + + +SCHEMA = "action-aware-constraint-pilot-manifest-v0" +CONFIGS = ( + {"id": "b_base", "mns": 64, "mbbt": 256, "repetition_order": [1, 2, 3]}, + {"id": "a_base", "mns": 16, "mbbt": 8192, "repetition_order": [2, 3, 1]}, + {"id": "shared", "mns": 64, "mbbt": 8192, "repetition_order": [3, 1, 2]}, + {"id": "b_mns", "mns": 128, "mbbt": 256, "repetition_order": [1, 3, 2]}, + {"id": "a_mbbt", "mns": 16, "mbbt": 16384, "repetition_order": [2, 1, 3]}, +) + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def atomic_json(path: Path, payload: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text( + json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + os.replace(temporary, path) + + +def build(base_path: Path) -> dict[str, Any]: + base = json.loads(base_path.read_text(encoding="utf-8")) + if base.get("schema") != "intervention-response-phase-aware-pilot-manifest-v3": + raise ValueError("unexpected base manifest schema") + if base.get("status") != "PASS": + raise ValueError("base manifest did not pass its preflight") + if sorted(int(key) for key in base["repetitions"]) != [1, 2, 3]: + raise ValueError("base manifest must contain exactly three repetitions") + + repetitions = {} + selection_hashes = [] + for repetition in (1, 2, 3): + source = base["repetitions"][str(repetition)] + selection = dict(source["selections"]["mid"]) + selection_hashes.append(selection["request_id_order_sha256"]) + repetitions[str(repetition)] = { + "study": source["study"], + "study_sha256": source["study_sha256"], + "selection": selection, + "merged_trace": source["merged_trace"], + } + + config_ids = [str(config["id"]) for config in CONFIGS] + payload = { + "schema": SCHEMA, + "status": "PASS", + "source": { + "base_manifest": str(base_path.resolve()), + "base_manifest_sha256": sha256_file(base_path), + "window_id": base["source"]["window_id"], + "source_trace": base["source"]["source_trace"], + "source_trace_sha256": base["source"]["source_trace_sha256"], + }, + "engine": { + "tp": 4, + "duration_s": 300.0, + "disable_slo_early_stop": True, + "client_timeout_s": 450.0, + }, + "burnin": base["burnin"], + "repetitions": repetitions, + "configs": [dict(config) for config in CONFIGS], + "regimes": { + "A": { + "source": "a_base", + "actions": {"mns": "shared", "mbbt": "a_mbbt"}, + }, + "B": { + "source": "b_base", + "actions": {"mns": "b_mns", "mbbt": "shared"}, + }, + }, + "budget": { + "hard_cap_h20_hours": 8.0, + "session_estimate_h20_hours": 1.35, + "safety_h20_hours": 0.25, + "expected_h20_hours": [6.0, 7.2], + "expected_wall_minutes": [90, 110], + }, + "gates": { + "minimum_relative_winner_margin": 0.10, + "minimum_exclusive_fraction": 0.10, + "minimum_exclusive_ratio": 5.0, + "phase_fractions": [0.25, 0.50, 0.75, 1.0], + "material_kv_usage": 0.90, + }, + "sanity": { + "invariants": { + "five_unique_configs": len(config_ids) == len(set(config_ids)) == 5, + "three_disjoint_repetitions": len(set(selection_hashes)) == 3, + "same_load_all_repetitions": len( + { + float(item["selection"]["offered_req_s_per_gpu"]) + for item in repetitions.values() + } + ) + == 1, + "all_repetition_orders_are_permutations": all( + sorted(config["repetition_order"]) == [1, 2, 3] + for config in CONFIGS + ), + } + }, + } + payload["sanity"]["invariants"]["shared_endpoint_reused_by_both_regimes"] = ( + payload["regimes"]["A"]["actions"]["mns"] + == payload["regimes"]["B"]["actions"]["mbbt"] + == "shared" + ) + payload["sanity"]["red_flags"] = [ + name + for name, passed in payload["sanity"]["invariants"].items() + if not passed + ] + if payload["sanity"]["red_flags"]: + payload["status"] = "FAIL" + return payload + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--base-manifest", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + payload = build(args.base_manifest) + atomic_json(args.output, payload) + print(json.dumps(payload["sanity"], sort_keys=True)) + if payload["status"] != "PASS": + raise SystemExit("manifest preflight failed") + + +if __name__ == "__main__": + main() diff --git a/runs/action-aware-v0/test_pilot.py b/runs/action-aware-v0/test_pilot.py new file mode 100644 index 0000000..5b2f3af --- /dev/null +++ b/runs/action-aware-v0/test_pilot.py @@ -0,0 +1,191 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import copy +import importlib.util +from pathlib import Path +from types import SimpleNamespace + + +HERE = Path(__file__).resolve().parent +ROOT = HERE.parents[1] + + +def load(name: str, filename: str): + spec = importlib.util.spec_from_file_location(name, HERE / filename) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +def record(*, waiting: int, running: int, tokens: int) -> dict: + return { + "queues": {"waiting": waiting, "deferred": 0, "running": running}, + "prefill_tokens": tokens, + "decode_tokens": 0, + "kv": {"usage": 0.5}, + "preemptions": 0, + } + + +def fake_run( + config: str, + repetition: int, + *, + goodput: float, + mns_score: float = 0.0, + mbbt_score: float = 0.0, + ambiguous: float = 0.0, +) -> dict: + binding = { + "mns_exclusive_fraction": mns_score, + "mbbt_exclusive_fraction": mbbt_score, + "both_fraction": ambiguous, + "waiting_unresolved_fraction": 0.0, + "kv_usage_max": 0.5, + "preemptions": 0, + } + phases = { + phase: { + "mns_exclusive_fraction": mns_score, + "mbbt_exclusive_fraction": mbbt_score, + } + for phase in ("0.25", "0.50", "0.75", "1.00") + } + return { + "config_id": config, + "repetition": repetition, + "outcome": {"slo_goodput_req_s": goodput}, + "binding": binding, + "phases": phases, + } + + +def main() -> None: + analysis = load("action_aware_analysis", "analyze_pilot.py") + summary = analysis.binding_summary( + [ + record(waiting=1, running=16, tokens=8), + record(waiting=1, running=8, tokens=32), + record(waiting=1, running=16, tokens=32), + record(waiting=1, running=8, tokens=8), + record(waiting=0, running=8, tokens=8), + ], + mns=16, + mbbt=32, + ) + assert summary["mns_exclusive_count"] == 1 + assert summary["mbbt_exclusive_count"] == 1 + assert summary["both_count"] == 1 + assert summary["waiting_unresolved_count"] == 1 + assert summary["waiting_count"] == 4 + + manifest = { + "repetitions": {str(index): {} for index in (1, 2, 3)}, + "regimes": { + "A": { + "source": "a_base", + "actions": {"mns": "shared", "mbbt": "a_mbbt"}, + }, + "B": { + "source": "b_base", + "actions": {"mns": "b_mns", "mbbt": "shared"}, + }, + }, + "gates": { + "minimum_relative_winner_margin": 0.10, + "minimum_exclusive_fraction": 0.10, + "minimum_exclusive_ratio": 5.0, + "material_kv_usage": 0.90, + }, + } + runs = [] + for repetition in (1, 2, 3): + runs.extend( + [ + fake_run( + "a_base", + repetition, + goodput=1.0, + mns_score=0.8, + mbbt_score=0.01, + ), + fake_run( + "b_base", + repetition, + goodput=1.0, + mns_score=0.01, + mbbt_score=0.7, + ), + fake_run("shared", repetition, goodput=3.0), + fake_run("a_mbbt", repetition, goodput=1.5), + fake_run("b_mns", repetition, goodput=1.2), + ] + ) + result = analysis.evaluate_decisions(runs, manifest) + assert result["decision"] == "STOP_NO_NEW_INSTRUMENTATION_NEEDED" + assert result["baselines"] == { + "always_mns_correct": 3, + "always_mbbt_correct": 3, + "binding_correct": 6, + "decision_count": 6, + } + + ambiguous = copy.deepcopy(runs) + for run in ambiguous: + if run["config_id"] == "b_base": + run["binding"]["both_fraction"] = 0.8 + assert ( + analysis.evaluate_decisions(ambiguous, manifest)["decision"] + == "OPEN_EXACT_ATTRIBUTION_ABLATION" + ) + + wrong = copy.deepcopy(runs) + for run in wrong: + if run["config_id"] == "b_base": + run["binding"]["mns_exclusive_fraction"] = 0.8 + run["binding"]["mbbt_exclusive_fraction"] = 0.01 + for phase in run["phases"].values(): + phase["mns_exclusive_fraction"] = 0.8 + phase["mbbt_exclusive_fraction"] = 0.01 + assert ( + analysis.evaluate_decisions(wrong, manifest)["decision"] + == "STOP_BINDING_NOT_PREDICTIVE" + ) + + prepare = load("action_aware_prepare", "prepare_pilot.py") + frozen = prepare.build( + ROOT / "runs/intervention-response-v2/pilot-manifest-v3.json" + ) + assert frozen["status"] == "PASS" + assert frozen["sanity"]["red_flags"] == [] + assert [config["id"] for config in frozen["configs"]] == [ + "b_base", + "a_base", + "shared", + "b_mns", + "a_mbbt", + ] + + controller = load("action_aware_controller", "pilot_controller.py") + args = SimpleNamespace( + manifest=Path("/tmp/manifest.json"), + run_root=Path("/tmp/action-aware"), + aituner_root=Path("/tmp/aituner"), + vllm_source=Path("/tmp/vllm"), + venv=Path("/tmp/venv"), + model=Path("/tmp/model"), + client=Path("/tmp/client.py"), + ) + controller.configure(args, frozen) + plan = controller.dry_run_plan(args, frozen) + assert plan["status"] == "PASS" + assert len(plan["sessions"]) == 5 + assert plan["projected_h20_hours"] == 7.0 + assert "--max-num-batched-tokens 256" in plan["sessions"][0]["commands"]["server"] + print("action-aware constraint pilot: PASS") + + +if __name__ == "__main__": + main()