"""Wiring for a real desktop: preflight checks, then one-call agent construction. This module exists because "make it work on my computer" is mostly a diagnosis problem. Rather than failing deep inside an episode, :func:`doctor` reports exactly which piece is missing - display, capture backend, input backend, OCR engine, model runtime - and what to do about each. """ from __future__ import annotations import importlib.util import os import platform from dataclasses import dataclass from typing import Any, Callable, Iterable, Sequence from tiny_computer_agent.act import DryRunExecutor, Executor, PyAutoGuiExecutor from tiny_computer_agent.brain import Brain, PlannerBrain from tiny_computer_agent.loop import AgentLoop, Step from tiny_computer_agent.observe import GrabScreen, ScreenSource from tiny_computer_agent.perception import ( Detector, IndexedScreen, detector_available, make_detector, ) from tiny_computer_agent.gates import Gate from tiny_computer_agent.safety import Policy from tiny_computer_agent.skills import SkillBrain, SkillLibrary DEFAULT_KILL_SWITCH = "/tmp/tiny-computer-agent.STOP" BRAIN_NAMES = ("planner", "text-planner", "vlm") @dataclass(frozen=True) class Check: """One preflight result: what was checked, whether it works, and how to fix it.""" name: str ok: bool detail: str fix: str = "" required: bool = True def render(self) -> str: marker = "ok " if self.ok else ("FAIL" if self.required else "warn") line = f" [{marker}] {self.name}: {self.detail}" if not self.ok and self.fix: line += f"\n fix: {self.fix}" return line def _installed(module: str) -> bool: return importlib.util.find_spec(module) is not None def display_available() -> tuple[bool, str]: """Whether a windowing session looks reachable from this process.""" system = platform.system() if system in ("Darwin", "Windows"): return True, f"{system} desktop session assumed" display = os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY") if display: return True, f"DISPLAY={display}" return False, "no DISPLAY or WAYLAND_DISPLAY set (headless container or SSH session)" def run_checks(*, brain: str = "planner", model_id: str | None = None, detector: str = "ocr", real_input: bool = False, kill_switch_path: str | None = DEFAULT_KILL_SWITCH) -> list[Check]: """Every precondition for driving a real desktop, in one list.""" checks: list[Check] = [] ok, detail = display_available() checks.append(Check( "display server", ok, detail, fix="run this on the desktop session itself (a headless server cannot be driven)", )) checks.append(Check( "pillow", _installed("PIL"), "installed" if _installed("PIL") else "not installed", fix="pip install 'tiny-computer-agent[desktop]'", )) has_mss = _installed("mss") checks.append(Check( "screen capture", has_mss or _installed("PIL"), "mss (fast, multi-monitor)" if has_mss else "pillow ImageGrab fallback", fix="pip install mss for faster multi-monitor capture", required=False, )) has_pyautogui = _installed("pyautogui") checks.append(Check( "input backend (pyautogui)", has_pyautogui, "installed" if has_pyautogui else "not installed", fix="pip install 'tiny-computer-agent[desktop]'", required=real_input, )) detector_ok, detector_detail = detector_available(detector) checks.append(Check( f"element detector ({detector})", detector_ok, detector_detail, fix="pip install 'tiny-computer-agent[ocr]'", )) if brain == "planner": checks.append(Check( "model runtime", True, "not needed: the planner is rule-based (no weights, no download)", required=False, )) else: from tiny_computer_agent.models import ( DEFAULT_TEXT_MODEL, DEFAULT_VISION_MODEL, model_cached, model_dependencies, ) deps_ok, deps_detail = model_dependencies() model = model_id or (DEFAULT_TEXT_MODEL if brain == "text-planner" else DEFAULT_VISION_MODEL) checks.append(Check( "model runtime", deps_ok, deps_detail, fix="pip install 'tiny-computer-agent[vision]'", )) checks.append(Check( f"model weights ({model})", model_cached(model), "cached locally" if model_cached(model) else "not downloaded yet", fix=f"first run will download {model} from the Hub (one time)", required=False, )) checks.append(Check( "execution mode", True, "REAL mouse and keyboard input" if real_input else "dry run (nothing is clicked)", fix="pass --real-input AND set TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT=1", required=False, )) checks.append(Check( "kill switch", True, f"create {kill_switch_path} to abort the next action" if kill_switch_path else "disabled", required=False, )) return checks def doctor(*, brain: str = "planner", model_id: str | None = None, detector: str = "ocr", real_input: bool = False, kill_switch_path: str | None = DEFAULT_KILL_SWITCH) -> str: """A human-readable preflight report.""" checks = run_checks(brain=brain, model_id=model_id, detector=detector, real_input=real_input, kill_switch_path=kill_switch_path) blocking = [c for c in checks if c.required and not c.ok] lines = [ f"tiny-computer-agent preflight ({platform.system()} {platform.machine()}, python {platform.python_version()})", "", *(c.render() for c in checks), "", ] if blocking: lines.append(f"NOT READY: {len(blocking)} blocking problem(s): " + ", ".join(c.name for c in blocking)) else: lines.append("READY: the harness can drive this desktop.") if not real_input: lines.append("Dry run is on: nothing will be clicked. Read the plan first.") return "\n".join(lines) def build_agent(*, brain: Brain | str | None = None, screen: ScreenSource | None = None, detector: Detector | str | None = None, dry_run: bool = True, max_steps: int = 10, kill_switch_path: str | None = DEFAULT_KILL_SWITCH, allow_grid_clicks: bool = False, on_step: Callable[[Step], None] | None = None, detector_options: dict[str, Any] | None = None, skills: SkillLibrary | None = None, gate: Gate | None = None) -> AgentLoop: """Wire a real desktop into an :class:`AgentLoop` in one call. Defaults are the safe ones: dry run, a step budget of 10, and a kill-switch file on disk that aborts the next action if it appears. ``screen``/``detector``/ ``brain`` can be injected, which is how the tests drive this without a desktop. Pass ``skills`` and the brain becomes a :class:`~tiny_computer_agent.skills.SkillBrain` that answers from the library first and only falls back to the model when no learned skill applies; pass ``gate`` and the operator stands in front of every action. """ if detector is None or isinstance(detector, str): detector = make_detector(detector or "ocr", **(detector_options or {})) if screen is None: screen = IndexedScreen(GrabScreen(), detector) if brain is None or isinstance(brain, str): name = brain or "planner" if name == "planner": brain = PlannerBrain() else: from tiny_computer_agent.models import make_model_brain brain = make_model_brain(name) if skills is not None: brain = SkillBrain(skills, fallback=brain) executor: Executor = DryRunExecutor() if dry_run else PyAutoGuiExecutor() policy = Policy(dry_run=dry_run, step_budget=max_steps, kill_switch_path=kill_switch_path, allow_grid_clicks=allow_grid_clicks) return AgentLoop(screen=screen, brain=brain, executor=executor, policy=policy, on_step=on_step, gate=gate)