"""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 # a half-written line from a killed run 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: # noqa: BLE001 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")