diff --git a/scripts/native_eval/fleet.py b/scripts/native_eval/fleet.py index 74e0776..59ae30e 100644 --- a/scripts/native_eval/fleet.py +++ b/scripts/native_eval/fleet.py @@ -55,6 +55,11 @@ CRABBOX_READY_ATTEMPTS = 30 CRABBOX_READY_BACKOFF_SECONDS = 10 CRABBOX_INSPECT_ATTEMPTS = 4 +# Coordinator inspect is a single GET. Bound it so a hung broker cannot +# pin the fleet controller. +CRABBOX_INSPECT_TIMEOUT_SECONDS = 45 +# Warmup is the lease create POST. Provision readiness is polled separately. +CRABBOX_WARMUP_TIMEOUT_SECONDS = 2 * 60 class CommandExecutor(Protocol): @@ -65,6 +70,14 @@ def run( capture_output: bool = False, ) -> subprocess.CompletedProcess[str]: ... + def run_with_timeout( + self, + command: Sequence[str], + *, + capture_output: bool, + timeout: float, + ) -> subprocess.CompletedProcess[str]: ... + class SubprocessExecutor: def run( @@ -695,7 +708,11 @@ def _wait_for_lease_ready(self, identifier: str) -> Lease: def _warmup_lease(self, command: Sequence[str], slug: str) -> None: for attempt in range(1, self.config.warmup_capacity_attempts + 1): - result = self.executor.run(command, capture_output=True) + result = self.executor.run_with_timeout( + command, + capture_output=True, + timeout=CRABBOX_WARMUP_TIMEOUT_SECONDS, + ) if result.returncode == 0: return detail = (result.stderr or result.stdout or "").strip() @@ -728,7 +745,11 @@ def _inspect_lease( "--json", ] for attempt in range(1, CRABBOX_INSPECT_ATTEMPTS + 1): - result = self.executor.run(command, capture_output=True) + result = self.executor.run_with_timeout( + command, + capture_output=True, + timeout=CRABBOX_INSPECT_TIMEOUT_SECONDS, + ) if result.returncode == 0: break detail = (result.stderr or result.stdout or "").strip() diff --git a/tests/test_native_eval_fleet.py b/tests/test_native_eval_fleet.py index 2304ff7..1356b07 100644 --- a/tests/test_native_eval_fleet.py +++ b/tests/test_native_eval_fleet.py @@ -262,6 +262,45 @@ def timeout(*args: object, **kwargs: object) -> None: assert result.stderr == "partial stderr" +@pytest.mark.parametrize("action", ["inspect", "warmup"]) +def test_inspect_and_warmup_honor_subprocess_timeout( + monkeypatch: pytest.MonkeyPatch, + action: str, +) -> None: + def fake_run(*args: object, **kwargs: object) -> subprocess.CompletedProcess[str]: + timeout = kwargs.get("timeout") + if timeout is None: + raise AssertionError(f"{action} invoked subprocess.run without a timeout") + raise subprocess.TimeoutExpired( + args[0] if args else [action], + timeout, + output=b"", + stderr=b"", + ) + + monkeypatch.setattr(subprocess, "run", fake_run) + controller = object.__new__(FleetController) + controller.config = FleetConfig( + run_index=Path("/tmp/unused-index"), + local_root=Path("/tmp/unused-root"), + runner_root=Path("/tmp/unused-runner"), + task_archive=Path("/tmp/unused-tasks"), + env_file=Path("/tmp/unused.env"), + warmup_capacity_attempts=1, + warmup_capacity_backoff_seconds=0, + ) + controller.executor = SubprocessExecutor() + + with pytest.raises(FleetError, match="timed out"): + if action == "inspect": + controller._inspect_lease("cbx_hung", required=True) + else: + controller._warmup_lease( + ["crabbox", "warmup", "--slug", "hung"], + "hung", + ) + + def test_optional_inspect_treats_stopped_lease_as_absent(tmp_path: Path) -> None: run_index = tmp_path / "manifests" / "run_index.json" _write_index(run_index, []) @@ -401,6 +440,16 @@ def __init__( self.active_leases = 0 self.max_active_leases = 0 + def run_with_timeout( + self, + command: Sequence[str], + *, + capture_output: bool, + timeout: float, + ) -> subprocess.CompletedProcess[str]: + del timeout + return self.run(command, capture_output=capture_output) + def run( self, command: Sequence[str],