| """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, |
| ) |
|
|
|
|
| |
|
|
| 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() |
| 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 |
|
|
|
|
| |
|
|
| 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 |
|
|
