File size: 5,728 Bytes
e2f048d | 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 | """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"
|