EvoScientist-Multi/tests/test_web_tool_registry.py

from __future__ import annotations

import pytest

from EvoScientist.llm.contracts import EvoRuntimeError
from EvoScientist.web_runtime import _ToolRegistryFenceMiddleware


def test_web_registry_describes_tavily_as_controlled_live_search(monkeypatch):
    monkeypatch.setenv("TAVILY_API_KEY", "test-key")

    from EvoScientist.web_runtime import web_tool_registry_manifest

    manifest, _revision = web_tool_registry_manifest()
    tavily = next(item for item in manifest if item["name"] == "tavily_search")

    assert "live public web" in tavily["description"].lower()
    assert "execute" in tavily["description"].lower()


def test_base_kwargs_install_tavily_on_main_agent(monkeypatch):
    import EvoScientist.EvoScientist as agent_module

    monkeypatch.setenv("TAVILY_API_KEY", "test-key")
    monkeypatch.setattr(agent_module, "_ensure_general_purpose_subagent", lambda _subs: None)
    monkeypatch.setattr(agent_module, "_inject_subagent_middleware", lambda *_args, **_kwargs: None)
    monkeypatch.setattr(agent_module, "_maybe_swap_async_subagents", lambda subs, *_args, **_kwargs: subs)
    monkeypatch.setattr(agent_module, "_configured_system_prompt", lambda _cfg: "prompt")
    monkeypatch.setattr("EvoScientist.utils.load_subagents", lambda *_args, **_kwargs: [])

    kwargs = agent_module._build_base_kwargs(
        object(),
        [],
        cfg=object(),
        chat_model=object(),
        workspace_dir="/workspace",
    )

    assert "tavily_search" in {getattr(tool, "name", "") for tool in kwargs["tools"]}


def _stub_agent_build(monkeypatch, agent_module, subagents=None):
    monkeypatch.setattr(agent_module, "_ensure_general_purpose_subagent", lambda _subs: None)
    monkeypatch.setattr(agent_module, "_inject_subagent_middleware", lambda *_args, **_kwargs: None)
    monkeypatch.setattr(agent_module, "_maybe_swap_async_subagents", lambda subs, *_args, **_kwargs: subs)
    monkeypatch.setattr(agent_module, "_configured_system_prompt", lambda _cfg: "prompt")
    monkeypatch.setattr(
        "EvoScientist.utils.load_subagents",
        lambda *_args, **_kwargs: list(subagents or []),
    )


@pytest.mark.parametrize(
    "reserved_name",
    [
        "skill_manager",
        "execute",
        "start_async_task",
        "check_async_task",
        "update_async_task",
        "cancel_async_task",
        "list_async_tasks",
    ],
)
def test_mcp_cannot_override_reserved_tool(monkeypatch, reserved_name):
    from types import SimpleNamespace

    import EvoScientist.EvoScientist as agent_module

    monkeypatch.delenv("TAVILY_API_KEY", raising=False)
    monkeypatch.setattr(
        agent_module,
        "_load_mcp_tools_cached",
        lambda **_kwargs: {"main": [SimpleNamespace(name=reserved_name)]},
    )
    _stub_agent_build(monkeypatch, agent_module)

    with pytest.raises(EvoRuntimeError, match="TOOL_REGISTRY_CONFLICT"):
        agent_module.load_mcp_and_build_kwargs(
            object(),
            [],
            cfg=object(),
            chat_model=object(),
            workspace_dir="/workspace",
        )


def test_mcp_cannot_duplicate_existing_subagent_tool(monkeypatch):
    from types import SimpleNamespace

    import EvoScientist.EvoScientist as agent_module

    existing = SimpleNamespace(name="shared_search")
    injected = SimpleNamespace(name="shared_search")
    monkeypatch.delenv("TAVILY_API_KEY", raising=False)
    monkeypatch.setattr(
        agent_module,
        "_load_mcp_tools_cached",
        lambda **_kwargs: {"research-agent": [injected]},
    )
    _stub_agent_build(
        monkeypatch,
        agent_module,
        subagents=[{"name": "research-agent", "tools": [existing]}],
    )

    with pytest.raises(EvoRuntimeError, match="TOOL_REGISTRY_CONFLICT"):
        agent_module.load_mcp_and_build_kwargs(
            object(),
            [],
            cfg=object(),
            chat_model=object(),
            workspace_dir="/workspace",
        )


def test_mcp_cannot_override_builtin_tavily(monkeypatch):
    from types import SimpleNamespace

    import EvoScientist.EvoScientist as agent_module

    monkeypatch.setenv("TAVILY_API_KEY", "test-key")
    monkeypatch.setattr(
        agent_module,
        "_load_mcp_tools_cached",
        lambda **_kwargs: {"main": [SimpleNamespace(name="tavily_search")]},
    )
    monkeypatch.setattr(agent_module, "_ensure_general_purpose_subagent", lambda _subs: None)
    monkeypatch.setattr(agent_module, "_inject_subagent_middleware", lambda *_args, **_kwargs: None)
    monkeypatch.setattr(agent_module, "_maybe_swap_async_subagents", lambda subs, *_args, **_kwargs: subs)
    monkeypatch.setattr(agent_module, "_configured_system_prompt", lambda _cfg: "prompt")
    monkeypatch.setattr("EvoScientist.utils.load_subagents", lambda *_args, **_kwargs: [])

    with pytest.raises(EvoRuntimeError, match="TOOL_REGISTRY_CONFLICT"):
        agent_module.load_mcp_and_build_kwargs(
            object(),
            [],
            cfg=object(),
            chat_model=object(),
            workspace_dir="/workspace",
        )


def test_same_mcp_tool_can_be_routed_to_multiple_agents(monkeypatch):
    from types import SimpleNamespace

    import EvoScientist.EvoScientist as agent_module

    shared_main = SimpleNamespace(name="shared_search")
    shared_research = SimpleNamespace(name="shared_search")
    monkeypatch.delenv("TAVILY_API_KEY", raising=False)
    monkeypatch.setattr(
        agent_module,
        "_load_mcp_tools_cached",
        lambda **_kwargs: {
            "main": [shared_main],
            "research-agent": [shared_research],
        },
    )
    monkeypatch.setattr(agent_module, "_ensure_general_purpose_subagent", lambda _subs: None)
    monkeypatch.setattr(agent_module, "_inject_subagent_middleware", lambda *_args, **_kwargs: None)
    monkeypatch.setattr(agent_module, "_maybe_swap_async_subagents", lambda subs, *_args, **_kwargs: subs)
    monkeypatch.setattr(agent_module, "_configured_system_prompt", lambda _cfg: "prompt")
    monkeypatch.setattr(
        "EvoScientist.utils.load_subagents",
        lambda *_args, **_kwargs: [
            {"name": "research-agent", "tools": []}
        ],
    )

    kwargs = agent_module.load_mcp_and_build_kwargs(
        object(),
        [],
        cfg=object(),
        chat_model=object(),
        workspace_dir="/workspace",
    )

    assert shared_main in kwargs["tools"]
    research = next(
        subagent for subagent in kwargs["subagents"]
        if subagent["name"] == "research-agent"
    )
    assert shared_research in research["tools"]


def test_tool_dispatch_fence_rejects_changed_registry(monkeypatch):
    monkeypatch.setattr(
        "EvoScientist.web_runtime.web_tool_registry_manifest",
        lambda: ((), "registry-v2"),
    )
    middleware = _ToolRegistryFenceMiddleware("registry-v1")

    with pytest.raises(EvoRuntimeError, match="TOOL_REGISTRY_STALE"):
        middleware._require_current()