from __future__ import annotations

import hashlib
import inspect
import json
import os
import pathlib
import re
import sys
import tempfile
import time
from dataclasses import asdict, dataclass
from typing import Any

from . import process_control


ROOT = pathlib.Path(__file__).resolve().parents[2]
BASELINE_SOURCE_SHA256 = "d22ef241c7b26e95e73435a0cd8ae0dde0377adeb16494671e6c5ca6668a3294"


@dataclass(frozen=True)
class R7Result:
    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], fields: list[str]) -> R7Result:
    return R7Result(test_id, test_id.lower() + "-101", bool(passed), test_id + ":primary", test_id + ":contract", assertion, fields, "observed R7 behavior satisfies every declared field", observed, round((time.perf_counter() - started) * 1000, 3))


def _hazard(*, stdout_bytes: int = 0, stderr_bytes: int = 0, ready_delay: float = 0.0, sleep_seconds: float = 0.0, detached: bool = False, total: float = 5.0, bootstrap_timeout: float | None = None, ready_timeout: float | None = None, execution_timeout: float = 0.35, max_output: int = 131072, nonce: int = 0) -> tuple[dict[str, Any], dict[str, int]]:
    from scripts import build_evidence as evidence
    from .suite_isolation import phase_budget_from_environment

    budget = phase_budget_from_environment(ROOT)
    bootstrap_timeout = budget.bootstrap_budget_seconds if bootstrap_timeout is None else bootstrap_timeout
    ready_timeout = budget.ready_budget_seconds if ready_timeout is None else ready_timeout
    with tempfile.TemporaryDirectory(prefix="r7-hazard-") as directory:
        root = pathlib.Path(directory)
        boot = root / "bootstrapped.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" / "process_hazards.py"), "--",
            "--ready-path", str(ready),
            "--stdout-bytes", str(stdout_bytes), "--stderr-bytes", str(stderr_bytes),
            "--ready-delay", str(ready_delay), "--sleep-seconds", str(sleep_seconds),
        ]
        if detached:
            command.append("--detached-child")
        row = evidence.run_command(command, ROOT, bootstrapped_path=boot, bootstrap_timeout_seconds=bootstrap_timeout, ready_path=ready, ready_timeout_seconds=ready_timeout, execution_timeout_seconds=execution_timeout, total_timeout_seconds=total, max_output_bytes=max_output)
        times = {"bootstrapped_mtime_ns": boot.stat().st_mtime_ns if boot.is_file() else 0, "ready_mtime_ns": ready.stat().st_mtime_ns if ready.is_file() else 0}
        return row, times


def _structured(row: dict[str, Any]) -> dict[str, Any]:
    keys = (
        "exit_code", "elapsed_ms", "timed_out", "total_timeout_seconds", "total_deadline_exceeded",
        "deadline_overrun_ms", "stdout_total_bytes", "stderr_total_bytes", "stdout_truncated",
        "stderr_truncated", "output_drain_threads_stopped", "output_drain_errors", "orphan_process_count",
        "zombie_process_count", "tracked_descendant_count", "tracker_thread_stopped", "bootstrap_handshake",
        "ready_handshake", "process_tree_strategy", "process_marker",
    )
    return {key: row.get(key) for key in keys}


def run_r7_rg(test_id: str) -> R7Result:
    started = time.perf_counter()
    number = int(test_id.rsplit("-", 1)[1])
    try:
        from scripts import build_evidence as evidence

        if number in {1, 2, 3}:
            stdout_bytes = 256 * 1024 if number in {1, 3} else 0
            stderr_bytes = 256 * 1024 if number in {2, 3} else 0
            row, _ = _hazard(stdout_bytes=stdout_bytes, stderr_bytes=stderr_bytes)
            passed = row["exit_code"] == 0 and row["ready_handshake"]["ready"] is True and row["stdout_total_bytes"] == stdout_bytes and row["stderr_total_bytes"] == stderr_bytes
            return _done(test_id, started, passed, "stdout and stderr are drained concurrently before READY without pipe deadlock", _structured(row), ["ready_handshake", "stdout_total_bytes", "stderr_total_bytes"])
        if number == 4:
            row, _ = _hazard(stdout_bytes=2 * 1024 * 1024, stderr_bytes=2 * 1024 * 1024, max_output=65536)
            passed = row["stdout_truncated"] and row["stderr_truncated"] and len(row["stdout"].encode("utf-8")) < 70000 and len(row["stderr"].encode("utf-8")) < 70000
            return _done(test_id, started, passed, "large output is drained fully but retained under a deterministic bound", {**_structured(row), "retained_stdout_bytes": len(row["stdout"].encode("utf-8")), "retained_stderr_bytes": len(row["stderr"].encode("utf-8"))}, ["stdout_truncated", "stderr_truncated", "retained_stdout_bytes", "retained_stderr_bytes"])
        if number == 5:
            row, _ = _hazard(stdout_bytes=1048576, stderr_bytes=524288)
            return _done(test_id, started, row["stdout_total_bytes"] == 1048576 and row["stderr_total_bytes"] == 524288, "output byte counters reflect drained raw bytes rather than retained text", _structured(row), ["stdout_total_bytes", "stderr_total_bytes"])
        if number == 6:
            row, _ = _hazard(stdout_bytes=2097152, stderr_bytes=2097152)
            return _done(test_id, started, row["output_drain_threads_stopped"] and not row["output_drain_errors"], "both bounded drain threads terminate without reader errors", _structured(row), ["output_drain_threads_stopped", "output_drain_errors"])
        if number == 7:
            sources = [ROOT / "scripts" / "build_evidence.py", ROOT / "scripts" / "self_check.py", ROOT / "src" / "reliability_core" / "process_control.py"]
            violations = [path.relative_to(ROOT).as_posix() for path in sources if ".communicate(" in path.read_text(encoding="utf-8")]
            return _done(test_id, started, not violations, "production runner and selfcheck contain no communicate call", {"violations": violations}, ["violations"])
        if number in {8, 9, 10}:
            row, times = _hazard(ready_delay=0.08)
            checks = {8: row["bootstrap_handshake"]["bootstrapped"] is True, 9: row["ready_handshake"]["ready"] is True, 10: times["bootstrapped_mtime_ns"] < times["ready_mtime_ns"]}
            return _done(test_id, started, checks[number], "BOOTSTRAPPED and READY are distinct ordered observed events", {**_structured(row), **times}, ["bootstrap_handshake", "ready_handshake", "bootstrapped_mtime_ns", "ready_mtime_ns"])
        if number == 11:
            with tempfile.TemporaryDirectory(prefix="r7-missing-bootstrap-") as directory:
                root = pathlib.Path(directory); boot = root / "missing.json"; ready = root / "ready.json"
                row = evidence.run_command([sys.executable, str(ROOT / "tests" / "fixtures" / "process_hazards.py"), "--ready-path", str(ready), "--sleep-seconds", "1"], ROOT, bootstrapped_path=boot, ready_path=ready, ready_timeout_seconds=.2, total_timeout_seconds=.6)
            passed = row["exit_code"] == 125 and row["bootstrap_handshake"]["bootstrapped"] is False and row["ready_handshake"]["status"] == "NOT_REQUESTED"
            return _done(test_id, started, passed, "READY cannot bypass a missing BOOTSTRAPPED event", _structured(row), ["exit_code", "bootstrap_handshake", "ready_handshake"])
        if number == 12:
            with tempfile.TemporaryDirectory(prefix="r7-missing-ready-") as directory:
                root = pathlib.Path(directory); boot = root / "boot.json"; ready = root / "missing.json"
                boot.write_text(json.dumps({"status":"BOOTSTRAPPED","bootstrapped":True,"pid":os.getpid()}),encoding="utf-8")
                row = evidence.run_command([sys.executable, "-c", "import time;time.sleep(1)"], ROOT, bootstrapped_path=boot, ready_path=ready, ready_timeout_seconds=.2, total_timeout_seconds=.7)
            passed = row["exit_code"] == 125 and row["bootstrap_handshake"]["bootstrapped"] is True and row["ready_handshake"]["ready"] is False
            return _done(test_id, started, passed, "BOOTSTRAPPED alone cannot satisfy READY", _structured(row), ["exit_code", "bootstrap_handshake", "ready_handshake"])
        if number in {13, 15, 16}:
            with tempfile.TemporaryDirectory(prefix="r7-strict-event-") as directory:
                root = pathlib.Path(directory); boot = root / "boot.json"; ready = root / "ready.json"
                boot.write_text(json.dumps({"status": "READY" if number == 15 else "BOOTSTRAPPED", "bootstrapped": True}), encoding="utf-8")
                ready.write_text(json.dumps({"status": "BOOTSTRAPPED" if number == 16 else "READY", "ready": True}), encoding="utf-8")
                if number == 13:
                    boot.write_text("{", encoding="utf-8")
                row = evidence.run_command([sys.executable, "-c", "import time;time.sleep(1)"], ROOT, bootstrapped_path=boot, ready_path=ready, ready_timeout_seconds=.15, total_timeout_seconds=.5)
            expected = row["bootstrap_handshake"]["bootstrapped"] is False if number in {13, 15} else row["ready_handshake"]["ready"] is False
            return _done(test_id, started, expected and row["exit_code"] == 125, "malformed or status-mismatched handshake payload is rejected", _structured(row), ["exit_code", "bootstrap_handshake", "ready_handshake"])
        if number == 14:
            blocked = False
            with tempfile.TemporaryDirectory(prefix="r7-same-event-") as directory:
                same = pathlib.Path(directory) / "same.json"
                try:
                    process_control.run_supervised_process([sys.executable, "-c", "pass"], ROOT, total_timeout_seconds=1, bootstrapped_path=same, ready_path=same)
                except ValueError:
                    blocked = True
            return _done(test_id, started, blocked, "BOOTSTRAPPED and READY cannot share one file", {"same_path_blocked": blocked}, ["same_path_blocked"])
        if number in {17, 19}:
            row = evidence.run_command([sys.executable, "-c", "import time;time.sleep(60)"], ROOT, total_timeout_seconds=.65)
            passed = row["timed_out"] and row["exit_code"] == 124 and row["elapsed_ms"] <= 850 and not row["total_deadline_exceeded"]
            return _done(test_id, started, passed, "one wall-clock deadline bounds execution and cleanup", _structured(row), ["timed_out", "elapsed_ms", "total_deadline_exceeded"])
        if number in {18, 20}:
            with tempfile.TemporaryDirectory(prefix="r7-total-handshake-") as directory:
                root = pathlib.Path(directory); boot = root / "boot.json"; ready = root / "ready.json"
                command = [sys.executable, str(ROOT / "tests" / "fixtures" / "process_hazards.py"), "--bootstrapped-path", str(boot), "--ready-path", str(ready), "--ready-delay", "2"]
                row = evidence.run_command(command, ROOT, bootstrapped_path=boot, ready_path=ready, ready_timeout_seconds=20, total_timeout_seconds=.65)
            passed = row["exit_code"] == 125 and row["elapsed_ms"] <= 850 and row["total_timeout_seconds"] == .65
            return _done(test_id, started, passed, "handshake timeout cannot extend the single total deadline", _structured(row), ["exit_code", "elapsed_ms", "total_timeout_seconds"])
        if number == 21:
            row, _ = _hazard(total=2.0, sleep_seconds=60, execution_timeout=.25)
            passed = row["deadline_overrun_ms"] <= 50 and not row["total_deadline_exceeded"]
            return _done(test_id, started, passed, "deadline accounting reports no hidden cleanup extension", _structured(row), ["deadline_overrun_ms", "total_deadline_exceeded"])
        if number in {22, 24, 25, 27}:
            row, _ = _hazard(detached=True, sleep_seconds=60, total=5.0, execution_timeout=.3)
            ready_payload=row["ready_handshake"].get("payload") or {}
            reached_ready=row["bootstrap_handshake"]["bootstrapped"] is True and row["ready_handshake"]["ready"] is True and isinstance(ready_payload.get("detached_pid"),int)
            checks = {22: reached_ready and row["tracked_descendant_count"] >= 1 and row["orphan_process_count"] == 0, 24: reached_ready and row["zombie_process_count"] == 0, 25: reached_ready and row["tracker_thread_stopped"] is True, 27: reached_ready and row["orphan_process_count"] == 0 and row["zombie_process_count"] == 0 and row["output_drain_threads_stopped"]}
            return _done(test_id, started, checks[number], "detached descendant is tracked, force-terminated and reaped under the deadline", _structured(row), ["tracked_descendant_count", "orphan_process_count", "zombie_process_count", "tracker_thread_stopped"])
        if number == 23:
            rows = [_hazard(detached=True, sleep_seconds=60, total=5.0,execution_timeout=.3,nonce=nonce)[0] for nonce in range(1,4)]
            observed = {"runs": len(rows), "ready_runs":sum(row["ready_handshake"]["ready"] is True and isinstance((row["ready_handshake"].get("payload") or {}).get("detached_pid"),int) for row in rows), "orphan_total": sum(row["orphan_process_count"] for row in rows), "zombie_total": sum(row["zombie_process_count"] for row in rows), "timeouts": sum(row["timed_out"] for row in rows)}
            return _done(test_id, started, observed == {"runs": 3,"ready_runs":3, "orphan_total": 0, "zombie_total": 0, "timeouts": 3}, "three detached attacks each reach READY before timeout and leave no orphan or zombie", observed, ["runs","ready_runs", "orphan_total", "zombie_total", "timeouts"])
        if number == 26:
            row, _ = _hazard(detached=True, sleep_seconds=60, total=5.0,execution_timeout=.3)
            marker = str(row["process_marker"])
            return _done(test_id, started, bool(re.fullmatch(r"[0-9a-f]{64}", marker)) and process_control.PROCESS_MARKER_ENV not in json.dumps(row), "descendant correlation exposes only a marker digest", {"marker_digest": marker, "raw_marker_exposed": process_control.PROCESS_MARKER_ENV in json.dumps(row)}, ["marker_digest", "raw_marker_exposed"])
        if number == 28:
            current = process_control.process_snapshot().get(os.getpid()); fake = dict(current or {"pid": os.getpid(), "started": "missing", "state": "RUNNING"}); fake["started"] = str(fake["started"]) + "-reused"
            survivors = process_control.surviving_processes([fake])
            return _done(test_id, started, not survivors, "PID reuse cannot match a different process start identity", {"fake": fake, "survivors": survivors}, ["fake", "survivors"])
        if number == 29:
            row = process_control.run_supervised_process([sys.executable, str(ROOT / "tests" / "fixtures" / "process_hazards.py"), "--sleep-seconds", "60"], ROOT, total_timeout_seconds=.55)
            return _done(test_id, started, row["timed_out"] and row["exit_code"] == 124 and row["orphan_process_count"] == 0, "a dangerous fixture executes only inside a sacrificial watchdog process", _structured(row), ["timed_out", "exit_code", "orphan_process_count"])
        if number == 30:
            row = process_control.run_supervised_process([sys.executable, "run_all.py", "--child-smoke"], ROOT, total_timeout_seconds=30, max_output_bytes=1048576)
            return _done(test_id, started, row["exit_code"] == 0 and not row["timed_out"] and row["orphan_process_count"] == 0, "watchdog supervision preserves a normal bounded child test path", _structured(row), ["exit_code", "timed_out", "orphan_process_count"])
        if number in {31, 32}:
            source = (ROOT / "run_all.py").read_text(encoding="utf-8")
            observed = {"test_worker_route": "--test-worker" in source, "per_sensitive_test": "sacrificial_subprocess_per_sensitive_test" in source, "adaptive_total_budget": "suite_total_budget" in source and "suite_deadline" in source, "per_worker_limit": "per_worker_limit" in source, "supervisor": "run_supervised_process" in source, "recursive_tests_only_worker": "--tests-only" in source[source.find("def run_tests_worker"):source.find("def run_tests(")]}
            checks = {31: observed["test_worker_route"] and observed["per_sensitive_test"], 32: observed["adaptive_total_budget"] and observed["per_worker_limit"] and observed["supervisor"] and not observed["recursive_tests_only_worker"]}
            return _done(test_id, started, checks[number], "full tests route through adaptive bounded per-test sacrificial workers without recursive tests-only entry", observed, list(observed))
        if number == 33:
            files = [ROOT / "scripts" / "build_evidence.py", ROOT / "scripts" / "self_check.py", ROOT / "src" / "reliability_core" / "process_control.py"]
            violations = []
            for path in files:
                for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
                    if ".communicate(" in line:
                        violations.append(f"{path.relative_to(ROOT).as_posix()}:{line_number}")
            return _done(test_id, started, not violations, "production execution paths have no bounded or unbounded communicate dependency", {"violations": violations}, ["violations"])
        if number == 34:
            source = (ROOT / "src" / "reliability_core" / "process_control.py").read_text(encoding="utf-8")
            unbounded_join = [line.strip() for line in source.splitlines() if ".join(" in line and "timeout=" not in line]
            unbounded_wait = [line.strip() for line in source.splitlines() if re.search(r"\.wait\(\s*\)", line)]
            return _done(test_id, started, not unbounded_join and not unbounded_wait, "all thread joins and waits in process control are deadline-bounded", {"unbounded_join": unbounded_join, "unbounded_wait": unbounded_wait}, ["unbounded_join", "unbounded_wait"])
        if number in {35, 36}:
            source = (ROOT / "scripts" / "self_check.py").read_text(encoding="utf-8")
            fields = ["full_suite_rerun_passed", "full_suite_rerun_exit_code", "full_suite_rerun_timed_out", "full_suite_rerun_test_count", "full_suite_rerun_orphan_process_count", "full_suite_rerun_zombie_process_count", "full_suite_rerun_source_tree_sha256"]
            checks = {35: all(field in source for field in fields), 36: '"--tests-only"' in source and "TemporaryDirectory" in source and "archive.read" in source and "full_suite_rerun_stdout_sha256" in source}
            return _done(test_id, started, checks[number], "independent selfcheck extracts packaged source and records full-suite fields and raw hashes", {"fields_present": [field for field in fields if field in source], "tests_only": '"--tests-only"' in source, "temporary_extract": "TemporaryDirectory" in source}, ["fields_present", "tests_only", "temporary_extract"])
        if number == 37:
            current = evidence.source_tree_hash(); required = [ROOT / "src" / "reliability_core" / "process_control.py", ROOT / "src" / "reliability_core" / "regressions_r7.py", ROOT / "tests" / "test_r7_rg.py"]
            passed = current != BASELINE_SOURCE_SHA256 and all(path.is_file() for path in required) and "R7-RG-" in (ROOT / "scripts" / "build_evidence.py").read_text(encoding="utf-8")
            return _done(test_id, started, passed, "current source is fresh and R7 code, tests and dynamic evidence mapping are bound to it", {"baseline_source_sha256": BASELINE_SOURCE_SHA256, "current_source_sha256": current, "required_files": [path.relative_to(ROOT).as_posix() for path in required], "required_files_present": all(path.is_file() for path in required)}, ["baseline_source_sha256", "current_source_sha256", "required_files_present"])
        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)}, ["error", "message"])


def run_all_r7() -> list[dict[str, Any]]:
    return [asdict(run_r7_rg(f"R7-RG-{number:02d}")) for number in range(1, 38)]
