| """A queue that survives being killed, and a report that counts what was verified.""" |
|
|
| import json |
|
|
| from tiny_computer_agent import ( |
| AgentLoop, DryRunExecutor, Element, FakeScreen, MockBrain, Policy, RunReport, Task, |
| TaskOutcome, TaskRunner, |
| ) |
|
|
|
|
| def element(label="Save", x=10, y=10): |
| return Element(label, x=x, y=y, width=60, height=20) |
|
|
|
|
| def build_loop(task): |
| """Fresh loop per task, exactly as the runner's contract requires.""" |
| brain = MockBrain([{"kind": "click", "label": "Save"}]) |
| return AgentLoop(screen=FakeScreen([(element(),)], text="Saved"), brain=brain, |
| executor=DryRunExecutor(), policy=Policy(step_budget=2)) |
|
|
|
|
| def test_a_task_names_itself_and_carries_the_harness_check(): |
| task = Task("click 'Save' to save the document", verify_text="Saved") |
|
|
| assert task.name == "save-document" |
| assert task.line().startswith("save-document") |
| verify = task.verifier() |
| assert verify is not None and verify.__call__ is not None |
| assert verify(type("O", (), {"text": "Saved"})()) is True |
| assert Task("no check here").verifier() is None |
| assert Task("explicit", name="mine").name == "mine" |
|
|
|
|
| def test_a_verified_task_is_reported_as_a_success(): |
| runner = TaskRunner(build_loop) |
| report = runner.run([Task("click 'Save' to save the document", verify_text="Saved")]) |
|
|
| assert isinstance(report, RunReport) |
| assert report.total == 1 and report.passed == 1 and report.accuracy == 1.0 |
| assert report.outcomes[0].success is True |
| assert report.outcomes[0].steps >= 1 |
| assert "verified" in report.render() or "1/1" in report.render() |
|
|
|
|
| def test_a_task_that_never_verifies_is_a_failure_not_a_guess(): |
| report = TaskRunner(build_loop).run([Task("click 'Save'", verify_text="NeverAppears", budget=1)]) |
|
|
| assert report.failed == 1 and report.accuracy == 0.0 |
| assert report.outcomes[0].success is False |
| assert "budget" in report.outcomes[0].stop_reason |
|
|
|
|
| def test_a_broken_loop_is_an_outcome_not_a_traceback(): |
| def exploding(task): |
| raise RuntimeError("no display here") |
|
|
| outcome = TaskRunner(exploding).run_one(Task("anything")) |
|
|
| assert outcome.success is False |
| assert outcome.error == "RuntimeError: no display here" |
| assert "caught" in outcome.stop_reason |
|
|
|
|
| def test_the_state_file_lets_a_killed_run_be_resumed(tmp_path): |
| state = tmp_path / "runs" / "today.jsonl" |
| tasks = [Task("click 'Save'", name="first", verify_text="Saved"), |
| Task("click 'Cancel'", name="second", verify_text="Saved")] |
| first = TaskRunner(build_loop, state_path=state).run(tasks) |
|
|
| assert first.total == 2 and first.passed == 2 |
| lines = [json.loads(line) for line in state.read_text().splitlines()] |
| assert [row["name"] for row in lines] == ["first", "second"] |
|
|
| seen = [] |
| def tracks(task): |
| seen.append(task.name) |
| return build_loop(task) |
|
|
| second = TaskRunner(tracks, state_path=state).run(tasks) |
| assert seen == [], "a resumed run does not repeat finished work" |
| assert second.total == 2 and second.passed == 2 |
| assert all("already run" in o.stop_reason for o in second.outcomes) |
|
|
|
|
| def test_a_half_written_state_line_does_not_break_a_resume(tmp_path): |
| state = tmp_path / "today.jsonl" |
| state.write_text(json.dumps(TaskOutcome("done", "g", True, 1, "verified", 0.5).to_dict()) + "\n" |
| + '{"name": "torn", "suc') |
| runner = TaskRunner(build_loop, state_path=state) |
| completed = runner.completed() |
|
|
| assert set(completed) == {"done"}, "a killed run leaves a torn line, and it is ignored" |
| report = runner.run([Task("g", name="done"), Task("click 'Save'", name="fresh", verify_text="Saved")]) |
| assert report.passed == 2 |
|
|
|
|
| def test_the_report_saves_and_lists(tmp_path): |
| report = TaskRunner(build_loop).run([Task("click 'Save'", verify_text="Saved")]) |
| path = report.save(tmp_path / "report.json") |
| saved = json.loads(path.read_text()) |
|
|
| assert saved["passed"] == 1 and saved["total"] == 1 |
| assert saved["outcomes"][0]["name"] == "save" |
| assert set(report.to_rows()[0]) == {"task", "goal", "success", "steps", "seconds", "reason"} |
| assert "report" in report.render() and report.steps >= 1 |
|
|
|
|
| def test_the_runner_tells_the_caller_about_each_task_as_it_finishes(): |
| heard = [] |
| runner = TaskRunner(build_loop, on_task=heard.append) |
| runner.run([Task("click 'Save'", verify_text="Saved")]) |
|
|
| assert len(heard) == 1 and heard[0].name == "save" |
|
|
|
|
| def test_a_task_can_be_verified_by_an_element_going_away(): |
| """The check for a task whose success is a dialog closing, where text says nothing.""" |
| task = Task("click 'Save' to save the document", verify_gone="Name") |
| verify = task.verifier() |
|
|
| assert "'Name' to go" in task.line() |
| assert verify is not None |
| assert verify.__call__ is not None |
|
|
| from tiny_computer_agent import Element, Observation |
|
|
| dialog = Observation(elements=(Element("Name", kind="input", x=10, y=10), |
| Element("Save", x=90, y=10))) |
| closed = Observation(elements=(Element("Save", x=90, y=10),)) |
| assert verify(dialog) is False, "the first frame only establishes the baseline" |
| assert verify(closed) is True |
|
|
|
|
| def test_a_task_can_be_verified_by_an_element_arriving(): |
| task = Task("open the File menu", verify_appeared="Save As") |
| verify = task.verifier() |
| from tiny_computer_agent import Element, Observation |
|
|
| menu = Observation(elements=(Element("File", x=0, y=0), |
| Element("Save As", x=0, y=30))) |
| assert verify(Observation(elements=(Element("File", x=0, y=0),))) is False |
| assert verify(menu) is True |
| assert task.to_dict()["verify_appeared"] == "Save As" |
|
|
