tiny-computer-agent / tests /test_loop.py
hkblockchaintechnology's picture
Publish tests/ from marimo
627c6cf verified
Raw History Blame Contribute Delete
6.5 kB
import pytest
from tiny_computer_agent import (
Action, AgentLoop, DryRunExecutor, Element, Episode, FakeScreen,
GoalUnreachable, MockBrain, PlannerBrain, Policy,
)
from tiny_computer_agent.__main__ import demo_screen
def test_scripted_episode_succeeds_and_records_every_step():
screen = FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)])
brain = MockBrain([
{"kind": "type", "text": "report"},
{"kind": "key", "keys": "ctrl+s"},
{"kind": "done", "reason": "file saved"},
])
episode = AgentLoop(screen=screen, brain=brain).run("save the report")
assert episode.success
assert [s.action.kind for s in episode.steps] == ["type", "key", "done"]
assert "file saved" in episode.stop_reason
assert len(episode.to_rows()) == 3
def test_planner_solves_a_real_click_task_with_no_model():
episode = AgentLoop(screen=demo_screen(), brain=PlannerBrain()).run(
"click 'Save' to save the document"
)
assert episode.success
clicks = [s.action for s in episode.steps if s.action.kind == "click"]
assert [c.label for c in clicks] == ["Save"]
def test_screen_is_reread_after_every_executed_action():
"""The verify half of the loop: never decide twice on the same frame."""
screen = FakeScreen([(Element("Save"),)])
brain = MockBrain([{"kind": "click", "label": "Save"}, {"kind": "click", "label": "Save"}])
episode = AgentLoop(screen=screen, brain=brain).run("click Save")
executed = [s for s in episode.steps if s.result is not None and s.action.kind != "done"]
assert len(executed) == 2
assert screen.captures == len(executed) + 1 # one initial read, one per action
assert episode.steps[0].observation.index == 0
assert episode.steps[1].observation.index == 1
def test_step_budget_is_enforced():
screen = FakeScreen([(Element("Save"),)])
# distinct actions, so the budget is what stops it rather than the no-progress guard
brain = MockBrain([{"kind": "wait", "seconds": 0.5 + i} for i in range(10)])
episode = AgentLoop(
screen=screen, brain=brain, policy=Policy(step_budget=3)
).run("click Save")
assert episode.success is False
assert len(episode.steps) == 3
assert "budget of 3" in episode.stop_reason
def test_a_flailing_agent_is_stopped_instead_of_burning_the_budget():
"""Repeating one action against an unchanged screen is stuck, not persistent."""
brain = MockBrain([{"kind": "click", "label": "Save"}] * 10)
episode = AgentLoop(
screen=FakeScreen([(Element("Save"),)]), brain=brain,
policy=Policy(step_budget=10),
).run("click Save")
assert episode.success is False
assert "no progress" in episode.stop_reason
assert len(episode.steps) < 10
assert "unchanged screen" in episode.steps[-1].error
def test_progress_resets_the_repeat_counter():
"""Two different actions on an unchanged screen is not flailing."""
brain = MockBrain([
{"kind": "wait", "seconds": 0.5},
{"kind": "click", "label": "Save"},
{"kind": "wait", "seconds": 0.5},
{"kind": "click", "label": "Save"},
{"kind": "done", "reason": "assumed saved"},
])
episode = AgentLoop(
screen=FakeScreen([(Element("Save"),)]), brain=brain,
policy=Policy(step_budget=10),
).run("save")
assert episode.success
assert len(episode.steps) == 5
def test_a_dead_screen_is_reported_as_an_episode_not_a_traceback():
class Blind:
name = "blind"
def capture(self):
raise RuntimeError("X connection failed")
episode = AgentLoop(screen=Blind(), brain=MockBrain([{"kind": "done"}])).run("anything")
assert episode.success is False
assert episode.stop_reason.startswith("capture failed: RuntimeError")
assert episode.steps == []
def test_policy_refusal_stops_the_run_and_nothing_executes():
executor = DryRunExecutor()
episode = AgentLoop(
screen=FakeScreen([(Element("Terminal"),)]),
brain=MockBrain([{"kind": "type", "text": "sudo rm -rf /"}]),
executor=executor,
).run("clean up")
assert episode.success is False
assert "policy refusal" in episode.stop_reason
assert episode.steps[-1].error.startswith("refused:")
assert executor.performed == []
def test_kill_switch_aborts_before_the_first_action(tmp_path):
marker = tmp_path / "STOP"
marker.write_text("stop")
episode = AgentLoop(
screen=FakeScreen([(Element("Save"),)]),
brain=MockBrain([{"kind": "click", "label": "Save"}]),
policy=Policy(kill_switch_path=str(marker)),
).run("click Save")
assert episode.steps == []
assert episode.stop_reason == "kill switch engaged"
def test_a_broken_brain_stops_the_run_instead_of_crashing_it():
class Exploding:
name = "exploding"
def decide(self, goal, observation):
raise RuntimeError("model died")
episode = AgentLoop(screen=FakeScreen([(Element("Save"),)]), brain=Exploding()).run("save")
assert episode.success is False
assert episode.stop_reason == "brain error"
assert "RuntimeError: model died" in episode.steps[-1].error
def test_failed_action_ends_the_episode():
episode = AgentLoop(
screen=FakeScreen([(Element("Save"),)]),
brain=MockBrain([{"kind": "click", "label": "Nowhere"}]),
).run("click Nowhere")
assert episode.success is False
assert "action failed" in episode.stop_reason
def test_on_step_callback_sees_every_step_and_render_mentions_the_goal():
seen = []
episode = AgentLoop(
screen=demo_screen(), brain=PlannerBrain(), on_step=seen.append
).run("press Save")
assert len(seen) == len(episode.steps)
assert "press Save" in episode.render()
assert "success" in episode.render()
def test_a_brain_that_gives_up_is_a_failure_not_a_success():
"""The harness owns success: "I give up" must never read as "goal met"."""
episode = AgentLoop(
screen=FakeScreen([(Element("Files"),)]),
brain=PlannerBrain(max_waits=0),
).run("click 'Save'")
assert episode.success is False
assert episode.stop_reason.startswith("brain gave up:")
assert episode.steps[-1].error.startswith("gave up:")
def test_planner_raises_rather_than_inventing_a_target():
with pytest.raises(GoalUnreachable, match="no element name"):
PlannerBrain().decide("do the thing", FakeScreen([(Element("Save"),)]).capture())