path: root/debug/risc-v-gsp-rm-vm/harness/rm_capture.py

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