fix: hold the session to its wall-clock budget

selfcheck_gates.py

@@ -0,0 +1,124 @@
+"""Throwaway self-review: sweep the real pricing functions for disagreements."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from hyperloom.orchestrator.phases import machine_state as ms
+
+BOOT, COLD_BENCH, HOT = 350.0, 550.0, 400.0
+COLD_ROUND = BOOT + COLD_BENCH
+
+
+def state(usable, *, phase="PRELUDE", warm=HOT, post_ready=COLD_BENCH, marked=False, double=True):
+    return SimpleNamespace(
+        phase=phase,
+        max_minutes=180,
+        baseline_tput=1000.0,
+        baseline_runtime_sec=COLD_ROUND,
+        baseline_post_ready_runtime_sec=post_ready,
+        baseline_warm_runtime_sec=warm,
+        baseline_measure_round_dropped=marked,
+        baseline_double_run=double,
+        session_budget_usable_sec=lambda: usable,
+    )
+
+
+def gate1_need(s):
+    """What the pre-ignition gate demands, mirroring _round_affordable."""
+    cold = ms.measured_seconds(s, "baseline_runtime_sec")
+    rnd = ms.baseline_round_cost_sec(s, double_run=bool(s.baseline_double_run))
+    if cold is None or rnd is None:
+        return None
+    use = (ms.one_more_measurement_sec(s) or cold) if s.phase == "PRELUDE" else 0.0
+    return rnd + use
+
+
+def gate2_need(s, *, warmup_sec, warmup_post_ready):
+    """What the post-warmup gate demands, mirroring _measure_round_affordable."""
+    bench = ms.measured_seconds(s, "baseline_warm_runtime_sec")
+    if bench is None:
+        bench = warmup_post_ready
+    if bench is None or warmup_sec is None:
+        return None
+    use = (ms.one_more_measurement_sec(s) or warmup_sec) if s.phase == "PRELUDE" else 0.0
+    return bench + use
+
+
+def main() -> int:
+    bad = 0
+
+    # 1. The band: is there a budget gate 1 admits and gate 2 then certainly refuses?
+    # Gate 2 is asked after the warmup has spent a cold pass.
+    band = []
+    for usable in range(0, 6001, 10):
+        s = state(float(usable))
+        need1 = gate1_need(s)
+        admitted = need1 is not None and usable >= need1
+        if not admitted:
+            continue
+        after = state(float(usable) - COLD_ROUND)
+        need2 = gate2_need(after, warmup_sec=COLD_ROUND, warmup_post_ready=COLD_BENCH)
+        if need2 is not None and (usable - COLD_ROUND) < need2:
+            band.append(usable)
+    if band:
+        bad += 1
+        print(f"BAND      gate 1 admits and gate 2 refuses for usable in {band[0]}..{band[-1]}")
+    else:
+        print("ok        no budget is admitted before ignition only to be refused after the cold pass")
+
+    # 2. Livelock: with the mark set, does every budget either close or admit a retry?
+    stuck = []
+    for usable in range(0, 8001, 10):
+        s = state(float(usable), marked=True)
+        closes = ms.exit_cold_anchor_prelude(s) is not None
+        need1 = gate1_need(s)
+        admits = need1 is not None and usable >= need1
+        if not closes and not admits:
+            stuck.append(usable)
+    if stuck:
+        bad += 1
+        print(f"LIVELOCK  neither closes nor admits for usable in {stuck[0]}..{stuck[-1]}")
+    else:
+        print("ok        a marked session always either closes or may retry")
+
+    # 3. A session with no split measured (multi-node / scriptable shape).
+    s = state(3000.0, warm=0.0, post_ready=0.0)
+    need = gate1_need(s)
+    if need is None:
+        bad += 1
+        print("UNGATED   a round with no boot boundary is waved through")
+    else:
+        print(f"ok        a round with no split is priced at {need:.0f}s (whole cold rounds)")
+
+    # 4. A first baseline must never be judged.
+    first = SimpleNamespace(
+        phase="PRELUDE",
+        max_minutes=180,
+        baseline_tput=0.0,
+        baseline_runtime_sec=0.0,
+        baseline_post_ready_runtime_sec=0.0,
+        baseline_warm_runtime_sec=0.0,
+        baseline_measure_round_dropped=False,
+        baseline_double_run=True,
+        session_budget_usable_sec=lambda: 60.0,
+    )
+    if gate1_need(first) is not None:
+        bad += 1
+        print("PREDICTED a first baseline was priced from measurements it cannot have")
+    else:
+        print("ok        a first baseline is not judged")
+
+    # 5. Later phases ask only whether the round fits.
+    later = state(2000.0, phase="EXPLORE")
+    if gate1_need(later) != ms.baseline_round_cost_sec(later, double_run=True):
+        bad += 1
+        print("SCOPED    a re-baseline outside PRELUDE was charged for a successor")
+    else:
+        print("ok        a re-baseline outside PRELUDE pays only for itself")
+
+    return bad
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/hyperloom/agents/robustness/role/envelope.py

@@ -114,6 +114,7 @@ class IntentType(str, Enum):
         "start_ts",
         "resumed_ts",
         "max_minutes",
+        "closing_grace_sec",
         "optimization_stack",
         "gain_per_stack_entry",
         "schema_version",

src/hyperloom/inference_optimizer/actions/roofline.md

@@ -92,8 +92,11 @@ analysis task.
 
 ## Cost / runtime
 
-* `cost_minutes_p50=8` / `cost_minutes_p75=15` — dominated by profile
-  (Magpie + torch profiler overhead) + trace_analyze (TraceLens
-  subprocess); `trace_split` inside TraceLens adds < 30s.
+* `typical_runtime_min=10` — dominated by profile (Magpie + torch profiler
+  overhead) + trace_analyze (TraceLens subprocess); `trace_split` inside
+  TraceLens adds < 30s. The estimate is calibrated on small models: a field
+  session measured an 81-minute roofline, so the budget guards prefer this
+  session's own measured baseline round once one exists and fall back to this
+  number only before that.
 * `requires_lanes=[profile_lane]` — same lane as `profile` so we
   don't run two profile-class tasks concurrently against the server.

src/hyperloom/inference_optimizer/tests/conftest.py

@@ -46,6 +46,43 @@
 _bootstrap_kernel_agent_env()
 
 
+def enable_multi_node(monkeypatch, nodes: int = 2) -> None:
+    """Put the executors in multi-node mode with a no-op per-round server restart.
+
+    Multi-node is what puts a discarded client-warmup pass in front of a measured
+    round, so it is the mode in which one round launches more than one benchmark
+    process -- and the restart between them is the part that needs a cluster.
+
+    Args:
+        monkeypatch: The requesting test's monkeypatch fixture.
+        nodes: How many nodes to claim, which is what the executors read.
+    """
+    from hyperloom.orchestrator.actions.executors import _multi_node_server_lifecycle as mnl
+
+    async def _no_restart(*_args, **_kwargs) -> None:
+        return None
+
+    monkeypatch.setenv("INFERENCE_OPTIMIZER_NODES", str(nodes))
+    monkeypatch.setattr(mnl, "restart_server_for_round", _no_restart)
+
+
+def launches_by_round_slot(recorded: list[dict]) -> dict[str, dict]:
+    """Index recorded benchmark launches by the output slot each round ran in.
+
+    A round is identified by the slot it writes into rather than by its position
+    in the launch order, so a test can assert on one pass of a round without
+    encoding how many passes precede it.
+
+    Args:
+        recorded: Launch records, each carrying the ``round_slot`` name the
+            subprocess doubles stamp on every round they see.
+
+    Returns:
+        dict[str, dict]: The last launch recorded per slot name.
+    """
+    return {launch["round_slot"]: launch for launch in recorded}
+
+
 def seed_target_analysis_marker(session_dir: Path) -> Path:
     """Write a ``no_target_gpu_configured`` marker JSON at the session dir."""
     from hyperloom.inference_optimizer.session.session_paths import target_baseline_json
@@ -333,3 +370,99 @@
 
     monkeypatch.setenv("INFERENCE_OPTIMIZER_NODES", str(nodes))
     monkeypatch.setattr(mnl, "restart_server_for_round", _no_restart)
+
+
+class _RayDoubleActorClass:
+    """The ``@ray.remote`` class: ``.options(...)`` then ``.remote()`` for a handle."""
+
+    def __init__(self, cls: type) -> None:
+        self._cls = cls
+        self._options: dict = {}
+
+    def options(self, **opts):
+        self._options = dict(opts)
+        return self
+
+    def remote(self, *args, **kwargs) -> "_RayDoubleActorHandle":
+        return _RayDoubleActorHandle(self._cls(*args, **kwargs), self._options)
+
+
+class _RayDoubleActorHandle:
+    """An actor handle whose methods run in a pool sized like the real actor's.
+
+    ``max_concurrency`` is read from the options the production code passed, not
+    assumed: an actor left at Ray's single method slot gets a single-worker pool
+    here too, so a method that has to reach a call already running blocks behind
+    it exactly as it would on a real cluster.
+    """
+
+    def __init__(self, obj, options: dict) -> None:
+        from concurrent.futures import ThreadPoolExecutor
+
+        self._obj = obj
+        self._pool = ThreadPoolExecutor(max_workers=int(options.get("max_concurrency", 1) or 1))
+        self.killed = False
+
+    def __getattr__(self, name: str):
+        from types import SimpleNamespace
+
+        method = getattr(self._obj, name)
+        return SimpleNamespace(remote=lambda *a, **kw: self._pool.submit(method, *a, **kw))
+
+
+class RayDouble:
+    """A ``ray`` module stand-in that runs actor methods in real threads.
+
+    Ray is not a test dependency, and what these tests are about is what a lease
+    and its actor do to each other while work is in flight. So the transport is
+    the only thing faked: the actor is the real class, running real subprocesses,
+    and an ``ObjectRef`` is a :class:`~concurrent.futures.Future`.
+    """
+
+    class exceptions:  # noqa: N801 — mirrors the ray.exceptions namespace
+        class RayActorError(Exception):
+            pass
+
+        class RayTaskError(Exception):
+            pass
+
+        class GetTimeoutError(Exception):
+            pass
+
+    def __init__(self) -> None:
+        self.killed: list = []
+
+    def remote(self, cls: type) -> _RayDoubleActorClass:
+        return _RayDoubleActorClass(cls)
+
+    def cluster_resources(self) -> dict:
+        return {"CPU": 8.0, "GPU": 8.0, "serving_slot": 1.0}
+
+    def get(self, ref, timeout: float | None = None):
+        return ref.result(timeout)
+
+    def wait(self, refs: list, *, num_returns: int = 1, timeout: float | None = None):
+        from concurrent.futures import FIRST_COMPLETED
+        from concurrent.futures import wait as futures_wait
+
+        done, not_done = futures_wait(refs, timeout=timeout, return_when=FIRST_COMPLETED)
+        return list(done)[:num_returns], list(not_done)
+
+    def kill(self, actor) -> None:
+        actor.killed = True
+        self.killed.append(actor)
+
+
+@pytest.fixture
+def serving_lease_on_a_ray_double(monkeypatch):
+    """A real :class:`ServingLease` over :class:`RayDouble`, closed on teardown."""
+    import sys
+    from types import SimpleNamespace
+
+    from hyperloom.orchestrator.actions.executors import _ray_backend as rb
+    from hyperloom.orchestrator.actions.executors import _ray_serving as rs
+
+    monkeypatch.setitem(sys.modules, "ray", RayDouble())
+    monkeypatch.setattr(rb, "get_ray_backend", lambda: SimpleNamespace(ensure=lambda **_kw: None))
+    with rs.ServingLease(num_gpus=1) as lease:
+        yield lease

src/hyperloom/inference_optimizer/tests/test_agent_roles_and_policy.py

@@ -711,6 +711,16 @@ def test_gate_update_state_cannot_move_the_resume_boundary(gate):
     assert exc.value.rule == "state_field"
 
 
+def test_the_model_cannot_rewrite_the_budget_the_closing_reserve_leaves_it(gate):
+    """The reserve decides how much of ``max_minutes`` is spendable, so it is budget too."""
+    with pytest.raises(PolicyDenied) as exc:
+        gate.validate_intent(
+            "orchestration",
+            Intent(type=IntentType.UPDATE_STATE, payload={"changes": {"closing_grace_sec": 0.0}}),
+        )
+    assert exc.value.rule == "state_field"
+
+
 def test_gate_update_state_cannot_move_a_session_end_time(gate):
     # stop_ts is the timestamp half of stop_reason, written by the same setter:
     # locking only the reason lets a model post-date the session's end.
@@ -726,6 +736,21 @@ def test_gate_update_state_cannot_move_a_session_end_time(gate):
     assert exc.value.rule == "state_field"
 
 
+def test_a_forged_closing_reserve_would_have_spent_the_session_outright():
+    """Names what the lock prevents: one field, and the run has no usable time left."""
+    state = SharedState(session_id="s", max_minutes=100)
+    # Freeze elapsed time: two live ``session_budget_usable_sec`` reads race
+    # the clock by tens of microseconds, which is enough for ``==`` to fail.
+    state.elapsed_minutes = lambda **_kw: 90.0  # type: ignore[method-assign]
+    honest = state.session_budget_usable_sec()
+
+    applied = state.apply_changes({"closing_grace_sec": 1e9}, allow_core=False)
+
+    assert applied == {}
+    assert honest > 0.0
+    assert state.session_budget_usable_sec() == honest
+
+
 def test_core_state_fields_synced_with_robustness_envelope():
     # gate.CORE_STATE_FIELDS and the robustness
     # envelope copy must stay byte-identical. This direct assertion never skips