"""The read-only MCP bridge, and the ways someone would try to get past it.

The bridge is the one place in the assistant where a large, automatically
growing surface becomes reachable, so most of what follows is adversarial:
name the mutating tool directly, name the destructive one, ask for a tool that
does not exist, hand the model's own identity claim to the runtime. All of it
must land on a refusal, and the refusal must be by allowlist rather than by
naming the bad cases.
"""
from __future__ import annotations

import json

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant.mcp_bridge import (
    TOOL_CALL,
    TOOL_SEARCH,
    _lexical_search,
    bridge_tools,
    is_callable_tool,
    make_dispatch,
    readonly_catalogue,
)
from app.mcp.annotations import DESTRUCTIVE_TOOLS, mutating_tools_from_source

# ── the catalogue is derived, not curated ───────────────────────────


def test_catalogue_is_non_trivial_and_all_read_only():
    """Derived from the platform's own ``mutates=`` ground truth."""
    flags = mutating_tools_from_source()
    catalogue = readonly_catalogue()

    assert len(catalogue) > 100, "bridge should expose the real read-only surface"
    for name in catalogue:
        assert flags[name] is False, f"{name} is mutating but reachable"


def test_no_mutating_tool_is_reachable():
    flags = mutating_tools_from_source()
    mutating = [name for name, mutates in flags.items() if mutates]

    assert mutating, "sanity: the platform does have mutating tools"
    for name in mutating:
        assert not is_callable_tool(name), f"{name} must not be reachable"


def test_destructive_tools_are_excluded_even_if_flagged_read_only():
    for name in DESTRUCTIVE_TOOLS:
        assert not is_callable_tool(name)


def test_the_assistants_own_audit_tools_are_not_reachable_from_chat():
    """Authority-consistent, but the wrong thing to make chat-reachable.

    Only an admin or auditor could call these at all — yet pulling a
    colleague's verbatim transcript into a model prompt is precisely what
    narrowing that surface was meant to prevent.
    """
    for name in (
        "assistant_list_conversations",
        "assistant_get_conversation",
        "assistant_usage_stats",
    ):
        assert name in mutating_tools_from_source(), f"{name} should be registered"
        assert not is_callable_tool(name)


def test_unknown_tool_names_are_not_callable():
    assert not is_callable_tool("delete_everything")
    assert not is_callable_tool("")


# ── search ──────────────────────────────────────────────────────────


@pytest.mark.parametrize(
    ("query", "expected"),
    [
        ("my assets laptop", "list_my_assets"),
        ("software seats", "list_my_software_seats"),
    ],
)
def test_search_finds_the_obvious_tool(query, expected):
    assert expected in [t.name for t in _lexical_search(query)]


def test_search_is_bounded_and_ordered():
    results = _lexical_search("my list get")
    assert len(results) <= 6
    assert len(results) == len({t.name for t in results})


def test_search_on_noise_returns_nothing():
    assert _lexical_search("") == []
    assert _lexical_search("   ") == []


def test_the_model_sees_two_meta_tools_not_the_whole_registry():
    """400+ schemas do not fit a prompt; two do."""
    tools = bridge_tools()
    assert [t["name"] for t in tools] == [TOOL_SEARCH, TOOL_CALL]
    for tool in tools:
        assert tool["input_schema"]["additionalProperties"] is False


# ── dispatch ────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_search_dispatch_returns_names_and_parameters(
    db_session: AsyncSession, feishu_person,
):
    dispatch = make_dispatch(person=feishu_person)

    out = json.loads(
        await dispatch(TOOL_SEARCH, {"query": "my assets"}, db_session, feishu_person),
    )

    assert out["tools"]
    assert all("name" in t and "parameters" in t for t in out["tools"])


@pytest.mark.asyncio
async def test_calling_a_read_only_tool_runs_as_the_resolved_person(
    client, db_session: AsyncSession, feishu_person,
):
    """Identity comes from the closure, and the tool really executes.

    Takes the ``client`` fixture so the tool runtime's own session factory is
    bound to the test database — without it the tool opens a session elsewhere
    and every assertion here would pass on an error path.
    """
    dispatch = make_dispatch(person=feishu_person)

    out = json.loads(
        await dispatch(
            TOOL_CALL, {"name": "list_my_assets", "arguments": {}},
            db_session, feishu_person,
        ),
    )

    assert "error" not in out, out
    assert "assets" in out


@pytest.mark.asyncio
async def test_a_tool_that_refuses_is_not_reported_as_bad_arguments(
    client, db_session: AsyncSession, feishu_person,
):
    """FastMCP wraps both causes in ToolError; mislabelling sends the model
    off rewriting arguments that were already correct."""
    dispatch = make_dispatch(person=feishu_person)

    out = json.loads(
        await dispatch(
            TOOL_CALL,
            {"name": "list_all_assets", "arguments": {}},
            db_session,
            feishu_person,
        ),
    )

    assert out.get("error") in (None, "tool_failed", "tool_not_available")


@pytest.mark.asyncio
async def test_naming_a_mutating_tool_directly_is_refused(
    db_session: AsyncSession, feishu_person,
):
    """The filter is an allowlist, so this needs no knowledge of the tool."""
    flags = mutating_tools_from_source()
    mutating = next(name for name, mutates in flags.items() if mutates)
    dispatch = make_dispatch(person=feishu_person)

    out = json.loads(
        await dispatch(
            TOOL_CALL, {"name": mutating, "arguments": {}}, db_session, feishu_person,
        ),
    )

    assert out["error"] == "tool_not_available"


@pytest.mark.asyncio
async def test_naming_a_destructive_tool_is_refused(
    db_session: AsyncSession, feishu_person,
):
    dispatch = make_dispatch(person=feishu_person)

    out = json.loads(
        await dispatch(
            TOOL_CALL, {"name": "revoke_api_key", "arguments": {"id": "x"}},
            db_session, feishu_person,
        ),
    )

    assert out["error"] == "tool_not_available"


@pytest.mark.asyncio
async def test_a_made_up_tool_name_is_refused(
    db_session: AsyncSession, feishu_person,
):
    dispatch = make_dispatch(person=feishu_person)

    out = json.loads(
        await dispatch(
            TOOL_CALL, {"name": "drop_all_tables"}, db_session, feishu_person,
        ),
    )

    assert out["error"] == "tool_not_available"


@pytest.mark.asyncio
async def test_the_real_tool_name_reaches_the_transcript_not_the_meta_tool(
    db_session: AsyncSession, feishu_person,
):
    """"call_helix_tool" in an audit record answers no question anyone has."""
    seen: list[str] = []
    dispatch = make_dispatch(person=feishu_person, called=seen)

    await dispatch(
        TOOL_CALL, {"name": "list_my_assets", "arguments": {}},
        db_session, feishu_person,
    )

    assert seen == ["list_my_assets"]


@pytest.mark.asyncio
async def test_a_missing_required_argument_comes_back_as_guidance(
    client, db_session: AsyncSession, feishu_person,
):
    """The model can fix this itself if we say so plainly.

    Note ``get_my_wrapped`` is used because it actually declares a required
    typed parameter — an *unknown* key is silently dropped by the schema, so
    it is not a validation failure and would prove nothing here.
    """
    dispatch = make_dispatch(person=feishu_person)

    out = json.loads(
        await dispatch(
            TOOL_CALL,
            {"name": "get_my_wrapped", "arguments": {}},
            db_session,
            feishu_person,
        ),
    )

    assert out["error"] == "invalid_arguments"
    assert "year" in out["detail"]


@pytest.mark.asyncio
async def test_an_unknown_meta_tool_is_refused(
    db_session: AsyncSession, feishu_person,
):
    dispatch = make_dispatch(person=feishu_person)

    out = json.loads(await dispatch("something_else", {}, db_session, feishu_person))

    assert "error" in out


@pytest.mark.asyncio
async def test_the_context_var_is_restored_after_a_call(
    db_session: AsyncSession, feishu_person,
):
    """A leaked identity would make the NEXT caller run as this one."""
    from app.mcp.context import mcp_user_id

    before = mcp_user_id.get()
    dispatch = make_dispatch(person=feishu_person)

    await dispatch(
        TOOL_CALL, {"name": "list_my_assets", "arguments": {}},
        db_session, feishu_person,
    )

    assert mcp_user_id.get() == before


@pytest.mark.asyncio
async def test_the_context_var_is_restored_even_when_the_tool_fails(
    db_session: AsyncSession, feishu_person,
):
    from app.mcp.context import mcp_user_id

    before = mcp_user_id.get()
    dispatch = make_dispatch(person=feishu_person)

    await dispatch(
        TOOL_CALL, {"name": "list_my_assets", "arguments": {"bogus": 1}},
        db_session, feishu_person,
    )

    assert mcp_user_id.get() == before


# ── an over-long result must SAY it was cut ─────────────────────────


class _HugeTool:
    """Stands in for a tool whose real result overflows the cap.

    Swapped into the registry rather than into the derived catalogue, so the
    allowlist check still runs against the real read-only name — and so the
    bridge still reads the REAL tool's parameter schema, which is what decides
    between the two over-size shapes below.
    """

    async def run(self, _arguments):
        return {"items": ["x" * 200 for _ in range(200)]}


def _bridge_oversized(monkeypatch, name: str):
    from app.mcp.server import mcp_server

    monkeypatch.setitem(mcp_server._tool_manager._tools, name, _HugeTool())


@pytest.mark.asyncio
async def test_an_oversized_result_the_caller_could_not_narrow_is_marked_partial(
    client, db_session: AsyncSession, feishu_person, monkeypatch,
):
    """A sliced result reads to the model as a complete one.

    The bridge used to hand back ``json.dumps(result)[:6000]``. The software
    catalogue serialises to ~36k characters, so a listing arrived cut off after
    a third of the products with nothing saying so — and "it is not in the
    catalogue" is exactly the answer a model gives when the rest of the list
    was never shown to it. Truncation is unavoidable; hiding it is not.

    ``list_my_assets`` is the case where that still holds after #3512 added the
    zero-row refusal: its real schema is ``page`` / ``page_size`` and nothing
    else, so "retry with a narrowing argument" is an instruction it cannot
    answer. Refusing here would leave the model with no data AND no legal
    conclusion — a dead end on an employee asking about their own kit.
    """
    tool = readonly_catalogue()["list_my_assets"]
    assert set(tool.schema.get("properties", {})) == {"page", "page_size"}, (
        "premise changed: list_my_assets grew a filter, pick another "
        "pagination-only tool for this case"
    )
    _bridge_oversized(monkeypatch, tool.name)
    dispatch = make_dispatch(person=feishu_person)

    raw = await dispatch(
        TOOL_CALL, {"name": tool.name, "arguments": {}}, db_session, feishu_person,
    )

    # Still valid JSON — a chopped payload is not something a model can parse.
    out = json.loads(raw)
    assert out["truncated"] is True
    assert out["total_chars"] > out["shown_chars"]
    assert "PARTIAL" in out["warning"]
    assert "do not" in out["warning"].lower() or "Do NOT" in out["warning"]
    assert out["partial_result"].startswith("{")
    # …and the retry it suggests is one this tool can actually be given.
    assert "search term" not in out["warning"], (
        "told the model to filter a tool that has no filter — it will come "
        "back with invalid_arguments and still have no data"
    )
    assert "page" in out["warning"]


@pytest.mark.asyncio
async def test_an_oversized_unnarrowed_result_on_a_filterable_tool_returns_no_rows(
    client, db_session: AsyncSession, feishu_person, monkeypatch,
):
    """The 2026-09-07 shape: a global listing asked for with nothing narrowed.

    ``list_cloud_scopes`` is the production tool from that incident and it does
    take ``search``. The model called it three times without one and answered
    "no such cloud account" about an account that was in the database with an
    owner — it was row 197 of 271 and the page held 200. So this call gets no
    rows at all: a wrong negative cannot be picked out of a list that was never
    handed over (#3512).
    """
    tool = readonly_catalogue()["list_cloud_scopes"]
    assert "search" in tool.schema.get("properties", {}), (
        "premise changed: list_cloud_scopes lost its search parameter"
    )
    _bridge_oversized(monkeypatch, tool.name)
    dispatch = make_dispatch(person=feishu_person)

    out = json.loads(
        await dispatch(
            TOOL_CALL,
            {"name": tool.name, "arguments": {"page": 1, "page_size": 200}},
            db_session, feishu_person,
        ),
    )

    assert out["error"] == "too_many_rows"
    assert out["shown_rows"] == 0
    assert "partial_result" not in out, "still handed over half a list"
    assert "items" not in out
    assert "search" in out["hint"]
    assert "没有找到" in out["hint"] and "不存在" in out["hint"]


@pytest.mark.asyncio
async def test_an_oversized_result_after_narrowing_is_marked_partial(
    client, db_session: AsyncSession, feishu_person, monkeypatch,
):
    """A narrowed call that is still too big gets rows back, marked partial.

    The refusal is about HOW the tool was asked, not about size: once the caller
    has said what it is looking for, the truncation is the data being large and
    a marked partial is the honest answer.
    """
    tool = readonly_catalogue()["list_cloud_scopes"]
    _bridge_oversized(monkeypatch, tool.name)
    dispatch = make_dispatch(person=feishu_person)

    out = json.loads(
        await dispatch(
            TOOL_CALL,
            {"name": tool.name, "arguments": {"search": "pingcap"}},
            db_session, feishu_person,
        ),
    )

    assert out["truncated"] is True
    assert out["partial_result"].startswith("{")
    assert "PARTIAL" in out["warning"]
    assert "search term" in out["warning"]


def test_a_tool_with_only_paging_parameters_cannot_be_narrowed() -> None:
    """The predicate behind the split, on the real catalogue's own schemas.

    Measured 2026-09-08: 118 of the 360 read-only tools have no narrowable
    parameter — every ``get_my_*`` / ``list_my_*`` scoped to the caller. The
    zero-row refusal must never be the answer for those.
    """
    from app.assistant.mcp_bridge import _accepts_narrowing

    assert _accepts_narrowing({"properties": {"page": {}, "page_size": {}}}) is False
    assert _accepts_narrowing({"properties": {}}) is False
    assert _accepts_narrowing({"properties": {"search": {}, "page": {}}}) is True
    assert _accepts_narrowing({"properties": {"requestable": {}}}) is True
    # 读不出 schema 时按「可收窄」处理 —— 宁可少给数据。
    assert _accepts_narrowing({}) is True

    catalogue = readonly_catalogue()
    assert _accepts_narrowing(catalogue["list_my_assets"].schema) is False
    assert _accepts_narrowing(catalogue["whoami"].schema) is False
    assert _accepts_narrowing(catalogue["list_cloud_scopes"].schema) is True


def test_the_retry_advice_never_asks_for_a_parameter_the_tool_lacks() -> None:
    """Every branch's advice has to be followable by that tool."""
    from app.assistant.mcp_bridge import _retry_advice

    filterable = _retry_advice({"properties": {"search": {}}})
    assert "search term" in filterable

    pageable = _retry_advice({"properties": {"page": {}, "page_size": {}}})
    assert "search" not in pageable
    assert "page" in pageable

    fixed = _retry_advice({"properties": {}})
    assert "search" not in fixed and "page" not in fixed
    # 既收窄不了也翻不了页时,唯一诚实的话是「剩下的看不到」。
    assert "unknown" in fixed and "absent" in fixed


@pytest.mark.asyncio
async def test_a_result_within_the_cap_is_returned_verbatim(
    client, db_session: AsyncSession, feishu_person,
):
    """The envelope is for the over-long case only."""
    dispatch = make_dispatch(person=feishu_person)

    out = json.loads(
        await dispatch(
            TOOL_CALL, {"name": "whoami", "arguments": {}},
            db_session, feishu_person,
        ),
    )

    assert "truncated" not in out


@pytest.mark.asyncio
async def test_an_empty_search_points_at_the_knowledge_base_not_at_memory(
    db_session: AsyncSession, feishu_person, monkeypatch,
):
    """「打印机怎么连」 has no data tool — but it does have a handbook article.

    The note used to read "Answer from what you know", which for an IT
    assistant is an instruction to invent the company's own setup.
    """
    from app.assistant import mcp_bridge

    async def _nothing(_db, _query, *, limit=6):
        return []

    monkeypatch.setattr(mcp_bridge, "search", _nothing)
    dispatch = make_dispatch(person=feishu_person)

    out = json.loads(
        await dispatch(TOOL_SEARCH, {"query": "zzz"}, db_session, feishu_person),
    )

    assert out["tools"] == []
    assert "search_knowledge_base" in out["note"]
    assert "from what you know" not in out["note"]


def test_the_assistants_own_audit_reads_stay_off_the_chat_bridge() -> None:
    """🔴 Four tools carrying colleagues' verbatim text were bridged.

    ``readonly_catalogue`` admits every MCP tool the AST flag marks
    non-mutating, minus a hand-curated exclusion set — and that set exists
    precisely to stop "pulling a colleague's verbatim transcript into a model
    prompt". Its own comment records ``assistant_unhelpful_answers`` being
    missed once for this reason. Four more were added a release later and
    missed the same way.

    What they carry: ``list_assistant_eval_failures`` returns the question, the
    reply and the documents read for an attempt run under a PERSONA employee's
    identity; the knowledge-loop queue's ``evidence`` is
    ``{"employee_question": <a colleague's exact words>}``, and unlike the
    evaluation miner the lint path does not redact it. Bridged, an admin asking
    the bot 「队列里现在有什么」 sends that to the model provider AND writes it
    into an ``assistant_messages`` transcript row.
    """
    from app.assistant.mcp_bridge import readonly_catalogue

    catalogue = readonly_catalogue()
    for name in (
        "assistant_unhelpful_answers",
        "list_assistant_eval_failures",
        "get_assistant_eval_run",
        "list_knowledge_proposals",
        "get_knowledge_proposal",
    ):
        assert name not in catalogue, name

    # And the bridge is not simply empty — the exclusion is targeted, not a
    # blanket refusal that would make this assertion meaningless.
    assert len(catalogue) > 100