path: root/debug/risc-v-gsp-rm-vm/sim/devices/observe.py

"""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