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

#!/usr/bin/env python3
"""Version diff: old vs new firmware targets (roadmap H4.1, planned docs/21).

Compares two firmware images (a built ELF or a raw ``.fwimage``) section by
section; every section degrades to ``status: skipped`` with a reason instead
of aborting the report:

1. image/shape       -- entry, PT_LOAD layout, ELF section sizes, hashes, span;
2. symbol inventory  -- ELF ``.symtab`` plus Ghidra names/index/exports TSVs
                        (including the 3-column ``ExportForGreenEye`` form) and
                        green-eye JSON DBs; added/removed/moved/resized with
                        the anonymous-shift matching discipline of
                        ``userspace/cuda/tools/diff_corpus.py``;
3. decompiled hints  -- a corpus diff over two ``decompiled_index.tsv`` files,
                        an attached corpus report, or an explicit "not
                        supplied";
4. probe             -- bounded QEMU first-touch probe of both targets with
                        ``parse_qemu_faults`` and firmware-vaddr translation of
                        the epc and access addresses;
5. behavior          -- the shared demo scenario (or a ``--stimulus`` script)
                        under QEMU, ``analyze.compare_sessions`` plus
                        hit/watch/write deltas and the docs/06 M1-M6 skeleton.

Everything is local and stdlib-only: nothing is downloaded. The RM 610 gate
(docs/23) is enforced before any target file is opened.

Run from the tool root:

    python3 harness/version_diff.py --old old.elf --new new.elf --out dumps/vdiff
"""
from __future__ import annotations

import argparse
import contextlib
import os
import re
import shutil
import statistics
import struct
import sys
import tempfile
from collections import defaultdict
from dataclasses import dataclass
from pathlib import Path

try:
    from .common import (
        LOAD_ADDR,
        MAX_FLAT_BYTES,
        MAX_LOG_BYTES,
        SymbolDB,
        counter_delta,
        ensure_dir,
        memory_mb_for,
        qemu_available,
        read_log_window,
        sha256_file,
        write_json,
        write_text,
    )
    from .portability_matrix import (
        CTRL_RPC_COUNT,
        METRIC_DEFS,
        NO_REFERENCE,
        install_scenario,
        metric_rows,
        profile_from_events,
        read_ctrl,
    )
except ImportError:  # direct script execution (python3 harness/version_diff.py)
    from common import (  # type: ignore[no-redef]
        LOAD_ADDR,
        MAX_FLAT_BYTES,
        MAX_LOG_BYTES,
        SymbolDB,
        counter_delta,
        ensure_dir,
        memory_mb_for,
        qemu_available,
        read_log_window,
        sha256_file,
        write_json,
        write_text,
    )
    from portability_matrix import (  # type: ignore[no-redef]
        CTRL_RPC_COUNT,
        METRIC_DEFS,
        NO_REFERENCE,
        install_scenario,
        metric_rows,
        profile_from_events,
        read_ctrl,
    )

from greeneye.analyze import compare_sessions, parse_qemu_faults  # noqa: E402
from greeneye.elf import EM_RISCV, FirmwareImage, Segment, load_elf, write_flat_kernel  # noqa: E402
from greeneye.errors import GreenEyeError  # noqa: E402
from greeneye.record import load_events  # noqa: E402
from greeneye.rsp import RspClient  # noqa: E402
from greeneye.session import Session  # noqa: E402
from greeneye.symbols import Symbol  # noqa: E402
from greeneye.target import QemuTarget  # noqa: E402

FORMAT = "greeneye-version-diff/1"
RM610_DOC = "docs/23-RM610-READINESS.md"
RM610_TOKEN = "RM610-APPROVED"
RM610_ENV = "GREENEYE_RM610_APPROVED"
RM610_RE = re.compile(r"rm[\s._-]*610", re.IGNORECASE)
PT_LOAD = 1


class Rm610GateError(GreenEyeError):
    """A target/release carries an RM 610 label without explicit approval."""


def is_rm610_label(value: str | None) -> bool:
    return bool(value) and RM610_RE.search(str(value)) is not None


def rm610_approved(token: str | None = None, environ: dict | None = None) -> bool:
    """Explicit approval is ``--approval-token RM610-APPROVED`` or the env flag."""
    env = os.environ if environ is None else environ
    token_ok = (token or "").strip() == RM610_TOKEN
    env_ok = str(env.get(RM610_ENV, "")).strip().lower() in ("1", "true", "yes", "approved")
    return token_ok or env_ok


def check_rm610_gate(
    labels: list[str] | tuple[str, ...],
    *,
    approval_token: str | None = None,
    environ: dict | None = None,
) -> None:
    """Refuse RM 610 labels before anything is read, opened or downloaded."""
    blocked = sorted({str(label) for label in labels if is_rm610_label(label)})
    if not blocked:
        return
    if rm610_approved(approval_token, environ):
        return
    raise Rm610GateError(
        "target/release label(s) " + ", ".join(repr(label) for label in blocked) + " indicate RM 610. "
        f"Refused without explicit approval: pass --approval-token {RM610_TOKEN} or set "
        f"{RM610_ENV}=1. See {RM610_DOC}. Nothing was downloaded and no target file was opened."
    )


# ---------------------------------------------------------------------------
# target loading (ELF or raw)
# ---------------------------------------------------------------------------


@dataclass
class LoadedTarget:
    path: str
    name: str
    kind: str  # elf | raw | unsupported | missing
    image: FirmwareImage | None = None
    sha256: str = ""
    size: int = 0
    entry: int = 0
    bias: int = 0
    reason: str | None = None

    @property
    def ok(self) -> bool:
        return self.kind in ("elf", "raw")


def _read_bytes(path: str) -> bytes:
    with open(path, "rb") as fh:
        return fh.read()


def load_any_target(path: str) -> LoadedTarget:
    """Read a target as RISC-V ELF or as a raw relocated image (no download)."""
    name = Path(path).stem
    if not os.path.isfile(path):
        return LoadedTarget(path=path, name=name, kind="missing", reason="file not found")
    size = os.path.getsize(path)
    sha = sha256_file(path)
    with open(path, "rb") as fh:
        magic = fh.read(4)
    if magic == b"\x7fELF":
        try:
            image = load_elf(path)
        except Exception as exc:  # malformed ELF -> skipped, never fatal
            return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=size,
                                reason=f"{type(exc).__name__}: {exc}")
        if not image.is_riscv:
            return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=size,
                                reason=f"not a RISC-V ELF (e_machine={image.e_machine})")
        span = image.max_vaddr - image.min_vaddr
        if span > MAX_FLAT_BYTES:
            return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=size,
                                reason=f"image span {span:#x} exceeds the {MAX_FLAT_BYTES:#x} harness cap")
        return LoadedTarget(path=path, name=name, kind="elf", image=image, sha256=sha, size=size,
                            entry=image.entry, bias=LOAD_ADDR - image.min_vaddr)
    if size == 0:
        return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=0,
                            reason="empty raw image")
    if size > MAX_FLAT_BYTES:
        return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=size,
                            reason=f"raw image {size:#x} exceeds the {MAX_FLAT_BYTES:#x} harness cap")
    data = _read_bytes(path)
    segment = Segment(p_type=PT_LOAD, p_flags=0x7, p_offset=0, p_vaddr=0, p_paddr=0,
                      p_filesz=len(data), p_memsz=len(data), p_align=0x1000)
    image = FirmwareImage(path=path, is64=True, e_machine=EM_RISCV, entry=0, segments=[segment],
                          symbols=[], _blob=data)
    return LoadedTarget(path=path, name=name, kind="raw", image=image, sha256=sha, size=size,
                        entry=0, bias=LOAD_ADDR)


# ---------------------------------------------------------------------------
# 1. image / shape
# ---------------------------------------------------------------------------


def _cstr(blob: bytes, offset: int) -> str:
    if offset >= len(blob):
        return ""
    end = blob.find(b"\x00", offset)
    if end == -1:
        end = len(blob)
    return blob[offset:end].decode("utf-8", "replace")


def elf_sections(data: bytes) -> list[dict]:
    """Section name/type/addr/size table (minimal reader; [] when unusable)."""
    if len(data) < 64 or data[:4] != b"\x7fELF" or data[4] not in (1, 2):
        return []
    is64 = data[4] == 2
    try:
        if is64:
            sh_off = struct.unpack_from("<Q", data, 40)[0]
            shentsize, shnum, shstrndx = struct.unpack_from("<HHH", data, 58)
            sh_fmt = "<IIQQQQIIQQ"
        else:
            sh_off = struct.unpack_from("<I", data, 32)[0]
            shentsize, shnum, shstrndx = struct.unpack_from("<HHH", data, 46)
            sh_fmt = "<IIIIIIIIII"
    except struct.error:
        return []
    sh_size = struct.calcsize(sh_fmt)
    if not sh_off or sh_off + sh_size > len(data):
        return []
    shstr = b""
    if 0 <= shstrndx < shnum:
        base = sh_off + shstrndx * shentsize
        if base + sh_size <= len(data):
            sh = struct.unpack_from(sh_fmt, data, base)
            shstr = data[sh[4] : sh[4] + sh[5]]
    sections: list[dict] = []
    for index in range(shnum):
        base = sh_off + index * shentsize
        if base + sh_size > len(data):
            break
        sh = struct.unpack_from(sh_fmt, data, base)
        sections.append(
            {
                "name": _cstr(shstr, sh[0]) or f"<section-{index}>",
                "type": sh[1],
                "addr": sh[3],
                "offset": sh[4],
                "size": sh[5],
            }
        )
    return sections


def shape_payload(loaded: LoadedTarget) -> dict:
    payload = {
        "kind": loaded.kind,
        "sha256": loaded.sha256,
        "size": loaded.size,
        "reason": loaded.reason,
    }
    if not loaded.ok or loaded.image is None:
        return payload
    image = loaded.image
    payload.update(
        {
            "entry": loaded.entry,
            "bias": loaded.bias,
            "min_vaddr": image.min_vaddr,
            "max_vaddr": image.max_vaddr,
            "span": image.max_vaddr - image.min_vaddr,
            "loads": [
                {
                    "index": index,
                    "vaddr": seg.p_vaddr,
                    "paddr": seg.p_paddr,
                    "filesz": seg.p_filesz,
                    "memsz": seg.p_memsz,
                    "flags": seg.p_flags,
                    "align": seg.p_align,
                }
                for index, seg in enumerate(image.load_segments)
            ],
        }
    )
    if loaded.kind == "elf":
        payload["sections"] = elf_sections(image.read())
    else:
        payload["sections"] = []
        payload["note"] = "raw image: one implicit load at vaddr 0; no ELF section table"
    return payload


def _scalar_delta(a, b, name: str) -> dict:
    return {"a": a, "b": b, "delta": b - a if isinstance(a, int) and isinstance(b, int) else None,
            "changed": a != b}


def _cap(rows: list, max_rows: int) -> list:
    return rows[:max_rows]


def image_diff(old: LoadedTarget, new: LoadedTarget, max_rows: int) -> dict:
    a, b = shape_payload(old), shape_payload(new)
    result = {"status": "skipped", "reason": None, "a": a, "b": b, "same": None, "deltas": None}
    if not old.ok or not new.ok:
        result["reason"] = "; ".join(
            f"{side}: {loaded.reason or loaded.kind}" for side, loaded in (("old", old), ("new", new)) if not loaded.ok
        )
        return result

    loads_a, loads_b = a["loads"], b["loads"]
    load_added = [payload for index, payload in enumerate(loads_b) if index >= len(loads_a)]
    load_removed = [payload for index, payload in enumerate(loads_a) if index >= len(loads_b)]
    load_changed = []
    for index, (la, lb) in enumerate(zip(loads_a, loads_b)):
        fields = {key: _scalar_delta(la[key], lb[key], key) for key in ("vaddr", "filesz", "memsz", "flags")}
        if any(row["changed"] for row in fields.values()):
            load_changed.append({"index": index, "fields": fields})

    sec_a = {row["name"]: row for row in a.get("sections", [])}
    sec_b = {row["name"]: row for row in b.get("sections", [])}
    sec_added = [{key: sec_b[name][key] for key in ("name", "type", "addr", "size")} for name in sorted(set(sec_b) - set(sec_a))]
    sec_removed = [{key: sec_a[name][key] for key in ("name", "type", "addr", "size")} for name in sorted(set(sec_a) - set(sec_b))]
    sec_moved, sec_resized = [], []
    for name in sorted(set(sec_a) & set(sec_b)):
        ra, rb = sec_a[name], sec_b[name]
        if ra["addr"] != rb["addr"]:
            sec_moved.append({"name": name, "old_addr": ra["addr"], "new_addr": rb["addr"]})
        if ra["size"] != rb["size"]:
            sec_resized.append({"name": name, "old_size": ra["size"], "new_size": rb["size"]})

    deltas = {
        "entry": _scalar_delta(a["entry"], b["entry"], "entry"),
        "span": _scalar_delta(a["span"], b["span"], "span"),
        "file_size": _scalar_delta(a["size"], b["size"], "size"),
        "loads": {
            "added": _cap(load_added, max_rows),
            "removed": _cap(load_removed, max_rows),
            "changed": _cap(load_changed, max_rows),
        },
        "sections": {
            "added": _cap(sec_added, max_rows),
            "removed": _cap(sec_removed, max_rows),
            "moved": _cap(sec_moved, max_rows),
            "resized": _cap(sec_resized, max_rows),
        },
    }
    same = (
        not deltas["entry"]["changed"]
        and not deltas["span"]["changed"]
        and not deltas["file_size"]["changed"]
        and not any(deltas["loads"][key] for key in ("added", "removed", "changed"))
        and not any(deltas["sections"][key] for key in ("added", "removed", "moved", "resized"))
    )
    result.update({"status": "ok", "same": same, "deltas": deltas})
    return result


# ---------------------------------------------------------------------------
# 2. symbol inventory (diff_corpus.py matching discipline)
# ---------------------------------------------------------------------------


def looks_like_three_column_export(path: str) -> bool:
    """``addr<TAB>name<TAB>size`` (ExportForGreenEye) vs names/exports TSVs."""
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            for line in fh:
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                cols = line.split("\t")
                if len(cols) >= 3 and [c.strip() for c in cols[:3]] == ["addr", "name", "size"]:
                    return True
                if len(cols) == 3:
                    try:
                        int(cols[0], 16)
                        int(cols[2], 0)
                        return True
                    except ValueError:
                        return False
                return False
    except OSError:
        return False
    return False


def load_three_column_export(db: SymbolDB, path: str, bias: int = 0) -> int:
    count = 0
    with open(path, "r", encoding="utf-8", errors="replace") as fh:
        for line in fh:
            line = line.rstrip("\n")
            if not line or line.startswith("#"):
                continue
            cols = line.split("\t")
            if len(cols) != 3 or cols[0].strip() == "addr":
                continue
            try:
                addr = int(cols[0], 16)
                size = int(cols[2], 0)
            except ValueError:
                continue
            if not cols[1]:
                continue
            db.add(Symbol(addr=addr + bias, name=cols[1], size=size, source="ghidra-export"))
            count += 1
    return count


def load_symbol_sources(
    loaded: LoadedTarget,
    *,
    symbols_json: str | None = None,
    ghidra_names: str | None = None,
    ghidra_index: str | None = None,
    ghidra_exports: str | None = None,
    ghidra_bias: int = 0,
) -> tuple[SymbolDB, list[str], list[str]]:
    """Merge every supplied symbol source for one side into one SymbolDB."""
    db = SymbolDB()
    sources: list[str] = []
    notes: list[str] = []
    if loaded.ok and loaded.image is not None and loaded.image.symbols:
        count = db.load_elf(loaded.image, bias=loaded.bias, source="elf")
        if count:
            sources.append(f"elf:{count}")
    if symbols_json:
        if os.path.isfile(symbols_json):
            count = db.load_json(symbols_json)
            sources.append(f"{os.path.basename(symbols_json)}:{count}")
        else:
            notes.append(f"symbol DB not found: {symbols_json}")
    if ghidra_names:
        if os.path.isfile(ghidra_names):
            count = db.load_names_tsv(ghidra_names, bias=ghidra_bias)
            sources.append(f"{os.path.basename(ghidra_names)}:{count}")
        else:
            notes.append(f"Ghidra names TSV not found: {ghidra_names}")
    if ghidra_index:
        if os.path.isfile(ghidra_index):
            count = db.load_index_tsv(ghidra_index, bias=ghidra_bias)
            sources.append(f"{os.path.basename(ghidra_index)}:{count}")
        else:
            notes.append(f"Ghidra index TSV not found: {ghidra_index}")
    if ghidra_exports:
        if os.path.isfile(ghidra_exports):
            if looks_like_three_column_export(ghidra_exports):
                count = load_three_column_export(db, ghidra_exports, bias=ghidra_bias)
                sources.append(f"{os.path.basename(ghidra_exports)}:{count} (3-column export)")
            else:
                count = db.load_index_tsv(ghidra_exports, bias=ghidra_bias)
                sources.append(f"{os.path.basename(ghidra_exports)}:{count}")
        else:
            notes.append(f"Ghidra exports TSV not found: {ghidra_exports}")
    return db, sources, notes


def inventory_records(db: SymbolDB) -> list[dict]:
    records = [
        {
            "name": symbol.name,
            "addr": symbol.addr,
            "size": symbol.size,
            "source": symbol.source,
        }
        for symbol in db
    ]
    records.sort(key=lambda row: (row["addr"], row["name"], row["size"]))
    return records


def _is_named(name: str) -> bool:
    return bool(name) and not name.startswith("FUN_")


def _record_brief(record: dict | None) -> dict | None:
    if record is None:
        return None
    return {"addr": record["addr"], "name": record["name"], "size": record["size"]}


def inventory_diff(a_records: list[dict], b_records: list[dict], max_rows: int = 40) -> dict:
    """Named-by-name + anonymous-at-shifted-address matching (diff_corpus.py)."""
    a_by_addr = {row["addr"]: row for row in a_records}
    b_by_addr = {row["addr"]: row for row in b_records}
    named_a: dict[str, list[int]] = defaultdict(list)
    named_b: dict[str, list[int]] = defaultdict(list)
    for row in a_records:
        if _is_named(row["name"]):
            named_a[row["name"]].append(row["addr"])
    for row in b_records:
        if _is_named(row["name"]):
            named_b[row["name"]].append(row["addr"])

    matched_named: list[tuple[str, int, int]] = []
    for name in sorted(set(named_a) & set(named_b)):
        for addr_a, addr_b in zip(sorted(named_a[name]), sorted(named_b[name])):
            matched_named.append((name, addr_a, addr_b))

    deltas = [addr_b - addr_a for _, addr_a, addr_b in matched_named]
    shift = int(statistics.median(deltas)) if deltas else 0
    dominant = (sum(1 for value in deltas if abs(value - shift) <= 0x40) / len(deltas)) if deltas else 0.0

    used_a = {addr_a for _, addr_a, _ in matched_named}
    used_b = {addr_b for _, _, addr_b in matched_named}
    moved = [(name, addr_a, addr_b) for name, addr_a, addr_b in matched_named if addr_a != addr_b]
    resized_named = [
        (name, addr_a, addr_b)
        for name, addr_a, addr_b in matched_named
        if a_by_addr[addr_a]["size"] != b_by_addr[addr_b]["size"]
    ]

    shifted: list[tuple[int, int]] = []
    for row in sorted(a_records, key=lambda item: item["addr"]):
        addr_a = row["addr"]
        if addr_a in used_a or _is_named(row["name"]):
            continue
        addr_b = addr_a + shift
        if addr_b in b_by_addr and addr_b not in used_b:
            shifted.append((addr_a, addr_b))
            used_a.add(addr_a)
            used_b.add(addr_b)
    shifted_resized = [(fa, fb) for fa, fb in shifted if a_by_addr[fa]["size"] != b_by_addr[fb]["size"]]
    shifted_ok = [(fa, fb) for fa, fb in shifted if a_by_addr[fa]["size"] == b_by_addr[fb]["size"]]

    removed = sorted(addr for addr in a_by_addr if addr not in used_a)
    added = sorted(addr for addr in b_by_addr if addr not in used_b)
    matched_total = len(matched_named) + len(shifted)
    named_count_a = sum(len(v) for v in named_a.values())
    named_count_b = sum(len(v) for v in named_b.values())

    return {
        "shift": shift,
        "shift_dominant": round(dominant, 4),
        "matched_named": len(matched_named),
        "matched_shifted": len(shifted_ok),
        "matched_shifted_resized": len(shifted_resized),
        "moved": _cap(
            [{"name": name, "old_addr": fa, "new_addr": fb, "delta": fb - fa} for name, fa, fb in moved],
            max_rows,
        ),
        "resized": _cap(
            [
                {
                    "name": name,
                    "old_addr": fa,
                    "new_addr": fb,
                    "old_size": a_by_addr[fa]["size"],
                    "new_size": b_by_addr[fb]["size"],
                }
                for name, fa, fb in resized_named
            ],
            max_rows,
        ),
        "resized_anonymous": _cap(
            [
                {
                    "old_addr": fa,
                    "new_addr": fb,
                    "old_size": a_by_addr[fa]["size"],
                    "new_size": b_by_addr[fb]["size"],
                }
                for fa, fb in shifted_resized
            ],
            max_rows,
        ),
        "added": _cap([_record_brief(b_by_addr[addr]) for addr in added], max_rows),
        "removed": _cap([_record_brief(a_by_addr[addr]) for addr in removed], max_rows),
        "totals": {
            "a": len(a_records),
            "b": len(b_records),
            "named_a": named_count_a,
            "named_b": named_count_b,
            "anonymous_a": sum(1 for row in a_records if not _is_named(row["name"])),
            "anonymous_b": sum(1 for row in b_records if not _is_named(row["name"])),
            "added": len(added),
            "removed": len(removed),
            "moved": len(moved),
            "resized_named": len(resized_named),
        },
        "rates": {
            "match_a": round(matched_total / len(a_records), 4) if a_records else 0.0,
            "match_b": round(matched_total / len(b_records), 4) if b_records else 0.0,
            "named_transfer": round(len(matched_named) / named_count_a, 4) if named_count_a else 0.0,
        },
    }


def symbol_section(
    old: LoadedTarget,
    new: LoadedTarget,
    old_db: SymbolDB,
    old_sources: list[str],
    old_notes: list[str],
    new_db: SymbolDB,
    new_sources: list[str],
    new_notes: list[str],
    options: "VersionDiffOptions",
    out_dir: str,
) -> dict:
    result = {
        "status": "skipped",
        "reason": None,
        "a": {"count": 0, "sources": old_sources, "notes": old_notes, "db": None},
        "b": {"count": 0, "sources": new_sources, "notes": new_notes, "db": None},
    }
    if not old.ok or not new.ok:
        result["reason"] = "; ".join(
            f"{side}: {loaded.reason or loaded.kind}" for side, loaded in (("old", old), ("new", new)) if not loaded.ok
        )
        return result
    a_records = inventory_records(old_db)
    b_records = inventory_records(new_db)
    result["a"]["count"] = len(a_records)
    result["b"]["count"] = len(b_records)
    if not a_records and not b_records:
        result["reason"] = (
            "no symbols on either side: no ELF .symtab and no Ghidra/JSON symbol sources supplied"
        )
        return result
    result.update(inventory_diff(a_records, b_records, options.max_rows))
    result["status"] = "ok"
    for key, db, loaded, sources, notes in (
        ("a", old_db, old, old_sources, old_notes),
        ("b", new_db, new, new_sources, new_notes),
    ):
        if not len(db):
            continue
        rel = os.path.join("symbols", f"{loaded.name}.json")
        path = os.path.join(out_dir, rel)
        ensure_dir(os.path.dirname(path))
        db.save_json(path, meta={"firmware": os.path.basename(loaded.path), "bias": loaded.bias,
                                 "sources": sources, "notes": notes})
        result[key]["db"] = rel
    return result


# ---------------------------------------------------------------------------
# 3. decompiled-change hints
# ---------------------------------------------------------------------------


def corpus_records(path: str) -> list[dict]:
    db = SymbolDB()
    if looks_like_three_column_export(path):
        load_three_column_export(db, path, bias=0)
    else:
        db.load_index_tsv(path, bias=0)
    return inventory_records(db)


def corpus_diff(path_a: str, path_b: str, max_rows: int = 40) -> dict:
    diff = inventory_diff(corpus_records(path_a), corpus_records(path_b), max_rows)
    diff["a"] = os.path.basename(path_a)
    diff["b"] = os.path.basename(path_b)
    return diff


def corpus_section(options: "VersionDiffOptions", out_dir: str) -> dict:
    if options.corpus_report:
        if not os.path.isfile(options.corpus_report):
            return {"status": "skipped", "reason": f"corpus report not found: {options.corpus_report}",
                    "mode": "corpus-report"}
        data = _read_bytes(options.corpus_report)
        preview = data.decode("utf-8", "replace")
        return {
            "status": "ok",
            "reason": None,
            "mode": "corpus-report",
            "report": {
                "path": os.path.basename(options.corpus_report),
                "sha256": sha256_file(options.corpus_report),
                "bytes": len(data),
                "preview": preview[:4000],
                "truncated": len(preview) > 4000,
            },
        }
    if options.corpus_diff:
        path_a, path_b = options.corpus_diff
        missing = [path for path in (path_a, path_b) if not os.path.isfile(path)]
        if missing:
            return {"status": "skipped", "mode": "corpus-diff",
                    "reason": "corpus TSV not found: " + ", ".join(missing)}
        diff = corpus_diff(path_a, path_b, options.max_rows)
        rel = os.path.join("corpus", "corpus-diff.json")
        write_json(os.path.join(out_dir, rel), diff)
        return {"status": "ok", "reason": None, "mode": "corpus-diff", "diff": diff, "report": rel}
    return {
        "status": "skipped",
        "reason": "not supplied: pass --corpus-diff OLD_TSV NEW_TSV or --corpus-report FILE",
        "mode": None,
    }


# ---------------------------------------------------------------------------
# QEMU sessions for ELF and raw targets
# ---------------------------------------------------------------------------


@dataclass
class OpenTarget:
    loaded: LoadedTarget
    session: Session
    target: QemuTarget
    workdir: str
    owns_workdir: bool = False


@contextlib.contextmanager
def open_target_session(
    loaded: LoadedTarget,
    *,
    record: str | bool | None,
    name: str,
    symbols: SymbolDB | None = None,
    log_faults: bool = True,
    workdir: str | None = None,
):
    """Relocate the target to LOAD_ADDR, start QEMU and yield a live session."""
    owns_workdir = workdir is None
    if workdir is None:
        workdir = tempfile.mkdtemp(prefix="greeneye-vdiff-")
    flat_kernel = os.path.join(workdir, "flat.bin")
    assert loaded.image is not None
    if loaded.kind == "elf":
        write_flat_kernel(loaded.path, flat_kernel, load_addr=LOAD_ADDR)
    else:
        with open(flat_kernel, "wb") as fh:
            fh.write(loaded.image.read())
    target = QemuTarget(
        kernel=flat_kernel,
        memory_mb=memory_mb_for(loaded.image),
        dlog="guest_errors,unimp,int" if log_faults else None,
        workdir=workdir,
    )
    target.start(paused=True)
    client = RspClient(target.host, target.port)
    client.connect()
    client.sync()
    db = symbols if symbols is not None else SymbolDB()
    session = Session(
        client=client,
        target=target,
        symbols=db,
        record=record,
        name=name,
        meta={
            "backend": "qemu",
            "bias": loaded.bias,
            "firmware": loaded.path,
            "kind": loaded.kind,
            "harness": "version_diff",
        },
    )
    session.load_image(loaded.image, bias=loaded.bias, load_symbols=symbols is None)
    try:
        yield OpenTarget(loaded=loaded, session=session, target=target, workdir=workdir,
                         owns_workdir=owns_workdir)
    finally:
        try:
            session.close()
        finally:
            if owns_workdir:
                shutil.rmtree(workdir, ignore_errors=True)


def enrich_fault(fault: dict, bias: int, span: int) -> dict:
    """Add firmware-vaddr translations for epc and the access address."""
    out = {
        "cause": fault.get("cause"),
        "desc": fault.get("desc"),
        "epc": fault.get("epc"),
        "tval": fault.get("tval"),
        "access": fault.get("access"),
    }
    epc = fault.get("epc")
    out["epc_firmware"] = epc - bias if isinstance(epc, int) else None
    access = fault.get("access") or {}
    access_addr = access.get("access_addr")
    if isinstance(access_addr, int):
        if LOAD_ADDR <= access_addr < LOAD_ADDR + span:
            out["access_firmware"] = access_addr - bias
        else:
            out["access_firmware"] = None
    else:
        out["access_firmware"] = None
    return out


# ---------------------------------------------------------------------------
# 4. probe first-touch diff
# ---------------------------------------------------------------------------


def _skipped_probe(reason: str | None) -> dict:
    return {
        "status": "skipped",
        "reason": reason,
        "session": None,
        "instructions": None,
        "stopped_pc": None,
        "stopped_symbol": None,
        "fault_count": 0,
        "faults_capped": False,
        "first_fault": None,
    }


def run_probe(
    loaded: LoadedTarget,
    db: SymbolDB,
    options: "VersionDiffOptions",
    out_dir: str,
    qemu_ok: bool,
) -> dict:
    if not qemu_ok:
        return _skipped_probe("qemu-system-riscv64 not found")
    if not loaded.ok:
        return _skipped_probe(loaded.reason)
    result = _skipped_probe(None)
    result["instructions"] = options.probe_instructions
    session_rel = os.path.join("sessions", f"{loaded.name}-probe")
    try:
        with open_target_session(
            loaded,
            record=os.path.join(out_dir, session_rel),
            name=f"{loaded.name}-probe",
            symbols=db,
            log_faults=True,
        ) as ot:
            session = ot.session
            if options.probe_instructions > 0:
                session.step(options.probe_instructions)
            else:
                session.run(duration=options.probe_duration)
            result["stopped_pc"] = session.pc
            symbol = session.symbols.resolve(session.pc)
            result["stopped_symbol"] = symbol.name if symbol else None
            text, capped = read_log_window(ot.target.qemu_log, 0, options.log_bytes)
            faults = parse_qemu_faults(text)
            result["fault_count"] = len(faults)
            result["faults_capped"] = capped
            span = loaded.image.max_vaddr - loaded.image.min_vaddr if loaded.image else 0
            result["first_fault"] = enrich_fault(faults[0], loaded.bias, span) if faults else None
            result["session"] = session_rel
            if text:
                session.recorder.meta["qemu_log_blob"] = session.recorder.blob("qemu.log", text.encode("utf-8"))
                session.recorder.meta["faults_capped"] = capped
            result["status"] = "ok"
    except Exception as exc:  # a failed probe must not abort the diff
        result["status"] = "error"
        result["reason"] = f"{type(exc).__name__}: {exc}"
    return result


def _probe_signature(probe: dict) -> tuple:
    first = probe.get("first_fault") or {}
    return (
        probe.get("stopped_pc"),
        first.get("desc"),
        first.get("cause"),
        first.get("epc_firmware"),
        first.get("access_firmware"),
        (first.get("access") or {}).get("access_addr"),
    )


def probe_diff(
    old: LoadedTarget,
    new: LoadedTarget,
    old_db: SymbolDB,
    new_db: SymbolDB,
    options: "VersionDiffOptions",
    out_dir: str,
    qemu_ok: bool,
) -> dict:
    a = run_probe(old, old_db, options, out_dir, qemu_ok)
    b = run_probe(new, new_db, options, out_dir, qemu_ok)
    statuses = {a["status"], b["status"]}
    if statuses == {"ok"}:
        status, reason = "ok", None
    elif "error" in statuses:
        status = "error"
        reason = "; ".join(row["reason"] for row in (a, b) if row.get("reason")) or "probe failed"
    else:
        status = "skipped"
        reason = "; ".join(row["reason"] for row in (a, b) if row.get("reason")) or "probe not run"
    changed = _probe_signature(a) != _probe_signature(b)
    if status == "ok":
        summary = "first fault and stopped PC unchanged" if not changed else "first fault or stopped PC changed"
    else:
        summary = None
    return {"status": status, "reason": reason, "a": a, "b": b, "changed": changed, "summary": summary}


# ---------------------------------------------------------------------------
# 5. behavior diff
# ---------------------------------------------------------------------------


def _skipped_behavior(reason: str | None) -> dict:
    return {
        "status": "skipped",
        "reason": reason,
        "session": None,
        "hooks": [],
        "uart_banner": None,
        "ctrl": None,
        "event_counts": {},
        "bp": {},
        "watch": {},
        "writes": {},
        "fault_count": 0,
        "faults_capped": False,
        "first_fault": None,
    }


def run_behavior(
    loaded: LoadedTarget,
    db: SymbolDB,
    options: "VersionDiffOptions",
    out_dir: str,
    qemu_ok: bool,
) -> dict:
    if not qemu_ok:
        return _skipped_behavior("qemu-system-riscv64 not found")
    if not loaded.ok:
        return _skipped_behavior(loaded.reason)
    result = _skipped_behavior(None)
    session_rel = os.path.join("sessions", f"{loaded.name}-behavior")
    try:
        with open_target_session(
            loaded,
            record=os.path.join(out_dir, session_rel),
            name=f"{loaded.name}-behavior",
            symbols=db,
            log_faults=True,
        ) as ot:
            session = ot.session
            result["hooks"] = install_scenario(session)
            ctrl_symbol = session.symbols.get("gsp_ctrl")
            ctrl_addr = ctrl_symbol.addr if ctrl_symbol else None
            if ctrl_addr is not None:
                session.watch(ctrl_addr + CTRL_RPC_COUNT, size=8, kind="w",
                              listener="version-diff:rpc_count")
            if options.stimulus:
                import greeneye as ge

                script_path = Path(options.stimulus)
                namespace = {
                    "ge": ge,
                    "greeneye": ge,
                    "session": session,
                    "s": session,
                    "memory": session.memory,
                    "symbols": session.symbols,
                    "recorder": session.recorder,
                }
                exec(compile(script_path.read_text(encoding="utf-8"), str(script_path), "exec"), namespace)
            else:
                session.run(duration=options.behavior_duration, max_stops=options.max_stops)
                if ctrl_addr is not None:
                    session.memory.write_u64(ctrl_addr + 32, 1)  # PING
                    session.memory.write_u64(ctrl_addr + 40, 0x1234)
                    session.run(duration=options.behavior_duration, max_stops=options.max_stops)
            result["ctrl"] = read_ctrl(session, ctrl_symbol)
            uart = ot.target.read_uart()
            result["uart_banner"] = uart.strip().splitlines()[0] if uart.strip() else None
            text, capped = read_log_window(ot.target.qemu_log, 0, options.log_bytes)
            faults = parse_qemu_faults(text)
            result["fault_count"] = len(faults)
            result["faults_capped"] = capped
            span = loaded.image.max_vaddr - loaded.image.min_vaddr if loaded.image else 0
            result["first_fault"] = enrich_fault(faults[0], loaded.bias, span) if faults else None
        result.update(profile_from_events(load_events(os.path.join(out_dir, session_rel))))
        result["session"] = session_rel
        result["status"] = "ok"
    except Exception as exc:  # a failed scenario must not abort the diff
        result["status"] = "error"
        result["reason"] = f"{type(exc).__name__}: {exc}"
    return result


def behavior_diff(
    old: LoadedTarget,
    new: LoadedTarget,
    old_db: SymbolDB,
    new_db: SymbolDB,
    probe_a: dict,
    probe_b: dict,
    options: "VersionDiffOptions",
    out_dir: str,
    qemu_ok: bool,
) -> dict:
    a = run_behavior(old, old_db, options, out_dir, qemu_ok)
    b = run_behavior(new, new_db, options, out_dir, qemu_ok)
    entry = {
        "status": "skipped",
        "reason": None,
        "a": a,
        "b": b,
        "compare_report": None,
        "deltas": {"bp": [], "watch": [], "writes": []},
        "all_zero": None,
        "metrics": [],
    }
    a["metrics"] = metric_rows(probe_a, a)
    b["metrics"] = metric_rows(probe_b, b)
    if a["status"] != "ok" or b["status"] != "ok":
        reasons = [row.get("reason") for row in (a, b) if row.get("reason")]
        entry["reason"] = "; ".join(reasons) or "a behavior session is not available on both targets"
        entry["status"] = "error" if "error" in {a["status"], b["status"]} else "skipped"
        return entry
    a_dir = os.path.join(out_dir, a["session"])
    b_dir = os.path.join(out_dir, b["session"])
    try:
        report = compare_sessions(a_dir, b_dir, max_rows=options.max_rows)
        rel = os.path.join("compare", f"{old.name}--{new.name}.md")
        write_text(os.path.join(out_dir, rel), report)
        entry["compare_report"] = rel
    except Exception as exc:
        entry["reason"] = f"{type(exc).__name__}: {exc}"
    deltas = {
        name: counter_delta(a.get(name) or {}, b.get(name) or {})
        for name in ("bp", "watch", "writes")
    }
    entry["deltas"] = deltas
    entry["all_zero"] = all(row["delta"] == 0 for rows in deltas.values() for row in rows)
    a_metrics = {row["id"]: row for row in a["metrics"]}
    b_metrics = {row["id"]: row for row in b["metrics"]}
    entry["metrics"] = [
        {
            "id": metric_id,
            "name": name,
            "status": "not measured",
            "reason": NO_REFERENCE,
            "a_observed": (a_metrics.get(metric_id) or {}).get("observed"),
            "b_observed": (b_metrics.get(metric_id) or {}).get("observed"),
        }
        for metric_id, name in METRIC_DEFS
    ]
    entry["status"] = "ok"
    return entry


# ---------------------------------------------------------------------------
# report assembly + rendering
# ---------------------------------------------------------------------------


@dataclass
class VersionDiffOptions:
    old: str
    new: str
    out: str
    old_label: str | None = None
    new_label: str | None = None
    release: str | None = None
    old_symbols: str | None = None
    new_symbols: str | None = None
    old_ghidra_names: str | None = None
    new_ghidra_names: str | None = None
    old_ghidra_index: str | None = None
    new_ghidra_index: str | None = None
    old_ghidra_exports: str | None = None
    new_ghidra_exports: str | None = None
    old_ghidra_bias: int = 0
    new_ghidra_bias: int = 0
    corpus_diff: tuple[str, str] | None = None
    corpus_report: str | None = None
    stimulus: str | None = None
    probe_instructions: int = 2000
    probe_duration: float = 0.5
    behavior_duration: float = 0.4
    max_stops: int = 5000
    max_rows: int = 40
    log_bytes: int = MAX_LOG_BYTES
    approval_token: str | None = None

    def labels(self) -> list[str]:
        return [
            value
            for value in (
                self.release,
                self.old_label,
                self.new_label,
                os.path.basename(self.old),
                os.path.basename(self.new),
            )
            if value
        ]

    def label_for(self, side: str) -> str:
        explicit = self.old_label if side == "old" else self.new_label
        path = self.old if side == "old" else self.new
        return explicit or Path(path).stem


def _target_brief(label: str, loaded: LoadedTarget, qemu_ok: bool, options: VersionDiffOptions) -> dict:
    return {
        "label": label,
        "path": os.path.basename(loaded.path),
        "kind": loaded.kind,
        "status": "ok" if loaded.ok else "skipped",
        "reason": loaded.reason,
        "sha256": loaded.sha256,
        "size": loaded.size,
        "entry": loaded.entry if loaded.ok else None,
        "bias": loaded.bias if loaded.ok else None,
        "probe_instructions": options.probe_instructions,
        "qemu_available": qemu_ok,
    }


def run_version_diff(options: VersionDiffOptions) -> dict:
    """Run every section, write JSON+markdown under ``--out``, return the report."""
    check_rm610_gate(options.labels(), approval_token=options.approval_token)
    ensure_dir(options.out)
    old_label, new_label = options.label_for("old"), options.label_for("new")
    qemu_ok = qemu_available()
    old = load_any_target(options.old)
    new = load_any_target(options.new)

    if old.ok:
        old_db, old_sources, old_notes = load_symbol_sources(
            old, symbols_json=options.old_symbols, ghidra_names=options.old_ghidra_names,
            ghidra_index=options.old_ghidra_index, ghidra_exports=options.old_ghidra_exports,
            ghidra_bias=options.old_ghidra_bias,
        )
    else:
        old_db, old_sources, old_notes = SymbolDB(), [], []
    if new.ok:
        new_db, new_sources, new_notes = load_symbol_sources(
            new, symbols_json=options.new_symbols, ghidra_names=options.new_ghidra_names,
            ghidra_index=options.new_ghidra_index, ghidra_exports=options.new_ghidra_exports,
            ghidra_bias=options.new_ghidra_bias,
        )
    else:
        new_db, new_sources, new_notes = SymbolDB(), [], []

    image = image_diff(old, new, options.max_rows)
    symbols = symbol_section(old, new, old_db, old_sources, old_notes, new_db, new_sources, new_notes,
                             options, options.out)
    corpus = corpus_section(options, options.out)
    probe = probe_diff(old, new, old_db, new_db, options, options.out, qemu_ok)
    behavior = behavior_diff(old, new, old_db, new_db, probe["a"], probe["b"], options, options.out, qemu_ok)

    sections = {"image": image, "symbols": symbols, "corpus": corpus, "probe": probe, "behavior": behavior}
    report = {
        "format": FORMAT,
        "tool": "harness/version_diff.py",
        "backend": "qemu",
        "qemu_available": qemu_ok,
        "old": _target_brief(old_label, old, qemu_ok, options),
        "new": _target_brief(new_label, new, qemu_ok, options),
        "options": {
            "probe_instructions": options.probe_instructions,
            "probe_duration": options.probe_duration,
            "behavior_duration": options.behavior_duration,
            "max_rows": options.max_rows,
            "stimulus": os.path.basename(options.stimulus) if options.stimulus else None,
            "corpus_report": os.path.basename(options.corpus_report) if options.corpus_report else None,
            "corpus_diff": [os.path.basename(p) for p in options.corpus_diff] if options.corpus_diff else None,
        },
        "sections": sections,
        "summary": {
            "sections_ok": sorted(name for name, section in sections.items() if section["status"] == "ok"),
            "sections_skipped": {
                name: section.get("reason")
                for name, section in sections.items()
                if section["status"] == "skipped"
            },
            "sections_error": {
                name: section.get("reason")
                for name, section in sections.items()
                if section["status"] == "error"
            },
        },
    }
    write_json(os.path.join(options.out, "version-diff.json"), report)
    write_text(os.path.join(options.out, "version-diff.md"), render_markdown(report, options))
    return report


# ---------------------------------------------------------------------------
# markdown rendering
# ---------------------------------------------------------------------------


def _table(headers: list[str], rows: list[list]) -> list[str]:
    lines = ["| " + " | ".join(headers) + " |", "|" + "|".join("---" for _ in headers) + "|"]
    for row in rows:
        lines.append("| " + " | ".join(str(cell) for cell in row) + " |")
    return lines


def _hex(value) -> str:
    return f"`{value:#x}`" if isinstance(value, int) else "-"


def render_markdown(report: dict, options: VersionDiffOptions) -> str:
    old, new = report["old"], report["new"]
    lines = [
        "# green-eye version diff",
        "",
        "> **# TOOL IN DEVELOPMENT - NOT READY YET!** Generated by `harness/version_diff.py`.",
        "",
        f"* format: `{report['format']}`",
        f"* targets: `{old['label']}` -> `{new['label']}`",
        f"* backend: `{report['backend']}` (qemu available: {report['qemu_available']})",
        "",
        "## Targets",
        "",
    ]
    lines += _table(
        ["side", "label", "kind", "size", "sha256", "entry"],
        [
            [side, f"`{item['label']}`", item["kind"], item["size"], f"`{(item['sha256'] or '')[:16]}`",
             _hex(item["entry"])]
            for side, item in (("old", old), ("new", new))
        ],
    )
    lines += [""]

    image = report["sections"]["image"]
    lines += ["## 1. Image / shape", "", f"* status: {image['status']}"]
    if image.get("reason"):
        lines += [f"* reason: {image['reason']}"]
    if image["status"] == "ok":
        deltas = image["deltas"]
        lines += _table(
            ["field", "old", "new", "delta"],
            [
                ["entry", _hex(deltas["entry"]["a"]), _hex(deltas["entry"]["b"]), _hex(deltas["entry"]["delta"])],
                ["span", _hex(deltas["span"]["a"]), _hex(deltas["span"]["b"]), _hex(deltas["span"]["delta"])],
                ["file size", deltas["file_size"]["a"], deltas["file_size"]["b"], deltas["file_size"]["delta"]],
            ],
        )
        lines += ["", f"* identical shape: {'yes' if image['same'] else 'no'}", ""]
        lines += _table(
            ["load delta", "count"],
            [[key, len(image["deltas"]["loads"][key])] for key in ("added", "removed", "changed")]
            + [["section " + key, len(image["deltas"]["sections"][key])]
               for key in ("added", "removed", "moved", "resized")],
        )
        lines += [""]

    symbols = report["sections"]["symbols"]
    lines += ["## 2. Symbol inventory", "", f"* status: {symbols['status']}"]
    if symbols.get("reason"):
        lines += [f"* reason: {symbols['reason']}"]
    if symbols["status"] == "ok":
        lines += _table(
            ["side", "symbols", "named", "anonymous", "sources"],
            [
                [side, symbols[side]["count"], symbols["totals"][f"named_{side}"],
                 symbols["totals"][f"anonymous_{side}"], ", ".join(symbols[side]["sources"]) or "-"]
                for side in ("a", "b")
            ],
        )
        lines += [""]
        lines += _table(
            ["address shift", "dominant", "matched named", "matched anon", "moved", "resized", "added", "removed"],
            [[
                _hex(symbols["shift"]), f"{symbols['shift_dominant'] * 100:.1f}%", symbols["matched_named"],
                symbols["matched_shifted"], symbols["totals"]["moved"], symbols["totals"]["resized_named"],
                symbols["totals"]["added"], symbols["totals"]["removed"],
            ]],
        )
        lines += [""]
        for key, title in (("moved", "Moved named"), ("resized", "Resized named"), ("added", "Added"),
                           ("removed", "Removed")):
            rows = symbols.get(key) or []
            lines += [f"**{title} (first {options.max_rows})**", ""]
            if not rows:
                lines += ["_none_", ""]
                continue
            if key in ("added", "removed"):
                lines += _table(["addr", "name", "size"], [[_hex(r["addr"]), f"`{r['name']}`", r["size"]] for r in rows])
            elif key == "moved":
                lines += _table(["name", "old addr", "new addr", "delta"],
                                [[f"`{r['name']}`", _hex(r["old_addr"]), _hex(r["new_addr"]), _hex(r["delta"])] for r in rows])
            else:
                lines += _table(["name", "old addr", "new addr", "old size", "new size"],
                                [[f"`{r['name']}`", _hex(r["old_addr"]), _hex(r["new_addr"]), r["old_size"], r["new_size"]]
                                 for r in rows])
            lines += [""]
    else:
        lines += [""]

    corpus = report["sections"]["corpus"]
    lines += ["## 3. Decompiled-change hints", "", f"* status: {corpus['status']}",
              f"* mode: {corpus.get('mode') or '-'}"]
    if corpus.get("reason"):
        lines += [f"* reason: {corpus['reason']}"]
    if corpus["status"] == "ok" and corpus.get("mode") == "corpus-diff":
        diff = corpus["diff"]
        lines += _table(
            ["functions A", "functions B", "shift", "added", "removed", "moved", "resized"],
            [[diff["totals"]["a"], diff["totals"]["b"], _hex(diff["shift"]), diff["totals"]["added"],
              diff["totals"]["removed"], diff["totals"]["moved"], diff["totals"]["resized_named"]]],
        )
    elif corpus["status"] == "ok" and corpus.get("mode") == "corpus-report":
        info = corpus["report"]
        lines += [f"* report: `{info['path']}` ({info['bytes']} bytes, sha256 `{info['sha256'][:16]}`)"]
    lines += [""]

    probe = report["sections"]["probe"]
    lines += ["## 4. Probe first-touch", "", f"* status: {probe['status']}"]
    if probe.get("reason"):
        lines += [f"* reason: {probe['reason']}"]
    if probe["status"] == "ok":
        lines += [f"* {probe['summary']}", ""]
        lines += _table(
            ["side", "stopped pc", "stopped symbol", "fault", "epc (fw vaddr)", "access", "access (fw vaddr)"],
            [
                [
                    side,
                    _hex(probe[side].get("stopped_pc")),
                    probe[side].get("stopped_symbol") or "-",
                    (probe[side].get("first_fault") or {}).get("desc") or "none",
                    _hex((probe[side].get("first_fault") or {}).get("epc_firmware")),
                    _hex(((probe[side].get("first_fault") or {}).get("access") or {}).get("access_addr")),
                    _hex((probe[side].get("first_fault") or {}).get("access_firmware")),
                ]
                for side in ("a", "b")
            ],
        )
    lines += [""]

    behavior = report["sections"]["behavior"]
    lines += ["## 5. Behavior", "", f"* status: {behavior['status']}"]
    if behavior.get("reason"):
        lines += [f"* reason: {behavior['reason']}"]
    if behavior["status"] == "ok":
        lines += _table(
            ["side", "session", "hooks", "uart banner", "faults", "events"],
            [
                [
                    side,
                    f"`{behavior[side].get('session')}`",
                    ", ".join(behavior[side].get("hooks") or []),
                    behavior[side].get("uart_banner") or "-",
                    behavior[side].get("fault_count"),
                    sum((behavior[side].get("event_counts") or {}).values()),
                ]
                for side in ("a", "b")
            ],
        )
        lines += [f"* zero deltas: {'yes' if behavior['all_zero'] else 'no'}"]
        if behavior.get("compare_report"):
            lines += [f"* compare report: `{behavior['compare_report']}`"]
        lines += [""]
        for name in ("bp", "watch", "writes"):
            rows = [row for row in behavior["deltas"][name] if row["delta"] != 0]
            rows.sort(key=lambda row: (-abs(row["delta"]), str(row["key"])))
            lines += [f"**{name} delta (nonzero, first {options.max_rows})**", ""]
            if rows:
                lines += _table(["key", "old", "new", "delta"],
                                [[f"`{row['key']}`", row["a"], row["b"], row["delta"]] for row in rows[: options.max_rows]])
            else:
                lines += ["_no deltas_"]
            lines += [""]
        lines += [f"> {NO_REFERENCE}", ""]
        lines += _table(
            ["#", "metric", "old observed", "new observed"],
            [[row["id"], row["name"], _short_json(row["a_observed"]), _short_json(row["b_observed"])]
             for row in behavior["metrics"]],
        )
    lines += [""]

    summary = report["summary"]
    lines += ["## Summary", ""]
    lines += [f"* ok: {', '.join(summary['sections_ok']) or '-'}"]
    if summary["sections_skipped"]:
        for name, reason in sorted(summary["sections_skipped"].items()):
            lines += [f"* skipped `{name}`: {reason}"]
    if summary["sections_error"]:
        for name, reason in sorted(summary["sections_error"].items()):
            lines += [f"* error `{name}`: {reason}"]
    return "\n".join(lines) + "\n"


def _short_json(value, limit: int = 72) -> str:
    import json

    if value is None:
        return "-"
    text = json.dumps(value, sort_keys=True)
    return f"`{text}`" if len(text) <= limit else f"`{text[: limit - 1]}...`"


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def _add_side_symbols(parser: argparse.ArgumentParser, side: str) -> None:
    parser.add_argument(f"--{side}-symbols", help="green-eye symbol DB JSON for the %s target" % side)
    parser.add_argument(f"--{side}-ghidra-names", help="Ghidra names.tsv for the %s target" % side)
    parser.add_argument(f"--{side}-ghidra-index", help="Ghidra decompiled_index.tsv for the %s target" % side)
    parser.add_argument(
        f"--{side}-ghidra-exports",
        help="Ghidra exports.tsv or 3-column ExportForGreenEye TSV for the %s target" % side,
    )
    parser.add_argument(f"--{side}-ghidra-bias", type=lambda value: int(value, 0), default=0,
                        help="bias added to %s Ghidra symbol addresses" % side)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="version_diff", description=__doc__.splitlines()[0])
    parser.add_argument("--old", required=True, help="old target (RISC-V ELF or raw .fwimage)")
    parser.add_argument("--new", required=True, help="new target (RISC-V ELF or raw .fwimage)")
    parser.add_argument("--out", required=True, help="output directory for the report, sessions and DBs")
    parser.add_argument("--old-label", help="report label for the old target (default: file stem)")
    parser.add_argument("--new-label", help="report label for the new target (default: file stem)")
    parser.add_argument("--release", help="release label; an rm610 release is gated by docs/23")
    parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})")
    _add_side_symbols(parser, "old")
    _add_side_symbols(parser, "new")
    parser.add_argument("--corpus-diff", nargs=2, metavar=("OLD_TSV", "NEW_TSV"),
                        help="diff two decompiled_index.tsv corpus exports")
    parser.add_argument("--corpus-report", help="attach a previously produced corpus diff report")
    parser.add_argument("--stimulus", help="green-eye script replacing the builtin demo scenario")
    parser.add_argument("--probe-instructions", type=int, default=2000,
                        help="probe step budget (0 = wall-clock --probe-duration)")
    parser.add_argument("--probe-duration", type=float, default=0.5,
                        help="probe wall budget when --probe-instructions=0")
    parser.add_argument("--behavior-duration", type=float, default=0.4,
                        help="budget for each behavior phase")
    parser.add_argument("--max-stops", type=int, default=5000, help="safety bound on serviced stops per phase")
    parser.add_argument("--max-rows", type=int, default=40, help="table row cap in reports")
    parser.add_argument("--log-bytes", type=int, default=MAX_LOG_BYTES, help="QEMU log window per fault check")
    return parser


def options_from_args(args: argparse.Namespace) -> VersionDiffOptions:
    corpus = tuple(args.corpus_diff) if args.corpus_diff else None
    return VersionDiffOptions(
        old=args.old,
        new=args.new,
        out=os.path.abspath(args.out),
        old_label=args.old_label,
        new_label=args.new_label,
        release=args.release,
        old_symbols=args.old_symbols,
        new_symbols=args.new_symbols,
        old_ghidra_names=args.old_ghidra_names,
        new_ghidra_names=args.new_ghidra_names,
        old_ghidra_index=args.old_ghidra_index,
        new_ghidra_index=args.new_ghidra_index,
        old_ghidra_exports=args.old_ghidra_exports,
        new_ghidra_exports=args.new_ghidra_exports,
        old_ghidra_bias=args.old_ghidra_bias,
        new_ghidra_bias=args.new_ghidra_bias,
        corpus_diff=corpus,  # type: ignore[arg-type]
        corpus_report=args.corpus_report,
        stimulus=args.stimulus,
        probe_instructions=args.probe_instructions,
        probe_duration=args.probe_duration,
        behavior_duration=args.behavior_duration,
        max_stops=args.max_stops,
        max_rows=args.max_rows,
        log_bytes=args.log_bytes,
        approval_token=args.approval_token,
    )


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    options = options_from_args(args)
    try:
        report = run_version_diff(options)
    except Rm610GateError as exc:
        print(f"[version-diff] RM610 gate refused: {exc}", file=sys.stderr)
        return 3
    except GreenEyeError as exc:
        print(f"[version-diff] error: {exc}", file=sys.stderr)
        return 2
    print(f"[version-diff] {report['old']['label']} -> {report['new']['label']}")
    for name in ("image", "symbols", "corpus", "probe", "behavior"):
        section = report["sections"][name]
        print(f"[version-diff] {name}: status={section['status']}"
              + (f" reason={section['reason']}" if section.get("reason") else ""))
    print(f"[version-diff] wrote {os.path.join(options.out, 'version-diff.json')}")
    print(f"[version-diff] wrote {os.path.join(options.out, 'version-diff.md')}")
    return 0


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())