diff --git a/docs/opprof/oracle-gap-protocol.md b/docs/opprof/oracle-gap-protocol.md index ac37f0e..602fdaa 100644 --- a/docs/opprof/oracle-gap-protocol.md +++ b/docs/opprof/oracle-gap-protocol.md @@ -1,11 +1,32 @@ # Static-policy oracle-gap protocol -Status: **FROZEN WITH A-OG-1, A-OG-2, AND A-OG-3 AMENDMENTS**. +Status: **FROZEN WITH A-OG-1 THROUGH A-OG-4 AMENDMENTS**. Date frozen: 2026-07-13 (Asia/Singapore). Existing Phase-3 measurements were inspected only to choose the workload pair and rate brackets. They are exploratory calibration data, not primary observations in this protocol. +### A-OG-4 — close a majority-shifted boundary (before confirmation scores) + +After all four primary frontiers and 70 scores were complete, the controller +was interrupted before the first confirmation produced a score. The partial +`P01-C10-r26-rep1` attempt contained no result, sanity file, or score and is +archived separately. This amendment is therefore blind to confirmation +outcomes. + +The original confirmation schedule still runs first. Afterward, each cell's +majority-vote frontier is recomputed. If either side of the *actual* final +boundary has fewer than three trials because the provisional boundary moved, +the controller runs only enough repetitions at that existing rate to reach +three, high-to-low, on a fresh server for that config. It then recomputes the +boundary and repeats for at most three closure rounds. No new rate anchor may +be added. Failure to obtain a monotone, bracketed boundary with three trials +per side within three rounds or the 6 H20-hour cap makes the experiment +inconclusive. + +This fills a stopping-rule omission: it does not change any existing score, +majority rule, phase, config, rate grid, SLO, or oracle calculation. + ### A-OG-3 — freeze a per-logical-trial transport retry rule (after 62 scores) The first attempt at `P06-C10-r1.9-rep0` reproduced the same isolated local diff --git a/scripts/oracle_gap/run_frontier.py b/scripts/oracle_gap/run_frontier.py index 05ea898..74a7a09 100644 --- a/scripts/oracle_gap/run_frontier.py +++ b/scripts/oracle_gap/run_frontier.py @@ -16,14 +16,21 @@ import urllib.request from pathlib import Path from typing import Any -from analyze import CONFIGS, PHASES, atomic_json, score_trial, summarize_trials +from analyze import ( + CONFIGS, + PHASES, + atomic_json, + frontier_for_cell, + score_trial, + summarize_trials, +) SCHEMA = 1 -AMENDMENT = "A-OG-3" +AMENDMENT = "A-OG-4" AMENDMENT_REASON = ( - "apply the frozen per-logical-trial quarantine rule to an isolated " - "local transport-invalid attempt" + "close any majority-shifted final boundary to three trials per side " + "without adding rate anchors" ) REMOTE_ROOT = Path("/home/admin/cpfs/wjh/oracle-gap-20260713") RUN_ROOT = REMOTE_ROOT / "runs" @@ -39,6 +46,7 @@ GPU = 0 CPU_MASK = "0-19" PORT = 8820 GPU_HOUR_LIMIT = 6.0 +MAX_CLOSURE_ROUNDS = 3 PRIMARY_ORDER = ("C11", "C00", "C01", "C10") CONFIRM_ORDER = tuple(reversed(PRIMARY_ORDER)) CONFIG_DETAILS = { @@ -640,6 +648,136 @@ def run_confirm_config(state: dict[str, Any], config: str) -> None: save_state(state) +def load_cell_scores(phase: str, config: str) -> list[dict[str, Any]]: + paths = sorted((RUN_ROOT / "trials" / f"{phase}-{config}").glob("**/score.json")) + return [json.loads(path.read_text()) for path in paths] + + +def boundary_closure_needs(rows: list[dict[str, Any]]) -> list[float]: + frontier = frontier_for_cell(rows) + if not frontier["monotone"]: + raise RuntimeError("cannot close a non-monotone frontier") + lower = frontier["lower_feasible_rps"] + upper = frontier["upper_infeasible_rps"] + if lower is None or upper is None: + raise RuntimeError("cannot close an unbracketed frontier") + trials_by_rate = { + float(row["rate_rps"]): int(row["trials"]) + for row in frontier["rates"] + } + # Preserve the confirmation convention of running high-to-low. + return [ + float(rate) + for rate in (upper, lower) + if trials_by_rate[float(rate)] < 3 + ] + + +def run_closure_config( + state: dict[str, Any], + round_index: int, + config: str, + needs: dict[str, list[float]], +) -> None: + stage = f"closure-r{round_index}-{config}" + if state["stages"].get(stage, {}).get("status") == "complete": + return + state["stages"][stage] = { + "status": "starting", + "started_at": time.time(), + "needs": needs, + } + save_state(state) + server = start_server(config, stage, state) + failure = None + try: + for phase in PHASES: + for rate in needs.get(phase, []): + rows = load_cell_scores(phase, config) + repetitions = { + int(row["repetition"]) + for row in rows + if float(row["target_rate_rps"]) == rate + } + repetition = 0 + while len(repetitions) < 3: + while repetition in repetitions: + repetition += 1 + run_trial( + state, + server, + phase, + rate, + repetition, + "boundary-closure", + ) + repetitions.add(repetition) + except Exception as error: + failure = error + finally: + try: + stop_server(server, state) + except Exception as error: + failure = failure or error + if failure is not None: + state["stages"][stage]["status"] = "failed" + state["stages"][stage]["failure"] = repr(failure) + state["status"] = "failed" + save_state(state) + raise failure + state["stages"][stage]["status"] = "complete" + state["stages"][stage]["completed_at"] = time.time() + save_state(state) + + +def run_boundary_closure(state: dict[str, Any]) -> None: + stage = "boundary-closure" + closure = state["stages"].setdefault( + stage, {"status": "starting", "started_at": time.time(), "rounds": []} + ) + save_state(state) + for round_index in range(1, MAX_CLOSURE_ROUNDS + 1): + needs = { + config: { + phase: boundary_closure_needs(load_cell_scores(phase, config)) + for phase in PHASES + } + for config in CONFIGS + } + closure["rounds"].append({"round": round_index, "needs": needs}) + save_state(state) + if not any(rates for config in needs.values() for rates in config.values()): + closure["status"] = "complete" + closure["completed_at"] = time.time() + save_state(state) + return + for config in CONFIRM_ORDER: + config_needs = { + phase: rates for phase, rates in needs[config].items() if rates + } + if not config_needs: + continue + if float(state["gpu_hours"]) >= GPU_HOUR_LIMIT: + raise RuntimeError("GPU-hour hard cap reached during boundary closure") + run_closure_config(state, round_index, config, config_needs) + remaining = { + config: { + phase: boundary_closure_needs(load_cell_scores(phase, config)) + for phase in PHASES + } + for config in CONFIGS + } + closure["final_needs"] = remaining + if not any(rates for config in remaining.values() for rates in config.values()): + closure["status"] = "complete" + closure["completed_at"] = time.time() + save_state(state) + return + raise RuntimeError( + f"failed to close final boundaries in {MAX_CLOSURE_ROUNDS} rounds" + ) + + def cleanup_recorded(state: dict[str, Any]) -> None: for pgid in state.get("owned_pgids", []): try: @@ -690,6 +828,7 @@ def execute(resume: bool) -> None: if float(state["gpu_hours"]) >= GPU_HOUR_LIMIT: raise RuntimeError("GPU-hour hard cap reached before confirmations") run_confirm_config(state, config) + run_boundary_closure(state) score_paths = sorted((RUN_ROOT / "trials").glob("**/score.json")) summary = summarize_trials([json.loads(path.read_text()) for path in score_paths]) summary["gpu_hours"] = state["gpu_hours"] @@ -712,6 +851,7 @@ def plan() -> dict[str, Any]: "confirm_order": list(CONFIRM_ORDER), "primary_trials_without_extensions": primary, "confirmation_trials": confirmations, + "max_boundary_closure_rounds": MAX_CLOSURE_ROUNDS, "expected_total_trials": primary + confirmations, "expected_h20_hours": "3.0-4.0", "hard_cap_h20_hours": GPU_HOUR_LIMIT, diff --git a/tests/test_oracle_gap.py b/tests/test_oracle_gap.py index 177310a..4e1beea 100644 --- a/tests/test_oracle_gap.py +++ b/tests/test_oracle_gap.py @@ -9,7 +9,12 @@ ORACLE_GAP = Path(__file__).resolve().parents[1] / "scripts/oracle_gap" sys.path.insert(0, str(ORACLE_GAP)) from analyze import score_trial, summarize_trials # noqa: E402 -from run_frontier import AMENDMENT, UP_EXTENSIONS, resume_compatible # noqa: E402 +from run_frontier import ( # noqa: E402 + AMENDMENT, + UP_EXTENSIONS, + boundary_closure_needs, + resume_compatible, +) def _request( @@ -175,7 +180,7 @@ def test_a_og_1_extends_only_mutable_resume_fields() -> None: ) -def test_a_og_3_requires_the_amended_grid_to_stay_fixed() -> None: +def test_a_og_4_requires_the_amended_grid_to_stay_fixed() -> None: old = { "analyzer_sha256": "analyzer", "p5_client_sha256": "p5", @@ -190,9 +195,21 @@ def test_a_og_3_requires_the_amended_grid_to_stay_fixed() -> None: "up_extensions": {"P01": [38.0], "P06": [2.1, 2.2, 2.3, 2.4]}, } - assert AMENDMENT == "A-OG-3" + assert AMENDMENT == "A-OG-4" assert resume_compatible(old, {**old, "controller_sha256": "new"}) assert not resume_compatible( old, {**old, "up_extensions": {"P01": [38.0], "P06": [2.1, 2.2, 2.3, 2.5]}}, ) + + +def test_boundary_closure_follows_a_majority_shift() -> None: + rows = [] + rows.append(_frontier_row("P06", "C00", 2.3, True)) + rows.extend(_frontier_row("P06", "C00", 2.4, value) for value in (True, False, False)) + rows.extend(_frontier_row("P06", "C00", 2.5, False) for _ in range(3)) + + assert boundary_closure_needs(rows) == [2.3] + + rows.extend(_frontier_row("P06", "C00", 2.3, True) for _ in range(2)) + assert boundary_closure_needs(rows) == []