From 655f3d2cc53c66c46ef650ed50b31501df48e628 Mon Sep 17 00:00:00 2001 From: yeyitech Date: Tue, 14 Apr 2026 12:40:23 +0800 Subject: [PATCH] fix: harden cron tool contract and repeat guard --- nanobot/agent/runner.py | 42 ++++++++---- nanobot/agent/tools/cron.py | 103 +++++++++++++++++++++--------- nanobot/utils/runtime.py | 34 ++++++++++ tests/agent/test_runner.py | 42 ++++++++++++ tests/cron/test_cron_tool_list.py | 42 +++++++++++- tests/utils/test_runtime.py | 20 ++++++ 6 files changed, 242 insertions(+), 41 deletions(-) create mode 100644 tests/utils/test_runtime.py diff --git a/nanobot/agent/runner.py b/nanobot/agent/runner.py index 592af9de..c7006caa 100644 --- a/nanobot/agent/runner.py +++ b/nanobot/agent/runner.py @@ -3,15 +3,14 @@ from __future__ import annotations import asyncio -from dataclasses import dataclass, field import inspect +from dataclasses import dataclass, field from pathlib import Path from typing import Any from loguru import logger from nanobot.agent.hook import AgentHook, AgentHookContext -from nanobot.utils.prompt_templates import render_template from nanobot.agent.tools.registry import ToolRegistry from nanobot.providers.base import LLMProvider, ToolCallRequest from nanobot.utils.helpers import ( @@ -22,6 +21,7 @@ from nanobot.utils.helpers import ( maybe_persist_tool_result, truncate_text, ) +from nanobot.utils.prompt_templates import render_template from nanobot.utils.runtime import ( EMPTY_FINAL_RESPONSE_MESSAGE, build_finalization_retry_message, @@ -29,6 +29,7 @@ from nanobot.utils.runtime import ( ensure_nonempty_tool_result, is_blank_text, repeated_external_lookup_error, + repeated_tool_call_error, ) _DEFAULT_ERROR_MESSAGE = "Sorry, I encountered an error calling the AI model." @@ -233,6 +234,7 @@ class AgentRunner: stop_reason = "completed" tool_events: list[dict[str, str]] = [] external_lookup_counts: dict[str, int] = {} + tool_call_counts: dict[str, int] = {} empty_content_retries = 0 length_recovery_count = 0 had_injections = False @@ -303,6 +305,7 @@ class AgentRunner: spec, response.tool_calls, external_lookup_counts, + tool_call_counts, ) tool_events.extend(new_events) context.tool_results = list(results) @@ -616,18 +619,21 @@ class AgentRunner: spec: AgentRunSpec, tool_calls: list[ToolCallRequest], external_lookup_counts: dict[str, int], + tool_call_counts: dict[str, int], ) -> tuple[list[Any], list[dict[str, str]], BaseException | None]: batches = self._partition_tool_batches(spec, tool_calls) tool_results: list[tuple[Any, dict[str, str], BaseException | None]] = [] for batch in batches: if spec.concurrent_tools and len(batch) > 1: tool_results.extend(await asyncio.gather(*( - self._run_tool(spec, tool_call, external_lookup_counts) + self._run_tool(spec, tool_call, external_lookup_counts, tool_call_counts) for tool_call in batch ))) else: for tool_call in batch: - tool_results.append(await self._run_tool(spec, tool_call, external_lookup_counts)) + tool_results.append( + await self._run_tool(spec, tool_call, external_lookup_counts, tool_call_counts) + ) results: list[Any] = [] events: list[dict[str, str]] = [] @@ -644,8 +650,9 @@ class AgentRunner: spec: AgentRunSpec, tool_call: ToolCallRequest, external_lookup_counts: dict[str, int], + tool_call_counts: dict[str, int], ) -> tuple[Any, dict[str, str], BaseException | None]: - _HINT = "\n\n[Analyze the error above and try a different approach.]" + _hint = "\n\n[Analyze the error above and try a different approach.]" lookup_error = repeated_external_lookup_error( tool_call.name, tool_call.arguments, @@ -658,8 +665,22 @@ class AgentRunner: "detail": "repeated external lookup blocked", } if spec.fail_on_tool_error: - return lookup_error + _HINT, event, RuntimeError(lookup_error) - return lookup_error + _HINT, event, None + return lookup_error + _hint, event, RuntimeError(lookup_error) + return lookup_error + _hint, event, None + repeat_error = repeated_tool_call_error( + tool_call.name, + tool_call.arguments, + tool_call_counts, + ) + if repeat_error: + event = { + "name": tool_call.name, + "status": "error", + "detail": "repeated identical tool call blocked", + } + if spec.fail_on_tool_error: + return repeat_error + _hint, event, RuntimeError(repeat_error) + return repeat_error + _hint, event, None prepare_call = getattr(spec.tools, "prepare_call", None) tool, params, prep_error = None, tool_call.arguments, None if callable(prepare_call): @@ -675,7 +696,7 @@ class AgentRunner: "status": "error", "detail": prep_error.split(": ", 1)[-1][:120], } - return prep_error + _HINT, event, RuntimeError(prep_error) if spec.fail_on_tool_error else None + return prep_error + _hint, event, RuntimeError(prep_error) if spec.fail_on_tool_error else None try: if tool is not None: result = await tool.execute(**params) @@ -700,8 +721,8 @@ class AgentRunner: "detail": result.replace("\n", " ").strip()[:120], } if spec.fail_on_tool_error: - return result + _HINT, event, RuntimeError(result) - return result + _HINT, event, None + return result + _hint, event, RuntimeError(result) + return result + _hint, event, None detail = "" if result is None else str(result) detail = detail.replace("\n", " ").strip() @@ -966,4 +987,3 @@ class AgentRunner: if current: batches.append(current) return batches - diff --git a/nanobot/agent/tools/cron.py b/nanobot/agent/tools/cron.py index f0d3ddab..dd2d9d43 100644 --- a/nanobot/agent/tools/cron.py +++ b/nanobot/agent/tools/cron.py @@ -5,39 +5,72 @@ from datetime import datetime from typing import Any from nanobot.agent.tools.base import Tool, tool_parameters -from nanobot.agent.tools.schema import BooleanSchema, IntegerSchema, StringSchema, tool_parameters_schema +from nanobot.agent.tools.schema import ( + BooleanSchema, + IntegerSchema, + StringSchema, + tool_parameters_schema, +) from nanobot.cron.service import CronService from nanobot.cron.types import CronJob, CronJobState, CronSchedule +_CRON_PARAMETERS = tool_parameters_schema( + action=StringSchema("Action to perform", enum=["add", "list", "remove"]), + name=StringSchema( + "Optional short human-readable label for the job " + "(e.g., 'weather-monitor', 'daily-standup'). Defaults to first 30 chars of message." + ), + message=StringSchema( + "Instruction for the agent to execute when the job triggers. " + "Required when action='add' " + "(e.g., 'Send a reminder to WeChat: xxx' or 'Check system status and report')" + ), + every_seconds=IntegerSchema(0, description="Interval in seconds (for recurring tasks)"), + cron_expr=StringSchema("Cron expression like '0 9 * * *' (for scheduled tasks)"), + tz=StringSchema( + "Optional IANA timezone for cron expressions (e.g. 'America/Vancouver'). " + "When omitted with cron_expr, the tool's default timezone applies." + ), + at=StringSchema( + "ISO datetime for one-time execution (e.g. '2026-02-12T10:30:00'). " + "Naive values use the tool's default timezone." + ), + deliver=BooleanSchema( + description="Whether to deliver the execution result to the user channel (default true)", + default=True, + ), + job_id=StringSchema("Job ID (for remove)"), + required=["action"], + description=( + "Action-specific parameters: add requires a non-empty message plus one schedule " + "(every_seconds, cron_expr, or at); remove requires job_id; list only needs action." + ), +) +_CRON_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"], + }, +] + @tool_parameters( - tool_parameters_schema( - action=StringSchema("Action to perform", enum=["add", "list", "remove"]), - name=StringSchema( - "Optional short human-readable label for the job " - "(e.g., 'weather-monitor', 'daily-standup'). Defaults to first 30 chars of message." - ), - message=StringSchema( - "Instruction for the agent to execute when the job triggers " - "(e.g., 'Send a reminder to WeChat: xxx' or 'Check system status and report')" - ), - every_seconds=IntegerSchema(0, description="Interval in seconds (for recurring tasks)"), - cron_expr=StringSchema("Cron expression like '0 9 * * *' (for scheduled tasks)"), - tz=StringSchema( - "Optional IANA timezone for cron expressions (e.g. 'America/Vancouver'). " - "When omitted with cron_expr, the tool's default timezone applies." - ), - at=StringSchema( - "ISO datetime for one-time execution (e.g. '2026-02-12T10:30:00'). " - "Naive values use the tool's default timezone." - ), - deliver=BooleanSchema( - description="Whether to deliver the execution result to the user channel (default true)", - default=True, - ), - job_id=StringSchema("Job ID (for remove)"), - required=["action"], - ) + _CRON_PARAMETERS ) class CronTool(Tool): """Tool to schedule reminders and recurring tasks.""" @@ -94,6 +127,15 @@ class CronTool(Tool): f"If tz is omitted, cron expressions and naive ISO times default to {self._default_timezone}." ) + def validate_params(self, params: dict[str, Any]) -> list[str]: + errors = super().validate_params(params) + action = params.get("action") + if action == "add" and not str(params.get("message") or "").strip(): + errors.append("message is required when action='add'") + if action == "remove" and not str(params.get("job_id") or "").strip(): + errors.append("job_id is required when action='remove'") + return errors + async def execute( self, action: str, @@ -128,7 +170,10 @@ class CronTool(Tool): deliver: bool = True, ) -> str: if not message: - return "Error: message is required for add" + return ( + "Error: cron action='add' requires a non-empty 'message' parameter " + "describing what to do when the job triggers. Retry including message=\"...\"." + ) if not self._channel or not self._chat_id: return "Error: no session context (channel/chat_id)" if tz and not cron_expr: diff --git a/nanobot/utils/runtime.py b/nanobot/utils/runtime.py index 39822fd4..0101443e 100644 --- a/nanobot/utils/runtime.py +++ b/nanobot/utils/runtime.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json from typing import Any from loguru import logger @@ -9,6 +10,7 @@ from loguru import logger from nanobot.utils.helpers import stringify_text_blocks _MAX_REPEAT_EXTERNAL_LOOKUPS = 2 +_MAX_REPEAT_TOOL_CALLS = 2 EMPTY_FINAL_RESPONSE_MESSAGE = ( "I completed the tool steps but couldn't produce a final answer. " @@ -73,6 +75,15 @@ def external_lookup_signature(tool_name: str, arguments: dict[str, Any]) -> str return None +def tool_call_signature(tool_name: str, arguments: dict[str, Any]) -> str: + """Stable signature for repeated tool calls across retries.""" + try: + args_json = json.dumps(arguments, sort_keys=True, default=str, ensure_ascii=True) + except Exception: + args_json = repr(sorted(arguments.items())) + return f"{tool_name}:{args_json}" + + def repeated_external_lookup_error( tool_name: str, arguments: dict[str, Any], @@ -95,3 +106,26 @@ def repeated_external_lookup_error( "Error: repeated external lookup blocked. " "Use the results you already have to answer, or try a meaningfully different source." ) + + +def repeated_tool_call_error( + tool_name: str, + arguments: dict[str, Any], + seen_counts: dict[str, int], +) -> str | None: + """Block repeated identical tool calls after a small retry budget.""" + signature = tool_call_signature(tool_name, arguments) + count = seen_counts.get(signature, 0) + 1 + seen_counts[signature] = count + if count <= _MAX_REPEAT_TOOL_CALLS: + return None + logger.warning( + "Blocking repeated tool call {} on attempt {}", + signature[:160], + count, + ) + return ( + f"Error: repeated identical call to '{tool_name}' blocked after {count - 1} attempts. " + "The previous attempts used the same arguments. Change the arguments or try a different " + "approach." + ) diff --git a/tests/agent/test_runner.py b/tests/agent/test_runner.py index 74025d77..69a5f51e 100644 --- a/tests/agent/test_runner.py +++ b/tests/agent/test_runner.py @@ -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) diff --git a/tests/cron/test_cron_tool_list.py b/tests/cron/test_cron_tool_list.py index 86f3055c..a3ee9b1a 100644 --- a/tests/cron/test_cron_tool_list.py +++ b/tests/cron/test_cron_tool_list.py @@ -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( diff --git a/tests/utils/test_runtime.py b/tests/utils/test_runtime.py new file mode 100644 index 00000000..aa25e155 --- /dev/null +++ b/tests/utils/test_runtime.py @@ -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