"""Tests for server graph MCP loading behavior."""

from __future__ import annotations

import importlib
import os
import sys
import threading
from types import ModuleType, SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from deepagents_code._env_vars import SERVER_ENV_PREFIX
from deepagents_code._server_config import ServerConfig


def _import_fresh_server_graph() -> ModuleType:
    """Import `deepagents_code.server_graph` from a clean module state."""
    sys.modules.pop("deepagents_code.server_graph", None)
    return importlib.import_module("deepagents_code.server_graph")


def _module_with_attrs(name: str, **attrs: object) -> ModuleType:
    """Create a module stub with dynamically assigned attributes."""
    module = ModuleType(name)
    for key, value in attrs.items():
        setattr(module, key, value)
    return module


class TestServerGraph:
    """Tests for server-mode graph bootstrap."""

    async def test_make_graph_caches_first_constructed_graph(self) -> None:
        """Repeated factory access should preserve process-lifetime resources."""
        graph_obj = object()
        module = _import_fresh_server_graph()

        with patch.object(
            module, "_make_graph", new=AsyncMock(return_value=graph_obj)
        ) as make_graph:
            assert await module.make_graph() is graph_obj
            assert await module.make_graph() is graph_obj

        make_graph.assert_awaited_once_with()

    async def test_make_graph_emits_marker_and_exits_on_failure(
        self, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """A construction failure must emit the startup marker, then exit non-zero."""
        from deepagents_code._startup_error import STARTUP_ERROR_MARKER

        module = _import_fresh_server_graph()

        with (
            patch.object(
                module,
                "_make_graph",
                new=AsyncMock(side_effect=ValueError("boom: bad model")),
            ),
            pytest.raises(SystemExit) as exc_info,
        ):
            await module.make_graph()

        assert exc_info.value.code == 1
        captured = capsys.readouterr()
        assert f"{STARTUP_ERROR_MARKER}ValueError: boom: bad model" in captured.err

    async def test_auto_discovery_loads_mcp_without_explicit_config(self) -> None:
        """Server mode should auto-discover MCP configs when the graph is built."""
        graph_obj = object()
        model_obj = object()
        fetch_tool = object()
        thread_tool = object()
        mcp_tool = object()
        mcp_server_info = [SimpleNamespace(name="docs")]
        loop_thread_id = threading.get_ident()
        create_cli_agent_thread_ids: list[int] = []
        create_model_thread_ids: list[int] = []
        warm_import_thread_ids: list[int] = []

        def create_cli_agent_side_effect(**_: object) -> tuple[object, object]:
            create_cli_agent_thread_ids.append(threading.get_ident())
            return graph_obj, object()

        def create_model_side_effect(*_: object, **__: object) -> object:
            create_model_thread_ids.append(threading.get_ident())
            return model_result

        def warm_import_side_effect() -> None:
            warm_import_thread_ids.append(threading.get_ident())

        create_cli_agent = MagicMock(side_effect=create_cli_agent_side_effect)
        agent_module = _module_with_attrs(
            "deepagents_code.agent",
            DEFAULT_AGENT_NAME="agent",
            create_cli_agent=create_cli_agent,
            load_async_subagents=MagicMock(return_value=None),
        )

        model_result = SimpleNamespace(
            model=model_obj,
            apply_to_settings=MagicMock(),
        )
        config_module = _module_with_attrs(
            "deepagents_code.config",
            create_model=MagicMock(side_effect=create_model_side_effect),
            settings=SimpleNamespace(
                has_tavily=False,
                reload_from_environment=MagicMock(),
            ),
        )

        tools_module = _module_with_attrs(
            "deepagents_code.tools",
            fetch_url=fetch_tool,
            get_current_thread_id=thread_tool,
            web_search=object(),
        )

        class FakeSessionManager:
            async def cleanup(self) -> None:
                return None

        resolve_mcp_tools = AsyncMock(return_value=([mcp_tool], None, mcp_server_info))
        mcp_module = _module_with_attrs(
            "deepagents_code.mcp_tools",
            MCPSessionManager=FakeSessionManager,
            resolve_and_load_mcp_tools=resolve_mcp_tools,
        )

        config = ServerConfig(no_mcp=False)
        env_overrides = {}
        for suffix, value in config.to_env().items():
            if value is not None:
                env_overrides[f"{SERVER_ENV_PREFIX}{suffix}"] = value

        with (
            patch.dict(os.environ, env_overrides, clear=False),
            patch.dict(
                sys.modules,
                {
                    "deepagents_code.agent": agent_module,
                    "deepagents_code.config": config_module,
                    "deepagents_code.tools": tools_module,
                    "deepagents_code.mcp_tools": mcp_module,
                },
            ),
            patch(
                "deepagents_code.project_utils.get_server_project_context",
                return_value=None,
            ),
        ):
            for suffix in (
                "MCP_CONFIG_PATH",
                "TRUST_PROJECT_MCP",
                "CWD",
                "PROJECT_ROOT",
            ):
                os.environ.pop(f"{SERVER_ENV_PREFIX}{suffix}", None)

            module = _import_fresh_server_graph()
            resolve_mcp_tools.assert_not_awaited()
            with patch.object(
                module,
                "_warm_mcp_adapter_imports",
                side_effect=warm_import_side_effect,
            ):
                assert await module.make_graph() is graph_obj

        resolve_mcp_tools.assert_awaited_once()
        assert warm_import_thread_ids
        assert warm_import_thread_ids[0] != loop_thread_id
        assert create_cli_agent_thread_ids
        assert create_cli_agent_thread_ids[0] != loop_thread_id
        # `create_model` must run off the loop thread: it does blocking disk IO
        # for some providers (e.g. the `openai_codex` token store calls
        # `os.mkdir`), which `blockbuster` rejects on the server event loop.
        assert create_model_thread_ids
        assert create_model_thread_ids[0] != loop_thread_id
        kwargs = resolve_mcp_tools.await_args_list[0].kwargs
        assert kwargs["explicit_config_path"] is None
        assert kwargs["no_mcp"] is False
        assert kwargs["trust_project_mcp"] is None
        assert kwargs["project_context"] is None
        assert kwargs["stateless"] is True
        assert isinstance(kwargs["session_manager"], FakeSessionManager)
        create_cli_agent.assert_called_once_with(
            model=model_obj,
            assistant_id="agent",
            tools=[fetch_tool, thread_tool, mcp_tool],
            sandbox=None,
            sandbox_type=None,
            system_prompt=None,
            interactive=True,
            auto_approve=False,
            interrupt_shell_only=False,
            shell_allow_list=None,
            enable_ask_user=False,
            enable_memory=True,
            enable_skills=True,
            enable_shell=True,
            enable_interpreter=False,
            mcp_server_info=mcp_server_info,
            cwd=None,
            project_context=None,
            async_subagents=None,
        )

    async def test_build_tools_skips_mcp_when_disabled(self) -> None:
        """`no_mcp=True` should not warm imports or call the MCP resolver."""
        fetch_tool = object()
        thread_tool = object()
        resolve_mcp_tools = AsyncMock()
        config_module = _module_with_attrs(
            "deepagents_code.config",
            settings=SimpleNamespace(has_tavily=False),
        )
        tools_module = _module_with_attrs(
            "deepagents_code.tools",
            fetch_url=fetch_tool,
            get_current_thread_id=thread_tool,
            web_search=object(),
        )
        mcp_module = _module_with_attrs(
            "deepagents_code.mcp_tools",
            resolve_and_load_mcp_tools=resolve_mcp_tools,
        )

        with patch.dict(
            sys.modules,
            {
                "deepagents_code.config": config_module,
                "deepagents_code.tools": tools_module,
                "deepagents_code.mcp_tools": mcp_module,
            },
        ):
            module = _import_fresh_server_graph()
            warm_imports = MagicMock(side_effect=AssertionError("MCP warmup ran"))
            with patch.object(module, "_warm_mcp_adapter_imports", warm_imports):
                tools, mcp_server_info = await module._build_tools(
                    ServerConfig(no_mcp=True),
                    None,
                )

        assert tools == [fetch_tool, thread_tool]
        assert mcp_server_info is None
        warm_imports.assert_not_called()
        resolve_mcp_tools.assert_not_awaited()

    async def test_interpreter_settings_apply_before_agent_construction(self) -> None:
        """Server config settings writes should be visible to `create_cli_agent`."""
        graph_obj = object()
        model_obj = object()
        observed: dict[str, object] = {}

        def create_cli_agent_side_effect(**_: object) -> tuple[object, object]:
            from deepagents_code.config import settings

            observed["interpreter_ptc"] = settings.interpreter_ptc
            observed["acknowledge"] = settings.interpreter_ptc_acknowledge_unsafe
            observed["enable_interpreter"] = settings.enable_interpreter
            return graph_obj, object()

        settings_obj = SimpleNamespace(
            has_tavily=False,
            interpreter_ptc=None,
            interpreter_ptc_acknowledge_unsafe=False,
            enable_interpreter=False,
        )
        config_module = _module_with_attrs(
            "deepagents_code.config",
            create_model=MagicMock(
                return_value=SimpleNamespace(
                    model=model_obj,
                    apply_to_settings=MagicMock(),
                ),
            ),
            settings=settings_obj,
        )
        agent_module = _module_with_attrs(
            "deepagents_code.agent",
            create_cli_agent=MagicMock(side_effect=create_cli_agent_side_effect),
            load_async_subagents=MagicMock(return_value=None),
        )
        tools_module = _module_with_attrs(
            "deepagents_code.tools",
            fetch_url=object(),
            get_current_thread_id=object(),
            web_search=object(),
        )
        config = ServerConfig(
            no_mcp=True,
            enable_interpreter=True,
            interpreter_ptc=["js_eval"],
            interpreter_ptc_acknowledge_unsafe=True,
        )
        env_overrides = {
            f"{SERVER_ENV_PREFIX}{suffix}": value
            for suffix, value in config.to_env().items()
            if value is not None
        }

        with (
            patch.dict(os.environ, env_overrides, clear=False),
            patch.dict(
                sys.modules,
                {
                    "deepagents_code.agent": agent_module,
                    "deepagents_code.config": config_module,
                    "deepagents_code.tools": tools_module,
                },
            ),
            patch(
                "deepagents_code.project_utils.get_server_project_context",
                return_value=None,
            ),
        ):
            module = _import_fresh_server_graph()
            assert await module.make_graph() is graph_obj

        assert observed == {
            "interpreter_ptc": ["js_eval"],
            "acknowledge": True,
            "enable_interpreter": True,
        }

    async def test_mcp_adapter_warmup_runs_before_mcp_resolver(self) -> None:
        """MCP imports must be warmed before resolver imports adapter modules."""
        events: list[str] = []
        fetch_tool = object()
        thread_tool = object()

        def resolve_mcp_tools(
            **_: object,
        ) -> tuple[list[object], None, list[object]]:
            events.append("resolver")
            return [], None, []

        config_module = _module_with_attrs(
            "deepagents_code.config",
            settings=SimpleNamespace(has_tavily=False),
        )
        tools_module = _module_with_attrs(
            "deepagents_code.tools",
            fetch_url=fetch_tool,
            get_current_thread_id=thread_tool,
            web_search=object(),
        )

        class FakeSessionManager:
            pass

        mcp_module = _module_with_attrs(
            "deepagents_code.mcp_tools",
            MCPSessionManager=FakeSessionManager,
            resolve_and_load_mcp_tools=AsyncMock(side_effect=resolve_mcp_tools),
        )

        with patch.dict(
            sys.modules,
            {
                "deepagents_code.config": config_module,
                "deepagents_code.tools": tools_module,
                "deepagents_code.mcp_tools": mcp_module,
            },
        ):
            module = _import_fresh_server_graph()

            def warm_imports() -> None:
                events.append("warmup")

            with patch.object(module, "_warm_mcp_adapter_imports", warm_imports):
                tools, mcp_server_info = await module._build_tools(
                    ServerConfig(no_mcp=False),
                    None,
                )

        assert events == ["warmup", "resolver"]
        assert tools == [fetch_tool, thread_tool]
        assert mcp_server_info == []


class TestStartupErrorMarker:
    """`emit_startup_failure` must produce the parser marker on stderr.

    The marker is the contract `wait_for_server_healthy` parses to surface
    a one-line summary instead of "Server process exited with code N".
    """

    def test_emits_marker_with_type_and_summary(
        self,
        capsys: pytest.CaptureFixture[str],
    ) -> None:
        from deepagents_code._startup_error import (
            STARTUP_ERROR_MARKER,
            emit_startup_failure,
        )

        emit_startup_failure(ValueError("boom: details"))
        captured = capsys.readouterr()
        assert f"{STARTUP_ERROR_MARKER}ValueError: boom: details" in captured.err
        assert "Failed to initialize server graph: boom: details" in captured.err

    def test_marker_collapses_multiline_exception(
        self,
        capsys: pytest.CaptureFixture[str],
    ) -> None:
        from deepagents_code._startup_error import (
            STARTUP_ERROR_MARKER,
            emit_startup_failure,
        )

        emit_startup_failure(ValueError("first line\nsecond line"))
        captured = capsys.readouterr()
        marker_line = next(
            line
            for line in captured.err.splitlines()
            if line.startswith(STARTUP_ERROR_MARKER)
        )
        assert marker_line == f"{STARTUP_ERROR_MARKER}ValueError: first line"

    def test_marker_handles_empty_exception_message(
        self,
        capsys: pytest.CaptureFixture[str],
    ) -> None:
        from deepagents_code._startup_error import (
            STARTUP_ERROR_MARKER,
            emit_startup_failure,
        )

        emit_startup_failure(RuntimeError())
        captured = capsys.readouterr()
        assert f"{STARTUP_ERROR_MARKER}RuntimeError: <no message>" in captured.err
