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, 217 insertions, 0 deletions

diff --git a/debug/risc-v-gsp-rm-vm/sim/libos.py b/debug/risc-v-gsp-rm-vm/sim/libos.py
new file mode 100644
index 0000000..e8280fd
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/sim/libos.py
@@ -0,0 +1,217 @@
+"""Minimal, honest libos syscall surface for the native GSP boot.
+
+Evidence kind: **firmware use + RTOS syscall table.**  The GSP application is
+a libos task that requests kernel services with ``ecall`` (the syscall number
+travels in ``t0``); the native ``Machine`` stops on ``ecall`` by default, so
+booting the real image past kernel init needs a deterministic service model.
+
+Rules:
+
+* every requested syscall is recorded (number, name, arguments, pc,
+  instruction count) so the session shows exactly what the firmware asked for;
+* only *no-ops and modelled* services are acknowledged: task yield, cache
+  invalidation, external-interrupt arming, profiler enable/query, and the
+  timer read (returned from the modeled instruction count, labelled
+  modeled);
+* **waits are never faked**: ``PROCESSOR_SUSPEND``,
+  ``TASK_PORT_SEND_RECV_WAIT``, ``TASK_PANIC`` and every unimplemented
+  syscall return *not handled*, which stops the run with ``R_ECALL`` so the
+  caller can classify the stop (message-loop wait / panic / unsupported
+  service) instead of inventing progress;
+* task switching, DMA, memory queries, pagetable registration, partition
+  switch and interrupt delivery are not modelled.
+
+No timing, scheduling or security claim is made; this is development
+tooling for the boot ladder (docs/35 §3.3).
+"""
+from __future__ import annotations
+
+# Syscall numbers (RTOS ``libos_syscalls.h``).
+TASK_PORT_SEND_RECV_WAIT = 1
+TASK_YIELD = 2
+PROCESSOR_SUSPEND = 3
+DCACHE_INVALIDATE = 4
+MEMORY_QUERY = 5
+TASK_STATE_QUERY = 6
+SHUTTLE_RESET = 7
+TASK_DMA_REGISTER_TCM = 8
+TASK_DMA_COPY = 9
+TASK_DMA_FLUSH = 10
+TASK_MEMORY_READ = 11
+INIT_REGISTER_PAGETABLES = 12
+PROFILER_ENABLE = 13
+TASK_EXIT = 14
+TASK_PANIC = 15
+INTERRUPT_ARM_EXTERNAL = 16
+TIMER_GET_NS = 17
+DCACHE_INVALIDATE_ALL = 18
+PROFILER_GET_COUNTER_MASK = 19
+PARTITION_SWITCH = 20
+LIMIT_POW2 = 32
+
+NAMES = {
+    TASK_PORT_SEND_RECV_WAIT: "task_port_send_recv_wait",
+    TASK_YIELD: "task_yield",
+    PROCESSOR_SUSPEND: "processor_suspend",
+    DCACHE_INVALIDATE: "dcache_invalidate",
+    MEMORY_QUERY: "memory_query",
+    TASK_STATE_QUERY: "task_state_query",
+    SHUTTLE_RESET: "shuttle_reset",
+    TASK_DMA_REGISTER_TCM: "task_dma_register_tcm",
+    TASK_DMA_COPY: "task_dma_copy",
+    TASK_DMA_FLUSH: "task_dma_flush",
+    TASK_MEMORY_READ: "task_memory_read",
+    INIT_REGISTER_PAGETABLES: "init_register_pagetables",
+    PROFILER_ENABLE: "profiler_enable",
+    TASK_EXIT: "task_exit",
+    TASK_PANIC: "task_panic",
+    INTERRUPT_ARM_EXTERNAL: "interrupt_arm_external",
+    TIMER_GET_NS: "timer_get_ns",
+    DCACHE_INVALIDATE_ALL: "dcache_invalidate_all",
+    PROFILER_GET_COUNTER_MASK: "profiler_get_counter_mask",
+    PARTITION_SWITCH: "partition_switch",
+    LIMIT_POW2: "limit_pow2",
+}
+
+#: Services acknowledged as no-ops (result 0); everything else stops the run.
+_NOOP_SYSCALLS = frozenset({
+    TASK_YIELD, DCACHE_INVALIDATE, DCACHE_INVALIDATE_ALL, PROFILER_ENABLE,
+})
+
+
+class LibosSyscalls:
+    """Deterministic libos service model: record, acknowledge, or stop."""
+
+    name = "libos-syscalls"
+
+    def __init__(self) -> None:
+        self.calls: list[dict] = []
+        self.counts: dict[int, int] = {}
+        self.armed_mask = 0
+        #: Last registered TCM bounce buffer (``task_dma_register_tcm``).
+        self.tcm: dict | None = None
+        #: Last inverted-pagetable registration (``init_register_pagetables``).
+        self.pagetables: dict | None = None
+        self.dma_copies = 0
+        self.dma_bytes = 0
+        #: (number, a0, a1, pc, insns) of the last request.
+        self.last: tuple | None = None
+        #: The request that stopped the run (not handled), if any.
+        self.stopped: dict | None = None
+
+    def handle(self, machine, pc: int = 0):
+        """``Machine`` ecall hook: ``(handled, result)``.
+
+        Unhandled requests leave ``machine.run`` stopping with ``R_ECALL``;
+        the number is still recorded here and in ``stopped``.
+        """
+        num = machine.x[5]
+        a0, a1 = machine.x[10], machine.x[11]
+        entry = {
+            "num": num,
+            "name": NAMES.get(num, f"unknown_{num}"),
+            "a0": a0,
+            "a1": a1,
+            "pc": pc,
+            "insns": machine.insns,
+        }
+        self.calls.append(entry)
+        self.counts[num] = self.counts.get(num, 0) + 1
+        self.last = (num, a0, a1, pc, machine.insns)
+        if num == INTERRUPT_ARM_EXTERNAL:
+            self.armed_mask |= a0
+            return True, 0
+        if num == TIMER_GET_NS:
+            # Modeled time: instruction count * 1 ns (docs/36: modeled, not
+            # calibrated; no timing claim).
+            return True, machine.insns
+        if num == TASK_DMA_REGISTER_TCM:
+            # The kernel validates the VA against its pagetable and stores it
+            # as the DMA bounce buffer; without a paging model the region is
+            # recorded and the observed result (LIBOS_OK) returned.
+            self.tcm = {"va": a0, "size": a1}
+            return True, 0
+        if num == TASK_DMA_COPY:
+            return self._dma_copy(machine, a0, a1, machine.x[12], pc)
+        if num == INIT_REGISTER_PAGETABLES:
+            # The kernel builds its inverted page table from this registration
+            # and returns LIBOS_OK.  The simulator does not model the
+            # inverted table (DMA resolution goes through the flat RAM + MPU
+            # + aperture seam); the registration is recorded as evidence.
+            self.pagetables = {"base": a0, "entries": a1, "probe_count": machine.x[12]}
+            return True, 0
+        if num == TASK_DMA_FLUSH:
+            return True, 0
+        if num in _NOOP_SYSCALLS:
+            return True, 0
+        if num == PROFILER_GET_COUNTER_MASK:
+            return True, 0
+        self.stopped = entry
+        return False, None
+
+    # -- DMA service model -------------------------------------------------
+
+    def _resolve(self, machine, va: int):
+        """VA -> simulator RAM offset through the MPU and aperture seam."""
+        if machine.mpu.active:
+            translated = machine.mpu.translate(va)
+            if translated is not None:
+                va = translated
+        off = va - machine.ram_base
+        if 0 <= off < machine.ram_size:
+            return off
+        translator = machine.translator
+        if translator is not None:
+            sim = translator.translate(va)
+            if sim is not None:
+                off = sim - machine.ram_base
+                if 0 <= off < machine.ram_size:
+                    return off
+        return None
+
+    def _dma_copy(self, machine, dst_va: int, src_va: int, size: int, pc: int = 0):
+        """Modeled DMA copy: equivalent to a RAM memcpy in the flat model.
+
+        The real kernel drives the falcon DMA engine through TCM; the
+        simulator keeps all regions in one RAM, so the copy effect is the
+        same.  Both sides must resolve or the run stops (never a faked copy).
+        """
+        if size == 0:
+            return True, 0
+        dst = self._resolve(machine, dst_va)
+        src = self._resolve(machine, src_va)
+        if dst is None or src is None:
+            self.stopped = {"num": TASK_DMA_COPY, "name": "task_dma_copy",
+                            "a0": dst_va, "a1": src_va, "size": size,
+                            "reason": "unresolved", "pc": pc}
+            return False, None
+        ram = machine.ram
+        if (0 <= dst <= machine.ram_size - size
+                and 0 <= src <= machine.ram_size - size):
+            ram[dst:dst + size] = bytes(ram[src:src + size])
+        else:
+            # Straddling or out-of-range: byte-wise through the bounds.
+            for i in range(size):
+                if dst + i >= machine.ram_size or src + i >= machine.ram_size:
+                    self.stopped = {"num": TASK_DMA_COPY, "name": "task_dma_copy",
+                                    "a0": dst_va, "a1": src_va, "size": size,
+                                    "reason": "out-of-range", "pc": pc}
+                    return False, None
+                ram[dst + i] = ram[src + i]
+        machine.invalidate()
+        self.dma_copies += 1
+        self.dma_bytes += size
+        return True, 0
+
+    def state(self) -> dict:
+        return {
+            "calls": len(self.calls),
+            "counts": {str(k): v for k, v in sorted(self.counts.items())},
+            "armed_mask": self.armed_mask,
+            "tcm": self.tcm,
+            "pagetables": self.pagetables,
+            "dma_copies": self.dma_copies,
+            "dma_bytes": self.dma_bytes,
+            "last": list(self.last) if self.last else None,
+            "stopped": self.stopped,
+        }