From c0b40af24ff3a30c60f4bfb36ad4de14b1a02c67 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Tue, 14 Jul 2026 17:35:58 +0800 Subject: [PATCH] Enforce phase-stable telemetry pilot gates --- ...sponse-v2-phase-aware-protocol-20260714.md | 19 +- .../intervention-response-v2/analyze_pilot.py | 230 ++++++++++++++++-- .../intervention-response-v2/test_analysis.py | 64 +++++ 3 files changed, 292 insertions(+), 21 deletions(-) diff --git a/docs/intervention-response-v2-phase-aware-protocol-20260714.md b/docs/intervention-response-v2-phase-aware-protocol-20260714.md index d87025b..a29e535 100644 --- a/docs/intervention-response-v2-phase-aware-protocol-20260714.md +++ b/docs/intervention-response-v2-phase-aware-protocol-20260714.md @@ -58,6 +58,9 @@ The pilot is a mechanism gate, not paper evidence. 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. +- A measured Layer-1 interval is complete only when its start-boundary, + end-boundary, and maximum internal record gaps are each at most one second; + timestamps must be monotonic. - 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, @@ -71,14 +74,16 @@ 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. +response exceeds same-config repeat noise at the same pair of consecutive +checkpoints under the unchanged v1 response thresholds. Those features must +also 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. +Decision evidence additionally requires both action-efficacy classes and at +least one of the phase-stable mechanism features to reach +leave-one-repetition-out balanced accuracy at least 0.75 and exceed the best +external prefix outcome by at least 0.15 at two adjacent predeclared +checkpoints from 25% onward. 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. diff --git a/runs/intervention-response-v2/analyze_pilot.py b/runs/intervention-response-v2/analyze_pilot.py index edbfbe1..7cd539a 100644 --- a/runs/intervention-response-v2/analyze_pilot.py +++ b/runs/intervention-response-v2/analyze_pilot.py @@ -19,6 +19,7 @@ SCHEMA = "intervention-response-phase-aware-pilot-analysis-v2" EXPECTED_ACTION_PAIRS = 9 EXPECTED_REPEAT_PAIRS = 12 MIN_EFFICACY_CLASS = 3 +MAX_LAYER1_GAP_S = 1.0 def _load_p1(): @@ -54,6 +55,7 @@ def _trial_record( requests_path: Path, state: dict[str, float], outcome: dict[str, float], + telemetry_coverage: dict[str, float], ) -> dict[str, Any]: return { "trial_id": str(result_path.relative_to(run_root)), @@ -74,9 +76,34 @@ def _trial_record( "early_stopped": bool(result["early_stopped"]), "state": state, "outcome": outcome, + "telemetry_coverage": telemetry_coverage, } +def telemetry_coverage( + records: list[dict[str, Any]], *, start_ns: int, end_ns: int +) -> dict[str, float]: + layer1 = [record for record in records if "step_index" in record] + timestamps = [int(record["submit_mono_ns"]) for record in layer1] + if timestamps != sorted(timestamps): + raise ValueError("Layer-1 timestamps are not monotonic") + selected = [timestamp for timestamp in timestamps if start_ns <= timestamp <= end_ns] + if not selected: + raise ValueError("Layer-1 coverage interval contains no records") + internal_gaps = [ + (right - left) / 1e9 + for left, right in zip(selected, selected[1:], strict=False) + ] + coverage = { + "start_gap_s": (selected[0] - start_ns) / 1e9, + "end_gap_s": (end_ns - selected[-1]) / 1e9, + "max_internal_gap_s": max(internal_gaps, default=0.0), + } + if any(value > MAX_LAYER1_GAP_S for value in coverage.values()): + raise ValueError(f"Layer-1 coverage gap exceeds {MAX_LAYER1_GAP_S}s: {coverage}") + return coverage + + def validate_result_against_manifest( *, result: Mapping[str, Any], @@ -147,11 +174,16 @@ def load_interval_trials( start_ns = int(result["interval"]["start_mono_ns"]) for interval in intervals_s: start_s, end_s = interval + interval_start_ns = start_ns + int(start_s * 1e9) + interval_end_ns = start_ns + int(end_s * 1e9) + coverage = telemetry_coverage( + stream, start_ns=interval_start_ns, end_ns=interval_end_ns + ) 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), + start_ns=interval_start_ns, + end_ns=interval_end_ns, request_count=int(result["selection"]["count"]), ) ) @@ -166,6 +198,7 @@ def load_interval_trials( requests_path=requests_path, state=state, outcome=outcome, + telemetry_coverage=coverage, ) ) return by_interval, streams @@ -221,6 +254,31 @@ def analyze_window( * float(trial["outcome"]["completed_over_admitted"]) for trial in trials ] + state_vectors = { + tuple(round(float(trial["state"][feature]), 12) for feature in P1.V0.ALL_FEATURES) + for trial in trials + } + per_cell_vectors: dict[str, set[tuple[float, ...]]] = defaultdict(set) + for trial in trials: + per_cell_vectors[str(trial["cell"])].add( + tuple( + round(float(trial["state"][feature]), 12) + for feature in P1.V0.ALL_FEATURES + ) + ) + ratio_features = ( + "prefill_token_fraction", + "kv_usage_mean", + "kv_usage_max", + "graph_none_share", + "graph_full_share", + "graph_padding_fraction", + ) + nonnegative_features = tuple( + feature + for feature in P1.V0.ALL_FEATURES + if feature != "kv_usage_end_minus_start" + ) action_metadata = [ { "group": pair["group"], @@ -243,6 +301,25 @@ def analyze_window( for value in pair[key].values() ), "all_results_uncensored": all(not trial["early_stopped"] for trial in trials), + "state_vectors_not_all_identical": len(state_vectors) > 1, + "per_cell_state_vectors_not_all_identical": all( + len(vectors) > 1 for vectors in per_cell_vectors.values() + ), + "ratios_bounded": all( + 0.0 <= float(trial["state"][feature]) <= 1.0 + for trial in trials + for feature in ratio_features + ), + "nonnegative_counters": all( + float(trial["state"][feature]) >= 0.0 + for trial in trials + for feature in nonnegative_features + ), + "layer1_boundary_and_internal_gaps_bounded": all( + value <= MAX_LAYER1_GAP_S + for trial in trials + for value in trial["telemetry_coverage"].values() + ), } return { "start_s": start_s, @@ -255,6 +332,21 @@ def analyze_window( "action_pairs": len(actions), "repeat_pairs": len(repeats), "actions": action_metadata, + "trial_sanity": [ + { + "trial_id": trial["trial_id"], + "cell": trial["cell"], + "level": trial["level"], + "replicate": trial["replicate"], + "admitted_fraction": trial["outcome"]["admitted_fraction"], + "completed_fraction": ( + trial["outcome"]["admitted_fraction"] + * trial["outcome"]["completed_over_admitted"] + ), + "telemetry_coverage": trial["telemetry_coverage"], + } + for trial in trials + ], "response_statistics": response, "qualifying_response_features": response_qualifying, "efficacy": { @@ -290,6 +382,20 @@ def stable_adjacent_features(windows: list[dict[str, Any]]) -> dict[str, list[st return result +def stable_adjacent_efficacy_features( + windows: list[dict[str, Any]], +) -> dict[str, list[str]]: + eligible = [window for window in windows if window["end_fraction"] >= 0.25] + result = {} + for left, right in zip(eligible, eligible[1:], strict=False): + key = f"{left['end_fraction']:.2f}->{right['end_fraction']:.2f}" + result[key] = sorted( + set(left["efficacy"]["telemetry_qualifying_features"]) + & set(right["efficacy"]["telemetry_qualifying_features"]) + ) + return result + + def consistent_load_regimes( windows: list[dict[str, Any]], stable: dict[str, list[str]] ) -> dict[str, Any]: @@ -322,6 +428,96 @@ def consistent_load_regimes( return result +def mechanism_gate( + stable: Mapping[str, list[str]], load_consistency: Mapping[str, Mapping[str, Any]] +) -> dict[str, Any]: + by_transition = {} + for transition, features in stable.items(): + qualifying = sorted( + feature + for feature in features + if load_consistency[f"{transition}:{feature}"]["passes_two_regimes"] + ) + by_transition[transition] = qualifying + passing_transitions = sorted( + transition + for transition, features in by_transition.items() + if len(features) >= 2 + ) + return { + "minimum_features": 2, + "by_transition": by_transition, + "passing_transitions": passing_transitions, + "passes": bool(passing_transitions), + } + + +def controller_gate( + run_root: Path, manifest: Mapping[str, Any] +) -> dict[str, Any]: + path = run_root / "controller-state.json" + state = json.loads(path.read_text(encoding="utf-8")) + expected_sessions = {str(session["session"]) for session in manifest["sessions"]} + actual_sessions = set(state.get("sessions", {})) + session_invariants = [ + passed + for session in state.get("sessions", {}).values() + for passed in session.get("validation", {}).get("invariants", {}).values() + ] + invariants = { + "controller_complete": state.get("status") == "complete", + "completed_session_count": int(state.get("completed_sessions", -1)) + == len(expected_sessions), + "exact_session_set": actual_sessions == expected_sessions, + "all_sessions_complete": all( + session.get("status") == "complete" + for session in state.get("sessions", {}).values() + ), + "no_controller_failures": not state.get("failures"), + "under_h20_hour_cap": float(state.get("gpu_hours_total", math.inf)) + <= float(manifest["budget"]["hard_cap_h20_hours"]), + "all_stream_validation_invariants_pass": bool(session_invariants) + and all(session_invariants), + } + return { + "path": str(path.resolve()), + "sha256": sha256_file(path), + "gpu_hours_total": float(state.get("gpu_hours_total", math.nan)), + "invariants": invariants, + "red_flags": [name for name, passed in invariants.items() if not passed], + } + + +def cumulative_coverage_gate(windows: list[dict[str, Any]]) -> dict[str, Any]: + trajectories: dict[str, list[tuple[float, float]]] = defaultdict(list) + trial_sets = [] + for window in windows: + trial_sets.append({str(trial["trial_id"]) for trial in window["trial_sanity"]}) + for trial in window["trial_sanity"]: + trajectories[str(trial["trial_id"])].append( + ( + float(trial["admitted_fraction"]), + float(trial["completed_fraction"]), + ) + ) + invariants = { + "same_trials_at_every_checkpoint": bool(trial_sets) + and all(trial_set == trial_sets[0] for trial_set in trial_sets[1:]), + "admitted_fraction_monotonic": all( + all(right[0] + 1e-12 >= left[0] for left, right in zip(values, values[1:])) + for values in trajectories.values() + ), + "completed_fraction_monotonic": all( + all(right[1] + 1e-12 >= left[1] for left, right in zip(values, values[1:])) + for values in trajectories.values() + ), + } + return { + "invariants": invariants, + "red_flags": [name for name, passed in invariants.items() if not passed], + } + + 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": @@ -354,33 +550,34 @@ def audit(*, run_root: Path, manifest_path: Path, output_path: Path) -> dict[str ] stable = stable_adjacent_features(cumulative) load_consistency = consistent_load_regimes(cumulative, stable) + mechanism = mechanism_gate(stable, load_consistency) mechanism_features = sorted( { - key.split(":", 1)[1] - for key, item in load_consistency.items() - if item["passes_two_regimes"] + feature + for transition in mechanism["passing_transitions"] + for feature in mechanism["by_transition"][transition] } ) full = cumulative[-1] - efficacy_features = sorted( - set.intersection( - *( - set(window["efficacy"]["telemetry_qualifying_features"]) - for window in cumulative - if window["end_fraction"] >= 0.25 - ) - ) + efficacy_stable = stable_adjacent_efficacy_features(cumulative) + efficacy_candidates = sorted( + {feature for features in efficacy_stable.values() for feature in features} ) + efficacy_features = sorted(set(efficacy_candidates) & set(mechanism_features)) + controller = controller_gate(run_root, manifest) + coverage = cumulative_coverage_gate(cumulative) red_flags = sorted( { flag for window in [*cumulative, *quarter_blocks] for flag in window["sanity"]["red_flags"] } + | set(controller["red_flags"]) + | set(coverage["red_flags"]) ) if red_flags: decision = "STOP_DATA_INVALID" - elif not mechanism_features: + elif not mechanism["passes"]: decision = "STOP_NO_PHASE_STABLE_RESPONSE" elif not full["efficacy"]["label_balance_sufficient"]: decision = "MECHANISM_ONLY_NO_LABEL_BALANCE" @@ -394,9 +591,12 @@ def audit(*, run_root: Path, manifest_path: Path, output_path: Path) -> dict[str "decision": decision, "claim_boundary": "Development mechanism pilot; not a held-out paper claim.", "mechanism_features": mechanism_features, + "mechanism_gate": mechanism, "stable_adjacent_features": stable, "load_consistency": load_consistency, "stable_incremental_efficacy_features": efficacy_features, + "stable_incremental_efficacy_candidates": efficacy_candidates, + "stable_adjacent_efficacy_features": efficacy_stable, "cumulative": cumulative, "quarter_blocks": quarter_blocks, "provenance": { @@ -410,6 +610,8 @@ def audit(*, run_root: Path, manifest_path: Path, output_path: Path) -> dict[str "sanity": { "streams": len(streams), "stream_bytes": numeric(item["bytes"] for item in streams), + "controller": controller, + "cumulative_coverage": coverage, "red_flags": red_flags, }, } diff --git a/runs/intervention-response-v2/test_analysis.py b/runs/intervention-response-v2/test_analysis.py index 932dec4..03ef524 100644 --- a/runs/intervention-response-v2/test_analysis.py +++ b/runs/intervention-response-v2/test_analysis.py @@ -129,6 +129,70 @@ def main() -> None: ] ) assert stable == {"0.10->0.25": ["queue"], "0.25->0.50": ["queue"]} + load_consistency = { + "0.10->0.25:queue": {"passes_two_regimes": True}, + "0.25->0.50:queue": {"passes_two_regimes": True}, + } + mechanism = pilot_analysis.mechanism_gate(stable, load_consistency) + assert mechanism["passes"] is False + stable["0.25->0.50"].append("kv") + load_consistency["0.25->0.50:kv"] = {"passes_two_regimes": True} + mechanism = pilot_analysis.mechanism_gate(stable, load_consistency) + assert mechanism["passes"] is True + assert mechanism["passing_transitions"] == ["0.25->0.50"] + efficacy = pilot_analysis.stable_adjacent_efficacy_features( + [ + { + "end_fraction": 0.1, + "efficacy": {"telemetry_qualifying_features": ["early"]}, + }, + { + "end_fraction": 0.25, + "efficacy": {"telemetry_qualifying_features": ["queue"]}, + }, + { + "end_fraction": 0.5, + "efficacy": {"telemetry_qualifying_features": ["kv", "queue"]}, + }, + ] + ) + assert efficacy == {"0.25->0.50": ["queue"]} + coverage = pilot_analysis.telemetry_coverage( + [ + {"step_index": 1, "submit_mono_ns": 100_000_000}, + {"step_index": 2, "submit_mono_ns": 200_000_000}, + ], + start_ns=0, + end_ns=300_000_000, + ) + assert coverage == { + "start_gap_s": 0.1, + "end_gap_s": 0.1, + "max_internal_gap_s": 0.1, + } + coverage_gate = pilot_analysis.cumulative_coverage_gate( + [ + { + "trial_sanity": [ + { + "trial_id": "a", + "admitted_fraction": 0.25, + "completed_fraction": 0.2, + } + ] + }, + { + "trial_sanity": [ + { + "trial_id": "a", + "admitted_fraction": 0.5, + "completed_fraction": 0.4, + } + ] + }, + ] + ) + assert coverage_gate["red_flags"] == [] print("phase-aware intervention response v2 analysis: PASS")