File size: 6,167 Bytes
03c1ae9 | 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 149 150 151 152 153 154 155 156 157 158 159 | """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
|