from __future__ import annotations

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

from .harness import run_capability_ablation,valid_documents
from .runtime import WorkflowRuntime
from .storage import Store

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


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


def _tree(evidence:Any,escape:bool=False)->dict[str,Any]:
    with tempfile.TemporaryDirectory(prefix="r6-tree-") as directory:
        ready=pathlib.Path(directory)/"ready.json"; command=[sys.executable,str(ROOT/"tests"/"fixtures"/"spawn_tree.py"),str(ready)]
        if escape: command.append("--escape-session")
        return evidence.run_command(command,ROOT,timeout_seconds=.15,ready_path=ready,ready_timeout_seconds=5)


def _producer_fixture(evidence:Any,root:pathlib.Path,source_hash:str="b"*64)->dict[str,Any]:
    post=root/"postpatch_evidence"; post.mkdir(parents=True); raw=post/"raw_runs.jsonl"; raw.write_text('{"test_id":"R6","passed":true}\n',encoding="utf-8"); created=raw.stat().st_mtime
    producer=[sys.executable,"run_all.py","--"+"evidence-build"]
    row=evidence._binding(raw,"postpatch_evidence/raw_runs.jsonl",source_hash,producer,"test",created-1,created+1,0,row_count=1)
    row.update({"producer_command":producer,"producer_started_epoch":created-1,"producer_ended_epoch":created+1,"producer_exit_code":0,"producer_platform":"test","producer_source_tree_sha256":source_hash,"raw_created_epoch":created,"verification_command":[sys.executable,"run_all.py","--tests-only"]})
    evidence.json_write(post/"raw_runs.sha256.json",row); evidence.json_write(post/"source_binding.json",row); return row


def _linux_fixture(evidence:Any,root:pathlib.Path,source_hash:str="c"*64,version:str="3.13")->pathlib.Path:
    root.mkdir(parents=True,exist_ok=True); stdout=root/"python313.stdout.txt"; stderr=root/"python313.stderr.txt"; stdout.write_text("",encoding="utf-8"); stderr.write_text("Ran 160 tests in 1.000s\n\nOK\n",encoding="utf-8")
    row={"command":["python","run_all.py","--tests-only"],"exit_code":0,"timed_out":False,"orphan_process_count":0,"test_count":160,"source_tree_sha256":source_hash,"platform":"linux-test","python_version":version,"image_digest":"sha256:"+"1"*64,"stdout_file":stdout.name,"stderr_file":stderr.name,"stdout_sha256":hashlib.sha256(stdout.read_bytes()).hexdigest(),"stderr_sha256":hashlib.sha256(stderr.read_bytes()).hexdigest()}
    command=root/"python313_command.json"; evidence.json_write(command,row); evidence.json_write(root/(command.name+".sha256.json"),{"sha256":evidence.sha256(command),"size":command.stat().st_size}); return command


def _classification_row(path:str,symbol:str,test_id:str,before:str,after:str,origin:str="R6-F06",purpose:str="unique semantic purpose")->dict[str,Any]:
    return {"path":path,"origin":origin,"purpose":purpose,"requirements":["R6-F06"],"tests":[test_id],"symbol":symbol,"code_location":f"{path}:1-1","before_sha256":before,"after_sha256":after,"decision":"KEEP_INTEGRATED","decision_reason":"direct semantic evidence"}


def run_r6_rg(test_id:str)->R6Result:
    started=time.perf_counter(); number=int(test_id.rsplit("-",1)[1])
    try:
        from scripts import build_evidence as evidence
        if number==1:
            from .portable_r8 import run_sacrificial_batch
            observed=run_sacrificial_batch(10)
            passed=observed["started_count"]==10 and observed["pass_count"]==10 and observed["unique_victim_pids"]==10 and all(row["stdout_sha256"] and row["stderr_sha256"] for row in observed["rows"])
            return _done(test_id,started,passed,"R4 process-tree regression passes ten isolated sacrificial interpreters with watchdog and digest evidence",observed,["started_count","pass_count","unique_victim_pids","rows"])
        if number in {2,3,4}:
            version={2:"3.11",3:"3.12",4:"3.13"}[number]
            with tempfile.TemporaryDirectory() as directory:
                root=pathlib.Path(directory); _linux_fixture(evidence,root,version=version); observed=evidence.validate_linux_evidence_directory(root,"c"*64,(version,))
            return _done(test_id,started,observed["passed"],f"Python {version} platform evidence requires transcript-bound full-suite facts",observed,["required_versions","passed"])
        if number in {5,6,7}:
            row=_tree(evidence,escape=number==6); passed=row["orphan_process_count"]==0 and row["zombie_process_count"]==0 and row["tracker_thread_stopped"] and row["tracked_descendant_count"]>=2
            return _done(test_id,started,passed,"tracked descendants terminate, zombies are reaped and tracker stops",{key:row[key] for key in ("tracked_descendant_count","orphan_process_count","zombie_process_count","reaped_descendant_count","tracker_thread_stopped","linux_subreaper_enabled")},["orphan_process_count","zombie_process_count","tracker_thread_stopped"])
        if number==8:
            current=evidence._process_snapshot().get(__import__("os").getpid()); fake=dict(current or {"pid":__import__("os").getpid(),"started":"missing","state":"RUNNING"}); fake["started"]=str(fake["started"])+"-reused"
            return _done(test_id,started,not evidence.surviving_processes([fake]),"same PID with a different start identity is never treated as the tracked process",{"fake_identity":fake,"survivors":evidence.surviving_processes([fake])},["fake_identity","survivors"])
        if number==9:
            rows=[evidence.run_command([sys.executable,str(ROOT/"tests"/"fixtures"/"minimal_suite.py")],ROOT,timeout_seconds=10) for _ in range(20)]; passed=all(row["orphan_process_count"]==0 and row["zombie_process_count"]==0 and row["tracker_thread_stopped"] for row in rows)
            return _done(test_id,started,passed,"twenty bounded fixture processes leave no running or zombie residue",{"count":len(rows),"orphan_total":sum(row["orphan_process_count"] for row in rows),"zombie_total":sum(row["zombie_process_count"] for row in rows)},["count","orphan_total","zombie_total"])
        if number==10:
            with warnings.catch_warnings(record=True) as caught:
                warnings.simplefilter("always",ResourceWarning)
                for index in range(100):
                    with WorkflowRuntime(run_id=f"r6-life-{index}") as runtime: runtime.execute(f"case-{index}",valid_documents(),stop_after_ready=True)
                gc.collect()
            leaks=[str(item.message) for item in caught if issubclass(item.category,ResourceWarning)]
            return _done(test_id,started,not leaks,"one hundred owned stores close without ResourceWarning",{"runtime_count":100,"resource_warning_count":len(leaks)},["runtime_count","resource_warning_count"])
        if number==11:
            store=Store(); runtime=WorkflowRuntime(store=store,run_id="r6-external"); runtime.execute("case-external",valid_documents(),stop_after_ready=True); runtime.close(); state=store.state("case-external","r6-external"); store.close()
            return _done(test_id,started,state=="READY_FOR_REVIEW" and not runtime.owns_store,"runtime close preserves caller-owned Store usability",{"owns_store":runtime.owns_store,"state_after_runtime_close":state},["owns_store","state_after_runtime_close"])
        if number==12:
            runtime=WorkflowRuntime(run_id="r6-owned"); store=runtime.store; runtime.close(); blocked=False
            try: runtime.execute("closed",valid_documents())
            except RuntimeError: blocked=True
            return _done(test_id,started,store.closed and blocked,"owned Store closes and closed runtime rejects work",{"store_closed":store.closed,"work_blocked":blocked},["store_closed","work_blocked"])
        if number==13:
            runtime=WorkflowRuntime(); runtime.close(); runtime.close()
            return _done(test_id,started,runtime.closed and runtime.store.closed,"runtime and Store close are idempotent",{"runtime_closed":runtime.closed,"store_closed":runtime.store.closed},["runtime_closed","store_closed"])
        if number==14:
            runtime=None
            try:
                with WorkflowRuntime() as runtime: raise ValueError("fixture")
            except ValueError: pass
            return _done(test_id,started,bool(runtime and runtime.closed and runtime.store.closed),"context manager closes owned Store on exception",{"runtime_closed":runtime.closed if runtime else None,"store_closed":runtime.store.closed if runtime else None},["runtime_closed","store_closed"])
        if number==15:
            with warnings.catch_warnings(record=True) as caught:
                warnings.simplefilter("error",ResourceWarning)
                for index in range(30):
                    runtime=WorkflowRuntime(run_id=f"r6-gc-{index}"); runtime.execute(f"case-gc-{index}",valid_documents(),stop_after_ready=True); runtime.close()
                gc.collect()
            return _done(test_id,started,not caught,"runtime lifecycle passes ResourceWarning error-mode probe",{"resource_warning_count":len(caught)},["resource_warning_count"])
        if 16<=number<=22:
            scenario="transient_before_commit" if number==17 else "timeout_after_commit"; rows=run_capability_ablation((101,),(scenario,)); by={row["profile"]:row for row in rows}; a1,a2=by["A1"],by["A2"]
            checks={16:evidence.a1_a2_behavior_gate(rows),17:a2["retry_attempt_count"]>0 and a2["final_state"]=="READY_FOR_REVIEW",18:a2["committed_result_reuse_count"]>0,19:a2["duplicate_side_effects"]==0,20:a1["retry_attempt_count"]==0 and a1["retry_pending_count"]==0 and a1["committed_result_reuse_count"]==0,21:a2["approval_request_count"]==0 and a2["business_updates"]==0,22:not evidence.a1_a2_behavior_gate([{**a1,"capability_branches":{"retry":False}},{**a1,"profile":"A2","capability_branches":{"retry":True}}])}
            return _done(test_id,started,checks[number],"A1 and A2 differ through safe retry/idempotency behavior without approval bypass",{"scenario":scenario,"a1":a1,"a2":a2,"gate":checks[number]},["a1","a2","gate"])
        if 23<=number<=27:
            with tempfile.TemporaryDirectory() as directory:
                root=pathlib.Path(directory); row=_producer_fixture(evidence,root); binding=root/"postpatch_evidence"/"raw_runs.sha256.json"
                if number==24: row["producer_command"]=[sys.executable,"run_all.py","--tests-only"]
                if number==25: row["producer_command"]=[sys.executable,"run_all.py","--unknown"]
                if number==26: row["raw_created_epoch"]=row["producer_ended_epoch"]+10
                if number==27: row["verification_command"]=list(row["producer_command"])
                evidence.json_write(binding,row); observed=evidence.validate_raw_producer_binding(root,"b"*64)
            expected=number==23
            return _done(test_id,started,observed["passed"] is expected,"raw producer and tests-only verifier are truthfully separated and time-bound",observed,["passed","checks"])
        if 28<=number<=32:
            with tempfile.TemporaryDirectory() as directory:
                root=pathlib.Path(directory); command=_linux_fixture(evidence,root); row=json.loads(command.read_text(encoding="utf-8"))
                if number==28: row["command"]=["python","run_all.py","--child-smoke"]
                if number==29: (root/row["stderr_file"]).write_text("Ran 160 tests\nFAILED\n",encoding="utf-8")
                if number==30: row["test_count"]=159
                if number==31: row["orphan_process_count"]=1
                if number==32: row["source_tree_sha256"]="d"*64
                evidence.json_write(command,row); evidence.json_write(root/(command.name+".sha256.json"),{"sha256":evidence.sha256(command),"size":command.stat().st_size}); observed=evidence.validate_linux_evidence_directory(root,"c"*64,("3.13",))
            return _done(test_id,started,not observed["passed"],"Linux command, transcript, counts, limits and source binding reject tamper",observed,["passed","rows"])
        if 33<=number<=37:
            with tempfile.TemporaryDirectory() as directory:
                root=pathlib.Path(directory); existing={f"R6-RG-{n:02d}" for n in range(1,38)}
                if number==35:
                    before={}; after={}; rows=[]
                    for idx in range(4):
                        path=f"m{idx}.py"; (root/path).write_text("def symbol():\n    return True\n",encoding="utf-8"); before[path]="a"*64; after[path]="b"*64; rows.append(_classification_row(path,"symbol","R6-RG-35","a"*64,"b"*64,origin="same",purpose="same"))
                else:
                    path="regressions_r5.py" if number==36 else "m.py"; (root/path).write_text("def real_symbol():\n    return True\n",encoding="utf-8"); before={path:"a"*64}; after={path:"b"*64}; rows=[_classification_row(path,"missing" if number==34 else "real_symbol","MISSING" if number==33 else f"R6-RG-{number:02d}","a"*64,"b"*64,origin="R4 correction" if number==36 else "R6-F06")]
                observed=evidence.validate_r6_integration_classification(rows,before,after,existing,root)
            expected=number==37
            return _done(test_id,started,observed["passed"] is expected,"classification validates real tests, symbols, unique meaning, generation and digests",observed,["passed","checks","boilerplate_detected"])
        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_r6()->list[dict[str,Any]]:
    return [asdict(run_r6_rg(f"R6-RG-{number:02d}")) for number in range(1,38)]
