"""MCP surface completeness: ToolAnnotations + Resources + Prompts.

Guards the three spec surfaces added 2026-07-09 (AI 接入体系 PR3):

1. every registered tool carries a ToolAnnotations derived from its
   ``tool_runtime(mutates=…)`` ground truth (incl. delegation-pattern
   tools like the ``server_power_*`` family);
2. the server's DESTRUCTIVE_TOOLS stays in lockstep with the CLI's
   ``_DESTRUCTIVE_MCP_TOOLS`` confirmation list (same judgement, two
   enforcement points);
3. the two platform resources and five prompt playbooks stay
   registered and readable.
"""
from __future__ import annotations

import ast
import re
from pathlib import Path

import pytest

# Import for registration side-effects (tools/resources/prompts).
import app.mcp.prompts  # noqa: F401
import app.mcp.resources  # noqa: F401
import app.mcp.tools  # noqa: F401
from app.mcp.annotations import (
    DESTRUCTIVE_TOOLS,
    NON_IDEMPOTENT_TOOLS,
    apply_annotations,
    mutating_tools_from_source,
)
from app.mcp.server import mcp_server

_REPO_ROOT = Path(__file__).resolve().parents[3]


@pytest.fixture(scope="module", autouse=True)
def _annotated():
    apply_annotations(mcp_server)


def _tools():
    return mcp_server._tool_manager._tools


def test_every_tool_is_annotated():
    missing = [n for n, t in _tools().items() if t.annotations is None]
    assert missing == []


def test_read_write_hints_match_source_flags():
    flags = mutating_tools_from_source()
    tools = _tools()
    assert set(flags) == set(tools)
    for name, tool in tools.items():
        ann = tool.annotations
        assert ann.readOnlyHint is (not flags[name]), name
        assert ann.openWorldHint is False, name
        if flags[name]:
            # Platform invariant: state-changing operations are idempotent,
            # except the curated cost-bearing set (see NON_IDEMPOTENT_TOOLS).
            assert ann.idempotentHint is (name not in NON_IDEMPOTENT_TOOLS), name


def test_cost_bearing_generate_tools_are_not_advertised_idempotent():
    """A retry of a forced AI generation is NOT a no-op.

    ``force=True`` bypasses the cache, debits the caller's daily quota and
    bills a fresh model call, so a client that trusts ``idempotentHint`` and
    replays a timed-out request pays twice. The hint must say so.
    """
    tools = _tools()
    flags = mutating_tools_from_source()
    assert NON_IDEMPOTENT_TOOLS <= set(flags), "non-idempotent set names a missing tool"
    for name in NON_IDEMPOTENT_TOOLS:
        assert flags[name] is True, f"{name} is a read, cannot be non-idempotent"
        assert tools[name].annotations.idempotentHint is False, name
        assert tools[name].annotations.readOnlyHint is False, name
    # The invariant still holds for everything else — this is a two-name
    # exception, not a general loosening.
    others = [
        name for name, mutates in flags.items()
        if mutates and name not in NON_IDEMPOTENT_TOOLS
    ]
    assert all(tools[name].annotations.idempotentHint is True for name in others)


def test_spot_checks():
    tools = _tools()
    assert tools["list_my_assets"].annotations.readOnlyHint is True
    assert tools["borrow_asset"].annotations.readOnlyHint is False
    assert tools["borrow_asset"].annotations.destructiveHint is True
    # Approval-gated additive write: mutating but NOT destructive.
    assert tools["create_support_ticket"].annotations.readOnlyHint is False
    assert tools["create_support_ticket"].annotations.destructiveHint is False
    # Delegation pattern resolves through the shared lifecycle helper.
    assert tools["server_power_on"].annotations.readOnlyHint is False
    # Conditionally-mutating annotates conservatively as mutating.
    assert tools["list_approval_flow_stages"].annotations.readOnlyHint is False


def test_destructive_is_subset_of_mutating_and_registered():
    flags = mutating_tools_from_source()
    for name in DESTRUCTIVE_TOOLS:
        assert name in flags, f"{name} not a registered tool"
        assert flags[name] is True, f"{name} marked destructive but is a read"


def test_destructive_set_matches_cli_confirmation_list():
    """Server annotations and CLI confirm-prompt must agree on what is
    destructive. Parses the sibling cli/ checkout; skips gracefully when
    the server is checked out standalone (same convention as the CLI's
    mirror test reading tools.py)."""
    cli_main = _REPO_ROOT / "cli" / "helix_cli" / "main.py"
    if not cli_main.is_file():
        pytest.skip("cli/ not present in this checkout")
    tree = ast.parse(cli_main.read_text(encoding="utf-8"))
    cli_set: set[str] | None = None
    for node in ast.walk(tree):
        if (
            isinstance(node, ast.Assign)
            and any(
                isinstance(t, ast.Name) and t.id == "_DESTRUCTIVE_MCP_TOOLS"
                for t in node.targets
            )
        ):
            cli_set = {
                elt.value
                for call in ast.walk(node)
                if isinstance(call, ast.Call)
                for arg in call.args
                for elt in ast.walk(arg)
                if isinstance(elt, ast.Constant) and isinstance(elt.value, str)
            }
    assert cli_set, "could not locate _DESTRUCTIVE_MCP_TOOLS in cli main.py"
    assert cli_set == set(DESTRUCTIVE_TOOLS)


async def test_resources_registered_and_readable():
    resources = await mcp_server.list_resources()
    uris = {str(r.uri) for r in resources}
    assert {"helix://platform/overview", "helix://platform/tool-catalog"} <= uris

    overview = list(await mcp_server.read_resource("helix://platform/overview"))
    assert "approval chain" in overview[0].content

    catalog = list(await mcp_server.read_resource("helix://platform/tool-catalog"))
    text = catalog[0].content
    assert "`list_my_assets`" in text
    assert "**[DESTRUCTIVE]**" in text
    # Header count reflects the live registry.
    m = re.search(r"^(\d+) tools —", text.splitlines()[2])
    assert m and int(m.group(1)) == len(_tools())


async def test_prompts_registered():
    prompts = await mcp_server.list_prompts()
    names = {p.name for p in prompts}
    assert {
        "it_overview",
        "approvals_review",
        "cost_review",
        "offboarding_handover",
        "onboarding_progress",
    } <= names
    result = await mcp_server.get_prompt("offboarding_handover", None)
    joined = " ".join(
        m.content.text for m in result.messages if hasattr(m.content, "text")
    )
    assert "get_offboarding_preview" in joined