import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

from langchain_core.messages import AIMessage

from app.agent.orchestrator import MoviePilotAgent
from app.agent.memory import memory_manager
from app.runtime.config import settings
from app.schemas.types import ChainEventType, EventType


class _FakeGraphState:
    """提供 LangGraph get_state 测试替身。"""

    def __init__(self, messages):
        self.values = {"messages": messages}


class _FakeAgent:
    """提供非流式 Agent 执行测试替身。"""

    def __init__(self, messages):
        self._messages = messages

    async def ainvoke(self, _payload, config=None):
        """模拟成功完成 Agent 调用。"""
        return None

    def get_state(self, _config):
        """返回测试消息状态。"""
        return _FakeGraphState(self._messages)


class _FakeFailingAgent(_FakeAgent):
    """提供失败 Agent 执行测试替身。"""

    async def ainvoke(self, _payload, config=None):
        """模拟 Agent 调用失败。"""
        raise RuntimeError("llm failed")


def test_initialize_llm_uses_chain_event_selection(monkeypatch) -> None:
    """Agent 初始化 LLM 时应优先使用链式事件返回的供应商配置。"""
    monkeypatch.setattr(settings, "LLM_THINKING_LEVEL", "xhigh")
    agent = MoviePilotAgent(session_id="agent-tokens-test", user_id="user-1")
    fake_llm = object()

    async def select_provider(event_type, event_data):
        """模拟 Agent Tokens 插件写入供应商配置。"""
        assert event_type == ChainEventType.AgentLLMProvider
        event_data.provider = "openai"
        event_data.base_url = "https://tokens.example.com/v1"
        event_data.api_key = "sk-agent-token"
        event_data.model = "free-model"
        event_data.base_url_preset = None
        event_data.user_agent = "AgentTokens-UA/1.0"
        event_data.selected_provider_id = "provider-1"
        event_data.selected_provider_name = "Free Provider"
        event_data.source = "AgentTokens"
        return SimpleNamespace(event_data=event_data)

    with (
        patch(
            "app.agent.orchestrator.eventmanager.async_send_event",
            new=AsyncMock(side_effect=select_provider),
        ) as send_event,
        patch(
            "app.agent.orchestrator.LLMHelper.get_llm",
            new=AsyncMock(return_value=fake_llm),
        ) as get_llm,
    ):
        result = asyncio.run(agent._initialize_llm(streaming=True))
        second_result = asyncio.run(agent._initialize_llm(streaming=False))

    assert result is fake_llm
    assert second_result is fake_llm
    send_event.assert_awaited_once()
    assert get_llm.await_count == 2
    get_llm.assert_any_await(
        streaming=True,
        provider="openai",
        model="free-model",
        api_key="sk-agent-token",
        base_url="https://tokens.example.com/v1",
        base_url_preset=None,
        user_agent="AgentTokens-UA/1.0",
        use_proxy=True,
        thinking_level="xhigh",
        api_protocol="auto",
        web_search_mode="local",
        prompt_cache_key=agent._build_prompt_cache_key(),
    )
    assert agent._llm_provider_selection["selected_provider_id"] == "provider-1"


def test_record_usage_emits_low_cardinality_token_metrics() -> None:
    """Token 指标只暴露归一供应商类别与输入输出方向。"""
    agent = MoviePilotAgent(session_id="usage-metrics", user_id="user-1")
    agent._llm_provider_selection = {"provider": "private-provider-name"}

    with patch("app.agent.orchestrator.record_metric") as record_metric:
        agent._record_usage(
            {
                "has_usage": True,
                "input_usage_available": True,
                "input_tokens": 120,
                "output_tokens": 30,
                "total_tokens": 150,
            }
        )

    record_metric.assert_any_call(
        "agent.token_usage",
        120,
        provider_type="custom",
        direction="input",
    )
    record_metric.assert_any_call(
        "agent.token_usage",
        30,
        provider_type="custom",
        direction="output",
    )


def test_execute_agent_broadcasts_usage_on_success() -> None:
    """Agent 执行成功后应广播聚合 token 用量事件。"""
    agent = MoviePilotAgent(session_id="usage-success", user_id="user-1")
    agent._should_stream = lambda: False
    agent.stream_handler = SimpleNamespace(
        stop_streaming=AsyncMock(return_value=(False, ""))
    )
    agent.send_agent_message = AsyncMock()

    async def create_agent(_streaming=False, streaming=False):
        """模拟创建 Agent 时完成供应商选择和用量统计。"""
        agent._llm_provider_selection = {
            "selected_provider_id": "provider-1",
            "selected_provider_name": "Free Provider",
            "provider": "openai",
            "base_url": "https://tokens.example.com/v1",
            "model": "free-model",
            "source": "AgentTokens",
        }
        agent._record_usage(
            {
                "has_usage": True,
                "model": "free-model",
                "input_tokens": 12,
                "output_tokens": 8,
                "total_tokens": 20,
                "cache_usage_available": True,
                "cache_read_input_tokens": 8,
                "cache_write_input_tokens": 2,
                "uncached_input_tokens": 2,
                "cache_hit_ratio": 8 / 12,
            }
        )
        return _FakeAgent([AIMessage(content="ok")])

    with (
        patch.object(agent, "_create_agent", new=create_agent),
        patch.object(memory_manager, "save_agent_messages"),
        patch("app.agent.orchestrator.eventmanager.send_event") as send_event,
    ):
        asyncio.run(agent._execute_agent([]))

    send_event.assert_called_once()
    assert send_event.call_args.args[0] == EventType.AgentTokensUsage
    usage = send_event.call_args.args[1]
    assert usage.success
    assert usage.selected_provider_id == "provider-1"
    assert usage.input_tokens == 12
    assert usage.output_tokens == 8
    assert usage.total_tokens == 20
    assert usage.cache_read_input_tokens == 8
    assert usage.cache_write_input_tokens == 2
    assert usage.uncached_input_tokens == 2
    assert usage.cache_hit_ratio == 8 / 12
    assert usage.cache_usage_available


def test_execute_agent_broadcasts_usage_on_failure() -> None:
    """Agent 执行失败后仍应广播用量事件。"""
    agent = MoviePilotAgent(session_id="usage-failure", user_id="user-1")
    agent._should_stream = lambda: False
    agent.stream_handler = SimpleNamespace(
        stop_streaming=AsyncMock(return_value=(False, ""))
    )
    agent.send_agent_message = AsyncMock()

    async def create_agent(_streaming=False, streaming=False):
        """模拟创建 Agent 时已选中供应商但执行失败。"""
        agent._llm_provider_selection = {
            "selected_provider_id": "provider-2",
            "selected_provider_name": "Backup Provider",
            "provider": "openai",
            "base_url": "https://backup.example.com/v1",
            "model": "backup-model",
            "source": "AgentTokens",
        }
        return _FakeFailingAgent([])

    with (
        patch.object(agent, "_create_agent", new=create_agent),
        patch("app.agent.orchestrator.eventmanager.send_event") as send_event,
    ):
        result, _ = asyncio.run(agent._execute_agent([]))

    assert "智能助手执行失败" in result
    send_event.assert_called_once()
    assert send_event.call_args.args[0] == EventType.AgentTokensUsage
    usage = send_event.call_args.args[1]
    assert not usage.success
    assert usage.selected_provider_id == "provider-2"
    assert "llm failed" in usage.error
