"""Unit tests for navi.core.tool_executor."""
import asyncio
import json
import pytest
from navi.core import tasks as tasks_mod
from navi.core.registry import ToolRegistry
from navi.core.tool_executor import ToolExecutor
from navi.llm.base import Message, ToolCallRequest
from navi.tools._internal.base import ToolResult
from tests.conftest_factory import FakeTool
def patch_settings(monkeypatch, **overrides):
import navi.config as config_mod
from navi.config import Settings
new_settings = Settings(**overrides)
monkeypatch.setattr(config_mod, "settings", new_settings)
return new_settings
class _Ctx:
"""Minimal ToolContext stand-in carrying only the session id."""
def __init__(self, session_id: str = "s1") -> None:
self.session_id = session_id
class RecordingTool:
"""Fake tool that captures its arguments before answering."""
def __init__(self, name: str, output: str = "ok") -> None:
self.name = name
self.output = output
self.calls: list[dict] = []
async def execute(self, arguments: dict, ctx=None) -> ToolResult:
self.calls.append(dict(arguments))
return ToolResult(success=True, output=self.output)
class TestToolExecutorMcpAliases:
async def test_executes_bare_mcp_tool_alias(self):
registry = ToolRegistry()
tool = FakeTool("mcp__gnexus_book__search_docs")
registry.register(tool, builtin=True)
executor = ToolExecutor(registry)
messages, images = await executor._execute_tool_calls(
[ToolCallRequest(id="1", name="search_docs", arguments={"query": "git"})],
[tool],
)
assert images == []
assert messages[0].name == "mcp__gnexus_book__search_docs"
assert messages[0].content == "executed mcp__gnexus_book__search_docs"
async def test_executes_mcp_tool_alias_with_dash_variant(self):
registry = ToolRegistry()
tool = FakeTool("mcp__gnexus_book__search_docs")
registry.register(tool, builtin=True)
executor = ToolExecutor(registry)
messages, images = await executor._execute_tool_calls(
[ToolCallRequest(id="1", name="mcp__gnexus-book__search_docs", arguments={"query": "git"})],
[tool],
)
assert images == []
assert messages[0].name == "mcp__gnexus_book__search_docs"
assert messages[0].content == "executed mcp__gnexus_book__search_docs"
async def test_executes_old_underscore_format_fallback(self):
registry = ToolRegistry()
tool = FakeTool("mcp__gnexus_book__search_docs")
registry.register(tool, builtin=True)
executor = ToolExecutor(registry)
messages, images = await executor._execute_tool_calls(
[ToolCallRequest(id="1", name="mcp_gnexus_book_search_docs", arguments={"query": "git"})],
[tool],
)
assert images == []
assert messages[0].name == "mcp__gnexus_book__search_docs"
assert messages[0].content == "executed mcp__gnexus_book__search_docs"
async def test_executes_legacy_colon_format_fallback(self):
registry = ToolRegistry()
tool = FakeTool("mcp__gnexus_book__search_docs")
registry.register(tool, builtin=True)
executor = ToolExecutor(registry)
messages, images = await executor._execute_tool_calls(
[ToolCallRequest(id="1", name="mcp:gnexus_book:search_docs", arguments={"query": "git"})],
[tool],
)
assert images == []
assert messages[0].name == "mcp__gnexus_book__search_docs"
assert messages[0].content == "executed mcp__gnexus_book__search_docs"
class TestBackgroundInterception:
@pytest.fixture(autouse=True)
def _fresh_manager(self, monkeypatch):
from navi.core.tasks import TaskManager
manager = TaskManager()
monkeypatch.setattr(tasks_mod, "_manager", manager)
self.manager = manager
yield manager
for job in list(manager._jobs.values()):
if job.task is not None and not job.task.done():
job.task.cancel()
async def test_background_flag_detaches_call(self, monkeypatch):
patch_settings(monkeypatch, backgroundable_tools="slow_tool")
tool = RecordingTool("slow_tool", output="long result")
registry = ToolRegistry()
registry.register(tool, builtin=True)
executor = ToolExecutor(registry)
event, msg, image = await executor._execute_one(
ToolCallRequest(id="tc1", name="slow_tool",
arguments={"query": "x", "background": True}),
{"slow_tool": tool},
ctx=_Ctx(),
)
# immediate synthetic result with a task id
assert msg.metadata.get("background") is True
payload = json.loads(msg.content)
task_id = payload["task_id"]
assert task_id.startswith("bt-")
assert payload["status"] == "running"
assert "tasks check" in payload["hint"]
assert event.success is True
# background flag is stripped from the args passed to the tool
job = self.manager.get(task_id, "s1")
assert job is not None
assert "background" not in job.args
await job.done.wait()
assert tool.calls == [{"query": "x"}]
assert job.result.output == "long result"
async def test_non_backgroundable_tool_runs_inline(self, monkeypatch):
patch_settings(monkeypatch, backgroundable_tools="other_tool")
tool = RecordingTool("slow_tool")
registry = ToolRegistry()
registry.register(tool, builtin=True)
executor = ToolExecutor(registry)
event, msg, _ = await executor._execute_one(
ToolCallRequest(id="tc1", name="slow_tool",
arguments={"query": "x", "background": True}),
{"slow_tool": tool},
ctx=_Ctx(),
)
assert msg.content == "ok" # ran inline, real output
assert msg.metadata.get("background") is None
assert tool.calls == [{"query": "x", "background": True}] # flag intact
async def test_background_false_runs_inline(self, monkeypatch):
patch_settings(monkeypatch, backgroundable_tools="slow_tool")
tool = RecordingTool("slow_tool")
executor = ToolExecutor({"slow_tool": tool})
_, msg, _ = await executor._execute_one(
ToolCallRequest(id="tc1", name="slow_tool",
arguments={"background": False}),
{"slow_tool": tool},
ctx=_Ctx(),
)
assert msg.content == "ok"
assert self.manager.list("s1") == []
async def test_cap_rejection_is_inline_failure(self, monkeypatch):
patch_settings(monkeypatch, backgroundable_tools="slow_tool",
tasks_max_per_session=0)
tool = RecordingTool("slow_tool")
executor = ToolExecutor({"slow_tool": tool})
_, msg, _ = await executor._execute_one(
ToolCallRequest(id="tc1", name="slow_tool",
arguments={"background": True}),
{"slow_tool": tool},
ctx=_Ctx(),
)
assert "Cannot run in background" in msg.content
assert msg.metadata.get("background") is None
assert tool.calls == [] # was not executed
async def test_terminal_open_not_hijacked(self, monkeypatch):
patch_settings(monkeypatch, backgroundable_tools="terminal")
tool = RecordingTool("terminal")
executor = ToolExecutor({"terminal": tool})
_, msg, _ = await executor._execute_one(
ToolCallRequest(id="tc1", name="terminal",
arguments={"action": "open", "background": True}),
{"terminal": tool},
ctx=_Ctx(),
)
# native terminal open(background=true) semantics preserved
assert msg.content == "ok"
assert tool.calls == [{"action": "open", "background": True}]
assert self.manager.list("s1") == []
async def test_terminal_run_is_backgroundable(self, monkeypatch):
patch_settings(monkeypatch, backgroundable_tools="terminal")
tool = RecordingTool("terminal")
executor = ToolExecutor({"terminal": tool})
_, msg, _ = await executor._execute_one(
ToolCallRequest(id="tc1", name="terminal",
arguments={"action": "run", "command": "sleep 5",
"background": True}),
{"terminal": tool},
ctx=_Ctx(),
)
assert msg.metadata.get("background") is True
job = list(self.manager.list("s1"))[0]
# timeout lifted to 300s for the detached run (foreground default is 20s)
assert job.args == {"action": "run", "command": "sleep 5", "timeout": 300}
await job.done.wait()
async def test_bg_timeout_lift_respects_explicit_timeout(self, monkeypatch):
patch_settings(monkeypatch, backgroundable_tools="terminal")
tool = RecordingTool("terminal")
executor = ToolExecutor({"terminal": tool})
_, msg, _ = await executor._execute_one(
ToolCallRequest(id="tc1", name="terminal",
arguments={"action": "run", "command": "sleep 5",
"timeout": 45, "background": True}),
{"terminal": tool},
ctx=_Ctx(),
)
job = list(self.manager.list("s1"))[0]
assert job.args["timeout"] == 45
await job.done.wait()
async def test_bg_timeout_not_injected_for_spawn_agent(self, monkeypatch):
patch_settings(monkeypatch, backgroundable_tools="spawn_agent")
tool = RecordingTool("spawn_agent")
executor = ToolExecutor({"spawn_agent": tool})
_, msg, _ = await executor._execute_one(
ToolCallRequest(id="tc1", name="spawn_agent",
arguments={"task": "x", "background": True}),
{"spawn_agent": tool},
ctx=_Ctx(),
)
job = list(self.manager.list("s1"))[0]
assert "timeout" not in job.args
await job.done.wait()
async def test_spawn_agent_gets_deep_ring(self, monkeypatch):
"""ะค2: detached sub-agents get a 200-event ring for live progress."""
from types import SimpleNamespace
patch_settings(monkeypatch, backgroundable_tools="spawn_agent")
tool = RecordingTool("spawn_agent")
executor = ToolExecutor({"spawn_agent": tool})
_, msg, _ = await executor._execute_one(
ToolCallRequest(id="tc1", name="spawn_agent",
arguments={"task": "research", "background": True}),
{"spawn_agent": tool},
ctx=SimpleNamespace(session_id="s1"),
)
task_id = json.loads(msg.content)["task_id"]
job = self.manager.get(task_id, "s1")
assert job.ring.maxsize == 200
assert job.parent_tool_call_id == "tc1"
await job.done.wait()
