"""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()
class TestUnresolvedTool:
"""A tool the live tool_map does not hold must leave a trace.
The agent gets an error string back and the user sees it, but the server
logged nothing at all โ which is how a stale toolset after switch_profile
(all the new tools "not found") stayed invisible in the journal.
"""
async def test_missing_tool_is_logged_with_the_live_profile(self, capfd):
from types import SimpleNamespace
executor = ToolExecutor(ToolRegistry())
ctx = SimpleNamespace(session_id="s1", profile_id="developer")
event, msg, image = await executor._execute_one(
ToolCallRequest(id="1", name="ssh_exec", arguments={}),
{"todo": FakeTool("todo")},
ctx=ctx,
)
assert event.success is False
assert msg.content == "Error: tool 'ssh_exec' not found."
assert image is None
logged = capfd.readouterr().out
assert "tool.not_found" in logged
assert "ssh_exec" in logged
assert "developer" in logged
class TestSecretRedactionInLogs:
"""A credential must not reach the journal through the tool.args log line.
``ssh_exec`` takes a ``password`` parameter, and both the main loop and the
sub-agent loop logged the whole argument dict โ the prod journal held SSH
passwords in plaintext. Masking happens at the log call, so what the tool
receives and what the client is shown stay intact.
"""
async def test_password_is_masked_in_the_log_but_not_for_the_tool(self, capfd):
tool = RecordingTool("ssh_exec", output="ok")
executor = ToolExecutor(ToolRegistry())
event, _msg, _image = await executor._execute_one(
ToolCallRequest(
id="1",
name="ssh_exec",
arguments={"host": "10.0.0.1", "username": "ubuntu", "password": "hunter2"},
),
{"ssh_exec": tool},
ctx=_Ctx(),
)
logged = capfd.readouterr().out
assert "hunter2" not in logged
assert "'password': '***'" in logged
# The neighbours are still there โ the log has to stay worth reading.
assert "10.0.0.1" in logged and "ubuntu" in logged
assert tool.calls == [
{"host": "10.0.0.1", "username": "ubuntu", "password": "hunter2"}
]
assert event.arguments["password"] == "hunter2"
