import json
import math
import uuid
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Final

import httpx
import pytest
import yaml
from integration._support.client import Gateway, Scenario, eventually, object_value, string_value
from integration._support.database import read_rows
from integration._support.ports import allocate_port
from integration._support.process import owned_proxy_process
from integration._support.upstream import ScenarioHandle, delete_scenario, register_scenario
from integration.cost_calculation.cost_tracking_case import JsonResponse
from pydantic import JsonValue, TypeAdapter

import litellm

_API_KEY: Final = "synthetic-decisions-key"
_ENV_KEY: Final = "synthetic-decisions-env-key"
_AZURE_AI_ENTRA_SCENARIO: Final = "providers"
_AZURE_AI_ENTRA_TOKEN: Final = "synthetic-decisions-entra-token"
_AZURE_AI_MISSING_CREDENTIALS_MESSAGE: Final = (
    "Missing Azure AI credentials - set an API key (`api_key` or AZURE_AI_API_KEY), or Entra ID / OAuth "
    "credentials (`tenant_id` + `client_id` + `client_secret`, `azure_ad_token`, an OIDC token, or a managed "
    "identity with `litellm.enable_azure_ad_token_refresh = True`)"
)
_PASS_THROUGH_MODEL: Final = "gpt-6-luna"
_PASS_THROUGH_AUTHORIZATION: Final = "Bearer customer-held-upstream-key"
_PASS_THROUGH_NEIGHBOUR: Final = "decisions-beside-a-pass-through"
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
_QUESTION_MAPPINGS: Final = TypeAdapter(dict[str, dict[str, object]])
_USAGE: Final[dict[str, JsonValue]] = {"input_tokens": 367, "output_tokens": 3}
_STATE: Final[dict[str, JsonValue]] = {"ticket": "The export job hangs at 99%", "component": "billing"}
_QUESTIONS: Final[dict[str, JsonValue]] = {
    "defect": {"type": "noul", "instructions": "Is this a defect?"},
    "severity": {"type": "choice", "criteria": {"low": "cosmetic", "high": "blocks users"}, "weight": 2},
    "confidence": {"type": "score", "instructions": "How sure are you?", "criteria": ["unsure", "sure"]},
}
_SDK_QUESTIONS: Final = _QUESTION_MAPPINGS.validate_python(_QUESTIONS)
_ANSWERS: Final[dict[str, JsonValue]] = {
    "defect": {"type": "noul", "noul": 0.93},
    "severity": {"type": "choice", "choice": "high", "confidence": 0.8, "probabilities": {"low": 0.2, "high": 0.8}},
    "confidence": {
        "type": "score",
        "score": 1.0,
        "confidence": 0.7,
        "legend": {"0": "unsure", "1": "sure"},
        "probabilities": {"0": 0.3, "1": 0.7},
    },
}
_CHAT_BODY: Final[dict[str, JsonValue]] = {"messages": [{"role": "user", "content": "hi"}]}
_CHAT_REPLY: Final[dict[str, JsonValue]] = {
    "id": "chatcmpl-decisions-parity",
    "object": "chat.completion",
    "created": 1700000000,
    "model": "pplx-decider-v1-27b",
    "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
    "usage": {"prompt_tokens": 3, "completion_tokens": 1, "total_tokens": 4},
}
_SPEND_QUERY: Final = (
    "SELECT spend, status, call_type, model_group, custom_llm_provider, api_base, prompt_tokens, completion_tokens, "
    'request_tags FROM "LiteLLM_SpendLogs" WHERE request_id = %s'
)


@dataclass(frozen=True, slots=True)
class _Provider:
    name: str
    model: str
    path: str
    body_model: str
    api_key: str | None
    wraps_result: bool
    cost_map_key: str | None
    provider_reported_cost: float | None = None


_PROVIDERS: Final = (
    _Provider(
        "perplexity",
        "perplexity/pplx-decider-v1-27b",
        "/v1/decisions",
        "pplx-decider-v1-27b",
        _API_KEY,
        False,
        "perplexity/pplx-decider-v1-27b",
    ),
    _Provider("typesafe", "typesafe/jev-1.13.0", "/v1/systemone", "jev-1.13.0", _API_KEY, False, "typesafe/jev-1.13.0"),
    _Provider("nace", "nace/drex-v2.0", "/v1/systemone", "drex-v2.0", _API_KEY, False, "nace/drex-v2.0"),
    _Provider(
        "openrouter",
        "openrouter/typesafe/jev-1.13",
        "/alpha/decisions",
        "typesafe/jev-1.13",
        _API_KEY,
        False,
        None,
        1.5834e-5,
    ),
    _Provider(
        "strands_decider", "strands_decider/systemone-decider", "/v1/systemone", "systemone-decider", None, False, None
    ),
    _Provider("laya", "laya/english", "/v1/systemone", "english", None, False, "laya/english"),
    _Provider(
        "bespoke", "bespoke/nimble-latest", "/v1/systemone", "nimble-latest", None, False, "bespoke/nimble-latest"
    ),
    _Provider("hosted_vllm", "hosted_vllm/Qwen/Qwen3-0.6B", "/v1/systemone", "Qwen/Qwen3-0.6B", None, False, None),
    _Provider(
        "fireworks_ai",
        "fireworks_ai/accounts/fireworks/routers/rune",
        "/v1/systemone",
        "accounts/fireworks/routers/rune",
        _API_KEY,
        False,
        "fireworks_ai/accounts/fireworks/routers/rune",
    ),
    _Provider(
        "cloudflare",
        "cloudflare/clef",
        "/ai/run/@cf/cloudflare/clef",
        "clef",
        _API_KEY,
        True,
        "cloudflare/@cf/cloudflare/clef",
    ),
    _Provider(
        "azure_ai", "azure_ai/decision-1", "/providers/microsoft/v1/systemone", "decision-1", _API_KEY, False, None
    ),
    _Provider(
        "databricks",
        "databricks/databricks-openjev-qwen35-4b",
        "/databricks-openjev-qwen35-4b/invocations",
        "databricks-openjev-qwen35-4b",
        _API_KEY,
        False,
        "databricks/databricks-openjev-qwen35-4b",
    ),
)
_LOCAL_PROVIDERS: Final = (
    pytest.param(
        _Provider("ollama", "ollama/d3:4b", "/v1/systemone", "d3:4b", None, False, None), id="regression-ollama"
    ),
    pytest.param(
        _Provider("ollama_chat", "ollama_chat/d3:4b", "/v1/systemone", "d3:4b", None, False, None),
        id="regression-ollama_chat",
    ),
    pytest.param(
        _Provider("hosted_vllm", "hosted_vllm/d3:4b", "/v1/systemone", "d3:4b", None, False, None),
        id="guard-hosted_vllm",
    ),
    pytest.param(
        _Provider("lm_studio", "lm_studio/d3:4b", "/v1/systemone", "d3:4b", None, False, None),
        id="guard-lm_studio",
    ),
)
_LOCAL_STATE: Final = "I was charged twice for order 4417."
_LOCAL_QUESTIONS: Final[Mapping[str, JsonValue]] = {
    "billing": {"type": "noul", "instructions": "Is this a billing issue?"}
}
_LOCAL_ANSWERS: Final[Mapping[str, JsonValue]] = {"billing": {"type": "noul", "noul": 0.91}}
_LOCAL_OPENAI_QUESTIONS: Final[Sequence[JsonValue]] = (
    {"type": "predicate", "name": "billing", "instructions": "Is this a billing issue?"},
)
_LOCAL_OPENAI_ANSWERS: Final[Sequence[JsonValue]] = ({"type": "predicate", "name": "billing", "probability": 0.91},)
_LOCAL_OPENAI_USAGE: Final[Mapping[str, JsonValue]] = {
    **_USAGE,
    "input_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 0},
    "output_tokens_details": {"reasoning_tokens": 0},
    "total_tokens": 370,
}
_PERPLEXITY: Final = _PROVIDERS[0]
_OPENROUTER: Final = next(provider for provider in _PROVIDERS if provider.name == "openrouter")
_OPENROUTER_CHAT_MODEL: Final = "openrouter/openai/gpt-5-mini"
_UNSUPPORTED_PROVIDER_MODEL: Final = "anthropic/claude-opus-5-5"
_CONNECTION_ERROR: Final = "litellm.APIConnectionError"
_GENERIC_API_ERROR: Final = "litellm.APIError"
_INVALID_BODIES: Final[tuple[tuple[str, dict[str, JsonValue]], ...]] = (
    ("missing questions", {"state": _STATE}),
    ("missing state", {"questions": _QUESTIONS}),
    ("numeric state", {"state": 5, "questions": _QUESTIONS}),
    ("empty questions", {"state": _STATE, "questions": {}}),
    ("noul without instructions or criteria", {"state": _STATE, "questions": {"q": {"type": "noul"}}}),
    ("choice without criteria", {"state": _STATE, "questions": {"q": {"type": "choice", "criteria": {}}}}),
    (
        "score with eleven criteria",
        {"state": _STATE, "questions": {"q": {"type": "score", "criteria": [f"level-{index}" for index in range(11)]}}},
    ),
    ("unknown question type", {"state": _STATE, "questions": {"q": {"type": "ranking", "criteria": ["a"]}}}),
)


def _number(value: JsonValue) -> float:
    assert isinstance(value, (int, float)) and not isinstance(value, bool), value
    return float(value)


def _expected_spend(provider: _Provider) -> float:
    if provider.provider_reported_cost is not None:
        return provider.provider_reported_cost
    if provider.cost_map_key is None:
        return 0.0
    prices: Final = object_value(
        json.loads(Path("model_prices_and_context_window.json").read_text())[provider.cost_map_key]
    )
    return _number(_USAGE["input_tokens"]) * _number(prices["input_cost_per_token"]) + _number(
        _USAGE["output_tokens"]
    ) * _number(prices["output_cost_per_token"])


def _usage(provider: _Provider) -> dict[str, JsonValue]:
    return {
        **_USAGE,
        **({"cost": provider.provider_reported_cost} if provider.provider_reported_cost is not None else {}),
    }


def _answer_body(provider: _Provider) -> dict[str, JsonValue]:
    answer: Final[dict[str, JsonValue]] = {
        "model": provider.body_model,
        "answers": _ANSWERS,
        "usage": _usage(provider),
    }
    return {"result": answer, "success": True} if provider.wraps_result else answer


def _register(scenario: Scenario, body: dict[str, JsonValue], *, status: int = 200) -> ScenarioHandle:
    handle: Final = register_scenario(
        f"decisions-{uuid.uuid4().hex[:12]}", JsonResponse(content_type="application/json", body=body, status=status)
    )
    scenario.cleanups.callback(delete_scenario, handle)
    return handle


def _deployment(scenario: Scenario, handle: ScenarioHandle, provider: _Provider) -> str:
    return scenario.model(model=provider.model, api_base=handle.api_base(), api_key=provider.api_key)


def _decide(gateway: Gateway, model: str, *, key: str | None = None, **extra: JsonValue) -> httpx.Response:
    return gateway.request(
        "POST", "/v1/systemone", {"model": model, "state": _STATE, "questions": _QUESTIONS, **extra}, key=key
    )


def _chat(gateway: Gateway, model: str, *, key: str | None = None, **extra: JsonValue) -> httpx.Response:
    return gateway.request("POST", "/v1/chat/completions", {"model": model, **_CHAT_BODY, **extra}, key=key)


def _observed_requests(gateway: Gateway) -> tuple[dict[str, JsonValue], ...]:
    with httpx.Client(base_url=gateway.upstream_url, timeout=5, trust_env=False) as upstream:
        return tuple(map(object_value, upstream.get("/__observations").json()["requests"]))


def _calls_to(requests: Sequence[dict[str, JsonValue]], handle: ScenarioHandle) -> list[dict[str, JsonValue]]:
    return [request for request in requests if string_value(request["path"]).startswith(f"/{handle.scenario_id}/")]


def _upstream_calls(gateway: Gateway, handle: ScenarioHandle) -> list[dict[str, JsonValue]]:
    return _calls_to(_observed_requests(gateway), handle)


def _sdk_inference_call(gateway: Gateway, handle: ScenarioHandle, provider: _Provider) -> dict[str, JsonValue]:
    observed: Final = _upstream_calls(gateway, handle)
    inference_path: Final = f"/{handle.scenario_id}/v1/systemone"
    (call,) = tuple(request for request in observed if request["path"] == inference_path)
    metadata_calls: Final = tuple(request for request in observed if request["path"] != inference_path)
    assert metadata_calls in (
        (),
        (
            {
                "path": f"{inference_path}/api/show",
                "authorization": "",
                "body": {"name": provider.body_model},
                "method": "POST",
                "api_key": "",
            },
        ),
    ), observed
    return call


def _spend_row(call_id: str) -> dict[str, JsonValue]:
    rows: Final = eventually(lambda: read_rows(_SPEND_QUERY, (call_id,)), lambda found: len(found) == 1, seconds=70)
    return rows[0]


def _assert_connection_error(response: httpx.Response) -> None:
    assert 500 <= response.status_code < 600, response.text
    assert _CONNECTION_ERROR in response.text, response.text
    assert _GENERIC_API_ERROR not in response.text, response.text


def test_azure_ai_entra_auth_reaches_the_scripted_upstream_for_both_decisions_routes(
    gateway: Gateway, scenario: Scenario
) -> None:
    provider: Final = next(provider for provider in _PROVIDERS if provider.name == "azure_ai")
    handle: Final = register_scenario(
        _AZURE_AI_ENTRA_SCENARIO,
        JsonResponse(content_type="application/json", body=_answer_body(provider)),
    )
    scenario.cleanups.callback(delete_scenario, handle)
    model: Final = "azure_ai/decision-1-entra"
    body: Final = {"model": model, "state": _STATE, "questions": _QUESTIONS}
    decisions_body: Final = {
        "model": model,
        "input": "The export job hangs at 99%",
        "questions": [{"type": "predicate", "name": "defect", "instructions": "Is this a defect?"}],
    }

    no_credentials_body: Final = {**body, "model": "azure_ai/decision-1-no-credentials"}
    systemone_auth_error: Final = gateway.request("POST", "/v1/systemone", no_credentials_body)
    assert systemone_auth_error.status_code == 401, systemone_auth_error.text
    assert _AZURE_AI_MISSING_CREDENTIALS_MESSAGE in string_value(
        object_value(object_value(systemone_auth_error.json())["error"])["message"]
    )
    no_credentials_decisions_body: Final = {
        **decisions_body,
        "model": "azure_ai/decision-1-no-credentials",
    }
    decisions_auth_error: Final = gateway.request("POST", "/v1/decisions", no_credentials_decisions_body)
    assert decisions_auth_error.status_code == 401, decisions_auth_error.text
    assert _AZURE_AI_MISSING_CREDENTIALS_MESSAGE in string_value(
        object_value(object_value(decisions_auth_error.json())["error"])["message"]
    )
    assert not _upstream_calls(gateway, handle)

    systemone_response: Final = gateway.request("POST", "/v1/systemone", body)
    assert systemone_response.status_code == 200, systemone_response.text
    decisions_response: Final = gateway.request("POST", "/v1/decisions", decisions_body)
    assert decisions_response.status_code == 200, decisions_response.text

    calls: Final = _upstream_calls(gateway, handle)
    assert tuple(string_value(call["authorization"]) for call in calls) == (
        f"Bearer {_AZURE_AI_ENTRA_TOKEN}",
        f"Bearer {_AZURE_AI_ENTRA_TOKEN}",
    )
    assert tuple(string_value(call["path"]) for call in calls) == (
        "/providers/microsoft/v1/systemone",
        "/providers/microsoft/v1/systemone",
    )


def _pass_through_config(directory: Path, pass_through_target: str, native_api_base: str) -> Path:
    base: Final = _JSON_OBJECT.validate_python(yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()))
    config: Final = {
        **base,
        "general_settings": {
            **object_value(base["general_settings"]),
            "pass_through_endpoints": [
                {
                    "path": "/v1/decisions",
                    "target": pass_through_target,
                    "headers": {"Authorization": _PASS_THROUGH_AUTHORIZATION},
                }
            ],
        },
        "model_list": [
            {
                "model_name": _PASS_THROUGH_NEIGHBOUR,
                "litellm_params": {"model": _PERPLEXITY.model, "api_base": native_api_base, "api_key": _API_KEY},
            }
        ],
    }
    path: Final = directory / "decisions-pass-through.yaml"
    path.write_text(yaml.safe_dump(config))
    return path


@pytest.mark.parametrize("provider", _PROVIDERS, ids=lambda provider: provider.name)
def test_each_provider_gets_its_own_path_key_and_body_and_is_billed(gateway: Gateway, provider: _Provider) -> None:
    expected_spend: Final = _expected_spend(provider)
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(provider))
        model: Final = _deployment(scenario, handle, provider)
        response: Final = _decide(gateway, model)
        assert response.status_code == 200, response.text
        assert response.json() == {"model": provider.body_model, "answers": _ANSWERS, "usage": _usage(provider)}
        assert response.headers["x-litellm-model-group"] == model
        assert math.isclose(float(response.headers.get("x-litellm-response-cost", "0")), expected_spend, rel_tol=1e-9)
        (call,) = _upstream_calls(gateway, handle)
        assert call["path"] == f"/{handle.scenario_id}{provider.path}"
        assert call["authorization"] == (f"Bearer {provider.api_key}" if provider.api_key else "")
        assert call["body"] == {"model": provider.body_model, "state": _STATE, "questions": _QUESTIONS}
        row: Final = _spend_row(response.headers["x-litellm-call-id"])
        assert (
            row["status"],
            row["call_type"],
            row["custom_llm_provider"],
            row["model_group"],
            row["api_base"],
            row["prompt_tokens"],
            row["completion_tokens"],
        ) == ("success", "asystemone", provider.name, model, f"{handle.api_base()}{provider.path}", 367, 3)
        assert math.isclose(_number(row["spend"]), expected_spend, rel_tol=1e-9), row


@pytest.fixture(autouse=True)
def _without_ambient_ollama_settings(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("OLLAMA_API_KEY", raising=False)
    monkeypatch.delenv("OLLAMA_API_BASE", raising=False)


@pytest.mark.parametrize("provider", _LOCAL_PROVIDERS)
@pytest.mark.parametrize("route", ("/v1/systemone", "/v1/decisions"), ids=("systemone", "decisions"))
@pytest.mark.parametrize(
    ("base_suffix", "api_key"),
    (
        pytest.param("", None, id="root-no-key"),
        pytest.param("/", _API_KEY, id="root-slash-key"),
        pytest.param("/v1", None, id="v1-no-key"),
        pytest.param("/v1/", _API_KEY, id="v1-slash-key"),
    ),
)
def test_local_decisions_routes_preserve_wire_auth_usage_and_spend(
    gateway: Gateway, provider: _Provider, route: str, base_suffix: str, api_key: str | None
) -> None:
    native_reply: Final[Mapping[str, JsonValue]] = {
        "model": provider.body_model,
        "answers": dict(_LOCAL_ANSWERS),
        "usage": _USAGE,
    }
    request_body: Final[Mapping[str, JsonValue]] = (
        {"state": _LOCAL_STATE, "questions": dict(_LOCAL_QUESTIONS)}
        if route == "/v1/systemone"
        else {"input": _LOCAL_STATE, "questions": list(_LOCAL_OPENAI_QUESTIONS)}
    )
    expected_reply: Final = (
        native_reply
        if route == "/v1/systemone"
        else {"model": provider.body_model, "answers": list(_LOCAL_OPENAI_ANSWERS), "usage": dict(_LOCAL_OPENAI_USAGE)}
    )
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, dict(native_reply))
        model: Final = scenario.model(
            model=provider.model, api_base=f"{handle.api_base()}{base_suffix}", api_key=api_key
        )
        response: Final = gateway.request("POST", route, {"model": model, **request_body})
        assert response.status_code == 200, response.text
        assert response.json() == expected_reply
        assert response.headers["x-litellm-model-group"] == model
        assert float(response.headers.get("x-litellm-response-cost", "0")) == 0.0
        (call,) = _upstream_calls(gateway, handle)
        assert (call["path"], call["authorization"], call["body"]) == (
            f"/{handle.scenario_id}/v1/systemone",
            f"Bearer {api_key}" if api_key else "",
            {"model": provider.body_model, "state": _LOCAL_STATE, "questions": _LOCAL_QUESTIONS},
        )
        row: Final = _spend_row(response.headers["x-litellm-call-id"])
        assert {key: value for key, value in row.items() if key != "request_tags"} == {
            "spend": 0.0,
            "status": "success",
            "call_type": "asystemone" if route == "/v1/systemone" else "adecisions",
            "model_group": model,
            "custom_llm_provider": provider.name,
            "api_base": f"{handle.api_base()}/v1/systemone",
            "prompt_tokens": _USAGE["input_tokens"],
            "completion_tokens": _USAGE["output_tokens"],
        }


@pytest.mark.parametrize("provider", _LOCAL_PROVIDERS)
@pytest.mark.parametrize("asynchronous", (False, True), ids=("sync", "async"))
async def test_local_systemone_sdk_preserves_native_wire(
    gateway: Gateway, provider: _Provider, asynchronous: bool
) -> None:
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(provider))
        response: Final = (
            await litellm.asystemone(
                model=provider.model, state=_STATE, questions=_SDK_QUESTIONS, api_base=f"{handle.api_base()}/v1/"
            )
            if asynchronous
            else litellm.systemone(
                model=provider.model, state=_STATE, questions=_SDK_QUESTIONS, api_base=handle.api_base()
            )
        )
        assert response.model_dump(mode="json") == _answer_body(provider)
        call: Final = _sdk_inference_call(gateway, handle, provider)
        assert (call["path"], call["authorization"], call["body"]) == (
            f"/{handle.scenario_id}/v1/systemone",
            "",
            {"model": provider.body_model, "state": _STATE, "questions": _QUESTIONS},
        )


@pytest.mark.parametrize("provider", _LOCAL_PROVIDERS)
@pytest.mark.parametrize("asynchronous", (False, True), ids=("sync", "async"))
async def test_local_decisions_sdk_translates_openai_format(
    gateway: Gateway, provider: _Provider, asynchronous: bool
) -> None:
    questions: Final = TypeAdapter(list[dict[str, object]]).validate_python(_LOCAL_OPENAI_QUESTIONS)
    native_reply: Final[Mapping[str, JsonValue]] = {
        "model": provider.body_model,
        "answers": dict(_LOCAL_ANSWERS),
        "usage": _USAGE,
    }
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, dict(native_reply))
        response: Final = (
            await litellm.adecisions(
                model=provider.model, input=_LOCAL_STATE, questions=questions, api_base=f"{handle.api_base()}/v1/"
            )
            if asynchronous
            else litellm.decisions(
                model=provider.model, input=_LOCAL_STATE, questions=questions, api_base=handle.api_base()
            )
        )
        assert response.model_dump(mode="json") == {
            "model": provider.body_model,
            "answers": list(_LOCAL_OPENAI_ANSWERS),
            "usage": dict(_LOCAL_OPENAI_USAGE),
        }
        call: Final = _sdk_inference_call(gateway, handle, provider)
        assert (call["path"], call["authorization"], call["body"]) == (
            f"/{handle.scenario_id}/v1/systemone",
            "",
            {"model": provider.body_model, "state": _LOCAL_STATE, "questions": _LOCAL_QUESTIONS},
        )


@pytest.mark.parametrize("provider", _LOCAL_PROVIDERS[:2])
@pytest.mark.parametrize(
    ("environment_base", "api_key", "environment_key", "authorization"),
    (
        pytest.param(True, None, _ENV_KEY, f"Bearer {_ENV_KEY}", id="environment-base-and-key"),
        pytest.param(False, _API_KEY, _ENV_KEY, f"Bearer {_API_KEY}", id="explicit-overrides-environment"),
        pytest.param(False, "fake-api-key", _ENV_KEY, "", id="fake-key-suppresses-environment"),
        pytest.param(False, None, None, "", id="unrelated-keys-not-forwarded"),
    ),
)
def test_ollama_decisions_sdk_respects_environment_and_explicit_auth(
    gateway: Gateway,
    monkeypatch: pytest.MonkeyPatch,
    provider: _Provider,
    environment_base: bool,
    api_key: str | None,
    environment_key: str | None,
    authorization: str,
) -> None:
    monkeypatch.setenv("OPENAI_API_KEY", "synthetic-unrelated-openai-key")
    monkeypatch.setenv("LITELLM_API_KEY", "synthetic-unrelated-litellm-key")
    if environment_key is not None:
        monkeypatch.setenv("OLLAMA_API_KEY", environment_key)
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(provider))
        unused: Final = _register(scenario, _answer_body(provider))
        monkeypatch.setenv("OLLAMA_API_BASE", handle.api_base() if environment_base else unused.api_base())
        response: Final = litellm.systemone(
            model=provider.model,
            state=_STATE,
            questions=_SDK_QUESTIONS,
            api_base=None if environment_base else f"{handle.api_base()}/v1/",
            api_key=api_key,
        )
        assert response.model_dump(mode="json") == _answer_body(provider)
        call: Final = _sdk_inference_call(gateway, handle, provider)
        assert (call["path"], call["authorization"], call["body"]) == (
            f"/{handle.scenario_id}/v1/systemone",
            authorization,
            {"model": provider.body_model, "state": _STATE, "questions": _QUESTIONS},
        )
        assert _upstream_calls(gateway, unused) == []


def test_test_connection_evaluation_mode_uses_typesafe_decisions_path(gateway: Gateway) -> None:
    provider: Final = _PROVIDERS[1]
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(provider))
        response: Final = gateway.request(
            "POST",
            "/health/test_connection",
            {
                "litellm_params": {"model": provider.model, "api_base": handle.api_base(), "api_key": provider.api_key},
                "mode": "evaluation",
            },
        )

        assert response.status_code == 200, response.text
        assert response.json()["status"] == "success", response.text
        (call,) = _upstream_calls(gateway, handle)
        assert call["path"] == f"/{handle.scenario_id}{provider.path}"


def test_test_connection_by_configured_alias_probes_the_stored_model(gateway: Gateway) -> None:
    pytest.skip("BUG: /health/test_connection given only a configured alias sends the alias as the model")
    provider: Final = _PROVIDERS[1]
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(provider))
        model: Final = _deployment(scenario, handle, provider)
        response: Final = gateway.request(
            "POST",
            "/health/test_connection",
            {"litellm_params": {"model": model}, "mode": "evaluation"},
        )

        assert response.status_code == 200, response.text
        assert response.json()["status"] == "success", response.text
        (call,) = _upstream_calls(gateway, handle)
        assert call["path"] == f"/{handle.scenario_id}{provider.path}"


def test_repeated_identical_requests_each_reach_the_upstream_and_are_each_billed(gateway: Gateway) -> None:
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(_PERPLEXITY))
        model: Final = _deployment(scenario, handle, _PERPLEXITY)
        responses: Final = tuple(_decide(gateway, model) for _ in range(2))
        assert [response.status_code for response in responses] == [200, 200], [r.text for r in responses]
        call_ids: Final = tuple(response.headers["x-litellm-call-id"] for response in responses)
        assert len(set(call_ids)) == 2, call_ids
        assert len(_upstream_calls(gateway, handle)) == 2
        for call_id in call_ids:
            assert _spend_row(call_id)["status"] == "success"


async def test_sdk_sync_and_async_clients_send_the_same_request(gateway: Gateway) -> None:
    provider: Final = _PROVIDERS[1]
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(provider))
        synchronous: Final = litellm.decisions(
            model=provider.model, state=_STATE, questions=_SDK_QUESTIONS, api_base=handle.api_base(), api_key=_API_KEY
        )
        asynchronous: Final = await litellm.asystemone(
            model=provider.model, state=_STATE, questions=_SDK_QUESTIONS, api_base=handle.api_base(), api_key=_API_KEY
        )
        for response in (synchronous, asynchronous):
            assert response.model_dump(mode="json") == {
                "model": provider.body_model,
                "answers": _ANSWERS,
                "usage": _USAGE,
            }
        calls: Final = _upstream_calls(gateway, handle)
        assert len(calls) == 2, calls
        for call in calls:
            assert call["path"] == f"/{handle.scenario_id}{provider.path}"
            assert call["authorization"] == f"Bearer {_API_KEY}"
            assert call["body"] == {"model": provider.body_model, "state": _STATE, "questions": _QUESTIONS}


def test_gateway_only_fields_stay_at_the_gateway_and_tags_reach_the_spend_log(gateway: Gateway) -> None:
    tag: Final = f"decisions-audit-{uuid.uuid4().hex[:8]}"
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(_PERPLEXITY))
        model: Final = _deployment(scenario, handle, _PERPLEXITY)
        response: Final = _decide(
            gateway, model, user="auditor", num_retries=0, temperature=0.2, metadata={"tags": [tag]}
        )
        assert response.status_code == 200, response.text
        (call,) = _upstream_calls(gateway, handle)
        assert call["body"] == {"model": _PERPLEXITY.body_model, "state": _STATE, "questions": _QUESTIONS}
        row: Final = _spend_row(response.headers["x-litellm-call-id"])
        tags: Final = row["request_tags"]
        assert isinstance(tags, list) and tag in tags, row


def test_invalid_bodies_are_refused_at_the_gateway_without_an_upstream_call(gateway: Gateway) -> None:
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(_PERPLEXITY))
        model: Final = _deployment(scenario, handle, _PERPLEXITY)
        for label, body in _INVALID_BODIES:
            response: Final = gateway.request("POST", "/v1/systemone", {"model": model, **body})
            assert response.status_code == 400, (label, response.text)
            assert "Invalid Decisions request" in response.text, (label, response.text)
        assert _upstream_calls(gateway, handle) == []


def test_unknown_model_is_refused_like_chat(gateway: Gateway) -> None:
    model: Final = f"missing-{uuid.uuid4().hex}"
    decisions: Final = _decide(gateway, model)
    chat: Final = _chat(gateway, model)
    assert 400 <= decisions.status_code < 500, decisions.text
    assert decisions.status_code == chat.status_code, (decisions.text, chat.text)


def test_key_checks_match_chat(gateway: Gateway) -> None:
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(_PERPLEXITY))
        model: Final = _deployment(scenario, handle, _PERPLEXITY)
        anonymous: Final = gateway.client.post(
            "/v1/systemone", json={"model": model, "state": _STATE, "questions": _QUESTIONS}
        )
        assert anonymous.status_code == 401, anonymous.text
        restricted: Final = scenario.key(models=[f"other-{uuid.uuid4().hex}"])
        refused: Final = _decide(gateway, model, key=restricted)
        assert 400 <= refused.status_code < 500, refused.text
        assert refused.status_code == _chat(gateway, model, key=restricted).status_code, refused.text
        assert _upstream_calls(gateway, handle) == []
        spender: Final = scenario.key(max_budget=1e-06)
        first: Final = _decide(gateway, model, key=spender)
        assert first.status_code == 200, first.text
        blocked: Final = eventually(
            lambda: _decide(gateway, model, key=spender), lambda response: response.status_code != 200, seconds=70
        )
        assert 400 <= blocked.status_code < 500, blocked.text
        assert blocked.status_code == _chat(gateway, model, key=spender).status_code, blocked.text


def test_request_body_api_base_is_refused_like_chat_without_an_upstream_call(gateway: Gateway) -> None:
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(_PERPLEXITY))
        model: Final = _deployment(scenario, handle, _PERPLEXITY)
        decisions: Final = _decide(gateway, model, api_base=f"http://127.0.0.1:{allocate_port()}")
        chat: Final = _chat(gateway, model, api_base=f"http://127.0.0.1:{allocate_port()}")
        assert 400 <= decisions.status_code < 500, decisions.text
        assert decisions.status_code == chat.status_code, (decisions.text, chat.text)
        assert _upstream_calls(gateway, handle) == []


def test_a_deployment_without_a_key_sends_the_provider_env_key_to_its_configured_api_base(gateway: Gateway) -> None:
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(_PERPLEXITY))
        model: Final = scenario.model(model=_PERPLEXITY.model, api_base=handle.api_base(), api_key=None)
        response: Final = _decide(gateway, model)
        assert response.status_code == 200, response.text
        (call,) = _upstream_calls(gateway, handle)
        assert (call["path"], call["authorization"]) == (f"/{handle.scenario_id}/v1/decisions", f"Bearer {_ENV_KEY}")


def test_a_deployment_opted_into_client_api_base_sends_decisions_and_chat_to_the_body_api_base(
    gateway: Gateway,
) -> None:
    with gateway.scenario() as scenario:
        configured: Final = _register(scenario, _answer_body(_PERPLEXITY))
        decisions_target: Final = _register(scenario, _answer_body(_PERPLEXITY))
        chat_target: Final = _register(scenario, _CHAT_REPLY)
        model: Final = scenario.model(
            model=_PERPLEXITY.model,
            api_base=configured.api_base(),
            api_key=_API_KEY,
            configurable_clientside_auth_params=["api_base"],
        )
        decisions: Final = _decide(gateway, model, api_base=decisions_target.api_base())
        chat: Final = _chat(gateway, model, api_base=chat_target.api_base())
        assert decisions.status_code == 200, decisions.text
        assert chat.status_code == 200, chat.text
        observed: Final = _observed_requests(gateway)
        assert [call["path"] for call in _calls_to(observed, decisions_target)] == [
            f"/{decisions_target.scenario_id}/v1/decisions"
        ]
        assert [call["path"] for call in _calls_to(observed, chat_target)] == [
            f"/{chat_target.scenario_id}/chat/completions"
        ]
        assert _calls_to(observed, configured) == []


def test_a_config_pass_through_at_v1_decisions_keeps_answering_and_the_native_api_serves_system_one(
    gateway: Gateway, tmp_path: Path
) -> None:
    with gateway.scenario() as scenario:
        pass_through_target: Final = _register(
            scenario, {"model": _PASS_THROUGH_MODEL, "answers": _ANSWERS, "usage": _USAGE}
        )
        native_target: Final = _register(scenario, _answer_body(_PERPLEXITY))
        config: Final = _pass_through_config(
            tmp_path, f"{pass_through_target.api_base()}/v1/decisions", native_target.api_base()
        )
        with owned_proxy_process(gateway, tmp_path, {}, config=config) as owned:
            through: Final = owned.gateway.request(
                "POST", "/v1/decisions", {"model": _PASS_THROUGH_MODEL, "state": _STATE, "questions": _QUESTIONS}
            )
            native: Final = owned.gateway.request(
                "POST", "/systemone", {"model": _PASS_THROUGH_NEIGHBOUR, "state": _STATE, "questions": _QUESTIONS}
            )
        assert through.status_code == 200, through.text
        assert through.json() == {"model": _PASS_THROUGH_MODEL, "answers": _ANSWERS, "usage": _USAGE}
        observed: Final = _observed_requests(gateway)
        (forwarded,) = _calls_to(observed, pass_through_target)
        assert (forwarded["path"], forwarded["authorization"], object_value(forwarded["body"])["model"]) == (
            f"/{pass_through_target.scenario_id}/v1/decisions",
            _PASS_THROUGH_AUTHORIZATION,
            _PASS_THROUGH_MODEL,
        )
        assert native.status_code == 200, native.text
        assert [call["path"] for call in _calls_to(observed, native_target)] == [
            f"/{native_target.scenario_id}/v1/decisions"
        ]


@pytest.mark.parametrize("status", (401, 429, 500))
def test_upstream_errors_keep_their_status_and_log_an_unbilled_failure(gateway: Gateway, status: int) -> None:
    marker: Final = f"scripted-{status}-{uuid.uuid4().hex[:8]}"
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, {"error": {"message": marker}}, status=status)
        model: Final = _deployment(scenario, handle, _PERPLEXITY)
        response: Final = _decide(gateway, model, num_retries=0)
        assert response.status_code == status, response.text
        assert marker in response.text
        row: Final = _spend_row(response.headers["x-litellm-call-id"])
        assert (row["status"], row["call_type"], row["model_group"], _number(row["spend"])) == (
            "failure",
            "asystemone",
            model,
            0.0,
        )


@pytest.mark.parametrize(
    ("reply", "field"),
    (
        pytest.param({"model": _PERPLEXITY.body_model, "usage": _USAGE}, "answers", id="missing_answers"),
        pytest.param(
            {"model": _PERPLEXITY.body_model, "answers": {"defect": {"type": "Request timed out"}}, "usage": _USAGE},
            "answers.defect",
            id="answer_type_timeout_text",
        ),
    ),
)
def test_a_malformed_upstream_success_is_a_server_error_naming_the_field(
    gateway: Gateway, reply: dict[str, JsonValue], field: str
) -> None:
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, reply)
        model: Final = _deployment(scenario, handle, _PERPLEXITY)
        response: Final = _decide(gateway, model, num_retries=0)
        assert response.status_code == 500, response.text
        assert field in response.text
        assert _spend_row(response.headers["x-litellm-call-id"])["status"] == "failure"


@pytest.mark.parametrize("provider", _PROVIDERS, ids=lambda provider: provider.name)
def test_unreachable_upstream_fails_only_its_own_deployment(gateway: Gateway, provider: _Provider) -> None:
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(provider))
        healthy: Final = _deployment(scenario, handle, provider)
        dead: Final = scenario.model(
            model=provider.model, api_base=f"http://127.0.0.1:{allocate_port()}", api_key=provider.api_key
        )
        failed: Final = _decide(gateway, dead, num_retries=0)
        _assert_connection_error(failed)
        assert _spend_row(failed.headers["x-litellm-call-id"])["status"] == "failure"
        served: Final = _decide(gateway, healthy)
        assert served.status_code == 200, served.text
        assert len(_upstream_calls(gateway, handle)) == 1


def test_an_unreachable_openrouter_deployment_reports_a_connection_error_on_chat_embeddings_and_decisions(
    gateway: Gateway,
) -> None:
    with gateway.scenario() as scenario:
        dead_api_base: Final = f"http://127.0.0.1:{allocate_port()}"
        chat_model: Final = scenario.model(model=_OPENROUTER_CHAT_MODEL, api_base=dead_api_base, api_key=_API_KEY)
        decisions_model: Final = scenario.model(model=_OPENROUTER.model, api_base=dead_api_base, api_key=_API_KEY)
        chat: Final = _chat(gateway, chat_model, num_retries=0)
        streamed: Final = _chat(gateway, chat_model, num_retries=0, stream=True)
        embeddings: Final = gateway.request(
            "POST", "/v1/embeddings", {"model": chat_model, "input": "hi", "num_retries": 0}
        )
        decisions: Final = _decide(gateway, decisions_model, num_retries=0)
        for response in (chat, streamed, embeddings, decisions):
            _assert_connection_error(response)
        for response in (chat, decisions):
            assert _spend_row(response.headers["x-litellm-call-id"])["status"] == "failure"


async def test_sdk_openrouter_connection_failures_raise_a_connection_error(gateway: Gateway) -> None:
    dead_api_base: Final = f"http://127.0.0.1:{allocate_port()}"
    with pytest.raises(litellm.APIConnectionError):
        litellm.completion(
            model=_OPENROUTER_CHAT_MODEL,
            messages=[{"role": "user", "content": "hi"}],
            api_base=dead_api_base,
            api_key=_API_KEY,
        )
    with pytest.raises(litellm.APIConnectionError):
        await litellm.acompletion(
            model=_OPENROUTER_CHAT_MODEL,
            messages=[{"role": "user", "content": "hi"}],
            api_base=dead_api_base,
            api_key=_API_KEY,
        )
    with pytest.raises(litellm.APIConnectionError) as decisions_error:
        litellm.decisions(
            model=_OPENROUTER.model, state=_STATE, questions=_SDK_QUESTIONS, api_base=dead_api_base, api_key=_API_KEY
        )
    assert decisions_error.value.llm_provider == "openrouter"
    with pytest.raises(litellm.APIConnectionError) as systemone_error:
        await litellm.asystemone(
            model=_OPENROUTER.model, state=_STATE, questions=_SDK_QUESTIONS, api_base=dead_api_base, api_key=_API_KEY
        )
    assert systemone_error.value.llm_provider == "openrouter"


def test_a_deployment_whose_provider_has_no_decisions_support_is_refused_naming_every_supported_provider(
    gateway: Gateway,
) -> None:
    with gateway.scenario() as scenario:
        handle: Final = _register(scenario, _answer_body(_PERPLEXITY))
        model: Final = scenario.model(model=_UNSUPPORTED_PROVIDER_MODEL, api_base=handle.api_base(), api_key=_API_KEY)
        response: Final = _decide(gateway, model)
        assert response.status_code == 400, response.text
        for provider in _PROVIDERS:
            assert provider.name in response.text, response.text
        assert _upstream_calls(gateway, handle) == []


def test_typesafe_gateway_preserves_missing_usage(gateway: Gateway) -> None:
    with gateway.scenario() as scenario:
        body = {"model": "jev-1.13.0", "answers": {"exp": {"type": "noul", "noul": 0.8}}}
        handle = _register(scenario, body)
        model = _deployment(scenario, handle, _PROVIDERS[1])
        response = gateway.request(
            "POST",
            "/v1/systemone",
            {
                "model": model,
                "state": {"finding": "x"},
                "questions": {"exp": {"type": "noul", "instructions": "Is it exploitable?"}},
            },
        )
        assert response.status_code == 200, response.text
        assert response.json()["usage"] is None
        assert response.json()["answers"] == body["answers"]