| """Run a queue of goals, survive being killed, and report what actually happened. |
| |
| The loop runs one episode. An operator runs a day of them: several goals against one screen, |
| some of which are slow, some of which fail, and any of which may be interrupted when the |
| sandbox dies. So this is the boring part done properly - a queue, a resume file, and a report |
| that counts what was verified rather than what was attempted. |
| |
| tasks = [Task("click 'Save'", verify_text="Saved"), Task("click 'Cancel'")] |
| runner = TaskRunner(build_loop, state_path=Path("runs/today.jsonl")) |
| report = runner.run(tasks) # kill it and run again: finished tasks are skipped |
| print(report.render()) |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import time |
| from dataclasses import dataclass, field |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Any, Callable, Iterable, Sequence |
|
|
| from .diff import ChangeVerifier |
| from .loop import AgentLoop |
| from .observe import Observation, normalise_label |
| from .skills import slug |
|
|
|
|
| def _stamp() -> str: |
| return datetime.now(timezone.utc).isoformat(timespec="seconds") |
|
|
|
|
| @dataclass(frozen=True) |
| class Task: |
| """One goal, and how to tell whether it worked.""" |
|
|
| goal: str |
| name: str = "" |
| budget: int | None = None |
| verify_text: str = "" |
| verify_gone: str = "" |
| """A label that must *disappear* for this task to count - the dialog closed, verified.""" |
| verify_appeared: str = "" |
| """A label that must *appear* - the menu opened, the row was added.""" |
|
|
| def __post_init__(self) -> None: |
| if not self.name: |
| object.__setattr__(self, "name", slug(self.goal)) |
|
|
| def verifier(self) -> Callable[[Observation], bool] | None: |
| """The harness's own check, so success is evidence rather than a brain's claim. |
| |
| Text on screen is the simplest check; "this element went away" is the strongest one |
| for a task whose success is a dialog closing, which is exactly the case where text |
| alone cannot tell a winner from a no-op. |
| """ |
| if self.verify_text: |
| wanted = normalise_label(self.verify_text) |
|
|
| def by_text(observation: Observation) -> bool: |
| return wanted in normalise_label(observation.text) |
|
|
| return by_text |
| if self.verify_gone or self.verify_appeared: |
| return ChangeVerifier(appeared=(self.verify_appeared,) if self.verify_appeared else (), |
| disappeared=(self.verify_gone,) if self.verify_gone else ()) |
| return None |
|
|
| def line(self) -> str: |
| if self.verify_text: |
| check = f" expect {self.verify_text!r}" |
| elif self.verify_gone or self.verify_appeared: |
| bits = [] |
| if self.verify_appeared: |
| bits.append(f"{self.verify_appeared!r} to appear") |
| if self.verify_gone: |
| bits.append(f"{self.verify_gone!r} to go") |
| check = " expect " + " and ".join(bits) |
| else: |
| check = "" |
| return f"{self.name:28s} {self.goal[:56]}{check}" |
|
|
| def to_dict(self) -> dict[str, Any]: |
| return {"name": self.name, "goal": self.goal, "budget": self.budget, |
| "verify_text": self.verify_text, "verify_gone": self.verify_gone, |
| "verify_appeared": self.verify_appeared} |
|
|
|
|
| @dataclass |
| class TaskOutcome: |
| """What happened to one task, in the form a report and a resume file both need.""" |
|
|
| name: str |
| goal: str |
| success: bool = False |
| steps: int = 0 |
| stop_reason: str = "" |
| seconds: float = 0.0 |
| error: str = "" |
|
|
| def line(self) -> str: |
| verdict = "success" if self.success else "failed" |
| detail = self.error or self.stop_reason |
| return f"{self.name:28s} {verdict:8s} {self.steps:2d} step(s) {self.seconds:6.1f}s {detail}" |
|
|
| def to_dict(self) -> dict[str, Any]: |
| return {"name": self.name, "goal": self.goal, "success": self.success, "steps": self.steps, |
| "stop_reason": self.stop_reason, "seconds": round(self.seconds, 3), "error": self.error} |
|
|
|
|
| @dataclass |
| class RunReport: |
| """The day's outcomes, with the only number this project counts.""" |
|
|
| outcomes: list[TaskOutcome] = field(default_factory=list) |
| started: str = "" |
| finished: str = "" |
|
|
| @property |
| def total(self) -> int: |
| return len(self.outcomes) |
|
|
| @property |
| def passed(self) -> int: |
| return sum(1 for outcome in self.outcomes if outcome.success) |
|
|
| @property |
| def failed(self) -> int: |
| return self.total - self.passed |
|
|
| @property |
| def accuracy(self) -> float: |
| return self.passed / self.total if self.total else 0.0 |
|
|
| @property |
| def steps(self) -> int: |
| return sum(outcome.steps for outcome in self.outcomes) |
|
|
| def to_rows(self) -> list[dict[str, Any]]: |
| return [{"task": o.name, "goal": o.goal, "success": o.success, "steps": o.steps, |
| "seconds": round(o.seconds, 1), "reason": o.error or o.stop_reason} |
| for o in self.outcomes] |
|
|
| def render(self) -> str: |
| lines = [f"report {self.started} -> {self.finished}", |
| f"{self.passed}/{self.total} verified ({self.accuracy:.0%}), {self.steps} step(s) total"] |
| lines += [f" {outcome.line()}" for outcome in self.outcomes] |
| return "\n".join(lines) |
|
|
| def save(self, path: Path | str) -> Path: |
| target = Path(path) |
| target.parent.mkdir(parents=True, exist_ok=True) |
| target.write_text(json.dumps({"started": self.started, "finished": self.finished, |
| "passed": self.passed, "total": self.total, |
| "outcomes": [o.to_dict() for o in self.outcomes]}, indent=2) + "\n") |
| return target |
|
|
|
|
| @dataclass |
| class TaskRunner: |
| """A queue that can be interrupted and resumed. |
| |
| ``build_loop`` is a callable returning a *fresh* loop per task, so state like "already |
| clicked this" cannot leak between goals - the same rule the benchmark uses. The state file |
| is JSONL, appended as each task finishes, so a run that is killed at task seven of nine |
| resumes at task eight rather than starting over. |
| """ |
|
|
| build_loop: Callable[[Task], AgentLoop] |
| state_path: Path | None = None |
| on_task: Callable[[TaskOutcome], None] | None = None |
| clock: Callable[[], float] = time.monotonic |
|
|
| def completed(self) -> dict[str, dict[str, Any]]: |
| """The tasks already recorded in the state file, by name.""" |
| if self.state_path is None or not Path(self.state_path).exists(): |
| return {} |
| done: dict[str, dict[str, Any]] = {} |
| for line in Path(self.state_path).read_text().splitlines(): |
| if not line.strip(): |
| continue |
| try: |
| entry = json.loads(line) |
| except json.JSONDecodeError: |
| continue |
| if "name" in entry: |
| done[entry["name"]] = entry |
| return done |
|
|
| def run(self, tasks: Sequence[Task], *, resume: bool = True, |
| max_steps: int | None = None) -> RunReport: |
| """Run every task that has not already finished, and report on all of them.""" |
| done = self.completed() if resume else {} |
| report = RunReport(started=_stamp()) |
| for task in tasks: |
| if task.name in done: |
| entry = done[task.name] |
| report.outcomes.append(TaskOutcome( |
| name=task.name, goal=task.goal, success=bool(entry.get("success")), |
| steps=int(entry.get("steps", 0)), |
| stop_reason=f"already run ({entry.get('stop_reason', '')})".strip(), |
| seconds=float(entry.get("seconds", 0.0)), error=entry.get("error", ""))) |
| continue |
| outcome = self.run_one(task, max_steps=max_steps) |
| report.outcomes.append(outcome) |
| self._remember(outcome) |
| if self.on_task is not None: |
| self.on_task(outcome) |
| report.finished = _stamp() |
| return report |
|
|
| def run_one(self, task: Task, *, max_steps: int | None = None) -> TaskOutcome: |
| """One task, with every failure turned into an outcome rather than a traceback.""" |
| outcome = TaskOutcome(name=task.name, goal=task.goal) |
| started = self.clock() |
| try: |
| loop = self.build_loop(task) |
| if task.verifier() is not None: |
| loop.verify = task.verifier() |
| episode = loop.run(task.goal, max_steps=task.budget or max_steps) |
| outcome.success = bool(episode.success) |
| outcome.steps = len(episode.steps) |
| outcome.stop_reason = episode.stop_reason |
| except Exception as exc: |
| outcome.error = f"{type(exc).__name__}: {exc}" |
| outcome.stop_reason = "the runner caught a failure" |
| outcome.seconds = self.clock() - started |
| return outcome |
|
|
| def _remember(self, outcome: TaskOutcome) -> None: |
| if self.state_path is None: |
| return |
| target = Path(self.state_path) |
| target.parent.mkdir(parents=True, exist_ok=True) |
| with open(target, "a") as handle: |
| handle.write(json.dumps(outcome.to_dict()) + "\n") |
|
|
