Publish tiny_computer_agent/ from marimo

tiny_computer_agent/__init__.py

@@ -0,0 +1,63 @@
+"""Tiny Computer Agent - an agentic harness that drives a real computer with a
+tiny, fully local model.
+
+The harness is deliberately boring: it owns the loop, the closed action
+vocabulary and the safety gates, so the model only ever answers one question -
+"which action, from this menu, comes next?"  Nothing here needs a network call
+at run time, and the core is standard-library only.
+
+    from tiny_computer_agent import (
+        AgentLoop, DryRunExecutor, Element, FakeScreen, MockBrain, Policy,
+    )
+
+    screen = FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)])
+    loop = AgentLoop(screen=screen, brain=MockBrain([{"kind": "click", "label": "Save"}]))
+    episode = loop.run("save the document")
+    print(episode.render())
+"""
+
+from tiny_computer_agent.act import (
+    ActionResult,
+    DryRunExecutor,
+    Executor,
+    InputUnavailable,
+    PyAutoGuiExecutor,
+    resolve_point,
+)
+from tiny_computer_agent.actions import (
+    ACTION_KINDS,
+    ACTION_MENU,
+    Action,
+    ActionError,
+    parse_action,
+)
+from tiny_computer_agent.brain import (
+    Brain,
+    BrainError,
+    GoalUnreachable,
+    MockBrain,
+    PlannerBrain,
+    VLMBrain,
+    extract_json_object,
+)
+from tiny_computer_agent.loop import AgentLoop, Episode, Step
+from tiny_computer_agent.observe import (
+    Element,
+    FakeScreen,
+    GrabScreen,
+    Observation,
+    ScreenSource,
+    ScreenUnavailable,
+)
+from tiny_computer_agent.safety import Policy, PolicyViolation
+
+__version__ = "0.1.0"
+
+__all__ = [
+    "ACTION_KINDS", "ACTION_MENU", "Action", "ActionError", "ActionResult",
+    "AgentLoop", "Brain", "BrainError", "DryRunExecutor", "Element", "Episode",
+    "Executor", "FakeScreen", "GoalUnreachable", "GrabScreen", "InputUnavailable", "MockBrain",
+    "Observation", "PlannerBrain", "Policy", "PolicyViolation", "PyAutoGuiExecutor",
+    "ScreenSource", "ScreenUnavailable", "Step", "VLMBrain", "extract_json_object",
+    "parse_action", "resolve_point", "__version__",
+]

tiny_computer_agent/__main__.py

@@ -0,0 +1,105 @@
+"""Command line entry point.
+
+    python -m tiny_computer_agent                        # scripted demo, dry-run
+    python -m tiny_computer_agent --live                 # your real screen, dry-run
+    python -m tiny_computer_agent "click 'Save'"        # planner brain
+    python -m tiny_computer_agent --script '[{"kind": "done"}]'
+
+Real mouse and keyboard input additionally requires ``--real-input`` *and*
+``TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT=1`` in the environment. That is deliberate:
+dry-run is the default so you can read a whole episode first.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+
+from tiny_computer_agent.act import DryRunExecutor, PyAutoGuiExecutor
+from tiny_computer_agent.brain import MockBrain, PlannerBrain
+from tiny_computer_agent.loop import AgentLoop, Step
+from tiny_computer_agent.observe import Element, FakeScreen, GrabScreen
+from tiny_computer_agent.safety import Policy
+
+ALLOW_REAL_INPUT_ENV = "TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT"
+DEFAULT_KILL_SWITCH = "/tmp/tiny-computer-agent.STOP"
+
+
+def demo_screen() -> FakeScreen:
+    """A three-frame desktop: a dialog, then an editor, then a saved file."""
+    return FakeScreen(
+        [
+            (Element("Terminal", x=0, y=0, width=40, height=24),
+             Element("Files", x=60, y=0, width=40, height=24)),
+            (Element("Name", kind="input", x=100, y=120, width=200, height=24),
+             Element("Save", x=320, y=120, width=80, height=24)),
+            (Element("Save", x=320, y=120, width=80, height=24),),
+        ],
+        text="Untitled document - unsaved changes",
+    )
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="tiny-computer-agent",
+        description="Drive a computer with a tiny local model (dry-run by default).",
+    )
+    parser.add_argument("goal", nargs="?", default="click 'Save' to save the document")
+    parser.add_argument("--max-steps", type=int, default=6)
+    parser.add_argument("--script", default=None,
+                        help="JSON list of actions to replay instead of using a brain")
+    parser.add_argument("--live", action="store_true",
+                        help="capture your real screen instead of the scripted demo")
+    parser.add_argument("--real-input", action="store_true",
+                        help="actually move the mouse and type (requires the allow env var)")
+    parser.add_argument("--kill-switch", default=DEFAULT_KILL_SWITCH,
+                        help="create this file to abort the run before the next action")
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+
+    if args.real_input and os.environ.get(ALLOW_REAL_INPUT_ENV) != "1":
+        print(f"refusing --real-input: set {ALLOW_REAL_INPUT_ENV}=1 to confirm you "
+              f"understand this will drive your actual mouse and keyboard.", file=sys.stderr)
+        return 2
+
+    screen = GrabScreen() if args.live else demo_screen()
+
+    if args.script:
+        try:
+            script = json.loads(args.script)
+        except json.JSONDecodeError as exc:
+            print(f"--script is not valid JSON: {exc}", file=sys.stderr)
+            return 2
+        brain = MockBrain(script)
+    else:
+        brain = PlannerBrain(done_when_text_contains=None)
+
+    policy = Policy(
+        dry_run=not args.real_input,
+        step_budget=args.max_steps,
+        kill_switch_path=args.kill_switch,
+    )
+    executor = PyAutoGuiExecutor() if args.real_input else DryRunExecutor()
+
+    mode = "REAL INPUT (mouse and keyboard live)" if args.real_input else "dry-run"
+    print(f"goal: {args.goal}\nmode: {mode}  |  screen: "
+          f"{'real desktop' if args.live else 'scripted demo'}  |  "
+          f"kill switch: create {args.kill_switch} to stop\n")
+
+    def on_step(step: Step) -> None:
+        print("  " + step.line())
+
+    episode = AgentLoop(screen=screen, brain=brain, executor=executor,
+                        policy=policy, on_step=on_step).run(args.goal)
+    print()
+    print(episode.render())
+    return 0 if episode.success else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tiny_computer_agent/act.py

@@ -0,0 +1,148 @@
+"""Execution: the only module in the harness that touches the machine."""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+from typing import Protocol
+
+from tiny_computer_agent.actions import Action
+from tiny_computer_agent.observe import Observation
+
+
+@dataclass(frozen=True)
+class ActionResult:
+    action: Action
+    ok: bool
+    note: str = ""
+
+    def summary(self) -> str:
+        marker = "ok" if self.ok else "FAILED"
+        return f"{marker}: {self.note}"
+
+
+class InputUnavailable(RuntimeError):
+    """Raised when real input is requested but pyautogui cannot drive the machine."""
+
+
+def resolve_point(
+    action: Action, observation: Observation | None = None
+) -> tuple[int, int] | None:
+    """Turn a click into pixels, using the element index for label clicks."""
+    if action.kind != "click":
+        return None
+    if action.x is not None and action.y is not None:
+        return (action.x, action.y)
+    if observation is not None and action.label:
+        element = observation.find(action.label)
+        if element is not None:
+            return element.center
+    return None
+
+
+class Executor(Protocol):
+    dry_run: bool
+
+    def execute(
+        self, action: Action, observation: Observation | None = None
+    ) -> ActionResult: ...
+
+
+class DryRunExecutor:
+    """Records actions instead of performing them. The default everywhere.
+
+    Lets you read an entire episode — including which pixel each label resolved
+    to — before a single event reaches the machine.
+    """
+
+    dry_run = True
+
+    def __init__(self) -> None:
+        self.performed: list[Action] = []
+        self.log: list[str] = []
+
+    def execute(
+        self, action: Action, observation: Observation | None = None
+    ) -> ActionResult:
+        note = f"(dry-run) would {action.describe()}"
+        ok = True
+        if action.kind == "click":
+            point = resolve_point(action, observation)
+            if point is not None:
+                note += f" -> pixel {point}"
+            elif action.label:
+                # Fail exactly where live input would, so a dry run never gives
+                # false confidence about a plan the executor would refuse.
+                ok = False
+                note += " (label absent from the element index; live input refuses this)"
+        self.performed.append(action)
+        self.log.append(note)
+        return ActionResult(action=action, ok=ok, note=note)
+
+
+class PyAutoGuiExecutor:
+    """Real mouse and keyboard input.
+
+    Refuses to click when a label cannot be resolved to an element: a tiny model
+    guessing pixel coordinates is exactly how you click "Delete account".
+    """
+
+    dry_run = False
+
+    def __init__(self, *, pause: float = 0.1, typing_interval: float = 0.02,
+                 scroll_ticks: int = 100) -> None:
+        self.pause = pause
+        self.typing_interval = typing_interval
+        self.scroll_ticks = scroll_ticks
+
+    @staticmethod
+    def _pyautogui():
+        try:
+            import pyautogui
+        except ImportError as exc:  # pragma: no cover - depends on the machine
+            raise InputUnavailable(
+                "real input needs pyautogui: pip install \'tiny-computer-agent[desktop]\'"
+            ) from exc
+        return pyautogui
+
+    def execute(
+        self, action: Action, observation: Observation | None = None
+    ) -> ActionResult:
+        if action.kind == "done":
+            return ActionResult(action=action, ok=True, note="goal reported met")
+
+        point: tuple[int, int] | None = None
+        if action.kind == "click":
+            point = resolve_point(action, observation)
+            if point is None:
+                return ActionResult(
+                    action=action,
+                    ok=False,
+                    note=f"no element matching {action.label!r}; refusing to guess pixels",
+                )
+
+        pyautogui = self._pyautogui()
+        try:
+            if action.kind == "click":
+                pyautogui.click(*point)
+                note = f"clicked {point}"
+            elif action.kind == "type":
+                pyautogui.write(action.text or "", interval=self.typing_interval)
+                note = f"typed {len(action.text or '')} characters"
+            elif action.kind == "key":
+                chord = [k.strip() for k in (action.keys or "").split("+") if k.strip()]
+                pyautogui.hotkey(*chord)
+                note = f"pressed {action.keys}"
+            elif action.kind == "scroll":
+                pyautogui.scroll(action.amount * self.scroll_ticks)
+                note = f"scrolled {action.amount} ticks"
+            elif action.kind == "wait":
+                time.sleep(action.seconds)
+                note = f"waited {action.seconds}s"
+            else:  # unreachable: Action already validated the vocabulary
+                return ActionResult(action=action, ok=False, note=f"unhandled kind {action.kind!r}")
+        except Exception as exc:
+            return ActionResult(action=action, ok=False, note=f"{type(exc).__name__}: {exc}")
+        if self.pause:
+            time.sleep(self.pause)
+        return ActionResult(action=action, ok=True, note=note)

tiny_computer_agent/actions.py

@@ -0,0 +1,126 @@
+"""The closed action vocabulary.
+
+A tiny model is only dependable when its output space is small and every mistake
+is cheap to catch. The harness therefore accepts exactly six verbs, and validates
+each field *before* anything touches the machine.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Final, Mapping
+
+ACTION_KINDS: Final[tuple[str, ...]] = ("click", "type", "key", "scroll", "wait", "done")
+
+
+class ActionError(ValueError):
+    """Raised when a brain proposes something outside the vocabulary."""
+
+
+@dataclass(frozen=True)
+class Action:
+    """One step of computer use: exactly one verb, validated on construction.
+
+    label   : click target *by element label* — the preferred form, because it
+              survives window moves that would break raw pixel coordinates
+    x, y    : pixel fallback, used only when no label matches
+    text    : literal text to type
+    keys    : key chord, e.g. ``ctrl+s``
+    amount  : scroll ticks; negative scrolls up
+    seconds : how long to wait
+    reason  : the brain's stated rationale — kept for traces, never executed
+    """
+
+    kind: str
+    label: str | None = None
+    x: int | None = None
+    y: int | None = None
+    text: str | None = None
+    keys: str | None = None
+    amount: int = 0
+    seconds: float = 0.0
+    reason: str = ""
+
+    def __post_init__(self) -> None:
+        if self.kind not in ACTION_KINDS:
+            raise ActionError(f"unknown action {self.kind!r}; expected one of {ACTION_KINDS}")
+        if self.kind == "click":
+            if not self.label and not (self.x is not None and self.y is not None):
+                raise ActionError("click needs a label, or x and y")
+            if (self.x is not None and self.x < 0) or (self.y is not None and self.y < 0):
+                raise ActionError("click coordinates must be non-negative")
+        if self.kind == "type" and not self.text:
+            raise ActionError("type needs non-empty text")
+        if self.kind == "key" and not self.keys:
+            raise ActionError("key needs a chord, e.g. 'ctrl+s'")
+        if self.kind == "scroll" and self.amount == 0:
+            raise ActionError("scroll needs a non-zero amount")
+        if self.kind == "wait" and self.seconds <= 0:
+            raise ActionError("wait needs a positive number of seconds")
+
+    def describe(self) -> str:
+        """One-line, human-readable form used in traces and dry-run logs."""
+        parts: list[str] = [self.kind]
+        if self.label:
+            parts.append(f"label={self.label!r}")
+        if self.x is not None and self.y is not None:
+            parts.append(f"at=({self.x},{self.y})")
+        if self.text is not None:
+            parts.append(f"text={self.text!r}")
+        if self.keys:
+            parts.append(f"keys={self.keys!r}")
+        if self.kind == "scroll":
+            parts.append(f"amount={self.amount}")
+        if self.kind == "wait":
+            parts.append(f"seconds={self.seconds}")
+        if self.reason:
+            parts.append(f"reason={self.reason!r}")
+        return " ".join(parts)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "kind": self.kind,
+            "label": self.label,
+            "x": self.x,
+            "y": self.y,
+            "text": self.text,
+            "keys": self.keys,
+            "amount": self.amount,
+            "seconds": self.seconds,
+            "reason": self.reason,
+        }
+
+
+ACTION_MENU: Final[str] = """\
+Reply with exactly one JSON object and nothing else. Allowed verbs:
+
+  {"kind": "click",  "label": "<text of the element to click>", "reason": "..."}
+  {"kind": "click",  "x": 640, "y": 400, "reason": "..."}   // pixels, only if no label fits
+  {"kind": "type",   "text": "...", "reason": "..."}
+  {"kind": "key",    "keys": "ctrl+s", "reason": "..."}
+  {"kind": "scroll", "amount": -3, "reason": "..."}
+  {"kind": "wait",   "seconds": 1.5, "reason": "..."}
+  {"kind": "done",   "reason": "why the goal is met"}
+
+Rules: one action per reply; prefer a label over guessing pixels; reply "done"
+as soon as the goal is met; never invent new verbs."""
+
+_FIELDS: Final[tuple[str, ...]] = (
+    "label", "x", "y", "text", "keys", "amount", "seconds", "reason",
+)
+
+
+def parse_action(payload: Mapping[str, Any]) -> Action:
+    """Turn a brain's raw mapping into a validated :class:`Action`.
+
+    Unknown keys are ignored, so a chatty model cannot crash the loop. An unknown
+    verb or a missing field raises :class:`ActionError`, which the loop records as
+    a failed step *without executing anything*.
+    """
+    if not isinstance(payload, Mapping):
+        raise ActionError(f"expected a mapping, got {type(payload).__name__}")
+    kind = payload.get("kind")
+    if kind not in ACTION_KINDS:
+        raise ActionError(f"unknown action {kind!r}; expected one of {ACTION_KINDS}")
+    kwargs = {k: payload[k] for k in _FIELDS if k in payload}
+    return Action(kind=str(kind), **kwargs)

tiny_computer_agent/brain.py

@@ -0,0 +1,174 @@
+"""Decision making.
+
+Three brains ship here, and only one of them is a model:
+
+* :class:`MockBrain`    - replays a scripted action list (tests, demos, replays)
+* :class:`PlannerBrain` - rule-based: clicks the element whose label matches the
+  goal. No model, no download, yet it solves real "click the thing labelled X"
+  tasks, because the *element index* does the grounding a VLM would otherwise do
+* :class:`VLMBrain`     - the seam for a real tiny VLM. Prompt building and reply
+  validation are done; you implement ``_generate()``
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import Any, Mapping, Protocol, Sequence
+
+from tiny_computer_agent.actions import ACTION_MENU, Action, ActionError, parse_action
+from tiny_computer_agent.observe import Observation
+
+
+class BrainError(RuntimeError):
+    """Raised when a brain cannot produce a valid action."""
+
+
+class GoalUnreachable(BrainError):
+    """Raised by a brain that can decide nothing further, without claiming success.
+
+    Deliberately distinct from a crash: the harness records the episode as a
+    *failure*, because a brain admitting "I give up" must never be logged as
+    "goal met" - and the harness, not the brain, owns that judgement.
+    """
+
+
+def extract_json_object(raw: str) -> dict[str, Any]:
+    """Pull the first JSON object out of a reply, ignoring any surrounding prose."""
+    start = raw.find("{")
+    end = raw.rfind("}")
+    if start == -1 or end <= start:
+        raise BrainError(f"no JSON object in the model reply: {raw[:200]!r}")
+    try:
+        payload = json.loads(raw[start : end + 1])
+    except json.JSONDecodeError as exc:
+        raise BrainError(f"malformed JSON in the model reply: {exc}") from exc
+    if not isinstance(payload, dict):
+        raise BrainError(f"expected a JSON object, got {type(payload).__name__}")
+    return payload
+
+
+class Brain(Protocol):
+    """Anything that can turn (goal, observation) into one validated action."""
+
+    name: str
+
+    def decide(self, goal: str, observation: Observation) -> Action: ...
+
+
+class MockBrain:
+    """Replays a fixed script of actions, exactly once. Deterministic and offline."""
+
+    name = "mock"
+
+    def __init__(self, script: Sequence[Action | Mapping[str, Any]]) -> None:
+        self._script: list[Action] = [
+            a if isinstance(a, Action) else parse_action(a) for a in script
+        ]
+        self._cursor = 0
+
+    @property
+    def remaining(self) -> int:
+        return max(0, len(self._script) - self._cursor)
+
+    def decide(self, goal: str, observation: Observation) -> Action:
+        if self._cursor >= len(self._script):
+            return Action("done", reason="script exhausted")
+        action = self._script[self._cursor]
+        self._cursor += 1
+        return action
+
+
+class PlannerBrain:
+    """Rule-based baseline: click the element matching the goal, then stop.
+
+    Its purpose is twofold - it proves the harness works end to end with no model
+    at all, and it shows how much of a "computer use" task is really an indexing
+    problem. It waits when the target is not on screen yet, refuses to click the
+    same label twice, and gives up rather than flailing. It will not generalise
+    the way a VLM can.
+    """
+
+    name = "planner"
+
+    def __init__(self, target: str | None = None, *,
+                 done_when_text_contains: str | None = None,
+                 max_waits: int = 3, wait_seconds: float = 0.5) -> None:
+        self.target = target
+        self.done_when_text_contains = done_when_text_contains
+        self.max_waits = max_waits
+        self.wait_seconds = wait_seconds
+        self._clicked: set[str] = set()
+        self._waits = 0
+
+    def decide(self, goal: str, observation: Observation) -> Action:
+        if self.done_when_text_contains:
+            needle = self.done_when_text_contains.lower()
+            if needle in observation.text.lower():
+                return Action("done", reason=f"page text contains {self.done_when_text_contains!r}")
+
+        target = self.target or self._target_from(goal)
+        if target:
+            element = observation.find(target)
+            if element is None:
+                if self._waits < self.max_waits:
+                    self._waits += 1
+                    return Action("wait", seconds=self.wait_seconds,
+                                  reason=f"{target!r} is not on screen yet")
+                raise GoalUnreachable(f"{target!r} never appeared on screen")
+            key = element.label.strip().lower()
+            if key in self._clicked:
+                return Action("done", reason=f"already clicked {element.label!r}; goal assumed met")
+            self._clicked.add(key)
+            return Action("click", label=element.label,
+                          reason=f"matched {target!r} in the element index")
+        raise GoalUnreachable(f"no element name found in the goal {goal!r}")
+
+
+    @staticmethod
+    def _target_from(goal: str) -> str | None:
+        """Best effort: quoted text wins, else the last capitalised word."""
+        quoted = re.search(r"[\"\']([^\"\']{2,40})[\"\']", goal)
+        if quoted:
+            return quoted.group(1)
+        words = [w for w in goal.split() if w[:1].isupper()]
+        return words[-1].strip(".,!?") if words else None
+
+
+class VLMBrain:
+    """Seam for a real (small) vision-language model.
+
+    ``build_prompt`` and reply validation are implemented; subclass this and
+    implement ``_generate()`` to return the raw model text. Swap the model by
+    changing ``model_id`` - nothing else in the harness cares which weights you use.
+    """
+
+    name = "vlm"
+
+    def __init__(self, model_id: str = "vikhyatk/moondream2", *,
+                 max_new_tokens: int = 128) -> None:
+        self.model_id = model_id
+        self.max_new_tokens = max_new_tokens
+
+    def build_prompt(self, goal: str, observation: Observation) -> str:
+        return (
+            "You are controlling a computer. Choose the single next action.\n\n"
+            f"Goal: {goal}\n\n"
+            f"{observation.summary()}\n\n"
+            f"{ACTION_MENU}"
+        )
+
+    def _generate(self, prompt: str, observation: Observation) -> str:
+        raise NotImplementedError(
+            "No vision backend is wired up yet. Subclass VLMBrain and implement "
+            "_generate(prompt, observation) to return raw model text, or use "
+            "MockBrain / PlannerBrain while the harness is being tested."
+        )
+
+    def decide(self, goal: str, observation: Observation) -> Action:
+        raw = self._generate(self.build_prompt(goal, observation), observation)
+        payload = extract_json_object(raw)
+        try:
+            return parse_action(payload)
+        except ActionError as exc:
+            raise BrainError(f"model proposed an invalid action: {exc}") from exc

tiny_computer_agent/loop.py

@@ -0,0 +1,152 @@
+"""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, or a brain error
+    * 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) -> None:
+        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
+
+    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")
+        observation = self.screen.capture()
+
+        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
+
+            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
+
+            observation = self.screen.capture()  # verify: what did that change?
+
+        return episode
+
+    def _record(self, episode: Episode, step: Step) -> None:
+        episode.steps.append(step)
+        if self.on_step is not None:
+            self.on_step(step)

tiny_computer_agent/observe.py

@@ -0,0 +1,141 @@
+"""Perception: turn the world into an :class:`Observation` the brain can read.
+
+The harness never asks the model to *find* anything. OCR or the accessibility tree
+produces a labelled element index; the model only chooses among those labels.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Protocol, Sequence
+
+
+@dataclass(frozen=True)
+class Element:
+    """A clickable thing, located by the platform — not by the model."""
+
+    label: str
+    kind: str = "button"
+    x: int = 0
+    y: int = 0
+    width: int = 0
+    height: int = 0
+
+    def __post_init__(self) -> None:
+        if not self.label.strip():
+            raise ValueError("Element.label must be non-empty")
+        if min(self.x, self.y, self.width, self.height) < 0:
+            raise ValueError("Element geometry must be non-negative")
+
+    @property
+    def center(self) -> tuple[int, int]:
+        if self.width and self.height:
+            return (self.x + self.width // 2, self.y + self.height // 2)
+        return (self.x, self.y)
+
+    def matches(self, needle: str) -> bool:
+        return needle.strip().lower() in self.label.strip().lower()
+
+
+@dataclass(frozen=True)
+class Observation:
+    """One frame of the world: pixels plus whatever index we could build."""
+
+    index: int = 0
+    frame: Any = None
+    elements: tuple[Element, ...] = ()
+    text: str = ""
+
+    def find(self, label: str) -> Element | None:
+        """Exact label first, then substring; None if nothing matches."""
+        want = label.strip().lower()
+        exact = [e for e in self.elements if e.label.strip().lower() == want]
+        if exact:
+            return exact[0]
+        loose = [e for e in self.elements if e.matches(label)]
+        return loose[0] if loose else None
+
+    def summary(self, max_elements: int = 40) -> str:
+        """Compact text form — this is what a model would be shown."""
+        lines = [f"step {self.index}"]
+        if self.text:
+            lines.append("visible text: " + " ".join(self.text.split())[:400])
+        if self.elements:
+            lines.append("clickable elements:")
+            for e in self.elements[:max_elements]:
+                x, y = e.center
+                lines.append(f"  - {e.label!r} ({e.kind}) at {x},{y}")
+            if len(self.elements) > max_elements:
+                lines.append(f"  ... and {len(self.elements) - max_elements} more")
+        if not self.elements and not self.text:
+            lines.append("no element index available (pixels only)")
+        return "\n".join(lines)
+
+
+class ScreenSource(Protocol):
+    def capture(self) -> Observation: ...
+
+
+class FakeScreen:
+    """Scripted screen: each capture() advances one frame.
+
+    Used to exercise the loop with no display at all. The final frame repeats
+    forever, so a runaway agent cannot exhaust the script and crash.
+    """
+
+    def __init__(self, frames: Sequence[Sequence[Element]], *, text: str = "") -> None:
+        if not frames:
+            raise ValueError("FakeScreen needs at least one frame")
+        self._frames: tuple[tuple[Element, ...], ...] = tuple(tuple(f) for f in frames)
+        self._text = text
+        self.captures = 0
+
+    def capture(self) -> Observation:
+        i = min(self.captures, len(self._frames) - 1)
+        observation = Observation(
+            index=self.captures,
+            frame=f"frame-{i}",
+            elements=self._frames[i],
+            text=self._text,
+        )
+        self.captures += 1
+        return observation
+
+
+class ScreenUnavailable(RuntimeError):
+    """Raised when real capture is requested but the machine cannot provide it."""
+
+
+class GrabScreen:
+    """Real desktop capture via ``mss`` + Pillow.
+
+    Pixels only: nothing here builds an element index. Wiring OCR (tesseract) or
+    the accessibility tree (AT-SPI on Linux) is what turns frames into labels.
+    Needs a display, so it is untested in a headless sandbox.
+    """
+
+    def __init__(self, *, monitor: int = 1, max_width: int = 1280) -> None:
+        self.monitor = monitor
+        self.max_width = max_width
+        self._index = 0
+
+    def capture(self) -> Observation:
+        try:
+            import mss
+            from PIL import Image
+        except ImportError as exc:  # pragma: no cover - depends on the machine
+            raise ScreenUnavailable(
+                "GrabScreen needs 'mss' and 'pillow': pip install \'tiny-computer-agent[desktop]\'"
+            ) from exc
+        try:
+            with mss.mss() as sct:
+                shot = sct.grab(sct.monitors[self.monitor])
+                image = Image.frombytes("RGB", shot.size, shot.bgra, "raw", "BGRX")
+        except Exception as exc:  # pragma: no cover - depends on the machine
+            raise ScreenUnavailable(f"screen capture failed (no display?): {exc}") from exc
+        if image.width > self.max_width:
+            ratio = self.max_width / image.width
+            image = image.resize((self.max_width, int(image.height * ratio)))
+        observation = Observation(index=self._index, frame=image, elements=(), text="")
+        self._index += 1
+        return observation

tiny_computer_agent/safety.py

@@ -0,0 +1,93 @@
+"""The gates between a small model's mistake and your filesystem."""
+
+from __future__ import annotations
+
+import os
+import re
+from dataclasses import dataclass, field
+from typing import Pattern
+
+from tiny_computer_agent.actions import Action
+from tiny_computer_agent.observe import Observation
+
+DEFAULT_DENY_TEXT = (
+    r"\bsudo\b",
+    r"\brm\s+-rf\b",
+    r"\bmkfs\b",
+    r"\bdd\s+if=",
+    r"\bchmod\s+777\b",
+    r"\bgit\s+push\s+--force\b",
+    r"password",
+    r"api[_-]?key",
+    r"\bseed\s+phrase\b",
+)
+
+DEFAULT_DENY_LABELS = (
+    "terminal",
+    "shell",
+    "keychain",
+    "password manager",
+    "bank",
+    "wallet",
+    "credit card",
+)
+
+DEFAULT_DENY_KEYS = ("ctrl+alt+t", "ctrl+alt+delete", "ctrl+shift+escape")
+
+
+class PolicyViolation(RuntimeError):
+    """Raised when an action is refused. The loop stops; nothing is executed."""
+
+
+@dataclass
+class Policy:
+    """Rules an action must pass before it is executed.
+
+    dry_run            : when True (the default) the loop uses a DryRunExecutor
+    step_budget        : hard cap on actions per episode
+    deny_text_patterns : regexes a `type` action may not contain
+    deny_labels        : substrings that block a click on a matching element
+    deny_keys          : key chords that are never pressed
+    kill_switch_path   : if this file exists, the episode aborts before the next
+                         action — create it to stop a live run
+    """
+
+    dry_run: bool = True
+    step_budget: int = 20
+    deny_text_patterns: tuple[str, ...] = DEFAULT_DENY_TEXT
+    deny_labels: tuple[str, ...] = DEFAULT_DENY_LABELS
+    deny_keys: tuple[str, ...] = DEFAULT_DENY_KEYS
+    kill_switch_path: str | None = None
+    _compiled: tuple[Pattern[str], ...] = field(default=(), init=False, repr=False)
+
+    def __post_init__(self) -> None:
+        self._compiled = tuple(re.compile(p, re.IGNORECASE) for p in self.deny_text_patterns)
+        if self.step_budget < 1:
+            raise ValueError("step_budget must be at least 1")
+
+    def should_stop(self) -> bool:
+        """True when the kill switch exists. Checked before every action."""
+        return bool(self.kill_switch_path) and os.path.exists(self.kill_switch_path)
+
+    def check(self, action: Action, observation: Observation | None = None) -> None:
+        """Raise :class:`PolicyViolation` if the action is not allowed."""
+        if action.kind == "type" and action.text:
+            for pattern in self._compiled:
+                if pattern.search(action.text):
+                    raise PolicyViolation(
+                        f"typing {pattern.pattern!r} is on the deny list"
+                    )
+
+        if action.kind == "key" and action.keys:
+            chord = "+".join(k.strip().lower() for k in action.keys.split("+") if k.strip())
+            if chord in {k.lower() for k in self.deny_keys}:
+                raise PolicyViolation(f"key chord {action.keys!r} is on the deny list")
+
+        if action.kind == "click" and action.label:
+            element = observation.find(action.label) if observation is not None else None
+            haystack = (element.label if element is not None else action.label).lower()
+            for denied in self.deny_labels:
+                if denied.lower() in haystack:
+                    raise PolicyViolation(
+                        f"clicking {action.label!r} is on the deny list ({denied!r})"
+                    )