#!/usr/bin/env python3
"""Run/validate a small MN103S firmware CPU smoke proof for GDR-8050L READ LBA ingestion.

This is intentionally offline.  It does not change the live iSCSI target.  It
reuses the current package config's packet-session model and asks mn103s_emu.py
to run the known small persistent command sequence until the firmware READ(10)
path assembles the host LBA into SRAM 0x05EC.  The result is a current-version
report that can travel with the live bridge artifacts.
"""
from __future__ import annotations

import argparse
import csv
import json
import os
import re
import subprocess
import sys
from copy import deepcopy
from pathlib import Path
from typing import Any

VERSION = "v227"
EXPECTED_HOST_LBA = 0x0004ABBF
EXPECTED_HOST_LBA_HEX = "0x0004ABBF"
EXPECTED_05EC_LE = [0xBF, 0xAB, 0x04, 0x00]
READ_HANDLER_PCS = {"0x9001E3E0", "0x9001E4E8", "0x9001E500"}
LBA_BYTE_PCS = {"0x9001E527", "0x9001E52D", "0x9001E533", "0x9001E539"}
ASSEMBLED_PC = "0x9001E541"


def load_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as f:
        return json.load(f)


def write_json(path: Path, obj: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8") as f:
        json.dump(obj, f, indent=2)
        f.write("\n")


def parse_int(v: Any, default: int = 0) -> int:
    if v is None or v == "":
        return default
    if isinstance(v, int):
        return v
    return int(str(v), 0)


def norm_hex(v: Any) -> str:
    try:
        return f"0x{parse_int(v):08X}"
    except Exception:
        return str(v or "")


def rel_or_abs(path_text: str, base_dir: Path) -> Path:
    p = Path(path_text)
    if p.is_absolute():
        return p
    return (base_dir / p).resolve()


EXPECTED_GDR8050L_0L23_SHA256 = "7cab0f5cd77533256bd902df254bdf1690c41fb5def859d32b1270fbc555d698"
EXPECTED_GDR8050L_0L23_BASENAME = "GDR-8050L_0L23_5af3ae9b_firmware_4c61eb12_plaintext_7cab0f5c.bin"


def _path_candidates_for_text(text: str, config_path: Path, workspace: Path) -> list[Path]:
    text = str(text or "")
    if not text:
        return []
    normalized = text.replace("\\", "/")
    basename = normalized.rsplit("/", 1)[-1]
    out: list[Path] = []
    raw = Path(text)
    out.append(raw)
    out.append(config_path.parent / text.replace("\\", os.sep).replace("/", os.sep))
    out.append(workspace / text.replace("\\", os.sep).replace("/", os.sep))
    for root in [config_path.parent, workspace, Path.cwd(), Path("/mnt/data")]:
        out.append(root / basename)
        out.append(root / "firmware" / basename)
        out.append(root / "decrypted_payloads" / basename)
    # Walk upward from the tool dir to support repo layouts such as
    # <repo>/scripts/toolchain/mn103s_emu_trace_v17 plus <repo>/decrypted_payloads.
    for parent in [config_path.parent, *config_path.parent.parents]:
        out.append(parent / "firmware" / basename)
        out.append(parent / "decrypted_payloads" / basename)
    # Do not bundle/copy firmware in update packages.  Still allow a user-created
    # local ./firmware copy or analyzer-produced decrypted_payloads output to be found.
    # Preserve order while de-duping.
    seen = set()
    uniq = []
    for p in out:
        try:
            key = str(p.resolve()) if p.exists() else str(p)
        except Exception:
            key = str(p)
        if key not in seen:
            seen.add(key)
            uniq.append(p)
    return uniq


def resolve_firmware(config: dict[str, Any], config_path: Path, firmware_arg: str | None) -> str:
    candidates = []
    if firmware_arg:
        candidates.append(firmware_arg)
    candidates += [
        config.get("custom_bin"),
        config.get("proof_firmware_image"),
        EXPECTED_GDR8050L_0L23_BASENAME,
    ]
    workspace = Path(str(config.get("workspace") or config_path.parent))
    first_text = ""
    for c in candidates:
        if not c:
            continue
        if not first_text:
            first_text = str(c)
        for p in _path_candidates_for_text(str(c), config_path, workspace):
            if p.exists() and p.is_file():
                return str(p)
    return first_text


def proof_watchpoints() -> list[dict[str, Any]]:
    return [
        {"name": "watch_dfa0_cdb", "start": "0x0000DFA0", "size": "0x10", "kind": "access", "description": "firmware-visible packet/CDB bytes"},
        {"name": "watch_df30_packet_frame", "start": "0x0000DF30", "size": "0x10", "kind": "access", "description": "packet frame/tail source"},
        {"name": "watch_05d0_cdb_shadow", "start": "0x000005D0", "size": "0x10", "kind": "access", "description": "firmware shadow of CDB bytes"},
        {"name": "watch_05ec_assembled_lba", "start": "0x000005EC", "size": "0x04", "kind": "access", "description": "assembled LBA candidate"},
        {"name": "watch_05f0_packet_state", "start": "0x000005F0", "size": "0x20", "kind": "access", "description": "packet/transfer state candidate"},
        {"name": "watch_06f1_packet_phase", "start": "0x000006F1", "size": "0x20", "kind": "access", "description": "packet phase/DMA state candidate"},
        {"name": "watch_0760_lba_state", "start": "0x00000760", "size": "0x08", "kind": "access", "description": "current/cached LBA state candidates"},
        {"name": "watch_d904_d905_status", "start": "0x0000D904", "size": "0x02", "kind": "access", "description": "packet status"},
        {"name": "watch_d950_host_status", "start": "0x0000D950", "size": "0x01", "kind": "access", "description": "host packet status"},
    ]


def proof_breakpoints() -> list[dict[str, Any]]:
    cap = [
        {"name": "dfa0_cdb", "address": "0x0000DFA0", "length": 12},
        {"name": "shadow_05d0", "address": "0x000005D0", "length": 12},
        {"name": "assembled_05ec", "address": "0x000005EC", "length": 4},
        {"name": "packet_05f0", "address": "0x000005F0", "length": 16},
        {"name": "phase_06f1", "address": "0x000006F1", "length": 16},
        {"name": "lba_state_0760", "address": "0x00000760", "length": 8},
    ]
    pcs = [
        ("0x9001E3E0", "lba_read10_opcode_compare_0x28_0x9001E3E0"),
        ("0x9001E4E8", "lba_read10_opcode_compare_0x28_0x9001E4E8"),
        ("0x9001E500", "lba_read12_opcode_compare_0xA8_0x9001E500"),
        ("0x9001E527", "lba_candidate_addr_dfa2_packet_cdb_lba_b0_or_shadow_0x9001E527"),
        ("0x9001E52D", "lba_candidate_addr_dfa3_packet_cdb_lba_b1_or_shadow_0x9001E52D"),
        ("0x9001E533", "lba_candidate_addr_dfa4_packet_cdb_lba_b2_or_shadow_0x9001E533"),
        ("0x9001E539", "lba_candidate_addr_dfa5_packet_cdb_lba_b3_or_shadow_0x9001E539"),
        ("0x9001E541", "lba_candidate_addr_05ec_assembled_lba_candidate_0x9001E541"),
        ("0x9000606B", "main_loop_9000606B"),
        ("0x90008824", "packet_timer_done_90008824"),
        ("0x9001CD4D", "data_phase_helper_9001CD4D"),
    ]
    return [{"address": a, "name": n, "capture_memory": cap if a.startswith("0x9001E") else []} for a, n in pcs]


def proof_route_patch() -> dict[str, Any]:
    """Known-small natural route settings that reach the GDR READ CDB/LBA parser.

    This is the v189-v191 route reduced to the knobs needed by mn103s_emu.py.
    It is not the live bridge behavior; it is an offline CPU proof harness.
    """
    return {
        "main_loop_pc": "0x9000606B",
        "timer_done_pc": "0x90008824",
        "host_event_gate": {
            "enabled": True,
            "force_016d_active": True,
            "force_016d_value": "0x01",
            "force_00de_pollable": True,
            "force_00de_value": "0x01",
            "force_08a8": "0x04",
            "force_d950_value": "0x10",
            "post_frame_d950_value": "0x40",
            "clear_08a8_after_frame": True,
            "clear_08a8_after_tail": True,
            "clear_d950_after_tail": True,
            "d950_zero_after_post_frame_reads": "4",
            "irq_after_d108_data_phase": True,
            "complete_data_phase_on_d108": True,
            "seed_mem8": {
                "0x000005B1": "0x20",
                "0x000000DE": "0x01",
                "0x0000016D": "0x01",
                "0x000008A8": "0x04",
                "0x000008A9": "0x00",
                "0x0000DA18": "0x00"
            },
            "seed_dfa0_after_packet_tail": True,
            "seed_dfa0_after_packet_tail_opcodes": ["0x28", "0xA8", "0xBE"],
            "seed_memory_after_latch": [
                {"addr": "0x0A8D", "or": "0x08", "size": 1, "opcodes": ["0x28", "0xA8", "0xBE"]},
                {"addr": "0x0728", "value": "0x07", "size": 1, "opcodes": ["0x28", "0xA8", "0xBE"]},
                {"addr": "0x06F1", "value": "0x04", "size": 1, "opcodes": ["0x28", "0xA8", "0xBE"]}
            ],
            "d954_ready_read": {
                "enabled": True,
                "addr": "0xD954",
                "pc": "0x9001CA1A",
                "mask": "0x04",
                "max_reads": "1",
                "return_pc": "0x9001E4CA",
                "ack_clears": True,
                "opcodes": ["0x28", "0xA8", "0xBE"]
            },
            "df38_fifo": {
                "enabled": True,
                "base": "0xDF38",
                "width": "8",
                "cdb_offset": "0",
                "opcodes": ["0x28", "0xA8", "0xBE"]
            }
        },
        "firmware_ingress": {"enabled": False}
    }


def proof_read_command() -> dict[str, Any]:
    return {
        "tag": "lba_trace_read10_lba306111_blocks1",
        "title": "lba_trace_read10_lba306111_blocks1",
        "direction": "in",
        "cdb_hex": "28 00 00 04 AB BF 00 00 01 00",
        "data_in_hex": " ".join(f"{i & 0xFF:02X}" for i in range(2048))
    }


def make_smoke_config(config: dict[str, Any], config_path: Path, out_prefix: Path, firmware: str) -> dict[str, Any]:
    cfg = deepcopy(config)
    cfg["test_name"] = "GDR-8050L v227 offline firmware READ CDB CPU sweep proof"
    cfg["config_version"] = VERSION
    cfg["requires_emulator_version"] = "V7-direct-config-trace-v120-gcc-version-sync-gdr163-v227-media-preflight"
    cfg["custom_bin"] = firmware
    cfg["proof_firmware_image"] = firmware
    if str(firmware).startswith("/mnt/data"):
        cfg["workspace"] = "/mnt/data"
    cfg["max_cycles"] = int(parse_int(cfg.get("max_cycles"), 240000))
    cfg.setdefault("initial_registers", {})["SP"] = "0x003FF000"
    cfg.setdefault("memory_aliases", [{"name":"SRAM alias 0x80000000->0x00000000","from":"0x80000000","to":"0x00000000","size":"0x400000"}])
    trace = cfg.setdefault("trace", {})
    trace["enabled"] = True
    trace["live_instruction_log"] = False
    trace["out_csv"] = str(out_prefix.with_suffix(".trace.csv"))
    trace["watchpoints"] = proof_watchpoints()
    trace["breakpoints"] = proof_breakpoints()
    # The current live config still carries the historical packet-session model
    # under trace.  Keep it, but force useful event emission and completion.
    gdr = trace.setdefault("gdr_hardware_model", {})
    gdr["enabled"] = True
    gdr.setdefault("timer", {"enabled": True, "addr": "0x0000DC54", "base": "0xF000", "ticks_per_cycle": "0x2000"})
    gdr.setdefault("packet_status", {"enabled": True})
    psm = trace.get("gdr_packet_session_model") or cfg.get("gdr_packet_session_model")
    if not isinstance(psm, dict) or not psm.get("commands"):
        raise SystemExit("config does not contain a usable gdr_packet_session_model with commands")
    trace["gdr_packet_session_model"] = psm
    psm["enabled"] = True
    psm["emit_events"] = True
    psm["stop_on_complete"] = True

    # Force the known small CPU proof route.  The live bridge may use a newer
    # main-loop surface that is fine for stateful iSCSI but does not currently
    # walk the firmware CDB/LBA parser.  This offline proof explicitly selects
    # the reduced natural route that reaches the READ handler candidates.
    psm.update(proof_route_patch())

    cmds = list(psm.get("commands") or [])
    setup = cmds[:10] if len(cmds) >= 10 else cmds
    psm["commands"] = setup + [proof_read_command()]
    return cfg


def parse_trace(trace_csv: Path) -> dict[str, Any]:
    rows: list[dict[str, str]] = []
    with trace_csv.open("r", encoding="utf-8", newline="") as f:
        rows = list(csv.DictReader(f))
    session_complete = any(r.get("event") == "gdr_session_complete" for r in rows)
    command_complete = [r for r in rows if r.get("event") == "gdr_session_command_complete"]
    handler_hits = sorted({norm_hex(r.get("pc")) for r in rows if r.get("event") == "break" and norm_hex(r.get("pc")) in READ_HANDLER_PCS})
    lba_byte_hits = sorted({norm_hex(r.get("pc")) for r in rows if r.get("event") == "break" and norm_hex(r.get("pc")) in LBA_BYTE_PCS})
    assembled_hits = [r for r in rows if r.get("event") == "break" and norm_hex(r.get("pc")) == ASSEMBLED_PC]
    # Reconstruct the last 4 byte writes at 0x05EC..0x05EF from the assembled PC.
    writes = []
    for r in rows:
        if r.get("event") != "watch_write" or r.get("name") != "watch_05ec_assembled_lba":
            continue
        if norm_hex(r.get("pc")) != ASSEMBLED_PC:
            continue
        addr = parse_int(r.get("addr"), -1)
        if 0x05EC <= addr <= 0x05EF:
            writes.append({
                "seq": parse_int(r.get("seq"), 0),
                "pc": norm_hex(r.get("pc")),
                "addr": f"0x{addr:08X}",
                "value": parse_int(r.get("value"), 0) & 0xFF,
                "bytes_hex": str(r.get("bytes_hex") or ""),
            })
    last_by_addr: dict[int, dict[str, Any]] = {}
    for w in writes:
        last_by_addr[parse_int(w["addr"])] = w
    assembled_bytes = [last_by_addr.get(0x05EC + i, {}).get("value") for i in range(4)]
    if all(v is not None for v in assembled_bytes):
        assembled_lba = int(assembled_bytes[0]) | (int(assembled_bytes[1]) << 8) | (int(assembled_bytes[2]) << 16) | (int(assembled_bytes[3]) << 24)
    else:
        assembled_lba = None
    dfa_cdb_seen = any((
        ("dfa0_cdb@0x0000DFA0" in (r.get("notes") or "") and "28000004ABBF" in (r.get("notes") or "").replace(" ", "").upper())
        or (r.get("event") == "gdr_session_dfa0_latch" and "28000004ABBF" in (r.get("bytes_hex") or "").replace(" ", "").upper())
    ) for r in rows)
    source_decisions = [r for r in rows if r.get("event") == "gdr_session_read_source_decision"]
    outcome = (
        session_complete
        and len(command_complete) > 0
        and bool(handler_hits)
        and len(lba_byte_hits) >= 4
        and bool(assembled_hits)
        and assembled_lba == EXPECTED_HOST_LBA
        and dfa_cdb_seen
    )
    return {
        "version": VERSION,
        "trace_csv": str(trace_csv),
        "rows": len(rows),
        "session_complete": session_complete,
        "command_complete_count": len(command_complete),
        "expected_host_lba": EXPECTED_HOST_LBA,
        "expected_host_lba_hex": EXPECTED_HOST_LBA_HEX,
        "assembled_05ec_lba": assembled_lba,
        "assembled_05ec_lba_hex": None if assembled_lba is None else f"0x{assembled_lba:08X}",
        "assembled_05ec_bytes_le": [None if v is None else f"{int(v)&0xFF:02X}" for v in assembled_bytes],
        "assembled_05ec_match": assembled_lba == EXPECTED_HOST_LBA,
        "read_handler_hits": handler_hits,
        "lba_byte_read_pc_hits": lba_byte_hits,
        "assembled_lba_pc_hits": sorted({norm_hex(r.get("pc")) for r in assembled_hits}),
        "dfa0_cdb_seen": dfa_cdb_seen,
        "source_decision_rows": len(source_decisions),
        "watch_05ec_write_count": len(writes),
        "sample_05ec_writes": writes[:8],
        "firmware_derived_host_lba": bool(assembled_lba == EXPECTED_HOST_LBA),
        "firmware_derived_delta": False,
        "ok": outcome,
        "note": "This proves firmware-side READ(10) CDB/LBA ingestion into SRAM 0x05EC for the sampled host LBA. It does not yet derive the 306080 media-window delta from firmware state.",
    }


def write_md(path: Path, report: dict[str, Any]) -> None:
    lines = [
        "# GDR-8050L v227 firmware READ CDB CPU sweep proof",
        "",
        f"- ok: `{str(report.get('ok')).lower()}`",
        f"- rows: `{report.get('rows')}`",
        f"- session_complete: `{str(report.get('session_complete')).lower()}`",
        f"- command_complete_count: `{report.get('command_complete_count')}`",
        f"- expected_host_lba: `{report.get('expected_host_lba_hex')}`",
        f"- assembled_05ec_lba: `{report.get('assembled_05ec_lba_hex')}`",
        f"- assembled_05ec_bytes_le: `{report.get('assembled_05ec_bytes_le')}`",
        f"- assembled_05ec_match: `{str(report.get('assembled_05ec_match')).lower()}`",
        f"- dfa0_cdb_seen: `{str(report.get('dfa0_cdb_seen')).lower()}`",
        f"- read_handler_hits: `{', '.join(report.get('read_handler_hits') or [])}`",
        f"- lba_byte_read_pc_hits: `{', '.join(report.get('lba_byte_read_pc_hits') or [])}`",
        f"- assembled_lba_pc_hits: `{', '.join(report.get('assembled_lba_pc_hits') or [])}`",
        f"- firmware_derived_host_lba: `{str(report.get('firmware_derived_host_lba')).lower()}`",
        f"- firmware_derived_delta: `{str(report.get('firmware_derived_delta')).lower()}`",
        "",
        "## Scope",
        "",
        str(report.get("note") or ""),
        "",
    ]
    if report.get("sample_05ec_writes"):
        lines += ["## Sample 0x05EC writes", "", "| seq | pc | addr | value |", "|---:|---|---|---|"]
        for w in report["sample_05ec_writes"]:
            lines.append(f"| {w.get('seq')} | `{w.get('pc')}` | `{w.get('addr')}` | `0x{int(w.get('value')):02X}` |")
        lines.append("")
    path.write_text("\n".join(lines), encoding="utf-8")


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Run GDR-8050L v227 firmware READ CDB CPU sweep proof")
    ap.add_argument("--config", default="gdr8050l_auth_v227_stateful_relock_keep_loaded_iscsi_config.json")
    ap.add_argument("--firmware")
    ap.add_argument("--mn103s-emu", default="mn103s_emu.py")
    ap.add_argument("--out-prefix", default="gdr8050l_auth_v227_firmware_read_cdb_cpu_sweep")
    ap.add_argument("--json")
    ap.add_argument("--md")
    ap.add_argument("--no-run", action="store_true", help="Only parse an existing trace at <out-prefix>.trace.csv")
    args = ap.parse_args(argv)

    config_path = Path(args.config).resolve()
    config = load_json(config_path)
    out_prefix = Path(args.out_prefix).resolve()
    json_path = Path(args.json).resolve() if args.json else out_prefix.with_suffix(".json")
    md_path = Path(args.md).resolve() if args.md else out_prefix.with_suffix(".md")
    smoke_config_path = out_prefix.with_suffix(".config.json")
    run_log_path = out_prefix.with_suffix(".run.log")
    trace_csv = out_prefix.with_suffix(".trace.csv")

    firmware = resolve_firmware(config, config_path, args.firmware)
    if not firmware or not Path(str(firmware)).exists():
        report = {
            "version": VERSION,
            "ok": False,
            "skipped": True,
            "reason": f"firmware image not found: {firmware}",
            "firmware_derived_host_lba": False,
            "firmware_derived_delta": False,
        }
        write_json(json_path, report)
        write_md(md_path, report)
        print(f"[FIRMWARE-LBA-CPU-SMOKE] skipped firmware_missing ok=false")
        return 2

    if not args.no_run:
        smoke_config = make_smoke_config(config, config_path, out_prefix, str(firmware))
        smoke_config["trace"]["out_csv"] = str(trace_csv)
        write_json(smoke_config_path, smoke_config)
        emu_path = rel_or_abs(args.mn103s_emu, config_path.parent)
        cmd = [sys.executable, str(emu_path), str(smoke_config_path)]
        with run_log_path.open("w", encoding="utf-8") as log:
            proc = subprocess.run(cmd, cwd=str(config_path.parent), text=True, stdout=log, stderr=subprocess.STDOUT)
        if proc.returncode != 0:
            report = {
                "version": VERSION,
                "ok": False,
                "returncode": proc.returncode,
                "run_log": str(run_log_path),
                "firmware_derived_host_lba": False,
                "firmware_derived_delta": False,
            }
            write_json(json_path, report)
            write_md(md_path, report)
            print(f"[FIRMWARE-LBA-CPU-SMOKE] emulator_failed rc={proc.returncode} ok=false")
            return proc.returncode or 1

    report = parse_trace(trace_csv)
    report.update({
        "smoke_config": str(smoke_config_path),
        "run_log": str(run_log_path),
        "firmware": str(firmware),
    })
    write_json(json_path, report)
    write_md(md_path, report)
    print(f"[FIRMWARE-LBA-CPU-SMOKE] rows={report['rows']} assembled_05ec={report.get('assembled_05ec_lba_hex')} firmware_derived_host_lba={str(report.get('firmware_derived_host_lba')).lower()} firmware_derived_delta=false ok={str(report.get('ok')).lower()}")
    return 0 if report.get("ok") else 1


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