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