Publish tiny_computer_agent/ from marimo

tiny_computer_agent/__init__.py

@@ -25,6 +25,15 @@ An episode can be saved as a portable *trace*, replayed offline, and scored, so
     print(benchmark(scenarios, {"planner": lambda s: PlannerBrain()}).render())
 """
 
+from tiny_computer_agent.accessibility import (
+    INTERACTIVE_ROLES,
+    TEXT_ROLES,
+    AccessibleNode,
+    AtspiDetector,
+    AtspiReader,
+    TreeReader,
+    accessibility_available,
+)
 from tiny_computer_agent.act import (
     ActionResult,
     DryRunExecutor,
@@ -115,23 +124,25 @@ from tiny_computer_agent.trace import (
 __version__ = "0.3.0"
 
 __all__ = [
-    "ACTION_KINDS", "ACTION_MENU", "Action", "ActionError",
-    "ActionResult", "AgentLoop", "AutoDetector", "BRAIN_NAMES",
-    "Brain", "BrainError", "Check", "DEFAULT_KILL_SWITCH",
-    "DEFAULT_TEXT_MODEL", "DEFAULT_VISION_MODEL", "DecisionMenu", "Detector",
-    "DetectorUnavailable", "DryRunExecutor", "Element", "Episode",
-    "Executor", "FakeScreen", "GoalUnreachable", "GrabScreen",
-    "GridDetector", "ImageScreen", "IndexedScreen", "InputUnavailable",
+    "ACTION_KINDS", "ACTION_MENU", "AccessibleNode", "Action",
+    "ActionError", "ActionResult", "AgentLoop", "AtspiDetector",
+    "AtspiReader", "AutoDetector", "BRAIN_NAMES", "Brain",
+    "BrainError", "Check", "DEFAULT_KILL_SWITCH", "DEFAULT_TEXT_MODEL",
+    "DEFAULT_VISION_MODEL", "DecisionMenu", "Detector", "DetectorUnavailable",
+    "DryRunExecutor", "Element", "Episode", "Executor",
+    "FakeScreen", "GoalUnreachable", "GrabScreen", "GridDetector",
+    "INTERACTIVE_ROLES", "ImageScreen", "IndexedScreen", "InputUnavailable",
     "MODEL_BRAINS", "MenuItem", "MockBrain", "ModelBrain",
     "Observation", "OcrDetector", "PlannerBrain", "Policy",
     "PolicyViolation", "PyAutoGuiExecutor", "Scenario", "Scoreboard",
-    "ScreenSource", "ScreenUnavailable", "Step", "TextPlannerBrain",
-    "Trace", "TraceStep", "VLMBrain", "Verdict",
-    "__version__", "benchmark", "benchmark_traces", "build_agent",
-    "detector_available", "display_available", "doctor", "extract_json_object",
-    "extract_json_objects", "fake_desktop", "load_traces", "make_detector",
-    "make_model_brain", "model_cached", "model_dependencies", "normalise_label",
-    "parse_action", "parse_decision", "pick_dtype", "resolve_device",
-    "resolve_point", "run_checks", "run_scenario", "score",
-    "score_replay", "score_trace", "write_fake_desktop",
+    "ScreenSource", "ScreenUnavailable", "Step", "TEXT_ROLES",
+    "TextPlannerBrain", "Trace", "TraceStep", "TreeReader",
+    "VLMBrain", "Verdict", "__version__", "accessibility_available",
+    "benchmark", "benchmark_traces", "build_agent", "detector_available",
+    "display_available", "doctor", "extract_json_object", "extract_json_objects",
+    "fake_desktop", "load_traces", "make_detector", "make_model_brain",
+    "model_cached", "model_dependencies", "normalise_label", "parse_action",
+    "parse_decision", "pick_dtype", "resolve_device", "resolve_point",
+    "run_checks", "run_scenario", "score", "score_replay",
+    "score_trace", "write_fake_desktop",
 ]

tiny_computer_agent/__main__.py

@@ -68,8 +68,10 @@ def build_parser() -> argparse.ArgumentParser:
                              f"'{EXAMPLE_IMAGE}' uses the built-in one")
     parser.add_argument("--brain", default="planner", choices=BRAIN_NAMES,
                         help="planner (rule-based, no weights) | text-planner | vlm")
-    parser.add_argument("--detector", default="auto", choices=("auto", "ocr", "grid"),
-                        help="how to build the element index (auto = OCR, grid fallback)")
+    parser.add_argument("--detector", default="auto",
+                        choices=("auto", "best", "atspi", "ocr", "grid"),
+                        help="how to build the element index "
+                             "(best = accessibility tree, then OCR, then grid)")
     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",

tiny_computer_agent/accessibility.py

@@ -0,0 +1,266 @@
+"""Elements from the operating system's accessibility tree.
+
+OCR reads pixels and *guesses* which marks are widgets. The accessibility tree
+simply knows: every application that publishes one advertises each button, entry,
+menu item and label with a role, a name, and its real screen extents. That is
+faster than OCR, exact on small text the pixels lose, and - the part that matters
+for a tiny model - it produces the same short, labelled menu the rest of the
+harness already expects.
+
+The catch is honest and load-bearing: not every application publishes a tree.
+GTK/Qt/Electron and most native Windows apps do; Tk barely does, and a raw canvas
+does not at all. So this is a detector, not a replacement - when the tree is
+empty, `AutoDetector` falls through to OCR, and then to the numbered grid.
+
+Backends:
+
+* **AT-SPI** (Linux) - via PyGObject's ``Atspi`` typelib, implemented here and
+  exercised against a real GTK window.
+* **UI Automation** (Windows) - *not implemented*: :func:`accessibility_available`
+  reports it plainly rather than pretending, so an agent on Windows keeps using
+  OCR instead of silently indexing nothing.
+
+The coordinates AT-SPI reports are already absolute screen pixels, which is why
+:class:`AtspiDetector` sets ``absolute_coordinates``: :class:`IndexedScreen` must
+not re-map them through a downscale, or every click would land in the wrong place.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Protocol, Sequence
+
+from tiny_computer_agent.observe import Element
+
+# Roles that are worth clicking. Deliberately not "everything with a name": a
+# frame, a panel or a scrollbar is named too, and offering them to a model is how
+# you get an agent that clicks the window furniture.
+INTERACTIVE_ROLES: frozenset[str] = frozenset({
+    "button", "push button", "toggle button", "check box", "radio button",
+    "menu item", "check menu item", "radio menu item", "page tab", "list item",
+    "combo box", "entry", "text", "password text", "spin button", "slider",
+    "link", "icon", "table cell", "tree item", "tool bar", "menu",
+})
+
+# Named things that are not clickable, but are worth *reading* (a verifier, or a
+# VLM, needs the words on screen).
+TEXT_ROLES: frozenset[str] = frozenset({
+    "label", "paragraph", "heading", "static", "status bar", "text",
+})
+
+
+class AccessibleNode:
+    """One node of an accessibility tree, reduced to what a decision needs."""
+
+    __slots__ = ("role", "name", "x", "y", "width", "height", "depth")
+
+    def __init__(self, role: str, name: str, x: int, y: int, width: int, height: int,
+                 depth: int = 0) -> None:
+        self.role = role
+        self.name = name
+        self.x = x
+        self.y = y
+        self.width = width
+        self.height = height
+        self.depth = depth
+
+    def __repr__(self) -> str:                                    # pragma: no cover
+        return (f"AccessibleNode({self.role!r}, {self.name!r}, "
+                f"{self.x},{self.y} {self.width}x{self.height})")
+
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, AccessibleNode) and self.__slots__ and all(
+            getattr(self, slot) == getattr(other, slot) for slot in self.__slots__
+        )
+
+    def __hash__(self) -> int:
+        return hash((self.role, self.name, self.x, self.y, self.width, self.height))
+
+
+class TreeReader(Protocol):
+    """Reads the accessibility tree of the whole desktop."""
+
+    name: str
+
+    def read(self) -> tuple[AccessibleNode, ...]: ...
+
+
+class AtspiReader:
+    """Reads the live AT-SPI desktop through PyGObject.
+
+    Everything here is guarded: a machine without the typelib, without a session
+    bus, or without the accessibility bridge raises :class:`DetectorUnavailable`
+    with the fix, rather than returning an empty index that would look like "the
+    screen is empty" when it is really "the bridge is off".
+    """
+
+    name = "atspi"
+
+    def __init__(self, *, desktop: int = 0, max_depth: int = 10,
+                 max_nodes: int = 600, timeout_ms: int = 500) -> None:
+        self.desktop = desktop
+        self.max_depth = max_depth
+        self.max_nodes = max_nodes
+        self.timeout_ms = timeout_ms
+
+    @staticmethod
+    def _atspi():
+        try:
+            import gi
+        except ImportError as exc:                       # pragma: no cover - env
+            raise _unavailable(
+                "the AT-SPI reader needs PyGObject: pip install PyGObject, and "
+                "apt install gir1.2-atspi-2.0 at-spi2-core"
+            ) from exc
+        try:
+            gi.require_version("Atspi", "2.0")
+            from gi.repository import Atspi
+        except Exception as exc:                          # pragma: no cover - env
+            raise _unavailable(
+                "the Atspi typelib is missing: apt install gir1.2-atspi-2.0 "
+                "at-spi2-core (and run inside a session bus)"
+            ) from exc
+        return Atspi
+
+    def read(self) -> tuple[AccessibleNode, ...]:
+        Atspi = self._atspi()
+        try:
+            desktop = Atspi.get_desktop(self.desktop)
+        except Exception as exc:                          # pragma: no cover - env
+            raise _unavailable(
+                f"could not reach the accessibility bus: {type(exc).__name__}: {exc}. "
+                "Start a session bus (dbus-run-session) and leave NO_AT_BRIDGE unset."
+            ) from exc
+
+        nodes: list[AccessibleNode] = []
+        for index in range(desktop.get_child_count()):
+            self._walk(Atspi, desktop.get_child_at_index(index), 0, nodes)
+            if len(nodes) >= self.max_nodes:
+                break
+        return tuple(nodes)
+
+    def _walk(self, Atspi, node: Any, depth: int, out: list[AccessibleNode]) -> None:
+        if depth > self.max_depth or len(out) >= self.max_nodes:
+            return
+        try:
+            role = (node.get_role_name() or "").strip()
+            name = " ".join((node.get_name() or "").split())
+        except Exception:
+            return
+        extents = self._extents(Atspi, node)
+        if extents is not None:
+            x, y, width, height = extents
+            out.append(AccessibleNode(role, name, x, y, width, height, depth))
+        try:
+            children = node.get_child_count()
+        except Exception:
+            children = 0
+        for index in range(children):
+            try:
+                child = node.get_child_at_index(index)
+            except Exception:
+                continue
+            self._walk(Atspi, child, depth + 1, out)
+
+    def _extents(self, Atspi, node: Any) -> tuple[int, int, int, int] | None:
+        try:
+            component = node.get_component_iface()
+            if component is None:
+                return None
+            box = component.get_extents(Atspi.CoordType.SCREEN)
+        except Exception:
+            return None
+        if box is None:
+            return None
+        return int(box.x), int(box.y), int(box.width), int(box.height)
+
+
+def _unavailable(detail: str):
+    from tiny_computer_agent.perception import DetectorUnavailable
+    return DetectorUnavailable(detail)
+
+
+@dataclass
+class AtspiDetector:
+    """Turns the accessibility tree into the harness's element index.
+
+    ``kind`` is the AT-SPI role (``button``, ``check box``, ...), so a policy can
+    distinguish a button from a text field, and ``absolute_coordinates`` tells
+    :class:`IndexedScreen` these are already screen pixels.
+    """
+
+    name: str = "atspi"
+    absolute_coordinates: bool = True
+    reader: TreeReader | None = field(default=None)
+    interactive_roles: frozenset[str] = INTERACTIVE_ROLES
+    include_text: bool = False
+    min_label_chars: int = 1
+    max_label_chars: int = 64
+    min_size: int = 4
+    max_nodes: int = 120
+    _last_roles: tuple[str, ...] = field(default=(), init=False, repr=False)
+
+    def detect(self, image: Any = None) -> tuple[Element, ...]:
+        """Index the live tree. The image is accepted and deliberately ignored."""
+        reader = self.reader if self.reader is not None else AtspiReader()
+        nodes = reader.read()
+        wanted = self.interactive_roles | (TEXT_ROLES if self.include_text else frozenset())
+        elements: list[Element] = []
+        seen: set[tuple[str, int, int]] = set()
+        for node in nodes:
+            role = node.role.lower()
+            if role not in wanted:
+                continue
+            label = node.name
+            if not label or not (self.min_label_chars <= len(label) <= self.max_label_chars):
+                continue
+            if node.width < self.min_size or node.height < self.min_size:
+                continue
+            if node.x < 0 or node.y < 0:                  # off screen
+                continue
+            key = (label.lower(), node.x, node.y)
+            if key in seen:
+                continue
+            seen.add(key)
+            elements.append(Element(label=label, kind=role, x=node.x, y=node.y,
+                                    width=node.width, height=node.height))
+            if len(elements) >= self.max_nodes:
+                break
+        self._last_roles = tuple(element.kind for element in elements)
+        return tuple(elements)
+
+    def text(self) -> str:
+        """The words on screen, for a verifier - straight from the tree."""
+        reader = self.reader if self.reader is not None else AtspiReader()
+        return " ".join(node.name for node in reader.read()
+                        if node.name and node.role.lower() in (TEXT_ROLES | self.interactive_roles))
+
+
+def accessibility_available(backend: str = "atspi") -> tuple[bool, str]:
+    """(available, detail) for preflight and the doctor, honestly reported."""
+    if backend == "uia":
+        return False, ("UI Automation is not implemented yet: on Windows the harness "
+                       "falls back to OCR (see accessibility.py)")
+    if backend != "atspi":
+        return False, f"unknown accessibility backend {backend!r}"
+    try:
+        Atspi = AtspiReader._atspi()
+    except Exception as exc:
+        return False, str(exc)
+    try:
+        desktop = Atspi.get_desktop(0)
+        applications = desktop.get_child_count()
+    except Exception as exc:
+        return False, f"Atspi is installed but the a11y bus is unreachable: {exc}"
+    return True, f"AT-SPI via PyGObject ({applications} application(s) on the desktop)"
+
+
+__all__ = [
+    "INTERACTIVE_ROLES",
+    "TEXT_ROLES",
+    "AccessibleNode",
+    "AtspiDetector",
+    "AtspiReader",
+    "TreeReader",
+    "accessibility_available",
+]

tiny_computer_agent/brain.py

@@ -23,6 +23,15 @@ from dataclasses import dataclass
 from typing import Any, Iterator, Mapping, Protocol, Sequence
 
 from tiny_computer_agent.actions import ACTION_KINDS, Action, ActionError, parse_action
+from tiny_computer_agent.decoding import (
+    brevity_ladder,
+    choice_token_ids,
+    digit_token_ids,
+    majority_choice,
+    pair_token_ids,
+    read_choice,
+    read_verdict,
+)
 from tiny_computer_agent.observe import Element, Observation, normalise_label
 
 
@@ -348,12 +357,27 @@ class ModelBrain:
 
     name = "model"
     requires_index = False
+    # a backend that can mask its own logits opts in, and then decision_mode
+    # "choice" is available: ask for a menu number instead of asking for JSON
+    supports_constrained = False
+    # "json"   - free text, parsed leniently (the original behaviour)
+    # "choice" - one masked generation, a menu number
+    # "scan"   - one yes/no question per item, masked, first yes wins
+    # "rank"   - the same questions, but ranked by the yes/no log-odds margin,
+    #            which does not reward whichever item happened to be asked first
+    decision_mode = "json"
 
     def __init__(self, model_id: str, *, max_menu_items: int = 12,
-                 max_new_tokens: int = 64) -> None:
+                 max_new_tokens: int = 64, decision_mode: str = "json",
+                 retries: int = 3, votes: int = 1) -> None:
         self.model_id = model_id
         self.max_menu_items = max_menu_items
         self.max_new_tokens = max_new_tokens
+        self.decision_mode = decision_mode
+        self.retries = max(1, retries)
+        self.votes = max(1, votes)
+        self.last_choice: int | None = None
+        self.echoes_refused = 0
         self.last_prompt: str | None = None
         self.last_reply: str | None = None
         # "what I already tried", newest last; each entry's outcome is filled in at
@@ -367,13 +391,21 @@ class ModelBrain:
                 observation.text[:200])
 
     def build_prompt(self, goal: str, menu: DecisionMenu, observation: Observation,
-                     *, history: Sequence[str] = (), screen_changed: bool = True) -> str:
+                     *, history: Sequence[str] = (), screen_changed: bool = True,
+                     brevity: int = 0) -> str:
         """A short prompt with a strict output shape.
 
         Deliberately contains placeholders (``<number>``) rather than worked
         examples: the weakest models echo the prompt back, and an echoed
         placeholder fails validation instead of being mistaken for a decision.
+
+        ``brevity`` is the retry ladder: echoing is driven by how much text the
+        model has to copy, so each step upwards drops a section - first the screen
+        text and the history, then everything but the menu.
         """
+        if self.decision_mode == "choice":
+            return self._choice_prompt(goal, menu, observation,
+                                       history=history, brevity=brevity)
         lines = [
             "You are controlling a computer. Choose exactly one action.",
             "",
@@ -396,6 +428,27 @@ class ModelBrain:
                           "Choose a DIFFERENT action."]
         return "\n".join(lines)
 
+    def _choice_prompt(self, goal: str, menu: DecisionMenu, observation: Observation,
+                       *, history: Sequence[str] = (), brevity: int = 0) -> str:
+        """The constrained-mode prompt: one number, and as little text as possible.
+
+        There is no JSON shape to copy and no worked example - the whole point of
+        the masked path is that the model has nothing to echo.
+        """
+        lines = [
+            "You are controlling a computer. Reply with ONE NUMBER and nothing else.",
+            "0 means the goal is already met. Any other number is the item to click.",
+            "",
+            f"Goal: {goal}",
+        ]
+        if brevity == 0 and observation.text:
+            lines += ["", f"Text on screen: {observation.text[:200]}"]
+        lines += ["", f"Items ({len(menu)}):", menu.render() if menu else "  (none detected)"]
+        if brevity == 0 and history:
+            lines += ["", "Already tried, and it did not work:"]
+            lines += [f"  - {entry}" for entry in history]
+        return "\n".join(lines)
+
     def decide(self, goal: str, observation: Observation) -> Action:
         menu = DecisionMenu(observation, max_items=self.max_menu_items)
         if self.requires_index and not menu:
@@ -420,14 +473,182 @@ class ModelBrain:
             goal, menu, observation, history=tuple(self.history),
             screen_changed=screen_changed,
         )
-        self.last_reply = self._generate(self.last_prompt, observation)
-        action = parse_decision(self.last_reply, observation,
-                                max_menu_items=self.max_menu_items)
+        if self.decision_mode == "rank" and self.supports_constrained:
+            action = self._decide_by_ranking(goal, menu, observation)
+        elif self.decision_mode == "scan" and self.supports_constrained:
+            action = self._decide_by_scanning(goal, menu, observation)
+        elif self.decision_mode == "choice" and self.supports_constrained:
+            action = self._decide_by_choice(goal, menu, observation)
+        else:
+            self.last_reply = self._generate(self.last_prompt, observation)
+            action = parse_decision(self.last_reply, observation,
+                                    max_menu_items=self.max_menu_items)
         self.history.append(action.describe())
         del self.history[:-3]
         self._last_fingerprint = fingerprint
         return action
 
+    def _decide_by_choice(self, goal: str, menu: DecisionMenu,
+                          observation: Observation) -> Action:
+        """Ask for a menu number with every other token masked out.
+
+        The echo failure cannot happen here: the model is unable to emit the prompt
+        back, only one of the numbers we allow. When even that fails, the prompt
+        gets shorter (echoing is driven by prompt length) and, if ``votes`` is set,
+        the answer is taken by majority over several samples.
+        """
+        if not menu:
+            raise GoalUnreachable("no clickable items to choose from")
+        tokenizer = self._tokenizer_for_choice()
+        allowed = choice_token_ids(tokenizer, len(menu))
+        if not allowed:
+            allowed = digit_token_ids(tokenizer)
+        if not allowed:
+            raise BrainError(
+                "this tokenizer cannot express a menu number as a single token, so "
+                "the choice cannot be constrained; use decision_mode='json'"
+            )
+
+        last_error: BrainError | None = None
+        for brevity in brevity_ladder(self.retries):
+            self.last_prompt = self.build_prompt(
+                goal, menu, observation, history=tuple(self.history), brevity=brevity)
+            replies = [
+                self._generate_constrained(self.last_prompt, observation, allowed, 4,
+                                           sample=index > 0)
+                for index in range(self.votes)
+            ]
+            self.last_reply = replies[0] if replies else ""
+            choices = [read_choice(reply, len(menu)) for reply in replies]
+            number = majority_choice(choices)
+            if number is None:
+                self.echoes_refused += sum(1 for reply in replies
+                                           if reply and reply.strip() == self.last_prompt.strip())
+                last_error = BrainError(
+                    f"the constrained reply was not a menu number: "
+                    f"{self.last_reply[:80]!r} (brevity {brevity}, {len(replies)} vote(s))"
+                )
+                continue
+            self.last_choice = number
+            if number == 0:
+                return Action("done", reason="the model chose 0: the goal is already met")
+            item = menu.item(number)
+            if item is None:                                   # pragma: no cover
+                last_error = BrainError(f"the model chose {number}, outside the menu")
+                continue
+            return Action("click", label=item.label,
+                          reason=f"constrained choice {number} of {len(menu)}")
+
+        raise last_error or BrainError("the model produced no constrained choice")
+
+    def _scan_prompt(self, goal: str, label: str) -> str:
+        """The prompt for one binary question: does *this* item do the job?
+
+        Short on purpose. Every word a tiny model is given is a word it can copy
+        instead of answering, and there is nothing here worth copying back.
+        """
+        return "\n".join([
+            "You are controlling a computer. Answer with one word: yes or no.",
+            "",
+            f"Goal: {goal}",
+            f"Candidate item: {label}",
+            "Does clicking this item achieve the goal?",
+        ])
+
+    def _decide_by_scanning(self, goal: str, menu: DecisionMenu,
+                            observation: Observation) -> Action:
+        """Ask one masked yes/no question per menu item, and take the first yes.
+
+        Why this rather than one N-way choice: a 256M model cannot hold twelve
+        options in mind, but "does this button do what the goal says?" is a
+        question it can answer. Each answer is one token out of two, so the mask
+        makes the reply space as small as it can possibly be - and the scan stops
+        at the first yes, so it is usually cheaper than it looks.
+        """
+        if not menu:
+            raise GoalUnreachable("no clickable items to choose from")
+        tokenizer = self._tokenizer_for_choice()
+        yes_ids, no_ids = pair_token_ids(tokenizer)
+        if not yes_ids or not no_ids:
+            raise BrainError(
+                "this tokenizer cannot express yes/no as single tokens, so the scan "
+                "cannot be constrained; use decision_mode='json'"
+            )
+        allowed = sorted(set(yes_ids) | set(no_ids))
+        scanned: list[tuple[str, str | None]] = []
+        for item in menu:
+            self.last_prompt = self._scan_prompt(goal, item.label)
+            self.last_reply = self._generate_constrained(
+                self.last_prompt, observation, allowed, 1, sample=False)
+            verdict = read_verdict(self.last_reply)
+            scanned.append((item.label, verdict))
+            if verdict is None:
+                # a reply that is neither yes nor no means the mask is not being
+                # applied; refuse rather than fall through to a guess
+                raise BrainError(
+                    f"the masked reply to a yes/no question was {self.last_reply[:40]!r}"
+                )
+            if verdict == "yes":
+                return Action("click", label=item.label,
+                              reason=f"the model said yes to {item.label!r} "
+                                     f"({len(scanned)} of {len(menu)} scanned)")
+        self.last_reply = ""
+        raise BrainError(
+            f"the model said no to all {len(menu)} item(s): "
+            f"{[label for label, _ in scanned][:6]}"
+        )
+
+    def _decide_by_ranking(self, goal: str, menu: DecisionMenu,
+                           observation: Observation) -> Action:
+        """Ask every item, then click the one the model liked most.
+
+        ``scan`` stops at the first yes, which quietly rewards position. Here the
+        model is asked about every candidate and the winner is the largest
+        yes-minus-no log-odds margin - the model's own opinion, read off the
+        distribution instead of off a sampled token. Costs N short generations,
+        and it is the cheapest way to find out whether the weights can do it at all.
+        """
+        if not menu:
+            raise GoalUnreachable("no clickable items to choose from")
+        tokenizer = self._tokenizer_for_choice()
+        yes_ids, no_ids = pair_token_ids(tokenizer)
+        if not yes_ids or not no_ids:
+            raise BrainError(
+                "this tokenizer cannot express yes/no as single tokens, so the ranking "
+                "cannot be constrained; use decision_mode='json'"
+            )
+        scored: list[tuple[Any, float]] = []
+        for item in menu:
+            self.last_prompt = self._scan_prompt(goal, item.label)
+            margin = self._score_pair(self.last_prompt, observation, yes_ids, no_ids)
+            scored.append((item, margin))
+        scored.sort(key=lambda pair: pair[1], reverse=True)
+        best, margin = scored[0]
+        self.last_reply = f"margin {margin:+.2f}"
+        if margin <= 0:
+            raise BrainError(
+                f"the model preferred no item: best was {best.label!r} at {margin:+.2f}; "
+                f"runner-up {scored[1][0].label!r} at {scored[1][1]:+.2f}"
+                if len(scored) > 1 else
+                f"the model preferred no item (only {best.label!r}, {margin:+.2f})"
+            )
+        return Action("click", label=best.label,
+                      reason=f"highest yes/no margin {margin:+.2f} of {len(scored)} item(s)")
+
+    def _score_pair(self, prompt: str, observation: Observation,
+                    yes_ids: Sequence[int], no_ids: Sequence[int]) -> float:
+        """How much more likely "yes" is than "no". Backends implement this."""
+        raise NotImplementedError("a constrained backend implements _score_pair")
+
+    def _tokenizer_for_choice(self) -> Any:
+        """The tokenizer used to build the mask. Subclasses load it lazily."""
+        raise NotImplementedError("a constrained backend provides its tokenizer")
+
+    def _generate_constrained(self, prompt: str, observation: Observation,
+                              allowed_ids: Sequence[int], max_new_tokens: int,
+                              *, sample: bool = False) -> str:
+        raise NotImplementedError("a constrained backend implements this")
+
     @property
     def loaded(self) -> bool:
         """True once the weights are in memory - they load lazily, on the first decide()."""

tiny_computer_agent/decoding.py

@@ -0,0 +1,179 @@
+"""Making a tiny model answer with a choice instead of repeating the question.
+
+A 256M model asked for JSON on a busy screen does the thing it is best at: it
+continues the text it was handed, and gives the prompt back. The harness already
+*detects* that (echo detection refuses the reply) but detecting it is not fixing
+it - the episode simply fails.
+
+The fix is to stop asking for free text. A menu of twelve items has twelve legal
+answers, so mask every other token: an echo becomes impossible rather than merely
+caught, and a model that cannot write a sentence can still pick a number.
+
+Three tools, all model-agnostic:
+
+* :func:`digit_token_ids` / :func:`choice_token_ids` - the token ids a menu answer
+  can possibly need
+* :class:`MenuMask` - a transformers ``LogitsProcessor`` that forbids everything
+  else, so "hello, here is your prompt back" is not representable
+* :func:`brevity_ladder` and :func:`majority_choice` - a shorter prompt on the
+  retry (echoing is driven by prompt length) and a vote over k attempts
+"""
+
+from __future__ import annotations
+
+from typing import Any, Iterable, Sequence
+
+DIGITS = tuple(str(d) for d in range(10))
+
+
+def _single_token_ids(tokenizer: Any, text: str) -> list[int]:
+    """Ids that render ``text`` as exactly one token, with and without a space."""
+    ids: list[int] = []
+    for variant in (text, " " + text):
+        try:
+            encoded = tokenizer.encode(variant, add_special_tokens=False)
+        except Exception:
+            continue
+        if len(encoded) == 1:
+            ids.append(int(encoded[0]))
+    return ids
+
+
+def digit_token_ids(tokenizer: Any) -> list[int]:
+    """Every single-token digit, so any menu number can be spelled out.
+
+    The loosest useful mask: ten or twenty tokens instead of a 50k vocabulary.
+    A model restricted to these cannot answer with prose or with the prompt.
+    """
+    ids: list[int] = []
+    for digit in DIGITS:
+        ids += _single_token_ids(tokenizer, digit)
+    return sorted(set(ids))
+
+
+def choice_token_ids(tokenizer: Any, count: int, *, allow_done: bool = True) -> list[int]:
+    """The single-token ids for ``0..count`` (0 means "the goal is already met").
+
+    Stricter than :func:`digit_token_ids`: the model may emit exactly one of the
+    legal answers, and nothing else, in a single token.
+    """
+    ids: list[int] = []
+    for number in range(0 if allow_done else 1, count + 1):
+        ids += _single_token_ids(tokenizer, str(number))
+    return sorted(set(ids))
+
+
+def pair_token_ids(tokenizer: Any, *, yes: Sequence[str] = ("yes", "Yes"),
+                   no: Sequence[str] = ("no", "No")) -> tuple[list[int], list[int]]:
+    """Single-token ids for a yes/no answer.
+
+    Two tokens instead of a whole vocabulary: a 256M model that cannot pick 1-of-12
+    can often still answer "does this label do what the goal asks?".
+    """
+    yes_ids: list[int] = []
+    no_ids: list[int] = []
+    for text in yes:
+        yes_ids += _single_token_ids(tokenizer, text)
+    for text in no:
+        no_ids += _single_token_ids(tokenizer, text)
+    return sorted(set(yes_ids)), sorted(set(no_ids))
+
+
+def read_verdict(text: str) -> str | None:
+    """'yes', 'no', or None when the reply is neither."""
+    cleaned = (text or "").strip().lower().lstrip(":").strip()
+    if cleaned.startswith("yes"):
+        return "yes"
+    if cleaned.startswith("no"):
+        return "no"
+    return None
+
+
+def MenuMask(allowed_ids: Sequence[int], *, eos_token_id: int | None = None,
+             force_first: bool = True):
+    """A ``LogitsProcessor`` that permits only the ids in ``allowed_ids``.
+
+    Returns a plain callable with the transformers signature
+    ``(input_ids, scores) -> scores``, so it needs no torch import at module load
+    and can be tested without a model.
+
+    ``force_first`` closes a hole that a real 256M model found immediately: if the
+    end-of-sequence token is legal from the start, the cheapest continuation is to
+    stop, and the answer comes back *empty*. The mask therefore withholds EOS until
+    one real token has been emitted, so the model has to say something.
+    """
+    allowed = sorted(set(int(i) for i in allowed_ids))
+    state = {"calls": 0}
+
+    def process(input_ids: Any, scores: Any) -> Any:
+        mask = scores.new_full(scores.shape, float("-inf"))
+        for index in allowed:
+            if 0 <= index < scores.shape[-1]:
+                mask[..., index] = scores[..., index]
+        if eos_token_id is not None:
+            stop = int(eos_token_id)
+            if not force_first or state["calls"] > 0:
+                if 0 <= stop < scores.shape[-1]:
+                    mask[..., stop] = scores[..., stop]
+        state["calls"] += 1
+        return mask
+
+    process.allowed_ids = tuple(allowed)          # for tests and for the trace
+    process.eos_token_id = eos_token_id
+    process.force_first = force_first
+    return process
+
+
+def read_choice(text: str, count: int, *, allow_done: bool = True) -> int | None:
+    """The menu number in a reply, or None if there is not exactly one.
+
+    Only the digits are read: with the mask in place the reply *is* a number, and
+    anything else means the model is not being constrained (so it is refused
+    rather than guessed at).
+    """
+    if not text:
+        return None
+    digits = "".join(character for character in text if character.isdigit())
+    if not digits:
+        return None
+    try:
+        number = int(digits)
+    except ValueError:                            # pragma: no cover - defensive
+        return None
+    low = 0 if allow_done else 1
+    return number if low <= number <= count else None
+
+
+def brevity_ladder(levels: int = 3) -> tuple[int, ...]:
+    """The brevity settings to try, in order: full prompt first, terse last."""
+    return tuple(range(0, max(1, levels)))
+
+
+def majority_choice(choices: Iterable[int | None]) -> int | None:
+    """The most common choice, ignoring the failures. Ties go to the first seen."""
+    counts: dict[int, int] = {}
+    order: list[int] = []
+    for choice in choices:
+        if choice is None:
+            continue
+        if choice not in counts:
+            counts[choice] = 0
+            order.append(choice)
+        counts[choice] += 1
+    if not counts:
+        return None
+    best = max(counts.values())
+    return next(choice for choice in order if counts[choice] == best)
+
+
+__all__ = [
+    "DIGITS",
+    "MenuMask",
+    "brevity_ladder",
+    "choice_token_ids",
+    "digit_token_ids",
+    "majority_choice",
+    "pair_token_ids",
+    "read_choice",
+    "read_verdict",
+]

tiny_computer_agent/loop.py

@@ -97,97 +97,170 @@ class AgentLoop:
         self.on_step = on_step
         self.verify = verify
         self.max_repeats = max_repeats
-
-    def run(self, goal: str, *, max_steps: int | None = None) -> Episode:
+        # 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
-        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
+        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:
-            observation = self.screen.capture()
+            self.observation = self.screen.capture()
         except Exception as exc:
-            episode.stop_reason = f"capture failed: {type(exc).__name__}: {exc}"
-            return episode
+            self.observation = None
+            self._finish(f"capture failed: {type(exc).__name__}: {exc}")
+        return self
 
-        for index in range(budget):
-            if self.policy.should_stop():
-                episode.stop_reason = "kill switch engaged"
-                break
+    @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.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
+        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":
-                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
+        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:
-                self._record(episode, Step(index, observation, action=action,
-                                           error=f"refused: {exc}"))
-                episode.stop_reason = f"policy refusal: {exc}"
-                break
+        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())
 
-            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:
+            return self._record_step(
+                Step(index, observation, action=action,
+                     error=f"executor failed: {type(exc).__name__}: {exc}"),
+                "executor error")
 
-            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
+        step = Step(index, observation, action=action, result=result)
+        self._append(step)
+        if not result.ok:
+            self._finish(f"action failed: {result.note}")
+            return step
 
-            self._record(episode, Step(index, observation, action=action, result=result))
-            if not result.ok:
-                episode.stop_reason = f"action failed: {result.note}"
+        try:
+            self.observation = self.screen.capture()     # verify: what did that change?
+        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
 
-            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
+    def _append(self, step: Step) -> None:
+        self.episode.steps.append(step)
+        if self.on_step is not None:
+            self.on_step(step)
 
-            if self.verify is not None and self._verified(observation):
-                episode.success = True
-                episode.stop_reason = f"verified after {index + 1} action(s)"
-                break
+    def _record_step(self, step: Step, stop_reason: str) -> Step:
+        self._append(step)
+        self._finish(stop_reason)
+        return step
 
-        return episode
+    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."""

tiny_computer_agent/models.py

@@ -20,12 +20,37 @@ import os
 from typing import Any
 
 from tiny_computer_agent.brain import BrainError, ModelBrain
+from tiny_computer_agent.decoding import MenuMask
 from tiny_computer_agent.observe import Observation
 
 DEFAULT_TEXT_MODEL = "Qwen/Qwen2.5-0.5B-Instruct"
 DEFAULT_VISION_MODEL = "HuggingFaceTB/SmolVLM-256M-Instruct"
 
 
+def _pair_margin(logits: Any, yes_ids: Any, no_ids: Any) -> float:
+    """log P(yes) - log P(no) for one position, over the whole vocabulary."""
+    import torch
+
+    yes = torch.logsumexp(logits[list(yes_ids)], dim=0)
+    no = torch.logsumexp(logits[list(no_ids)], dim=0)
+    return float((yes - no).item())
+
+
+def _masked_generation(model: Any, allowed_ids: Any, eos_token_id: Any) -> Any:
+    """A transformers ``LogitsProcessorList`` holding the menu mask.
+
+    Imported here, not at module scope, so the harness still imports with no
+    transformers installed at all.
+    """
+    try:
+        from transformers import LogitsProcessorList
+    except ImportError:                                   # pragma: no cover - env
+        raise BrainError(
+            "constrained decoding needs transformers: pip install 'tiny-computer-agent[vision]'"
+        )
+    return LogitsProcessorList([MenuMask(allowed_ids, eos_token_id=eos_token_id)])
+
+
 def resolve_device(prefer: str | None = None) -> str:
     """Pick a torch device: explicit preference, then CUDA, then CPU."""
     try:
@@ -73,6 +98,7 @@ class TextPlannerBrain(ModelBrain):
 
     name = "text-planner"
     requires_index = True
+    supports_constrained = True
 
     def __init__(self, model_id: str = DEFAULT_TEXT_MODEL, *, device: str | None = None,
                  **kwargs: Any) -> None:
@@ -101,6 +127,53 @@ class TextPlannerBrain(ModelBrain):
         self._model = None
         self._tokenizer = None
 
+    def _tokenizer_for_choice(self) -> Any:
+        self._load()
+        return self._tokenizer
+
+    def _generate_constrained(self, prompt: str, observation: Observation,
+                              allowed_ids: Any, max_new_tokens: int,
+                              *, sample: bool = False) -> str:
+        """One short generation that can only contain a menu number."""
+        import torch
+
+        self._load()
+        messages = [{"role": "user", "content": prompt}]
+        text = self._tokenizer.apply_chat_template(
+            messages, add_generation_prompt=True, tokenize=False
+        )
+        inputs = self._tokenizer(text, return_tensors="pt").to(self.device)
+        settings: dict[str, Any] = {
+            "max_new_tokens": max_new_tokens,
+            "pad_token_id": self._tokenizer.eos_token_id,
+            "logits_processor": _masked_generation(
+                self._model, allowed_ids, self._tokenizer.eos_token_id),
+        }
+        if sample:
+            settings.update(do_sample=True, temperature=0.7, top_p=0.9)
+        else:
+            settings["do_sample"] = False
+        with torch.inference_mode():
+            output = self._model.generate(**inputs, **settings)
+        return self._tokenizer.decode(
+            output[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True
+        ).strip()
+
+    def _score_pair(self, prompt: str, observation: Observation,
+                    yes_ids: Any, no_ids: Any) -> float:
+        import torch
+
+        self._load()
+        messages = [{"role": "user", "content": prompt}]
+        text = self._tokenizer.apply_chat_template(
+            messages, add_generation_prompt=True, tokenize=False)
+        inputs = self._tokenizer(text, return_tensors="pt").to(self.device)
+        with torch.inference_mode():
+            out = self._model.generate(
+                **inputs, max_new_tokens=1, do_sample=False, return_dict_in_generate=True,
+                output_scores=True, pad_token_id=self._tokenizer.eos_token_id)
+        return _pair_margin(out.scores[0][0], yes_ids, no_ids)
+
     def _generate(self, prompt: str, observation: Observation) -> str:
         import torch
 
@@ -127,6 +200,7 @@ class VLMBrain(ModelBrain):
 
     name = "vlm"
     requires_index = False
+    supports_constrained = True
 
     def __init__(self, model_id: str = DEFAULT_VISION_MODEL, *, device: str | None = None,
                  **kwargs: Any) -> None:
@@ -155,6 +229,68 @@ class VLMBrain(ModelBrain):
         self._model = None
         self._processor = None
 
+    def _tokenizer_for_choice(self) -> Any:
+        self._load()
+        return self._processor.tokenizer
+
+    def _generate_constrained(self, prompt: str, observation: Observation,
+                              allowed_ids: Any, max_new_tokens: int,
+                              *, sample: bool = False) -> str:
+        """A vision generation that can only contain a menu number.
+
+        This is the fix for the echo: the model is no longer able to reply with
+        the prompt, because the prompt's tokens are masked out of the logits.
+        """
+        import torch
+
+        image = observation.frame
+        if image is None or isinstance(image, str):
+            raise BrainError(
+                "the vision backend needs a real screenshot but the observation carries "
+                f"{type(image).__name__} - run with --live, or use the text planner"
+            )
+        self._load()
+        messages = [{"role": "user", "content": [{"type": "image"},
+                                                 {"type": "text", "text": prompt}]}]
+        text = self._processor.apply_chat_template(messages, add_generation_prompt=True)
+        inputs = self._processor(text=text, images=[image], return_tensors="pt").to(self.device)
+        settings: dict[str, Any] = {
+            "max_new_tokens": max_new_tokens,
+            "logits_processor": _masked_generation(
+                self._model, allowed_ids,
+                getattr(self._processor.tokenizer, "eos_token_id", None)),
+        }
+        if sample:
+            settings.update(do_sample=True, temperature=0.7, top_p=0.9)
+        else:
+            settings["do_sample"] = False
+        with torch.inference_mode():
+            output = self._model.generate(**inputs, **settings)
+        return self._processor.batch_decode(
+            output[:, inputs["input_ids"].shape[1]:], skip_special_tokens=True
+        )[0].strip()
+
+    def _score_pair(self, prompt: str, observation: Observation,
+                    yes_ids: Any, no_ids: Any) -> float:
+        """log P(yes) - log P(no) for one image question, read off the logits."""
+        import torch
+
+        image = observation.frame
+        if image is None or isinstance(image, str):
+            raise BrainError(
+                "the vision backend needs a real screenshot but the observation carries "
+                f"{type(image).__name__} - run with --live, or use the text planner"
+            )
+        self._load()
+        messages = [{"role": "user", "content": [{"type": "image"},
+                                                 {"type": "text", "text": prompt}]}]
+        text = self._processor.apply_chat_template(messages, add_generation_prompt=True)
+        inputs = self._processor(text=text, images=[image], return_tensors="pt").to(self.device)
+        with torch.inference_mode():
+            out = self._model.generate(**inputs, max_new_tokens=1, do_sample=False,
+                                       return_dict_in_generate=True, output_scores=True)
+        return _pair_margin(out.scores[0][0], yes_ids, no_ids)
+
     def _generate(self, prompt: str, observation: Observation) -> str:
         import torch
 

tiny_computer_agent/perception.py

@@ -118,8 +118,11 @@ class IndexedScreen:
             )
             return observation
         mapper = getattr(self.source, "map_to_screen", None)
-        if mapper is not None:
+        already_absolute = getattr(self.detector, "absolute_coordinates", False)
+        if mapper is not None and not already_absolute:
             # Frame pixels -> absolute screen pixels, so a click cannot drift.
+            # A detector that already reports screen coordinates (the accessibility
+            # tree) opts out of this, or a downscaled capture would double-map it.
             elements = tuple(mapper(element) for element in elements)
         text = observation.text
         if self.fill_text and not text:
@@ -176,8 +179,18 @@ class AutoDetector:
     name: str = "auto"
     ocr: Detector = field(default_factory=OcrDetector)
     grid: Detector = field(default_factory=GridDetector)
+    # opt in (the "best" detector does): the accessibility tree is tried first,
+    # because it is exact where OCR is a guess - and skipped when it is empty
+    accessibility: Detector | None = None
 
     def detect(self, image: Any) -> tuple[Element, ...]:
+        if self.accessibility is not None:
+            try:
+                elements = self.accessibility.detect(image)
+            except DetectorUnavailable:
+                elements = ()
+            if elements:
+                return elements
         try:
             elements = self.ocr.detect(image)
         except DetectorUnavailable:
@@ -201,6 +214,18 @@ def detector_available(name: str = "auto") -> tuple[bool, str]:
     """(available, detail) for preflight and the ``doctor`` command."""
     if name == "grid":
         return True, "numbered grid cells (no dependencies)"
+    if name == "atspi":
+        from tiny_computer_agent.accessibility import accessibility_available
+
+        return accessibility_available("atspi")
+    if name == "best":
+        from tiny_computer_agent.accessibility import accessibility_available
+
+        reachable, detail = accessibility_available("atspi")
+        if reachable:
+            return True, (f"accessibility tree when it has content, then OCR, then a "
+                          f"numbered grid ({detail})")
+        return True, (f"OCR, then a numbered grid (no accessibility tree: {detail})")
     if name == "auto":
         available, detail = _ocr_available()
         if available:
@@ -218,6 +243,15 @@ def make_detector(name: str = "auto", **kwargs: Any) -> Detector:
         raise DetectorUnavailable(detail)
     if name == "grid":
         return GridDetector(**kwargs)
+    if name == "atspi":
+        from tiny_computer_agent.accessibility import AtspiDetector
+
+        return AtspiDetector(**kwargs)
+    if name == "best":
+        from tiny_computer_agent.accessibility import AtspiDetector, accessibility_available
+
+        accessibility = AtspiDetector() if accessibility_available("atspi")[0] else None
+        return AutoDetector(accessibility=accessibility, **kwargs)
     if name == "auto":
         if _ocr_available()[0]:
             return AutoDetector(**kwargs)