"""Finding the flow an employee described, in the words they actually used.

Discovery failing is indistinguishable, from the employee's side, from the flow
not existing: production asked for a 「邮**箱**别名」, the catalogue title says
「邮**件**别名」, lexical search returned nothing, and the assistant filed a
support ticket for something the platform does properly.

A synonym table was the first patch and does not scale — 「软件坐席」/「license」/
「授权」/「licnese」 are all the same request, and no hand-maintained list
survives contact with how people type. So the ranking is semantic, reusing the
embedding path the knowledge base already runs on (DashScope via the org's own
key), with the lexical matcher kept as the fallback for when no embedding
backend is configured or the call fails.

Cache: the catalogue changes only on deploy, so its vectors are embedded once
per process and reused. A query costs one embedding call.
"""
from __future__ import annotations

import logging
import math
import time
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

logger = logging.getLogger(__name__)

#: Below this the match is noise. Same bar the KB semantic search uses, for the
#: same reason: a wrong flow offered confidently is worse than a ticket.
MIN_SCORE = 0.30

#: One embedding round-trip sits inside a user-facing turn, so it is bounded
#: tightly; the lexical fallback is always there when it lapses.
_EMBED_TIMEOUT = 8.0

#: Per-corpus embedding index, built lazily and kept for the process.
_vectors: dict[str, dict[str, list[float]]] = {}
#: Corpus → when its warm-up last failed. Remembered so a dead embedding
#: backend costs one attempt, not a 32-second stall on every later question —
#: but remembered with an EXPIRY, because the first shape never retried: one
#: DashScope blip, or a key configured five minutes after boot, and every
#: question for the rest of the process's life fell back to lexical matching.
#: Discovery quietly getting worse is indistinguishable, from the employee's
#: side, from the flow not existing.
_failed: dict[str, float] = {}
#: Long enough that a hard outage is not re-probed per question, short enough
#: that a fixed backend is picked up without a deploy.
_FAILURE_TTL_SECONDS = 600.0


def _cosine(a: list[float], b: list[float]) -> float:
    if not a or not b or len(a) != len(b):
        return 0.0
    dot = sum(x * y for x, y in zip(a, b, strict=False))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(y * y for y in b))
    if na == 0.0 or nb == 0.0:
        return 0.0
    return dot / (na * nb)


def _corpus_text(name: str, action: Any) -> str:
    """What an action is embedded as.

    Every localized title, not just the default one: the employee may type any
    of the three languages, and the tool name itself carries the English words
    a bilingual colleague is likely to reach for ("seat", "alias").
    """
    titles = " ".join(action.localized_title(lang) for lang in ("zh", "en", "ja"))
    return f"{name.replace('_', ' ')}\n{titles}\n{titles}"


#: The two things a turn searches: the flows it may OFFER, and the read tools
#: it may CALL. One implementation, because the reason semantic search exists
#: applies to both — an employee types 「软件目录」 and the corpus is English.
CORPUS_ACTIONS = "actions"
CORPUS_READONLY = "readonly"

#: How much of a tool's description is embedded. It used to be 400 characters,
#: which was not a budget so much as an accident: 116 of the 273 read-only tool
#: descriptions are longer than that, and what sits at the end of a description
#: is exactly the part written to be found — ``whoami`` closes with 「我的工号/
#: GitHub/办公地是什么」 at character 420, so the one tool that answers 「我的
#: 工号是多少」 was embedded without ever mentioning 工号 and ranked below three
#: admin tools about employee-number ALLOCATION. Embedding is a one-off per
#: process; the cap is only here so a pathological description cannot dominate
#: its own vector.
_CORPUS_DESCRIPTION_CHARS = 2000


def _corpus(kind: str) -> dict[str, str]:
    """Name → the text that describes it, for one corpus."""
    if kind == CORPUS_ACTIONS:
        from app.assistant.derived_actions import derived_catalogue  # noqa: PLC0415

        return {n: _corpus_text(n, a) for n, a in derived_catalogue().items()}

    from app.assistant.mcp_bridge import readonly_catalogue  # noqa: PLC0415

    return {
        name: (
            f"{name.replace('_', ' ')}. "
            f"{(tool.description or '')[:_CORPUS_DESCRIPTION_CHARS]}"
        )
        for name, tool in readonly_catalogue().items()
    }


async def _ensure_vectors(db: AsyncSession, kind: str) -> bool:
    """Embed one corpus once per process. False when unavailable.

    A failure is remembered. The first shape left ``_vectors`` at ``None`` on
    failure, so every later query re-attempted a whole-catalogue embed with a
    32-second timeout **inside a user-facing turn** — one unreachable backend
    turned every question into a stall.
    """
    if kind in _vectors:
        return True
    failed_at = _failed.get(kind)
    if failed_at is not None:
        if time.monotonic() - failed_at < _FAILURE_TTL_SECONDS:
            return False
        # The cool-off lapsed: try again rather than degrade forever.
        del _failed[kind]

    from app.domains.support.copilot import embed_texts  # noqa: PLC0415

    corpus = _corpus(kind)
    names = list(corpus)
    if not names:
        _failed[kind] = time.monotonic()
        return False
    embedded = await embed_texts(
        db,
        [corpus[n] for n in names],
        timeout=_EMBED_TIMEOUT * 4,  # a one-off warm-up, not a per-query cost
    )
    if embedded is None or len(embedded[0]) != len(names):
        # Logged in BOTH cases. It used to be silent when ``embedded`` was None,
        # and that silence is why the whole semantic layer could be dead in
        # production for a day without anyone noticing: DashScope caps an
        # embeddings request at 10 inputs, this warm-up sent 273, every call was
        # a 400, and the only visible symptom was search quietly getting worse.
        # "Degraded to keyword matching" is a fact an operator has to be able to
        # find in a log.
        logger.warning(
            "assistant: %s corpus (%d entries) could not be embedded — "
            "semantic search is DEGRADED to lexical matching%s",
            kind, len(names),
            "" if embedded is None else " (vector count mismatch)",
        )
        _failed[kind] = time.monotonic()
        return False
    vectors, _model = embedded
    _vectors[kind] = dict(zip(names, vectors, strict=False))
    logger.info("assistant: embedded %d %s entries for search", len(names), kind)
    return True


async def semantic_search(
    db: AsyncSession, query: str, *, kind: str = CORPUS_ACTIONS, limit: int = 6,
) -> list[str] | None:
    """Action names ranked by meaning, or None when embeddings are unavailable.

    None (not an empty list) is the "fall back to lexical" signal — an empty
    list means the corpus genuinely had nothing close enough.
    """
    text = (query or "").strip()
    if not text:
        return None
    try:
        if not await _ensure_vectors(db, kind):
            return None
        from app.domains.support.copilot import embed_texts  # noqa: PLC0415

        embedded = await embed_texts(db, [text], timeout=_EMBED_TIMEOUT)
        if embedded is None or not embedded[0]:
            return None
        probe = embedded[0][0]
    except Exception as exc:  # noqa: BLE001 — discovery must never break a turn
        logger.warning("assistant: semantic action search failed: %s", exc)
        return None

    scored = [
        (_cosine(probe, vector), name)
        for name, vector in _vectors.get(kind, {}).items()
    ]
    scored.sort(key=lambda pair: (-pair[0], pair[1]))
    return [name for score, name in scored[:limit] if score >= MIN_SCORE]


def reset_cache() -> None:
    """Drop the embedded catalogue — for tests, and for a settings change that
    swaps the embedding model underneath us."""
    _vectors.clear()
    _failed.clear()