"""Let the assistant reach the platform's read-only MCP tools.

**Why this is the right surface.** The platform already has a rule that every
new self-service flow ships with its MCP tool in the same PR, so the MCP
registry tracks the product automatically. A hand-maintained chat tool list
would spend the rest of its life catching up with that registry; borrowing it
means the assistant learns new capabilities the day they ship, at zero
maintenance cost.

**Why it is safe to borrow.** What the old chat/MCP isolation actually
protected was two things, and neither depends on where a tool comes from:

* *Privilege* — enforced inside each tool by the domain's own RBAC, against the
  caller the server resolved. Borrowing a tool cannot widen what its caller may
  see, and a caller reaching for an admin tool gets that domain's own refusal.
* *Unintended writes* — enforced here, before dispatch, by a hard filter: only
  tools the platform's own ``mutates=`` ground truth marks read-only are
  reachable, and :data:`app.mcp.annotations.DESTRUCTIVE_TOOLS` is excluded
  outright. There is no argument, prompt or tool name that reaches a write.

Under that filter the worst outcome of prompt injection is a wrong answer built
from data the employee could already read — not an action, and not someone
else's data.

**Why two stages.** 400+ tool schemas do not fit in a prompt, and a model asked
to choose among them chooses badly. The assistant instead gets one search tool
and one call tool: find the handful that match the question, then call one.
"""
from __future__ import annotations

import json
import logging
import uuid
from dataclasses import dataclass
from functools import lru_cache
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

TOOL_SEARCH = "search_helix_tools"
TOOL_CALL = "call_helix_tool"

#: The assistant's own audit reads. Authority-consistent — only an admin or an
#: auditor can call them at all — but pulling a colleague's verbatim transcript
#: into a model prompt is the opposite of why that surface was narrowed. Kept
#: off the chat bridge for the same reason the confirm path is.
_SELF_AUDIT_TOOLS = frozenset({
    "assistant_list_conversations",
    "assistant_get_conversation",
    "assistant_usage_stats",
    # Returns other employees' verbatim questions and answers. Shipped in the
    # same change as the gap queue and missed here, which is exactly the drift
    # this set exists to prevent — an admin asking the bot 「哪些问题没解决」
    # would have sent colleagues' free text to the model provider.
    "assistant_unhelpful_answers",
    # Same class, missed the same way one release later (#3409 / #3411).
    # ``list_assistant_eval_failures`` returns, by its own description, "the
    # question, the assistant's reply, the documents it actually read" for an
    # attempt run under a PERSONA employee's identity — caller-scoped data about
    # someone who is not the reader. ``get_assistant_eval_run`` is blocked for a
    # DIFFERENT reason worth stating: its own serialiser is pure aggregates, but
    # the run view carries the comparison against the previous run, and that
    # names each regressed case's QUESTION — text mined from real employees.
    "list_assistant_eval_failures",
    "get_assistant_eval_run",
    # The knowledge-loop queue is worse than it looks: ``evidence`` on a
    # backfill proposal is ``{"employee_question": <a colleague's exact words>}``
    # and, unlike the evaluation miner, the lint path does NOT redact it.
    "list_knowledge_proposals",
    "get_knowledge_proposal",
})


#: How many candidates a search returns. Enough to cover a vague question,
#: few enough that the model still has to choose rather than call them all.
_SEARCH_LIMIT = 6
#: How much of a candidate's description the model gets to read. The tail of a
#: description is where the caveats live — which role it needs, which sibling
#: tool to use instead, what a filter actually means — and a model that never
#: sees them calls the wrong tool with confident arguments.
_SEARCH_DESCRIPTION_CHARS = 900
#: Cap on a single tool result handed back to the model. A large listing would
#: otherwise blow the context and push the employee's own question out of it.
_RESULT_CHARS = 6000
#: How much of an over-long result is still worth showing, leaving room for the
#: envelope that says it is partial.
_TRUNCATED_PAYLOAD_CHARS = 5000


@dataclass(frozen=True)
class BridgedTool:
    name: str
    description: str
    schema: dict[str, Any]
    haystack: str


@lru_cache(maxsize=1)
def readonly_catalogue() -> dict[str, BridgedTool]:
    """Every MCP tool the assistant may call, keyed by name.

    Derived from the platform's own ground truth rather than a second list:
    ``mutates=`` on each tool body drives the MCP read-only hint already, so a
    tool that starts mutating leaves this catalogue on the same commit that
    changes its flag.
    """
    from app.mcp import tools as _tools  # noqa: F401, PLC0415 — force registration
    from app.mcp.annotations import DESTRUCTIVE_TOOLS, mutating_tools_from_source
    from app.mcp.server import mcp_server  # noqa: PLC0415

    flags = mutating_tools_from_source()
    catalogue: dict[str, BridgedTool] = {}
    for name, tool in mcp_server._tool_manager._tools.items():  # noqa: SLF001
        if flags.get(name, True):
            continue  # unknown or mutating → not reachable
        if name in DESTRUCTIVE_TOOLS:
            continue  # belt and braces: never reachable even if flagged read-only
        if name in _SELF_AUDIT_TOOLS:
            continue
        description = (getattr(tool, "description", "") or "").strip()
        schema = getattr(tool, "parameters", None) or {}
        catalogue[name] = BridgedTool(
            name=name,
            description=description,
            schema=schema,
            haystack=f"{name} {description}".lower(),
        )
    logger.info("assistant: %d read-only MCP tools bridged", len(catalogue))
    return catalogue


def is_callable_tool(name: str) -> bool:
    return name in readonly_catalogue()


def _score(tool: BridgedTool, terms: list[str]) -> int:
    score = 0
    for term in terms:
        if term in tool.name:
            score += 3
        if term in tool.haystack:
            score += 1
    return score


async def search(
    db: AsyncSession, query: str, *, limit: int = _SEARCH_LIMIT,
) -> list[BridgedTool]:
    """Rank read-only tools against a free-text query, by meaning first.

    It used to be lexical only, on the argument that tool names here are
    descriptive enough (``get_my_assets``, ``list_my_software_seats``) for term
    overlap to work. That argument silently assumed the query is English. Every
    name and description in this registry is English; the employee is not.
    Production, 2026-08-18: asked 「我要申请 OPEN AI 的坐席」 — a product that
    IS in the software catalogue, with a working 申请坐席 button on the web —
    the model searched 「软件目录」, matched **nothing**, and told the employee
    OpenAI is not in Helix. 「OpenAI」 did match, but only the API-key tools, so
    it helpfully offered an API key instead. The employee saw a confident wrong
    answer about their own platform.

    CJK bigrams cannot fix this the way they fixed the action catalogue: there
    the titles are localized, so a Chinese query meets Chinese text. Here there
    is no Chinese to meet. Embeddings are the only thing that crosses that gap
    without a hand-maintained zh→en term list, which would rot against a
    registry that changes weekly — this module's own stated reason for
    preferring derivation over lists.

    Lexical stays as the fallback for when no embedding backend is configured.
    """
    from app.assistant import action_search  # noqa: PLC0415 — cycle

    ranked_names = await action_search.semantic_search(
        db, query, kind=action_search.CORPUS_READONLY, limit=limit,
    )
    if ranked_names is not None:
        catalogue = readonly_catalogue()
        found = [catalogue[n] for n in ranked_names if n in catalogue]
        if found:
            return found
        # Nothing close enough by meaning — try the words themselves before
        # telling the model the platform cannot do it.
    return _lexical_search(query, limit=limit)


def _lexical_search(query: str, *, limit: int = _SEARCH_LIMIT) -> list[BridgedTool]:
    """Term overlap. Works for an English query; the net under the above."""
    terms = [t for t in query.lower().replace(",", " ").split() if len(t) > 1]
    if not terms:
        return []
    ranked = [
        (score, tool)
        for tool in readonly_catalogue().values()
        if (score := _score(tool, terms)) > 0
    ]
    ranked.sort(key=lambda pair: (-pair[0], pair[1].name))
    return [tool for _score_value, tool in ranked[:limit]]


def bridge_tools() -> list[dict[str, Any]]:
    """The two meta-tools the model actually sees."""
    return [
        {
            "name": TOOL_SEARCH,
            "description": (
                "Search Helix's read-only data tools by keyword when the "
                "employee asks about something your built-in tools do not "
                "cover — devices, accounts, software seats, expenses, leave, "
                "onboarding, cloud resources, costs. Returns tool names with "
                "their parameters. Search first, then call one with "
                f"{TOOL_CALL}."
            ),
            "input_schema": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Keywords, e.g. 'laptop warranty' or 'software seats'.",
                    },
                },
                "required": ["query"],
                "additionalProperties": False,
            },
        },
        {
            "name": TOOL_CALL,
            "description": (
                "Call one read-only Helix tool found via "
                f"{TOOL_SEARCH}. Runs as the employee you are talking to, so it "
                "returns only what they are allowed to see. Read-only: it "
                "cannot change anything."
            ),
            "input_schema": {
                "type": "object",
                "properties": {
                    "name": {"type": "string", "description": "Exact tool name."},
                    "arguments": {
                        "type": "object",
                        "description": "Arguments matching that tool's parameters.",
                        "additionalProperties": True,
                    },
                },
                "required": ["name"],
                "additionalProperties": False,
            },
        },
    ]


def make_dispatch(*, person: IdentityPerson, called: list[str] | None = None):
    """Dispatcher for the two bridge tools, bound to one resolved person.

    ``called`` collects the REAL tool names invoked. Without it the transcript
    and the admin filters would only ever show ``call_helix_tool``, which tells
    an auditor nothing about what the answer was actually built from.

    The identity comes from this closure — set from the signed Feishu envelope —
    and is pushed into the MCP context var the tool runtime reads. A tool
    therefore runs as the employee who sent the message, whatever the model
    puts in its arguments.
    """

    async def _dispatch(
        name: str,
        tool_input: dict[str, Any],
        db: AsyncSession,
        _person: IdentityPerson,
    ) -> str:
        if name == TOOL_SEARCH:
            query = str((tool_input or {}).get("query") or "")[:200]
            found = await search(db, query)
            if not found:
                # NOT "answer from what you know" — that is what this note used
                # to say, and it is the wrong instruction for an IT assistant:
                # the questions that reach here are 「打印机怎么连」-shaped, whose
                # real answer is in the company's own handbook, not in the
                # model's memory of how companies generally work.
                return json.dumps(
                    {
                        "tools": [],
                        "note": (
                            "No data tool matches. If this is a how-to or "
                            "troubleshooting question, search the company "
                            "knowledge base with search_knowledge_base — it "
                            "holds the real IT handbook. Only if that is also "
                            "empty, say Helix does not have it."
                        ),
                    },
                    ensure_ascii=False,
                )
            return json.dumps(
                {
                    "tools": [
                        {
                            "name": t.name,
                            "description": t.description[:_SEARCH_DESCRIPTION_CHARS],
                            "parameters": t.schema,
                        }
                        for t in found
                    ],
                },
                ensure_ascii=False,
            )

        if name == TOOL_CALL:
            requested = str((tool_input or {}).get("name") or "")
            if called is not None and requested:
                called.append(requested)
            return await _call_tool(tool_input or {}, person=person)

        return json.dumps({"error": f"unknown tool {name}"}, ensure_ascii=False)

    return _dispatch


async def _audit_refusal(requested: str, *, person: IdentityPerson) -> None:
    """Record one refused tool call on its own session.

    Independent session and immediate commit, matching the platform's
    denied-action pattern: the caller's transaction may roll back, and a
    refusal that disappears with it is a refusal nobody can audit.
    """
    from app.audit.service import AuditService  # noqa: PLC0415
    from app.core.database import async_session_factory  # noqa: PLC0415

    try:
        async with async_session_factory() as db:
            await AuditService(db).log(
                trace_id="",
                actor_type="user",
                actor_user_id=person.id,
                action="assistant.tool.refused",
                resource_type="assistant_tool",
                resource_id=requested[:128] or None,
                status="denied",
                detail={"tool": requested[:200], "reason": "not_read_only"},
            )
            await db.commit()
    except Exception as exc:  # noqa: BLE001 — auditing must not break the reply
        logger.warning("assistant: refusal audit failed: %s", exc)


#: 纯翻页参数 —— 它们不收窄「查的是什么」,只决定「看第几段」。翻页不算收窄:
#: 那正是 2026-09-07 那次失败的形状(三次调用、只在翻页、一次没给 search)。
_PAGINATION_ONLY_ARGS = frozenset({"page", "page_size", "limit", "offset", "cursor"})


def _has_narrowing_argument(arguments: dict[str, Any]) -> bool:
    """调用方有没有告诉工具「我要找的是什么」。

    判据刻意宽松:任何一个非翻页、非空的参数都算收窄。目的不是审查参数质量,
    而是把「什么都没说就要全量」这一种调用拦下来 —— 它是唯一会让模型拿着一页
    残缺清单去断言「不存在」的形状。
    """
    for key, value in (arguments or {}).items():
        if key in _PAGINATION_ONLY_ARGS:
            continue
        if value is None or value is False or value == "":
            continue
        return True
    return False


def _accepts_narrowing(schema: dict[str, Any]) -> bool:
    """这个工具**有没有**一个可以用来收窄的入参。

    只有翻页参数(或干脆没有参数)的工具答不了「带上 search 重试」这条指令 ——
    它的 schema 里根本没有那个键,模型照做只会换来一条 ``invalid_arguments``,
    第二次仍然是零行。那时拒绝就不是「拦住一个错结论」,而是**把员工问自己的
    东西问成了死胡同**:既没有数据,又被明令禁止说「没有」。

    这不是个别情况:2026-09-08 实测只读目录 360 个工具里有 118 个是这个形状,
    包括 ``list_my_assets`` / ``get_my_it_footprint`` / ``list_my_software_seats``
    这类「我自己的东西」类工具 —— 它们本来就按调用者收窄,没有第二个维度可给。

    对这些工具,#3512 那条推理的前提反过来成立:模型已经尽到了它能尽的事
    (没有参数可传),超限是数据本身太大,而不是提问方式不对 —— 所以走
    「部分结果 + 警告」,和给了收窄条件的调用同一条路。
    """
    properties = schema.get("properties")
    if not isinstance(properties, dict):
        # schema 读不出来时按「可收窄」处理 —— 2026-09-07 那个错答案来自
        # 「给了半份清单」那条路,拿不准的时候宁可少给数据。
        return True
    return any(key not in _PAGINATION_ONLY_ARGS for key in properties)


def _retry_advice(schema: dict[str, Any]) -> str:
    """截断后该告诉模型「怎么再来一次」—— 只说这个工具真的做得到的那件事。

    给一条工具答不了的建议,和上面那条死胡同拒绝是同一个缺陷:模型照做,换来
    ``invalid_arguments``,第二次仍然没有数据。所以三种 schema 给三句话。
    """
    if _accepts_narrowing(schema):
        return (
            "Call the tool again with a narrower argument (a search term, a "
            "filter, a smaller page_size, or a later page)."
        )
    properties = schema.get("properties")
    if isinstance(properties, dict) and properties:
        return (
            "This tool takes no filter — page through the rest with "
            "page / page_size to see the remaining rows."
        )
    # 一个入参都没有:既收窄不了也翻不了页。这时候唯一诚实的话是「剩下的你
    # 看不到」—— 不能让模型把「没看到」说成「没有」。
    return (
        "This tool takes no arguments at all, so the missing rows cannot be "
        "retrieved — treat them as unknown, never as absent."
    )


def oversized_refusal() -> dict[str, Any]:
    """超限、工具**能**被收窄、而调用方一个收窄条件都没给时返回的载荷。

    它不含任何数据行。

    为什么是「拿不到行」而不是「给一半 + 警告」:那条警告一直都在,措辞也很直白
    (「Do NOT tell the employee something does not exist based on this call」),
    而 2026-09-07 的生产实测是**模型照样无视它** —— 员工问「380838443567 这个
    云账号谁负责」,模型三次调 ``list_cloud_scopes``、一次没传 ``search``,然后
    回答「在当前可查询的云账号列表中,没有找到」。那个账号在库里、有负责人、
    状态 active,只是按名字排在第 197 位而工具一页给 200 行、共 271 行。

    所以改成结构性的:**模型无法从一份不存在的清单里挑出一个错结论。**
    """
    return {
        "error": "too_many_rows",
        "shown_rows": 0,
        "hint": (
            "这次调用没有任何收窄条件,返回的数据超出可读上限,"
            "所以**一行都没有返回**。请带上收窄参数重试 —— 通常是 "
            "`search=<名称或编号>`(多数列表工具的 search 同时匹配"
            "名称和外部编号,例如云账号号码)。\n\n"
            "⚠️ 你现在**没有**看到任何数据,因此绝对不能据此回答"
            "「没有找到」「不存在」「查不到」—— 那会是一个凭空捏造的"
            "否定结论。先用 search 精确查。"
        ),
    }


async def call_readonly_tool(
    name: str, arguments: dict[str, Any], *, person: IdentityPerson,
) -> str:
    """跑一个只读工具、以 ``person`` 的身份 —— 供**服务端自己**做预检索用。

    存在的理由是 RBAC 不该有第二套实现:预检索(``identifier_grounding``)要
    的是「这个编号,在提问者的可见范围内是什么」,而每个工具体内已经写着那道
    门(``can_read_cloud`` / 只看自己拥有或参与的 scope / 资产域的 office 范围)。
    绕过工具直接写 SQL 就等于把这些谓词抄第二遍 —— 抄错一次就是越权读。

    返回值是**工具原样的 JSON 字符串**(含超限 / 拒绝时的那些结构),和模型走
    ``call_helix_tool`` 拿到的完全一致。
    """
    return await _call_tool({"name": name, "arguments": arguments}, person=person)


async def _call_tool(tool_input: dict[str, Any], *, person: IdentityPerson) -> str:
    """Run one read-only tool as ``person``.

    The allowlist check happens before anything else: a tool that is not in the
    read-only catalogue is refused by name, so a mutating or destructive tool
    is unreachable regardless of how it was asked for.
    """
    from pydantic import ValidationError  # noqa: PLC0415

    from app.mcp.context import mcp_trace_id, mcp_user_id  # noqa: PLC0415
    from app.mcp.server import mcp_server  # noqa: PLC0415

    requested = str(tool_input.get("name") or "")
    if not is_callable_tool(requested):
        logger.info(
            "assistant: refused non-readonly tool %r for person=%s",
            requested, person.id,
        )
        # Audited, not merely logged: a refusal is the signal that something
        # reached past the read-only boundary, and that is precisely what an
        # administrator should be able to notice and count.
        await _audit_refusal(requested, person=person)
        return json.dumps(
            {
                "error": "tool_not_available",
                "detail": (
                    "That tool is not available here. Only read-only tools are, "
                    f"and only ones returned by {TOOL_SEARCH}."
                ),
            },
            ensure_ascii=False,
        )

    arguments = tool_input.get("arguments")
    if not isinstance(arguments, dict):
        arguments = {}

    tool = mcp_server._tool_manager._tools.get(requested)  # noqa: SLF001
    if tool is None:  # pragma: no cover — catalogue is built from this registry
        return json.dumps({"error": "tool_not_available"}, ensure_ascii=False)

    # The tool runtime reads the caller from this context var, and everything
    # it enforces — RBAC, audit, rate-limit class, the leaver write gate —
    # hangs off that. Set it to the person the SERVER resolved, never to
    # anything the model produced.
    # ``tool.run()`` rather than ``tool.fn(**arguments)``: FastMCP validates and
    # coerces arguments against the tool's own schema on that path. Calling the
    # raw function would hand model-supplied values straight to a service body
    # untyped, and would leak the Python signature back through TypeError.
    user_token = mcp_user_id.set(person.id)
    # The turn's trace id too, so a bridged call is findable from the transcript
    # the administrator is reading — otherwise the runtime mints a fresh random
    # one and the two records can never be joined.
    trace_token = mcp_trace_id.set(mcp_trace_id.get() or uuid.uuid4())
    try:
        result = await tool.run(arguments)
    except Exception as exc:  # noqa: BLE001 — a tool error is data, not a 500
        # FastMCP wraps BOTH schema rejection and a failure inside the tool body
        # in ``ToolError``, so the cause decides the label. Calling a permission
        # refusal "invalid_arguments" would send the model off rewriting
        # arguments that were already correct.
        # A schema rejection arrives as ValidationError; an unknown or missing
        # keyword arrives as TypeError from the call itself. Both mean "fix the
        # arguments". A TypeError raised deeper inside a tool body would be
        # mislabelled by this — an acceptable trade, since the model's only
        # response either way is to retry differently or give up.
        cause = exc.__cause__
        kind = (
            "invalid_arguments"
            if isinstance(cause, (ValidationError, TypeError))
            else "tool_failed"
        )
        if kind == "tool_failed":
            logger.warning(
                "assistant: tool %s failed for person=%s: %s",
                requested, person.id, exc,
            )
        return json.dumps(
            {"error": kind, "detail": str(exc)[:300]}, ensure_ascii=False,
        )
    finally:
        mcp_trace_id.reset(trace_token)
        mcp_user_id.reset(user_token)

    blob = json.dumps(result, ensure_ascii=False, default=str)
    if len(blob) <= _RESULT_CHARS:
        return blob

    # 太大而且**调用方一个收窄参数都没给** —— 这种情况不返回行,只返回一条指令。
    #
    # 为什么不能只是「警告 + 给一半」:那条警告已经存在,措辞也已经很直白
    # (「Do NOT tell the employee something does not exist based on this
    # call」),而 2026-09-07 的生产实测是**模型照样无视它**:员工问
    # 「380838443567 这个云账号谁负责」,模型三次调 ``list_cloud_scopes``、
    # 一次都没传 ``search``,然后回答「在当前可查询的云账号列表中,没有找到」。
    # 那个账号在库里、有负责人、状态 active —— 它只是按 display_name 排在第
    # 197 位,而工具一页给 200 行、总共 271 行。
    #
    # 所以这里改成「拿不到行」:模型无法从一份不存在的清单里挑出一个错结论。
    # 给了收窄参数的调用仍然走下面的「部分结果 + 警告」—— 那时模型已经尽到了
    # 它该做的事,截断是数据本身太大,而不是提问方式不对。
    #
    # 而「尽到了它该做的事」还有第二种形状,#3512 漏掉的那种:**工具压根没有
    # 可收窄的入参**。只有 page/page_size 的工具答不了「带上 search 重试」,
    # 拒绝对它就只是一条死胡同 —— 零行 + 不准说「没有」。这类工具占只读目录
    # 118/360(2026-09-08 实测),``list_my_assets`` 就是其中一个。判据取工具
    # 自己的 schema,不取本次传参,见 :func:`_accepts_narrowing`。
    schema = readonly_catalogue()[requested].schema
    if _accepts_narrowing(schema) and not _has_narrowing_argument(arguments):
        logger.info(
            "assistant: refused unnarrowed oversized result for %s (%d chars)",
            requested, len(blob),
        )
        return json.dumps(oversized_refusal(), ensure_ascii=False)

    # A silent slice is the shape that produces confident wrong answers: the
    # model receives the first N characters of a listing, sees nothing after
    # them, and reports that the platform has no such thing. Production,
    # 2026-08-19: the software catalogue serialises to 29k characters, so an
    # unfiltered listing showed 17 of 57 products and everything alphabetically
    # after them was invisible. Truncation is unavoidable; hiding it is not.
    logger.info(
        "assistant: tool %s result truncated (%d chars)", requested, len(blob),
    )
    return json.dumps(
        {
            "truncated": True,
            "total_chars": len(blob),
            "shown_chars": _TRUNCATED_PAYLOAD_CHARS,
            "warning": (
                "PARTIAL result — rows are missing from the end. Do NOT tell "
                "the employee something does not exist based on this call. "
                f"{_retry_advice(schema)}"
            ),
            "partial_result": blob[:_TRUNCATED_PAYLOAD_CHARS],
        },
        ensure_ascii=False,
    )