"""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.audit import AuditLog from tiny_computer_agent.brain import Brain, GoalUnreachable from tiny_computer_agent.observe import Observation, ScreenSource from tiny_computer_agent.gates import Approval, Gate from tiny_computer_agent.safety import Policy, PolicyViolation from tiny_computer_agent.watchdog import Watchdog, WatchdogTimeout @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 **and** any gate have 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, gate: Gate | None = None, watchdog: Watchdog | None = None, audit: AuditLog | 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.gate = gate self.watchdog = watchdog self.audit = audit self._audit_closed = False self.max_repeats = max_repeats # the stepwise state: an episode in progress, and the guards around it 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 # -- stepwise control -------------------------------------------------- # 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 self._audit_closed = False try: self.observation = self._guard("capture", self.screen.capture) except WatchdogTimeout as exc: self.observation = None self._finish(f"watchdog stopped the episode: {exc}") 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: # a screen that never worked self._done = True return None index, self._turns = self._turns, self._turns + 1 try: action = self._guard("decide", self.brain.decide, episode.goal, observation) except WatchdogTimeout as exc: return self._record_step(Step(index, observation, error=f"watchdog: {exc}"), f"watchdog stopped the episode: {exc}") 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}") if self.gate is not None: # the operator's yes, before anything moves approval: Approval = self.gate.approve(action, observation) if not approval.approved: return self._record_step( Step(index, observation, action=action, error=f"not approved: {approval.reason}"), f"operator refused: {approval.reason}") 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._guard("execute", self.executor.execute, action, observation) except WatchdogTimeout as exc: return self._record_step( Step(index, observation, action=action, error=f"watchdog: {exc}"), f"watchdog stopped the episode: {exc}") 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._guard("capture", self.screen.capture) except WatchdogTimeout as exc: # verify: what did that change? self._append(Step(index + 1, observation, error=f"watchdog: {exc}")) self._finish(f"watchdog stopped the episode: {exc}") return step 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 **and** any gate have 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.audit is not None: self.audit.append(step, self.episode.goal) 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 if self.audit is not None and not self._audit_closed: # one closing entry, ever self._audit_closed = True self.audit.close(self.episode) def _guard(self, where: str, call: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: """One guarded call: with a watchdog, a hung call becomes a named failure.""" if self.watchdog is None: return call(*args, **kwargs) return self.watchdog.call(where, call, *args, **kwargs) 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)