From 6c3fd5df61ff2f5cb8833ead6c05eaff9fdc92bf Mon Sep 17 00:00:00 2001
From: Arthur Hanson Rasmusson <arthur@vgpu.io>
Date: Tue, 6 Oct 2026 19:28:39 -0400
Subject: risc-v-gsp-rm-debugger: land docs 45-53 + harness/sim tool set with
 tests

rm_capture (chunked decode), launch_audit, state_seed/state_decode,
valprobe, rehost, contracts+run_contracts, host_traffic, gmmu_walk,
msgq_attach, libos_port, log_decode, server_snapshot; doc index updates
in the tool overview, operations pages and the repo README.
---
 debug/risc-v-gsp-rm-vm/harness/server_snapshot.py | 698 ++++++++++++++++++++++
 1 file changed, 698 insertions(+)
 create mode 100644 debug/risc-v-gsp-rm-vm/harness/server_snapshot.py

(limited to 'debug/risc-v-gsp-rm-vm/harness/server_snapshot.py')

diff --git a/debug/risc-v-gsp-rm-vm/harness/server_snapshot.py b/debug/risc-v-gsp-rm-vm/harness/server_snapshot.py
new file mode 100644
index 0000000..1099fb0
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/server_snapshot.py
@@ -0,0 +1,698 @@
+#!/usr/bin/env python3
+"""Collect a canonical server-state snapshot from cycle evidence directories.
+
+The differential loop (live run vs emulated run, run N vs run N+1) needs one
+machine-readable view of the server state a cycle produced.  Cycles already
+write that state as text: labelled field variables (``LABEL[offset]=0xVALUE``),
+labelled hexdump blocks (an ``INSTANCE[0000]: w w w w``-style row, or a header
+line naming the label followed by ``OFFSET: w w w w`` rows -- e.g.
+``LOGINIT-raw`` / ``LOGRM-raw`` raw task-log dumps), and ``vgsp-nocat`` record
+lines carrying ``recType=`` / ``errorCode=``.  This harness discovers those
+files under evidence/session directories, normalizes every dump to
+little-endian u32 words, and emits one JSON snapshot that two runs can be
+compared with ``diff``.
+
+Schemas
+-------
+
+``greeneye-server-snapshot/1``::
+
+    {"schema": "greeneye-server-snapshot/1",
+     "source": "<string>",
+     "created": "<ISO-8601>",
+     "chunks": [{"label": "INSTANCE", "base": "0x0",
+                 "words": ["0x00000000", ...]}],
+     "records": [{"kind": "record", "raw": "...",
+                  "rec_type": 5, "error_code": 0x57f1344,
+                  "source_name": "ASSERT"}]}
+
+``greeneye-server-snapshot-diff/1``::
+
+    {"schema": "greeneye-server-snapshot-diff/1",
+     "a": {"source": ..., "created": ...}, "b": {...},
+     "chunks": [{"label": "INSTANCE", "status": "changed|added|removed|unchanged",
+                 "base": {"a": "0x0", "b": "0x0"},
+                 "words": {"a": 1024, "b": 1024},
+                 "added": [{"index": 8, "value": "0x00000000"}],
+                 "removed": [{"index": 8, "value": "0x00000001"}],
+                 "changed": [{"index": 8, "a": "0x...", "b": "0x..."}]}],
+     "records": {"added": [...], "removed": [...]},
+     "summary": {"identical": false, "chunks_added": 0, ...}}
+
+Normalization rules (the consumer depends on them):
+
+* words are 8-hex-digit lowercase strings of little-endian u32 values;
+* a ``LABEL[offset]=0xVALUE`` value wider than 32 bits becomes two words
+  (low word first); its offset orders the entries and selects ``base``, and
+  the values are appended in offset order (the offset is a table index there,
+  not a byte address);
+* hexdump rows are placed by byte offset (``(offset - base) // 4``) when
+  4-byte aligned, so a row missing in one run still aligns the rest;
+* ``base`` is the smallest offset seen for the chunk (``0x0`` when none);
+* chunks are sorted by label; a repeated label is merged only when its base
+  and word list are byte-identical, otherwise it is kept as ``LABEL#2``,
+  ``LABEL#3`` ... and the collision is reported in the run stats;
+* non-empty records are deduplicated by raw line and sorted by raw line;
+* ``created`` is the only variable field: pin it with ``--created`` or
+  ``SOURCE_DATE_EPOCH`` for deterministic output.
+
+Honesty: this is an evidence normalizer, not a state oracle.  It preserves
+what the cycle wrote -- offsets are honoured, identical duplicates are merged,
+differing duplicates are renamed, and dump-looking lines that fail to parse
+are counted (``unparsed``) instead of being silently dropped.  It never
+executes anything, touches no GPU and reads only files given on the command
+line.
+"""
+from __future__ import annotations
+
+import argparse
+import fnmatch
+import json
+import os
+import re
+import sys
+import tempfile
+import time
+from typing import Any, Iterable
+
+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)
+
+SNAPSHOT_FORMAT = "greeneye-server-snapshot/1"
+DIFF_FORMAT = "greeneye-server-snapshot-diff/1"
+DEMO_FORMAT = "greeneye-server-snapshot-demo/1"
+
+#: File-name globs discovered under each ``--dir`` unless extended with
+#: ``--include``.  fnmatch's ``*`` matches leading dots, so the rolling host
+#: dmesg captures (``.dmesg-*.rolling``) are covered by ``*.rolling``.
+DEFAULT_INCLUDES = ("*.txt", "*.log", "*.rolling")
+
+#: ``INSTANCE[0000]: w w w w`` -- one labelled dump row inside an arbitrary
+#: prefix (dmesg timestamp, subsystem, case id).  Offsets are hexadecimal.
+_INLINE_ROW_RE = re.compile(
+    r"(?P<label>[A-Za-z_][A-Za-z0-9_.]*)\[(?P<offset>(?:0x)?[0-9A-Fa-f]+)\]:"
+    r"[ \t]*(?P<tail>[0-9A-Fa-f]{1,8}(?:[ \t]+[0-9A-Fa-f]{1,8})*)[ \t]*$"
+)
+
+#: ``LABEL[offset]=0xVALUE`` (optionally ``LABEL[0].suffix=``), several per
+#: line allowed (``root[0]=... pd2[0]=... pd0[0].small=...``).
+_LABELED_VALUE_RE = re.compile(
+    r"(?P<label>[A-Za-z_][A-Za-z0-9_.]*)\[(?P<offset>(?:0x)?[0-9A-Fa-f]+)\]"
+    r"(?P<suffix>\.[A-Za-z_][A-Za-z0-9_]*)?[ \t]*=[ \t]*"
+    r"(?P<value>0[xX][0-9A-Fa-f]+)\b"
+)
+
+#: A bare hexdump row (``00000010: w w w w``) inside a labelled block.
+_BLOCK_ROW_RE = re.compile(
+    r"^(?P<offset>[0-9A-Fa-f]{4,16}):[ \t]+"
+    r"(?P<tail>[0-9A-Fa-f]{1,8}(?:[ \t]+[0-9A-Fa-f]{1,8})*)[ \t]*$"
+)
+
+#: A block header naming the label for the rows that follow: ``LOGINIT-raw:``,
+#: ``LOGRM-raw:``, ``RAMFC:`` ...  Used only as one-line lookahead.
+_BLOCK_HEADER_RE = re.compile(r"^(?P<label>[A-Za-z_][A-Za-z0-9_.-]*)[ \t]*:")
+
+#: A GSP-RM error record line (``rec seq=1 recType=5 errorCode=0x...``).
+_RECORD_RE = re.compile(
+    r"\brecType=(?P<rec_type>\d+)\b.*?\berrorCode=(?P<error_code>0[xX][0-9A-Fa-f]+)"
+)
+
+_SOURCE_RE = re.compile(r'\bsource="(?P<source_name>[^"]*)"')
+
+#: Loose shape check for a line that looks like a dump but failed to parse;
+#: these are counted so silent data loss is visible in the run stats.
+_UNPARSED_HINT_RE = re.compile(
+    r"\[[0-9A-Fa-fx]+\][ \t]*[=:]|^[0-9A-Fa-f]{4,}:|\brecType="
+)
+
+
+class SnapshotError(ValueError):
+    """An input file or snapshot the tool cannot honour."""
+
+
+def _new_stats() -> dict[str, Any]:
+    return {
+        "files": 0,
+        "lines": 0,
+        "chunks": 0,
+        "words": 0,
+        "records": 0,
+        "unparsed": 0,
+        "merged": 0,
+        "collisions": 0,
+        "conflicts": 0,
+        "misaligned": 0,
+        "collision_list": [],
+    }
+
+
+class _Builder:
+    """Words captured for one label occurrence before normalization.
+
+    Entries are ``(offset, [u32 word, ...])`` in encounter order.  Hexdump
+    builders are positional: entries are placed at ``(offset - base) // 4``
+    when the byte offset is 4-byte aligned; otherwise they are appended after
+    the highest placed word.  Value-line builders (``LABEL[offset]=0xVALUE``)
+    are sequential: the offset only orders the entries and selects ``base``
+    (its unit is a table entry index there, not a byte), so each entry's words
+    are appended in offset order.  A later value that would overwrite a
+    different value at the same slot is dropped and counted as a
+    ``conflict``.
+    """
+
+    __slots__ = ("label", "positional", "entries", "conflicts", "misaligned")
+
+    def __init__(self, label: str, positional: bool = True):
+        self.label = label
+        self.positional = positional
+        self.entries: list[tuple[int, list[str]]] = []
+        self.conflicts = 0
+        self.misaligned = 0
+
+    def add(self, offset: int, words: list[str]) -> None:
+        if not words:
+            return
+        for existing_offset, existing_words in self.entries:
+            if existing_offset == offset and existing_words == words:
+                return
+        self.entries.append((offset, words))
+
+    def build(self) -> tuple[int, list[str]]:
+        base = min(offset for offset, _ in self.entries)
+        placed: dict[int, str] = {}
+        append_at = 0
+        conflicts = 0
+        misaligned = 0
+        for offset, words in sorted(self.entries, key=lambda entry: entry[0]):
+            if not self.positional:
+                index = append_at
+            elif (offset - base) % 4 == 0:
+                index = (offset - base) // 4
+            else:
+                index = append_at
+                misaligned += 1
+            for step, word in enumerate(words):
+                slot = index + step
+                if slot in placed and placed[slot] != word:
+                    conflicts += 1
+                elif slot not in placed:
+                    placed[slot] = word
+            append_at = max(append_at, index + len(words))
+        self.conflicts = conflicts
+        self.misaligned = misaligned
+        return base, [placed[index] for index in sorted(placed)]
+
+
+def _word(value: int) -> str:
+    return f"0x{value & 0xFFFFFFFF:08x}"
+
+
+def _value_words(value: int) -> list[str]:
+    """Split a value into little-endian u32 words (low word first)."""
+    words = []
+    while True:
+        words.append(_word(value))
+        value >>= 32
+        if value == 0:
+            return words
+
+
+def _split_words(tail: str) -> list[str]:
+    return [f"0x{int(token, 16):08x}" for token in tail.split()]
+
+
+def _row_offset(text: str) -> int:
+    return int(text, 16)
+
+
+def _labeled_offset(text: str) -> int:
+    if text.lower().startswith("0x"):
+        return int(text, 16)
+    try:
+        return int(text, 10)
+    except ValueError:
+        return int(text, 16)
+
+
+def _match_record(line: str) -> dict[str, Any] | None:
+    match = _RECORD_RE.search(line)
+    if not match:
+        return None
+    record: dict[str, Any] = {
+        "kind": "record",
+        "raw": line,
+        "rec_type": int(match.group("rec_type")),
+        "error_code": int(match.group("error_code"), 16),
+    }
+    source = _SOURCE_RE.search(line)
+    if source:
+        record["source_name"] = source.group("source_name")
+    return record
+
+
+def _parse_text(text: str, stats: dict[str, Any]) -> tuple[list[_Builder], list[dict]]:
+    occurrences: list[_Builder] = []
+    records: list[dict] = []
+    by_value_label: dict[str, _Builder] = {}
+    current_block: _Builder | None = None
+    inline_block: _Builder | None = None
+    pending_header: str | None = None
+
+    for line in text.splitlines():
+        stripped = line.strip()
+        if not stripped:
+            current_block = pending_header = inline_block = None
+            continue
+
+        record = _match_record(stripped)
+        if record is not None:
+            records.append(record)
+            current_block = pending_header = inline_block = None
+            continue
+
+        match = _INLINE_ROW_RE.search(stripped)
+        if match:
+            label = match.group("label")
+            words = _split_words(match.group("tail"))
+            if inline_block is not None and inline_block.label == label:
+                inline_block.add(_row_offset(match.group("offset")), words)
+            else:
+                inline_block = _Builder(label)
+                inline_block.add(_row_offset(match.group("offset")), words)
+                occurrences.append(inline_block)
+            current_block = pending_header = None
+            continue
+
+        match = _BLOCK_ROW_RE.match(stripped)
+        if match:
+            if current_block is None:
+                if pending_header is None:
+                    stats["unparsed"] += 1
+                    continue
+                current_block = _Builder(pending_header)
+                occurrences.append(current_block)
+            current_block.add(int(match.group("offset"), 16),
+                             _split_words(match.group("tail")))
+            pending_header = inline_block = None
+            continue
+
+        value_matches = list(_LABELED_VALUE_RE.finditer(stripped))
+        if value_matches:
+            for value_match in value_matches:
+                label = value_match.group("label")
+                if value_match.group("suffix"):
+                    label += value_match.group("suffix")
+                builder = by_value_label.get(label)
+                if builder is None:
+                    builder = _Builder(label, positional=False)
+                    by_value_label[label] = builder
+                    occurrences.append(builder)
+                builder.add(_labeled_offset(value_match.group("offset")),
+                            _value_words(int(value_match.group("value"), 16)))
+            current_block = pending_header = inline_block = None
+            continue
+
+        if _UNPARSED_HINT_RE.search(stripped):
+            stats["unparsed"] += 1
+
+        current_block = inline_block = None
+        header = _BLOCK_HEADER_RE.match(stripped)
+        pending_header = header.group("label") if header else None
+
+    return occurrences, records
+
+
+def _chunk_sort_key(label: str) -> tuple[str, int, str]:
+    stem, separator, tail = label.rpartition("#")
+    if separator and tail.isdigit():
+        return stem, int(tail), label
+    return label, 0, label
+
+
+def _finalize(occurrences: Iterable[_Builder],
+              stats: dict[str, Any]) -> list[dict[str, Any]]:
+    chunks: dict[str, dict[str, Any]] = {}
+    for builder in occurrences:
+        base, words = builder.build()
+        stats["conflicts"] += builder.conflicts
+        stats["misaligned"] += builder.misaligned
+        if not words:
+            continue
+        key = builder.label
+        suffix = 1
+        merged = False
+        while key in chunks:
+            existing = chunks[key]
+            if existing["base"] == f"{base:#x}" and existing["words"] == words:
+                stats["merged"] += 1
+                merged = True
+                break
+            suffix += 1
+            key = f"{builder.label}#{suffix}"
+        if merged:
+            continue
+        chunks[key] = {"label": key, "base": f"{base:#x}", "words": words}
+        if suffix > 1:
+            stats["collisions"] += 1
+            stats["collision_list"].append(
+                {"label": builder.label, "renamed": key,
+                 "base": f"{base:#x}", "words": len(words)})
+    return [chunks[key] for key in sorted(chunks, key=_chunk_sort_key)]
+
+
+def discover(dirs: Iterable[str] | None, files: Iterable[str] | None,
+             includes: Iterable[str] | None = None) -> list[str]:
+    """Return the sorted absolute paths matched by the dirs/files/includes."""
+    patterns = tuple(DEFAULT_INCLUDES) + tuple(includes or ())
+    found: set[str] = set()
+    for directory in dirs or ():
+        if not os.path.isdir(directory):
+            raise SnapshotError(f"not a directory: {directory}")
+        for root, _subdirs, names in os.walk(directory):
+            for name in names:
+                if any(fnmatch.fnmatch(name, pattern) for pattern in patterns):
+                    found.add(os.path.abspath(os.path.join(root, name)))
+    for path in files or ():
+        if not os.path.isfile(path):
+            raise SnapshotError(f"not a file: {path}")
+        found.add(os.path.abspath(path))
+    return sorted(found)
+
+
+def _default_source(dirs: Iterable[str] | None,
+                    files: Iterable[str] | None) -> str:
+    roots = sorted(os.path.abspath(path)
+                   for path in list(dirs or ()) + list(files or ()))
+    return ",".join(roots) if roots else "empty"
+
+
+def _created_iso() -> str:
+    """Build timestamp; stable when ``SOURCE_DATE_EPOCH`` is set."""
+    raw = os.environ.get("SOURCE_DATE_EPOCH")
+    try:
+        epoch = int(raw) if raw else int(time.time())
+    except ValueError:
+        epoch = 0
+    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(epoch))
+
+
+def collect(dirs: Iterable[str] | None = None,
+            files: Iterable[str] | None = None,
+            includes: Iterable[str] | None = None,
+            source: str | None = None,
+            created: str | None = None) -> tuple[dict[str, Any], dict[str, Any]]:
+    """Collect every dump in ``dirs``/``files`` into one snapshot.
+
+    Returns ``(snapshot, stats)``; ``stats`` carries the parse accounting
+    (files, lines, chunks, words, records, unparsed candidates, merged
+    duplicates, renamed collisions, conflicts, misaligned offsets) and is
+    printed by the CLI but is deliberately not part of the consumed schema.
+    """
+    paths = discover(dirs, files, includes)
+    if not paths:
+        raise SnapshotError(
+            "no input files matched the requested --dir/--file/--include set")
+    stats = _new_stats()
+    occurrences: list[_Builder] = []
+    records: dict[str, dict] = {}
+    for path in paths:
+        try:
+            with open(path, "r", encoding="utf-8", errors="replace") as handle:
+                text = handle.read()
+        except OSError as exc:
+            raise SnapshotError(f"cannot read {path}: {exc}") from exc
+        stats["files"] += 1
+        stats["lines"] += len(text.splitlines())
+        file_occurrences, file_records = _parse_text(text, stats)
+        occurrences.extend(file_occurrences)
+        for record in file_records:
+            records.setdefault(record["raw"], record)
+    chunks = _finalize(occurrences, stats)
+    stats["chunks"] = len(chunks)
+    stats["words"] = sum(len(chunk["words"]) for chunk in chunks)
+    stats["records"] = len(records)
+    snapshot = {
+        "schema": SNAPSHOT_FORMAT,
+        "source": source if source is not None else _default_source(dirs, files),
+        "created": created if created is not None else _created_iso(),
+        "chunks": chunks,
+        "records": sorted(records.values(), key=lambda record: record["raw"]),
+    }
+    return snapshot, stats
+
+
+def diff_snapshots(a: dict[str, Any], b: dict[str, Any]) -> dict[str, Any]:
+    """Compare two ``greeneye-server-snapshot/1`` snapshots word by word.
+
+    Emits a ``greeneye-server-snapshot-diff/1`` report: per shared chunk the
+    added/removed/changed word lists (with word indices), whole-chunk
+    added/removed rows, per-chunk and total counts, and record deltas.  Output
+    is sorted by label and by word index, so two diffs of the same inputs are
+    byte-identical.
+    """
+    for name, snapshot in (("a", a), ("b", b)):
+        if not isinstance(snapshot, dict) or snapshot.get("schema") != SNAPSHOT_FORMAT:
+            raise SnapshotError(f"snapshot {name} is not {SNAPSHOT_FORMAT}")
+    a_chunks = {chunk["label"]: chunk for chunk in a.get("chunks", [])}
+    b_chunks = {chunk["label"]: chunk for chunk in b.get("chunks", [])}
+    rows: list[dict[str, Any]] = []
+    counts = {
+        "chunks_added": 0, "chunks_removed": 0, "chunks_changed": 0,
+        "chunks_unchanged": 0, "words_added": 0, "words_removed": 0,
+        "words_changed": 0,
+    }
+    for label in sorted(set(a_chunks) | set(b_chunks), key=_chunk_sort_key):
+        left = a_chunks.get(label)
+        right = b_chunks.get(label)
+        if left is None:
+            added = [{"index": index, "value": word}
+                     for index, word in enumerate(right["words"])]
+            row = {"label": label, "status": "added",
+                   "base": {"a": None, "b": right["base"]},
+                   "words": {"a": 0, "b": len(right["words"])},
+                   "added": added, "removed": [], "changed": []}
+            counts["chunks_added"] += 1
+            counts["words_added"] += len(added)
+        elif right is None:
+            removed = [{"index": index, "value": word}
+                       for index, word in enumerate(left["words"])]
+            row = {"label": label, "status": "removed",
+                   "base": {"a": left["base"], "b": None},
+                   "words": {"a": len(left["words"]), "b": 0},
+                   "added": [], "removed": removed, "changed": []}
+            counts["chunks_removed"] += 1
+            counts["words_removed"] += len(removed)
+        else:
+            a_words, b_words = left["words"], right["words"]
+            added, removed, changed = [], [], []
+            for index in range(max(len(a_words), len(b_words))):
+                if index >= len(a_words):
+                    added.append({"index": index, "value": b_words[index]})
+                elif index >= len(b_words):
+                    removed.append({"index": index, "value": a_words[index]})
+                elif a_words[index] != b_words[index]:
+                    changed.append({"index": index, "a": a_words[index],
+                                    "b": b_words[index]})
+            unchanged = (not added and not removed and not changed
+                         and left["base"] == right["base"])
+            row = {"label": label,
+                   "status": "unchanged" if unchanged else "changed",
+                   "base": {"a": left["base"], "b": right["base"]},
+                   "words": {"a": len(a_words), "b": len(b_words)},
+                   "added": added, "removed": removed, "changed": changed}
+            if unchanged:
+                counts["chunks_unchanged"] += 1
+            else:
+                counts["chunks_changed"] += 1
+            counts["words_added"] += len(added)
+            counts["words_removed"] += len(removed)
+            counts["words_changed"] += len(changed)
+        rows.append(row)
+
+    a_records = {record.get("raw", "") for record in a.get("records", [])}
+    b_records = {record.get("raw", "") for record in b.get("records", [])}
+    records_added = sorted(b_records - a_records)
+    records_removed = sorted(a_records - b_records)
+    counts["records_added"] = len(records_added)
+    counts["records_removed"] = len(records_removed)
+    identical = (counts["chunks_added"] == 0 and counts["chunks_removed"] == 0
+                 and counts["chunks_changed"] == 0
+                 and not records_added and not records_removed)
+    return {
+        "schema": DIFF_FORMAT,
+        "a": {"source": a.get("source"), "created": a.get("created")},
+        "b": {"source": b.get("source"), "created": b.get("created")},
+        "chunks": rows,
+        "records": {"added": records_added, "removed": records_removed},
+        "summary": {"identical": identical, **counts},
+    }
+
+
+# ---------------------------------------------------------------------------
+# Demo: two tiny snapshots in a tempdir, collected and diffed
+# ---------------------------------------------------------------------------
+
+DEMO_A = """\
+[demo] MOCK[0000]: 00000001 00000002 00000003 00000004
+alpha[0]=0x00000011
+alpha[4]=0x00000022
+"""
+
+DEMO_B = """\
+[demo] MOCK[0000]: 00000001 000000ff 00000003 00000004 00000005
+alpha[0]=0x00000011
+alpha[4]=0x00000033
+"""
+
+
+def _write_text(path: str, text: str) -> str:
+    with open(path, "w", encoding="utf-8") as handle:
+        handle.write(text)
+    return os.path.abspath(path)
+
+
+def _write_json(path: str, payload: dict[str, Any]) -> str:
+    with open(path, "w", encoding="utf-8") as handle:
+        json.dump(payload, handle, indent=2, sort_keys=True)
+        handle.write("\n")
+    return os.path.abspath(path)
+
+
+def run_demo(workdir: str) -> dict[str, Any]:
+    """Craft two tiny evidence files in ``workdir``, collect both and diff."""
+    a_path = _write_text(os.path.join(workdir, "run-a.txt"), DEMO_A)
+    b_path = _write_text(os.path.join(workdir, "run-b.txt"), DEMO_B)
+    snapshot_a, _stats_a = collect(files=[a_path], source="demo-a",
+                                   created="2026-01-01T00:00:00Z")
+    snapshot_b, _stats_b = collect(files=[b_path], source="demo-b",
+                                   created="2026-01-01T00:00:00Z")
+    return {
+        "schema": DEMO_FORMAT,
+        "workdir": os.path.abspath(workdir),
+        "snapshot_a": snapshot_a,
+        "snapshot_b": snapshot_b,
+        "diff": diff_snapshots(snapshot_a, snapshot_b),
+    }
+
+
+def _load_json(path: str) -> dict[str, Any]:
+    try:
+        with open(path, "r", encoding="utf-8") as handle:
+            return json.load(handle)
+    except (OSError, json.JSONDecodeError) as exc:
+        raise SnapshotError(f"cannot read {path}: {exc}") from exc
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="server_snapshot.py",
+        description="Collect and diff canonical server-state snapshots "
+                    "(offline evidence normalizer).",
+    )
+    sub = parser.add_subparsers(dest="command")
+
+    collect_cmd = sub.add_parser(
+        "collect", help="normalize evidence files into one snapshot")
+    collect_cmd.add_argument("--dir", action="append", default=[],
+                             help="evidence/session directory (repeatable)")
+    collect_cmd.add_argument("--file", action="append", default=[],
+                             help="explicit evidence file (repeatable)")
+    collect_cmd.add_argument(
+        "--include", action="append", default=[],
+        help="additional file-name glob under --dir (repeatable); "
+             f"defaults: {', '.join(DEFAULT_INCLUDES)}")
+    collect_cmd.add_argument("--source", help="override the snapshot source")
+    collect_cmd.add_argument(
+        "--created", help="override the ISO-8601 timestamp (SOURCE_DATE_EPOCH "
+                          "is honoured otherwise)")
+    collect_cmd.add_argument("--json", help="write the snapshot here instead "
+                                            "of stdout")
+
+    diff_cmd = sub.add_parser("diff", help="compare two snapshot JSON files")
+    diff_cmd.add_argument("--a", required=True, help="first snapshot JSON")
+    diff_cmd.add_argument("--b", required=True, help="second snapshot JSON")
+    diff_cmd.add_argument("--json", help="write the diff here instead of stdout")
+
+    demo_cmd = sub.add_parser(
+        "demo", help="craft two tiny snapshots in a tempdir and diff them")
+    demo_cmd.add_argument("--out", help="keep the crafted files in this directory")
+    return parser
+
+
+def _collect_command(args) -> int:
+    if not args.dir and not args.file:
+        raise SnapshotError("collect needs at least one --dir or --file")
+    snapshot, stats = collect(dirs=args.dir, files=args.file,
+                              includes=args.include, source=args.source,
+                              created=args.created)
+    if args.json:
+        path = _write_json(args.json, snapshot)
+        summary = {
+            "schema": snapshot["schema"],
+            "source": snapshot["source"],
+            "created": snapshot["created"],
+            "chunks": len(snapshot["chunks"]),
+            "words": stats["words"],
+            "records": len(snapshot["records"]),
+            "unparsed": stats["unparsed"],
+            "collisions": stats["collisions"],
+            "snapshot_json": path,
+        }
+        print(json.dumps(summary, indent=2, sort_keys=True))
+    else:
+        print(json.dumps(snapshot, indent=2, sort_keys=True))
+    print(f"server_snapshot: {stats['files']} file(s), {stats['lines']} line(s), "
+          f"{stats['chunks']} chunk(s), {stats['words']} word(s), "
+          f"{stats['records']} record(s), {stats['unparsed']} unparsed, "
+          f"{stats['merged']} merged, {stats['collisions']} collision(s), "
+          f"{stats['conflicts']} conflict(s)", file=sys.stderr)
+    return 0
+
+
+def _diff_command(args) -> int:
+    report = diff_snapshots(_load_json(args.a), _load_json(args.b))
+    if args.json:
+        path = _write_json(args.json, report)
+        print(json.dumps({"schema": report["schema"], **report["summary"],
+                          "diff_json": path}, indent=2, sort_keys=True))
+    else:
+        print(json.dumps(report, indent=2, sort_keys=True))
+    return 0 if report["summary"]["identical"] else 3
+
+
+def _demo_command(args) -> int:
+    if args.out:
+        os.makedirs(args.out, exist_ok=True)
+        demo = run_demo(args.out)
+    else:
+        with tempfile.TemporaryDirectory(prefix="server-snapshot-demo-") as workdir:
+            demo = run_demo(workdir)
+    for row in demo["diff"]["chunks"]:
+        if row["status"] != "unchanged":
+            print(f"{row['status']:<9} {row['label']:<16} "
+                  f"+{len(row['added'])} -{len(row['removed'])} "
+                  f"~{len(row['changed'])}")
+    print(json.dumps(demo["diff"]["summary"], indent=2, sort_keys=True))
+    return 0
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    try:
+        if args.command == "collect":
+            return _collect_command(args)
+        if args.command == "diff":
+            return _diff_command(args)
+        if args.command == "demo":
+            return _demo_command(args)
+        parser.print_help()
+        return 2
+    except SnapshotError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
-- 
cgit v1.2.3