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

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