#!/usr/bin/env python3
"""s07 - version diff: same stimulus against two firmware versions.

Real scenario (Handoff-vGPU-IO.md PART 91, 2026-10-01 cycles 2-3): the staged
570.211 image and a verified 570.144 image both failed ``GSP_INIT_DONE`` with
``rpc_result=0x59``; disproving the image-pairing hypothesis cost one full
supervised boot cycle per version, and docs/14 section 4.1 exists because
"the code changed" and "the behaviour moved" must be separated before the
emulation is touched.  Sorting a new image into the decision table
(static changed x hit/write profile changed) is the H4.1 triage.

Analog: byte-patch the demo's PING reply constant (0xbeef -> 0xbef0) to make a
labelled "new version", run the *same* recorded stimulus on both, and diff the
sessions and the decoded control block.  The answer is one live-path field.
"""
from __future__ import annotations

import os
import shutil

try:  # package import (tests) or direct script execution (run_all)
    from . import _common as C
except ImportError:
    import sys as _sys

    _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    import _common as C

#: addiw a1, a1, -0x111  (the 0xbeef PING key) -> -0x110 (0xbef0)
ADDW_OLD = bytes.fromhex("9b85f5ee")
ADDW_NEW = bytes.fromhex("9b8505ef")
STRING_OLD = b"hello-risc-v-gsp-rm-debugger"
STRING_NEW = b"hello-risc-v-gsp-rm-debuggef"


def _make_variant(src: str, dst: str) -> dict:
    with open(src, "rb") as fh:
        data = bytearray(fh.read())
    count = data.count(ADDW_OLD)
    if count == 1:
        offset = data.find(ADDW_OLD)
        old_bytes = bytes(data[offset : offset + 4])
        data[offset : offset + 4] = ADDW_NEW
        kind = "ping-key-0xbeef->0xbef0"
    else:
        count = data.count(STRING_OLD)
        if count != 1:
            raise C.DemoError(f"no unique patch site (addiw x{data.count(ADDW_OLD)}, string x{count})")
        offset = data.find(STRING_OLD)
        old_bytes = bytes(data[offset : offset + len(STRING_OLD)])
        data[offset : offset + len(STRING_OLD)] = STRING_NEW
        kind = "echo-string"
    with open(dst, "wb") as fh:
        fh.write(data)
    return {
        "kind": kind,
        "offset": offset,
        "old_bytes": old_bytes.hex(),
        "source_sha256": C.sha256_file(src),
        "variant_sha256": C.sha256_file(dst),
    }


def run() -> dict:
    import greeneye as ge
    from greeneye.analyze import compare_sessions

    backend = C.choose_backend("native")
    out = C.artifact_dir("s07_version_diff")
    va = os.path.join(out, "uproc_demo_vA.elf")
    vb = os.path.join(out, "uproc_demo_vB.elf")
    shutil.copyfile(C.DEMO_ELF, va)
    patch = _make_variant(C.DEMO_ELF, vb)

    commands = [
        "byte-patch PING key 0xbeef -> 0xbef0 to create version B",
        'for v in (A,B): s = ge.launch(v, backend="%s", record=...)' % backend,
        "identical stimulus: mailbox cmd=1 arg0=0x1234; snapshot+decode gsp_ctrl",
        "compare_sessions(A, B, symbols) + decoded field diff",
    ]
    scenario = C.scenario(
        ids=[
            "Handoff-vGPU-IO.md PART 91 (2026-10-01) cycles 2-3: 570.211 vs 570.144 pairing disproven",
            "PART 91 ADDENDUM 6 (2026-10-01): LOGRM id + fn72 identity changed behaviour between cycles",
            "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 4.1: version-diff decision table",
        ],
        baseline=(
            "one full supervised boot (~5-6 min) per version plus manual log "
            "comparison; the pairing hypothesis took cycles 2 and 3 back-to-back"
        ),
        projection=(
            "risc-v-gsp-rm-debugger cannot boot the real image without the H2 device model; "
            "the same same-stimulus/compare-sessions loop runs on re-hosted "
            "firmware and the produced symbol/session deltas are the H4.1 artifact"
        ),
        blocked=True,
        device="H2.2 device model for real .fwimage versions",
    )
    C.render_banner(
        "s07_version_diff",
        "On the same stimulus, which live-path behaviour moved between two firmware versions?",
        "570.211 vs 570.144 image pairing; semantic diff before touching the emulation",
        scenario["static_baseline"],
        commands,
    )

    schema = C.load_schema(ge)
    results = {}
    session_dirs = {}
    for label, path in (("A", va), ("B", vb)):
        session_dirs[label] = C.session_dir(out, f"session-{label}")
        session = ge.launch(kernel=path, backend=backend, record=session_dirs[label], name=f"vd-s07-{label}")
        try:
            ctrl = session.symbols.lookup("gsp_ctrl")
            dispatches = []
            session.hook(
                "rpc_dispatch",
                on_enter=lambda hit, d=dispatches: d.append({"cmd": hit.arg(0), "arg0": hit.arg(1)}),
            )
            session.watch(ctrl + C.CTRL_RESULT, size=8, kind="w")
            session.run(duration=0.25)
            session.memory.write_u64(ctrl + C.CTRL_ARG0, 0x1234)
            session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 1)
            session.run(duration=0.25)
            decoded = C.decode_ctrl(ge, session.snapshot(ctrl, C.CTRL_SIZE, f"ctrl-{label}"), schema)
            session.note("version", label=label, patch=patch["kind"], reply=decoded.get("result"))
            results[label] = {
                "path": os.path.relpath(path, C.TOOL_ROOT),
                "sha256": C.sha256_file(path),
                "dispatches": dispatches,
                "reply": decoded.get("result"),
                "rpc_count": decoded.get("rpc_count"),
                "last_rpc": decoded.get("last_rpc"),
            }
        finally:
            session.close()

    compare_md = compare_sessions(
        session_dirs["A"], session_dirs["B"], symbols=session.symbols, max_rows=20
    )
    C.write_text(os.path.join(out, "compare-sessions.md"), compare_md)
    moved = sorted(
        key
        for key in ("reply", "rpc_count", "last_rpc")
        if results["A"][key] != results["B"][key]
    )
    payload = {
        "success": results["A"]["reply"] == 0xACDB and results["B"]["reply"] == 0xACC4 and moved == ["reply"],
        "backend": backend,
        "patch": patch,
        "version_A": results["A"],
        "version_B": results["B"],
        "moved_fields": moved,
        "decision_table_row": {
            "static_code_changed": True,
            "live_profile_changed": bool(moved),
            "reading": "new logic on a live path - prioritise (docs/14 4.1)",
        },
        "compare_report": os.path.relpath(os.path.join(out, "compare-sessions.md"), C.TOOL_ROOT),
        "question": "separate 'the code changed' from 'the behaviour moved' on a fixed stimulus",
        "answer": (
            f"version A replies {results['A']['reply']:#x}, version B replies {results['B']['reply']:#x}; "
            f"moved fields {moved}; the patch site is {patch['kind']} at file offset {patch['offset']:#x} - "
            "the decision-table row is 'code changed + profile changed -> prioritise'"
        ),
        "success_criteria": {
            "pass": "exactly the patched reply field moves on the live path; both sessions replay from the same stimulus",
            "fail": "unrelated fields move or a version cannot be replayed",
        },
        "real_case": scenario,
    }
    C.write_json(os.path.join(out, "version-diff.json"), payload)
    return payload


def main() -> int:
    return 0 if C.timed_demo(run, C.artifact_dir("s07_version_diff"))["success"] else 1


if __name__ == "__main__":
    raise SystemExit(main())