from __future__ import annotations

import asyncio
import hashlib
from contextlib import contextmanager

import pytest

from backplane_runtime.backends.base import BackendInput
from backplane_runtime.backends.transformers import (
    TransformersBackend,
    _peft_adapter_name,
    _receipt_model_label,
    adapter_directory_sha256,
)
from backplane_runtime.backends.vllm import VLLMBackend
from backplane_runtime.errors import UnsupportedCapability


def test_vllm_capability_boundary_is_explicit():
    backend = VLLMBackend("http://127.0.0.1:9", "local-model")
    assert backend.capabilities.kv_cache is True
    assert backend.capabilities.per_request_adapters is True
    assert backend.capabilities.activation_cards is False


def test_vllm_rejects_activation_cards_before_network():
    backend = VLLMBackend("http://127.0.0.1:9", "local-model")
    request = BackendInput(
        prompt="hello",
        max_new_tokens=8,
        tenant_id="tenant",
        request_id="request",
        stack_key="stack",
        activation_specs=({"kind": "steering", "token": "safe"},),
    )
    async def scenario():
        with pytest.raises(UnsupportedCapability, match="residual-stream"):
            await backend.generate_batch([request])

    asyncio.run(scenario())


def test_adapter_directory_hash_commits_config_and_weights(tmp_path):
    (tmp_path / "adapter_config.json").write_bytes(b'{"r":4}')
    (tmp_path / "adapter_model.safetensors").write_bytes(b"weights")
    expected = hashlib.sha256(b'{"r":4}' + b"weights").hexdigest()
    assert adapter_directory_sha256(tmp_path) == expected


def test_peft_adapter_name_is_deterministic_and_module_safe():
    first = _peft_adapter_name("semantic.global-anchor-v1")
    assert first == _peft_adapter_name("semantic.global-anchor-v1")
    assert first != _peft_adapter_name("semantic.global-anchor-v2")
    assert first.startswith("bp_")
    assert "." not in first


def test_local_model_source_requires_path_free_receipt_label(tmp_path):
    assert _receipt_model_label("publisher/model", None) == "publisher/model"
    assert _receipt_model_label(str(tmp_path), "private-model@revision") == (
        "private-model@revision"
    )
    with pytest.raises(ValueError, match="path-free model_label"):
        _receipt_model_label(str(tmp_path), None)
    with pytest.raises(ValueError, match="absolute filesystem path"):
        _receipt_model_label("publisher/model", "C:/private/model")


def test_transformers_adapter_selection_is_registry_and_hash_bound(tmp_path):
    class FakeModel:
        def __init__(self):
            self.selected = []
            self.disabled = 0

        def set_adapter(self, name):
            self.selected.append(name)

        @contextmanager
        def disable_adapter(self):
            self.disabled += 1
            yield

    backend = TransformersBackend.__new__(TransformersBackend)
    backend.model = FakeModel()
    backend._adapter_paths = {"capability.route": tmp_path}
    backend._adapter_digests = {"capability.route": "a" * 64}
    backend._peft_adapter_names = {
        "capability.route": _peft_adapter_name("capability.route")
    }
    backend.system_prompt_sha256 = "c" * 64
    spec = (
        {
            "kind": "adapter",
            "card_id": "capability.route",
            "artifact_sha256": "a" * 64,
            "system_prompt_sha256": "c" * 64,
        },
    )
    with backend._adapter_context(spec) as selected:
        assert selected == "capability.route"
    assert backend.model.selected == [_peft_adapter_name("capability.route")]
    with backend._adapter_context(()) as selected:
        assert selected is None
    assert backend.model.disabled == 1
    with pytest.raises(UnsupportedCapability, match="artifact hash mismatch"):
        with backend._adapter_context(
            ({**spec[0], "artifact_sha256": "b" * 64},)
        ):
            pass
    with pytest.raises(UnsupportedCapability, match="system prompt hash mismatch"):
        with backend._adapter_context(
            ({**spec[0], "system_prompt_sha256": "d" * 64},)
        ):
            pass


def test_transformers_system_prompt_uses_chat_template():
    class Tokenizer:
        chat_template = "available"

        def apply_chat_template(self, messages, **kwargs):
            assert kwargs["tokenize"] is False
            assert kwargs["add_generation_prompt"] is True
            return f"{messages[0]['content']}::{messages[1]['content']}"

    backend = TransformersBackend.__new__(TransformersBackend)
    backend.system_prompt = "system"
    backend.tokenizer = Tokenizer()
    assert backend._format_prompt("user") == "system::user"


def test_vllm_adapter_card_selects_signed_served_model():
    backend = VLLMBackend("http://127.0.0.1:9", "base-model")

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {
                "choices": [{"message": {"content": "ok"}}],
                "usage": {"prompt_tokens": 1, "completion_tokens": 1},
            }

    class Client:
        def __init__(self):
            self.body = None

        async def post(self, path, json):
            self.body = json
            return Response()

    client = Client()
    request = BackendInput(
        prompt="hello",
        max_new_tokens=8,
        tenant_id="tenant",
        request_id="request",
        stack_key="stack",
        activation_specs=(
            {
                "kind": "adapter",
                "card_id": "capability.route",
                "served_model": "tenant-lora",
            },
        ),
    )
    result = asyncio.run(backend._one(client, request))
    assert client.body["model"] == "tenant-lora"
    assert result.metadata["served_model"] == "tenant-lora"


def test_vllm_ignores_unsigned_model_metadata_and_requires_signed_adapter_target():
    backend = VLLMBackend("http://127.0.0.1:9", "base-model")

    class Response:
        def raise_for_status(self):
            return None

        def json(self):
            return {
                "choices": [{"message": {"content": "ok"}}],
                "usage": {"prompt_tokens": 1, "completion_tokens": 1},
            }

    class Client:
        def __init__(self):
            self.body = None

        async def post(self, path, json):
            self.body = json
            return Response()

    client = Client()
    unsigned = BackendInput(
        prompt="hello",
        max_new_tokens=8,
        tenant_id="tenant",
        request_id="request",
        stack_key="stack",
        metadata={"vllm_model": "caller-controlled"},
    )
    asyncio.run(backend._one(client, unsigned))
    assert client.body["model"] == "base-model"
    missing_target = BackendInput(
        prompt="hello",
        max_new_tokens=8,
        tenant_id="tenant",
        request_id="request",
        stack_key="stack",
        activation_specs=(
            {"kind": "adapter", "card_id": "capability.route"},
        ),
    )
    with pytest.raises(UnsupportedCapability, match="signed served_model"):
        asyncio.run(backend._one(client, missing_target))
