1 files changed, 372 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/sim/devices/observe.py b/debug/risc-v-gsp-rm-vm/sim/devices/observe.py
new file mode 100644
index 0000000..d43cfe6
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/sim/devices/observe.py
@@ -0,0 +1,372 @@
+"""Observer-only MMIO windows: record accesses, change no semantics.
+
+Evidence rule (see ``sim/devices/base.py`` and ``docs/36``): this module
+carries **no modelled register behaviour** beyond optional read-back of
+values written to registers the header proves plain read/write.  Every access
+is recorded -- address, window-relative offset, size, direction,
+value-if-known, PC, virtual time, window label and register name -- and the
+observer then **declines** the access, so the machine's existing
+device-map/lenient/unknown path serves it exactly as before.  A read value is
+never invented: an observe-only read records ``value: null`` and returns
+``None``; a storage read before any write is declined the same way.
+
+Modes
+=====
+
+``observe-only`` (default)
+    Read and write are recorded and declined.  Nothing changes.
+
+``storage``
+    A declared register whose header access type is plain ``rw`` (or ``wo``)
+    stores the last written value; a later read of a register that has been
+    written returns that value.  Undeclared offsets, ``ro`` registers and
+    never-written storage registers are still declined.
+
+Register specs come from the profile (or a runtime window list)::
+
+    "0x24": {"name": "NV_PFB_PRI_MMU_TLB_CTRL",
+             "width": 4, "access": "rw",
+             "fields": {"L2_SIZE": [1, 0], ...}}
+
+``fields`` is optional and only used to decode known values in the trace
+(``hi``/``lo`` bit positions, header field names); no field semantics are
+implied.
+
+The access record is the unit of evidence.  It lives in
+:attr:`ObservedWindow.accesses` for the live run and is exported through the
+``greeneye-gmmu-fbif-trace/1`` document built by :func:`build_trace`;
+:meth:`ObservedWindow.state` deliberately carries only counters and storage
+values so a device snapshot stays small.
+
+Nothing here is a silicon claim and nothing here authorises hardware: these
+windows are simulator-only observability for the GMMU/FBIF configuration
+surface (``docs/42-GMMU-FBIF-OBSERVABILITY.md``).
+"""
+from __future__ import annotations
+
+from .base import Device
+
+#: Accepted observer modes.
+MODES = ("observe-only", "storage")
+#: Accepted per-register access kinds (header access type, plain meaning).
+ACCESS_KINDS = ("rw", "ro", "wo")
+#: Schema tag for the GMMU/FBIF trace document.
+TRACE_SCHEMA = "greeneye-gmmu-fbif-trace/1"
+
+DEFAULT_MAX_ACCESS_LOG = 65536
+
+
+def _int(value, default=None):
+    if value is None:
+        return default
+    if isinstance(value, str):
+        return int(value, 0)
+    return int(value)
+
+
+def _normalise_spec(offset, spec) -> dict:
+    """Accept ``"NAME"`` or ``{name, width, access, fields}`` register specs."""
+    if isinstance(spec, str):
+        spec = {"name": spec}
+    if not isinstance(spec, dict) or not spec.get("name"):
+        raise ValueError(f"observer register {offset:#x}: spec needs a name")
+    access = spec.get("access", "rw")
+    if access not in ACCESS_KINDS:
+        raise ValueError(f"observer register {offset:#x}: unknown access {access!r}")
+    width = int(spec.get("width", 4))
+    if width not in (1, 2, 4, 8):
+        raise ValueError(f"observer register {offset:#x}: unsupported width {width}")
+    fields = {}
+    for name, span in (spec.get("fields") or {}).items():
+        if isinstance(span, str):
+            hi, lo = span.split(":")
+            hi, lo = int(hi, 0), int(lo, 0)
+        else:
+            hi, lo = int(span[0]), int(span[1])
+        if not 0 <= lo <= hi <= 8 * width - 1:
+            raise ValueError(
+                f"observer register {offset:#x}: bad field {name} {hi}:{lo}")
+        fields[str(name)] = (hi, lo)
+    return {"name": str(spec["name"]), "width": width,
+            "access": access, "fields": fields}
+
+
+class ObservedWindow(Device):
+    """One observe-only (or storage) MMIO window; a :class:`~sim.devices.base.Device`.
+
+    ``clock`` is an optional zero-argument callable returning virtual time
+    (``Machine.insn_count`` follows the existing ``PtimerBlock`` pattern).
+    The counter advances at :meth:`sim.machine.Machine.run` boundaries, so all
+    accesses inside one bounded run share the run-start value; that granularity
+    is documented, not hidden.
+    """
+
+    name = "observe"
+    #: Opt-in: the registry forwards ``pc=`` to devices that set this.
+    wants_pc = True
+
+    def __init__(self, *, base: int = 0, size: int = 0x1000, label: str | None = None,
+                 mode: str = "observe-only", registers=None, clock=None,
+                 evidence: str = "", max_access_log: int = DEFAULT_MAX_ACCESS_LOG):
+        super().__init__(base=base, size=size)
+        self.label = str(label or "observe")
+        if mode not in MODES:
+            raise ValueError(f"observer window {self.label}: unknown mode {mode!r}")
+        self.mode = mode
+        self.evidence = str(evidence)
+        self.clock = clock
+        self.registers: dict[int, dict] = {}
+        for off, spec in (registers or {}).items():
+            self.registers[_int(off)] = _normalise_spec(_int(off), spec)
+        self.max_access_log = max(0, int(max_access_log))
+        self.accesses: list[dict] = []
+        self.dropped = 0
+        self.counts = {"r": 0, "w": 0, "served": 0, "declined": 0}
+        self.written: dict[int, int] = {}
+
+    # -- recording ---------------------------------------------------------
+
+    def _record(self, offset: int, size: int, rw: str, value, served: bool,
+                pc: int) -> None:
+        spec = self.registers.get(offset)
+        entry = {
+            "addr": self.base + offset,
+            "offset": offset,
+            "size": size,
+            "rw": rw,
+            "value": value,
+            "pc": pc,
+            "vt": self.clock() if self.clock is not None else None,
+            "window": self.label,
+            "register": spec["name"] if spec is not None else None,
+            "served": bool(served),
+        }
+        if value is not None and spec is not None and spec["fields"]:
+            entry["decoded"] = {
+                name: (value >> lo) & ((1 << (hi - lo + 1)) - 1)
+                for name, (hi, lo) in spec["fields"].items()
+            }
+        self.counts[rw] += 1
+        self.counts["served" if served else "declined"] += 1
+        if len(self.accesses) < self.max_access_log:
+            self.accesses.append(entry)
+        else:
+            self.dropped += 1
+
+    # -- guest-visible semantics ------------------------------------------
+
+    def read(self, offset: int, size: int, pc: int = 0) -> int | None:
+        spec = self.registers.get(offset)
+        value = None
+        served = False
+        if (self.mode == "storage" and spec is not None
+                and spec["width"] == size and spec["access"] == "rw"
+                and offset in self.written):
+            value = self.written[offset]
+            served = True
+        self._record(offset, size, "r", value, served, pc)
+        return value
+
+    def write(self, offset: int, size: int, value: int, pc: int = 0) -> bool:
+        spec = self.registers.get(offset)
+        served = False
+        if (self.mode == "storage" and spec is not None
+                and spec["width"] == size and spec["access"] in ("rw", "wo")):
+            mask = (1 << (8 * size)) - 1
+            self.written[offset] = value & mask
+            served = True
+        self._record(offset, size, "w", value & ((1 << (8 * size)) - 1), served, pc)
+        return served
+
+    def peek(self, offset: int, size: int) -> int | None:
+        spec = self.registers.get(offset)
+        if (self.mode == "storage" and spec is not None
+                and spec["width"] == size and spec["access"] == "rw"
+                and offset in self.written):
+            return self.written[offset]
+        return None
+
+    # -- state -------------------------------------------------------------
+
+    def state(self) -> dict:
+        return {
+            "label": self.label,
+            "mode": self.mode,
+            "counts": dict(self.counts),
+            "written": {str(k): v for k, v in sorted(self.written.items())},
+            "dropped_accesses": self.dropped,
+        }
+
+    def load_state(self, data: dict) -> None:
+        counts = data.get("counts", {})
+        self.counts = {k: int(counts.get(k, 0)) for k in ("r", "w", "served", "declined")}
+        self.written = {int(k): int(v) for k, v in data.get("written", {}).items()}
+        self.dropped = int(data.get("dropped_accesses", 0))
+
+    def reset(self) -> None:
+        self.accesses.clear()
+        self.dropped = 0
+        self.counts = {"r": 0, "w": 0, "served": 0, "declined": 0}
+        self.written.clear()
+
+    def __repr__(self) -> str:  # pragma: no cover - cosmetic
+        return (f"ObservedWindow({self.label} @ {self.base:#x}+{self.size:#x}, "
+                f"{self.mode}, {len(self.registers)} registers)")
+
+
+def windows_from_section(section: dict) -> list[dict]:
+    """Expand a profile observer section into per-window declarations.
+
+    A window entry may carry ``count`` and ``stride`` (header-proven register
+    instances, e.g. the 16 FBPA partitions); labels get an ``-<i>`` suffix
+    when more than one instance is declared.  Window-level ``mode`` and
+    ``registers`` override the section-level values.
+    """
+    section_regs = section.get("registers") or {}
+    section_mode = section.get("mode", "observe-only")
+    section_label = section.get("label", "observe")
+    out: list[dict] = []
+    for win in section.get("windows", ()):
+        base = _int(win["base"])
+        size = _int(win.get("size", 0x1000))
+        count = int(win.get("count", 1))
+        stride = _int(win.get("stride", size))
+        label = str(win.get("label", section_label))
+        mode = win.get("mode", section_mode)
+        regs = win.get("registers", section_regs)
+        if count < 1:
+            raise ValueError(f"observer window {label}: count must be >= 1")
+        for i in range(count):
+            out.append({
+                "base": base + i * stride,
+                "size": size,
+                "label": label if count == 1 else f"{label}-{i}",
+                "mode": mode,
+                "registers": regs,
+            })
+    return out
+
+
+def parse_window_spec(text: str) -> dict:
+    """Parse ``BASE:SIZE[:LABEL[:MODE]]`` for a runtime observer window.
+
+    Bases/sizes accept ``0x`` forms.  This is the escape hatch for windows
+    whose base cannot be established from evidence yet (docs/42); it never
+    invents a value, it only records accesses.
+    """
+    parts = str(text).split(":")
+    if len(parts) < 2:
+        raise ValueError(f"observer window spec {text!r}: expected BASE:SIZE[:LABEL[:MODE]]")
+    base, size = _int(parts[0]), _int(parts[1])
+    if base < 0 or size <= 0:
+        raise ValueError(f"observer window spec {text!r}: bad base/size")
+    label = parts[2] if len(parts) > 2 and parts[2] else "runtime-observe"
+    mode = parts[3] if len(parts) > 3 and parts[3] else "observe-only"
+    if mode not in MODES:
+        raise ValueError(f"observer window spec {text!r}: unknown mode {mode!r}")
+    return {"base": base, "size": size, "label": label, "mode": mode, "registers": {}}
+
+
+def collect_observations(registry) -> list[ObservedWindow]:
+    """All observer windows in a device registry, in registration order."""
+    return [d for d in getattr(registry, "devices", ()) if isinstance(d, ObservedWindow)]
+
+
+def _counts(entries) -> dict:
+    out = {"r": 0, "w": 0, "served": 0}
+    for entry in entries:
+        out[entry["rw"]] += 1
+        if entry["served"]:
+            out["served"] += 1
+    return out
+
+
+def _decorate(entry: dict, resolve) -> dict:
+    out = dict(entry)
+    out["addr"] = hex(entry["addr"])
+    out["offset"] = hex(entry["offset"])
+    if entry["value"] is not None:
+        out["value"] = hex(entry["value"])
+    if entry["pc"] is not None:
+        out["pc"] = hex(entry["pc"])
+    out["pc_symbol"] = None
+    if resolve is not None and entry["pc"] is not None:
+        out["pc_symbol"] = resolve(entry["pc"])
+    return out
+
+
+def build_trace(registry, *, image_sha256: str | None = None,
+                profile: str | None = None, profile_name: str | None = None,
+                symbols_path: str | None = None, resolve=None) -> dict:
+    """Build the ``greeneye-gmmu-fbif-trace/1`` document.
+
+    Accesses are grouped per window and per declared register; offsets inside
+    an observed window that the profile does not name go to ``unclassified``
+    (recorded, never guessed).  ``resolve`` is an optional ``pc -> str``
+    callable (``SymbolDB``-backed in :mod:`sim.gsp_boot`).
+
+    Truncation is never silent: a window's access log is capped at
+    ``ObservedWindow.max_access_log`` entries (default 65536);
+    ``dropped_accesses`` reports the accesses beyond the cap in each window
+    and in the summary, and the summary separates ``accesses`` (total
+    observed) from ``retained_accesses`` (present in this document).
+    """
+    windows = []
+    total = {"windows": 0, "windows_touched": 0, "accesses": 0,
+             "retained_accesses": 0, "dropped_accesses": 0,
+             "reads": 0, "writes": 0, "served": 0, "registers_touched": 0}
+    for dev in collect_observations(registry):
+        total["windows"] += 1
+        win = {
+            "label": dev.label,
+            "base": hex(dev.base),
+            "size": hex(dev.size),
+            "mode": dev.mode,
+            "evidence": dev.evidence,
+            "counts": dict(dev.counts),
+            "dropped_accesses": dev.dropped,
+            "retained_accesses": len(dev.accesses),
+            "registers": [],
+            "unclassified": {"counts": {"r": 0, "w": 0, "served": 0},
+                             "accesses": []},
+        }
+        grouped: dict[int, list] = {}
+        unclassified: list = []
+        for entry in dev.accesses:
+            if entry["offset"] in dev.registers:
+                grouped.setdefault(entry["offset"], []).append(entry)
+            else:
+                unclassified.append(_decorate(entry, resolve))
+        for offset in sorted(grouped):
+            spec = dev.registers[offset]
+            entries = [_decorate(e, resolve) for e in grouped[offset]]
+            win["registers"].append({
+                "offset": hex(offset),
+                "name": spec["name"],
+                "access": spec["access"],
+                "counts": _counts(entries),
+                "accesses": entries,
+            })
+        win["unclassified"]["accesses"] = unclassified
+        win["unclassified"]["counts"] = _counts(unclassified)
+        observed = dev.counts["r"] + dev.counts["w"]
+        total["dropped_accesses"] += dev.dropped
+        if observed:
+            total["windows_touched"] += 1
+            total["accesses"] += observed
+            total["retained_accesses"] += len(dev.accesses)
+            total["reads"] += dev.counts["r"]
+            total["writes"] += dev.counts["w"]
+            total["served"] += dev.counts["served"]
+            total["registers_touched"] += len(grouped)
+        windows.append(win)
+    doc = {
+        "schema": TRACE_SCHEMA,
+        "image_sha256": image_sha256,
+        "profile": profile,
+        "profile_name": profile_name,
+        "symbols": symbols_path,
+        "summary": total,
+        "windows": windows,
+    }
+    return doc