Publish showcase_catalog.py from marimo

showcase_catalog.py

@@ -14,11 +14,14 @@ from showcase_core import (
 )
 import os
 import subprocess
+import time
+from pathlib import Path
 
 
-def term(sid, category, title, shows, note="", argv=None, code=None, env=None):
+def term(sid, category, title, shows, note="", argv=None, code=None, env=None,
+         prepare=None):
     CATALOG.append(Shot(id=sid, category=category, title=title, shows=shows, note=note,
-                        kind="terminal", argv=argv, code=code, env=env))
+                        kind="terminal", argv=argv, code=code, env=env, prepare=prepare))
 
 
 def desk(sid, category, title, shows, note="", scene_name=None, make=None):
@@ -1491,3 +1494,229 @@ _edit("exec-refuses-unresolved",
       'print("   ", InputUnavailable("real input needs pyautogui and a live display: '
       'pip install \'tiny-computer-agent[desktop]\'"))\nprint()\n'
       'resolved = executor.execute(Action("done", reason="nothing to do"), frame)')
+
+# =========================================================================== #
+# 15. the stepwise loop, the masked decoder, and the accessibility tree
+# =========================================================================== #
+
+term("stepwise", "The loop",
+     "The same loop, one turn at a time, with pause and resume",
+     ["AgentLoop", "AgentLoop.begin", "AgentLoop.step_once", "AgentLoop.pause",
+      "AgentLoop.resume", "AgentLoop.finished", "AgentLoop.run", "Step.line"],
+     note="run() is batch; a console needs the caller to own the loop. These are the same rules.",
+     code="""
+from tiny_computer_agent import AgentLoop, Element, FakeScreen, MockBrain, Policy
+
+screen = FakeScreen([
+    (Element("Files", x=10, y=6, width=60, height=20),),
+    (Element("Save", x=320, y=120, width=80, height=24),),
+])
+loop = AgentLoop(
+    screen=screen,
+    brain=MockBrain([{"kind": "wait", "seconds": 0.1},
+                     {"kind": "click", "label": "Save"},
+                     {"kind": "done", "reason": "saved"}]),
+    policy=Policy(step_budget=6),
+)
+loop.begin("save the report")
+print("after begin():      finished =", loop.finished, "| steps =", len(loop.episode.steps))
+
+print("step_once() ->     ", loop.step_once().line())
+print("                    finished =", loop.finished, "| steps =", len(loop.episode.steps))
+
+loop.pause()
+loop.run()                                  # returns at once: the caller said pause
+print("after pause():      paused =", loop.paused, "| steps =", len(loop.episode.steps))
+
+loop.resume()
+episode = loop.run()
+print("after resume():     finished =", loop.finished, "| success =", episode.success)
+print()
+print(episode.render())
+""")
+
+term("decoding", "Models",
+     "Constrained decoding: make an echo impossible, not merely detected",
+     ["MenuMask", "digit_token_ids", "choice_token_ids", "pair_token_ids", "read_choice",
+      "read_verdict", "brevity_ladder", "majority_choice", "DIGITS"],
+     note="The 256M model's failure was answering with the prompt. A mask removes that option.",
+     code="""
+from tiny_computer_agent.decoding import (DIGITS, MenuMask, brevity_ladder,
+                                          choice_token_ids, digit_token_ids,
+                                          majority_choice, pair_token_ids, read_choice,
+                                          read_verdict)
+from tiny_computer_agent.models import DEFAULT_VISION_MODEL
+
+try:
+    from transformers import AutoTokenizer
+    tokenizer = AutoTokenizer.from_pretrained(DEFAULT_VISION_MODEL)
+    source = DEFAULT_VISION_MODEL
+except Exception as exc:                                   # pragma: no cover - env
+    class _StandIn:
+        table = {str(d): 100 + d for d in range(10)}
+        def encode(self, text, add_special_tokens=False):
+            return [self.table[text]] if text in self.table else [7, 7]
+    tokenizer, source = _StandIn(), f"a stand-in ({type(exc).__name__})"
+
+print("tokenizer:", source)
+print("digits we allow      :", DIGITS)
+print("digit token ids      :", digit_token_ids(tokenizer)[:10], "...")
+print("ids for a 4-item menu:", choice_token_ids(tokenizer, 4))
+yes_ids, no_ids = pair_token_ids(tokenizer)
+print("yes / no token ids   :", yes_ids[:3], "/", no_ids[:3])
+mask = MenuMask(choice_token_ids(tokenizer, 4), eos_token_id=2)
+print("the mask allows      :", mask.allowed_ids, "(everything else is -inf)")
+print()
+print("reading an answer, and refusing anything that is not one:")
+for reply in ["3", " 3 ", "1000", "The goal is to click Save", "3 or 4", ""]:
+    print(f"   read_choice({reply[:26]!r:30}) -> {read_choice(reply, 4)}")
+for reply in ["Yes", " no", "maybe", ""]:
+    print(f"   read_verdict({reply!r:9}) -> {read_verdict(reply)!r}")
+print()
+print("the retry ladder (echoing grows with prompt length):", brevity_ladder(3))
+print("a vote over samples:", majority_choice([2, 3, 2]), "| ties go to the first seen:",
+      majority_choice([3, 2]))
+""")
+
+# --- the accessibility half needs a real GTK window and a session bus --------
+_GTK = {"process": None}
+GTK_BOX = (140, 140, 660, 420)
+
+
+def open_gtk_scene():
+    """A session bus plus the GTK window, so AT-SPI has something to read."""
+    if _GTK["process"] is not None:
+        return
+    address = subprocess.run(["dbus-daemon", "--session", "--fork", "--print-address=1"],
+                             capture_output=True, text=True).stdout.strip()
+    os.environ["DBUS_SESSION_BUS_ADDRESS"] = address
+    os.environ["NO_AT_BRIDGE"] = "0"
+    ready = Path("/tmp/gallery-gtk.ready")
+    ready.unlink(missing_ok=True)
+    _GTK["process"] = subprocess.Popen(
+        [PY, str(ROOT / "showcase_scenes" / "gtk_dialog.py"),
+         "/tmp/gallery-gtk.marker", str(ready)],
+        env=dict(os.environ), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    for _ in range(60):
+        if ready.exists():
+            break
+        time.sleep(0.25)
+    time.sleep(2.5)
+
+
+def close_gtk_scene():
+    process, _GTK["process"] = _GTK["process"], None
+    if process is not None:
+        process.terminate()
+        try:
+            process.wait(timeout=10)
+        except subprocess.TimeoutExpired:                  # pragma: no cover
+            process.kill()
+    time.sleep(0.5)
+
+
+def _with_gtk(tmp=None):
+    open_gtk_scene()
+    return close_gtk_scene
+
+
+term("atspi-index", "Perception",
+     "The accessibility tree: elements the OS already knows about",
+     ["AtspiDetector", "AtspiDetector.detect", "AtspiDetector.text", "AtspiReader",
+      "AtspiReader.read", "AccessibleNode", "TreeReader", "TreeReader.read",
+      "accessibility_available",
+      "INTERACTIVE_ROLES", "TEXT_ROLES"],
+     note="No pixels are read: GTK publishes every button's name and screen extents.",
+     prepare=_with_gtk,
+     code="""
+from tiny_computer_agent import (INTERACTIVE_ROLES, TEXT_ROLES, AtspiDetector, AtspiReader,
+                                 accessibility_available)
+
+available, detail = accessibility_available("atspi")
+print("accessibility_available('atspi') ->", available)
+print("   ", detail)
+print()
+print("interactive roles offered to a model:")
+print("   ", ", ".join(sorted(INTERACTIVE_ROLES)))
+print("text roles - readable, but not clickable:")
+print("   ", ", ".join(sorted(TEXT_ROLES)))
+print()
+
+reader = AtspiReader()
+nodes = reader.read()
+print(f"AtspiReader().read() -> {len(nodes)} AccessibleNode(s) on the live desktop")
+print()
+
+detector = AtspiDetector(reader=reader)
+elements = detector.detect()
+print(f"AtspiDetector().detect() -> {len(elements)} clickable elements, with no OCR at all:")
+for element in elements:
+    print(f"   {element.label!r:12} kind={element.kind:8} "
+          f"centre={str(element.center):12} box={element.width}x{element.height}")
+print()
+print("the words on screen, straight from the tree:")
+print("   ", detector.text()[:150])
+""")
+
+term("models-decision-modes", "Models",
+     "Four ways to ask a tiny model, and what each one costs",
+     ["ModelBrain", "ModelBrain.loaded", "TextPlannerBrain", "VLMBrain", "make_model_brain"],
+     note="json=free text, choice=one masked number, scan=yes/no per item, rank=log-odds order.",
+     code="""
+from tiny_computer_agent import ModelBrain, make_model_brain
+from tiny_computer_agent.models import DEFAULT_TEXT_MODEL, DEFAULT_VISION_MODEL
+
+print("decision_mode decides how the model is asked:")
+print("   json   - reply with JSON, parsed leniently (the original behaviour)")
+print("   choice - one generation, masked to a menu number")
+print("   scan   - one masked yes/no question per item, first yes wins")
+print("   rank   - the same questions, ranked by the yes/no log-odds margin")
+print()
+for name in ("text-planner", "vlm"):
+    brain = make_model_brain(name, decision_mode="scan")
+    print(f"{name:13} -> {type(brain).__name__}")
+    print(f"   model            : {brain.model_id}")
+    print(f"   supports masking : {brain.supports_constrained}")
+    print(f"   decision_mode    : {brain.decision_mode} | retries {brain.retries} | votes {brain.votes}")
+    print(f"   weights loaded   : {brain.loaded}")
+    brain.unload()
+    print()
+print("defaults are unchanged:", make_model_brain("vlm").decision_mode)
+""")
+
+# --- the accessibility tree, on a real desktop, with a real click ------------
+def make_atspi_click(tmp):
+    from tiny_computer_agent import (AgentLoop, AtspiDetector, GrabScreen, IndexedScreen,
+                                     PlannerBrain, Policy, PyAutoGuiExecutor)
+    open_gtk_scene()
+    try:
+        elements = AtspiDetector().detect()
+        before = crop(annotate(grab(), elements), GTK_BOX)
+        episode = AgentLoop(
+            screen=IndexedScreen(GrabScreen(monitor=1), AtspiDetector()),
+            brain=PlannerBrain(), executor=PyAutoGuiExecutor(),
+            policy=Policy(dry_run=False, step_budget=3),
+        ).run("click 'Save' to save the document")
+        target = next((e for e in elements if e.matches("Save")), None)
+        after = crop(annotate(grab(), [], click=target.center if target else None), GTK_BOX)
+        marker = Path("/tmp/gallery-gtk.marker")
+        lines = [
+            "the window's own status label now reads 'last action: Save',",
+            "and it wrote " + (repr(marker.read_text()) if marker.exists() else "nothing")
+            + " to " + marker.name,
+            "result: " + ("success" if episode.success else "failure")
+            + " - " + episode.stop_reason,
+        ]
+        return overlay_lines(side_by_side(
+            before, after,
+            title="AT-SPI index -> a real click: not one pixel was read to find the button"), lines)
+    finally:
+        close_gtk_scene()
+
+
+desk("desk-atspi", "Real desktop",
+     "Indexed by the accessibility tree, acted on with the real mouse",
+     ["AtspiDetector", "AtspiDetector.detect", "AtspiReader", "IndexedScreen",
+      "PyAutoGuiExecutor", "PlannerBrain"],
+     note="Roles and extents come from GTK itself; the click is a real one and the app reacts.",
+     scene_name="gtk", make=make_atspi_click)