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.

16 files changed, 4042 insertions, 0 deletions

diff --git a/debug/risc-v-gsp-rm-vm/greeneye/__init__.py b/debug/risc-v-gsp-rm-vm/greeneye/__init__.py
new file mode 100644
index 0000000..8b1c1e5
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/__init__.py
@@ -0,0 +1,115 @@
+"""green-eye: Frida-like runtime instrumentation for GSP/uproc firmware in a
+RISC-V QEMU VM.
+
+green-eye talks to QEMU's gdbstub directly (no GDB dependency) and layers a
+scriptable instrumentation API on top: breakpoints, hooks, watchpoints,
+memory inspection, structure decoding, session recording and post-run analysis.
+
+Public API (stable-ish for v0.1):
+
+    import greeneye as ge
+    s = ge.launch(kernel="firmware/uproc_demo/build/uproc_demo.elf")
+    fn = s.symbols.lookup("rpc_sum")
+    h = s.hook(fn, on_enter=lambda hit: print(hit))
+    s.run(duration=2.0)
+    s.close()
+
+Everything is development tooling; see docs/00-OVERVIEW.md.
+"""
+from __future__ import annotations
+
+__version__ = "0.1.0"
+
+from .errors import (  # noqa: F401
+    GreenEyeError,
+    RspError,
+    RspTimeout,
+    RspProtocolError,
+    SymbolNotFound,
+    TargetError,
+    StructError,
+)
+from .rsp import RspClient, StopReply, Register  # noqa: F401
+from .elf import FirmwareImage, Segment, ElfSymbol, load_elf  # noqa: F401
+from .symbols import Symbol, SymbolDB  # noqa: F401
+from .memory import Memory  # noqa: F401
+from .structs import Field, Schema, decode, schema_from_json  # noqa: F401
+from .record import Recorder, NullRecorder, load_events  # noqa: F401
+from .session import Session, Hit, HookHandle, Breakpoint, Watchpoint  # noqa: F401
+from .target import QemuTarget  # noqa: F401
+
+__all__ = [
+    "attach",
+    "launch",
+    "Session",
+    "Hit",
+    "HookHandle",
+    "Breakpoint",
+    "Watchpoint",
+    "QemuTarget",
+    "RspClient",
+    "SymbolDB",
+    "Symbol",
+    "Memory",
+    "Recorder",
+]
+
+
+def attach(
+    host: str = "127.0.0.1",
+    port: int = 4444,
+    *,
+    timeout: float = 10.0,
+    record: str | None = None,
+    symbols: SymbolDB | None = None,
+    name: str = "attach",
+) -> Session:
+    """Attach to an already-running QEMU gdbstub and return a Session."""
+    client = RspClient(host, port, timeout=timeout)
+    client.connect()
+    client.sync()
+    return Session(client=client, symbols=symbols, record=record, name=name)
+
+
+def launch(
+    *,
+    kernel: str | None = None,
+    machine: str = "virt",
+    memory_mb: int = 256,
+    gdb_port: int = 0,
+    bios: str = "none",
+    extra_args: list[str] | None = None,
+    timeout: float = 10.0,
+    record: str | None = None,
+    symbols: SymbolDB | None = None,
+    name: str = "session",
+    load_symbols: bool = True,
+    keep_alive: bool = False,
+) -> Session:
+    """Launch QEMU (paused), attach, and return a ready Session.
+
+    The target starts paused at its reset vector; use Session.load_image() to
+    load an ELF (which also sets PC), or QemuTarget.run_kernel() style helper
+    launch with kernel= for the fast `-kernel` path.
+    """
+    target = QemuTarget(
+        machine=machine,
+        memory_mb=memory_mb,
+        gdb_port=gdb_port,
+        bios=bios,
+        extra_args=extra_args,
+        kernel=kernel,
+    )
+    if keep_alive:
+        target.keep_alive = True
+    target.start(paused=True)
+    client = RspClient(target.host, target.port, timeout=timeout)
+    client.connect()
+    client.sync()
+
+    symdb = symbols or SymbolDB()
+    sess = Session(client=client, target=target, symbols=symdb, record=record, name=name)
+    if kernel:
+        image = load_elf(kernel)
+        sess.load_image(image, load_symbols=load_symbols)
+    return sess
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/__main__.py b/debug/risc-v-gsp-rm-vm/greeneye/__main__.py
new file mode 100644
index 0000000..eb53e2f
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/__main__.py
@@ -0,0 +1,3 @@
+from .cli import main
+
+raise SystemExit(main())
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/analyze.py b/debug/risc-v-gsp-rm-vm/greeneye/analyze.py
new file mode 100644
index 0000000..07c24eb
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/analyze.py
@@ -0,0 +1,293 @@
+"""Post-run analysis: turn a session directory into a symbol-annotated report.
+
+The report is the "complete map" half of the deliverable: hits by function,
+watch hits, memory writes, coverage and snapshot inventory, ready for
+offline comparison between runs (bare-metal vs VFIO, old firmware vs new).
+"""
+from __future__ import annotations
+
+import os
+import re
+from collections import Counter, defaultdict
+
+from .record import load_events, load_meta
+
+_ACCESS_RE = re.compile(
+    r"Invalid (read|write) at addr (0x[0-9a-fA-F]+), size (\d+), region '([^']*)', reason: (\w+)"
+)
+_INTERRUPT_RE = re.compile(
+    r"riscv_cpu_do_interrupt: hart:(\d+), async:(\d+), cause:([0-9a-fA-F]+), "
+    r"epc:(0x[0-9a-fA-F]+), tval:(0x[0-9a-fA-F]+), desc=([\w_]+)"
+)
+
+
+def parse_qemu_faults(log_text: str) -> list[dict]:
+    """Parse QEMU ``-d guest_errors,unimp,int`` logs into fault records.
+
+    Each record pairs the most recent invalid access (if any) with the
+    exception it raised, which is exactly the "first divergence" report a
+    firmware probe needs.
+    """
+    faults: list[dict] = []
+    pending_access: dict | None = None
+    for line in log_text.splitlines():
+        access = _ACCESS_RE.search(line)
+        if access:
+            pending_access = {
+                "rw": access.group(1),
+                "access_addr": int(access.group(2), 16),
+                "size": int(access.group(3)),
+                "region": access.group(4),
+                "reason": access.group(5),
+            }
+            continue
+        interrupt = _INTERRUPT_RE.search(line)
+        if interrupt:
+            faults.append(
+                {
+                    "hart": int(interrupt.group(1)),
+                    "async": int(interrupt.group(2)),
+                    "cause": int(interrupt.group(3), 16),
+                    "epc": int(interrupt.group(4), 16),
+                    "tval": int(interrupt.group(5), 16),
+                    "desc": interrupt.group(6),
+                    "access": pending_access,
+                }
+            )
+            pending_access = None
+    return faults
+
+
+_RISCV_CAUSES = {
+    0: "instruction address misaligned",
+    1: "instruction access fault",
+    2: "illegal instruction",
+    3: "breakpoint",
+    4: "load address misaligned",
+    5: "load access fault",
+    6: "store address misaligned",
+    7: "store access fault",
+}
+
+
+def describe_fault(fault: dict, bias: int = 0) -> str:
+    epc = fault["epc"]
+    fw_pc = epc - bias
+    cause = fault["cause"]
+    desc = _RISCV_CAUSES.get(cause & 0xFF, fault.get("desc", "exception"))
+    lines = [
+        f"first exception: {fault.get('desc', '?')} (cause {cause:#x}: {desc})",
+        f"  epc:      {epc:#x}" + (f"  (firmware vaddr {fw_pc:#x})" if bias else ""),
+        f"  tval:     {fault.get('tval', 0):#x}",
+    ]
+    access = fault.get("access")
+    if access:
+        lines.insert(
+            1,
+            f"  access:   {access['rw']} at {access['access_addr']:#x} "
+            f"({access['size']} bytes, reason={access['reason']})",
+        )
+    return "\n".join(lines)
+
+
+def render_summary(session_dir: str, symbols=None, max_rows: int = 40) -> str:
+    meta = load_meta(session_dir)
+    events = load_events(session_dir)
+    return render_summary_from(meta, events, symbols=symbols, max_rows=max_rows)
+
+
+def render_summary_from(meta: dict, events: list[dict], symbols=None, max_rows: int = 40) -> str:
+    lines: list[str] = []
+    lines.append(f"# green-eye session summary: {meta.get('name', '?')}")
+    lines.append("")
+    lines.append(f"* format: `{meta.get('format', '?')}`  green-eye {meta.get('greeneye', '?')}")
+    lines.append(f"* directory: `{meta.get('dir', '.')}`")
+    lines.append(f"* events: {len(events)}")
+    counts = Counter(event.get("type", "?") for event in events)
+    lines.append("* event counts: " + ", ".join(f"{k}={v}" for k, v in sorted(counts.items())))
+    counts_meta = meta.get("counts")
+    if counts_meta:
+        lines.append("* recorder counts: " + ", ".join(f"{k}={v}" for k, v in sorted(counts_meta.items())))
+    lines.append("")
+
+    def annotate(addr):
+        if not isinstance(addr, int):
+            return ""
+        if symbols is not None:
+            return symbols.annotate(addr)
+        return f"{addr:#x}"
+
+    # Breakpoint hits
+    bp_hits = [e for e in events if e.get("type") == "bp_hit"]
+    if bp_hits:
+        by_site: Counter = Counter()
+        by_symbol: Counter = Counter()
+        for event in bp_hits:
+            key = event.get("symbol") or f"{event.get('pc', 0):#x}"
+            by_site[(event.get("pc"), key)] += 1
+            by_symbol[key] += 1
+        lines.append("## Breakpoint hits")
+        lines.append("")
+        lines.append("| hits | pc | symbol |")
+        lines.append("|---:|---|---|")
+        for (pc, key), count in by_site.most_common(max_rows):
+            lines.append(f"| {count} | `{pc:#x}` | `{key}` |")
+        lines.append("")
+
+    # Watchpoint hits
+    watch_hits = [e for e in events if e.get("type") == "watch_hit"]
+    if watch_hits:
+        by_addr: Counter = Counter()
+        for event in watch_hits:
+            by_addr[(event.get("addr"), event.get("kind"))] += 1
+        lines.append("## Watchpoint hits")
+        lines.append("")
+        lines.append("| hits | address | kind | symbol |")
+        lines.append("|---:|---|---|---|")
+        for (addr, kind), count in by_addr.most_common(max_rows):
+            sym = symbols.resolve(addr) if symbols else None
+            lines.append(f"| {count} | `{addr:#x}` | {kind} | `{sym.name if sym else ''}` |")
+        lines.append("")
+
+    # Memory writes
+    writes = [e for e in events if e.get("type") == "mem_write"]
+    if writes:
+        regions = Counter()
+        for event in writes:
+            regions[(event.get("addr", 0) >> 12) << 12] += 1
+        lines.append("## Memory writes")
+        lines.append("")
+        lines.append(f"* writes: {len(writes)}, distinct 4 KiB pages: {len(regions)}")
+        lines.append("")
+        lines.append("| page | writes | symbol |")
+        lines.append("|---|---:|---|")
+        for page, count in regions.most_common(max_rows):
+            sym = symbols.resolve(page) if symbols else None
+            lines.append(f"| `{page:#x}` | {count} | `{sym.name if sym else ''}` |")
+        lines.append("")
+
+    # Coverage
+    steps = [e for e in events if e.get("type") == "step"]
+    if steps:
+        pcs = {(e.get("pc")) for e in steps}
+        lines.append("## Stepped coverage")
+        lines.append("")
+        lines.append(f"* steps: {len(steps)}, unique PCs: {len(pcs)}")
+        by_sym = Counter()
+        for pc in pcs:
+            sym = symbols.resolve(pc) if symbols else None
+            by_sym[sym.name if sym else f"{pc:#x}"] += 1
+        lines.append("")
+        lines.append("| PCs | symbol |")
+        lines.append("|---:|---|")
+        for name, count in by_sym.most_common(max_rows):
+            lines.append(f"| {count} | `{name}` |")
+        lines.append("")
+
+    # Snapshots
+    snapshots = meta.get("snapshots") or [e for e in events if e.get("type") == "snapshot_saved"]
+    if snapshots:
+        lines.append("## Snapshots")
+        lines.append("")
+        lines.append("| label | address | size | path | sha256 |")
+        lines.append("|---|---|---:|---|---|")
+        for snap in snapshots[:max_rows]:
+            lines.append(
+                f"| {snap.get('label')} | `{snap.get('addr', 0):#x}` | {snap.get('size', 0)} | "
+                f"`{snap.get('path')}` | `{(snap.get('sha256') or '')[:16]}` |"
+            )
+        lines.append("")
+
+    # Timeline excerpt
+    notable = [e for e in events if e.get("type") in ("bp_hit", "leave", "watch_hit", "mem_write", "note", "exit")]
+    if notable:
+        lines.append("## Timeline (first hits)")
+        lines.append("")
+        for event in notable[:max_rows]:
+            t = event.get("t", 0)
+            kind = event.get("type")
+            if kind in ("bp_hit", "leave"):
+                site = event.get("symbol") or f"{event.get('pc', 0):#x}"
+                extra = ""
+                regs = event.get("regs") or {}
+                if "a0" in regs:
+                    extra = f" a0={regs['a0']:#x}"
+                if kind == "leave" and "retval" in event:
+                    extra += f" -> {event['retval']:#x}"
+                lines.append(f"* `{t:8.3f}s` {kind}: {site}{extra}")
+            elif kind == "watch_hit":
+                lines.append(f"* `{t:8.3f}s` watch: {annotate(event.get('addr'))} ({event.get('kind')})")
+            elif kind == "mem_write":
+                lines.append(
+                    f"* `{t:8.3f}s` write: {annotate(event.get('addr'))} "
+                    f"{event.get('size')}B {event.get('data', '')[:32]}"
+                )
+            elif kind == "note":
+                lines.append(f"* `{t:8.3f}s` note: {event.get('text')}")
+            elif kind == "exit":
+                lines.append(f"* `{t:8.3f}s` exit status={event.get('status')}")
+        lines.append("")
+
+    # Symbol-annotation table for automation
+    lines.append("## Annotated sites")
+    lines.append("")
+    sites = Counter()
+    for event in events:
+        for key in ("pc", "addr"):
+            value = event.get(key)
+            if isinstance(value, int) and value and event.get("type") in ("bp_hit", "watch_hit", "mem_write"):
+                sites[value] += 1
+    lines.append("| address | count | annotation |")
+    lines.append("|---|---:|---|")
+    for addr, count in sites.most_common(max_rows):
+        lines.append(f"| `{addr:#x}` | {count} | `{annotate(addr)}` |")
+    lines.append("")
+    return "\n".join(lines) + "\n"
+
+
+def compare_sessions(a_dir: str, b_dir: str, symbols=None, max_rows: int = 40) -> str:
+    """Diff two recorded sessions at the symbolic event level (A/B tooling)."""
+    a_meta, b_meta = load_meta(a_dir), load_meta(b_dir)
+    a_events, b_events = load_events(a_dir), load_events(b_dir)
+    lines = [f"# green-eye session comparison", "", f"* A: {a_meta.get('name')} ({a_dir})", f"* B: {b_meta.get('name')} ({b_dir})", ""]
+
+    def profile(events):
+        bp = Counter()
+        watch = Counter()
+        writes = Counter()
+        for event in events:
+            if event.get("type") == "bp_hit":
+                bp[event.get("symbol") or f"{event.get('pc', 0):#x}"] += 1
+            elif event.get("type") == "watch_hit":
+                watch[(event.get("addr"), event.get("kind"))] += 1
+            elif event.get("type") == "mem_write":
+                writes[(event.get("addr", 0) >> 12) << 12] += 1
+        return bp, watch, writes
+
+    a_bp, a_watch, a_writes = profile(a_events)
+    b_bp, b_watch, b_writes = profile(b_events)
+
+    lines.append("## Breakpoint profile delta (A vs B)")
+    lines.append("")
+    lines.append("| symbol | A hits | B hits | delta |")
+    lines.append("|---|---:|---:|---:|")
+    for name in sorted(set(a_bp) | set(b_bp), key=lambda k: -(a_bp.get(k, 0) + b_bp.get(k, 0)))[:max_rows]:
+        lines.append(f"| `{name}` | {a_bp.get(name, 0)} | {b_bp.get(name, 0)} | {b_bp.get(name, 0) - a_bp.get(name, 0):+d} |")
+    lines.append("")
+
+    lines.append("## Watchpoint delta")
+    lines.append("")
+    lines.append("| address | kind | A | B |")
+    lines.append("|---|---|---:|---:|")
+    for key in sorted(set(a_watch) | set(b_watch), key=lambda k: -(a_watch.get(k, 0) + b_watch.get(k, 0)))[:max_rows]:
+        lines.append(f"| `{key[0]:#x}` | {key[1]} | {a_watch.get(key, 0)} | {b_watch.get(key, 0)} |")
+    lines.append("")
+
+    lines.append("## Written pages delta")
+    lines.append("")
+    lines.append("| page | A writes | B writes |")
+    lines.append("|---|---:|---:|")
+    for page in sorted(set(a_writes) | set(b_writes), key=lambda k: -(a_writes.get(k, 0) + b_writes.get(k, 0)))[:max_rows]:
+        lines.append(f"| `{page:#x}` | {a_writes.get(page, 0)} | {b_writes.get(page, 0)} |")
+    lines.append("")
+    return "\n".join(lines) + "\n"
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())
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/elf.py b/debug/risc-v-gsp-rm-vm/greeneye/elf.py
new file mode 100644
index 0000000..abfccd9
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/elf.py
@@ -0,0 +1,322 @@
+"""Minimal ELF64/ELF32 reader (stdlib only).
+
+Needed for three things:
+  * symbol resolution for the demo uproc firmware (breakpoints by name),
+  * program-header extraction so the real GSP ``.fwimage`` can be placed into
+    QEMU RAM (and relocated when a probe needs it),
+  * entry-point detection for the first-fault firmware probe.
+
+This is not a full ELF library; it parses headers, PT_LOAD segments and
+``.symtab`` symbols, which is all green-eye needs.
+"""
+from __future__ import annotations
+
+import struct
+from dataclasses import dataclass, field as _field
+
+from .errors import GreenEyeError
+
+EM_RISCV = 243
+PT_LOAD = 1
+SHT_SYMTAB = 2
+SHT_STRTAB = 3
+STT_FUNC = 2
+STT_OBJECT = 1
+STT_NOTYPE = 0
+
+
+class ElfError(GreenEyeError):
+    """Malformed or unsupported ELF file."""
+
+
+@dataclass
+class Segment:
+    p_type: int
+    p_flags: int
+    p_offset: int
+    p_vaddr: int
+    p_paddr: int
+    p_filesz: int
+    p_memsz: int
+    p_align: int
+
+    @property
+    def is_load(self) -> bool:
+        return self.p_type == PT_LOAD
+
+    @property
+    def writable(self) -> bool:
+        return bool(self.p_flags & 0x2)
+
+    @property
+    def executable(self) -> bool:
+        return bool(self.p_flags & 0x1)
+
+
+@dataclass
+class ElfSymbol:
+    name: str
+    value: int
+    size: int
+    bind: int
+    type: int
+
+    @property
+    def is_function(self) -> bool:
+        return self.type == STT_FUNC
+
+    @property
+    def is_object(self) -> bool:
+        return self.type == STT_OBJECT
+
+
+@dataclass
+class FirmwareImage:
+    path: str
+    is64: bool
+    e_machine: int
+    entry: int
+    segments: list[Segment] = _field(default_factory=list)
+    symbols: list[ElfSymbol] = _field(default_factory=list)
+    _blob: bytes | None = None
+
+    @property
+    def is_riscv(self) -> bool:
+        return self.e_machine == EM_RISCV
+
+    @property
+    def load_segments(self) -> list[Segment]:
+        return [s for s in self.segments if s.is_load]
+
+    @property
+    def min_vaddr(self) -> int:
+        loads = self.load_segments
+        return min((s.p_vaddr for s in loads), default=0)
+
+    @property
+    def max_vaddr(self) -> int:
+        loads = self.load_segments
+        return max((s.p_vaddr + s.p_memsz for s in loads), default=0)
+
+    def read(self) -> bytes:
+        if self._blob is None:
+            with open(self.path, "rb") as fh:
+                self._blob = fh.read()
+        return self._blob
+
+    def segment_bytes(self, seg: Segment) -> bytes:
+        return self.read()[seg.p_offset : seg.p_offset + seg.p_filesz]
+
+    def flat_load_plan(self, bias: int = 0) -> list[tuple[int, bytes]]:
+        """Return ``(load_addr, data)`` pairs for every PT_LOAD segment.
+
+        ``bias`` is added to each segment's virtual address (use it to place a
+        firmware linked for GPU physical addresses into QEMU RAM).
+        """
+        plan: list[tuple[int, bytes]] = []
+        for seg in self.load_segments:
+            data = bytearray(self.segment_bytes(seg))
+            if len(data) < seg.p_memsz:
+                data += b"\x00" * (seg.p_memsz - len(data))
+            plan.append((seg.p_vaddr + bias, bytes(data)))
+        return plan
+
+    def relocated_entry(self, bias: int = 0) -> int:
+        return self.entry + bias
+
+    def summary(self) -> str:
+        return (
+            f"{self.path}: ELF{'64' if self.is64 else '32'} "
+            f"machine={self.e_machine} entry={self.entry:#x} "
+            f"segments={len(self.segments)} loads={len(self.load_segments)} "
+            f"symbols={len(self.symbols)} "
+            f"vaddr=[{self.min_vaddr:#x},{self.max_vaddr:#x})"
+        )
+
+
+def load_elf(path: str) -> FirmwareImage:
+    with open(path, "rb") as fh:
+        data = fh.read()
+    return parse_elf(data, path=path)
+
+
+def parse_elf(data: bytes, path: str = "<memory>") -> FirmwareImage:
+    if len(data) < 52 or data[:4] != b"\x7fELF":
+        raise ElfError(f"{path}: not an ELF file")
+    ei_class = data[4]
+    if ei_class == 2:
+        is64 = True
+        hdr_fmt = "<HHIQQQIHHHHHH"
+    elif ei_class == 1:
+        is64 = False
+        hdr_fmt = "<HHIIIIIHHHHHH"
+    else:
+        raise ElfError(f"{path}: unknown EI_CLASS {ei_class}")
+    hdr_size = struct.calcsize(hdr_fmt)
+    fields = struct.unpack_from(hdr_fmt, data, 16)
+    (
+        _e_type,
+        e_machine,
+        _e_version,
+        e_entry,
+        e_phoff,
+        e_shoff,
+        _e_flags,
+        _e_ehsize,
+        e_phentsize,
+        e_phnum,
+        e_shentsize,
+        e_shnum,
+        e_shstrndx,
+    ) = fields
+
+    image = FirmwareImage(path=path, is64=is64, e_machine=e_machine, entry=e_entry)
+
+    if is64:
+        ph_fmt = "<IIQQQQQQ"
+        sh_fmt = "<IIQQQQIIQQ"
+        sym_fmt = "<IBBHQQ"
+    else:
+        ph_fmt = "<IIIIIIII"
+        sh_fmt = "<IIIIIIIIII"
+        sym_fmt = "<IIIBBH"
+
+    ph_size = struct.calcsize(ph_fmt)
+    for i in range(e_phnum):
+        base = e_phoff + i * e_phentsize
+        if base + ph_size > len(data):
+            break
+        p = struct.unpack_from(ph_fmt, data, base)
+        image.segments.append(Segment(*p[:8]))
+
+    # Sections -> symbols
+    sh_size = struct.calcsize(sh_fmt)
+    sections: list[tuple[int, int, int, int]] = []  # type, offset, size, link
+    for i in range(e_shnum):
+        base = e_shoff + i * e_shentsize
+        if base + sh_size > len(data):
+            break
+        sh = struct.unpack_from(sh_fmt, data, base)
+        sh_type, sh_offset, sh_size_val, sh_link = sh[1], sh[4], sh[5], sh[6]
+        sections.append((sh_type, sh_offset, sh_size_val, sh_link))
+
+    for sh_type, sh_offset, sh_size_val, sh_link in sections:
+        if sh_type != SHT_SYMTAB:
+            continue
+        strtab_off, strtab_size = 0, 0
+        if 0 <= sh_link < len(sections):
+            _, strtab_off, strtab_size, _ = sections[sh_link]
+        strtab = data[strtab_off : strtab_off + strtab_size]
+        count = sh_size_val // struct.calcsize(sym_fmt)
+        for i in range(count):
+            base = sh_offset + i * struct.calcsize(sym_fmt)
+            if base + struct.calcsize(sym_fmt) > len(data):
+                break
+            st = struct.unpack_from(sym_fmt, data, base)
+            if is64:
+                st_name, st_info, _st_other, _st_shndx, st_value, st_size_val = st
+            else:
+                st_name, st_value, st_size_val, st_info, _st_other, _st_shndx = st
+            if st_name == 0 or st_value == 0:
+                continue
+            name = _read_cstr(strtab, st_name)
+            if not name:
+                continue
+            image.symbols.append(
+                ElfSymbol(
+                    name=name,
+                    value=st_value,
+                    size=st_size_val,
+                    bind=(st_info >> 4) & 0xF,
+                    type=st_info & 0xF,
+                )
+            )
+    return image
+
+
+def _read_cstr(blob: bytes, offset: int) -> str:
+    if offset >= len(blob):
+        return ""
+    end = blob.find(b"\x00", offset)
+    if end == -1:
+        end = len(blob)
+    return blob[offset:end].decode("utf-8", "replace")
+
+
+def write_relocated_elf(path: str, out_path: str, bias: int) -> str:
+    """Write a copy of an ELF64 whose PT_LOADs and entry are shifted by bias.
+
+    ``p_paddr`` is forced to ``p_vaddr + bias`` so the layout matches the
+    virtual addresses.  Useful for formats/tools that honour e_entry.
+    """
+    with open(path, "rb") as fh:
+        data = bytearray(fh.read())
+    if len(data) < 64 or data[:4] != b"\x7fELF" or data[4] != 2:
+        raise ElfError(f"{path}: relocation requires a 64-bit ELF")
+    e_entry = struct.unpack_from("<Q", data, 24)[0]
+    e_phoff = struct.unpack_from("<Q", data, 32)[0]
+    e_phentsize = struct.unpack_from("<H", data, 54)[0]
+    e_phnum = struct.unpack_from("<H", data, 56)[0]
+    struct.pack_into("<Q", data, 24, e_entry + bias)
+    for i in range(e_phnum):
+        base = e_phoff + i * e_phentsize
+        if base + 56 > len(data):
+            break
+        p_type = struct.unpack_from("<I", data, base)[0]
+        if p_type != PT_LOAD:
+            continue
+        p_vaddr = struct.unpack_from("<Q", data, base + 16)[0]
+        struct.pack_into("<Q", data, base + 16, p_vaddr + bias)
+        struct.pack_into("<Q", data, base + 24, p_vaddr + bias)
+    with open(out_path, "wb") as fh:
+        fh.write(data)
+    return out_path
+
+
+def _trampoline(entry_offset: int) -> bytes:
+    """RV64 code at offset 0 that jumps to ``PC + entry_offset``.
+
+    QEMU's RISC-V ``-kernel`` raw/ELF path starts the CPU at the RAM base for
+    images whose e_entry it does not honour; this 8-byte stub (auipc t0, hi;
+    jalr x0, t0, lo) makes the firmware entry reachable from the base without
+    patching the image itself.
+    """
+    if not -0x80000000 <= entry_offset < 0x80000000:
+        raise ElfError("entry offset out of trampoline range")
+    hi20 = ((entry_offset + 0x800) >> 12) & 0xFFFFF
+    lo12 = entry_offset & 0xFFF
+    auipc = (hi20 << 12) | (5 << 7) | 0x17  # auipc t0, hi20
+    jalr = (lo12 << 20) | (5 << 15) | 0x67  # jalr x0, t0, lo12
+    return struct.pack("<II", auipc, jalr)
+
+
+def write_flat_kernel(
+    path: str,
+    out_path: str,
+    *,
+    load_addr: int = 0x80000000,
+    trampoline: bool = True,
+) -> tuple[str, int]:
+    """Flatten an ELF's PT_LOADs into a raw image QEMU can boot as `-kernel`.
+
+    Returns ``(out_path, entry_addr)`` where ``entry_addr = load_addr +
+    (e_entry - min_vaddr)``.  When the entry is not at offset 0 a tiny
+    PC-relative trampoline is written at offset 0 (QEMU starts at the load
+    base for raw kernels).  ``load_addr`` is not encoded anywhere except in
+    the returned entry address: all code is position-independent relative to
+    the image base, exactly as it executes on the device.
+    """
+    image = load_elf(path)
+    span = image.max_vaddr - image.min_vaddr
+    flat = bytearray(span)
+    for seg in image.load_segments:
+        start = seg.p_vaddr - image.min_vaddr
+        end = start + seg.p_filesz
+        flat[start:end] = image.segment_bytes(seg)
+    entry_offset = image.entry - image.min_vaddr
+    if trampoline and entry_offset != 0:
+        stub = _trampoline(entry_offset)
+        flat[0 : len(stub)] = stub
+    with open(out_path, "wb") as fh:
+        fh.write(flat)
+    return out_path, load_addr + entry_offset
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/errors.py b/debug/risc-v-gsp-rm-vm/greeneye/errors.py
new file mode 100644
index 0000000..fab5874
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/errors.py
@@ -0,0 +1,30 @@
+"""green-eye error types. Keep them few and meaningful."""
+from __future__ import annotations
+
+
+class GreenEyeError(Exception):
+    """Base class for all green-eye errors."""
+
+
+class RspError(GreenEyeError):
+    """Base class for gdbstub RSP transport errors."""
+
+
+class RspTimeout(RspError):
+    """The stub did not answer within the timeout."""
+
+
+class RspProtocolError(RspError):
+    """The stub answered with a malformed packet or an error reply."""
+
+
+class TargetError(GreenEyeError):
+    """QEMU target lifecycle/configuration error."""
+
+
+class SymbolNotFound(GreenEyeError):
+    """A requested symbol is not in the symbol database."""
+
+
+class StructError(GreenEyeError):
+    """Structure schema/decode error."""
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/ghidra.py b/debug/risc-v-gsp-rm-vm/greeneye/ghidra.py
new file mode 100644
index 0000000..6986d74
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/ghidra.py
@@ -0,0 +1,80 @@
+"""Ghidra bridge: symbols in, hit annotations out.
+
+Two flows:
+
+*out of Ghidra*: export function names/addresses/sizes to a green-eye JSON
+symbol database (from ``names.tsv``, ``decompiled_index.tsv``, exports or ELF);
+*hits*: turn a recorded session into ``greeneye_hits.tsv``
+(``addr<TAB>hits<TAB>symbol``) that ``MarkGreenEyeHits.java`` can annotate in
+the Ghidra project.
+
+This is the static/dynamic join: Ghidra owns the address space and
+decompilation, green-eye owns what actually executed.
+"""
+from __future__ import annotations
+
+import os
+from collections import Counter
+
+from .record import load_events
+from .symbols import SymbolDB
+
+
+def build_symbol_db(
+    *,
+    names_tsv: str | None = None,
+    index_tsv: str | None = None,
+    exports_tsv: str | None = None,
+    callgraph_tsv: str | None = None,
+    elf: str | None = None,
+    bias: int = 0,
+) -> SymbolDB:
+    db = SymbolDB()
+    if elf:
+        db.load_elf(elf, bias=0)  # ELF symbol values are already runtime addresses
+    if names_tsv:
+        db.load_names_tsv(names_tsv, bias=bias)
+    if index_tsv:
+        db.load_index_tsv(index_tsv, bias=bias)
+    if exports_tsv:
+        db.load_index_tsv(exports_tsv, bias=bias)
+    if callgraph_tsv:
+        db.load_callgraph(callgraph_tsv)
+    return db
+
+
+def export_hits(session_dir: str, out_tsv: str, symbols: SymbolDB | None = None) -> str:
+    events = load_events(session_dir)
+    hits: Counter = Counter()
+    for event in events:
+        if event.get("type") == "bp_hit":
+            hits[event.get("pc")] += 1
+        elif event.get("type") == "watch_hit":
+            hits[event.get("addr")] += 1
+    with open(out_tsv, "w", encoding="utf-8") as fh:
+        fh.write("addr\thits\tsymbol\tannotation\n")
+        for addr, count in sorted(hits.items(), key=lambda kv: -kv[1]):
+            if not isinstance(addr, int):
+                continue
+            sym = symbols.resolve(addr) if symbols else None
+            name = sym.name if sym else ""
+            annotation = symbols.annotate(addr) if symbols else f"{addr:#x}"
+            fh.write(f"{addr:#x}\t{count}\t{name}\t{annotation}\n")
+    return os.path.abspath(out_tsv)
+
+
+def corpus_location(symbol, corpus_root: str) -> str | None:
+    """Return ``<bucket>:<line>`` for a symbol with Ghidra index extras."""
+    bucket = symbol.extras.get("bucket")
+    line = symbol.extras.get("line")
+    if not bucket:
+        return None
+    return os.path.join(corpus_root, bucket) + (f":{line}" if line else "")
+
+
+def render_ghidra_command(project_dir: str, project_name: str, program: str, hits_tsv: str) -> str:
+    return (
+        "analyzeHeadless {proj} {name} -process {program} -noanalysis "
+        "-scriptPath <open-iov>/debug/risc-v-gsp-rm-vm/ghidra "
+        "-postScript MarkGreenEyeHits.java {hits}"
+    ).format(proj=project_dir, name=project_name, program=program, hits=hits_tsv)
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/memory.py b/debug/risc-v-gsp-rm-vm/greeneye/memory.py
new file mode 100644
index 0000000..23d9ca0
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/memory.py
@@ -0,0 +1,190 @@
+"""Memory inspection helpers: read/write, hexdump, search, snapshots, diffs."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .errors import GreenEyeError
+
+
+@dataclass
+class Diff:
+    addr: int
+    old: bytes
+    new: bytes
+
+    @property
+    def size(self) -> int:
+        return len(self.old)
+
+
+class Memory:
+    """A convenience facade over an RSP client (and optional symbol DB)."""
+
+    def __init__(self, client, symbols=None, recorder=None):
+        self.client = client
+        self.symbols = symbols
+        self.recorder = recorder
+
+    # -- basic access ------------------------------------------------------
+
+    def read(self, addr: int, size: int) -> bytes:
+        data = self.client.read_mem(addr, size)
+        if self.recorder is not None:
+            self.recorder.event("mem_read", addr=addr, size=size)
+        return data
+
+    def write(self, addr: int, data: bytes | int | str) -> None:
+        if isinstance(data, int):
+            data = bytes([data & 0xFF])
+        elif isinstance(data, str):
+            data = data.encode("utf-8")
+        self.client.write_mem(addr, data)
+        if self.recorder is not None:
+            self.recorder.event("mem_write", addr=addr, size=len(data), data=data.hex())
+
+    def read_u8(self, addr: int) -> int:
+        return self.read(addr, 1)[0]
+
+    def read_u16(self, addr: int) -> int:
+        return int.from_bytes(self.read(addr, 2), "little")
+
+    def read_u32(self, addr: int) -> int:
+        return int.from_bytes(self.read(addr, 4), "little")
+
+    def read_u64(self, addr: int) -> int:
+        return int.from_bytes(self.read(addr, 8), "little")
+
+    def write_u32(self, addr: int, value: int) -> None:
+        self.write(addr, value.to_bytes(4, "little"))
+
+    def write_u64(self, addr: int, value: int) -> None:
+        self.write(addr, value.to_bytes(8, "little"))
+
+    def read_cstring(self, addr: int, max_len: int = 256) -> str:
+        return self.read(addr, max_len).split(b"\x00", 1)[0].decode("utf-8", "replace")
+
+    def read_ptr(self, addr: int, width: int = 8) -> int:
+        return int.from_bytes(self.read(addr, width), "little")
+
+    # -- display -----------------------------------------------------------
+
+    def hexdump(self, addr: int, size: int, width: int = 16, annotate: bool = True) -> str:
+        data = self.read(addr, size)
+        lines: list[str] = []
+        if annotate and self.symbols is not None:
+            sym = self.symbols.resolve(addr)
+            if sym is not None:
+                lines.append(f"; {sym.label(addr)} (source={sym.source or '?'})")
+        for offset in range(0, len(data), width):
+            chunk = data[offset : offset + width]
+            hex_part = " ".join(f"{b:02x}" for b in chunk)
+            ascii_part = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk)
+            lines.append(f"{addr + offset:08x}  {hex_part:<{width * 3 - 1}}  |{ascii_part}|")
+        return "\n".join(lines)
+
+    def hexdump_annotated(
+        self,
+        addr: int,
+        size: int,
+        struct_schema=None,
+        width: int = 16,
+    ) -> str:
+        """Hexdump plus a decoded structure overlay when a schema is given."""
+        lines = [self.hexdump(addr, size, width=width, annotate=True)]
+        if struct_schema is not None:
+            from .structs import decode, render
+
+            values = decode(struct_schema, self.read(addr, struct_schema.size), base=addr)
+            lines.append("")
+            lines.append(render(struct_schema, values))
+        return "\n".join(lines)
+
+    # -- search ------------------------------------------------------------
+
+    def search(
+        self,
+        addr: int,
+        size: int,
+        pattern: bytes | int,
+        *,
+        mask: bytes | None = None,
+        max_hits: int = 64,
+        chunk: int = 0x10000,
+    ) -> list[int]:
+        if isinstance(pattern, int):
+            pattern = bytes([pattern & 0xFF])
+        if not pattern:
+            raise GreenEyeError("empty search pattern")
+        if mask is not None and len(mask) != len(pattern):
+            raise GreenEyeError("mask length must match pattern length")
+        hits: list[int] = []
+        offset = 0
+        overlap = len(pattern) - 1
+        carry = b""
+        while offset < size and len(hits) < max_hits:
+            want = min(chunk, size - offset)
+            blob = carry + self.read(addr + offset - len(carry), want + len(carry)) if carry else self.read(addr + offset, want)
+            base = addr + offset - len(carry)
+            for index in range(0, len(blob) - len(pattern) + 1):
+                candidate = blob[index : index + len(pattern)]
+                if mask is None:
+                    if candidate == pattern:
+                        hits.append(base + index)
+                else:
+                    if all((candidate[i] & mask[i]) == (pattern[i] & mask[i]) for i in range(len(pattern))):
+                        hits.append(base + index)
+                if len(hits) >= max_hits:
+                    break
+            carry = blob[-overlap:] if overlap else b""
+            offset += want
+        return hits
+
+    def search_u32(self, addr: int, size: int, value: int, **kw) -> list[int]:
+        return self.search(addr, size, value.to_bytes(4, "little"), **kw)
+
+    def search_string(self, addr: int, size: int, text: str, **kw) -> list[int]:
+        return self.search(addr, size, text.encode("utf-8"), **kw)
+
+    # -- capture / diff ----------------------------------------------------
+
+    def snapshot(self, addr: int, size: int, name: str = "region") -> bytes:
+        data = self.client.read_mem(addr, size)
+        if self.recorder is not None:
+            path = self.recorder.snapshot(name, addr, data)
+            self.recorder.event("snapshot", label=name, addr=addr, size=size, path=path)
+        return data
+
+    @staticmethod
+    def diff(old: bytes, new: bytes, base: int = 0) -> list[Diff]:
+        length = min(len(old), len(new))
+        changed = [i for i in range(length) if old[i] != new[i]]
+        if not changed:
+            return []
+        # coalesce adjacent bytes into runs
+        runs: list[Diff] = []
+        start = changed[0]
+        prev = changed[0]
+        for index in changed[1:]:
+            if index == prev + 1:
+                prev = index
+                continue
+            runs.append(Diff(base + start, old[start : prev + 1], new[start : prev + 1]))
+            start = prev = index
+        runs.append(Diff(base + start, old[start : prev + 1], new[start : prev + 1]))
+        return runs
+
+    @staticmethod
+    def render_diff(diffs: list[Diff], symbols=None, max_lines: int = 64) -> str:
+        lines: list[str] = []
+        for diff in diffs[:max_lines]:
+            old = " ".join(f"{b:02x}" for b in diff.old[:8])
+            new = " ".join(f"{b:02x}" for b in diff.new[:8])
+            label = ""
+            if symbols is not None:
+                sym = symbols.resolve(diff.addr)
+                if sym is not None:
+                    label = f"  ; {sym.label(diff.addr)}"
+            lines.append(f"{diff.addr:08x}  {old:<24} -> {new:<24}{label}")
+        if len(diffs) > max_lines:
+            lines.append(f"... {len(diffs) - max_lines} more changed runs")
+        return "\n".join(lines)
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/plugins.py b/debug/risc-v-gsp-rm-vm/greeneye/plugins.py
new file mode 100644
index 0000000..af4545f
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/plugins.py
@@ -0,0 +1,121 @@
+"""green-eye plugin system.
+
+A plugin is a normal Python module that exposes::
+
+    def register(ge):
+        ge.log("my plugin loaded")
+        ge.add_command("mycmd", run_mycmd, help="…")
+        ge.hook("rpc_sum", on_enter=lambda hit: ge.log(f"sum({hit.arg(0)})"))
+
+Plugins can extend green-eye with:
+
+* new CLI commands (including an MCP server later - roadmap);
+* runtime RE tools (extra watchpoint strategies, structure probes, coverage);
+* virtualisation-environment updates (e.g. record/replay harnesses);
+* AI-harness integration and reporting.
+
+Discovery order:
+
+1. explicit ``--plugin PATH`` arguments;
+2. colon-separated paths in ``GREENEYE_PLUGINS``;
+3. ``<tool>/plugins/*.py`` (the bundled examples).
+"""
+from __future__ import annotations
+
+import importlib.util
+import os
+import sys
+from typing import Callable
+
+
+class GreenEyeAPI:
+    """Handle passed to ``register(ge)``."""
+
+    def __init__(self, session, cli=None):
+        self.session = session
+        self.cli = cli
+        self.commands: dict[str, tuple[Callable, str]] = {}
+        self.loaded: list[str] = []
+
+    # -- convenience --------------------------------------------------------
+
+    @property
+    def memory(self):
+        return self.session.memory
+
+    @property
+    def symbols(self):
+        return self.session.symbols
+
+    @property
+    def recorder(self):
+        return self.session.recorder
+
+    def log(self, message: str, **fields) -> None:
+        self.session.recorder.event("plugin", message=message, **fields)
+
+    def hook(self, target, on_enter=None, on_leave=None, **kw):
+        return self.session.hook(target, on_enter=on_enter, on_leave=on_leave, **kw)
+
+    def watch(self, target, size: int = 8, kind: str = "w", callback=None, **kw):
+        return self.session.watch(target, size=size, kind=kind, callback=callback, **kw)
+
+    def on_event(self, kind: str, callback) -> None:
+        self.session.on_event(kind, callback)
+
+    def resolve(self, target):
+        return self.session.resolve(target)
+
+    def add_command(self, name: str, fn: Callable, help: str = "") -> None:
+        self.commands[name] = (fn, help)
+
+    def run_command(self, name: str, args: list[str]) -> int:
+        fn, _help = self.commands[name]
+        return int(fn(self.session, args) or 0)
+
+
+class PluginManager:
+    def __init__(self, session, cli=None):
+        self.session = session
+        self.api = GreenEyeAPI(session, cli=cli)
+        self.loaded: list[str] = []
+
+    def default_paths(self) -> list[str]:
+        here = os.path.dirname(os.path.abspath(__file__))
+        bundled = os.path.join(os.path.dirname(here), "plugins")
+        paths = []
+        env = os.environ.get("GREENEYE_PLUGINS", "")
+        paths += [p for p in env.split(os.pathsep) if p]
+        if os.path.isdir(bundled):
+            paths.append(bundled)
+        return paths
+
+    def load_all(self, explicit: list[str] | None = None) -> list[str]:
+        for path in (explicit or []) + self.default_paths():
+            if os.path.isdir(path):
+                for entry in sorted(os.listdir(path)):
+                    if entry.endswith(".py") and not entry.startswith("_"):
+                        self._load_file(os.path.join(path, entry))
+            elif path.endswith(".py") and os.path.isfile(path):
+                self._load_file(path)
+        return self.loaded
+
+    def _load_file(self, path: str) -> None:
+        name = os.path.splitext(os.path.basename(path))[0]
+        mod_name = f"greeneye_plugin_{name}"
+        try:
+            spec = importlib.util.spec_from_file_location(mod_name, path)
+            if spec is None or spec.loader is None:
+                return
+            module = importlib.util.module_from_spec(spec)
+            sys.modules[mod_name] = module
+            spec.loader.exec_module(module)
+            register = getattr(module, "register", None)
+            if register is None:
+                self.session.recorder.event("plugin_error", plugin=name, error="no register(ge)")
+                return
+            register(self.api)
+            self.loaded.append(name)
+            self.session.recorder.event("plugin_loaded", plugin=name)
+        except Exception as exc:  # plugins must never break the session
+            self.session.recorder.event("plugin_error", plugin=name, error=repr(exc))
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/probes.py b/debug/risc-v-gsp-rm-vm/greeneye/probes.py
new file mode 100644
index 0000000..5837bab
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/probes.py
@@ -0,0 +1,166 @@
+"""Declarative probe specifications (the Frida-script-like layer).
+
+A probe spec is JSON so it can be generated by another tool (or an AI
+harness) and replayed unchanged:
+
+    {
+      "probes": [
+        {
+          "name": "sum-entry",
+          "symbol": "rpc_sum",
+          "when": "a1 > 0",
+          "on_enter": {
+            "log": ["a0", "a1", "a2"],
+            "dump": [{"reg": "a0", "len": 32, "label": "sum-arg"}],
+            "note": "sum called"
+          },
+          "on_leave": {"log": ["retval"]}
+        }
+      ]
+    }
+
+Conditions are evaluated over a register dictionary (``a0``..``a7``, ``pc``,
+``ra``, or raw ``x10`` names).  The evaluator is deliberately tiny: it only
+accepts comparisons/boolean arithmetic over those names and integer
+literals, never arbitrary Python.
+"""
+from __future__ import annotations
+
+import json
+import re
+
+from .errors import GreenEyeError
+
+_ALLOWED_CHARS = re.compile(r"^[\s0-9a-fA-FxX<>=!()+\-*/%&|^~]*$")
+_NAME_RE = re.compile(r"\b[A-Za-z_][A-Za-z0-9_]*\b")
+_ALLOWED_NAMES = re.compile(
+    r"^(retval|pc|ra|sp|gp|tp|fp|and|or|not|True|False|None"
+    r"|[a-z][0-9]?|x[0-9]{1,2}|[stga][0-9]{1,2})$"
+)
+
+
+def load_spec(path: str) -> dict:
+    with open(path, "r", encoding="utf-8") as fh:
+        spec = json.load(fh)
+    if "probes" not in spec or not isinstance(spec["probes"], list):
+        raise GreenEyeError("probe spec must be an object with a 'probes' list")
+    return spec
+
+
+def compile_condition(expr: str | None):
+    if not expr:
+        return None
+    for name in _NAME_RE.findall(expr):
+        if not _ALLOWED_NAMES.match(name):
+            raise GreenEyeError(f"probe condition references disallowed name {name!r}")
+    if not _ALLOWED_CHARS.match(_NAME_RE.sub("0", expr)):
+        raise GreenEyeError(f"probe condition contains disallowed characters: {expr!r}")
+    code = compile(expr, "<probe-when>", "eval")
+
+    def check(regs: dict) -> bool:
+        scope = {
+            **{k: v for k, v in regs.items() if isinstance(v, int)},
+            "retval": regs.get("a0", 0),
+        }
+        try:
+            return bool(eval(code, {"__builtins__": {}}, scope))  # noqa: S307 - whitelisted above
+        except Exception:
+            return False
+
+    return check
+
+
+def install(session, spec: dict) -> list:
+    """Install every probe in the spec; returns the HookHandles."""
+    handles = []
+    for entry in spec["probes"]:
+        target = entry.get("symbol") or entry.get("addr")
+        if target is None:
+            raise GreenEyeError(f"probe {entry.get('name', '?')}: needs 'symbol' or 'addr'")
+        when = compile_condition(entry.get("when"))
+        name = entry.get("name", str(target))
+        on_enter_actions = entry.get("on_enter")
+        on_leave_actions = entry.get("on_leave")
+
+        def make_enter(name=name, when=when, actions=on_enter_actions):
+            def callback(hit):
+                if when is not None and not when(hit.regs):
+                    return
+                _run_actions(hit, name, actions)
+
+            return callback
+
+        def make_leave(name=name, actions=on_leave_actions):
+            def callback(hit):
+                _run_actions(hit, name, actions, is_leave=True)
+
+            return callback
+
+        handle = session.hook(
+            target,
+            on_enter=make_enter() if on_enter_actions else None,
+            on_leave=make_leave() if on_leave_actions else None,
+            listener=f"probe:{name}",
+        )
+        session.note(f"probe installed: {name}", target=str(target))
+        handles.append(handle)
+    return handles
+
+
+def _run_actions(hit, probe_name: str, actions: dict | None, is_leave: bool = False) -> None:
+    session = hit.session
+    if actions is None:
+        return
+    values = {}
+    for expr in actions.get("log", []):
+        values[expr] = _read_value(hit, expr)
+    if values:
+        session.recorder.event(
+            "probe_log",
+            probe=probe_name,
+            phase="leave" if is_leave else "enter",
+            addr=hit.addr,
+            symbol=hit.symbol.name if hit.symbol else None,
+            values=values,
+        )
+    for dump in actions.get("dump", []):
+        addr = _read_value(hit, dump.get("reg", "a0"))
+        if isinstance(addr, str):
+            continue
+        length = int(dump.get("len", 64))
+        label = dump.get("label", f"{probe_name}-dump")
+        try:
+            data = session.memory.read(addr & 0xFFFFFFFFFFFFFFFF, length)
+        except Exception as exc:
+            session.recorder.event("probe_dump_error", probe=probe_name, addr=addr, error=str(exc))
+            continue
+        path = session.recorder.blob(f"{label}-{addr:#x}.bin", data)
+        session.recorder.event(
+            "probe_dump",
+            probe=probe_name,
+            name=label,
+            addr=addr,
+            size=len(data),
+            path=path,
+            preview=data[:32].hex(),
+        )
+    for snapshot in actions.get("snapshot", []):
+        addr = _read_value(hit, snapshot.get("reg", "a0"))
+        if isinstance(addr, str):
+            continue
+        session.snapshot(addr & 0xFFFFFFFFFFFFFFFF, int(snapshot.get("len", 4096)), name=snapshot.get("label", probe_name))
+    if actions.get("note"):
+        session.note(actions["note"], probe=probe_name, addr=hit.addr)
+
+
+def _read_value(hit, expr: str):
+    if expr == "retval":
+        return hit.retval if hit.retval is not None else hit.regs.get("a0", 0)
+    if expr in hit.regs:
+        return hit.regs[expr]
+    if expr.startswith("0x"):
+        return int(expr, 16)
+    try:
+        return int(expr, 0)
+    except ValueError:
+        return f"<unknown:{expr}>"
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/record.py b/debug/risc-v-gsp-rm-vm/greeneye/record.py
new file mode 100644
index 0000000..bf70e58
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/record.py
@@ -0,0 +1,156 @@
+"""Session recording: events.jsonl + snapshots + summary.
+
+A recorded session is the durable artifact green-eye produces: a complete,
+symbol-annotatable map of what the firmware did (hook hits, watch hits,
+memory writes, snapshots, coverage steps) that can be replayed offline,
+diffed between runs (bare-metal vs VFIO-mdev, old GSP vs new GSP) and
+annotated back into Ghidra.
+"""
+from __future__ import annotations
+
+import json
+import os
+import time
+import zlib
+
+FORMAT = "greeneye-session/1"
+
+
+class NullRecorder:
+    dir: str | None = None
+
+    def event(self, event_type: str, **fields) -> None:
+        pass
+
+    def snapshot(self, label: str, addr: int, data: bytes) -> None:
+        pass
+
+    def blob(self, name: str, data: bytes) -> None:
+        pass
+
+    def close(self) -> None:
+        pass
+
+
+class Recorder:
+    def __init__(self, root: str, *, name: str = "session", meta: dict | None = None):
+        self.dir = root
+        os.makedirs(self.dir, exist_ok=True)
+        self.name = name
+        self.meta = dict(meta or {})
+        self.meta.update({"format": FORMAT, "name": name, "created": time.time(), "greeneye": _version()})
+        self._events_path = os.path.join(self.dir, "events.jsonl")
+        self._fh = open(self._events_path, "w", encoding="utf-8")
+        self._t0 = time.monotonic()
+        self._counts: dict[str, int] = {}
+        self._snapshots: list[dict] = []
+        self._closed = False
+
+    @classmethod
+    def new(cls, *, base: str | None = None, name: str = "session", meta: dict | None = None) -> "Recorder":
+        base = base or os.environ.get("GREENEYE_DUMPS") or os.path.join(os.getcwd(), "dumps")
+        stamp = time.strftime("%Y%m%d-%H%M%S")
+        return cls(os.path.join(base, f"{stamp}-{name}"), name=name, meta=meta)
+
+    # -- writing -----------------------------------------------------------
+
+    def event(self, event_type: str, **fields) -> None:
+        if self._closed:
+            return
+        record = {"t": round(time.monotonic() - self._t0, 6), "type": event_type}
+        for key, value in fields.items():
+            record[key] = _jsonable(value)
+        self._fh.write(json.dumps(record, sort_keys=True) + "\n")
+        self._counts[event_type] = self._counts.get(event_type, 0) + 1
+
+    def snapshot(self, label: str, addr: int, data: bytes) -> str:
+        rel = os.path.join("snapshots", f"{label}-{addr:#x}-{len(data)}B.bin")
+        path = os.path.join(self.dir, rel)
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "wb") as fh:
+            fh.write(zlib.compress(data))
+        info = {"label": label, "addr": addr, "size": len(data), "path": rel, "sha256": _sha(data)}
+        self._snapshots.append(info)
+        self.event("snapshot_saved", **info)
+        return rel
+
+    def blob(self, name: str, data: bytes) -> str:
+        rel = os.path.join("blobs", name)
+        path = os.path.join(self.dir, rel)
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "wb") as fh:
+            fh.write(data)
+        self.event("blob_saved", path=rel, size=len(data))
+        return rel
+
+    # -- closing -----------------------------------------------------------
+
+    def close(self, write_summary: bool = True) -> None:
+        if self._closed:
+            return
+        self.event("session_end")
+        self._fh.close()
+        self._closed = True
+        self.meta["counts"] = self._counts
+        self.meta["snapshots"] = self._snapshots
+        with open(os.path.join(self.dir, "session.json"), "w", encoding="utf-8") as fh:
+            json.dump(self.meta, fh, indent=1, sort_keys=True)
+        if write_summary:
+            try:
+                from .analyze import render_summary
+
+                report = render_summary(self.dir)
+                with open(os.path.join(self.dir, "summary.md"), "w", encoding="utf-8") as fh:
+                    fh.write(report)
+            except Exception as exc:  # pragma: no cover - never break close
+                with open(os.path.join(self.dir, "summary-error.txt"), "w", encoding="utf-8") as fh:
+                    fh.write(f"summary failed: {exc}\n")
+
+    def __enter__(self) -> "Recorder":
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self.close()
+
+
+def load_events(session_dir: str) -> list[dict]:
+    path = session_dir
+    if os.path.isdir(session_dir):
+        path = os.path.join(session_dir, "events.jsonl")
+    events: list[dict] = []
+    with open(path, "r", encoding="utf-8") as fh:
+        for line in fh:
+            line = line.strip()
+            if line:
+                events.append(json.loads(line))
+    return events
+
+
+def load_meta(session_dir: str) -> dict:
+    with open(os.path.join(session_dir, "session.json"), "r", encoding="utf-8") as fh:
+        return json.load(fh)
+
+
+def _jsonable(value):
+    if isinstance(value, bytes):
+        return value.hex()
+    if isinstance(value, (set, tuple)):
+        return list(value)
+    if isinstance(value, dict):
+        return {str(k): _jsonable(v) for k, v in value.items()}
+    return value
+
+
+def _sha(data: bytes) -> str:
+    import hashlib
+
+    return hashlib.sha256(data).hexdigest()
+
+
+def _version() -> str:
+    try:
+        from . import __version__
+
+        return __version__
+    except Exception:  # pragma: no cover
+        return "unknown"
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/rsp.py b/debug/risc-v-gsp-rm-vm/greeneye/rsp.py
new file mode 100644
index 0000000..0263c63
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/rsp.py
@@ -0,0 +1,583 @@
+"""Pure-Python GDB Remote Serial Protocol (RSP) client for QEMU's gdbstub.
+
+green-eye deliberately avoids GDB: the whole instrumentation stack is a Python
+library, so sessions, probes and plugins are scriptable end to end.  This
+module implements the RSP subset QEMU needs: packet framing/acks, no-ack mode,
+registers (with layout read from ``target.xml``), memory, software/hardware
+breakpoints, watchpoints, continue/step/interrupt and monitor commands.
+
+Tested against QEMU 8.2 (``qemu-system-riscv64``).
+"""
+from __future__ import annotations
+
+import socket
+import time
+from dataclasses import dataclass, field as _field
+
+from .errors import RspProtocolError, RspTimeout
+
+# ---------------------------------------------------------------------------
+# Packet framing
+# ---------------------------------------------------------------------------
+
+
+def checksum(data: bytes) -> int:
+    """RSP 8-bit checksum: sum of bytes mod 256."""
+    return sum(data) & 0xFF
+
+
+def encode_packet(data: bytes) -> bytes:
+    """Frame a payload as ``$<data>#<checksum>``."""
+    return b"$" + data + b"#" + f"{checksum(data):02x}".encode("ascii")
+
+
+def _hex_value(payload: str) -> int:
+    try:
+        return int(payload, 16)
+    except ValueError as exc:  # pragma: no cover - defensive
+        raise RspProtocolError(f"malformed hex value: {payload!r}") from exc
+
+
+# ---------------------------------------------------------------------------
+# Stop replies
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class StopReply:
+    """A parsed stop reply (``T``/``S``/``W``/``X`` packet)."""
+
+    raw: str
+    signal: int | None = None
+    thread: str | None = None
+    regs: dict[int, int] = _field(default_factory=dict)
+    watch_addr: int | None = None
+    watch_kind: str | None = None
+    exit_status: int | None = None
+
+    @property
+    def is_exit(self) -> bool:
+        return self.raw[:1] in ("W", "X")
+
+    @property
+    def is_stop(self) -> bool:
+        return self.raw[:1] in ("T", "S")
+
+    def __str__(self) -> str:  # pragma: no cover - cosmetic
+        if self.is_exit:
+            return f"exit(status={self.exit_status})"
+        bits = [f"signal={self.signal}"]
+        if self.watch_addr is not None:
+            bits.append(f"{self.watch_kind}@{self.watch_addr:#x}")
+        if self.thread:
+            bits.append(f"thread={self.thread}")
+        return "stop(" + ", ".join(bits) + ")"
+
+
+def parse_stop_reply(raw: str) -> StopReply:
+    """Parse a stop packet body (without the leading channel)."""
+    reply = StopReply(raw=raw)
+    if not raw:
+        return reply
+    head = raw[0]
+    if head in ("T", "S"):
+        if len(raw) >= 3:
+            try:
+                reply.signal = int(raw[1:3], 16)
+            except ValueError:
+                pass
+        rest = raw[3:]
+        for item in rest.split(";"):
+            if not item or ":" not in item:
+                continue
+            key, value = item.split(":", 1)
+            if key == "thread":
+                reply.thread = value
+            elif key in ("watch", "rwatch", "awatch"):
+                reply.watch_kind = key
+                reply.watch_addr = _hex_value(value)
+            else:
+                try:
+                    reply.regs[int(key, 16)] = _hex_value(value)
+                except ValueError:
+                    continue
+    elif head in ("W", "X"):
+        if len(raw) >= 3:
+            try:
+                reply.exit_status = int(raw[1:3], 16)
+            except ValueError:
+                pass
+    return reply
+
+
+# ---------------------------------------------------------------------------
+# Register layout
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class Register:
+    number: int
+    name: str
+    bitsize: int
+    offset: int = 0
+
+    @property
+    def nbytes(self) -> int:
+        return (self.bitsize + 7) // 8
+
+
+#: Fallback layout for RV64 when target.xml is unavailable: x0..x31 + pc.
+RV64_FALLBACK_REGS = [
+    Register(n, name, 64, n * 8)
+    for n, name in enumerate(
+        ["x0", "ra", "sp", "gp", "tp", "t0", "t1", "t2"]
+        + [f"s{i}" for i in range(0, 2)]
+        + [f"a{i}" for i in range(0, 8)]
+        + [f"s{i}" for i in range(2, 12)]
+        + [f"t{i}" for i in range(3, 7)]
+        + ["pc"]
+    )
+]
+
+
+# ---------------------------------------------------------------------------
+# Client
+# ---------------------------------------------------------------------------
+
+
+class RspClient:
+    """A blocking RSP client over TCP."""
+
+    def __init__(self, host: str = "127.0.0.1", port: int = 4444, timeout: float = 10.0):
+        self.host = host
+        self.port = port
+        self.timeout = timeout
+        self.sock: socket.socket | None = None
+        self.no_ack = False
+        self.packet_size = 0x1000
+        self.registers: list[Register] = []
+        self._by_name: dict[str, Register] = {}
+        self._by_number: dict[int, Register] = {}
+        self.features: str = ""
+
+    # -- lifecycle ---------------------------------------------------------
+
+    def connect(self, retries: int = 1, retry_delay: float = 0.1) -> None:
+        last: Exception | None = None
+        for _ in range(max(1, retries)):
+            try:
+                self.sock = socket.create_connection((self.host, self.port), timeout=self.timeout)
+                self.sock.settimeout(self.timeout)
+                return
+            except OSError as exc:
+                last = exc
+                time.sleep(retry_delay)
+        raise RspTimeout(f"cannot connect to gdbstub {self.host}:{self.port}: {last}")
+
+    def close(self) -> None:
+        if self.sock is not None:
+            try:
+                self.sock.close()
+            except OSError:
+                pass
+            self.sock = None
+
+    @property
+    def connected(self) -> bool:
+        return self.sock is not None
+
+    def __enter__(self) -> "RspClient":
+        if not self.connected:
+            self.connect()
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self.close()
+
+    # -- low level I/O -----------------------------------------------------
+
+    def _send_raw(self, data: bytes) -> None:
+        if self.sock is None:
+            raise RspTimeout("not connected")
+        self.sock.sendall(data)
+
+    def _read_exact(self, count: int, deadline: float) -> bytes:
+        if self.sock is None:
+            raise RspTimeout("not connected")
+        chunks = bytearray()
+        while len(chunks) < count:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise RspTimeout(f"timeout waiting for {count} bytes")
+            self.sock.settimeout(remaining)
+            try:
+                part = self.sock.recv(count - len(chunks))
+            except socket.timeout as exc:
+                raise RspTimeout(f"timeout waiting for {count} bytes") from exc
+            if not part:
+                raise RspProtocolError("gdbstub closed the connection")
+            chunks += part
+        return bytes(chunks)
+
+    def _send_packet(self, payload: bytes) -> None:
+        packet = encode_packet(payload)
+        self._send_raw(packet)
+        if self.no_ack:
+            return
+        deadline = time.monotonic() + self.timeout
+        while True:
+            ack = self._read_exact(1, deadline)
+            if ack == b"+":
+                return
+            if ack == b"-":
+                self._send_raw(packet)
+                continue
+            # Some stubs answer with a stop packet instead of an ack when a
+            # run-control command stops immediately; push it back for the
+            # packet reader by buffering is complex, so treat as protocol
+            # error only if it is not a start-of-packet marker.
+            if ack == b"$":
+                raise RspProtocolError("unexpected packet while waiting for ack")
+
+    def _read_packet(self, timeout: float | None = None) -> str:
+        eff = self.timeout if timeout is None else timeout
+        deadline = time.monotonic() + eff
+        # Skip stale acks and interrupt notifications until a packet starts.
+        while True:
+            byte = self._read_exact(1, deadline)
+            if byte == b"$":
+                break
+            if byte == b"+":
+                continue
+            if byte == b"\x03":
+                return "S02"  # async interrupt notification
+        body = bytearray()
+        while True:
+            byte = self._read_exact(1, deadline)
+            if byte == b"#":
+                break
+            body += byte
+        got = self._read_exact(2, deadline)
+        try:
+            expected = int(got.decode("ascii"), 16)
+        except ValueError as exc:
+            raise RspProtocolError(f"malformed checksum bytes {got!r}") from exc
+        if checksum(bytes(body)) != expected:
+            if not self.no_ack:
+                self._send_raw(b"-")
+            raise RspProtocolError(f"bad checksum for {bytes(body)!r}")
+        if not self.no_ack:
+            self._send_raw(b"+")
+        return bytes(body).decode("latin1")
+
+    def command(self, payload: str | bytes, timeout: float | None = None) -> str:
+        """Send one packet and return the reply payload."""
+        data = payload.encode("ascii") if isinstance(payload, str) else payload
+        self._send_packet(data)
+        return self._read_packet(timeout)
+
+    # -- handshake ---------------------------------------------------------
+
+    def sync(self) -> None:
+        """Negotiate features and register layout; consume the initial stop."""
+        # QEMU acknowledges QStartNoAckMode with an EMPTY packet ($#00), then
+        # switches to no-ack mode once the client acks that reply.
+        try:
+            self._send_packet(b"QStartNoAckMode")
+            reply = self._read_packet()
+            if reply in ("", "OK"):
+                self.no_ack = True
+        except RspProtocolError:
+            pass  # older stub: stay in ack mode
+
+        self.packet_size = 0x1000
+        try:
+            self.features = self.command("qSupported:multiprocess+;swbreak+;hwbreak+;qXfer:features:read+")
+            for token in self.features.split(";"):
+                if token.startswith("PacketSize="):
+                    try:
+                        self.packet_size = int(token.split("=", 1)[1], 16)
+                    except ValueError:
+                        pass
+        except RspProtocolError:
+            self.features = ""
+
+        # Consume the initial stop reply (target is paused).
+        try:
+            self.command("?")
+        except RspProtocolError:
+            pass
+        # Select CPU 0 for register access in system mode.
+        for selection in ("Hg0", "Hc-1"):
+            try:
+                self.command(selection)
+            except RspProtocolError:
+                pass
+        self._load_registers()
+
+    def _load_registers(self) -> None:
+        self.registers = []
+        try:
+            xml = self._read_target_xml()
+        except RspProtocolError:
+            xml = ""
+        if xml:
+            self.registers = _parse_target_xml(xml)
+        if not self.registers:
+            self.registers = list(RV64_FALLBACK_REGS)
+        self._by_name = {r.name: r for r in self.registers}
+        self._by_number = {r.number: r for r in self.registers}
+
+    def _read_target_xml(self, depth: int = 0) -> str:
+        """Fetch target.xml and recursively inline its xi:include files."""
+        return self._collect_annex("target.xml", depth)
+
+    def _collect_annex(self, annex: str, depth: int) -> str:
+        import re
+
+        try:
+            xml = self._read_xfer("features", annex)
+        except RspProtocolError:
+            return ""
+        if depth >= 4:
+            return xml
+        parts = [xml]
+        for href in re.findall(r'xi:include[^>]*href="([^"]+)"', xml):
+            if href.startswith(("http:", "https:", "file:")):
+                continue
+            child = self._collect_annex(href, depth + 1)
+            if child:
+                parts.append(child)
+        return "\n".join(parts)
+
+    def _read_xfer(self, kind: str, annex: str) -> str:
+        data = bytearray()
+        offset = 0
+        while True:
+            reply = self.command(f"qXfer:{kind}:read:{annex}:{offset:x},fff")
+            if reply.startswith("E"):
+                raise RspProtocolError(f"qXfer {kind}:{annex} failed: {reply}")
+            if not reply or reply[0] not in ("l", "m"):
+                raise RspProtocolError(f"qXfer {kind}:{annex} malformed: {reply[:16]!r}")
+            chunk = reply[1:].encode("latin1")
+            data += chunk
+            offset += len(chunk)
+            if reply[0] == "l":
+                break
+        return data.decode("latin1", "replace")
+
+    # -- registers ---------------------------------------------------------
+
+    def read_registers(self) -> dict[str, int]:
+        payload = self.command("g")
+        raw = _unhex(payload)
+        out: dict[str, int] = {}
+        for reg in self.registers:
+            start = reg.offset
+            chunk = raw[start : start + reg.nbytes]
+            if len(chunk) < reg.nbytes:
+                break
+            out[reg.name] = int.from_bytes(chunk, "little", signed=False)
+        return out
+
+    def read_reg(self, name_or_number: str | int) -> int:
+        reg = self._reg(name_or_number)
+        reply = self.command(f"p{reg.number:x}")
+        if reply.startswith("E"):
+            raise RspProtocolError(f"read register {reg.name} failed: {reply}")
+        return int.from_bytes(_unhex(reply), "little", signed=False)
+
+    def write_reg(self, name_or_number: str | int, value: int) -> None:
+        reg = self._reg(name_or_number)
+        payload = value.to_bytes(reg.nbytes, "little", signed=False).hex()
+        reply = self.command(f"P{reg.number:x}={payload}")
+        if reply != "OK":
+            raise RspProtocolError(f"write register {reg.name} failed: {reply}")
+
+    def _reg(self, name_or_number: str | int) -> Register:
+        if isinstance(name_or_number, int):
+            if name_or_number in self._by_number:
+                return self._by_number[name_or_number]
+            # allow ABI-numbered convenience: 0..31 map to x0..x31
+            if name_or_number < len(self.registers):
+                return self.registers[name_or_number]
+            raise RspProtocolError(f"unknown register number {name_or_number}")
+        if name_or_number in self._by_name:
+            return self._by_name[name_or_number]
+        aliases = {"pc": "pc", "a0": "a0", "return": "ra"}
+        name_or_number = aliases.get(name_or_number, name_or_number)
+        if name_or_number in self._by_name:
+            return self._by_name[name_or_number]
+        raise RspProtocolError(f"unknown register {name_or_number!r}")
+
+    @property
+    def pc(self) -> int:
+        return self.read_reg("pc")
+
+    # -- memory ------------------------------------------------------------
+
+    def _mem_chunk(self) -> int:
+        # Each data byte costs two hex chars; leave room for the header.
+        return max(64, (self.packet_size - 128) // 2)
+
+    def read_mem(self, addr: int, size: int, chunk: int = 0) -> bytes:
+        if chunk <= 0:
+            chunk = self._mem_chunk()
+        out = bytearray()
+        while len(out) < size:
+            want = min(chunk, size - len(out))
+            here = addr + len(out)
+            reply = self.command(f"m{here:x},{want:x}")
+            if reply.startswith("E"):
+                raise RspProtocolError(f"memory read {here:#x}+{want:#x} failed: {reply}")
+            data = _unhex(reply)
+            if not data:
+                raise RspProtocolError(f"short memory read at {here:#x}")
+            out += data
+        return bytes(out)
+
+    def write_mem(self, addr: int, data: bytes, chunk: int = 0) -> None:
+        if chunk <= 0:
+            chunk = self._mem_chunk()
+        view = memoryview(data)
+        offset = 0
+        while offset < len(view):
+            part = bytes(view[offset : offset + chunk])
+            here = addr + offset
+            reply = self.command(f"M{here:x},{len(part):x}:{part.hex()}")
+            if reply != "OK":
+                raise RspProtocolError(f"memory write {here:#x} failed: {reply}")
+            offset += len(part)
+
+    def read_u32(self, addr: int) -> int:
+        return int.from_bytes(self.read_mem(addr, 4), "little")
+
+    def read_u64(self, addr: int) -> int:
+        return int.from_bytes(self.read_mem(addr, 8), "little")
+
+    def write_u32(self, addr: int, value: int) -> None:
+        self.write_mem(addr, value.to_bytes(4, "little"))
+
+    def write_u64(self, addr: int, value: int) -> None:
+        self.write_mem(addr, value.to_bytes(8, "little"))
+
+    def read_cstring(self, addr: int, max_len: int = 256) -> str:
+        raw = self.read_mem(addr, max_len)
+        return raw.split(b"\x00", 1)[0].decode("utf-8", "replace")
+
+    # -- breakpoints / watchpoints ----------------------------------------
+
+    def set_sw_break(self, addr: int, kind: int = 4) -> None:
+        reply = self.command(f"Z0,{addr:x},{kind:x}")
+        if reply != "OK":
+            raise RspProtocolError(f"set software breakpoint {addr:#x} failed: {reply}")
+
+    def clear_sw_break(self, addr: int, kind: int = 4) -> None:
+        reply = self.command(f"z0,{addr:x},{kind:x}")
+        if reply != "OK":
+            raise RspProtocolError(f"clear software breakpoint {addr:#x} failed: {reply}")
+
+    def set_hw_break(self, addr: int, kind: int = 4) -> None:
+        reply = self.command(f"Z1,{addr:x},{kind:x}")
+        if reply != "OK":
+            raise RspProtocolError(f"set hardware breakpoint {addr:#x} failed: {reply}")
+
+    def clear_hw_break(self, addr: int, kind: int = 4) -> None:
+        reply = self.command(f"z1,{addr:x},{kind:x}")
+        if reply != "OK":
+            raise RspProtocolError(f"clear hardware breakpoint {addr:#x} failed: {reply}")
+
+    _WATCH_KINDS = {"w": 2, "r": 3, "a": 4}
+
+    def set_watch(self, addr: int, size: int = 8, kind: str = "w") -> None:
+        code = self._WATCH_KINDS[kind]
+        reply = self.command(f"Z{code},{addr:x},{size:x}")
+        if reply != "OK":
+            raise RspProtocolError(f"set watchpoint {kind}@{addr:#x} failed: {reply}")
+
+    def clear_watch(self, addr: int, size: int = 8, kind: str = "w") -> None:
+        code = self._WATCH_KINDS[kind]
+        reply = self.command(f"z{code},{addr:x},{size:x}")
+        if reply != "OK":
+            raise RspProtocolError(f"clear watchpoint {kind}@{addr:#x} failed: {reply}")
+
+    # -- run control -------------------------------------------------------
+
+    def continue_(self, timeout: float | None = None) -> StopReply:
+        self._send_packet(b"c")
+        raw = self._read_packet(timeout)
+        return parse_stop_reply(raw)
+
+    def step(self, timeout: float | None = None) -> StopReply:
+        self._send_packet(b"s")
+        raw = self._read_packet(timeout)
+        return parse_stop_reply(raw)
+
+    def interrupt(self, timeout: float | None = None) -> StopReply:
+        self._send_raw(b"\x03")
+        raw = self._read_packet(timeout)
+        return parse_stop_reply(raw)
+
+    # -- monitor -----------------------------------------------------------
+
+    def monitor(self, cmd: str) -> str:
+        payload = ("qRcmd," + cmd.encode("ascii").hex()).encode("ascii")
+        self._send_packet(payload)
+        out = bytearray()
+        while True:
+            reply = self._read_packet()
+            if reply == "OK":
+                break
+            if reply.startswith("E"):
+                raise RspProtocolError(f"monitor {cmd!r} failed: {reply}")
+            if reply.startswith("O"):
+                out += bytes.fromhex(reply[1:])
+            else:  # pragma: no cover - defensive
+                raise RspProtocolError(f"unexpected monitor reply: {reply[:32]!r}")
+        return out.decode("utf-8", "replace")
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _unhex(payload: str) -> bytes:
+    if len(payload) % 2:
+        payload = payload[:-1]  # tolerate odd trailing nibble
+    try:
+        return bytes.fromhex(payload)
+    except ValueError as exc:
+        raise RspProtocolError(f"malformed hex payload: {payload[:32]!r}") from exc
+
+
+_RE_REG = None
+
+
+def _parse_target_xml(xml: str) -> list[Register]:
+    import re
+
+    global _RE_REG
+    if _RE_REG is None:
+        _RE_REG = re.compile(r"<reg\b[^>]*?/?>", re.IGNORECASE)
+    attrs_re = re.compile(r'(\w+)\s*=\s*"([^"]*)"')
+    regs: list[Register] = []
+    next_number = 0
+    offset = 0
+    for match in _RE_REG.finditer(xml):
+        attrs = {k: v for k, v in attrs_re.findall(match.group(0))}
+        name = attrs.get("name")
+        if not name:
+            continue
+        bitsize = int(attrs.get("bitsize", "64"))
+        number = int(attrs["regnum"], 0) if "regnum" in attrs else next_number
+        if "offset" in attrs:
+            try:
+                offset = int(attrs["offset"], 0)
+            except ValueError:
+                pass
+        regs.append(Register(number=number, name=name, bitsize=bitsize, offset=offset))
+        offset += (bitsize + 7) // 8
+        next_number = number + 1
+    return regs
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/session.py b/debug/risc-v-gsp-rm-vm/greeneye/session.py
new file mode 100644
index 0000000..7ad6444
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/session.py
@@ -0,0 +1,562 @@
+"""green-eye instrumentation session: hooks, breakpoints, watchpoints, events.
+
+This is the Frida-like core.  It is deliberately transport-agnostic; the RSP
+client is the default backend, and a Session can also be constructed around a
+plugin-supplied backend later.
+"""
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field as _field
+from typing import Callable
+
+from .errors import GreenEyeError, RspTimeout, SymbolNotFound
+from .memory import Memory
+from .record import NullRecorder, Recorder
+from .rsp import RspClient, StopReply
+from .symbols import Symbol, SymbolDB
+
+
+@dataclass
+class Hit:
+    """Everything known about one stop event."""
+
+    kind: str  # breakpoint | leave | watchpoint | step | fault | exit | stop
+    addr: int
+    regs: dict = _field(default_factory=dict)
+    symbol: Symbol | None = None
+    retval: int | None = None
+    watch: "Watchpoint | None" = None
+    stop: StopReply | None = None
+    session: "Session | None" = None
+    data: dict = _field(default_factory=dict)
+
+    @property
+    def pc(self) -> int:
+        return self.regs.get("pc", self.addr)
+
+    def arg(self, index: int) -> int:
+        """RISC-V ABI argument a0..a7 -> x10..x17."""
+        return self.regs.get(f"a{index}", 0)
+
+    def label(self) -> str:
+        if self.symbol is not None:
+            return self.symbol.label(self.addr)
+        return f"{self.addr:#x}"
+
+    def __str__(self) -> str:  # pragma: no cover - cosmetic
+        base = f"[{self.kind}] {self.label()}"
+        if self.kind in ("breakpoint", "leave"):
+            base += f" a0={self.arg(0):#x} a1={self.arg(1):#x}"
+        if self.kind == "leave" and self.retval is not None:
+            base += f" -> {self.retval:#x}"
+        if self.kind == "watchpoint" and self.watch is not None:
+            base += f" {self.watch.kind}@{self.watch.addr:#x}"
+        return base
+
+
+@dataclass
+class Breakpoint:
+    addr: int
+    kind: str = "sw"  # sw | hw
+    temporary: bool = False
+    purpose: str = "user"  # user | leave
+    on_enter: Callable[[Hit], None] | None = None
+    on_leave: Callable[[Hit], None] | None = None
+    symbol: Symbol | None = None
+    attached: bool = True
+    hits: int = 0
+    listener: str | None = None
+
+    @property
+    def label(self) -> str:
+        return self.symbol.label(self.addr) if self.symbol else f"{self.addr:#x}"
+
+
+@dataclass
+class Watchpoint:
+    addr: int
+    size: int = 8
+    kind: str = "w"  # w | r | a
+    callback: Callable[[Hit], None] | None = None
+    symbol: Symbol | None = None
+    hits: int = 0
+    listener: str | None = None
+
+    @property
+    def label(self) -> str:
+        return self.symbol.label(self.addr) if self.symbol else f"{self.addr:#x}"
+
+
+class HookHandle:
+    def __init__(self, session: "Session", breakpoint: Breakpoint):
+        self._session = session
+        self._bp = breakpoint
+
+    @property
+    def addr(self) -> int:
+        return self._bp.addr
+
+    @property
+    def hits(self) -> int:
+        return self._bp.hits
+
+    @property
+    def label(self) -> str:
+        return self._bp.label
+
+    def detach(self) -> None:
+        self._session.remove_breakpoint(self._bp)
+
+
+class Session:
+    def __init__(
+        self,
+        *,
+        client: RspClient,
+        target=None,
+        symbols: SymbolDB | None = None,
+        record: str | bool | None = None,
+        name: str = "session",
+        meta: dict | None = None,
+    ):
+        self.client = client
+        self.target = target
+        self.symbols = symbols or SymbolDB()
+        self.name = name
+        self.image = None
+        self._owns_target = target is not None
+        if isinstance(record, str):
+            self.recorder = Recorder(record, name=name, meta=meta or {})
+        elif record:
+            self.recorder = Recorder.new(name=name, meta=meta or {})
+        else:
+            self.recorder = NullRecorder()
+        self.memory = Memory(client, symbols=self.symbols, recorder=self.recorder)
+        self._bps: dict[int, Breakpoint] = {}
+        self._bp_seq = 0
+        self._bps_by_addr: dict[int, list[Breakpoint]] = {}
+        self._watches: dict[tuple[int, int, str], Watchpoint] = {}
+        self._watches_by_addr: dict[int, Watchpoint] = {}
+        self._leave_pending: dict[int, list[Breakpoint]] = {}
+        self._stopped = True
+        self._event_hooks: list[tuple[str, Callable[[Hit], None]]] = []
+        self.recorder.event("session_start", name=name)
+
+    # -- convenience proxies ----------------------------------------------
+
+    def load_image(
+        self,
+        image,
+        *,
+        write_segments: bool = False,
+        set_pc: bool = False,
+        bias: int = 0,
+        load_symbols: bool = True,
+        source: str = "elf",
+    ):
+        """Register a firmware image with the session.
+
+        ``write_segments`` copies every PT_LOAD through the RSP client (for
+        targets started without their loader devices); ``set_pc`` points the
+        CPU at the (relocated) entry.  With QEMU's ``-kernel`` fast path both
+        are already handled by QEMU, so the default only loads symbols.
+        """
+        self.image = image
+        if load_symbols:
+            self.symbols.load_elf(image, bias=bias, source=source)
+        if write_segments:
+            for addr, data in image.flat_load_plan(bias=bias):
+                self.client.write_mem(addr, data)
+        if set_pc:
+            self.client.write_reg("pc", image.relocated_entry(bias=bias))
+        self.recorder.event(
+            "image_loaded",
+            path=getattr(image, "path", None),
+            entry=image.entry,
+            bias=bias,
+            write_segments=write_segments,
+        )
+        return image
+
+    @property
+    def pc(self) -> int:
+        return self.client.pc
+
+    def read_regs(self) -> dict[str, int]:
+        return self.client.read_registers()
+
+    def snapshot(self, addr: int, size: int, name: str = "region") -> bytes:
+        return self.memory.snapshot(addr, size, name=name)
+
+    def note(self, text: str, **fields) -> None:
+        self.recorder.event("note", text=text, **fields)
+
+    # -- symbol helpers ----------------------------------------------------
+
+    def resolve(self, target: int | str) -> int:
+        if isinstance(target, int):
+            return target
+        if isinstance(target, str):
+            if target.lower().startswith("0x"):
+                return int(target, 16)
+            sym = self.symbols.get(target)
+            if sym is None:
+                raise SymbolNotFound(f"symbol {target!r} not found in symbol DB")
+            return sym.addr
+        raise GreenEyeError(f"cannot resolve target {target!r}")
+
+    def symbol_at(self, addr: int) -> Symbol | None:
+        return self.symbols.resolve(addr)
+
+    # -- breakpoints / hooks ----------------------------------------------
+
+    def hook(
+        self,
+        target: int | str,
+        on_enter: Callable[[Hit], None] | None = None,
+        on_leave: Callable[[Hit], None] | None = None,
+        *,
+        kind: str = "sw",
+        temporary: bool = False,
+        symbol: Symbol | None = None,
+        listener: str | None = None,
+    ) -> HookHandle:
+        addr = self.resolve(target)
+        sym = symbol or self.symbols.resolve(addr)
+        bp = Breakpoint(
+            addr=addr,
+            kind=kind,
+            temporary=temporary,
+            on_enter=on_enter,
+            on_leave=on_leave,
+            symbol=sym,
+            listener=listener,
+        )
+        self.add_breakpoint(bp)
+        self.recorder.event(
+            "hook",
+            addr=addr,
+            symbol=sym.name if sym else None,
+            kind=kind,
+            on_leave=bool(on_leave),
+            listener=listener,
+        )
+        return HookHandle(self, bp)
+
+    def breakpoint(self, target: int | str, callback: Callable[[Hit], None] | None = None, **kw) -> Breakpoint:
+        addr = self.resolve(target)
+        bp = Breakpoint(addr=addr, on_enter=callback, symbol=self.symbols.resolve(addr), **kw)
+        self.add_breakpoint(bp)
+        return bp
+
+    def add_breakpoint(self, bp: Breakpoint) -> None:
+        self._bp_seq += 1
+        self._bps[self._bp_seq] = bp
+        self._bps_by_addr.setdefault(bp.addr, []).append(bp)
+        if bp.kind == "hw":
+            self.client.set_hw_break(bp.addr)
+        else:
+            self.client.set_sw_break(bp.addr)
+
+    def remove_breakpoint(self, bp: Breakpoint) -> None:
+        for key, candidate in list(self._bps.items()):
+            if candidate is bp:
+                del self._bps[key]
+        lst = self._bps_by_addr.get(bp.addr, [])
+        if bp in lst:
+            lst.remove(bp)
+        if not lst:
+            self._bps_by_addr.pop(bp.addr, None)
+            self._leave_pending.pop(bp.addr, None)
+            try:
+                if bp.kind == "hw":
+                    self.client.clear_hw_break(bp.addr)
+                else:
+                    self.client.clear_sw_break(bp.addr)
+            except Exception:
+                pass
+        bp.attached = False
+
+    def on_event(self, kind: str, callback: Callable[[Hit], None]) -> None:
+        self._event_hooks.append((kind, callback))
+
+    # -- watchpoints -------------------------------------------------------
+
+    def watch(
+        self,
+        target: int | str,
+        size: int = 8,
+        kind: str = "w",
+        callback: Callable[[Hit], None] | None = None,
+        listener: str | None = None,
+    ) -> Watchpoint:
+        addr = self.resolve(target)
+        wp = Watchpoint(
+            addr=addr,
+            size=size,
+            kind=kind,
+            callback=callback,
+            symbol=self.symbols.resolve(addr),
+            listener=listener,
+        )
+        key = (addr, size, kind)
+        if key in self._watches:
+            raise GreenEyeError(f"watchpoint {kind}@{addr:#x}/{size} already exists")
+        self._watches[key] = wp
+        self._watches_by_addr[addr] = wp
+        self.client.set_watch(addr, size=size, kind=kind)
+        self.recorder.event(
+            "watch",
+            addr=addr,
+            size=size,
+            kind=kind,
+            symbol=wp.symbol.name if wp.symbol else None,
+            listener=listener,
+        )
+        return wp
+
+    def unwatch(self, wp: Watchpoint) -> None:
+        key = (wp.addr, wp.size, wp.kind)
+        self._watches.pop(key, None)
+        self._watches_by_addr.pop(wp.addr, None)
+        try:
+            self.client.clear_watch(wp.addr, size=wp.size, kind=wp.kind)
+        except Exception:
+            pass
+
+    # -- execution ---------------------------------------------------------
+
+    def step(self, count: int = 1) -> list[Hit]:
+        hits: list[Hit] = []
+        for _ in range(count):
+            stop = self.client.step()
+            hit = self._dispatch_stop(stop, from_step=True)
+            if hit is not None:
+                hits.append(hit)
+            if stop.is_exit:
+                break
+        return hits
+
+    def run(
+        self,
+        duration: float | None = None,
+        max_stops: int | None = None,
+        timeout: float | None = None,
+    ) -> int:
+        """Resume and service events until the deadline/budget/exit.
+
+        ``duration`` (or ``timeout`` when duration is None) is a wall-clock
+        budget: when it expires the target is interrupted once, the resulting
+        stop is dispatched, and the call returns.  Returns the number of stops
+        serviced.
+        """
+        limit = duration if duration is not None else timeout
+        deadline = time.monotonic() + limit if limit is not None else None
+        stops = 0
+        while True:
+            if deadline is not None:
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    break  # budget spent while stopped: leave the target stopped
+                remaining = max(0.001, remaining)
+            else:
+                remaining = None
+            interrupted = False
+            try:
+                stop = self.client.continue_(timeout=remaining)
+            except RspTimeout:
+                if deadline is None:
+                    raise
+                stop = self.client.interrupt()
+                interrupted = True
+            stops += 1
+            self._dispatch_stop(stop)
+            if stop.is_exit:
+                break
+            if max_stops is not None and stops >= max_stops:
+                break
+            if interrupted or (deadline is not None and time.monotonic() >= deadline):
+                break
+        return stops
+
+    def stop(self) -> Hit | None:
+        stop = self.client.interrupt()
+        return self._dispatch_stop(stop)
+
+    # -- stop dispatch -----------------------------------------------------
+
+    def _dispatch_stop(self, stop: StopReply, from_step: bool = False) -> Hit | None:
+        if stop.is_exit:
+            hit = Hit(kind="exit", addr=0, stop=stop, session=self)
+            self.recorder.event("exit", status=stop.exit_status)
+            self._fire_event_hooks(hit)
+            return hit
+
+        regs = self.client.read_registers()
+        pc = regs.get("pc", 0)
+        sym = self.symbols.resolve(pc)
+
+        if stop.watch_addr is not None:
+            wp = self._watches_by_addr.get(stop.watch_addr)
+            hit = Hit(
+                kind="watchpoint",
+                addr=stop.watch_addr,
+                regs=regs,
+                symbol=self.symbols.resolve(stop.watch_addr),
+                watch=wp,
+                stop=stop,
+                session=self,
+            )
+            if wp is not None:
+                wp.hits += 1
+                if wp.callback is not None:
+                    wp.callback(hit)
+            self.recorder.event(
+                "watch_hit",
+                addr=stop.watch_addr,
+                kind=wp.kind if wp else stop.watch_kind,
+                size=wp.size if wp else None,
+                symbol=hit.symbol.name if hit.symbol else None,
+                pc=pc,
+                regs=_json_regs(regs),
+            )
+            self._fire_event_hooks(hit)
+            # QEMU stops *at* the instruction that performed the watched
+            # access.  Step it once with the watchpoint temporarily cleared so
+            # resuming does not re-trigger on the same store forever.
+            try:
+                if wp is not None:
+                    self.client.clear_watch(wp.addr, size=wp.size, kind=wp.kind)
+                self.client.step()
+            except RspTimeout:
+                pass
+            finally:
+                if wp is not None:
+                    try:
+                        self.client.set_watch(wp.addr, size=wp.size, kind=wp.kind)
+                    except Exception:
+                        pass
+            return hit
+
+        hit = Hit(
+            kind="step" if from_step else "stop",
+            addr=pc,
+            regs=regs,
+            symbol=sym,
+            stop=stop,
+            session=self,
+        )
+
+        bps = list(self._bps_by_addr.get(pc, []))
+        leave_queue = self._leave_pending.get(pc, [])
+        if not bps and not leave_queue:
+            self.recorder.event(
+                "stop",
+                pc=pc,
+                symbol=sym.name if sym else None,
+                signal=stop.signal,
+                reason="step" if from_step else "halt",
+            )
+            self._fire_event_hooks(hit)
+            return hit
+
+        hit.kind = "breakpoint"
+        self.recorder.event(
+            "bp_hit",
+            pc=pc,
+            addr=pc,
+            symbol=sym.name if sym else None,
+            signal=stop.signal,
+            regs=_json_regs(regs),
+        )
+        for bp in bps:
+            bp.hits += 1
+            if bp.on_enter is not None:
+                bp.on_enter(hit)
+            if bp.on_leave is not None and not bp.temporary:
+                self._arm_leave(regs.get("ra", 0), bp)
+            if bp.temporary:
+                self.remove_breakpoint(bp)
+
+        # Return hooks: pop one pending hook for this return address.
+        if leave_queue:
+            bp = leave_queue.pop()
+            if not leave_queue:
+                self._leave_pending.pop(pc, None)
+                for candidate in list(self._bps_by_addr.get(pc, [])):
+                    if candidate.purpose == "leave":
+                        self.remove_breakpoint(candidate)
+            hit.kind = "leave"
+            hit.retval = regs.get("a0", 0)
+            if bp.symbol is not None:
+                hit.symbol = bp.symbol
+                hit.addr = bp.addr
+            if bp.on_leave is not None:
+                bp.on_leave(hit)
+
+        # Software breakpoints need a single-step over them before resuming.
+        self._step_over(pc)
+        self._fire_event_hooks(hit)
+        return hit
+
+    def _arm_leave(self, retaddr: int, bp: Breakpoint) -> None:
+        if not retaddr:
+            return
+        queue = self._leave_pending.get(retaddr)
+        if queue is None:
+            queue = []
+            self._leave_pending[retaddr] = queue
+            temp = Breakpoint(
+                addr=retaddr,
+                kind="sw",
+                temporary=True,
+                purpose="leave",
+                symbol=self.symbols.resolve(retaddr),
+            )
+            self.add_breakpoint(temp)
+        queue.append(bp)
+
+    def _step_over(self, pc: int) -> None:
+        sw = [bp for bp in self._bps_by_addr.get(pc, []) if bp.kind == "sw"]
+        if not sw:
+            return
+        for bp in sw:
+            self.client.clear_sw_break(pc)
+        self.client.step()
+        for bp in sw:
+            if bp.attached:
+                self.client.set_sw_break(pc)
+
+    def _fire_event_hooks(self, hit: Hit) -> None:
+        for kind, callback in self._event_hooks:
+            if kind in (hit.kind, "*"):
+                callback(hit)
+
+    # -- lifecycle ---------------------------------------------------------
+
+    def close(self, kill_target: bool | None = None) -> None:
+        try:
+            for bp in list(self._bps.values()):
+                self.remove_breakpoint(bp)
+            for wp in list(self._watches.values()):
+                self.unwatch(wp)
+        except Exception:
+            pass
+        self.client.close()
+        if kill_target is None:
+            kill_target = self._owns_target
+        if kill_target and self.target is not None and not getattr(self.target, "keep_alive", False):
+            self.target.kill()
+        self.recorder.close()
+        self._owns_target = False
+
+    def __enter__(self) -> "Session":
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self.close()
+
+
+def _json_regs(regs: dict[str, int]) -> dict[str, int]:
+    return {name: value for name, value in regs.items() if isinstance(value, int)}
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/structs.py b/debug/risc-v-gsp-rm-vm/greeneye/structs.py
new file mode 100644
index 0000000..3081bc9
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/structs.py
@@ -0,0 +1,197 @@
+"""Declarative structure decoding for firmware memory.
+
+A schema is plain data (JSON-friendly), so the same description can come from
+the repo, a Ghidra export, or a plugin:
+
+    {
+      "name": "gsp_ctrl",
+      "fields": [
+        {"name": "magic",   "offset": 0,  "fmt": "u32"},
+        {"name": "state",   "offset": 8,  "fmt": "u32", "enum": {"0": "IDLE", "1": "RUN"}},
+        {"name": "buf",     "offset": 64, "fmt": "bytes", "size": 256},
+        {"name": "name",    "offset": 64, "fmt": "cstr",  "size": 32}
+      ]
+    }
+
+Supported formats: u8/u16/u32/u64, s8/s16/s32/s64, f32/f64, bool, bytes,
+cstr, ptr (u64 by default; override with ``size``).
+"""
+from __future__ import annotations
+
+import struct as _struct
+from dataclasses import dataclass, field as _field
+
+from .errors import StructError
+
+_INT_FORMATS = {
+    "u8": ("B", 1),
+    "u16": ("H", 2),
+    "u32": ("I", 4),
+    "u64": ("Q", 8),
+    "s8": ("b", 1),
+    "s16": ("h", 2),
+    "s32": ("i", 4),
+    "s64": ("q", 8),
+    "bool": ("B", 1),
+}
+_FLOAT_FORMATS = {"f32": ("f", 4), "f64": ("d", 8)}
+
+
+@dataclass
+class Field:
+    name: str
+    offset: int
+    fmt: str
+    size: int = 0
+    enum: dict | None = None
+    note: str | None = None
+    count: int = 1
+
+    @property
+    def nbytes(self) -> int:
+        if self.fmt in _INT_FORMATS:
+            return _INT_FORMATS[self.fmt][1]
+        if self.fmt in _FLOAT_FORMATS:
+            return _FLOAT_FORMATS[self.fmt][1]
+        if self.fmt in ("bytes", "cstr"):
+            return self.size
+        if self.fmt == "ptr":
+            return self.size or 8
+        if self.fmt == "u64x":
+            return 8
+        raise StructError(f"field {self.name!r}: unknown format {self.fmt!r}")
+
+
+@dataclass
+class Schema:
+    name: str
+    fields: list[Field] = _field(default_factory=list)
+    size: int = 0
+
+    def __post_init__(self) -> None:
+        computed = max((f.offset + f.nbytes for f in self.fields), default=0)
+        if not self.size:
+            self.size = computed
+        elif self.size < computed:
+            raise StructError(
+                f"schema {self.name!r}: declared size {self.size:#x} < fields end {computed:#x}"
+            )
+        self.validate()
+
+    def validate(self) -> None:
+        seen: set[str] = set()
+        for field in self.fields:
+            if field.name in seen:
+                raise StructError(f"schema {self.name!r}: duplicate field {field.name!r}")
+            seen.add(field.name)
+            if field.offset < 0:
+                raise StructError(f"schema {self.name!r}: negative offset for {field.name!r}")
+            if field.nbytes <= 0:
+                raise StructError(f"schema {self.name!r}: zero-size field {field.name!r}")
+
+    def decode(self, data: bytes, base: int = 0) -> dict:
+        if len(data) < self.size:
+            raise StructError(
+                f"schema {self.name!r}: need {self.size} bytes, got {len(data)}"
+            )
+        out: dict = {}
+        for field in self.fields:
+            out[field.name] = _decode_field(field, data, base)
+        return out
+
+
+def _decode_field(field: Field, data: bytes, base: int):
+    start = field.offset
+    if field.fmt in _INT_FORMATS:
+        code, width = _INT_FORMATS[field.fmt]
+        value = _struct.unpack_from("<" + code, data, start)[0]
+        if field.fmt == "bool":
+            value = bool(value)
+        return value
+    if field.fmt in _FLOAT_FORMATS:
+        code, _width = _FLOAT_FORMATS[field.fmt]
+        return _struct.unpack_from("<" + code, data, start)[0]
+    if field.fmt == "ptr":
+        return int.from_bytes(data[start : start + field.nbytes], "little")
+    if field.fmt == "bytes":
+        return bytes(data[start : start + field.size])
+    if field.fmt == "cstr":
+        raw = data[start : start + field.size]
+        return raw.split(b"\x00", 1)[0].decode("utf-8", "replace")
+    raise StructError(f"unknown format {field.fmt!r}")
+
+
+def decode(schema: Schema, data: bytes, base: int = 0) -> dict:
+    return schema.decode(data, base=base)
+
+
+def schema_from_json(payload: dict) -> Schema:
+    if not isinstance(payload, dict) or "fields" not in payload:
+        raise StructError("schema JSON must be an object with a 'fields' list")
+    fields = []
+    for item in payload["fields"]:
+        try:
+            fields.append(
+                Field(
+                    name=item["name"],
+                    offset=int(item["offset"]),
+                    fmt=item.get("fmt", "bytes"),
+                    size=int(item.get("size", 0)),
+                    enum=item.get("enum"),
+                    note=item.get("note"),
+                    count=int(item.get("count", 1)),
+                )
+            )
+        except KeyError as exc:
+            raise StructError(f"schema field missing key: {exc}") from exc
+    return Schema(name=payload.get("name", "schema"), fields=fields, size=int(payload.get("size", 0)))
+
+
+def schema_to_json(schema: Schema) -> dict:
+    return {
+        "name": schema.name,
+        "size": schema.size,
+        "fields": [
+            {
+                k: v
+                for k, v in {
+                    "name": f.name,
+                    "offset": f.offset,
+                    "fmt": f.fmt,
+                    "size": f.size or None,
+                    "enum": f.enum,
+                    "note": f.note,
+                }.items()
+                if v is not None
+            }
+            for f in schema.fields
+        ],
+    }
+
+
+def render(schema: Schema, values: dict, indent: str = "  ") -> str:
+    lines = [f"{schema.name} ({{base}}+{schema.size:#x} bytes)"]
+    width = max((len(f.name) for f in schema.fields), default=0)
+    for field in schema.fields:
+        value = values.get(field.name)
+        lines.append(f"{indent}{field.name:<{width}} @+0x{field.offset:03x} = {_format_value(field, value)}")
+        if field.note:
+            lines.append(f"{indent}{'':<{width}}   ; {field.note}")
+    return "\n".join(lines)
+
+
+def _format_value(field: Field, value) -> str:
+    if isinstance(value, bool):
+        return str(value)
+    if isinstance(value, float):
+        return f"{value:g}"
+    if isinstance(value, int):
+        text = f"{value:#x}" if not field.enum else f"{value} ({field.enum.get(str(value), '?')})"
+        if field.enum and field.fmt not in ("bool",):
+            text = f"{value:#x} ({field.enum.get(str(value), field.enum.get(value, '?'))})"
+        return text
+    if isinstance(value, (bytes, bytearray)):
+        head = bytes(value[:16]).hex(" ")
+        suffix = " ..." if len(value) > 16 else ""
+        return f"[{len(value)}B] {head}{suffix}"
+    return repr(value)
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/symbols.py b/debug/risc-v-gsp-rm-vm/greeneye/symbols.py
new file mode 100644
index 0000000..7b2e9dd
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/symbols.py
@@ -0,0 +1,313 @@
+"""Symbol resolution and address translation for green-eye.
+
+Inputs supported (all optional, merged into one database):
+
+* ELF ``.symtab`` symbols (demo firmware);
+* the project's Ghidra exports:
+  - ``names.tsv``   ``<file_vaddr>\t<name>[\t<source>]`` (comments with #),
+  - ``decompiled_index.tsv`` / bucket TSVs
+    ``<program_addr>\t<file_addr>\t<size>\t<name>\t<bucket>\t<line>\t<status>``,
+  - ``exports.tsv``  ``addr\tfile_addr\tsize\tname``,
+* green-eye native JSON (``greeneye-symbols.json``).
+
+Address conventions (matching the project docs):
+  * Ghidra program address = file vaddr + image base (``0x100000`` for libcuda);
+  * runtime address        = file vaddr + runtime load bias;
+  * for firmware images linked at their load address the bias is 0.
+
+``SymbolDB`` stores runtime addresses by default; each loader takes a ``bias``
+so a relocated firmware or a different image base can be normalised.
+"""
+from __future__ import annotations
+
+import bisect
+import json
+import os
+from dataclasses import dataclass, field as _field
+
+from .errors import SymbolNotFound
+
+
+@dataclass
+class Symbol:
+    addr: int
+    name: str
+    size: int = 0
+    source: str = ""
+    subsystem: str | None = None
+    extras: dict = _field(default_factory=dict)
+
+    @property
+    def end(self) -> int:
+        return self.addr + self.size if self.size else self.addr
+
+    def contains(self, addr: int) -> bool:
+        if self.size:
+            return self.addr <= addr < self.addr + self.size
+        return addr == self.addr
+
+    def label(self, addr: int | None = None) -> str:
+        if addr is None or addr == self.addr or not self.size:
+            return self.name
+        return f"{self.name}+{addr - self.addr:#x}"
+
+    def to_json(self) -> dict:
+        out = {"addr": self.addr, "name": self.name}
+        if self.size:
+            out["size"] = self.size
+        if self.source:
+            out["source"] = self.source
+        if self.subsystem:
+            out["subsystem"] = self.subsystem
+        if self.extras:
+            out.update(self.extras)
+        return out
+
+
+class SymbolDB:
+    def __init__(self) -> None:
+        self._symbols: list[Symbol] = []
+        self._sorted = False
+        self._by_name: dict[str, Symbol] = {}
+        self._addr_index: list[int] = []
+
+    # -- loading -----------------------------------------------------------
+
+    def add(self, symbol: Symbol) -> None:
+        self._symbols.append(symbol)
+        self._sorted = False
+        existing = self._by_name.get(symbol.name)
+        if existing is None or (symbol.source and not existing.source):
+            self._by_name[symbol.name] = symbol
+
+    def load_elf(self, image, bias: int = 0, source: str = "elf", only_code: bool = True) -> int:
+        """Load ``.symtab`` symbols from a FirmwareImage/ELF path."""
+        from .elf import ElfSymbol, load_elf
+
+        if isinstance(image, str):
+            image = load_elf(image)
+        count = 0
+        for sym in image.symbols:  # type: ignore[union-attr]
+            if only_code and not (sym.is_function or sym.is_object):
+                continue
+            self.add(
+                Symbol(
+                    addr=sym.value + bias,
+                    name=sym.name,
+                    size=sym.size,
+                    source=source,
+                    extras={"type": "func" if sym.is_function else "object"},
+                )
+            )
+            count += 1
+        return count
+
+    def load_names_tsv(self, path: str, bias: int = 0) -> int:
+        """Load the project's Ghidra ``names.tsv`` (file vaddrs)."""
+        count = 0
+        with open(path, "r", encoding="utf-8", errors="replace") as fh:
+            for line in fh:
+                line = line.rstrip("\n")
+                if not line or line.startswith("#"):
+                    continue
+                cols = line.split("\t")
+                if len(cols) < 2:
+                    continue
+                try:
+                    addr = int(cols[0], 16)
+                except ValueError:
+                    continue
+                self.add(
+                    Symbol(addr=addr + bias, name=cols[1], source=cols[2] if len(cols) > 2 else "ghidra-names")
+                )
+                count += 1
+        return count
+
+    def load_index_tsv(self, path: str, bias: int = 0) -> int:
+        """Load ``decompiled_index.tsv``/bucket TSVs/exports.tsv heuristically."""
+        count = 0
+        with open(path, "r", encoding="utf-8", errors="replace") as fh:
+            for line in fh:
+                line = line.rstrip("\n")
+                if not line or line.startswith("#"):
+                    continue
+                cols = line.split("\t")
+                if cols[0] in ("addr", "caller"):
+                    continue
+                if len(cols) >= 7:
+                    program, file_addr, size, name, bucket, lineno, status = cols[:7]
+                    addr = _parse_hex(file_addr)
+                    extras = {"program_addr": _parse_hex(program), "bucket": bucket, "line": lineno, "status": status}
+                elif len(cols) >= 4:
+                    program, file_addr, size, name = cols[:4]
+                    addr = _parse_hex(file_addr)
+                    extras = {"program_addr": _parse_hex(program)}
+                elif len(cols) >= 2:
+                    addr = _parse_hex(cols[0])
+                    name = cols[1]
+                    size = "0"
+                    extras = {}
+                else:
+                    continue
+                if addr is None or not name:
+                    continue
+                try:
+                    size_val = int(size, 0)
+                except ValueError:
+                    size_val = 0
+                self.add(
+                    Symbol(addr=addr + bias, name=name, size=size_val, source="ghidra-index", extras=extras)
+                )
+                count += 1
+        return count
+
+    def load_callgraph(self, path: str) -> int:
+        """Attach caller/callee lists from ``callgraph.tsv`` (program addrs)."""
+        by_program: dict[int, Symbol] = {}
+        for sym in self._symbols:
+            prog = sym.extras.get("program_addr")
+            if prog is not None:
+                by_program[int(prog)] = sym
+        if not by_program:
+            return 0
+        for sym in self._symbols:
+            sym.extras.setdefault("callers", [])
+            sym.extras.setdefault("callees", [])
+        count = 0
+        with open(path, "r", encoding="utf-8", errors="replace") as fh:
+            for line in fh:
+                line = line.rstrip("\n")
+                if not line or line.startswith("#") or line.startswith("caller"):
+                    continue
+                cols = line.split("\t")
+                if len(cols) < 2:
+                    continue
+                caller, callee = _parse_hex(cols[0]), _parse_hex(cols[1])
+                if caller is None or callee is None:
+                    continue
+                if caller in by_program:
+                    by_program[caller].extras["callees"].append(callee)
+                    count += 1
+                if callee in by_program:
+                    by_program[callee].extras["callers"].append(caller)
+        return count
+
+    def load_json(self, path: str) -> int:
+        with open(path, "r", encoding="utf-8") as fh:
+            payload = json.load(fh)
+        symbols = payload.get("symbols", payload if isinstance(payload, list) else [])
+        count = 0
+        for item in symbols:
+            try:
+                addr = int(item["addr"], 0) if isinstance(item["addr"], str) else int(item["addr"])
+            except (KeyError, TypeError, ValueError):
+                continue
+            extras = {k: v for k, v in item.items() if k not in ("addr", "name", "size", "source", "subsystem")}
+            self.add(
+                Symbol(
+                    addr=addr,
+                    name=item.get("name", f"sub_{addr:x}"),
+                    size=int(item.get("size", 0)),
+                    source=item.get("source", "json"),
+                    subsystem=item.get("subsystem"),
+                    extras=extras,
+                )
+            )
+            count += 1
+        return count
+
+    def save_json(self, path: str, meta: dict | None = None) -> None:
+        self._ensure_sorted()
+        payload = {
+            "format": "greeneye-symbols/1",
+            "count": len(self._symbols),
+            "meta": meta or {},
+            "symbols": [s.to_json() for s in self._symbols],
+        }
+        with open(path, "w", encoding="utf-8") as fh:
+            json.dump(payload, fh, indent=1, sort_keys=True)
+
+    # -- queries -----------------------------------------------------------
+
+    def _ensure_sorted(self) -> None:
+        if self._sorted:
+            return
+        self._symbols.sort(key=lambda s: (s.addr, -s.size))
+        self._addr_index = [s.addr for s in self._symbols]
+        self._sorted = True
+
+    def __len__(self) -> int:
+        return len(self._symbols)
+
+    def __iter__(self):
+        return iter(self._symbols)
+
+    def get(self, name: str) -> Symbol | None:
+        return self._by_name.get(name)
+
+    def lookup(self, name: str) -> int:
+        sym = self._by_name.get(name)
+        if sym is None:
+            raise SymbolNotFound(f"symbol {name!r} not found")
+        return sym.addr
+
+    def resolve(self, addr: int, max_delta: int = 0x10000) -> Symbol | None:
+        """Resolve an address to a containing or nearest preceding symbol."""
+        self._ensure_sorted()
+        if not self._symbols:
+            return None
+        pos = bisect.bisect_right(self._addr_index, addr) - 1
+        if pos < 0:
+            return None
+        candidate = self._symbols[pos]
+        if candidate.contains(addr):
+            return candidate
+        if candidate.size and addr < candidate.addr + candidate.size:
+            return candidate
+        # nearest-within-delta fallback for zero-size symbols (Ghidra names)
+        if not candidate.size and 0 <= addr - candidate.addr <= max_delta:
+            return candidate
+        return None
+
+    def annotate(self, addr: int, max_delta: int = 0x10000) -> str:
+        sym = self.resolve(addr, max_delta=max_delta)
+        if sym is None:
+            return f"{addr:#x} <unknown>"
+        return f"{addr:#x} {sym.label(addr)}"
+
+    def nearest(self, addr: int, max_delta: int = 0x10000) -> Symbol | None:
+        return self.resolve(addr, max_delta=max_delta)
+
+    def names(self) -> list[str]:
+        return sorted(self._by_name)
+
+    def by_subsystem(self, subsystem: str) -> list[Symbol]:
+        return [s for s in self._symbols if s.subsystem == subsystem]
+
+
+def _parse_hex(value: str) -> int | None:
+    value = value.strip()
+    if not value:
+        return None
+    try:
+        return int(value, 16)
+    except ValueError:
+        try:
+            return int(value, 0)
+        except ValueError:
+            return None
+
+
+def translation_for(
+    *,
+    file_vaddr: int,
+    program_addr: int | None = None,
+    runtime_addr: int | None = None,
+) -> dict:
+    """Explain the three address spaces for a debug session (docs aid)."""
+    out = {"file_vaddr": file_vaddr}
+    if program_addr is not None:
+        out["image_base"] = program_addr - file_vaddr
+    if runtime_addr is not None:
+        out["load_bias"] = runtime_addr - file_vaddr
+    return out
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/target.py b/debug/risc-v-gsp-rm-vm/greeneye/target.py
new file mode 100644
index 0000000..efcb1d1
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/target.py
@@ -0,0 +1,223 @@
+"""QEMU target lifecycle for green-eye.
+
+Owns the ``qemu-system-riscv64`` process, the gdbstub endpoint, the serial
+capture and the QEMU trace log.  Both firmware paths are supported:
+
+* ``kernel=`` fast path - QEMU loads an ELF linked for RAM and sets the PC;
+* ``segments=`` probe path - each PT_LOAD segment is placed with a loader
+  device (raw), so a firmware linked for GPU physical addresses can be
+  relocated into RAM and observed.
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import signal
+import socket
+import subprocess
+import tempfile
+import time
+
+from .errors import TargetError
+
+DEFAULT_QEMU = "qemu-system-riscv64"
+
+
+def find_qemu() -> str:
+    env = os.environ.get("GREENEYE_QEMU")
+    if env:
+        return env
+    found = shutil.which(DEFAULT_QEMU)
+    if not found:
+        raise TargetError(
+            f"{DEFAULT_QEMU} not found. Install it (Ubuntu: qemu-system-misc) "
+            "or set GREENEYE_QEMU to the binary."
+        )
+    return found
+
+
+def free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+class QemuTarget:
+    def __init__(
+        self,
+        *,
+        machine: str = "virt",
+        memory_mb: int = 256,
+        gdb_port: int = 0,
+        bios: str = "none",
+        extra_args: list[str] | None = None,
+        qemu: str | None = None,
+        kernel: str | None = None,
+        segments: list[tuple[int, bytes]] | None = None,
+        dlog: str | None = None,
+        uart_log: str | None = None,
+        qemu_log: str | None = None,
+        workdir: str | None = None,
+        keep_alive: bool = False,
+    ):
+        self.qemu = qemu or find_qemu()
+        self.machine = machine
+        self.memory_mb = memory_mb
+        self.host = "127.0.0.1"
+        self.port = gdb_port or free_port()
+        self.bios = bios
+        self.extra_args = list(extra_args or [])
+        self.kernel = os.path.abspath(kernel) if kernel else None
+        self.segments = list(segments or [])
+        self.dlog = dlog
+        self.workdir = workdir or tempfile.mkdtemp(prefix="greeneye-qemu-")
+        os.makedirs(self.workdir, exist_ok=True)
+        self.uart_log = uart_log or os.path.join(self.workdir, "uart.log")
+        self.qemu_log = qemu_log or os.path.join(self.workdir, "qemu.log")
+        self.keep_alive = keep_alive
+        self.process: subprocess.Popen | None = None
+        self._segment_files: list[str] = []
+        self._cmd: list[str] = []
+
+    # -- lifecycle ---------------------------------------------------------
+
+    def _build_command(self, paused: bool) -> list[str]:
+        cmd = [
+            self.qemu,
+            "-machine",
+            self.machine,
+            "-m",
+            f"{self.memory_mb}M",
+            "-display",
+            "none",
+            "-monitor",
+            "none",
+            "-serial",
+            f"file:{self.uart_log}",
+            "-gdb",
+            f"tcp:{self.host}:{self.port}",
+        ]
+        if self.bios:
+            cmd += ["-bios", self.bios]
+        if paused:
+            cmd += ["-S"]
+        if self.kernel:
+            cmd += ["-kernel", self.kernel]
+        for index, (addr, blob) in enumerate(self.segments):
+            seg_path = os.path.join(self.workdir, f"segment-{index:02d}-{addr:#x}.bin")
+            with open(seg_path, "wb") as fh:
+                fh.write(blob)
+            self._segment_files.append(seg_path)
+            cmd += ["-device", f"loader,file={seg_path},addr={addr:#x},force-raw=on"]
+        if self.dlog:
+            cmd += ["-d", self.dlog, "-D", self.qemu_log]
+        cmd += self.extra_args
+        return cmd
+
+    def start(self, paused: bool = True, timeout: float = 15.0) -> "QemuTarget":
+        self._cmd = self._build_command(paused)
+        try:
+            self.process = subprocess.Popen(
+                self._cmd,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.PIPE,
+                cwd=self.workdir,
+                start_new_session=True,
+            )
+        except OSError as exc:
+            raise TargetError(f"cannot start QEMU: {exc}") from exc
+        try:
+            self.wait_for_port(timeout=timeout)
+        except TargetError:
+            self.kill()
+            raise
+        return self
+
+    def wait_for_port(self, timeout: float = 15.0) -> None:
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if self.process is not None and self.process.poll() is not None:
+                stderr = b""
+                if self.process.stderr is not None:
+                    try:
+                        stderr = self.process.stderr.read() or b""
+                    except OSError:
+                        pass
+                raise TargetError(
+                    f"QEMU exited early rc={self.process.returncode}: "
+                    f"{stderr.decode('utf-8', 'replace')[-2000:]}"
+                )
+            try:
+                with socket.create_connection((self.host, self.port), timeout=0.25):
+                    return
+            except OSError:
+                time.sleep(0.05)
+        raise TargetError(f"gdbstub port {self.port} not ready after {timeout}s")
+
+    def alive(self) -> bool:
+        return self.process is not None and self.process.poll() is None
+
+    def kill(self, timeout: float = 5.0) -> None:
+        if self.process is None:
+            return
+        if self.process.poll() is None:
+            try:
+                os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
+            except (ProcessLookupError, PermissionError):
+                try:
+                    self.process.terminate()
+                except ProcessLookupError:
+                    pass
+            try:
+                self.process.wait(timeout=timeout)
+            except subprocess.TimeoutExpired:
+                try:
+                    os.killpg(os.getpgid(self.process.pid), signal.SIGKILL)
+                except (ProcessLookupError, PermissionError):
+                    try:
+                        self.process.kill()
+                    except ProcessLookupError:
+                        pass
+                self.process.wait(timeout=timeout)
+        if self.process.stderr is not None:
+            try:
+                self.process.stderr.close()
+            except OSError:
+                pass
+        self.process = None
+
+    # -- observation -------------------------------------------------------
+
+    def read_uart(self) -> str:
+        try:
+            with open(self.uart_log, "rb") as fh:
+                return fh.read().decode("utf-8", "replace")
+        except FileNotFoundError:
+            return ""
+
+    def wait_for_uart(self, marker: str, timeout: float = 5.0, interval: float = 0.05) -> bool:
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if marker in self.read_uart():
+                return True
+            time.sleep(interval)
+        return False
+
+    def read_qemu_log(self) -> str:
+        try:
+            with open(self.qemu_log, "rb") as fh:
+                return fh.read().decode("utf-8", "replace")
+        except FileNotFoundError:
+            return ""
+
+    @property
+    def command_line(self) -> list[str]:
+        return list(self._cmd)
+
+    def shell_command(self) -> str:
+        import shlex
+
+        return " ".join(shlex.quote(part) for part in self._cmd)
+
+    def __repr__(self) -> str:  # pragma: no cover - cosmetic
+        return f"QemuTarget(port={self.port}, alive={self.alive()})"