fix: harden cron tool contract and repeat guard

This commit is contained in:
yeyitech
2026-04-14 12:40:23 +08:00
parent a38bc637bd
commit 655f3d2cc5
6 changed files with 242 additions and 41 deletions
+42
View File
@@ -798,6 +798,48 @@ async def test_runner_blocks_repeated_external_fetches():
assert "repeated external lookup blocked" in blocked_tool_message["content"]
@pytest.mark.asyncio
async def test_runner_blocks_repeated_identical_tool_calls():
from nanobot.agent.runner import AgentRunSpec, AgentRunner
provider = MagicMock()
captured_final_call: list[dict] = []
call_count = {"n": 0}
async def chat_with_retry(*, messages, **kwargs):
call_count["n"] += 1
if call_count["n"] <= 3:
return LLMResponse(
content="working",
tool_calls=[ToolCallRequest(id=f"call_{call_count['n']}", name="read_file", arguments={"path": "memory/history.jsonl", "limit": 50, "offset": 1})],
usage={},
)
captured_final_call[:] = messages
return LLMResponse(content="done", tool_calls=[], usage={})
provider.chat_with_retry = chat_with_retry
tools = MagicMock()
tools.get_definitions.return_value = []
tools.execute = AsyncMock(return_value="file content")
runner = AgentRunner(provider)
result = await runner.run(AgentRunSpec(
initial_messages=[{"role": "user", "content": "what happened recently?"}],
tools=tools,
model="test-model",
max_iterations=4,
max_tool_result_chars=_MAX_TOOL_RESULT_CHARS,
))
assert result.final_content == "done"
assert tools.execute.await_count == 2
blocked_tool_message = [
msg for msg in captured_final_call
if msg.get("role") == "tool" and msg.get("tool_call_id") == "call_3"
][0]
assert "repeated identical call to 'read_file' blocked" in blocked_tool_message["content"]
@pytest.mark.asyncio
async def test_loop_max_iterations_message_stays_stable(tmp_path):
loop = _make_loop(tmp_path)
+41 -1
View File
@@ -7,7 +7,6 @@ import pytest
from nanobot.agent.tools.cron import CronTool
from nanobot.cron.service import CronService
from nanobot.cron.types import CronJob, CronJobState, CronPayload, CronSchedule
from tests.test_openai_api import pytest_plugins
def _make_tool(tmp_path) -> CronTool:
@@ -346,6 +345,47 @@ def test_add_job_can_disable_delivery(tmp_path) -> None:
assert job.payload.deliver is False
def test_cron_schema_advertises_action_specific_requirements(tmp_path) -> None:
tool = _make_tool(tmp_path)
assert tool.parameters["required"] == ["action"]
assert tool.parameters["oneOf"] == [
{
"properties": {
"action": {"enum": ["add"]},
"message": {"type": "string", "minLength": 1},
},
"required": ["action", "message"],
},
{
"properties": {"action": {"enum": ["list"]}},
"required": ["action"],
},
{
"properties": {"action": {"enum": ["remove"]}},
"required": ["action", "job_id"],
},
]
def test_validate_params_requires_message_only_for_add(tmp_path) -> None:
tool = _make_tool(tmp_path)
assert "message is required when action='add'" in tool.validate_params({"action": "add"})
assert tool.validate_params({"action": "list"}) == []
assert "job_id is required when action='remove'" in tool.validate_params({"action": "remove"})
def test_add_job_empty_message_returns_actionable_error(tmp_path) -> None:
tool = _make_tool(tmp_path)
tool.set_context("telegram", "chat-1")
result = tool._add_job(None, "", 60, None, None, None)
assert "action='add' requires a non-empty 'message'" in result
assert "Retry including message=" in result
def test_list_excludes_disabled_jobs(tmp_path) -> None:
tool = _make_tool(tmp_path)
job = tool._cron.add_job(
+20
View File
@@ -0,0 +1,20 @@
from nanobot.utils.runtime import repeated_tool_call_error, tool_call_signature
def test_tool_call_signature_sorts_arguments_stably() -> None:
first = tool_call_signature("read_file", {"offset": 1, "path": "memory/history.jsonl"})
second = tool_call_signature("read_file", {"path": "memory/history.jsonl", "offset": 1})
assert first == second
def test_repeated_tool_call_error_blocks_after_two_attempts() -> None:
seen: dict[str, int] = {}
assert repeated_tool_call_error("read_file", {"path": "a.txt"}, seen) is None
assert repeated_tool_call_error("read_file", {"path": "a.txt"}, seen) is None
error = repeated_tool_call_error("read_file", {"path": "a.txt"}, seen)
assert error is not None
assert "repeated identical call to 'read_file' blocked after 2 attempts" in error