fix(runner): bound container wait so one wedged run can't stall a batch

src/clawbench/runner/batch.py

@@ -14,10 +14,16 @@
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
+from typing import Protocol
 
 import yaml
 
 from clawbench.utils.paths import ASSET_ROOT, WORKSPACE_ROOT, ensure_workspace_templates
+from clawbench.utils.timeouts import (
+    BATCH_JOB_GRACE_S,
+    DEFAULT_TIME_LIMIT_S,
+    JOB_KILL_GRACE_S,
+)
 
 
 def detect_engine() -> str:
@@ -104,6 +110,69 @@ def _resolve_cases_dir(cases_dir: str | Path) -> Path:
     return path
 
 
+def job_timeout_s(
+    case_dir: Path, override_minutes: float | None = None
+) -> float | None:
+    """Wall-clock bound for one job, or None when the user disabled it."""
+    if override_minutes is not None:
+        return None if override_minutes <= 0 else override_minutes * 60
+
+    task_file = case_dir if case_dir.is_file() else case_dir / "task.json"
+    limit_s = DEFAULT_TIME_LIMIT_S
+    try:
+        task = json.loads(task_file.read_text(encoding="utf-8"))
+        limit_s = int(float(task["time_limit"]) * 60)
+    except (OSError, ValueError, KeyError, TypeError):
+        pass
+    return limit_s + BATCH_JOB_GRACE_S
+
+
+class _Stoppable(Protocol):
+    """The part of asyncio.subprocess.Process that stop_wedged_job uses.
+
+    Narrow enough that a test can stand in a stub run which honours only
+    the signals it chooses, which is how the SIGTERM-then-SIGKILL sequence
+    is made observable without a real container.
+    """
+
+    @property
+    def pid(self) -> int: ...
+
+    async def communicate(self, input: bytes | None = None) -> tuple[bytes, bytes]: ...
+
+
+async def stop_wedged_job(
+    proc: _Stoppable, grace_s: float = JOB_KILL_GRACE_S
+) -> tuple[bytes, bool]:
+    """Stop a job that blew through its bound. Returns (output, escalated).
+
+    SIGTERM first. clawbench-run installs a SIGTERM handler that raises
+    KeyboardInterrupt and unwinds through `finally: docker_rm(container)`
+    (run.py), so the container it started actually goes away. SIGKILL cannot
+    be caught: it reaps the Python child while the container keeps running
+    under the engine daemon, holding the CPU and memory this bound exists to
+    reclaim and leaving one stale container behind per timed-out job.
+
+    Escalates to SIGKILL only if the child is still alive after `grace_s`; a
+    run wedged badly enough to ignore SIGTERM must not hold its slot open.
+    The second element reports whether that escalation happened, so callers
+    can log what actually occurred instead of claiming a clean teardown.
+    """
+    for sig in (signal.SIGTERM, signal.SIGKILL):
+        try:
+            os.killpg(proc.pid, sig)
+        except (ProcessLookupError, OSError):
+            # The group is already gone, so clawbench-run reached the end of
+            # its own cleanup; nothing was escalated past SIGTERM.
+            return b"", False
+        try:
+            stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=grace_s)
+            return stdout or b"", sig is signal.SIGKILL
+        except (asyncio.TimeoutError, ProcessLookupError, OSError):
+            continue
+    return b"", True
+
+
 def _flat_case_files(base: Path) -> list[Path]:
     return [
         p
@@ -248,6 +317,7 @@ async def run_job(
     browser_runtime_options: str | None = None,
     judge: str | None = None,
     no_judge: bool = False,
+    job_timeout: float | None = None,
 ) -> None:
     assert shutdown_event is not None
     try:
@@ -312,17 +382,50 @@ async def run_job(
                 )
                 job.proc = proc
                 running_procs.append(proc)
+                bound = job_timeout_s(job.case_dir, job_timeout)
+                host_timed_out = False
+                killed_hard = False
                 try:
-                    stdout, _ = await proc.communicate()
+                    stdout, _ = await asyncio.wait_for(
+                        proc.communicate(), timeout=bound
+                    )
+                except asyncio.TimeoutError:
+                    # The child is wedged past even its own host-side
+                    # deadline. Signal it down through its own cleanup and
+                    # keep the batch moving; stop_wedged_job explains why
+                    # this is not a SIGKILL.
+                    host_timed_out = True
+                    print(
+                        f"[{ts()}] HOST TIMEOUT {job.case_name} ({job.model}) "
+                        f"after {int(bound or 0)}s — stopping"
+                    )
+                    stdout, killed_hard = await stop_wedged_job(proc)
                 finally:
                     if proc in running_procs:
                         running_procs.remove(proc)
                     job.proc = None
 
                 job.duration = time.monotonic() - start
+                if host_timed_out:
+                    if killed_hard:
+                        # Say so plainly: SIGKILL skipped clawbench-run's
+                        # own cleanup, so its container may still be up.
+                        detail = (
+                            f"SIGTERM ignored for {int(JOB_KILL_GRACE_S)}s, "
+                            "escalated to SIGKILL; a container may have "
+                            "survived this job"
+                        )
+                    else:
+                        detail = "SIGTERM sent; clawbench-run cleaned up"
+                    marker = (
+                        f"\nbatch.py: host_timeout after {int(bound or 0)}s; {detail}\n"
+                    )
+                    stdout = (stdout or b"") + marker.encode()
                 log_path.write_bytes(stdout or b"")
 
-                if proc.returncode == 0:
+                if host_timed_out:
+                    job.status = "error"
+                elif proc.returncode == 0:
                     job.status = "passed"
                 elif proc.returncode == 1:
                     job.status = "failed"
@@ -735,6 +838,7 @@ async def _noop() -> None:
                 browser_runtime_options=getattr(args, "browser_runtime_options", None),
                 judge=args.judge,
                 no_judge=args.no_judge,
+                job_timeout=args.job_timeout,
             )
         )
         for j in jobs
@@ -828,6 +932,13 @@ def main() -> None:
         help="Max parallel jobs (default: 1 for managed runtimes, otherwise 2)",
     )
     p.add_argument("--output-dir", default="test-output", help="Base output directory")
+    p.add_argument(
+        "--job-timeout",
+        type=float,
+        default=None,
+        help="Per-job wall-clock limit in minutes. Default: the case's own "
+        "time_limit plus head-room for pull/copy/judge. 0 disables the bound.",
+    )
     p.add_argument(
         "--stagger-delay",
         type=float,

src/clawbench/runner/run.py

@@ -51,6 +51,7 @@
     step,
 )
 from clawbench.runner.run_support.email import create_email, delete_email
+from clawbench.utils.timeouts import HOST_TIMEOUT_GRACE_S
 from clawbench.runner.run_support.metadata import make_run_meta, write_run_meta
 from clawbench.runner.run_support.results import (
     classify_run,
@@ -250,6 +251,7 @@ def main():
     time_limit_s = 1800
     extra_info_warnings: list[str] = []
     intercepted = False
+    host_timeout_reason: str | None = None
     host_port: int | None = None
     judge_cfg: dict | None = startup_judge_cfg
     personal_info_metadata: dict[str, Any] | None = None
@@ -579,11 +581,21 @@ def handle_sigint(sig, frame):
             step(f"Agent running (max {task['time_limit']}min)")
 
         phase = "waiting_for_container"
-        docker_wait(
+        # Host-side backstop: the in-container watchdog gets time_limit_s to
+        # stop the agent; if it never fires we kill the container ourselves
+        # rather than blocking forever. --human runs are unbounded by design.
+        host_timed_out = docker_wait(
             container,
             model_cfg=None if args.human else model_cfg,
             harness=None if args.human else args.harness,
+            timeout_s=None if args.human else time_limit_s + HOST_TIMEOUT_GRACE_S,
         )
+        if host_timed_out:
+            host_timeout_reason = (
+                f"host_timeout: container did not exit within "
+                f"{time_limit_s + HOST_TIMEOUT_GRACE_S}s"
+            )
+            print(f"WARNING: {host_timeout_reason}")
 
         phase = "container_logs"
         step("Container logs")
@@ -669,6 +681,10 @@ def handle_sigint(sig, frame):
         classification = classify_run(
             output_dir,
             intercepted,
+            # A killed container is an infra failure, not the model's fault, so
+            # it stays out of adjusted scoring. The specific cause goes in
+            # failure_reason, matching "infra_failure: ..." elsewhere here.
+            "infra_failure" if host_timeout_reason else None,
             model_cfg=model_cfg,
             recording_required=_recording_required(),
         )
@@ -693,6 +709,7 @@ def handle_sigint(sig, frame):
             classification=classification,
             browser_runtime=_browser_runtime_meta(),
             extra_info_warnings=extra_info_warnings,
+            failure_reason=host_timeout_reason,
         )
         if judge_result is not None:
             meta["judge"] = judge_result

src/clawbench/runner/run_support/docker.py

@@ -29,6 +29,7 @@
     format_usage_status,
     summarize_usage_text,
 )
+from clawbench.utils.timeouts import HOST_TIMEOUT_GRACE_S  # noqa: F401
 from clawbench.utils.paths import DOCKER_CONTEXT_ROOT
 
 console = Console()
@@ -523,13 +524,22 @@ def docker_wait(
     name: str,
     model_cfg: dict | None = None,
     harness: str | None = None,
-) -> None:
-    """Block until the container exits, showing a live status line."""
+    timeout_s: float | None = None,
+) -> bool:
+    """Block until the container exits, showing a live status line.
+
+    Returns True if the host deadline expired and the container had to be
+    killed. The only time limit otherwise lives inside the container
+    (entrypoint.sh's MAX_WAIT); if that watchdog never fires — entrypoint
+    crash, wedged Chromium, zombie container — the host would wait forever,
+    and in batch mode the job would hold a concurrency slot indefinitely.
+    """
     start = time.time()
     proc = subprocess.Popen(
         [ENGINE, "wait", name], stdout=subprocess.PIPE, stderr=subprocess.PIPE
     )
     last_actions = 0
+    timed_out = False
     usage_summary: dict | None = None
     pricing_models: dict[str, dict] | None = None
     if model_cfg and "openrouter.ai" in str(model_cfg.get("base_url", "")):
@@ -569,6 +579,19 @@ def docker_wait(
                 f"[dim]{mins:02d}:{secs:02d}  •  {last_actions} actions  •  "
                 f"{usage_part}[/]"
             )
+            if timeout_s is not None and time.time() - start > timeout_s:
+                timed_out = True
+                console.print(
+                    f"  [yellow]Host timeout after {int(time.time() - start)}s "
+                    f"(limit {int(timeout_s)}s) — killing container[/]"
+                )
+                subprocess.run([ENGINE, "kill", name], capture_output=True, timeout=60)
+                proc.terminate()
+                try:
+                    proc.wait(timeout=10)
+                except subprocess.TimeoutExpired:
+                    proc.kill()
+                break
             try:
                 proc.wait(timeout=5)
             except subprocess.TimeoutExpired:
@@ -580,9 +603,11 @@ def docker_wait(
         if usage_summary is not None and usage_summary.get("total_tokens")
         else ""
     )
+    verb = "killed after host timeout" if timed_out else "exited"
     console.print(
-        f"  Container exited ({mins}m{secs:02d}s, {last_actions} actions{usage_part})"
+        f"  Container {verb} ({mins}m{secs:02d}s, {last_actions} actions{usage_part})"
     )
+    return timed_out
 
 
 def docker_copy(name: str, output_dir: Path) -> None:

src/clawbench/utils/timeouts.py

@@ -0,0 +1,28 @@
+"""Host-side deadlines shared by the single-run and batch drivers.
+
+Kept in `utils` because `batch.py` needs the same numbers `run_support.docker`
+does, and importing that module would drag in `run_support.config`, which
+resolves a container engine at import time and exits when neither Docker nor
+Podman is installed. `batch.py` must stay importable without one.
+"""
+
+# Head-room over the in-container watchdog (entrypoint.sh's MAX_WAIT) before
+# the host kills the container itself. Only reached when that watchdog never
+# fires: entrypoint crash, wedged Chromium, engine hiccup, zombie container.
+HOST_TIMEOUT_GRACE_S = 300
+
+# Head-room over a run's own host-side deadline, covering the work that happens
+# outside docker_wait: image pull, result copy, judge call, upload. Larger than
+# HOST_TIMEOUT_GRACE_S so clawbench-run reports its own timeout first and the
+# batch bound stays a backstop for a child process that is itself wedged.
+BATCH_JOB_GRACE_S = 900
+
+# How long a wedged clawbench-run gets to tear itself down after SIGTERM
+# before the batch escalates to SIGKILL. Its handler raises KeyboardInterrupt
+# and unwinds through `finally: docker_rm(container)`, which also deletes the
+# disposable mailbox and the browser runtime, so this covers a few short
+# subprocess and network calls rather than any agent work.
+JOB_KILL_GRACE_S = 60
+
+# Fallback when a task file has no readable time_limit.
+DEFAULT_TIME_LIMIT_S = 1800