diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 0612e59cf7..7d018f2b1d 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/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: diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py index 8b8597418a..2b49ac1945 100644 --- a/benchmark/runtime/harbor.py +++ b/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( diff --git a/benchmark/runtime/provider_request_admission.py b/benchmark/runtime/provider_request_admission.py index 41d3757757..eff68ac58b 100644 --- a/benchmark/runtime/provider_request_admission.py +++ b/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) diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py index 79e63e645b..37f9546931 100644 --- a/benchmark/runtime/scheduler.py +++ b/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, diff --git a/benchmark/runtime/sforge.py b/benchmark/runtime/sforge.py index c3b3a325c0..82df81dc0b 100644 --- a/benchmark/runtime/sforge.py +++ b/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, diff --git a/benchmark/runtime/sforge_entry.py b/benchmark/runtime/sforge_entry.py index 5e6e8c953d..ac8b57792e 100644 --- a/benchmark/runtime/sforge_entry.py +++ b/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. diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index 17b7dab84b..968539ed90 100644 --- a/benchmark/runtime/worker.py +++ b/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, diff --git a/benchmark/tests/test_provider_request_admission.py b/benchmark/tests/test_provider_request_admission.py index ceaa80b4d9..7a2b3bf064 100644 --- a/benchmark/tests/test_provider_request_admission.py +++ b/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") diff --git a/benchmark/tests/test_sforge_runtime.py b/benchmark/tests/test_sforge_runtime.py index 1a4ed9b521..e35ec3374c 100644 --- a/benchmark/tests/test_sforge_runtime.py +++ b/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 diff --git a/benchmark/tests/test_shared_codex_runtime.py b/benchmark/tests/test_shared_codex_runtime.py index 1235f89639..e82124dd98 100644 --- a/benchmark/tests/test_shared_codex_runtime.py +++ b/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 diff --git a/benchmark/tests/test_task_entry.py b/benchmark/tests/test_task_entry.py index 6cce32fab3..3ba270be27 100644 --- a/benchmark/tests/test_task_entry.py +++ b/benchmark/tests/test_task_entry.py @@ -580,7 +580,7 @@ async def read_result(**kwargs): if executions: command, env = executions[0] assert "--kill-after=30 300s" in command - assert float(env["LOOPX_CODEX_TURN_TIMEOUT_SEC"]) == 140 + assert float(env["LOOPX_CODEX_TURN_TIMEOUT_SEC"]) == 250 assert "LOOPX_PHASE_DEADLINE_EPOCH=$(( $(date +%s) + 300 ))" in command @@ -631,7 +631,7 @@ def test_real_scheduler_stops_after_one_budget_exhausted_wake(planning_env, mode env = planning_env | { "LOOPX_EXECUTION_MODE": mode, "LOOPX_TASK_STAGE": "execute", "LOOPX_TASK_ENTRY": "seeded-todo", - "LOOPX_PHASE_DEADLINE_EPOCH": str(time.time() + 100), + "LOOPX_PHASE_DEADLINE_EPOCH": str(time.time() + (0 if mode == "heartbeat" else 100)), } if mode == "turn": env["LOOPX_VALIDATION_COMMAND_JSON"] = '["python", "check.py"]' @@ -687,7 +687,7 @@ class CapturedWindow(Exception): pass def capture(home, *, execution, **kwargs): - assert execution.timeout_seconds == 40 + assert execution.timeout_seconds == 60 raise CapturedWindow monkeypatch.setattr(worker.time, "time", lambda: 100) @@ -875,14 +875,14 @@ async def result(**kwargs): assert "LOOPX_PHASE_DEADLINE_EPOCH=$(( $(date +%s) + 1000 ))" in command assert options["timeout_sec"] >= 1000 assert "--kill-after=30 600s" in execution[0][0] - assert float(execution[0][1]["LOOPX_CODEX_TURN_TIMEOUT_SEC"]) == 440 + assert float(execution[0][1]["LOOPX_CODEX_TURN_TIMEOUT_SEC"]) == 500 def test_sforge_post_plan_deadline_never_marks_execution_started(tmp_path, monkeypatch): from benchmark.runtime import sforge_entry monkeypatch.setattr(sforge_entry, "prepare_entry", lambda env: True) - monkeypatch.setenv("LOOPX_PHASE_DEADLINE_EPOCH", str(time.time() + 150)) + monkeypatch.setenv("LOOPX_PHASE_DEADLINE_EPOCH", str(time.time() - 1)) monkeypatch.setenv("LOOPX_PLANNING_RESULT", str(tmp_path / "planning.json")) monkeypatch.setattr(sys, "argv", ["sforge-entry", "execution-must-not-start"]) monkeypatch.setattr(os, "execvpe", lambda *args: pytest.fail("Execution after exhausted planning")) @@ -892,7 +892,7 @@ def test_sforge_post_plan_deadline_never_marks_execution_started(tmp_path, monke def test_sforge_planning_exhausted_budget_never_invokes_host(planning_env): from benchmark.runtime.sforge_entry import prepare_entry - env = planning_env | {"LOOPX_PHASE_DEADLINE_EPOCH": str(time.time() + 150)} + env = planning_env | {"LOOPX_PHASE_DEADLINE_EPOCH": str(time.time() - 1)} assert not prepare_entry(env) assert not Path(env["LOOPX_PLANNING_RESULT"]).exists() assert not Path(env["LOOPX_WAKE_LOG_DIR"]).exists() diff --git a/benchmark/tests/test_worker_phase_budget.py b/benchmark/tests/test_worker_phase_budget.py index 2a17024e5f..5963aa76db 100644 --- a/benchmark/tests/test_worker_phase_budget.py +++ b/benchmark/tests/test_worker_phase_budget.py @@ -44,7 +44,7 @@ def _phase_worker(tmp_path, monkeypatch, *, work_seconds=1, independent=None, monkeypatch.setattr(worker, "time", SimpleNamespace( time=lambda: clock[0], time_ns=time.time_ns, monotonic=time.monotonic)) monkeypatch.setattr(worker, "heartbeat_body", lambda *a, **kw: "synthetic work") - env.update(LOOPX_PHASE_DEADLINE_EPOCH=str(1160 + work_seconds), LOOPX_EXECUTION_MODE="heartbeat") + env.update(LOOPX_PHASE_DEADLINE_EPOCH=str(1000 + work_seconds), LOOPX_EXECUTION_MODE="heartbeat") env.pop("LOOPX_CODEX_TURN_TIMEOUT_SEC", None) if independent is not None: env["LOOPX_CODEX_TURN_TIMEOUT_SEC"] = str(independent) @@ -96,6 +96,17 @@ def listen(self): return listeners +def test_heartbeat_can_use_a_positive_allowance_smaller_than_the_old_reserve(tmp_path, monkeypatch): + env, project, *_ = _phase_worker(tmp_path, monkeypatch, work_seconds=140, output="success") + deadline = env["LOOPX_PHASE_DEADLINE_EPOCH"] + receipt = worker.run_once(env) + assert receipt["host_invoked"] and receipt["ok"] + assert not receipt.get("budget_exhausted") + assert env["LOOPX_PHASE_DEADLINE_EPOCH"] == deadline + assert (project / "native-starts").read_text().splitlines() == ["native"] + _assert_reaped(project) + + @pytest.mark.parametrize("listener", [False, True]) def test_started_phase_timeout_is_budget_stop_and_preserves_partial_output(tmp_path, monkeypatch, listener): env, project, _runtime, _source, state, _clock = _phase_worker(