diff --git a/CHANGELOG.md b/CHANGELOG.md index cb130c2c..a996371d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] ### Added +- Added `clawbench-batch --auto-restart N`, a supervisor that re-invokes an aborted batch with `--resume` into the same directory up to N times, and `clawbench-batch-watch`, which posts to a Slack/Discord webhook when a batch aborts, completes, or every N tasks. See [`docs/operations.md`](docs/operations.md). - Added `scripts/export_openeval.py`, an additive script exporting a batch's `rescore-summary.json` as an [EvalPort](https://github.com/adhabnr-ux/evalport) `ResultSet` Thanks to [@adhabnr-ux](https://github.com/adhabnr-ux). - Added a `--browser-runtime kernel` mode to the Harbor adapter that runs each task against one Kernel cloud browser, exposing only a credential-free CDP bridge to the agent, and finalizes the replay and deletes the browser during verification. diff --git a/docs/cli.md b/docs/cli.md index 9f3bb254..c13d8fbb 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -7,6 +7,7 @@ Every ClawBench command. From a PyPI install run them directly (`clawbench-run | `clawbench` | Interactive TUI — guided model and test-case selection. Needs a TTY. | | `clawbench-run` | One task, one model (or human mode). | | `clawbench-batch` | A matrix of models × cases. | +| `clawbench-batch-watch` | Post to a webhook when a batch aborts or finishes — see [`operations.md`](operations.md). | | `clawbench-rescore` | Re-judge trajectories you already have, without re-running agents. | | `clawbench-reproduce` | Download published traces for one leaderboard row and check you reproduce it. | | `clawbench-harbor-adapt` | Convert V2 into a Harbor dataset — see [`harbor.md`](harbor.md). | @@ -74,6 +75,8 @@ Execution: | `--max-concurrent ` | 2 local, 1 Kernel/Browserbase | Parallel jobs | | `--stagger-delay ` | 15 | Minimum seconds between consecutive container starts (rolling start) | | `--resume ` | — | Reuse a previous batch's output directory and skip finished runs | +| `--auto-restart ` | `0` | Supervise the batch: re-invoke with `--resume` up to n times if it exits non-zero — see [`operations.md`](operations.md) | +| `--auto-restart-delay ` | `30` | Seconds to wait before each re-invocation | | `--dry-run` | off | Print the job matrix without running anything | | `--output-dir ` | `test-output` | Base output directory | diff --git a/docs/operations.md b/docs/operations.md new file mode 100644 index 00000000..8ecef242 --- /dev/null +++ b/docs/operations.md @@ -0,0 +1,99 @@ +# Operating long batches + +A full V2 sweep is 8–20 hours per (model × harness). Over that span something +outside the agent's control will go wrong — a container gets OOM-killed, a +provider returns 502 for half a minute, a queue cap trips at task 75. This page +covers what to do about it: how to resume, how to make resuming automatic, how +to find out promptly, and what an abort does and does not cost you. + +## How to resume a run + +Every batch writes into one directory, `test-output/batch-/`, with +one run directory per (case × model) and a `batch-logs/` folder beside them. +That directory *is* the checkpoint. To continue an interrupted batch, re-run +the same command with `--resume` pointing at it: + +```bash +uv run clawbench-batch --models deepseek-v4-flash --cases-suite v2 --all-cases \ + --harness hermes --resume test-output/batch-20260912-081500 +``` + +Jobs the batch already finished are skipped; everything else runs. The rest of +the command line must match the original — the batch directory records +outcomes, not the flags that produced them. + +## Making resume automatic + +```bash +uv run clawbench-batch --models deepseek-v4-flash --cases-suite v2 --all-cases \ + --harness hermes --auto-restart 3 +``` + +`--auto-restart N` runs the batch under a small supervisor. If the batch +process exits non-zero — including being killed outright — the supervisor +waits (`--auto-restart-delay`, default 30s) and re-invokes it with `--resume` +into the same directory, up to N times. The supervisor holds no browser, no +containers, and no model calls, which is exactly why it survives the failures +the batch does not. + +Some things to know: + +- The batch directory is created *before* the first attempt, so every attempt + — including the first — runs as a resume into one place. You can also pass + `--resume` yourself to supervise an existing batch. +- **Ctrl-C is never retried.** An interrupt stops the supervisor after the + current attempt winds down; only failures restart. +- The final exit status is the last attempt's: `0` only when an attempt + finished with every job in a terminal state. +- It is single-host. If the machine itself goes away, so does the supervisor; + resume by hand when it comes back. + +## Getting told when something happens + +```bash +uv run clawbench-batch-watch test-output/batch-20260912-081500 \ + --pid --heartbeat-every 10 +``` + +`clawbench-batch-watch` polls a batch directory and posts to a webhook when +the batch **aborts** (with the last task, an error tail from the newest log, and +elapsed time), when it **completes** (with pass/total and wall-clock), and +optionally a **heartbeat** every N completed tasks. It reads only artifacts the +batch already writes, so it needs nothing from the batch process. + +Abort is detected two ways: if you pass `--pid`, the batch process exiting +without a `batch-summary.json` is an abort; either way, `--stall-timeout` +(default 45 minutes) with no new artifacts is treated as one. + +Configure routing once per operator in `~/.config/clawbench/notify.toml`: + +```toml +webhook_url = "https://hooks.slack.com/services/…" # or a Discord webhook +interval_s = 60 +heartbeat_every = 10 +stall_timeout_s = 2700 +``` + +Flags override the file. With no webhook configured the watcher still prints +each notification, which is enough for a terminal or a log. + +## What survives an abort and what does not + +**Survives.** Every run directory that reached `run-meta.json` — its trace +bundle, its interception result, its judge verdict if the judge ran. Those are +what `--resume` skips. + +**Does not survive.** The task that was in flight when the process died. Its +run directory may exist with a partial recording and no `run-meta.json`; on +resume that task is re-run from scratch and the partial output is discarded. +There is no action-level checkpointing inside a task, and there is no plan for +one — a task is the unit of work. + +**Is not restored.** Batch-level bookkeeping from the aborted attempt. +`batch-summary.json` is written at the end of an attempt, so an aborted attempt +leaves none; the summary you get is the final attempt's, and jobs it skipped +because they were already done appear there as `skipped`. Use +`clawbench-rescore` on the batch directory for a full accounting across +attempts. + +Related: [`docs/cli.md`](cli.md) · [`eval/scoring.md`](../eval/scoring.md) diff --git a/pyproject.toml b/pyproject.toml index 2b6a4d4c..9c0f0002 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,7 @@ Paper = "https://arxiv.org/abs/2604.08523" clawbench = "clawbench.tui:main" clawbench-run = "clawbench.runner.run:main" clawbench-batch = "clawbench.runner.batch:main" +clawbench-batch-watch = "clawbench.runner.watch:main" clawbench-rescore = "clawbench.eval.rescore:main" clawbench-analyze = "clawbench.eval.analyze:main" clawbench-reproduce = "clawbench.eval.reproduce:main" diff --git a/src/clawbench/runner/batch.py b/src/clawbench/runner/batch.py index 056c4557..18484662 100644 --- a/src/clawbench/runner/batch.py +++ b/src/clawbench/runner/batch.py @@ -949,6 +949,24 @@ def main() -> None: "any (case x model) job whose batch-logs/-.log already exists." ), ) + p.add_argument( + "--auto-restart", + type=int, + default=0, + metavar="N", + help=( + "Supervise the batch: if it exits non-zero (OOM-killed container, " + "provider outage, queue cap), wait and re-invoke it with --resume up " + "to N times. Ctrl-C is never retried." + ), + ) + p.add_argument( + "--auto-restart-delay", + type=float, + default=30.0, + metavar="SECONDS", + help="Seconds to wait before each --auto-restart re-invocation (default: 30)", + ) from clawbench.runner.run import DEFAULT_HARNESS, HARNESSES p.add_argument( @@ -998,6 +1016,23 @@ def main() -> None: suite = args.cases_suite or DEFAULT_CASES_SUITE args.cases_dir = CASE_SUITES[suite] + if args.auto_restart > 0: + # The batch cannot retry its own death, so a supervisor re-invokes it. + from clawbench.runner.supervise import ( + resolve_batch_dir, + run_supervised, + strip_supervisor_flags, + ) + + sys.exit( + run_supervised( + strip_supervisor_flags(sys.argv[1:]), + batch_dir=resolve_batch_dir(args.resume, args.output_dir), + max_restarts=args.auto_restart, + delay_s=args.auto_restart_delay, + ) + ) + rc = asyncio.run(async_main(args)) sys.exit(rc) diff --git a/src/clawbench/runner/supervise.py b/src/clawbench/runner/supervise.py new file mode 100644 index 00000000..59206215 --- /dev/null +++ b/src/clawbench/runner/supervise.py @@ -0,0 +1,136 @@ +"""Re-invoke an aborted batch until it finishes — `clawbench-batch --auto-restart`. + +A multi-hour batch dies for reasons that have nothing to do with the agent: +Chromium balloons and the kernel OOM-kills the container, a provider returns +502 for thirty seconds, the queue cap trips at task 75. The batch process is +gone, so nothing inside it can retry. This runs *outside* it: a small +supervisor that spawns `clawbench-batch` as a child, and when the child exits +non-zero waits and spawns it again with `--resume` pointing at the same batch +directory, up to a bounded number of times. + +The supervisor owns nothing heavy — no browser, no containers, no model +calls — which is exactly why it survives the failures the batch does not. +""" + +from __future__ import annotations + +import subprocess +import sys +import time +from collections.abc import Callable +from datetime import datetime, timezone +from pathlib import Path + +DEFAULT_RESTART_DELAY_S = 30.0 +BATCH_MODULE = "clawbench.runner.batch" + +# Exit status for "the operator interrupted us", mirroring the shell's 128+SIGINT. +INTERRUPTED = 130 + +Spawn = Callable[[list[str]], int] +Sleep = Callable[[float], None] + + +def _default_spawn(cmd: list[str]) -> int: + """Run the batch as a child and return its exit status. + + A KeyboardInterrupt here means the operator hit Ctrl-C: the child shares + our process group and received it too, so wait for it to wind down rather + than orphaning its containers, then re-raise so the loop stops. + """ + proc = subprocess.Popen(cmd) + try: + return proc.wait() + except KeyboardInterrupt: + try: + proc.wait(timeout=60) + except subprocess.TimeoutExpired: + proc.terminate() + raise + + +def strip_supervisor_flags(argv: list[str]) -> list[str]: + """Remove the flags the supervisor consumes so the child does not recurse.""" + out: list[str] = [] + skip_next = False + for arg in argv: + if skip_next: + skip_next = False + continue + if arg in ("--auto-restart", "--auto-restart-delay", "--resume"): + skip_next = True + continue + if arg.startswith(("--auto-restart=", "--auto-restart-delay=", "--resume=")): + continue + out.append(arg) + return out + + +def resolve_batch_dir(resume: str | None, output_dir: str | Path) -> Path: + """The one batch directory every attempt resumes into. + + `--resume` carries over an existing batch; otherwise a fresh batch + directory is created here so that the *first* attempt already runs under + `--resume` and every later attempt lands in the same place. + """ + if resume: + return Path(resume).resolve() + ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") + batch_dir = Path(output_dir).resolve() / f"batch-{ts}" + batch_dir.mkdir(parents=True, exist_ok=True) + return batch_dir + + +def run_supervised( + child_argv: list[str], + *, + batch_dir: Path, + max_restarts: int, + delay_s: float = DEFAULT_RESTART_DELAY_S, + spawn: Spawn = _default_spawn, + sleep: Sleep = time.sleep, + log: Callable[[str], None] = print, +) -> int: + """Run the batch, restarting on non-zero exit up to ``max_restarts`` times. + + Returns the final child's exit status: 0 only when an attempt finished + with every job in a terminal state. A child that was interrupted by the + operator is not restarted. + """ + attempts = max_restarts + 1 + cmd = [ + sys.executable, + "-m", + BATCH_MODULE, + *child_argv, + "--resume", + str(batch_dir), + ] + rc = 1 + for attempt in range(1, attempts + 1): + log( + f"[SUPERVISOR] attempt {attempt}/{attempts}: " + f"clawbench-batch --resume {batch_dir}" + ) + started = time.monotonic() + try: + rc = spawn(cmd) + except KeyboardInterrupt: + log("[SUPERVISOR] interrupted by operator; not restarting") + return INTERRUPTED + elapsed = time.monotonic() - started + if rc == 0: + log(f"[SUPERVISOR] batch finished cleanly on attempt {attempt}") + return 0 + log( + f"[SUPERVISOR] attempt {attempt} exited with status {rc} " + f"after {elapsed / 60:.1f} min" + ) + if attempt < attempts: + log(f"[SUPERVISOR] restarting in {delay_s:.0f}s") + sleep(delay_s) + log( + f"[SUPERVISOR] giving up after {attempts} attempt(s); " + f"resume manually with: clawbench-batch --resume {batch_dir} ..." + ) + return rc diff --git a/src/clawbench/runner/watch.py b/src/clawbench/runner/watch.py new file mode 100644 index 00000000..7a1533f7 --- /dev/null +++ b/src/clawbench/runner/watch.py @@ -0,0 +1,394 @@ +"""``clawbench-batch-watch`` — tell someone when a long batch aborts or finishes. + +A V2 sweep runs 8–20 hours. When it dies at task 61 nobody finds out until +someone looks, and a partial cell sits on the leaderboard until then. This +polls a batch directory and posts to a webhook on abort, on completion, and +optionally every N completed tasks — so the operator learns within a minute, +not the next morning. + +It reads only artifacts the batch already writes (``run-meta.json`` per run, +``batch-logs/*.log``, ``batch-summary.json`` at the end) and needs nothing +from the batch process itself. Liveness comes from ``--pid`` when given, else +from a stall timeout on artifact activity. + +Routing is per operator: ``~/.config/clawbench/notify.toml``, overridden by +flags. Slack and Discord webhooks both accept the payload sent here. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import time +import tomllib +import urllib.error +import urllib.request +from collections.abc import Callable +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +DEFAULT_INTERVAL_S = 60.0 +# Longer than any single V2 task limit plus judge and teardown, so a quiet +# batch dir means the process is gone, not merely busy. +DEFAULT_STALL_TIMEOUT_S = 45 * 60 +CONFIG_PATH = Path.home() / ".config" / "clawbench" / "notify.toml" + +Post = Callable[[str, str], None] + + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class NotifyConfig: + webhook_url: str | None = None + interval_s: float = DEFAULT_INTERVAL_S + heartbeat_every: int = 0 + stall_timeout_s: float = DEFAULT_STALL_TIMEOUT_S + + +def load_config(path: Path = CONFIG_PATH) -> NotifyConfig: + """Read ``notify.toml``; a missing or empty file means defaults.""" + if not path.is_file(): + return NotifyConfig() + try: + raw = tomllib.loads(path.read_text(encoding="utf-8")) + except (OSError, tomllib.TOMLDecodeError) as e: + raise SystemExit(f"ERROR: cannot read {path}: {e}") from None + return NotifyConfig( + webhook_url=raw.get("webhook_url") or None, + interval_s=float(raw.get("interval_s", DEFAULT_INTERVAL_S)), + heartbeat_every=int(raw.get("heartbeat_every", 0)), + stall_timeout_s=float(raw.get("stall_timeout_s", DEFAULT_STALL_TIMEOUT_S)), + ) + + +# --------------------------------------------------------------------------- +# Reading the batch directory +# --------------------------------------------------------------------------- + + +@dataclass +class BatchSnapshot: + done: int = 0 + passed: int = 0 + intercepted: int = 0 + last_case: str | None = None + last_activity: float = 0.0 + started: float | None = None + finished: bool = False + summary: dict[str, Any] = field(default_factory=dict) + + +def snapshot(batch_dir: Path) -> BatchSnapshot: + snap = BatchSnapshot() + summary_file = batch_dir / "batch-summary.json" + if summary_file.is_file(): + snap.finished = True + try: + snap.summary = json.loads(summary_file.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + snap.summary = {} + + newest: tuple[float, str | None] = (0.0, None) + for meta_file in batch_dir.glob("*/*/run-meta.json"): + try: + meta = json.loads(meta_file.read_text(encoding="utf-8")) + mtime = meta_file.stat().st_mtime + except (OSError, json.JSONDecodeError): + continue + snap.done += 1 + if meta.get("intercepted"): + snap.intercepted += 1 + if meta.get("judge_match") is True: + snap.passed += 1 + if mtime > newest[0]: + newest = (mtime, meta.get("test_case") or meta_file.parent.name) + snap.last_case = newest[1] + + activity = [newest[0]] + starts: list[float] = [] + for log_file in (batch_dir / "batch-logs").glob("*.log"): + try: + stat = log_file.stat() + except OSError: + continue + activity.append(stat.st_mtime) + starts.append(stat.st_mtime) + try: + starts.append(batch_dir.stat().st_mtime) + except OSError: + pass + snap.last_activity = max(activity) + snap.started = min(starts) if starts else None + return snap + + +def error_tail(batch_dir: Path, lines: int = 15) -> str: + """Last lines of the most recently written batch log — where the abort is.""" + logs = sorted( + (batch_dir / "batch-logs").glob("*.log"), + key=lambda p: p.stat().st_mtime if p.exists() else 0, + ) + if not logs: + return "" + try: + text = logs[-1].read_text(encoding="utf-8", errors="replace") + except OSError: + return "" + return "\n".join(text.splitlines()[-lines:]) + + +def pid_alive(pid: int) -> bool: + """Whether ``pid`` is still running. + + Not ``os.kill(pid, 0)``: on Windows that call *terminates* the process, + because any signal other than the Ctrl events is passed to + TerminateProcess as an exit code. + """ + if sys.platform == "win32": + import ctypes + + kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] + process_query_limited_information = 0x1000 + still_active = 259 + handle = kernel32.OpenProcess(process_query_limited_information, False, pid) + if not handle: + return False + try: + code = ctypes.c_ulong() + if not kernel32.GetExitCodeProcess(handle, ctypes.byref(code)): + return False + return code.value == still_active + finally: + kernel32.CloseHandle(handle) + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + return True + + +# --------------------------------------------------------------------------- +# Posting +# --------------------------------------------------------------------------- + + +def post_webhook(url: str, text: str) -> None: + """POST a message; Slack reads ``text``, Discord reads ``content``.""" + payload = json.dumps({"text": text, "content": text}).encode() + request = urllib.request.Request( + url, + data=payload, + method="POST", + headers={"Content-Type": "application/json"}, + ) + try: + with urllib.request.urlopen(request, timeout=15): + pass + except (urllib.error.URLError, TimeoutError, OSError) as e: + # A failed notification must not take the watcher down with it. + print(f"[watch] webhook post failed: {e}", file=sys.stderr) + + +def _fmt_hours(seconds: float) -> str: + return f"{seconds / 3600:.1f}h" + + +def completion_message(name: str, snap: BatchSnapshot, now: float) -> str: + totals = snap.summary.get("totals") or {} + elapsed = snap.summary.get("elapsed_seconds") + if not isinstance(elapsed, (int, float)): + elapsed = now - snap.started if snap.started else 0 + jobs = snap.summary.get("jobs") + total = len(jobs) if isinstance(jobs, list) else snap.done + passed = totals.get("passed", snap.passed) + errors = totals.get("error", 0) + text = f"✅ {name} · finished · {passed}/{total} passed · {_fmt_hours(elapsed)}" + if errors: + text += f" · {errors} infra error(s)" + return text + + +def abort_message( + name: str, snap: BatchSnapshot, now: float, reason: str, tail: str +) -> str: + elapsed = now - snap.started if snap.started else 0 + text = ( + f"🛑 {name} · aborted ({reason}) · {snap.done} done · " + f"last task {snap.last_case or '?'} · {_fmt_hours(elapsed)} elapsed" + ) + if tail: + text += f"\n```\n{tail}\n```" + return text + + +def heartbeat_message( + name: str, snap: BatchSnapshot, now: float, total: int | None +) -> str: + elapsed = now - snap.started if snap.started else 0 + denominator = f"/{total}" if total else "" + return ( + f"💓 {name} · {snap.done}{denominator} done · {_fmt_hours(elapsed)} elapsed · " + f"{snap.passed} passes · {snap.intercepted} intercepted" + ) + + +# --------------------------------------------------------------------------- +# The watch loop +# --------------------------------------------------------------------------- + + +@dataclass +class Watcher: + batch_dir: Path + config: NotifyConfig + pid: int | None = None + total: int | None = None + post: Post = post_webhook + now: Callable[[], float] = time.time + log: Callable[[str], None] = print + _last_heartbeat_at: int = 0 + + @property + def name(self) -> str: + return self.batch_dir.name + + def _notify(self, text: str) -> None: + self.log(text) + if self.config.webhook_url: + self.post(self.config.webhook_url, text) + + def poll(self) -> int | None: + """One observation. Returns an exit status once the batch is over.""" + snap = snapshot(self.batch_dir) + now = self.now() + + if snap.finished: + self._notify(completion_message(self.name, snap, now)) + return 0 + + if self.pid is not None and not pid_alive(self.pid): + reason = f"pid {self.pid} exited without a batch summary" + self._notify( + abort_message(self.name, snap, now, reason, error_tail(self.batch_dir)) + ) + return 1 + + if ( + snap.last_activity + and now - snap.last_activity > self.config.stall_timeout_s + ): + quiet = _fmt_hours(now - snap.last_activity) + reason = f"no artifact activity for {quiet}" + self._notify( + abort_message(self.name, snap, now, reason, error_tail(self.batch_dir)) + ) + return 1 + + every = self.config.heartbeat_every + if every > 0 and snap.done // every > self._last_heartbeat_at // every: + self._last_heartbeat_at = snap.done + self._notify(heartbeat_message(self.name, snap, now, self.total)) + return None + + def run(self, sleep: Callable[[float], None] = time.sleep) -> int: + while True: + status = self.poll() + if status is not None: + return status + sleep(self.config.interval_s) + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="clawbench-batch-watch", + description=( + "Poll a running batch directory and post to a webhook when it " + "aborts or finishes." + ), + ) + parser.add_argument("batch_dir", type=Path, help="the batch- directory") + parser.add_argument( + "--webhook", help="Slack/Discord webhook URL (overrides notify.toml)" + ) + parser.add_argument( + "--interval", type=float, help="seconds between polls (default: 60)" + ) + parser.add_argument( + "--heartbeat-every", + type=int, + help="post a progress line every N completed tasks (default: off)", + ) + parser.add_argument( + "--stall-timeout", + type=float, + help="seconds without artifact activity before declaring an abort (default: 2700)", + ) + parser.add_argument( + "--pid", type=int, help="batch process id; its exit means abort" + ) + parser.add_argument( + "--total", type=int, help="expected job count, for heartbeat denominators" + ) + parser.add_argument( + "--config", + type=Path, + default=CONFIG_PATH, + help=f"notify.toml to read (default: {CONFIG_PATH})", + ) + parser.add_argument("--once", action="store_true", help="poll once and exit") + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + if not args.batch_dir.is_dir(): + print(f"ERROR: not a directory: {args.batch_dir}", file=sys.stderr) + return 2 + base = load_config(args.config) + config = NotifyConfig( + webhook_url=args.webhook or base.webhook_url, + interval_s=args.interval if args.interval is not None else base.interval_s, + heartbeat_every=( + args.heartbeat_every + if args.heartbeat_every is not None + else base.heartbeat_every + ), + stall_timeout_s=( + args.stall_timeout + if args.stall_timeout is not None + else base.stall_timeout_s + ), + ) + if not config.webhook_url: + print( + "[watch] no webhook configured; printing notifications only " + f"(set webhook_url in {args.config} or pass --webhook)" + ) + watcher = Watcher( + batch_dir=args.batch_dir.resolve(), + config=config, + pid=args.pid, + total=args.total, + ) + if args.once: + status = watcher.poll() + return 0 if status is None else status + return watcher.run() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_batch_resilience.py b/tests/test_batch_resilience.py new file mode 100644 index 00000000..5ba3d3a0 --- /dev/null +++ b/tests/test_batch_resilience.py @@ -0,0 +1,340 @@ +"""`clawbench-batch --auto-restart` and `clawbench-batch-watch` (#160).""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from clawbench.runner import supervise, watch + +# --------------------------------------------------------------------------- +# Supervisor +# --------------------------------------------------------------------------- + + +def _fake_spawn(exit_codes: list[int]) -> tuple[list[list[str]], supervise.Spawn]: + """A child that exits with each code in turn, recording every command.""" + calls: list[list[str]] = [] + remaining = list(exit_codes) + + def spawn(cmd: list[str]) -> int: + calls.append(list(cmd)) + return remaining.pop(0) + + return calls, spawn + + +def test_supervisor_restarts_into_the_same_batch_dir_until_success( + tmp_path: Path, +) -> None: + calls, spawn = _fake_spawn([1, 137, 0]) + sleeps: list[float] = [] + lines: list[str] = [] + + rc = supervise.run_supervised( + ["--models", "m", "--all-cases"], + batch_dir=tmp_path / "batch-x", + max_restarts=3, + delay_s=30, + spawn=spawn, + sleep=sleeps.append, + log=lines.append, + ) + + assert rc == 0 + assert len(calls) == 3 + for cmd in calls: + assert cmd[:3] == [sys.executable, "-m", supervise.BATCH_MODULE] + assert cmd[-2:] == ["--resume", str(tmp_path / "batch-x")] + assert "--auto-restart" not in cmd + # Waited before each retry, never after the success. + assert sleeps == [30, 30] + assert any("finished cleanly on attempt 3" in line for line in lines) + + +def test_supervisor_gives_up_after_n_restarts(tmp_path: Path) -> None: + calls, spawn = _fake_spawn([1, 1, 1, 1]) + lines: list[str] = [] + + rc = supervise.run_supervised( + [], + batch_dir=tmp_path, + max_restarts=2, + spawn=spawn, + sleep=lambda _s: None, + log=lines.append, + ) + + assert rc == 1 + # N restarts means N+1 attempts, no more. + assert len(calls) == 3 + assert any("giving up after 3 attempt(s)" in line for line in lines) + assert any(f"--resume {tmp_path}" in line for line in lines) + + +def test_supervisor_does_not_restart_after_operator_interrupt(tmp_path: Path) -> None: + calls: list[list[str]] = [] + + def spawn(cmd: list[str]) -> int: + calls.append(cmd) + raise KeyboardInterrupt + + rc = supervise.run_supervised( + [], + batch_dir=tmp_path, + max_restarts=5, + spawn=spawn, + sleep=lambda _s: pytest.fail("must not sleep after an interrupt"), + log=lambda _line: None, + ) + + assert rc == supervise.INTERRUPTED + assert len(calls) == 1 + + +def test_supervisor_flags_are_stripped_from_the_child_command() -> None: + argv = [ + "--models", + "m", + "--auto-restart", + "3", + "--auto-restart-delay=5", + "--resume", + "/old/batch", + "--all-cases", + ] + + assert supervise.strip_supervisor_flags(argv) == ["--models", "m", "--all-cases"] + + +def test_resolve_batch_dir_creates_a_fresh_dir_or_reuses_resume(tmp_path: Path) -> None: + fresh = supervise.resolve_batch_dir(None, tmp_path / "out") + assert fresh.is_dir() + assert fresh.parent == (tmp_path / "out").resolve() + assert fresh.name.startswith("batch-") + + existing = tmp_path / "existing" + assert ( + supervise.resolve_batch_dir(str(existing), tmp_path / "out") + == existing.resolve() + ) + + +def test_batch_cli_hands_off_to_the_supervisor(tmp_path: Path) -> None: + """`--auto-restart N` never runs the batch in-process.""" + env = os.environ.copy() + env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1] / "src") + code = ( + "import sys\n" + "from clawbench.runner import batch, supervise\n" + "seen = {}\n" + "def fake(argv, *, batch_dir, max_restarts, delay_s):\n" + " seen.update(argv=argv, batch_dir=str(batch_dir), n=max_restarts, d=delay_s)\n" + " return 7\n" + "supervise.run_supervised = fake\n" + f"sys.argv = ['clawbench-batch', '--models', 'm', '--all-cases', " + f"'--output-dir', {str(tmp_path)!r}, '--auto-restart', '2', " + "'--auto-restart-delay', '1']\n" + "try:\n" + " batch.main()\n" + "except SystemExit as e:\n" + " print('rc', e.code)\n" + "print(seen)\n" + ) + result = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + env=env, + timeout=60, + ) + + assert result.returncode == 0, result.stderr + assert "rc 7" in result.stdout + assert "'n': 2" in result.stdout and "'d': 1.0" in result.stdout + assert "--auto-restart" not in result.stdout.split("'argv': ")[1].split("]")[0] + assert "batch-" in result.stdout + + +# --------------------------------------------------------------------------- +# Watcher +# --------------------------------------------------------------------------- + + +def _run( + base: Path, + model: str, + case: str, + *, + intercepted: bool, + judge_match: object = "absent", +) -> Path: + run_dir = base / model / case + run_dir.mkdir(parents=True) + meta: dict = {"test_case": case, "model": model, "intercepted": intercepted} + if judge_match != "absent": + meta["judge_match"] = judge_match + (run_dir / "run-meta.json").write_text(json.dumps(meta)) + return run_dir + + +def _batch(tmp_path: Path) -> Path: + base = tmp_path / "batch-20260912-000000" + (base / "batch-logs").mkdir(parents=True) + (base / "batch-logs" / "case-1-m.log").write_text("started\nok\n") + _run(base, "m", "case-1", intercepted=True, judge_match=True) + _run(base, "m", "case-2", intercepted=True, judge_match=False) + _run(base, "m", "case-3", intercepted=False) + return base + + +def _watcher( + base: Path, + *, + heartbeat_every: int = 0, + stall_timeout_s: float = 3600, + now: float | None = None, + pid: int | None = None, + total: int | None = None, +) -> tuple[watch.Watcher, list[tuple[str, str]]]: + posts: list[tuple[str, str]] = [] + cfg = watch.NotifyConfig( + webhook_url="https://hooks.example.test/x", + interval_s=1, + heartbeat_every=heartbeat_every, + stall_timeout_s=stall_timeout_s, + ) + clock = now if now is not None else base.stat().st_mtime + 60 + watcher = watch.Watcher( + batch_dir=base, + config=cfg, + pid=pid, + total=total, + post=lambda url, text: posts.append((url, text)), + now=lambda: clock, + log=lambda _line: None, + ) + return watcher, posts + + +def test_snapshot_counts_both_stages_and_finds_the_last_task(tmp_path: Path) -> None: + snap = watch.snapshot(_batch(tmp_path)) + + assert snap.done == 3 + assert snap.intercepted == 2 + assert snap.passed == 1 + assert snap.last_case in {"case-1", "case-2", "case-3"} + assert snap.finished is False + assert snap.started is not None and snap.last_activity >= snap.started + + +def test_watcher_keeps_quiet_while_the_batch_is_healthy(tmp_path: Path) -> None: + watcher, posts = _watcher(_batch(tmp_path)) + + assert watcher.poll() is None + assert posts == [] + + +def test_watcher_posts_on_completion(tmp_path: Path) -> None: + base = _batch(tmp_path) + (base / "batch-summary.json").write_text( + json.dumps( + { + "elapsed_seconds": 7200, + "jobs": [{}] * 3, + "totals": {"passed": 1, "failed": 2, "error": 0}, + } + ) + ) + watcher, posts = _watcher(base) + + assert watcher.poll() == 0 + ((url, text),) = posts + assert url == "https://hooks.example.test/x" + assert "finished" in text and "1/3 passed" in text and "2.0h" in text + + +def test_watcher_posts_abort_when_the_pid_is_gone(tmp_path: Path) -> None: + base = _batch(tmp_path) + (base / "batch-logs" / "case-1-m.log").write_text("started\nTraceback: boom\n") + dead = subprocess.Popen([sys.executable, "-c", "pass"]) + dead.wait() + watcher, posts = _watcher(base, pid=dead.pid) + + assert watcher.poll() == 1 + ((_url, text),) = posts + assert "aborted" in text and f"pid {dead.pid}" in text + assert "3 done" in text + assert "Traceback: boom" in text + + +def test_watcher_posts_abort_on_stall(tmp_path: Path) -> None: + base = _batch(tmp_path) + stale_now = base.stat().st_mtime + 10 * 3600 + watcher, posts = _watcher(base, stall_timeout_s=3600, now=stale_now) + + assert watcher.poll() == 1 + ((_url, text),) = posts + assert "no artifact activity" in text + + +def test_watcher_heartbeats_every_n_tasks_without_repeating(tmp_path: Path) -> None: + base = _batch(tmp_path) + watcher, posts = _watcher(base, heartbeat_every=2, total=10) + + assert watcher.poll() is None + assert len(posts) == 1 + assert "3/10 done" in posts[0][1] + # Same count again: no second heartbeat until another 2 tasks land. + assert watcher.poll() is None + assert len(posts) == 1 + _run(base, "m", "case-4", intercepted=True) + assert watcher.poll() is None + assert len(posts) == 2 + + +def test_pid_alive_for_this_process_and_a_dead_one() -> None: + assert watch.pid_alive(os.getpid()) is True + dead = subprocess.Popen([sys.executable, "-c", "pass"]) + dead.wait() + assert watch.pid_alive(dead.pid) is False + + +def test_notify_config_reads_toml_and_flags_override(tmp_path: Path) -> None: + cfg_file = tmp_path / "notify.toml" + cfg_file.write_text( + 'webhook_url = "https://hooks.example.test/from-file"\n' + "interval_s = 15\nheartbeat_every = 5\n" + ) + cfg = watch.load_config(cfg_file) + assert cfg.webhook_url == "https://hooks.example.test/from-file" + assert cfg.interval_s == 15 + assert cfg.heartbeat_every == 5 + assert cfg.stall_timeout_s == watch.DEFAULT_STALL_TIMEOUT_S + + assert watch.load_config(tmp_path / "missing.toml") == watch.NotifyConfig() + + +def test_watch_cli_once_reports_status( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + base = _batch(tmp_path) + (base / "batch-summary.json").write_text( + json.dumps({"elapsed_seconds": 60, "jobs": [{}] * 3, "totals": {"passed": 1}}) + ) + + rc = watch.main([str(base), "--once", "--config", str(tmp_path / "none.toml")]) + + assert rc == 0 + out = capsys.readouterr().out + assert "no webhook configured" in out + assert "finished" in out + + +def test_watch_cli_rejects_a_missing_dir(tmp_path: Path) -> None: + assert watch.main([str(tmp_path / "nope"), "--once"]) == 2 diff --git a/tests/test_cli_entrypoints.py b/tests/test_cli_entrypoints.py index fdeef64a..3ddae75d 100644 --- a/tests/test_cli_entrypoints.py +++ b/tests/test_cli_entrypoints.py @@ -17,6 +17,7 @@ "clawbench.tui", "clawbench.runner.run", "clawbench.runner.batch", + "clawbench.runner.watch", "clawbench.eval.harbor_adapter", )