#!/usr/bin/env python3
"""s08 - mutation triage: mutate one field, classify the outcome.

Real scenario (docs/18-HARNESS-AUTOMATION.md R5 and
Handoff-vGPU-IO.md PART 77 / docs/19 3.4, 2026-09-30..2026-10-02): the
error-path corpus (sentinel/boundary pointers on ECHO/SUM, invalid command
IDs) is how the emulation's fail-closed rules are derived.  Statically the
demo's 4-of-24 outcomes were invisible; the discovery sweep found
``rpc_echo``/``rpc_sum`` fault on sentinel/boundary pointers while
``rpc_ping``/``rpc_state``/``rpc_fill`` stay on the handler path.

Analog: a seeded mutation table over command/arguments, one fresh session per
attempt (the documented safety strategy), classified as ok / trap / fault /
no-progress from stop signals, ``rpc_count`` and the result field.
"""
from __future__ import annotations

import json
import os

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

FAULT_SIGNALS = {4, 7, 11}  # SIGILL, SIGBUS, SIGSEGV
TRAP_SIGNAL = 5  # SIGTRAP (ebreak)


def _probe(ge, mutation: dict, backend: str) -> dict:
    session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s08")
    try:
        ctrl = session.symbols.lookup("gsp_ctrl")
        signals = []
        session.on_event("stop", lambda hit, sig=signals: sig.append(hit.stop.signal) if hit.stop else None)
        session.run(duration=0.15)
        literal = 0
        if mutation.get("arg0") == "hello":
            hits = session.memory.search(0x80000000, 0x2000, b"hello-green-eye")
            literal = hits[0] if hits else ctrl + C.CTRL_BUF
        before = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT)
        session.memory.write_u64(ctrl + C.CTRL_ARG0, literal if mutation.get("arg0") == "hello" else mutation.get("arg0", 0))
        session.memory.write_u64(ctrl + C.CTRL_ARG1, mutation.get("arg1", 0))
        session.memory.write_u64(ctrl + C.CTRL_MAILBOX, mutation["cmd"])
        # max_stops=1: a faulted target keeps re-faulting at the same PC, so
        # stop at the first signal instead of spinning the fault loop.
        session.run(duration=0.2, max_stops=1)
        after = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT)
        if after == before and not any(signals):
            session.run(duration=0.3, max_stops=1)
            after = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT)
        result = session.memory.read_u64(ctrl + C.CTRL_RESULT)
        if any(s == TRAP_SIGNAL for s in signals):
            outcome = "trap"
        elif any(s in FAULT_SIGNALS for s in signals):
            outcome = "fault"
        elif after > before:
            outcome = "ok"
        else:
            outcome = "no-progress"
        return {
            "outcome": outcome,
            "rpc_count_delta": after - before,
            "result": result,
            "result_hex": f"{result:#x}",
            "signals": [s for s in signals if s],
            "consumed": after > before,
        }
    finally:
        session.close()


def run() -> dict:
    import greeneye as ge

    out = C.artifact_dir("s08_mutation_triage")
    backend = C.choose_backend("native")
    mutations = [
        {"strategy": "arg-boundary", "cmd": 2, "arg0": 0, "arg1": 0, "expect": "fault"},
        {"strategy": "valid", "cmd": 2, "arg0": "hello", "arg1": 0, "expect": "ok"},
        {"strategy": "arg-boundary", "cmd": 2, "arg0": 0xFFFFFFFFFFFFFFFF, "arg1": 0, "expect": "fault"},
        {"strategy": "valid", "cmd": 4, "arg0": "hello", "arg1": 5, "expect": "ok"},
        {"strategy": "buffer-length", "cmd": 4, "arg0": "hello", "arg1": 0x8000000000000000, "expect": "no-progress"},
        {"strategy": "buffer-length", "cmd": 3, "arg0": 0x41, "arg1": 4096, "expect": "ok"},
        {"strategy": "cmd-invalid", "cmd": 7, "arg0": 0, "arg1": 0, "expect": "ok"},
    ]
    commands = [
        "for each seeded mutation: fresh native session, settle, one mailbox write",
        "run bounded; classify from stop signals + rpc_count + result",
        "write findings.jsonl (deterministic order) for corpus diffing",
    ]
    scenario = C.scenario(
        ids=[
            "docs/18-HARNESS-AUTOMATION.md 2 (R5): bounded seeded mutation, outcome classes",
            "docs/19-CONTROL-DISCOVERY.md 3.4 (2026-10-02): sentinel/boundary pointer faults found by the sweep",
            "Handoff-vGPU-IO.md PARTs 3-9 (2026-09-21..23): hand-derived assert/error paths",
        ],
        baseline=(
            "statically the fault surface is invisible; the project derived error "
            "paths by hand from asserts and NOCAT records across many physical "
            "cycles; a 64-iteration QEMU mutation run is the tracked baseline"
        ),
        projection=(
            "real RM controls land in the same mutation harness once the RM "
            "surface exists (docs/19 section 5); one device-write experiment "
            "per physical cycle is the current cap"
        ),
        blocked=True,
        device="H1.3 RM control surface for mutation of real controls",
    )
    C.render_banner(
        "s08_mutation_triage",
        "Mutate one input field deterministically and classify the outcome (ok/trap/fault/no-progress).",
        "error-path corpus for fail-closed emulation rules (R5/H4.3)",
        scenario["static_baseline"],
        commands,
    )

    findings = []
    for index, mutation in enumerate(mutations):
        observed = _probe(ge, mutation, backend)
        finding = {"iteration": index, **mutation, **observed, "matched_expectation": observed["outcome"] == mutation["expect"]}
        findings.append(finding)
    classes: dict[str, int] = {}
    for finding in findings:
        classes[finding["outcome"]] = classes.get(finding["outcome"], 0) + 1
    payload = {
        "success": all(f["matched_expectation"] for f in findings),
        "backend": backend,
        "mutations": len(findings),
        "outcome_classes": classes,
        "findings": findings,
        "question": "enumerate and classify one-field mutations without hardware",
        "answer": (
            f"{len(findings)} mutations -> {classes}; all expectations matched; "
            "faults localised to the dereferencing handlers, the oversized SUM "
            "stays no-progress, FILL clamps at the buffer size - a corpus entry "
            "per mutation with no device writes"
        ),
        "success_criteria": {
            "pass": "every seeded mutation produces its documented outcome class",
            "fail": "an outcome contradicts the case table or a session leaks",
        },
        "real_case": scenario,
    }
    with open(os.path.join(out, "findings.jsonl"), "w", encoding="utf-8") as fh:
        for finding in findings:
            fh.write(json.dumps(finding, sort_keys=True) + "\n")
    C.write_json(os.path.join(out, "mutation.json"), payload)
    return payload


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


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