| """Refuse to wait forever. |
| |
| The loop already stops a *repeated* action against an unchanged screen, and a step budget |
| bounds how many turns an episode gets. Neither of them covers the other failure: one call |
| that never returns. A model that is loading weights, an OCR pass over a 4K screen, an |
| executor waiting on a window manager - the harness sits inside `brain.decide()` or |
| `executor.execute()` and there is no step, no reason and no evidence, because nothing has |
| finished. A console in that state looks exactly like a console that is thinking. |
| |
| A watchdog puts a deadline on each call and on the episode as a whole, and turns a missed |
| deadline into the same kind of object as every other failure in this harness: a recorded |
| step with a reason, never a traceback and never a hang. |
| |
| loop = AgentLoop(screen=..., brain=..., |
| watchdog=Watchdog(step_timeout=30, wall_clock=600)) |
| |
| Honest limits, because this is not a scheduler. Python cannot interrupt a thread, so a call |
| that has missed its deadline is **abandoned, not killed**: it keeps running in a daemon |
| thread until it returns or the process exits, and the watchdog records its name in |
| `orphans`. What is guaranteed is that the *loop* stops waiting, says which call it stopped |
| waiting for, and gives control back to the caller. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import threading |
| import time |
| from dataclasses import dataclass, field |
| from typing import Any, Callable |
|
|
| __all__ = ["DEFAULT_STEP_TIMEOUT", "Watchdog", "WatchdogTimeout"] |
|
|
| DEFAULT_STEP_TIMEOUT = 60.0 |
| """Seconds one guarded call may take. Generous, because a model's first decision loads.""" |
|
|
|
|
| class WatchdogTimeout(RuntimeError): |
| """A guarded call missed its deadline. |
| |
| Carries the two facts a reader needs - *where* it hung and *how long* it was given - |
| so the step line says something more useful than "failed". |
| """ |
|
|
| def __init__(self, where: str, seconds: float, *, abandoned: bool = True) -> None: |
| self.where = where |
| self.seconds = seconds |
| self.abandoned = abandoned |
| how = "abandoned after" if abandoned else "refused: the wall clock left" |
| super().__init__(f"{where} {how} {seconds:g}s") |
|
|
| def __str__(self) -> str: |
| return (f"{self.where} did not finish within {self.seconds:g}s" |
| if self.abandoned else |
| f"{self.where} was not attempted: the episode's wall clock is spent") |
|
|
|
|
| @dataclass |
| class Watchdog: |
| """A deadline on every call the loop cannot otherwise escape. |
| |
| ``step_timeout=None`` disables per-call deadlines entirely and runs calls inline, with |
| no thread and no overhead - which is the honest default for a harness driving a local |
| desktop, where the process *is* the thing being watched. ``wall_clock`` bounds the |
| whole episode instead, and is checked before each call rather than during one. |
| """ |
|
|
| step_timeout: float | None = DEFAULT_STEP_TIMEOUT |
| wall_clock: float | None = None |
| clock: Callable[[], float] = time.monotonic |
| threads_started: int = field(default=0, init=False) |
| orphans: list[str] = field(default_factory=list, init=False) |
| _started_at: float = field(default=0.0, init=False, repr=False) |
| _ticking: bool = field(default=False, init=False, repr=False) |
|
|
| |
|
|
| def start(self) -> "Watchdog": |
| """Begin timing this episode. Idempotent, so a re-run cannot double-charge it.""" |
| if not self._ticking: |
| self._started_at = self.clock() |
| self._ticking = True |
| return self |
|
|
| @property |
| def elapsed(self) -> float: |
| return self.clock() - self._started_at if self._ticking else 0.0 |
|
|
| @property |
| def remaining(self) -> float | None: |
| """Seconds left on the wall clock, or ``None`` when there is no wall clock.""" |
| if self.wall_clock is None: |
| return None |
| return max(0.0, self.wall_clock - self.elapsed) |
|
|
| @property |
| def expired(self) -> bool: |
| return self.wall_clock is not None and self.elapsed >= self.wall_clock |
|
|
| |
|
|
| def call(self, where: str, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: |
| """Call ``fn`` under the deadline, or raise :class:`WatchdogTimeout`. |
| |
| The wall clock is checked *before* the call, so a game over is never reported as a |
| timeout of whatever happened to be next. Per-call deadlines run on a daemon thread; |
| with none configured the call is made inline. |
| """ |
| if self.expired: |
| raise WatchdogTimeout(where, self.wall_clock or 0.0, abandoned=False) |
| if self.step_timeout is None: |
| return fn(*args, **kwargs) |
|
|
| self.threads_started += 1 |
| outcome: dict[str, Any] = {} |
|
|
| def run() -> None: |
| try: |
| outcome["value"] = fn(*args, **kwargs) |
| except BaseException as exc: |
| outcome["error"] = exc |
|
|
| worker = threading.Thread(target=run, name=f"watchdog-{where}", daemon=True) |
| worker.start() |
| worker.join(self.step_timeout) |
| if worker.is_alive(): |
| self.orphans.append(where) |
| raise WatchdogTimeout(where, self.step_timeout) |
| if "error" in outcome: |
| raise outcome["error"] |
| return outcome.get("value") |
|
|
| |
|
|
| def render(self) -> str: |
| limits = [] |
| if self.step_timeout is None: |
| limits.append("per-call deadline: none") |
| else: |
| limits.append(f"per call <= {self.step_timeout:g}s") |
| limits.append("wall clock: none" if self.wall_clock is None |
| else f"wall clock <= {self.wall_clock:g}s") |
| if self._ticking: |
| limits.append(f"{self.elapsed:.1f}s used") |
| line = "watchdog: " + ", ".join(limits) |
| if self.orphans: |
| line += f" | abandoned: {', '.join(self.orphans)}" |
| return line |
|
|
