Let heartbeat benchmarks use their full remaining phase budget

benchmark/runtime/RUNTIME.md

@@ -359,13 +359,22 @@ state or force the agent active. An unresolved Turn must be recovered before
 another phase can replace its task input. These wait/recovery rules apply to
 both entry policies; they correct the earlier unconditional phase reset.
 Every scheduler wake caps its host timeout against the remaining phase budget
-before opening an execution. If only startup and settlement reserve remains,
-it records a budget-exhausted no-op without creating a pending Turn. In the
+before opening an execution. Heartbeat workers may use the entire positive
+remainder, including a continuation shorter than 160 seconds. Their bounded
+process cleanup is not a solver-time deduction. The adapters do not derive an
+independent wake limit from the initial trial budget; an explicitly supplied
+Harbor wake limit remains an independent limit and retains its failure semantics.
+Native Goal and managed-turn transports retain their existing reserve.
+When the applicable work allowance is exhausted, the worker records a
+budget-exhausted no-op without creating a pending Turn. In the
 heartbeat/turn scheduler path, that receipt exits the wake with code 75 and the
 configured shell worker stops normally. It does not re-admit empty wakes, mark
 the task complete or spend quota. Direct plain/native-Goal calls retain exit 0
 for normal budget exhaustion.
 The deadline uses the task environment's clock, including remote Harbor backends.
+Budget exhaustion is distinct from task completion: this clock does not override
+provider failures, Stop, blocked planning, or a valid terminal LoopX decision.
+Campaign duration coverage must still classify an early exit as missing time.
 
 To compare entry policies, hold the execution mode, session policy, model,
 effort, tools, feedback and total budget fixed, and use separate trials:

benchmark/runtime/harbor.py

@@ -23,6 +23,7 @@
 from .codex_offline import MODEL_CATALOG_REMOTE, CodexOffline, stage_portable_python
 from .codex import DEFAULT_REPLAN_AFTER_TURNS, Execution, turn_envelope_receipt
 from .layout import _ROOT, _SRC, _CONTROL, _CODEX_HOME, _REGISTRY, _LOOPX_RUNTIME
+from .scheduler import phase_work_reserve
 
 
 _PYTHON = f"{_ROOT}/python"
@@ -94,7 +95,7 @@ def __init__(
             execution_mode,
             iteration_context,
             codex_sandbox,
-            float(scheduler_timeout_sec) - 160 if turn_timeout_sec is None else float(turn_timeout_sec),
+            None if turn_timeout_sec is None else float(turn_timeout_sec),
             validation_command if validation_command is not None else (),
             task_entry,
             turn_envelope,
@@ -110,7 +111,9 @@ def __init__(
                 "Remove planning_timeout_sec; planning shares scheduler_timeout_sec with execution"
             )
         self.scheduler_timeout = int(scheduler_timeout_sec)
-        if self.scheduler_timeout <= self.execution.timeout_seconds + 150:
+        if self.scheduler_timeout <= 0:
+            raise ValueError("scheduler timeout must be positive")
+        if self.execution.timeout_seconds is not None and self.scheduler_timeout <= self.execution.timeout_seconds + 150:
             raise ValueError(
                 "scheduler timeout must exceed turn timeout plus cleanup allowance"
             )
@@ -645,7 +648,8 @@ def _worker_env(self, *, cwd: str) -> dict[str, str]:
                 "LOOPX_VALIDATION_COMMAND_JSON": json.dumps(
                     self.execution.validation_command
                 ),
-                "LOOPX_CODEX_TURN_TIMEOUT_SEC": str(self.execution.timeout_seconds),
+                **({"LOOPX_CODEX_TURN_TIMEOUT_SEC": str(self.execution.timeout_seconds)}
+                   if self.execution.timeout_seconds is not None else {}),
                 "CODEX_BIN": "/usr/local/bin/codex",
                 "MODEL_NAME": (self.model_name or "").split("/", 1)[-1],
                 "REASONING_EFFORT": str(
@@ -811,7 +815,7 @@ async def run(
             if self.execution.task_entry == "loopx-planned":
                 result_path = f"{_CONTROL}/planning-phase-{self._phase_number:03d}.json"
                 remaining = int(deadline - time.monotonic())
-                if remaining <= 160:
+                if remaining <= phase_work_reserve(self.execution.mode):
                     raise TimeoutError("Task budget exhausted before planning")
                 await self.exec_as_agent(
                     environment,
@@ -834,13 +838,14 @@ async def run(
                 if entry["status"] == "blocked":
                     return
             remaining = int(deadline - time.monotonic())
-            if remaining <= 160:
+            reserve = phase_work_reserve(self.execution.mode)
+            if remaining <= reserve:
                 raise TimeoutError("Task budget exhausted before execution handoff")
-            # Planning consumes the phase budget, including when the host later resumes.
-            # Keep ten seconds for scheduler startup before the worker checks
-            # its execution window plus the existing 150-second settlement reserve.
-            host_timeout = min(self.execution.timeout_seconds, remaining - 160)
-            env["LOOPX_CODEX_TURN_TIMEOUT_SEC"] = str(host_timeout)
+            # Planning, admission and work share one absolute phase clock.
+            host_timeout = remaining - reserve
+            if self.execution.timeout_seconds is not None:
+                host_timeout = min(self.execution.timeout_seconds, host_timeout)
+                env["LOOPX_CODEX_TURN_TIMEOUT_SEC"] = str(host_timeout)
             from .scheduler import worker_command
 
             command = worker_command(

benchmark/runtime/provider_request_admission.py

@@ -50,7 +50,8 @@ class ProviderRequestAdmission:
     """One immutable owner plus a one-shot process attachment."""
 
     def __init__(self, env, session_wake, receipt, wake, *, timeout=None, original_quota=None):
-        from .worker import PHASE_WORK_RESERVE_SECONDS, heartbeat_launch_check
+        from .scheduler import phase_work_reserve
+        from .worker import heartbeat_launch_check
 
         if session_wake is None or session_wake.admission is None:
             raise ValueError("provider request admission requires the original heartbeat session")
@@ -67,8 +68,9 @@ def __init__(self, env, session_wake, receipt, wake, *, timeout=None, original_q
             epoch = float(captured_env["LOOPX_PHASE_DEADLINE_EPOCH"])
             if not math.isfinite(epoch):
                 raise ValueError("provider phase deadline must be finite")
+            reserve = phase_work_reserve(captured_env.get("LOOPX_EXECUTION_MODE", "heartbeat"))
             deadline = min(deadline,
-                time.monotonic() + epoch - PHASE_WORK_RESERVE_SECONDS - time.time())
+                time.monotonic() + epoch - reserve - time.time())
         self._invocation = _Invocation(
             session_wake.admission.with_launch_decision(check), captured_env["LOOPX_TURN"],
             session_wake.session_id, deadline)

benchmark/runtime/scheduler.py

@@ -5,6 +5,12 @@
 # Private process transport between the benchmark wake and its shell scheduler.
 # The benchmark worker remains the owner of phase-budget exhaustion.
 BUDGET_EXHAUSTED_EXIT_CODE = 75
+PHASE_WORK_RESERVE_SECONDS = 160
+
+
+def phase_work_reserve(mode: str) -> int:
+    """Heartbeat cleanup must not consume the remaining solver allowance."""
+    return 0 if mode == "heartbeat" else PHASE_WORK_RESERVE_SECONDS
 
 
 def worker_command(env: dict[str, str], *, python: str, source: str,

benchmark/runtime/sforge.py

@@ -26,7 +26,6 @@
     BenchmarkCodex, _CONTROL, _CODEX_HOME, _PYTHON, _SCHEDULER_STATE, _SRC,
 )
 from .scheduler import worker_command
-from .worker import PHASE_WORK_RESERVE_SECONDS
 
 
 # Experiment profiles, local to the research runner; not new LoopX modes.
@@ -125,8 +124,8 @@ def __init__(self, config, *, profile: str, cwd: str,
             if (os.environ.get("CODEX_WIRE_API") or "responses") != "responses":
                 raise ValueError("provider_request_admission requires the Responses wire API")
         self.provider_request_admission = provider_request_admission
-        if timeout_seconds <= PHASE_WORK_RESERVE_SECONDS:
-            raise ValueError("Worker budget must exceed the 160s startup/settlement reserve")
+        if timeout_seconds <= 0:
+            raise ValueError("Worker budget must be positive")
         if not config.agent_model or not config.agent_effort:
             raise ValueError("Explicit model and reasoning effort are required")
         if config.agent_api_base_url and not config.agent_api_key:
@@ -218,9 +217,9 @@ def __init__(self, config, *, profile: str, cwd: str,
         self.instructions_installed = False
         self.prompt_installed = False
         self.timeout_seconds = timeout_seconds
-        # Let one call use the trial budget. The shared worker recomputes the
-        # remaining deadline on every continuation, retaining cleanup time.
-        self.turn_timeout = timeout_seconds - PHASE_WORK_RESERVE_SECONDS
+        # The phase owns the total budget. No separate per-wake cap is needed;
+        # the scheduler's transport watchdog may outlive bounded child cleanup.
+        self.turn_timeout = timeout_seconds
         self.runtime = None
         self.prepared = False
         # Only the official profile delegates continuation to SForge. Heartbeat
@@ -310,7 +309,7 @@ def _install_worker(self, backend, handle, log_dir, logger):
             self.runtime = BenchmarkCodex(
                 **common, execution_mode=mode,
                 iteration_context="resume" if mode == "heartbeat" else "fresh",
-                turn_timeout_sec=self.turn_timeout,
+                turn_timeout_sec=None,
                 scheduler_timeout_sec=self.timeout_seconds,
                 task_entry=self.task_entry,
                 turn_envelope=self.turn_envelope,

benchmark/runtime/sforge_entry.py

@@ -11,7 +11,8 @@
 import time
 
 from .planning import task_plan_packet, validate_plan_readback
-from .worker import PHASE_WORK_RESERVE_SECONDS, loopx_command, run_once
+from .scheduler import phase_work_reserve
+from .worker import loopx_command, run_once
 
 # One planning wake whose native process exits non-zero without a harness
 # error or a validated plan (a dropped model stream, a failed turn) must not
@@ -32,7 +33,8 @@ def prepare_entry(env: dict[str, str]) -> bool:
     remaining = float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) - time.time()
     if not math.isfinite(remaining):
         raise ValueError("Invalid planning budget")
-    if remaining <= PHASE_WORK_RESERVE_SECONDS:
+    reserve = phase_work_reserve(env.get("LOOPX_EXECUTION_MODE", "heartbeat"))
+    if remaining <= reserve:
         return False
     path = Path(env["LOOPX_PLANNING_RESULT"])
     if path.exists():
@@ -67,7 +69,7 @@ def prepare_entry(env: dict[str, str]) -> bool:
                          "return_code": receipt.get("return_code"),
                          **({"error_kind": receipt["error_kind"]} if "error_kind" in receipt else {})})
         remaining = float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) - time.time()
-        if not _retryable_planning_wake(receipt) or remaining <= PHASE_WORK_RESERVE_SECONDS:
+        if not _retryable_planning_wake(receipt) or remaining <= reserve:
             break
     path.with_name("planning-attempts.json").write_text(json.dumps(attempts))
     if receipt.get("ok") is not True:
@@ -91,7 +93,7 @@ def cancelled(signum, frame):
         # Native process completion must not qualify it as a finished trial.
         return 1
     # A successful plan may itself consume the remaining execution allowance.
-    if float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) - time.time() <= PHASE_WORK_RESERVE_SECONDS:
+    if float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) - time.time() <= phase_work_reserve(env.get("LOOPX_EXECUTION_MODE", "heartbeat")):
         return 1
     # Native SForge owns environment filtering and the outer phase deadline;
     # retain its proxy/isolation inputs rather than launching from setup hooks.

benchmark/runtime/worker.py

@@ -16,7 +16,8 @@
 
 from benchmark.runtime.codex import Execution, prepare_codex_home, process_environment
 
-PHASE_WORK_RESERVE_SECONDS = 160
+from benchmark.runtime.scheduler import PHASE_WORK_RESERVE_SECONDS as PHASE_WORK_RESERVE_SECONDS
+from benchmark.runtime.scheduler import phase_work_reserve
 
 
 def read_json(text: str) -> dict:
@@ -556,16 +557,18 @@ def run_once(env: dict[str, str]) -> dict:
     try:
         if stage == "execute" and env.get("LOOPX_PHASE_DEADLINE_EPOCH"):
             remaining = float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) - time.time()
-            # Reserve startup and settlement on every wake, then allow the
-            # remaining time for work instead of reusing the initial timeout.
-            if remaining <= PHASE_WORK_RESERVE_SECONDS:
+            # Heartbeat work uses the whole remaining phase. Cleanup is bounded
+            # by its existing host transport, not subtracted from each wake.
+            # Native Goal and managed-turn transports retain their own reserve.
+            reserve = phase_work_reserve(execution.mode)
+            if remaining <= reserve:
                 receipt.update(ok=True, budget_exhausted=True, host_invoked=False)
                 return receipt
             if execution.mode == "heartbeat":
-                phase_work_deadline = float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) - PHASE_WORK_RESERVE_SECONDS
+                phase_work_deadline = float(env["LOOPX_PHASE_DEADLINE_EPOCH"])
             execution = replace(
-                execution, timeout_seconds=(remaining - PHASE_WORK_RESERVE_SECONDS if execution.timeout_seconds is None
-                                             else min(execution.timeout_seconds, remaining - PHASE_WORK_RESERVE_SECONDS))
+                execution, timeout_seconds=(remaining - reserve if execution.timeout_seconds is None
+                                             else min(execution.timeout_seconds, remaining - reserve))
             )
         prepare_codex_home(
             home,

benchmark/tests/test_provider_request_admission.py

@@ -43,7 +43,8 @@ class _Authority:
     root: Path
     api_key: str = "fixture-provider-key"
 
-    def invocation(self, upstream, name, *, timeout=None, receipt=None, wake=None, turn_id=None):
+    def invocation(self, upstream, name, *, timeout=None, receipt=None, wake=None, turn_id=None,
+                   extra_env=None):
         if wake is None:
             wake = self.root / name
             wake.mkdir()
@@ -56,6 +57,8 @@ def invocation(self, upstream, name, *, timeout=None, receipt=None, wake=None, t
             "OPENAI_BASE_URL": upstream,
             "OPENAI_API_KEY": self.api_key,
         }
+        if extra_env is not None:
+            env.update(extra_env)
         session = SimpleNamespace(
             admission=FirstPartyHostGoalAdmission.for_plan(
                 registry_path=self.registry, goal_id=quota_fixture.GOAL_ID,
@@ -290,6 +293,23 @@ def observed_transaction(*args, **kwargs):
             assert not upstream.rows and _finished(wake)[0]["forwarded"] is False
 
 
+def test_heartbeat_short_phase_uses_the_same_zero_reserve_as_worker(authority):
+    with _upstream() as upstream, _process() as process:
+        guard, receipt, wake = authority.invocation(
+            upstream.base_url, "short-heartbeat",
+            extra_env={
+                "LOOPX_EXECUTION_MODE": "heartbeat",
+                "LOOPX_PHASE_DEADLINE_EPOCH": str(time.time() + 140),
+            },
+        )
+        with guard.serve(["native", "-"]):
+            guard.started(process)
+            assert _post(guard.base_url)[0] == 200
+            assert len(upstream.rows) == 1
+            assert _finished(wake)[0]["forwarded"] is True
+            assert receipt["provider_request_invocations"][0]["sent"] == 1
+
+
 def test_process_attachment_is_one_shot(authority):
     with _upstream() as upstream, _process() as old, _process() as replacement:
         guard, _receipt, _wake = authority.invocation(upstream.base_url, "one-shot")

benchmark/tests/test_sforge_runtime.py

@@ -130,8 +130,8 @@ def test_invalid_worker_inputs_fail_before_install(monkeypatch):
     config = SForgeConfig(agent_model="fixture-model", agent_effort="xhigh")
     with pytest.raises(ValueError, match="Unknown"):
         SForgeWorker(config, profile="unknown", cwd="/task")
-    with pytest.raises(ValueError, match="160s"):
-        SForgeWorker(config, profile="single", cwd="/task", timeout_seconds=160)
+    with pytest.raises(ValueError, match="positive"):
+        SForgeWorker(config, profile="single", cwd="/task", timeout_seconds=0)
     monkeypatch.delenv("CODEX_AUTH_JSON_PATH", raising=False)
     with pytest.raises(ValueError, match="credential"):
         SForgeWorker(config, profile="single", cwd="/task")
@@ -386,8 +386,8 @@ async def installed(self, environment):
 
 
 @pytest.mark.parametrize("profile,total,expected", [
-    ("native-goal", 64800, 64640), ("native-goal", 1800, 1640),
-    ("heartbeat-resume", 64800, 64640), ("heartbeat-explore", 64800, 64640),
+    ("native-goal", 64800, 64800), ("native-goal", 1800, 1800),
+    ("heartbeat-resume", 140, 140), ("heartbeat-explore", 64800, 64800),
 ])
 @pytest.mark.parametrize("turns", [None, 3])
 def test_native_goal_and_heartbeat_use_trial_budget_without_independent_wake_limit(
@@ -410,7 +410,7 @@ async def installed(self, environment):
                           replan_after_turns=turns)
     worker.install_stop_hook(None, None, tmp_path, None)
     env = worker.runtime._worker_env(cwd="/task")
-    assert float(env["LOOPX_CODEX_TURN_TIMEOUT_SEC"]) == expected
+    assert "LOOPX_CODEX_TURN_TIMEOUT_SEC" not in env
     assert worker.runtime.scheduler_timeout == total
     expected_cadence = ({"replan_after_effective_turns": turns or 6}
                         if profile.startswith("heartbeat-") else
@@ -1072,7 +1072,7 @@ async def installed(self, environment):
         profile=profile, cwd="/task", timeout_seconds=1800)
     worker.install_stop_hook(None, None, tmp_path, None)
     assert worker.runtime.scheduler_timeout == 1800
-    assert worker.turn_timeout == 1640
+    assert worker.turn_timeout == 1800
     receipt = json.loads((tmp_path / "worker-profile.json").read_text())
     assert receipt["planning_timeout_seconds"] is None
     assert receipt["timeout_seconds"] == 1800

benchmark/tests/test_shared_codex_runtime.py

@@ -871,7 +871,8 @@ def test_harbor_default_execution_budget_tracks_total_trial(tmp_path, total):
     from benchmark.runtime.harbor import BenchmarkCodex
     agent = BenchmarkCodex(logs_dir=tmp_path, model_name="fixture",
                            scheduler_timeout_sec=total)
-    assert agent.execution.timeout_seconds == total - 160
+    assert agent.execution.timeout_seconds is None
+    assert "LOOPX_CODEX_TURN_TIMEOUT_SEC" not in agent._worker_env(cwd="/task")
     assert agent.scheduler_timeout == total