from __future__ import annotations

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

from .stage_control_r10r3r2 import cleanup_wait_seconds


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


@dataclass(frozen=True)
class R5Result:
    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]) -> R5Result:
    return R5Result(test_id,f"{test_id.lower()}-{time.time_ns()}",bool(passed),test_id+":primary",test_id+":contract",assertion,fields,"the declared R5 behavior is observed, not inferred from names or file presence",observed,round((time.perf_counter()-started)*1000,3))


def _tree_run(
    evidence: Any,
    *extra: str,
    ready_delay: float = 0.0,
    bootstrap_timeout: float | None = None,
    ready_timeout: float | None = None,
    execution_timeout: float = 0.35,
    single_process_ready: bool = False,
    source_tree_sha256: str | None = None,
) -> dict[str, Any]:
    from .suite_isolation import phase_budget_from_environment

    budget = phase_budget_from_environment(ROOT)
    ready_timeout = budget.ready_budget_seconds if ready_timeout is None else ready_timeout
    if bootstrap_timeout is None:
        ready_phase_capacity = max(0.0, float(ready_timeout) - float(ready_delay))
        bootstrap_timeout = max(
            budget.bootstrap_budget_seconds,
            min(budget.ready_budget_seconds, ready_phase_capacity),
        )
    with tempfile.TemporaryDirectory(prefix="r5-ready-") as directory:
        bootstrapped_path=pathlib.Path(directory)/"bootstrapped.json"
        ready_path=pathlib.Path(directory)/"ready.json"
        target_arguments=[str(ready_path),"--ready-delay",str(ready_delay),*extra]
        if single_process_ready:
            target_arguments.append("--single-process-ready")
        command=[sys.executable,str(ROOT/"tests"/"fixtures"/"minimal_bootstrap_launcher.py"),"--bootstrapped-path",str(bootstrapped_path),"--target",str(ROOT/"tests"/"fixtures"/"spawn_tree.py"),"--",*target_arguments]
        result=evidence.run_command(command,ROOT,timeout_seconds=execution_timeout,bootstrapped_path=bootstrapped_path,bootstrap_timeout_seconds=bootstrap_timeout,ready_path=ready_path,ready_timeout_seconds=ready_timeout,source_tree_sha256=source_tree_sha256)
        result["bootstrapped_file_exists_after_run"]=bootstrapped_path.is_file()
        result["ready_file_exists_after_run"]=ready_path.is_file()
        return result


def _binding_fixture(evidence: Any, root: pathlib.Path, pre_hash: str = "a"*64, post_hash: str = "b"*64) -> None:
    pre=root/"prepatch_evidence"; post=root/"postpatch_evidence"; pre.mkdir(parents=True); post.mkdir(parents=True)
    manifest={"phase":"PRE_PATCH","source_tree_sha256":pre_hash,"command":["probe"],"platform":"test","started_epoch":1.0,"ended_epoch":2.0,"exit_code":2,"file_count":1,"files":[{"path":"raw/probe.txt","sha256":"0"*64,"size":0}]}
    evidence.json_write(pre/"integration_evidence_manifest.json",manifest)
    pre_side=evidence._binding(pre/"integration_evidence_manifest.json","prepatch_evidence/integration_evidence_manifest.json",pre_hash,["probe"],"test",1,2,2,file_count=1)
    evidence.json_write(pre/"integration_evidence_manifest.sha256.json",pre_side)
    raw=post/"raw_runs.jsonl"; raw.write_text(json.dumps({"test_id":"R5","passed":True})+"\n",encoding="utf-8")
    post_side=evidence._binding(raw,"postpatch_evidence/raw_runs.jsonl",post_hash,["tests"],"test",3,4,0,row_count=1)
    evidence.json_write(post/"raw_runs.sha256.json",post_side); evidence.json_write(post/"source_binding.json",post_side)


def _valid_rollback_proof() -> dict[str, Any]:
    return {"restore_full_test_command":[sys.executable,"run_all.py","--tests-only"],"restored_test_count":160,"restored_failures":0,"restored_errors":0,"restored_exit_code":0,"restore_timed_out":False,"restore_orphan_process_count":0,"hash_restored":True,"original_source_unchanged":True}


def run_r5_rg(test_id: str, *, source_tree_sha256: str | None = None) -> R5Result:
    started=time.perf_counter(); number=int(test_id.rsplit("-",1)[1])
    try:
        from scripts import build_evidence as evidence
        if number == 1:
            row=_tree_run(evidence); passed=row["ready_handshake"]["ready"] is True and row["tracked_descendant_count"]>=2 and row["timed_out"] and row["orphan_process_count"]==0
            return _done(test_id,started,passed,"READY-gated parent, child and grandchild are identity-tracked and terminated",{key:row[key] for key in ("ready_handshake","tracked_descendant_count","timed_out","orphan_process_count","surviving_descendant_pids")},["ready_handshake","tracked_descendant_count","orphan_process_count"])
        if number == 2:
            row=_tree_run(evidence,"--escape-session"); payload=row["ready_handshake"].get("payload") or {}; passed=row["ready_handshake"]["ready"] is True and payload.get("escaped_session") is True and row["tracked_descendant_count"]>=2 and row["orphan_process_count"]==0
            return _done(test_id,started,passed,"a descendant that creates a new POSIX session or Windows process group cannot escape cleanup",{"platform":os.name,"ready":row["ready_handshake"],"tracked_descendant_count":row["tracked_descendant_count"],"orphan_process_count":row["orphan_process_count"],"strategy":row["process_tree_strategy"]},["ready","tracked_descendant_count","orphan_process_count","strategy"])
        if number == 3:
            row=_tree_run(evidence,ready_delay=0.45,ready_timeout=3.0,execution_timeout=0.1,single_process_ready=True,source_tree_sha256=source_tree_sha256)
            stage=row["stage_timings"]; boot_epoch=stage["bootstrapped_event_epoch"]; ready_epoch=stage["ready_event_epoch"]
            intended_delay_ms=(ready_epoch-boot_epoch)*1000 if isinstance(boot_epoch,(int,float)) and isinstance(ready_epoch,(int,float)) else -1
            ready_payload=row["ready_handshake"].get("payload") or {}
            declared_delay=ready_payload.get("intended_ready_delay_seconds")
            ready_deadline_ms=float(row["ready_timeout_seconds"])*1000
            passed=(
                row["bootstrap_handshake"]["bootstrapped"] is True
                and row["ready_handshake"]["ready"] is True
                and isinstance(declared_delay,(int,float))
                and abs(float(declared_delay)-0.45) <= 0.001
                and 400 <= intended_delay_ms <= ready_deadline_ms
                and row["timed_out"]
            )
            return _done(test_id,started,passed,"Python startup, the declared intentional delay, the observed READY deadline, and workflow timeout are independently measured",{"bootstrap":row["bootstrap_handshake"],"ready":row["ready_handshake"],"declared_delay_seconds":declared_delay,"intended_delay_ms":round(intended_delay_ms,3),"ready_deadline_ms":ready_deadline_ms,"execution_timeout_seconds":row["execution_timeout_seconds"],"workflow_timed_out":row["timed_out"],"tracked_descendant_count":row["tracked_descendant_count"]},["bootstrap","ready","declared_delay_seconds","intended_delay_ms","ready_deadline_ms","execution_timeout_seconds","workflow_timed_out","tracked_descendant_count"])
        if number == 4:
            row=_tree_run(evidence,"--never-ready",ready_timeout=0.3,execution_timeout=2.0); passed=row["ready_handshake"]["status"]=="FAIL" and row["exit_code"]==125 and not row["timed_out"] and row["orphan_process_count"]==0
            return _done(test_id,started,passed,"missing READY returns a structured preparation failure and cleans descendants",{key:row[key] for key in ("ready_handshake","exit_code","timed_out","orphan_process_count")},["ready_handshake","exit_code","timed_out","orphan_process_count"])
        if number == 5:
            row=_tree_run(evidence,"--escape-session"); passed=row["tracked_descendant_count"]>=2 and row["surviving_descendant_pids"]==[] and row["orphan_process_count"]==0
            return _done(test_id,started,passed,"every captured process identity is rechecked after timeout",{"tracked":row["tracked_descendants"],"survivors":row["surviving_descendant_pids"],"orphan_process_count":row["orphan_process_count"]},["tracked","survivors","orphan_process_count"])
        if number == 6:
            kwargs={"start_new_session":True} if os.name!="nt" else {"creationflags":getattr(subprocess,"CREATE_NEW_PROCESS_GROUP",0)}
            child=subprocess.Popen([sys.executable,"-c","import time; time.sleep(60)"],**kwargs)
            try:
                deadline=time.monotonic()+2; identity=None
                while time.monotonic()<deadline and identity is None:
                    identity=evidence._process_snapshot().get(child.pid); time.sleep(0.02)
                survivors=evidence.surviving_processes([identity] if identity else []); reported=len(survivors)
            finally:
                if identity: evidence._terminate_identity(identity,True)
                try: child.wait(timeout=cleanup_wait_seconds())
                except subprocess.TimeoutExpired: child.kill(); child.wait(timeout=cleanup_wait_seconds())
            return _done(test_id,started,identity is not None and reported==1,"a real live detached process identity produces a nonzero orphan count",{"identity_captured":identity is not None,"reported_survivors":reported},["identity_captured","reported_survivors"])
        if number == 7:
            row=_tree_run(evidence,"--escape-session"); payload=row["ready_handshake"].get("payload") or {}; identities=row["tracked_descendants"]; passed=len(payload)>=3 and not evidence.surviving_processes(identities)
            return _done(test_id,started,passed,"fixture PIDs leave no live matching process identity after the attack",{"payload":payload,"tracked":identities,"survivors":evidence.surviving_processes(identities)},["payload","tracked","survivors"])
        if 8 <= number <= 13:
            with tempfile.TemporaryDirectory(prefix="r5-binding-") as directory:
                root=pathlib.Path(directory); _binding_fixture(evidence,root)
                if number == 8: (root/"prepatch_evidence"/"integration_evidence_manifest.json").unlink()
                elif number == 9: (root/"prepatch_evidence"/"integration_evidence_manifest.json").write_text("{}",encoding="utf-8")
                elif number == 10:
                    path=root/"prepatch_evidence"/"integration_evidence_manifest.sha256.json"; row=json.loads(path.read_text(encoding="utf-8")); row["path"]="postpatch_evidence/raw_runs.jsonl"; evidence.json_write(path,row)
                elif number == 11:
                    for name in ("raw_runs.sha256.json","source_binding.json"):
                        path=root/"postpatch_evidence"/name; row=json.loads(path.read_text(encoding="utf-8")); row["source_tree_sha256"]="a"*64; evidence.json_write(path,row)
                elif number == 12: pass
                elif number == 13: (root/"postpatch_evidence"/"raw_runs.jsonl").write_text('{"test_id":"R5","passed":false}\n',encoding="utf-8")
                packaged_hash="c"*64 if number==12 else "b"*64; observed=evidence.validate_evidence_bindings(root,packaged_hash)
            labels={8:"missing referenced evidence",9:"evidence digest mismatch",10:"shared PRE/POST raw path",11:"equal PRE/POST source hashes",12:"packaged source binding mismatch",13:"one-row raw tamper"}
            return _done(test_id,started,not observed["passed"],f"{labels[number]} is rejected by content validation",observed,["passed","checks"])
        if number == 14:
            source=inspect.getsource(evidence.run_canary_rollback_drill); proof=_valid_rollback_proof(); passed='"--tests-only"' in source and evidence.validate_rollback_proof(proof)
            return _done(test_id,started,passed,"the default restored command is tests-only and the proof requires its full result",{"default_tests_only":'"--tests-only"' in source,"proof_valid":evidence.validate_rollback_proof(proof)},["default_tests_only","proof_valid"])
        if number == 15:
            proof=_valid_rollback_proof(); proof["restored_failures"]=1
            return _done(test_id,started,not evidence.validate_rollback_proof(proof),"one restored-suite failure invalidates rollback",proof,["restored_failures"])
        if number == 16:
            proof=_valid_rollback_proof(); proof["restored_test_count"]=0
            return _done(test_id,started,not evidence.validate_rollback_proof(proof),"matching hashes without a full restored test count cannot pass",proof,["restored_test_count","hash_restored"])
        if number == 17:
            with tempfile.TemporaryDirectory(prefix="r5-rollback-") as directory:
                proof=evidence.run_canary_rollback_drill(ROOT,pathlib.Path(directory)/"drill",[sys.executable,"run_all.py","--child-smoke"])
            passed=proof["original_source_unchanged"] and proof["original_source_before_sha256"]==proof["original_source_after_sha256"]
            return _done(test_id,started,passed,"rollback attack leaves the original active source byte-identical",proof,["original_source_unchanged","original_source_before_sha256","original_source_after_sha256"])
        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_r5() -> list[dict[str,Any]]:
    return [asdict(run_r5_rg(f"R5-RG-{number:02d}")) for number in range(1,18)]
