debug: add green-eye GSP/uproc RISC-V VM runtime analysis tooling

Adds open-iov/debug/risc-v-gsp-rm-vm: a GDB-free instrumentation stack that boots uproc/GSP firmware in qemu-system-riscv64 and provides Frida-like hooks, breakpoints, watchpoints, memory dumps, structure decoding, snapshot diffs, recorded sessions, declarative probes, a plugin API and a Ghidra symbol/hit bridge.

Verified: 25 tests green (including a live QEMU smoke test); the demo uproc boots with correct hook/leave/watch/schema/snapshot/fault evidence; the optional TCG trace plugin builds and recorded 15M trace lines in 3 s; the TU104 570.144 fwimage probe reports its first exception (load access fault at firmware vaddr 0x4009014 reading 0x2000000000000A00).

Documentation: lean tool docs under debug/risc-v-gsp-rm-vm/docs/00-10, a new docs/05-operations/03-risc-v-gsp-rm-vm.md debugging document, and '# TOOL IN DEVELOPMENT - NOT READY YET!' markers in docs/README.md, the onboarding tooling table and the debugging workflows doc.

Vendors qemu-plugin.h from QEMU v8.2.2 (GPL-2.0-or-later), with its license header intact, so the optional TCG plugin builds without a QEMU source tree.

6 files changed, 550 insertions, 0 deletions

diff --git a/debug/risc-v-gsp-rm-vm/tests/__init__.py b/debug/risc-v-gsp-rm-vm/tests/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/__init__.py
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_core.py b/debug/risc-v-gsp-rm-vm/tests/test_core.py
new file mode 100644
index 0000000..ab27451
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_core.py
@@ -0,0 +1,131 @@
+"""Unit tests for ELF parsing, symbol resolution and structure decoding."""
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+
+from greeneye.elf import EM_RISCV, parse_elf, load_elf
+from greeneye.structs import Field, Schema, decode, render, schema_from_json
+from greeneye.symbols import Symbol, SymbolDB
+
+DEMO_ELF = os.path.join(os.path.dirname(__file__), "..", "firmware", "uproc_demo", "build", "uproc_demo.elf")
+
+
+class ElfTests(unittest.TestCase):
+    def test_parse_demo_elf(self):
+        if not os.path.exists(DEMO_ELF):
+            self.skipTest("demo firmware not built")
+        image = load_elf(DEMO_ELF)
+        self.assertTrue(image.is_riscv)
+        self.assertEqual(image.e_machine, EM_RISCV)
+        self.assertEqual(image.entry, 0x80000000)
+        self.assertGreaterEqual(len(image.load_segments), 1)
+        names = {s.name for s in image.symbols}
+        self.assertIn("main", names)
+        self.assertIn("rpc_ping", names)
+        self.assertIn("gsp_ctrl", names)
+        ctrl = next(s for s in image.symbols if s.name == "gsp_ctrl")
+        self.assertEqual(ctrl.value, 0x80000840)
+        self.assertEqual(ctrl.size, 0x188)
+        plan = image.flat_load_plan()
+        self.assertTrue(plan and plan[0][0] == 0x80000000)
+
+    def test_not_elf(self):
+        with self.assertRaises(Exception):
+            parse_elf(b"not an elf")
+
+
+class SymbolTests(unittest.TestCase):
+    def setUp(self):
+        self.db = SymbolDB()
+        self.db.add(Symbol(addr=0x1000, name="alpha", size=0x100, source="test"))
+        self.db.add(Symbol(addr=0x1200, name="beta", size=0x20, source="test"))
+
+    def test_resolve(self):
+        self.assertEqual(self.db.resolve(0x1010).name, "alpha")
+        self.assertEqual(self.db.resolve(0x1210).name, "beta")
+        self.assertIsNone(self.db.resolve(0x5000))
+        self.assertEqual(self.db.annotate(0x1010), "0x1010 alpha+0x10")
+        self.assertEqual(self.db.lookup("beta"), 0x1200)
+
+    def test_names_tsv(self):
+        with tempfile.NamedTemporaryFile("w", suffix=".tsv", delete=False) as fh:
+            fh.write("# comment\n0x00279000\t_DT_INIT\tghidra\n0x00279020\tFUN_x\n")
+            path = fh.name
+        try:
+            count = self.db.load_names_tsv(path, bias=0x100000)
+            self.assertEqual(count, 2)
+            self.assertEqual(self.db.lookup("_DT_INIT"), 0x279000 + 0x100000)
+        finally:
+            os.unlink(path)
+
+    def test_index_tsv(self):
+        with tempfile.NamedTemporaryFile("w", suffix=".tsv", delete=False) as fh:
+            fh.write("addr\tfile_addr\tsize\tname\n")
+            fh.write("0x00279000\t0x00179000\t24\t_DT_INIT\n")
+            path = fh.name
+        try:
+            count = self.db.load_index_tsv(path, bias=0)
+            self.assertEqual(count, 1)
+            sym = self.db.get("_DT_INIT")
+            self.assertEqual(sym.addr, 0x179000)
+            self.assertEqual(sym.size, 24)
+            self.assertEqual(sym.extras["program_addr"], 0x279000)
+        finally:
+            os.unlink(path)
+
+    def test_json_roundtrip(self):
+        with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh:
+            path = fh.name
+        try:
+            self.db.save_json(path, meta={"unit": "test"})
+            other = SymbolDB()
+            self.assertEqual(other.load_json(path), 2)
+            self.assertEqual(other.lookup("alpha"), 0x1000)
+            payload = json.load(open(path))
+            self.assertEqual(payload["format"], "greeneye-symbols/1")
+        finally:
+            os.unlink(path)
+
+
+class StructTests(unittest.TestCase):
+    SCHEMA = {
+        "name": "demo",
+        "fields": [
+            {"name": "magic", "offset": 0, "fmt": "u32"},
+            {"name": "state", "offset": 4, "fmt": "u32", "enum": {"0": "IDLE", "1": "RUN"}},
+            {"name": "count", "offset": 8, "fmt": "u64"},
+            {"name": "name", "offset": 16, "fmt": "cstr", "size": 16},
+            {"name": "buf", "offset": 32, "fmt": "bytes", "size": 8},
+        ],
+    }
+
+    def test_decode_render(self):
+        schema = schema_from_json(self.SCHEMA)
+        self.assertEqual(schema.size, 40)
+        data = bytearray(40)
+        data[0:4] = (0x47535031).to_bytes(4, "little")
+        data[4:8] = (1).to_bytes(4, "little")
+        data[8:16] = (42).to_bytes(8, "little")
+        data[16:22] = b"green\x00"
+        data[32:40] = bytes(range(8))
+        values = decode(schema, bytes(data), base=0x80000000)
+        self.assertEqual(values["magic"], 0x47535031)
+        self.assertEqual(values["state"], 1)
+        self.assertEqual(values["count"], 42)
+        self.assertEqual(values["name"], "green")
+        self.assertEqual(values["buf"], bytes(range(8)))
+        text = render(schema, values)
+        self.assertIn("magic", text)
+        self.assertIn("0x47535031", text)
+        self.assertIn("green", text)
+
+    def test_invalid_schema(self):
+        with self.assertRaises(Exception):
+            Field(name="x", offset=0, fmt="nope").nbytes
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_memory_record.py b/debug/risc-v-gsp-rm-vm/tests/test_memory_record.py
new file mode 100644
index 0000000..24b360d
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_memory_record.py
@@ -0,0 +1,101 @@
+"""Unit tests for memory helpers and session recording."""
+from __future__ import annotations
+
+import os
+import shutil
+import tempfile
+import unittest
+
+from greeneye.memory import Memory
+from greeneye.record import NullRecorder, Recorder, load_events, load_meta
+from greeneye.symbols import Symbol, SymbolDB
+
+
+class FakeClient:
+    def __init__(self, size=0x1000):
+        self.data = bytearray(size)
+        for i in range(size):
+            self.data[i] = i & 0xFF
+
+    def read_mem(self, addr, size, chunk=0):
+        return bytes(self.data[addr : addr + size])
+
+    def write_mem(self, addr, data, chunk=0):
+        self.data[addr : addr + len(data)] = data
+
+
+class MemoryTests(unittest.TestCase):
+    def setUp(self):
+        self.mem = Memory(FakeClient())
+
+    def test_read_write_hexdump(self):
+        self.assertEqual(self.mem.read(0x10, 4), bytes(range(0x10, 0x14)))
+        self.mem.write(0x20, b"AB")
+        self.assertEqual(self.mem.read(0x20, 2), b"AB")
+        text = self.mem.hexdump(0, 16)
+        self.assertIn("00000000", text)
+        self.assertIn("00 01 02", text)
+        self.assertIn("|", text)
+
+    def test_search(self):
+        self.assertEqual(self.mem.search(0, 0x40, b"\x10\x11"), [0x10])
+        self.assertEqual(self.mem.search_string(0, 0x40, "\x20\x21"), [0x20])
+        masked = self.mem.search(0, 0x100, b"\xa0", mask=b"\xf0")
+        self.assertEqual(masked[0], 0xA0)
+        self.mem.write(0x30, b"GREEN")
+        self.assertEqual(self.mem.search_string(0, 0x100, "GREEN"), [0x30])
+
+    def test_snapshot_and_diff(self):
+        before = self.mem.snapshot(0, 32, "before")
+        self.mem.write(0x10, b"\xff")
+        after = self.mem.snapshot(0, 32, "after")
+        diffs = Memory.diff(before, after, base=0)
+        self.assertEqual(len(diffs), 1)
+        self.assertEqual(diffs[0].addr, 0x10)
+        self.assertEqual(diffs[0].new, b"\xff")
+        text = Memory.render_diff(diffs)
+        self.assertIn("00000010", text)
+
+    def test_hexdump_annotated(self):
+        db = SymbolDB()
+        db.add(Symbol(addr=0, name="block", size=0x100, source="test"))
+        mem = Memory(FakeClient(), symbols=db)
+        text = mem.hexdump(0, 16, annotate=True)
+        self.assertIn("block", text)
+
+
+class RecorderTests(unittest.TestCase):
+    def setUp(self):
+        self.dir = tempfile.mkdtemp(prefix="greeneye-test-")
+        self.recorder = Recorder(self.dir, name="unit", meta={"note": "test"})
+        self.recorder.event("alpha", value=1)
+        self.recorder.event("alpha", value=2)
+        self.recorder.snapshot("regs", 0x80000000, b"\x00" * 16)
+
+    def tearDown(self):
+        shutil.rmtree(self.dir, ignore_errors=True)
+
+    def test_events_and_meta(self):
+        self.recorder.close()
+        events = load_events(self.dir)
+        kinds = [e["type"] for e in events]
+        self.assertEqual(kinds.count("alpha"), 2)
+        self.assertIn("snapshot_saved", kinds)
+        self.assertIn("session_end", kinds)
+        meta = load_meta(self.dir)
+        self.assertEqual(meta["name"], "unit")
+        self.assertEqual(meta["note"], "test")
+        self.assertTrue(os.path.exists(os.path.join(self.dir, "summary.md")))
+        summary = open(os.path.join(self.dir, "summary.md")).read()
+        self.assertIn("unit", summary)
+
+    def test_null_recorder(self):
+        null = NullRecorder()
+        null.event("x")
+        null.snapshot("y", 0, b"")
+        null.close()
+        self.assertIsNone(null.dir)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_probe.py b/debug/risc-v-gsp-rm-vm/tests/test_probe.py
new file mode 100644
index 0000000..f97a4bf
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_probe.py
@@ -0,0 +1,65 @@
+"""Unit tests for the ELF trampoline/flat-kernel path and QEMU fault parsing."""
+from __future__ import annotations
+
+import os
+import struct
+import tempfile
+import unittest
+
+from greeneye.analyze import describe_fault, parse_qemu_faults
+from greeneye.elf import _trampoline, load_elf, write_flat_kernel
+
+DEMO_ELF = os.path.join(os.path.dirname(__file__), "..", "firmware", "uproc_demo", "build", "uproc_demo.elf")
+
+SAMPLE_LOG = """
+Invalid read at addr 0x2000000000000a00, size 4, region '(null)', reason: rejected
+riscv_cpu_do_interrupt: hart:0, async:0, cause:0000000000000005, epc:0x0000000084009014, tval:0x2000000000000a00, desc=fault_load
+Invalid read at addr 0x0, size 2, region '(null)', reason: rejected
+riscv_cpu_do_interrupt: hart:0, async:0, cause:0000000000000001, epc:0x0000000000000000, tval:0x0000000000000000, desc=fault_fetch
+"""
+
+
+class TrampolineTests(unittest.TestCase):
+    def test_encoding(self):
+        stub = _trampoline(0x4009000)
+        auipc, jalr = struct.unpack("<II", stub)
+        self.assertEqual(auipc & 0x7F, 0x17)  # AUIPC
+        self.assertEqual((auipc >> 7) & 0x1F, 5)  # t0
+        self.assertEqual((auipc >> 12) & 0xFFFFF, 0x4009)
+        self.assertEqual(jalr & 0x7F, 0x67)  # JALR
+        self.assertEqual((jalr >> 15) & 0x1F, 5)
+        self.assertEqual(jalr >> 20, 0)
+
+    def test_flat_demo(self):
+        if not os.path.exists(DEMO_ELF):
+            self.skipTest("demo firmware not built")
+        with tempfile.TemporaryDirectory() as tmp:
+            out = os.path.join(tmp, "flat.bin")
+            path, entry = write_flat_kernel(DEMO_ELF, out, load_addr=0x80000000)
+            self.assertEqual(entry, 0x80000000)
+            image = load_elf(DEMO_ELF)
+            self.assertEqual(os.path.getsize(path), image.max_vaddr - image.min_vaddr)
+            with open(path, "rb") as fh:
+                # entry at offset 0: the real first instruction must be present
+                first = fh.read(4)
+            self.assertNotEqual(first, b"\x00\x00\x00\x00")
+
+
+class FaultParseTests(unittest.TestCase):
+    def test_parse_and_describe(self):
+        faults = parse_qemu_faults(SAMPLE_LOG)
+        self.assertEqual(len(faults), 2)
+        first = faults[0]
+        self.assertEqual(first["cause"], 5)
+        self.assertEqual(first["epc"], 0x84009014)
+        self.assertEqual(first["access"]["access_addr"], 0x2000000000000A00)
+        self.assertEqual(first["access"]["size"], 4)
+        text = describe_fault(first, bias=0x80000000)
+        self.assertIn("0x4009014", text)
+        self.assertIn("load access fault", text)
+        self.assertIn("0x2000000000000a00", text)
+        self.assertEqual(faults[1]["desc"], "fault_fetch")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_rsp.py b/debug/risc-v-gsp-rm-vm/tests/test_rsp.py
new file mode 100644
index 0000000..70b9f45
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_rsp.py
@@ -0,0 +1,175 @@
+"""Unit tests for the RSP client using an in-process fake stub."""
+from __future__ import annotations
+
+import socket
+import threading
+import unittest
+
+from greeneye.rsp import (
+    RV64_FALLBACK_REGS,
+    RspClient,
+    checksum,
+    encode_packet,
+    parse_stop_reply,
+    _parse_target_xml,
+)
+
+TARGET_XML = """<?xml version="1.0"?>
+<target><architecture>riscv:rv64</architecture>
+<feature name="org.gnu.gdb.riscv.cpu">
+<reg name="x0" bitsize="64" regnum="0"/>
+<reg name="ra" bitsize="64" regnum="1"/>
+<reg name="pc" bitsize="64" regnum="32"/>
+</feature></target>"""
+
+
+class FakeStub(threading.Thread):
+    """Minimal RSP server for unit tests."""
+
+    def __init__(self, sock: socket.socket):
+        super().__init__(daemon=True)
+        self.sock = sock
+        self.no_ack = False
+        self.stopped = True
+        self.regs = bytearray(33 * 8)
+        self.regs[8:16] = (0xAA).to_bytes(8, "little")  # ra (x1)
+        self.regs[16:24] = (0x80000194).to_bytes(8, "little")  # pc (3-reg test layout)
+        self.memory = bytearray(0x10000)
+        for i in range(len(self.memory)):
+            self.memory[i] = i & 0xFF
+        self.bps: set[int] = set()
+
+    def run(self):
+        buf = b""
+        while True:
+            try:
+                chunk = self.sock.recv(4096)
+            except OSError:
+                return
+            if not chunk:
+                return
+            buf += chunk
+            while b"#" in buf and b"$" in buf:
+                start = buf.index(b"$")
+                end = buf.index(b"#", start)
+                if len(buf) < end + 3:
+                    break
+                payload = buf[start + 1 : end]
+                buf = buf[end + 3 :]
+                self.handle(payload)
+
+    def send(self, text: str):
+        data = encode_packet(text.encode())
+        self.sock.sendall(data)
+
+    def handle(self, payload: bytes):
+        cmd = payload.decode("latin1")
+        if not self.no_ack:
+            self.sock.sendall(b"+")
+        if cmd.startswith("qSupported"):
+            self.send("PacketSize=1000;qXfer:features:read+;multiprocess+")
+        elif cmd == "QStartNoAckMode":
+            self.send("")
+            self.no_ack = True
+        elif cmd == "?":
+            self.send("T05thread:p01.01;")
+        elif cmd in ("Hg0", "Hc-1"):
+            self.send("OK")
+        elif cmd.startswith("qXfer:features:read:target.xml"):
+            self.send("l" + TARGET_XML)
+        elif cmd.startswith("qXfer"):
+            self.send("E01")
+        elif cmd == "g":
+            self.send(bytes(self.regs).hex())
+        elif cmd.startswith("p"):
+            regno = int(cmd[1:], 16)
+            if regno == 32:
+                self.send(self.regs[16:24].hex())
+            else:
+                self.send(self.regs[regno * 8 : regno * 8 + 8].hex())
+        elif cmd.startswith("m"):
+            addr_s, len_s = cmd[1:].split(",")
+            addr, length = int(addr_s, 16), int(len_s, 16)
+            self.send(bytes(self.memory[addr : addr + length]).hex())
+        elif cmd.startswith("M"):
+            head, data = cmd[1:].split(":", 1)
+            addr_s, len_s = head.split(",")
+            addr, length = int(addr_s, 16), int(len_s, 16)
+            blob = bytes.fromhex(data)
+            self.memory[addr : addr + length] = blob
+            self.send("OK")
+        elif cmd.startswith("Z0"):
+            self.bps.add(int(cmd.split(",")[1], 16))
+            self.send("OK")
+        elif cmd.startswith("z0"):
+            self.bps.discard(int(cmd.split(",")[1], 16))
+            self.send("OK")
+        elif cmd.startswith("Z2") or cmd.startswith("z2"):
+            self.send("OK")
+        elif cmd in ("c", "s"):
+            self.stopped = True
+            self.send("T05thread:p01.01;")
+        else:
+            self.send("")
+
+
+class RspTests(unittest.TestCase):
+    def setUp(self):
+        self.client_sock, self.server_sock = socket.socketpair()
+        self.stub = FakeStub(self.server_sock)
+        self.stub.start()
+        self.client = RspClient(timeout=2.0)
+        self.client.sock = self.client_sock
+        self.client.sync()
+
+    def tearDown(self):
+        self.client.close()
+        self.server_sock.close()
+
+    def test_checksum_and_encode(self):
+        self.assertEqual(checksum(b"OK"), (ord("O") + ord("K")) & 0xFF)
+        self.assertEqual(encode_packet(b"OK"), b"$OK#9a")
+
+    def test_parse_stop_reply(self):
+        stop = parse_stop_reply("T05watch:0000000080001000;thread:p01.01;")
+        self.assertEqual(stop.signal, 5)
+        self.assertEqual(stop.watch_addr, 0x80001000)
+        self.assertEqual(stop.watch_kind, "watch")
+        self.assertEqual(stop.thread, "p01.01")
+        self.assertTrue(stop.is_stop)
+        exit_stop = parse_stop_reply("W00")
+        self.assertTrue(exit_stop.is_exit)
+
+    def test_target_xml_parse(self):
+        regs = _parse_target_xml(TARGET_XML)
+        self.assertEqual([r.name for r in regs], ["x0", "ra", "pc"])
+        self.assertEqual([r.number for r in regs], [0, 1, 32])
+        by_name = {r.name: r for r in regs}
+        self.assertEqual(by_name["ra"].offset, 8)
+        self.assertEqual(by_name["pc"].offset, 16)
+        self.assertEqual(len(RV64_FALLBACK_REGS), 33)
+        self.assertEqual(RV64_FALLBACK_REGS[32].name, "pc")
+
+    def test_registers_and_memory(self):
+        regs = self.client.read_registers()
+        self.assertEqual(regs["ra"], 0xAA)
+        self.assertEqual(regs["pc"], 0x80000194)
+        self.assertEqual(self.client.read_reg("pc"), 0x80000194)
+        data = self.client.read_mem(0x10, 8)
+        self.assertEqual(data, bytes(range(0x10, 0x18)))
+        self.client.write_mem(0x20, b"\xde\xad\xbe\xef")
+        self.assertEqual(self.client.read_mem(0x20, 4), b"\xde\xad\xbe\xef")
+        self.assertEqual(self.client.read_u32(0x20), 0xEFBEADDE)
+
+    def test_breakpoints_watchpoints_continue(self):
+        self.client.set_sw_break(0x80000048)
+        self.client.clear_sw_break(0x80000048)
+        self.client.set_watch(0x80000850, size=8, kind="w")
+        stop = self.client.continue_()
+        self.assertEqual(stop.signal, 5)
+        stop = self.client.step()
+        self.assertEqual(stop.signal, 5)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_smoke_qemu.py b/debug/risc-v-gsp-rm-vm/tests/test_smoke_qemu.py
new file mode 100644
index 0000000..c0354d2
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_smoke_qemu.py
@@ -0,0 +1,78 @@
+"""Live integration smoke test: boots the demo uproc in QEMU.
+
+Skipped cleanly when qemu-system-riscv64 or the demo firmware is absent.
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import unittest
+
+import greeneye as ge
+from greeneye.errors import GreenEyeError
+from greeneye.probes import compile_condition, install
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+DEMO_ELF = os.path.join(HERE, "..", "firmware", "uproc_demo", "build", "uproc_demo.elf")
+
+
+def qemu_available() -> bool:
+    return bool(os.environ.get("GREENEYE_QEMU") or shutil.which("qemu-system-riscv64"))
+
+
+@unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed")
+@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)")
+class QemuSmokeTests(unittest.TestCase):
+    def test_boot_hook_watch_snapshot_analyze(self):
+        session = ge.launch(kernel=DEMO_ELF, record=True, name="unittest-smoke")
+        hits = []
+        try:
+            ctrl = session.symbols.lookup("gsp_ctrl")
+            session.hook("rpc_sum", on_enter=lambda hit: hits.append(hit))
+            session.watch(ctrl + 16, size=8, kind="w")
+            session.run(duration=0.6)
+            self.assertTrue(session.target.wait_for_uart("GREEN-EYE UPROC DEMO", timeout=1.0))
+            self.assertGreaterEqual(len(hits), 1, "rpc_sum hook never fired")
+            after = session.snapshot(ctrl, 0x188, "ctrl")
+            self.assertEqual(int.from_bytes(after[0:4], "little"), 0x47535031)
+            self.assertGreaterEqual(int.from_bytes(after[16:24], "little"), 7)
+        finally:
+            directory = getattr(session.recorder, "dir", None)
+            session.close()
+        if directory:
+            from greeneye.analyze import render_summary
+
+            report = render_summary(directory)
+            self.assertIn("rpc_sum", report)
+
+    def test_probe_spec_install(self):
+        session = ge.launch(kernel=DEMO_ELF, name="unittest-probes")
+        try:
+            spec = {
+                "probes": [
+                    {
+                        "name": "sum",
+                        "symbol": "rpc_sum",
+                        "when": "a1 > 0",
+                        "on_enter": {"log": ["a0", "a1"], "note": "sum called"},
+                    }
+                ]
+            }
+            handles = install(session, spec)
+            self.assertEqual(len(handles), 1)
+            session.run(duration=0.6)
+            self.assertEqual(handles[0].hits, 1)  # self-test calls rpc_sum once
+        finally:
+            session.close()
+
+    def test_condition_guard(self):
+        self.assertIsNone(compile_condition(None))
+        check = compile_condition("a0 == 0x2a and a1 < 4")
+        self.assertTrue(check({"a0": 0x2A, "a1": 2}))
+        self.assertFalse(check({"a0": 0x2B, "a1": 2}))
+        with self.assertRaises(GreenEyeError):
+            compile_condition("__import__('os').system('true')")
+
+
+if __name__ == "__main__":
+    unittest.main()