From ba58dcdc5ad6ca021c723d279f33e3f14e1184df Mon Sep 17 00:00:00 2001
From: Arthur Hanson Rasmusson <arthur@vgpu.io>
Date: Fri, 2 Oct 2026 08:36:35 -0400
Subject: debug: add green-eye GSP/uproc RISC-V VM runtime analysis tooling

Adds open-iov/debug/risc-v-gsp-rm-vm: a GDB-free instrumentation stack that boots uproc/GSP firmware in qemu-system-riscv64 and provides Frida-like hooks, breakpoints, watchpoints, memory dumps, structure decoding, snapshot diffs, recorded sessions, declarative probes, a plugin API and a Ghidra symbol/hit bridge.

Verified: 25 tests green (including a live QEMU smoke test); the demo uproc boots with correct hook/leave/watch/schema/snapshot/fault evidence; the optional TCG trace plugin builds and recorded 15M trace lines in 3 s; the TU104 570.144 fwimage probe reports its first exception (load access fault at firmware vaddr 0x4009014 reading 0x2000000000000A00).

Documentation: lean tool docs under debug/risc-v-gsp-rm-vm/docs/00-10, a new docs/05-operations/03-risc-v-gsp-rm-vm.md debugging document, and '# TOOL IN DEVELOPMENT - NOT READY YET!' markers in docs/README.md, the onboarding tooling table and the debugging workflows doc.

Vendors qemu-plugin.h from QEMU v8.2.2 (GPL-2.0-or-later), with its license header intact, so the optional TCG plugin builds without a QEMU source tree.
---
 debug/risc-v-gsp-rm-vm/greeneye/cli.py | 688 +++++++++++++++++++++++++++++++++
 1 file changed, 688 insertions(+)
 create mode 100644 debug/risc-v-gsp-rm-vm/greeneye/cli.py

(limited to 'debug/risc-v-gsp-rm-vm/greeneye/cli.py')

diff --git a/debug/risc-v-gsp-rm-vm/greeneye/cli.py b/debug/risc-v-gsp-rm-vm/greeneye/cli.py
new file mode 100644
index 0000000..5bfcee2
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/cli.py
@@ -0,0 +1,688 @@
+"""green-eye command line interface.
+
+Commands:
+  info       environment/tooling check
+  demo       boot the bundled uproc demo and run a canned instrumentation pass
+  run        run a Python instrumentation script against a target
+  attach     interactive REPL against a QEMU gdbstub
+  probe      run an arbitrary firmware (default: the real GSP fwimage) for N
+             instructions and report the first divergence
+  record     boot a firmware and record a memory-activity session
+  analyze    render a recorded session to a symbol-annotated report
+  compare    A/B two recorded sessions at the symbolic level
+  symbols    build/inspect a green-eye symbol database from Ghidra exports
+  plugins    list discoverable plugins
+"""
+from __future__ import annotations
+
+import argparse
+import cmd
+import os
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+import greeneye as ge
+from . import analyze as _analyze
+from . import ghidra as _ghidra
+from .elf import load_elf
+from .errors import GreenEyeError
+from .plugins import PluginManager
+from .structs import schema_from_json
+from .symbols import SymbolDB
+from .target import QemuTarget
+
+TOOL_ROOT = Path(__file__).resolve().parent.parent
+DUMPS = TOOL_ROOT / "dumps"
+DEMO_ELF = TOOL_ROOT / "firmware" / "uproc_demo" / "build" / "uproc_demo.elf"
+DEMO_SCHEMA = TOOL_ROOT / "firmware" / "uproc_demo" / "gsp_ctrl.schema.json"
+DEFAULT_FWIMAGE = TOOL_ROOT.parents[1] / "firmware" / "57X-GSP" / "570.144" / "gsp-570.144.fwimage.bin"
+
+RPC_HANDLERS = ["rpc_ping", "rpc_echo", "rpc_fill", "rpc_sum", "rpc_state", "rpc_fault"]
+
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+
+def _ensure_dumps_env() -> None:
+    os.environ.setdefault("GREENEYE_DUMPS", str(DUMPS))
+
+
+def _build_demo(quiet: bool = False) -> Path:
+    import subprocess
+
+    fw_dir = TOOL_ROOT / "firmware" / "uproc_demo"
+    result = subprocess.run(
+        ["make", "-C", str(fw_dir)],
+        capture_output=quiet,
+        text=True,
+    )
+    if result.returncode != 0:
+        output = (result.stdout or "") + (result.stderr or "")
+        raise GreenEyeError(f"demo firmware build failed:\n{output}")
+    if not DEMO_ELF.exists():
+        raise GreenEyeError(f"demo build produced no {DEMO_ELF}")
+    return DEMO_ELF
+
+
+def _load_schema(path: Path):
+    import json
+
+    with open(path, "r", encoding="utf-8") as fh:
+        return schema_from_json(json.load(fh))
+
+
+def _start_session(
+    *,
+    kernel: str | None,
+    port: int | None,
+    record,
+    name: str,
+    plugins: list[str] | None = None,
+):
+    if port:
+        db = SymbolDB()
+        if kernel:
+            db.load_elf(kernel)
+        session = ge.attach(port=port, record=record, symbols=db, name=name)
+    else:
+        session = ge.launch(kernel=kernel, record=record, name=name)
+    manager = PluginManager(session)
+    manager.load_all(plugins)
+    return session, manager
+
+
+def _print_session_dir(session) -> None:
+    directory = getattr(session.recorder, "dir", None)
+    if directory:
+        print(f"[green-eye] session: {directory}")
+
+
+# ---------------------------------------------------------------------------
+# commands
+# ---------------------------------------------------------------------------
+
+
+def cmd_info(args) -> int:
+    from .target import DEFAULT_QEMU
+
+    print(f"green-eye {ge.__version__}")
+    print(f"tool root: {TOOL_ROOT}")
+    qemu = os.environ.get("GREENEYE_QEMU", DEFAULT_QEMU)
+    import shutil
+
+    resolved = shutil.which(qemu) if os.path.sep not in qemu else (qemu if os.path.exists(qemu) else None)
+    print(f"qemu: {resolved or 'NOT FOUND (install qemu-system-misc)'}")
+    if resolved:
+        import subprocess
+
+        version = subprocess.run([resolved, "--version"], capture_output=True, text=True).stdout.strip()
+        print(f"qemu version: {version}")
+    for label, path in [
+        ("demo firmware", DEMO_ELF),
+        ("demo schema", DEMO_SCHEMA),
+        ("default fwimage", DEFAULT_FWIMAGE),
+    ]:
+        print(f"{label}: {path} ({'present' if Path(path).exists() else 'missing'})")
+    cuda_names = Path(TOOL_ROOT).parents[1] / "userspace" / "cuda" / "570.144" / "annotations" / "names.tsv"
+    print(f"CUDA Ghidra names.tsv: {cuda_names} ({'present' if cuda_names.exists() else 'missing'})")
+    plugins_dir = TOOL_ROOT / "plugins"
+    plugins = sorted(p.name for p in plugins_dir.glob("*.py") if not p.name.startswith("_"))
+    print(f"bundled plugins: {', '.join(plugins) or 'none'}")
+    return 0
+
+
+def cmd_demo(args) -> int:
+    _ensure_dumps_env()
+    elf = args.kernel or (_build_demo() if not args.no_build else DEMO_ELF)
+    if not Path(elf).exists():
+        raise GreenEyeError(f"demo firmware not built: {elf}")
+    schema = _load_schema(DEMO_SCHEMA)
+    record = True if args.record is None else args.record
+    session, _manager = _start_session(kernel=str(elf), port=None, record=record, name="demo")
+    hits: list[str] = []
+    try:
+        ctrl = session.symbols.lookup("gsp_ctrl")
+        for name in RPC_HANDLERS:
+            sym = session.symbols.get(name)
+            if sym is not None:
+                session.hook(
+                    sym.addr,
+                    on_enter=lambda hit: hits.append(f"enter {hit.label()} a0={hit.arg(0):#x}"),
+                    on_leave=lambda hit: hits.append(f"leave {hit.label()} -> {hit.retval:#x}"),
+                )
+        session.watch(ctrl + 16, size=8, kind="w", listener="rpc_count")
+        print(f"[demo] ctrl={ctrl:#x} schema={schema.name} size={schema.size:#x}")
+        print("[demo] running the firmware self-test ...")
+        session.run(duration=args.duration)
+        banner = session.target.read_uart() if session.target else ""
+        print("[demo] uart:", banner.strip().splitlines()[0] if banner.strip() else "<empty>")
+        before = session.snapshot(ctrl, schema.size, "ctrl-after-selftest")
+
+        print("[demo] host stimulus: PING")
+        session.memory.write_u64(ctrl + 32, 1)
+        session.memory.write_u64(ctrl + 40, 0x1234)
+        session.run(duration=0.5)
+        after = session.snapshot(ctrl, schema.size, "ctrl-after-ping")
+
+        values = schema.decode(after, base=ctrl)
+        print("[demo] control block after stimulus:")
+        from .structs import render
+
+        print(render(schema, values))
+        diffs = ge.Memory.diff(before, after, base=ctrl)
+        if diffs:
+            print("[demo] snapshot diff (self-test -> ping):")
+            print(ge.Memory.render_diff(diffs, session.symbols, max_lines=16))
+
+        print("[demo] host stimulus: FAULT (ebreak) to demonstrate trap capture")
+        session.memory.write_u64(ctrl + 32, 6)
+        session.run(duration=1.0, max_stops=1)
+        pc = session.pc
+        sym = session.symbols.resolve(pc)
+        print(f"[demo] stopped at {pc:#x} {sym.name if sym else '<unknown>'} (ebreak)")
+        session.step()  # step past the trap
+        print("[demo] stepped past the trap; hits:")
+        for line in hits[:24]:
+            print("   ", line)
+        print(f"[demo] total hook lines: {len(hits)}")
+        _print_session_dir(session)
+    finally:
+        session.close()
+    return 0
+
+
+def cmd_run(args) -> int:
+    _ensure_dumps_env()
+    script = Path(args.script)
+    if not script.exists():
+        raise GreenEyeError(f"script not found: {script}")
+    kernel = args.kernel
+    if not kernel and not args.port:
+        kernel = str(_build_demo())
+    session, manager = _start_session(
+        kernel=kernel, port=args.port, record=args.record or True, name=script.stem, plugins=args.plugin
+    )
+    namespace = {
+        "ge": ge,
+        "session": session,
+        "s": session,
+        "memory": session.memory,
+        "symbols": session.symbols,
+        "recorder": session.recorder,
+        "greeneye": ge,
+    }
+    try:
+        source = script.read_text(encoding="utf-8")
+        exec(compile(source, str(script), "exec"), namespace)
+        return 0
+    except SystemExit as exc:
+        return int(exc.code or 0)
+    finally:
+        session.close()
+
+
+# -- interactive REPL -------------------------------------------------------
+
+
+class GreenEyeRepl(cmd.Cmd):
+    intro = "green-eye interactive instrumentation. Type 'help' for commands."
+    prompt = "greeneye> "
+
+    def __init__(self, session, manager):
+        super().__init__()
+        self.session = session
+        self.manager = manager
+        self._listener_hooks = []
+
+    # utility
+    def _print_hit(self, hit):
+        print("  " + str(hit))
+
+    def _say(self, text):
+        print(text)
+
+    # commands
+    def do_regs(self, arg):
+        """regs -- dump all registers."""
+        regs = self.session.read_regs()
+        for name, value in regs.items():
+            print(f"  {name:<5} = {value:#018x}")
+
+    def do_reg(self, arg):
+        """reg <name> [value] -- read or write one register."""
+        parts = arg.split()
+        if not parts:
+            return self._say("usage: reg <name> [value]")
+        if len(parts) == 1:
+            print(f"  {parts[0]} = {self.session.client.read_reg(parts[0]):#x}")
+        else:
+            self.session.client.write_reg(parts[0], int(parts[1], 0))
+            self._say("ok")
+
+    def do_x(self, arg):
+        """x <addr> [len] -- hexdump memory (annotated)."""
+        parts = arg.split()
+        if not parts:
+            return self._say("usage: x <addr> [len]")
+        addr = self.session.resolve(parts[0])
+        length = int(parts[1], 0) if len(parts) > 1 else 64
+        print(self.session.memory.hexdump(addr, length))
+
+    def do_mem(self, arg):
+        """mem <addr> <len> -- raw memory read."""
+        addr, length = arg.split()
+        print(self.session.memory.read(self.session.resolve(addr), int(length, 0)).hex())
+
+    def do_w(self, arg):
+        """w <addr> <hexbytes> -- write memory."""
+        addr, data = arg.split()
+        self.session.memory.write(self.session.resolve(addr), bytes.fromhex(data))
+        self._say("ok")
+
+    def do_wstr(self, arg):
+        """wstr <addr> <text> -- write a string plus NUL."""
+        addr, _, text = arg.partition(" ")
+        self.session.memory.write(self.session.resolve(addr), text.encode() + b"\x00")
+        self._say("ok")
+
+    def do_bp(self, arg):
+        """bp <addr|symbol> [sw|hw] -- set a breakpoint that prints on hit."""
+        parts = arg.split()
+        if not parts:
+            return self._say("usage: bp <addr|symbol> [sw|hw]")
+        kind = parts[1] if len(parts) > 1 else "sw"
+        handle = self.session.hook(parts[0], on_enter=self._print_hit, kind=kind, listener="repl")
+        self._listener_hooks.append(handle)
+        self._say(f"breakpoint @ {handle.addr:#x} ({handle.label})")
+
+    def do_hook(self, arg):
+        """hook <addr|symbol> -- alias for bp with enter/leave printing."""
+        parts = arg.split()
+        if not parts:
+            return self._say("usage: hook <addr|symbol>")
+        handle = self.session.hook(
+            parts[0], on_enter=self._print_hit, on_leave=self._print_hit, listener="repl"
+        )
+        self._listener_hooks.append(handle)
+        self._say(f"hooked {handle.label}")
+
+    def do_watch(self, arg):
+        """watch <addr|symbol> [size] [w|r|a] -- set a watchpoint."""
+        parts = arg.split()
+        if not parts:
+            return self._say("usage: watch <addr|symbol> [size] [w|r|a]")
+        size = int(parts[1], 0) if len(parts) > 1 else 8
+        kind = parts[2] if len(parts) > 2 else "w"
+        wp = self.session.watch(parts[0], size=size, kind=kind, callback=self._print_hit, listener="repl")
+        self._say(f"watch {kind}@{wp.addr:#x} size={size}")
+
+    def do_s(self, arg):
+        """s [n] -- step n instructions (default 1)."""
+        count = int(arg) if arg.strip() else 1
+        for hit in self.session.step(count):
+            self._print_hit(hit)
+
+    def do_c(self, arg):
+        """c [seconds] -- continue and service events for N seconds (default 1)."""
+        seconds = float(arg) if arg.strip() else 1.0
+        stops = self.session.run(duration=seconds)
+        self._say(f"resumed for {seconds}s, serviced {stops} stops")
+
+    def do_events(self, arg):
+        """events [n] -- show the last n recorded events (default 20)."""
+        from .record import load_events
+
+        directory = getattr(self.session.recorder, "dir", None)
+        if not directory:
+            return self._say("this session is not recording")
+        events = load_events(directory)
+        for event in events[-int(arg or 20) :]:
+            print("  " + str(event))
+
+    def do_snap(self, arg):
+        """snap <label> <addr> <len> -- snapshot a memory region into the session."""
+        parts = arg.split()
+        if len(parts) < 3:
+            return self._say("usage: snap <label> <addr> <len>")
+        data = self.session.memory.snapshot(self.session.resolve(parts[1]), int(parts[2], 0), name=parts[0])
+        self._say(f"snapshot {parts[0]}: {len(data)} bytes")
+
+    def do_sym(self, arg):
+        """sym <addr|name> -- resolve a symbol or an address."""
+        target = arg.strip()
+        if not target:
+            return self._say("usage: sym <addr|name>")
+        try:
+            print("  " + self.session.symbols.annotate(self.session.resolve(target)))
+        except GreenEyeError as exc:
+            self._say(f"error: {exc}")
+
+    def do_note(self, arg):
+        """note <text> -- add a note to the session recording."""
+        self.session.note(arg)
+        self._say("noted")
+
+    def do_quit(self, arg):
+        """quit -- close the session and exit."""
+        return True
+
+    do_EOF = do_quit
+
+    def onecmd(self, line):
+        name = line.split()[0] if line.split() else ""
+        if name in self.manager.api.commands:
+            fn, _help = self.manager.api.commands[name]
+            try:
+                fn(self.session, line.split()[1:])
+            except Exception as exc:
+                self._say(f"plugin error: {exc}")
+            return False
+        if name and name not in self.get_names() and not hasattr(self, f"do_{name}"):
+            return super().onecmd(line)
+        return super().onecmd(line)
+
+    def get_names(self):
+        names = super().get_names()
+        names += [f"do_{cmd}" for cmd in self.manager.api.commands]
+        return names
+
+    def default(self, line):
+        self._say(f"unknown command: {line.split()[0] if line.split() else line!r}")
+
+
+def cmd_attach(args) -> int:
+    _ensure_dumps_env()
+    kernel = args.kernel
+    if kernel:
+        kernel = str(Path(kernel))
+    elif not args.port:
+        kernel = str(_build_demo())
+    session, manager = _start_session(
+        kernel=kernel, port=args.port, record=args.record, name="repl", plugins=args.plugin
+    )
+    try:
+        GreenEyeRepl(session, manager).cmdloop()
+    finally:
+        session.close()
+    return 0
+
+
+def cmd_launch(args) -> int:
+    """Start a target, print the gdbstub endpoint, and keep QEMU alive."""
+    _ensure_dumps_env()
+    kernel = args.kernel or str(_build_demo())
+    if not Path(kernel).exists():
+        raise GreenEyeError(f"kernel not found: {kernel}")
+    session = ge.launch(kernel=str(kernel), record=args.record, name="launch", keep_alive=True)
+    port = session.target.port
+    print(f"[green-eye] target running: {kernel}")
+    print(f"[green-eye] gdbstub: 127.0.0.1:{port}")
+    print(f"[green-eye] attach:  ./greeneye.sh attach --port {port} --kernel {kernel}")
+    print("[green-eye] Ctrl-C stops waiting; QEMU keeps running for attach.")
+    try:
+        while True:
+            time.sleep(1)
+    except KeyboardInterrupt:
+        print("[green-eye] detached (QEMU still running)")
+    return 0
+
+
+def cmd_record(args) -> int:
+    _ensure_dumps_env()
+    elf = args.kernel or str(_build_demo())
+    session, _manager = _start_session(kernel=str(elf), port=None, record=args.record or True, name=args.name)
+    try:
+        for spec in args.watch or []:
+            addr_or_sym, size = (spec.split(":", 1) + ["8"])[:2]
+            session.watch(addr_or_sym, size=int(size, 0))
+        for spec in args.snapshot or []:
+            addr_or_sym, size = spec.split(":", 1)
+            session.snapshot(session.resolve(addr_or_sym), int(size, 0), name=addr_or_sym)
+        session.run(duration=args.duration)
+        print(f"[record] serviced run for {args.duration}s")
+        _print_session_dir(session)
+    finally:
+        session.close()
+    return 0
+
+
+def cmd_probe(args) -> int:
+    _ensure_dumps_env()
+    elf = Path(args.elf)
+    if not elf.exists():
+        raise GreenEyeError(f"firmware not found: {elf}")
+    image = load_elf(str(elf))
+    print(f"[probe] {image.summary()}")
+    relocate = not args.no_relocate
+    bias = (0x80000000 - image.min_vaddr) if relocate else 0
+    kernel_path = str(elf)
+    if relocate:
+        from .elf import write_flat_kernel
+
+        out = Path(tempfile.mkdtemp(prefix="greeneye-probe-")) / (elf.name + ".flat.bin")
+        kernel_path, entry_addr = write_flat_kernel(str(elf), str(out), load_addr=0x80000000)
+        size = Path(kernel_path).stat().st_size
+        print(
+            f"[probe] flat kernel {kernel_path} ({size} bytes); "
+            f"load 0x80000000 entry {entry_addr:#x} (original {image.entry:#x})"
+        )
+    target = QemuTarget(
+        kernel=kernel_path,
+        dlog="guest_errors,unimp,int",
+        memory_mb=max(256, (image.max_vaddr - image.min_vaddr) // (1024 * 1024) + 64),
+    )
+    target.start(paused=True)
+    from .rsp import RspClient
+
+    client = RspClient(target.host, target.port)
+    client.connect()
+    client.sync()
+    symbols = SymbolDB()
+    if args.symbols:
+        symbols.load_json(args.symbols)
+    session = ge.Session(
+        client=client,
+        target=target,
+        symbols=symbols,
+        record=args.record or True,
+        name=f"probe-{elf.stem}",
+        meta={"firmware": str(elf), "entry": image.entry, "bias": bias, "kernel": kernel_path},
+    )
+    session.load_image(image, bias=bias)
+    try:
+        if args.instructions:
+            print(f"[probe] stepping {args.instructions} instructions ...")
+            session.step(args.instructions)
+        else:
+            print(f"[probe] running for {args.duration}s ...")
+            session.run(duration=args.duration)
+        pc = session.pc
+        sym = session.symbols.resolve(pc)
+        print(f"[probe] stopped at {pc:#x} {sym.name if sym else '<unknown>'}")
+        from .analyze import describe_fault, parse_qemu_faults
+
+        qlog = target.read_qemu_log()
+        faults = parse_qemu_faults(qlog)
+        if faults:
+            print("[probe] " + describe_fault(faults[0], bias=bias))
+            print(f"[probe] total exceptions: {len(faults)}")
+        else:
+            print("[probe] no exception recorded (execution stayed in mapped memory)")
+        tail = qlog.strip().splitlines()
+        if tail:
+            print("[probe] qemu log tail:")
+            for line in tail[-6:]:
+                print("   ", line)
+        _print_session_dir(session)
+    finally:
+        session.close()
+    return 0
+
+
+def cmd_analyze(args) -> int:
+    symbols = None
+    if args.symbols:
+        symbols = SymbolDB()
+        symbols.load_json(args.symbols)
+    report = _analyze.render_summary(args.session, symbols=symbols)
+    if args.out:
+        Path(args.out).write_text(report, encoding="utf-8")
+        print(f"wrote {args.out}")
+    else:
+        print(report)
+    return 0
+
+
+def cmd_compare(args) -> int:
+    symbols = SymbolDB() if args.symbols else None
+    report = _analyze.compare_sessions(args.a, args.b, symbols=symbols)
+    print(report)
+    return 0
+
+
+def cmd_symbols(args) -> int:
+    if args.symbols_command == "export":
+        db = _ghidra.build_symbol_db(
+            names_tsv=args.names,
+            index_tsv=args.index,
+            exports_tsv=args.exports,
+            callgraph_tsv=args.callgraph,
+            elf=args.elf,
+            bias=args.bias,
+        )
+        db.save_json(args.out, meta={"bias": args.bias})
+        print(f"wrote {args.out}: {len(db)} symbols")
+        return 0
+    if args.symbols_command == "resolve":
+        db = SymbolDB()
+        if args.db:
+            db.load_json(args.db)
+        print(db.annotate(int(args.addr, 0)))
+        return 0
+    return 1
+
+
+def cmd_plugins(args) -> int:
+    plugins_dir = TOOL_ROOT / "plugins"
+    print("bundled plugins:")
+    for path in sorted(plugins_dir.glob("*.py")):
+        if path.name.startswith("_"):
+            continue
+        print(f"  {path.name}")
+    env = os.environ.get("GREENEYE_PLUGINS")
+    if env:
+        print(f"GREENEYE_PLUGINS: {env}")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# parser
+# ---------------------------------------------------------------------------
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="greeneye", description=__doc__.splitlines()[0])
+    parser.add_argument("--version", action="version", version=f"green-eye {ge.__version__}")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    sub.add_parser("info", help="environment/tooling check").set_defaults(fn=cmd_info)
+
+    demo = sub.add_parser("demo", help="boot the demo uproc and instrument it")
+    demo.add_argument("--kernel", help="firmware ELF (default: built demo)")
+    demo.add_argument("--no-build", action="store_true", help="do not rebuild the demo")
+    demo.add_argument("--record", nargs="?", const=True, default=None, help="record to a path (or default dumps/)")
+    demo.add_argument("--duration", type=float, default=1.0)
+    demo.set_defaults(fn=cmd_demo)
+
+    run = sub.add_parser("run", help="run a Python instrumentation script")
+    run.add_argument("script")
+    run.add_argument("--kernel")
+    run.add_argument("--port", type=int)
+    run.add_argument("--record", nargs="?", const=True, default=None)
+    run.add_argument("--plugin", action="append", default=[])
+    run.set_defaults(fn=cmd_run)
+
+    attach = sub.add_parser("attach", help="interactive REPL over a gdbstub (launches if no --port)")
+    attach.add_argument("--port", type=int, help="attach to an existing gdbstub instead of launching")
+    attach.add_argument("--kernel", help="firmware ELF (default: built demo)")
+    attach.add_argument("--record", nargs="?", const=True, default=None)
+    attach.add_argument("--plugin", action="append", default=[])
+    attach.set_defaults(fn=cmd_attach)
+
+    launch = sub.add_parser("launch", help="start a target and keep it alive for attach")
+    launch.add_argument("--kernel", help="firmware ELF (default: built demo)")
+    launch.add_argument("--record", nargs="?", const=True, default=None)
+    launch.set_defaults(fn=cmd_launch)
+
+    probe = sub.add_parser("probe", help="run a firmware image and report first divergence")
+    probe.add_argument("elf", nargs="?", default=str(DEFAULT_FWIMAGE))
+    probe.add_argument("--no-relocate", action="store_true")
+    probe.add_argument("--instructions", type=int, default=0)
+    probe.add_argument("--duration", type=float, default=1.0)
+    probe.add_argument("--symbols", help="green-eye symbol DB JSON to annotate the stop")
+    probe.add_argument("--record", nargs="?", const=True, default=None)
+    probe.set_defaults(fn=cmd_probe)
+
+    record = sub.add_parser("record", help="boot and record a memory-activity session")
+    record.add_argument("--kernel")
+    record.add_argument("--name", default="record")
+    record.add_argument("--duration", type=float, default=2.0)
+    record.add_argument("--watch", action="append", help="addr|symbol[:size]")
+    record.add_argument("--snapshot", action="append", help="addr|symbol:len")
+    record.add_argument("--record", nargs="?", const=True, default=None)
+    record.set_defaults(fn=cmd_record)
+
+    analyze = sub.add_parser("analyze", help="render a recorded session report")
+    analyze.add_argument("session")
+    analyze.add_argument("--symbols")
+    analyze.add_argument("--out")
+    analyze.set_defaults(fn=cmd_analyze)
+
+    compare = sub.add_parser("compare", help="A/B two recorded sessions")
+    compare.add_argument("a")
+    compare.add_argument("b")
+    compare.add_argument("--symbols", action="store_true")
+    compare.set_defaults(fn=cmd_compare)
+
+    symbols = sub.add_parser("symbols", help="build/inspect a symbol database")
+    symbols_sub = symbols.add_subparsers(dest="symbols_command", required=True)
+    exp = symbols_sub.add_parser("export")
+    exp.add_argument("--names")
+    exp.add_argument("--index")
+    exp.add_argument("--exports")
+    exp.add_argument("--callgraph")
+    exp.add_argument("--elf")
+    exp.add_argument("--bias", type=lambda v: int(v, 0), default=0)
+    exp.add_argument("-o", "--out", required=True)
+    exp.set_defaults(fn=cmd_symbols)
+    res = symbols_sub.add_parser("resolve")
+    res.add_argument("addr")
+    res.add_argument("--db")
+    res.set_defaults(fn=cmd_symbols)
+
+    plugins = sub.add_parser("plugins", help="list discoverable plugins")
+    plugins.set_defaults(fn=cmd_plugins)
+
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    _ensure_dumps_env()
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    try:
+        return args.fn(args)
+    except GreenEyeError as exc:
+        print(f"green-eye error: {exc}", file=sys.stderr)
+        return 2
+    except KeyboardInterrupt:
+        print("interrupted", file=sys.stderr)
+        return 130
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(main())
-- 
cgit v1.2.3