From e5fd463f05e3351ad931ed2773289b46056d157f Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Wed, 15 Jul 2026 02:03:38 +0800 Subject: [PATCH] Add prospective active intervention experiment --- ...ctive-intervention-v0-protocol-20260715.md | 129 +++++ runs/action-aware-v0/pilot_controller.py | 14 +- .../analyze_prospective.py | 325 +++++++++++++ .../prepare_prospective.py | 362 ++++++++++++++ .../prospective_controller.py | 198 ++++++++ .../prospective_decision.py | 441 ++++++++++++++++++ runs/active-intervention-v0/test_pipeline.py | 7 +- .../test_prospective.py | 190 ++++++++ runs/active-intervention-v0/train_policy.py | 225 ++++++++- 9 files changed, 1875 insertions(+), 16 deletions(-) create mode 100644 docs/active-intervention-v0-protocol-20260715.md create mode 100644 runs/active-intervention-v0/analyze_prospective.py create mode 100644 runs/active-intervention-v0/prepare_prospective.py create mode 100644 runs/active-intervention-v0/prospective_controller.py create mode 100644 runs/active-intervention-v0/prospective_decision.py create mode 100644 runs/active-intervention-v0/test_prospective.py diff --git a/docs/active-intervention-v0-protocol-20260715.md b/docs/active-intervention-v0-protocol-20260715.md new file mode 100644 index 0000000..cf6a5e4 --- /dev/null +++ b/docs/active-intervention-v0-protocol-20260715.md @@ -0,0 +1,129 @@ +# Active intervention + measurement v0 protocol + +Date: 2026-07-15 (Asia/Singapore) + +Status: **FROZEN BEFORE THE `chat_w20260313_1000` GPU RUN**. + +## Research question + +This experiment asks whether a tuner conditioned on direct engine-state +trajectories can choose both a measurement horizon and a coupled configuration +intervention with lower real-GPU cost than the same tuner using only external +prefix outcomes. + +The contribution is not the controller, legality checks, telemetry collection, +or the ridge model. The route remains open only if engine state changes an +actual decision and reduces cost-to-near-oracle on unseen workloads. + +## Development result that motivates, but does not pass, the route + +The frozen trace-12 dataset contains 72 examples: six source decisions, four +measurement checkpoints, and `noop/MNS/MBBT` actions. Features are direct +continuous Layer-1 state summaries; cap-exclusive and bottleneck labels are +excluded. Leave-one-repetition-out sequential replay uses the same model, +candidate set, confidence rule, and checkpoint set for both modes. + +The external-outcome policy and telemetry policy both put all six decisions +within 2% regret. Outcome-only selected a mean 262.5-second source measurement +and cost 3.750 replay H20-hours across the six replayed decisions; telemetry +selected 275 seconds and cost 3.833 H20-hours. Telemetry therefore increased +the replay lower-bound cost by 2.22%, with no regret reduction. This is a +negative result. It does not settle the question because the dataset has only +two source regimes, one source is at the offered ceiling, and there is no joint +MNS+MBBT action. + +Sanity: n=6 decisions; regret min=0, max=0.009412, distinct=3; source cutoff +min=150s, max=300s, distinct=3 across the two policies; all costs are +non-negative, regrets are in `[0,1]`, target results are not all identical, and +the six decisions are complete exact-workload pairs. + +## Frozen prospective setup + +- Host: `dash0`, 8 NVIDIA H20 GPUs available; each TP4 server runs alone on + GPUs 0-3. Co-location is prohibited for SLO verdicts. +- Engine: patched vLLM `0.24.1.dev3+opprof` from + `/home/admin/cpfs/wjh/vllm-opprof-phase3` in + `/tmp/wjh/venvs/vllm-0.20.0-cu129`. +- Model: `/home/admin/cpfs/wjh/models/Qwen/Qwen3-30B-A3B`, BF16. +- Workload: unseen `chat_w20260313_1000`; input 0-8192; output exactly 128; + replay scale 0.5; 300-second arrival window. +- Three disjoint repetitions: source rows are assigned by a deterministic + SHA-256 modulo-3 partition before input filtering. Each repetition selects + approximately 3300 requests, or 2.75 requests/s/GPU at TP4. +- SLO: at least 95% pass; stepped TTFT 2/4/6 seconds; TPOT at most 50 ms. +- Checkpoints: 75, 150, 225, and 300 seconds. +- Full 2x2 surface: + - source: `MNS=32, MBBT=4096`; + - MNS action: `64,4096`; + - MBBT action: `32,8192`; + - joint action: `64,8192`; + - `noop` retains the source. +- Four config sessions are serialized. Each session uses a fresh server, + warm-up, burn-in, and counter-rotated repetition order. +- Expected campaign cost: 4.6-5.5 H20-hours; hard cap: 6.0 H20-hours; + expected wall time: 75-100 minutes. + +The source is executed first. The frozen telemetry policy selects the next +real config session; all remaining cells are then measured only to construct +the exact finite-surface oracle. Oracle annotation after the selected action +is reported separately from tuner cost. + +## Frozen policies + +Both policies fit the paired treatment effect + +```text +target normalized SLO-goodput - source normalized SLO-goodput +``` + +from source config, full config delta, offered load, and external prefix +outcomes. The telemetry policy additionally receives fixed direct Layer-1 +summaries and their interactions with `delta_log2(MNS)` and +`delta_log2(MBBT)`. It does not receive a bottleneck label or a +diagnosis-to-knob rule. + +At each checkpoint, jackknife models produce an effect distribution for +`noop`, MNS, MBBT, and joint actions. Measurement stops at the earliest second +consecutive checkpoint with the same confident best action; otherwise it uses +the full 300 seconds. Confidence requires a predicted margin of at least 0.02 +and the best lower bound to exceed the second-best upper bound. If the final +choice is not confident, the next run is the positive-UCB action, explicitly +marked as a diagnostic intervention. The exact same rule is used for the +outcome-only baseline. + +## Hypotheses and gates + +### H1: action value + +Engine state must change the selected intervention or its ranking and reduce +real action regret. Prediction error or bottleneck-label accuracy is not a +success metric. + +### H2: measurement value + +Engine state must select a shorter stable source measurement without increasing +action regret. A shorter reconstructed prefix is only a trigger; it is not an +actual GPU-cost claim until an early-terminated confirmation run measures +startup, warm-up, drain, and cleanup. + +### H3: end-to-end cost + +Primary development metric is H20-hours to first reach a configuration within +2% of the exact median-goodput oracle. The outcome-only and telemetry policies +use the same measured config costs and differ only in source information. + +- At least 10% prospective replay cost reduction, telemetry regret at most 2%, + and no outcome-only-to-telemetry harm triggers an actual early-stop + confirmation. +- At least 20% measured all-in H20-hour reduction is required for a contribution + claim. This one task can only establish development feasibility; a paper + claim additionally requires task-held-out replication. +- Source median normalized goodput at or above 0.98 stops the surface before + target runs because the workload has no material improvement headroom. +- Any hash mismatch, missing/censored result, telemetry drop, non-monotonic + phase, negative cost, ratio outside `[0,1]`, or all-identical config outcomes + is a red flag and stops analysis. + +If the 10% trigger fails, this route is closed for the current engine-state +representation. The experimental control plane is not retained as a fallback +research contribution. diff --git a/runs/action-aware-v0/pilot_controller.py b/runs/action-aware-v0/pilot_controller.py index 14668f9..7ecddb3 100644 --- a/runs/action-aware-v0/pilot_controller.py +++ b/runs/action-aware-v0/pilot_controller.py @@ -453,11 +453,21 @@ def execute_session( ) if float(state["gpu_hours_total"]) + projection > base.GPU_LIMIT: raise RuntimeError(f"projected cost exceeds cap before {name}") + load_values = { + float(item["selection"]["offered_req_s_per_gpu"]) + for item in manifest["repetitions"].values() + } + load_text = ( + f"{next(iter(load_values)):.6g}" + if len(load_values) == 1 + else ",".join(f"{value:.6g}" for value in sorted(load_values)) + ) 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"workload={manifest['source']['window_id']} load_per_gpu={load_text} " + f"duration_s={manifest['engine']['duration_s']} " + f"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}" diff --git a/runs/active-intervention-v0/analyze_prospective.py b/runs/active-intervention-v0/analyze_prospective.py new file mode 100644 index 0000000..3442769 --- /dev/null +++ b/runs/active-intervention-v0/analyze_prospective.py @@ -0,0 +1,325 @@ +#!/usr/bin/env python3 +"""Audit held-out action/measurement choices against the exact 2x2 surface.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import os +import statistics +from pathlib import Path +from typing import Any, Mapping + + +SCHEMA = "active-intervention-prospective-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: list[float]) -> dict[str, Any]: + finite = [float(value) for value in values] + if not finite or any(not math.isfinite(value) for value in finite): + raise ValueError("numeric summary requires finite values") + return { + "n": len(finite), + "min": min(finite), + "max": max(finite), + "distinct_n": len(set(finite)), + } + + +def load_surface( + manifest: Mapping[str, Any], run_root: Path +) -> tuple[dict[str, Any], list[dict[str, Any]]]: + rows = [] + aggregate = {} + duration_s = float(manifest["engine"]["duration_s"]) + tp = int(manifest["engine"]["tp"]) + for config in manifest["configs"]: + config_id = str(config["id"]) + values = [] + for repetition in sorted(int(key) for key in manifest["repetitions"]): + expected = manifest["repetitions"][str(repetition)]["selection"] + result_path = ( + run_root / "sessions" / config_id / f"rep{repetition}" / "result.json" + ) + result = json.loads(result_path.read_text(encoding="utf-8")) + if result["selection"]["request_id_order_sha256"] != expected[ + "request_id_order_sha256" + ]: + raise ValueError(f"request hash mismatch: {config_id} rep{repetition}") + offered_total = float(expected["offered_req_s_per_gpu"]) * tp + normalized = float(result["slo_pass_count"]) / duration_s / offered_total + values.append(normalized) + rows.append( + { + "config_id": config_id, + "mns": int(config["mns"]), + "mbbt": int(config["mbbt"]), + "repetition": repetition, + "normalized_slo_goodput": normalized, + "slo_goodput_req_s": float(result["slo_pass_count"]) / duration_s, + "pass_rate": float(result["pass_rate"]), + "elapsed_s": float(result["interval"]["elapsed_s"]), + "result": str(result_path), + "result_sha256": sha256_file(result_path), + } + ) + aggregate[config_id] = { + "normalized_slo_goodput_values": values, + "median_normalized_slo_goodput": float(statistics.median(values)), + "sanity": numeric(values), + } + return aggregate, rows + + +def source_cost_estimate( + *, + source_session: Mapping[str, Any], + source_rows: list[Mapping[str, Any]], + cutoff_s: float, + tp: int, +) -> dict[str, float]: + actual_h20_hours = float(source_session["gpu_hours"]) + measured_replay_h20_hours = ( + tp * sum(float(row["elapsed_s"]) for row in source_rows) / 3600.0 + ) + fixed_h20_hours = max(0.0, actual_h20_hours - measured_replay_h20_hours) + prefix_replay_h20_hours = tp * len(source_rows) * cutoff_s / 3600.0 + return { + "actual_full_session_h20_hours": actual_h20_hours, + "fixed_startup_warmup_burnin_cleanup_h20_hours": fixed_h20_hours, + "prefix_replay_h20_hours_lower_bound": prefix_replay_h20_hours, + "counterfactual_all_in_h20_hours_lower_bound": fixed_h20_hours + + prefix_replay_h20_hours, + } + + +def replay_policy( + *, + mode: str, + manifest: Mapping[str, Any], + decision: Mapping[str, Any], + surface: Mapping[str, Any], + session_costs: Mapping[str, float], + source_cost: Mapping[str, float], +) -> dict[str, Any]: + acceptable_regret = float(manifest["gates"]["acceptable_regret"]) + source_id = str(manifest["source_config_id"]) + oracle = max( + float(item["median_normalized_slo_goodput"]) for item in surface.values() + ) + cumulative = float(source_cost["counterfactual_all_in_h20_hours_lower_bound"]) + source_score = float(surface[source_id]["median_normalized_slo_goodput"]) + source_regret = 1.0 - source_score / oracle if oracle > 0 else 0.0 + points = [ + { + "action_id": "noop", + "config_id": source_id, + "score": source_score, + "regret": source_regret, + "cumulative_h20_hours_lower_bound": cumulative, + } + ] + hit = points[0] if source_regret <= acceptable_regret + 1e-12 else None + seen = {source_id} + for action_id in decision["decisions"][mode]["intervention_order"]: + config_id = str(manifest["actions"][action_id]) + if config_id in seen: + continue + seen.add(config_id) + cumulative += float(session_costs[config_id]) + score = float(surface[config_id]["median_normalized_slo_goodput"]) + regret = 1.0 - score / oracle if oracle > 0 else 0.0 + point = { + "action_id": action_id, + "config_id": config_id, + "score": score, + "regret": regret, + "cumulative_h20_hours_lower_bound": cumulative, + } + points.append(point) + if hit is None and regret <= acceptable_regret + 1e-12: + hit = point + return { + "mode": mode, + "measurement_cutoff_s": float( + decision["decisions"][mode]["selected_cutoff_s"] + ), + "selected_action": decision["decisions"][mode]["selected_action"], + "decision_kind": decision["decisions"][mode]["decision_kind"], + "intervention_order": decision["decisions"][mode]["intervention_order"], + "source_cost": dict(source_cost), + "oracle_normalized_slo_goodput": oracle, + "cost_to_acceptable": hit, + "reached_acceptable": hit is not None, + "points": points, + } + + +def build_audit( + *, manifest_path: Path, decision_path: Path, run_root: Path +) -> dict[str, Any]: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + decision = json.loads(decision_path.read_text(encoding="utf-8")) + state_path = run_root / "controller-state.json" + state = json.loads(state_path.read_text(encoding="utf-8")) + if manifest.get("schema") != "active-intervention-prospective-manifest-v0": + raise ValueError("unexpected prospective manifest schema") + if decision.get("schema") != "active-intervention-prospective-decision-v0": + raise ValueError("unexpected prospective decision schema") + if decision["manifest_sha256"] != sha256_file(manifest_path): + raise ValueError("decision does not match prospective manifest") + surface, rows = load_surface(manifest, run_root) + source_id = str(manifest["source_config_id"]) + sessions = state["sessions"] + session_costs = { + config_id: float(sessions[config_id]["gpu_hours"]) + for config_id in surface + } + source_rows = [row for row in rows if row["config_id"] == source_id] + policies = {} + for mode in ("outcome_only", "telemetry"): + cost = source_cost_estimate( + source_session=sessions[source_id], + source_rows=source_rows, + cutoff_s=float(decision["decisions"][mode]["selected_cutoff_s"]), + tp=int(manifest["engine"]["tp"]), + ) + policies[mode] = replay_policy( + mode=mode, + manifest=manifest, + decision=decision, + surface=surface, + session_costs=session_costs, + source_cost=cost, + ) + outcome_hit = policies["outcome_only"]["cost_to_acceptable"] + telemetry_hit = policies["telemetry"]["cost_to_acceptable"] + if outcome_hit is None or telemetry_hit is None: + reduction = None + else: + outcome_cost = float(outcome_hit["cumulative_h20_hours_lower_bound"]) + telemetry_cost = float(telemetry_hit["cumulative_h20_hours_lower_bound"]) + reduction = 1.0 - telemetry_cost / outcome_cost if outcome_cost > 0 else 0.0 + confirmation_trigger = bool( + reduction is not None + and reduction + >= float(manifest["gates"]["confirmation_trigger_gpu_cost_reduction"]) + and policies["telemetry"]["reached_acceptable"] + ) + contribution_gate = bool( + reduction is not None + and reduction >= float(manifest["gates"]["contribution_gpu_cost_reduction"]) + and policies["telemetry"]["reached_acceptable"] + ) + status = ( + "TRIGGER_ACTUAL_EARLY_STOP_CONFIRMATION" + if confirmation_trigger + else "STOP_NO_PROSPECTIVE_GPU_COST_SIGNAL" + ) + normalized_values = [float(row["normalized_slo_goodput"]) for row in rows] + costs = list(session_costs.values()) + invariants = { + "controller_complete": state.get("status") == "complete", + "four_sessions_complete": len(sessions) == 4 + and all(item.get("status") == "complete" for item in sessions.values()), + "twelve_surface_outcomes": len(rows) == 12, + "nonnegative_goodput": all(value >= 0.0 for value in normalized_values), + "normalized_goodput_bounded": all(value <= 1.0 + 1e-12 for value in normalized_values), + "surface_not_all_identical": len(set(normalized_values)) > 1, + "nonnegative_session_costs": all(value >= 0.0 for value in costs), + "policy_replay_reaches_oracle_surface": all( + policy["reached_acceptable"] for policy in policies.values() + ), + } + red_flags = [name for name, passed in invariants.items() if not passed] + if red_flags: + status = "STOP_SANITY" + return { + "schema": SCHEMA, + "status": status, + "claim_boundary": ( + "Prospective exact-surface replay. Prefix source costs reconstruct the " + "measured fixed overhead plus selected replay seconds; actual early-stop " + "confirmation is required before claiming GPU-cost reduction." + ), + "manifest": str(manifest_path), + "manifest_sha256": sha256_file(manifest_path), + "decision": str(decision_path), + "decision_sha256": sha256_file(decision_path), + "controller_state": str(state_path), + "controller_state_sha256": sha256_file(state_path), + "surface": surface, + "rows": rows, + "session_costs_h20_hours": session_costs, + "annotation_campaign_h20_hours": float(state["gpu_hours_total"]), + "policies": policies, + "comparison": { + "telemetry_gpu_cost_reduction_fraction": reduction, + "confirmation_trigger": confirmation_trigger, + "contribution_gate": contribution_gate, + "confirmation_trigger_threshold": manifest["gates"][ + "confirmation_trigger_gpu_cost_reduction" + ], + "contribution_threshold": manifest["gates"][ + "contribution_gpu_cost_reduction" + ], + "action_changed": policies["outcome_only"]["selected_action"] + != policies["telemetry"]["selected_action"], + "measurement_changed": policies["outcome_only"]["measurement_cutoff_s"] + != policies["telemetry"]["measurement_cutoff_s"], + }, + "sanity": { + "invariants": invariants, + "red_flags": red_flags, + "normalized_slo_goodput": numeric(normalized_values), + "session_h20_hours": numeric(costs), + }, + } + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--manifest", type=Path, required=True) + parser.add_argument("--decision", type=Path, required=True) + parser.add_argument("--run-root", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + audit = build_audit( + manifest_path=args.manifest, + decision_path=args.decision, + run_root=args.run_root, + ) + atomic_json(args.output, audit) + print( + json.dumps( + { + "status": audit["status"], + "comparison": audit["comparison"], + "sanity": audit["sanity"], + }, + sort_keys=True, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/runs/active-intervention-v0/prepare_prospective.py b/runs/active-intervention-v0/prepare_prospective.py new file mode 100644 index 0000000..0268537 --- /dev/null +++ b/runs/active-intervention-v0/prepare_prospective.py @@ -0,0 +1,362 @@ +#!/usr/bin/env python3 +"""Freeze the unseen-trace 2x2 active intervention development surface.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import os +import sys +from pathlib import Path +from typing import Any + + +AITUNER_ROOT = Path(os.environ.get("AITUNER_ROOT", Path(__file__).resolve().parents[2])) +sys.path.insert(0, str(AITUNER_ROOT / "src")) + +from aituner.spec import load_study_spec # noqa: E402 +from aituner.trace import load_trace_requests, select_requests_for_threshold # noqa: E402 + + +SCHEMA = "active-intervention-prospective-manifest-v0" +TP = 4 +REPETITIONS = (1, 2, 3) +DURATION_S = 300.0 +REPLAY_TIME_SCALE = 0.5 +OFFERED_RATE_PER_GPU = 2.75 +TARGET_COUNT = round(OFFERED_RATE_PER_GPU * DURATION_S * TP) +WINDOW_ID = "chat_w20260313_1000" + + +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 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 order_hash(values: list[str]) -> str: + return hashlib.sha256("\n".join(values).encode()).hexdigest() + + +def configs() -> list[dict[str, Any]]: + return [ + { + "id": "source_mns32_mbbt4096", + "mns": 32, + "mbbt": 4096, + "repetition_order": [1, 2, 3], + }, + { + "id": "mns64_mbbt4096", + "mns": 64, + "mbbt": 4096, + "repetition_order": [2, 3, 1], + }, + { + "id": "mns32_mbbt8192", + "mns": 32, + "mbbt": 8192, + "repetition_order": [3, 1, 2], + }, + { + "id": "joint_mns64_mbbt8192", + "mns": 64, + "mbbt": 8192, + "repetition_order": [1, 3, 2], + }, + ] + + +def partition_trace(source: Path, output_root: Path) -> dict[str, Any]: + source_sha = sha256_file(source) + output_root.mkdir(parents=True, exist_ok=True) + paths = {rep: output_root / f"rep{rep}.jsonl" for rep in REPETITIONS} + temporary = {rep: path.with_suffix(".jsonl.tmp") for rep, path in paths.items()} + handles = {rep: temporary[rep].open("w", encoding="utf-8") for rep in REPETITIONS} + counts = {rep: 0 for rep in REPETITIONS} + id_digests = {rep: hashlib.sha256() for rep in REPETITIONS} + total = 0 + try: + with source.open(encoding="utf-8") as input_file: + for line_number, line in enumerate(input_file, start=1): + if not line.strip(): + continue + row = json.loads(line) + original_id = str(row.get("request_id") or row.get("id") or line_number) + digest = hashlib.sha256( + f"{source_sha}:{line_number}:{original_id}".encode() + ).hexdigest() + repetition = int(digest[:16], 16) % len(REPETITIONS) + 1 + row["request_id"] = f"active-r{repetition}-{digest}" + handles[repetition].write(json.dumps(row, ensure_ascii=False) + "\n") + counts[repetition] += 1 + total += 1 + id_digests[repetition].update(row["request_id"].encode() + b"\n") + finally: + for handle in handles.values(): + handle.close() + for repetition in REPETITIONS: + os.replace(temporary[repetition], paths[repetition]) + partitions = { + str(rep): { + "path": str(paths[rep]), + "rows": counts[rep], + "bytes": paths[rep].stat().st_size, + "sha256": sha256_file(paths[rep]), + "request_id_order_sha256": id_digests[rep].hexdigest(), + } + for rep in REPETITIONS + } + return { + "source": str(source), + "source_sha256": source_sha, + "source_rows": total, + "partition_rule": "sha256(source_sha:line_number:original_id) modulo 3", + "partitions": partitions, + } + + +def materialize_study( + base_study: Path, + target: Path, + *, + repetition: int, + trace_path: Path, + windows_path: Path, +) -> None: + payload = json.loads(base_study.read_text(encoding="utf-8")) + payload["study_id"] = f"active-intervention-trace13-rep{repetition}" + payload["hardware"]["host_candidates"] = ["dash0"] + payload["engine"]["engine_version"] = "0.24.1.dev3+opprof" + trace = payload["trace"] + trace.update( + { + "windows_path": str(windows_path), + "window_id": WINDOW_ID, + "trace_file_override": str(trace_path), + "completion_tokens_override": 128, + "replay_time_scale": REPLAY_TIME_SCALE, + "early_stop_max_lag_s": None, + "early_stop_max_elapsed_s": 360.0, + "restart_engine_after_early_stop": False, + "adaptive_stop": {"enabled": False}, + } + ) + atomic_json(target, payload) + + +def attainable_anchor(requests: list[Any], target_count: int) -> tuple[float, list[Any]]: + ordered = sorted(float(request.sampling_u) for request in requests) + if target_count <= 0 or target_count > len(ordered): + raise ValueError( + f"target count {target_count} is outside available range 1..{len(ordered)}" + ) + candidates = [] + for index in sorted({target_count - 1, min(target_count, len(ordered) - 1)}): + anchor = ordered[index] + selected = select_requests_for_threshold(requests, threshold=anchor) + candidates.append((abs(len(selected) - target_count), len(selected), anchor, selected)) + _error, _count, anchor, selected = min( + candidates, key=lambda item: (item[0], item[1], item[2]) + ) + return anchor, selected + + +def selection_record(selected: list[Any]) -> dict[str, Any]: + return { + "anchor": max(float(request.sampling_u) for request in selected), + "selected_count": len(selected), + "target_count": TARGET_COUNT, + "offered_req_s": len(selected) / DURATION_S, + "offered_req_s_per_gpu": len(selected) / DURATION_S / TP, + "request_id_order_sha256": order_hash([request.row_id for request in selected]), + "arrival_order_sha256": order_hash( + [f"{request.arrival_s:.12f}" for request in selected] + ), + "input_length_order_sha256": order_hash( + [str(request.prompt_tokens_hint) for request in selected] + ), + } + + +def build( + *, + base_study: Path, + base_action_manifest: Path, + source_trace: Path, + windows_path: Path, + private_root: Path, + policy_path: Path, +) -> dict[str, Any]: + base_manifest = json.loads(base_action_manifest.read_text(encoding="utf-8")) + if base_manifest.get("status") != "PASS": + raise ValueError("base action-aware manifest did not pass") + policy = json.loads(policy_path.read_text(encoding="utf-8")) + if policy.get("schema") != "active-intervention-policy-v0": + raise ValueError("unexpected frozen policy schema") + if policy.get("sanity", {}).get("red_flags"): + raise ValueError("frozen policy contains red flags") + + partition = partition_trace(source_trace, private_root / "traces") + repetitions = {} + selected_sets: list[set[str]] = [] + for repetition in REPETITIONS: + trace_path = Path(partition["partitions"][str(repetition)]["path"]) + study_path = private_root / "studies" / f"rep{repetition}-tp4.json" + materialize_study( + base_study, + study_path, + repetition=repetition, + trace_path=trace_path, + windows_path=windows_path, + ) + study = load_study_spec(study_path) + window, requests = load_trace_requests(study, study_spec_path=study_path) + duration_s = float(window.window_end - window.window_start) + if not math.isclose(duration_s, DURATION_S, abs_tol=1e-9): + raise ValueError(f"rep{repetition}: duration {duration_s} != {DURATION_S}") + _anchor, selected = attainable_anchor(requests, TARGET_COUNT) + record = selection_record(selected) + selected_sets.append({request.row_id for request in selected}) + repetitions[str(repetition)] = { + "study": str(study_path), + "study_sha256": sha256_file(study_path), + "trace": partition["partitions"][str(repetition)], + "available_filtered_requests": len(requests), + "selection": record, + } + + frozen_configs = configs() + config_ids = {str(config["id"]) for config in frozen_configs} + invariants = { + "three_nonempty_trace_partitions": all( + int(item["rows"]) > 0 for item in partition["partitions"].values() + ), + "partition_rows_conserved": sum( + int(item["rows"]) for item in partition["partitions"].values() + ) + == int(partition["source_rows"]), + "selected_sets_disjoint": all( + not selected_sets[left] & selected_sets[right] + for left in range(len(selected_sets)) + for right in range(left + 1, len(selected_sets)) + ), + "target_count_attained": all( + abs(int(item["selection"]["selected_count"]) - TARGET_COUNT) <= 1 + for item in repetitions.values() + ), + "four_unique_configs": len(config_ids) == 4, + "two_by_two_surface": { + (int(config["mns"]), int(config["mbbt"])) + for config in frozen_configs + } + == {(32, 4096), (64, 4096), (32, 8192), (64, 8192)}, + "repetition_orders_are_permutations": all( + sorted(config["repetition_order"]) == list(REPETITIONS) + for config in frozen_configs + ), + } + red_flags = [name for name, passed in invariants.items() if not passed] + return { + "schema": SCHEMA, + "status": "PASS" if not red_flags else "STOP", + "source": { + "window_id": WINDOW_ID, + "source_trace": str(source_trace), + "source_trace_sha256": partition["source_sha256"], + "windows_path": str(windows_path), + "base_study": str(base_study), + "base_study_sha256": sha256_file(base_study), + "base_action_manifest": str(base_action_manifest), + "base_action_manifest_sha256": sha256_file(base_action_manifest), + }, + "policy": { + "path": str(policy_path), + "sha256": sha256_file(policy_path), + "status": policy["status"], + "training": policy["training"], + "measurement_policy": policy["measurement_policy"], + "launch_reason": ( + "bounded unseen-trace joint-action test after a negative narrow " + "retrospective replay" + ), + }, + "engine": { + "tp": TP, + "duration_s": DURATION_S, + "client_timeout_s": 450.0, + "burnin_max_elapsed_s": 90.0, + "disable_slo_early_stop": True, + }, + "burnin": base_manifest["burnin"], + "private": {"trace_partition": partition}, + "repetitions": repetitions, + "configs": frozen_configs, + "source_config_id": "source_mns32_mbbt4096", + "actions": { + "noop": "source_mns32_mbbt4096", + "mns": "mns64_mbbt4096", + "mbbt": "mns32_mbbt8192", + "joint": "joint_mns64_mbbt8192", + }, + "checkpoints": { + "fractions": [0.25, 0.50, 0.75, 1.0], + "seconds": [75.0, 150.0, 225.0, 300.0], + }, + "gates": { + "acceptable_regret": 0.02, + "source_ceiling_normalized_goodput": 0.98, + "confirmation_trigger_gpu_cost_reduction": 0.10, + "contribution_gpu_cost_reduction": 0.20, + "maximum_task_regret": 0.05, + }, + "budget": { + "hard_cap_h20_hours": 6.0, + "session_estimate_h20_hours": 1.3, + "safety_h20_hours": 0.3, + "expected_h20_hours": [4.6, 5.5], + "expected_wall_minutes": [75, 100], + }, + "sanity": {"invariants": invariants, "red_flags": red_flags}, + } + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--base-study", type=Path, required=True) + parser.add_argument("--base-action-manifest", type=Path, required=True) + parser.add_argument("--source-trace", type=Path, required=True) + parser.add_argument("--windows-path", type=Path, required=True) + parser.add_argument("--private-root", type=Path, required=True) + parser.add_argument("--policy", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + payload = build( + base_study=args.base_study, + base_action_manifest=args.base_action_manifest, + source_trace=args.source_trace, + windows_path=args.windows_path, + private_root=args.private_root, + policy_path=args.policy, + ) + atomic_json(args.output, payload) + print(json.dumps({"status": payload["status"], "sanity": payload["sanity"]})) + if payload["status"] != "PASS": + raise SystemExit("prospective manifest preflight failed") + + +if __name__ == "__main__": + main() diff --git a/runs/active-intervention-v0/prospective_controller.py b/runs/active-intervention-v0/prospective_controller.py new file mode 100644 index 0000000..086ea93 --- /dev/null +++ b/runs/active-intervention-v0/prospective_controller.py @@ -0,0 +1,198 @@ +#!/usr/bin/env python3 +"""Run source first, select the next intervention, then annotate the 2x2 surface.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any, Mapping + + +HERE = Path(__file__).resolve().parent +ACTION_DIR = HERE.parent / "action-aware-v0" +sys.path.insert(0, str(ACTION_DIR)) +sys.path.insert(0, str(HERE)) + +import pilot_controller as action_controller # noqa: E402 +import prospective_decision # noqa: E402 + + +SCHEMA = "active-intervention-prospective-state-v0" + + +def validate_inputs(args: argparse.Namespace, manifest: Mapping[str, Any]) -> None: + if manifest.get("schema") != "active-intervention-prospective-manifest-v0": + raise RuntimeError("unexpected active intervention manifest schema") + if manifest.get("status") != "PASS" or manifest["sanity"]["red_flags"]: + raise RuntimeError("active intervention manifest did not pass preflight") + required = { + "manifest": args.manifest, + "policy": args.policy, + "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["trace"]["path"]) + missing = {name: str(path) for name, path in required.items() if not path.exists()} + if missing: + raise RuntimeError(f"active intervention input paths missing: {missing}") + if prospective_decision.sha256_file(args.policy) != manifest["policy"]["sha256"]: + raise RuntimeError("active intervention policy hash mismatch") + + +def dry_run(args: argparse.Namespace, manifest: Mapping[str, Any]) -> dict[str, Any]: + plan = action_controller.dry_run_plan(args, manifest) + return { + "schema": "active-intervention-prospective-dry-run-v0", + "status": "PASS", + "manifest": str(args.manifest), + "policy": str(args.policy), + "source_first": manifest["source_config_id"], + "post_source_order": "selected by telemetry policy; all remaining cells then annotated", + "candidate_actions": manifest["actions"], + "projected_h20_hours": plan["projected_h20_hours"], + "hard_cap_h20_hours": plan["hard_cap_h20_hours"], + "sessions": plan["sessions"], + } + + +def load_or_build_decision( + *, args: argparse.Namespace, run_root: Path +) -> dict[str, Any]: + path = run_root / "active-decision.json" + if path.exists(): + decision = json.loads(path.read_text(encoding="utf-8")) + if decision.get("manifest_sha256") != prospective_decision.sha256_file( + args.manifest + ): + raise RuntimeError("existing active decision has a different manifest") + if decision.get("policy_sha256") != prospective_decision.sha256_file(args.policy): + raise RuntimeError("existing active decision has a different policy") + return decision + decision = prospective_decision.build_decision( + manifest_path=args.manifest, + policy_path=args.policy, + run_root=run_root, + ) + prospective_decision.atomic_json(path, decision) + return decision + + +def parser() -> argparse.ArgumentParser: + result = argparse.ArgumentParser() + result.add_argument("--manifest", type=Path, required=True) + result.add_argument("--policy", 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) + action_controller.configure(args, manifest) + action_controller.base.MARKER = "active-intervention-prospective-v0" + if args.dry_run: + print(json.dumps(dry_run(args, manifest), indent=2, sort_keys=True)) + return + + args.run_root.mkdir(parents=True, exist_ok=True) + copied_manifest = args.run_root / "prospective-manifest.json" + if not copied_manifest.exists(): + action_controller.atomic_json(copied_manifest, manifest) + state_path = args.run_root / "controller-state.json" + state = action_controller.load_state( + state_path, float(manifest["budget"]["hard_cap_h20_hours"]) + ) + state["schema"] = SCHEMA + state["status"] = "running" + action_controller.atomic_json(state_path, state) + + configs = {str(item["id"]): dict(item) for item in manifest["configs"]} + config_indexes = { + str(item["id"]): index for index, item in enumerate(manifest["configs"]) + } + source_id = str(manifest["source_config_id"]) + action_controller.execute_session( + args=args, + manifest=manifest, + config=configs[source_id], + index=config_indexes[source_id], + state=state, + state_path=state_path, + ) + decision = load_or_build_decision(args=args, run_root=args.run_root) + state["active_decision"] = { + "path": str(args.run_root / "active-decision.json"), + "status": decision["status"], + "outcome_only": { + key: decision["decisions"]["outcome_only"][key] + for key in ("selected_cutoff_s", "decision_kind", "selected_action") + }, + "telemetry": { + key: decision["decisions"]["telemetry"][key] + for key in ("selected_cutoff_s", "decision_kind", "selected_action") + }, + } + action_controller.atomic_json(state_path, state) + if decision["status"] != "SELECTED": + state["status"] = decision["status"].lower() + state["completed_at"] = action_controller.time.time() + action_controller.atomic_json(state_path, state) + action_controller.wait_all_idle() + print(json.dumps({"status": state["status"], "decision": decision["status"]})) + return + + action_order = decision["decisions"]["telemetry"]["intervention_order"] + execution_order = [source_id] + for action_id in action_order: + target_id = str(manifest["actions"][action_id]) + if target_id not in execution_order: + execution_order.append(target_id) + for config_id in configs: + if config_id not in execution_order: + execution_order.append(config_id) + state["execution_order"] = execution_order + action_controller.atomic_json(state_path, state) + for config_id in execution_order[1:]: + action_controller.execute_session( + args=args, + manifest=manifest, + config=configs[config_id], + index=config_indexes[config_id], + state=state, + state_path=state_path, + ) + state["status"] = "complete" + state["completed_at"] = action_controller.time.time() + action_controller.atomic_json(state_path, state) + action_controller.wait_all_idle() + print( + json.dumps( + { + "status": state["status"], + "completed_sessions": state["completed_sessions"], + "gpu_hours_total": state["gpu_hours_total"], + "execution_order": execution_order, + }, + sort_keys=True, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/runs/active-intervention-v0/prospective_decision.py b/runs/active-intervention-v0/prospective_decision.py new file mode 100644 index 0000000..7b489d7 --- /dev/null +++ b/runs/active-intervention-v0/prospective_decision.py @@ -0,0 +1,441 @@ +#!/usr/bin/env python3 +"""Choose measurement horizon and next intervention from a completed source run.""" + +from __future__ import annotations + +import argparse +import hashlib +import importlib.util +import json +import math +import os +import statistics +import sys +from pathlib import Path +from typing import Any, Mapping, Sequence + +import numpy as np + + +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 = "active-intervention-prospective-decision-v0" + + +def load_module(name: str, path: Path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +MODEL = load_module("active_intervention_prospective_model", HERE / "model.py") +EXTRACT = load_module( + "active_intervention_prospective_extract", HERE / "extract_training.py" +) + + +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: Sequence[float]) -> dict[str, Any]: + finite = [float(value) for value in values] + if not finite or any(not math.isfinite(value) for value in finite): + raise ValueError("numeric summary requires finite values") + return { + "n": len(finite), + "min": min(finite), + "max": max(finite), + "distinct_n": len(set(finite)), + } + + +def load_engine_records(source_root: Path) -> tuple[list[dict[str, Any]], Path]: + streams = sorted((source_root / "opprof").glob("*.jsonl")) + if len(streams) != 1: + raise ValueError(f"expected one source engine stream, found {len(streams)}") + records = [ + row for row in EXTRACT.load_jsonl(streams[0]) if "step_index" in row + ] + if not records: + raise ValueError("source engine stream has no Layer-1 records") + return records, streams[0] + + +def candidate_example( + *, + source_config: Mapping[str, Any], + target_config: Mapping[str, Any], + action_id: str, + offered_rate_per_gpu: float, + outcome: Mapping[str, float], + telemetry: Mapping[str, float], +) -> dict[str, Any]: + return { + "source": { + "mns": int(source_config["mns"]), + "mbbt": int(source_config["mbbt"]), + "offered_rate_per_gpu": float(offered_rate_per_gpu), + "outcome": dict(outcome), + "telemetry": dict(telemetry), + }, + "action": { + "id": action_id, + "target_mns": int(target_config["mns"]), + "target_mbbt": int(target_config["mbbt"]), + }, + } + + +def aggregate_checkpoint( + *, + models: Sequence[Any], + examples_by_action: Mapping[str, Sequence[Mapping[str, Any]]], + include_telemetry: bool, + confidence_z: float, + minimum_margin: float, +) -> dict[str, Any]: + rows = [] + for action_id, examples in sorted(examples_by_action.items()): + raw = [] + for example in examples: + source = example["source"] + action = example["action"] + noop = ( + int(source["mns"]) == int(action["target_mns"]) + and int(source["mbbt"]) == int(action["target_mbbt"]) + ) + if noop: + raw.extend(0.0 for _model in models) + continue + names, values = MODEL.feature_vector( + example, include_telemetry=include_telemetry + ) + if any(model.feature_names != tuple(names) for model in models): + raise ValueError("prospective feature schema does not match frozen model") + raw.extend(model.predict(values) for model in models) + clipped = np.clip(np.asarray(raw, dtype=np.float64), -1.0, 1.0) + prediction = { + "mean": float(clipped.mean()), + "std": float(clipped.std(ddof=0)), + "min": float(clipped.min()), + "max": float(clipped.max()), + "distinct_n": len(set(float(value) for value in clipped)), + "sample_n": int(clipped.size), + } + rows.append( + { + "action_id": action_id, + "prediction": prediction, + "lower": prediction["mean"] - confidence_z * prediction["std"], + "upper": prediction["mean"] + confidence_z * prediction["std"], + } + ) + rows.sort(key=lambda row: (-row["prediction"]["mean"], row["action_id"])) + best, second = rows[:2] + margin = float(best["prediction"]["mean"] - second["prediction"]["mean"]) + confident = bool( + margin >= minimum_margin and best["lower"] > second["upper"] + ) + return { + "selected_action": best["action_id"], + "confident": confident, + "predicted_margin": margin, + "candidates": rows, + } + + +def apply_measurement_and_acquisition(checkpoints: list[dict[str, Any]]) -> dict[str, Any]: + selected = checkpoints[-1] + stop_reason = "full_measurement_fallback" + for previous, current in zip(checkpoints, checkpoints[1:], strict=False): + if ( + previous["confident"] + and current["confident"] + and previous["selected_action"] == current["selected_action"] + ): + selected = current + stop_reason = "two_consecutive_confident_checkpoints" + break + candidates = selected["candidates"] + mean_best = candidates[0] + non_noop = [row for row in candidates if row["action_id"] != "noop"] + if selected["confident"]: + chosen = mean_best + decision_kind = "exploit" + else: + positive_ucb = [row for row in non_noop if float(row["upper"]) > 0.0] + if positive_ucb: + chosen = max( + positive_ucb, + key=lambda row: (float(row["upper"]), row["action_id"]), + ) + decision_kind = "diagnostic_ucb" + else: + chosen = next(row for row in candidates if row["action_id"] == "noop") + decision_kind = "abstain_no_positive_ucb" + remaining = [row for row in candidates if row["action_id"] != chosen["action_id"]] + remaining.sort(key=lambda row: (-float(row["upper"]), row["action_id"])) + order = [chosen["action_id"], *(row["action_id"] for row in remaining)] + return { + "selected_phase": selected["phase"], + "selected_cutoff_s": selected["cutoff_s"], + "measurement_stop_reason": stop_reason, + "decision_kind": decision_kind, + "selected_action": chosen["action_id"], + "intervention_order": order, + "selected_checkpoint": selected, + "checkpoints": checkpoints, + } + + +def build_decision( + *, manifest_path: Path, policy_path: Path, run_root: Path +) -> dict[str, Any]: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + policy = json.loads(policy_path.read_text(encoding="utf-8")) + if manifest.get("schema") != "active-intervention-prospective-manifest-v0": + raise ValueError("unexpected prospective manifest schema") + if policy.get("schema") != "active-intervention-policy-v0": + raise ValueError("unexpected frozen policy schema") + if sha256_file(policy_path) != manifest["policy"]["sha256"]: + raise ValueError("frozen policy hash changed after manifest preparation") + configs = {str(item["id"]): item for item in manifest["configs"]} + source_id = str(manifest["source_config_id"]) + source_config = configs[source_id] + source_root = run_root / "sessions" / source_id + engine_records, stream_path = load_engine_records(source_root) + phases = [f"{fraction:.2f}" for fraction in manifest["checkpoints"]["fractions"]] + confidence_z = float(policy["measurement_policy"]["confidence_z"]) + minimum_margin = float(policy["measurement_policy"]["minimum_margin"]) + + examples: dict[str, dict[str, dict[str, Mapping[str, Any]]]] = {} + source_measurements: dict[str, dict[str, Any]] = {} + source_normalized = [] + telemetry_values = [] + for repetition in sorted(int(key) for key in manifest["repetitions"]): + item = manifest["repetitions"][str(repetition)] + result_root = source_root / f"rep{repetition}" + result = json.loads((result_root / "result.json").read_text(encoding="utf-8")) + if result["selection"]["request_id_order_sha256"] != item["selection"][ + "request_id_order_sha256" + ]: + raise ValueError(f"source request hash mismatch: rep{repetition}") + requests = EXTRACT.load_jsonl(result_root / "requests.jsonl") + offered_rate = float(item["selection"]["offered_req_s_per_gpu"]) + offered_total = offered_rate * int(manifest["engine"]["tp"]) + source_normalized.append( + float(result["slo_pass_count"]) + / float(manifest["engine"]["duration_s"]) + / offered_total + ) + start_ns = int(result["interval"]["start_mono_ns"]) + examples[str(repetition)] = {} + source_measurements[str(repetition)] = { + "result": str(result_root / "result.json"), + "result_sha256": sha256_file(result_root / "result.json"), + "request_sha256": sha256_file(result_root / "requests.jsonl"), + "phases": {}, + } + for phase, cutoff_s in zip( + phases, manifest["checkpoints"]["seconds"], strict=True + ): + outcome = EXTRACT.prefix_outcome( + requests, cutoff_s=float(cutoff_s), offered_total=offered_total + ) + admitted_count = sum( + float(request["arrival_s"]) <= float(cutoff_s) + for request in requests + ) + state = summarize_engine( + engine_records, + start_ns=start_ns, + end_ns=start_ns + round(float(cutoff_s) * 1e9), + request_count=admitted_count, + ) + if not all(state["sanity"]["invariants"].values()): + raise ValueError( + f"source engine state invariant failed: rep{repetition} {phase}" + ) + telemetry = EXTRACT.telemetry_record(state) + telemetry_values.extend(float(value) for value in telemetry.values()) + source_measurements[str(repetition)]["phases"][phase] = { + "cutoff_s": float(cutoff_s), + "outcome": outcome, + "telemetry": telemetry, + "engine_sanity": state["sanity"], + } + examples[str(repetition)][phase] = { + action_id: candidate_example( + source_config=source_config, + target_config=configs[str(target_id)], + action_id=action_id, + offered_rate_per_gpu=offered_rate, + outcome=outcome, + telemetry=telemetry, + ) + for action_id, target_id in manifest["actions"].items() + } + + decisions = {} + for mode, include_telemetry in (("outcome_only", False), ("telemetry", True)): + checkpoints = [] + for phase, cutoff_s in zip( + phases, manifest["checkpoints"]["seconds"], strict=True + ): + models = MODEL.models_from_json(policy["phases"][phase][mode]["models"]) + examples_by_action = { + action_id: [ + examples[str(repetition)][phase][action_id] + for repetition in sorted(int(key) for key in manifest["repetitions"]) + ] + for action_id in manifest["actions"] + } + checkpoint = aggregate_checkpoint( + models=models, + examples_by_action=examples_by_action, + include_telemetry=include_telemetry, + confidence_z=confidence_z, + minimum_margin=minimum_margin, + ) + checkpoints.append( + {"phase": phase, "cutoff_s": float(cutoff_s), **checkpoint} + ) + decisions[mode] = apply_measurement_and_acquisition(checkpoints) + + ceiling = float(manifest["gates"]["source_ceiling_normalized_goodput"]) + source_median = float(statistics.median(source_normalized)) + status = "STOP_SOURCE_CEILING" if source_median >= ceiling else "SELECTED" + phase_admission_monotonic = all( + all( + left <= right + 1e-12 + for left, right in zip(values, values[1:], strict=False) + ) + for repetition in source_measurements.values() + for values in ( + [ + float(repetition["phases"][phase]["outcome"]["admitted_fraction"]) + for phase in phases + ], + ) + ) + telemetry_ratio_keys = { + "prefill_token_fraction", + "kv_usage_mean", + "kv_usage_max", + "graph_none_share", + "graph_full_share", + "graph_padding_fraction", + } + telemetry_records = [ + measurement["telemetry"] + for repetition in source_measurements.values() + for measurement in repetition["phases"].values() + ] + invariants = { + "three_source_repetitions": len(source_normalized) == 3, + "source_goodput_nonnegative": all(value >= 0.0 for value in source_normalized), + "source_goodput_bounded": all( + value <= 1.0 + 1e-12 for value in source_normalized + ), + "four_actions": set(manifest["actions"]) == {"noop", "mns", "mbbt", "joint"}, + "four_checkpoints": len(phases) == 4, + "finite_telemetry": all(math.isfinite(value) for value in telemetry_values), + "nonnegative_telemetry": all( + float(value) >= 0.0 + for record in telemetry_records + for key, value in record.items() + if key != "kv_usage_end_minus_start" + ), + "telemetry_ratios_bounded": all( + 0.0 <= float(record[key]) <= 1.0 + 1e-12 + for record in telemetry_records + for key in telemetry_ratio_keys + ), + "telemetry_not_all_identical": len(set(telemetry_values)) > 1, + "phase_admission_monotonic": phase_admission_monotonic, + "orders_are_permutations": all( + set(decisions[mode]["intervention_order"]) == set(manifest["actions"]) + for mode in decisions + ), + } + red_flags = [name for name, passed in invariants.items() if not passed] + if red_flags: + status = "STOP_SANITY" + return { + "schema": SCHEMA, + "status": status, + "manifest": str(manifest_path), + "manifest_sha256": sha256_file(manifest_path), + "policy": str(policy_path), + "policy_sha256": sha256_file(policy_path), + "source_stream": str(stream_path), + "source_stream_sha256": sha256_file(stream_path), + "source_measurements": source_measurements, + "source_normalized_goodput": { + "values": source_normalized, + "median": source_median, + **numeric(source_normalized), + }, + "decisions": decisions, + "sanity": { + "invariants": invariants, + "red_flags": red_flags, + "telemetry_values": numeric(telemetry_values), + }, + } + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--manifest", type=Path, required=True) + parser.add_argument("--policy", type=Path, required=True) + parser.add_argument("--run-root", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + decision = build_decision( + manifest_path=args.manifest, policy_path=args.policy, run_root=args.run_root + ) + atomic_json(args.output, decision) + print( + json.dumps( + { + "status": decision["status"], + "source_normalized_goodput": decision["source_normalized_goodput"], + "outcome_only": { + key: decision["decisions"]["outcome_only"][key] + for key in ("selected_cutoff_s", "decision_kind", "selected_action") + }, + "telemetry": { + key: decision["decisions"]["telemetry"][key] + for key in ("selected_cutoff_s", "decision_kind", "selected_action") + }, + }, + sort_keys=True, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/runs/active-intervention-v0/test_pipeline.py b/runs/active-intervention-v0/test_pipeline.py index ee7805c..a6597eb 100644 --- a/runs/active-intervention-v0/test_pipeline.py +++ b/runs/active-intervention-v0/test_pipeline.py @@ -185,9 +185,12 @@ def main() -> None: write_json(dataset_path, dataset) policy = trainer.build_policy(dataset_path) assert policy["status"] in { - "RETROSPECTIVE_INCREMENTAL_SIGNAL", - "NO_RETROSPECTIVE_INCREMENTAL_SIGNAL", + "RETROSPECTIVE_GPU_COST_SIGNAL", + "NO_RETROSPECTIVE_GPU_COST_SIGNAL", } + assert policy["training"]["acceptable_regret"] == 0.02 + assert policy["sequential_replay"]["outcome_only"]["decision_n"] == 6 + assert policy["sequential_replay"]["telemetry"]["decision_n"] == 6 assert not policy["sanity"]["red_flags"] print("active intervention pipeline: PASS") diff --git a/runs/active-intervention-v0/test_prospective.py b/runs/active-intervention-v0/test_prospective.py new file mode 100644 index 0000000..7c95331 --- /dev/null +++ b/runs/active-intervention-v0/test_prospective.py @@ -0,0 +1,190 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import importlib.util +import json +import sys +import tempfile +from pathlib import Path + + +HERE = Path(__file__).resolve().parent + + +def load(name: str, path: Path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def write_json(path: Path, payload) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload) + "\n", encoding="utf-8") + + +def main() -> None: + prepare = load("active_intervention_prepare_test", HERE / "prepare_prospective.py") + decision_module = load( + "active_intervention_decision_test", HERE / "prospective_decision.py" + ) + analyzer = load("active_intervention_audit_test", HERE / "analyze_prospective.py") + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "source.jsonl" + source.write_text( + "".join( + json.dumps( + { + "request_id": f"request-{index}", + "timestamp": float(index), + "sampling_u": index / 100.0, + } + ) + + "\n" + for index in range(60) + ), + encoding="utf-8", + ) + partition = prepare.partition_trace(source, root / "partitions") + assert sum(item["rows"] for item in partition["partitions"].values()) == 60 + ids = [] + for item in partition["partitions"].values(): + assert item["rows"] > 0 + ids.extend( + json.loads(line)["request_id"] + for line in Path(item["path"]).read_text(encoding="utf-8").splitlines() + ) + assert len(ids) == len(set(ids)) == 60 + + checkpoints = [ + { + "phase": "0.25", + "cutoff_s": 75.0, + "selected_action": "joint", + "confident": True, + "candidates": [ + {"action_id": "joint", "upper": 0.5, "prediction": {"mean": 0.4}}, + {"action_id": "mns", "upper": 0.2, "prediction": {"mean": 0.1}}, + {"action_id": "mbbt", "upper": 0.1, "prediction": {"mean": 0.05}}, + {"action_id": "noop", "upper": 0.0, "prediction": {"mean": 0.0}}, + ], + }, + { + "phase": "0.50", + "cutoff_s": 150.0, + "selected_action": "joint", + "confident": True, + "candidates": [ + {"action_id": "joint", "upper": 0.45, "prediction": {"mean": 0.4}}, + {"action_id": "mns", "upper": 0.2, "prediction": {"mean": 0.1}}, + {"action_id": "mbbt", "upper": 0.1, "prediction": {"mean": 0.05}}, + {"action_id": "noop", "upper": 0.0, "prediction": {"mean": 0.0}}, + ], + }, + ] + selected = decision_module.apply_measurement_and_acquisition(checkpoints) + assert selected["selected_cutoff_s"] == 150.0 + assert selected["selected_action"] == "joint" + + configs = prepare.configs() + repetitions = { + str(rep): { + "selection": { + "offered_req_s_per_gpu": 0.25, + "request_id_order_sha256": f"hash-{rep}", + } + } + for rep in (1, 2, 3) + } + manifest = { + "schema": "active-intervention-prospective-manifest-v0", + "engine": {"duration_s": 300.0, "tp": 4}, + "repetitions": repetitions, + "configs": configs, + "source_config_id": "source_mns32_mbbt4096", + "actions": { + "noop": "source_mns32_mbbt4096", + "mns": "mns64_mbbt4096", + "mbbt": "mns32_mbbt8192", + "joint": "joint_mns64_mbbt8192", + }, + "gates": { + "acceptable_regret": 0.02, + "confirmation_trigger_gpu_cost_reduction": 0.10, + "contribution_gpu_cost_reduction": 0.20, + }, + } + manifest_path = root / "manifest.json" + write_json(manifest_path, manifest) + run_root = root / "run" + scores = { + "source_mns32_mbbt4096": 0.5, + "mns64_mbbt4096": 0.8, + "mns32_mbbt8192": 0.7, + "joint_mns64_mbbt8192": 1.0, + } + sessions = {} + for config in configs: + config_id = config["id"] + sessions[config_id] = {"status": "complete", "gpu_hours": 1.2} + for repetition in (1, 2, 3): + result = { + "selection": { + "request_id_order_sha256": f"hash-{repetition}" + }, + "slo_pass_count": round(scores[config_id] * 300), + "pass_rate": scores[config_id], + "interval": {"elapsed_s": 300.0}, + } + write_json( + run_root + / "sessions" + / config_id + / f"rep{repetition}" + / "result.json", + result, + ) + state = { + "status": "complete", + "gpu_hours_total": 4.8, + "sessions": sessions, + } + write_json(run_root / "controller-state.json", state) + mode_base = { + "selected_cutoff_s": 300.0, + "selected_action": "mns", + "decision_kind": "exploit", + "intervention_order": ["mns", "mbbt", "joint", "noop"], + } + mode_telemetry = { + "selected_cutoff_s": 150.0, + "selected_action": "joint", + "decision_kind": "exploit", + "intervention_order": ["joint", "mns", "mbbt", "noop"], + } + decision = { + "schema": "active-intervention-prospective-decision-v0", + "manifest_sha256": analyzer.sha256_file(manifest_path), + "decisions": { + "outcome_only": mode_base, + "telemetry": mode_telemetry, + }, + } + decision_path = root / "decision.json" + write_json(decision_path, decision) + audit = analyzer.build_audit( + manifest_path=manifest_path, + decision_path=decision_path, + run_root=run_root, + ) + assert audit["status"] == "TRIGGER_ACTUAL_EARLY_STOP_CONFIRMATION" + assert audit["comparison"]["telemetry_gpu_cost_reduction_fraction"] > 0.10 + assert not audit["sanity"]["red_flags"] + print("active intervention prospective pipeline: PASS") + + +if __name__ == "__main__": + main() diff --git a/runs/active-intervention-v0/train_policy.py b/runs/active-intervention-v0/train_policy.py index 8bbae02..bd8359c 100644 --- a/runs/active-intervention-v0/train_policy.py +++ b/runs/active-intervention-v0/train_policy.py @@ -33,6 +33,7 @@ MODEL = _load_model() REGULARIZATION = 10.0 MINIMUM_MARGIN = 0.02 CONFIDENCE_Z = 1.0 +ACCEPTABLE_REGRET = 0.02 def sha256_file(path: Path) -> str: @@ -110,13 +111,20 @@ def evaluate_grouped_cv( best_actions = { row["action_id"] for row in predictions if math.isclose(row["real"], oracle) } + acceptable_actions = { + row["action_id"] + for row in predictions + if oracle <= 0 + or 1.0 - float(row["real"]) / oracle <= ACCEPTABLE_REGRET + 1e-12 + } decision_rows.append( { "holdout": held_out, "decision_id": decision_id, "selected_action": selected["action_id"], "best_actions": sorted(best_actions), - "correct": selected["action_id"] in best_actions, + "acceptable_actions": sorted(acceptable_actions), + "correct": regret <= ACCEPTABLE_REGRET + 1e-12, "selected_real": selected["real"], "oracle_real": oracle, "regret": regret, @@ -129,6 +137,7 @@ def evaluate_grouped_cv( return { "status": "VALID", "holdout_key": holdout_key, + "acceptable_regret": ACCEPTABLE_REGRET, "decision_n": len(decision_rows), "correct_n": sum(bool(row["correct"]) for row in decision_rows), "accuracy": sum(bool(row["correct"]) for row in decision_rows) / len(decision_rows), @@ -172,6 +181,173 @@ def paired_delta(outcome: Mapping[str, Any], telemetry: Mapping[str, Any]) -> di } +def evaluate_sequential_measurement_cv( + examples: Sequence[Mapping[str, Any]], + *, + include_telemetry: bool, + holdout_key: str, +) -> dict[str, Any]: + """Replay a two-consecutive-confident-checkpoint measurement policy.""" + + phases = sorted({str(example["phase"]) for example in examples}, key=float) + holdouts = grouped(examples, holdout_key) + rows = [] + full_duration_s = max(float(example["cutoff_s"]) for example in examples) + for held_out, test_examples in sorted(holdouts.items()): + training = [ + example for example in examples if str(example[holdout_key]) != held_out + ] + if len({str(example["decision_id"]) for example in training}) < 3: + continue + phase_models = {} + for phase in phases: + phase_training = [ + example for example in training if str(example["phase"]) == phase + ] + phase_models[phase] = MODEL.fit_jackknife_ensemble( + phase_training, + include_telemetry=include_telemetry, + regularization=REGULARIZATION, + ) + for decision_id, decision_examples in sorted( + grouped(test_examples, "decision_id").items() + ): + checkpoints = [] + by_phase = grouped(decision_examples, "phase") + for phase in phases: + candidates = by_phase[phase] + decision = MODEL.select_action( + phase_models[phase], + candidates, + include_telemetry=include_telemetry, + confidence_z=CONFIDENCE_Z, + minimum_margin=MINIMUM_MARGIN, + ) + checkpoints.append( + { + "phase": phase, + "cutoff_s": float(candidates[0]["cutoff_s"]), + **decision, + } + ) + selected_checkpoint = checkpoints[-1] + stop_reason = "full_measurement_fallback" + for previous, current in zip(checkpoints, checkpoints[1:], strict=False): + if ( + previous["confident"] + and current["confident"] + and previous["selected_action"] == current["selected_action"] + ): + selected_checkpoint = current + stop_reason = "two_consecutive_confident_checkpoints" + break + candidates = by_phase[str(selected_checkpoint["phase"])] + real_by_action = { + str(candidate["action"]["id"]): float( + candidate["target_normalized_goodput"] + ) + for candidate in candidates + } + target_by_action = { + str(candidate["action"]["id"]): str( + candidate["action"]["target_config_id"] + ) + for candidate in candidates + } + selected_action = str(selected_checkpoint["selected_action"]) + oracle = max(real_by_action.values()) + selected_real = real_by_action[selected_action] + regret = 1.0 - selected_real / oracle if oracle > 0 else 0.0 + source_tp = 4 + target_s = 0.0 if selected_action == "noop" else full_duration_s + replay_gpu_seconds = source_tp * ( + float(selected_checkpoint["cutoff_s"]) + target_s + ) + rows.append( + { + "holdout": held_out, + "decision_id": decision_id, + "selected_phase": str(selected_checkpoint["phase"]), + "selected_cutoff_s": float(selected_checkpoint["cutoff_s"]), + "stop_reason": stop_reason, + "selected_action": selected_action, + "selected_target_config_id": target_by_action[selected_action], + "selected_real": selected_real, + "oracle_real": oracle, + "regret": regret, + "acceptable": regret <= ACCEPTABLE_REGRET + 1e-12, + "replay_gpu_seconds_lower_bound": replay_gpu_seconds, + "checkpoints": checkpoints, + } + ) + if not rows: + return {"status": "INSUFFICIENT_GROUPS", "decisions": []} + regrets = [float(row["regret"]) for row in rows] + cutoffs = [float(row["selected_cutoff_s"]) for row in rows] + costs = [float(row["replay_gpu_seconds_lower_bound"]) for row in rows] + return { + "status": "VALID", + "holdout_key": holdout_key, + "measurement_rule": "earliest two consecutive confident checkpoints; otherwise full", + "acceptable_regret": ACCEPTABLE_REGRET, + "decision_n": len(rows), + "acceptable_n": sum(bool(row["acceptable"]) for row in rows), + "mean_regret": sum(regrets) / len(regrets), + "max_regret": max(regrets), + "mean_cutoff_s": sum(cutoffs) / len(cutoffs), + "total_replay_gpu_seconds_lower_bound": sum(costs), + "total_replay_h20_hours_lower_bound": sum(costs) / 3600.0, + "decisions": rows, + } + + +def paired_sequential_delta( + outcome: Mapping[str, Any], telemetry: Mapping[str, Any] +) -> dict[str, Any]: + if outcome.get("status") != "VALID" or telemetry.get("status") != "VALID": + return {"status": "INSUFFICIENT_GROUPS"} + before_by_id = {row["decision_id"]: row for row in outcome["decisions"]} + after_by_id = {row["decision_id"]: row for row in telemetry["decisions"]} + rows = [] + for decision_id in sorted(set(before_by_id) & set(after_by_id)): + before = before_by_id[decision_id] + after = after_by_id[decision_id] + rows.append( + { + "decision_id": decision_id, + "outcome_action": before["selected_action"], + "telemetry_action": after["selected_action"], + "outcome_cutoff_s": before["selected_cutoff_s"], + "telemetry_cutoff_s": after["selected_cutoff_s"], + "outcome_regret": before["regret"], + "telemetry_regret": after["regret"], + "regret_delta": float(after["regret"]) - float(before["regret"]), + "gpu_seconds_delta": float( + after["replay_gpu_seconds_lower_bound"] + ) + - float(before["replay_gpu_seconds_lower_bound"]), + "telemetry_corrected": (not before["acceptable"]) + and bool(after["acceptable"]), + "telemetry_harmed": bool(before["acceptable"]) + and (not after["acceptable"]), + } + ) + outcome_cost = float(outcome["total_replay_gpu_seconds_lower_bound"]) + telemetry_cost = float(telemetry["total_replay_gpu_seconds_lower_bound"]) + return { + "status": "VALID", + "decision_n": len(rows), + "corrected_n": sum(row["telemetry_corrected"] for row in rows), + "harmed_n": sum(row["telemetry_harmed"] for row in rows), + "outcome_replay_gpu_seconds_lower_bound": outcome_cost, + "telemetry_replay_gpu_seconds_lower_bound": telemetry_cost, + "gpu_cost_reduction_fraction": ( + 1.0 - telemetry_cost / outcome_cost if outcome_cost > 0 else 0.0 + ), + "rows": rows, + } + + def build_policy(dataset_path: Path) -> dict[str, Any]: dataset = json.loads(dataset_path.read_text(encoding="utf-8")) if dataset.get("status") != "VALID" or dataset["sanity"]["red_flags"]: @@ -211,6 +387,10 @@ def build_policy(dataset_path: Path) -> dict[str, Any]: and int(delta["harmed_n"]) == 0 and float(delta["mean_regret_delta"]) < -1e-12 and float(telemetry_cv["max_regret"]) <= 0.05 + and telemetry_regime.get("status") == "VALID" + and float(telemetry_regime["mean_regret"]) + <= float(outcome_regime["mean_regret"]) + 1e-12 + and float(telemetry_regime["max_regret"]) <= 0.05 ) if incremental: incremental_candidates.append(phase) @@ -229,11 +409,27 @@ def build_policy(dataset_path: Path) -> dict[str, Any]: "paired_incremental": delta, "incremental_gate": incremental, } - selected_phase = incremental_candidates[0] if incremental_candidates else phases[-1] + outcome_sequential = evaluate_sequential_measurement_cv( + examples, include_telemetry=False, holdout_key="repetition" + ) + telemetry_sequential = evaluate_sequential_measurement_cv( + examples, include_telemetry=True, holdout_key="repetition" + ) + sequential_delta = paired_sequential_delta( + outcome_sequential, telemetry_sequential + ) + retrospective_cost_gate = bool( + sequential_delta.get("status") == "VALID" + and int(sequential_delta["harmed_n"]) == 0 + and int(telemetry_sequential["acceptable_n"]) + >= int(outcome_sequential["acceptable_n"]) + and float(telemetry_sequential["max_regret"]) <= 0.05 + and float(sequential_delta["gpu_cost_reduction_fraction"]) >= 0.10 + ) status = ( - "RETROSPECTIVE_INCREMENTAL_SIGNAL" - if incremental_candidates - else "NO_RETROSPECTIVE_INCREMENTAL_SIGNAL" + "RETROSPECTIVE_GPU_COST_SIGNAL" + if retrospective_cost_gate + else "NO_RETROSPECTIVE_GPU_COST_SIGNAL" ) target_values = [float(example["target_normalized_goodput"]) for example in examples] effect_values = [ @@ -265,15 +461,20 @@ def build_policy(dataset_path: Path) -> dict[str, Any]: "regularization": REGULARIZATION, "confidence_z": CONFIDENCE_Z, "minimum_margin": MINIMUM_MARGIN, + "acceptable_regret": ACCEPTABLE_REGRET, }, "measurement_policy": { - "selected_phase": selected_phase, - "selected_cutoff_s": phase_results[selected_phase]["cutoff_s"], - "selection_reason": ( - "earliest phase passing the frozen incremental gate" - if incremental_candidates - else "no incremental phase; retain full measurement for exploratory held-out test" - ), + "rule": "earliest two consecutive confident checkpoints; otherwise full", + "checkpoints": [phase_results[phase]["cutoff_s"] for phase in phases], + "confidence_z": CONFIDENCE_Z, + "minimum_margin": MINIMUM_MARGIN, + }, + "sequential_replay": { + "outcome_only": outcome_sequential, + "telemetry": telemetry_sequential, + "paired_delta": sequential_delta, + "retrospective_gpu_cost_gate": retrospective_cost_gate, + "minimum_cost_reduction_fraction": 0.10, }, "phases": phase_results, "sanity": {