diff --git a/src/clawbench/runner/batch.py b/src/clawbench/runner/batch.py index 8d3637e5..37ea52c6 100644 --- a/src/clawbench/runner/batch.py +++ b/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, diff --git a/src/clawbench/runner/run.py b/src/clawbench/runner/run.py index 4c9bf3d3..f4b8e2e7 100644 --- a/src/clawbench/runner/run.py +++ b/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 diff --git a/src/clawbench/runner/run_support/docker.py b/src/clawbench/runner/run_support/docker.py index 904e3c2b..acd8233b 100644 --- a/src/clawbench/runner/run_support/docker.py +++ b/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: diff --git a/src/clawbench/utils/timeouts.py b/src/clawbench/utils/timeouts.py new file mode 100644 index 00000000..1faf236d --- /dev/null +++ b/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 diff --git a/tests/test_host_timeout.py b/tests/test_host_timeout.py new file mode 100644 index 00000000..69867a11 --- /dev/null +++ b/tests/test_host_timeout.py @@ -0,0 +1,302 @@ +"""Host-side deadlines: a wedged container must not stall a run or a batch.""" + +from __future__ import annotations + +import asyncio +import json +import os +import shutil +import signal +import subprocess +import sys +import time +from pathlib import Path +from types import ModuleType + +import pytest + +from clawbench.runner.batch import job_timeout_s, stop_wedged_job +from clawbench.utils.timeouts import ( + BATCH_JOB_GRACE_S, + DEFAULT_TIME_LIMIT_S, + HOST_TIMEOUT_GRACE_S, +) + + +def _import_docker(monkeypatch: pytest.MonkeyPatch) -> ModuleType: + """docker.py resolves a container engine at import time.""" + import importlib + + for name in ( + "clawbench.runner.run_support.docker", + "clawbench.runner.run_support.config", + ): + sys.modules.pop(name, None) + monkeypatch.delenv("CONTAINER_ENGINE", raising=False) + monkeypatch.setattr(shutil, "which", lambda cmd: cmd if cmd == "docker" else None) + return importlib.import_module("clawbench.runner.run_support.docker") + + +# --- per-job bound in batch.py ------------------------------------------------ + + +def test_job_timeout_reads_the_time_limit_from_a_task_directory( + tmp_path: Path, +) -> None: + (tmp_path / "task.json").write_text(json.dumps({"time_limit": 12})) + + assert job_timeout_s(tmp_path) == 12 * 60 + BATCH_JOB_GRACE_S + + +def test_job_timeout_reads_a_flat_claw_eval_task_file(tmp_path: Path) -> None: + """claw-eval stores tasks as flat /.json, not /task.json.""" + flat = tmp_path / "ce-T001-example.json" + flat.write_text(json.dumps({"time_limit": 5})) + + assert job_timeout_s(flat) == 5 * 60 + BATCH_JOB_GRACE_S + + +@pytest.mark.parametrize( + "payload", + ["", "not json", json.dumps({}), json.dumps({"time_limit": "soon"})], + ids=["empty", "garbage", "no-time-limit", "non-numeric"], +) +def test_job_timeout_falls_back_when_the_task_is_unreadable( + tmp_path: Path, payload: str +) -> None: + (tmp_path / "task.json").write_text(payload) + + assert job_timeout_s(tmp_path) == DEFAULT_TIME_LIMIT_S + BATCH_JOB_GRACE_S + + +def test_job_timeout_override_and_disable(tmp_path: Path) -> None: + (tmp_path / "task.json").write_text(json.dumps({"time_limit": 1})) + + assert job_timeout_s(tmp_path, 30) == 1800 + assert job_timeout_s(tmp_path, 0) is None # explicitly disabled + + +def test_batch_job_bound_exceeds_the_runs_own_deadline() -> None: + """The batch bound is a backstop: clawbench-run must hit its own first.""" + assert BATCH_JOB_GRACE_S > HOST_TIMEOUT_GRACE_S + + +def test_batch_stays_importable_without_a_container_engine( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """batch.py takes its deadlines from utils.timeouts, not run_support.docker, + which resolves a container engine at import time and exits without one.""" + import importlib + + monkeypatch.setattr(shutil, "which", lambda cmd: None) + sys.modules.pop("clawbench.runner.batch", None) + + batch = importlib.import_module("clawbench.runner.batch") + + assert batch.job_timeout_s is not None + assert "clawbench.runner.run_support.docker" not in sys.modules + + +def test_wait_for_actually_bounds_a_hanging_child() -> None: + """Guard the mechanism itself: asyncio.wait_for must interrupt the wait.""" + + async def scenario() -> bool: + async def never_returns() -> None: + await asyncio.sleep(3600) + + try: + await asyncio.wait_for(never_returns(), timeout=0.05) + except asyncio.TimeoutError: + return True + return False + + assert asyncio.run(scenario()) is True + + +# --- docker_wait deadline ----------------------------------------------------- + + +class _NeverExits: + """Stand-in for `docker wait` against a container that never exits.""" + + pid = 4242 + + def poll(self) -> None: + return None + + def wait(self, timeout: float | None = None) -> int: + raise subprocess.TimeoutExpired("docker wait", timeout or 0) + + def terminate(self) -> None: + pass + + def kill(self) -> None: + pass + + +def test_docker_wait_kills_the_container_when_the_deadline_passes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + docker = _import_docker(monkeypatch) + + monkeypatch.setattr(docker.subprocess, "Popen", lambda *a, **k: _NeverExits()) + monkeypatch.setattr(docker, "_container_usage_summary", lambda *a, **k: None) + + calls: list[list[str]] = [] + + def fake_run(cmd, *a, **k): # type: ignore[no-untyped-def] + calls.append(list(cmd)) + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(docker.subprocess, "run", fake_run) + + started = time.time() + timed_out = docker.docker_wait("wedged-container", timeout_s=0.2) + + assert timed_out is True + assert time.time() - started < 30 # returned promptly, did not hang + assert any(c[1:] == ["kill", "wedged-container"] for c in calls), calls + + +def test_docker_wait_without_a_deadline_still_waits( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """timeout_s=None keeps the old unbounded behaviour (--human runs).""" + docker = _import_docker(monkeypatch) + + class ExitsOnSecondPoll: + pid = 1 + polls = 0 + + def poll(self): # type: ignore[no-untyped-def] + ExitsOnSecondPoll.polls += 1 + return None if ExitsOnSecondPoll.polls < 2 else 0 + + def wait(self, timeout: float | None = None) -> int: + return 0 + + monkeypatch.setattr(docker.subprocess, "Popen", lambda *a, **k: ExitsOnSecondPoll()) + monkeypatch.setattr(docker, "_container_usage_summary", lambda *a, **k: None) + monkeypatch.setattr( + docker.subprocess, + "run", + lambda cmd, *a, **k: subprocess.CompletedProcess(cmd, 1, "", ""), + ) + + assert docker.docker_wait("healthy-container", timeout_s=None) is False + + +# --- stopping a wedged job without orphaning its container -------------------- + + +posix_kill_only = pytest.mark.skipif( + not hasattr(signal, "SIGKILL"), + reason="the batch kill path is os.killpg + SIGKILL, neither of which " + "exists on Windows; the runner does not use it there either", +) + + +class _FakeRun: + """Stand-in for a wedged clawbench-run: exits only on signals it honours. + + SIGTERM is catchable and run.py handles it (raising KeyboardInterrupt so + the `finally: docker_rm(container)` runs); SIGKILL is not. A stub that + honours only some signals is what makes the difference observable. + """ + + def __init__(self, honours: set[int], output: bytes = b"") -> None: + self.honours = honours + self.output = output + self.signals: list[int] = [] + self.pid = 31337 + self._exited: asyncio.Event | None = None + + def _event(self) -> asyncio.Event: + if self._exited is None: + self._exited = asyncio.Event() + return self._exited + + async def communicate(self, input: bytes | None = None) -> tuple[bytes, bytes]: + await self._event().wait() + return self.output, b"" + + def deliver(self, sig: int) -> None: + self.signals.append(sig) + if sig in self.honours: + self._event().set() + + +def _run_stop( + proc: _FakeRun, monkeypatch: pytest.MonkeyPatch, grace_s: float = 0.05 +) -> tuple[bytes, bool]: + # raising=False: os.killpg is POSIX-only and absent on Windows, where the + # batch driver's kill path does not run either. + monkeypatch.setattr( + os, + "killpg", + lambda pid, sig: proc.deliver(sig) if pid == proc.pid else None, + raising=False, + ) + return asyncio.run(stop_wedged_job(proc, grace_s=grace_s)) + + +@posix_kill_only +def test_a_timed_out_job_is_asked_to_stop_before_it_is_killed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """SIGKILL cannot be caught, so killing the group outright reaps the + Python child and leaves its container running under the engine daemon. + The first signal must be one clawbench-run can act on.""" + proc = _FakeRun(honours={signal.SIGTERM}, output=b"partial log\n") + + stdout, escalated = _run_stop(proc, monkeypatch) + + assert proc.signals == [signal.SIGTERM] + assert signal.SIGKILL not in proc.signals + assert escalated is False + assert stdout == b"partial log\n" + + +@posix_kill_only +def test_a_job_that_ignores_sigterm_is_still_killed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Graceful teardown must not become a second way to hang the batch.""" + proc = _FakeRun(honours=set()) + + _, escalated = _run_stop(proc, monkeypatch) + + assert proc.signals == [signal.SIGTERM, signal.SIGKILL] + assert escalated is True + + +@posix_kill_only +def test_escalation_is_reported_so_the_job_log_can_say_so( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A SIGKILLed run skipped its own cleanup, so the log must not claim the + container was removed. The flag is what lets batch.py tell them apart.""" + graceful = _FakeRun(honours={signal.SIGTERM}) + stubborn = _FakeRun(honours={signal.SIGKILL}) + + assert _run_stop(graceful, monkeypatch)[1] is False + assert _run_stop(stubborn, monkeypatch)[1] is True + + +@posix_kill_only +def test_an_already_dead_job_is_not_reported_as_escalated( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """If the group is gone the run finished its own teardown; saying it was + SIGKILLed would send a reader hunting for a container that isn't there.""" + proc = _FakeRun(honours=set()) + + def gone(pid: int, sig: int) -> None: + raise ProcessLookupError + + monkeypatch.setattr(os, "killpg", gone, raising=False) + + stdout, escalated = asyncio.run(stop_wedged_job(proc, grace_s=0.05)) + + assert escalated is False + assert stdout == b""