#!/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 = "") -> 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- : 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())