summaryrefslogtreecommitdiffstats
path: root/debug/risc-v-gsp-rm-vm/harness/rm_capture.py
diff options
context:
space:
mode:
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/rm_capture.py')
-rw-r--r--debug/risc-v-gsp-rm-vm/harness/rm_capture.py1251
1 files changed, 1251 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/rm_capture.py b/debug/risc-v-gsp-rm-vm/harness/rm_capture.py
new file mode 100644
index 0000000..4aa459b
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/rm_capture.py
@@ -0,0 +1,1251 @@
+#!/usr/bin/env python3
+"""Real-client GSP RPC capture ingestion, comparison and replay prep.
+
+Ingests the text output of the supervised host-client capture (the GSP RPC
+send-path instrumentation described by the CAPTURE-RUNBOOK) and turns it into
+structured, replayable artifacts:
+
+* **decode** — a raw capture log (dmesg text with the documented ``GE-SEND``,
+ ``GE-CTRL``, ``GE-ALLOC``, ``GE-GOLDEN-*`` and ``GE-CTRL-DONE`` tags) into a
+ ``greeneye-rm-rpc-capture/1`` document: one ordered record per RPC/alloc,
+ with the wire function id, the named control/class id, handles, caller
+ return address, and the captured parameter bytes.
+* **compare** — two capture documents (for example, the real host client's
+ sequence vs. the open-iov host ladder's replay, or two firmware/driver
+ versions) aligned by record identity, with per-record parameter agreement,
+ coverage and explicit unmatched lists.
+* **stimulus** — a ``greeneye-rm-rpc-stimulus/1`` ordered message list for the
+ simulator's B4/B5 CSB/message-queue seam (not wired yet).
+
+Honesty rules:
+
+* Truncated payloads are flagged, never silently padded. The runbook's hex
+ helper caps a dump at 192 bytes; a capture produced by that helper will mark
+ every larger payload ``truncated`` so a reader can see what is missing.
+* Handles, physical addresses and caller addresses are capture-specific; the
+ compare report labels them, and the stimulus output says so explicitly.
+* This tool only reads files and writes reports. It never touches a GPU,
+ kernel module, mdev, guest or network. The capture it ingests is a
+ supervised, human-run physical action (RUNBOOK P1/P7); this tool never runs
+ or instructs it beyond printing the tag grammar.
+* No timing/cycle claim is made anywhere; fidelity is L0/L1 (docs/06).
+"""
+from __future__ import annotations
+
+import argparse
+import difflib
+import hashlib
+import json
+import os
+import re
+import sys
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if TOOL_ROOT not in sys.path:
+ sys.path.insert(0, TOOL_ROOT)
+
+from greeneye.errors import GreenEyeError # noqa: E402
+
+#: Schema tag of the decoded capture document.
+CAPTURE_FORMAT = "greeneye-rm-rpc-capture/1"
+#: Schema tag of the comparison report.
+COMPARE_FORMAT = "greeneye-rm-rpc-compare/1"
+#: Schema tag of the replay stimulus document.
+STIMULUS_FORMAT = "greeneye-rm-rpc-stimulus/1"
+#: Schema tag of the demo result line.
+DEMO_FORMAT = "greeneye-rm-rpc-capture-demo/1"
+
+#: Size of ``rpc_message_header_v``: the wire payload starts after it.
+RPC_HEADER_BYTES = 32
+
+# Wire function ids (``rpc_global_enums.h``): used to derive the embedded
+# control/class id from a ``GE-SEND`` payload when no named ``GE-CTRL`` line
+# was captured for it.
+GSP_RM_CONTROL = 76
+GSP_RM_ALLOC = 103
+# Payload offsets inside the flat parameter blob (runbook §4).
+CONTROL_CMD_OFFSET = 8
+ALLOC_CLASS_OFFSET = 12
+
+KINDS = ("send", "control", "alloc", "golden", "ctrl-done")
+HEX_KIND_TAG = {"send": "SEND", "control": "CTRL", "alloc": "ALLOC"}
+TAG_KIND = {v: k for k, v in HEX_KIND_TAG.items()}
+
+# FROZEN BEFORE THE FIRST MEASUREMENT (2026-10-06). Do not move silently:
+# a threshold change requires a new recorded run and a note in docs/45.
+#: Conservative ESTIMATE (not measured on a human) of decoding one golden
+#: sequence by eye from a capture log: grep the tags, count records, walk each
+#: parameter blob against the RPC structs, transcribe into a replay table.
+MANUAL_DECODE_BASELINE_SECONDS = 1800
+MANUAL_BASELINE_LABEL = "conservative estimate (not measured on a human)"
+#: The tool path must stay at or below this fraction of the estimate.
+WORKFLOW_THRESHOLD_FRACTION = 0.05
+
+_HEX_RE = re.compile(
+ r"GE-(SEND|CTRL|ALLOC)\s+([0-9a-fA-F]{1,4})\s*:\s*((?:[0-9a-fA-F]{2}\s*)+)"
+)
+_KV_RE = re.compile(r"([A-Za-z_][A-Za-z0-9_]*)=(\S+)")
+_TS_RE = re.compile(r"^\[([^\]]+)\]\s*")
+_GOLDEN_TRIGGER_RE = re.compile(
+ r"GE-GOLDEN-TRIGGER\s+gspclient=(\d+)(?:\s+mig=(\d+))?"
+)
+_GOLDEN_END_RE = re.compile(r"GE-GOLDEN-END\s+status=(\S+)")
+_GOLDEN_BEGIN_RE = re.compile(r"GE-GOLDEN-BEGIN")
+_CTRL_DONE_RE = re.compile(r"GE-CTRL-DONE\b")
+
+
+class RmCaptureError(GreenEyeError):
+ """A capture input is absent, unreadable or not in a supported schema."""
+
+
+def _parse_int(value: Any, default: int | None = None) -> int | None:
+ """Parse ints, decimal strings and ``0x`` strings; ``default`` on failure."""
+ if value is None:
+ return default
+ if isinstance(value, bool):
+ return int(value)
+ if isinstance(value, int):
+ return value
+ if isinstance(value, float):
+ return int(value)
+ if isinstance(value, str):
+ text = value.strip()
+ if not text:
+ return default
+ try:
+ return int(text, 0)
+ except ValueError:
+ try:
+ return int(text, 16)
+ except ValueError:
+ return default
+ return default
+
+
+def _sha256_bytes(data: bytes) -> str:
+ return hashlib.sha256(data).hexdigest()
+
+
+def _sha256_file(path: str) -> str:
+ digest = hashlib.sha256()
+ with open(path, "rb") as fh:
+ for chunk in iter(lambda: fh.read(1 << 20), b""):
+ digest.update(chunk)
+ return digest.hexdigest()
+
+
+def canonical_json(payload: Any) -> str:
+ """Deterministic JSON text (sorted keys, fixed indent, trailing newline)."""
+ return json.dumps(payload, indent=2, sort_keys=True) + "\n"
+
+
+@dataclass
+class CaptureRecord:
+ """One decoded capture record (an RPC, an alloc, or a golden marker)."""
+
+ index: int
+ kind: str
+ line: int
+ ts: str = ""
+ fn: int | None = None
+ length: int | None = None
+ size: int | None = None
+ cmd: int | None = None
+ hclass: int | None = None
+ hclient: int | None = None
+ hobject: int | None = None
+ hparent: int | None = None
+ caller: int | None = None
+ caller_raw: str = ""
+ marker: str = ""
+ gspclient: int | None = None
+ mig: int | None = None
+ status: int | None = None
+ rpc_status: int | None = None
+ params_status: int | None = None
+ chunks: dict[int, bytes] = field(default_factory=dict)
+ anomalies: list[str] = field(default_factory=list)
+
+ # -- parameter assembly -------------------------------------------------
+ @property
+ def params(self) -> bytes:
+ """Captured parameter bytes, assembled from the hex chunks in order.
+
+ A gap between chunks is zero-filled and recorded as an anomaly; the
+ raw chunk map stays available on the record.
+ """
+ if not self.chunks:
+ return b""
+ end = max(offset + len(data) for offset, data in self.chunks.items())
+ out = bytearray(end)
+ for offset, data in sorted(self.chunks.items()):
+ out[offset : offset + len(data)] = data
+ return bytes(out)
+
+ @property
+ def expected_params_size(self) -> int | None:
+ """Declared parameter size, or ``None`` when the tag does not carry one."""
+ if self.kind == "send":
+ if self.length is None or self.length < RPC_HEADER_BYTES:
+ return None
+ return self.length - RPC_HEADER_BYTES
+ if self.kind in ("control", "alloc"):
+ return self.size
+ return None
+
+ @property
+ def truncated(self) -> bool:
+ expected = self.expected_params_size
+ return expected is not None and len(self.params) < expected
+
+ @property
+ def overrun(self) -> bool:
+ expected = self.expected_params_size
+ return expected is not None and len(self.params) > expected
+
+ @property
+ def wire_id(self) -> int | None:
+ """Named id: ``cmd``/``hclass`` when captured, else derived from payload."""
+ if self.kind == "control":
+ return self.cmd
+ if self.kind == "alloc":
+ return self.hclass
+ if self.kind == "send":
+ payload = self.params
+ if self.fn == GSP_RM_CONTROL and len(payload) >= CONTROL_CMD_OFFSET + 4:
+ return int.from_bytes(
+ payload[CONTROL_CMD_OFFSET : CONTROL_CMD_OFFSET + 4], "little"
+ )
+ if self.fn == GSP_RM_ALLOC and len(payload) >= ALLOC_CLASS_OFFSET + 4:
+ return int.from_bytes(
+ payload[ALLOC_CLASS_OFFSET : ALLOC_CLASS_OFFSET + 4], "little"
+ )
+ return None
+
+ @property
+ def signature(self) -> str:
+ """Identity used for compare alignment (kind + id + declared size)."""
+ if self.kind == "golden":
+ return f"golden:{self.marker}"
+ if self.kind == "ctrl-done":
+ return f"ctrl-done:{self.cmd:#x}" if self.cmd is not None else "ctrl-done:?"
+ identifier = self.wire_id
+ id_text = f"{identifier:#x}" if identifier is not None else "?"
+ fn_text = f"{self.fn:#x}" if self.fn is not None else "?"
+ size = self.expected_params_size
+ size_text = f"{size:#x}" if size is not None else "?"
+ return f"{self.kind}:{id_text}:{fn_text}:{size_text}"
+
+ @property
+ def label(self) -> str:
+ """Short human label for reports."""
+ if self.kind == "golden":
+ return f"GOLDEN {self.marker}" + (
+ f" status={self.status:#x}" if self.status is not None else ""
+ )
+ identifier = self.wire_id
+ id_text = f"{identifier:#x}" if identifier is not None else "?"
+ if self.kind == "send":
+ return f"SEND fn={self.fn:#x} id={id_text}" if self.fn is not None else "SEND"
+ if self.kind == "control":
+ return f"CTRL cmd={id_text}"
+ if self.kind == "alloc":
+ return f"ALLOC cls={id_text}"
+ return f"CTRL-DONE cmd={id_text}"
+
+
+@dataclass
+class RmCapture:
+ """A decoded capture document."""
+
+ path: str
+ sha256: str
+ records: list[CaptureRecord] = field(default_factory=list)
+ source: str = ""
+ format: str = CAPTURE_FORMAT
+ warnings: list[str] = field(default_factory=list)
+
+ @property
+ def truncated_records(self) -> list[CaptureRecord]:
+ return [r for r in self.records if r.truncated]
+
+ def kind_counts(self) -> dict[str, int]:
+ counts: dict[str, int] = {}
+ for record in self.records:
+ counts[record.kind] = counts.get(record.kind, 0) + 1
+ return counts
+
+ def fn_counts(self) -> dict[str, int]:
+ counts: dict[str, int] = {}
+ for record in self.records:
+ if record.kind == "send" and record.fn is not None:
+ key = f"0x{record.fn:x}"
+ counts[key] = counts.get(key, 0) + 1
+ return counts
+
+ def golden_status(self) -> int | None:
+ for record in self.records:
+ if record.kind == "golden" and record.marker == "end":
+ return record.status
+ return None
+
+
+# ---------------------------------------------------------------------------
+# Decode
+# ---------------------------------------------------------------------------
+
+
+def _kv_fields(line: str) -> dict[str, str]:
+ return {key: value for key, value in _KV_RE.findall(line)}
+
+
+def _new_record(index: int, kind: str, line_no: int, ts: str) -> CaptureRecord:
+ return CaptureRecord(index=index, kind=kind, line=line_no, ts=ts)
+
+
+def decode_text(text: str, source: str = "<text>") -> RmCapture:
+ """Decode capture-log text into a capture document.
+
+ Lines are matched anywhere in the line (dmesg/NVRM prefixes are fine).
+ Hex lines attach to the most recent record of their own tag; a hex line
+ with no open record of that tag is recorded as an ``orphan_hex`` warning
+ instead of being guessed onto some other record.
+ """
+ records: list[CaptureRecord] = []
+ warnings: list[str] = []
+ open_records: dict[str, CaptureRecord] = {}
+
+ for line_no, raw_line in enumerate(text.splitlines(), start=1):
+ ts_match = _TS_RE.search(raw_line)
+ ts = ts_match.group(1) if ts_match else ""
+
+ hex_match = _HEX_RE.search(raw_line)
+ if hex_match:
+ tag = hex_match.group(1)
+ kind = TAG_KIND[tag]
+ offset = int(hex_match.group(2), 16)
+ data = bytes.fromhex(re.sub(r"\s+", "", hex_match.group(3)))
+ record = open_records.get(kind)
+ if record is None:
+ warnings.append(f"line {line_no}: orphan {tag} hex block (no header)")
+ continue
+ if offset in record.chunks:
+ record.anomalies.append(f"duplicate hex offset {offset:#x}")
+ record.chunks[offset] = data
+ continue
+
+ if _GOLDEN_TRIGGER_RE.search(raw_line):
+ match = _GOLDEN_TRIGGER_RE.search(raw_line)
+ record = _new_record(len(records), "golden", line_no, ts)
+ record.marker = "trigger"
+ record.gspclient = int(match.group(1)) if match else None
+ record.mig = int(match.group(2)) if match and match.group(2) else None
+ records.append(record)
+ continue
+
+ if _GOLDEN_BEGIN_RE.search(raw_line):
+ record = _new_record(len(records), "golden", line_no, ts)
+ record.marker = "begin"
+ records.append(record)
+ continue
+
+ if _GOLDEN_END_RE.search(raw_line):
+ match = _GOLDEN_END_RE.search(raw_line)
+ record = _new_record(len(records), "golden", line_no, ts)
+ record.marker = "end"
+ record.status = _parse_int(match.group(1) if match else None)
+ records.append(record)
+ continue
+
+ if _CTRL_DONE_RE.search(raw_line):
+ fields = _kv_fields(raw_line)
+ record = _new_record(len(records), "ctrl-done", line_no, ts)
+ record.cmd = _parse_int(fields.get("cmd"))
+ record.rpc_status = _parse_int(fields.get("rpc_status"))
+ record.params_status = _parse_int(fields.get("params_status"))
+ records.append(record)
+ continue
+
+ header_kind = None
+ for tag, kind in TAG_KIND.items():
+ if re.search(rf"GE-{tag}\b", raw_line):
+ header_kind = kind
+ break
+ if header_kind is None:
+ continue
+
+ fields = _kv_fields(raw_line)
+ record = _new_record(len(records), header_kind, line_no, ts)
+ if header_kind == "send":
+ record.fn = _parse_int(fields.get("fn"))
+ record.length = _parse_int(fields.get("len"))
+ elif header_kind == "control":
+ record.cmd = _parse_int(fields.get("cmd"))
+ record.hclient = _parse_int(fields.get("hClient"))
+ record.hobject = _parse_int(fields.get("hObject"))
+ record.size = _parse_int(fields.get("size"))
+ elif header_kind == "alloc":
+ record.hclass = _parse_int(fields.get("cls"))
+ record.hclient = _parse_int(fields.get("hClient"))
+ record.hparent = _parse_int(fields.get("hParent"))
+ record.hobject = _parse_int(fields.get("hObject"))
+ record.size = _parse_int(fields.get("size"))
+ caller_raw = fields.get("ret", "")
+ record.caller_raw = caller_raw
+ record.caller = _parse_int(caller_raw)
+ records.append(record)
+ open_records[header_kind] = record
+
+ # Post-pass: chunk-gap and size anomalies.
+ for record in records:
+ expected = record.expected_params_size
+ if record.chunks and 0 not in record.chunks:
+ record.anomalies.append("hex chunks do not start at offset 0")
+ if record.overrun:
+ record.anomalies.append(
+ f"captured {len(record.params)} bytes > declared {expected} bytes"
+ )
+
+ return RmCapture(
+ path=os.path.abspath(source),
+ sha256=_sha256_bytes(text.encode("utf-8", "replace")),
+ records=records,
+ source=source,
+ warnings=warnings,
+ )
+
+
+def decode_file(path: str) -> RmCapture:
+ if not os.path.isfile(path):
+ raise RmCaptureError(f"capture log not found: {path}")
+ try:
+ with open(path, "r", encoding="utf-8", errors="replace") as fh:
+ text = fh.read()
+ except OSError as exc:
+ raise RmCaptureError(f"cannot read capture log {path}: {exc}") from exc
+ return decode_text(text, source=path)
+
+
+def load_capture(path: str) -> RmCapture:
+ """Load a ``greeneye-rm-rpc-capture/1`` document or decode a raw log."""
+ if not os.path.isfile(path):
+ raise RmCaptureError(f"capture not found: {path}")
+ with open(path, "r", encoding="utf-8", errors="replace") as fh:
+ head = fh.read(4096)
+ if not head.lstrip().startswith("{"):
+ return decode_file(path)
+ try:
+ with open(path, "r", encoding="utf-8", errors="replace") as fh:
+ payload = json.load(fh)
+ except (OSError, json.JSONDecodeError) as exc:
+ raise RmCaptureError(f"cannot read capture {path}: {exc}") from exc
+ if not isinstance(payload, dict):
+ raise RmCaptureError(f"{path}: capture must be an object")
+ fmt = str(payload.get("format") or payload.get("schema") or "")
+ if fmt != CAPTURE_FORMAT:
+ raise RmCaptureError(f"{path}: format {fmt!r} is not {CAPTURE_FORMAT!r}")
+ raw_records = payload.get("records")
+ if not isinstance(raw_records, list):
+ raise RmCaptureError(f"{path}: missing 'records' list")
+ records: list[CaptureRecord] = []
+ for index, raw in enumerate(raw_records):
+ if not isinstance(raw, dict):
+ continue
+ record = _new_record(
+ index, str(raw.get("kind") or "unknown"), int(raw.get("line") or 0),
+ str(raw.get("ts") or ""),
+ )
+ record.fn = _parse_int(raw.get("fn"))
+ record.length = _parse_int(raw.get("length"))
+ record.size = _parse_int(raw.get("size"))
+ record.cmd = _parse_int(raw.get("cmd"))
+ record.hclass = _parse_int(raw.get("hclass"))
+ record.hclient = _parse_int(raw.get("hclient"))
+ record.hobject = _parse_int(raw.get("hobject"))
+ record.hparent = _parse_int(raw.get("hparent"))
+ record.caller = _parse_int(raw.get("caller"))
+ record.caller_raw = str(raw.get("caller_raw") or "")
+ record.marker = str(raw.get("marker") or "")
+ record.gspclient = _parse_int(raw.get("gspclient"))
+ record.mig = _parse_int(raw.get("mig"))
+ record.status = _parse_int(raw.get("status"))
+ record.rpc_status = _parse_int(raw.get("rpc_status"))
+ record.params_status = _parse_int(raw.get("params_status"))
+ params_hex = raw.get("params_hex")
+ if isinstance(params_hex, str) and params_hex:
+ try:
+ record.chunks[0] = bytes.fromhex(re.sub(r"\s+", "", params_hex))
+ except ValueError as exc:
+ raise RmCaptureError(f"{path}: record {index}: bad params hex: {exc}") from exc
+ records.append(record)
+ capture = RmCapture(
+ path=os.path.abspath(path),
+ sha256=_sha256_file(path),
+ records=records,
+ source=str(payload.get("source") or path),
+ )
+ warnings = payload.get("warnings")
+ if isinstance(warnings, list):
+ capture.warnings = [str(w) for w in warnings]
+ return capture
+
+
+# ---------------------------------------------------------------------------
+# Compare
+# ---------------------------------------------------------------------------
+
+
+def _param_agreement(a: CaptureRecord, b: CaptureRecord) -> dict:
+ pa, pb = a.params, b.params
+ matched = sum(1 for x, y in zip(pa, pb) if x == y)
+ first_diff = None
+ for offset, (x, y) in enumerate(zip(pa, pb)):
+ if x != y:
+ first_diff = offset
+ break
+ if first_diff is None and len(pa) != len(pb):
+ first_diff = min(len(pa), len(pb))
+ denominator = max(len(pa), len(pb))
+ return {
+ "a_bytes": len(pa),
+ "b_bytes": len(pb),
+ "matched_bytes": matched,
+ "byte_agreement_pct": (
+ round(100.0 * matched / denominator, 2) if denominator else None
+ ),
+ "first_diff_offset": first_diff,
+ "a_truncated": a.truncated,
+ "b_truncated": b.truncated,
+ }
+
+
+def _pair_status(a: CaptureRecord, b: CaptureRecord) -> tuple[str, dict]:
+ if a.kind == "golden":
+ if (
+ a.status == b.status
+ and a.gspclient == b.gspclient
+ and a.mig == b.mig
+ ):
+ return "agree", {}
+ return "mismatch", {
+ "a_status": a.status,
+ "b_status": b.status,
+ "a_gspclient": a.gspclient,
+ "b_gspclient": b.gspclient,
+ "a_mig": a.mig,
+ "b_mig": b.mig,
+ }
+ if a.kind == "ctrl-done":
+ if a.rpc_status == b.rpc_status and a.params_status == b.params_status:
+ return "agree", {}
+ return "mismatch", {
+ "a_rpc_status": a.rpc_status,
+ "b_rpc_status": b.rpc_status,
+ "a_params_status": a.params_status,
+ "b_params_status": b.params_status,
+ }
+ agreement = _param_agreement(a, b)
+ if a.truncated or b.truncated:
+ return "partial", agreement
+ if agreement["a_bytes"] == agreement["b_bytes"] and agreement["first_diff_offset"] is None:
+ return "agree", agreement
+ return "shape-agree", agreement
+
+
+def compare(a: RmCapture, b: RmCapture) -> dict:
+ """Ordered, per-record comparison of two captures; no timing claims."""
+ sig_a = [record.signature for record in a.records]
+ sig_b = [record.signature for record in b.records]
+ matcher = difflib.SequenceMatcher(a=sig_a, b=sig_b, autojunk=False)
+ rows: list[dict] = []
+ a_only: list[dict] = []
+ b_only: list[dict] = []
+ counts = {
+ "agree": 0,
+ "shape-agree": 0,
+ "partial": 0,
+ "mismatch": 0,
+ "a-only": 0,
+ "b-only": 0,
+ }
+ for opcode, i1, i2, j1, j2 in matcher.get_opcodes():
+ if opcode == "equal":
+ for offset in range(i2 - i1):
+ record_a = a.records[i1 + offset]
+ record_b = b.records[j1 + offset]
+ status, detail = _pair_status(record_a, record_b)
+ counts[status] += 1
+ rows.append(
+ {
+ "a_index": record_a.index,
+ "b_index": record_b.index,
+ "a_label": record_a.label,
+ "b_label": record_b.label,
+ "signature": record_a.signature,
+ "status": status,
+ "detail": detail,
+ "a_truncated": record_a.truncated,
+ "b_truncated": record_b.truncated,
+ "a_caller": record_a.caller,
+ "b_caller": record_b.caller,
+ }
+ )
+ else:
+ for index in range(i1, i2):
+ counts["a-only"] += 1
+ record = a.records[index]
+ a_only.append(
+ {
+ "index": record.index,
+ "label": record.label,
+ "signature": record.signature,
+ "params_bytes": len(record.params),
+ "line": record.line,
+ }
+ )
+ for index in range(j1, j2):
+ counts["b-only"] += 1
+ record = b.records[index]
+ b_only.append(
+ {
+ "index": record.index,
+ "label": record.label,
+ "signature": record.signature,
+ "params_bytes": len(record.params),
+ "line": record.line,
+ }
+ )
+ matched = counts["agree"] + counts["shape-agree"] + counts["partial"] + counts["mismatch"]
+ byte_rows = [
+ row["detail"].get("byte_agreement_pct")
+ for row in rows
+ if isinstance(row.get("detail"), dict)
+ and row["detail"].get("byte_agreement_pct") is not None
+ ]
+ return {
+ "format": COMPARE_FORMAT,
+ "inputs": {
+ "a": {
+ "path": a.path,
+ "sha256": a.sha256,
+ "source": a.source,
+ "records": len(a.records),
+ "kinds": a.kind_counts(),
+ "fn_counts": a.fn_counts(),
+ "golden_status": a.golden_status(),
+ "truncated_records": len(a.truncated_records),
+ },
+ "b": {
+ "path": b.path,
+ "sha256": b.sha256,
+ "source": b.source,
+ "records": len(b.records),
+ "kinds": b.kind_counts(),
+ "fn_counts": b.fn_counts(),
+ "golden_status": b.golden_status(),
+ "truncated_records": len(b.truncated_records),
+ },
+ },
+ "summary": {
+ "a_records": len(a.records),
+ "b_records": len(b.records),
+ "matched": matched,
+ "agree": counts["agree"],
+ "shape_agree": counts["shape-agree"],
+ "partial": counts["partial"],
+ "mismatch": counts["mismatch"],
+ "a_only": counts["a-only"],
+ "b_only": counts["b-only"],
+ "agreement_pct": round(100.0 * counts["agree"] / matched, 2) if matched else None,
+ "coverage_pct": round(100.0 * matched / len(a.records), 2) if a.records else None,
+ "mean_byte_agreement_pct": (
+ round(sum(byte_rows) / len(byte_rows), 2) if byte_rows else None
+ ),
+ "claim": (
+ "per-record identity and parameter agreement on the aligned intersection; "
+ "not a timing, silicon or full-coverage claim"
+ ),
+ },
+ "rows": rows,
+ "a_only": a_only,
+ "b_only": b_only,
+ "fidelity": {
+ "a": f"file-ingested capture ({a.source or 'unlabelled'})",
+ "b": f"file-ingested capture ({b.source or 'unlabelled'})",
+ "levels": "L0/L1 only; L2 timing and L3 device environment not claimed (docs/06)",
+ },
+ "limits": [
+ "This tool never ran the capture: a supervised human produced both inputs.",
+ "Handles, physical addresses and caller return addresses are capture-specific and are labelled, not normalised.",
+ "A truncated payload is reported as partial/truncated, never padded to look complete.",
+ "A-only/b-only records are ordering or presence differences under this alignment; they are not proof of a wrong value.",
+ ],
+ "warnings": (a.warnings if a.warnings else []) + (b.warnings if b.warnings else []),
+ }
+
+
+def render_compare_markdown(report: dict) -> str:
+ summary = report["summary"]
+ lines = [
+ "# GSP RPC capture comparison",
+ "",
+ f"> Format `{report['format']}`. Fidelity: {report['fidelity']['levels']}",
+ "> No cycle/timing claim. Underclaim every number (docs/06, docs/45).",
+ "",
+ "## Inputs",
+ "",
+ f"* A: `{report['inputs']['a']['path']}` "
+ f"({report['inputs']['a']['records']} records, "
+ f"sha256 `{report['inputs']['a']['sha256'][:16]}…`"
+ + (
+ f", golden status `{report['inputs']['a']['golden_status']:#x}`"
+ if report["inputs"]["a"]["golden_status"] is not None
+ else ""
+ )
+ + ")",
+ f"* B: `{report['inputs']['b']['path']}` "
+ f"({report['inputs']['b']['records']} records, "
+ f"sha256 `{report['inputs']['b']['sha256'][:16]}…`"
+ + (
+ f", golden status `{report['inputs']['b']['golden_status']:#x}`"
+ if report["inputs"]["b"]["golden_status"] is not None
+ else ""
+ )
+ + ")",
+ "",
+ "## Summary",
+ "",
+ "| matched | agree | shape-agree | partial | mismatch | A-only | B-only | agreement % | coverage % | mean byte agreement % |",
+ "|---|---|---|---|---|---|---|---|---|---|",
+ f"| {summary['matched']} | {summary['agree']} | {summary['shape_agree']} | "
+ f"{summary['partial']} | {summary['mismatch']} | {summary['a_only']} | {summary['b_only']} | "
+ f"{summary['agreement_pct'] if summary['agreement_pct'] is not None else 'n/a'} | "
+ f"{summary['coverage_pct'] if summary['coverage_pct'] is not None else 'n/a'} | "
+ f"{summary['mean_byte_agreement_pct'] if summary['mean_byte_agreement_pct'] is not None else 'n/a'} |",
+ "",
+ f"* {summary['claim']}",
+ "",
+ "## Per-record rows",
+ "",
+ "| A | B | status | bytes A/B | byte agreement % | first diff | truncated A/B |",
+ "|---|---|---|---|---|---|---|",
+ ]
+ for row in report["rows"]:
+ detail = row.get("detail") if isinstance(row.get("detail"), dict) else {}
+ a_bytes = detail.get("a_bytes", "")
+ b_bytes = detail.get("b_bytes", "")
+ byte_pct = detail.get("byte_agreement_pct")
+ first_diff = detail.get("first_diff_offset")
+ lines.append(
+ f"| `{row['a_label']}` | `{row['b_label']}` | {row['status']} | "
+ f"{a_bytes}/{b_bytes} | {byte_pct if byte_pct is not None else ''} | "
+ f"{first_diff if first_diff is not None else ''} | "
+ f"{'Y' if row.get('a_truncated') else '-'}/{'Y' if row.get('b_truncated') else '-'} |"
+ )
+ lines += ["", "## A-only records", ""]
+ if report["a_only"]:
+ for item in report["a_only"]:
+ lines.append(
+ f"* #{item['index']} `{item['label']}` ({item['params_bytes']} B, line {item['line']})"
+ )
+ else:
+ lines.append("* none")
+ lines += ["", "## B-only records", ""]
+ if report["b_only"]:
+ for item in report["b_only"]:
+ lines.append(
+ f"* #{item['index']} `{item['label']}` ({item['params_bytes']} B, line {item['line']})"
+ )
+ else:
+ lines.append("* none")
+ lines += ["", "## Limits", ""]
+ for limit in report["limits"]:
+ lines.append(f"* {limit}")
+ if report.get("warnings"):
+ lines += ["", "## Decode warnings", ""]
+ for warning in report["warnings"]:
+ lines.append(f"* {warning}")
+ lines.append("")
+ return "\n".join(lines)
+
+
+def run_decode(log_path: str, out_dir: str | None) -> dict:
+ capture = decode_file(log_path)
+ payload = capture_to_dict(capture)
+ if out_dir:
+ os.makedirs(out_dir, exist_ok=True)
+ json_path = os.path.join(out_dir, "rm-capture.json")
+ with open(json_path, "w", encoding="utf-8") as fh:
+ fh.write(canonical_json(payload))
+ md_path = os.path.join(out_dir, "rm-capture.md")
+ with open(md_path, "w", encoding="utf-8") as fh:
+ fh.write(render_capture_markdown(capture))
+ payload["artifacts"] = {
+ "capture_json": os.path.abspath(json_path),
+ "capture_md": os.path.abspath(md_path),
+ }
+ return payload
+
+
+def run_compare(a_path: str, b_path: str, out_dir: str) -> dict:
+ capture_a = load_capture(a_path)
+ capture_b = load_capture(b_path)
+ report = compare(capture_a, capture_b)
+ os.makedirs(out_dir, exist_ok=True)
+ json_path = os.path.join(out_dir, "rm-capture-compare.json")
+ md_path = os.path.join(out_dir, "rm-capture-compare.md")
+ with open(json_path, "w", encoding="utf-8") as fh:
+ fh.write(canonical_json(report))
+ with open(md_path, "w", encoding="utf-8") as fh:
+ fh.write(render_compare_markdown(report))
+ report["artifacts"] = {
+ "report_json": os.path.abspath(json_path),
+ "report_md": os.path.abspath(md_path),
+ }
+ return report
+
+
+def run_stimulus(capture_path: str, out_path: str | None) -> dict:
+ capture = load_capture(capture_path)
+ payload = stimulus_from_capture(capture)
+ if out_path:
+ os.makedirs(os.path.dirname(os.path.abspath(out_path)), exist_ok=True)
+ with open(out_path, "w", encoding="utf-8") as fh:
+ fh.write(canonical_json(payload))
+ payload["artifacts"] = {"stimulus_json": os.path.abspath(out_path)}
+ return payload
+
+
+# ---------------------------------------------------------------------------
+# Serialisation / stimulus
+# ---------------------------------------------------------------------------
+
+
+def record_to_dict(record: CaptureRecord) -> dict:
+ out: dict[str, Any] = {
+ "index": record.index,
+ "kind": record.kind,
+ "line": record.line,
+ }
+ if record.ts:
+ out["ts"] = record.ts
+ for key in (
+ "fn",
+ "length",
+ "size",
+ "cmd",
+ "hclass",
+ "hclient",
+ "hobject",
+ "hparent",
+ "caller",
+ ):
+ value = getattr(record, key)
+ if value is not None:
+ out[key] = value
+ if record.caller_raw:
+ out["caller_raw"] = record.caller_raw
+ if record.marker:
+ out["marker"] = record.marker
+ if record.gspclient is not None:
+ out["gspclient"] = record.gspclient
+ if record.mig is not None:
+ out["mig"] = record.mig
+ for key in ("status", "rpc_status", "params_status"):
+ value = getattr(record, key)
+ if value is not None:
+ out[key] = value
+ params = record.params
+ if params:
+ out["params_hex"] = params.hex()
+ out["params_bytes"] = len(params)
+ out["params_size_expected"] = record.expected_params_size
+ out["truncated"] = record.truncated
+ if record.anomalies:
+ out["anomalies"] = list(record.anomalies)
+ return out
+
+
+def capture_to_dict(capture: RmCapture) -> dict:
+ return {
+ "format": CAPTURE_FORMAT,
+ "source": capture.source,
+ "sha256": capture.sha256,
+ "records": [record_to_dict(record) for record in capture.records],
+ "summary": {
+ "records": len(capture.records),
+ "kinds": capture.kind_counts(),
+ "fn_counts": capture.fn_counts(),
+ "truncated_records": len(capture.truncated_records),
+ "golden_status": capture.golden_status(),
+ },
+ "warnings": list(capture.warnings),
+ "limits": [
+ "Decoded from the documented capture tags only (GE-SEND/GE-CTRL/GE-ALLOC/GE-GOLDEN-*/GE-CTRL-DONE).",
+ "Handles and addresses are capture-specific; do not reuse them across boots.",
+ "Truncated payloads are flagged (the runbook hex helper caps at 192 bytes unless extended).",
+ "This tool never ran the capture; the log is a supervised human artifact.",
+ ],
+ }
+
+
+def render_capture_markdown(capture: RmCapture) -> str:
+ summary = capture_to_dict(capture)["summary"]
+ lines = [
+ "# GSP RPC capture",
+ "",
+ f"> Format `{CAPTURE_FORMAT}`; source `{capture.source}`; "
+ f"sha256 `{capture.sha256[:16]}…`. Fidelity L0/L1; no timing claim.",
+ "",
+ "## Summary",
+ "",
+ f"* records: {summary['records']} {summary['kinds']}",
+ f"* wire fn counts: {summary['fn_counts']}",
+ f"* truncated records: {summary['truncated_records']}",
+ f"* golden end status: {summary['golden_status']}",
+ "",
+ "## Records",
+ "",
+ "| # | kind | ts | id | size | captured B | expected B | truncated | caller |",
+ "|---|---|---|---|---|---|---|---|---|",
+ ]
+ for record in capture.records:
+ identifier = record.wire_id
+ id_text = f"{identifier:#x}" if identifier is not None else ""
+ expected = record.expected_params_size
+ lines.append(
+ f"| {record.index} | {record.kind} | {record.ts} | {id_text} | "
+ f"{record.size if record.size is not None else ''} | {len(record.params)} | "
+ f"{expected if expected is not None else ''} | "
+ f"{'Y' if record.truncated else '-'} | "
+ f"{f'{record.caller:#x}' if record.caller is not None else record.caller_raw} |"
+ )
+ if capture.warnings:
+ lines += ["", "## Decode warnings", ""]
+ for warning in capture.warnings:
+ lines.append(f"* {warning}")
+ lines += ["", "## Limits", ""]
+ for limit in capture_to_dict(capture)["limits"]:
+ lines.append(f"* {limit}")
+ lines.append("")
+ return "\n".join(lines)
+
+
+def stimulus_from_capture(capture: RmCapture) -> dict:
+ messages: list[dict] = []
+ markers: list[dict] = []
+ for record in capture.records:
+ if record.kind == "golden":
+ markers.append(
+ {
+ "line": record.line,
+ "marker": record.marker,
+ "status": record.status,
+ "gspclient": record.gspclient,
+ }
+ )
+ continue
+ if record.kind == "ctrl-done":
+ continue
+ message: dict[str, Any] = {
+ "seq": len(messages),
+ "kind": record.kind,
+ "params_hex": record.params.hex(),
+ "params_size": len(record.params),
+ "params_size_expected": record.expected_params_size,
+ "truncated": record.truncated,
+ }
+ if record.fn is not None:
+ message["fn"] = record.fn
+ if record.cmd is not None:
+ message["cmd"] = record.cmd
+ if record.hclass is not None:
+ message["hclass"] = record.hclass
+ if record.kind == "send" and record.wire_id is not None:
+ message["wire_id"] = record.wire_id
+ messages.append(message)
+ return {
+ "format": STIMULUS_FORMAT,
+ "source": {
+ "path": capture.path,
+ "sha256": capture.sha256,
+ "records": len(capture.records),
+ },
+ "markers": markers,
+ "messages": messages,
+ "limits": [
+ "Handles, physical addresses and caller addresses are capture-specific and deliberately omitted; a replay must re-map or allocate them.",
+ "Truncated messages (params_size < params_size_expected) are not replayable as-is; extend the capture dump first.",
+ "Consumption by the simulator B4/B5 CSB/message-queue seam is not wired yet (docs/45).",
+ "No ordering claim beyond the captured log order; server replies were not captured by the send-side probes.",
+ ],
+ }
+
+
+# ---------------------------------------------------------------------------
+# Demo (synthetic; measured value contract)
+# ---------------------------------------------------------------------------
+
+
+def _hex_lines(tag: str, payload: bytes, chunk: int = 192) -> list[str]:
+ lines: list[str] = []
+ for start in range(0, len(payload), chunk):
+ piece = payload[start : start + chunk]
+ lines.append(f"GE-{tag} {start:03x}:" + "".join(f" {b:02x}" for b in piece))
+ return lines
+
+
+def _synthetic_log() -> str:
+ allocs = [
+ ("0x90f1", 0x30),
+ ("0x003e", 0x100),
+ ("0x50a0", 0x40),
+ ("0x0040", 0x1000),
+ ("0xc46f", 0x180),
+ ("0xc597", 0x0),
+ ]
+ controls = [
+ ("0x2080012b", 560),
+ ("0x20801201", 64),
+ ("0x20800a2a", 32),
+ ("0x20809001", 16),
+ ]
+ lines = [
+ "[ 100.000000] NVRM: GE-GOLDEN-TRIGGER gspclient=1 mig=0",
+ "[ 100.000010] NVRM: GE-GOLDEN-BEGIN",
+ ]
+ step = 1
+ for hclass, size in allocs:
+ hclass_value = int(hclass, 16)
+ params = bytes((i * 7 + hclass_value) & 0xFF for i in range(size))
+ lines.append(
+ f"[ 100.{step:06d}] NVRM: GE-ALLOC cls={hclass} hClient=0xbaba0001 "
+ f"hParent=0x0 hObject=0xbaba10{step:02x} size={size:#x} "
+ f"paramp=0xffffffffc0a00000 ret=0xffffffffc0b00000"
+ )
+ lines.extend(f"[ 100.{step:06d}] NVRM: {line}" for line in _hex_lines("ALLOC", params))
+ step += 1
+ for cmd, size in controls:
+ cmd_value = int(cmd, 16)
+ params = bytes((i * 3 + cmd_value) & 0xFF for i in range(size))
+ lines.append(
+ f"[ 100.{step:06d}] NVRM: GE-CTRL cmd={cmd} hClient=0xc2000006 "
+ f"hObject=0xabcd0080 size={size:#x} paramp=0xffffffffc0a10000 "
+ f"ret=0xffffffffc0b00000"
+ )
+ lines.extend(f"[ 100.{step:06d}] NVRM: {line}" for line in _hex_lines("CTRL", params))
+ step += 1
+ payload = (
+ (0xC2000006).to_bytes(4, "little")
+ + (0xABCD0080).to_bytes(4, "little")
+ + (0x2080012B).to_bytes(4, "little")
+ + bytes(range(60))
+ )
+ lines.append(
+ f"[ 100.{step:06d}] NVRM: GE-SEND fn=0x0000004c len={len(payload) + RPC_HEADER_BYTES:#x} "
+ f"ret=0xffffffffc0b00000"
+ )
+ lines.extend(f"[ 100.{step:06d}] NVRM: {line}" for line in _hex_lines("SEND", payload))
+ step += 1
+ lines.append(
+ f"[ 100.{step:06d}] NVRM: GE-CTRL-DONE cmd=0x2080012b rpc_status=0x0 params_status=0x0"
+ )
+ lines.append("[ 100.000999] NVRM: GE-GOLDEN-END status=0x0")
+ return "\n".join(lines) + "\n"
+
+
+def _mutate_log(text: str) -> str:
+ """Replay-side variant: one parameter byte differs, one alloc is dropped."""
+ kept: list[str] = []
+ dropping = False
+ dropped = False
+ for line in text.splitlines():
+ if not dropped and "GE-ALLOC cls=0x003e" in line:
+ dropped = True
+ dropping = True
+ continue
+ if dropping:
+ if re.search(r"GE-ALLOC\s+[0-9a-fA-F]{1,4}:", line):
+ continue
+ dropping = False
+ kept.append(line)
+ text_b = "\n".join(kept) + "\n"
+
+ def flip(match: re.Match) -> str:
+ value = int(match.group(2), 16) ^ 0x01
+ return f"{match.group(1)}{value:02x}"
+
+ text_b, _ = re.subn(r"(GE-CTRL 000: )([0-9a-fA-F]{2})", flip, text_b, count=1)
+ return text_b
+
+
+def run_demo(out_dir: str | None = None) -> dict:
+ started = time.perf_counter()
+ capture_a = decode_text(_synthetic_log(), source="demo-capture-a")
+ capture_b = decode_text(_mutate_log(_synthetic_log()), source="demo-replay-b")
+ report = compare(capture_a, capture_b)
+ stimulus = stimulus_from_capture(capture_a)
+ wall = time.perf_counter() - started
+ threshold_seconds = MANUAL_DECODE_BASELINE_SECONDS * WORKFLOW_THRESHOLD_FRACTION
+ ok = (
+ wall <= threshold_seconds
+ and report["summary"]["agree"] + report["summary"]["shape_agree"] >= 1
+ and report["summary"]["a_only"] + report["summary"]["b_only"] >= 1
+ and not capture_a.truncated_records
+ )
+ payload = {
+ "format": DEMO_FORMAT,
+ "ok": ok,
+ "wall_seconds": round(wall, 4),
+ "baseline_seconds": MANUAL_DECODE_BASELINE_SECONDS,
+ "baseline_label": MANUAL_BASELINE_LABEL,
+ "threshold_fraction": WORKFLOW_THRESHOLD_FRACTION,
+ "threshold_seconds": threshold_seconds,
+ "records_a": len(capture_a.records),
+ "records_b": len(capture_b.records),
+ "messages": len(stimulus["messages"]),
+ "summary": report["summary"],
+ }
+ if out_dir:
+ os.makedirs(out_dir, exist_ok=True)
+ capture_path = os.path.join(out_dir, "demo-capture-a.json")
+ replay_path = os.path.join(out_dir, "demo-replay-b.json")
+ with open(capture_path, "w", encoding="utf-8") as fh:
+ fh.write(canonical_json(capture_to_dict(capture_a)))
+ with open(replay_path, "w", encoding="utf-8") as fh:
+ fh.write(canonical_json(capture_to_dict(capture_b)))
+ compare_path = os.path.join(out_dir, "demo-compare.json")
+ with open(compare_path, "w", encoding="utf-8") as fh:
+ fh.write(canonical_json(report))
+ stimulus_path = os.path.join(out_dir, "demo-stimulus.json")
+ with open(stimulus_path, "w", encoding="utf-8") as fh:
+ fh.write(canonical_json(stimulus))
+ payload["artifacts"] = {
+ "capture_a": os.path.abspath(capture_path),
+ "replay_b": os.path.abspath(replay_path),
+ "compare": os.path.abspath(compare_path),
+ "stimulus": os.path.abspath(stimulus_path),
+ }
+ return payload
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+CAPTURE_HINTS = """\
+================================================================================
+GSP RPC CAPTURE TAG GRAMMAR — SUPERVISED, HUMAN-RUN PHYSICAL ACTION
+================================================================================
+This command prints the tag contract the decoder understands; it executes
+NOTHING and never runs the capture.
+
+The capture itself is supervised-only (RUNBOOK P1/P7, SIMULATOR-WORK-
+BOUNDARY.txt): driver swap on the physical GPU, one attempt per boot, stop on
+Zid/WPR2/pstore; the authored procedure lives in the capture runbook
+(authored, not executed). This tool only ingests the resulting log.
+
+Tags (anywhere in the line; dmesg/NVRM prefixes are fine):
+ GE-GOLDEN-TRIGGER gspclient=<0|1> mig=<0|1>
+ GE-GOLDEN-BEGIN
+ GE-GOLDEN-END status=0x...
+ GE-SEND fn=0x... len=0x... ret=0x...
+ GE-CTRL cmd=0x... hClient=0x... hObject=0x... size=0x... paramp=0x... ret=0x...
+ GE-ALLOC cls=0x... hClient=0x... hParent=0x... hObject=0x... size=0x... paramp=0x... ret=0x...
+ GE-CTRL-DONE cmd=0x... rpc_status=0x... params_status=0x...
+ Hex lines attach to the most recent record of their own tag:
+ GE-<TAG> <offset-hex>: <hex bytes...>
+
+NOTE — the runbook's hex helper currently caps a dump at 192 bytes; payloads
+larger than that (e.g. a 560-byte PROMOTE_CTX) decode as `truncated` and are
+not replayable until the helper is extended to chunk. The decoder flags this
+instead of padding.
+
+Offline use (safe):
+ python3 harness/rm_capture.py decode --log capture.log --out dumps/rm-capture
+ python3 harness/rm_capture.py compare --a real.json --b replay.json --out dumps/rm-compare
+ python3 harness/rm_capture.py stimulus --capture real.json --out dumps/rm-stimulus.json
+ python3 harness/rm_capture.py demo --out /tmp/rm-capture-demo
+================================================================================
+"""
+
+
+def print_capture_hints(out=None) -> None:
+ """Print the supervised capture tag grammar; execute nothing."""
+ print(CAPTURE_HINTS, file=out or sys.stdout)
+
+
+def build_parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(
+ prog="rm_capture.py",
+ description=(
+ "Decode, compare and replay-prep a supervised GSP RPC capture. "
+ "Offline only; never executes physical commands."
+ ),
+ )
+ parser.add_argument(
+ "--print-capture-hints",
+ action="store_true",
+ help="print the capture tag grammar (executes nothing)",
+ )
+ sub = parser.add_subparsers(dest="command")
+ decode_cmd = sub.add_parser("decode", help="decode a raw capture log")
+ decode_cmd.add_argument("--log", required=True, help="capture log (text)")
+ decode_cmd.add_argument("--out", help="output directory (optional)")
+ compare_cmd = sub.add_parser("compare", help="compare two captures")
+ compare_cmd.add_argument("--a", required=True, help="capture JSON or raw log")
+ compare_cmd.add_argument("--b", required=True, help="capture JSON or raw log")
+ compare_cmd.add_argument("--out", required=True, help="output directory")
+ stimulus_cmd = sub.add_parser("stimulus", help="emit a replay stimulus document")
+ stimulus_cmd.add_argument("--capture", required=True, help="capture JSON or raw log")
+ stimulus_cmd.add_argument("--out", help="output JSON path (optional)")
+ demo_cmd = sub.add_parser("demo", help="synthetic measured end-to-end demo")
+ demo_cmd.add_argument("--out", help="output directory (optional)")
+ return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+ parser = build_parser()
+ args = parser.parse_args(argv)
+ if args.print_capture_hints:
+ print_capture_hints()
+ return 0
+ try:
+ if args.command == "decode":
+ payload = run_decode(args.log, args.out)
+ elif args.command == "compare":
+ payload = run_compare(args.a, args.b, args.out)
+ elif args.command == "stimulus":
+ payload = run_stimulus(args.capture, args.out)
+ elif args.command == "demo":
+ payload = run_demo(args.out)
+ else:
+ parser.print_help()
+ return 2
+ except RmCaptureError as exc:
+ print(f"error: {exc}", file=sys.stderr)
+ return 2
+ summary = payload.get("summary", {})
+ print(
+ json.dumps(
+ {
+ "format": payload.get("format"),
+ "ok": payload.get("ok"),
+ "wall_seconds": payload.get("wall_seconds"),
+ "summary": summary,
+ "artifacts": payload.get("artifacts"),
+ },
+ indent=2,
+ sort_keys=True,
+ )
+ )
+ if args.command == "demo" and not payload.get("ok"):
+ return 3
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-12 07:45:26 +0000