| """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 |
| |
| self.episode: Episode | None = None |
| self.observation: Observation | None = None |
| self.paused = False |
| self._done = False |
| self._budget = self.policy.step_budget |
| self._turns = 0 |
| self._previous: tuple | None = None |
| self._repeats = 0 |
|
|
| |
|
|
| def begin(self, goal: str, *, max_steps: int | None = None) -> "AgentLoop": |
| """Start a new episode: read the screen once, and reset the guards. |
| |
| The stepwise half of the API. :meth:`run` is the batch form and is built |
| from exactly this; a console calls ``begin`` then :meth:`step_once`, so one |
| button press can advance exactly one turn. |
| """ |
| budget = max_steps or self.policy.step_budget |
| self.episode = Episode(goal=goal, stop_reason=f"step budget of {budget} exhausted") |
| self._budget = budget |
| self._turns = 0 |
| self._previous = None |
| self._repeats = 0 |
| self._done = False |
| self.paused = False |
| try: |
| self.observation = self.screen.capture() |
| except Exception as exc: |
| self.observation = None |
| self._finish(f"capture failed: {type(exc).__name__}: {exc}") |
| return self |
|
|
| @property |
| def finished(self) -> bool: |
| """True once this episode has ended - success, failure, budget or switch.""" |
| return self._done |
|
|
| def pause(self) -> "AgentLoop": |
| """Stop between steps: :meth:`run` returns and the episode stays open.""" |
| self.paused = True |
| return self |
|
|
| def resume(self) -> "AgentLoop": |
| self.paused = False |
| return self |
|
|
| def step_once(self) -> Step | None: |
| """One observe -> think -> act -> verify turn. |
| |
| Returns the step it recorded, or ``None`` once the episode is over, so a |
| caller can loop on it without knowing any of the stopping rules. |
| """ |
| if self.episode is None or self._done or self.paused: |
| return None |
| episode, observation = self.episode, self.observation |
|
|
| if self._turns >= self._budget: |
| self._done = True |
| return None |
| if self.policy.should_stop(): |
| self._finish("kill switch engaged") |
| return None |
| if observation is None: |
| self._done = True |
| return None |
|
|
| index, self._turns = self._turns, self._turns + 1 |
|
|
| try: |
| action = self.brain.decide(episode.goal, observation) |
| except GoalUnreachable as exc: |
| return self._record_step(Step(index, observation, error=f"gave up: {exc}"), |
| f"brain gave up: {exc}") |
| except Exception as exc: |
| return self._record_step( |
| Step(index, observation, error=f"brain failed: {type(exc).__name__}: {exc}"), |
| "brain error") |
|
|
| if action.kind == "done": |
| episode.success = True |
| return self._record_step( |
| Step(index, observation, action=action, |
| result=ActionResult(action=action, ok=True, note="goal reported met")), |
| f"brain reported done: {action.reason or 'no reason given'}") |
|
|
| try: |
| self.policy.check(action, observation) |
| except PolicyViolation as exc: |
| return self._record_step(Step(index, observation, action=action, |
| error=f"refused: {exc}"), |
| f"policy refusal: {exc}") |
|
|
| signature = (self._fingerprint(observation), action.describe()) |
| self._repeats = self._repeats + 1 if signature == self._previous else 0 |
| self._previous = signature |
| if self._repeats > self.max_repeats: |
| return self._record_step( |
| Step(index, observation, action=action, |
| error=f"no progress: repeating {action.describe()} " |
| f"against an unchanged screen"), |
| f"no progress after {self._repeats} identical actions on an unchanged " |
| f"screen ({action.kind} {action.label or action.text or ''})".strip()) |
|
|
| try: |
| result = self.executor.execute(action, observation) |
| except Exception as exc: |
| return self._record_step( |
| Step(index, observation, action=action, |
| error=f"executor failed: {type(exc).__name__}: {exc}"), |
| "executor error") |
|
|
| step = Step(index, observation, action=action, result=result) |
| self._append(step) |
| if not result.ok: |
| self._finish(f"action failed: {result.note}") |
| return step |
|
|
| try: |
| self.observation = self.screen.capture() |
| except Exception as exc: |
| self._append(Step(index + 1, observation, |
| error=f"capture failed: {type(exc).__name__}: {exc}")) |
| self._finish("perception failed mid-episode") |
| return step |
|
|
| if self.verify is not None and self._verified(self.observation): |
| episode.success = True |
| self._finish(f"verified after {self._turns} action(s)") |
| return step |
|
|
| def run(self, goal: str | None = None, *, max_steps: int | None = None) -> Episode: |
| """Run to the end of the episode - or continue one that was paused. |
| |
| 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 |
| """ |
| if self.episode is None or self._done or (goal is not None and goal != self.episode.goal): |
| self.begin(goal or "", max_steps=max_steps) |
| while not self.paused and not self._done: |
| if self.step_once() is None: |
| break |
| return self.episode |
|
|
| def _append(self, step: Step) -> None: |
| self.episode.steps.append(step) |
| if self.on_step is not None: |
| self.on_step(step) |
|
|
| def _record_step(self, step: Step, stop_reason: str) -> Step: |
| self._append(step) |
| self._finish(stop_reason) |
| return step |
|
|
| def _finish(self, stop_reason: str) -> None: |
| self.episode.stop_reason = stop_reason |
| self._done = True |
|
|
| 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) |
|
|
