"""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") # A small model that repeats an action against an unchanged screen is stuck, # not persistent. Tolerate `max_repeats` such repeats, then stop - burning the # whole budget on the same click teaches the user nothing. 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() # verify: what did that change? 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)