rename (cosmetic): green-eye -> risc-v-gsp-rm-debugger in user-facing names, docs, CLI, launcher and CI workflow; internal greeneye package/module identifiers unchanged (includes in-flight greeneye module edits present in the tree)

8 files changed, 270 insertions, 6 deletions

diff --git a/debug/risc-v-gsp-rm-vm/tests/test_bridge.py b/debug/risc-v-gsp-rm-vm/tests/test_bridge.py
index 7eeb46c..966399a 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_bridge.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_bridge.py
@@ -5,6 +5,7 @@ import json
 import os
 import shutil
 import socket
+import time
 import unittest
 
 import greeneye as ge
@@ -23,6 +24,36 @@ class GhidraCommandTests(unittest.TestCase):
         self.assertIn("MarkGreenEyeHits.java /tmp/hits.tsv", cmd)
 
 
+class BridgeLifecycleTests(unittest.TestCase):
+    """No target needed: start/stop/close are transport-level."""
+
+    def test_refuses_non_loopback_bind(self):
+        bridge = GhidraBridge(object(), host="0.0.0.0", port=0)
+        with self.assertRaises(ValueError):
+            bridge.start()
+
+    def test_close_command_takes_down_the_listener(self):
+        bridge = GhidraBridge(object(), port=0)
+        port = bridge.start()
+        with socket.create_connection(("127.0.0.1", port), timeout=10) as sock:
+            stream = sock.makefile("rwb")
+            stream.write((json.dumps({"id": 1, "cmd": "close", "args": {}}) + "\n").encode())
+            stream.flush()
+            response = json.loads(stream.readline())
+        self.assertTrue(response["ok"])
+        self.assertTrue(response["close"])
+        deadline = time.monotonic() + 5.0
+        while True:
+            try:
+                with socket.create_connection(("127.0.0.1", port), timeout=0.5):
+                    pass
+            except OSError:
+                break
+            if time.monotonic() > deadline:
+                self.fail("listener still accepting connections after <close>")
+            time.sleep(0.05)
+
+
 def qemu_available() -> bool:
     return bool(os.environ.get("GREENEYE_QEMU") or shutil.which("qemu-system-riscv64"))
 
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_harness.py b/debug/risc-v-gsp-rm-vm/tests/test_harness.py
index 55938ae..2e079cd 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_harness.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_harness.py
@@ -1,4 +1,4 @@
-"""Tests for the green-eye automation harness (roadmap R5/R6, H4.1/H4.3).
+"""Tests for the risc-v-gsp-rm-debugger automation harness (roadmap R5/R6, H4.1/H4.3).
 
 The integration tests drive the two harness CLIs as subprocesses against the
 bundled demo firmware and skip cleanly when qemu-system-riscv64 or the demo
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_mcp.py b/debug/risc-v-gsp-rm-vm/tests/test_mcp.py
index 614d397..2df8a4a 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_mcp.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_mcp.py
@@ -1,4 +1,4 @@
-"""Tests for the green-eye MCP server plugin (roadmap R3).
+"""Tests for the risc-v-gsp-rm-debugger MCP server plugin (roadmap R3).
 
 The protocol and tool tests drive an in-memory Session over a stub RSP-shaped
 client, so they need no QEMU and no simulator.  A standalone subprocess test
@@ -278,7 +278,7 @@ class ReadOnlyToolTests(McpTestCase):
                 )
                 payload, is_error = self.call_tool(server, "analyze_current")
                 self.assertFalse(is_error)
-                self.assertIn("# green-eye session summary", payload["report"])
+                self.assertIn("# risc-v-gsp-rm-debugger session summary", payload["report"])
             finally:
                 session.close()
 
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_session_call.py b/debug/risc-v-gsp-rm-vm/tests/test_session_call.py
new file mode 100644
index 0000000..aab173c
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_session_call.py
@@ -0,0 +1,195 @@
+"""Session.call and Session.close invariant tests with a fake RSP client.
+
+These lock in two contracts that hardware behavior alone did not cover:
+
+* ``Session.call`` restores every ABI register plus ``pc`` after both a normal
+  return and a timeout.  QEMU names x8 ``fp`` (not ``s0``), so both spellings
+  must be handled.
+* ``Session.close`` clears every breakpoint and watchpoint it registered and
+  closes the transport.
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import unittest
+
+import greeneye as ge
+from greeneye.errors import RspTimeout
+from greeneye.rsp import StopReply
+from greeneye.session import RESTORE_REGS, Session
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
+DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
+
+#: QEMU's target.xml spelling: x8 is ``fp`` and ``s0`` is not present.
+QEMU_REG_NAMES = (
+    ["zero", "ra", "sp", "gp", "tp", "t0", "t1", "t2", "fp", "s1"]
+    + [f"s{i}" for i in range(2, 12)]
+    + [f"a{i}" for i in range(8)]
+    + [f"t{i}" for i in range(3, 7)]
+    + ["pc"]
+)
+
+
+class FakeClient:
+    """Minimal RSP client double with breakpoint/watchpoint accounting."""
+
+    def __init__(self, *, timeout_on_continue: bool = False, retval: int = 0x1234):
+        self.regs = {name: 0x100 + index for index, name in enumerate(QEMU_REG_NAMES)}
+        self.regs["pc"] = 0x80000000
+        self.writes: list[tuple[str, int]] = []
+        self.sw_breaks: set[int] = set()
+        self.hw_breaks: set[int] = set()
+        self.watches: set[tuple[int, int, str]] = set()
+        self.closed = False
+        self.timeout_on_continue = timeout_on_continue
+        self.retval = retval
+        self.continue_count = 0
+
+    def read_registers(self) -> dict[str, int]:
+        return dict(self.regs)
+
+    def read_reg(self, name: str) -> int:
+        return self.regs["fp" if name == "s0" and "s0" not in self.regs else name]
+
+    def write_reg(self, name: str, value: int) -> None:
+        key = "fp" if name == "s0" and "s0" not in self.regs else name
+        self.regs[key] = value
+        self.writes.append((name, value))
+
+    @property
+    def pc(self) -> int:
+        return self.regs["pc"]
+
+    def set_sw_break(self, addr: int, kind: int = 4) -> None:
+        self.sw_breaks.add(addr)
+
+    def clear_sw_break(self, addr: int, kind: int = 4) -> None:
+        self.sw_breaks.discard(addr)
+
+    def set_hw_break(self, addr: int, kind: int = 4) -> None:
+        self.hw_breaks.add(addr)
+
+    def clear_hw_break(self, addr: int, kind: int = 4) -> None:
+        self.hw_breaks.discard(addr)
+
+    def set_watch(self, addr: int, size: int = 8, kind: str = "w") -> None:
+        self.watches.add((addr, size, kind))
+
+    def clear_watch(self, addr: int, size: int = 8, kind: str = "w") -> None:
+        self.watches.discard((addr, size, kind))
+
+    def continue_(self, timeout: float | None = None) -> StopReply:
+        self.continue_count += 1
+        if self.timeout_on_continue:
+            raise RspTimeout("target never returned")
+        self.regs["pc"] = self.regs["ra"]  # returns to the scratch trampoline
+        self.regs["a0"] = self.retval
+        return StopReply(raw="T05")
+
+    def interrupt(self, timeout: float | None = None) -> StopReply:
+        return StopReply(raw="T02")
+
+    def close(self) -> None:
+        self.closed = True
+
+
+class CallRestoreTests(unittest.TestCase):
+    def _restored(self, client: FakeClient, saved: dict[str, int]) -> list[str]:
+        missing = []
+        for name, value in saved.items():
+            if name not in RESTORE_REGS:
+                continue
+            if client.regs.get(name) != value:
+                missing.append(name)
+        if saved.get("fp") != client.regs.get("fp"):
+            missing.append("fp")
+        return missing
+
+    def test_call_success_restores_abi_regs_and_pc(self):
+        client = FakeClient(retval=0xCAFE)
+        session = Session(client=client, name="fake-success")
+        saved = session.read_regs()
+        result = session.call(0x80001000, (0x41, 0x42), timeout=1.0, scratch=0x801FF000)
+        self.assertEqual(result, 0xCAFE)
+        self.assertEqual(self._restored(client, saved), [])
+        self.assertEqual(client.sw_breaks, set(), "temporary call breakpoint must be gone")
+        session.close()
+
+    def test_call_timeout_restores_abi_regs_and_pc(self):
+        client = FakeClient(timeout_on_continue=True)
+        session = Session(client=client, name="fake-timeout")
+        saved = session.read_regs()
+        with self.assertRaises(RspTimeout):
+            session.call(0x80001000, (), timeout=0.05, scratch=0x801FF000)
+        self.assertEqual(self._restored(client, saved), [])
+        self.assertIn("fp", [name for name, _ in client.writes], "x8/fp must be restored under QEMU's name")
+        self.assertEqual(client.sw_breaks, set(), "temporary call breakpoint must be gone")
+        session.close()
+
+    @unittest.skipUnless(shutil.which("qemu-system-riscv64"), "qemu-system-riscv64 not installed")
+    @unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built")
+    def test_real_qemu_register_names_include_fp_not_s0(self):
+        session = ge.launch(kernel=DEMO_ELF, name="unittest-call-names")
+        try:
+            regs = session.read_regs()
+            self.assertIn("fp", regs)
+            self.assertNotIn("s0", regs)
+            self.assertLessEqual(set(RESTORE_REGS) - {"s0"}, set(regs))
+        finally:
+            session.close()
+
+    @unittest.skipUnless(shutil.which("qemu-system-riscv64"), "qemu-system-riscv64 not installed")
+    @unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built")
+    def test_real_qemu_loop_call_timeout_restores_registers(self):
+        session = ge.launch(kernel=DEMO_ELF, name="unittest-call-loop")
+        loop_addr = 0x80200000  # free RAM past the firmware segments
+        try:
+            session.client.write_mem(loop_addr, b"\x6f\x00\x00\x00")  # jal x0, 0
+            saved = session.read_regs()
+            with self.assertRaises(RspTimeout):
+                session.call(loop_addr, (), timeout=0.3, scratch=0x801FF000)
+            after = session.read_regs()
+            for name in RESTORE_REGS:
+                if name in saved and name != "s0":
+                    self.assertEqual(after.get(name), saved[name], f"{name} not restored after timeout")
+            self.assertEqual(after.get("fp"), saved.get("fp"), "fp (x8/s0) not restored after timeout")
+        finally:
+            session.close()
+
+
+class CloseCleanupTests(unittest.TestCase):
+    def test_close_clears_breakpoints_watchpoints_and_transport(self):
+        client = FakeClient()
+        session = Session(client=client, name="fake-close")
+        session.hook(0x80001000)
+        session.hook(0x80001000)  # two on one address: only drop after the last
+        session.breakpoint(0x80002000, kind="hw")
+        session.watch(0x80003000, size=8, kind="w")
+        session.watch(0x80003040, size=4, kind="r")
+        self.assertTrue(client.sw_breaks)
+        self.assertTrue(client.hw_breaks)
+        self.assertEqual(len(client.watches), 2)
+
+        session.close()
+
+        self.assertEqual(client.sw_breaks, set())
+        self.assertEqual(client.hw_breaks, set())
+        self.assertEqual(client.watches, set())
+        self.assertTrue(client.closed)
+
+    def test_close_removes_only_last_breakpoint_at_address(self):
+        client = FakeClient()
+        session = Session(client=client, name="fake-last-bp")
+        first = session.hook(0x80001000)
+        session.hook(0x80001000)
+        first.detach()
+        self.assertEqual(client.sw_breaks, {0x80001000})
+        session.close()
+        self.assertEqual(client.sw_breaks, set())
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py
index a0635ab..1ccfb04 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py
@@ -118,7 +118,7 @@ class NativeSmokeTests(unittest.TestCase):
             watch_hits = []
             session.watch(ctrl + 16, size=8, kind="w", callback=lambda hit: watch_hits.append(hit))
             session.run(duration=0.6)
-            self.assertTrue(session.target.wait_for_uart("GREEN-EYE UPROC DEMO", timeout=1.0))
+            self.assertTrue(session.target.wait_for_uart("RISC-V GSP-RM DEBUGGER UPROC DEMO", timeout=1.0))
             self.assertGreaterEqual(len(hits), 1, "rpc_sum hook never fired")
             self.assertGreaterEqual(len(watch_hits), 7, "rpc_count watchpoint never fired")
             self.assertEqual(watch_hits[0].addr, ctrl + 16)
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
index c0354d2..850942f 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_smoke_qemu.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_smoke_qemu.py
@@ -31,7 +31,7 @@ class QemuSmokeTests(unittest.TestCase):
             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.assertTrue(session.target.wait_for_uart("RISC-V GSP-RM DEBUGGER 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)
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_value_demos.py b/debug/risc-v-gsp-rm-vm/tests/test_value_demos.py
index 7640e13..2a8e957 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_value_demos.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_value_demos.py
@@ -73,7 +73,7 @@ class ValueDemoTests(unittest.TestCase):
         result = _run_demo("s05_snapshot_diff.py", "native")
         self.assertTrue(result["success"], result)
         self.assertTrue({"result", "rpc_count", "last_rpc", "buf"}.issubset(set(result["changed_field_names"])))
-        self.assertEqual(result["after"]["result"], 15)
+        self.assertEqual(result["after"]["result"], len(b"hello-risc-v-gsp-rm-debugger"))
 
     def test_s09_ghidra_join(self):
         result = _run_demo("s09_grabbing_a_ghidra_address.py", "qemu")
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_version_diff.py b/debug/risc-v-gsp-rm-vm/tests/test_version_diff.py
index df704df..d20e780 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_version_diff.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_version_diff.py
@@ -262,6 +262,44 @@ class RunVersionDiffTests(unittest.TestCase):
             self.assertTrue(os.path.exists(os.path.join(out, "version-diff.json")))
 
 
+class VersionDiffDryRunTests(unittest.TestCase):
+    def test_dry_run_exits_zero_and_touches_nothing(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            old = os.path.join(tmp, "old.bin")
+            new = os.path.join(tmp, "new.bin")
+            for path in (old, new):
+                with open(path, "wb") as fh:
+                    fh.write(NOP * 8)
+            before = {}
+            for name in ("old.bin", "new.bin"):
+                path = os.path.join(tmp, name)
+                with open(path, "rb") as fh:
+                    before[name] = (os.path.getmtime(path), fh.read())
+            out = os.path.join(tmp, "out")
+            rc, stdout, _ = _run_main(
+                version_diff_main,
+                ["--old", old, "--new", new, "--out", out, "--dry-run", "--probe-instructions", "4"],
+            )
+            self.assertEqual(rc, 0)
+            plan = json.loads(stdout[: stdout.find("\n[version-diff]")])
+            self.assertEqual(plan["format"], "greeneye-version-diff-plan/1")
+            self.assertEqual(plan["status"], "planned")
+            self.assertFalse(os.path.exists(out), "dry-run must not create --out")
+            for name, (mtime, data) in before.items():
+                path = os.path.join(tmp, name)
+                self.assertEqual(os.path.getmtime(path), mtime)
+                with open(path, "rb") as fh:
+                    self.assertEqual(fh.read(), data)
+
+    def test_dry_run_rm610_refused(self):
+        rc, _, err = _run_main(
+            version_diff_main,
+            ["--old", "rm610.elf", "--new", "new.elf", "--dry-run"],
+        )
+        self.assertEqual(rc, 3)
+        self.assertIn("docs/23", err)
+
+
 class CiReleaseTests(unittest.TestCase):
     def test_dry_run_exits_zero_and_touches_nothing(self):
         with tempfile.TemporaryDirectory() as tmp: