fix(agent): reject duplicate client tool names

src/openrouter_agent/model_result.py

@@ -29,6 +29,7 @@
     StepResult,
     Tool,
     UnsentToolResult,
+    assert_unique_client_tool_names,
     get_tool_function,
     is_auto_resolvable_tool,
     is_client_tool,
@@ -100,6 +101,7 @@ def _find_latest_user_string_index(items: Sequence[Any]) -> int:
 
 class ModelResult:
     def __init__(self, options: Mapping[str, Any]) -> None:
+        assert_unique_client_tool_names(options.get("tools"))
         self.options = dict(options)
         self._run_task: Optional[asyncio.Task[Any]] = None
         self._final_response: Any = None

src/openrouter_agent/tool_types.py

@@ -2,7 +2,7 @@
 
 from dataclasses import dataclass
 from enum import Enum
-from typing import Any, Callable, Dict, List, Mapping, Optional, Protocol, Sequence, Union
+from typing import Any, Callable, Dict, List, Mapping, Optional, Protocol, Sequence, Set, Union
 
 from typing_extensions import Literal, TypeAlias, TypedDict
 
@@ -275,6 +275,36 @@ def get_tool_function(tool: Mapping[str, Any]) -> Dict[str, Any]:
     return value if isinstance(value, dict) else {}
 
 
+def find_duplicate_client_tool_name(tools: Sequence[Mapping[str, Any]]) -> Optional[str]:
+    """Return the first client tool name that appears more than once, or `None`.
+    Server tools have no client name and are skipped."""
+    seen: Set[str] = set()
+    for candidate in tools:
+        if is_server_tool(candidate):
+            continue
+        name = get_tool_function(candidate).get("name")
+        if not isinstance(name, str):
+            continue
+        if name in seen:
+            return name
+        seen.add(name)
+    return None
+
+
+def assert_unique_client_tool_names(tools: Optional[Sequence[Mapping[str, Any]]]) -> None:
+    """Raise when two client tools share a name. Calls are dispatched by name, so a
+    duplicate would let whichever tool is listed first receive arguments the model
+    (or an approver) meant for the other."""
+    if not tools:
+        return
+    name = find_duplicate_client_tool_name(tools)
+    if name is not None:
+        raise ValueError(
+            f'Duplicate tool name "{name}": every tool passed to call_model must have a unique name. '
+            "Rename one, or prefix the names of MCP tools when combining them with local tools."
+        )
+
+
 def has_execute_function(tool: Mapping[str, Any]) -> bool:
     fn = get_tool_function(tool)
     return is_client_tool(tool) and callable(fn.get("execute"))

tests/unit/test_duplicate_tool_names.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+import pytest
+
+from openrouter_agent import call_model, tool
+from openrouter_agent.model_result import ModelResult
+from tests._fixtures import MemoryStateAccessor, QueuedClient, text_response, tool_call_response
+
+
+def _tool(name: str, hits: List[Dict[str, Any]], **kwargs: Any) -> Any:
+    def execute(params: Dict[str, Any], ctx: Any) -> Dict[str, Any]:
+        hits.append(params)
+        return {"ok": True}
+
+    return tool(name=name, input_schema=dict, output_schema=dict, execute=execute, **kwargs)
+
+
+def test_rejects_two_client_tools_with_the_same_name() -> None:
+    first = _tool("read_value", [])
+    second = _tool("read_value", [])
+
+    with pytest.raises(ValueError, match='Duplicate tool name "read_value"'):
+        ModelResult({"model": "test", "input": "hi", "tools": [first, second]})
+    with pytest.raises(ValueError, match='Duplicate tool name "read_value"'):
+        call_model(QueuedClient([]), {"model": "test", "input": "hi", "tools": [first, second]})
+
+
+def test_allows_distinct_names() -> None:
+    ModelResult({"model": "test", "input": "hi", "tools": [_tool("a", []), _tool("b", [])]})
+
+
+async def test_approved_call_does_not_run_on_a_same_named_tool_added_before_resume() -> None:
+    state = MemoryStateAccessor()
+    approved_hits: List[Dict[str, Any]] = []
+    other_hits: List[Dict[str, Any]] = []
+    approved = _tool("read_value", approved_hits, require_approval=True)
+    other = _tool("read_value", other_hits)
+
+    first = call_model(
+        QueuedClient([tool_call_response("resp_1", "read_value", call_id="call_1")]),
+        {"model": "test", "input": "read it", "tools": [approved], "state": state},
+    )
+    await first.get_response()
+    assert state.stored is not None
+    assert state.stored.status == "awaiting_approval"
+
+    with pytest.raises(ValueError, match="Duplicate tool name"):
+        call_model(
+            QueuedClient([text_response("resp_done", "done")]),
+            {
+                "model": "test",
+                "input": [],
+                "tools": [other, approved],
+                "state": state,
+                "approve_tool_calls": ["call_1"],
+            },
+        )
+
+    assert other_hits == []
+    assert approved_hits == []
+    assert state.stored.status == "awaiting_approval"