path: root/debug/risc-v-gsp-rm-vm/tests/test_state_decode.py

"""Tests for the engine-visible state decoder (harness/state_decode.py).

Offline only: text/hexdump/binary/snapshot-JSON/aggregate-group parsing, the
bundled INSTANCE/RAMFC field table, map override/extension, decode/diff
reports and the CLI.  The real-evidence tests parse frozen live ladder traces
and an aggregate GSP physical-state dump, and skip when the backups are
absent.  No hardware, no module, no guest, no network.

Run::

    python3 -m unittest tests.test_state_decode -v
"""
from __future__ import annotations

import contextlib
import io
import json
import os
import struct
import sys
import tempfile
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
TOOL_ROOT = os.path.abspath(os.path.join(HERE, ".."))
if TOOL_ROOT not in sys.path:
    sys.path.insert(0, TOOL_ROOT)

from harness.state_decode import (  # noqa: E402
    AGGREGATE_FORMAT,
    DECODE_FORMAT,
    DEMO_FORMAT,
    DIFF_FORMAT,
    MAP_FORMAT,
    SNAPSHOT_FORMAT,
    StateDecodeError,
    decode_chunk,
    decode_specs,
    diff_reports,
    load_input,
    load_map,
    main,
    render_diff,
    run_demo,
)

EVIDENCE = os.environ.get("GREENEYE_STATE_DECODE_EVIDENCE") or os.path.normpath(
    os.path.join(TOOL_ROOT, "..", "..", "backups",
                 "evidence-20261002-victory", "06-ladder-m12.txt"))

# Our own evidence artifact directory (not a vendor path).  The aggregate
# dump is one group per line: ``stage: primed=1 kicked=1 tx_seq_num=106``.
AGGREGATE_EVIDENCE = os.environ.get(
    "GREENEYE_STATE_DECODE_EVIDENCE_AGG") or os.path.normpath(
    os.path.join(TOOL_ROOT, "..", "..", "backups",
                 "evidence-20261002-victory", "20-gsp-phys-state-m26.txt"))
AGGREGATE_EVIDENCE_PEER = os.environ.get(
    "GREENEYE_STATE_DECODE_EVIDENCE_AGG_PEER") or os.path.normpath(
    os.path.join(TOOL_ROOT, "..", "..", "backups",
                 "evidence-20261002-victory", "20-gsp-phys-state-m29.txt"))


def _run_main(argv: list[str]) -> tuple[int, str, str]:
    out, err = io.StringIO(), io.StringIO()
    with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
        rc = main(argv)
    return rc, out.getvalue(), err.getvalue()


def _write(path: str, text: str) -> str:
    with open(path, "w", encoding="utf-8") as handle:
        handle.write(text)
    return path


def _write_snapshot(tmp: str, name: str, chunks: list[dict]) -> str:
    path = os.path.join(tmp, name)
    with open(path, "w", encoding="utf-8") as handle:
        json.dump({"schema": SNAPSHOT_FORMAT, "source": name,
                   "chunks": chunks}, handle)
    return path


def _write_map(tmp: str, name: str, doc) -> str:
    path = os.path.join(tmp, name)
    with open(path, "w", encoding="utf-8") as handle:
        json.dump(doc, handle)
    return path


def _report_from_chunks(chunks: list[dict], map_path: str | None = None) -> dict:
    entries, source = load_map(map_path)
    return {
        "format": DECODE_FORMAT,
        "map_source": source,
        "sources": [],
        "chunks": [decode_chunk(chunk, entries) for chunk in chunks],
        "limits": [],
    }


def _chunk_field(report: dict, label: str, name: str) -> dict:
    for chunk in report["chunks"]:
        if chunk["label"] != label:
            continue
        for field in chunk["fields"]:
            if field["name"] == name:
                return field
    raise AssertionError(f"field {name} not found in chunk {label}")


def _spec(path: str, label: str | None = None, base=None,
          binary: bool = False) -> dict:
    return {"path": path, "label": label, "base": base, "binary": binary}


class ParsingTests(unittest.TestCase):
    def test_text_lines_parse_labels_offsets_and_values(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write(os.path.join(tmp, "lines.txt"), "\n".join([
                "banner line that must be ignored",
                "INSTANCE[0x10]=0xface",
                "mid-text noise",
                "Instance[00ac]=0x00030000",
                "INSTANCE[72]=0x1",      # decimal offset 72 == 0x48
                "INSTANCE[0x4c]=64206",  # decimal value 64206 == 0xface
                "",
            ]))
            document = load_input(_spec(path))
            self.assertEqual(document["format"], "text")
            self.assertEqual(len(document["chunks"]), 1)
            chunk = document["chunks"][0]
            self.assertEqual(chunk["label"], "INSTANCE")  # first spelling wins
            self.assertEqual(chunk["words"][0x10], 0xFACE)
            self.assertEqual(chunk["words"][0xAC], 0x00030000)
            self.assertEqual(chunk["words"][0x48], 0x1)
            self.assertEqual(chunk["words"][0x4C], 64206)

    def test_hexdump_rows_parse_with_label_and_base(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write(os.path.join(tmp, "dump.txt"), "\n".join([
                "00000000: 000002d2 00000000 00000000 00000000",
                "00000010: 0000face 00000000 00000000 00000000",
            ]))
            document = load_input(_spec(path, label="INSTANCE",
                                       base="0x1f4000000"))
            chunk = document["chunks"][0]
            self.assertEqual(chunk["label"], "INSTANCE")
            self.assertEqual(chunk["base"], 0x1F4000000)
            self.assertEqual(chunk["words"][0x0], 0x2D2)
            self.assertEqual(chunk["words"][0x4], 0x0)
            self.assertEqual(chunk["words"][0x10], 0xFACE)

    def test_raw_binary_parses_little_endian_words(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "ramfc.bin")
            with open(path, "wb") as handle:
                handle.write(struct.pack("<4I", 0x2D2, 0, 0x30000, 0))
            document = load_input(_spec(path, label="RAMFC", binary=True))
            self.assertEqual(document["format"], "binary")
            chunk = document["chunks"][0]
            self.assertEqual(chunk["label"], "RAMFC")
            self.assertEqual(chunk["words"][0x0], 0x2D2)
            self.assertEqual(chunk["words"][0x8], 0x30000)

    def test_raw_binary_is_auto_detected(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "auto.bin")
            with open(path, "wb") as handle:
                handle.write(struct.pack("<8I", 1, 2, 3, 4, 5, 6, 7, 8))
            document = load_input(_spec(path, label="WINDOW"))
            self.assertEqual(document["format"], "binary")
            chunk = document["chunks"][0]
            self.assertEqual(chunk["words"][0x1C], 8)

    def test_snapshot_json_parses_exactly(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write_snapshot(tmp, "snap.json", [{
                "label": "RAMFC", "base": "0x1000",
                "words": ["0x2d2", 0, "0x30000", "0x0"],
            }])
            document = load_input(_spec(path))
            self.assertEqual(document["format"], "snapshot-json")
            chunk = document["chunks"][0]
            self.assertEqual(chunk["label"], "RAMFC")
            self.assertEqual(chunk["base"], 0x1000)
            self.assertEqual(chunk["words"][0x0], 0x2D2)
            self.assertEqual(chunk["words"][0x8], 0x30000)

    def test_snapshot_schema_rejected(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "other.json")
            with open(path, "w", encoding="utf-8") as handle:
                json.dump({"schema": "something-else/1", "chunks": []}, handle)
            with self.assertRaises(StateDecodeError):
                load_input(_spec(path))

    def test_unparseable_text_rejected(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write(os.path.join(tmp, "nothing.txt"),
                          "just prose\nwith no fields at all\n")
            with self.assertRaises(StateDecodeError):
                load_input(_spec(path))

    def test_bad_binary_size_rejected(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "odd.bin")
            with open(path, "wb") as handle:
                handle.write(b"\x01\x02\x03")
            with self.assertRaises(StateDecodeError):
                load_input(_spec(path, binary=True))


class AggregateParsingTests(unittest.TestCase):
    def _decode(self, tmp: str, name: str, text: str) -> dict:
        entries, source = load_map(None)
        path = _write(os.path.join(tmp, name), text)
        return decode_specs([_spec(path)], entries, source)

    def test_aggregate_line_decodes_types_and_preserves_raw(self):
        text = "\n".join([
            "stage: primed=1 kicked=1 tx_seq_num=106 variant=V6",
            "ctx: pa=0x1f4000000 size=0x133000 last_ret=-5 "
            "userd=sysmem (host-writable)",
        ])
        with tempfile.TemporaryDirectory() as tmp:
            path = _write(os.path.join(tmp, "agg.txt"), text)
            document = load_input(_spec(path))
            self.assertEqual(document["format"], AGGREGATE_FORMAT)
            self.assertEqual([chunk["label"] for chunk in document["chunks"]],
                             ["PHYS_stage", "PHYS_ctx"])
            stage = document["chunks"][0]
            self.assertEqual(stage["format"], AGGREGATE_FORMAT)
            self.assertIn("stage: primed=1", stage["raw"])
            entries, source = load_map(None)
            report = decode_specs([_spec(path)], entries, source)
            chunks = {chunk["label"]: chunk for chunk in report["chunks"]}
            stage = chunks["PHYS_stage"]
            self.assertEqual(stage["input_format"], AGGREGATE_FORMAT)
            self.assertEqual(stage["word_count"], 0)
            self.assertEqual(stage["kv_count"], 4)
            self.assertEqual(stage["raw"],
                             "stage: primed=1 kicked=1 tx_seq_num=106 "
                             "variant=V6")
            fields = {field["name"]: field for field in stage["fields"]}
            self.assertEqual(fields["primed"]["value"], 1)
            self.assertTrue(fields["primed"]["numeric"])
            self.assertEqual(fields["tx_seq_num"]["value"], 106)
            self.assertEqual(fields["variant"]["value"], "V6")
            self.assertFalse(fields["variant"]["numeric"])
            self.assertEqual(fields["variant"]["raw"], "V6")
            self.assertEqual(fields["variant"]["kind"], "kv")
            self.assertIsNone(fields["variant"]["offset"])
            ctx = {field["name"]: field for field in chunks["PHYS_ctx"]["fields"]}
            self.assertEqual(ctx["pa"]["value"], 0x1F4000000)
            self.assertEqual(ctx["size"]["value"], 0x133000)
            self.assertEqual(ctx["last_ret"]["value"], -5)
            self.assertEqual(ctx["userd"]["value"], "sysmem")
            self.assertNotIn("host-writable", ctx)

    def test_aggregate_missing_value_and_unparsed_lines(self):
        with tempfile.TemporaryDirectory() as tmp:
            report = self._decode(tmp, "agg.txt", "\n".join([
                "banner line that must be ignored",
                "status_ring: writePtr slot=29; last produced elements "
                "(max 8):",
                "peer: host status readPtr @cmd+0x20 = 8",
                "empty: a= b=1",
            ]))
            self.assertEqual([chunk["label"] for chunk in report["chunks"]],
                             ["PHYS_empty"])
            fields = {field["name"]: field for field in
                      report["chunks"][0]["fields"]}
            self.assertEqual(fields["a"]["value"], "")
            self.assertEqual(fields["a"]["raw"], "")
            self.assertFalse(fields["a"]["numeric"])
            self.assertEqual(fields["b"]["value"], 1)

    def test_aggregate_single_kv_group_is_captured(self):
        with tempfile.TemporaryDirectory() as tmp:
            report = self._decode(tmp, "agg.txt", "guest_ram: active=0\n")
            self.assertEqual([chunk["label"] for chunk in report["chunks"]],
                             ["PHYS_guest_ram"])
            field = report["chunks"][0]["fields"][0]
            self.assertEqual(field["name"], "active")
            self.assertEqual(field["value"], 0)

    def test_aggregate_pipe_groups_and_duplicate_keys(self):
        with tempfile.TemporaryDirectory() as tmp:
            report = self._decode(tmp, "agg.txt", "\n".join([
                "rm_call_userd: userd=sysmem (host-writable) | "
                "rm_call_pdb: vaBase=0x100000 vaGpfifo=0x200000",
                "queue_dma: shared=0x1a86b6000 ptes=129 | "
                "radix3=0x105370000 fmc=0x142d25000",
                "guest_ram: active=1 gpa=0x0 gpa=0x100000000 "
                "size=0x80000000 size=0x80000000",
            ]))
            self.assertEqual([chunk["label"] for chunk in report["chunks"]], [
                "PHYS_guest_ram", "PHYS_queue_dma", "PHYS_rm_call_pdb",
                "PHYS_rm_call_userd",
            ])
            chunks = {chunk["label"]: chunk for chunk in report["chunks"]}
            userd = {field["name"]: field for field in
                     chunks["PHYS_rm_call_userd"]["fields"]}
            self.assertEqual(userd["userd"]["value"], "sysmem")
            pdb = [field["name"] for field in chunks["PHYS_rm_call_pdb"]["fields"]]
            self.assertEqual(pdb, ["vaBase", "vaGpfifo"])
            queue = [field["name"] for field in chunks["PHYS_queue_dma"]["fields"]]
            self.assertEqual(queue, ["shared", "ptes", "radix3", "fmc"])
            gpa = [field for field in chunks["PHYS_guest_ram"]["fields"]
                   if field["name"] == "gpa"]
            self.assertEqual([field["index"] for field in gpa], [0, 1])
            self.assertEqual([field["value"] for field in gpa],
                             [0, 0x100000000])

    def test_mixed_word_and_aggregate_lines(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write(os.path.join(tmp, "mixed.txt"), "\n".join([
                "INSTANCE[0x10]=0xface",
                "stage: a=1 b=2",
            ]))
            document = load_input(_spec(path))
            self.assertEqual(document["format"], "text")
            self.assertEqual([chunk["label"] for chunk in document["chunks"]],
                             ["INSTANCE", "PHYS_stage"])

    def test_unstructured_text_still_rejected(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write(os.path.join(tmp, "none.txt"), "\n".join([
                "status_ring: writePtr slot=29; last produced elements:",
                "peer: host status readPtr @cmd+0x20 = 8",
            ]))
            with self.assertRaises(StateDecodeError):
                load_input(_spec(path))

    def test_aggregate_decode_is_deterministic(self):
        with tempfile.TemporaryDirectory() as tmp:
            text = "stage: a=1 b=0x10 c=up\n"
            first = self._decode(tmp, "one.txt", text)
            second = self._decode(tmp, "two.txt", text)
            stripped = [
                {key: value for key, value in chunk.items()
                 if key != "source"}
                for chunk in first["chunks"]
            ]
            other = [
                {key: value for key, value in chunk.items()
                 if key != "source"}
                for chunk in second["chunks"]
            ]
            self.assertEqual(json.dumps(stripped, sort_keys=True),
                             json.dumps(other, sort_keys=True))


class DecodeTests(unittest.TestCase):
    def _ramfc_target(self, word48: int) -> dict:
        report = _report_from_chunks([
            {"label": "RAMFC", "base": 0, "words": {0x48: word48}},
        ])
        return _chunk_field(report, "RAMFC", "RAMFC_TARGET")

    def test_ramfc_target_both_bits_set(self):
        field = self._ramfc_target(0x00030000)
        self.assertEqual(field["offset"], "0x48")
        self.assertEqual(field["raw"], "0x00030000")
        self.assertEqual(field["kind"], "bits")
        bits = {bit["name"]: bit for bit in field["bitfields"]}
        self.assertEqual(bits["ENG_CTX_VALID"]["value"], 1)
        self.assertEqual(bits["ENG_CTX_VALID"]["state"], "set")
        self.assertEqual(bits["CE_CTX_VALID"]["value"], 1)
        self.assertEqual(bits["CE_CTX_VALID"]["state"], "set")

    def test_ramfc_target_only_eng_bit16(self):
        field = self._ramfc_target(0x00010000)
        bits = {bit["name"]: bit for bit in field["bitfields"]}
        self.assertEqual(bits["ENG_CTX_VALID"]["state"], "set")
        self.assertEqual(bits["CE_CTX_VALID"]["state"], "clear")

    def test_ramfc_target_only_ce_bit17(self):
        field = self._ramfc_target(0x00020000)
        bits = {bit["name"]: bit for bit in field["bitfields"]}
        self.assertEqual(bits["ENG_CTX_VALID"]["state"], "clear")
        self.assertEqual(bits["CE_CTX_VALID"]["state"], "set")

    def test_instance_fields_and_unmapped_words(self):
        report = _report_from_chunks([{
            "label": "INSTANCE", "base": 0,
            "words": {0x0: 0, 0x10: 0xFACE, 0x48: 0x100000, 0x4C: 0x30000},
        }])
        chunk = report["chunks"][0]
        names = {field["name"] for field in chunk["fields"]}
        self.assertEqual(names, {"MAGIC_0FACE", "GPFIFO_VA"})
        self.assertEqual(chunk["word_count"], 4)
        self.assertEqual(chunk["unmapped_count"], 2)
        self.assertEqual({word["offset"] for word in chunk["unmapped"]},
                         {"0x0", "0x4c"})
        self.assertEqual(chunk["words"]["0x48"], "0x00100000")

    def test_decode_report_format_and_determinism(self):
        report = _report_from_chunks([{
            "label": "INSTANCE", "base": 0, "words": {0x48: 0x100000},
        }])
        self.assertEqual(report["format"], DECODE_FORMAT)
        again = _report_from_chunks([{
            "label": "INSTANCE", "base": 0, "words": {0x48: 0x100000},
        }])
        self.assertEqual(json.dumps(report, sort_keys=True),
                         json.dumps(again, sort_keys=True))


class MapTests(unittest.TestCase):
    def test_override_replaces_bundled_and_extends(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write_map(tmp, "map.json", {
                "format": MAP_FORMAT,
                "entries": [
                    {"chunk": "INSTANCE", "label": "GPFIFO_VA_ALT",
                     "offset": "0x48", "width": 4, "note": "override"},
                    {"chunk": "INSTANCE", "label": "CTX_MARK",
                     "offset": "0x4c", "width": 4,
                     "bits": [
                         {"name": "ENG", "lo": 16, "hi": 16},
                         {"name": "CE", "lo": 17, "hi": 17},
                         {"name": "MODE", "lo": 0, "hi": 7,
                          "values": {"0": "zero", "1": "one"}},
                     ]},
                ],
            })
            report = _report_from_chunks([{
                "label": "INSTANCE", "base": 0,
                "words": {0x10: 0xFACE, 0x48: 0x100000, 0x4C: 0x00030001},
            }], map_path=path)
            chunk = report["chunks"][0]
            names = [field["name"] for field in chunk["fields"]]
            self.assertIn("GPFIFO_VA_ALT", names)
            self.assertNotIn("GPFIFO_VA", names)
            self.assertIn("MAGIC_0FACE", names)  # bundled table still extends
            mark = _chunk_field(report, "INSTANCE", "CTX_MARK")
            bits = {bit["name"]: bit for bit in mark["bitfields"]}
            self.assertEqual(bits["MODE"]["state"], "one")
            self.assertEqual(bits["ENG"]["state"], "set")
            self.assertEqual(bits["CE"]["state"], "set")
            self.assertEqual(report["map_source"], os.path.abspath(path))

    def test_replace_mode_drops_bundled_table(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write_map(tmp, "replace.json", {
                "format": MAP_FORMAT,
                "mode": "replace",
                "entries": [
                    {"chunk": "RAMFC", "label": "TARGET_ALT",
                     "offset": "0x48", "width": 4,
                     "bits": [{"name": "VALID", "lo": 16, "hi": 16}]},
                ],
            })
            report = _report_from_chunks([{
                "label": "INSTANCE", "base": 0, "words": {0x10: 0xFACE},
            }, {
                "label": "RAMFC", "base": 0, "words": {0x48: 0x10000},
            }], map_path=path)
            self.assertEqual(report["chunks"][0]["fields"], [])
            field = _chunk_field(report, "RAMFC", "TARGET_ALT")
            self.assertEqual(field["bitfields"][0]["state"], "set")

    def test_bad_map_rejected(self):
        cases = [
            {"entries": [{"chunk": "X", "label": "Y", "offset": "0x48",
                          "width": 3}]},
            {"entries": [{"chunk": "X", "label": "Y", "offset": "0x48",
                          "width": 4,
                          "bits": [{"name": "B", "lo": 0, "hi": 99}]}]},
            {"entries": [{"chunk": "X", "label": "Y", "offset": "0x0",
                          "width": 4}], "mode": "sideways"},
        ]
        for doc in cases:
            with self.subTest(doc=doc):
                with tempfile.TemporaryDirectory() as tmp:
                    path = _write_map(tmp, "bad.json", doc)
                    with self.assertRaises(StateDecodeError):
                        load_map(path)


class DiffTests(unittest.TestCase):
    def test_diff_detects_changed_word_and_named_field(self):
        left = _report_from_chunks([{
            "label": "INSTANCE", "base": 0,
            "words": {0x10: 0xFACE, 0x48: 0x100000},
        }])
        right = _report_from_chunks([{
            "label": "INSTANCE", "base": 0,
            "words": {0x10: 0xFACE, 0x48: 0x200000},
        }])
        report = diff_reports(left, right)
        self.assertEqual(report["format"], DIFF_FORMAT)
        self.assertFalse(report["identical"])
        chunk = report["chunks"][0]
        self.assertEqual(chunk["status"], "changed")
        self.assertEqual(chunk["changed"],
                         [{"offset": "0x48", "old": "0x00100000",
                           "new": "0x00200000"}])
        self.assertEqual(len(chunk["changed_fields"]), 1)
        field = chunk["changed_fields"][0]
        self.assertEqual(field["name"], "GPFIFO_VA")
        self.assertEqual(field["old"], "0x00100000")
        self.assertEqual(field["new"], "0x00200000")
        self.assertEqual(report["summary"]["changed_words"], 1)
        self.assertEqual(report["summary"]["changed_fields"], 1)

    def test_diff_reports_bitfield_state_change(self):
        left = _report_from_chunks([
            {"label": "RAMFC", "base": 0, "words": {0x48: 0x00030000}}])
        right = _report_from_chunks([
            {"label": "RAMFC", "base": 0, "words": {0x48: 0x00010000}}])
        report = diff_reports(left, right)
        field = report["chunks"][0]["changed_fields"][0]
        self.assertEqual(field["name"], "RAMFC_TARGET")
        bits = {bit["name"]: bit for bit in field["bits"]}
        self.assertEqual(bits["ENG_CTX_VALID"]["old"], "set")
        self.assertEqual(bits["ENG_CTX_VALID"]["new"], "set")
        self.assertEqual(bits["CE_CTX_VALID"]["old"], "set")
        self.assertEqual(bits["CE_CTX_VALID"]["new"], "clear")

    def test_diff_identical_is_empty_and_deterministic(self):
        left = _report_from_chunks([{
            "label": "INSTANCE", "base": 0, "words": {0x10: 0xFACE},
        }])
        right = _report_from_chunks([{
            "label": "INSTANCE", "base": 0, "words": {0x10: 0xFACE},
        }])
        report = diff_reports(left, right)
        self.assertTrue(report["identical"])
        self.assertEqual(report["summary"]["changed_words"], 0)
        again = diff_reports(left, right)
        self.assertEqual(json.dumps(report, sort_keys=True),
                         json.dumps(again, sort_keys=True))

    def test_diff_added_and_removed_words(self):
        left = _report_from_chunks([{
            "label": "INSTANCE", "base": 0, "words": {0x0: 1, 0x4: 2},
        }])
        right = _report_from_chunks([{
            "label": "INSTANCE", "base": 0, "words": {0x0: 1, 0x8: 3},
        }])
        report = diff_reports(left, right)
        chunk = report["chunks"][0]
        self.assertEqual([row["offset"] for row in chunk["removed"]], ["0x4"])
        self.assertEqual([row["offset"] for row in chunk["added"]], ["0x8"])
        self.assertEqual(report["summary"]["added"], 1)
        self.assertEqual(report["summary"]["removed"], 1)


class AggregateDiffTests(unittest.TestCase):
    def _pair(self, tmp: str, left_text: str, right_text: str) -> tuple:
        entries, source = load_map(None)
        left = _write(os.path.join(tmp, "left.txt"), left_text)
        right = _write(os.path.join(tmp, "right.txt"), right_text)
        return (
            decode_specs([_spec(left)], entries, source),
            decode_specs([_spec(right)], entries, source),
        )

    def test_diff_classifies_added_removed_changed_kvs(self):
        with tempfile.TemporaryDirectory() as tmp:
            left, right = self._pair(
                tmp,
                "stage: a=1 b=2 gone=1 variant=V6\n",
                "stage: a=1 b=3 variant=V7 extra=4\n")
            report = diff_reports(left, right)
            self.assertEqual(report["format"], DIFF_FORMAT)
            self.assertFalse(report["identical"])
            chunk = report["chunks"][0]
            self.assertEqual(chunk["label"], "PHYS_stage")
            self.assertEqual(chunk["status"], "changed")
            self.assertEqual(chunk["added"], [])
            self.assertEqual(chunk["removed"], [])
            self.assertEqual(chunk["changed"], [])
            rows = {row["name"]: row for row in chunk["changed_fields"]}
            self.assertEqual(rows["b"]["status"], "changed")
            self.assertEqual((rows["b"]["old"], rows["b"]["new"]), (2, 3))
            self.assertEqual(rows["variant"]["status"], "changed")
            self.assertEqual((rows["variant"]["old"], rows["variant"]["new"]),
                             ("V6", "V7"))
            self.assertEqual(rows["extra"]["status"], "added")
            self.assertIsNone(rows["extra"]["old"])
            self.assertEqual(rows["extra"]["new"], 4)
            self.assertEqual(rows["gone"]["status"], "removed")
            self.assertEqual(rows["gone"]["old"], 1)
            self.assertIsNone(rows["gone"]["new"])
            self.assertEqual([row["name"] for row in chunk["fields_added"]],
                             ["extra"])
            self.assertEqual([row["name"] for row in chunk["fields_removed"]],
                             ["gone"])
            self.assertEqual([row["name"] for row in chunk["fields_changed"]],
                             ["b", "variant"])
            self.assertEqual(report["summary"]["fields_added"], 1)
            self.assertEqual(report["summary"]["fields_removed"], 1)
            self.assertEqual(report["summary"]["fields_changed"], 2)
            self.assertEqual(report["summary"]["changed_fields"], 4)
            again = diff_reports(left, right)
            self.assertEqual(json.dumps(report, sort_keys=True),
                             json.dumps(again, sort_keys=True))

    def test_diff_aggregate_identical_and_one_sided(self):
        with tempfile.TemporaryDirectory() as tmp:
            left, right = self._pair(tmp, "stage: a=1 b=up\n",
                                     "stage: a=1 b=up\n")
            report = diff_reports(left, right)
            self.assertTrue(report["identical"])
            self.assertEqual(report["summary"]["fields_added"], 0)
            self.assertEqual(report["summary"]["fields_removed"], 0)
            self.assertEqual(report["summary"]["fields_changed"], 0)
            only_left, only_right = self._pair(
                tmp, "stage: a=1 b=2\n", "ctx: x=1\n")
            report = diff_reports(only_left, only_right)
            labels = {chunk["label"]: chunk for chunk in report["chunks"]}
            self.assertEqual(labels["PHYS_stage"]["status"], "left_only")
            self.assertEqual(
                [row["name"] for row in labels["PHYS_stage"]["fields_removed"]],
                ["a", "b"])
            self.assertEqual(labels["PHYS_ctx"]["status"], "right_only")
            self.assertEqual(
                [row["name"] for row in labels["PHYS_ctx"]["fields_added"]],
                ["x"])

    def test_diff_aggregate_duplicate_key_occurrence(self):
        with tempfile.TemporaryDirectory() as tmp:
            left, right = self._pair(
                tmp, "mem: size=0x10 size=0x20\n",
                "mem: size=0x10 size=0x30\n")
            chunk = diff_reports(left, right)["chunks"][0]
            self.assertEqual(len(chunk["fields_changed"]), 1)
            row = chunk["fields_changed"][0]
            self.assertEqual(row["name"], "size")
            self.assertEqual(row["index"], 1)
            self.assertEqual((row["old"], row["new"]), (0x20, 0x30))

    def test_cli_diff_aggregate_exit_codes_and_kv_rendering(self):
        with tempfile.TemporaryDirectory() as tmp:
            left = _write(os.path.join(tmp, "left.txt"),
                          "stage: a=1 b=2 state=up\n")
            right = _write(os.path.join(tmp, "right.txt"),
                           "stage: a=1 b=3 state=up\n")
            rc, out, err = _run_main(["diff", left, right])
            self.assertEqual(rc, 3, err)
            self.assertIn("PHYS_stage", out)
            self.assertIn("kv b changed: 2 -> 3", out)
            rc, out, err = _run_main(["diff", left, left])
            self.assertEqual(rc, 0, err)
            self.assertIn("identical=True", out)

    def test_render_diff_aggregate_handles_added_and_removed(self):
        with tempfile.TemporaryDirectory() as tmp:
            left, right = self._pair(tmp, "stage: gone=1\n",
                                     "stage: extra=2\n")
            text = render_diff(diff_reports(left, right))
            self.assertIn("kv extra added: - -> 2", text)
            self.assertIn("kv gone removed: 1 -> -", text)

    def test_cli_decode_aggregate_writes_report(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write(os.path.join(tmp, "agg.txt"),
                          "stage: primed=1 variant=V6\n")
            out_path = os.path.join(tmp, "agg.json")
            rc, out, err = _run_main(["decode", "--input", path,
                                      "--json", out_path])
            self.assertEqual(rc, 0, err)
            self.assertIn("PHYS_stage", out)
            self.assertIn("(aggregate)", out)
            with open(out_path, encoding="utf-8") as handle:
                report = json.load(handle)
            self.assertEqual(report["format"], DECODE_FORMAT)
            self.assertEqual(report["sources"][0]["format"], AGGREGATE_FORMAT)
            self.assertEqual(report["chunks"][0]["input_format"],
                             AGGREGATE_FORMAT)
            self.assertEqual(report["chunks"][0]["fields"][1]["value"], "V6")


class CliAndDemoTests(unittest.TestCase):
    def test_cli_decode_prints_table_and_writes_json(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write(os.path.join(tmp, "dump.txt"), "\n".join([
                "00000000: 00000000 00000000 00000000 00000000",
                "00000010: 0000face 00000000 00000000 00000000",
                "00000020: 00000000 00000000 00000000 00000000",
                "00000030: 00000000 00000000 00000000 00000000",
                "00000040: 00000000 00000000 00100000 00030000",
            ]))
            out_path = os.path.join(tmp, "decoded.json")
            rc, out, err = _run_main(["decode", "--input", path,
                                      "--label", "INSTANCE",
                                      "--json", out_path])
            self.assertEqual(rc, 0, err)
            self.assertIn("GPFIFO_VA", out)
            self.assertIn("0x00100000", out)
            with open(out_path, encoding="utf-8") as handle:
                report = json.load(handle)
            self.assertEqual(report["format"], DECODE_FORMAT)
            self.assertEqual(len(report["chunks"]), 1)
            self.assertEqual(_chunk_field(report, "INSTANCE",
                                         "GPFIFO_VA")["raw"], "0x00100000")

    def test_cli_multiple_inputs_with_per_input_labels(self):
        with tempfile.TemporaryDirectory() as tmp:
            left = _write(os.path.join(tmp, "left.txt"),
                          "00000000: 00000001 00000000 00000000 00000000\n")
            right = _write(os.path.join(tmp, "right.txt"),
                           "00000000: 00000002 00000000 00000000 00000000\n")
            rc, out, err = _run_main([
                "decode", "--input", left, "--label", "LEFT",
                "--input", right, "--label", "RIGHT"])
            self.assertEqual(rc, 0, err)
            self.assertIn("chunk LEFT", out)
            self.assertIn("chunk RIGHT", out)

    def test_cli_decode_text_lines_reports_canonical_words(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write(os.path.join(tmp, "lines.txt"),
                          "INSTANCE[0x48]=0x100000\n")
            out_path = os.path.join(tmp, "decoded.json")
            rc, _, err = _run_main(["decode", "--input", path,
                                    "--json", out_path])
            self.assertEqual(rc, 0, err)
            with open(out_path, encoding="utf-8") as handle:
                report = json.load(handle)
            chunk = report["chunks"][0]
            self.assertEqual(chunk["label"], "INSTANCE")
            self.assertEqual(chunk["words"]["0x48"], "0x00100000")
            self.assertEqual(chunk["unmapped_count"], 0)

    def test_cli_diff_exit_codes(self):
        with tempfile.TemporaryDirectory() as tmp:
            left = _write_snapshot(tmp, "left.json", [{
                "label": "RAMFC", "base": "0x0",
                "words": ["0x0", "0x0", "0x0", "0x0", "0x0", "0x0", "0x0",
                          "0x0", "0x0", "0x0", "0x0", "0x0", "0x0", "0x0",
                          "0x0", "0x0", "0x0", "0x0", "0x30000", "0x0"],
            }])
            right = _write_snapshot(tmp, "right.json", [{
                "label": "RAMFC", "base": "0x0",
                "words": ["0x0", "0x0", "0x0", "0x0", "0x0", "0x0", "0x0",
                          "0x0", "0x0", "0x0", "0x0", "0x0", "0x0", "0x0",
                          "0x0", "0x0", "0x0", "0x0", "0x10000", "0x0"],
            }])
            rc, out, err = _run_main(["diff", left, right])
            self.assertEqual(rc, 3, err)
            self.assertIn("CE_CTX_VALID: set -> clear", out)
            rc, out, err = _run_main(["diff", left, left])
            self.assertEqual(rc, 0, err)
            self.assertIn("identical=True", out)

    def test_cli_diff_accepts_decoded_reports(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write_snapshot(tmp, "snap.json", [{
                "label": "INSTANCE", "base": "0x0",
                "words": ["0xface", "0x0", "0x0", "0x0"],
            }])
            report_path = os.path.join(tmp, "decoded.json")
            rc, _, err = _run_main(["decode", "--input", path,
                                    "--json", report_path])
            self.assertEqual(rc, 0, err)
            rc, out, err = _run_main(["diff", report_path, report_path])
            self.assertEqual(rc, 0, err)
            self.assertIn("identical=True", out)

    def test_cli_demo_green(self):
        rc, out, err = _run_main(["demo"])
        self.assertEqual(rc, 0, err)
        self.assertIn("ok=True", out)
        self.assertIn("RAMFC_TARGET", out)
        self.assertIn("CE_CTX_VALID: set -> clear", out)

    def test_demo_api_is_self_contained_and_deterministic(self):
        first = run_demo()
        second = run_demo()
        self.assertEqual(first["format"], DEMO_FORMAT)
        self.assertTrue(first["ok"])
        self.assertEqual(first["diff"]["summary"]["changed_words"], 2)
        self.assertEqual(first["diff"]["summary"]["changed_fields"], 2)
        self.assertEqual(json.dumps(first["diff"], sort_keys=True),
                         json.dumps(second["diff"], sort_keys=True))


@unittest.skipUnless(os.path.isfile(EVIDENCE), "ladder evidence absent")
class EvidenceTests(unittest.TestCase):
    def test_ladder_trace_decodes_instance_fields(self):
        entries, source = load_map(None)
        report = decode_specs([_spec(EVIDENCE)], entries, source)
        self.assertGreaterEqual(len(report["chunks"]), 1)
        chunks = {chunk["label"]: chunk for chunk in report["chunks"]}
        self.assertIn("INSTANCE", chunks)
        instance = chunks["INSTANCE"]
        self.assertGreater(instance["word_count"], 0)
        fields = {field["name"]: field for field in instance["fields"]}
        self.assertEqual(fields["GPFIFO_VA"]["raw"], "0x00100000")
        self.assertEqual(fields["MAGIC_0FACE"]["raw"], "0x0000face")


@unittest.skipUnless(os.path.isfile(AGGREGATE_EVIDENCE),
                     "aggregate GSP physical-state evidence absent")
class AggregateEvidenceTests(unittest.TestCase):
    """Frozen GSP physical-state dumps from our own evidence directory.

    Override with GREENEYE_STATE_DECODE_EVIDENCE_AGG (and ..._PEER); skipped
    when the files are absent.  No hardware, no module, no guest, no network.
    """

    def test_aggregate_evidence_decodes_groups(self):
        entries, source = load_map(None)
        report = decode_specs([_spec(AGGREGATE_EVIDENCE)], entries, source)
        self.assertGreater(len(report["chunks"]), 0)
        self.assertEqual(report["sources"][0]["format"], AGGREGATE_FORMAT)
        chunks = {chunk["label"]: chunk for chunk in report["chunks"]}
        for label in ("PHYS_stage", "PHYS_rm_call", "PHYS_guest_ram",
                      "PHYS_rm_call_ctx"):
            self.assertIn(label, chunks)
        stage = {field["name"]: field for field in
                 chunks["PHYS_stage"]["fields"]}
        self.assertEqual(stage["primed"]["value"], 1)
        self.assertTrue(stage["tx_seq_num"]["numeric"])

    def test_aggregate_evidence_diff_peer_shows_changed_kvs(self):
        if not os.path.isfile(AGGREGATE_EVIDENCE_PEER):
            self.skipTest("aggregate peer evidence absent")
        entries, source = load_map(None)
        left = decode_specs([_spec(AGGREGATE_EVIDENCE)], entries, source)
        right = decode_specs([_spec(AGGREGATE_EVIDENCE_PEER)], entries, source)
        report = diff_reports(left, right)
        self.assertFalse(report["identical"])
        self.assertGreater(report["summary"]["fields_changed"], 0)
        self.assertEqual(report["summary"]["changed_words"], 0)
        chunks = {chunk["label"]: chunk for chunk in report["chunks"]}
        rows = {row["name"]: row for row in
                chunks["PHYS_guest_ram"]["changed_fields"]}
        self.assertEqual(rows["active"]["status"], "changed")
        self.assertEqual((rows["active"]["old"], rows["active"]["new"]),
                         (1, 0))
        render_diff(report)


if __name__ == "__main__":
    unittest.main()