from __future__ import annotations

import ast
import hashlib
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 valid_documents
from .models import Budget, CheckpointIncompatible, State, canonical_hash
from .policy import ExternalApprovalService, PolicyGateway
from .runtime import WorkflowRuntime
from .storage import Store
from .tools import ToolExecutor


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


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


def _table_counts(store: Store) -> dict[str, int]:
    return {name: int(store.db.execute(f"SELECT COUNT(*) FROM {name}").fetchone()[0]) for name in ("cases", "business_state", "transitions", "effects", "approvals")}


def _run(command: list[str], timeout: float, env: dict[str, str] | None = None) -> dict[str, Any]:
    started = time.perf_counter()
    try:
        process = subprocess.run(command, cwd=ROOT, text=True, capture_output=True, timeout=timeout, env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1", **(env or {})})
        return {"exit_code": process.returncode, "elapsed": round(time.perf_counter() - started, 3), "stdout_digest": hashlib.sha256(process.stdout.encode()).hexdigest(), "stderr_digest": hashlib.sha256(process.stderr.encode()).hexdigest()}
    except subprocess.TimeoutExpired:
        return {"exit_code": 124, "elapsed": round(time.perf_counter() - started, 3), "timeout": True}


def _approval_effect_counts(runtime: WorkflowRuntime, case_id: str) -> dict[str, int]:
    approvals = runtime.store.rows_for_case("approvals", case_id, runtime.run_id)
    business = runtime.store.business(case_id, runtime.run_id, "synthetic-review-record")
    apply_effects = [row for row in runtime.store.rows_for_case("effects", case_id, runtime.run_id) if row["tool"] == "apply_mock_update"]
    return {"used": sum(int(row["used"]) for row in approvals), "effects": len(apply_effects), "updates": int((business or {}).get("update_count", 0))}


def run_r3_rg(test_id: str) -> R3Result:
    started = time.perf_counter(); number = int(test_id.rsplit("-", 1)[1]); docs = valid_documents()
    try:
        if number == 1:
            observed = _run([sys.executable, "run_all.py", "--child-smoke"], 30)
            return _done(test_id, started, observed["exit_code"] == 0, "native current-OS Python child-smoke passes without a container", observed, ["exit_code", "elapsed"])
        if number == 2:
            import reliability_core.regressions_r2 as r2
            original = r2.shutil.which
            try:
                r2.shutil.which = lambda _: None
                observed = r2._linux_probe()
            finally: r2.shutil.which = original
            return _done(test_id, started, observed.get("passed") is True and observed.get("docker_status") == "SKIPPED_NONBLOCKING", "Docker absence is separately recorded and cannot fail native execution", observed, ["passed", "docker_status"])
        if number == 3:
            observed = _run([sys.executable, str(ROOT/"tests"/"fixtures"/"minimal_suite.py")], 30, {"PATH": str(pathlib.Path(sys.executable).parent)})
            return _done(test_id, started, observed["exit_code"] == 0 and observed["elapsed"] < 30, "separate minimal suite passes with PowerShell and Docker removed from PATH", observed, ["exit_code", "elapsed"])
        if number == 4:
            from scripts.build_evidence import run_command
            observed = run_command([sys.executable, str(ROOT/"tests"/"fixtures"/"minimal_suite.py")], ROOT, timeout_seconds=20)
            passed = observed["exit_code"] == 0 and not observed.get("timed_out") and observed.get("orphan_process_count") == 0 and observed["elapsed_ms"] < 20000
            return _done(test_id, started, passed, "evidence subprocess is bounded and leaves zero children", {key: value for key, value in observed.items() if key not in {"stdout", "stderr"}}, ["exit_code", "timed_out", "orphan_process_count", "elapsed_ms"])
        if number == 5:
            store = Store(); a = WorkflowRuntime(store=store, run_id="run-a"); b = WorkflowRuntime(store=store, run_id="run-b")
            wa = a.execute("same-case", docs); wb = b.execute("same-case", docs)
            ExternalApprovalService(store, a.policy).decide(wa["pending_approval_id"], "external-a"); ExternalApprovalService(store, b.policy).decide(wb["pending_approval_id"], "external-b")
            errors: list[str] = []; results = []
            for runtime in (a, b):
                try: results.append(runtime.continue_ready("same-case", docs))
                except Exception as error: errors.append(type(error).__name__)
            passed = not errors and [row["state"] for row in results] == [State.COMPLETED.value, State.COMPLETED.value] and all((store.business("same-case", run, "synthetic-review-record") or {})["update_count"] == 1 for run in ("run-a", "run-b"))
            return _done(test_id, started, passed, "same case identifier completes and returns normally in both isolated runs", {"errors": errors, "states": [row["state"] for row in results]}, ["errors", "states"])
        if number == 6:
            violations = []
            for relative in ("src/reliability_core/runtime.py", "src/reliability_core/tools.py", "src/reliability_core/evaluator.py"):
                tree = ast.parse((ROOT / relative).read_text(encoding="utf-8"))
                for node in ast.walk(tree):
                    if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr in {"state", "effect", "add_effect", "approval"}:
                        minimum = 3 if node.func.attr in {"effect", "add_effect"} else 2
                        if len(node.args) < minimum: violations.append(f"{relative}:{node.lineno}:{node.func.attr}")
            return _done(test_id, started, not violations, "core Store reads and effects always carry case and run identity", {"violations": violations}, ["violations"])
        if number in {7, 8}:
            store = Store(); a = WorkflowRuntime(store=store, run_id="run-a"); a.execute("case-a", docs); checkpoint = store.checkpoint("run-a", "case-a")
            if number == 7:
                target = WorkflowRuntime(store=store, run_id="run-b"); target.execute("case-a", docs); expected_case = "case-a"
            else:
                target = WorkflowRuntime(store=store, run_id="run-a-2"); target.execute("case-b", docs); checkpoint = dict(checkpoint or {}); checkpoint["run_id"] = "run-a-2"; expected_case = "case-b"
            blocked = False
            try: target.resume(checkpoint or {}, docs, expected_case)
            except CheckpointIncompatible: blocked = True
            return _done(test_id, started, blocked, "checkpoint run, case, key and active runtime ownership must all match", {"blocked": blocked, "expected_case": expected_case}, ["blocked", "expected_case"])
        if number in {9, 10, 11}:
            from scripts.build_evidence import build_semantic_traceability, verify_semantic_traceability
            with tempfile.TemporaryDirectory() as directory:
                root = pathlib.Path(directory); source = root / "m.py"; source.write_text("def real_symbol():\n    return True\n", encoding="utf-8")
                raw = [{"test_id":"REAL","run_id":"r1","passed":True,"assertion_id":"REAL:primary","assertion_contract_id":"REAL:contract","assertion":"real","observed":{"value":True}}]
                mapping = {"REQ":{"code_locations":["m.py:1-2"],"symbols":["real_symbol"],"test_ids":["REAL"],"assertion_ids":["REAL:primary"],"assertion_contracts":{"REAL":{"contract_id":"REAL:contract","observed_fields":["value"],"predicate":"passed_is_true"}},"expected_semantics":"real behavior"}}
                if number == 9:
                    mapping["REQ"]["test_ids"].append("MISSING"); mapping["REQ"]["assertion_ids"].append("MISSING:primary"); mapping["REQ"]["assertion_contracts"]["MISSING"] = {"contract_id":"MISSING:contract","observed_fields":["value"],"predicate":"passed_is_true"}
                if number == 10: raw[0]["assertion_contract_id"] = "UNRELATED:contract"
                records = build_semantic_traceability(mapping, raw, root)
                if number == 11:
                    source.write_text("def real_symbol():\n    return False\n", encoding="utf-8"); passed = not verify_semantic_traceability(records, raw, root)
                else: passed = records[0]["status"] == "FAIL"
                observed = {"status": records[0]["status"], "linked_test_ids": records[0].get("linked_test_ids", [])}
            return _done(test_id, started, passed, "missing tests, mismatched assertion contracts and changed code are rejected", observed, ["status", "linked_test_ids"])
        if number == 12:
            import reliability_core.regressions_r2 as r2
            original = r2._seed_signature
            try:
                r2._seed_signature = lambda seed: {"seed": seed, "agent_trace_digest":"constant", "fault_plan_digest":"constant", "fault_plan":{}, "trigger_attempts":{}}
                mutant = r2.run_r2_rg("R2-RG-23")
            finally: r2._seed_signature = original
            return _done(test_id, started, not mutant.passed, "different seed numbers cannot rescue constant agent and fault behavior", {"mutant_passed": mutant.passed}, ["mutant_passed"])
        if number == 13:
            import reliability_core.regressions_r2 as r2
            same = [r2._behavior_digest(r2._seed_signature(101)) for _ in range(3)]; diverse = [r2._behavior_digest(r2._seed_signature(seed)) for seed in (101, 202, 303)]
            return _done(test_id, started, len(set(same)) == 1 and len(set(diverse)) >= 2, "behavior-only digests are deterministic and diverse", {"same_unique": len(set(same)), "different_unique": len(set(diverse)), "digests": diverse}, ["same_unique", "different_unique"])
        if number == 14:
            payload = json.loads((ROOT / "fixtures" / "r2_prepatch_raw.json").read_text(encoding="utf-8")); required = {"command","captured_at","platform","source_result_zip_sha256","raw_output_ref","raw_output_digest"}; rows = payload.get("rows", [])
            valid = all(required.issubset(row) and (ROOT / row["raw_output_ref"]).is_file() and hashlib.sha256((ROOT / row["raw_output_ref"]).read_bytes()).hexdigest() == row["raw_output_digest"] for row in rows)
            return _done(test_id, started, valid and len({row["raw_output_digest"] for row in rows}) == len(rows), "every PRE_PATCH row replays to a distinct referenced raw output", {"rows":len(rows),"valid":valid,"unique_digests":len({row.get("raw_output_digest") for row in rows})}, ["rows", "valid", "unique_digests"])
        if number in {15, 16}:
            variants = [Budget(max_steps=0), Budget(max_calls=0), Budget(max_agent_calls=0), Budget(max_tool_calls=0), Budget(max_cost_units=0)] if number == 15 else [Budget()]
            observations = []
            for index, budget in enumerate(variants):
                store = Store(); runtime = WorkflowRuntime(store=store, run_id=f"reject-{number}-{index}", budget=budget, deadline_seconds=(-1 if number == 16 else 30)); blocked = False
                try: runtime.execute(f"case-{index}", docs)
                except Exception: blocked = True
                observations.append({"blocked":blocked,"counts":_table_counts(store)})
            passed = all(item["blocked"] and all(value == 0 for value in item["counts"].values()) for item in observations)
            return _done(test_id, started, passed, "resource and deadline rejection creates no normal durable workflow rows", {"observations":observations}, ["observations"])
        if number == 17:
            runtime = WorkflowRuntime(run_id="atomic"); waiting = runtime.execute("case-atomic", docs); ExternalApprovalService(runtime.store, runtime.policy).decide(waiting["pending_approval_id"], "external")
            original = runtime.store.set_state
            def fail_complete(case_id, old, new, reason, run_id=None, created_at=0.0):
                if new == State.COMPLETED.value: raise ValueError("simulated_cas_failure")
                return original(case_id, old, new, reason, run_id, created_at)
            runtime.store.set_state = fail_complete
            try: runtime.continue_ready("case-atomic", docs)
            except Exception: pass
            runtime.store.set_state = original
            tokens_before = runtime.store.rows_for_case("completion_tokens", "case-atomic", "atomic"); state_before = runtime.store.state("case-atomic", "atomic")
            evaluation = runtime.evaluator.evaluate_precompletion(runtime.store, "case-atomic", "atomic", {State.PARSING.value,State.EXTRACTING.value,State.VALIDATING.value,State.POLICY_CHECK.value,State.READY_FOR_REVIEW.value,State.APPROVED.value})
            token = canonical_hash({"retry": evaluation.evaluation_digest}); retried = False
            try: runtime.transition("case-atomic", State.COMPLETED, "atomic retry", completion_token=token, evaluation_digest=evaluation.evaluation_digest); retried = True
            except Exception: pass
            passed = state_before == State.APPROVED.value and all(int(row["used"]) == 0 for row in tokens_before) and retried and runtime.store.state("case-atomic", "atomic") == State.COMPLETED.value
            return _done(test_id, started, passed, "completion token issue, consume and state transition rollback and retry atomically", {"state_before":state_before,"tokens_before":tokens_before,"retried":retried}, ["state_before", "tokens_before", "retried"])
        if number == 18:
            case = "same"; store = Store(); store.create_case(case, State.APPROVED.value, run_id="run-a"); store.init_business(case,"run-a","synthetic-review-record"); store.create_case(case, State.APPROVED.value, run_id="run-b"); store.init_business(case,"run-b","synthetic-review-record")
            args={"case_id":case,"target":"synthetic-review-record","value":"REVIEW_RECORDED","expected_version":0}; store.request_approval({"id":"foreign","run_id":"run-b","case_id":case,"tool":"apply_mock_update","args_hash":canonical_hash(args),"requested_at":0.0,"policy_version":"policy-v1"}); store.decide_approval("foreign","APPROVED","external",0.0,10.0)
            approval_blocked=False
            try: PolicyGateway(store, clock=lambda:1.0).validate_approval("foreign",case,"run-a","apply_mock_update",args)
            except Exception: approval_blocked=True
            store.add_effect(case,"save_review_draft","foreign-effect",canonical_hash({}),{"ok":True},"run-b"); effect_blocked=store.effect("foreign-effect","run-a",case) is None
            digest=canonical_hash("foreign"); store.issue_completion_token("foreign-token","run-b",case,digest); token_blocked=False
            try: store.consume_completion_token("foreign-token","run-a",case,digest)
            except Exception: token_blocked=True
            return _done(test_id, started, approval_blocked and effect_blocked and token_blocked, "foreign approval, effect and completion token attacks are independently blocked", {"approval_blocked":approval_blocked,"effect_blocked":effect_blocked,"token_blocked":token_blocked}, ["approval_blocked", "effect_blocked", "token_blocked"])
        if number == 19:
            results=[]
            for decision in ("APPROVED", "REJECTED"):
                clock_value=[0.0]; runtime=WorkflowRuntime(run_id="decision-"+decision.lower(),clock=lambda:clock_value[0],sleeper=lambda _:None); waiting=runtime.execute("case",docs); ExternalApprovalService(runtime.store,runtime.policy).decide(waiting["pending_approval_id"],"external",decision=decision,ttl=1); clock_value[0]=2.0
                try: runtime.continue_ready("case",docs)
                except Exception: pass
                counts=_approval_effect_counts(runtime,"case"); results.append({"decision":decision,"state":runtime.store.state("case",runtime.run_id),**counts})
            passed=all(row["state"]!=State.COMPLETED.value and row["updates"]==0 and row["effects"]==0 and row["used"]==0 for row in results)
            return _done(test_id, started, passed, "expired or rejected approval cannot add business effect or consume approval", {"results":results}, ["results"])
        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_r3() -> list[dict[str, Any]]:
    return [asdict(run_r3_rg(f"R3-RG-{number:02d}")) for number in range(1, 20)]
