Add prospective active intervention experiment

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

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

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

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

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

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

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": {