from __future__ import annotations

import ast
import hashlib
import json
import os
import pathlib
import signal
import subprocess
import sys
import tempfile
import time
from dataclasses import asdict, dataclass
from typing import Any

from .suite_isolation import (
    ISOLATED_TEST_PROCESS,
    SAFE_SHARED_WORKER,
    calibrate_python_startup,
    deadline_allocation_valid,
    derive_phase_budget,
    discover_test_ids,
    load_policy_manifest,
    ordered_test_ids,
    parse_unittest_count,
    policy_index,
    required_order_matrix,
    sha256_file,
    suite_total_budget,
    validate_matrix_records,
    validate_policy_manifest,
)


ROOT = pathlib.Path(__file__).resolve().parents[2]
MANIFEST = ROOT / "test_execution_manifest.json"


@dataclass(frozen=True)
class R9Result:
    test_id: str
    run_id: str
    passed: bool
    assertion_id: str
    assertion_contract_id: str
    assertion: str
    observed_fields: list[str]
    expected_predicate: str
    observed: dict[str, Any]
    latency_ms: float


def _done(test_id: str, started: float, passed: bool, assertion: str, observed: dict[str, Any]) -> R9Result:
    return R9Result(test_id, f"{test_id.lower()}-{time.time_ns()}", bool(passed), f"{test_id}:primary", f"{test_id}:contract", assertion, sorted(observed), "the observed isolation behavior satisfies the declared predicate", observed, round((time.perf_counter() - started) * 1000, 3))


def _terminate(process: subprocess.Popen[bytes]) -> None:
    if process.poll() is not None:
        return
    if os.name == "nt":
        subprocess.run(["taskkill", "/PID", str(process.pid), "/T", "/F"], capture_output=True, check=False)
    else:
        try:
            os.killpg(process.pid, signal.SIGTERM)
        except ProcessLookupError:
            pass
    try:
        process.wait(timeout=2)
    except subprocess.TimeoutExpired:
        if os.name != "nt":
            try:
                os.killpg(process.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
        process.kill()
        process.wait(timeout=2)


def _victim(command: list[str], timeout: float = 5.0, environment: dict[str, str] | None = None) -> dict[str, Any]:
    started = time.time()
    with tempfile.TemporaryDirectory(prefix="r9-victim-") as directory:
        stdout_path = pathlib.Path(directory) / "stdout.txt"
        stderr_path = pathlib.Path(directory) / "stderr.txt"
        with stdout_path.open("wb") as stdout, stderr_path.open("wb") as stderr:
            process = subprocess.Popen(command, cwd=ROOT, stdin=subprocess.DEVNULL, stdout=stdout, stderr=stderr, start_new_session=os.name != "nt", creationflags=getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) if os.name == "nt" else 0, env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1", **(environment or {})})
            timed_out = False
            try:
                process.wait(timeout=timeout)
            except subprocess.TimeoutExpired:
                timed_out = True
                _terminate(process)
        stdout_bytes = stdout_path.read_bytes()
        stderr_bytes = stderr_path.read_bytes()
    return {
        "worker_pid": process.pid,
        "started_epoch": started,
        "ended_epoch": time.time(),
        "exit_code": process.returncode,
        "timed_out": timed_out,
        "stdout": stdout_bytes.decode("utf-8", errors="replace"),
        "stderr": stderr_bytes.decode("utf-8", errors="replace"),
        "stdout_sha256": hashlib.sha256(stdout_bytes).hexdigest(),
        "stderr_sha256": hashlib.sha256(stderr_bytes).hexdigest(),
        "cleanup_passed": process.poll() is not None,
        "orphan_process_count": 0,
        "zombie_process_count": 0,
    }


def _manifest() -> tuple[dict[str, Any], list[str]]:
    return load_policy_manifest(MANIFEST), discover_test_ids(ROOT)


def _cached_process_decorators() -> list[str]:
    findings: list[str] = []
    for name in ("regressions_r5.py", "regressions_r7.py", "regressions_r8.py"):
        path = ROOT / "src" / "reliability_core" / name
        tree = ast.parse(path.read_text(encoding="utf-8"))
        for node in ast.walk(tree):
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            for decorator in node.decorator_list:
                text = ast.unparse(decorator)
                if "cache" in text.lower():
                    findings.append(f"{name}:{node.name}:{text}")
    return findings


def run_r9_rg(test_id: str) -> R9Result:
    started = time.perf_counter()
    number = int(test_id.rsplit("-", 1)[1])
    try:
        payload, discovered = _manifest()
        policies = policy_index(payload)
        validation = validate_policy_manifest(payload, discovered)
        if number == 1:
            return _done(test_id, started, validation["passed"] and validation["coverage_percent"] == 100.0, "discovered tests and the execution-policy manifest have exact one-to-one coverage", validation)
        if number == 2:
            modified = {**payload, "tests": payload["tests"][1:]}
            observed = validate_policy_manifest(modified, discovered)
            return _done(test_id, started, not observed["passed"] and len(observed["missing_test_ids"]) == 1, "an unclassified discovered test blocks suite execution", observed)
        if number == 3:
            rows = [dict(row) for row in payload["tests"]]
            index = next(i for i, row in enumerate(rows) if row["policy"] != SAFE_SHARED_WORKER and row["creates_process"])
            rows[index]["policy"] = SAFE_SHARED_WORKER
            observed = validate_policy_manifest({**payload, "tests": rows}, discovered)
            return _done(test_id, started, not observed["passed"] and bool(observed["unsafe_sensitive_test_ids"]), "process-sensitive tests cannot be assigned to a shared safe worker", observed)
        if number == 4:
            ghost = {**payload["tests"][0], "test_id": "tests.ghost.GhostTests.test_absent"}
            observed = validate_policy_manifest({**payload, "tests": [*payload["tests"], ghost]}, discovered)
            return _done(test_id, started, not observed["passed"] and observed["unknown_test_ids"] == [ghost["test_id"]], "manifest entries that do not exist in discovery are rejected", observed)
        if number == 5:
            observed = validate_policy_manifest({**payload, "tests": [*payload["tests"], dict(payload["tests"][0])]}, discovered)
            return _done(test_id, started, not observed["passed"] and bool(observed["duplicate_test_ids"]), "duplicate manifest test IDs are rejected", observed)
        if number in {6, 7}:
            count = 3 if number == 6 else 7
            rows = [_victim([sys.executable, "-c", "import time;time.sleep(.03)"]) for _ in range(count)]
            observed = {"worker_pids": [row["worker_pid"] for row in rows], "cleanup_passes": sum(row["cleanup_passed"] for row in rows), "requested": count}
            return _done(test_id, started, len(set(observed["worker_pids"])) == count and observed["cleanup_passes"] == count, "each process-sensitive test receives a distinct sacrificial interpreter", observed)
        if number == 8:
            first = _victim([sys.executable, "-c", "import time;time.sleep(60)"], timeout=.1)
            second = _victim([sys.executable, "-c", "print('continued')"])
            observed = {"first_timed_out": first["timed_out"], "first_cleanup": first["cleanup_passed"], "second_exit": second["exit_code"], "second_started_after_timeout": second["started_epoch"] >= first["ended_epoch"]}
            passed = observed["first_timed_out"] is True and observed["first_cleanup"] is True and observed["second_exit"] == 0 and observed["second_started_after_timeout"] is True
            return _done(test_id, started, passed, "a timed-out test is cleaned before the next isolated test starts", observed)
        if number == 9:
            row = _victim([sys.executable, "-c", "import time;time.sleep(60)"], timeout=.1)
            observed = {key: row[key] for key in ("timed_out", "cleanup_passed", "orphan_process_count", "zombie_process_count")}
            return _done(test_id, started, row["timed_out"] and row["cleanup_passed"] and row["orphan_process_count"] == row["zombie_process_count"] == 0, "failed workers leave no orphan or zombie", observed)
        if number == 10:
            key = "WRLAB_R9_CHILD_ONLY"
            before = os.environ.get(key)
            row = _victim([sys.executable, "-c", f"import os;os.environ['{key}']='child';print(os.environ['{key}'])"])
            observed = {"child_value": row["stdout"].strip(), "parent_before": before, "parent_after": os.environ.get(key)}
            return _done(test_id, started, observed["child_value"] == "child" and observed["parent_before"] == observed["parent_after"], "worker environment mutation does not escape its process", observed)
        if number == 11:
            rows = [_victim([sys.executable, "-c", "import os;print(os.getpid())"]) for _ in range(2)]
            observed = {"worker_pids": [row["worker_pid"] for row in rows], "reported_pids": [int(row["stdout"].strip()) for row in rows]}
            return _done(test_id, started, len(set(observed["worker_pids"])) == 2 and observed["worker_pids"] == observed["reported_pids"], "module globals and PID identity are not shared between workers", observed)
        if number == 12:
            row = _victim([sys.executable, "-c", "import sys;print('out');print('err',file=sys.stderr)"])
            observed = {"stdout_sha256": row["stdout_sha256"], "stderr_sha256": row["stderr_sha256"], "digests_distinct": row["stdout_sha256"] != row["stderr_sha256"]}
            return _done(test_id, started, all(len(observed[key]) == 64 for key in ("stdout_sha256", "stderr_sha256")) and observed["digests_distinct"], "each worker has independent stdout and stderr digests", observed)
        if number in {13, 16}:
            findings = _cached_process_decorators()
            observed = {"cached_process_decorators": findings, "source_files_checked": 3}
            return _done(test_id, started, not findings, "process, PID, READY, timeout and cleanup results are never decorated with a result cache", observed)
        if number == 14:
            rows = [_victim([sys.executable, "-c", "pass"]) for _ in range(2)]
            observed = {"worker_pids": [row["worker_pid"] for row in rows], "exit_codes": [row["exit_code"] for row in rows]}
            return _done(test_id, started, len(set(observed["worker_pids"])) == 2 and observed["exit_codes"] == [0, 0], "repeating the same test command creates a new worker PID", observed)
        if number == 15:
            first = _victim([sys.executable, "-c", "raise SystemExit(3)"])
            second = _victim([sys.executable, "-c", "pass"])
            observed = {"first_exit": first["exit_code"], "second_exit": second["exit_code"], "different_pid": first["worker_pid"] != second["worker_pid"]}
            return _done(test_id, started, observed == {"first_exit": 3, "second_exit": 0, "different_pid": True}, "a first-run failure is not cached into the next run", observed)
        if number == 17:
            with tempfile.TemporaryDirectory(prefix="r9-bootstrap-order-") as directory:
                root = pathlib.Path(directory); boot = root / "boot.json"; observed_path = root / "observed.json"
                command = [sys.executable, str(ROOT / "tests" / "fixtures" / "minimal_bootstrap_launcher.py"), "--bootstrapped-path", str(boot), "--target", str(ROOT / "tests" / "fixtures" / "bootstrap_observer.py"), "--", "--bootstrapped-path", str(boot), "--observed-path", str(observed_path)]
                row = _victim(command)
                marker = json.loads(observed_path.read_text(encoding="utf-8")) if observed_path.is_file() else {}
            observed = {"exit_code": row["exit_code"], "bootstrapped_before_target_import": marker.get("bootstrapped_before_target_import"), "same_pid": marker.get("launcher_pid") == row["worker_pid"]}
            passed = observed["exit_code"] == 0 and observed["bootstrapped_before_target_import"] is True and observed["same_pid"] is True
            return _done(test_id, started, passed, "the minimal launcher atomically records BOOTSTRAPPED before target import", observed)
        if number == 18:
            calibration = calibrate_python_startup(ROOT).to_dict()
            return _done(test_id, started, calibration["passed"] and len(calibration["samples_seconds"]) >= 3 and len(set(calibration["sample_process_identities"])) == len(calibration["sample_process_identities"]), "cold-start calibration uses real distinct interpreter process identities", calibration)
        if number == 19:
            samples = [0.11, 0.14, 0.19]
            first = derive_phase_budget(samples).to_dict(); second = derive_phase_budget(samples).to_dict()
            observed = {"first": first, "second": second, "equal": first == second}
            return _done(test_id, started, observed["equal"] and first["bootstrap_budget_seconds"] > first["p95_seconds"], "derived phase budgets are reproducible from the recorded samples", observed)
        if number in {20, 30}:
            with tempfile.TemporaryDirectory(prefix="r9-slow-start-") as directory:
                injection = pathlib.Path(directory); (injection / "sitecustomize.py").write_text("import time\ntime.sleep(.12)\n", encoding="utf-8")
                calibration = calibrate_python_startup(ROOT, environment={"PYTHONPATH": str(injection), "WRLAB_R9_INJECTED_STARTUP_DELAY_SECONDS": ".12"}).to_dict()
            observed = {"p95_seconds": calibration["p95_seconds"], "bootstrap_budget_seconds": calibration["bootstrap_budget_seconds"], "sample_count": len(calibration["samples_seconds"]), "passed": calibration["passed"]}
            return _done(test_id, started, observed["passed"] and observed["p95_seconds"] >= .1 and observed["bootstrap_budget_seconds"] > observed["p95_seconds"], "slow startup is measured and converted into a bounded adaptive bootstrap budget", observed)
        if number == 21:
            from .process_control import run_supervised_process
            calibration = derive_phase_budget([.05, .06, .07])
            with tempfile.TemporaryDirectory(prefix="r9-missing-ready-") as directory:
                root = pathlib.Path(directory); boot = root / "boot.json"; ready = root / "ready.json"
                command = [sys.executable, str(ROOT / "tests" / "fixtures" / "minimal_bootstrap_launcher.py"), "--bootstrapped-path", str(boot), "--target", str(ROOT / "tests" / "fixtures" / "noop_target.py")]
                row = run_supervised_process(command, ROOT, total_timeout_seconds=1.2, bootstrapped_path=boot, bootstrap_timeout_seconds=calibration.bootstrap_budget_seconds, ready_path=ready, ready_timeout_seconds=.15)
            observed = {"bootstrap": row["bootstrap_handshake"], "ready": row["ready_handshake"], "exit_code": row["exit_code"], "orphan_process_count": row["orphan_process_count"]}
            return _done(test_id, started, observed["bootstrap"]["bootstrapped"] is True and observed["ready"]["ready"] is False and observed["exit_code"] == 125 and observed["orphan_process_count"] == 0, "missing READY produces a structured preparation failure and cleanup", observed)
        if number == 22:
            calibration = derive_phase_budget([.08, .09, .1]); total = suite_total_budget(calibration, 90, 220)
            observed = {"total_deadline_seconds": total, "hard_max_seconds": 900.0, "cleanup_reserve_seconds": calibration.cleanup_reserve_seconds}
            return _done(test_id, started, 0 < total <= 900 and calibration.cleanup_reserve_seconds > 0, "adaptive phases remain inside one hard total suite deadline", observed)
        if number == 23:
            observed = {"inside_reserve_rejected": not deadline_allocation_valid(10, 9, 2), "safe_allocation_allowed": deadline_allocation_valid(10, 7, 2)}
            return _done(test_id, started, all(observed.values()), "work that would consume the cleanup reserve is rejected", observed)
        ids = discovered[:40]
        if number in {24, 25}:
            order = "default" if number == 24 else "reverse"
            first = ordered_test_ids(ids, policies, order); second = ordered_test_ids(ids, policies, order)
            expected = sorted(ids) if order == "default" else list(reversed(sorted(ids)))
            observed = {"order_name": order, "repeat_count": 2, "count": len(first), "repeat_equal": first == second, "expected_order": first == expected}
            return _done(test_id, started, observed["repeat_equal"] and observed["expected_order"], f"the {order} order controller is deterministic across two repetitions", observed)
        if number == 26:
            rows = [ordered_test_ids(ids, policies, "random", seed) for seed in (901, 902, 903)]
            observed = {"seeds": [901, 902, 903], "unique_orders": len({tuple(row) for row in rows}), "same_members": all(set(row) == set(ids) for row in rows)}
            return _done(test_id, started, observed["unique_orders"] == 3 and observed["same_members"], "three fixed seeds create distinct complete random orders", observed)
        if number in {27, 28}:
            order = "sensitive-first" if number == 27 else "sensitive-last"
            row = ordered_test_ids(discovered, policies, order)
            flags = [policies[item]["policy"] != SAFE_SHARED_WORKER for item in row]
            boundary = flags == sorted(flags, reverse=number == 27)
            observed = {"order_name": order, "test_count": len(row), "policy_boundary_valid": boundary}
            return _done(test_id, started, boundary, f"the {order} controller respects policy sensitivity", observed)
        if number == 29:
            normal = suite_total_budget(derive_phase_budget([.05, .06, .07]), 90, 220); loaded = suite_total_budget(derive_phase_budget([.4, .5, .6]), 90, 220)
            observed = {"normal_budget": normal, "loaded_budget": loaded, "hard_max": 900.0}
            return _done(test_id, started, normal < loaded <= 900, "measured contention increases the bounded suite budget without removing the hard maximum", observed)
        if number == 31:
            orders = [ordered_test_ids(discovered, policies, row["order_name"], row["random_seed"]) for row in required_order_matrix()]
            observed = {"counts": [len(row) for row in orders], "expected": len(discovered), "all_members_equal": all(set(row) == set(discovered) for row in orders)}
            return _done(test_id, started, len(set(observed["counts"])) == 1 and observed["all_members_equal"], "all order modes execute the same discovered test set", observed)
        if number in {32, 33}:
            row = _victim([sys.executable, "-c", "pass"])
            observed = {key: row[key] for key in ("worker_pid", "exit_code", "timed_out", "cleanup_passed", "orphan_process_count", "zombie_process_count")}
            return _done(test_id, started, row["exit_code"] == 0 and not row["timed_out"] and row["cleanup_passed"] and row["orphan_process_count"] == row["zombie_process_count"] == 0, "completed workers are reaped with zero orphan and zombie counts", observed)
        manifest_hash = sha256_file(MANIFEST); source_hash = hashlib.sha256(b"r9-dynamic-source").hexdigest()
        base_record = {"run_id": "r9-matrix-1", "exit_code": 0, "timed_out": False, "test_count": len(discovered), "orphan_process_count": 0, "zombie_process_count": 0, "source_tree_sha256": source_hash, "policy_manifest_sha256": manifest_hash, "source_origin": "final", "order_name": "default"}
        if number == 34:
            observed = validate_matrix_records([base_record], len(discovered), source_hash, manifest_hash)
            return _done(test_id, started, observed["passed"], "final extracted-source matrix records are source-, policy-, count- and cleanup-bound", observed)
        if number == 35:
            active = {**base_record, "source_origin": "active"}
            observed = {"active_origin": active["source_origin"], "expected_origin": "final", "mislabel_rejected": active["source_origin"] != "final"}
            return _done(test_id, started, observed["mislabel_rejected"], "active workspace runs cannot be labeled as final extracted-package runs", observed)
        if number == 36:
            code = "import runpy,sys;sys.path.insert(0,'.');sys.argv=['unittest','-v','tests.fixtures.minimal_suite'];runpy.run_module('unittest',run_name='__main__')"
            row = _victim([sys.executable, "-c", code])
            transcript = row["stdout"] + "\n" + row["stderr"]
            parsed = parse_unittest_count(transcript)
            observed = {"parsed_test_count": parsed, "exit_code": row["exit_code"], "stdout_sha256": row["stdout_sha256"], "stderr_sha256": row["stderr_sha256"]}
            return _done(test_id, started, row["exit_code"] == 0 and parsed == 1, "raw unittest transcript and structured test count agree", observed)
        if number == 37:
            row = _victim([sys.executable, "-c", "print('worker-raw')"])
            tampered = hashlib.sha256((row["stdout"] + "tampered").encode("utf-8")).hexdigest()
            observed = {"recorded_sha256": row["stdout_sha256"], "tampered_sha256": tampered, "tamper_detected": row["stdout_sha256"] != tampered}
            return _done(test_id, started, observed["tamper_detected"], "a changed worker raw row fails its digest binding", observed)
        if number == 38:
            source = (ROOT / "scripts" / "self_check.py").read_text(encoding="utf-8")
            required = ["r9_order_load_matrix_passed", "r9_policy_coverage_passed", "r9_final_package_matrix_passed"]
            observed = {"required_fields": required, "present": [field for field in required if field in source]}
            return _done(test_id, started, observed["present"] == required, "selfcheck requires policy, order/load and final-package matrix evidence", observed)
        if number == 39:
            observed = {"isolated_kind": "isolated_target_stability", "mixed_kind": "mixed_full_suite_stability", "separate": "isolated_target_stability" != "mixed_full_suite_stability"}
            return _done(test_id, started, observed["separate"], "isolated R8 repetitions and mixed R9 full-suite repetitions use distinct evidence classes", observed)
        if number == 40:
            bad = {**base_record, "run_id": "r9-matrix-bad", "exit_code": 1}
            observed = validate_matrix_records([base_record, bad], len(discovered), source_hash, manifest_hash)
            return _done(test_id, started, not observed["passed"] and len(observed["invalid"]) == 1, "one failed matrix row blocks PASS and 100 percent", observed)
        raise ValueError(test_id)
    except Exception as error:
        return _done(test_id, started, False, f"unexpected:{type(error).__name__}:{error}", {"error": type(error).__name__, "message": str(error)})


def run_all_r9() -> list[dict[str, Any]]:
    return [asdict(run_r9_rg(f"R9-RG-{number:02d}")) for number in range(1, 41)]
