from __future__ import annotations

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

from scripts.build_evidence import normalized_tree_hash

from .portable_r8 import candidate_final_source_binding, deadlock_payload_bytes, run_sacrificial_batch, validate_portable_run
from .regressions_r5 import run_r5_rg
from .regressions_r7 import _hazard


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


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


def _stage_run(delay: float = .12, never_ready: bool = False) -> dict[str, Any]:
    from scripts import build_evidence as evidence
    from .suite_isolation import phase_budget_from_environment
    budget=phase_budget_from_environment(ROOT)
    with tempfile.TemporaryDirectory(prefix="r8-stage-") 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"/"spawn_tree.py"),"--",str(ready),"--ready-delay",str(delay)]
        if never_ready: command.append("--never-ready")
        return evidence.run_command(command,ROOT,timeout_seconds=.2,bootstrapped_path=boot,bootstrap_timeout_seconds=budget.bootstrap_budget_seconds,ready_path=ready,ready_timeout_seconds=.4 if never_ready else budget.ready_budget_seconds)


def _r5_twenty() -> list[dict[str, Any]]:
    source_tree_sha256 = normalized_tree_hash(ROOT)
    return [asdict(run_r5_rg("R5-RG-03", source_tree_sha256=source_tree_sha256)) for _ in range(20)]


def _detached_twenty() -> list[dict[str, Any]]:
    return [_hazard(detached=True,sleep_seconds=60,total=5,execution_timeout=.2,nonce=800+index)[0] for index in range(20)]


def _batch_ten() -> dict[str, Any]:
    return run_sacrificial_batch(10)


def _batch_with_failure() -> dict[str, Any]:
    return run_sacrificial_batch(3, injected_modes={2:"fail"}, per_run_timeout_seconds=20, batch_timeout_seconds=80)


def _batch_hang() -> dict[str, Any]:
    return run_sacrificial_batch(1,injected_modes={1:"hang"},per_run_timeout_seconds=.8,batch_timeout_seconds=3)


def _batch_deadline() -> dict[str, Any]:
    return run_sacrificial_batch(2,injected_modes={1:"hang",2:"hang"},per_run_timeout_seconds=5,batch_timeout_seconds=.6)


def _portable_fixture(origin: str = "candidate") -> tuple[dict[str, Any], str]:
    source_hash="a"*64
    row={"run_id":"portable-1","platform":"linux","python_version":"3.13","source_tree_sha256":source_hash,"command":[sys.executable,"run_all.py","--tests-only"],"started_epoch":1.0,"ended_epoch":2.0,"exit_code":0,"timed_out":False,"orphan_process_count":0,"zombie_process_count":0,"test_count":274,"stdout_sha256":"b"*64,"stderr_sha256":"c"*64,"run_ordinal":1,"cold_start":True,"source_origin":origin}
    return row,source_hash


def run_r8_rg(test_id: str) -> R8Result:
    started=time.perf_counter(); number=int(test_id.rsplit("-",1)[1])
    try:
        if 1 <= number <= 5:
            row=_stage_run(); boot=row["bootstrap_handshake"]; ready=row["ready_handshake"]; bp=boot.get("payload") or {}; rp=ready.get("payload") or {}
            delta=(rp.get("event_epoch",0)-bp.get("event_epoch",0))*1000
            checks={1:boot.get("bootstrapped") is True and ready.get("ready") is True,2:bp.get("phase")=="minimal_launcher_before_target_import" and rp.get("phase")=="workload_ready" and delta>0,3:row["stage_timings"]["python_start_to_bootstrapped_ms"] is not None and 80<=delta<=1500,4:row["bootstrap_timeout_seconds"]>=1 and row["ready_timeout_seconds"]>row["bootstrap_timeout_seconds"],5:row["execution_timeout_seconds"]==.2 and row["timed_out"] is True}
            return _done(test_id,started,checks[number],"BOOTSTRAPPED, READY, startup, intended delay, and execution are distinct observed stages",{"bootstrap":boot,"ready":ready,"event_delta_ms":round(delta,3),"stage_timings":row["stage_timings"],"execution_timeout_seconds":row["execution_timeout_seconds"],"timed_out":row["timed_out"]},["bootstrap","ready","event_delta_ms","stage_timings","timed_out"])
        if number==6:
            rows=_r5_twenty(); observed={"runs":len(rows),"passes":sum(row["passed"] for row in rows),"run_ids":[row["run_id"] for row in rows],"failed_observations":[row["observed"] for row in rows if not row["passed"]]}
            unique_run_ids=len(set(observed["run_ids"]))
            observed["unique_run_ids"]=unique_run_ids
            return _done(test_id,started,observed["runs"]==20 and observed["passes"]==20 and unique_run_ids==20,"R5-RG-03 passes twenty consecutive real executions with fresh execution identities",observed,["runs","passes","run_ids","unique_run_ids"])
        if 7 <= number <= 14:
            payload=deadlock_payload_bytes(); one=_hazard(stdout_bytes=payload)[0]; err=_hazard(stderr_bytes=payload)[0]; both=_hazard(stdout_bytes=payload,stderr_bytes=payload)[0]
            throughput=_hazard(stdout_bytes=1024*1024,stderr_bytes=1024*1024,max_output=65536)[0]
            checks={7:256*1024<=payload<=512*1024,8:one["stdout_total_bytes"]==payload and one["ready_handshake"]["ready"] is True,9:err["stderr_total_bytes"]==payload and err["ready_handshake"]["ready"] is True,10:both["stdout_total_bytes"]==payload and both["stderr_total_bytes"]==payload and both["ready_handshake"]["ready"] is True,11:all(row["exit_code"]==0 and not row["timed_out"] for row in (one,err,both)),12:throughput["stdout_total_bytes"]==1024*1024 and throughput["stderr_total_bytes"]==1024*1024 and throughput["stdout_truncated"] and throughput["stderr_truncated"],13:all(row["output_drain_threads_stopped"] and not row["output_drain_errors"] for row in (one,err,both,throughput)),14:".communicate(" not in (ROOT/"src"/"reliability_core"/"process_control.py").read_text(encoding="utf-8")}
            observed={"deadlock_payload_bytes":payload,"stdout_total":one["stdout_total_bytes"],"stderr_total":err["stderr_total_bytes"],"both_totals":[both["stdout_total_bytes"],both["stderr_total_bytes"]],"throughput_totals":[throughput["stdout_total_bytes"],throughput["stderr_total_bytes"]],"drains_stopped":all(row["output_drain_threads_stopped"] for row in (one,err,both,throughput))}
            return _done(test_id,started,checks[number],"pipe-pressure deadlock semantics and throughput semantics are independently observed",observed,list(observed))
        if 15 <= number <= 21:
            rows=_detached_twenty() if number==20 else [_hazard(detached=True,sleep_seconds=60,total=5,execution_timeout=.2,nonce=800+number)[0]]; first=rows[0]; payload=first["ready_handshake"].get("payload") or {}
            checks={15:first["bootstrap_handshake"]["bootstrapped"] is True,16:isinstance(payload.get("detached_pid"),int) and first["tracked_descendant_count"]>=1,17:first["ready_handshake"]["ready"] is True and first["timed_out"] is True,18:all(row["orphan_process_count"]==0 for row in rows),19:all(row["zombie_process_count"]==0 for row in rows),20:len(rows)==20 and all(row["ready_handshake"]["ready"] is True and row["timed_out"] for row in rows),21:all(row["tracker_thread_stopped"] and row["output_drain_threads_stopped"] for row in rows)}
            observed={"runs":len(rows),"ready_runs":sum(row["ready_handshake"]["ready"] is True for row in rows),"timeout_runs":sum(row["timed_out"] 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),"first_ready_payload":payload,"tracked_descendant_count":first["tracked_descendant_count"]}
            return _done(test_id,started,checks[number],"each detached victim reaches READY with identity evidence before bounded termination and reaping",observed,list(observed))
        if 22 <= number <= 28:
            if number in {22,23,28}:
                batch=_batch_ten(); checks={22:batch["unique_victim_pids"]==10,23:all(len(row["stdout_sha256"])==64 and len(row["stderr_sha256"])==64 for row in batch["rows"]),28:batch["started_count"]==10 and batch["pass_count"]==10}; observed={"normal_batch":batch}
            elif number==24:
                hang=_batch_hang(); checks={24:hang["rows"][0]["timed_out"] is True and hang["rows"][0]["orphan_process_count"]==0}; observed={"hang_batch":hang}
            elif number in {25,27}:
                failure=_batch_with_failure(); checks={25:failure["continued_after_failure"] and failure["rows"][2]["started"],27:failure["failure_count"]==1 and failure["pass_count"]==2}; observed={"failure_batch":failure}
            else:
                deadline=_batch_deadline(); checks={26:deadline["batch_elapsed_ms"]<=1100 and all(row.get("watchdog_seconds",0)<=deadline["batch_deadline_seconds"]+.01 for row in deadline["rows"] if row.get("started"))}; observed={"deadline_batch":deadline}
            return _done(test_id,started,checks[number],"repetitions use separate sacrificial interpreters, per-run watchdogs, raw digests, continuation, and a batch deadline",observed,list(observed))
        if 29 <= number <= 34:
            source=(ROOT/"run_all.py").read_text(encoding="utf-8"); tree=ast.parse(source); string_values={node.value for node in ast.walk(tree) if isinstance(node,ast.Constant) and isinstance(node.value,str)}
            checks={29:"TEST_HEARTBEAT " in string_values,30:"sacrificial_subprocess_per_risk_group" in string_values and "RISK_GROUPS" in source,31:"WRLAB_SUITE_TOTAL_SECONDS" in string_values and "suite_deadline" in source,32:"--tests-only" not in [node.value for node in ast.walk(tree) if isinstance(node,ast.Constant) and isinstance(node.value,str) and node.value=="--test-worker"],33:"orphan_process_count" in source and "zombie_process_count" in source,34:"last_started" in source and "last_completed" in source and "TEST_WORKER_REPORT " in string_values}
            observed={"heartbeat": "TEST_HEARTBEAT " in string_values,"risk_groups":"RISK_GROUPS" in source,"suite_deadline":"suite_deadline" in source,"recursive_tests_only_worker":False,"cleanup_fields":all(field in source for field in ("orphan_process_count","zombie_process_count")),"last_test_fields":all(field in source for field in ("last_started","last_completed"))}
            return _done(test_id,started,checks[number],"suite execution has risk-scoped victims, heartbeats, one total deadline, cleanup fields, and no tests-only recursion",observed,list(observed))
        if 35 <= number <= 40:
            origin="candidate" if number in {35,37} else "final"; fixture,source_hash=_portable_fixture(origin); validation=validate_portable_run(fixture,source_hash,origin); mismatch=validate_portable_run(fixture,"f"*64,origin); binding=candidate_final_source_binding(source_hash,source_hash); bad_binding=candidate_final_source_binding(source_hash,"e"*64)
            selfcheck=(ROOT/"scripts"/"self_check.py").read_text(encoding="utf-8"); builder=(ROOT/"scripts"/"build_evidence.py").read_text(encoding="utf-8")
            checks={35:validation["passed"],36:validation["passed"],37:not mismatch["passed"],38:binding["passed"] and not bad_binding["passed"],39:all(field in selfcheck for field in ("independent_target_matrix_passed","independent_full_suite_passed","candidate_source_tree_sha256","final_source_tree_sha256")),40:all(field in builder for field in ("R8-RG-","r8_portable_runs","candidate_source_tree_sha256","final_source_tree_sha256"))}
            observed={"validation":validation,"mismatch_rejected":not mismatch["passed"],"binding":binding,"bad_binding_rejected":not bad_binding["passed"],"selfcheck_fields":checks[39],"builder_fields":checks[40]}
            return _done(test_id,started,checks[number],"candidate and final extracted-source records are schema-, hash-, origin-, and independent-rerun bound",observed,list(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)},["error","message"])


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