Package: 2.0 - watchdog and audit modules

tiny_computer_agent/__init__.py

@@ -36,6 +36,13 @@ from tiny_computer_agent.accessibility import (
     UiaReader,
     accessibility_available,
 )
+from tiny_computer_agent.audit import (
+    AuditEntry,
+    AuditLog,
+    observation_digest,
+    read_audit,
+    verify_audit,
+)
 from tiny_computer_agent.act import (
     ActionResult,
     DryRunExecutor,
@@ -118,6 +125,7 @@ from tiny_computer_agent.gates import (
     DenyAll,
     Gate,
     InteractiveGate,
+    gate_from_spec,
     mutating,
 )
 from tiny_computer_agent.runner import RunReport, Task, TaskOutcome, TaskRunner
@@ -131,6 +139,7 @@ from tiny_computer_agent.skills import (
     record_skill,
     slug,
 )
+from tiny_computer_agent.watchdog import Watchdog, WatchdogTimeout
 from tiny_computer_agent.trace import (
     Scenario,
     Scoreboard,
@@ -152,29 +161,31 @@ __all__ = [
     "ACTION_KINDS",             "ACTION_MENU",              "AccessibleNode",           "Action",
     "ActionError",              "ActionResult",             "AgentLoop",                "AllOf",
     "AllowAll",                 "AllowOnlyLabels",          "AnyOf",                    "Approval",
-    "AtspiDetector",            "AtspiReader",              "AutoDetector",             "BRAIN_NAMES",
-    "Brain",                    "BrainError",               "ChangeVerifier",           "Check",
-    "DEFAULT_KILL_SWITCH",      "DEFAULT_SKILLS_PATH",      "DEFAULT_TEXT_MODEL",       "DEFAULT_VISION_MODEL",
-    "DecisionMenu",             "Delta",                    "DenyAll",                  "Detector",
-    "DetectorUnavailable",      "DryRunExecutor",           "Element",                  "Episode",
-    "Executor",                 "FakeScreen",               "Gate",                     "GoalUnreachable",
-    "GrabScreen",               "GridDetector",             "INTERACTIVE_ROLES",        "ImageScreen",
-    "IndexedScreen",            "InputUnavailable",         "InteractiveGate",          "MODEL_BRAINS",
-    "MenuItem",                 "MockBrain",                "ModelBrain",               "Observation",
-    "OcrDetector",              "PlannerBrain",             "Policy",                   "PolicyViolation",
-    "PyAutoGuiExecutor",        "READ_ONLY_KINDS",          "RunReport",                "Scenario",
-    "Scoreboard",               "ScreenSource",             "ScreenUnavailable",        "Skill",
-    "SkillBrain",               "SkillLibrary",             "SkillStep",                "Step",
-    "TEXT_ROLES",               "Task",                     "TaskOutcome",              "TaskRunner",
-    "TextPlannerBrain",         "Trace",                    "TraceStep",                "TreeReader",
-    "UiaDetector",              "UiaReader",                "VLMBrain",                 "Verdict",
+    "AtspiDetector",            "AtspiReader",              "AuditEntry",               "AuditLog",
+    "AutoDetector",             "BRAIN_NAMES",              "Brain",                    "BrainError",
+    "ChangeVerifier",           "Check",                    "DEFAULT_KILL_SWITCH",      "DEFAULT_SKILLS_PATH",
+    "DEFAULT_TEXT_MODEL",       "DEFAULT_VISION_MODEL",     "DecisionMenu",             "Delta",
+    "DenyAll",                  "Detector",                 "DetectorUnavailable",      "DryRunExecutor",
+    "Element",                  "Episode",                  "Executor",                 "FakeScreen",
+    "Gate",                     "GoalUnreachable",          "GrabScreen",               "GridDetector",
+    "INTERACTIVE_ROLES",        "ImageScreen",              "IndexedScreen",            "InputUnavailable",
+    "InteractiveGate",          "MODEL_BRAINS",             "MenuItem",                 "MockBrain",
+    "ModelBrain",               "Observation",              "OcrDetector",              "PlannerBrain",
+    "Policy",                   "PolicyViolation",          "PyAutoGuiExecutor",        "READ_ONLY_KINDS",
+    "RunReport",                "Scenario",                 "Scoreboard",               "ScreenSource",
+    "ScreenUnavailable",        "Skill",                    "SkillBrain",               "SkillLibrary",
+    "SkillStep",                "Step",                     "TEXT_ROLES",               "Task",
+    "TaskOutcome",              "TaskRunner",               "TextPlannerBrain",         "Trace",
+    "TraceStep",                "TreeReader",               "UiaDetector",              "UiaReader",
+    "VLMBrain",                 "Verdict",                  "Watchdog",                 "WatchdogTimeout",
     "__version__",              "accessibility_available",  "benchmark",                "benchmark_traces",
     "build_agent",              "detector_available",       "diff",                     "display_available",
     "doctor",                   "element_key",              "extract_json_object",      "extract_json_objects",
-    "fake_desktop",             "load_traces",              "make_detector",            "make_model_brain",
-    "model_cached",             "model_dependencies",       "mutating",                 "normalise_label",
-    "parse_action",             "parse_decision",           "pick_dtype",               "record_skill",
-    "resolve_device",           "resolve_point",            "run_checks",               "run_scenario",
-    "score",                    "score_replay",             "score_trace",              "slug",
+    "fake_desktop",             "gate_from_spec",           "load_traces",              "make_detector",
+    "make_model_brain",         "model_cached",             "model_dependencies",       "mutating",
+    "normalise_label",          "observation_digest",       "parse_action",             "parse_decision",
+    "pick_dtype",               "read_audit",               "record_skill",             "resolve_device",
+    "resolve_point",            "run_checks",               "run_scenario",             "score",
+    "score_replay",             "score_trace",              "slug",                     "verify_audit",
     "write_fake_desktop",
 ]

tiny_computer_agent/__main__.py

@@ -42,6 +42,8 @@ from tiny_computer_agent.safety import Policy
 from tiny_computer_agent.runner import Task, TaskRunner
 from tiny_computer_agent.skills import DEFAULT_SKILLS_PATH, SkillBrain, SkillLibrary, record_skill
 from tiny_computer_agent.trace import Trace, score_trace
+from tiny_computer_agent.audit import AuditLog, verify_audit
+from tiny_computer_agent.gates import Gate, gate_from_spec
 
 ALLOW_REAL_INPUT_ENV = "TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT"
 EXAMPLE_IMAGE = "example"
@@ -103,6 +105,13 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--trace", default=None, metavar="PATH",
                         help="write a portable trace of the run (frames, no pixels) so it "
                              "can be replayed and scored offline")
+    parser.add_argument("--gate", default=None,
+                        help="the operator's yes, before each action: read-only | interactive "
+                             "| labels:Save,Cancel | allow-all")
+    parser.add_argument("--audit", default=None, metavar="PATH",
+                        help="append a hash-chained record of this run to a JSONL file")
+    parser.add_argument("--verify-audit", default=None, metavar="PATH",
+                        help="check an audit log's chain and exit (0 intact, 1 tampered)")
     parser.add_argument("--expected", default=None, metavar="LABEL",
                         help="the element label that is the right answer for this goal; "
                              "recorded in the trace so the run can be scored")
@@ -141,6 +150,23 @@ def _probe_live(args: argparse.Namespace) -> Check:
                      fix="run this on the desktop session itself, not a headless shell")
 
 
+def _gate_from_spec_arg(args: argparse.Namespace) -> tuple[Gate | None, str | None]:
+    """The operator's gate, from the one-line spelling on the command line."""
+    spec = getattr(args, "gate", None)
+    if not spec:
+        return None, None
+    try:
+        return gate_from_spec(spec), None
+    except ValueError as exc:
+        return None, f"bad --gate: {exc}"
+
+
+def _audit_from_arg(args: argparse.Namespace) -> AuditLog | None:
+    """Append-only evidence for this run, if the operator asked for it."""
+    path = getattr(args, "audit", None)
+    return AuditLog(path) if path else None
+
+
 def _build_screen(args: argparse.Namespace):
     """(screen, description). Raises ScreenUnavailable / OSError on failure."""
     detector_name = "grid" if args.no_ocr else args.detector
@@ -345,6 +371,17 @@ def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
     detector_name = "grid" if args.no_ocr else args.detector
 
+    if args.verify_audit:
+        intact, detail = verify_audit(args.verify_audit)
+        print(f"{'ok  ' if intact else 'FAIL'} audit chain: {detail}")
+        return 0 if intact else 1
+
+    gate, gate_error = _gate_from_spec_arg(args)
+    if gate_error is not None:
+        print(gate_error, file=sys.stderr)
+        return 2
+    audit = _audit_from_arg(args)
+
     if args.doctor:
         checks = run_checks(brain=args.brain, detector=detector_name,
                             real_input=args.real_input, kill_switch_path=args.kill_switch)
@@ -410,6 +447,10 @@ def main(argv: list[str] | None = None) -> int:
     print(f"goal: {args.goal}")
     print(f"mode: {mode}  |  screen: {description}  |  brain: {args.brain}  |  "
           f"kill switch: create {args.kill_switch} to stop")
+    if gate is not None:
+        print(f"gate: {args.gate} - every action needs the operator's yes")
+    if audit is not None:
+        print(f"audit: recording this run to {audit.path}")
     if args.image is not None or args.live:
         _, index_detail = detector_available(detector_name)
         print(f"index: {detector_name} - {index_detail}")
@@ -445,7 +486,8 @@ def main(argv: list[str] | None = None) -> int:
 
     try:
         episode = AgentLoop(screen=screen, brain=brain, executor=executor,
-                            policy=policy, on_step=on_step, verify=verify).run(args.goal)
+                            policy=policy, on_step=on_step, verify=verify,
+                            gate=gate, audit=audit).run(args.goal)
     except ScreenUnavailable as exc:
         print(f"cannot see the screen: {exc}", file=sys.stderr)
         print("run `python -m tiny_computer_agent --doctor --live` for details.", file=sys.stderr)
@@ -453,6 +495,8 @@ def main(argv: list[str] | None = None) -> int:
 
     print()
     print(episode.render())
+    if audit is not None:
+        print(audit.render())
 
     if args.trace:
         trace = Trace.record(episode, expected=args.expected, brain=args.brain,

tiny_computer_agent/audit.py

@@ -0,0 +1,202 @@
+"""A run's record, chained so that editing it is louder than deleting it.
+
+The harness already refuses to take a brain's word for success, and a trace already keeps a
+run's decisions so they can be replayed and scored. What neither does is make the *record*
+checkable. A JSON file can be edited by whoever would prefer a better result, and nothing
+notices; a directory of screenshots can lose the one frame that proves a failure, and nothing
+notices either.
+
+An audit log is an append-only JSONL whose every line carries the digest of the line before
+it. Change one entry and its own digest stops matching; delete one and the next entry's `prev`
+points at an entry that is not there; reorder them and the indices give it away.
+:func:`verify_audit` says which entry is at fault, by number, because "the log is fine" is
+only useful if a reader can check it.
+
+    log = AuditLog(Path("runs/today.jsonl"))
+    loop = AgentLoop(screen=..., brain=..., audit=log)     # one entry per step
+    loop.run("click 'Save' to save the document")
+
+    ok, detail = verify_audit("runs/today.jsonl")          # (True, "3 entries, chain intact")
+
+What this is not: a signature. A hash chain catches edits, not an author who simply writes a
+whole new chain. It is evidence against accident and casual tampering, and it says so.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Iterable
+
+from tiny_computer_agent.observe import Observation
+
+__all__ = ["AuditEntry", "AuditLog", "observation_digest", "read_audit", "verify_audit"]
+
+
+def _canonical(payload: dict[str, Any]) -> str:
+    """One stable spelling per entry, so a digest means the same thing twice."""
+    return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def observation_digest(observation: Observation) -> str:
+    """A digest of what was on screen: the element index, not the pixels.
+
+    Two frames with the same labels in the same places hash the same, which is the point -
+    a step is tied to the screen it was decided on without storing a screenshot.
+    """
+    index = [[element.label, element.kind, element.x, element.y, element.width, element.height]
+             for element in observation.elements]
+    return hashlib.sha256(_canonical({"index": index, "text": observation.text[:200]})
+                          .encode("utf-8")).hexdigest()[:16]
+
+
+@dataclass(frozen=True)
+class AuditEntry:
+    """One line of the log: a step, or the episode that closed it."""
+
+    kind: str                       # "step" | "episode"
+    index: int
+    at: str
+    goal: str
+    screen: str
+    action: str
+    outcome: str
+    ok: bool
+    prev: str
+    digest: str = ""
+
+    def payload(self) -> dict[str, Any]:
+        """Everything the digest covers - which is everything except the digest."""
+        return {"kind": self.kind, "index": self.index, "at": self.at, "goal": self.goal,
+                "screen": self.screen, "action": self.action, "outcome": self.outcome,
+                "ok": self.ok, "prev": self.prev}
+
+    def compute_digest(self) -> str:
+        return hashlib.sha256(_canonical(self.payload()).encode("utf-8")).hexdigest()
+
+    def to_dict(self) -> dict[str, Any]:
+        return {**self.payload(), "digest": self.digest}
+
+    @classmethod
+    def from_dict(cls, raw: dict[str, Any]) -> "AuditEntry":
+        return cls(kind=raw.get("kind", "step"), index=raw.get("index", -1),
+                   at=raw.get("at", ""), goal=raw.get("goal", ""), screen=raw.get("screen", ""),
+                   action=raw.get("action", ""), outcome=raw.get("outcome", ""),
+                   ok=bool(raw.get("ok", False)), prev=raw.get("prev", ""),
+                   digest=raw.get("digest", ""))
+
+
+@dataclass
+class AuditLog:
+    """An append-only, hash-chained record of what an episode did."""
+
+    path: Path | str
+    _last: str = field(default="", init=False, repr=False)
+    _count: int = field(default=0, init=False, repr=False)
+
+    def __post_init__(self) -> None:
+        self.path = Path(self.path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        # resuming a log keeps its chain: pick up the last digest rather than starting over
+        if self.path.exists():
+            existing = read_audit(self.path)
+            self._count = len(existing)
+            if existing:
+                self._last = existing[-1].digest
+
+    # -- writing ----------------------------------------------------------- #
+
+    def _write(self, kind: str, *, goal: str, screen: str = "", action: str = "",
+               outcome: str = "", ok: bool = False) -> AuditEntry:
+        entry = AuditEntry(kind=kind, index=self._count,
+                           at=datetime.now(timezone.utc).isoformat(timespec="seconds"),
+                           goal=goal, screen=screen, action=action, outcome=outcome,
+                           ok=ok, prev=self._last)
+        entry = AuditEntry(**{**entry.payload(), "digest": entry.compute_digest()})
+        with open(self.path, "a", encoding="utf-8") as handle:
+            handle.write(_canonical(entry.to_dict()) + "\n")
+        self._last, self._count = entry.digest, self._count + 1
+        return entry
+
+    def append(self, step: Any, goal: str = "") -> AuditEntry:
+        """Record one step. Accepts the loop's :class:`~tiny_computer_agent.loop.Step`."""
+        result = getattr(step, "result", None)
+        outcome = (getattr(step, "error", None) or (result.summary() if result else "")) or ""
+        return self._write("step", goal=goal or "",
+                           screen=observation_digest(step.observation),
+                           action=step.action.describe() if step.action else "",
+                           outcome=str(outcome), ok=bool(step.ok))
+
+    def close(self, episode: Any) -> AuditEntry:
+        """Record how the episode ended - verified, failed, refused, or out of budget."""
+        return self._write("episode", goal=getattr(episode, "goal", ""),
+                           outcome=f"{'success' if episode.success else 'failure'}: "
+                                   f"{episode.stop_reason}",
+                           ok=bool(episode.success))
+
+    # -- reading ----------------------------------------------------------- #
+
+    def entries(self) -> list[AuditEntry]:
+        return read_audit(self.path)
+
+    def verify(self) -> tuple[bool, str]:
+        return verify_audit(self.path)
+
+    def render(self) -> str:
+        entries = self.entries()
+        steps = [e for e in entries if e.kind == "step"]
+        ok, detail = self.verify()
+        counts: dict[str, int] = {}
+        for entry in steps:
+            label = "ok" if entry.ok else "failed"
+            counts[label] = counts.get(label, 0) + 1
+        tally = ", ".join(f"{n} {label}" for label, n in sorted(counts.items())) or "no steps"
+        return (f"audit: {len(entries)} entries ({len(steps)} steps, {tally}) | "
+                f"chain {'intact' if ok else 'BROKEN'}: {detail}")
+
+
+def read_audit(path: Path | str) -> list[AuditEntry]:
+    """Every entry in the log, in order. A malformed line is an entry that fails to load."""
+    path = Path(path)
+    if not path.exists():
+        return []
+    entries: list[AuditEntry] = []
+    with open(path, encoding="utf-8") as handle:
+        for line in handle:
+            if line.strip():
+                entries.append(AuditEntry.from_dict(json.loads(line)))
+    return entries
+
+
+def verify_audit(path: Path | str) -> tuple[bool, str]:
+    """Check the chain. Returns ``(ok, detail)``, naming the first entry at fault.
+
+    Deliberately reports *which* entry broke and *how* - a digest that does not match, a
+    `prev` pointing at nothing, or an index out of order - because those three failures mean
+    different things to whoever has to decide whether to trust the run.
+    """
+    path = Path(path)
+    if not path.exists():
+        return False, f"no audit log at {path}"
+
+    through = ""
+    count = 0
+    for number, line in enumerate((l for l in path.read_text(encoding="utf-8").splitlines()
+                                   if l.strip()), start=1):
+        entry = AuditEntry.from_dict(json.loads(line))
+        if entry.index != number - 1:
+            return False, (f"entry {number} is out of order (index {entry.index} where "
+                           f"{number - 1} was expected): lines were removed or reordered")
+        if entry.prev != through:
+            return False, (f"entry {number} does not follow entry {number - 1}: its link "
+                           f"points at an entry that is not there")
+        if entry.compute_digest() != entry.digest:
+            return False, f"entry {number} was edited after it was written"
+        through = entry.digest
+        count += 1
+    if count == 0:
+        return False, f"{path} is empty"
+    return True, f"{count} entries, chain intact"

tiny_computer_agent/gates.py

@@ -161,3 +161,37 @@ class AnyOf:
             if approval.approved:
                 return approval
         return Approval(False, self.reason)
+
+
+GATE_SPECS = ("allow-all", "read-only", "interactive", "labels:")
+"""The spellings :func:`gate_from_spec` accepts, in the order its error lists them."""
+
+
+def gate_from_spec(spec: str) -> Gate:
+    """Build a gate from a one-line spelling, the one the command line takes.
+
+    A gate is the operator's yes, and an operator should not have to write Python to give
+    one::
+
+        allow-all              everything policy already allowed
+        read-only              nothing that changes the machine, whatever policy says
+        interactive            ask me, once per action, and remember "always"/"never"
+        labels:Save,Cancel     only these labels, and nothing else
+
+    An unknown spec raises ``ValueError`` naming the accepted spellings, because a typo in
+    a safety switch must never quietly mean "allow everything".
+    """
+    wanted = (spec or "").strip().lower()
+    if wanted in ("allow-all", "allow", "none"):
+        return AllowAll()
+    if wanted in ("read-only", "readonly"):
+        return DenyAll(reason="read-only gate: nothing may change the machine")
+    if wanted in ("interactive", "ask"):
+        return InteractiveGate()
+    if wanted.startswith("labels:"):
+        labels = [part.strip() for part in spec.split(":", 1)[1].split(",") if part.strip()]
+        if not labels:
+            raise ValueError("labels: needs at least one label, e.g. labels:Save,Cancel")
+        return AllowOnlyLabels(tuple(labels))
+    accepted = " | ".join(GATE_SPECS)
+    raise ValueError(f"unknown gate {spec!r}: expected one of {accepted}")

tiny_computer_agent/loop.py

@@ -7,10 +7,12 @@ from typing import Any, Callable
 
 from tiny_computer_agent.act import ActionResult, DryRunExecutor, Executor, PyAutoGuiExecutor
 from tiny_computer_agent.actions import Action
+from tiny_computer_agent.audit import AuditLog
 from tiny_computer_agent.brain import Brain, GoalUnreachable
 from tiny_computer_agent.observe import Observation, ScreenSource
 from tiny_computer_agent.gates import Approval, Gate
 from tiny_computer_agent.safety import Policy, PolicyViolation
+from tiny_computer_agent.watchdog import Watchdog, WatchdogTimeout
 
 
 @dataclass(frozen=True)
@@ -86,7 +88,9 @@ class AgentLoop:
                  on_step: Callable[[Step], None] | None = None,
                  max_repeats: int = 1,
                  verify: Callable[[Observation], bool] | None = None,
-                 gate: Gate | None = None) -> None:
+                 gate: Gate | None = None,
+                 watchdog: Watchdog | None = None,
+                 audit: AuditLog | None = None) -> None:
         """``verify`` runs after every action: the harness, not the model, decides
         whether the goal was reached. A brain saying "done" is a claim; a verifier
         looking at the screen is evidence."""
@@ -99,6 +103,9 @@ class AgentLoop:
         self.on_step = on_step
         self.verify = verify
         self.gate = gate
+        self.watchdog = watchdog
+        self.audit = audit
+        self._audit_closed = False
         self.max_repeats = max_repeats
         # the stepwise state: an episode in progress, and the guards around it
         self.episode: Episode | None = None
@@ -127,8 +134,12 @@ class AgentLoop:
         self._repeats = 0
         self._done = False
         self.paused = False
+        self._audit_closed = False
         try:
-            self.observation = self.screen.capture()
+            self.observation = self._guard("capture", self.screen.capture)
+        except WatchdogTimeout as exc:
+            self.observation = None
+            self._finish(f"watchdog stopped the episode: {exc}")
         except Exception as exc:
             self.observation = None
             self._finish(f"capture failed: {type(exc).__name__}: {exc}")
@@ -171,7 +182,10 @@ class AgentLoop:
         index, self._turns = self._turns, self._turns + 1
 
         try:
-            action = self.brain.decide(episode.goal, observation)
+            action = self._guard("decide", self.brain.decide, episode.goal, observation)
+        except WatchdogTimeout as exc:
+            return self._record_step(Step(index, observation, error=f"watchdog: {exc}"),
+                                     f"watchdog stopped the episode: {exc}")
         except GoalUnreachable as exc:
             return self._record_step(Step(index, observation, error=f"gave up: {exc}"),
                                      f"brain gave up: {exc}")
@@ -214,7 +228,11 @@ class AgentLoop:
                 f"screen ({action.kind} {action.label or action.text or ''})".strip())
 
         try:
-            result = self.executor.execute(action, observation)
+            result = self._guard("execute", self.executor.execute, action, observation)
+        except WatchdogTimeout as exc:
+            return self._record_step(
+                Step(index, observation, action=action, error=f"watchdog: {exc}"),
+                f"watchdog stopped the episode: {exc}")
         except Exception as exc:
             return self._record_step(
                 Step(index, observation, action=action,
@@ -228,7 +246,11 @@ class AgentLoop:
             return step
 
         try:
-            self.observation = self.screen.capture()     # verify: what did that change?
+            self.observation = self._guard("capture", self.screen.capture)
+        except WatchdogTimeout as exc:                   # verify: what did that change?
+            self._append(Step(index + 1, observation, error=f"watchdog: {exc}"))
+            self._finish(f"watchdog stopped the episode: {exc}")
+            return step
         except Exception as exc:
             self._append(Step(index + 1, observation,
                               error=f"capture failed: {type(exc).__name__}: {exc}"))
@@ -261,6 +283,8 @@ class AgentLoop:
 
     def _append(self, step: Step) -> None:
         self.episode.steps.append(step)
+        if self.audit is not None:
+            self.audit.append(step, self.episode.goal)
         if self.on_step is not None:
             self.on_step(step)
 
@@ -272,6 +296,15 @@ class AgentLoop:
     def _finish(self, stop_reason: str) -> None:
         self.episode.stop_reason = stop_reason
         self._done = True
+        if self.audit is not None and not self._audit_closed:   # one closing entry, ever
+            self._audit_closed = True
+            self.audit.close(self.episode)
+
+    def _guard(self, where: str, call: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
+        """One guarded call: with a watchdog, a hung call becomes a named failure."""
+        if self.watchdog is None:
+            return call(*args, **kwargs)
+        return self.watchdog.call(where, call, *args, **kwargs)
 
     def _verified(self, observation: Observation) -> bool:
         """A broken verifier must never fake a success."""

tiny_computer_agent/watchdog.py

@@ -0,0 +1,147 @@
+"""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