fix(fleet): timeout crabbox inspect and warmup

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()

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],