debug: green-eye wave 3 - RM harness, local simulator, device map, MCP plugin, value-validation demos, CI pipeline, toolchain docs; add host RM tools rmtool/rmtop

1 files changed, 946 insertions, 0 deletions

diff --git a/debug/risc-v-gsp-rm-vm/harness/reach.py b/debug/risc-v-gsp-rm-vm/harness/reach.py
new file mode 100644
index 0000000..9df0c4c
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/reach.py
@@ -0,0 +1,946 @@
+#!/usr/bin/env python3
+"""Reachability answers for a decompiled firmware target (H4.1/H4.5).
+
+Given a symbol name or address taken from Ghidra or the symbol DB, answer the
+practical questions a reimplementation needs: *is this code reachable, how do
+I trigger it, what arguments does it need, and what happens when it runs?*
+
+Methods attempted (``--method auto`` tries them in this order and stops at the
+first hit unless ``--all-methods`` is given):
+
+* ``stimulus`` -- run a supplied green-eye script (``--stimulus script.py``)
+  or probe spec (``--probe probe.json``) under instrumentation;
+* ``trigger``  -- discover the dispatch entry that routes to the target (the
+  demo's handler table), apply that command, and run bounded;
+* ``run``      -- ``Session.run_until(target)`` from launch, which catches the
+  natural boot/self-test path;
+* ``call``     -- ``Session.call(target, args)`` with ``--args`` and
+  ``--arg-count`` auto-fill (zeros or sentinels).
+
+On a hit the report records the entry register state, the prior hook events
+that led there, the caller, and the effects after resume (field deltas and
+watchpoint stores, optional recorded-session trace summary). A repeatable
+probe spec is written to ``--emit-probe`` (default ``<out>/probe.json``) so a
+later run (or docs/21 CI) can replay the exact route.
+
+Demonstrated on the demo firmware: ``rpc_ping`` via ``call``, ``rpc_sum`` via
+``run`` (self-test), ``rpc_fault`` via ``trigger`` (mailbox 6).  See
+docs/19-CONTROL-DISCOVERY.md.
+
+Run from the tool root:
+
+    python3 harness/reach.py --target rpc_fault --method trigger --out dumps/reach-fault
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import shutil
+import sys
+import time
+from collections import Counter
+from dataclasses import dataclass, field
+from pathlib import Path
+
+try:
+    from .common import DEMO_ELF, ensure_dir, qemu_available, read_faults_window, sha256_file
+    from .control_discovery import DemoSurface, SETTLE_MARKER, SETTLE_SLICE
+except ImportError:  # direct script execution (python3 harness/reach.py)
+    from common import DEMO_ELF, ensure_dir, qemu_available, read_faults_window, sha256_file  # type: ignore[no-redef]
+    from control_discovery import DemoSurface, SETTLE_MARKER, SETTLE_SLICE  # type: ignore[no-redef]
+
+from greeneye import launch  # noqa: E402
+from greeneye.errors import GreenEyeError, RspTimeout  # noqa: E402
+from greeneye.record import load_events  # noqa: E402
+from greeneye.symbols import SymbolDB  # noqa: E402
+
+FORMAT = "greeneye-reach-report/1"
+PROBE_FORMAT = "greeneye-reach-probe/1"
+SENTINEL_BASE = 0xA5A5_0000_0000_0000
+RUN_SLICE = 0.02
+GRACE = 0.10
+AUTO_ORDER = ("stimulus", "trigger", "run", "call")
+TRAP_SIGNALS = (5,)
+FAULT_SIGNALS = (4, 7, 11)
+
+
+@dataclass
+class ReachOptions:
+    kernel: str = DEMO_ELF
+    out: str = ""
+    target: str = ""
+    method: str = "auto"
+    args: list[int] = field(default_factory=list)
+    arg_count: int = 0
+    arg_fill: str = "zero"
+    timeout: float = 0.5
+    settle: float = 2.0
+    stimulus: str | None = None
+    probe: str | None = None
+    emit_probe: str | None = None
+    record: str | None = None
+    backend: str = "qemu"
+    max_prior: int = 12
+    all_methods: bool = False
+    allow_foreign_elf: bool = False
+    watch: list[dict] = field(default_factory=list)
+
+
+@dataclass
+class Attempt:
+    method: str
+    ok: bool = False
+    reason: str | None = None
+    entry: dict | None = None
+    retval: int | None = None
+    dispatch: list[dict] = field(default_factory=list)
+    prior: list[dict] = field(default_factory=list)
+    effects: dict = field(default_factory=dict)
+    trap: bool = False
+    fault: bool = False
+    trigger: dict | None = None
+    session: str | None = None
+    hit_count: int = 0
+
+    def summary(self) -> dict:
+        return {
+            "method": self.method,
+            "ok": self.ok,
+            "reason": self.reason,
+            "entry": self.entry,
+            "retval": self.retval,
+            "dispatch": self.dispatch[:4],
+            "prior_events": self.prior,
+            "effects": self.effects,
+            "trap": self.trap,
+            "fault": self.fault,
+            "trigger": self.trigger,
+            "session": self.session,
+            "hit_count": self.hit_count,
+        }
+
+
+def build_args(options: ReachOptions) -> list[int]:
+    """Explicit --args plus deterministic --arg-count auto-fill."""
+    args = list(options.args)
+    count = max(options.arg_count, len(args))
+    if count > 8:
+        raise GreenEyeError(f"--arg-count {count}: the RISC-V ABI passes at most 8 integer args")
+    while len(args) < count:
+        if options.arg_fill == "sentinel":
+            args.append(SENTINEL_BASE | len(args))
+        else:
+            args.append(0)
+    return args
+
+
+class Collector:
+    """Ordered hook/watch evidence for one reachability attempt."""
+
+    def __init__(
+        self,
+        session,
+        surface: DemoSurface | None,
+        target_addr: int,
+        target_name: str,
+        options: ReachOptions,
+        fault_log: str | None = None,
+    ):
+        self.session = session
+        self.surface = surface
+        self.target_addr = target_addr
+        self.target_name = target_name
+        self.options = options
+        self.fault_log = fault_log
+        self.fault_offset = 0
+        self.faults: list[dict] = []
+        self.hits: list[dict] = []
+        self.entry: dict | None = None
+        self.prior: list[dict] = []
+        self.dispatch: list[dict] = []
+        self.signals: list[dict] = []
+        self.watch: list[dict] = []
+        self.observer = surface.observer() if surface is not None else None
+        self.before: dict = {}
+        self.base = 0
+
+    # -- installation ------------------------------------------------------
+
+    def install(self) -> None:
+        self.session.on_event("stop", self._on_stop)
+        if self.fault_log and os.path.exists(self.fault_log):
+            self.fault_offset = os.path.getsize(self.fault_log)
+        if self.surface is not None and self.session.symbols.get("gsp_ctrl"):
+            self.base = self.session.symbols.lookup("gsp_ctrl")
+            if self.observer is not None:
+                self.observer.reset()
+                self.observer.arm(self.session, self.base)
+            for spec in self.options.watch:
+                self._install_watch(spec)
+        hooked: set[int] = set()
+        self._hook_addr(self.target_addr, self.target_name)
+        hooked.add(self.target_addr)
+        if self.surface is not None:
+            command_hook = self.surface.command_hook(self.session)
+            if command_hook:
+                addr = self.session.symbols.lookup(command_hook)
+                self._hook_addr(addr, command_hook, stage="dispatch")
+                hooked.add(addr)
+            for cmd, (entry, name) in sorted(self.surface.handler_map(self.session).items()):
+                if entry in hooked:
+                    continue
+                self._hook_addr(entry, name or f"handler-{entry:#x}", stage="handler", cmd=cmd)
+                hooked.add(entry)
+
+    def _install_watch(self, spec: dict) -> None:
+        target = spec.get("symbol", spec.get("addr"))
+        if target is None:
+            return
+        if isinstance(target, str) and not target.lower().startswith("0x"):
+            base = self.session.symbols.lookup(target)
+        else:
+            base = int(target, 0) if isinstance(target, str) else int(target)
+        offset = int(spec.get("offset", 0))
+        size = int(spec.get("size", 8))
+        kind = spec.get("kind", "w")
+        label = spec.get("name") or f"{target}+{offset:#x}"
+
+        def callback(hit, label=label):
+            self.watch.append({"name": label, "pc": hit.pc, "symbol": hit.symbol.name if hit.symbol else None})
+
+        self.session.watch(base + offset, size=size, kind=kind, callback=callback)
+
+    def _hook_addr(self, addr: int, name: str, stage: str = "handler", cmd: int | None = None) -> None:
+        def callback(hit, stage=stage, cmd=cmd):
+            item = {
+                "stage": stage,
+                "symbol": hit.symbol.name if hit.symbol else name,
+                "addr": hit.addr,
+                "args": [hit.arg(i) for i in range(4)],
+            }
+            if stage == "dispatch":
+                self.dispatch.append(item)
+            if hit.addr == self.target_addr:
+                self.entry = {
+                    "symbol": hit.symbol.name if hit.symbol else name,
+                    "pc": hit.addr,
+                    "args": [hit.arg(i) for i in range(4)],
+                    "ra": hit.regs.get("ra", 0),
+                    "sp": hit.regs.get("sp", 0),
+                    "cmd": cmd,
+                    # The demo compiler's jump table clobbers a0 at handler entry;
+                    # the dispatcher's own args are the authoritative call model.
+                    "dispatch_args": self.dispatch[-1]["args"] if self.dispatch else None,
+                }
+                self.prior = list(self.hits[-self.options.max_prior :])
+            self.hits.append(item)
+
+        self.session.hook(addr, on_enter=callback, listener=f"reach:{name}")
+
+    def _on_stop(self, hit) -> None:
+        stop = hit.stop
+        if stop is not None and stop.signal is not None:
+            self.signals.append({"signal": stop.signal, "pc": hit.pc})
+
+    # -- accounting --------------------------------------------------------
+
+    def snapshot_before(self) -> None:
+        if self.observer is not None:
+            self.before = self.observer.snapshot(self.session, self.base)
+
+    def effects(self, retval: int | None) -> dict:
+        before = self.before
+        after = self.observer.snapshot(self.session, self.base) if self.observer is not None else {}
+        changed = [
+            {"field": name, "before": before.get(name), "after": after.get(name)}
+            for name in before
+            if before.get(name) != after.get(name)
+        ]
+        faults: list[dict] = []
+        if self.fault_log:
+            faults, _capped = read_faults_window(self.fault_log, self.fault_offset)
+        causes = [int(fault.get("cause", 0)) for fault in faults]
+        trap = any(cause == 3 for cause in causes) or any(
+            sig.get("signal") in TRAP_SIGNALS for sig in self.signals
+        ) or self.session.pc < 0x100
+        # A trap often drags a fetch-fault loop behind it (mtvec=0); do not
+        # double-label the consequence as an independent fault.
+        fault = (not trap) and (
+            any(cause != 3 for cause in causes)
+            or any(sig.get("signal") in FAULT_SIGNALS for sig in self.signals)
+        )
+        effects = {
+            "changed": changed,
+            "watch": list(self.watch) + (list(self.observer.hits) if self.observer is not None else []),
+            "retval": retval,
+            "trap": bool(trap),
+            "fault": bool(fault),
+            "pc_after": self.session.pc,
+        }
+        if faults:
+            effects["faults"] = [
+                {"cause": fault.get("cause"), "desc": fault.get("desc"), "epc": fault.get("epc"),
+                 "access": fault.get("access")}
+                for fault in faults[:4]
+            ]
+        return effects
+
+
+# ---------------------------------------------------------------------------
+# session plumbing
+# ---------------------------------------------------------------------------
+
+
+class SessionHandle:
+    def __init__(self, session, record_path: str | None, fault_log: str | None = None):
+        self.session = session
+        self.path = record_path
+        self.fault_log = fault_log
+
+
+def open_session(options: ReachOptions, method: str, record: bool) -> SessionHandle:
+    record_path = os.path.join(options.record, method) if record and options.record else None
+    if record_path:
+        ensure_dir(record_path)
+    fault_log = None
+    extra_args = None
+    if options.backend == "qemu":
+        fault_log = os.path.join(options.out, ".faults", f"{method}.log")
+        ensure_dir(os.path.dirname(fault_log))
+        extra_args = ["-d", "guest_errors,unimp,int", "-D", fault_log]
+    session = launch(
+        kernel=options.kernel,
+        backend=options.backend,
+        name=f"reach-{method}",
+        record=record_path,
+        extra_args=extra_args,
+    )
+    return SessionHandle(session, record_path, fault_log)
+
+
+def settle(session, budget: float) -> bool:
+    deadline = time.monotonic() + budget
+    while time.monotonic() < deadline:
+        if SETTLE_MARKER in session.target.read_uart():
+            return True
+        session.run(duration=SETTLE_SLICE, max_stops=1)
+    return SETTLE_MARKER in session.target.read_uart()
+
+
+def run_window(session, collector: Collector, timeout: float) -> None:
+    """Run bounded until the target entry is hit, then a short grace window."""
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if collector.entry is not None:
+            break
+        if not session.target.alive():
+            return
+        remaining = max(0.005, min(RUN_SLICE, deadline - time.monotonic()))
+        session.run(duration=remaining, max_stops=64)
+    if collector.entry is not None:
+        session.run(duration=min(GRACE, max(0.01, timeout)), max_stops=64)
+
+
+# ---------------------------------------------------------------------------
+# methods
+# ---------------------------------------------------------------------------
+
+
+def attempt_call(options: ReachOptions, surface: DemoSurface | None, target_addr: int, target_name: str,
+                 record: bool) -> Attempt:
+    attempt = Attempt(method="call")
+    args = build_args(options)
+    handle = open_session(options, "call", record)
+    session = handle.session
+    attempt.session = handle.path
+    try:
+        collector = Collector(session, surface, target_addr, target_name, options, fault_log=handle.fault_log)
+        collector.install()
+        collector.snapshot_before()
+        retval: int | None = None
+        try:
+            retval = session.call(target_addr, args, timeout=options.timeout)
+        except RspTimeout as exc:
+            attempt.reason = f"call did not return: {exc}"
+        except GreenEyeError as exc:
+            attempt.reason = f"call failed: {exc}"
+        attempt.entry = collector.entry
+        attempt.prior = collector.prior
+        attempt.dispatch = collector.dispatch
+        attempt.hit_count = len(collector.hits)
+        attempt.retval = retval
+        attempt.effects = collector.effects(retval)
+        attempt.trap = attempt.effects["trap"]
+        attempt.fault = attempt.effects["fault"]
+        attempt.ok = collector.entry is not None
+        if not attempt.ok and attempt.reason is None:
+            attempt.reason = "target entry hook never fired"
+    finally:
+        session.close()
+    return attempt
+
+
+def attempt_run(options: ReachOptions, surface: DemoSurface | None, target_addr: int, target_name: str,
+                record: bool) -> Attempt:
+    attempt = Attempt(method="run")
+    handle = open_session(options, "run", record)
+    session = handle.session
+    attempt.session = handle.path
+    try:
+        collector = Collector(session, surface, target_addr, target_name, options, fault_log=handle.fault_log)
+        collector.install()
+        collector.snapshot_before()
+        try:
+            session.run_until(target_addr, timeout=options.timeout)
+        except RspTimeout as exc:
+            attempt.reason = f"run_until not reached: {exc}"
+        except GreenEyeError as exc:
+            attempt.reason = f"run_until failed: {exc}"
+        if collector.entry is not None:
+            session.run(duration=min(GRACE, options.timeout), max_stops=64)
+        attempt.entry = collector.entry
+        attempt.prior = collector.prior
+        attempt.dispatch = collector.dispatch
+        attempt.hit_count = len(collector.hits)
+        attempt.effects = collector.effects(None)
+        attempt.trap = attempt.effects["trap"]
+        attempt.fault = attempt.effects["fault"]
+        attempt.ok = collector.entry is not None
+        if not attempt.ok and attempt.reason is None:
+            attempt.reason = "target never executed on the boot path"
+    finally:
+        session.close()
+    return attempt
+
+
+def attempt_trigger(options: ReachOptions, surface: DemoSurface | None, target_addr: int, target_name: str,
+                    record: bool) -> Attempt:
+    attempt = Attempt(method="trigger")
+    if surface is None:
+        attempt.reason = "no command surface available (gsp_ctrl/rpc_dispatch missing)"
+        return attempt
+    args = build_args(options)[:3]
+    while len(args) < 3:
+        args.append(0)
+    handle = open_session(options, "trigger", record)
+    session = handle.session
+    attempt.session = handle.path
+    try:
+        if not settle(session, options.settle):
+            attempt.reason = f"{SETTLE_MARKER!r} not observed; trigger state unknown"
+            return attempt
+        candidates = surface.command_for_addr(session, target_addr)
+        if not candidates:
+            attempt.reason = "target is not in the declared handler/dispatch table"
+            return attempt
+        # Prefer the documented switch case over the injected-slot alias.
+        candidates.sort(key=lambda item: (item["id"] >= 16, item["id"]))
+        chosen = candidates[0]
+        collector = Collector(session, surface, target_addr, target_name, options, fault_log=handle.fault_log)
+        collector.install()
+        collector.snapshot_before()
+        surface.probe(session, int(chosen["id"]), (args[0], args[1], args[2]), timeout=options.timeout)
+        run_window(session, collector, options.timeout)
+        attempt.entry = collector.entry
+        attempt.prior = collector.prior
+        attempt.dispatch = collector.dispatch
+        attempt.hit_count = len(collector.hits)
+        attempt.effects = collector.effects(None)
+        attempt.trap = attempt.effects["trap"]
+        attempt.fault = attempt.effects["fault"]
+        attempt.ok = collector.entry is not None
+        attempt.trigger = {"surface": surface.name, "id": chosen["id"], "handler": chosen["handler"], "args": args,
+                           "candidates": candidates}
+        if not attempt.ok and attempt.reason is None:
+            attempt.reason = f"mailbox command {chosen['id']} did not reach the target within the budget"
+    finally:
+        session.close()
+    return attempt
+
+
+def attempt_stimulus(options: ReachOptions, surface: DemoSurface | None, target_addr: int, target_name: str,
+                     record: bool) -> Attempt:
+    attempt = Attempt(method="stimulus")
+    if not options.stimulus:
+        attempt.reason = "no --stimulus script supplied"
+        return attempt
+    handle = open_session(options, "stimulus", record)
+    session = handle.session
+    attempt.session = handle.path
+    try:
+        collector = Collector(session, surface, target_addr, target_name, options, fault_log=handle.fault_log)
+        collector.install()
+        collector.snapshot_before()
+        import greeneye as ge
+
+        script_path = Path(options.stimulus)
+        namespace = {
+            "ge": ge,
+            "greeneye": ge,
+            "session": session,
+            "s": session,
+            "memory": session.memory,
+            "symbols": session.symbols,
+            "recorder": session.recorder,
+        }
+        try:
+            exec(compile(script_path.read_text(encoding="utf-8"), str(script_path), "exec"), namespace)
+        except Exception as exc:  # a stimulus failure is a finding, not a crash
+            attempt.reason = f"stimulus raised {type(exc).__name__}: {exc}"
+        if collector.entry is None:
+            run_window(session, collector, options.timeout)
+        attempt.entry = collector.entry
+        attempt.prior = collector.prior
+        attempt.dispatch = collector.dispatch
+        attempt.hit_count = len(collector.hits)
+        attempt.effects = collector.effects(None)
+        attempt.trap = attempt.effects["trap"]
+        attempt.fault = attempt.effects["fault"]
+        attempt.ok = collector.entry is not None
+        if not attempt.ok and attempt.reason is None:
+            attempt.reason = "stimulus completed without reaching the target"
+    finally:
+        session.close()
+    return attempt
+
+
+METHOD_FUNCS = {
+    "call": attempt_call,
+    "run": attempt_run,
+    "trigger": attempt_trigger,
+    "stimulus": attempt_stimulus,
+}
+
+
+# ---------------------------------------------------------------------------
+# orchestration, report, probe spec
+# ---------------------------------------------------------------------------
+
+
+def attempt_order(options: ReachOptions, target_addr: int, surface: DemoSurface | None) -> list[str]:
+    if options.method != "auto":
+        return [options.method]
+    order = list(AUTO_ORDER)
+    if not options.stimulus:
+        order.remove("stimulus")
+    return order
+
+
+def run_reach(options: ReachOptions) -> dict:
+    if options.backend == "qemu" and not qemu_available():
+        raise GreenEyeError("qemu-system-riscv64 not found; install qemu-system-misc or set GREENEYE_QEMU")
+    if not os.path.isfile(options.kernel):
+        raise GreenEyeError(f"kernel not found: {options.kernel}")
+    if not options.allow_foreign_elf and os.path.abspath(options.kernel) != os.path.abspath(DEMO_ELF):
+        raise GreenEyeError(
+            "reachability is restricted to the bundled demo firmware; "
+            "pass --allow-foreign-elf to run against another ELF under QEMU"
+        )
+    if not options.target:
+        raise GreenEyeError("--target is required (symbol name or 0x address)")
+    ensure_dir(options.out)
+    # Resolve the target once on a short-lived probe session so the report can
+    # name the address even if every method misses.
+    target_addr, target_name, symbols = resolve_target(options)
+    surface = DemoSurface() if symbols.get("gsp_ctrl") is not None else None
+    attempts: list[Attempt] = []
+    for method in attempt_order(options, target_addr, surface):
+        attempt = METHOD_FUNCS[method](options, surface, target_addr, target_name, bool(options.record))
+        attempts.append(attempt)
+        if attempt.ok and not options.all_methods:
+            break
+    success = next((attempt for attempt in attempts if attempt.ok), None)
+    report = build_report(options, target_addr, target_name, attempts, success)
+    shutil.rmtree(os.path.join(options.out, ".faults"), ignore_errors=True)
+    if report["reachable"]:
+        probe_path = options.emit_probe or os.path.join(options.out, "probe.json")
+        ensure_dir(os.path.dirname(os.path.abspath(probe_path)))
+        write_json(probe_path, suggested_probe(options, target_name, success))
+        report["probe_spec"] = probe_path
+    write_json(os.path.join(options.out, "reach-report.json"), report)
+    with open(os.path.join(options.out, "reach-report.md"), "w", encoding="utf-8") as fh:
+        fh.write(render_markdown(report, options))
+    return report
+
+
+def resolve_target(options: ReachOptions) -> tuple[int, str, SymbolDB]:
+    """Resolve the target address with a paused session, then close it."""
+    session = launch(kernel=options.kernel, backend=options.backend, name="reach-resolve")
+    try:
+        symbols = session.symbols
+        addr = session.resolve(options.target)
+        symbol = symbols.resolve(addr)
+        name = symbol.name if symbol else options.target
+        return addr, name, symbols
+    finally:
+        session.close()
+
+
+def suggested_probe(options: ReachOptions, target_name: str, success: Attempt | None) -> dict:
+    method = success.method if success is not None else options.method
+    spec = {
+        "format": PROBE_FORMAT,
+        "target": target_name,
+        "kernel": options.kernel,
+        "method": method,
+        "args": list(options.args),
+        "arg_count": options.arg_count,
+        "arg_fill": options.arg_fill,
+        "timeout": options.timeout,
+        "settle": options.settle,
+        "stimulus": options.stimulus,
+        "trigger": success.trigger if success is not None and success.method == "trigger" else None,
+        "watch": options.watch,
+    }
+    return spec
+
+
+def build_report(options: ReachOptions, target_addr: int, target_name: str, attempts: list[Attempt],
+                 success: Attempt | None) -> dict:
+    reachable = success is not None
+    entry = success.entry if success is not None else None
+    prior = success.prior if success is not None else []
+    return {
+        "format": FORMAT,
+        "tool": "harness/reach.py",
+        "target": target_name,
+        "target_addr": target_addr,
+        "kernel": options.kernel,
+        "kernel_sha256": sha256_file(options.kernel) if os.path.isfile(options.kernel) else None,
+        "backend": options.backend,
+        "reachable": reachable,
+        "method": success.method if success is not None else None,
+        "methods_tried": [attempt.summary() for attempt in attempts],
+        "entry": entry,
+        "prior_events": prior,
+        "dispatch": success.dispatch if success is not None else [],
+        "required_args": required_args(entry, success),
+        "effects": success.effects if success is not None else {},
+        "trigger": success.trigger if success is not None else None,
+        "trace_window": trace_window(success),
+        "session": success.session if success is not None else next(
+            (attempt.session for attempt in attempts if attempt.session), None
+        ),
+        "probe_spec": None,
+        "reproduce": reproduce_command(options),
+        "reason": None if reachable else (attempts[-1].reason if attempts else "no method attempted"),
+    }
+
+
+def required_args(entry: dict | None, success: Attempt | None) -> dict | None:
+    if entry is None:
+        return None
+    caller = entry.get("ra", 0)
+    dispatch_args = entry.get("dispatch_args")
+    if dispatch_args:
+        return {
+            "source": "firmware-dispatch",
+            "cmd": dispatch_args[0],
+            "a0": dispatch_args[1] if len(dispatch_args) > 1 else None,
+            "a1": dispatch_args[2] if len(dispatch_args) > 2 else None,
+            "a2": dispatch_args[3] if len(dispatch_args) > 3 else None,
+            "handler_entry_args": entry.get("args"),
+            "caller": f"{caller:#x}" if caller else None,
+        }
+    args = entry.get("args") or []
+    return {
+        "source": "call-args" if success is not None and success.method == "call" else "firmware-dispatch",
+        "a0": args[0] if len(args) > 0 else None,
+        "a1": args[1] if len(args) > 1 else None,
+        "a2": args[2] if len(args) > 2 else None,
+        "a3": args[3] if len(args) > 3 else None,
+        "caller": f"{caller:#x}" if caller else None,
+    }
+
+
+def trace_window(success: Attempt | None) -> dict | None:
+    if success is None:
+        return None
+    counts = Counter(item.get("stage", "hook") for item in success.prior)
+    window = {"hook_counts": dict(sorted(counts.items())), "prior": success.prior}
+    if success.session and os.path.isdir(success.session):
+        try:
+            events = load_events(success.session)
+        except Exception:
+            return window
+        window["session_event_counts"] = dict(sorted(Counter(event.get("type", "?") for event in events).items()))
+    return window
+
+
+def render_markdown(report: dict, options: ReachOptions) -> str:
+    lines = [
+        "# green-eye reachability report",
+        "",
+        "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/reach.py`.",
+        "",
+        f"* format: `{FORMAT}`",
+        f"* target: `{report['target']}` @ `{report['target_addr']:#x}`",
+        f"* kernel: `{report['kernel']}` (sha256 `{(report['kernel_sha256'] or '')[:16]}`)",
+        f"* backend: `{report['backend']}`",
+        f"* reachable: **{'yes' if report['reachable'] else 'no'}**"
+        + (f" via `{report['method']}`" if report["reachable"] else ""),
+        "",
+        "## Methods tried",
+        "",
+    ]
+    lines += _table(
+        ["method", "result", "hits", "reason"],
+        [
+            [
+                attempt["method"],
+                "reachable" if attempt["ok"] else "miss",
+                attempt["hit_count"],
+                attempt["reason"] or "-",
+            ]
+            for attempt in report["methods_tried"]
+        ],
+    )
+    if report["reachable"]:
+        entry = report["entry"] or {}
+        args = entry.get("args") or []
+        req = report["required_args"] or {}
+        lines += ["", "## Entry state", ""]
+        lines += _table(
+            ["field", "value"],
+            [
+                ["symbol", entry.get("symbol")],
+                ["pc", f"`{entry.get('pc', 0):#x}`"],
+                ["args a0..a3", ", ".join(f"`{value:#x}`" for value in args[:4]) or "-"],
+                ["ra", f"`{entry.get('ra', 0):#x}`"],
+                ["sp", f"`{entry.get('sp', 0):#x}`"],
+            ],
+        )
+        lines += ["", "## Required args discovered", ""]
+        lines += _table(
+            ["source", "cmd", "a0", "a1", "a2", "a3", "caller"],
+            [[
+                req.get("source"),
+                f"`{req['cmd']:#x}`" if req.get("cmd") is not None else "-",
+                *[f"`{req[key]:#x}`" if req.get(key) is not None else "-" for key in ("a0", "a1", "a2", "a3")],
+                f"`{req['caller']}`" if req.get("caller") else "-",
+            ]],
+        )
+        if report.get("trigger"):
+            trigger = report["trigger"]
+            lines += ["", "## Trigger", ""]
+            lines += _table(
+                ["surface", "command id", "handler", "args"],
+                [[trigger["surface"], trigger["id"], trigger["handler"],
+                  ", ".join(f"`{value:#x}`" for value in trigger["args"])]],
+            )
+        lines += ["", "## Prior events (hooks that led here)", ""]
+        if report["prior_events"]:
+            lines += _table(
+                ["#", "stage", "symbol", "addr", "args"],
+                [
+                    [index, item.get("stage"), item.get("symbol"), f"`{item.get('addr', 0):#x}`",
+                     ", ".join(f"`{value:#x}`" for value in item.get("args", []))]
+                    for index, item in enumerate(report["prior_events"])
+                ],
+            )
+        else:
+            lines += ["_none: the target was the first hooked event_"]
+        lines += ["", "## Effects after resume", ""]
+        effects = report["effects"]
+        changed = effects.get("changed") or []
+        lines += _table(
+            ["field", "before", "after"],
+            [[row["field"], _short(row["before"]), _short(row["after"])] for row in changed] or [["-", "-", "-"]],
+        )
+        lines += ["", f"* return value: {_hex(effects.get('retval'))}",
+                  f"* trap (ebreak) observed: {effects.get('trap')}",
+                  f"* fault observed: {effects.get('fault')}",
+                  f"* watchpoint stores: {len(effects.get('watch') or [])}"]
+        window = report.get("trace_window") or {}
+        if window.get("session_event_counts"):
+            lines += ["", "## Trace window (recorded session)", "",
+                      f"`{json.dumps(window['session_event_counts'], sort_keys=True)}`"]
+    else:
+        lines += ["", "## Not reached", "",
+                  f"* reason: {report.get('reason') or 'unknown'}",
+                  "* absence of a hit is evidence about this stimulus at this fidelity, not proof of dead code."]
+    lines += ["", "## Reproduce", "", "```bash", report["reproduce"], "```"]
+    if report.get("probe_spec"):
+        lines += ["", f"* repeatable probe spec: `{report['probe_spec']}`"]
+    lines += [""]
+    return "\n".join(lines)
+
+
+def _short(value) -> str:
+    text = json.dumps(value, sort_keys=True)
+    return f"`{text[:48]}`" if len(text) <= 48 else f"`{text[:47]}...`"
+
+
+def _hex(value) -> str:
+    return f"`{value:#x}`" if isinstance(value, int) else "-"
+
+
+def _table(headers: list[str], rows: list[list]) -> list[str]:
+    lines = ["| " + " | ".join(headers) + " |", "|" + "|".join("---" for _ in headers) + "|"]
+    for row in rows:
+        lines.append("| " + " | ".join(str(cell) for cell in row) + " |")
+    return lines
+
+
+def reproduce_command(options: ReachOptions) -> str:
+    args = f" --args {','.join(hex(value) for value in options.args)}" if options.args else ""
+    arg_count = f" --arg-count {options.arg_count}" if options.arg_count else ""
+    stimulus = f" --stimulus {options.stimulus}" if options.stimulus else ""
+    methods = " --all-methods" if options.all_methods else ""
+    foreign = " --allow-foreign-elf" if options.allow_foreign_elf else ""
+    record = f" --record {options.record}" if options.record else ""
+    return (
+        f"python3 harness/reach.py --target {options.target} --method {options.method} "
+        f"--kernel {options.kernel} --out {options.out} --timeout {options.timeout} "
+        f"--settle {options.settle}{args}{arg_count}{stimulus}{methods}{record}{foreign}"
+    )
+
+
+def write_json(path: str, payload: dict) -> str:
+    ensure_dir(os.path.dirname(path))
+    with open(path, "w", encoding="utf-8") as fh:
+        json.dump(payload, fh, indent=2, sort_keys=True)
+        fh.write("\n")
+    return path
+
+
+# ---------------------------------------------------------------------------
+# probe spec loading
+# ---------------------------------------------------------------------------
+
+
+def load_probe_spec(path: str) -> dict:
+    with open(path, "r", encoding="utf-8") as fh:
+        spec = json.load(fh)
+    if not isinstance(spec, dict) or "target" not in spec:
+        raise GreenEyeError(f"{path}: reachability probe spec must be an object with a 'target'")
+    return spec
+
+
+def apply_probe_spec(options: ReachOptions, spec: dict) -> None:
+    if not options.target:
+        options.target = str(spec.get("target", ""))
+    if spec.get("kernel") and os.path.abspath(options.kernel) == os.path.abspath(DEMO_ELF):
+        options.kernel = os.path.abspath(str(spec["kernel"]))
+    for attr, key in (
+        ("method", "method"),
+        ("stimulus", "stimulus"),
+    ):
+        if getattr(options, attr) in (None, "", "auto") and spec.get(key):
+            setattr(options, attr, spec[key])
+    if not options.args and spec.get("args"):
+        options.args = [int(value) for value in spec["args"]]
+    if options.arg_count == 0 and spec.get("arg_count"):
+        options.arg_count = int(spec["arg_count"])
+    if options.arg_fill == "zero" and spec.get("arg_fill"):
+        options.arg_fill = str(spec["arg_fill"])
+    if options.timeout == 0.5 and spec.get("timeout"):
+        options.timeout = float(spec["timeout"])
+    if options.settle == 2.0 and spec.get("settle"):
+        options.settle = float(spec["settle"])
+    if not options.watch and spec.get("watch"):
+        options.watch = list(spec["watch"])
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def parse_args_list(text: str) -> list[int]:
+    values: list[int] = []
+    for part in text.replace(" ", ",").split(","):
+        if part:
+            values.append(int(part, 0))
+    return values
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="reach", description=__doc__.splitlines()[0])
+    parser.add_argument("--target", help="symbol name or 0x address to reach")
+    parser.add_argument("--kernel", default=DEMO_ELF, help="demo/re-hosted ELF (default: bundled demo)")
+    parser.add_argument("--out", default=None, help="output directory (default: dumps/reach-<target>)")
+    parser.add_argument(
+        "--method",
+        choices=("auto", "stimulus", "trigger", "run", "call"),
+        default="auto",
+        help="methods to attempt; auto tries stimulus/trigger/run/call and stops at the first hit",
+    )
+    parser.add_argument("--all-methods", action="store_true", help="attempt every method, not just until the first hit")
+    parser.add_argument("--args", type=parse_args_list, default=[], help="comma-separated integer args (a0..a7)")
+    parser.add_argument("--arg-count", type=int, default=0, help="auto-fill a0..a(n-1) when --args is short")
+    parser.add_argument("--arg-fill", choices=("zero", "sentinel"), default="zero", help="auto-fill value policy")
+    parser.add_argument("--timeout", type=float, default=0.5, help="per-attempt budget (seconds)")
+    parser.add_argument("--settle", type=float, default=2.0, help="self-test settle budget before trigger")
+    parser.add_argument("--stimulus", help="green-eye script to run under instrumentation")
+    parser.add_argument("--probe", help="JSON probe spec to replay (see --emit-probe)")
+    parser.add_argument("--emit-probe", help="where to write the repeatable probe spec (default <out>/probe.json)")
+    parser.add_argument("--record", nargs="?", const=True, default=None, help="record attempts under DIR/<method>")
+    parser.add_argument("--backend", choices=("qemu", "native"), default="qemu")
+    parser.add_argument("--max-prior", type=int, default=12, help="prior hook events kept in the report")
+    parser.add_argument("--allow-foreign-elf", action="store_true", help="permit a non-demo ELF under QEMU")
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    if args.probe:
+        try:
+            spec = load_probe_spec(args.probe)
+        except (OSError, ValueError, GreenEyeError) as exc:
+            print(f"reach error: {exc}", file=sys.stderr)
+            return 2
+    else:
+        spec = {}
+    target_hint = args.target or str(spec.get("target", "target"))
+    out = args.out or os.path.join("dumps", "reach-" + _safe_name(target_hint))
+    if isinstance(args.record, str):
+        record_base: str | None = os.path.abspath(args.record)
+    elif args.record:
+        record_base = os.path.join(os.path.abspath(out), "sessions")
+    else:
+        record_base = None
+    options = ReachOptions(
+        kernel=os.path.abspath(args.kernel),
+        out=os.path.abspath(out),
+        target=args.target or "",
+        method=args.method,
+        args=args.args,
+        arg_count=args.arg_count,
+        arg_fill=args.arg_fill,
+        timeout=args.timeout,
+        settle=args.settle,
+        stimulus=args.stimulus,
+        probe=args.probe,
+        emit_probe=os.path.abspath(args.emit_probe) if args.emit_probe else None,
+        record=record_base,
+        backend=args.backend,
+        max_prior=args.max_prior,
+        all_methods=args.all_methods,
+        allow_foreign_elf=args.allow_foreign_elf,
+    )
+    if spec:
+        apply_probe_spec(options, spec)
+    if not options.target:
+        parser.error("--target is required (or provide a probe spec with 'target')")
+    try:
+        report = run_reach(options)
+    except GreenEyeError as exc:
+        print(f"reach error: {exc}", file=sys.stderr)
+        return 2
+    if report["reachable"]:
+        print(f"[reach] {report['target']} reachable via {report['method']} ({os.path.join(options.out, 'reach-report.md')})")
+    else:
+        print(f"[reach] {report['target']} not reached: {report['reason']}")
+    return 0 if report["reachable"] else 1
+
+
+def _safe_name(text: str) -> str:
+    return "".join(ch if ch.isalnum() or ch in "-_." else "-" for ch in text)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(main())