#!/usr/bin/env python3
"""S4 evidence harness: is the native GSP simulator above the QEMU line?

Runs one A/B pair on the real TU104 570.144 GSP image:

* **QEMU baseline** - the v0.1/v0.2 probe path (flat relocated load, bounded
  wall-clock run, ``guest_errors,unimp,int`` log): first exception, address and
  ladder rung, recorded as a normal ``greeneye-session/1``.
* **Native baseline** - ``sim.gsp_profile`` (deterministic RV64IMAC, modeled
  cycles): instructions, cycles, first modeled MMIO, stop reason and rung.
  If S1's device model/profile is present (``sim/profiles/``, ``sim/devices/``,
  or a Python ``sim.devices`` entry point) it is discovered defensively and
  used as the MMIO map; otherwise the run is the lenient B0 baseline.

The rung ladder is docs/35-GREENEYE-SIMULATOR-DESIGN.md section 3.3
(``pre-B0`` < B0 < B1 < ... < B5).  "Above the QEMU line" means the native
rung ordinal is at least one past QEMU's, twice identically, within the frozen
wall-clock threshold, and with a discovered S1 device model (docs/35 1.1);
requiring the device model means a lenient run reports ``above_lenient_only``,
not ``above``.  The frozen record and the first measured results live in
``docs/37-SIM-QEMU-LINE.md``.

The harness also exercises the native feature matrix (hooks/breakpoints/
watchpoints, memory/struct, sessions, compare, trace, injection) on the bundled
demo firmware.  MCP is deliberately ``not_run``: the native MCP surface is S3's
deliverable (see docs/37 section "Not verified").

Everything is stdlib-only, bounded (``<= 2M`` instructions per native run,
``<= 10 s`` of QEMU) and touches no hardware, GPU, guest, module or network.

CLI::

    python3 harness/sim_qemu_line.py
    python3 harness/sim_qemu_line.py --no-qemu --instructions 50000
    python3 harness/sim_qemu_line.py --device-map sim/profiles/tu104-570.144.json
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import shutil
import subprocess
import sys
import time

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

from greeneye.analyze import compare_sessions, parse_qemu_faults  # noqa: E402
from greeneye.device_map import (  # noqa: E402
    QEMU_REFERENCE,
    DeviceEntry,
    DeviceMap,
    firmware_sha256,
    harvest_from_text,
)
from greeneye.errors import GreenEyeError  # noqa: E402
from harness.common import (  # noqa: E402
    load_target,
    open_flat_session,
    qemu_available,
    read_log_window,
)
from sim.gsp_profile import DEFAULT_IMAGE, run_profile  # noqa: E402

SCHEMA = "greeneye-sim-qemu-line/1"

#: Boot-ladder rungs (docs/35 section 3.3), lowest first.  ``pre-B0`` is the
#: QEMU fault position: image loaded, first MMIO observed *as a fault*.
RUNG_ORDER = ("pre-B0", "B0", "B1", "B2", "B3", "B4", "B5")
RUNG_INDEX = {name: index for index, name in enumerate(RUNG_ORDER)}
QEMU_RUNG = "pre-B0"
NATIVE_LENIENT_RUNG = "B0"

#: Default frozen thresholds (recorded 2026-10-03, before the first
#: measurement).  The measured run and rationale live in
#: docs/37-SIM-QEMU-LINE.md section 3; the CLI uses these values when no
#: override is passed.  Never move silently.
FROZEN_THRESHOLDS = {
    "frozen_on": "2026-10-03",
    "native_rungs_above_qemu": 1,
    "identical_runs": 2,
    "native_pair_wall_s": 20.0,
    "qemu_first_fault_fw": QEMU_REFERENCE["firmware_vaddr"],
    "qemu_first_touch": QEMU_REFERENCE["access_addr"],
    # docs/35 1.1 requires the milestone to be reached "with the device model
    # and shims documented": lenient MMIO is observation, not a device model,
    # so it cannot satisfy the line contract on its own.
    "device_model_required": True,
}

#: Bounds (eng-ops / task discipline).
MAX_INSTRUCTIONS = 2_000_000
MAX_QEMU_DURATION_S = 10.0
DEFAULT_INSTRUCTIONS = 200_000
DEFAULT_QEMU_DURATION_S = 0.5

#: Default report directory (new, S4-owned).
DEFAULT_OUT = os.path.join(HERE, "qemu_line")

DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
DEMO_SCHEMA = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "gsp_ctrl.schema.json")
SDK_ELF = os.path.join(TOOL_ROOT, "sdk", "build", "injected_demo.elf")

#: S1 device-model discovery: JSON conventions first, then a duck-typed Python
#: API.  See docs/37 section "Re-running when S1 lands".
S1_JSON_DIRS = ("sim/profiles", "sim/devices")
S1_PY_API_NAMES = (
    "device_map_for_firmware",
    "profile_for_firmware",
    "load_device_map",
    "load_profile",
    "profile_for",
)
_RUNG_KEYS = ("rung", "highest_rung", "boot_ladder_rung", "milestone")


class HarnessError(RuntimeError):
    """A harness precondition failed in a way the caller should report."""


# -- small helpers ---------------------------------------------------------


def _relpath(path: str | None) -> str | None:
    if path is None:
        return None
    return os.path.relpath(os.path.abspath(path), TOOL_ROOT)


def _rm_tree(path: str) -> None:
    shutil.rmtree(path, ignore_errors=True)


def _write_text(path: str, text: str) -> str:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(text)
    return path


def _write_json(path: str, payload: dict) -> str:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = f"{path}.tmp-{os.getpid()}"
    with open(tmp, "w", encoding="utf-8") as fh:
        json.dump(payload, fh, indent=2, sort_keys=True)
        fh.write("\n")
    os.replace(tmp, path)
    return path


def session_hashes(session_dir: str | None) -> dict:
    """sha256 of a session's core evidence files ({} when the dir is absent)."""
    if not session_dir or not os.path.isdir(session_dir):
        return {}
    hashes: dict = {}
    for name in ("session.json", "events.jsonl", "summary.md"):
        path = os.path.join(session_dir, name)
        if not os.path.isfile(path):
            continue
        digest = hashlib.sha256()
        with open(path, "rb") as fh:
            for chunk in iter(lambda: fh.read(1 << 20), b""):
                digest.update(chunk)
        hashes[name] = digest.hexdigest()
    return hashes


def _x86_vms_running() -> int | None:
    """Count other agents' x86 QEMU VMs (never disturbed; recorded only)."""
    try:
        proc = subprocess.run(
            ["pgrep", "-f", "qemu-system-x86_64"],
            capture_output=True,
            text=True,
            timeout=5.0,
        )
    except (OSError, subprocess.SubprocessError):
        return None
    if proc.returncode not in (0, 1):
        return None
    pids = [line for line in proc.stdout.splitlines() if line.strip().isdigit()]
    return len(pids)


def rung_ordinal(rung: str | None) -> int | None:
    if rung in RUNG_INDEX:
        return RUNG_INDEX[rung]
    return None


def above_qemu_line(qemu_rung: str | None, native_rung: str | None, minimum: int = 1) -> bool:
    """True when ``native_rung`` is at least ``minimum`` rungs past QEMU's."""
    q = rung_ordinal(qemu_rung)
    n = rung_ordinal(native_rung)
    if q is None or n is None:
        return False
    return n >= q + minimum


def _summary_fault(fault: dict, bias: int) -> dict:
    access = fault.get("access") or {}
    epc = int(fault.get("epc", 0) or 0)
    return {
        "desc": fault.get("desc"),
        "cause": fault.get("cause"),
        "epc": epc,
        "epc_fw": (epc - bias) if (bias and epc and epc >= bias) else epc,
        "access_addr": access.get("access_addr"),
        "size": access.get("size"),
        "rw": access.get("rw"),
        "reason": access.get("reason"),
    }


# -- QEMU baseline ---------------------------------------------------------


def run_qemu_baseline(
    image: str = DEFAULT_IMAGE,
    out_dir: str = DEFAULT_OUT,
    duration: float = DEFAULT_QEMU_DURATION_S,
) -> dict:
    """Probe the real image under QEMU; skip cleanly when unavailable/absent."""
    result: dict = {
        "backend": "qemu",
        "ran": False,
        "reason": None,
        "rung": None,
        "session": None,
        "wall_s": 0.0,
    }
    if duration <= 0 or duration > MAX_QEMU_DURATION_S:
        raise HarnessError(f"qemu duration must be in (0, {MAX_QEMU_DURATION_S}], got {duration}")
    if not os.path.isfile(image):
        result["reason"] = f"firmware absent: {image}"
        return result
    if not qemu_available():
        result["reason"] = "qemu-system-riscv64 not installed"
        return result

    start = time.monotonic()
    image_obj = load_target(image)
    session_dir = os.path.join(out_dir, "sessions", "qemu-probe")
    _rm_tree(session_dir)
    with open_flat_session(image_obj, image, record=session_dir, name="s4-qemu-probe") as flat:
        flat.session.run(duration=duration)
        pc = flat.session.pc
        bias = flat.bias
        # Read the log window once: `-d int` trap loops flood the file, so the
        # first exception has to be parsed from the bounded head window.
        log_text, capped = read_log_window(flat.target.qemu_log)
        faults = parse_qemu_faults(log_text)
        first = faults[0] if faults else None
        device_map = harvest_from_text(log_text, bias=bias)
        meta = {
            "s4_qemu_rung": QEMU_RUNG,
            "qemu_fault_count_window": len(faults),
            "qemu_log_window_capped": capped,
            "qemu_first_fault": first,
        }
        recorder = getattr(flat.session, "recorder", None)
        if recorder is not None and hasattr(recorder, "meta"):
            recorder.meta.update(meta)
            if log_text and hasattr(recorder, "blob"):
                recorder.meta["qemu_log_blob"] = recorder.blob(
                    "qemu.log", log_text.encode("utf-8")
                )
    result.update(
        {
            "ran": True,
            "reason": "ok",
            "rung": QEMU_RUNG,
            "session": _relpath(session_dir),
            "session_hashes": session_hashes(session_dir),
            "stopped_pc": pc,
            "fault_count_window": len(faults),
            "log_window_capped": capped,
            "first_fault": _summary_fault(first, bias) if first else None,
            "first_touch_match": bool(
                first
                and (first.get("access") or {}).get("access_addr") == QEMU_REFERENCE["access_addr"]
            ),
            "device_map_entries": len(device_map),
            "device_map": device_map,
        }
    )
    result["wall_s"] = round(time.monotonic() - start, 4)
    return result


# -- S1 integration (defensive) -------------------------------------------


def _coerce_device_map(obj) -> DeviceMap | None:
    """Accept DeviceMap objects, paths, or device-map/profile dicts; else None."""
    if isinstance(obj, DeviceMap):
        return obj
    if isinstance(obj, (str, os.PathLike)):
        path = os.fspath(obj)
        if os.path.isfile(path):
            try:
                return DeviceMap.load(path)
            except GreenEyeError:
                return None
        return None
    if isinstance(obj, dict):
        if obj.get("schema") == "greeneye-device-map/1":
            try:
                return DeviceMap.from_dict(obj)
            except (KeyError, TypeError, ValueError, GreenEyeError):
                return None
        if "device_map" in obj:
            return _coerce_device_map(obj["device_map"])
        entries = obj.get("entries")
        if isinstance(entries, list):
            dm = DeviceMap(meta=dict(obj.get("meta") or {}))
            try:
                for raw in entries:
                    dm.add(DeviceEntry.from_dict(raw))
            except (KeyError, TypeError, ValueError):
                return None
            return dm
    return None


def _json_rung(data) -> str | None:
    if not isinstance(data, dict):
        return None
    for key in _RUNG_KEYS:
        value = data.get(key)
        if isinstance(value, str) and value in RUNG_INDEX:
            return value
        if isinstance(value, dict):
            inner = value.get("highest") or value.get("rung")
            if isinstance(inner, str) and inner in RUNG_INDEX:
                return inner
    return None


def discover_s1_device_model(image: str, sha: str | None = None) -> dict:
    """Find S1's device model/profile without assuming it exists.

    Returns ``{"device_map": DeviceMap|None, "profile": path|None,
    "declared_rung": str|None, "source": str|None, "notes": [...]}``.
    Discovery order: an explicit ``sim.devices`` Python entry point, then JSON
    files under ``sim/profiles/`` and ``sim/devices/`` (a
    ``greeneye-sim-profile/1`` profile, a ``greeneye-device-map/1`` file, a
    profile with a ``device_map`` key, or a raw ``entries`` list).  A profile
    is the semantic device model (``sim.gsp_boot`` runs it); a device map only
    replays observed accesses.  A candidate is preferred when its firmware
    hash matches the image; otherwise a single candidate is used and noted as
    unverified pairing.
    """
    notes: list[str] = []
    declared_rung: str | None = None

    try:
        import sim.devices as devices  # type: ignore

        for name in S1_PY_API_NAMES:
            fn = getattr(devices, name, None)
            if not callable(fn):
                continue
            for args in ((sha,), (image,), ()):
                if sha is None and args == (sha,):
                    continue
                try:
                    candidate = fn(*args)
                except TypeError:
                    continue
                except (OSError, UnicodeError):
                    # A path-taking helper handed a digest (or a helper handed
                    # the binary firmware) simply does not apply; not an error.
                    break
                except Exception as exc:  # discovery must never break the run
                    notes.append(f"sim.devices.{name}{args!r} failed: {exc}")
                    break
                if candidate is None:
                    continue
                dm = _coerce_device_map(candidate)
                rung = getattr(candidate, "rung", None)
                if isinstance(rung, str) and rung in RUNG_INDEX:
                    declared_rung = rung
                if dm is not None:
                    notes.append(f"device map via sim.devices.{name}{args!r}")
                    return {
                        "device_map": dm,
                        "profile": None,
                        "declared_rung": declared_rung,
                        "source": f"sim.devices.{name}",
                        "notes": notes,
                    }
    except ImportError:
        notes.append("sim.devices package absent")
    except Exception as exc:
        notes.append(f"sim.devices import failed: {exc}")

    candidates: list[tuple[str, DeviceMap, str | None, str | None]] = []
    profiles: list[tuple[str, str, str | None]] = []
    for rel in S1_JSON_DIRS:
        root = os.path.join(TOOL_ROOT, rel)
        if not os.path.isdir(root):
            notes.append(f"{rel}/ absent")
            continue
        for name in sorted(os.listdir(root)):
            if not name.endswith(".json"):
                continue
            path = os.path.join(root, name)
            try:
                with open(path, "r", encoding="utf-8") as fh:
                    data = json.load(fh)
            except (OSError, ValueError) as exc:
                notes.append(f"{rel}/{name}: unreadable ({exc})")
                continue
            if isinstance(data, dict) and data.get("schema") == "greeneye-sim-profile/1":
                firmware = data.get("firmware") or {}
                profiles.append((f"{rel}/{name}", path,
                                 str(firmware.get("sha256") or "") or None))
                continue
            dm = _coerce_device_map(data)
            rung = _json_rung(data)
            if rung and declared_rung is None:
                declared_rung = rung
            if dm is not None:
                candidates.append((f"{rel}/{name}", dm, rung, None))

    exact_profiles = [p for p in profiles if sha and p[2] == sha]
    if exact_profiles:
        rel, path, _digest = exact_profiles[0]
        notes.append(f"using {rel} (greeneye-sim-profile/1, firmware_sha256 matched)")
        return {
            "device_map": None,
            "profile": path,
            "declared_rung": declared_rung,
            "source": rel,
            "notes": notes,
        }

    if not candidates:
        if len(profiles) == 1:
            rel, path, _digest = profiles[0]
            notes.append(f"using {rel} (single profile, pairing unverified)")
            return {
                "device_map": None,
                "profile": path,
                "declared_rung": declared_rung,
                "source": rel,
                "notes": notes,
            }
        notes.append("no S1 device-map/profile JSON found")
        return {"device_map": None, "profile": None, "declared_rung": declared_rung,
                "source": None, "notes": notes}

    exact = [
        c for c in candidates
        if sha and str((c[1].meta or {}).get("firmware_sha256", "")) == sha
    ]
    chosen = exact[0] if exact else candidates[0]
    note = "firmware_sha256 matched" if exact else "single candidate, pairing unverified"
    notes.append(f"using {chosen[0]} ({note})")
    return {
        "device_map": chosen[1],
        "profile": None,
        "declared_rung": chosen[2] or declared_rung,
        "source": chosen[0],
        "notes": notes,
    }


# -- native baseline -------------------------------------------------------


def estimate_native_rung(run: dict, *, device_model_used: bool,
                         declared_rung: str | None = None) -> tuple[str, str]:
    """Conservative rung estimate from a bounded native run.

    * matching lenient first touch -> B0 (observed, no device semantics);
    * a device model that serves the QEMU first-touch and lets execution
      continue -> B1 (frontier moved past the QEMU fault);
    * a higher rung only when S1 declares it in profile data (honoured as
      ``rung_source: declared``).

    B2+ need boot-stage milestones, not just progress; absence of a fault is
    not evidence of a boot stage.  See docs/37 sections 3-4.
    """
    stop = run.get("stop") or {}
    stop_reason = stop.get("reason")
    first = run.get("first_unknown_mmio")
    reached_b0 = bool(
        (first and first.get("addr") == QEMU_REFERENCE["access_addr"])
        or run.get("instructions", 0) > 0
    )

    inferred = None
    basis = None
    if device_model_used and reached_b0:
        unmodelled_at_qemu = bool(first and first.get("addr") == QEMU_REFERENCE["access_addr"])
        if not unmodelled_at_qemu:
            inferred = "B1"
            basis = (
                "device model served the QEMU first-touch and execution continued "
                "past the QEMU fault to the run bound"
            )
        else:
            inferred = "B0"
            basis = "device model present but the QEMU first-touch was still served leniently"
    elif reached_b0:
        inferred = NATIVE_LENIENT_RUNG
        basis = "lenient first-touch observation; execution continued past the QEMU fault"

    if inferred is None:
        return "pre-B0", "no native progress observed"

    rung, source = inferred, "inferred"
    declared_index = rung_ordinal(declared_rung)
    if declared_index is not None and declared_index > RUNG_INDEX[inferred]:
        rung, source = declared_rung, "declared"
        basis = f"S1 profile declares {declared_rung}; inference was {inferred} ({basis})"
    return rung, f"{basis} [rung_source={source}, stop={stop_reason}]"


def run_native_baseline(
    image: str = DEFAULT_IMAGE,
    out_dir: str = DEFAULT_OUT,
    instructions: int = DEFAULT_INSTRUCTIONS,
    repeat: int = 2,
    device_map=None,
    declared_rung: str | None = None,
    profile: str | None = None,
    use_shim: bool = True,
) -> dict:
    """Run the bounded deterministic native baseline ``repeat`` times.

    ``profile`` selects the S1 semantic device model (``sim.gsp_boot`` with
    the versioned profile and the recorded boot shim); without it the run is
    the lenient ``sim.gsp_profile`` baseline.  ``device_map`` remains a
    legacy/observational MMIO map for the lenient path.
    """
    if instructions <= 0 or instructions > MAX_INSTRUCTIONS:
        raise HarnessError(f"instructions must be in (0, {MAX_INSTRUCTIONS}], got {instructions}")
    if repeat < 1:
        raise HarnessError(f"repeat must be >= 1, got {repeat}")
    if not os.path.isfile(image):
        return {
            "backend": "native",
            "ran": False,
            "reason": f"firmware absent: {image}",
            "runs": [],
        }

    s1_model = bool(profile) or device_map is not None
    runs: list[dict] = []
    start_all = time.monotonic()
    for index in range(repeat):
        session_dir = os.path.join(out_dir, "sessions", f"native-{index + 1}")
        _rm_tree(session_dir)
        start = time.monotonic()
        if profile:
            from sim.gsp_boot import run_boot  # deferred: only the S1 path needs it

            boot = run_boot(
                image=image,
                profile_path=profile,
                instructions=instructions,
                record=session_dir,
                name=f"s4-native-{index + 1}",
                use_shim=use_shim,
            )
            wall = time.monotonic() - start
            if not boot.ran:
                raise HarnessError(f"native boot refused to run: {boot.reason}")
            run = {
                "instructions": boot.instructions,
                "cycles": boot.cycles,
                "cycle_model": boot.cycle_model,
                "stop": {
                    "reason": boot.stop_reason,
                    "pc": boot.stop_pc,
                    "pc_fw": boot.stop_pc_fw,
                },
                "first_unknown_mmio": boot.first_unknown_mmio,
                "mmio_unknown": boot.mmio_unknown,
                "device_unhandled": boot.device_unhandled,
                "device_accesses": boot.device_accesses,
                "device_state_sha256": boot.device_state_sha256,
                "milestones": list(boot.milestones),
                "session": _relpath(session_dir),
                "session_hashes": session_hashes(session_dir),
                "wall_s": round(wall, 4),
            }
            run["rung"] = boot.rung
            run["rung_basis"] = (
                "observed S1 boot-ladder rung from sim.gsp_boot "
                f"(milestones={list(boot.milestones)}, mmio_unknown={boot.mmio_unknown}, "
                f"device_unhandled={boot.device_unhandled})"
            )
        else:
            result = run_profile(
                image=image,
                instructions=instructions,
                record=session_dir,
                device_map=device_map,
                name=f"s4-native-{index + 1}",
            )
            wall = time.monotonic() - start
            if not result.ran:
                raise HarnessError(f"native profile refused to run: {result.reason}")
            run = {
                "instructions": result.instructions,
                "cycles": result.cycles,
                "cycle_model": result.cycle_model,
                "stop": {
                    "reason": result.stop_reason,
                    "pc": result.stop_pc,
                    "pc_fw": result.stop_pc_fw,
                },
                "first_unknown_mmio": result.first_unknown_mmio,
                "mmio_unknown": result.mmio_unknown,
                "session": _relpath(session_dir),
                "session_hashes": session_hashes(session_dir),
                "wall_s": round(wall, 4),
            }
            run["rung"], run["rung_basis"] = estimate_native_rung(
                run, device_model_used=s1_model, declared_rung=declared_rung
            )
        runs.append(run)

    keys = (
        "instructions",
        "cycles",
        "stop",
        "first_unknown_mmio",
        "mmio_unknown",
        "rung",
        "device_state_sha256",
        "milestones",
    )
    identical = all(
        all(a.get(key) == runs[0].get(key) for key in keys) for a in runs[1:]
    )
    compare = {"ran": False}
    if repeat >= 2:
        try:
            text = compare_sessions(
                os.path.join(out_dir, "sessions", "native-1"),
                os.path.join(out_dir, "sessions", "native-2"),
            )
            compare = {
                "ran": True,
                "chars": len(text),
                "headers": [
                    line[3:].strip()
                    for line in text.splitlines()
                    if line.startswith("## ")
                ],
            }
        except Exception as exc:  # compare is evidence, never a fatal step
            compare = {"ran": False, "error": f"{type(exc).__name__}: {exc}"}

    return {
        "backend": "native",
        "ran": True,
        "reason": "ok",
        "instructions_budget": instructions,
        "repeat": repeat,
        "identical": identical,
        "rung": runs[0]["rung"],
        "rung_basis": runs[0]["rung_basis"],
        "runs": runs,
        "pair_wall_s": round(time.monotonic() - start_all, 4),
        "compare": compare,
        "device_model_used": s1_model,
        "profile": _relpath(profile),
    }


# -- native feature matrix -------------------------------------------------


def _events_summary(session_dir: str) -> dict:
    counts: dict[str, int] = {}
    path = os.path.join(session_dir, "events.jsonl")
    try:
        with open(path, "r", encoding="utf-8") as fh:
            for line in fh:
                line = line.strip()
                if not line:
                    continue
                try:
                    kind = json.loads(line).get("type")
                except ValueError:
                    continue
                if kind:
                    counts[kind] = counts.get(kind, 0) + 1
    except OSError:
        pass
    return dict(sorted(counts.items()))


def run_feature_matrix(out_dir: str = DEFAULT_OUT) -> dict:
    """Exercise the native debugger feature set on the bundled demo firmware."""
    import greeneye as ge

    matrix: dict = {}
    if not os.path.isfile(DEMO_ELF):
        return {
            "ran": False,
            "reason": f"demo firmware absent: {DEMO_ELF} (run make demo)",
            "features": {},
        }

    session_dir = os.path.join(out_dir, "sessions", "features")
    _rm_tree(session_dir)
    session = ge.launch(kernel=DEMO_ELF, backend="native", record=session_dir, name="s4-features")
    try:
        ctrl = session.symbols.lookup("gsp_ctrl")
        session.run(duration=0.2)
        magic = session.memory.read_u32(ctrl)
        matrix["memory"] = {
            "status": "pass" if magic == 0x47535031 else "fail",
            "detail": {"ctrl": ctrl, "magic": magic},
        }

        with open(DEMO_SCHEMA, "r", encoding="utf-8") as fh:
            schema = ge.schema_from_json(json.load(fh))
        decoded = schema.decode(session.memory.read(ctrl, schema.size), base=0)
        matrix["struct"] = {
            "status": "pass" if decoded.get("magic") == 0x47535031 else "fail",
            "detail": {"schema": os.path.basename(DEMO_SCHEMA), "fields": len(decoded)},
        }

        dispatches: list[int] = []
        stores: list[int] = []
        session.hook("rpc_dispatch", on_enter=lambda hit: dispatches.append(hit.arg(0)))
        session.watch(ctrl + 0x40, size=8, kind="w", callback=lambda hit: stores.append(hit.pc))
        session.memory.write_u64(ctrl + 0x28, 0x1234)
        session.memory.write_u64(ctrl + 0x20, 1)
        session.run(duration=0.25)
        matrix["hooks"] = {
            "status": "pass" if dispatches else "fail",
            "detail": {"rpc_dispatch_hits": len(dispatches), "cmds": dispatches},
        }
        matrix["watch"] = {
            "status": "pass" if stores else "fail",
            "detail": {"result_field_stores": len(stores), "pcs": stores[:4]},
        }

        session.memory.write_u64(ctrl + 0x28, 0x55)
        session.memory.write_u64(ctrl + 0x20, 2)
        hit = session.run_until("rpc_dispatch", timeout=1.0)
        bp = session.breakpoint("rpc_dispatch")
        attached = bool(bp.attached)
        session.remove_breakpoint(bp)
        matrix["breakpoints"] = {
            "status": "pass" if hit is not None and attached and not bp.attached else "fail",
            "detail": {
                "run_until_hit": hit.addr if hit is not None else None,
                "attached_then_removed": attached and not bp.attached,
                "kind": "sw",
            },
        }
    finally:
        session.close()

    events = _events_summary(session_dir)
    matrix["sessions"] = {
        "status": "pass"
        if os.path.isfile(os.path.join(session_dir, "session.json"))
        and os.path.isfile(os.path.join(session_dir, "events.jsonl"))
        else "fail",
        "detail": {
            "dir": _relpath(session_dir),
            "event_counts": events,
            "hashes": session_hashes(session_dir),
        },
    }

    trace_dir = os.path.join(out_dir, "sessions", "features-trace")
    _rm_tree(trace_dir)
    trace_session = ge.launch(
        kernel=DEMO_ELF,
        backend="native",
        record=trace_dir,
        name="s4-features-trace",
        trace={"exec": True, "max_events": 4000},
    )
    try:
        trace_session.run(duration=0.1)
    finally:
        trace_session.close()
    trace_path = os.path.join(trace_dir, "blobs", "trace.jsonl")
    trace_lines = 0
    if os.path.isfile(trace_path):
        with open(trace_path, "r", encoding="utf-8", errors="replace") as fh:
            trace_lines = sum(1 for _ in fh)
    matrix["trace"] = {
        "status": "pass" if trace_lines > 0 else "fail",
        "detail": {
            "dir": _relpath(trace_dir),
            "blob": _relpath(trace_path),
            "events": trace_lines,
            "max_events": 4000,
            "hashes": session_hashes(trace_dir),
        },
    }

    if os.path.isfile(SDK_ELF):
        from greeneye.inject import invoke_remote, load_service

        inject_dir = os.path.join(out_dir, "sessions", "features-inject")
        _rm_tree(inject_dir)
        inject_session = ge.launch(
            kernel=DEMO_ELF, backend="native", record=inject_dir, name="s4-features-inject"
        )
        try:
            inject_session.run(duration=0.25)
            service = load_service(inject_session, SDK_ELF, slot=6)
            value = invoke_remote(inject_session, service, 1, 0x41)
        finally:
            inject_session.close()
        matrix["injection"] = {
            "status": "pass" if value == (0x41 ^ 0xBEEF) else "fail",
            "detail": {
                "dir": _relpath(inject_dir),
                "blob": _relpath(SDK_ELF),
                "slot": 6,
                "ping_result": value,
                "expected": 0x41 ^ 0xBEEF,
                "hashes": session_hashes(inject_dir),
            },
        }
    else:
        matrix["injection"] = {
            "status": "not_run",
            "reason": "sdk/build/injected_demo.elf not built (run make sdk)",
        }

    matrix["mcp"] = {
        "status": "not_run",
        "reason": (
            "native MCP surface is S3's deliverable (plugins/mcp_server.py); "
            "S4 must not race that work. Re-run after S3 lands (docs/37)."
        ),
    }
    analyze_dir = os.path.join(out_dir, "sessions", "native-1")
    compare_dir = os.path.join(out_dir, "sessions", "native-2")
    if os.path.isfile(os.path.join(analyze_dir, "session.json")) and os.path.isfile(
        os.path.join(compare_dir, "session.json")
    ):
        try:
            from greeneye.analyze import render_summary

            summary = render_summary(analyze_dir)
            delta_text = compare_sessions(analyze_dir, compare_dir)
            matrix["analyze_compare"] = {
                "status": "pass" if summary and delta_text else "fail",
                "detail": {
                    "analyze_session": _relpath(analyze_dir),
                    "analyze_chars": len(summary),
                    "compare_pair": [_relpath(analyze_dir), _relpath(compare_dir)],
                    "compare_chars": len(delta_text),
                    "compare_headers": [
                        line[3:].strip()
                        for line in delta_text.splitlines()
                        if line.startswith("## ")
                    ],
                },
            }
        except Exception as exc:  # evidence step, never fatal to the matrix
            matrix["analyze_compare"] = {
                "status": "fail",
                "reason": f"{type(exc).__name__}: {exc}",
            }
    else:
        matrix["analyze_compare"] = {
            "status": "not_run",
            "reason": "native-1/native-2 sessions absent",
        }
    return {"ran": True, "reason": "ok", "features": matrix}


# -- report ----------------------------------------------------------------


def threshold_checks(report: dict, thresholds: dict | None = None) -> dict:
    """Evaluate the frozen thresholds against a report; returns check -> bool."""
    th = dict(FROZEN_THRESHOLDS)
    th.update(thresholds or {})
    qemu = report.get("qemu") or {}
    native = report.get("native") or {}
    checks: dict = {
        "qemu_first_fault_reproduced": bool(
            qemu.get("ran") and qemu.get("first_touch_match")
        ),
        "native_above_qemu": above_qemu_line(
            qemu.get("rung"), native.get("rung"), int(th["native_rungs_above_qemu"])
        ),
        "identical_runs": bool(
            native.get("ran")
            and native.get("repeat", 0) >= int(th["identical_runs"])
            and native.get("identical")
        ),
        "native_pair_wall_s": bool(
            native.get("ran")
            and float(native.get("pair_wall_s", 1e9)) <= float(th["native_pair_wall_s"])
        ),
        "device_model_documented": bool(
            not th.get("device_model_required", True) or native.get("device_model_used")
        ),
    }
    checks["all"] = all(checks.values())
    return checks


def build_report(
    *,
    image: str,
    qemu: dict,
    native: dict,
    features: dict | None,
    s1: dict,
    thresholds: dict,
) -> dict:
    checks = threshold_checks({"qemu": qemu, "native": native}, thresholds)
    rung_progress = bool(checks["native_above_qemu"] and checks["identical_runs"])
    if not qemu.get("ran"):
        verdict = "unverified_qemu_absent"
    elif not checks["qemu_first_fault_reproduced"]:
        verdict = "unverified_qemu_mismatch"
    elif checks["all"]:
        verdict = "above"
    elif rung_progress and not checks["device_model_documented"]:
        # Rung progress exists, but docs/35 1.1 also requires working device
        # semantics; lenient MMIO cannot satisfy the line contract.  Underclaim.
        verdict = "above_lenient_only"
    else:
        verdict = "not_above"
    return {
        "schema": SCHEMA,
        "harness": "harness/sim_qemu_line.py",
        "line_definition": "docs/35-GREENEYE-SIMULATOR-DESIGN.md 1.1 + 3.3",
        "firmware": {
            "name": os.path.basename(image),
            "sha256": firmware_sha256(image) if os.path.isfile(image) else None,
        },
        "rung_order": list(RUNG_ORDER),
        "thresholds": thresholds,
        "threshold_checks": checks,
        "verdict": verdict,
        "verdict_explanation": {
            "above": "all frozen checks passed: QEMU fault reproduced, native >= QEMU + 1 rung, identical runs, wall bound, documented device model",
            "above_lenient_only": "native progressed past the QEMU fault, but lenient MMIO is not a documented device model (docs/35 1.1) - contract NOT satisfied",
            "not_above": "one or more frozen checks failed and no rung-progress excuse applies",
            "unverified_qemu_absent": "QEMU baseline unavailable; no A/B comparison is possible",
            "unverified_qemu_mismatch": "QEMU ran but did not reproduce its documented first fault; the rungs are not comparable",
        }.get(verdict, ""),
        "rung_progress": rung_progress,
        "above_qemu_line": verdict == "above",
        "qemu": {k: v for k, v in qemu.items() if k != "device_map"},
        "native": native,
        "features": features,
        "s1_integration": {
            "source": s1.get("source"),
            "profile": _relpath(s1.get("profile")),
            "declared_rung": s1.get("declared_rung"),
            "notes": s1.get("notes", []),
        },
        "x86_vms_running": _x86_vms_running(),
        "not_verified": [
            "lenient MMIO (B0) is observation, not device semantics",
            "B3/B4 are not reached: the S1 device model stops at the B2 DMEM-runspace unknown",
            "the boot shim values (bootargs/WPR meta/DMEM mirror) are declared assumptions",
            "MCP on native is S3-owned and not exercised here",
            "L2 timing: cycles are modeled (docs/32), never silicon claims",
            "no hardware/GPU/guest/module involvement by design",
        ],
        "rerun_when_s1_lands": [
            "python3 harness/sim_qemu_line.py",
            "S1 landed (2026-10-03): sim/profiles/*.json is discovered and the native side runs sim.gsp_boot",
            "sim/devices/*.json with greeneye-device-map/1 (or a device_map key) is still picked up as an observational map",
        ],
    }


def render_markdown(report: dict) -> str:
    qemu = report.get("qemu") or {}
    native = report.get("native") or {}
    lines = [
        "# Simulator vs the QEMU line - measured report",
        "",
        f"* schema: `{report.get('schema')}`",
        f"* harness: `{report.get('harness')}`",
        f"* line: {report.get('line_definition')}",
        f"* firmware: `{(report.get('firmware') or {}).get('name')}` "
        f"sha256 `{((report.get('firmware') or {}).get('sha256') or '')[:16]}...`",
        f"* verdict: **{report.get('verdict')}** "
        f"(above_qemu_line={report.get('above_qemu_line')})",
        f"* verdict meaning: {report.get('verdict_explanation') or '-'}",
        "",
        "## Rung table",
        "",
        "| side | ran | rung | instructions | cycles (modeled) | stop | wall s | evidence |",
        "|---|---|---|---:|---:|---|---:|---|",
    ]
    runs = native.get("runs") or []
    lines.append(
        f"| QEMU probe | {qemu.get('ran')} | {qemu.get('rung')} | - | - | "
        f"{'fault' if qemu.get('first_fault') else '-'} | {qemu.get('wall_s')} | "
        f"`{qemu.get('session') or qemu.get('reason')}` |"
    )
    for index, run in enumerate(runs, start=1):
        stop = run.get("stop") or {}
        pc_fw = stop.get("pc_fw")
        pc_fw_text = f"{pc_fw:#x}" if pc_fw is not None else "-"
        lines.append(
            f"| native run {index} | True | {run.get('rung')} | {run.get('instructions')} | "
            f"{run.get('cycles')} | {stop.get('reason')}@{pc_fw_text} | "
            f"{run.get('wall_s')} | `{run.get('session')}` |"
        )
    if not runs:
        lines.append(f"| native | {native.get('ran')} | - | - | - | - | - | {native.get('reason')} |")

    first = qemu.get("first_fault") or {}
    lines += [
        "",
        "### QEMU first fault",
        "",
        f"* {first.get('desc')} at fw vaddr `{first.get('epc_fw') if first.get('epc_fw') is not None else 0:#x}` "
        f"accessing `{(first.get('access_addr') or 0):#x}` ({first.get('size')} bytes, {first.get('rw')})",
        f"* first-touch matches the documented reference: {qemu.get('first_touch_match')}",
        f"* fault records in the capped log window: {qemu.get('fault_count_window')}",
        "",
        "## Frozen thresholds",
        "",
        "| check | pass |",
        "|---|---|",
    ]
    for name, value in (report.get("threshold_checks") or {}).items():
        lines.append(f"| {name} | {value} |")
    lines += [
        "",
        f"Thresholds (frozen {report['thresholds'].get('frozen_on')}): "
        f"native >= QEMU + {report['thresholds'].get('native_rungs_above_qemu')} rung(s); "
        f"{report['thresholds'].get('identical_runs')} identical runs; "
        f"pair wall <= {report['thresholds'].get('native_pair_wall_s')} s.",
        "",
        "## Determinism",
        "",
        f"* identical across {native.get('repeat')} native runs: {native.get('identical')}",
        f"* session compare ran: {(native.get('compare') or {}).get('ran')} "
        f"({(native.get('compare') or {}).get('headers')})",
        "",
        "## Session hashes (sha256)",
        "",
        "| session | file | sha256 |",
        "|---|---|---|",
    ]
    hash_rows = [("qemu-probe", qemu.get("session"), qemu.get("session_hashes") or {})]
    for index, run in enumerate(runs, start=1):
        hash_rows.append((f"native-{index}", run.get("session"), run.get("session_hashes") or {}))
    features_detail = ((report.get("features") or {}).get("features") or {})
    for feature in ("sessions", "trace", "injection"):
        detail = (features_detail.get(feature) or {}).get("detail") or {}
        if isinstance(detail, dict):
            hashes = detail.get("hashes") or {}
            if hashes:
                hash_rows.append((feature, detail.get("dir") or detail.get("blob"), hashes))
    for label, session, hashes in hash_rows:
        if not hashes:
            lines.append(f"| {label} | - | (no hashes recorded) |")
            continue
        for name in sorted(hashes):
            lines.append(f"| {label} | {name} | `{hashes[name]}` |")
    lines += [
        "",
        "## Native feature matrix (demo firmware)",
        "",
        "| feature | status | detail |",
        "|---|---|---|",
    ]
    feature_block = (report.get("features") or {}).get("features") or {}
    for name in sorted(feature_block):
        entry = feature_block[name]
        detail = entry.get("detail") or entry.get("reason") or ""
        lines.append(f"| {name} | {entry.get('status')} | {detail} |")
    if not feature_block:
        lines.append(f"| - | not_run | {(report.get('features') or {}).get('reason')} |")
    lines += [
        "",
        "## S1 integration",
        "",
        f"* device source: {report['s1_integration'].get('source') or 'lenient (no S1 device model found)'}",
        f"* profile: {report['s1_integration'].get('profile') or '-'}",
        f"* declared rung: {report['s1_integration'].get('declared_rung')}",
    ]
    for note in report["s1_integration"].get("notes") or []:
        lines.append(f"* note: {note}")
    lines += [
        "",
        "## Not verified",
        "",
    ]
    for item in report.get("not_verified") or []:
        lines.append(f"* {item}")
    lines += [
        "",
        "## S1 integration notes",
        "",
    ]
    for item in report.get("rerun_when_s1_lands") or []:
        lines.append(f"* `{item}`" if item.startswith("python3") else f"* {item}")
    lines += [
        "",
        f"* other agents' x86 QEMU VMs observed (never touched): {report.get('x86_vms_running')}",
        "",
    ]
    return "\n".join(lines)


def run_harness(
    *,
    image: str = DEFAULT_IMAGE,
    out_dir: str = DEFAULT_OUT,
    instructions: int = DEFAULT_INSTRUCTIONS,
    qemu_duration: float = DEFAULT_QEMU_DURATION_S,
    repeat: int = 2,
    device_map: str | None = None,
    features: bool = True,
    thresholds: dict | None = None,
    run_qemu: bool = True,
) -> dict:
    """Run the full A/B harness and write ``sim-qemu-line.json`` + ``.md``."""
    os.makedirs(out_dir, exist_ok=True)
    frozen = dict(FROZEN_THRESHOLDS)
    frozen.update(thresholds or {})

    if device_map is not None:
        dm = DeviceMap.load(device_map)
        s1 = {
            "device_map": dm,
            "profile": None,
            "declared_rung": None,
            "source": _relpath(device_map),
            "notes": ["explicit --device-map override"],
        }
    else:
        sha = firmware_sha256(image) if os.path.isfile(image) else None
        s1 = discover_s1_device_model(image, sha)

    qemu = (
        run_qemu_baseline(image=image, out_dir=out_dir, duration=qemu_duration)
        if run_qemu
        else {"backend": "qemu", "ran": False, "reason": "qemu=False", "rung": None}
    )

    native = run_native_baseline(
        image=image,
        out_dir=out_dir,
        instructions=instructions,
        repeat=repeat,
        device_map=s1.get("device_map"),
        declared_rung=s1.get("declared_rung"),
        profile=s1.get("profile"),
    )
    if not native.get("ran"):
        report = build_report(
            image=image, qemu=qemu, native=native, features=None, s1=s1, thresholds=frozen
        )
    else:
        feature_result = run_feature_matrix(out_dir) if features else None
        report = build_report(
            image=image,
            qemu=qemu,
            native=native,
            features=feature_result,
            s1=s1,
            thresholds=frozen,
        )

    json_path = _write_json(os.path.join(out_dir, "sim-qemu-line.json"), report)
    md_path = _write_text(os.path.join(out_dir, "sim-qemu-line.md"), render_markdown(report))
    report["report_json"] = _relpath(json_path)
    report["report_md"] = _relpath(md_path)
    return report


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="python3 harness/sim_qemu_line.py",
        description=(
            "S4 A/B: QEMU probe vs native GSP simulator on the real 570.144 image; "
            "emits sim-qemu-line.json + .md. Bounded, no hardware."
        ),
    )
    parser.add_argument("--image", default=DEFAULT_IMAGE, help="flat GSP firmware image")
    parser.add_argument("--out", default=DEFAULT_OUT, help="report directory")
    parser.add_argument(
        "--instructions", type=int, default=DEFAULT_INSTRUCTIONS,
        help=f"native instruction budget per run (<= {MAX_INSTRUCTIONS})",
    )
    parser.add_argument(
        "--qemu-duration", type=float, default=DEFAULT_QEMU_DURATION_S,
        help=f"QEMU wall-clock probe bound (<= {MAX_QEMU_DURATION_S}s)",
    )
    parser.add_argument("--repeat", type=int, default=2, help="native runs for determinism")
    parser.add_argument("--device-map", help="explicit greeneye-device-map/1 file (skips discovery)")
    parser.add_argument("--no-qemu", action="store_true", help="native only (no baseline claim)")
    parser.add_argument("--no-features", action="store_true", help="skip the feature matrix")
    parser.add_argument("--json", action="store_true", help="print the report JSON")
    args = parser.parse_args(argv)

    try:
        report = run_harness(
            image=args.image,
            out_dir=args.out,
            instructions=args.instructions,
            qemu_duration=args.qemu_duration,
            repeat=args.repeat,
            device_map=args.device_map,
            features=not args.no_features,
            run_qemu=not args.no_qemu,
        )
    except (HarnessError, GreenEyeError) as exc:
        print(f"[sim-qemu-line] error: {exc}", file=sys.stderr)
        return 1

    if not os.path.isfile(args.image):
        print(f"[sim-qemu-line] skipped: firmware absent: {args.image}", file=sys.stderr)
        return 2
    if args.json:
        print(json.dumps(report, indent=1, sort_keys=True, default=str))
    native = report.get("native") or {}
    print(f"[sim-qemu-line] qemu rung: {(report.get('qemu') or {}).get('rung')}")
    print(f"[sim-qemu-line] native rung: {native.get('rung')} (identical={native.get('identical')})")
    print(f"[sim-qemu-line] verdict: {report.get('verdict')}")
    print(f"[sim-qemu-line] thresholds: {report.get('threshold_checks')}")
    print(f"[sim-qemu-line] report: {report.get('report_json')} + {report.get('report_md')}")
    return 0 if report.get("above_qemu_line") else 1


if __name__ == "__main__":
    sys.exit(main())