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

Replace undrainable telemetry load with fresh rerun

Browse Source
This commit is contained in:
Gahow Wang
2026-07-14 17:52:21 +08:00
parent c0b40af24f
commit 7fd9563550
6 changed files with 150 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
runs/intervention-response-v2/analyze_pilot.py
View File

@@ -15,10 +15,10 @@ from typing import Any, Iterable, Mapping
HERE = Path(__file__).resolve().parent
P1_PATH = HERE.parent / "intervention-response-v0" / "analyze_p1.py"
SCHEMA = "intervention-response-phase-aware-pilot-analysis-v2"
EXPECTED_ACTION_PAIRS = 9
EXPECTED_REPEAT_PAIRS = 12
MIN_EFFICACY_CLASS = 3
SCHEMA = "intervention-response-phase-aware-pilot-analysis-v3"
EXPECTED_ACTION_PAIRS = 6
EXPECTED_REPEAT_PAIRS = 8
MIN_EFFICACY_CLASS = 2
MAX_LAYER1_GAP_S = 1.0
@@ -520,7 +520,7 @@ def cumulative_coverage_gate(windows: list[dict[str, Any]]) -> dict[str, Any]:
def audit(*, run_root: Path, manifest_path: Path, output_path: Path) -> dict[str, Any]:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
if manifest.get("schema") != "intervention-response-phase-aware-pilot-manifest-v2":
if manifest.get("schema") != "intervention-response-phase-aware-pilot-manifest-v3":
raise ValueError("unexpected phase-aware pilot manifest schema")
fractions = [float(value) for value in manifest["checkpoints"]["fractions"]]
seconds = [float(value) for value in manifest["checkpoints"]["seconds"]]

32
runs/intervention-response-v2/pilot_controller.py
View File

@@ -12,7 +12,7 @@ import subprocess
import sys
import time
from pathlib import Path
from typing import Any
from typing import Any, Mapping
HERE = Path(__file__).resolve().parent
@@ -22,8 +22,8 @@ sys.path.insert(0, str(PHASE6))
import opprof_phase6_controller as base # noqa: E402
SCHEMA = "intervention-response-phase-aware-pilot-state-v2"
SESSION_ESTIMATE_H20_HOURS = 1.25
SCHEMA = "intervention-response-phase-aware-pilot-state-v3"
SESSION_ESTIMATE_H20_HOURS = 1.0
SAFETY_H20_HOURS = 0.20
CLIENT_TIMEOUT_S = 450.0
@@ -55,7 +55,7 @@ def configure(args: argparse.Namespace, manifest: dict[str, Any]) -> None:
base.MODEL = args.model
base.CLIENT = args.client
base.GPU_LIMIT = float(manifest["budget"]["hard_cap_h20_hours"])
base.MARKER = "intervention-response-phase-aware-v2"
base.MARKER = "intervention-response-phase-aware-v3"
base.CELLS = {
f"tp4_mns{mns}": {"tp": 4, "mns": int(mns)}
for mns in manifest["engine"]["mns_endpoints"]
@@ -63,10 +63,15 @@ def configure(args: argparse.Namespace, manifest: dict[str, Any]) -> None:
def validate_inputs(args: argparse.Namespace, manifest: dict[str, Any]) -> None:
if manifest.get("schema") != "intervention-response-phase-aware-pilot-manifest-v2":
if manifest.get("schema") != "intervention-response-phase-aware-pilot-manifest-v3":
raise RuntimeError("unexpected phase-aware pilot manifest schema")
if manifest.get("status") != "PASS":
raise RuntimeError("phase-aware pilot manifest did not pass preflight")
if abs(
float(manifest["budget"]["session_estimate_h20_hours"])
- SESSION_ESTIMATE_H20_HOURS
) > 1e-12:
raise RuntimeError("controller and manifest session-cost estimates disagree")
failed_invariants = [
name
for name, passed in manifest.get("sanity", {}).get("invariants", {}).items()
@@ -95,6 +100,13 @@ def validate_inputs(args: argparse.Namespace, manifest: dict[str, Any]) -> None:
raise RuntimeError(f"phase-aware pilot input paths missing: {missing}")
def warmup_selection(repetition: Mapping[str, Any]) -> Mapping[str, Any]:
return max(
repetition["selections"].values(),
key=lambda selection: float(selection["offered_req_s_per_gpu"]),
)
def dry_run_plan(args: argparse.Namespace, manifest: dict[str, Any]) -> dict[str, Any]:
sessions = []
for index, session in enumerate(manifest["sessions"]):
@@ -102,13 +114,13 @@ def dry_run_plan(args: argparse.Namespace, manifest: dict[str, Any]) -> dict[str
entry = {"cell": cell, "gpus": (0, 1, 2, 3), "port": 8950 + index}
repetition = manifest["repetitions"][str(session["replicate"])]
session_root = args.run_root / "sessions" / str(session["session"])
high = repetition["selections"]["high"]
warmup = warmup_selection(repetition)
commands = {
"server": base.server_command(cell, entry["gpus"], entry["port"]),
"warmup": client_command(
entry,
study=repetition["study"],
anchor=float(high["anchor"]),
anchor=float(warmup["anchor"]),
output=session_root / "warmup",
warmup=True,
),
@@ -145,7 +157,7 @@ def dry_run_plan(args: argparse.Namespace, manifest: dict[str, Any]) -> dict[str
}
)
return {
"schema": "intervention-response-phase-aware-pilot-dry-run-v2",
"schema": "intervention-response-phase-aware-pilot-dry-run-v3",
"status": "PASS",
"manifest": str(args.manifest),
"run_root": str(args.run_root),
@@ -423,14 +435,14 @@ def execute_session(
failure: Exception | None = None
try:
base.wait_ready(entry)
high = repetition["selections"]["high"]
warmup = warmup_selection(repetition)
session_state["status"] = "warmup"
save_state(state_path, state)
run_client(
entry=entry,
role="warmup",
study=repetition["study"],
selection=high,
selection=warmup,
output=entry["dir"] / "warmup",
state=state,
warmup=True,

49
runs/intervention-response-v2/prepare_pilot.py
View File

@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Prepare the uncensored 300-second TP4 matched pilot on dash0."""
"""Prepare the uncensored 300-second TP4 two-load matched pilot on dash0."""
from __future__ import annotations
@@ -20,18 +20,18 @@ from aituner.spec import load_study_spec # noqa: E402
from aituner.trace import load_trace_requests, select_requests_for_threshold # noqa: E402
SCHEMA = "intervention-response-phase-aware-pilot-manifest-v2"
SCHEMA = "intervention-response-phase-aware-pilot-manifest-v3"
TP = 4
MNS_ENDPOINTS = (16, 64)
REPLAY_TIME_SCALE = 0.5
EXPECTED_DURATION_S = 300.0
SAFETY_DEADLINE_S = 360.0
LOADS_REQ_S_GPU = {"low": 1.5, "mid": 2.125, "high": 3.125}
LOADS_REQ_S_GPU = {"low": 1.5, "mid": 2.125}
REPLICATE_ROLE_PAIRS = (("low1", "high1"), ("low2", "high2"), ("low3", "high3"))
LOAD_ORDERS = {
1: ("low", "mid", "high"),
2: ("high", "low", "mid"),
3: ("mid", "high", "low"),
1: ("low", "mid"),
2: ("mid", "low"),
3: ("low", "mid"),
}
SESSION_ORDER = (
(1, 16),
@@ -118,7 +118,7 @@ def private_request_id(
*, source_sha256: str, line_number: int, original_id: str
) -> str:
payload = f"{source_sha256}:{line_number}:{original_id}".encode()
return f"phase-v2-{hashlib.sha256(payload).hexdigest()}"
return f"phase-v3-{hashlib.sha256(payload).hexdigest()}"
def merge_role_traces(sources: tuple[Path, Path], target: Path) -> dict[str, Any]:
@@ -159,7 +159,7 @@ def materialize_study(
source: Path, target: Path, *, replicate: int, trace_override: Path
) -> Path:
payload = json.loads(source.read_text(encoding="utf-8"))
payload["study_id"] = f"phase-aware-telemetry-v2-rep{replicate}"
payload["study_id"] = f"phase-aware-telemetry-v3-rep{replicate}"
payload["hardware"]["host_candidates"] = ["dash0"]
payload["engine"]["engine_version"] = "0.24.1.dev3+opprof"
trace = payload["trace"]
@@ -174,7 +174,7 @@ def materialize_study(
def build_manifest(
*, base_manifest_path: Path, private_root: Path
*, base_manifest_path: Path, private_root: Path, prior_attempt_state: Path
) -> dict[str, Any]:
base = json.loads(base_manifest_path.read_text(encoding="utf-8"))
if base.get("schema") != "fidelity-prefix-pilot-manifest-v1":
@@ -223,7 +223,7 @@ def build_manifest(
selections[level] = record
selected_ids_by_level[level] = {request.row_id for request in selected}
selection_hashes.append(record["request_id_order_sha256"])
selected_ids_by_replicate[replicate] = selected_ids_by_level["high"]
selected_ids_by_replicate[replicate] = selected_ids_by_level["mid"]
repetitions[str(replicate)] = {
"source_roles": list(roles),
"merged_trace": merged_traces[str(replicate)],
@@ -250,8 +250,8 @@ def build_manifest(
invariants = {
"three_repetitions": len(repetitions) == 3,
"six_sessions": len(sessions) == 6,
"load_levels_three": all(
len(item["selections"]) == 3 for item in repetitions.values()
"load_levels_two": all(
len(item["selections"]) == 2 for item in repetitions.values()
),
"selection_hashes_unique": len(selection_hashes) == len(set(selection_hashes)),
"selection_sets_disjoint_across_repetitions": all(
@@ -267,6 +267,13 @@ def build_manifest(
),
}
red_flags = [name for name, passed in invariants.items() if not passed]
prior_state = json.loads(prior_attempt_state.read_text(encoding="utf-8"))
if prior_state.get("status") != "failed":
raise ValueError("prior three-load attempt was not recorded as failed")
prior_h20_hours = float(prior_state["gpu_hours_total"])
incremental_cap_h20_hours = 8.0 - prior_h20_hours
if incremental_cap_h20_hours < 6.5:
raise ValueError("insufficient global H20-hour budget for the two-load rerun")
return {
"schema": SCHEMA,
"status": "PASS" if not red_flags else "STOP",
@@ -276,6 +283,10 @@ def build_manifest(
"window_id": base["source"]["window_id"],
"source_trace": base["source"]["trace"],
"source_trace_sha256": base["source"]["trace_sha256"],
"prior_attempt_state": str(prior_attempt_state),
"prior_attempt_state_sha256": sha256_file(prior_attempt_state),
"prior_attempt_h20_hours": prior_h20_hours,
"prior_attempt_failure": prior_state["failures"],
},
"engine": {
"tp": TP,
@@ -294,9 +305,12 @@ def build_manifest(
"seconds": [30.0, 75.0, 150.0, 225.0, 300.0],
},
"budget": {
"hard_cap_h20_hours": 8.0,
"expected_wall_minutes": [95, 120],
"expected_h20_hours": [6.3, 8.0],
"global_hard_cap_h20_hours": 8.0,
"prior_attempt_h20_hours": prior_h20_hours,
"hard_cap_h20_hours": incremental_cap_h20_hours,
"session_estimate_h20_hours": 1.0,
"expected_wall_minutes": [75, 95],
"expected_h20_hours": [4.8, incremental_cap_h20_hours],
},
"sanity": {
"red_flags": red_flags,
@@ -312,9 +326,12 @@ def main() -> None:
parser.add_argument("--base-manifest", type=Path, required=True)
parser.add_argument("--private-root", type=Path, required=True)
parser.add_argument("--public-manifest", type=Path, required=True)
parser.add_argument("--prior-attempt-state", type=Path, required=True)
args = parser.parse_args()
manifest = build_manifest(
base_manifest_path=args.base_manifest, private_root=args.private_root
base_manifest_path=args.base_manifest,
private_root=args.private_root,
prior_attempt_state=args.prior_attempt_state,
)
atomic_json(args.public_manifest, manifest)
print(

14
runs/intervention-response-v2/test_analysis.py
View File

@@ -95,8 +95,8 @@ def main() -> None:
source_sha256="b" * 64, line_number=1, original_id="1"
)
controller = load_controller_module()
assert math.isclose(controller.remaining_projection(6, 0), 7.7)
assert math.isclose(controller.remaining_projection(6, 5), 1.45)
assert math.isclose(controller.remaining_projection(6, 0), 6.2)
assert math.isclose(controller.remaining_projection(6, 5), 1.2)
parsed = controller.parser().parse_args(
[
"--manifest",
@@ -193,7 +193,15 @@ def main() -> None:
]
)
assert coverage_gate["red_flags"] == []
print("phase-aware intervention response v2 analysis: PASS")
assert controller.warmup_selection(
{
"selections": {
"low": {"offered_req_s_per_gpu": 1.5},
"mid": {"offered_req_s_per_gpu": 2.125},
}
}
)["offered_req_s_per_gpu"] == 2.125
print("phase-aware intervention response v3 analysis: PASS")
if __name__ == "__main__":