from __future__ import annotations

import hashlib
import io
import json
import os
import subprocess
import sys
from importlib import util
from pathlib import Path
from types import SimpleNamespace

import pytest

ROOT = Path(__file__).resolve().parents[1]
MODEL_ROUTER_DIR = ROOT / "mcp" / "model-router"
SERVER = MODEL_ROUTER_DIR / "server_v2.py"

sys.path.insert(0, str(MODEL_ROUTER_DIR))

spec = util.spec_from_file_location("model_router_server_v2", SERVER)
assert spec is not None
server_v2 = util.module_from_spec(spec)
sys.modules["model_router_server_v2"] = server_v2
assert spec.loader is not None
spec.loader.exec_module(server_v2)

import tier_trust  # noqa: E402,I001


CHANGE_ID = "change-t3-server"
_TIER_SECRET = "test-tier-secret-server"


@pytest.fixture(autouse=True)
def _reset_runtime(monkeypatch, tmp_path: Path) -> None:
    monkeypatch.setenv("CLAUDE_PROJECT_DIR", str(tmp_path))
    monkeypatch.setenv("MODEL_ROUTER_TIER_SECRET", _TIER_SECRET)
    server_v2._RUNTIME = None


def _allowlist_path(tmp_path: Path) -> Path:
    path = tmp_path / "openspec" / "changes" / CHANGE_ID / "MODEL_ALLOWLIST.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(
            {
                "schema_version": "T3.allowlist.v1",
                "policy_version": "policy-t3.6",
                "models": [
                    {
                        "id": "qwen-max",
                        "vendor_family": "alibaba",
                        "deployment": "private://qwen-max",
                        "allowed_agent_roles": ["coder", "compliance", "docs", "maintainer"],
                        "allowed_data_levels": ["L1", "L2", "L3"],
                        "rate_limit_qps": 2,
                    },
                    {
                        "id": "claude-sonnet-4.6",
                        "vendor_family": "anthropic",
                        "deployment": "private://claude-sonnet-4.6",
                        "allowed_agent_roles": ["coder", "compliance", "reviewer", "docs"],
                        "allowed_data_levels": ["L1", "L2"],
                        "rate_limit_qps": 10,
                    },
                ],
            },
            ensure_ascii=False,
            indent=2,
        ),
        encoding="utf-8",
    )
    return path


def _route_payload(**overrides: object) -> dict[str, object]:
    payload: dict[str, object] = {
        "model_id": "qwen-max",
        "agent_role": "coder",
        "data_level": "L2",
        "change_id": CHANGE_ID,
        "caller_vendor_family": "openai",
        "desensitized": True,
    }
    payload.update(overrides)
    payload["tier_sig"] = tier_trust.sign_tier(payload, _TIER_SECRET.encode("utf-8"))
    return payload


def _audit_lines(tmp_path: Path) -> list[dict[str, object]]:
    audit_path = tmp_path / ".audit" / "routing_log.jsonl"
    assert audit_path.exists()
    return [
        json.loads(line) for line in audit_path.read_text(encoding="utf-8").splitlines() if line
    ]


class _FakeSocket:
    def __init__(self, request_bytes: bytes) -> None:
        self._rfile = io.BytesIO(request_bytes)
        self._wfile = io.BytesIO()

    def makefile(self, mode: str, buffering: int | None = None) -> io.BytesIO:
        del buffering
        if "r" in mode:
            return self._rfile
        return self._wfile

    def sendall(self, data: bytes) -> None:
        self._wfile.write(data)

    def close(self) -> None:
        return None


def _invoke_http_request(
    method: str,
    path: str,
    payload: dict[str, object] | None = None,
    raw_body: bytes | None = None,
) -> tuple[int, dict[str, object]]:
    server_v2._RUNTIME = None
    body = raw_body or b""
    headers = ["Host: 127.0.0.1", "Connection: close"]
    if payload is not None:
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        headers.append("Content-Type: application/json")
        headers.append(f"Content-Length: {len(body)}")
    elif raw_body is not None:
        headers.append("Content-Type: application/json")
        headers.append(f"Content-Length: {len(body)}")
    request_bytes = (f"{method} {path} HTTP/1.1\r\n" + "\r\n".join(headers) + "\r\n\r\n").encode(
        "utf-8"
    ) + body
    fake_socket = _FakeSocket(request_bytes)
    server = SimpleNamespace(server_name="127.0.0.1", server_port=0)
    server_v2._ModelRouterHTTPHandler(fake_socket, ("127.0.0.1", 12345), server)
    response = fake_socket._wfile.getvalue()
    status_line, _, remainder = response.partition(b"\r\n")
    headers_blob, _, body_bytes = remainder.partition(b"\r\n\r\n")
    del headers_blob
    status_code = int(status_line.split()[1])
    return status_code, json.loads(body_bytes.decode("utf-8"))


def test_route_health_and_stdio_smoke(tmp_path: Path) -> None:
    _allowlist_path(tmp_path)

    allow = server_v2.route_v2(_route_payload())
    health = server_v2.health_v2()

    env = os.environ.copy()
    env["CLAUDE_PROJECT_DIR"] = str(tmp_path)
    route_cli = subprocess.run(
        [sys.executable, str(SERVER), "route"],
        input=json.dumps(_route_payload(prompt="RAW-PHI-CLAIM"), ensure_ascii=False),
        capture_output=True,
        text=True,
        check=True,
        env=env,
        timeout=10,
    )
    route_stdio = subprocess.run(
        [sys.executable, str(SERVER), "serve", "--stdio"],
        input=json.dumps(
            {"id": "route-1", "method": "route", "params": _route_payload(prompt="RAW-PHI-CLAIM")},
            ensure_ascii=False,
        )
        + "\n"
        + json.dumps({"id": "health-1", "method": "health", "params": {}}, ensure_ascii=False)
        + "\n",
        capture_output=True,
        text=True,
        check=True,
        env=env,
        timeout=10,
    )

    cli_payload = json.loads(route_cli.stdout)
    stdio_payloads = [json.loads(line) for line in route_stdio.stdout.splitlines() if line]

    assert allow["decision"] == "allow"
    assert allow["model_id"] == "qwen-max"
    assert allow["vendor_family"] == "alibaba"
    assert allow["deployment"] == "private://qwen-max"
    assert allow["routing_log_id"]
    assert allow["policy_version"] == "policy-t3.6"
    assert allow["_meta"]["duration_ms"] >= 0

    assert health["status"] == "ok-v2"
    assert health["policy_version"] == "policy-t3.6"
    assert health["circuit_open_count"] == 0

    assert cli_payload["decision"] == "allow"
    assert cli_payload["routing_log_id"]
    assert route_stdio.returncode == 0
    assert stdio_payloads[0]["result"]["decision"] == "allow"
    assert stdio_payloads[1]["result"]["status"] == "ok-v2"

    audit_text = (tmp_path / ".audit" / "routing_log.jsonl").read_text(encoding="utf-8")
    assert "RAW-PHI-CLAIM" not in audit_text
    assert "prompt" not in audit_text


def test_http_health_and_signed_route_response(tmp_path: Path) -> None:
    _allowlist_path(tmp_path)

    health_code, health = _invoke_http_request("GET", "/health")
    route_code, result = _invoke_http_request("POST", "/route", _route_payload())

    signed_fields = {
        key: result[key]
        for key in (
            "decision",
            "reason",
            "layer_failed",
            "policy_version",
            "duration_us",
            "allowed_model_set",
            "lane",
            "max_data_level",
            "map_id",
        )
    }

    assert health_code == 200
    assert route_code == 200
    assert health["status"] == "ok-v2"
    assert result["decision"] == "allow"
    assert result["routing_log_id"]
    assert result["sig"]
    assert tier_trust.verify_decision(signed_fields, result["sig"], _TIER_SECRET.encode("utf-8"))
    assert result["vendor_family"] == "alibaba"
    assert result["deployment"] == "private://qwen-max"


def test_http_deny_is_generic_and_reason_free(tmp_path: Path) -> None:
    _allowlist_path(tmp_path)

    payload = {
        "model_id": "qwen-max",
        "agent_role": "coder",
        "data_level": "L2",
        "change_id": CHANGE_ID,
        "desensitized": True,
    }

    status, result = _invoke_http_request("POST", "/route", payload)

    assert status == 200
    assert result["decision"] == "deny"
    assert result["error"]["code"] == "route_denied"
    assert result["error"]["msg"] == "request denied by routing policy"
    assert "reason" not in result["error"]
    assert "layer_failed" not in result["error"]
    assert "reason" not in json.dumps(result, ensure_ascii=False)
    assert "layer_failed" not in json.dumps(result, ensure_ascii=False)


def test_http_forged_tier_denies_generic(tmp_path: Path) -> None:
    _allowlist_path(tmp_path)
    payload = _route_payload(data_level="L4")
    payload["data_level"] = "L1"

    status, result = _invoke_http_request("POST", "/route", payload)

    assert status == 200
    assert result["decision"] == "deny"
    assert result["error"]["code"] == "route_denied"
    assert "reason" not in result["error"]


def test_http_bad_json_rejects_without_echo(tmp_path: Path) -> None:
    _allowlist_path(tmp_path)

    status, body = _invoke_http_request("POST", "/route", raw_body=b'{"model_id": "broken"')

    assert status == 400
    assert body["error"]["code"] == "bad_request"
    assert "broken" not in json.dumps(body, ensure_ascii=False)


def test_missing_desensitized_marker_denies(tmp_path: Path) -> None:
    _allowlist_path(tmp_path)
    denied = server_v2.route_v2(_route_payload(desensitized=False))

    assert denied["decision"] == "deny"
    assert denied["error"]["layer_failed"] == "marker"
    assert denied["error"]["type"] == "PolicyDenyError"
    # H2: detail is in the audit record, NOT the external error message
    assert any(
        "must route through mcp-desensitize first" in str(r.get("reason", ""))
        for r in _audit_lines(tmp_path)
    )


def test_unsigned_tier_denied(tmp_path: Path) -> None:
    """B1: a caller cannot self-assert tier — unsigned tier is rejected (fail-closed)."""
    _allowlist_path(tmp_path)
    payload = {
        "model_id": "qwen-max",
        "agent_role": "coder",
        "data_level": "L2",
        "change_id": CHANGE_ID,
        "caller_vendor_family": "openai",
        "desensitized": True,
        # no tier_sig — caller is not the trusted gate middleware
    }
    denied = server_v2.route_v2(payload)

    assert denied["decision"] == "deny"
    assert denied["error"]["layer_failed"] == "tier"


def test_forged_tier_denied(tmp_path: Path) -> None:
    """B1: tampering a signed tier field (downgrade L4->L1) breaks the signature -> deny."""
    _allowlist_path(tmp_path)
    payload = _route_payload(data_level="L4")  # validly signed for L4
    payload["data_level"] = "L1"  # caller forges a downgrade after signing

    denied = server_v2.route_v2(payload)

    assert denied["decision"] == "deny"
    assert denied["error"]["layer_failed"] == "tier"


def test_allowlist_miss_denies(tmp_path: Path) -> None:
    _allowlist_path(tmp_path)
    denied = server_v2.route_v2(_route_payload(model_id="missing-model"))

    assert denied["decision"] == "deny"
    assert denied["error"]["layer_failed"] == "allowlist"
    assert any("missing-model" in str(r.get("reason", "")) for r in _audit_lines(tmp_path))


def test_data_level_over_policy_denies(tmp_path: Path) -> None:
    _allowlist_path(tmp_path)
    denied = server_v2.route_v2(_route_payload(data_level="L4"))

    assert denied["decision"] == "deny"
    assert denied["error"]["layer_failed"] == "data_level"
    assert denied["error"]["type"] == "PolicyDenyError"
    assert any("data_level='L4'" in str(r.get("reason", "")) for r in _audit_lines(tmp_path))
    # H2: external message is generic — must NOT leak the tier value or policy trace
    assert "L4" not in denied["error"]["message"]
    assert denied["error"]["message"] == "request denied by routing policy"


def test_same_family_compliance_call_denies(tmp_path: Path) -> None:
    _allowlist_path(tmp_path)
    denied = server_v2.route_v2(
        _route_payload(
            agent_role="compliance",
            caller_vendor_family="alibaba",
        )
    )

    assert denied["decision"] == "deny"
    assert denied["error"]["layer_failed"] == "heterogeneity"
    assert denied["error"]["severity"] == "WARN"
    assert any(
        "heterogeneity policy denied" in str(r.get("reason", "")) for r in _audit_lines(tmp_path)
    )


def test_five_denies_open_circuit_and_sixth_is_sev2(tmp_path: Path) -> None:
    _allowlist_path(tmp_path)
    for _ in range(5):
        denied = server_v2.route_v2(
            _route_payload(
                model_id="claude-sonnet-4.6",
                data_level="L2",
                agent_role="compliance",
                caller_vendor_family="anthropic",
            )
        )
        assert denied["decision"] == "deny"
        assert denied["error"]["layer_failed"] == "heterogeneity"

    opened = server_v2.route_v2(
        _route_payload(
            model_id="claude-sonnet-4.6",
            data_level="L2",
            agent_role="compliance",
            caller_vendor_family="anthropic",
        )
    )

    assert opened["decision"] == "deny"
    assert opened["error"]["layer_failed"] == "circuit"
    assert opened["error"]["type"] == "CircuitOpenError"
    assert opened["error"]["severity"] == "SEV-2"

    health = server_v2.health_v2()
    assert health["circuit_open_count"] >= 1


def test_rate_limit_denies_when_burst_exceeds_qps(tmp_path: Path) -> None:
    _allowlist_path(tmp_path)
    first = server_v2.route_v2(_route_payload())
    second = server_v2.route_v2(_route_payload())
    third = server_v2.route_v2(_route_payload())

    assert first["decision"] == "allow"
    assert second["decision"] == "allow"
    assert third["decision"] == "deny"
    assert third["error"]["layer_failed"] == "rate"
    assert third["error"]["type"] == "RateLimitError"


def test_routing_log_ids_are_unique_and_audit_records_are_structured(tmp_path: Path) -> None:
    _allowlist_path(tmp_path)
    first = server_v2.route_v2(_route_payload())
    second = server_v2.route_v2(
        _route_payload(model_id="claude-sonnet-4.6", caller_vendor_family="openai")
    )

    assert first["routing_log_id"] != second["routing_log_id"]

    records = _audit_lines(tmp_path)
    assert len(records) >= 2
    for record in records:
        assert "decision" in record
        assert "reason" in record
        assert "duration_ms" in record
        assert "prompt" not in record
        assert "RAW-PHI-CLAIM" not in json.dumps(record, ensure_ascii=False)


class FakeAuditServer:
    def __init__(self) -> None:
        self.appended_events: list[dict[str, object]] = []

    def append(self, event: dict[str, object]) -> dict[str, object]:
        self.appended_events.append(event)
        return {
            "event_id": event["event_id"],
            "row_id": len(self.appended_events) - 1,
            "prev_hash": "GENESIS" if len(self.appended_events) == 1 else "prev-hash",
            "current_hash": f"current-{len(self.appended_events)}",
            "state": "normal",
        }


def _routing_record(**overrides: object) -> dict[str, object]:
    record: dict[str, object] = {
        "routing_log_id": "route-0001",
        "decision": "allow",
        "reason": "allowed by policy",
        "policy_version": "policy-t4.8",
        "duration_ms": 12.345,
        "layer_failed": None,
        "severity": None,
        "error_type": None,
        "model_id": "qwen-max",
        "agent_role": "coder",
        "data_level": "L2",
        "change_id": "feat/T4.8-router-audit-adapter",
        "target_model_id": "qwen-max",
        "vendor_family": "alibaba",
        "deployment": "private://qwen-max",
        "caller_vendor_family": "openai",
        "desensitized": True,
    }
    record.update(overrides)
    return record


def test_clickhouse_audit_adapter_rejects_default_init() -> None:
    with pytest.raises(NotImplementedError, match="requires audit_server"):
        server_v2.ClickHouseAuditAdapter()


def test_clickhouse_audit_adapter_accepts_injected_server() -> None:
    audit_server = FakeAuditServer()
    adapter = server_v2.ClickHouseAuditAdapter(audit_server=audit_server)

    event_id = adapter.write_routing_decision(_routing_record())

    assert event_id == "route-0001"
    assert len(audit_server.appended_events) == 1


def test_clickhouse_audit_adapter_accepts_factory() -> None:
    audit_server = FakeAuditServer()
    factory_calls = 0

    def factory() -> FakeAuditServer:
        nonlocal factory_calls
        factory_calls += 1
        return audit_server

    adapter = server_v2.ClickHouseAuditAdapter(audit_server_factory=factory)

    assert factory_calls == 0
    assert adapter.write_routing_decision(_routing_record()) == "route-0001"
    assert factory_calls == 1
    assert (
        adapter.write_routing_decision(_routing_record(routing_log_id="route-0002")) == "route-0002"
    )
    assert factory_calls == 1


def test_write_routing_decision_delegates_to_audit_server_append() -> None:
    audit_server = FakeAuditServer()
    adapter = server_v2.ClickHouseAuditAdapter(audit_server=audit_server)
    record = _routing_record(decision="deny", reason="rate limit exceeded", duration_ms=3.21)

    event_id = adapter.write_routing_decision(record)

    assert event_id == "route-0001"
    assert len(audit_server.appended_events) == 1
    event = audit_server.appended_events[0]
    assert event["event_id"] == "route-0001"
    assert event["result"] == {
        "status": "deny",
        "reason": "rate limit exceeded",
        "duration_ms": 3.21,
    }


def test_record_to_event_maps_all_fields_correctly() -> None:
    record = _routing_record()

    event = server_v2.ClickHouseAuditAdapter._record_to_event(record)

    assert event["event_id"] == "route-0001"
    assert isinstance(event["timestamp"], str)
    assert event["actor"] == {
        "agent_role": "coder",
        "model_id": "qwen-max",
        "vendor_family": "alibaba",
        "session_id": "",
    }
    assert event["action"] == {
        "tool": "model-router",
        "skill": None,
        "operation": "route",
    }
    assert event["context"] == {
        "change_id": "feat/T4.8-router-audit-adapter",
        "step": None,
        "data_levels": ["L2"],
    }
    assert event["result"] == {
        "status": "allow",
        "reason": "allowed by policy",
        "duration_ms": 12.345,
    }
    assert (
        event["input_hash"]
        == hashlib.sha256(b"qwen-max|feat/T4.8-router-audit-adapter|L2").hexdigest()
    )
    assert event["output_hash"] == hashlib.sha256(b"route-0001|allow|allowed by policy").hexdigest()


def test_record_to_event_does_not_leak_raw_payload() -> None:
    record = _routing_record(
        prompt="RAW-PHI-CLAIM",
        raw_text="RAW-PHI-CLAIM",
        reason="deny synthetic routing request",
    )

    event = server_v2.ClickHouseAuditAdapter._record_to_event(record)
    event_text = json.dumps(event, ensure_ascii=False)

    assert "RAW-PHI-CLAIM" not in event_text
    assert "prompt" not in event
    assert "raw_text" not in event
    assert (
        event["input_hash"]
        == hashlib.sha256(b"qwen-max|feat/T4.8-router-audit-adapter|L2").hexdigest()
    )
