from __future__ import annotations

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

from .harness import AcceptanceHarness
from .regressions import run_rg
from .regressions_r2 import run_r2_rg
from .regressions_r3 import run_r3_rg


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


@dataclass(frozen=True)
class R4Result:
    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]) -> R4Result:
    return R4Result(test_id, test_id.lower()+"-101", bool(passed), test_id+":primary", test_id+":contract", assertion, fields, "all declared R4 safety conditions are true", observed, round((time.perf_counter()-started)*1000,3))


def _nested_entrypoint_calls() -> list[str]:
    violations=[]
    for path in sorted((ROOT/"src"/"reliability_core").glob("regressions*.py")):
        tree=ast.parse(path.read_text(encoding="utf-8"))
        for node in ast.walk(tree):
            if not isinstance(node, ast.Call): continue
            values=[]
            for arg in node.args:
                if isinstance(arg,(ast.List,ast.Tuple)):
                    values.extend(item.value for item in arg.elts if isinstance(item,ast.Constant) and isinstance(item.value,str))
                elif isinstance(arg,ast.Constant) and isinstance(arg.value,str): values.append(arg.value)
            if "run_all.py" in values and any(flag in values for flag in ("--tests-only","--evidence-build","--evidence")):
                violations.append(f"{path.name}:{node.lineno}:{','.join(values)}")
    return violations


def _pid_alive(pid: int) -> bool:
    if os.name == "nt":
        result=subprocess.run(["tasklist","/FI",f"PID eq {pid}","/FO","CSV","/NH"],capture_output=True,text=True,timeout=10)
        return str(pid) in result.stdout and "No tasks" not in result.stdout
    try:
        fields=(pathlib.Path("/proc")/str(pid)/"stat").read_text(encoding="utf-8").rsplit(")",1)[1].strip().split()
        return fields[0] != "Z"
    except (OSError,IndexError):
        return False


def run_r4_rg(test_id: str) -> R4Result:
    started=time.perf_counter(); number=int(test_id.rsplit("-",1)[1])
    try:
        from scripts import build_evidence as evidence
        if number == 1:
            selection=evidence.active_source_selection(ROOT)
            passed=selection["selected"] and selection["entrypoint_exists"] and selection["package_exists"] and len(selection["tree_sha256"])==64
            return _done(test_id,started,passed,"active source selection is content and entrypoint based",selection,["selected","tree_sha256"])
        if number == 2:
            blocked=False
            try: evidence.selective_patch_guard({"a.py":"1","b.py":"2"},{"a.py":"9","b.py":"2"},{"b.py"})
            except Exception: blocked=True
            return _done(test_id,started,blocked,"bulk or out-of-scope reference overwrite is rejected",{"blocked":blocked},["blocked"])
        if number == 3:
            rows=evidence.classify_integration_changes({"runtime.py":"a","meta.json":"b"},{"runtime.py":"x","meta.json":"b"},{"runtime.py":{"origin":"R4 fresh evidence","purpose":"runtime correction","tests":["R4-RG-03"]}})
            passed=all(row.get("status") and row.get("origin") and row.get("purpose") and row.get("tests") for row in rows)
            return _done(test_id,started,passed,"each changed file has origin, purpose, tests and classification",{"rows":rows},["rows"])
        if number == 4:
            violations=[row for row in _nested_entrypoint_calls() if "--tests-only" in row]
            return _done(test_id,started,not violations,"no regression test re-enters the complete test suite",{"violations":violations},["violations"])
        if number == 5:
            with tempfile.TemporaryDirectory() as directory:
                ready_path=pathlib.Path(directory)/"ready.json"
                result=evidence.run_command([sys.executable,str(ROOT/"tests"/"fixtures"/"spawn_tree.py"),str(ready_path)],ROOT,timeout_seconds=.1,ready_path=ready_path,ready_timeout_seconds=10)
                payload=result["ready_handshake"].get("payload") or {}; child_pids=[int(payload[key]) for key in ("child_pid","grandchild_pid") if key in payload]; alive=[pid for pid in child_pids if _pid_alive(pid)]
            passed=result["ready_handshake"]["ready"] is True and result["timed_out"] and result["tracked_descendant_count"]>=2 and result["orphan_process_count"]==0 and not alive and result["elapsed_ms"] < 15000
            observed={"ready_handshake":result["ready_handshake"],"timed_out":result["timed_out"],"execution_timeout_seconds":result["execution_timeout_seconds"],"tracked_descendant_count":result["tracked_descendant_count"],"orphan_process_count":result["orphan_process_count"],"alive_after_timeout":alive,"elapsed_ms":result["elapsed_ms"]}
            return _done(test_id,started,passed,"explicit READY precedes timeout and identity-tracked parent, child and grandchild termination",observed,["ready_handshake","timed_out","execution_timeout_seconds","tracked_descendant_count","orphan_process_count","alive_after_timeout","elapsed_ms"])
        if number == 6:
            violations=[row for row in _nested_entrypoint_calls() if "--evidence" in row]
            return _done(test_id,started,not violations,"tests never recursively invoke the complete evidence entrypoint",{"violations":violations},["violations"])
        if number == 7:
            result=subprocess.run([sys.executable,"run_all.py","--functional-smoke"],cwd=ROOT,text=True,capture_output=True,timeout=45)
            selfcheck_source=(ROOT/"scripts"/"self_check.py").read_text(encoding="utf-8")
            passed=result.returncode==0 and "functional_rerun_passed" in selfcheck_source
            return _done(test_id,started,passed,"selfcheck requires a fresh functional smoke beyond package integrity",{"exit_code":result.returncode,"functional_binding":"functional_rerun_passed" in selfcheck_source},["exit_code","functional_binding"])
        if number == 8:
            product=evidence.product_summary_contract(); engine=evidence.engine_inheritance_contract()
            forbidden={"CRD_ROUTE_CANDIDATE_GATE","mandatory_inheritance_contract","tokenmax_runtime_inheritance","capacity_cleanup_anchor"}
            passed=not forbidden.intersection(product) and forbidden.issubset(engine)
            return _done(test_id,started,passed,"engine-global inheritance is separated from product P0 metrics",{"product_keys":sorted(product),"engine_keys":sorted(engine)},["product_keys","engine_keys"])
        if number == 9:
            manifest=evidence.normalized_source_manifest(ROOT); forbidden=("artifacts/","delivery/","__pycache__/",".pyc",".db")
            hits=[name for name in manifest if any(token in name for token in forbidden)]
            return _done(test_id,started,not hits,"generated files never enter normalized source hash",{"file_count":len(manifest),"hits":hits},["file_count","hits"])
        if number == 10:
            results={"AT-01":AcceptanceHarness().run("AT-01",101).passed,"RG-01":run_rg("RG-01",101).passed,"R2-RG-14":run_r2_rg("R2-RG-14").passed,"R3-RG-17":run_r3_rg("R3-RG-17").passed}
            return _done(test_id,started,all(results.values()),"representative integrated safety gates execute directly",results,list(results))
        if number == 11:
            result=evidence.run_command([sys.executable,"run_all.py","--child-smoke"],ROOT,timeout_seconds=30)
            required={"source_tree_sha256","platform","pid","started_epoch","ended_epoch","command"}
            return _done(test_id,started,required.issubset(result),"raw command evidence binds source, platform, PID and time",{"required":sorted(required),"present":sorted(required.intersection(result))},["required","present"])
        if number == 12:
            fixture=json.loads((ROOT/"fixtures"/"r4_evidence_provenance.json").read_text(encoding="utf-8")); pre=fixture["prepatch"]; post=fixture["postpatch"]; post_tree=evidence.normalized_tree_hash(ROOT) if post["tree_sha256"]=="CURRENT_RUNTIME" else post["tree_sha256"]
            passed=pre["tree_sha256"]!=post_tree and pre["raw_ref"]!=post["raw_ref"]
            return _done(test_id,started,passed,"PRE_PATCH and POST_PATCH use different source hashes and raw references",{"pre_tree":pre["tree_sha256"],"post_tree":post_tree,"pre_raw":pre["raw_ref"],"post_raw":post["raw_ref"]},["pre_tree","post_tree","pre_raw","post_raw"])
        if number == 13:
            before={"protected.py":"same","target.py":"old"}; after={"protected.py":"same","target.py":"new"}; changed=evidence.selective_patch_guard(before,after,{"target.py"})
            return _done(test_id,started,changed==["target.py"],"selective patch preserves unrelated integrated files",{"changed":changed},["changed"])
        if number == 14:
            with tempfile.TemporaryDirectory() as directory: proof=evidence.run_canary_rollback_drill(ROOT,pathlib.Path(directory)/"drill",[sys.executable,"run_all.py","--child-smoke"])
            return _done(test_id,started,proof["passed"] and proof["before_tree_sha256"]==proof["restored_tree_sha256"],"rollback restores the exact integrated source hash",proof,["passed","before_tree_sha256","restored_tree_sha256"])
        if number == 15:
            gates={"A":True,"B":False}; issues=evidence.issues_from_gates(gates,[{"run_id":"raw-b","passed":False}]); passed=issues["verdict"]=="FAIL" and issues["issue_count"]==1 and issues["issues"][0]["gate"]=="B"
            return _done(test_id,started,passed,"issues and verdict are dynamically recomputed from raw gates",{"issues":issues},["issues"])
        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_r4() -> list[dict[str,Any]]:
    return [asdict(run_r4_rg(f"R4-RG-{number:02d}")) for number in range(1,16)]
