| """The harness itself: observe -> think -> act -> verify, bounded and traced.""" |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import dataclass, field |
| from typing import Any, Callable |
|
|
| from tiny_computer_agent.act import ActionResult, DryRunExecutor, Executor, PyAutoGuiExecutor |
| from tiny_computer_agent.actions import Action |
| from tiny_computer_agent.brain import Brain, GoalUnreachable |
| from tiny_computer_agent.observe import Observation, ScreenSource |
| from tiny_computer_agent.safety import Policy, PolicyViolation |
|
|
|
|
| @dataclass(frozen=True) |
| class Step: |
| """One turn of the loop, with everything needed to audit it afterwards.""" |
|
|
| index: int |
| observation: Observation |
| action: Action | None = None |
| result: ActionResult | None = None |
| error: str | None = None |
|
|
| @property |
| def ok(self) -> bool: |
| return self.error is None and (self.result is None or self.result.ok) |
|
|
| def line(self) -> str: |
| bits = [f"step {self.index}"] |
| if self.action is not None: |
| bits.append(self.action.describe()) |
| if self.error: |
| bits.append(f"ERROR {self.error}") |
| elif self.result is not None: |
| bits.append(self.result.summary()) |
| return " | ".join(bits) |
|
|
|
|
| @dataclass |
| class Episode: |
| """An append-only record of one run: the unit you replay and score.""" |
|
|
| goal: str |
| steps: list[Step] = field(default_factory=list) |
| stop_reason: str = "" |
| success: bool = False |
|
|
| def to_rows(self) -> list[dict[str, Any]]: |
| """Rows for a dataframe / notebook table.""" |
| rows = [] |
| for step in self.steps: |
| rows.append({ |
| "step": step.index, |
| "elements": len(step.observation.elements), |
| "action": step.action.describe() if step.action else "", |
| "outcome": step.error or (step.result.summary() if step.result else ""), |
| }) |
| return rows |
|
|
| def render(self) -> str: |
| lines = [ |
| f"goal: {self.goal}", |
| f"result: {'success' if self.success else 'failure'} - {self.stop_reason}", |
| ] |
| lines += [f" {step.line()}" for step in self.steps] |
| return "\n".join(lines) |
|
|
|
|
| class AgentLoop: |
| """Wires a screen, a brain, an executor and a policy into an agent. |
| |
| Guarantees, by construction rather than by prompting: |
| |
| * every action is one of six validated verbs |
| * nothing executes until policy has passed it |
| * the episode ends on a step budget, a kill switch, a brain error, or a |
| failed screen capture (reported as a step, never as a traceback) |
| * the screen is re-read after every action, so the next decision sees the |
| consequence of the last one |
| """ |
|
|
| def __init__(self, *, screen: ScreenSource, brain: Brain, |
| executor: Executor | None = None, policy: Policy | None = None, |
| on_step: Callable[[Step], None] | None = None, |
| max_repeats: int = 1, |
| verify: Callable[[Observation], bool] | None = None) -> None: |
| """``verify`` runs after every action: the harness, not the model, decides |
| whether the goal was reached. A brain saying "done" is a claim; a verifier |
| looking at the screen is evidence.""" |
| self.screen = screen |
| self.brain = brain |
| self.policy = policy or Policy() |
| if executor is None: |
| executor = DryRunExecutor() if self.policy.dry_run else PyAutoGuiExecutor() |
| self.executor = executor |
| self.on_step = on_step |
| self.verify = verify |
| self.max_repeats = max_repeats |
|
|
| def run(self, goal: str, *, max_steps: int | None = None) -> Episode: |
| budget = max_steps or self.policy.step_budget |
| episode = Episode(goal=goal, stop_reason=f"step budget of {budget} exhausted") |
| |
| |
| |
| previous: tuple[tuple, str] | None = None |
| repeats = 0 |
| try: |
| observation = self.screen.capture() |
| except Exception as exc: |
| episode.stop_reason = f"capture failed: {type(exc).__name__}: {exc}" |
| return episode |
|
|
| for index in range(budget): |
| if self.policy.should_stop(): |
| episode.stop_reason = "kill switch engaged" |
| break |
|
|
| try: |
| action = self.brain.decide(goal, observation) |
| except GoalUnreachable as exc: |
| self._record(episode, Step(index, observation, error=f"gave up: {exc}")) |
| episode.stop_reason = f"brain gave up: {exc}" |
| break |
| except Exception as exc: |
| self._record(episode, Step(index, observation, |
| error=f"brain failed: {type(exc).__name__}: {exc}")) |
| episode.stop_reason = "brain error" |
| break |
|
|
| if action.kind == "done": |
| self._record(episode, Step( |
| index, observation, action=action, |
| result=ActionResult(action=action, ok=True, note="goal reported met"), |
| )) |
| episode.success = True |
| episode.stop_reason = f"brain reported done: {action.reason or 'no reason given'}" |
| break |
|
|
| try: |
| self.policy.check(action, observation) |
| except PolicyViolation as exc: |
| self._record(episode, Step(index, observation, action=action, |
| error=f"refused: {exc}")) |
| episode.stop_reason = f"policy refusal: {exc}" |
| break |
|
|
| signature = (self._fingerprint(observation), action.describe()) |
| repeats = repeats + 1 if signature == previous else 0 |
| previous = signature |
| if repeats > self.max_repeats: |
| self._record(episode, Step( |
| index, observation, action=action, |
| error=f"no progress: repeating {action.describe()} " |
| f"against an unchanged screen", |
| )) |
| episode.stop_reason = ( |
| f"no progress after {repeats} identical actions on an unchanged " |
| f"screen ({action.kind} {action.label or action.text or ''})".strip() |
| ) |
| break |
|
|
| try: |
| result = self.executor.execute(action, observation) |
| except Exception as exc: |
| self._record(episode, Step(index, observation, action=action, |
| error=f"executor failed: {type(exc).__name__}: {exc}")) |
| episode.stop_reason = "executor error" |
| break |
|
|
| self._record(episode, Step(index, observation, action=action, result=result)) |
| if not result.ok: |
| episode.stop_reason = f"action failed: {result.note}" |
| break |
|
|
| try: |
| observation = self.screen.capture() |
| except Exception as exc: |
| self._record(episode, Step(index + 1, observation, |
| error=f"capture failed: {type(exc).__name__}: {exc}")) |
| episode.stop_reason = "perception failed mid-episode" |
| break |
|
|
| if self.verify is not None and self._verified(observation): |
| episode.success = True |
| episode.stop_reason = f"verified after {index + 1} action(s)" |
| break |
|
|
| return episode |
|
|
| def _verified(self, observation: Observation) -> bool: |
| """A broken verifier must never fake a success.""" |
| try: |
| return bool(self.verify(observation)) |
| except Exception: |
| return False |
|
|
| @staticmethod |
| def _fingerprint(observation: Observation) -> tuple: |
| """What the agent can see: enough to tell "nothing changed" from "it worked".""" |
| return (tuple(element.label for element in observation.elements), |
| observation.text[:200]) |
|
|
| def _record(self, episode: Episode, step: Step) -> None: |
| episode.steps.append(step) |
| if self.on_step is not None: |
| self.on_step(step) |
|
|
