"""E72 tests: resource-aware scheduling, real missions, cross-host evidence, recovery, memory, governance.

These tests read the artifacts produced by the E72 runners and exercise the deterministic modules directly. They do
not re-run the hour-long campaign or the model evaluations.
"""
from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
A = ROOT / "artifacts/e72"


def load(name):
    p = A / name
    if not p.is_file():
        pytest.skip(f"{name} not present")
    return json.loads(p.read_text())


def test_accounting_union_and_e70_reconciliation():
    from cain_workforce.e72.accounting import union_seconds
    assert union_seconds([(0, 10), (5, 15)]) == 15
    assert union_seconds([(0, 5), (10, 20)]) == 15
    b = load("baseline.json")
    r = b["e70_reconciliation"]
    assert r["original_e70_hours"] == 163.4645
    assert abs(r["recomputed_all_workers_hours"] - 163.894) < 0.02
    assert r["delta_matches_excluded_workers"] is True
    assert r["alternative_definitions"]["verified_productive_hours"] < r["alternative_definitions"]["all_workers_created_to_stop"]


def test_scheduling_admission_never_authority_and_never_zero_cap():
    from cain_workforce.e72.scheduling import AdaptiveResourceController
    c = AdaptiveResourceController(cpu_cap=8)
    c.soft_cap = 0.9999999999
    d = c.decide({"task_id": "t", "operation": "count_words", "params": {}, "timeout_s": 30}, in_flight=0)
    assert d["admit"] is True and d["grants_authority"] is False
    imp = c.decide({"task_id": "big", "operation": "hash_compute", "params": {"iterations": 90_000_000},
                    "timeout_s": 1.0}, in_flight=0)
    assert imp["reason"] == "ESTIMATE_EXCEEDS_LEASE"


def test_scheduling_results_show_oversubscription_fix():
    d = load("scheduling.json")
    assert all(r["mission_state"] == "COMPLETED" for r in d["levels"])
    assert d["comparison"]["max_busy_with_controller_at_64"] <= 8
    assert d["comparison"]["lease_expiries_controller"] == 0
    assert d["comparison"]["lease_expiries_naive_lease_failure"] > 0
    assert d["comparison"]["p95_naive_at_32"] > d["comparison"]["p95_controller_at_32"]


def test_model_discovery_does_not_invent_credentials():
    from cain_workforce.e72.models import discover_providers
    d = discover_providers()
    for name, p in d["remote"].items():
        assert set(p) >= {"credentials_present", "execution_class_if_called"}
        assert "key" not in json.dumps(p).lower() or "env" in json.dumps(p).lower()
    calls = load("models.json")
    assert calls["real_remote_calls"] == 0
    assert set(calls["execution_classes"]) == {"REAL_LOCAL_MODEL"}
    assert calls["fallback"]["recovered"] is True


def test_calibrated_routing_improves_on_heldout():
    d = load("router-policy.json")
    assert d["policies"]["P2_calibrated_min2"]["accuracy"] >= d["policies"]["P1_family_no_minimum"]["accuracy"]
    assert d["delta_P2_vs_P1"] >= 0


def test_five_real_missions_accepted_with_independent_proof():
    d = load("real-missions.json")
    assert d["accepted"] == 5 and d["completed"] == 5
    repo = d["missions"]["repo_software"]
    assert repo["test_failed_before"] and repo["test_passed_after"]
    assert repo["independent_verifier"]["buggy_failed"] and repo["independent_verifier"]["patched_passed"]
    sec = d["missions"]["security_review"]
    assert sec["reproduction_before"]["command_injection"] and not sec["reproduction_after_fix"]["command_injection"]
    assert sec["reproduction_before"]["path_traversal"] and not sec["reproduction_after_fix"]["path_traversal"]
    res = d["missions"]["research_literature"]
    assert all("arxiv.org" in s["id"] for s in res["sources"])
    data = d["missions"]["data_iris"]
    assert data["validation"]["rows"] == 150
    assert d["missions"]["infra_reliability"]["status"]["alive_after_kill"] is False


def test_cross_host_is_two_real_hosts_with_no_duplicates():
    d = load("cross-host.json")
    assert d["distinct_hosts"] is True
    assert d["distinct_public_ips"] is True
    r = d["result_from_host_b"]
    assert r["mission_state"] == "COMPLETED"
    assert len(r["verified_execute_by_host"]) == 2
    assert r["duplicate_results"] == 0
    assert d["local_evidence_verified"] is True


def test_recovery_worker_kill_and_cold_restart_but_reboot_blocked():
    d = load("recovery.json")
    assert d["status"] == "PASS"
    assert d["worker_kill"]["mission_state"] == "COMPLETED"
    assert d["worker_kill"]["max_attempt"] >= 2
    assert d["worker_kill"]["evidence_verified"] is True
    assert d["database_lock"]["failed_closed"] is True
    assert d["cold_restart"]["new_mission_state"] == "COMPLETED"
    assert d["host_reboot"] == "BLOCKED"


def test_memory_improves_heldout_and_never_authorizes():
    d = load("memory-eval.json")
    assert d["arms"]["B_verified_memory"]["accuracy"] > d["arms"]["A_no_memory"]["accuracy"]
    assert d["authority_violations"] == 0
    assert d["injection"]["quarantine_status"] == "QUARANTINED"
    assert d["injection"]["recalled"] is False
    assert d["supersession"]["current_answer_correct"] is True
    m = d.get("model_arm") or {}
    if "with_context" in m:
        assert m["with_context"]["correct"] >= m["no_context"]["correct"]


def test_self_improvement_is_measured_and_not_self_promoted():
    d = load("self-improvement.json")
    assert all(e["delta"] > 0 for e in d["experiments"])
    assert all(e["status"] == "MEASURED" for e in d["experiments"])
    assert d["history"]["ok"] is True
    ref = {r["case"]: r["refused"] for r in d["refusals"]}
    assert ref["protected_target"] and ref["self_promotion"] and ref["evaluator_substitution"]


def test_benchmark_reports_denominators_and_intervals():
    d = load("benchmark.json")
    for name, f in d["families"].items():
        if "n" not in f:
            continue
        assert f["n"] > 0
        assert abs(f["accuracy"] - f["correct"] / f["n"]) < 1e-9
        lo, hi = f["wilson95"]
        assert lo <= f["accuracy"] <= hi + 1e-9
    assert d["model"]["calls"] == len(d["calls"])
    assert d["model"]["execution_class"] == "REAL_LOCAL_MODEL"


def test_economics_smaller_team_is_more_efficient():
    d = load("economics.json")
    runs = {r["workers"]: r for r in d["runs"]}
    assert set(runs) == {4, 8}
    for r in runs.values():
        assert r["verified_productive_hours"] <= r["worker_hours_process_alive"]
        assert r["verified_execute"] == 240
    assert runs[4]["peak_pss_mb"] < runs[8]["peak_pss_mb"]


def test_security_lab_all_scenarios_blocked():
    d = load("security-lab.json")
    assert d["scenarios"] >= 72
    assert d["blocked"] == d["scenarios"]
    assert d["bypassed"] == 0


def test_clawx_is_local_and_tenant_isolated():
    d = load("clawx.json")
    assert d["deployed_publicly"] is False
    assert d["tenant_isolation"] is True
    assert d["server_check"]["authorized"] == 200
    assert d["server_check"]["wrong_token"] in (401, 403)
    assert d["server_check"]["other_tenant_denied"] in (401, 403)


def test_e72_offline_verifier_accepts_bundle_and_rejects_tamper(tmp_path):
    bundle = A / "bundle"
    if not bundle.is_dir():
        pytest.skip("E72 bundle not built")
    r = subprocess.run([sys.executable, str(ROOT / "scripts/evolution72/verify_e72_offline.py"), str(bundle), "--json"],
                       capture_output=True, text=True)
    assert r.returncode == 0, r.stdout + r.stderr
    out = json.loads(r.stdout)
    assert out["ok"] is True
    # tamper with a copied bundle
    import shutil
    bad = tmp_path / "bad"
    shutil.copytree(bundle, bad)
    target = bad / "artifacts/e72/baseline.json"
    target.write_text(target.read_text() + "\n ")
    r2 = subprocess.run([sys.executable, str(ROOT / "scripts/evolution72/verify_e72_offline.py"), str(bad), "--json"],
                        capture_output=True, text=True)
    assert r2.returncode != 0
    assert json.loads(r2.stdout)["ok"] is False


def test_mission_evidence_bundles_verify():
    verifier = ROOT / "scripts/evolution68/verify_e68_offline.py"
    for d in sorted((A / "missions").glob("*/evidence")):
        if not (d / "manifest.json").is_file():
            continue
        r = subprocess.run([sys.executable, str(verifier), str(d), "--json"], capture_output=True, text=True)
        assert r.returncode == 0, (d, r.stdout[-500:], r.stderr[-500:])


def test_agent_identity_cannot_promote_experiment(tmp_path):
    from cain_collective.evolution import EvolutionError, ExperimentRegistry
    reg = ExperimentRegistry(str(tmp_path / "exp.db"))
    try:
        e = reg.propose(hypothesis="h", target="model_routing", change="c", evaluator_hash="0" * 64,
                        heldout="x")
        # a measured experiment is required before promotion; an agent identity must be refused regardless
        try:
            reg.promote(e["experiment_id"], approver="agent:e72")
            raise AssertionError("agent identity was allowed to promote")
        except EvolutionError:
            pass
    finally:
        reg.close()


def test_hash_text_operation_is_deterministic_and_safe():
    from cain_workforce import ops
    a = ops.run_local("hash_text", {"text": "evidence"})
    b = ops.run_local("hash_text", {"text": "evidence"})
    assert a["sha256"] == b["sha256"] and a["grants_authority"] is False
    assert a["chars"] == 8