File size: 9,424 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 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 | """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")
|