fix(xai): reject FileSearch plus collections_search or file_search

livekit-plugins/livekit-plugins-xai/livekit/plugins/xai/realtime/realtime_model.py

@@ -45,7 +45,7 @@
 from livekit.plugins.openai.realtime.realtime_model import _DiscardedGeneration
 
 from ..log import logger
-from ..tools import XAITool
+from ..tools import XAITool, _raise_if_xai_tool_reserved_name_conflict
 from ..types import GrokRealtimeModels, GrokVoices
 
 XAI_BASE_URL = "wss://api.x.ai/v1/realtime"
@@ -231,6 +231,7 @@ def _wrap_session_update(
         return super()._wrap_session_update(event_id=event_id, session=session)
 
     def _create_tools_update_event(self, tools: list[llm.Tool]) -> dict[str, Any]:
+        _raise_if_xai_tool_reserved_name_conflict(tools)
         event = super()._create_tools_update_event(tools)
         xai_tools: list[dict[str, Any]] = []
         for tool in tools:

livekit-plugins/livekit-plugins-xai/livekit/plugins/xai/tools.py

@@ -1,20 +1,33 @@
 from abc import abstractmethod
+from collections.abc import Sequence
 from dataclasses import dataclass, field
-from typing import Any
+from typing import Any, ClassVar
 
 from livekit.agents import ProviderTool
+from livekit.agents.llm.tool_context import FunctionTool, RawFunctionTool, Tool
 
 
 class XAITool(ProviderTool):
     """Base class for xAI server-side provider tools."""
 
+    # function names the server answers to once this tool is enabled. Measured against
+    # grok-voice-latest: a client function of the same name makes the first
+    # response.create return server_error.
+    _reserved_function_names: ClassVar[frozenset[str]] = frozenset()
+
     @abstractmethod
     def to_dict(self) -> dict[str, Any]: ...
 
 
 @dataclass
 class WebSearch(XAITool):
-    """Enable web search tool for real-time internet searches."""
+    """Enable web search tool for real-time internet searches.
+
+    Do not also register a function named ``web_search`` or ``browse_page``;
+    xAI's WebSearch already uses those names.
+    """
+
+    _reserved_function_names: ClassVar[frozenset[str]] = frozenset({"web_search", "browse_page"})
 
     def __post_init__(self) -> None:
         super().__init__(id="xai_web_search")
@@ -25,8 +38,16 @@ def to_dict(self) -> dict[str, Any]:
 
 @dataclass
 class XSearch(XAITool):
-    """Enable X (Twitter) search tool for searching posts."""
+    """Enable X (Twitter) search tool for searching posts.
+
+    Do not also register a function named ``x_keyword_search``,
+    ``x_semantic_search``, ``x_user_search``, or ``x_thread_fetch``;
+    xAI's XSearch already uses those names.
+    """
 
+    _reserved_function_names: ClassVar[frozenset[str]] = frozenset(
+        {"x_keyword_search", "x_semantic_search", "x_user_search", "x_thread_fetch"}
+    )
     allowed_x_handles: list[str] | None = None
 
     def __post_init__(self) -> None:
@@ -41,8 +62,15 @@ def to_dict(self) -> dict[str, Any]:
 
 @dataclass
 class FileSearch(XAITool):
-    """Enable file search tool for searching uploaded document collections."""
+    """Enable file search tool for searching uploaded document collections.
 
+    Do not also register a function named ``collections_search`` or ``file_search``;
+    xAI's FileSearch already uses those names.
+    """
+
+    _reserved_function_names: ClassVar[frozenset[str]] = frozenset(
+        {"collections_search", "file_search"}
+    )
     vector_store_ids: list[str] = field(default_factory=list)
     max_num_results: int | None = None
 
@@ -58,3 +86,21 @@ def to_dict(self) -> dict[str, Any]:
             result["max_num_results"] = self.max_num_results
 
         return result
+
+
+def _raise_if_xai_tool_reserved_name_conflict(tools: Sequence[Tool]) -> None:
+    """Reject a client function whose name an enabled xAI provider tool already answers to."""
+    fnc_names = {
+        tool.info.name for tool in tools if isinstance(tool, (FunctionTool, RawFunctionTool))
+    }
+    for tool in tools:
+        if not isinstance(tool, XAITool):
+            continue
+        if conflicts := sorted(tool._reserved_function_names & fnc_names):
+            names = ", ".join(repr(name) for name in conflicts)
+            raise ValueError(
+                f"xAI {type(tool).__name__} already uses the function name(s) {names}. "
+                "Rename or remove them; a client function that shadows a provider tool's "
+                "own name makes grok-voice-latest return server_error/internal_error on "
+                "the first response.create."
+            )

tests/test_realtime/test_xai_realtime_model.py

@@ -31,10 +31,121 @@
     RealtimeModel,
     RealtimeSession,
 )
+from livekit.plugins.xai.tools import (
+    FileSearch,
+    WebSearch,
+    XSearch,
+    _raise_if_xai_tool_reserved_name_conflict,
+)
 
 pytestmark = pytest.mark.unit
 
 
+def _named(name: str) -> llm.FunctionTool:
+    @llm.function_tool(name=name)
+    async def tool() -> str:
+        return "ok"
+
+    return tool
+
+
+def _raw(name: str) -> llm.RawFunctionTool:
+    @llm.function_tool(
+        raw_schema={
+            "name": name,
+            "description": "test",
+            "parameters": {"type": "object", "properties": {}},
+        }
+    )
+    async def tool() -> str:
+        return "ok"
+
+    return tool
+
+
+@llm.function_tool
+async def collections_search() -> str:
+    """A function that collides with xAI FileSearch."""
+    return "hit"
+
+
+_RESERVED_PAIRS: list[tuple[llm.ProviderTool, str]] = [
+    (WebSearch(), "web_search"),
+    (WebSearch(), "browse_page"),
+    (XSearch(), "x_keyword_search"),
+    (XSearch(), "x_semantic_search"),
+    (XSearch(), "x_user_search"),
+    (XSearch(), "x_thread_fetch"),
+    (FileSearch(), "collections_search"),
+    (FileSearch(), "file_search"),
+]
+
+
+@pytest.mark.parametrize(
+    ("provider_tool", "function_name"),
+    _RESERVED_PAIRS,
+    ids=[f"{t.__class__.__name__}-{name}" for t, name in _RESERVED_PAIRS],
+)
+def test_provider_tool_plus_reserved_function_raises(
+    provider_tool: llm.ProviderTool, function_name: str
+) -> None:
+    with pytest.raises(ValueError, match="Rename or remove"):
+        _raise_if_xai_tool_reserved_name_conflict([provider_tool, _named(function_name)])
+
+
+def test_file_search_plus_plain_collections_search_raises() -> None:
+    with pytest.raises(ValueError, match="Rename or remove"):
+        _raise_if_xai_tool_reserved_name_conflict([FileSearch(), collections_search])
+
+
+def test_file_search_plus_raw_collections_search_raises() -> None:
+    with pytest.raises(ValueError, match="Rename or remove"):
+        _raise_if_xai_tool_reserved_name_conflict([FileSearch(), _raw("collections_search")])
+
+
+def test_file_search_plus_nested_toolset_reserved_name_raises() -> None:
+    toolset = llm.Toolset(id="nested", tools=[collections_search])
+    tools = llm.ToolContext([FileSearch(), toolset]).flatten()
+    with pytest.raises(ValueError, match="Rename or remove"):
+        _raise_if_xai_tool_reserved_name_conflict(tools)
+
+
+_HARMLESS_PAIRS: list[tuple[llm.ProviderTool | None, str]] = [
+    (WebSearch(), "view_image"),
+    (WebSearch(), "harmless_control"),
+    (WebSearch(), "collections_search"),
+    (XSearch(), "x_search"),
+    (XSearch(), "harmless_control"),
+    (FileSearch(), "view_document"),
+    (FileSearch(), "harmless_control"),
+    (FileSearch(), "web_search"),
+    (None, "collections_search"),
+    (None, "file_search"),
+    (None, "web_search"),
+    (None, "browse_page"),
+]
+
+
+@pytest.mark.parametrize(
+    ("provider_tool", "function_name"),
+    _HARMLESS_PAIRS,
+    ids=[
+        f"{t.__class__.__name__ if t is not None else 'none'}-{name}" for t, name in _HARMLESS_PAIRS
+    ],
+)
+def test_non_reserved_combinations_are_ok(
+    provider_tool: llm.ProviderTool | None, function_name: str
+) -> None:
+    tools: list[llm.Tool] = [_named(function_name)]
+    if provider_tool is not None:
+        tools.insert(0, provider_tool)
+    _raise_if_xai_tool_reserved_name_conflict(tools)
+
+
+def test_file_search_alone_is_ok() -> None:
+    _raise_if_xai_tool_reserved_name_conflict([FileSearch()])
+
+
 def test_default_model_is_grok_voice_latest() -> None:
     assert XAI_DEFAULT_MODEL == "grok-voice-latest"
     model = RealtimeModel(api_key="fake")