from __future__ import annotations

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

from .evaluator import IndependentEvaluator
from .harness import CAPABILITY_PROFILES, valid_documents
from .models import AgentAction, Budget, State, canonical_hash
from .policy import ExternalApprovalService, PolicyGateway
from .runtime import ScriptedAgent, WorkflowRuntime
from .storage import Store
from .tools import FaultInjector, ToolExecutor


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


class FakeClock:
    def __init__(self, value: float = 0.0) -> None: self.value = value
    def __call__(self) -> float: return self.value
    def advance(self, amount: float) -> None: self.value += amount


@dataclass(frozen=True)
class R2Result:
    test_id: str
    run_id: str
    passed: bool
    assertion_id: str
    assertion: str
    observed: dict[str, Any]
    latency_ms: float


def _done(test_id: str, started: float, passed: bool, assertion: str, observed: dict[str, Any]) -> R2Result:
    return R2Result(test_id, test_id.lower() + "-101", passed, test_id + ":primary", assertion, observed, round((time.perf_counter() - started) * 1000, 3))


def _flags(**changes: bool) -> dict[str, bool]:
    value = asdict(CAPABILITY_PROFILES["A4"]); value.update(changes); return value


def _seed_signature(seed: int) -> dict[str, Any]:
    agent = ScriptedAgent(seed); context = {"case_id": "case-seed", "run_id": "run-seed", "state": "READY_FOR_REVIEW"}
    actions = [asdict(agent.choose({**context, "phase": phase})) for phase in ("draft", "approval_request", "apply")]
    fault = FaultInjector.seeded(seed)
    return {"agent_trace_digest": canonical_hash(actions), "fault_plan": fault.plans, "trigger_attempts": fault.trigger_attempts}


def _behavior_digest(signature: dict[str, Any]) -> str:
    behavior = {
        "agent_trace_digest": signature.get("agent_trace_digest"),
        "fault_plan": signature.get("fault_plan", {}),
        "trigger_attempts": signature.get("trigger_attempts", {}),
    }
    return canonical_hash(behavior)


def _linux_probe() -> dict[str, Any]:
    native = subprocess.run([sys.executable, "run_all.py", "--child-smoke"], cwd=ROOT, text=True, capture_output=True, timeout=30)
    native_result = {"exit_code": native.returncode, "platform": sys.platform, "stdout_digest": hashlib.sha256(native.stdout.encode()).hexdigest(), "stderr_digest": hashlib.sha256(native.stderr.encode()).hexdigest()}
    docker = shutil.which("docker")
    if not docker: return {"passed": native.returncode == 0, "native": native_result, "docker_status": "SKIPPED_NONBLOCKING"}
    command = [docker, "run", "--rm", "-v", f"{ROOT}:/workspace", "-w", "/workspace", "python:3.12-slim", "python", "run_all.py", "--child-smoke"]
    try:
        result = subprocess.run(command, text=True, capture_output=True, timeout=45)
        return {"passed": native.returncode == 0, "native": native_result, "docker_status": "PASS_OPTIONAL" if result.returncode == 0 else "FAIL_NONBLOCKING", "docker": {"exit_code": result.returncode, "platform": "linux_container", "stdout_digest": hashlib.sha256(result.stdout.encode()).hexdigest(), "stderr_digest": hashlib.sha256(result.stderr.encode()).hexdigest()}}
    except Exception as error:
        return {"passed": native.returncode == 0, "native": native_result, "docker_status": "FAIL_NONBLOCKING", "docker_error": type(error).__name__}


def _baseline_rows() -> tuple[dict[str, Any], list[dict[str, Any]]]:
    payload = json.loads((ROOT / "fixtures" / "r2_prepatch_raw.json").read_text(encoding="utf-8"))
    return payload, list(payload["rows"])


def run_r2_rg(test_id: str) -> R2Result:
    started = time.perf_counter(); number = int(test_id.rsplit("-", 1)[1]); run_id = test_id.lower() + "-101"; case_id = "same-case"
    try:
        if number == 1:
            observed = _linux_probe(); return _done(test_id, started, observed["passed"], "native current-OS Python is required and Docker is optional", observed)
        if number == 2:
            p = subprocess.run([sys.executable, str(ROOT/"tests"/"fixtures"/"minimal_suite.py")], cwd=ROOT, text=True, capture_output=True, timeout=20, env={**os.environ,"PYTHONDONTWRITEBYTECODE":"1"})
            return _done(test_id, started, p.returncode == 0, "independent leaf fixture terminates without re-entering product suites", {"exit_code": p.returncode, "elapsed": round(time.perf_counter()-started,3)})
        if number == 3:
            from scripts.build_evidence import evidence_recursion_guard
            blocked = not evidence_recursion_guard(1)
            return _done(test_id, started, blocked, "nested evidence level is rejected by the pure recursion guard", {"blocked": blocked})
        if number == 4:
            from scripts.build_evidence import platform_gate, platform_matrix
            matrix = platform_matrix("", ROOT); decision_blocks_python_failure = not platform_gate(1, "SKIPPED_NONBLOCKING")
            return _done(test_id, started, matrix["powershell"]["status"] == "SKIPPED_NONBLOCKING" and matrix["passed"] and decision_blocks_python_failure, "PowerShell absence is nonblocking while Python remains required", matrix)
        if number == 5:
            runtime = WorkflowRuntime(run_id=run_id, capabilities=_flags(retry=False), faults=FaultInjector(7, {"apply_mock_update": "transient_before_commit"})); waiting = runtime.execute(case_id, valid_documents()); ExternalApprovalService(runtime.store, runtime.policy).decide(waiting["pending_approval_id"], "external"); final = runtime.continue_ready(case_id, valid_documents(), 2)
            starts = sum(row["tool"] == "apply_mock_update" and row["phase"] == "STARTED" for row in runtime.store.rows_for_case("journal", case_id, run_id))
            passed = final["state"] != State.COMPLETED.value and not final["backoff_history"] and starts == 1
            return _done(test_id, started, passed, "retry=false performs no retry or backoff", {"state": final["state"], "starts": starts, "backoff": final["backoff_history"]})
        if number == 6:
            runtime = WorkflowRuntime(run_id=run_id, capabilities=_flags(idempotency=False), faults=FaultInjector(7, {"apply_mock_update": "timeout_after_commit"})); waiting = runtime.execute(case_id, valid_documents()); ExternalApprovalService(runtime.store, runtime.policy).decide(waiting["pending_approval_id"], "external")
            try: runtime.continue_ready(case_id, valid_documents(), 1)
            except Exception: pass
            phases = [row["phase"] for row in runtime.store.rows_for_case("journal", case_id, run_id)]
            return _done(test_id, started, "IDEMPOTENCY_DISABLED" in phases and "RESULT_REUSED" not in phases, "idempotency=false disables result reuse path", {"disabled_events": phases.count("IDEMPOTENCY_DISABLED"), "reuse_events": phases.count("RESULT_REUSED")})
        if number == 7:
            off = WorkflowRuntime(run_id="policy-off", capabilities=_flags(state_policy=False, external_approval=False)); off.execute("case-policy", valid_documents()); on = WorkflowRuntime(run_id="policy-on", capabilities=_flags(external_approval=False)); on.execute("case-policy", valid_documents())
            off_count = sum(row["phase"] == "STATE_POLICY_DISABLED" for row in off.store.rows_for_case("journal", "case-policy", "policy-off")); on_count = sum(row["phase"] == "STATE_POLICY_DISABLED" for row in on.store.rows_for_case("journal", "case-policy", "policy-on"))
            return _done(test_id, started, off_count > 0 and on_count == 0, "state_policy flag changes actual policy trace", {"off": off_count, "on": on_count})
        if number == 8:
            runtime = WorkflowRuntime(run_id=run_id, capabilities=_flags(external_approval=False)); final = runtime.execute(case_id, valid_documents()); approvals = runtime.store.rows_for_case("approvals", case_id, run_id)
            return _done(test_id, started, not approvals and final["pending_approval_id"] is None, "external_approval=false creates no request or decision", {"approvals": len(approvals), "state": final["state"]})
        if number == 9:
            runtime = WorkflowRuntime(run_id=run_id, capabilities=_flags(checkpoint=False, external_approval=False)); runtime.execute(case_id, valid_documents()); blocked = False
            try: runtime.continue_ready(case_id, valid_documents())
            except Exception: blocked = True
            return _done(test_id, started, blocked and runtime.store.checkpoint(run_id, case_id) is None, "checkpoint=false cannot resume", {"blocked": blocked})
        if number == 10:
            runtime = WorkflowRuntime(run_id=run_id, capabilities=_flags(independent_evaluation=False)); waiting = runtime.execute(case_id, valid_documents()); ExternalApprovalService(runtime.store, runtime.policy).decide(waiting["pending_approval_id"], "external"); final = runtime.continue_ready(case_id, valid_documents())
            tokens = runtime.store.rows_for_case("completion_tokens", case_id, run_id)
            return _done(test_id, started, final["state"] != State.COMPLETED.value and not tokens, "independent evaluation disabled forbids token and completion", {"state": final["state"], "tokens": len(tokens)})
        if number == 11:
            from .harness import run_capability_ablation
            rows = run_capability_ablation((101,), ("normal",))
            by_profile = {row["profile"]: row for row in rows}
            expected = {
                "A0": lambda row: row["final_state"] == State.READY_FOR_REVIEW.value and not row["checkpoint_present"] and row["approval_request_count"] == 0,
                "A1": lambda row: row["final_state"] == State.READY_FOR_REVIEW.value and row["checkpoint_present"] and row["approval_request_count"] == 0,
                "A2": lambda row: row["final_state"] == State.READY_FOR_REVIEW.value and row["checkpoint_present"] and row["approval_request_count"] == 0,
                "A3": lambda row: row["final_state"] != State.COMPLETED.value and row["business_updates"] == 1 and row["completion_token_count"] == 0,
                "A4": lambda row: row["final_state"] == State.COMPLETED.value and row["business_updates"] == 1 and row["completion_token_count"] == 1,
            }
            behaviors = {name: bool(expected[name](by_profile[name])) and by_profile[name]["capability_branches"] == asdict(CAPABILITY_PROFILES[name]) for name in expected}
            observed = {name: {key: by_profile[name][key] for key in ("final_state", "business_updates", "checkpoint_present", "approval_request_count", "completion_token_count", "capability_branches")} for name in expected}
            return _done(test_id, started, all(behaviors.values()), "A0-A4 execute distinct expected state, persistence, approval and evaluation paths", {"behaviors": behaviors, "observed": observed})
        if number == 12:
            store = Store(); store.create_case(case_id, State.RECEIVED.value, run_id="run-a"); store.create_case(case_id, State.CANCELLED.value, run_id="run-b")
            a = ToolExecutor(store, PolicyGateway(store), "run-a").execute(case_id, case_id, "get_case", {"case_id": case_id}, "read-a"); b = ToolExecutor(store, PolicyGateway(store), "run-b").execute(case_id, case_id, "get_case", {"case_id": case_id}, "read-b")
            return _done(test_id, started, a["state"] != b["state"] and a["run_id"] != b["run_id"], "read tools isolate same case across runs", {"a": a, "b": b})
        if number == 13:
            store = Store(); a = WorkflowRuntime(store=store, run_id="run-a"); a.execute(case_id, valid_documents(), stop_after_ready=True); store.create_case(case_id, State.READY_FOR_REVIEW.value, run_id="run-b"); store.log("run-b", case_id, "COMPLETED", {}, "foreign-request", "get_case"); store.add_effect(case_id, "save_review_draft", "foreign-effect", "hash", {"ok": True}, "run-b"); cp = a.checkpoint(case_id); text = json.dumps(cp)
            return _done(test_id, started, "foreign-request" not in text and "foreign-effect" not in text, "checkpoint excludes foreign run identifiers", {"checkpoint_digest": canonical_hash(cp)})
        if number == 14:
            store = Store(); store.create_case(case_id, State.APPROVED.value, run_id="run-a"); store.init_business(case_id,"run-a","synthetic-review-record"); store.create_case(case_id, State.APPROVED.value, run_id="run-b"); store.init_business(case_id,"run-b","synthetic-review-record")
            args={"case_id":case_id,"target":"synthetic-review-record","value":"REVIEW_RECORDED","expected_version":0}; store.request_approval({"id":"foreign","run_id":"run-b","case_id":case_id,"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_id,"run-a","apply_mock_update",args)
            except Exception: approval_blocked=True
            store.add_effect(case_id,"save_review_draft","foreign-effect",canonical_hash({}),{"ok":True},"run-b"); effect_blocked=store.effect("foreign-effect","run-a",case_id) is None
            token = canonical_hash("run-b-token"); digest = canonical_hash("run-b-eval"); store.issue_completion_token(token, "run-b", case_id, digest); token_blocked = False
            try: store.consume_completion_token(token, "run-a", case_id, digest)
            except Exception: token_blocked = True
            observed={"approval_blocked":approval_blocked,"effect_blocked":effect_blocked,"token_blocked":token_blocked}
            return _done(test_id, started, all(observed.values()), "foreign run approval, effect and token are each independently blocked", observed)
        if number == 15:
            store = Store(); store.create_case(case_id, State.APPROVED.value, run_id="run-a"); store.init_business(case_id, "run-a", "synthetic-review-record"); store.create_case(case_id, State.APPROVED.value, run_id="run-b"); store.init_business(case_id, "run-b", "synthetic-review-record"); store.db.execute("UPDATE business_state SET value='REVIEW_RECORDED',update_count=1 WHERE case_id=? AND run_id='run-b'", (case_id,)); store.db.commit(); ev = IndependentEvaluator().evaluate_precompletion(store, case_id, "run-a", {State.APPROVED.value})
            return _done(test_id, started, not ev.passed and "business_target_not_reached" in ev.reasons, "evaluator ignores other run business state", {"reasons": ev.reasons})
        if number in {16, 17, 18}:
            clock = FakeClock();
            class ExpiringAgent(ScriptedAgent):
                def choose(self, context):
                    action = super().choose(context)
                    if number == 17 and context["phase"] == "apply": clock.advance(2)
                    return action
            runtime = WorkflowRuntime(run_id=run_id, clock=clock, sleeper=lambda _: None, agent=ExpiringAgent(1)); waiting = runtime.execute(case_id, valid_documents()); service = ExternalApprovalService(runtime.store, runtime.policy)
            if number == 18: service.decide(waiting["pending_approval_id"], "external", decision="REJECTED")
            else: service.decide(waiting["pending_approval_id"], "external", ttl=1)
            if number == 16: clock.advance(2)
            blocked = False
            try: runtime.continue_ready(case_id, valid_documents())
            except Exception: blocked = True
            business = runtime.store.business(case_id, run_id, "synthetic-review-record"); approvals=runtime.store.rows_for_case("approvals",case_id,run_id); apply_effects=[row for row in runtime.store.rows_for_case("effects",case_id,run_id) if row["tool"]=="apply_mock_update"]
            observed={"blocked":blocked,"state":runtime.store.state(case_id,run_id),"updates":business["update_count"],"apply_effects":len(apply_effects),"approval_used":sum(int(row["used"]) for row in approvals)}
            passed=observed["state"]!=State.COMPLETED.value and observed["updates"]==0 and observed["apply_effects"]==0 and observed["approval_used"]==0
            return _done(test_id, started, passed, "expired/rejected approval leaves state uncompleted with zero business update, apply effect and approval use", observed)
        if number in {19, 20, 21}:
            from scripts.build_evidence import build_semantic_traceability, verify_semantic_traceability
            with tempfile.TemporaryDirectory() as directory:
                root = pathlib.Path(directory); path = root / "module.py"; path.write_text("def real_symbol():\n    return True\n", encoding="utf-8")
                raw = [{"test_id":"TEST-X","run_id":"run-x","passed":True,"assertion":"real","assertion_id":"ASSERT-X"}]
                mapping = {"REQ-X":{"code_locations":["module.py:1-2"],"test_ids":["TEST-X"],"symbols":["real_symbol"],"assertion_ids":["ASSERT-X"],"expected_semantics":"real symbol"}}
                if number == 19: mapping["REQ-X"]["code_locations"] = ["module.py:99-100"]
                if number == 20: mapping["REQ-X"]["symbols"] = ["missing_symbol"]; raw[0]["assertion_id"] = "ASSERT-OTHER"
                records = build_semantic_traceability(mapping, raw, root)
                if number == 21:
                    path.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"
                return _done(test_id, started, passed, "traceability mutant is rejected", {"status": records[0]["status"]})
        if number in {22, 23}:
            signatures = [_seed_signature(seed) for seed in ([101,101,101] if number == 22 else [101,202,303])]; digests = [_behavior_digest(item) for item in signatures]
            passed = len(set(digests)) == (1 if number == 22 else len(set(digests))) and (number == 22 or len(set(digests)) >= 2)
            return _done(test_id, started, passed, "seed controls deterministic and diverse agent/fault plans", {"digests": digests, "unique": len(set(digests))})
        if number in {24, 25}:
            payload, rows = _baseline_rows(); digest = canonical_hash(rows); summary = {"row_count":len(rows),"failed":sum(not row["passed"] for row in rows),"findings":sorted(row["finding"] for row in rows),"source_hash":payload["source_tree_sha256"],"raw_digest":digest}
            if number == 24: passed = len(summary["findings"]) == len(set(summary["findings"])) and all(len(row["raw_output_digest"]) == 64 for row in rows) and len(summary["source_hash"]) == 64
            else: passed = canonical_hash(rows[:-1]) != summary["raw_digest"]
            return _done(test_id, started, passed, "PRE_PATCH report is recomputed from immutable raw rows", summary)
        if number in {26, 27, 28}:
            allowed = {"get_case", "save_review_draft"}; payload = {"tool":"get_case","arguments":{"case_id":"c"},"request_id":"r","idempotency_key":None}
            if number == 26: payload["request_id"] = 123
            if number == 27: payload["idempotency_key"] = ""
            if number == 28: payload = {"tool":"save_review_draft","arguments":{"case_id":"c","draft_version":1},"request_id":"r"}
            blocked = False
            try: AgentAction.from_payload(payload, allowed)
            except Exception: blocked = True
            return _done(test_id, started, blocked, "strict AgentAction rejects invalid identifiers before execution", {"blocked": blocked})
        if number in {29, 30, 31}:
            clock = FakeClock(10); budget = Budget()
            if number == 29: budget.max_steps = 0
            if number == 30: budget.max_agent_calls = budget.max_tool_calls = budget.max_cost_units = 0
            runtime = WorkflowRuntime(run_id=run_id, budget=budget, clock=clock, deadline_seconds=(-1 if number == 31 else 30)); blocked = False
            try: runtime.execute(case_id, valid_documents())
            except Exception: blocked = True
            counts={table:int(runtime.store.db.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]) for table in ("cases","business_state","transitions","effects","approvals","journal")}; rejections=int(runtime.store.db.execute("SELECT COUNT(*) FROM run_rejections").fetchone()[0])
            passed = blocked and all(value==0 for value in counts.values()) and rejections==1
            return _done(test_id, started, passed, "resource/deadline preflight creates only a rejection audit and no normal workflow rows", {"counts":counts,"rejections":rejections})
        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)})


def run_all_r2() -> list[dict[str, Any]]:
    return [asdict(run_r2_rg(f"R2-RG-{number:02d}")) for number in range(1, 32)]
