Altium - the Altium Bridge

#!/usr/bin/env python3
"""The altium-bridge verb battery: every altium_* verb in the LIVE catalog, against a real Altium.

    python3 run_verb_tests.py --target <machine> [--thread "<ai thread>"] [--only a,b] [--keep-open]
                              [--fixtures <dir with Adom_Test.SchLib/.PcbLib>] [--json out.json]

Coverage is ENFORCED: a verb in `altium_describe.allVerbs` with no entry in MATRIX turns the run red,
so a new verb cannot ship untested. Modelled on adom/kicad-bridge's run_verb_tests.py.

What a PASS is
  ok       the verb answered success:true (plus the entry's `check`, when it has one)
  refuse   the verb answered success:false with a NAMED reason (errorCode or data.error / error text).
           For verbs whose real inputs this battery does not have (an OutJob, an IntLib, an APS token)
           a clean, fast refusal IS the contract under test.
  skip     never run, with the reason printed. Only for verbs that would harm the machine or the
           user's account (sign-out, trial activation, uninstall, foreground steal, extension install).
What a FAIL is
  a timeout or transport error, success:false with NO reason, the wrong outcome for the entry,
  a dialog left on the user's screen after the verb, or the agent dead when it should be alive.

SAFETY
  - Works only on SCRATCH copies (Battery.SchLib / Battery.PcbLib / battery.SchDoc / battery.PcbDoc)
    in <Documents>/Adom/altium-bridge-battery on the target; never the user's documents.
  - Closes Altium when done unless --keep-open: an Altium 365 workspace has a fixed number of Author
    seats, and an Altium left running on a test box starves every other machine (view-only mode).
"""
import argparse
import json
import os
import subprocess
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))


def ab(target, thread, verb, args, timeout=240):
    args = dict(args)
    args.setdefault("reason", "altium-bridge verb battery")
    cmd = ["adom-bridge", "--ai-thread", thread, "--target", target, verb, json.dumps(args)]
    t0 = time.time()
    try:
        p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    except subprocess.TimeoutExpired:
        return {"success": False, "_transport": "battery_timeout", "_sec": time.time() - t0}
    out = p.stdout.strip()
    try:
        d, _ = json.JSONDecoder().raw_decode(out[out.index("{"):])
    except Exception:
        return {"success": False, "_transport": "unparseable", "_raw": (out or p.stderr)[:300], "_sec": time.time() - t0}
    d["_sec"] = round(time.time() - t0, 2)
    return d


IDLE_PS = r"""
Add-Type @'
using System; using System.Runtime.InteropServices;
public static class AdomIdle { [StructLayout(LayoutKind.Sequential)] struct LII { public uint cbSize; public uint dwTime; }
  [DllImport("user32.dll")] static extern bool GetLastInputInfo(ref LII p);
  public static uint Seconds() { LII l = new LII(); l.cbSize = (uint)Marshal.SizeOf(l); GetLastInputInfo(ref l); return ((uint)Environment.TickCount - l.dwTime) / 1000; } }
'@
[AdomIdle]::Seconds()
"""


def idle_seconds(target, thread):
    """Seconds since the last keyboard/mouse input on the target (None when it cannot be read)."""
    import base64
    enc = base64.b64encode(IDLE_PS.encode("utf-16-le")).decode()
    r = ab(target, thread, "shell_execute", {"command": f"powershell -NoProfile -EncodedCommand {enc}", "timeoutSeconds": 60,
                                             "reason": "verb battery: is anyone using this machine right now?"}, timeout=90)
    try:
        return int(str(r.get("stdout", "")).strip().splitlines()[-1])
    except Exception:
        return None


def reason_of(d):
    data = d.get("data") if isinstance(d.get("data"), dict) else {}
    return d.get("errorCode") or data.get("errorCode") or d.get("error") or data.get("error") or ""


def matrix(w):
    """verb -> dict(phase, args, expect='ok'|'refuse'|'any'|'skip', check=callable?, note, timeout)."""
    D, SL, PL, SD, PD = w["dir"], w["schlib"], w["pcblib"], w["schdoc"], w["pcbdoc"]
    ok = lambda **a: dict(expect="ok", args=a)
    ref = lambda note, **a: dict(expect="refuse", args=a, note=note)
    anyo = lambda note, **a: dict(expect="any", args=a, note=note)
    skip = lambda note: dict(expect="skip", args={}, note=note)
    M = {}

    # ---- phase 0: answers without the script engine ------------------------------------------
    for v in ["describe", "status", "readiness", "diagnostics", "instances", "agent_status", "progress",
              "check_dialogs", "window_info", "license_status", "signin", "signin_status", "install_receipt", "a365_status",
              "a365_oauth_status"]:
        M[v] = dict(phase=0, **ok())
    M["dialog_log"] = dict(phase=0, **ok(limit=5))
    M["install"] = dict(phase=0, **ok(stage="preflight"), note="read-only stage: disk, OS, elevation, UAC policy. Never launches an installer")
    M["list_commands"] = dict(phase=0, **ok(query="PCB", limit=5))
    M["uninstall"] = dict(phase=0, **ok(dryRun=True), note="SAFETY: dryRun only, never removes anything")
    M["extension_ping"] = dict(phase=0, **anyo("the Altium extension is optional; ok or a named 'not installed' both pass"))
    # Altium 365 API lane: no token on a test box, so the contract is a clean refusal
    for v, a in [("a365_refresh", {}), ("a365_query", {"query": "{ __typename }"}), ("a365_library_overview", {}),
                 ("a365_library_schema", {}), ("a365_files_upload", {"filePath": PL})]:
        M[v] = dict(phase=0, **anyo("no API token on a test box: a named not-configured refusal is the pass", **a))
    M["a365_oauth_cancel"] = dict(phase=0, **anyo("nothing pending: ok or a named refusal"))

    # ---- phase 1: lifecycle ------------------------------------------------------------------
    M["launch"] = dict(phase=1, timeout=300, **ok(waitSec=150))
    M["agent_start"] = dict(phase=1, **ok())
    M["ensure_licensed"] = dict(phase=1, timeout=300, **ok(), note="defaults only: never signs out, never starts a trial")
    M["state"] = dict(phase=1, **ok(screenshots=False))
    M["eval"] = dict(phase=1, **ok(text="Begin End;"))
    # NEVER a process that can open a dialog: PCB:Zoom raised Altium's "Zoom Value" box on the user's
    # screen (2026-09-20) and the bridge's dialog scan did not see it. No command given = a named refusal.
    M["run_command"] = dict(phase=1, **ref("no command given: must refuse by name (a real process can open a dialog)"))

    # ---- phase 2: library lane on SCRATCH copies ---------------------------------------------
    M["install_library"] = dict(phase=2, **ok(libraryPath=SL))
    M["open_document"] = dict(phase=2, **ok(filePath=SL))
    M["activate_document"] = dict(phase=2, **ok(path=SL))
    M["list_library"] = dict(phase=2, **ok(libraryPath=SL, limit=10))
    M["show_library"] = dict(phase=2, **ok(libraryPath=SL))
    M["read_symbol"] = dict(phase=2, **ok(libraryPath=SL, symbolName="TST_QFN24"))
    M["symbol_prims"] = dict(phase=2, **ok(libraryPath=SL, symbolName="TST_BGA36", models=True),
                             check=lambda d: (d.get("data") or {}).get("pins") == 36)
    M["show_symbol"] = dict(phase=2, **ok(libraryPath=SL, symbolName="TST_SOIC8"))
    M["symbol_mode"] = dict(phase=2, **anyo("single-mode test symbols: ok or a named refusal", libraryPath=SL, mode=0))
    M["set_params"] = dict(phase=2, **ok(libraryPath=SL, symbolName="TST_SOIC8", params={"BatteryParam": "1"}))
    M["set_visibility"] = dict(phase=2, **anyo("param visibility", libraryPath=SL, symbolName="TST_SOIC8", param="BatteryParam", visible=False))
    M["set_models"] = dict(phase=2, **ok(libraryPath=SL, symbolName="TST_QFN24", footprint="TST_QFN24"))
    M["add_models"] = dict(phase=2, **anyo("adds a footprint model row", libraryPath=SL, symbolName="TST_SOIC8", models=[{"type": "PCBLIB", "name": "TST_SOIC8"}]))
    M["create_symbol"] = dict(phase=2, **ok(libraryPath=SL, name="BAT_SYM", designator="U?", description="battery",
                                          pins=[{"number": "1", "name": "A", "side": "left"}, {"number": "2", "name": "B", "side": "right"}]))
    M["clone_symbol"] = dict(phase=2, **ok(libraryPath=SL, **{"from": "BAT_SYM", "to": "BAT_SYM_CLONE"}))
    M["rename_symbol"] = dict(phase=2, **ok(libraryPath=SL, **{"from": "BAT_SYM_CLONE", "to": "BAT_SYM_RENAMED"}))
    M["remove_symbol"] = dict(phase=2, **ok(libraryPath=SL, symbolName="BAT_SYM_RENAMED"))
    M["build_symbol_modes"] = dict(phase=2, **ref("no spec given: must refuse by name, not halt", libraryPath=SL))
    M["add_symbol_art"] = dict(phase=2, **ref("no kicadSym given: must refuse by name", libraryPath=SL, symbolName="TST_SOIC8"))

    M["show_footprint"] = dict(phase=2, **ok(libraryPath=PL, footprintName="TST_QFN24"))
    M["show_3d_chip"] = dict(phase=2, **anyo("test footprints carry no 3D body: ok or named refusal", libraryPath=PL, footprintName="TST_QFN24"))
    M["footprint_primitives"] = dict(phase=2, **anyo("reads a footprint's primitives", libraryPath=PL, footprintName="TST_QFN24"))
    M["plate_vias"] = dict(phase=2, **ok(libraryPath=PL), check=lambda d: (d.get("data") or {}).get("holes") == 9)
    M["normalize_bodies"] = dict(phase=2, **anyo("no bodies on the test footprints", libraryPath=PL))
    M["create_footprint"] = dict(phase=2, **ok(libraryPath=PL, name="BAT_FP", description="battery",
                                             pads="1,-40,0,24,24,rect,0;2,40,0,24,24,rect,0"))
    M["copy_footprint"] = dict(phase=2, **ok(libraryPath=PL, **{"from": "BAT_FP", "to": "BAT_FP_COPY"}))
    M["rename_footprint"] = dict(phase=2, **ok(libraryPath=PL, **{"from": "BAT_FP_COPY", "to": "BAT_FP_RENAMED"}))
    M["remove_footprint"] = dict(phase=2, **ok(libraryPath=PL, footprintName="BAT_FP_RENAMED"))
    M["xcopy_footprint"] = dict(phase=2, **anyo("cross-library copy onto itself: ok or named refusal", fromLib=PL, toLib=PL, **{"from": "BAT_FP", "to": "BAT_FP_X"}))
    M["add_body"] = dict(phase=2, **anyo("extruded body on the scratch footprint", libraryPath=PL, footprintName="BAT_FP", hMil=40))
    M["build_footprint"] = dict(phase=2, **ref("no spec given: must refuse by name", libraryPath=PL))
    M["cleanup_libs"] = dict(phase=2, **anyo("library housekeeping", libraryPath=SL))
    M["save"] = dict(phase=2, **ok(filePath=SL))
    M["list_installed_libraries"] = dict(phase=2, **ok())
    M["uninstall_library"] = dict(phase=2, **anyo("SAFETY: dryRun only, removes nothing; a named not-installed refusal also passes",
                                                  libraryName="Adom_Test", dryRun=True))

    # ---- phase 3: schematic + board built from the scratch libraries --------------------------
    M["place_on_sch"] = dict(phase=3, **ok(libraryPath=SL, symbolName="TST_SOIC8;TST_QFN24;TST_BGA36", schPath=SD, title="battery",
                                         number="B-1", revision="A", drawnBy="battery", description="battery sheet"),
                             check=lambda d: (d.get("data") or {}).get("count") == 3)
    M["test_board"] = dict(phase=3, **ok(libraryPath=PL, footprintName="TST_SOIC8;TST_QFN24;TST_BGA36", wMm="50", hMm="25", pcbPath=PD),
                           check=lambda d: (d.get("data") or {}).get("count") == 3)
    M["place_on_pcb"] = dict(phase=3, **anyo("place one more footprint on the battery board", libraryPath=PL, footprintName="TST_SOIC8", pcbPath=PD))
    M["swap_footprint"] = dict(phase=3, **anyo("swap on the battery board", libraryPath=PL, pcbPath=PD, footprintName="TST_SOIC8", assigns="U1=TST_SOIC8"))
    M["board_shape"] = dict(phase=3, **ok(pcbPath=PD, wMm=50, hMm=25))
    for v in ["read_board", "board_summary", "list_designators", "list_nets", "layers", "layers_in_use", "copper_layers",
              "read_rules", "rules_by_kind"]:
        M[v] = dict(phase=3, **ok(filePath=PD, pcbPath=PD, path=PD))
    M["component_data"] = dict(phase=3, **anyo("U1 on the battery board", filePath=PD, designator="U1"))
    M["component_pins"] = dict(phase=3, **anyo("U1 pins", filePath=PD, designator="U1"))
    M["net_connections"] = dict(phase=3, **anyo("the battery board has no nets: named refusal is a pass", filePath=PD, net="GND"))
    M["set_layer_visibility"] = dict(phase=3, **anyo("toggle Top Overlay", filePath=PD, layer="Top Overlay", visible=True))
    M["pcb_report"] = dict(phase=3, **anyo("board report", path=PD, want="summary"))
    M["show_schematic"] = dict(phase=3, **ok(filePath=SD))
    M["read_schematic"] = dict(phase=3, **ok(filePath=SD, path=SD))
    M["schematic_data"] = dict(phase=3, **anyo("schematic data", filePath=SD))
    M["cross_check"] = dict(phase=3, **anyo("sch vs pcb", board=PD, sheet=SD))
    M["review_board"] = dict(phase=3, timeout=300, **anyo("design review", board=PD, sheet=SD, limit=5))
    M["compile"] = dict(phase=3, **anyo("free documents have no project to compile: named refusal is a pass", filePath=SD))
    M["run_drc"] = dict(phase=3, timeout=300, **anyo("DRC on the battery board", filePath=PD, limit=5))
    M["show_2d_board"] = dict(phase=3, **ok(filePath=PD))
    M["show_3d_board"] = dict(phase=3, timeout=300, **ok(filePath=PD))
    M["animate_view"] = dict(phase=3, **anyo("short orbit", preset="orbit", steps=3, pauseMs=50))
    M["show_project"] = dict(phase=3, **anyo("no path = Altium's bundled demo project, by design ('show me altium')"))
    M["open_project"] = dict(phase=3, **ref("no project given: must refuse by name"))
    M["run_erc"] = dict(phase=3, **anyo("ERC needs a compiled project: named refusal is a pass", filePath=SD))

    # ---- phase 3.5: the AI Flow board lane on its OWN scratch board ----------------------------
    # build_board makes lane.PcbDoc from the fixture footprints (two SOIC8s, N1 = U1.8-U2.1 straight
    # across, GND = U1.4/U2.5), then every live-board verb edits it in turn: read, rules, move, route,
    # via, select/zoom, remove, rip up, pour (repoured outside the script engine: the agent stops and
    # restarts for a few seconds), outline, silkscreen, DRC. build_schematic then draws lane.SchDoc in
    # its own project with lane.PcbDoc, and export_manufacturing runs that project's OutJob.
    LB, LS, LP = D + "/lane.PcbDoc", D + "/lane.SchDoc", D + "/lane.PrjPcb"
    lane_nets = {"N1": ["U1.8", "U2.1"], "GND": ["U1.4", "U2.5"]}
    lane_parts = [{"ref": "U1", "footprint": "TST_SOIC8", "x": 10, "y": 10, "rotation": 0},
                  {"ref": "U2", "footprint": "TST_SOIC8", "x": 30, "y": 10, "rotation": 0}]
    n_comps = lambda d: len(d.get("components") or (d.get("data") or {}).get("components") or [])
    M["build_board"] = dict(phase=3.5, timeout=300, **ok(pcbPath=LB, libraryPath=PL, outline={"rect": [0, 0, 40, 20]},
                                                        parts=lane_parts, nets=lane_nets),
                            check=lambda d: d.get("placed") == 2 and not d.get("wrongPins"))
    M["routing_state"] = dict(phase=3.5, **ok(expectPath=LB), check=lambda d: n_comps(d) == 2)
    M["set_design_rules"] = dict(phase=3.5, timeout=300, **ok(clearance=0.2, trackWidth={"min": 0.15, "preferred": 0.25, "max": 2},
                                                             holeToHole=0.25, polygonConnect="relief"))
    M["add_rule"] = dict(phase=3.5, **ok(kind="clearance", name="Battery", scope1="InNet('N1')", gap=0.3))
    M["move_parts"] = dict(phase=3.5, **ok(parts=[{"ref": "U2", "x": 29, "y": 10, "rotation": 0}], expectPath=LB))
    M["route_net"] = dict(phase=3.5, **ok(net="N1", width=0.25, paths=[[{"pad": "U1.8"}, {"pad": "U2.1"}]], expectPath=LB))
    M["place_vias"] = dict(phase=3.5, **ok(vias=[{"net": "GND", "x": 20, "y": 16, "size": 0.6, "drill": 0.3}], expectPath=LB))
    M["select_net"] = dict(phase=3.5, **ok(net="N1", zoom=True))
    M["zoom_to"] = dict(phase=3.5, **ok(ref="U1"))
    M["remove_items"] = dict(phase=3.5, **ok(vias=[{"x": 20, "y": 16}], expectPath=LB))
    M["ripup_nets"] = dict(phase=3.5, **ok(nets=["N1"], expectPath=LB))
    M["apply_pours"] = dict(phase=3.5, timeout=300, **ok(pours=[{"net": "GND", "layer": "B.Cu", "points": [[1, 1], [39, 1], [39, 19], [1, 19]],
                                                                "connect": "relief"}]))
    M["pour_state"] = dict(phase=3.5, **ok())
    M["repour"] = dict(phase=3.5, timeout=300, **ok())
    M["remove_pours"] = dict(phase=3.5, **ok(pours="all"))
    M["set_outline"] = dict(phase=3.5, **ok(outline={"rect": [0, 0, 40, 20], "cornerRadius": 1}))
    M["silk_state"] = dict(phase=3.5, **ok())
    M["apply_silkscreen"] = dict(phase=3.5, **ok(texts=[{"text": "BATTERY", "x": 20, "y": 18, "size": 1}]))
    M["routing_drc"] = dict(phase=3.5, timeout=300, **ok())
    M["build_schematic"] = dict(phase=3.5, timeout=600, **ok(schPath=LS, libraryPath=SL, pcbLibPath=PL, pcbPath=LB, projectPath=LP, title="battery lane",
                                                           parts=[{"ref": "U1", "symbol": "TST_SOIC8", "block": "core", "footprint": "TST_SOIC8"},
                                                                  {"ref": "U2", "symbol": "TST_SOIC8", "block": "core", "footprint": "TST_SOIC8"}],
                                                           nets=lane_nets))
    M["export_manufacturing"] = dict(phase=3.5, timeout=400, **ok(projectPath=LP, pcbPath=LB))

    # ---- phase 4: outputs --------------------------------------------------------------------
    M["export_step"] = dict(phase=4, timeout=300, **anyo("STEP of the battery board", filePath=PD, output=D + "/battery.step"))
    M["export_bom"] = dict(phase=4, **anyo("missing OutJob: named refusal (the demo OutJob opens its .xlsx on screen)", outjobPath=D + "/missing.OutJob"))
    M["export_gerber"] = dict(phase=4, **anyo("missing OutJob: named refusal", outjobPath=D + "/missing.OutJob"))
    # NEVER the demo defaults here: the bundled OutJobs OPEN what they generate (an .xlsx app chooser
    # landed on the screen, 2026-09-20). A path that does not exist must be refused by name.
    M["run_outjob"] = dict(phase=4, **anyo("missing OutJob: named refusal", outjobPath=D + "/missing.OutJob"))
    M["compile_intlib"] = dict(phase=4, **ref("no LibPkg in the fixtures: must refuse by name"))
    M["extract_intlib"] = dict(phase=4, **ref("no IntLib in the fixtures: must refuse by name"))
    M["import_kicad"] = dict(phase=4, **ref("no KiCad file in the fixtures: must refuse by name"))
    M["import_part"] = dict(phase=4, **ref("no spec given: must refuse by name"))
    M["install_library_bundle"] = dict(phase=4, **ref("no bundle given: must refuse by name"))
    M["close_document"] = dict(phase=4, **anyo("close the battery sheet", filePath=SD, path=SD))

    # ---- phase 5: the script lanes and the safety net -----------------------------------------
    M["run_script"] = dict(phase=5, **ref("refused while the agent is resident (agent_resident)", proc="Nope"))
    M["dismiss_dialogs"] = dict(phase=5, **anyo("nothing to dismiss"))
    # The slowest verb by design, and the slowest on ADDevelop, which parks SILENTLY so the halt is
    # only seen when the wait expires (18 s to over 60 s on one machine). Keep this ABOVE the verb's
    # own declared budget (180 s), or the battery blames the bridge for its own cut-off.
    M["selftest_script_error"] = dict(phase=5, timeout=220, **ok(), check=lambda d: d.get("netWorks") is True)
    M["agent_recover"] = dict(phase=5, **ok())
    M["agent_stop"] = dict(phase=6, **ok())
    M["close"] = dict(phase=6, timeout=300, **ok(discardChanges=True, waitSec=90), note="releases the licence seat; every modified document is the battery's own scratch")

    # ---- never run ----------------------------------------------------------------------------
    # The skip list used to hold all nine of these. It was written when the only Altium lived on a
    # machine someone was working on; on a dedicated test box each one below either reads state or
    # refuses by name BEFORE it touches anything, and a refusal is exactly what the battery is for.
    # (John, 2026-09-21: "none of those verbs seem unsafe to me?" - he was right about eight of them.)

    # Colby's 0.2.40 verb (#12's ganged-up view): activates a docked panel tab in the background.
    # "any" because ui_select_gated is an honest outcome when ab's shell grant is off on the box.
    M["panel"] = dict(phase=2, **anyo("activates the SCH Library dock tab; ui_select_gated is a named refusal when ab's shell grant is off", name="SCH Library"))

    # A background beat with a screenshot. Only demo {live:true} takes the foreground.
    M["demo"] = dict(phase=5, timeout=200, **anyo("stage 0, background beat (live:true is the foreground show and is NOT run here)", stage=0))
    # The sanctioned foreground exception. Raises Altium on a box the runner already proved idle,
    # so keep it LAST in phase 5, right before the agent teardown.
    M["foreground"] = dict(phase=5, **ok(reason="altium-bridge verb battery: proving the foreground verb itself"))
    # Altium is CLOSED in phase 0, so this is the pure "next action" report - it cannot start a
    # trial from here. The stage that would act needs a signed-in, UNLICENSED Altium running.
    M["setup_trial"] = dict(phase=0, **anyo("Altium is closed: reports the next action, activates nothing"))
    # Same reason: with Altium closed this is a clean altium_not_running refusal, which covers the
    # contract without opening the workspace picker on someone's account.
    M["connect_workspace"] = dict(phase=0, **ref("Altium is closed: must refuse by name, never open the workspace picker"))
    # The seat-moving actions (use_license / start_trial / sign_out) are deliberately NOT exercised;
    # an unknown action proves the dispatch and the named refusal without touching the one seat.
    M["license"] = dict(phase=1, **ref("unknown action: must refuse by name, never claim or release the seat", action="battery_no_such_action"))
    # Altium is running by phase 2, and the verb refuses in that state (it holds AdomBridge.dll open).
    M["install_extension"] = dict(phase=2, **ref("Altium is running: must refuse with altium_running, installing nothing"))
    # Both of these validate BEFORE they write or open anything: no workspace = a named refusal.
    M["a365_configure"] = dict(phase=0, **ref("no workspace given: must refuse before writing the config"))
    M["a365_oauth_begin"] = dict(phase=0, **ref("no workspace configured: must refuse before opening a browser"))
    # THE ONE REAL SKIP. Nothing is configured on a test box, so this would be a no-op here - but the
    # battery must be safe to run on ANY machine, and on one with a live Altium 365 session it drops
    # the stored token and signs the user out. No read-only mode, no way to prove it without that.
    M["a365_logout"] = skip("SAFETY: signs the user out of Altium 365 - the one verb here with a real, unasked-for side effect on a machine that IS configured")
    return M


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--target", required=True)
    ap.add_argument("--thread", default="altium-bridge battery")
    ap.add_argument("--only", default="")
    ap.add_argument("--keep-open", action="store_true")
    ap.add_argument("--fixtures", default=os.path.join(HERE, "fixtures"))
    ap.add_argument("--json", default="")
    ap.add_argument("--min-idle", type=int, default=180, help="refuse to run unless the target has had no keyboard/mouse input for this many seconds")
    ap.add_argument("--user-is-away", action="store_true", help="the human told you this machine is free: skip the idle check")
    a = ap.parse_args()
    T, TH = a.target, a.thread
    # The battery launches Altium, opens documents and shows views: it WILL pull Altium in front of
    # whoever is at the keyboard (it did, 2026-09-20, while the user was typing). Never run it on a
    # machine somebody is using.
    if not a.user_is_away:
        idle = idle_seconds(T, TH)
        if idle is None or idle < a.min_idle:
            sys.exit(f"REFUSED: {T} had keyboard/mouse input {idle} s ago (need {a.min_idle} s idle). Someone is using it; "
                     "the battery would raise Altium over their work. Ask the user, then pass --user-is-away.")
        print(f"{T} idle for {idle} s: ok to run")
    # The battery starts from a CLOSED Altium: phase 0 tests the closed state, and an Altium left open
    # holds the previous run's scratch project in memory, which turned build_board and build_schematic
    # into save and reload prompts mid-run (2026-10-16). Closing it is the user's call: their documents.
    if not a.only:
        inst = ab(T, TH, "altium_instances", {})
        if (inst.get("count") or len(inst.get("instances") or [])) > 0:
            sys.exit(f"REFUSED: Altium is already running on {T}. The battery starts from a closed Altium; save anything "
                     "that matters and close it (altium_close), then run again.")
    home = "C:/Users/john"
    r = ab(T, TH, "shell_execute", {"command": "cmd /c echo %USERPROFILE%"})
    if r.get("stdout"):
        home = r["stdout"].strip().replace("\\", "/")
    D = home + "/Documents/Adom/altium-bridge-battery"
    w = dict(dir=D, schlib=D + "/Battery.SchLib", pcblib=D + "/Battery.PcbLib", schdoc=D + "/battery.SchDoc", pcbdoc=D + "/battery.PcbDoc")

    # fixtures -> scratch copies on the target
    fx = [os.path.join(a.fixtures, n) for n in ("Adom_Test.SchLib", "Adom_Test.PcbLib")]
    for f in fx:
        if not os.path.isfile(f):
            sys.exit(f"missing fixture {f}")
    ab(T, TH, "send_files", {"filePaths": fx}, timeout=300)
    dw = D.replace("/", "\\"); hw = home.replace("/", "\\")
    ab(T, TH, "shell_execute", {"command": f'cmd /c rmdir /S /Q "{dw}" 2>nul & mkdir "{dw}" & move /Y "{hw}\\Downloads\\Adom_Test.SchLib" "{dw}\\Battery.SchLib" & move /Y "{hw}\\Downloads\\Adom_Test.PcbLib" "{dw}\\Battery.PcbLib"'})

    live = ab(T, TH, "altium_describe", {})
    verbs = sorted(v[len("altium_"):] for v in (live.get("allVerbs") or []))
    if not verbs:
        sys.exit(f"altium_describe returned no allVerbs: {str(live)[:300]}")
    M = matrix(w)
    uncovered = [v for v in verbs if v not in M]
    stale = [v for v in M if v not in verbs]
    only = [x for x in a.only.split(",") if x]

    rows = []
    transport_run, first_blocked_after, aborted = 0, "", False
    order = sorted(M.items(), key=lambda kv: (kv[1].get("phase", 9), list(M).index(kv[0])))
    for verb, spec in order:
        if verb not in verbs or (only and verb not in only):
            continue
        if a.keep_open and verb in ("close", "agent_stop"):
            rows.append(dict(verb=verb, outcome="SKIP", note="--keep-open", sec=0)); continue
        if spec["expect"] == "skip":
            rows.append(dict(verb=verb, outcome="SKIP", note=spec.get("note", ""), sec=0)); continue
        d = ab(T, TH, "altium_" + verb, spec["args"], timeout=spec.get("timeout", 200))
        okv, why = d.get("success") is True, reason_of(d)
        if d.get("_transport") or "HTTP error" in str(d.get("error", "")) or d.get("errorCode") in ("timed_out", "agent_timeout"):
            outcome, detail = "FAIL", f"transport/timeout: {d.get('_transport') or d.get('errorCode') or d.get('error')}"
        elif okv:
            chk = spec.get("check")
            good = spec["expect"] in ("ok", "any") and (chk is None or chk(d))
            outcome = "PASS" if good else "FAIL"
            detail = "ok" if good else ("check failed" if spec["expect"] != "refuse" else "expected a refusal, got success")
        elif why:
            outcome = "PASS" if spec["expect"] in ("refuse", "any") else "FAIL"
            detail = ("refused: " if outcome == "PASS" else "unexpected failure: ") + str(why)[:150]
        else:
            outcome, detail = "FAIL", "success:false with NO reason (no errorCode, no error text)"
        # nothing may be left on the user's screen
        if verb not in ("close", "agent_stop", "launch"):
            dl = ab(T, TH, "altium_check_dialogs", {}, timeout=60)
            left = [x.get("title") for x in (dl.get("dialogs") or [])]
            if left:
                outcome, detail = "FAIL", f"left dialog(s) on screen: {left} | was: {detail}"
                ab(T, TH, "altium_agent_recover", {}, timeout=90)
        rows.append(dict(verb=verb, outcome=outcome, detail=detail, sec=d.get("_sec"), note=spec.get("note", "")))
        # Three transport failures in a row = the bridge is blocked, not three bugs. Stop: firing
        # the rest of the matrix into a blocked bridge tests nothing and hammers the user's Altium.
        transport_run = transport_run + 1 if detail.startswith("transport/timeout") else 0
        if transport_run == 1:
            first_blocked_after = rows[-2]["verb"] if len(rows) > 1 else "(start)"
        if transport_run >= 3:
            print(f"\nABORT: 3 transport failures in a row. The bridge stopped answering right after '{first_blocked_after}'. "
                  f"That verb (or the one that failed first, '{rows[-3]['verb']}') is the defect; the rest of the run is void.", flush=True)
            aborted = True
            break
        print(f"{outcome:5s} {verb:28s} {str(d.get('_sec')):>6}s  {detail[:130]}", flush=True)

    n = {k: sum(1 for r in rows if r["outcome"] == k) for k in ("PASS", "FAIL", "SKIP")}
    red = n["FAIL"] or uncovered or aborted
    print("\n==== altium-bridge verb battery on", T, "| bridge", live.get("bridgeVersion"), "====")
    print(f"verbs in catalog {len(verbs)} | PASS {n['PASS']} | FAIL {n['FAIL']} | SKIP {n['SKIP']} (safety) | uncovered {len(uncovered)} | stale {len(stale)}")
    if uncovered: print("UNCOVERED (add to MATRIX):", uncovered)
    if stale: print("stale matrix entries (verb gone):", stale)
    for r in rows:
        if r["outcome"] == "FAIL": print("  FAIL", r["verb"], "-", r["detail"])
    if a.json:
        json.dump(dict(target=T, bridgeVersion=live.get("bridgeVersion"), counts=n, uncovered=uncovered, stale=stale, rows=rows), open(a.json, "w"), indent=1)
    sys.exit(1 if red else 0)


if __name__ == "__main__":
    main()