gahow/aituner
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add crossed-constraint action-aware pilot

Browse Source
This commit is contained in:
Gahow Wang
2026-07-14 20:26:54 +08:00
parent 26c2cdab2b
commit 823c550e53
7 changed files with 1988 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

179
docs/action-aware-constraint-pilot-v0-protocol-20260714.md Normal file
View File

@@ -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.

63
runs/action-aware-v0/action_aware_client.py Normal file
View File

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

584
runs/action-aware-v0/analyze_pilot.py Normal file
View File

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

223
runs/action-aware-v0/pilot-manifest.json Normal file
View File

@@ -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"
}

595
runs/action-aware-v0/pilot_controller.py Normal file
View File

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

153
runs/action-aware-v0/prepare_pilot.py Normal file
View File

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

191
runs/action-aware-v0/test_pilot.py Normal file
View File

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