"""Synthetic coordinates/evidence test fusion semantics, not domain/model quality."""

import math
from dataclasses import replace
from uuid import UUID, uuid4

import pytest
from sqlalchemy import func, select, text
from sqlalchemy.orm import Session
from test_indexing import build
from test_retrieval import add_item
from test_vectors import ToyProvider, text_version, vector_build, vector_publish

from app.models.retrieval import RetrievalRun
from app.schemas.hybrid import RankedCandidate
from app.services import vectors
from app.services.embedding import EmbeddingError
from app.services.embedding_runtime import get_embedding_provider
from app.services.hybrid import fuse
from app.services.indexing import publish_index


@pytest.fixture
def provider(client):
    value = ToyProvider()
    client.app.dependency_overrides[get_embedding_provider] = lambda: value
    return value


def search(client, path="hybrid", **kwargs):
    return client.post(f"/api/retrieval/{path}", json={"query": "query", **kwargs})


def publish_vectors(client):
    text_ = text_version(client)
    version = vector_build(client, text_)
    assert vector_publish(client, version).status_code == 200
    return text_, version


def test_rrf_hand_calculation_absent_leg_and_raw_score_independence():
    a, b, c = uuid4(), uuid4(), uuid4()
    lexical = [
        RankedCandidate(chunk_id=a, rank=1, score=1000),
        RankedCandidate(chunk_id=b, rank=2, score=1),
    ]
    vector = [
        RankedCandidate(chunk_id=b, rank=1, score=-0.9),
        RankedCandidate(chunk_id=c, rank=2, score=-1),
    ]
    scores = fuse(lexical, vector)
    assert scores == {str(a): 1 / 61, str(b): 1 / 62 + 1 / 61, str(c): 1 / 62}
    for candidate in lexical + vector:
        candidate.score *= 10000
    assert fuse(lexical, vector) == scores
    assert fuse([], []) == {}


def test_same_chunk_scope_standalone_rankings_formula_and_history(client, provider):
    items = [
        add_item(client, content="alpha query extra words words"),
        add_item(client, content="query"),
        add_item(client, content="beta"),
    ]
    add_item(client, status="pending", content="query alpha")
    text_, version = publish_vectors(client)
    response = search(client, top_k=3, candidate_k=3)
    assert response.status_code == 200, response.text
    result = response.json()
    lexical = search(client, "chunks", top_k=3).json()
    vector = search(client, "vector", top_k=3).json()
    assert result["corpus_size"] == lexical["corpus_size"] == vector["corpus_size"] == 3
    assert (
        result["text_index_version_id"]
        == lexical["text_index_version_id"]
        == vector["text_index_version_id"]
        == text_["id"]
    )
    assert result["vector_version_id"] == version["id"]
    assert result["config"]["space"]["id"] == provider.spec.space_id
    assert result["config"]["lexical"]["title_weight"] == 0
    assert result["config"]["lexical"]["k1"] == 1.2
    assert result["config"]["lexical"]["b"] == 0.75
    assert result["config"]["chunking"] == text_["config"]
    assert result["lexical_total_hits"] == 2 and result["vector_total_hits"] == 3
    assert [c["chunk_id"] for c in result["lexical_candidates"]] == [
        c["chunk_id"] for c in lexical["results"]
    ]
    assert [c["chunk_id"] for c in result["vector_candidates"]] == [
        c["chunk_id"] for c in vector["results"]
    ]
    # Independently calculate positive-log-IDF BM25 on bodies of lengths 5, 1, 1.
    expected = math.log(1 + 1.5 / 2.5) * 2.2 / (1 + 1.2 * (0.25 + 0.75 / (7 / 3)))
    assert lexical["results"][0]["knowledge_id"] == items[1]["id"]
    assert lexical["results"][0]["score"] == pytest.approx(expected)
    for hit in result["results"]:
        left = next(
            (c for c in result["lexical_candidates"] if c["chunk_id"] == hit["chunk_id"]), None
        )
        right = next(
            (c for c in result["vector_candidates"] if c["chunk_id"] == hit["chunk_id"]), None
        )
        assert hit["score"] == pytest.approx(
            (1 / (60 + left["rank"]) if left else 0) + (1 / (60 + right["rank"]) if right else 0)
        )
        assert hit["score"] == hit["lexical_contribution"] + hit["vector_contribution"]
        assert hit["source_url"] == "https://example.com/"
    assert search(client, candidate_k=3, top_k=3).json()["results"] == result["results"]
    for value in (result, lexical, vector):
        assert client.get(f"/api/retrieval/runs/{value['run_id']}").json() == value
    old = client.post("/api/retrieval", json={"query": "query"}).json()
    assert old["mode"] == "keyword" and "chunk_id" not in old["results"][0]
    history = client.get("/api/retrieval/runs?page_size=100").json()
    assert {r["mode"] for r in history["items"]} == {
        "keyword",
        "vector",
        "keyword_chunks",
        "hybrid",
    }
    client.cookies.clear()
    assert search(client).status_code == search(client, "chunks").status_code == 200
    assert client.get("/api/retrieval/runs").status_code == 401


def test_candidate_cutoffs_disjoint_legs_and_deterministic_tie(client, provider):
    first = add_item(client, content="alpha")
    second = add_item(client, content="query")
    publish_vectors(client)
    small = search(client, candidate_k=1, top_k=1).json()
    assert small["total_hits"] == 2 and len(small["results"]) == 1
    assert small["results"][0]["knowledge_id"] == first["id"]
    assert small["results"][0]["lexical_rank"] is None
    assert small["results"][0]["lexical_score"] is None
    assert small["results"][0]["lexical_contribution"] == 0
    assert small["results"][0]["score"] == 1 / 61
    full = search(client, candidate_k=2, top_k=2).json()
    assert full["results"][0]["knowledge_id"] == second["id"]
    assert full["results"][0]["lexical_rank"] == 1 and full["results"][0]["vector_rank"] == 2


def test_chunk_bodies_offsets_overlap_and_category_apply_before_both_rankings(client, provider):
    item = add_item(client, title="uniquetitle", content="alpha query 文本。" * 30, category="交通")
    add_item(client, content="query", category="其他")
    publish_vectors(client)
    result = search(client, category="交通", top_k=20, candidate_k=20).json()
    assert result["corpus_size"] > 1 and result["total_hits"] == result["corpus_size"]
    assert len({h["chunk_id"] for h in result["results"]}) == len(result["results"])
    for hit in result["results"]:
        assert hit["knowledge_id"] == item["id"]
        assert hit["snippet"] == item["content"][hit["snippet_start"] : hit["snippet_end"]]
    assert search(client, "chunks", query="uniquetitle").json()["status"] == "no_evidence"
    empty = search(client, category="不存在分类").json()
    assert empty["status"] == "no_evidence" and empty["results"] == []
    assert empty["lexical_candidates"] == empty["vector_candidates"] == []


def test_empty_lexical_leg_and_empty_corpus_are_explicit(client, provider):
    add_item(client, content="alpha")
    publish_vectors(client)
    for query in ["unmatchedword", "！！"]:
        result = search(client, query=query).json()
        assert result["lexical_candidates"] == [] and result["status"] == "ok"
        assert result["results"][0]["score"] == 1 / 61
    assert search(client, "chunks", query="！！").status_code == 422


def test_empty_published_corpus(client, provider):
    publish_vectors(client)
    for path in ["chunks", "hybrid"]:
        result = search(client, path).json()
        assert result["status"] == "no_evidence" and result["total_hits"] == 0


def test_keyword_chunks_need_no_runtime_or_vector_publication(client):
    add_item(client, content="query")
    assert search(client, "chunks").status_code == 409
    text_version(client)
    assert search(client, "chunks").status_code == 200
    assert search(client).status_code == 503


@pytest.mark.parametrize(
    "payload",
    [
        {"candidate_k": 1, "top_k": 2},
        {"candidate_k": 201},
        {"candidate_k": 0},
        {"top_k": 21},
        {"top_k": 0},
        {"mode": "vector"},
        {"rrf_k": 10},
        {"query": " "},
        {"query": "x" * 501},
    ],
)
def test_request_budget_and_fixed_config_validation(client, provider, payload):
    assert search(client, **payload).status_code == 422


@pytest.mark.parametrize("change", ["catalog", "text", "vector"])
def test_query_encoding_releases_locks_and_rechecks_publication(
    client, provider, migrated_database, change
):
    item = add_item(client, content="alpha query")
    text_, first = publish_vectors(client)
    replacement_text = build(client)
    replacement_vector = vector_build(client, text_)

    def mutate(query):
        assert query
        with Session(migrated_database) as other:
            other.execute(text("SET LOCAL lock_timeout='1s'"))
            if change == "catalog":
                state = vectors.locked_state(other)
                state.catalog_revision += 1
                other.execute(
                    text(
                        "UPDATE knowledge_items SET content='changed synthetic evidence' WHERE id=:id"
                    ),
                    {"id": item["id"]},
                )
                other.commit()
            elif change == "text":
                publish_index(other, UUID(replacement_text["id"]), 1, first["created_by"])
            else:
                vectors.publish_vectors(
                    other, UUID(replacement_vector["id"]), 1, first["created_by"]
                )

    provider.callback = mutate
    assert search(client).status_code == 409
    with Session(migrated_database) as db:
        assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0


@pytest.mark.parametrize("change", ["body", "chunks", "vectors", "space"])
def test_corrupt_or_incompatible_snapshot_fails_closed(client, provider, migrated_database, change):
    add_item(client, content="alpha query")
    publish_vectors(client)
    if change == "space":
        provider.spec = replace(provider.spec, revision="b" * 40)
    else:
        with migrated_database.begin() as connection:
            connection.execute(
                text(
                    {
                        "body": "UPDATE knowledge_items SET content='out of band change'",
                        "chunks": "UPDATE chunks SET text='tampered'",
                        "vectors": "DELETE FROM chunk_vectors",
                    }[change]
                )
            )
    assert search(client).status_code == 409
    with Session(migrated_database) as db:
        assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0


@pytest.mark.parametrize("failure,status", [("token", 422), ("runtime", 503)])
def test_model_failure_has_no_fallback_or_partial_history(
    client, provider, migrated_database, failure, status
):
    add_item(client, content="alpha query")
    publish_vectors(client)

    def fail(*args, **kwargs):
        if failure == "token":
            raise EmbeddingError("token budget")
        raise OSError("private-runtime-path")

    provider.encode = fail
    response = search(client)
    assert response.status_code == status and "private-runtime-path" not in response.text
    with Session(migrated_database) as db:
        assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0


def test_failed_history_insert_rolls_back_both_leg_result(client, provider, migrated_database):
    add_item(client, content="alpha query")
    publish_vectors(client)
    with migrated_database.begin() as connection:
        connection.execute(
            text("ALTER TABLE retrieval_runs ADD CONSTRAINT test_no_runs CHECK (false) NOT VALID")
        )
    assert search(client).status_code == 409
    with Session(migrated_database) as db:
        assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0


def test_stale_history_survives_evidence_change_and_capacity_is_explicit(
    client, provider, migrated_database, monkeypatch
):
    item = add_item(client, content="alpha query")
    publish_vectors(client)
    result = search(client).json()
    monkeypatch.setattr(vectors, "MAX_VECTOR_CHUNKS", 0)
    assert search(client).status_code == search(client, "chunks").status_code == 503
    monkeypatch.undo()
    assert (
        client.patch(
            f"/api/knowledge/{item['id']}",
            json={"expected_version": item["version"], "content": "updated synthetic body"},
        ).status_code
        == 200
    )
    assert search(client).status_code == search(client, "chunks").status_code == 409
    assert client.get(f"/api/retrieval/runs/{result['run_id']}").json() == result