File size: 5,363 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
"""The record must be checkable, and must say when it is not.

A run's JSON can be edited by whoever would prefer a better result. These tests write a real
log from a real episode, then do the three things a dishonest reader would try - edit a line,
delete a line, reorder them - and require the verifier to name the entry at fault.
"""

import json

from tiny_computer_agent import (
    AgentLoop, AuditLog, Element, FakeScreen, MockBrain, observation_digest, verify_audit,
)


def _screen():
    return FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)])


def _brain():
    return MockBrain([{"kind": "click", "label": "Save"},
                      {"kind": "done", "reason": "file saved"}])


def _run(tmp_path, name="audit.jsonl", **kwargs):
    log = AuditLog(tmp_path / name)
    loop = AgentLoop(screen=_screen(), brain=_brain(), audit=log, **kwargs)
    return loop.run("click 'Save' to save the document"), log


# -- writing ------------------------------------------------------------------- #

def test_a_run_writes_one_chained_entry_per_step_plus_a_closing_one(tmp_path):
    episode, log = _run(tmp_path)
    entries = log.entries()
    assert [e.kind for e in entries] == ["step", "step", "episode"]
    assert [e.index for e in entries] == [0, 1, 2]
    assert entries[0].prev == "", "the first entry follows nothing"
    assert entries[1].prev == entries[0].digest
    assert all(e.digest and len(e.digest) == 64 for e in entries)
    assert "click label='Save'" in entries[0].action


def test_the_chain_verifies_after_a_real_run(tmp_path):
    _run(tmp_path)
    ok, detail = verify_audit(tmp_path / "audit.jsonl")
    assert ok, detail
    assert detail == "3 entries, chain intact"


def test_the_closing_entry_carries_the_verdict_not_the_brains_word(tmp_path):
    episode, log = _run(tmp_path)
    assert episode.success
    closing = log.entries()[-1]
    assert closing.kind == "episode"
    assert closing.ok is True
    assert "success" in closing.outcome and episode.stop_reason in closing.outcome


def test_render_counts_the_steps_and_the_chain(tmp_path):
    _, log = _run(tmp_path)
    line = log.render()
    assert "2 steps" in line
    assert "chain intact" in line


def test_a_run_with_no_audit_writes_nothing(tmp_path):
    episode = AgentLoop(screen=_screen(), brain=_brain()).run("save")
    assert episode.success
    assert list(tmp_path.iterdir()) == []


def test_resuming_a_log_continues_its_chain(tmp_path):
    """A runner that is killed and restarted must not start a second, unrelated log."""
    _run(tmp_path)
    episode, log = _run(tmp_path)                      # same path, fresh AuditLog
    assert episode.success
    entries = log.entries()
    assert [e.index for e in entries] == [0, 1, 2, 3, 4, 5]
    assert entries[3].prev == entries[2].digest
    assert verify_audit(tmp_path / "audit.jsonl")[0] is True


# -- tampering ----------------------------------------------------------------- #

def _rewrite(tmp_path, lines, name="audit.jsonl"):
    (tmp_path / name).write_text("\n".join(lines) + "\n", encoding="utf-8")


def test_an_edited_entry_is_caught_and_named(tmp_path):
    _run(tmp_path)
    path = tmp_path / "audit.jsonl"
    lines = path.read_text(encoding="utf-8").splitlines()
    entry = json.loads(lines[1])
    entry["outcome"] = "clicked without incident"       # a nicer story
    lines[1] = json.dumps(entry, sort_keys=True, separators=(",", ":"))
    _rewrite(tmp_path, lines)

    ok, detail = verify_audit(path)
    assert ok is False
    assert "entry 2" in detail and "was edited" in detail


def test_a_deleted_line_is_caught(tmp_path):
    _run(tmp_path)
    path = tmp_path / "audit.jsonl"
    lines = path.read_text(encoding="utf-8").splitlines()
    _rewrite(tmp_path, [lines[0], lines[2]])
    ok, detail = verify_audit(path)
    assert ok is False
    assert "entry 2" in detail and "out of order" in detail


def test_reordered_lines_are_caught(tmp_path):
    _run(tmp_path)
    path = tmp_path / "audit.jsonl"
    lines = path.read_text(encoding="utf-8").splitlines()
    _rewrite(tmp_path, [lines[1], lines[0], lines[2]])
    ok, detail = verify_audit(path)
    assert ok is False
    assert "entry 1" in detail


def test_a_missing_log_is_not_reported_as_fine(tmp_path):
    ok, detail = verify_audit(tmp_path / "nothing-here.jsonl")
    assert ok is False and "no audit log" in detail


def test_an_empty_log_is_not_reported_as_fine(tmp_path):
    path = tmp_path / "empty.jsonl"
    path.write_text("")
    ok, detail = verify_audit(path)
    assert ok is False and "empty" in detail


# -- the screen digest --------------------------------------------------------- #

def test_the_digest_identifies_the_index_not_the_pixels(tmp_path):
    same = FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)]).capture()
    again = FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)]).capture()
    moved = FakeScreen([(Element("Save", x=400, y=120, width=80, height=24),)]).capture()
    renamed = FakeScreen([(Element("Cancel", x=320, y=120, width=80, height=24),)]).capture()

    assert observation_digest(same) == observation_digest(again)
    assert observation_digest(same) != observation_digest(moved)
    assert observation_digest(same) != observation_digest(renamed)
    assert len(observation_digest(same)) == 16