"""DashScope caps an embeddings request at 10 inputs. We sent 273.

The assistant's semantic search warms up by embedding its whole tool catalogue
in one call. Every one of those calls came back
``400 InternalError.Algo.InvalidParameter: batch size is invalid, it should not
be larger than 10`` — so the semantic layer was dead from the day it shipped and
the only symptom was search quietly falling back to keyword matching. 「软件目录」
matched nothing (the corpus is English, the query is not) and the assistant told
an employee OpenAI was not in Helix.

The stub here REJECTS an oversized batch exactly as DashScope does. A stub that
accepted any batch size would have let the original bug through unnoticed, which
is the whole reason this file exists.
"""
from __future__ import annotations

import httpx
import pytest
import respx

from app.core.exceptions import AppException
from app.domains.bailian.llm_client import (
    DEFAULT_EMBED_MODEL,
    BailianLLMClient,
    embed_endpoint_for_region,
)

#: The real limit, measured against the live endpoint: 10 succeeds, 11 is a 400.
WIRE_MAX_BATCH = 10


def _endpoint() -> str:
    return embed_endpoint_for_region("intl")


def _stub_like_dashscope(calls: list[int]):
    """Respond the way DashScope does, batch cap included."""

    def _handler(request: httpx.Request) -> httpx.Response:
        import json

        inputs = json.loads(request.content)["input"]
        calls.append(len(inputs))
        if len(inputs) > WIRE_MAX_BATCH:
            return httpx.Response(
                400,
                json={"error": {
                    "message": (
                        "<400> InternalError.Algo.InvalidParameter: Value "
                        "error, batch size is invalid, it should not be larger "
                        "than 10.: input.contents"
                    ),
                    "code": "InvalidParameter",
                }},
            )
        return httpx.Response(200, json={"data": [
            # A vector that encodes its own position, so a reordering bug in the
            # chunking cannot hide behind identical values.
            {"index": i, "embedding": [float(hash(text) % 1000), float(i)]}
            for i, text in enumerate(inputs)
        ]})

    return _handler


@pytest.mark.asyncio
@respx.mock
async def test_a_catalogue_sized_batch_is_chunked_and_succeeds() -> None:
    calls: list[int] = []
    respx.post(_endpoint()).mock(side_effect=_stub_like_dashscope(calls))

    client = BailianLLMClient(api_key="sk-test", region="intl")
    texts = [f"tool number {i}" for i in range(273)]

    vectors = await client.embed(texts, model=DEFAULT_EMBED_MODEL)

    assert len(vectors) == 273, "one vector per input, or the caller mis-pairs them"
    assert calls, "no request was made"
    assert max(calls) <= WIRE_MAX_BATCH, f"a batch of {max(calls)} would be a 400"
    assert sum(calls) == 273, "inputs were dropped or duplicated across chunks"


@pytest.mark.asyncio
@respx.mock
async def test_the_order_survives_chunking() -> None:
    """Callers zip these vectors against their own list of names."""
    calls: list[int] = []
    respx.post(_endpoint()).mock(side_effect=_stub_like_dashscope(calls))

    client = BailianLLMClient(api_key="sk-test", region="intl")
    texts = [f"text {i}" for i in range(25)]

    vectors = await client.embed(texts, model=DEFAULT_EMBED_MODEL)

    # The stub's second component is the within-chunk index; rebuilt in order it
    # has to read 0,1,…,9,0,1,… — anything else means chunks were interleaved.
    within_chunk = [v[1] for v in vectors]
    assert within_chunk == [float(i % WIRE_MAX_BATCH) for i in range(25)]
    # And the first component identifies the text itself.
    assert [v[0] for v in vectors] == [float(hash(t) % 1000) for t in texts]


@pytest.mark.asyncio
@respx.mock
async def test_a_batch_at_the_limit_is_one_call() -> None:
    calls: list[int] = []
    respx.post(_endpoint()).mock(side_effect=_stub_like_dashscope(calls))

    client = BailianLLMClient(api_key="sk-test", region="intl")
    await client.embed([f"t{i}" for i in range(WIRE_MAX_BATCH)])

    assert calls == [WIRE_MAX_BATCH], "chunking split a request that fit"


@pytest.mark.asyncio
@respx.mock
async def test_a_real_error_still_reaches_the_caller() -> None:
    """Chunking must not swallow a genuine failure into a partial result."""
    respx.post(_endpoint()).mock(return_value=httpx.Response(401, json={}))

    client = BailianLLMClient(api_key="sk-test", region="intl")
    with pytest.raises(AppException):
        await client.embed([f"t{i}" for i in range(15)])


@pytest.mark.asyncio
async def test_no_texts_makes_no_call() -> None:
    client = BailianLLMClient(api_key="sk-test", region="intl")
    assert await client.embed([]) == []


@pytest.mark.asyncio
@respx.mock
async def test_a_failing_batch_cancels_its_siblings() -> None:
    """A bare ``gather`` leaves the siblings running.

    The first bad batch of a 273-item warm-up returns in a second and the caller
    degrades to keyword matching — while 27 orphan tasks go on taking the
    semaphore and POSTing for another minute or two with nobody reading a single
    result. Counted here, because "bounded" has to be measurable.
    """
    import asyncio

    attempts: list[int] = []

    async def _slow_then_fail(request: httpx.Request) -> httpx.Response:
        import json

        inputs = json.loads(request.content)["input"]
        attempts.append(len(inputs))
        if len(attempts) == 1:
            return httpx.Response(500, json={"error": "boom"})
        # Slow enough that a leaked sibling would still be in flight when the
        # assertion runs, and would keep going.
        await asyncio.sleep(0.25)
        return httpx.Response(200, json={"data": [
            {"index": i, "embedding": [1.0]} for i in range(len(inputs))
        ]})

    respx.post(_endpoint()).mock(side_effect=_slow_then_fail)
    client = BailianLLMClient(api_key="sk-test", region="intl")

    with pytest.raises(AppException):
        await client.embed([f"t{i}" for i in range(120)])  # 12 batches

    # Whatever was already dispatched is finished or cancelled; nothing new may
    # start after the failure surfaced.
    settled = len(attempts)
    await asyncio.sleep(0.6)
    assert len(attempts) == settled, (
        f"{len(attempts) - settled} batches kept running after the failure"
    )
    assert settled < 12, "every batch ran even though the first one failed"


@pytest.mark.asyncio
@respx.mock
async def test_the_whole_call_deadline_is_a_ceiling_not_a_floor(monkeypatch) -> None:
    """Measured, not grepped.

    ``max(self._timeout, DEADLINE)`` made the documented ceiling a floor, so a
    caller passing ``timeout=120`` for its per-request bound silently bought a
    120-second whole-call budget. The first cut of this test asserted on
    ``inspect.getsource`` — the very anti-pattern the same change removed three
    of: it goes red on a reformat and green on a refactor that keeps the
    substring.
    """
    import asyncio
    import time

    from app.domains.bailian import llm_client

    async def _never_answers(request: httpx.Request) -> httpx.Response:
        await asyncio.sleep(30)
        return httpx.Response(200, json={"data": []})

    respx.post(_endpoint()).mock(side_effect=_never_answers)
    # Shortened so the test is quick; what matters is that the CONSTANT decides,
    # not the per-request timeout handed to the client.
    monkeypatch.setattr(llm_client, "_EMBED_TOTAL_DEADLINE_SECONDS", 0.4)

    client = BailianLLMClient(api_key="sk-test", region="intl", timeout=120.0)
    started = time.monotonic()
    with pytest.raises(AppException) as caught:
        await client.embed([f"t{i}" for i in range(30)])
    elapsed = time.monotonic() - started

    assert caught.value.code == "BAILIAN_EMBED_TIMEOUT"
    assert elapsed < 5, (
        f"waited {elapsed:.1f}s — the per-request timeout (120s) decided the "
        "whole-call budget instead of the deadline"
    )