path: root/debug/risc-v-gsp-rm-vm/harness/reach.py

#!/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 risc-v-gsp-rm-debugger 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 = [
        "# risc-v-gsp-rm-debugger reachability report",
        "",
        "> Status: implemented, documented and tested (audit: docs/27-TOOL-AUDIT.md; value demos: docs/25-VALUE-VALIDATION.md). 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="risc-v-gsp-rm-debugger 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())