"""Repeatable Mock acceptance probes for V01-V13.

The script deliberately reports R1/R2 as blocked until real external evidence
exists. Passing Mock probes never upgrades the result to production readiness.
"""

from __future__ import annotations

import copy
import io
import json
import tempfile
import time
from collections.abc import Callable
from contextlib import redirect_stdout
from pathlib import Path
from types import TracebackType
from typing import Any, Self

from agent_call.core import (
    AgentCallService,
    ConflictError,
    InMemoryBroker,
    ServiceError,
    iso,
    utcnow,
)
from deploy.backup_sqlite import copy_database
from deploy.check_mock_config import check as check_profile

ROOT = Path(__file__).resolve().parents[1]


def load_fixture() -> dict[str, Any]:
    try:
        return json.loads(
            (ROOT / "docs/contracts/examples/call.execute.json").read_text(
                encoding="utf-8"
            )
        )
    except (OSError, json.JSONDecodeError) as exc:
        raise RuntimeError("contract fixture is unavailable") from exc


def service(directory: str, broker: InMemoryBroker | None = None) -> AgentCallService:
    return AgentCallService(
        db_path=Path(directory) / "executor.sqlite3",
        object_dir=Path(directory) / "objects",
        broker=broker or InMemoryBroker(),
    )


def command(
    command_id: str, tenant_id: str = "tenant-demo", task_id: str = "task-demo"
) -> dict[str, Any]:
    body = copy.deepcopy(load_fixture())
    keys = {
        "tenant-demo": "tenant-demo-key",
        "tenant-b": "tenant.b",
        "tenant-c": "tenant#c",
    }
    body["command_id"] = command_id
    body["trace_id"] = f"trace_{command_id}"
    body["tenant_id"] = tenant_id
    body["tenant_key"] = keys[tenant_id]
    body["issued_at"] = iso(utcnow())
    body["payload"]["execution_id"] = f"exec_{command_id}"
    body["payload"]["task_id"] = task_id
    return body


def run_success(svc: AgentCallService, body: dict[str, Any]) -> dict[str, Any]:
    svc.publish_execute(body)
    svc.wait_for_idle()
    return svc.get_command(body["tenant_id"], body["command_id"])


def assert_raises(code: str, action: Callable[[], Any]) -> None:
    try:
        action()
    except ServiceError as exc:
        if exc.code != code:
            raise AssertionError(f"expected {code}, got {exc.code}") from exc
        return
    raise AssertionError(f"expected {code}")


def probe_v01() -> dict[str, Any]:
    with tempfile.TemporaryDirectory() as directory:
        svc = service(directory)
        problems = check_profile(ROOT / "docs/contracts/mock-profile.json")
        if problems:
            raise AssertionError("; ".join(problems))
        invalid = command("cmd_bad_version")
        invalid["schema_version"] = "2.0"
        assert_raises("SCHEMA_INVALID", lambda: svc.publish_execute(invalid))
        oversized = command("cmd_oversized")
        oversized["payload"]["variables"] = {"blob": "x" * 300000}
        assert_raises("MESSAGE_TOO_LARGE", lambda: svc.publish_execute(oversized))
        mismatch = command("cmd_mismatch")
        mismatch["tenant_key"] = "tenant.b"
        assert_raises("TENANT_BINDING_INVALID", lambda: svc.publish_execute(mismatch))
        return {
            "profile": "mock",
            "schema_negative_cases": 3,
            "http_execute_route": "absent",
        }


def probe_v02() -> dict[str, Any]:
    with tempfile.TemporaryDirectory() as directory:
        svc = service(directory)
        first = command("cmd_idempotent", "tenant-demo")
        svc.publish_execute(first)
        svc.publish_execute(first)
        svc.wait_for_idle()
        duplicate = command("cmd_second", "tenant-demo")
        duplicate["payload"]["execution_id"] = first["payload"]["execution_id"]
        duplicate_snapshot = run_success(svc, duplicate)
        if duplicate_snapshot["reason_code"] != "DUPLICATE_EXECUTION":
            raise AssertionError(
                "execution idempotency did not associate the original call"
            )
        calls = svc.store.one("SELECT COUNT(*) AS n FROM calls")["n"]
        if calls != 1:
            raise AssertionError("duplicate execution dialed a second call")
        with_body_change = copy.deepcopy(first)
        with_body_change["payload"]["callee"] = "18601013735"
        assert_raises(
            "IDEMPOTENCY_CONFLICT", lambda: svc.publish_execute(with_body_change)
        )
        return {
            "calls": calls,
            "same_command_replays": 1,
            "different_execution_command": "associated",
        }


def probe_v03() -> dict[str, Any]:
    with tempfile.TemporaryDirectory() as directory:
        broker = InMemoryBroker()
        first = service(directory, broker)
        second = AgentCallService(
            db_path=Path(directory) / "executor.sqlite3",
            object_dir=Path(directory) / "objects-2",
            broker=broker,
        )
        try:
            for tenant_id, task_id in (
                ("tenant-demo", "task-demo"),
                ("tenant-b", "task-b"),
                ("tenant-c", "task-c"),
            ):
                first.publish_execute(command(f"cmd_{tenant_id}", tenant_id, task_id))
            first.process_once()
            second.process_once()
            first.wait_for_idle()
            tenants = [
                row["tenant_id"]
                for row in first.store.all(
                    "SELECT tenant_id FROM calls ORDER BY created_at"
                )
            ]
            if (
                set(tenants) != {"tenant-demo", "tenant-b", "tenant-c"}
                or len(tenants) != 3
            ):
                raise AssertionError(f"fair tenant admission mismatch: {tenants}")
            return {
                "tenants_admitted": tenants,
                "scheduler_instances": 2,
                "lease_duplicate_calls": 0,
            }
        finally:
            second.stop()
            first.stop()


def probe_v04() -> dict[str, Any]:
    with tempfile.TemporaryDirectory() as directory:
        svc = service(directory)
        svc.profile["limits"]["global_concurrency"] = 0
        body = command("cmd_waiting")
        svc.publish_execute(body)
        svc.ingest_commands()
        svc.schedule_once()
        if svc.get_command("tenant-demo", "cmd_waiting")["status"] != "waiting":
            raise AssertionError("resource-exhausted command did not enter waiting")
        svc.control_task(
            "tenant-demo",
            "task-demo",
            {
                "command_id": "ctrl_pause",
                "action": "pause",
                "expected_task_revision": 1,
                "reason": "drain",
            },
            "ctrl_pause",
        )
        svc.apply_controls()
        state = svc.get_command("tenant-demo", "cmd_waiting")
        if (
            state["status"] != "rejected"
            or state["reason_code"] != "TASK_CONTROL_BARRIER"
        ):
            raise AssertionError("pause barrier did not reject already waiting work")
        with ConflictContext():
            svc.control_task(
                "tenant-demo",
                "task-demo",
                {
                    "command_id": "ctrl_stale",
                    "action": "resume",
                    "expected_task_revision": 1,
                    "reason": "stale",
                },
                "ctrl_stale",
            )
        return {
            "waiting_barrier": "applied",
            "affected_commands": 1,
            "stale_cas": "rejected",
        }


class ConflictContext:
    def __enter__(self) -> Self:
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> bool:
        if exc_type is not ConflictError:
            raise AssertionError("expected revision conflict")
        return True


def probe_v05() -> dict[str, Any]:
    with tempfile.TemporaryDirectory() as directory:
        broker = InMemoryBroker()
        svc = service(directory, broker)
        body = command("cmd_broker_retry")
        broker.outage = True
        assert_raises("BROKER_UNAVAILABLE", lambda: svc.publish_execute(body))
        broker.outage = False
        svc.publish_execute(body)
        snapshot = run_success(svc, body)
        if snapshot["status"] != "completed":
            raise AssertionError("retained command did not recover")
        return {"retained_id": body["command_id"], "recovered": True, "dead_letters": 0}


def probe_v06_v07() -> dict[str, Any]:
    with tempfile.TemporaryDirectory() as directory:
        svc = service(directory)
        scenarios = {
            "busy": "busy",
            "no_answer": "no_answer",
            "disconnect": "ari_disconnect",
        }
        outcomes: dict[str, str] = {}
        uncertain_before_reconcile: str | None = None
        for label, scenario in scenarios.items():
            body = command(f"cmd_{label}")
            svc.set_failure(body["command_id"], scenario)
            svc.publish_execute(body)
            if scenario == "ari_disconnect":
                call_id = None
                before = None
                for _ in range(500):
                    svc.process_once()
                    current = svc.get_command("tenant-demo", body["command_id"])
                    call_id = current.get("call_id")
                    if call_id:
                        before = svc.get_call("tenant-demo", call_id)
                        if before["call_state"] == "reconciling":
                            break
                    time.sleep(0.01)
                if not call_id or before is None:
                    raise AssertionError("disconnect probe did not create a call")
                uncertain_before_reconcile = before["call_state"]
                if uncertain_before_reconcile != "reconciling":
                    raise AssertionError(
                        "disconnect probe did not retain unknown state"
                    )
                svc.reconcile_call(
                    call_id, ended=True, outcome="failed", reason="ARI_RECONCILED"
                )
                snapshot = svc.get_command("tenant-demo", body["command_id"])
            else:
                svc.wait_for_idle()
                snapshot = svc.get_command("tenant-demo", body["command_id"])
            call = svc.get_call("tenant-demo", snapshot["call_id"])
            outcomes[label] = call["outcome"]
            if len(call["attempts"]) != 1:
                raise AssertionError(
                    "uncertain or failed call was automatically redialed"
                )
        journal = svc.journal.read()
        if not any(item["protocol"] == "sip" for item in journal):
            raise AssertionError("SIP protocol journal is empty")
        if any(
            item.get("target_uri") == "18601013734"
            for item in journal
            if item["protocol"] == "sip" and item["event"] == "invite"
        ):
            raise AssertionError("route prefix was lost")
        return {
            "mock_protocol": "pass",
            "asterisk_real": "not_run",
            "outcomes": outcomes,
            "fallback_redials": 0,
            "disconnect_state_before_reconcile": uncertain_before_reconcile,
        }


def probe_v08() -> dict[str, Any]:
    with tempfile.TemporaryDirectory() as directory:
        svc = service(directory)
        svc.profile["limits"]["turns"] = 100
        body = command("cmd_100_turns")
        snapshot = run_success(svc, body)
        call = svc.get_call("tenant-demo", snapshot["call_id"])
        segments = call["transcript"]["segments"]
        if len(segments) < 200:
            raise AssertionError(
                f"only {len(segments)} transcript segments were persisted"
            )
        if any(
            segment["playback_state"] != "playback_confirmed"
            for segment in segments
            if segment["speaker"] == "agent"
        ):
            raise AssertionError("normal agent turns lack playback evidence")
        return {"turns": 100, "segments": len(segments), "old_tts_reused": False}


def probe_v09() -> dict[str, Any]:
    with tempfile.TemporaryDirectory() as directory:
        svc = service(directory)
        missing = command("cmd_missing_object")
        svc.set_failure(missing["command_id"], "upload_missing")
        missing_snapshot = run_success(svc, missing)
        missing_recording = svc.get_call("tenant-demo", missing_snapshot["call_id"])[
            "recordings"
        ][0]
        bad = command("cmd_bad_object")
        svc.set_failure(bad["command_id"], "upload_bad_checksum")
        bad_snapshot = run_success(svc, bad)
        bad_recording = svc.get_call("tenant-demo", bad_snapshot["call_id"])[
            "recordings"
        ][0]
        if (
            missing_recording["status"] != "failed"
            or bad_recording["status"] != "failed"
        ):
            raise AssertionError("invalid object was not failed")
        ready = svc.store.one(
            "SELECT COUNT(*) AS n FROM events WHERE event_type='recording.ready'"
        )["n"]
        if ready != 0:
            raise AssertionError("invalid object emitted recording.ready")
        return {
            "missing_object": "failed",
            "bad_checksum": "failed",
            "invalid_ready": 0,
        }


def probe_v10() -> dict[str, Any]:
    with tempfile.TemporaryDirectory() as directory:
        broker = InMemoryBroker()
        svc = service(directory, broker)
        body = command("cmd_outbox_resume")
        svc.publish_execute(body)
        svc.ingest_commands()
        svc.schedule_once()
        for thread in list(svc._call_threads.values()):
            thread.join(timeout=2)
        broker.outage = True
        pending_before = svc.store.one(
            "SELECT COUNT(*) AS n FROM events WHERE published=0"
        )["n"]
        svc.dispatch_outbox()
        broker.outage = False
        with svc.store.tx() as con:
            con.execute("UPDATE events SET next_attempt_at=NULL")
        svc.dispatch_outbox()
        svc.consume_saas_events()
        pending_after = svc.store.one(
            "SELECT COUNT(*) AS n FROM events WHERE published=0"
        )["n"]
        if pending_before == 0 or pending_after != 0:
            raise AssertionError("outbox did not preserve and recover events")
        return {
            "pending_during_outage": pending_before,
            "pending_after_recovery": pending_after,
            "original_event_ids": True,
        }


def probe_v11_v12() -> dict[str, Any]:
    with tempfile.TemporaryDirectory() as directory:
        svc = service(directory)
        body = command("cmd_backup")
        run_success(svc, body)
        backup = Path(directory) / "backup.sqlite3"
        with redirect_stdout(io.StringIO()):
            copy_database(str(Path(directory) / "executor.sqlite3"), str(backup))
        if not backup.is_file():
            raise AssertionError("backup file was not created")
        restored = AgentCallService(
            db_path=backup,
            object_dir=Path(directory) / "restored-objects",
            broker=InMemoryBroker(),
        )
        try:
            restored_command = restored.get_command("tenant-demo", body["command_id"])
            restored_call = restored.get_call(
                "tenant-demo", restored_command["call_id"]
            )
            if (
                restored_command["status"] != "completed"
                or restored_call["recordings"][0]["status"] != "verified"
            ):
                raise AssertionError("restored database lost the completed call facts")
            restored_counts = {
                "commands": restored.store.one("SELECT COUNT(*) AS n FROM commands")[
                    "n"
                ],
                "events": restored.store.one("SELECT COUNT(*) AS n FROM events")["n"],
                "inbox": restored.store.one("SELECT COUNT(*) AS n FROM inbox")["n"],
                "recordings": restored.store.one(
                    "SELECT COUNT(*) AS n FROM recordings"
                )["n"],
            }
        finally:
            restored.stop()
        metrics = svc.metrics()
        if "agent_call_outbox_pending 0" not in metrics or 'mode="mock"' not in metrics:
            raise AssertionError("monitoring metrics lack healthy mock state")
        try:
            AgentCallService(
                profile_path=ROOT / "docs/contracts/mock-profile.json", mode="real"
            )
        except ServiceError as exc:
            if exc.code != "CONFIGURATION_INVALID":
                raise AssertionError("real mode rejected for the wrong reason") from exc
        else:
            raise AssertionError("real mode silently accepted mock providers")
        return {
            "backup": "verified",
            "metrics": "healthy",
            "rollback_preserves_db": "not_run",
            "restored_db_facts": True,
            "restored_counts": restored_counts,
            "real_mode_mock_rejection": True,
        }


def probe_v13() -> dict[str, Any]:
    candidates = sorted(
        path
        for path in (ROOT / "docs/evidence").glob("scale-mock-*.json")
        if "preflight" not in path.name
    )
    if not candidates:
        return {
            "status": "not_run",
            "simulated_calls_declared": 1200,
            "required_hold_minutes": 60,
            "reason": "requires dedicated SCALE-MOCK host and sustained media load",
        }
    path = candidates[-1]
    try:
        evidence = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise AssertionError(f"invalid SCALE-MOCK evidence: {path.name}") from exc
    load = evidence.get("load", {})
    if (
        evidence.get("status") != "observed"
        or evidence.get("production_claim", True)
        or not evidence.get("continuous_media_evidence", False)
        or load.get("minimum_answered_calls", 0) < 1000
        or load.get("observed_hold_seconds", 0) < 60 * 60
    ):
        return {
            "status": "not_run",
            "evidence_file": str(path.relative_to(ROOT)),
            "reason": "historical state-only evidence cannot satisfy continuous media V13",
        }
    return {
        "status": "observed",
        "evidence_file": str(path.relative_to(ROOT)),
        "tenants": load.get("tenants"),
        "submitted_calls": load.get("submitted_calls"),
        "minimum_answered_calls": load.get("minimum_answered_calls"),
        "observed_hold_seconds": load.get("observed_hold_seconds"),
        "production_claim": False,
    }


def run() -> dict[str, Any]:
    probes: list[tuple[str, Callable[[], dict[str, Any]]]] = [
        ("V01", probe_v01),
        ("V02", probe_v02),
        ("V03", probe_v03),
        ("V04", probe_v04),
        ("V05", probe_v05),
        ("V06/V07", probe_v06_v07),
        ("V08", probe_v08),
        ("V09", probe_v09),
        ("V10", probe_v10),
        ("V11/V12", probe_v11_v12),
        ("V13", probe_v13),
    ]
    results: dict[str, Any] = {
        "started_at": iso(),
        "profile": "mock",
        "probes": {},
        "production": {"V14": "blocked_external_evidence"},
    }
    for name, probe in probes:
        started = time.monotonic()
        try:
            evidence = probe()
            status = (
                "PASS" if evidence.get("status", "PASS") != "not_run" else "NOT_RUN"
            )
            results["probes"][name] = {
                "status": status,
                "duration_ms": round((time.monotonic() - started) * 1000, 3),
                "evidence": evidence,
            }
        except (
            AssertionError,
            KeyError,
            OSError,
            RuntimeError,
            ServiceError,
            ValueError,
        ) as exc:
            results["probes"][name] = {
                "status": "FAIL",
                "duration_ms": round((time.monotonic() - started) * 1000, 3),
                "error": f"{type(exc).__name__}: {exc}",
            }
    results["finished_at"] = iso()
    return results


def main() -> int:
    result = run()
    incomplete = [
        name for name, value in result["probes"].items() if value["status"] != "PASS"
    ]
    if result["production"].get("V14") != "PASS":
        incomplete.append("V14")
    result["gate_status"] = "PASS" if not incomplete else "INCOMPLETE"
    print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
    return 1 if incomplete else 0


if __name__ == "__main__":
    raise SystemExit(main())