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

#!/usr/bin/env python3
"""GMMU/FBIF static<->dynamic join (harness/gmmu_debug.py).

Answers "which firmware function programs which register, in what order"
from a recorded simulator trace, then writes a Ghidra-ready hit file.

Inputs
------

* A G1 trace: ``greeneye-gmmu-fbif-trace/1``.  The producer's document is
  nested (``windows[].registers[].accesses[]`` plus
  ``windows[].unclassified.accesses[]``); a flat ``events`` list and field
  aliases are also accepted so the harness survives producer naming choices
  (see :func:`load_sim_trace`).  Nested documents are flattened per
  window/register: per-register order is preserved, global interleaving
  across registers is not recoverable from that shape.
* A session directory containing such a trace under ``blobs/`` or named by
  ``session.json`` (``gmmu_fbif_trace`` / ``gmmu_fbif_trace_blob``).
* A symbol DB/map: ``greeneye-symbols/1`` JSON, a plain ``{addr: name}``
  JSON map, or a Ghidra ``addr<TAB>name[...]`` TSV.

Outputs
-------

* ``gmmu-debug-report.json`` / ``.md``: per-firmware-function register rows
  (function, register, window, offset, sequence of written values,
  first/last pc, write and read counts).
* ``greeneye_hits.tsv``: ``addr<TAB>hits<TAB>symbol<TAB>annotation``, the
  exact shape ``ghidra/MarkGreenEyeHits.java`` consumes.

The ``demo`` subcommand builds a small synthetic fixture, runs the whole
loop, and compares the measured wall time against thresholds frozen in this
file before the first measurement.  The manual baseline is a conservative
estimate, not a measurement on a human.

Development-only: this module only reads files and writes reports.  It never
touches a GPU, module, mdev, guest or network.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
import time
from dataclasses import dataclass, field
from typing import Any

TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if TOOL_ROOT not in sys.path:
    sys.path.insert(0, TOOL_ROOT)

from greeneye.errors import GreenEyeError  # noqa: E402
from greeneye.ghidra import render_ghidra_command  # noqa: E402
from greeneye.symbols import Symbol, SymbolDB  # noqa: E402

#: Schema tag of the G1 simulator observation file.
TRACE_FORMAT = "greeneye-gmmu-fbif-trace/1"
#: Schema tag of the report this module emits.
DEBUG_REPORT_FORMAT = "greeneye-gmmu-debug-report/1"
#: Schema tag of the demo result line.
DEMO_RESULT_FORMAT = "greeneye-gmmu-debug-demo/1"
#: Exact header expected by ``ghidra/MarkGreenEyeHits.java``.
HITS_HEADER = "addr\thits\tsymbol\tannotation"
#: Cap on the per-register value sequence kept in the JSON report.
MAX_SEQUENCE_VALUES = 32

# FROZEN BEFORE THE FIRST MEASUREMENT (2026-10-03).  Do not move silently:
# a threshold change requires a new recorded run and a note in docs/44.
#: Conservative ESTIMATE (not measured on a human) of the static path for
#: the demo's handful of functions: open the Ghidra project, locate each
#: function, read every register-write site by hand against the register
#: header.  ~20 minutes is deliberately generous to the manual baseline.
MANUAL_STATIC_BASELINE_SECONDS = 1200
MANUAL_BASELINE_LABEL = "conservative estimate (not measured on a human)"
#: The tool path must stay at or below this fraction of the estimate.
WORKFLOW_THRESHOLD_FRACTION = 0.05


class GmmuInputError(GreenEyeError):
    """A GMMU input file is absent, unreadable or not in a supported schema."""


def sha256_file(path: str) -> str:
    digest = hashlib.sha256()
    with open(path, "rb") as fh:
        for chunk in iter(lambda: fh.read(1 << 20), b""):
            digest.update(chunk)
    return digest.hexdigest()


def parse_int(value: Any, default: int | None = None) -> int | None:
    """Parse ints, decimal strings and ``0x`` strings; ``None`` on failure."""
    if value is None:
        return default
    if isinstance(value, bool):
        return int(value)
    if isinstance(value, int):
        return value
    if isinstance(value, float):
        return int(value)
    if isinstance(value, str):
        text = value.strip()
        if not text:
            return default
        try:
            return int(text, 0)
        except ValueError:
            try:
                return int(text, 16)
            except ValueError:
                return default
    return default


def _first_str(event: dict, keys: tuple[str, ...]) -> str:
    for key in keys:
        value = event.get(key)
        if isinstance(value, str) and value:
            return value
        if isinstance(value, int) and not isinstance(value, bool):
            return str(value)
    return ""


def _first_int(event: dict, keys: tuple[str, ...]) -> int | None:
    for key in keys:
        if key in event:
            parsed = parse_int(event.get(key))
            if parsed is not None:
                return parsed
    return None


def _first_bool(event: dict, keys: tuple[str, ...]) -> bool | None:
    for key in keys:
        value = event.get(key)
        if isinstance(value, bool):
            return value
    return None


_WRITE_WORDS = {"w", "write", "wr", "store", "st", "out"}
_READ_WORDS = {"r", "read", "rd", "load", "ld", "in"}
_OTHER_WORDS = {"exec", "summary", "note", "event"}


def _normalise_kind(event: dict) -> str:
    for key in ("kind", "rw", "direction", "access", "op", "type", "t"):
        value = event.get(key)
        if isinstance(value, str):
            low = value.strip().lower()
            if low in _WRITE_WORDS:
                return "write"
            if low in _READ_WORDS:
                return "read"
            if low in _OTHER_WORDS:
                return "other"
    write_flag = event.get("write")
    if isinstance(write_flag, bool):
        return "write" if write_flag else "read"
    return "other"


@dataclass
class TraceObservation:
    """One register access from a simulator trace."""

    seq: int
    kind: str
    name: str = ""
    window: str = ""
    offset: int | None = None
    addr: int | None = None
    size: int | None = None
    value: int | None = None
    pc: int | None = None
    function: str = ""
    handled: bool | None = None
    vt: int | None = None

    def to_json(self) -> dict:
        out: dict[str, Any] = {"seq": self.seq, "kind": self.kind}
        if self.name:
            out["name"] = self.name
        if self.window:
            out["window"] = self.window
        if self.offset is not None:
            out["offset"] = self.offset
        if self.addr is not None:
            out["addr"] = self.addr
        if self.size is not None:
            out["size"] = self.size
        if self.value is not None:
            out["value"] = self.value
        if self.pc is not None:
            out["pc"] = self.pc
        if self.function:
            out["function"] = self.function
        if self.handled is not None:
            out["handled"] = self.handled
        if self.vt is not None:
            out["vt"] = self.vt
        return out


@dataclass
class SimTrace:
    path: str
    format: str
    sha256: str
    observations: list[TraceObservation] = field(default_factory=list)
    source: dict = field(default_factory=dict)


def _observation_from_event(event: dict, seq: int) -> TraceObservation:
    return TraceObservation(
        seq=seq,
        kind=_normalise_kind(event),
        name=_first_str(event, ("name", "register", "reg", "signal")),
        window=_first_str(event, ("window", "block", "device", "bar", "region")),
        offset=_first_int(event, ("offset", "reg_offset", "index")),
        addr=_first_int(event, ("addr", "address", "pa", "guest_addr", "gpa")),
        size=_first_int(event, ("size", "width", "bytes")),
        value=_first_int(event, ("value", "data", "val")),
        pc=_first_int(event, ("pc", "pc_addr")),
        function=_first_str(event, ("function", "symbol", "func", "pc_symbol")),
        handled=_first_bool(event, ("handled", "served")),
        vt=_first_int(event, ("vt", "virtual_time")),
    )


_EVENT_KEYS = ("events", "trace", "accesses", "items", "observations")


def _flatten_windows(payload: dict) -> list[dict]:
    """Flatten the G1 nested document (windows -> registers -> accesses)."""
    events: list[dict] = []
    for window in payload.get("windows", []):
        if not isinstance(window, dict):
            continue
        label = window.get("label") or window.get("window") or ""
        for register in window.get("registers", []) or []:
            if not isinstance(register, dict):
                continue
            name = register.get("name")
            for access in register.get("accesses", []) or []:
                if not isinstance(access, dict):
                    continue
                event = dict(access)
                if label:
                    event.setdefault("window", label)
                if name:
                    event.setdefault("register", name)
                events.append(event)
        unclassified = window.get("unclassified")
        if isinstance(unclassified, dict):
            for access in unclassified.get("accesses", []) or []:
                if not isinstance(access, dict):
                    continue
                event = dict(access)
                if label:
                    event.setdefault("window", label)
                events.append(event)
    return events


def _source_metadata(payload: dict) -> dict:
    source: dict = {}
    if isinstance(payload.get("source"), dict):
        source.update(payload["source"])
    for key in ("image_sha256", "profile", "profile_name", "symbols"):
        if payload.get(key) is not None:
            source.setdefault(key, payload[key])
    summary = payload.get("summary")
    if isinstance(summary, dict):
        source.setdefault("summary", summary)
    return source


def _events_from_payload(payload: Any) -> tuple[str, list | None, dict]:
    if isinstance(payload, list):
        return "unlabeled-list", payload, {}
    if not isinstance(payload, dict):
        raise GmmuInputError("trace JSON must be an object or a list of events")
    fmt = str(payload.get("format") or payload.get("schema") or "")
    source = _source_metadata(payload)
    for key in _EVENT_KEYS:
        if isinstance(payload.get(key), list):
            return fmt, payload[key], source
    if isinstance(payload.get("windows"), list):
        return fmt, _flatten_windows(payload), source
    if fmt == TRACE_FORMAT:
        return fmt, [], source
    return fmt, None, source


def _parse_jsonl(text: str) -> tuple[str, list[dict], dict]:
    fmt = ""
    events: list[dict] = []
    for line in text.splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            item = json.loads(line)
        except json.JSONDecodeError:
            continue
        if not isinstance(item, dict):
            continue
        if item.get("format") or item.get("schema"):
            fmt = str(item.get("format") or item.get("schema"))
            if not any(k in item for k in ("addr", "pc", "offset", "name", "register")):
                continue
        if item.get("t") == "summary":
            continue
        events.append(item)
    return fmt, events, {}


def _find_session_trace(session_dir: str) -> str:
    session_json = os.path.join(session_dir, "session.json")
    explicit: list[str] = []
    if os.path.isfile(session_json):
        try:
            with open(session_json, "r", encoding="utf-8") as fh:
                meta = json.load(fh)
        except (OSError, json.JSONDecodeError):
            meta = {}
        if isinstance(meta, dict):
            for key in ("gmmu_fbif_trace", "gmmu_fbif_trace_blob", "trace_blob"):
                value = meta.get(key)
                if isinstance(value, str) and value:
                    explicit.append(os.path.join(session_dir, value))
    for candidate in explicit:
        if os.path.isfile(candidate):
            return candidate
    blobs = os.path.join(session_dir, "blobs")
    if os.path.isdir(blobs):
        for name in sorted(os.listdir(blobs)):
            if not name.endswith((".json", ".jsonl")):
                continue
            candidate = os.path.join(blobs, name)
            try:
                with open(candidate, "r", encoding="utf-8", errors="replace") as fh:
                    head = fh.read(1 << 16)
            except OSError:
                continue
            if TRACE_FORMAT in head:
                return candidate
        for name in sorted(os.listdir(blobs)):
            if name.endswith(".jsonl"):
                return os.path.join(blobs, name)
    raise GmmuInputError(
        f"{session_dir}: no {TRACE_FORMAT} trace found (looked for session.json "
        "pointers and blobs/*.json[l])"
    )


def load_sim_trace(path: str) -> SimTrace:
    """Load a G1 trace file, or a session directory containing one.

    Accepted shapes: the G1 nested document (``schema`` +
    ``windows[].registers[].accesses[]``), an object with a flat
    ``events``/``trace``/``accesses`` list, a bare list, and JSONL.  Event
    aliases: ``kind``/``rw``/``direction``/``type``/``t``;
    ``name``/``register``/``reg``; ``window``/``block``/``device``;
    ``offset``/``reg_offset``; ``addr``/``address``/``pa``;
    ``value``/``data``; ``pc``; ``function``/``symbol``/``pc_symbol``;
    ``handled``/``served``.  Unknown fields are ignored; an event with no
    recognisable direction is kept as ``other``.
    """
    if os.path.isdir(path):
        path = _find_session_trace(path)
    if not os.path.isfile(path):
        raise GmmuInputError(f"trace not found: {path}")
    digest = sha256_file(path)
    with open(path, "r", encoding="utf-8", errors="replace") as fh:
        text = fh.read()
    try:
        payload = json.loads(text)
    except json.JSONDecodeError:
        payload = None
    if payload is not None:
        fmt, events, source = _events_from_payload(payload)
        if events is None:
            raise GmmuInputError(
                f"{path}: no event list (expected {TRACE_FORMAT} with an 'events' key)"
            )
    else:
        fmt, events, source = _parse_jsonl(text)
    observations = [
        _observation_from_event(event, seq)
        for seq, event in enumerate(events)
        if isinstance(event, dict)
    ]
    return SimTrace(
        path=os.path.abspath(path),
        format=fmt or "unlabeled",
        sha256=digest,
        observations=observations,
        source=source,
    )


def load_symbols(path: str | None) -> SymbolDB:
    """Load a symbol DB (JSON) or a Ghidra-style TSV/map (``addr<TAB>name``)."""
    db = SymbolDB()
    if not path:
        return db
    if not os.path.isfile(path):
        raise GmmuInputError(f"symbol map not found: {path}")
    if path.endswith(".json"):
        try:
            with open(path, "r", encoding="utf-8") as fh:
                payload = json.load(fh)
        except (OSError, json.JSONDecodeError) as exc:
            raise GmmuInputError(f"cannot read symbol map {path}: {exc}") from exc
        if isinstance(payload, dict) and (
            "symbols" in payload or payload.get("format") == "greeneye-symbols/1"
        ):
            db.load_json(path)
            return db
        if isinstance(payload, list):
            db.load_json(path)
            return db
        if isinstance(payload, dict):
            for key, value in payload.items():
                addr = parse_int(key)
                if addr is None:
                    continue
                if isinstance(value, str):
                    db.add(Symbol(addr=addr, name=value, source="map"))
                elif isinstance(value, dict) and value.get("name"):
                    db.add(
                        Symbol(
                            addr=addr,
                            name=str(value["name"]),
                            size=parse_int(value.get("size"), 0) or 0,
                            source="map",
                        )
                    )
            return db
        raise GmmuInputError(f"{path}: unsupported symbol JSON shape")
    db.load_index_tsv(path)
    if len(db) == 0:
        db.load_names_tsv(path)
    if len(db) == 0:
        raise GmmuInputError(f"{path}: no symbols parsed")
    return db


def _function_identity(
    obs: TraceObservation, symbols: SymbolDB | None
) -> tuple[tuple | None, int | None, str]:
    """Resolve an observation to ``(group_key, entry_addr, display_name)``.

    An explicit ``function`` field groups by name (all events of one named
    function stay together even without a symbol DB); a resolved symbol
    groups by its entry address; an unresolved pc becomes ``sub_<pc>``.
    """
    if obs.function:
        symbol = symbols.get(obs.function) if symbols is not None else None
        return (
            ("name", obs.function),
            symbol.addr if symbol is not None else None,
            obs.function,
        )
    if obs.pc is None:
        return None, None, ""
    symbol = symbols.resolve(obs.pc) if symbols is not None else None
    if symbol is not None:
        return ("addr", symbol.addr), symbol.addr, symbol.name
    return ("sub", obs.pc), obs.pc, f"sub_{obs.pc:#x}"


def _register_identity(obs: TraceObservation) -> tuple:
    if obs.offset is not None:
        return (obs.window, obs.offset, obs.name)
    if obs.name:
        return (obs.window, None, obs.name)
    return (obs.window, obs.addr, "")


def _register_label(obs: TraceObservation) -> str:
    if obs.name:
        return obs.name
    if obs.offset is not None:
        return f"0x{obs.offset:x}"
    if obs.addr is not None:
        return f"0x{obs.addr:x}"
    return "?"


@dataclass
class RegisterRow:
    function: str
    function_addr: int | None
    register: str
    window: str
    offset: int | None
    addr: int | None
    write_count: int = 0
    read_count: int = 0
    first_pc: int | None = None
    last_pc: int | None = None
    first_seq: int | None = None
    last_seq: int | None = None
    values: list[int] = field(default_factory=list)
    values_truncated: bool = False

    def observe(self, obs: TraceObservation) -> None:
        if obs.kind == "write":
            self.write_count += 1
            if obs.value is not None:
                if len(self.values) < MAX_SEQUENCE_VALUES:
                    self.values.append(obs.value)
                else:
                    self.values_truncated = True
        elif obs.kind == "read":
            self.read_count += 1
        if obs.pc is not None:
            if self.first_pc is None:
                self.first_pc = obs.pc
            self.last_pc = obs.pc
        if self.first_seq is None:
            self.first_seq = obs.seq
        self.last_seq = obs.seq

    def to_json(self) -> dict:
        out: dict[str, Any] = {
            "register": self.register,
            "write_count": self.write_count,
            "read_count": self.read_count,
            "first_seq": self.first_seq,
            "last_seq": self.last_seq,
        }
        if self.window:
            out["window"] = self.window
        if self.offset is not None:
            out["offset"] = self.offset
        if self.addr is not None:
            out["addr"] = self.addr
        if self.first_pc is not None:
            out["first_pc"] = self.first_pc
        if self.last_pc is not None:
            out["last_pc"] = self.last_pc
        out["values"] = self.values
        out["values_truncated"] = self.values_truncated
        return out


def build_debug_report(trace: SimTrace, symbols: SymbolDB | None) -> dict:
    """Group trace observations by firmware function and register."""
    functions: dict[tuple, dict] = {}
    order: list[tuple] = []
    unknown_pc = 0
    for obs in trace.observations:
        fkey, addr, function = _function_identity(obs, symbols)
        if fkey is None:
            unknown_pc += 1
            continue
        entry = functions.get(fkey)
        if entry is None:
            entry = {
                "function": function,
                "addr": addr,
                "first_pc": None,
                "writes": 0,
                "reads": 0,
                "registers": {},
                "order": [],
            }
            functions[fkey] = entry
            order.append(fkey)
        rkey = _register_identity(obs)
        row = entry["registers"].get(rkey)
        if row is None:
            row = RegisterRow(
                function=function,
                function_addr=addr,
                register=_register_label(obs),
                window=obs.window,
                offset=obs.offset,
                addr=obs.addr,
            )
            entry["registers"][rkey] = row
            entry["order"].append(rkey)
        row.observe(obs)
        if obs.pc is not None and entry["first_pc"] is None:
            entry["first_pc"] = obs.pc
        if obs.kind == "write":
            entry["writes"] += 1
        elif obs.kind == "read":
            entry["reads"] += 1

    def sort_key(fkey: tuple):
        entry = functions[fkey]
        anchor = entry["addr"] if entry["addr"] is not None else entry["first_pc"]
        return (anchor is None, anchor or 0, entry["function"])

    function_rows = []
    for fkey in sorted(order, key=sort_key):
        entry = functions[fkey]
        register_rows = [entry["registers"][rkey] for rkey in entry["order"]]
        register_rows.sort(
            key=lambda r: (r.window, r.offset is None, r.offset or 0, r.register)
        )
        function_rows.append(
            {
                "function": entry["function"],
                "addr": entry["addr"],
                "first_pc": entry["first_pc"],
                "writes": entry["writes"],
                "reads": entry["reads"],
                "registers": [r.to_json() for r in register_rows],
            }
        )

    total_registers = sum(len(f["registers"]) for f in function_rows)
    return {
        "format": DEBUG_REPORT_FORMAT,
        "inputs": {
            "trace": {
                "path": trace.path,
                "sha256": trace.sha256,
                "format": trace.format,
                "events": len(trace.observations),
                "source": trace.source,
            }
        },
        "summary": {
            "functions": len(function_rows),
            "registers": total_registers,
            "writes": sum(f["writes"] for f in function_rows),
            "reads": sum(f["reads"] for f in function_rows),
            "unknown_pc_events": unknown_pc,
            "trace_events": len(trace.observations),
        },
        "functions": function_rows,
        "limits": [
            "L0/L1 evidence only: the trace records execution and register accesses, not timing.",
            "Absence of a write is not proof a register is untouched outside this run's stimulus.",
            "Synthetic demo fixture values are invented; real traces must come from the simulator.",
        ],
    }


def write_hits_tsv(report: dict, path: str) -> str:
    """Write a ``greeneye_hits.tsv`` compatible with MarkGreenEyeHits.java."""
    os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(HITS_HEADER + "\n")
        for function in report["functions"]:
            addr = function.get("addr")
            if addr is None:
                addr = function.get("first_pc")
            if addr is None:
                for register in function["registers"]:
                    if register.get("first_pc") is not None:
                        addr = register["first_pc"]
                        break
            if addr is None:
                continue
            hits = function["writes"] + function["reads"]
            annotation = (
                f"gmmu: {function['writes']} write(s), {function['reads']} read(s) "
                f"across {len(function['registers'])} register(s)"
            )
            fh.write(
                f"{addr:#x}\t{hits}\t{function['function']}\t{annotation}\n"
            )
    return os.path.abspath(path)


def render_debug_markdown(report: dict, hits_path: str, ghidra_command: str) -> str:
    summary = report["summary"]
    lines = [
        "# GMMU/FBIF per-function register-write report",
        "",
        f"> Format `{report['format']}`. Development-only static<->dynamic join;",
        "> no timing or silicon claim. Fidelity: L0/L1 (docs/06).",
        "",
        "## Inputs",
        "",
        f"* trace: `{report['inputs']['trace']['path']}` "
        f"({report['inputs']['trace']['events']} events, "
        f"format `{report['inputs']['trace']['format']}`, "
        f"sha256 `{report['inputs']['trace']['sha256'][:16]}…`)",
        "",
        "## Summary",
        "",
        "| functions | registers | writes | reads | unknown-pc events |",
        "|---|---|---|---|---|",
        f"| {summary['functions']} | {summary['registers']} | {summary['writes']} | "
        f"{summary['reads']} | {summary['unknown_pc_events']} |",
        "",
        "## Per function",
        "",
    ]
    for function in report["functions"]:
        addr = f"{function['addr']:#x}" if function["addr"] is not None else "unknown"
        lines.append(
            f"### `{function['function']}` @ {addr} — "
            f"{function['writes']} write(s), {function['reads']} read(s)"
        )
        lines.append("")
        lines.append("| register | window | offset | writes | reads | first pc | last pc | value sequence |")
        lines.append("|---|---|---|---|---|---|---|---|")
        for register in function["registers"]:
            offset = f"{register['offset']:#x}" if register.get("offset") is not None else ""
            first_pc = f"{register['first_pc']:#x}" if register.get("first_pc") is not None else ""
            last_pc = f"{register['last_pc']:#x}" if register.get("last_pc") is not None else ""
            sequence = ", ".join(f"{v:#x}" for v in register["values"])
            if register.get("values_truncated"):
                sequence += ", …"
            lines.append(
                f"| `{register['register']}` | {register.get('window', '')} | {offset} | "
                f"{register['write_count']} | {register['read_count']} | {first_pc} | {last_pc} | "
                f"{sequence or ''} |"
            )
        lines.append("")
    lines += [
        "## Ghidra annotation",
        "",
        f"Hits file: `{hits_path}`",
        "",
        "```bash",
        ghidra_command,
        "```",
        "",
        "## Limits",
        "",
    ]
    for limit in report["limits"]:
        lines.append(f"* {limit}")
    lines.append("")
    return "\n".join(lines)


def run_report(trace_path: str, symbols_path: str | None, out_dir: str) -> dict:
    """Run the full trace -> report -> hits-TSV loop into ``out_dir``."""
    trace = load_sim_trace(trace_path)
    symbols = load_symbols(symbols_path)
    report = build_debug_report(trace, symbols)
    os.makedirs(out_dir, exist_ok=True)
    report_json = os.path.join(out_dir, "gmmu-debug-report.json")
    report_md = os.path.join(out_dir, "gmmu-debug-report.md")
    hits_tsv = os.path.join(out_dir, "greeneye_hits.tsv")
    with open(report_json, "w", encoding="utf-8") as fh:
        json.dump(report, fh, indent=2, sort_keys=True)
        fh.write("\n")
    write_hits_tsv(report, hits_tsv)
    ghidra_command = render_ghidra_command(
        "<ghidra-project-dir>", "<project>", "<program>", os.path.abspath(hits_tsv)
    )
    with open(report_md, "w", encoding="utf-8") as fh:
        fh.write(render_debug_markdown(report, os.path.abspath(hits_tsv), ghidra_command))
    report["artifacts"] = {
        "report_json": os.path.abspath(report_json),
        "report_md": os.path.abspath(report_md),
        "hits_tsv": os.path.abspath(hits_tsv),
        "ghidra_command": ghidra_command,
    }
    return report


DEMO_FUNCTIONS = (
    (0x4001000, "gmmu_map_alloc", 0x120),
    (0x4002000, "fbif_setup", 0x80),
    (0x4003000, "gmmu_pte_write", 0x60),
)


def _fixture_events() -> list[dict]:
    raw = [
        (0x4002010, "write", "falcon", 0x600, "FBIF_TRANSCFG0", 4, 0x00000000),
        (0x4002020, "write", "falcon", 0x604, "FBIF_TRANSCFG1", 4, 0x00000001),
        (0x4002030, "write", "falcon", 0x608, "FBIF_TRANSCFG2", 4, 0x00000002),
        (0x4002040, "write", "falcon", 0x624, "FBIF_CTL", 4, 0x00000001),
        (0x4002050, "read", "falcon", 0x008, "IRQSTAT", 4, 0x00000000),
        (0x4001010, "write", "gmmu", 0x0000, "GMMU_PDB_LO", 4, 0x00400000),
        (0x4001020, "write", "gmmu", 0x0004, "GMMU_PDB_HI", 4, 0x00000000),
        (0x4001030, "write", "gmmu", 0x0010, "GMMU_MODE", 4, 0x00000007),
        (0x4001040, "write", "gmmu", 0x0010, "GMMU_MODE", 4, 0x0000000F),
        (0x4001050, "read", "gmmu", 0x0018, "GMMU_STATUS", 4, 0x00000000),
        (0x4003010, "write", "gmmu", 0x0200, "PTE_DATA", 8, 0x0000000000040001),
        (0x4003020, "write", "gmmu", 0x0208, "PTE_DATA", 8, 0x0000000000040001),
        (0x4004010, "write", "falcon", 0x110, "DMATRFBASE", 4, 0x00000000),
    ]
    return [
        {
            "seq": seq,
            "vt": 100 + seq,
            "pc": pc,
            "kind": kind,
            "window": window,
            "offset": offset,
            "name": name,
            "size": size,
            "value": value,
            "handled": True,
        }
        for seq, (pc, kind, window, offset, name, size, value) in enumerate(raw)
    ]


def build_fixture(out_dir: str) -> dict:
    """Write the synthetic demo fixture (symbols + trace); return the paths.

    Symbols carry sizes so a pc past every function stays unattributed
    (``sub_<pc>``) instead of being absorbed by the nearest-preceding name.
    """
    os.makedirs(out_dir, exist_ok=True)
    symbols_path = os.path.join(out_dir, "symbols.json")
    symbols_payload = {
        "format": "greeneye-symbols/1",
        "count": len(DEMO_FUNCTIONS),
        "symbols": [
            {"addr": addr, "name": name, "size": size, "source": "fixture"}
            for addr, name, size in DEMO_FUNCTIONS
        ],
    }
    with open(symbols_path, "w", encoding="utf-8") as fh:
        json.dump(symbols_payload, fh, indent=2, sort_keys=True)
        fh.write("\n")
    trace_path = os.path.join(out_dir, "trace.json")
    payload = {
        "format": TRACE_FORMAT,
        "source": {
            "kind": "synthetic-fixture",
            "note": "invented values; proves the join shape only, never a hardware claim",
        },
        "events": _fixture_events(),
    }
    with open(trace_path, "w", encoding="utf-8") as fh:
        json.dump(payload, fh, indent=2, sort_keys=True)
        fh.write("\n")
    return {
        "symbols": os.path.abspath(symbols_path),
        "trace": os.path.abspath(trace_path),
        "events": len(payload["events"]),
        "functions": len(DEMO_FUNCTIONS),
    }


def run_demo(out_dir: str) -> dict:
    """Build the fixture, run the loop, and check the frozen threshold."""
    fixture = build_fixture(out_dir)
    start = time.perf_counter()
    report = run_report(fixture["trace"], fixture["symbols"], out_dir)
    elapsed = time.perf_counter() - start
    threshold = MANUAL_STATIC_BASELINE_SECONDS * WORKFLOW_THRESHOLD_FRACTION
    result = {
        "format": DEMO_RESULT_FORMAT,
        "scenario": "trace -> per-function register report -> greeneye_hits.tsv -> Ghidra-ready",
        "fixture": fixture,
        "summary": report["summary"],
        "measured_seconds": round(elapsed, 4),
        "manual_baseline_seconds": MANUAL_STATIC_BASELINE_SECONDS,
        "manual_baseline_label": MANUAL_BASELINE_LABEL,
        "threshold_fraction": WORKFLOW_THRESHOLD_FRACTION,
        "threshold_seconds": round(threshold, 4),
        "pass": elapsed <= threshold,
        "artifacts": report["artifacts"],
        "limits": report["limits"],
    }
    with open(os.path.join(out_dir, "demo-result.json"), "w", encoding="utf-8") as fh:
        json.dump(result, fh, indent=2, sort_keys=True)
        fh.write("\n")
    print("RESULT_JSON: " + json.dumps(result, sort_keys=True))
    return result


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="gmmu_debug.py",
        description="GMMU/FBIF trace -> per-function register-write report -> Ghidra hits TSV.",
    )
    sub = parser.add_subparsers(dest="command")
    report = sub.add_parser("report", help="build the report from a trace and symbol map")
    report.add_argument("--trace", required=True, help="trace file or session directory")
    report.add_argument("--symbols", default=None, help="symbol DB/map (JSON or TSV)")
    report.add_argument("--out", required=True, help="output directory")
    demo = sub.add_parser("demo", help="run the synthetic end-to-end fixture")
    demo.add_argument(
        "--out",
        default=os.path.join(TOOL_ROOT, "harness", "gmmu_debug_fixture"),
        help="fixture/report directory (gitignored)",
    )
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    if args.command == "report":
        try:
            report = run_report(args.trace, args.symbols, args.out)
        except GmmuInputError as exc:
            print(f"error: {exc}", file=sys.stderr)
            return 2
        print(
            json.dumps(
                {
                    "format": report["format"],
                    "summary": report["summary"],
                    "artifacts": report["artifacts"],
                },
                indent=2,
                sort_keys=True,
            )
        )
        return 0
    if args.command == "demo":
        result = run_demo(args.out)
        return 0 if result["pass"] else 1
    parser.print_help()
    return 2


if __name__ == "__main__":
    raise SystemExit(main())