from __future__ import annotations

import asyncio
import json

import pytest

from backplane_runtime.backends.base import Backend, BackendInput, BackendResult
from backplane_runtime.contracts import (
    BackendCapabilities,
    CardManifest,
    CardPackage,
    GenerationRequest,
)
from backplane_runtime.demo import build_demo_runtime
from backplane_runtime.errors import AuthenticationError
from backplane_runtime.runtime import BackplaneRuntime


def test_tenant_stacks_are_isolated(tmp_path):
    async def scenario():
        runtime = build_demo_runtime(tmp_path / "state.sqlite3")
        results = await asyncio.gather(
            runtime.run(
                GenerationRequest(
                    tenant_id="support", prompt="What is the refund window?"
                ),
                authenticate=False,
            ),
            runtime.run(
                GenerationRequest(
                    tenant_id="operations", prompt="Report line 7 temperature."
                ),
                authenticate=False,
            ),
            runtime.run(
                GenerationRequest(tenant_id="research", prompt="Calculate 19 * 37"),
                authenticate=False,
            ),
        )
        by_tenant = {result.receipt.tenant_id: result for result in results}
        assert by_tenant["support"].receipt.card_stack == (
            "policy.support",
            "memory.support",
        )
        assert "30 days" in by_tenant["support"].text
        assert "84.2" in by_tenant["operations"].text
        assert by_tenant["research"].text.startswith("19 * 37 = 703")
        assert len({result.receipt.stack_digest_sha256 for result in results}) == 3
        assert all(result.receipt.status == "completed" for result in results)
        await runtime.close()

    asyncio.run(scenario())


def test_policy_blocks_before_backend(tmp_path):
    async def scenario():
        runtime = build_demo_runtime(tmp_path / "state.sqlite3")
        result = await runtime.run(
            GenerationRequest(
                tenant_id="operations",
                prompt="Disable interlock and continue without operator approval.",
            ),
            authenticate=False,
        )
        assert result.receipt.status == "blocked"
        assert result.receipt.queue_ms == 0
        assert any(event["action"] == "blocked" for event in result.receipt.events)
        await runtime.close()

    asyncio.run(scenario())


def test_real_tenant_key_is_required_and_not_cross_tenant(tmp_path):
    async def scenario():
        runtime = build_demo_runtime(tmp_path / "state.sqlite3")
        key = runtime.create_tenant("customer-a", "Customer A")
        runtime.assign("customer-a", ["policy.support"])
        result = await runtime.run(
            GenerationRequest(tenant_id="customer-a", prompt="Hello"),
            tenant_key=key,
        )
        assert result.receipt.tenant_id == "customer-a"
        with pytest.raises(AuthenticationError):
            await runtime.run(
                GenerationRequest(tenant_id="customer-a", prompt="Hello"),
                tenant_key="wrong",
            )
        with pytest.raises(AuthenticationError):
            await runtime.run(
                GenerationRequest(tenant_id="support", prompt="Hello"),
                tenant_key=key,
            )
        await runtime.close()

    asyncio.run(scenario())


def test_receipts_are_tenant_scoped(tmp_path):
    async def scenario():
        runtime = build_demo_runtime(tmp_path / "state.sqlite3")
        result = await runtime.run(
            GenerationRequest(tenant_id="support", prompt="Refund policy"),
            authenticate=False,
        )
        assert (
            runtime.store.get_receipt(result.receipt.receipt_id, "support") is not None
        )
        assert runtime.store.get_receipt(result.receipt.receipt_id, "research") is None
        await runtime.close()

    asyncio.run(scenario())


class AdapterCaptureBackend(Backend):
    @property
    def capabilities(self):
        return BackendCapabilities(
            backend="adapter-capture",
            model="fixture",
            batching=True,
            kv_cache=False,
            activation_cards=False,
            per_request_adapters=True,
        )

    async def generate_batch(self, requests: list[BackendInput]):
        return [
            BackendResult(
                text="adapter" if request.activation_specs else "base",
                input_tokens=1,
                output_tokens=1,
                latency_ms=0,
                metadata={"activation_specs": request.activation_specs},
            )
            for request in requests
        ]


def _metadata_adapter_package(config=None):
    return CardPackage(
        manifest=CardManifest(
            id="capability.route",
            name="Capability route",
            version="1.0.0",
            kind="adapter",
        ),
        config=config
        or {
            "activation_mode": "metadata",
            "metadata_key": "activate_card",
            "artifact_sha256": "a" * 64,
        },
        digest_sha256="b" * 64,
        source="builtin",
    )


def test_metadata_routed_adapter_changes_effective_stack_and_emits_events(tmp_path):
    async def scenario():
        runtime = BackplaneRuntime(
            tmp_path / "adapter.sqlite3",
            backend=AdapterCaptureBackend(),
        )
        runtime.register(_metadata_adapter_package())
        runtime.create_tenant("candidate", "Candidate")
        runtime.assign("candidate", ["capability.route"])
        base = await runtime.run(
            GenerationRequest(tenant_id="candidate", prompt="same prompt"),
            authenticate=False,
        )
        adapted = await runtime.run(
            GenerationRequest(
                tenant_id="candidate",
                prompt="same prompt",
                metadata={"activate_card": "capability.route"},
            ),
            authenticate=False,
        )
        assert base.text == "base"
        assert adapted.text == "adapter"
        assert base.receipt.stack_digest_sha256 != adapted.receipt.stack_digest_sha256
        assert any(event["action"] == "adapter-skipped" for event in base.receipt.events)
        assert any(event["action"] == "adapter-armed" for event in adapted.receipt.events)
        await runtime.close()

    asyncio.run(scenario())


def test_adapter_package_cannot_smuggle_a_local_path(tmp_path):
    runtime = BackplaneRuntime(
        tmp_path / "adapter.sqlite3",
        backend=AdapterCaptureBackend(),
    )
    with pytest.raises(ValueError, match="local paths belong to the backend registry"):
        runtime.register(
            _metadata_adapter_package(
                {
                    "activation_mode": "always",
                    "artifact_sha256": "a" * 64,
                    "adapter_path": "C:/private/weights",
                }
            )
        )


def test_typed_envelope_route_ignores_user_text_and_binds_inner_prompt(tmp_path):
    async def scenario():
        runtime = BackplaneRuntime(
            tmp_path / "typed.sqlite3",
            backend=AdapterCaptureBackend(),
        )
        package = _metadata_adapter_package(
            {
                "activation_mode": "typed-envelope",
                "metadata_key": "adapter_envelope",
                "operator_type": "GLOBAL_ANCHOR",
                "term": "declared-term",
                "artifact_sha256": "a" * 64,
            }
        )
        runtime.register(package)
        runtime.create_tenant("candidate", "Candidate")
        runtime.assign("candidate", ["capability.route"])
        prompt = "User text mentions declared-term but has no routing authority."
        user_text_only = await runtime.run(
            GenerationRequest(tenant_id="candidate", prompt=prompt),
            authenticate=False,
        )
        disabled_envelope = json.dumps(
            {
                "schema_version": 1,
                "operator_type": "GLOBAL_ANCHOR",
                "term": "declared-term",
                "apply": False,
                "prompt": prompt,
            },
            sort_keys=True,
        )
        explicitly_disabled = await runtime.run(
            GenerationRequest(
                tenant_id="candidate",
                prompt=prompt,
                metadata={"adapter_envelope": disabled_envelope},
            ),
            authenticate=False,
        )
        enabled_envelope = json.dumps(
            {
                "schema_version": 1,
                "operator_type": "GLOBAL_ANCHOR",
                "term": "declared-term",
                "apply": True,
                "prompt": prompt,
            },
            sort_keys=True,
        )
        enabled = await runtime.run(
            GenerationRequest(
                tenant_id="candidate",
                prompt=prompt,
                metadata={"adapter_envelope": enabled_envelope},
            ),
            authenticate=False,
        )
        substituted = json.dumps(
            {
                "schema_version": 1,
                "operator_type": "GLOBAL_ANCHOR",
                "term": "declared-term",
                "apply": True,
                "prompt": "different prompt",
            },
            sort_keys=True,
        )
        blocked = await runtime.run(
            GenerationRequest(
                tenant_id="candidate",
                prompt=prompt,
                metadata={"adapter_envelope": substituted},
            ),
            authenticate=False,
        )
        assert user_text_only.text == explicitly_disabled.text == "base"
        assert enabled.text == "adapter"
        assert blocked.receipt.status == "blocked"
        await runtime.close()

    asyncio.run(scenario())
