Blame for tiny_computer_agent/__main__.py

"""Command line entry point.

    python -m tiny_computer_agent                     # scripted demo, dry-run
    python -m tiny_computer_agent --doctor            # what can this machine do?
    python -m tiny_computer_agent --image example     # replay a screenshot, dry-run
    python -m tiny_computer_agent --image shot.png    # replay your own screenshot
    python -m tiny_computer_agent --live              # your real screen, dry-run
    python -m tiny_computer_agent --live --real-input "click 'OK'"

Real mouse and keyboard input additionally requires
``TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT=1``. Dry run is the default, so you can read a
whole episode before anything touches your pointer.
"""

from __future__ import annotations

import argparse
import json
import os
import platform
import sys

from tiny_computer_agent.act import DryRunExecutor, PyAutoGuiExecutor
from tiny_computer_agent.brain import MockBrain, PlannerBrain
from tiny_computer_agent.demo import fake_desktop
from tiny_computer_agent.desktop import (
    BRAIN_NAMES, DEFAULT_KILL_SWITCH, Check, display_available, run_checks,
)
from tiny_computer_agent.loop import AgentLoop, Step
from tiny_computer_agent.observe import (
    Element, FakeScreen, GrabScreen, ImageScreen, ScreenUnavailable,
)
from tiny_computer_agent.perception import IndexedScreen, detector_available, make_detector
from tiny_computer_agent.safety import Policy

ALLOW_REAL_INPUT_ENV = "TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT"
EXAMPLE_IMAGE = "example"


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("--doctor", action="store_true",
                        help="report what this machine can and cannot do, then exit")
    parser.add_argument("--image", default=None,
                        help="replay a screenshot instead of the screen; "
                             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("--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("--monitor", type=int, default=1, help="1 is the primary monitor")
    parser.add_argument("--scale", type=float, default=None,
                        help="screen pixels per captured pixel (HiDPI, secondary monitors)")
    parser.add_argument("--no-ocr", action="store_true",
                        help="skip OCR and index a numbered grid instead")
    parser.add_argument("--max-width", type=int, default=1600,
                        help="downscale captures wider than this before indexing")
    parser.add_argument("--verify-text", default=None, metavar="TEXT",
                        help="stop with success once this text appears on screen "
                             "(otherwise a successful action that removes its own "
                             "target is indistinguishable from a failure)")
    parser.add_argument("--allow-grid", action="store_true",
                        help="permit clicks on numbered grid cells with --real-input")
    parser.add_argument("--kill-switch", default=DEFAULT_KILL_SWITCH,
                        help="create this file to abort the run before the next action")
    return parser


def report(checks: list[Check]) -> int:
    """Print a preflight report; non-zero when something required is missing."""
    blocking = [c for c in checks if c.required and not c.ok]
    lines = [
        f"tiny-computer-agent preflight ({platform.system()} {platform.machine()}, "
        f"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.")
    print("\n".join(lines))
    return 1 if blocking else 0


def _probe_live(args: argparse.Namespace) -> Check:
    """Actually grab one frame, because only a real capture proves it works."""
    try:
        observation = GrabScreen(
            monitor=args.monitor, max_width=args.max_width, scale=args.scale
        ).capture()
        frame = observation.frame
        return Check("live capture", True, f"captured {frame.width}x{frame.height} pixels")
    except ScreenUnavailable as exc:
        return Check("live capture", False, str(exc),
                     fix="run this on the desktop session itself, not a headless shell")


def _build_screen(args: argparse.Namespace):
    """(screen, description). Raises ScreenUnavailable / OSError on failure."""
    detector_name = "grid" if args.no_ocr else args.detector
    if args.image is not None:
        source = ImageScreen(fake_desktop() if args.image == EXAMPLE_IMAGE else args.image)
        return IndexedScreen(source, make_detector(detector_name)), f"screenshot replay: {args.image}"
    if args.live:
        source = GrabScreen(monitor=args.monitor, max_width=args.max_width, scale=args.scale)
        return IndexedScreen(source, make_detector(detector_name)), f"real desktop (monitor {args.monitor})"
    return demo_screen(), "scripted demo"


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    detector_name = "grid" if args.no_ocr else args.detector

    if args.doctor:
        checks = run_checks(brain=args.brain, detector=detector_name,
                            real_input=args.real_input, kill_switch_path=args.kill_switch)
        if args.live:
            checks.append(_probe_live(args))
        return report(checks)

    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

    if args.brain == "vlm" and not (args.live or args.image):
        print("the vlm brain needs --live (or --image) so it has pixels to look at; "
              "use --brain planner to exercise the harness offline.", file=sys.stderr)
        return 2

    if args.live:
        available, detail = display_available()
        if not available:
            print(f"live capture needs a display session: {detail}")
            print(f"refusing to run live: {detail}. Try --image to replay a screenshot, "
                  f"or --doctor to diagnose.", file=sys.stderr)
            return 2

    try:
        screen, description = _build_screen(args)
    except ScreenUnavailable as exc:
        print(f"cannot see the screen: {exc}", file=sys.stderr)
        return 3
    except Exception as exc:
        print(f"could not start the screen: {type(exc).__name__}: {exc}", file=sys.stderr)
        return 2

    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)
    elif args.brain == "planner":
        brain = PlannerBrain()
    else:
        from tiny_computer_agent.models import make_model_brain

        try:
            brain = make_model_brain(args.brain)
        except Exception as exc:
            print(f"could not load the {args.brain} brain: {type(exc).__name__}: {exc}",
                  file=sys.stderr)
            return 2

    policy = Policy(
        dry_run=not args.real_input,
        step_budget=args.max_steps,
        kill_switch_path=args.kill_switch,
        allow_grid_clicks=args.allow_grid,
    )
    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}")
    print(f"mode: {mode}  |  screen: {description}  |  brain: {args.brain}  |  "
          f"kill switch: create {args.kill_switch} to stop")
    if args.image is not None or args.live:
        _, index_detail = detector_available(detector_name)
        print(f"index: {detector_name} - {index_detail}")
    if args.image is not None:
        # One read, so the header can state how big the action space is.
        try:
            preview = screen.capture()
        except ScreenUnavailable as exc:
            print(f"cannot see the screen: {exc}", file=sys.stderr)
            return 3
        print(f"elements detected: {len(preview.elements)}")
    print()

    warned = False

    verify = None
    if args.verify_text:
        _needle = args.verify_text.lower()

        def verify(observation) -> bool:  # noqa: F811 - the loop calls this
            return _needle in observation.text.lower()

    def on_step(step: Step) -> None:
        nonlocal warned
        elements = step.observation.elements
        if not warned and elements and all(e.kind == "grid" for e in elements):
            warned = True
            _, index_detail = detector_available(detector_name)
            print(f"warning: no text was found on screen, so the index is "
                  f"{len(elements)} numbered grid cells and the agent is choosing "
                  f"blind ({index_detail})", file=sys.stderr)
        print("  " + step.line())

    try:
        episode = AgentLoop(screen=screen, brain=brain, executor=executor,
                            policy=policy, on_step=on_step, verify=verify).run(args.goal)
    except ScreenUnavailable as exc:
        print(f"cannot see the screen: {exc}", file=sys.stderr)
        print("run `python -m tiny_computer_agent --doctor --live` for details.", file=sys.stderr)
        return 3

    print()
    print(episode.render())
    return 0 if episode.success else 1


if __name__ == "__main__":
    raise SystemExit(main())