diff options
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.py | 1251 |
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()) |