Close shifted oracle frontier boundaries

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,