1 files changed, 156 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/record.py b/debug/risc-v-gsp-rm-vm/greeneye/record.py
new file mode 100644
index 0000000..bf70e58
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/record.py
@@ -0,0 +1,156 @@
+"""Session recording: events.jsonl + snapshots + summary.
+
+A recorded session is the durable artifact green-eye produces: a complete,
+symbol-annotatable map of what the firmware did (hook hits, watch hits,
+memory writes, snapshots, coverage steps) that can be replayed offline,
+diffed between runs (bare-metal vs VFIO-mdev, old GSP vs new GSP) and
+annotated back into Ghidra.
+"""
+from __future__ import annotations
+
+import json
+import os
+import time
+import zlib
+
+FORMAT = "greeneye-session/1"
+
+
+class NullRecorder:
+    dir: str | None = None
+
+    def event(self, event_type: str, **fields) -> None:
+        pass
+
+    def snapshot(self, label: str, addr: int, data: bytes) -> None:
+        pass
+
+    def blob(self, name: str, data: bytes) -> None:
+        pass
+
+    def close(self) -> None:
+        pass
+
+
+class Recorder:
+    def __init__(self, root: str, *, name: str = "session", meta: dict | None = None):
+        self.dir = root
+        os.makedirs(self.dir, exist_ok=True)
+        self.name = name
+        self.meta = dict(meta or {})
+        self.meta.update({"format": FORMAT, "name": name, "created": time.time(), "greeneye": _version()})
+        self._events_path = os.path.join(self.dir, "events.jsonl")
+        self._fh = open(self._events_path, "w", encoding="utf-8")
+        self._t0 = time.monotonic()
+        self._counts: dict[str, int] = {}
+        self._snapshots: list[dict] = []
+        self._closed = False
+
+    @classmethod
+    def new(cls, *, base: str | None = None, name: str = "session", meta: dict | None = None) -> "Recorder":
+        base = base or os.environ.get("GREENEYE_DUMPS") or os.path.join(os.getcwd(), "dumps")
+        stamp = time.strftime("%Y%m%d-%H%M%S")
+        return cls(os.path.join(base, f"{stamp}-{name}"), name=name, meta=meta)
+
+    # -- writing -----------------------------------------------------------
+
+    def event(self, event_type: str, **fields) -> None:
+        if self._closed:
+            return
+        record = {"t": round(time.monotonic() - self._t0, 6), "type": event_type}
+        for key, value in fields.items():
+            record[key] = _jsonable(value)
+        self._fh.write(json.dumps(record, sort_keys=True) + "\n")
+        self._counts[event_type] = self._counts.get(event_type, 0) + 1
+
+    def snapshot(self, label: str, addr: int, data: bytes) -> str:
+        rel = os.path.join("snapshots", f"{label}-{addr:#x}-{len(data)}B.bin")
+        path = os.path.join(self.dir, rel)
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "wb") as fh:
+            fh.write(zlib.compress(data))
+        info = {"label": label, "addr": addr, "size": len(data), "path": rel, "sha256": _sha(data)}
+        self._snapshots.append(info)
+        self.event("snapshot_saved", **info)
+        return rel
+
+    def blob(self, name: str, data: bytes) -> str:
+        rel = os.path.join("blobs", name)
+        path = os.path.join(self.dir, rel)
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "wb") as fh:
+            fh.write(data)
+        self.event("blob_saved", path=rel, size=len(data))
+        return rel
+
+    # -- closing -----------------------------------------------------------
+
+    def close(self, write_summary: bool = True) -> None:
+        if self._closed:
+            return
+        self.event("session_end")
+        self._fh.close()
+        self._closed = True
+        self.meta["counts"] = self._counts
+        self.meta["snapshots"] = self._snapshots
+        with open(os.path.join(self.dir, "session.json"), "w", encoding="utf-8") as fh:
+            json.dump(self.meta, fh, indent=1, sort_keys=True)
+        if write_summary:
+            try:
+                from .analyze import render_summary
+
+                report = render_summary(self.dir)
+                with open(os.path.join(self.dir, "summary.md"), "w", encoding="utf-8") as fh:
+                    fh.write(report)
+            except Exception as exc:  # pragma: no cover - never break close
+                with open(os.path.join(self.dir, "summary-error.txt"), "w", encoding="utf-8") as fh:
+                    fh.write(f"summary failed: {exc}\n")
+
+    def __enter__(self) -> "Recorder":
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self.close()
+
+
+def load_events(session_dir: str) -> list[dict]:
+    path = session_dir
+    if os.path.isdir(session_dir):
+        path = os.path.join(session_dir, "events.jsonl")
+    events: list[dict] = []
+    with open(path, "r", encoding="utf-8") as fh:
+        for line in fh:
+            line = line.strip()
+            if line:
+                events.append(json.loads(line))
+    return events
+
+
+def load_meta(session_dir: str) -> dict:
+    with open(os.path.join(session_dir, "session.json"), "r", encoding="utf-8") as fh:
+        return json.load(fh)
+
+
+def _jsonable(value):
+    if isinstance(value, bytes):
+        return value.hex()
+    if isinstance(value, (set, tuple)):
+        return list(value)
+    if isinstance(value, dict):
+        return {str(k): _jsonable(v) for k, v in value.items()}
+    return value
+
+
+def _sha(data: bytes) -> str:
+    import hashlib
+
+    return hashlib.sha256(data).hexdigest()
+
+
+def _version() -> str:
+    try:
+        from . import __version__
+
+        return __version__
+    except Exception:  # pragma: no cover
+        return "unknown"