axelray-devandXubin Ren
851a0ff50c
Add fail_on_tool_error to AgentDefaults and wire it through AgentLoop -> SubagentManager -> AgentRunSpec. Previously hardcoded to True in SubagentManager._run_subagent. Now configurable via config.json with default True for backward compatibility. When set to False, subagents can retry on minor tool errors instead of immediately failing. Changes: - nanobot/config/schema.py: add fail_on_tool_error field (default True) - nanobot/agent/subagent.py: accept and forward fail_on_tool_error - nanobot/agent/loop.py: pass config through to SubagentManager - tests/agent/test_subagent.py: add regression test Signed-off-by: axelray-dev <110029405+axelray-dev@users.noreply.github.com>
113 lines
3.4 KiB
Python
113 lines
3.4 KiB
Python
"""Tests for SubagentManager."""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from nanobot.agent.runner import AgentRunResult
|
|
from nanobot.agent.subagent import SubagentManager, SubagentStatus
|
|
from nanobot.agent.tools.filesystem import FileToolsConfig
|
|
from nanobot.bus.queue import MessageBus
|
|
from nanobot.config.schema import ToolsConfig
|
|
from nanobot.providers.base import LLMProvider
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subagent_uses_tool_loader():
|
|
"""Verify subagent registers tools via ToolLoader, not hard-coded imports."""
|
|
provider = MagicMock(spec=LLMProvider)
|
|
provider.get_default_model.return_value = "test"
|
|
sm = SubagentManager(
|
|
provider=provider,
|
|
workspace=Path("/tmp"),
|
|
bus=MessageBus(),
|
|
model="test",
|
|
max_tool_result_chars=16_000,
|
|
)
|
|
tools = sm._build_tools()
|
|
assert tools.has("read_file")
|
|
assert tools.has("write_file")
|
|
assert not tools.has("message")
|
|
assert not tools.has("spawn")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subagent_build_tools_isolates_file_read_state(tmp_path):
|
|
"""Each spawned subagent needs a fresh file-state cache."""
|
|
(tmp_path / "note.txt").write_text("hello\n", encoding="utf-8")
|
|
provider = MagicMock(spec=LLMProvider)
|
|
provider.get_default_model.return_value = "test"
|
|
sm = SubagentManager(
|
|
provider=provider,
|
|
workspace=tmp_path,
|
|
bus=MessageBus(),
|
|
model="test",
|
|
max_tool_result_chars=16_000,
|
|
)
|
|
|
|
first_read = sm._build_tools().get("read_file")
|
|
second_read = sm._build_tools().get("read_file")
|
|
|
|
assert first_read is not second_read
|
|
assert (await first_read.execute(path="note.txt")).startswith("1| hello")
|
|
second_result = await second_read.execute(path="note.txt")
|
|
assert second_result.startswith("1| hello")
|
|
assert "File unchanged" not in second_result
|
|
|
|
|
|
def test_subagent_respects_file_tool_toggle(tmp_path):
|
|
provider = MagicMock(spec=LLMProvider)
|
|
provider.get_default_model.return_value = "test"
|
|
sm = SubagentManager(
|
|
provider=provider,
|
|
workspace=tmp_path,
|
|
bus=MessageBus(),
|
|
model="test",
|
|
max_tool_result_chars=16_000,
|
|
tools_config=ToolsConfig(file=FileToolsConfig(enable=False)),
|
|
)
|
|
|
|
tools = sm._build_tools()
|
|
|
|
file_tools = {
|
|
"apply_patch",
|
|
"edit_file",
|
|
"find_files",
|
|
"grep",
|
|
"list_dir",
|
|
"read_file",
|
|
"write_file",
|
|
}
|
|
assert file_tools.isdisjoint(tools.tool_names)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subagent_forwards_fail_on_tool_error_to_runner(tmp_path):
|
|
provider = MagicMock(spec=LLMProvider)
|
|
provider.get_default_model.return_value = "test"
|
|
sm = SubagentManager(
|
|
provider=provider,
|
|
workspace=tmp_path,
|
|
bus=MessageBus(),
|
|
model="test",
|
|
max_tool_result_chars=16_000,
|
|
fail_on_tool_error=False,
|
|
)
|
|
sm.runner.run = AsyncMock(
|
|
return_value=AgentRunResult(final_content="ok", messages=[], stop_reason="completed")
|
|
)
|
|
sm._announce_result = AsyncMock()
|
|
|
|
status = SubagentStatus(
|
|
task_id="t1",
|
|
label="label",
|
|
task_description="task",
|
|
started_at=0.0,
|
|
)
|
|
|
|
await sm._run_subagent("t1", "task", "label", {"channel": "cli", "chat_id": "direct"}, status)
|
|
|
|
spec = sm.runner.run.call_args.args[0]
|
|
assert spec.fail_on_tool_error is False
|