Make telemetry audit replay-phase aware

docs/intervention-response-v2-phase-aware-protocol-20260714.md

@@ -0,0 +1,84 @@
+# Phase-aware telemetry intervention-response v2 protocol
+
+Status: **FROZEN BEFORE THE SYSTEMATIC 10% DECILE AUDIT**.
+
+Date: 2026-07-14 (Asia/Singapore).
+
+## Correction to v0/v1
+
+The 5/10-second analyses tested an ultra-early verifier.  They did not test
+whether telemetry observed after the engine has developed queue, batch, and KV
+state can guide tuning.  The P1 replay lasts 60 seconds after time scaling, and
+the 5/10-second prefixes contain only a small fraction of its requests.
+
+V2 therefore replaces absolute cutoffs with replay phase.  The old audits and
+their negative decisions remain immutable, but their claim is narrowed to the
+first 5/10 seconds.
+
+## Historical corrective audit
+
+The historical audit is development-only and cannot become confirmatory after
+the horizon concern was observed.
+
+- Infer each trial's intended replay duration as selected requests divided by
+  offered requests per second.  All trials must agree.
+- Find every complete 10% replay decile supported by every trial.  Analyze all
+  such deciles; selecting only the best horizon is forbidden.
+- At each decile report both:
+  - cumulative state from replay start to the checkpoint; and
+  - the non-overlapping 10%-wide state block ending at the checkpoint.
+- Report admitted/completed request coverage, response-versus-repeat statistics,
+  telemetry versus external-outcome efficacy, and per-feature trajectory drift.
+- Reuse the frozen v1 action and repeat pairs and the frozen response and
+  incremental-efficacy thresholds.  These thresholds are descriptive in V2;
+  passing one post-hoc horizon does not open a contribution claim.
+
+If early stopping prevents complete observation of the replay phases, the
+historical decision is `REQUIRES_UNCENSORED_PHASE_AWARE_PILOT`, independent of
+which early decile looks best.
+
+## Uncensored matched pilot
+
+The pilot is a mechanism gate, not paper evidence.
+
+- Hardware/engine/model: solo placement on dash0, 4 NVIDIA H20 GPUs, patched
+  vLLM `0.24.1.dev3+opprof`, Qwen3-30B-A3B, fixed `TP=4`.
+- Action: `MNS 16 -> 64`; topology, model, engine build, workload, arrival
+  sequence, offered load, and all other settings remain fixed.
+- Workload: `chat_w20260312_1000` at replay-time scale `0.5`, hence 300 seconds.
+- Offered loads per GPU: `1.5`, `2.125`, and `3.125` requests/s.  These supply a
+  low control and the two already-observed P1 pressure regimes.
+- Repetitions: three disjoint session bands, exact request sequence matched
+  across action endpoints.  Endpoint order alternates `A/B`, `B/A`, `A/B`;
+  load order is counter-rotated across repetitions.
+- A fresh server receives the accepted long-request warm-up and a bounded
+  burn-in before each measured session.
+- SLO-unrecoverable early stop is disabled.  Every run must observe the full
+  300-second arrival window; a separate 360-second safety deadline may mark a
+  run invalid but cannot manufacture a full-run label.
+- Cumulative checkpoints: 10%, 25%, 50%, 75%, and 100%, or 30/75/150/225/300
+  seconds.  Quarter blocks are analyzed separately from cumulative means.
+- Placement is serialized.  Co-location remains forbidden because Phase 6
+  observed material co-location-induced outcome shifts.
+- Hard cap: 8 H20-hours including startup, warm-up, burn-in, invalid attempts,
+  and cleanup.
+
+## Gates
+
+Data validity requires complete 300-second Layer-1 coverage, zero dropped
+records, exact request/arrival/length hashes across action endpoints, monotonic
+timestamps, full request accounting, idle GPUs before and after each session,
+and no co-resident GPU process.
+
+Mechanism evidence requires at least two telemetry features whose matched action
+response exceeds same-config repeat noise at two consecutive checkpoints under
+the unchanged v1 response thresholds.  The same features must have a consistent
+direction in at least two of the three load regimes.
+
+Decision evidence additionally requires both action-efficacy classes and an
+instrumentation feature that reaches leave-one-repetition-out balanced accuracy
+at least 0.75 and exceeds the best external prefix outcome by at least 0.15.
+Without label balance the pilot can adjudicate mechanism evidence only.
+
+No H20 run is launched if the local analyzer/tests, manifest preflight, GPU
+probe, command dry-run, projected cost, or cleanup plan fails.

runs/intervention-response-v2/analyze_existing.py

@@ -0,0 +1,467 @@
+#!/usr/bin/env python3
+"""Audit telemetry responses over every uncensored replay decile.
+
+This corrective analysis keeps the frozen P1 pairs and thresholds, but replaces
+the absolute 5/10-second cutoff with cumulative and non-overlapping 10%-of-trace
+windows.  It deliberately reports every common decile instead of selecting the
+best-looking horizon.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import importlib.util
+import json
+import math
+from pathlib import Path
+from statistics import median
+from typing import Any, Iterable, Mapping
+
+
+HERE = Path(__file__).resolve().parent
+P1_PATH = HERE.parent / "intervention-response-v0" / "analyze_p1.py"
+SCHEMA = "intervention-response-phase-aware-existing-v2"
+DECILE_FRACTION = 0.1
+MAX_DECILES = 10
+
+
+def _load_p1():
+    spec = importlib.util.spec_from_file_location("intervention_response_p1", P1_PATH)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+P1 = _load_p1()
+
+
+def numeric(values: Iterable[float | int]) -> dict[str, Any]:
+    finite = [float(value) for value in values]
+    result = P1.V0.numeric(finite)
+    result["median"] = median(finite)
+    return result
+
+
+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 trial_directories(run_root: Path) -> list[Path]:
+    result = []
+    for cell in sorted((run_root / "cells").iterdir()):
+        if not cell.is_dir():
+            continue
+        for candidate in sorted(cell.iterdir()):
+            if candidate.is_dir() and P1.RUN_PATTERN.match(candidate.name):
+                result.append(candidate)
+    if not result:
+        raise ValueError("P1 run root contains no measured trial directories")
+    return result
+
+
+def load_metadata(run_root: Path) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    metadata = []
+    streams = []
+    for cell in sorted((run_root / "cells").iterdir()):
+        if not cell.is_dir():
+            continue
+        stream_paths = sorted((cell / "opprof").glob("*.jsonl"))
+        if len(stream_paths) != 1:
+            raise ValueError(f"{cell}: expected one Layer-1 stream")
+        stream_path = stream_paths[0]
+        streams.append(
+            {
+                "cell": cell.name,
+                "path": str(stream_path.resolve()),
+                "sha256": sha256_file(stream_path),
+                "bytes": stream_path.stat().st_size,
+            }
+        )
+    for run_dir in trial_directories(run_root):
+        match = P1.RUN_PATTERN.match(run_dir.name)
+        assert match is not None
+        level, replicate_text = match.groups()
+        result_path = run_dir / "result.json"
+        requests_path = run_dir / "requests.jsonl"
+        result = json.loads(result_path.read_text(encoding="utf-8"))
+        selected = int(result["selection"]["count"])
+        offered = float(result["selection"]["offered_req_s"])
+        if selected <= 0 or offered <= 0.0:
+            raise ValueError(f"{result_path}: invalid selected count or offered rate")
+        metadata.append(
+            {
+                "trial_id": str(result_path.relative_to(run_root)),
+                "cell": str(result["cell"]),
+                "tp": int(result["tp"]),
+                "mns": int(result["mns"]),
+                "level": level,
+                "replicate": int(replicate_text),
+                "elapsed_s": float(result["interval"]["elapsed_s"]),
+                "trace_duration_s": round(selected / offered, 9),
+                "early_stopped": bool(result["early_stopped"]),
+                "request_count": selected,
+                "result_sha256": sha256_file(result_path),
+                "requests_sha256": sha256_file(requests_path),
+            }
+        )
+    return metadata, streams
+
+
+def common_decile_fractions(
+    *, trace_duration_s: float, minimum_elapsed_s: float
+) -> tuple[float, ...]:
+    if trace_duration_s <= 0.0 or minimum_elapsed_s <= 0.0:
+        raise ValueError("trace duration and elapsed time must be positive")
+    supported = min(
+        MAX_DECILES,
+        int(math.floor((minimum_elapsed_s / trace_duration_s) * 10.0 + 1e-12)),
+    )
+    return tuple(
+        round(index * DECILE_FRACTION, 10) for index in range(1, supported + 1)
+    )
+
+
+def _trial_record(
+    *,
+    run_root: Path,
+    run_dir: Path,
+    result: Mapping[str, Any],
+    state: dict[str, float],
+    outcome: dict[str, float],
+) -> dict[str, Any]:
+    match = P1.RUN_PATTERN.match(run_dir.name)
+    assert match is not None
+    level, replicate_text = match.groups()
+    result_path = run_dir / "result.json"
+    requests_path = run_dir / "requests.jsonl"
+    return {
+        "trial_id": str(result_path.relative_to(run_root)),
+        "cell": str(result["cell"]),
+        "tp": int(result["tp"]),
+        "mns": int(result["mns"]),
+        "level": level,
+        "replicate": int(replicate_text),
+        "offered_rate_per_gpu": float(
+            result["selection"]["offered_req_s_per_gpu"]
+        ),
+        "request_hash": str(result["selection"]["request_id_order_sha256"]),
+        "request_count": int(result["selection"]["count"]),
+        "result_sha256": sha256_file(result_path),
+        "requests_sha256": sha256_file(requests_path),
+        "full_pass_rate": float(result["pass_rate"]),
+        "full_feasible": bool(result["feasible"]),
+        "early_stopped": bool(result["early_stopped"]),
+        "state": state,
+        "outcome": outcome,
+    }
+
+
+def load_interval_trials(
+    run_root: Path,
+    intervals_s: tuple[tuple[float, float], ...],
+) -> tuple[dict[tuple[float, float], list[dict[str, Any]]], list[dict[str, Any]]]:
+    by_interval = {interval: [] for interval in intervals_s}
+    stream_provenance = []
+    for cell in sorted((run_root / "cells").iterdir()):
+        if not cell.is_dir():
+            continue
+        stream_paths = sorted((cell / "opprof").glob("*.jsonl"))
+        if len(stream_paths) != 1:
+            raise ValueError(f"{cell}: expected one Layer-1 stream")
+        stream_path = stream_paths[0]
+        stream = P1.load_jsonl(stream_path)
+        stream_provenance.append(
+            {
+                "cell": cell.name,
+                "path": str(stream_path.resolve()),
+                "sha256": sha256_file(stream_path),
+                "bytes": stream_path.stat().st_size,
+            }
+        )
+        for run_dir in sorted(cell.iterdir()):
+            if not run_dir.is_dir() or P1.RUN_PATTERN.match(run_dir.name) is None:
+                continue
+            result_path = run_dir / "result.json"
+            requests_path = run_dir / "requests.jsonl"
+            result = json.loads(result_path.read_text(encoding="utf-8"))
+            requests = P1.load_jsonl(requests_path)
+            start_ns = int(result["interval"]["start_mono_ns"])
+            elapsed_s = float(result["interval"]["elapsed_s"])
+            for interval in intervals_s:
+                start_s, end_s = interval
+                if start_s < 0.0 or end_s <= start_s:
+                    raise ValueError(f"invalid analysis interval: {interval}")
+                if elapsed_s + 1e-9 < end_s:
+                    raise ValueError(
+                        f"{result_path}: elapsed {elapsed_s} shorter than {end_s}s"
+                    )
+                state = P1.V0.flatten_state(
+                    P1.summarize_engine(
+                        stream,
+                        start_ns=start_ns + int(start_s * 1e9),
+                        end_ns=start_ns + int(end_s * 1e9),
+                        request_count=int(result["selection"]["count"]),
+                    )
+                )
+                outcome = P1._prefix_outcome(result, requests, end_s)
+                by_interval[interval].append(
+                    _trial_record(
+                        run_root=run_root,
+                        run_dir=run_dir,
+                        result=result,
+                        state=state,
+                        outcome=outcome,
+                    )
+                )
+    return by_interval, stream_provenance
+
+
+def coverage(trials: list[dict[str, Any]]) -> dict[str, Any]:
+    admitted = [float(trial["outcome"]["admitted_fraction"]) for trial in trials]
+    completed = [
+        float(trial["outcome"]["admitted_fraction"])
+        * float(trial["outcome"]["completed_over_admitted"])
+        for trial in trials
+    ]
+    return {
+        "admitted_fraction_of_total": numeric(admitted),
+        "completed_fraction_of_total": numeric(completed),
+    }
+
+
+def slim_window_analysis(
+    trials: list[dict[str, Any]], *, start_s: float, end_s: float, fraction: float
+) -> dict[str, Any]:
+    analysis = P1.analyze_horizon(trials, end_s)
+    return {
+        "start_s": start_s,
+        "end_s": end_s,
+        "end_fraction": fraction,
+        "coverage_at_end": coverage(trials),
+        "action_pairs": len(analysis["actions"]),
+        "repeat_pairs": len(analysis["repeats"]),
+        "response_statistics": analysis["response_statistics"],
+        "qualifying_response_features": analysis["qualifying_response_features"],
+        "efficacy": analysis["efficacy"],
+        "sanity": analysis["sanity"],
+    }
+
+
+def _pearson(left: list[float], right: list[float]) -> float | None:
+    if len(left) != len(right) or not left:
+        raise ValueError("Pearson inputs must be non-empty and have equal length")
+    left_mean = sum(left) / len(left)
+    right_mean = sum(right) / len(right)
+    numerator = sum(
+        (x - left_mean) * (y - right_mean)
+        for x, y in zip(left, right, strict=True)
+    )
+    left_ss = sum((x - left_mean) ** 2 for x in left)
+    right_ss = sum((y - right_mean) ** 2 for y in right)
+    if left_ss == 0.0 or right_ss == 0.0:
+        return None
+    return numerator / math.sqrt(left_ss * right_ss)
+
+
+def trajectory_summary(
+    block_trials: list[tuple[tuple[float, float], list[dict[str, Any]]]]
+) -> dict[str, Any]:
+    if not block_trials:
+        raise ValueError("trajectory requires at least one block")
+    identities = []
+    states_by_block = []
+    for interval, trials in block_trials:
+        ordered = sorted(
+            trials,
+            key=lambda trial: (trial["cell"], trial["level"], trial["replicate"]),
+        )
+        current_identities = [
+            (trial["cell"], trial["level"], trial["replicate"]) for trial in ordered
+        ]
+        if identities and current_identities != identities:
+            raise ValueError("trajectory blocks do not contain identical trials")
+        identities = current_identities
+        states_by_block.append((interval, [trial["state"] for trial in ordered]))
+
+    features = {}
+    for feature in P1.V0.ALL_FEATURES:
+        block_values = [
+            [float(state[feature]) for state in states]
+            for _interval, states in states_by_block
+        ]
+        first = block_values[0]
+        last = block_values[-1]
+        delta = [right - left for left, right in zip(first, last, strict=True)]
+        features[feature] = {
+            "block_medians": [median(values) for values in block_values],
+            "first_to_last_delta": numeric(delta),
+            "first_to_last_abs_delta": numeric(abs(value) for value in delta),
+            "first_to_last_pearson": _pearson(first, last),
+            "changed_trials": sum(abs(value) > 1e-12 for value in delta),
+        }
+    return {
+        "trial_count": len(identities),
+        "blocks": [
+            {"start_s": interval[0], "end_s": interval[1]}
+            for interval, _states in states_by_block
+        ],
+        "features": features,
+    }
+
+
+def audit(*, run_root: Path, manifest_path: Path, output_path: Path) -> dict[str, Any]:
+    metadata, metadata_streams = load_metadata(run_root)
+    durations = [float(item["trace_duration_s"]) for item in metadata]
+    elapsed = [float(item["elapsed_s"]) for item in metadata]
+    duration = median(durations)
+    deciles = common_decile_fractions(
+        trace_duration_s=duration, minimum_elapsed_s=min(elapsed)
+    )
+    if not deciles:
+        raise ValueError("no complete replay decile is shared by all trials")
+    cumulative_intervals = tuple(
+        (0.0, round(duration * fraction, 9)) for fraction in deciles
+    )
+    block_intervals = tuple(
+        (
+            round(duration * (fraction - DECILE_FRACTION), 9),
+            round(duration * fraction, 9),
+        )
+        for fraction in deciles
+    )
+    all_intervals = tuple(dict.fromkeys([*cumulative_intervals, *block_intervals]))
+    trials_by_interval, streams = load_interval_trials(run_root, all_intervals)
+    manifest_validation = P1.validate_manifest(
+        trials_by_interval[cumulative_intervals[0]], manifest_path
+    )
+
+    cumulative = []
+    blocks = []
+    for fraction, cumulative_interval, block_interval in zip(
+        deciles, cumulative_intervals, block_intervals, strict=True
+    ):
+        cumulative.append(
+            slim_window_analysis(
+                trials_by_interval[cumulative_interval],
+                start_s=cumulative_interval[0],
+                end_s=cumulative_interval[1],
+                fraction=fraction,
+            )
+        )
+        blocks.append(
+            slim_window_analysis(
+                trials_by_interval[block_interval],
+                start_s=block_interval[0],
+                end_s=block_interval[1],
+                fraction=fraction,
+            )
+        )
+
+    invariants = {
+        "expected_trial_count": len(metadata) == 36,
+        "trace_duration_consistent": max(durations) - min(durations) <= 1e-9,
+        "all_intervals_uncensored": all(
+            item["elapsed_s"] + 1e-9 >= cumulative_intervals[-1][1]
+            for item in metadata
+        ),
+        "stream_provenance_consistent": metadata_streams == streams,
+        "manifest_trials_match": (
+            manifest_validation["expected_trials"]
+            == manifest_validation["matched_trials"]
+            == len(metadata)
+        ),
+        "all_window_sanity_pass": all(
+            not item["sanity"]["red_flags"] for item in [*cumulative, *blocks]
+        ),
+    }
+    red_flags = [name for name, passed in invariants.items() if not passed]
+    complete_full_trajectory = min(elapsed) + 1e-9 >= duration
+    if red_flags:
+        decision = "STOP_DATA_INVALID"
+    elif not complete_full_trajectory:
+        decision = "REQUIRES_UNCENSORED_PHASE_AWARE_PILOT"
+    else:
+        decision = "FULL_TRAJECTORY_AVAILABLE"
+
+    payload = {
+        "schema": SCHEMA,
+        "status": "COMPLETE",
+        "decision": decision,
+        "claim_boundary": (
+            "Post-hoc corrective audit over every common replay decile. It can "
+            "diagnose horizon sensitivity but cannot establish a held-out tuning claim."
+        ),
+        "design": {
+            "decile_fraction": DECILE_FRACTION,
+            "available_deciles": list(deciles),
+            "trace_duration_s": duration,
+            "maximum_common_end_s": cumulative_intervals[-1][1],
+            "maximum_common_fraction": deciles[-1],
+            "select_best_horizon": False,
+            "cumulative_and_nonoverlapping_blocks": True,
+        },
+        "cumulative": cumulative,
+        "blocks": blocks,
+        "trajectory": trajectory_summary(
+            [(interval, trials_by_interval[interval]) for interval in block_intervals]
+        ),
+        "provenance": {
+            "analysis_script": str(Path(__file__).resolve()),
+            "analysis_script_sha256": sha256_file(Path(__file__).resolve()),
+            "p1_analysis_script": str(P1_PATH.resolve()),
+            "p1_analysis_script_sha256": sha256_file(P1_PATH),
+            "run_root": str(run_root.resolve()),
+            "manifest": str(manifest_path.resolve()),
+            "manifest_sha256": sha256_file(manifest_path),
+            "manifest_validation": manifest_validation,
+            "streams": streams,
+            "trial_inputs": metadata,
+        },
+        "sanity": {
+            "trials": len(metadata),
+            "elapsed_s": numeric(elapsed),
+            "trace_duration_s": numeric(durations),
+            "early_stopped": sum(bool(item["early_stopped"]) for item in metadata),
+            "request_count": numeric(item["request_count"] for item in metadata),
+            "stream_bytes": numeric(item["bytes"] for item in streams),
+            "invariants": invariants,
+            "red_flags": red_flags,
+        },
+    }
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    return payload
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--run-root", type=Path, required=True)
+    parser.add_argument("--manifest", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    payload = audit(
+        run_root=args.run_root,
+        manifest_path=args.manifest,
+        output_path=args.output,
+    )
+    print(
+        json.dumps(
+            {
+                "decision": payload["decision"],
+                "design": payload["design"],
+                "sanity": payload["sanity"],
+            },
+            indent=2,
+            sort_keys=True,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

runs/intervention-response-v2/analyze_pilot.py

@@ -0,0 +1,449 @@
+#!/usr/bin/env python3
+"""Analyze the uncensored 300-second phase-aware matched pilot."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import importlib.util
+import json
+import math
+from collections import defaultdict
+from pathlib import Path
+from typing import Any, Iterable, Mapping
+
+
+HERE = Path(__file__).resolve().parent
+P1_PATH = HERE.parent / "intervention-response-v0" / "analyze_p1.py"
+SCHEMA = "intervention-response-phase-aware-pilot-analysis-v2"
+EXPECTED_ACTION_PAIRS = 9
+EXPECTED_REPEAT_PAIRS = 12
+MIN_EFFICACY_CLASS = 3
+
+
+def _load_p1():
+    spec = importlib.util.spec_from_file_location("intervention_response_p1", P1_PATH)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+P1 = _load_p1()
+
+
+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 numeric(values: Iterable[float | int]) -> dict[str, Any]:
+    return P1.V0.numeric(values)
+
+
+def _trial_record(
+    *,
+    run_root: Path,
+    session: Mapping[str, Any],
+    level: str,
+    result: Mapping[str, Any],
+    result_path: Path,
+    requests_path: Path,
+    state: dict[str, float],
+    outcome: dict[str, float],
+) -> dict[str, Any]:
+    return {
+        "trial_id": str(result_path.relative_to(run_root)),
+        "cell": str(result["cell"]),
+        "tp": int(result["tp"]),
+        "mns": int(result["mns"]),
+        "level": level,
+        "replicate": int(session["replicate"]),
+        "offered_rate_per_gpu": float(
+            result["selection"]["offered_req_s_per_gpu"]
+        ),
+        "request_hash": str(result["selection"]["request_id_order_sha256"]),
+        "request_count": int(result["selection"]["count"]),
+        "result_sha256": sha256_file(result_path),
+        "requests_sha256": sha256_file(requests_path),
+        "full_pass_rate": float(result["pass_rate"]),
+        "full_feasible": bool(result["feasible"]),
+        "early_stopped": bool(result["early_stopped"]),
+        "state": state,
+        "outcome": outcome,
+    }
+
+
+def validate_result_against_manifest(
+    *,
+    result: Mapping[str, Any],
+    selection: Mapping[str, Any],
+    session: Mapping[str, Any],
+    level: str,
+    expected_duration_s: float,
+) -> None:
+    identity = f"{session['session']}:{level}"
+    if int(result["mns"]) != int(session["mns"]) or int(result["tp"]) != 4:
+        raise ValueError(f"config mismatch: {identity}")
+    if bool(result["early_stopped"]):
+        raise ValueError(f"early-stopped measured result: {identity}")
+    if result.get("slo_early_stop_disabled") is not True:
+        raise ValueError(f"SLO early stop was enabled: {identity}")
+    if float(result["interval"]["elapsed_s"]) + 1e-9 < expected_duration_s:
+        raise ValueError(f"result does not cover full arrival window: {identity}")
+    if int(result["selection"]["count"]) != int(selection["selected_count"]):
+        raise ValueError(f"selection count mismatch: {identity}")
+    for result_key, manifest_key in (
+        ("request_id_order_sha256", "request_id_order_sha256"),
+        ("arrival_order_sha256", "arrival_order_sha256"),
+        ("raw_length_order_sha256", "input_length_order_sha256"),
+    ):
+        if result["selection"][result_key] != selection[manifest_key]:
+            raise ValueError(f"selection hash mismatch {result_key}: {identity}")
+    if int(result["observed_count"]) != int(selection["selected_count"]):
+        raise ValueError(f"request accounting mismatch: {identity}")
+
+
+def load_interval_trials(
+    *,
+    run_root: Path,
+    manifest: Mapping[str, Any],
+    intervals_s: tuple[tuple[float, float], ...],
+) -> tuple[dict[tuple[float, float], list[dict[str, Any]]], list[dict[str, Any]]]:
+    by_interval = {interval: [] for interval in intervals_s}
+    streams = []
+    duration_s = float(manifest["engine"]["duration_s"])
+    for session in manifest["sessions"]:
+        session_root = run_root / "sessions" / str(session["session"])
+        stream_paths = sorted((session_root / "opprof").glob("*.jsonl"))
+        if len(stream_paths) != 1:
+            raise ValueError(f"{session_root}: expected one Layer-1 stream")
+        stream_path = stream_paths[0]
+        stream = P1.load_jsonl(stream_path)
+        streams.append(
+            {
+                "session": str(session["session"]),
+                "path": str(stream_path.resolve()),
+                "sha256": sha256_file(stream_path),
+                "bytes": stream_path.stat().st_size,
+            }
+        )
+        repetition = manifest["repetitions"][str(session["replicate"])]
+        for level, selection in repetition["selections"].items():
+            result_path = session_root / level / "result.json"
+            requests_path = session_root / level / "requests.jsonl"
+            result = json.loads(result_path.read_text(encoding="utf-8"))
+            requests = P1.load_jsonl(requests_path)
+            validate_result_against_manifest(
+                result=result,
+                selection=selection,
+                session=session,
+                level=level,
+                expected_duration_s=duration_s,
+            )
+            start_ns = int(result["interval"]["start_mono_ns"])
+            for interval in intervals_s:
+                start_s, end_s = interval
+                state = P1.V0.flatten_state(
+                    P1.summarize_engine(
+                        stream,
+                        start_ns=start_ns + int(start_s * 1e9),
+                        end_ns=start_ns + int(end_s * 1e9),
+                        request_count=int(result["selection"]["count"]),
+                    )
+                )
+                outcome = P1._prefix_outcome(result, requests, end_s)
+                by_interval[interval].append(
+                    _trial_record(
+                        run_root=run_root,
+                        session=session,
+                        level=level,
+                        result=result,
+                        result_path=result_path,
+                        requests_path=requests_path,
+                        state=state,
+                        outcome=outcome,
+                    )
+                )
+    return by_interval, streams
+
+
+def analyze_window(
+    trials: list[dict[str, Any]], *, start_s: float, end_s: float, fraction: float
+) -> dict[str, Any]:
+    actions, repeats = P1.build_pairs(trials)
+    response = P1.V0.response_statistics(actions, repeats)
+    response_qualifying = sorted(
+        feature for feature, item in response.items() if item["qualifies"]
+    )
+    labels = [int(pair["full_action_efficacy"]) for pair in actions]
+    cross_validation_possible = all(
+        set(
+            int(pair["full_action_efficacy"])
+            for pair in actions
+            if pair["group"]["replicate"] != held_out
+        )
+        == {0, 1}
+        for held_out in (1, 2, 3)
+    )
+    if cross_validation_possible:
+        outcome_cv = P1.one_feature_leave_repeat_out(
+            actions, delta_key="delta_outcome", features=P1.OUTCOME_FEATURES
+        )
+        telemetry_cv = P1.one_feature_leave_repeat_out(
+            actions, delta_key="delta_state", features=P1.V0.GATE_FEATURES
+        )
+        outcome_best = float(outcome_cv["best_balanced_accuracy"])
+        efficacy_qualifying = sorted(
+            feature
+            for feature, item in telemetry_cv["features"].items()
+            if item["balanced_accuracy"] >= P1.MIN_EFFICACY_BALANCED_ACCURACY
+            and item["balanced_accuracy"]
+            >= outcome_best + P1.MIN_EFFICACY_DELTA_OVER_OUTCOME
+        )
+    else:
+        unavailable = {
+            "status": "UNAVAILABLE",
+            "reason": "each leave-one-repetition-out train fold needs both classes",
+        }
+        outcome_cv = unavailable
+        telemetry_cv = unavailable
+        efficacy_qualifying = []
+    transitions = defaultdict(int)
+    for pair in actions:
+        transitions[pair["full_feasibility_transition"]] += 1
+    admitted = [float(trial["outcome"]["admitted_fraction"]) for trial in trials]
+    completed = [
+        float(trial["outcome"]["admitted_fraction"])
+        * float(trial["outcome"]["completed_over_admitted"])
+        for trial in trials
+    ]
+    action_metadata = [
+        {
+            "group": pair["group"],
+            "source": pair["source"],
+            "target": pair["target"],
+            "full_action_efficacy": pair["full_action_efficacy"],
+            "full_feasibility_transition": pair["full_feasibility_transition"],
+            "delta_state": pair["delta_state"],
+            "delta_outcome": pair["delta_outcome"],
+        }
+        for pair in actions
+    ]
+    invariants = {
+        "expected_action_pair_count": len(actions) == EXPECTED_ACTION_PAIRS,
+        "expected_repeat_pair_count": len(repeats) == EXPECTED_REPEAT_PAIRS,
+        "finite_deltas": all(
+            math.isfinite(value)
+            for pair in [*actions, *repeats]
+            for key in ("delta_state", "delta_outcome")
+            for value in pair[key].values()
+        ),
+        "all_results_uncensored": all(not trial["early_stopped"] for trial in trials),
+    }
+    return {
+        "start_s": start_s,
+        "end_s": end_s,
+        "end_fraction": fraction,
+        "coverage_at_end": {
+            "admitted_fraction_of_total": numeric(admitted),
+            "completed_fraction_of_total": numeric(completed),
+        },
+        "action_pairs": len(actions),
+        "repeat_pairs": len(repeats),
+        "actions": action_metadata,
+        "response_statistics": response,
+        "qualifying_response_features": response_qualifying,
+        "efficacy": {
+            "labels": numeric(labels),
+            "positive": sum(labels),
+            "negative": len(labels) - sum(labels),
+            "label_balance_sufficient": (
+                sum(labels) >= MIN_EFFICACY_CLASS
+                and len(labels) - sum(labels) >= MIN_EFFICACY_CLASS
+            ),
+            "cross_validation_possible": cross_validation_possible,
+            "transitions": dict(sorted(transitions.items())),
+            "outcome_delta": outcome_cv,
+            "telemetry_delta": telemetry_cv,
+            "telemetry_qualifying_features": efficacy_qualifying,
+        },
+        "sanity": {
+            "trials": len(trials),
+            "invariants": invariants,
+            "red_flags": [name for name, passed in invariants.items() if not passed],
+        },
+    }
+
+
+def stable_adjacent_features(windows: list[dict[str, Any]]) -> dict[str, list[str]]:
+    result = {}
+    for left, right in zip(windows, windows[1:], strict=False):
+        key = f"{left['end_fraction']:.2f}->{right['end_fraction']:.2f}"
+        result[key] = sorted(
+            set(left["qualifying_response_features"])
+            & set(right["qualifying_response_features"])
+        )
+    return result
+
+
+def consistent_load_regimes(
+    windows: list[dict[str, Any]], stable: dict[str, list[str]]
+) -> dict[str, Any]:
+    by_end = {float(window["end_fraction"]): window for window in windows}
+    result = {}
+    for transition, features in stable.items():
+        _left_text, right_text = transition.split("->")
+        window = by_end[float(right_text)]
+        for feature in features:
+            deltas_by_level: dict[str, list[float]] = defaultdict(list)
+            all_deltas = []
+            for action in window["actions"]:
+                value = float(action["delta_state"][feature])
+                deltas_by_level[str(action["group"]["level"])].append(value)
+                all_deltas.append(value)
+            positive = sum(value > 1e-12 for value in all_deltas)
+            negative = sum(value < -1e-12 for value in all_deltas)
+            direction = 1 if positive >= negative else -1
+            consistent = []
+            for level, values in sorted(deltas_by_level.items()):
+                matching = sum(direction * value > 1e-12 for value in values)
+                nonzero = sum(abs(value) > 1e-12 for value in values)
+                if nonzero and matching / nonzero >= 2.0 / 3.0:
+                    consistent.append(level)
+            result[f"{transition}:{feature}"] = {
+                "direction": direction,
+                "consistent_load_regimes": consistent,
+                "passes_two_regimes": len(consistent) >= 2,
+            }
+    return result
+
+
+def audit(*, run_root: Path, manifest_path: Path, output_path: Path) -> dict[str, Any]:
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    if manifest.get("schema") != "intervention-response-phase-aware-pilot-manifest-v2":
+        raise ValueError("unexpected phase-aware pilot manifest schema")
+    fractions = [float(value) for value in manifest["checkpoints"]["fractions"]]
+    seconds = [float(value) for value in manifest["checkpoints"]["seconds"]]
+    cumulative_intervals = tuple((0.0, end_s) for end_s in seconds)
+    quarter_intervals = ((0.0, 75.0), (75.0, 150.0), (150.0, 225.0), (225.0, 300.0))
+    intervals = tuple(dict.fromkeys([*cumulative_intervals, *quarter_intervals]))
+    trials_by_interval, streams = load_interval_trials(
+        run_root=run_root, manifest=manifest, intervals_s=intervals
+    )
+    cumulative = [
+        analyze_window(
+            trials_by_interval[interval],
+            start_s=interval[0],
+            end_s=interval[1],
+            fraction=fraction,
+        )
+        for fraction, interval in zip(fractions, cumulative_intervals, strict=True)
+    ]
+    quarter_blocks = [
+        analyze_window(
+            trials_by_interval[interval],
+            start_s=interval[0],
+            end_s=interval[1],
+            fraction=interval[1] / 300.0,
+        )
+        for interval in quarter_intervals
+    ]
+    stable = stable_adjacent_features(cumulative)
+    load_consistency = consistent_load_regimes(cumulative, stable)
+    mechanism_features = sorted(
+        {
+            key.split(":", 1)[1]
+            for key, item in load_consistency.items()
+            if item["passes_two_regimes"]
+        }
+    )
+    full = cumulative[-1]
+    efficacy_features = sorted(
+        set.intersection(
+            *(
+                set(window["efficacy"]["telemetry_qualifying_features"])
+                for window in cumulative
+                if window["end_fraction"] >= 0.25
+            )
+        )
+    )
+    red_flags = sorted(
+        {
+            flag
+            for window in [*cumulative, *quarter_blocks]
+            for flag in window["sanity"]["red_flags"]
+        }
+    )
+    if red_flags:
+        decision = "STOP_DATA_INVALID"
+    elif not mechanism_features:
+        decision = "STOP_NO_PHASE_STABLE_RESPONSE"
+    elif not full["efficacy"]["label_balance_sufficient"]:
+        decision = "MECHANISM_ONLY_NO_LABEL_BALANCE"
+    elif not efficacy_features:
+        decision = "STOP_NO_INCREMENTAL_TUNING_SIGNAL"
+    else:
+        decision = "OPEN_E2E_POLICY_TEST"
+    payload = {
+        "schema": SCHEMA,
+        "status": "COMPLETE",
+        "decision": decision,
+        "claim_boundary": "Development mechanism pilot; not a held-out paper claim.",
+        "mechanism_features": mechanism_features,
+        "stable_adjacent_features": stable,
+        "load_consistency": load_consistency,
+        "stable_incremental_efficacy_features": efficacy_features,
+        "cumulative": cumulative,
+        "quarter_blocks": quarter_blocks,
+        "provenance": {
+            "analysis_script": str(Path(__file__).resolve()),
+            "analysis_script_sha256": sha256_file(Path(__file__).resolve()),
+            "manifest": str(manifest_path.resolve()),
+            "manifest_sha256": sha256_file(manifest_path),
+            "run_root": str(run_root.resolve()),
+            "streams": streams,
+        },
+        "sanity": {
+            "streams": len(streams),
+            "stream_bytes": numeric(item["bytes"] for item in streams),
+            "red_flags": red_flags,
+        },
+    }
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    return payload
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--run-root", type=Path, required=True)
+    parser.add_argument("--manifest", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    payload = audit(
+        run_root=args.run_root,
+        manifest_path=args.manifest,
+        output_path=args.output,
+    )
+    print(
+        json.dumps(
+            {
+                "decision": payload["decision"],
+                "mechanism_features": payload["mechanism_features"],
+                "stable_incremental_efficacy_features": payload[
+                    "stable_incremental_efficacy_features"
+                ],
+                "sanity": payload["sanity"],
+            },
+            indent=2,
+            sort_keys=True,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

runs/intervention-response-v2/pilot_controller.py

@@ -0,0 +1,471 @@
+#!/usr/bin/env python3
+"""Serialized, resumable controller for the 300-second phase-aware pilot."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import shlex
+import signal
+import subprocess
+import sys
+import time
+from pathlib import Path
+from typing import Any
+
+
+HERE = Path(__file__).resolve().parent
+PHASE6 = HERE.parent / "opprof-phase6"
+sys.path.insert(0, str(PHASE6))
+
+import opprof_phase6_controller as base  # noqa: E402
+
+
+SCHEMA = "intervention-response-phase-aware-pilot-state-v2"
+SESSION_ESTIMATE_H20_HOURS = 1.25
+SAFETY_H20_HOURS = 0.20
+CLIENT_TIMEOUT_S = 450.0
+
+
+def atomic_json(path: Path, payload: Any) -> None:
+    base.atomic_json(path, payload)
+
+
+def wait_all_idle(timeout_s: float = 30.0) -> None:
+    deadline = time.monotonic() + timeout_s
+    last_error: Exception | None = None
+    while time.monotonic() < deadline:
+        try:
+            base.assert_all_idle()
+            return
+        except RuntimeError as error:
+            last_error = error
+            time.sleep(1.0)
+    raise last_error or RuntimeError("GPU idle timeout")
+
+
+def configure(args: argparse.Namespace, manifest: dict[str, Any]) -> None:
+    base.WORKDIR = args.run_root.parent
+    base.RUN_ROOT = args.run_root
+    base.STATE = args.run_root / "controller-state.json"
+    base.SOURCE = args.vllm_source
+    base.VENV = args.venv
+    base.AITUNER = args.aituner_root
+    base.MODEL = args.model
+    base.CLIENT = args.client
+    base.GPU_LIMIT = float(manifest["budget"]["hard_cap_h20_hours"])
+    base.MARKER = "intervention-response-phase-aware-v2"
+    base.CELLS = {
+        f"tp4_mns{mns}": {"tp": 4, "mns": int(mns)}
+        for mns in manifest["engine"]["mns_endpoints"]
+    }
+
+
+def load_state(path: Path, hard_cap: float) -> dict[str, Any]:
+    if path.exists():
+        return json.loads(path.read_text(encoding="utf-8"))
+    return {
+        "schema": SCHEMA,
+        "status": "initialized",
+        "hard_cap_h20_hours": hard_cap,
+        "gpu_hours_total": 0.0,
+        "completed_sessions": 0,
+        "sessions": {},
+        "failures": [],
+        "started_at": time.time(),
+    }
+
+
+def save_state(path: Path, state: dict[str, Any]) -> None:
+    atomic_json(path, state)
+
+
+def append_echo(run_root: Path, line: str) -> None:
+    run_root.mkdir(parents=True, exist_ok=True)
+    with (run_root / "launch-echo.log").open("a", encoding="utf-8") as target:
+        target.write(line + "\n")
+    print(line, flush=True)
+
+
+def remaining_projection(session_count: int, index: int) -> float:
+    return (session_count - index) * SESSION_ESTIMATE_H20_HOURS + SAFETY_H20_HOURS
+
+
+def start_server(
+    *, session: dict[str, Any], index: int, run_root: Path
+) -> dict[str, Any]:
+    cell = f"tp4_mns{int(session['mns'])}"
+    gpus = (0, 1, 2, 3)
+    session_root = run_root / "sessions" / str(session["session"])
+    session_root.mkdir(parents=True, exist_ok=True)
+    port = 8950 + index
+    command = base.server_command(cell, gpus, port)
+    with (session_root / "commands.log").open("a", encoding="utf-8") as log:
+        log.write(f"SERVER {shlex.join(command)}\n")
+    server_log = (session_root / "server.log").open("ab", buffering=0)
+    environment = os.environ.copy()
+    environment.update(
+        {
+            "CUDA_VISIBLE_DEVICES": "0,1,2,3",
+            "VLLM_OPPROF_DIR": str(session_root / "opprof"),
+            "OPPROF_PHASE6_MARKER": base.MARKER,
+            "AITUNER_ROOT": str(base.AITUNER),
+            "HF_HUB_OFFLINE": "1",
+            "TRANSFORMERS_OFFLINE": "1",
+            "PYTHONUNBUFFERED": "1",
+        }
+    )
+    server = subprocess.Popen(
+        command,
+        cwd=base.SOURCE,
+        env=environment,
+        stdout=server_log,
+        stderr=subprocess.STDOUT,
+        start_new_session=True,
+    )
+    base.OWNED_PGIDS.add(server.pid)
+    return {
+        "cell": cell,
+        "gpus": gpus,
+        "port": port,
+        "dir": session_root,
+        "server": server,
+        "server_handle": server_log,
+        "spawned_at": time.time(),
+        "results": [],
+    }
+
+
+def client_command(
+    entry: dict[str, Any],
+    *,
+    study: str,
+    anchor: float,
+    output: Path,
+    warmup: bool,
+) -> list[str]:
+    config = base.CELLS[entry["cell"]]
+    command = [
+        "taskset",
+        "-c",
+        base.cpu_mask(entry["gpus"]),
+        str(base.VENV / "bin/python"),
+        str(base.CLIENT),
+        "warmup" if warmup else "run-anchor",
+        "--study",
+        study,
+        "--cell",
+        entry["cell"],
+        "--anchor",
+        str(anchor),
+        "--tp",
+        str(config["tp"]),
+        "--mns",
+        str(config["mns"]),
+        "--base-url",
+        f"http://127.0.0.1:{entry['port']}",
+        "--result-dir",
+        str(output),
+        "--disable-slo-early-stop",
+    ]
+    return command
+
+
+def run_client(
+    *,
+    entry: dict[str, Any],
+    role: str,
+    study: str,
+    selection: dict[str, Any],
+    output: Path,
+    state: dict[str, Any],
+    warmup: bool = False,
+) -> dict[str, Any]:
+    command = client_command(
+        entry,
+        study=study,
+        anchor=float(selection["anchor"]),
+        output=output,
+        warmup=warmup,
+    )
+    with (entry["dir"] / "commands.log").open("a", encoding="utf-8") as log:
+        log.write(f"CLIENT role={role} {shlex.join(command)}\n")
+    handle = (output.parent / f"{output.name}.log").open("ab", buffering=0)
+    environment = os.environ.copy()
+    environment.update({"AITUNER_ROOT": str(base.AITUNER), "PYTHONUNBUFFERED": "1"})
+    process = subprocess.Popen(
+        command,
+        cwd=base.WORKDIR,
+        env=environment,
+        stdout=handle,
+        stderr=subprocess.STDOUT,
+        start_new_session=True,
+    )
+    deadline = time.monotonic() + (180.0 if warmup else CLIENT_TIMEOUT_S)
+    try:
+        while process.poll() is None:
+            if time.monotonic() > deadline:
+                raise TimeoutError(f"client timeout: {entry['cell']} {role}")
+            if entry["server"].poll() is not None:
+                raise RuntimeError(f"server exited during {entry['cell']} {role}")
+            base.assert_no_other_compute()
+            if state["gpu_hours_total"] + base.live_gpu_hours([entry]) >= base.GPU_LIMIT:
+                raise RuntimeError("phase-aware pilot H20-hour hard cap reached")
+            time.sleep(1.0)
+    except Exception:
+        try:
+            os.killpg(process.pid, signal.SIGTERM)
+        except ProcessLookupError:
+            pass
+        try:
+            process.wait(timeout=10.0)
+        except subprocess.TimeoutExpired:
+            try:
+                os.killpg(process.pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+            process.wait(timeout=10.0)
+        raise
+    finally:
+        handle.close()
+    if process.returncode:
+        raise RuntimeError(
+            f"client failed: cell={entry['cell']} role={role} rc={process.returncode}"
+        )
+    result = json.loads((output / "result.json").read_text(encoding="utf-8"))
+    validate_result(
+        result=result,
+        selection=selection,
+        role=role,
+        warmup=warmup,
+    )
+    entry["results"].append(
+        {
+            "anchor": float(selection["anchor"]),
+            "dir": str(output),
+            "kind": result["kind"],
+        }
+    )
+    return result
+
+
+def validate_result(
+    *, result: dict[str, Any], selection: dict[str, Any], role: str, warmup: bool
+) -> None:
+    if result.get("slo_early_stop_disabled") is not True:
+        raise RuntimeError(f"SLO early stop was not disabled: {role}")
+    if warmup:
+        if result["kind"] != "warmup" or int(result["selection"]["count"]) != 16:
+            raise RuntimeError(f"invalid warmup result: {role}")
+        if not all(
+            result["invariants"].get(key, False)
+            for key in ("warmup_16", "warmup_exact_16", "warmup_long")
+        ):
+            raise RuntimeError(f"warmup invariant failed: {role}")
+        return
+    if bool(result["early_stopped"]):
+        raise RuntimeError(f"uncensored run early-stopped: {role}")
+    if int(result["selection"]["count"]) != int(selection["selected_count"]):
+        raise RuntimeError(f"selection count mismatch: {role}")
+    for key, manifest_key in (
+        ("request_id_order_sha256", "request_id_order_sha256"),
+        ("arrival_order_sha256", "arrival_order_sha256"),
+        ("raw_length_order_sha256", "input_length_order_sha256"),
+    ):
+        if result["selection"][key] != selection[manifest_key]:
+            raise RuntimeError(f"selection hash mismatch {key}: {role}")
+    if int(result["observed_count"]) != int(selection["selected_count"]):
+        raise RuntimeError(f"request accounting mismatch: {role}")
+
+
+def execute_session(
+    *,
+    index: int,
+    session: dict[str, Any],
+    manifest: dict[str, Any],
+    run_root: Path,
+    state_path: Path,
+    state: dict[str, Any],
+) -> None:
+    name = str(session["session"])
+    if state["sessions"].get(name, {}).get("status") == "complete":
+        return
+    projection = remaining_projection(len(manifest["sessions"]), index)
+    if state["gpu_hours_total"] + projection > base.GPU_LIMIT:
+        state["status"] = "budget_projection_stop"
+        state["budget_stop"] = {
+            "before_session": name,
+            "spent_h20_hours": state["gpu_hours_total"],
+            "remaining_projection_h20_hours": projection,
+            "hard_cap_h20_hours": base.GPU_LIMIT,
+        }
+        save_state(state_path, state)
+        raise RuntimeError(f"projected pilot cost exceeds hard cap before {name}")
+
+    replicate = str(session["replicate"])
+    repetition = manifest["repetitions"][replicate]
+    echo = (
+        f"PHASE_PILOT_SESSION_ECHO session={name} tp=4 mns={session['mns']} "
+        f"gpus=0-3 workload={manifest['source']['window_id']} duration_s=300 "
+        f"loads={','.join(repetition['load_order'])} disable_slo_early_stop=true "
+        f"spent_h20h={state['gpu_hours_total']:.6f} "
+        f"remaining_projection_h20h={projection:.3f} cap_h20h={base.GPU_LIMIT:.1f} "
+        f"manifest={run_root / 'pilot-manifest.json'}"
+    )
+    append_echo(run_root, echo)
+    wait_all_idle()
+    session_state = {
+        "status": "starting",
+        "replicate": int(replicate),
+        "mns": int(session["mns"]),
+        "started_at": time.time(),
+        "runs": [],
+    }
+    state["status"] = "running"
+    state["sessions"][name] = session_state
+    save_state(state_path, state)
+    entry = start_server(session=session, index=index, run_root=run_root)
+    failure: Exception | None = None
+    try:
+        base.wait_ready(entry)
+        high = repetition["selections"]["high"]
+        session_state["status"] = "warmup"
+        save_state(state_path, state)
+        run_client(
+            entry=entry,
+            role="warmup",
+            study=repetition["study"],
+            selection=high,
+            output=entry["dir"] / "warmup",
+            state=state,
+            warmup=True,
+        )
+        session_state["status"] = "burnin"
+        save_state(state_path, state)
+        burnin = manifest["burnin"]
+        burnin_result = run_client(
+            entry=entry,
+            role="burnin",
+            study=burnin["study"],
+            selection=burnin,
+            output=entry["dir"] / "burnin",
+            state=state,
+        )
+        session_state["burnin"] = {
+            "pass_rate": burnin_result["pass_rate"],
+            "feasible": burnin_result["feasible"],
+            "elapsed_s": burnin_result["interval"]["elapsed_s"],
+        }
+        session_state["status"] = "measured"
+        save_state(state_path, state)
+        for level in repetition["load_order"]:
+            selection = repetition["selections"][level]
+            result = run_client(
+                entry=entry,
+                role=level,
+                study=repetition["study"],
+                selection=selection,
+                output=entry["dir"] / level,
+                state=state,
+            )
+            session_state["runs"].append(
+                {
+                    "level": level,
+                    "selected_count": selection["selected_count"],
+                    "offered_req_s_per_gpu": selection["offered_req_s_per_gpu"],
+                    "pass_rate": result["pass_rate"],
+                    "feasible": result["feasible"],
+                    "elapsed_s": result["interval"]["elapsed_s"],
+                    "early_stopped": result["early_stopped"],
+                }
+            )
+            save_state(state_path, state)
+        session_state["status"] = "stopping"
+        save_state(state_path, state)
+    except Exception as error:  # noqa: BLE001
+        failure = error
+    finally:
+        try:
+            base.stop_entry(entry)
+        except Exception as error:  # noqa: BLE001
+            failure = failure or error
+        time.sleep(2.0)
+        try:
+            wait_all_idle()
+        except Exception as error:  # noqa: BLE001
+            failure = failure or error
+
+    session_hours = base.live_gpu_hours([entry])
+    state["gpu_hours_total"] += session_hours
+    session_state["gpu_hours"] = session_hours
+    if failure is not None:
+        session_state["status"] = "failed"
+        session_state["failure"] = repr(failure)
+        state["status"] = "failed"
+        state["failures"].append({"session": name, "failure": repr(failure)})
+        save_state(state_path, state)
+        raise failure
+    validation = base.validate_cell(entry)
+    session_state["validation"] = validation
+    session_state["status"] = "complete"
+    session_state["completed_at"] = time.time()
+    state["completed_sessions"] += 1
+    save_state(state_path, state)
+
+
+def parser() -> argparse.ArgumentParser:
+    result = argparse.ArgumentParser()
+    result.add_argument("--manifest", type=Path, required=True)
+    result.add_argument("--run-root", type=Path, required=True)
+    result.add_argument("--aituner-root", type=Path, required=True)
+    result.add_argument("--vllm-source", type=Path, required=True)
+    result.add_argument("--venv", type=Path, required=True)
+    result.add_argument("--model", type=Path, required=True)
+    result.add_argument("--client", type=Path, required=True)
+    return result
+
+
+def main() -> None:
+    args = parser().parse_args()
+    manifest = json.loads(args.manifest.read_text(encoding="utf-8"))
+    if manifest.get("schema") != "intervention-response-phase-aware-pilot-manifest-v2":
+        raise RuntimeError("unexpected phase-aware pilot manifest schema")
+    if manifest["status"] != "PASS":
+        raise RuntimeError("phase-aware pilot manifest did not pass preflight")
+    args.run_root.mkdir(parents=True, exist_ok=True)
+    copied_manifest = args.run_root / "pilot-manifest.json"
+    if not copied_manifest.exists():
+        atomic_json(copied_manifest, manifest)
+    configure(args, manifest)
+    state_path = args.run_root / "controller-state.json"
+    state = load_state(state_path, base.GPU_LIMIT)
+    state["status"] = "running"
+    save_state(state_path, state)
+    for index, session in enumerate(manifest["sessions"]):
+        execute_session(
+            index=index,
+            session=session,
+            manifest=manifest,
+            run_root=args.run_root,
+            state_path=state_path,
+            state=state,
+        )
+    state["status"] = "complete"
+    state["completed_at"] = time.time()
+    save_state(state_path, state)
+    wait_all_idle()
+    print(
+        json.dumps(
+            {
+                "status": state["status"],
+                "completed_sessions": state["completed_sessions"],
+                "gpu_hours_total": state["gpu_hours_total"],
+            },
+            sort_keys=True,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

runs/intervention-response-v2/prepare_pilot.py

@@ -0,0 +1,267 @@
+#!/usr/bin/env python3
+"""Prepare the uncensored 300-second TP4 matched pilot on dash0."""
+
+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 = "intervention-response-phase-aware-pilot-manifest-v2"
+TP = 4
+MNS_ENDPOINTS = (16, 64)
+REPLAY_TIME_SCALE = 0.5
+EXPECTED_DURATION_S = 300.0
+SAFETY_DEADLINE_S = 360.0
+LOADS_REQ_S_GPU = {"low": 1.5, "mid": 2.125, "high": 3.125}
+REPLICATE_ROLES = ("high1", "high2", "high3")
+LOAD_ORDERS = {
+    1: ("low", "mid", "high"),
+    2: ("high", "low", "mid"),
+    3: ("mid", "high", "low"),
+}
+SESSION_ORDER = (
+    (1, 16),
+    (1, 64),
+    (2, 64),
+    (2, 16),
+    (3, 16),
+    (3, 64),
+)
+
+
+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")
+    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 attainable_anchor(requests: list[Any], target_count: int) -> tuple[float, list[Any]]:
+    ordered = sorted(float(request.sampling_u) for request in requests)
+    if not ordered:
+        raise ValueError("no requests remain after study filtering")
+    if target_count <= 0 or target_count > len(ordered):
+        raise ValueError(
+            f"target count {target_count} is outside available range 1..{len(ordered)}"
+        )
+    indices = sorted(
+        {
+            max(0, min(len(ordered) - 1, target_count - 1)),
+            max(0, min(len(ordered) - 1, target_count)),
+        }
+    )
+    candidates = []
+    for index in indices:
+        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], *, duration_s: float) -> dict[str, Any]:
+    return {
+        "anchor": max(float(request.sampling_u) for request in selected),
+        "selected_count": len(selected),
+        "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 materialize_study(source: Path, target: Path, *, replicate: int) -> Path:
+    payload = json.loads(source.read_text(encoding="utf-8"))
+    payload["study_id"] = f"phase-aware-telemetry-v2-rep{replicate}"
+    payload["hardware"]["host_candidates"] = ["dash0"]
+    payload["engine"]["engine_version"] = "0.24.1.dev3+opprof"
+    trace = payload["trace"]
+    trace["replay_time_scale"] = REPLAY_TIME_SCALE
+    trace["early_stop_max_lag_s"] = None
+    trace["early_stop_max_elapsed_s"] = SAFETY_DEADLINE_S
+    trace["restart_engine_after_early_stop"] = False
+    trace["adaptive_stop"] = {"enabled": False}
+    atomic_json(target, payload)
+    return target
+
+
+def build_manifest(
+    *, base_manifest_path: Path, private_root: Path
+) -> dict[str, Any]:
+    base = json.loads(base_manifest_path.read_text(encoding="utf-8"))
+    if base.get("schema") != "fidelity-prefix-pilot-manifest-v1":
+        raise ValueError("unexpected base P1 manifest schema")
+
+    repetitions = {}
+    selection_hashes = []
+    selected_ids_by_replicate: dict[int, set[str]] = {}
+    for replicate, role in enumerate(REPLICATE_ROLES, start=1):
+        source_study = Path(base["private"]["studies"][role]["tp4"])
+        target_study = private_root / "studies" / f"rep{replicate}-tp4.json"
+        materialize_study(source_study, target_study, replicate=replicate)
+        study = load_study_spec(target_study)
+        window, requests = load_trace_requests(study, study_spec_path=target_study)
+        duration_s = float(window.window_end - window.window_start)
+        if not math.isclose(
+            duration_s, EXPECTED_DURATION_S, rel_tol=0.0, abs_tol=1e-9
+        ):
+            raise ValueError(
+                f"rep{replicate}: replay duration {duration_s} != {EXPECTED_DURATION_S}"
+            )
+        selections = {}
+        selected_ids_by_level: dict[str, set[str]] = {}
+        for level, target_rate in LOADS_REQ_S_GPU.items():
+            target_count = round(target_rate * duration_s * TP)
+            anchor, selected = attainable_anchor(requests, target_count)
+            record = selection_record(selected, duration_s=duration_s)
+            record.update(
+                {
+                    "anchor": anchor,
+                    "target_count": target_count,
+                    "target_req_s_per_gpu": target_rate,
+                }
+            )
+            selections[level] = record
+            selected_ids_by_level[level] = {request.row_id for request in selected}
+            selection_hashes.append(record["request_id_order_sha256"])
+        selected_ids_by_replicate[replicate] = selected_ids_by_level["high"]
+        repetitions[str(replicate)] = {
+            "source_role": role,
+            "study": str(target_study),
+            "study_sha256": sha256_file(target_study),
+            "duration_s": duration_s,
+            "load_order": list(LOAD_ORDERS[replicate]),
+            "selections": selections,
+        }
+
+    burnin = base["cells"]["tp4_mns16"]["targets"]["low"]["selections"][
+        "burnin"
+    ]
+    burnin = dict(burnin)
+    burnin["study_sha256"] = sha256_file(Path(burnin["study"]))
+    sessions = [
+        {
+            "session": f"rep{replicate}-mns{mns}",
+            "replicate": replicate,
+            "mns": mns,
+        }
+        for replicate, mns in SESSION_ORDER
+    ]
+    invariants = {
+        "three_repetitions": len(repetitions) == 3,
+        "six_sessions": len(sessions) == 6,
+        "load_levels_three": all(
+            len(item["selections"]) == 3 for item in repetitions.values()
+        ),
+        "selection_hashes_unique": len(selection_hashes) == len(set(selection_hashes)),
+        "selection_sets_disjoint_across_repetitions": all(
+            not selected_ids_by_replicate[left] & selected_ids_by_replicate[right]
+            for left in selected_ids_by_replicate
+            for right in selected_ids_by_replicate
+            if left < right
+        ),
+        "all_counts_positive": all(
+            selection["selected_count"] > 0
+            for item in repetitions.values()
+            for selection in item["selections"].values()
+        ),
+    }
+    red_flags = [name for name, passed in invariants.items() if not passed]
+    return {
+        "schema": SCHEMA,
+        "status": "PASS" if not red_flags else "STOP",
+        "source": {
+            "base_manifest": str(base_manifest_path),
+            "base_manifest_sha256": sha256_file(base_manifest_path),
+            "window_id": base["source"]["window_id"],
+            "source_trace": base["source"]["trace"],
+            "source_trace_sha256": base["source"]["trace_sha256"],
+        },
+        "engine": {
+            "tp": TP,
+            "mns_endpoints": list(MNS_ENDPOINTS),
+            "replay_time_scale": REPLAY_TIME_SCALE,
+            "duration_s": EXPECTED_DURATION_S,
+            "safety_deadline_s": SAFETY_DEADLINE_S,
+            "disable_slo_early_stop": True,
+        },
+        "burnin": burnin,
+        "repetitions": repetitions,
+        "sessions": sessions,
+        "checkpoints": {
+            "fractions": [0.1, 0.25, 0.5, 0.75, 1.0],
+            "seconds": [30.0, 75.0, 150.0, 225.0, 300.0],
+        },
+        "budget": {
+            "hard_cap_h20_hours": 8.0,
+            "expected_wall_minutes": [95, 120],
+            "expected_h20_hours": [6.3, 8.0],
+        },
+        "sanity": {
+            "red_flags": red_flags,
+            "invariants": invariants,
+            "selected_sets": len(selection_hashes),
+            "distinct_selected_sets": len(set(selection_hashes)),
+        },
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--base-manifest", type=Path, required=True)
+    parser.add_argument("--private-root", type=Path, required=True)
+    parser.add_argument("--public-manifest", type=Path, required=True)
+    args = parser.parse_args()
+    manifest = build_manifest(
+        base_manifest_path=args.base_manifest, private_root=args.private_root
+    )
+    atomic_json(args.public_manifest, manifest)
+    print(
+        json.dumps(
+            {
+                "status": manifest["status"],
+                "manifest": str(args.public_manifest),
+                "sanity": manifest["sanity"],
+            },
+            sort_keys=True,
+        )
+    )
+    if manifest["status"] != "PASS":
+        raise RuntimeError(f"phase-aware pilot preflight failed: {manifest['sanity']}")
+
+
+if __name__ == "__main__":
+    main()

runs/intervention-response-v2/test_analysis.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import importlib.util
+import math
+from pathlib import Path
+from types import SimpleNamespace
+
+
+HERE = Path(__file__).resolve().parent
+
+
+def load_module():
+    spec = importlib.util.spec_from_file_location(
+        "intervention_response_phase_aware_v2", HERE / "analyze_existing.py"
+    )
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def load_prepare_module():
+    spec = importlib.util.spec_from_file_location(
+        "intervention_response_phase_aware_prepare", HERE / "prepare_pilot.py"
+    )
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def load_controller_module():
+    spec = importlib.util.spec_from_file_location(
+        "intervention_response_phase_aware_controller", HERE / "pilot_controller.py"
+    )
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def load_pilot_analysis_module():
+    spec = importlib.util.spec_from_file_location(
+        "intervention_response_phase_aware_pilot_analysis", HERE / "analyze_pilot.py"
+    )
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def main() -> None:
+    module = load_module()
+    assert module.common_decile_fractions(
+        trace_duration_s=60.0, minimum_elapsed_s=19.448
+    ) == (0.1, 0.2, 0.3)
+    assert module.common_decile_fractions(
+        trace_duration_s=60.0, minimum_elapsed_s=60.0
+    )[-1] == 1.0
+    stats = module.numeric([0.0, 1.0, 2.0])
+    assert stats == {
+        "n": 3,
+        "min": 0.0,
+        "max": 2.0,
+        "distinct_n": 3,
+        "median": 1.0,
+    }
+    assert math.isclose(module._pearson([1.0, 2.0], [2.0, 4.0]), 1.0)
+    assert module._pearson([1.0, 1.0], [2.0, 3.0]) is None
+    prepare = load_prepare_module()
+    requests = [
+        SimpleNamespace(
+            sampling_u=index / 10.0,
+            row_id=f"r{index}",
+            arrival_s=float(index),
+            prompt_tokens_hint=100 + index,
+        )
+        for index in range(1, 6)
+    ]
+    _anchor, selected = prepare.attainable_anchor(requests, 3)
+    assert len(selected) == 3
+    record = prepare.selection_record(selected, duration_s=3.0)
+    assert record["selected_count"] == 3
+    assert record["offered_req_s_per_gpu"] == 0.25
+    assert len(prepare.SESSION_ORDER) == 6
+    assert {mns for _replicate, mns in prepare.SESSION_ORDER} == {16, 64}
+    controller = load_controller_module()
+    assert math.isclose(controller.remaining_projection(6, 0), 7.7)
+    assert math.isclose(controller.remaining_projection(6, 5), 1.45)
+    pilot_analysis = load_pilot_analysis_module()
+    stable = pilot_analysis.stable_adjacent_features(
+        [
+            {"end_fraction": 0.1, "qualifying_response_features": ["queue"]},
+            {
+                "end_fraction": 0.25,
+                "qualifying_response_features": ["kv", "queue"],
+            },
+            {"end_fraction": 0.5, "qualifying_response_features": ["queue"]},
+        ]
+    )
+    assert stable == {"0.10->0.25": ["queue"], "0.25->0.50": ["queue"]}
+    print("phase-aware intervention response v2 analysis: PASS")
+
+
+if __name__ == "__main__":
+    main()

runs/opprof-phase6/opprof_phase6_client.py

@@ -95,7 +95,11 @@ def run_replay(args: argparse.Namespace, *, warmup: bool) -> dict[str, Any]:
         base_url=args.base_url,
         timeout_s=study.engine.request_timeout_s,
         max_concurrency=study.trace.max_concurrency,
-        target_pass_rate=(0.0 if warmup else study.slo.target_pass_rate),
+        target_pass_rate=(
+            0.0
+            if warmup or args.disable_slo_early_stop
+            else study.slo.target_pass_rate
+        ),
         max_lag_s=study.trace.early_stop_max_lag_s,
         max_elapsed_s=(
             120.0 if warmup else _probe_drain_deadline(
@@ -162,6 +166,7 @@ def run_replay(args: argparse.Namespace, *, warmup: bool) -> dict[str, Any]:
         "feasible": bool(slo_summary["feasible"]),
         "early_stopped": early_stopped,
         "early_stop_reason": early_stop_reason,
+        "slo_early_stop_disabled": bool(args.disable_slo_early_stop),
         "ttft_ms": numeric([item.ttft_ms for item in outcomes]),
         "tpot_ms": numeric([item.tpot_ms for item in outcomes]),
         "invariants": {
@@ -240,6 +245,7 @@ def parser() -> argparse.ArgumentParser:
         q.add_argument("--mns", type=int, required=True)
         q.add_argument("--base-url", required=True)
         q.add_argument("--result-dir", required=True)
+        q.add_argument("--disable-slo-early-stop", action="store_true")
     return p
 
 

runs/opprof-phase6/test_phase6_tools.py

@@ -20,6 +20,7 @@ def main() -> None:
     controller = load("p6controller", HERE / "opprof_phase6_controller.py")
     solo = load("p6solo", HERE / "opprof_phase6_solo_controller.py")
     analysis = load("p6analysis", HERE / "analyze_phase6.py")
+    client = load("p6client", HERE / "opprof_phase6_client.py")
     assert len(controller.CELLS) == 12
     primary = sum(3 if item.get("trap") else 2 for item in controller.CELLS.values())
     assert primary == 25
@@ -36,6 +37,18 @@ def main() -> None:
     assert sum(solo.CELL_ESTIMATE.values()) + solo.SAFETY_HOURS == 3.44
     assert solo.next_below([.1, .2, .3], {.2, .3}) == .1
     assert solo.next_above([.1, .2, .3], {.1, .2}) == .3
+    parsed = client.parser().parse_args([
+        "run-anchor",
+        "--study", "study.json",
+        "--cell", "tp4_mns16",
+        "--anchor", "0.5",
+        "--tp", "4",
+        "--mns", "16",
+        "--base-url", "http://127.0.0.1:8000",
+        "--result-dir", "out",
+        "--disable-slo-early-stop",
+    ])
+    assert parsed.disable_slo_early_stop is True
     print("phase6 tools: PASS")