"""
L05: Validation tests for the unified LLM Provider interface (L01-L04).

Tests provider creation, ChatMessage/ChatResponse types, ConversationLam
with mock providers, context compaction, and history reset.

All tests use mocks — no real API calls, no anthropic/openai SDK imports.
"""
from __future__ import annotations

import unittest
from unittest.mock import MagicMock, patch
from typing import List, Dict

# Import the types and factory
from lambdagent.providers.base import (
    LLMProvider, ProviderConfig, ProviderError,
    ChatMessage, ChatResponse,
)
from lambdagent.providers import create_provider
from lambdagent.providers.anthropic_provider import AnthropicProvider
from lambdagent.providers.openai_compat_provider import OpenAICompatProvider
from lambdagent.conversation import ConversationLam
from lambdagent.core import Context


# ============================================================
# Mock provider for testing ConversationLam
# ============================================================

class MockProvider(LLMProvider):
    """Test double that returns canned responses without any API calls."""

    def __init__(self, responses: list[str] | None = None):
        config = ProviderConfig(model="mock-model", temperature=0.0, max_tokens=100)
        super().__init__(config)
        self._responses = list(responses or ["mock response"])
        self._call_count = 0

    def chat(self, messages: List[Dict[str, str]]) -> str:
        idx = min(self._call_count, len(self._responses) - 1)
        self._call_count += 1
        return self._responses[idx]

    def chat_typed(self, messages: List[ChatMessage],
                   model: str = "", temperature: float = 0.0,
                   max_tokens: int = 4096) -> ChatResponse:
        idx = min(self._call_count, len(self._responses) - 1)
        self._call_count += 1
        return ChatResponse(
            text=self._responses[idx],
            input_tokens=10,
            output_tokens=5,
            model=model or "mock-model",
            finish_reason="end_turn",
        )

    @property
    def provider_name(self) -> str:
        return "mock"

    @property
    def default_model(self) -> str:
        return "mock-model"


# ============================================================
# Test 1-4: create_provider factory
# ============================================================

class TestCreateProvider(unittest.TestCase):
    """Test create_provider() returns correct provider types."""

    def test_create_anthropic_provider(self):
        """create_provider('anthropic') returns AnthropicProvider."""
        provider = create_provider("anthropic")
        self.assertIsInstance(provider, AnthropicProvider)
        self.assertEqual(provider.provider_name, "anthropic")

    def test_create_dashscope_provider(self):
        """create_provider('dashscope') returns OpenAICompatProvider with correct base_url."""
        provider = create_provider("dashscope")
        self.assertIsInstance(provider, OpenAICompatProvider)
        self.assertIn("dashscope.aliyuncs.com", provider.base_url)
        self.assertEqual(provider.provider_name, "dashscope")

    def test_create_ollama_provider(self):
        """create_provider('ollama') returns OpenAICompatProvider with localhost URL."""
        provider = create_provider("ollama")
        self.assertIsInstance(provider, OpenAICompatProvider)
        self.assertTrue(
            "localhost:11434" in provider.base_url
            or "127.0.0.1:11434" in provider.base_url
        )
        self.assertEqual(provider.provider_name, "ollama")

    def test_create_openai_provider(self):
        """create_provider('openai') returns OpenAICompatProvider."""
        provider = create_provider("openai")
        self.assertIsInstance(provider, OpenAICompatProvider)
        self.assertEqual(provider.provider_name, "openai")

    def test_create_deepseek_provider(self):
        """create_provider('deepseek') returns OpenAICompatProvider."""
        provider = create_provider("deepseek")
        self.assertIsInstance(provider, OpenAICompatProvider)
        self.assertEqual(provider.provider_name, "deepseek")

    def test_create_unknown_raises_value_error(self):
        """create_provider('unknown') raises ValueError."""
        with self.assertRaises(ValueError) as cm:
            create_provider("unknown")
        self.assertIn("Unknown provider", str(cm.exception))
        self.assertIn("unknown", str(cm.exception))


# ============================================================
# Test 5: ChatMessage and ChatResponse creation
# ============================================================

class TestDataclasses(unittest.TestCase):
    """Test ChatMessage and ChatResponse can be created and used."""

    def test_chat_message_creation(self):
        msg = ChatMessage(role="user", content="hello")
        self.assertEqual(msg.role, "user")
        self.assertEqual(msg.content, "hello")

    def test_chat_message_roles(self):
        for role in ("system", "user", "assistant", "tool"):
            msg = ChatMessage(role=role, content=f"test {role}")
            self.assertEqual(msg.role, role)

    def test_chat_response_creation(self):
        resp = ChatResponse(text="hi", input_tokens=10, output_tokens=5,
                            model="test-model", finish_reason="stop")
        self.assertEqual(resp.text, "hi")
        self.assertEqual(resp.input_tokens, 10)
        self.assertEqual(resp.output_tokens, 5)
        self.assertEqual(resp.model, "test-model")
        self.assertEqual(resp.finish_reason, "stop")

    def test_chat_response_defaults(self):
        resp = ChatResponse(text="hi")
        self.assertEqual(resp.input_tokens, 0)
        self.assertEqual(resp.output_tokens, 0)
        self.assertEqual(resp.model, "")
        self.assertEqual(resp.finish_reason, "")


# ============================================================
# Test 6-9: ConversationLam with mock provider
# ============================================================

class TestConversationLam(unittest.TestCase):
    """Test ConversationLam instantiation, apply, compact, and reset."""

    def test_instantiation(self):
        """ConversationLam can be instantiated with a mock provider."""
        provider = MockProvider()
        lam = ConversationLam(
            name="test-agent",
            provider=provider,
            system_prompt="You are a test agent.",
        )
        self.assertEqual(lam._name, "test-agent")
        self.assertEqual(lam.system_prompt, "You are a test agent.")
        self.assertEqual(len(lam.messages), 1)  # Just system message

    def test_apply_returns_expected(self):
        """ConversationLam.apply() with a mock provider returns the mock response."""
        provider = MockProvider(["Hello from mock!"])
        lam = ConversationLam(
            name="test-agent",
            provider=provider,
            system_prompt="You are a test agent.",
        )
        result = lam.apply("Hi there")
        self.assertEqual(result, "Hello from mock!")

    def test_apply_builds_history(self):
        """Each apply() adds user + assistant messages to history."""
        provider = MockProvider(["r1", "r2", "r3"])
        lam = ConversationLam(
            name="test-agent",
            provider=provider,
            system_prompt="system",
        )
        lam.apply("q1")
        lam.apply("q2")
        # system + (user, assistant) * 2 = 5
        self.assertEqual(len(lam.messages), 5)
        self.assertEqual(lam.messages[1]["role"], "user")
        self.assertEqual(lam.messages[1]["content"], "q1")
        self.assertEqual(lam.messages[2]["role"], "assistant")
        self.assertEqual(lam.messages[2]["content"], "r1")

    def test_apply_typed_returns_chat_response(self):
        """ConversationLam.apply_typed() returns ChatResponse with metadata."""
        provider = MockProvider(["typed response"])
        lam = ConversationLam(
            name="test-agent",
            provider=provider,
            system_prompt="system",
        )
        response = lam.apply_typed("test input")
        self.assertIsInstance(response, ChatResponse)
        self.assertEqual(response.text, "typed response")
        self.assertEqual(response.input_tokens, 10)
        self.assertEqual(response.output_tokens, 5)

    def test_compact_trims_messages(self):
        """_compact_typed trims messages when over token budget."""
        provider = MockProvider()
        lam = ConversationLam(
            name="test-agent",
            provider=provider,
            system_prompt="s",
            max_history_tokens=50,  # Very low budget
        )
        # Build a list of messages that exceeds the budget
        messages = [ChatMessage(role="system", content="system prompt")]
        for i in range(20):
            messages.append(ChatMessage(role="user", content="x" * 100))
            messages.append(ChatMessage(role="assistant", content="y" * 100))

        compacted = lam._compact_typed(messages)
        # Should be fewer messages than original
        self.assertLess(len(compacted), len(messages))
        # Should always keep system message
        self.assertEqual(compacted[0].role, "system")

    def test_compact_no_trim_when_under_budget(self):
        """_compact_typed returns all messages when under budget."""
        provider = MockProvider()
        lam = ConversationLam(
            name="test-agent",
            provider=provider,
            system_prompt="s",
            max_history_tokens=100000,  # Large budget
        )
        messages = [
            ChatMessage(role="system", content="sys"),
            ChatMessage(role="user", content="hi"),
            ChatMessage(role="assistant", content="hello"),
        ]
        compacted = lam._compact_typed(messages)
        self.assertEqual(len(compacted), 3)

    def test_reset_clears_history(self):
        """ConversationLam.reset() clears both dict and typed history."""
        provider = MockProvider(["r1", "r2"])
        lam = ConversationLam(
            name="test-agent",
            provider=provider,
            system_prompt="system",
        )
        lam.apply("q1")
        lam.apply_typed("q2")
        self.assertGreater(len(lam.messages), 1)
        self.assertGreater(len(lam._typed_messages), 0)

        lam.reset()
        self.assertEqual(len(lam.messages), 1)  # Only system message
        self.assertEqual(lam.messages[0]["content"], "system")
        self.assertEqual(len(lam._typed_messages), 0)

    def test_context_trace_logging(self):
        """apply() logs trace entries to the Context."""
        provider = MockProvider(["response"])
        lam = ConversationLam(
            name="test-agent",
            provider=provider,
            system_prompt="system",
        )
        ctx = Context()
        lam.apply("input", ctx=ctx)
        self.assertEqual(len(ctx.trace), 1)
        self.assertEqual(ctx.trace[0].term_name, "test-agent")

    def test_history_property(self):
        """history property returns typed message list."""
        provider = MockProvider(["r1"])
        lam = ConversationLam(
            name="test-agent",
            provider=provider,
            system_prompt="system",
        )
        lam.apply_typed("q1")
        hist = lam.history
        self.assertEqual(len(hist), 2)  # user + assistant
        self.assertEqual(hist[0].role, "user")
        self.assertEqual(hist[0].content, "q1")
        self.assertEqual(hist[1].role, "assistant")
        self.assertEqual(hist[1].content, "r1")


# ============================================================
# Test: LLMProvider base class default implementations
# ============================================================

class TestLLMProviderDefaults(unittest.TestCase):
    """Test that LLMProvider base class provides working defaults."""

    def test_default_provider_name(self):
        """provider_name defaults to lowercased class name without 'Provider'."""
        provider = MockProvider()
        # MockProvider -> provider_name = "mock" (from override)
        self.assertEqual(provider.provider_name, "mock")

    def test_default_model(self):
        """default_model returns config.model."""
        provider = MockProvider()
        self.assertEqual(provider.default_model, "mock-model")

    def test_chat_typed_delegates_to_chat(self):
        """Base chat_typed() delegates to chat() for backward compat."""
        provider = MockProvider(["delegated response"])
        # Reset call count since MockProvider overrides chat_typed
        # Test the base class behavior by calling chat directly
        result = provider.chat([{"role": "user", "content": "test"}])
        self.assertEqual(result, "delegated response")


if __name__ == "__main__":
    unittest.main()


class TestReactProviderRetry(unittest.TestCase):
    """回归：provider 抛 retryable ProviderError(如 claude-code resume 卡死)时，
    react 引擎应重试恢复，而不是整个 run 崩掉。"""

    def test_react_retries_retryable_provider_error(self):
        from lambdagent.providers.base import ProviderError, ProviderConfig, ChatResponse, LLMProvider
        from lambdagent.fromconfig import compiler as _compiler

        class FlakeProvider(LLMProvider):
            """首次 chat 抛 retryable ProviderError，之后返回可终止的纯文本。"""
            def __init__(self):
                super().__init__(ProviderConfig(model="claude-code/sonnet"))
                self.calls = 0
            def _resp(self):
                self.calls += 1
                if self.calls == 1:
                    raise ProviderError("stalled (first-byte) on resume",
                                        "claude-code", retryable=True)
                return "任务已完成：测试通过。"
            def chat(self, messages):
                return self._resp()
            def chat_typed(self, messages, model="", temperature=0.0, max_tokens=4096):
                txt = self._resp()
                return ChatResponse(text=txt, input_tokens=5, output_tokens=5,
                                    model="claude-code/sonnet", finish_reason="end_turn")
            @property
            def provider_name(self):
                return "claude-code"

        flake = FlakeProvider()
        cfg = {
            "agentId": "t", "name": "t", "type": "react",
            "model": {"provider": "claude-code", "name": "sonnet"},
            "systemPrompt": "你是助手。",
            "react": {"maxSteps": 4},
            "mcp": {"localTools": ["terminate"]},
        }
        # 注入 flaky provider（session 模式，模拟 claude-code）
        with patch.object(_compiler, "_create_provider", return_value=(flake, True)):
            term = _compiler.build_agent(cfg)
            result = term.apply("做个小任务", Context())

        # 没有抛异常 = 引擎吞下首次 retryable 错误并重试恢复
        self.assertGreaterEqual(flake.calls, 2, "应至少重试一次（首调抛错、二调成功）")
        self.assertIn("完成", str(result))


if __name__ == "__main__":
    unittest.main()


class TestResumeSessionPreserved(unittest.TestCase):
    """回归：resume 首字节卡死应【保留】会话(只是 API 慢、会话仍在)，
    避免每次卡死换 fresh 会话导致 agent 丢记忆反复重分析。idle/hard 卡死才清。"""

    def _make_provider(self):
        from lambdagent.providers import claude_code_provider as ccp
        from lambdagent.providers.base import ProviderConfig
        p = ccp.ClaudeCodeProvider.__new__(ccp.ClaudeCodeProvider)  # 跳过 bin 校验
        p.config = ProviderConfig(model="sonnet", timeout=600, extra={})
        p.claude_bin = "claude"
        p._session_id = "sess-123"
        p._resume_count = 0
        p._resume_stall_streak = 0
        return p, ccp

    def test_first_byte_stall_keeps_session(self):
        from lambdagent.providers.base import ProviderError
        p, ccp = self._make_provider()
        def fake(*a, **k):
            raise ccp._StallError("first-byte", 45.0, "", "")
        with patch.object(ccp, "_run_with_idle_timeout", fake):
            with self.assertRaises(ProviderError):
                p._call_resume([{"role": "user", "content": "hi"}])
        self.assertEqual(p._session_id, "sess-123", "首字节卡死应保留会话")

    def test_idle_stall_clears_session(self):
        from lambdagent.providers.base import ProviderError
        p, ccp = self._make_provider()
        def fake(*a, **k):
            raise ccp._StallError("idle", 60.0, "", "")
        with patch.object(ccp, "_run_with_idle_timeout", fake):
            with self.assertRaises(ProviderError):
                p._call_resume([{"role": "user", "content": "hi"}])
        self.assertIsNone(p._session_id, "idle 卡死应清会话")


class TestSessionRotation(unittest.TestCase):
    """resume 成功 N 次后应轮换会话(清 session_id),防大会话 resume 卡死。"""

    def test_session_rotates_after_n_resumes(self):
        from lambdagent.providers import claude_code_provider as ccp
        from lambdagent.providers.base import ProviderConfig

        class FakeResult:
            returncode = 0
            stdout = "ok"
            stderr = ""

        p = ccp.ClaudeCodeProvider.__new__(ccp.ClaudeCodeProvider)
        p.config = ProviderConfig(model="sonnet", timeout=600,
                                  extra={"claude_session_rotate_after": 3})
        p.claude_bin = "claude"
        p._session_id = "sess-A"
        p._resume_count = 0

        with patch.object(ccp, "_run_with_idle_timeout", return_value=FakeResult()):
            # 前两次 resume:会话保留
            for i in range(2):
                p._call_resume([{"role": "user", "content": "x"}])
                self.assertEqual(p._session_id, "sess-A", f"第{i+1}次后不应轮换")
            # 第三次:到阈值 → 轮换,清 session_id
            p._call_resume([{"role": "user", "content": "x"}])
            self.assertIsNone(p._session_id, "第3次 resume 后应轮换(清会话)")
            self.assertEqual(p._resume_count, 0, "轮换后计数归零")


class TestResumeStallGiveup(unittest.TestCase):
    """连续首字节卡死必须熔断:连卡 K 次后清掉大 session 换新,
    否则会像那条 run 一样在同一大 session 上连卡 8 次打转到死。"""

    def test_consecutive_first_byte_stalls_clear_session(self):
        from lambdagent.providers import claude_code_provider as ccp
        from lambdagent.providers.base import ProviderConfig, ProviderError

        p = ccp.ClaudeCodeProvider.__new__(ccp.ClaudeCodeProvider)
        p.config = ProviderConfig(model="sonnet", timeout=600,
                                  extra={"claude_resume_stall_giveup": 2})
        p.claude_bin = "claude"
        p._session_id = "sess-BIG"
        p._resume_count = 0
        p._resume_stall_streak = 0

        def always_stall(*a, **k):
            raise ccp._StallError("first-byte", 45.0, "", "")

        with patch.object(ccp, "_run_with_idle_timeout", side_effect=always_stall), \
             patch.object(ccp.time, "sleep", lambda *_: None):
            # 第 1 次 _call_resume 全卡:保留会话(兜瞬时抖动)
            with self.assertRaises(ProviderError):
                p._call_resume([{"role": "user", "content": "x"}])
            self.assertEqual(p._session_id, "sess-BIG", "第1次连卡应保留会话")
            self.assertEqual(p._resume_stall_streak, 1)
            # 第 2 次 _call_resume 又全卡:达阈值 → 熔断清会话
            with self.assertRaises(ProviderError):
                p._call_resume([{"role": "user", "content": "x"}])
            self.assertIsNone(p._session_id, "连卡2次后应清会话换新")
            self.assertEqual(p._resume_stall_streak, 0, "熔断后计数归零")


class TestMcpIsolation(unittest.TestCase):
    """claude-code 子进程必须隔离用户 MCP(--strict-mcp-config),否则模型只看到
    Gmail/Calendar 等 MCP 工具、判定'没有工具写不了文件'而拒绝(run_f2a1eca9c6ca)。"""

    def _capture_cmd(self, session_id):
        from lambdagent.providers import claude_code_provider as ccp
        from lambdagent.providers.base import ProviderConfig

        class FakeResult:
            returncode = 0
            stdout = '{"result":"ok"}' if session_id is None else "ok"
            stderr = ""

        captured = {}

        def fake_run(cmd, **kw):
            captured["cmd"] = cmd
            return FakeResult()

        p = ccp.ClaudeCodeProvider.__new__(ccp.ClaudeCodeProvider)
        p.config = ProviderConfig(model="sonnet", timeout=600, extra={})
        p.claude_bin = "claude"
        p._session_id = session_id
        p._resume_count = 0
        p._resume_stall_streak = 0
        msgs = [{"role": "user", "content": "hi"}]
        with patch.object(ccp, "_run_with_idle_timeout", side_effect=fake_run):
            if session_id is None:
                p._call_new(msgs)
            else:
                p._call_resume(msgs)
        return captured["cmd"]

    def test_first_turn_isolates_mcp_and_native_tools(self):
        cmd = self._capture_cmd(session_id=None)
        self.assertIn("--strict-mcp-config", cmd)
        self.assertIn("--tools", cmd)
        # --tools 后面跟空串(禁原生工具)
        self.assertEqual(cmd[cmd.index("--tools") + 1], "")

    def test_resume_isolates_mcp_and_native_tools(self):
        cmd = self._capture_cmd(session_id="sess-1")
        self.assertIn("--strict-mcp-config", cmd)
        self.assertIn("--tools", cmd)


class TestAuthErrorTranslation(unittest.TestCase):
    """claude-code 401 必须翻译成人话提示、且 non-retryable;
    returncode 0 但 is_error 也要拦(否则把'Failed to authenticate'当答案返回)。"""

    def test_is_auth_error_detection(self):
        from lambdagent.providers import claude_code_provider as ccp
        self.assertTrue(ccp._is_auth_error("Failed to authenticate. API Error: 401 Invalid authentication credentials"))
        self.assertTrue(ccp._is_auth_error("oauth token has expired"))
        self.assertTrue(ccp._is_auth_error("401 Invalid authentication credentials"))
        # 不误伤:普通含 401 的内容(无 auth 语义)
        self.assertFalse(ccp._is_auth_error("found 401 matching files"))
        self.assertFalse(ccp._is_auth_error("the result is 401 widgets"))
        self.assertFalse(ccp._is_auth_error(""))

    def _provider(self):
        from lambdagent.providers import claude_code_provider as ccp
        from lambdagent.providers.base import ProviderConfig
        p = ccp.ClaudeCodeProvider.__new__(ccp.ClaudeCodeProvider)
        p.config = ProviderConfig(model="sonnet", timeout=600, extra={})
        p.claude_bin = "claude"
        p._session_id = None
        p._resume_count = 0
        p._resume_stall_streak = 0
        return p, ccp

    def test_first_turn_is_error_401_json_raises_hint(self):
        from lambdagent.providers.base import ProviderError
        p, ccp = self._provider()

        class FakeResult:
            returncode = 0
            stdout = '{"type":"result","is_error":true,"api_error_status":401,"result":"Failed to authenticate. API Error: 401 Invalid authentication credentials"}'
            stderr = ""

        with patch.object(ccp, "_run_with_idle_timeout", return_value=FakeResult()):
            with self.assertRaises(ProviderError) as ei:
                p._call_new([{"role": "user", "content": "hi"}])
        self.assertIn("登录已过期", str(ei.exception))
        self.assertFalse(ei.exception.retryable, "auth 错误必须 non-retryable")

    def test_first_turn_nonzero_exit_401_raises_hint(self):
        from lambdagent.providers.base import ProviderError
        p, ccp = self._provider()

        class FakeResult:
            returncode = 1
            stdout = "Failed to authenticate. API Error: 401 Invalid authentication credentials"
            stderr = ""

        with patch.object(ccp, "_run_with_idle_timeout", return_value=FakeResult()):
            with self.assertRaises(ProviderError) as ei:
                p._call_new([{"role": "user", "content": "hi"}])
        self.assertIn("登录", str(ei.exception))
        self.assertFalse(ei.exception.retryable)


class TestModelContextWindow(unittest.TestCase):
    """provider 必须按真实模型设 context_window — 治 qwen-max 被当 200000 → 历史超
    30720 → dashscope 400 'Range of input length should be [1, 30720]'。"""

    def test_qwen_max_is_30720(self):
        from lambdagent.providers.openai_compat_provider import OpenAICompatProvider as P
        self.assertEqual(P._resolve_context_window("qwen-max"), 30720)
        self.assertEqual(P._resolve_context_window("qwen-max-latest"), 30720)

    def test_other_models(self):
        from lambdagent.providers.openai_compat_provider import OpenAICompatProvider as P
        self.assertEqual(P._resolve_context_window("qwen-plus"), 131072)
        self.assertEqual(P._resolve_context_window("deepseek-chat"), 65536)
        self.assertIsNone(P._resolve_context_window("totally-unknown"))

    def test_provider_sets_context_window(self):
        from lambdagent.providers import create_provider
        p = create_provider("dashscope")  # default qwen-max
        self.assertEqual(p.context_window, 30720)


class TestTransientNetRetryable(unittest.TestCase):
    """瞬时网络断连必须判 retryable,否则一次抖动杀掉整条 run(dashscope
    'Remote end closed connection without response' 中断 59 分钟 run、写残 paper.tex)。"""

    def test_marker_and_type_detection(self):
        import http.client
        from lambdagent.providers.openai_compat_provider import _is_transient_net_error as f
        self.assertTrue(f(Exception("Remote end closed connection without response")))
        self.assertTrue(f(Exception("Connection reset by peer")))
        self.assertTrue(f(Exception("The read operation timed out")))
        self.assertTrue(f(ConnectionResetError("reset")))
        self.assertTrue(f(http.client.RemoteDisconnected("boom")))
        self.assertTrue(f(TimeoutError()))
        # 非网络错误不误判
        self.assertFalse(f(ValueError("bad json")))
        self.assertFalse(f(Exception("Range of input length should be [1, 30720]")))
