greeneye: GMMU/FBIF observability, PTE decoder and oracle, MPU/libos simulator wave

Simulator development wave (docs 42-44): GMMU page-table entry decode with its oracle/workflow harness, FBIF observability, and MPU/LIBOS device surfaces in the device model, plus refreshed boot/machine/profile data (tu104-570.144), the qemu-line results, and the new unit tests. Docs 00/06/35-41 updated to match; harness helpers and .gitignore included.

1 files changed, 924 insertions, 0 deletions

diff --git a/debug/risc-v-gsp-rm-vm/harness/gmmu_debug.py b/debug/risc-v-gsp-rm-vm/harness/gmmu_debug.py
new file mode 100644
index 0000000..794bb1f
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/gmmu_debug.py
@@ -0,0 +1,924 @@
+#!/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())