File size: 6,198 Bytes
42ee5cf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | """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: # keep the step line short
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)
# -- the clock for the episode as a whole ------------------------------- #
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
# -- the one method the loop calls ------------------------------------- #
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: # carried back to the caller intact
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")
# -- reporting --------------------------------------------------------- #
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
|