Tests: watchdog, audit and the AT-SPI guard

tests/conftest.py

@@ -20,3 +20,19 @@ def require_ocr() -> None:
     available, detail = detector_available("ocr")
     if not available:
         pytest.skip(f"no OCR engine installed: {detail}")
+
+
+def require_atspi() -> None:
+    """Skip the calling test unless the AT-SPI reader can actually be built.
+
+    The same trap as `require_ocr`, one layer down. `make_detector("atspi")` is
+    right to raise when PyGObject is missing - but a test that calls it on a
+    machine without the typelib *fails* where it should skip, and the failure
+    reads like a harness defect. Ask the backend that needs the typelib, never a
+    name that resolves anyway.
+    """
+    from tiny_computer_agent.accessibility import accessibility_available
+
+    available, detail = accessibility_available("atspi")
+    if not available:
+        pytest.skip(f"no AT-SPI reader on this machine: {detail}")

tests/test_accessibility.py

@@ -8,6 +8,8 @@ one test that drives actual AT-SPI is opt-in, like the live-desktop tests.
 import os
 
 import pytest
+
+from conftest import require_atspi
 from PIL import Image
 
 import platform
@@ -241,11 +243,12 @@ def test_the_auto_detector_still_defaults_to_ocr_then_grid():
 
 
 def test_the_new_detector_names_are_reachable():
-    assert type(make_detector("atspi")).__name__ == "AtspiDetector"
     assert type(make_detector("best")).__name__ == "AutoDetector"
+    assert TEXT_ROLES and "label" in TEXT_ROLES
+    require_atspi()                     # the atspi name needs the typelib to build
+    assert type(make_detector("atspi")).__name__ == "AtspiDetector"
     ok, detail = detector_available("atspi")
     assert isinstance(ok, bool) and detail
-    assert TEXT_ROLES and "label" in TEXT_ROLES
 
 
 # -- the Windows backend: its mapping, exercised without Windows ---------------
@@ -331,12 +334,17 @@ def test_the_uia_reader_refuses_rather_than_returning_an_empty_index():
         UiaReader().read()
 
 
-def test_the_detector_names_cover_both_backends():
-    assert type(make_detector("atspi")).__name__ == "AtspiDetector"
+def test_the_windows_detector_name_is_reachable_without_windows():
+    """The Windows backend's *mapping* is exercised anywhere; its tree is not."""
     ok, detail = detector_available("uia")
     assert isinstance(ok, bool) and detail
 
 
+def test_the_detector_names_cover_both_backends():
+    require_atspi()
+    assert type(make_detector("atspi")).__name__ == "AtspiDetector"
+
+
 def test_a_provider_that_throws_is_reported_as_unavailable():
     def broken():
         raise RuntimeError("COM is not initialised")

tests/test_api_surface.py

@@ -21,7 +21,7 @@ def test_no_public_callable_in_a_module_is_missing_from_all():
     modules = ["actions", "act", "observe", "perception", "brain", "models", "loop",
                "safety", "desktop", "demo", "trace",
                # 2.0: the operator layer, all of which must stay importable from the root
-               "diff", "gates", "runner", "skills"]
+               "diff", "gates", "runner", "skills", "watchdog", "audit"]
     unexported = []
     for name in modules:
         module = importlib.import_module(f"tiny_computer_agent.{name}")

tests/test_atspi_guard.py

@@ -0,0 +1,46 @@
+"""The AT-SPI guard must actually skip when there is no typelib.
+
+Same bug, one layer down from the OCR guard: `make_detector("atspi")` correctly
+raises without PyGObject, so two tests failed on a machine that has no AT-SPI
+reader - and a failing test reads like a harness defect rather than a missing
+system package. The guard now asks the backend that needs the typelib, and here
+is the proof that it skips, passes, and asks for the right thing.
+"""
+
+import pytest
+
+from conftest import require_atspi
+
+import tiny_computer_agent.accessibility as accessibility
+
+
+def test_the_guard_skips_when_the_typelib_is_missing(monkeypatch):
+    monkeypatch.setattr(accessibility, "accessibility_available",
+                        lambda name="atspi": (False, "simulated: no PyGObject"))
+    with pytest.raises(pytest.skip.Exception):
+        require_atspi()
+
+
+def test_the_guard_passes_when_the_reader_can_be_built(monkeypatch):
+    monkeypatch.setattr(accessibility, "accessibility_available",
+                        lambda name="atspi": (True, "simulated: AT-SPI reader ready"))
+    require_atspi()          # must not raise
+
+
+def test_the_guard_asks_for_atspi_and_not_a_name_that_always_resolves(monkeypatch):
+    asked = []
+
+    def record(name="atspi"):
+        asked.append(name)
+        return True, "recorded"
+
+    monkeypatch.setattr(accessibility, "accessibility_available", record)
+    require_atspi()
+    assert asked == ["atspi"]
+
+
+def test_uia_is_reported_unavailable_on_linux_and_that_is_not_a_failure():
+    """Why the Windows test had to be split rather than guarded as a whole."""
+    available, detail = accessibility.accessibility_available("uia")
+    assert available is False, "UI Automation is Windows-only"
+    assert detail, "an unavailable backend still has to say why"

tests/test_audit.py

@@ -0,0 +1,146 @@
+"""The record must be checkable, and must say when it is not.
+
+A run's JSON can be edited by whoever would prefer a better result. These tests write a real
+log from a real episode, then do the three things a dishonest reader would try - edit a line,
+delete a line, reorder them - and require the verifier to name the entry at fault.
+"""
+
+import json
+
+from tiny_computer_agent import (
+    AgentLoop, AuditLog, Element, FakeScreen, MockBrain, observation_digest, verify_audit,
+)
+
+
+def _screen():
+    return FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)])
+
+
+def _brain():
+    return MockBrain([{"kind": "click", "label": "Save"},
+                      {"kind": "done", "reason": "file saved"}])
+
+
+def _run(tmp_path, name="audit.jsonl", **kwargs):
+    log = AuditLog(tmp_path / name)
+    loop = AgentLoop(screen=_screen(), brain=_brain(), audit=log, **kwargs)
+    return loop.run("click 'Save' to save the document"), log
+
+
+# -- writing ------------------------------------------------------------------- #
+
+def test_a_run_writes_one_chained_entry_per_step_plus_a_closing_one(tmp_path):
+    episode, log = _run(tmp_path)
+    entries = log.entries()
+    assert [e.kind for e in entries] == ["step", "step", "episode"]
+    assert [e.index for e in entries] == [0, 1, 2]
+    assert entries[0].prev == "", "the first entry follows nothing"
+    assert entries[1].prev == entries[0].digest
+    assert all(e.digest and len(e.digest) == 64 for e in entries)
+    assert "click label='Save'" in entries[0].action
+
+
+def test_the_chain_verifies_after_a_real_run(tmp_path):
+    _run(tmp_path)
+    ok, detail = verify_audit(tmp_path / "audit.jsonl")
+    assert ok, detail
+    assert detail == "3 entries, chain intact"
+
+
+def test_the_closing_entry_carries_the_verdict_not_the_brains_word(tmp_path):
+    episode, log = _run(tmp_path)
+    assert episode.success
+    closing = log.entries()[-1]
+    assert closing.kind == "episode"
+    assert closing.ok is True
+    assert "success" in closing.outcome and episode.stop_reason in closing.outcome
+
+
+def test_render_counts_the_steps_and_the_chain(tmp_path):
+    _, log = _run(tmp_path)
+    line = log.render()
+    assert "2 steps" in line
+    assert "chain intact" in line
+
+
+def test_a_run_with_no_audit_writes_nothing(tmp_path):
+    episode = AgentLoop(screen=_screen(), brain=_brain()).run("save")
+    assert episode.success
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_resuming_a_log_continues_its_chain(tmp_path):
+    """A runner that is killed and restarted must not start a second, unrelated log."""
+    _run(tmp_path)
+    episode, log = _run(tmp_path)                      # same path, fresh AuditLog
+    assert episode.success
+    entries = log.entries()
+    assert [e.index for e in entries] == [0, 1, 2, 3, 4, 5]
+    assert entries[3].prev == entries[2].digest
+    assert verify_audit(tmp_path / "audit.jsonl")[0] is True
+
+
+# -- tampering ----------------------------------------------------------------- #
+
+def _rewrite(tmp_path, lines, name="audit.jsonl"):
+    (tmp_path / name).write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def test_an_edited_entry_is_caught_and_named(tmp_path):
+    _run(tmp_path)
+    path = tmp_path / "audit.jsonl"
+    lines = path.read_text(encoding="utf-8").splitlines()
+    entry = json.loads(lines[1])
+    entry["outcome"] = "clicked without incident"       # a nicer story
+    lines[1] = json.dumps(entry, sort_keys=True, separators=(",", ":"))
+    _rewrite(tmp_path, lines)
+
+    ok, detail = verify_audit(path)
+    assert ok is False
+    assert "entry 2" in detail and "was edited" in detail
+
+
+def test_a_deleted_line_is_caught(tmp_path):
+    _run(tmp_path)
+    path = tmp_path / "audit.jsonl"
+    lines = path.read_text(encoding="utf-8").splitlines()
+    _rewrite(tmp_path, [lines[0], lines[2]])
+    ok, detail = verify_audit(path)
+    assert ok is False
+    assert "entry 2" in detail and "out of order" in detail
+
+
+def test_reordered_lines_are_caught(tmp_path):
+    _run(tmp_path)
+    path = tmp_path / "audit.jsonl"
+    lines = path.read_text(encoding="utf-8").splitlines()
+    _rewrite(tmp_path, [lines[1], lines[0], lines[2]])
+    ok, detail = verify_audit(path)
+    assert ok is False
+    assert "entry 1" in detail
+
+
+def test_a_missing_log_is_not_reported_as_fine(tmp_path):
+    ok, detail = verify_audit(tmp_path / "nothing-here.jsonl")
+    assert ok is False and "no audit log" in detail
+
+
+def test_an_empty_log_is_not_reported_as_fine(tmp_path):
+    path = tmp_path / "empty.jsonl"
+    path.write_text("")
+    ok, detail = verify_audit(path)
+    assert ok is False and "empty" in detail
+
+
+# -- the screen digest --------------------------------------------------------- #
+
+def test_the_digest_identifies_the_index_not_the_pixels(tmp_path):
+    same = FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)]).capture()
+    again = FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)]).capture()
+    moved = FakeScreen([(Element("Save", x=400, y=120, width=80, height=24),)]).capture()
+    renamed = FakeScreen([(Element("Cancel", x=320, y=120, width=80, height=24),)]).capture()
+
+    assert observation_digest(same) == observation_digest(again)
+    assert observation_digest(same) != observation_digest(moved)
+    assert observation_digest(same) != observation_digest(renamed)
+    assert len(observation_digest(same)) == 16

tests/test_cli.py

@@ -3,6 +3,7 @@ import json
 import pytest
 
 from tiny_computer_agent import __main__ as cli
+from tiny_computer_agent import verify_audit
 
 
 def test_demo_runs_dry_run_by_default(capsys):
@@ -153,3 +154,61 @@ def test_batch_rejects_tasks_that_are_not_tasks(tmp_path, capsys):
     missing = tmp_path / "absent.json"
     assert cli.main(["batch", str(missing)]) == 2
     assert "cannot read" in capsys.readouterr().err
+
+
+# -- 2.0: the operator's gate, and a record of the run that can be checked ------- #
+
+SCRIPT = '[{"kind": "click", "label": "Save"}, {"kind": "done", "reason": "saved"}]'
+
+
+def test_the_audit_flag_writes_a_chain_that_verifies(tmp_path, capsys):
+    path = tmp_path / "run.jsonl"
+    assert cli.main(["--image", "example", "--script", SCRIPT, "--audit", str(path)]) == 0
+    out = capsys.readouterr().out
+    assert "recording this run to" in out
+    assert "audit:" in out and "chain intact" in out
+    ok, detail = verify_audit(path)
+    assert ok, detail
+
+
+def test_verify_audit_accepts_an_intact_chain(tmp_path, capsys):
+    path = tmp_path / "run.jsonl"
+    cli.main(["--image", "example", "--script", SCRIPT, "--audit", str(path)])
+    assert cli.main(["--verify-audit", str(path)]) == 0
+    assert "ok" in capsys.readouterr().out
+
+
+def test_verify_audit_fails_loudly_on_a_tampered_chain(tmp_path, capsys):
+    path = tmp_path / "run.jsonl"
+    cli.main(["--image", "example", "--script", SCRIPT, "--audit", str(path)])
+    lines = path.read_text(encoding="utf-8").splitlines()
+    entry = json.loads(lines[0])
+    entry["outcome"] = "nothing happened"                 # a nicer story
+    lines[0] = json.dumps(entry, sort_keys=True, separators=(",", ":"))
+    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+    assert cli.main(["--verify-audit", str(path)]) == 1
+    out = capsys.readouterr().out
+    assert "FAIL" in out and "entry 1" in out
+
+
+def test_a_bad_gate_spec_is_a_usage_error(capsys):
+    assert cli.main(["--gate", "yes-please"]) == 2
+    assert "bad --gate" in capsys.readouterr().err
+
+
+def test_the_read_only_gate_refuses_to_change_anything(capsys):
+    """The gate is the operator's yes; read-only means the answer is no to any change."""
+    code = cli.main(["--image", "example", "--script",
+                     '[{"kind": "click", "label": "Save"}]', "--gate", "read-only"])
+    out = capsys.readouterr().out
+    assert code == 1, "a refused action is a failed episode, not a success"
+    assert "gate: read-only" in out
+    assert "operator refused" in out and "read-only gate" in out
+
+
+def test_the_labels_gate_lets_a_whitelisted_click_through(capsys):
+    code = cli.main(["--image", "example", "--script", SCRIPT, "--gate", "labels:Save"])
+    out = capsys.readouterr().out
+    assert code == 0, out
+    assert "is whitelisted" in out or "would click" in out

tests/test_gates.py

@@ -1,8 +1,10 @@
+import pytest
+
 """A refusal from the operator is a step, not a traceback."""
 
 from tiny_computer_agent import (
     Action, AgentLoop, AllOf, AllowAll, AllowOnlyLabels, AnyOf, Approval, DenyAll, Element,
-    FakeScreen, InteractiveGate, MockBrain, Policy, mutating,
+    FakeScreen, InteractiveGate, MockBrain, Policy, gate_from_spec, mutating,
 )
 
 
@@ -95,3 +97,38 @@ def test_a_gate_refusal_stops_the_episode_and_is_recorded():
     assert episode.steps[0].error is not None and episode.steps[0].error.startswith("not approved")
     assert episode.steps[0].result is None, "nothing was executed"
     assert len(episode.steps) == 1, "the episode ended there"
+
+
+# -- the one-line spelling the command line takes -------------------------------- #
+
+def test_gate_from_spec_read_only_allows_a_read_and_refuses_a_change():
+    gate = gate_from_spec("read-only")
+    assert gate.approve(Action("wait", seconds=1)).approved
+    refusal = gate.approve(Action("click", label="Save"))
+    assert refusal.approved is False
+    assert "read-only gate" in refusal.reason
+
+
+def test_gate_from_spec_labels_normalise_what_the_operator_wrote():
+    gate = gate_from_spec("labels: save , Cancel")
+    assert gate.approve(Action("click", label="Save")).approved
+    assert gate.approve(Action("key", keys="ctrl+s")).approved is False
+
+
+def test_gate_from_spec_interactive_asks_the_operator():
+    assert type(gate_from_spec("interactive")).__name__ == "InteractiveGate"
+
+
+def test_gate_from_spec_allow_all_approves_whatever_policy_allowed():
+    assert gate_from_spec("allow-all").approve(Action("click", label="Save")).approved
+
+
+def test_an_unknown_spec_says_what_it_accepts():
+    """A typo in a safety switch must never quietly mean "allow everything"."""
+    with pytest.raises(ValueError, match="allow-all"):
+        gate_from_spec("yes-please")
+
+
+def test_labels_with_no_labels_is_a_configuration_error():
+    with pytest.raises(ValueError, match="at least one label"):
+        gate_from_spec("labels:  ,  ")

tests/test_watchdog.py

@@ -0,0 +1,158 @@
+"""A call that never returns must not be a harness that never returns.
+
+The loop stops a *repeated* action and a step budget stops a long episode, but nothing
+stopped a single call that hangs: no step, no reason, no evidence, and a console that looks
+like it is thinking. These tests pin the watchdog's two promises - the loop stops waiting and
+says which call it stopped waiting for - and its one honest limit, that a Python thread
+cannot be killed, only abandoned.
+"""
+
+import time
+
+import pytest
+
+from tiny_computer_agent import (
+    AgentLoop, DryRunExecutor, Element, FakeScreen, MockBrain, Watchdog, WatchdogTimeout,
+)
+
+
+# -- the unit: one guarded call ------------------------------------------------ #
+
+def test_a_call_that_returns_passes_its_value_through():
+    watchdog = Watchdog(step_timeout=5)
+    assert watchdog.call("decide", lambda a, b: a + b, 2, 3) == 5
+    assert watchdog.orphans == []
+
+
+def test_a_call_that_overruns_is_abandoned_and_named():
+    watchdog = Watchdog(step_timeout=0.05)
+    started = time.monotonic()
+    with pytest.raises(WatchdogTimeout) as raised:
+        watchdog.call("execute", lambda: time.sleep(3))
+    elapsed = time.monotonic() - started
+
+    assert elapsed < 1.5, "the watchdog waited for the deadline, not for the call"
+    assert raised.value.where == "execute"
+    assert raised.value.seconds == 0.05
+    assert "execute did not finish within 0.05s" in str(raised.value)
+    assert watchdog.orphans == ["execute"], "an abandoned call is recorded, not forgiven"
+
+
+def test_the_wall_clock_refuses_the_next_call_before_it_starts():
+    """A spent wall clock is reported as un-attempted, never as a timeout of what was next."""
+    watchdog = Watchdog(step_timeout=5, wall_clock=0).start()
+    called = []
+    with pytest.raises(WatchdogTimeout) as raised:
+        watchdog.call("decide", lambda: called.append("ran"))
+    assert called == [], "nothing may be attempted once the wall clock is spent"
+    assert raised.value.abandoned is False
+    assert "not attempted" in str(raised.value)
+
+
+def test_no_per_call_deadline_runs_the_call_inline():
+    """The honest default for a local desktop: this process *is* the thing being watched."""
+    watchdog = Watchdog(step_timeout=None)
+    assert watchdog.call("execute", lambda: "inline") == "inline"
+    assert watchdog.threads_started == 0
+
+
+def test_an_exception_inside_a_guarded_call_reaches_the_caller_unchanged():
+    def broken():
+        raise ValueError("the executor is broken, not slow")
+
+    with pytest.raises(ValueError, match="not slow"):
+        Watchdog(step_timeout=5).call("execute", broken)
+
+
+def test_render_names_the_limits_and_every_abandoned_call():
+    watchdog = Watchdog(step_timeout=0.05, wall_clock=60).start()
+    with pytest.raises(WatchdogTimeout):
+        watchdog.call("decide", lambda: time.sleep(3))
+    line = watchdog.render()
+    assert "per call <= 0.05s" in line
+    assert "wall clock <= 60s" in line
+    assert "abandoned: decide" in line
+
+
+def test_the_clock_starts_once_and_is_not_restarted_by_a_re_run():
+    watchdog = Watchdog(wall_clock=60).start()
+    time.sleep(0.02)
+    first = watchdog.elapsed
+    watchdog.start()                       # re-running a goal must not hand it a fresh budget
+    assert watchdog.elapsed >= first
+
+
+def test_remaining_counts_down_and_then_reports_expired():
+    watchdog = Watchdog(wall_clock=0.05).start()
+    assert watchdog.remaining is not None and watchdog.remaining <= 0.05
+    time.sleep(0.06)
+    assert watchdog.expired is True
+    assert Watchdog(wall_clock=None).remaining is None
+
+
+# -- the loop: a hung stage becomes a step, not a stall ------------------------- #
+
+class SlowExecutor(DryRunExecutor):
+    seconds = 2.0
+
+    def execute(self, action, observation):
+        time.sleep(self.seconds)
+        return super().execute(action, observation)
+
+
+class SlowScreen(FakeScreen):
+    def capture(self):
+        time.sleep(2.0)
+        return super().capture()
+
+
+class SlowBrain(MockBrain):
+    def decide(self, goal, observation):
+        time.sleep(2.0)
+        return super().decide(goal, observation)
+
+
+def _screen():
+    return FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)])
+
+
+def test_a_hung_executor_ends_the_episode_with_a_named_reason():
+    loop = AgentLoop(screen=_screen(), brain=MockBrain([{"kind": "click", "label": "Save"}]),
+                     executor=SlowExecutor(), watchdog=Watchdog(step_timeout=0.05))
+    started = time.monotonic()
+    episode = loop.run("click 'Save' to save the document")
+    assert time.monotonic() - started < 2.0, "the loop must not wait for the hung call"
+
+    assert episode.success is False
+    assert episode.stop_reason.startswith("watchdog stopped the episode")
+    assert "execute did not finish within 0.05s" in episode.steps[-1].error
+    assert loop.watchdog.orphans == ["execute"]
+
+
+def test_a_hung_brain_ends_the_episode_too():
+    loop = AgentLoop(screen=_screen(), brain=SlowBrain([{"kind": "done", "reason": "x"}]),
+                     watchdog=Watchdog(step_timeout=0.05))
+    episode = loop.run("save")
+    assert episode.success is False
+    assert "decide did not finish within 0.05s" in episode.steps[-1].error
+    assert loop.watchdog.orphans == ["decide"]
+
+
+def test_a_hung_capture_is_recorded_before_the_episode_starts():
+    loop = AgentLoop(screen=SlowScreen([(Element("Save", x=1, y=1, width=10, height=10),)]),
+                     brain=MockBrain([{"kind": "done", "reason": "x"}]),
+                     watchdog=Watchdog(step_timeout=0.05))
+    episode = loop.run("save")
+    assert episode.success is False
+    assert "watchdog stopped the episode" in episode.stop_reason
+    assert "capture did not finish within 0.05s" in episode.stop_reason
+
+
+def test_without_a_watchdog_a_slow_but_finite_call_still_completes():
+    """Guarding is opt-in: a local desktop should not pay for a thread per step."""
+    loop = AgentLoop(screen=_screen(), brain=MockBrain([{"kind": "done", "reason": "fine"}]),
+                     executor=SlowExecutor())
+    loop.executor.seconds = 0.05
+    episode = loop.run("save")
+    assert episode.success is True
+    assert loop.watchdog is None