| """A run's record, chained so that editing it is louder than deleting it. |
| |
| The harness already refuses to take a brain's word for success, and a trace already keeps a |
| run's decisions so they can be replayed and scored. What neither does is make the *record* |
| checkable. A JSON file can be edited by whoever would prefer a better result, and nothing |
| notices; a directory of screenshots can lose the one frame that proves a failure, and nothing |
| notices either. |
| |
| An audit log is an append-only JSONL whose every line carries the digest of the line before |
| it. Change one entry and its own digest stops matching; delete one and the next entry's `prev` |
| points at an entry that is not there; reorder them and the indices give it away. |
| :func:`verify_audit` says which entry is at fault, by number, because "the log is fine" is |
| only useful if a reader can check it. |
| |
| log = AuditLog(Path("runs/today.jsonl")) |
| loop = AgentLoop(screen=..., brain=..., audit=log) # one entry per step |
| loop.run("click 'Save' to save the document") |
| |
| ok, detail = verify_audit("runs/today.jsonl") # (True, "3 entries, chain intact") |
| |
| What this is not: a signature. A hash chain catches edits, not an author who simply writes a |
| whole new chain. It is evidence against accident and casual tampering, and it says so. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| from dataclasses import dataclass, field |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Any, Iterable |
|
|
| from tiny_computer_agent.observe import Observation |
|
|
| __all__ = ["AuditEntry", "AuditLog", "observation_digest", "read_audit", "verify_audit"] |
|
|
|
|
| def _canonical(payload: dict[str, Any]) -> str: |
| """One stable spelling per entry, so a digest means the same thing twice.""" |
| return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False) |
|
|
|
|
| def observation_digest(observation: Observation) -> str: |
| """A digest of what was on screen: the element index, not the pixels. |
| |
| Two frames with the same labels in the same places hash the same, which is the point - |
| a step is tied to the screen it was decided on without storing a screenshot. |
| """ |
| index = [[element.label, element.kind, element.x, element.y, element.width, element.height] |
| for element in observation.elements] |
| return hashlib.sha256(_canonical({"index": index, "text": observation.text[:200]}) |
| .encode("utf-8")).hexdigest()[:16] |
|
|
|
|
| @dataclass(frozen=True) |
| class AuditEntry: |
| """One line of the log: a step, or the episode that closed it.""" |
|
|
| kind: str |
| index: int |
| at: str |
| goal: str |
| screen: str |
| action: str |
| outcome: str |
| ok: bool |
| prev: str |
| digest: str = "" |
|
|
| def payload(self) -> dict[str, Any]: |
| """Everything the digest covers - which is everything except the digest.""" |
| return {"kind": self.kind, "index": self.index, "at": self.at, "goal": self.goal, |
| "screen": self.screen, "action": self.action, "outcome": self.outcome, |
| "ok": self.ok, "prev": self.prev} |
|
|
| def compute_digest(self) -> str: |
| return hashlib.sha256(_canonical(self.payload()).encode("utf-8")).hexdigest() |
|
|
| def to_dict(self) -> dict[str, Any]: |
| return {**self.payload(), "digest": self.digest} |
|
|
| @classmethod |
| def from_dict(cls, raw: dict[str, Any]) -> "AuditEntry": |
| return cls(kind=raw.get("kind", "step"), index=raw.get("index", -1), |
| at=raw.get("at", ""), goal=raw.get("goal", ""), screen=raw.get("screen", ""), |
| action=raw.get("action", ""), outcome=raw.get("outcome", ""), |
| ok=bool(raw.get("ok", False)), prev=raw.get("prev", ""), |
| digest=raw.get("digest", "")) |
|
|
|
|
| @dataclass |
| class AuditLog: |
| """An append-only, hash-chained record of what an episode did.""" |
|
|
| path: Path | str |
| _last: str = field(default="", init=False, repr=False) |
| _count: int = field(default=0, init=False, repr=False) |
|
|
| def __post_init__(self) -> None: |
| self.path = Path(self.path) |
| self.path.parent.mkdir(parents=True, exist_ok=True) |
| |
| if self.path.exists(): |
| existing = read_audit(self.path) |
| self._count = len(existing) |
| if existing: |
| self._last = existing[-1].digest |
|
|
| |
|
|
| def _write(self, kind: str, *, goal: str, screen: str = "", action: str = "", |
| outcome: str = "", ok: bool = False) -> AuditEntry: |
| entry = AuditEntry(kind=kind, index=self._count, |
| at=datetime.now(timezone.utc).isoformat(timespec="seconds"), |
| goal=goal, screen=screen, action=action, outcome=outcome, |
| ok=ok, prev=self._last) |
| entry = AuditEntry(**{**entry.payload(), "digest": entry.compute_digest()}) |
| with open(self.path, "a", encoding="utf-8") as handle: |
| handle.write(_canonical(entry.to_dict()) + "\n") |
| self._last, self._count = entry.digest, self._count + 1 |
| return entry |
|
|
| def append(self, step: Any, goal: str = "") -> AuditEntry: |
| """Record one step. Accepts the loop's :class:`~tiny_computer_agent.loop.Step`.""" |
| result = getattr(step, "result", None) |
| outcome = (getattr(step, "error", None) or (result.summary() if result else "")) or "" |
| return self._write("step", goal=goal or "", |
| screen=observation_digest(step.observation), |
| action=step.action.describe() if step.action else "", |
| outcome=str(outcome), ok=bool(step.ok)) |
|
|
| def close(self, episode: Any) -> AuditEntry: |
| """Record how the episode ended - verified, failed, refused, or out of budget.""" |
| return self._write("episode", goal=getattr(episode, "goal", ""), |
| outcome=f"{'success' if episode.success else 'failure'}: " |
| f"{episode.stop_reason}", |
| ok=bool(episode.success)) |
|
|
| |
|
|
| def entries(self) -> list[AuditEntry]: |
| return read_audit(self.path) |
|
|
| def verify(self) -> tuple[bool, str]: |
| return verify_audit(self.path) |
|
|
| def render(self) -> str: |
| entries = self.entries() |
| steps = [e for e in entries if e.kind == "step"] |
| ok, detail = self.verify() |
| counts: dict[str, int] = {} |
| for entry in steps: |
| label = "ok" if entry.ok else "failed" |
| counts[label] = counts.get(label, 0) + 1 |
| tally = ", ".join(f"{n} {label}" for label, n in sorted(counts.items())) or "no steps" |
| return (f"audit: {len(entries)} entries ({len(steps)} steps, {tally}) | " |
| f"chain {'intact' if ok else 'BROKEN'}: {detail}") |
|
|
|
|
| def read_audit(path: Path | str) -> list[AuditEntry]: |
| """Every entry in the log, in order. A malformed line is an entry that fails to load.""" |
| path = Path(path) |
| if not path.exists(): |
| return [] |
| entries: list[AuditEntry] = [] |
| with open(path, encoding="utf-8") as handle: |
| for line in handle: |
| if line.strip(): |
| entries.append(AuditEntry.from_dict(json.loads(line))) |
| return entries |
|
|
|
|
| def verify_audit(path: Path | str) -> tuple[bool, str]: |
| """Check the chain. Returns ``(ok, detail)``, naming the first entry at fault. |
| |
| Deliberately reports *which* entry broke and *how* - a digest that does not match, a |
| `prev` pointing at nothing, or an index out of order - because those three failures mean |
| different things to whoever has to decide whether to trust the run. |
| """ |
| path = Path(path) |
| if not path.exists(): |
| return False, f"no audit log at {path}" |
|
|
| through = "" |
| count = 0 |
| for number, line in enumerate((l for l in path.read_text(encoding="utf-8").splitlines() |
| if l.strip()), start=1): |
| entry = AuditEntry.from_dict(json.loads(line)) |
| if entry.index != number - 1: |
| return False, (f"entry {number} is out of order (index {entry.index} where " |
| f"{number - 1} was expected): lines were removed or reordered") |
| if entry.prev != through: |
| return False, (f"entry {number} does not follow entry {number - 1}: its link " |
| f"points at an entry that is not there") |
| if entry.compute_digest() != entry.digest: |
| return False, f"entry {number} was edited after it was written" |
| through = entry.digest |
| count += 1 |
| if count == 0: |
| return False, f"{path} is empty" |
| return True, f"{count} entries, chain intact" |
|
|
