fix: drop generic repeated tool-call guard

The global guard changed baseline agent and subagent behavior without
proving a real no-progress loop. Keep this PR focused on the cron
contract hardening and validation fixes.

Made-with: Cursor

nanobot/agent/runner.py

@@ -3,14 +3,15 @@
 from __future__ import annotations
 
 import asyncio
-import inspect
 from dataclasses import dataclass, field
+import inspect
 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 (
@@ -21,7 +22,6 @@ 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,7 +29,6 @@ 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."
@@ -235,7 +234,6 @@ 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
@@ -306,7 +304,6 @@ class AgentRunner:
                     spec,
                     response.tool_calls,
                     external_lookup_counts,
-                    tool_call_counts,
                 )
                 tool_events.extend(new_events)
                 context.tool_results = list(results)
@@ -627,21 +624,18 @@ 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, tool_call_counts)
+                    self._run_tool(spec, tool_call, external_lookup_counts)
                     for tool_call in batch
                 )))
             else:
                 for tool_call in batch:
-                    tool_results.append(
+                    tool_results.append(await self._run_tool(spec, tool_call, external_lookup_counts))
-                        await self._run_tool(spec, tool_call, external_lookup_counts, tool_call_counts)
-                    )
 
         results: list[Any] = []
         events: list[dict[str, str]] = []
@@ -658,9 +652,8 @@ 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,
@@ -673,22 +666,8 @@ 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, RuntimeError(lookup_error)
-            return lookup_error + _hint, event, None
+            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):
@@ -704,7 +683,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)
@@ -729,8 +708,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, RuntimeError(result)
-            return result + _hint, event, None
+            return result + _HINT, event, None
 
         detail = "" if result is None else str(result)
         detail = detail.replace("\n", " ").strip()
@@ -1005,3 +984,4 @@ class AgentRunner:
         if current:
             batches.append(current)
         return batches
+

nanobot/utils/runtime.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import json
 from typing import Any
 
 from loguru import logger
@@ -10,7 +9,6 @@ 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. "
@@ -75,15 +73,6 @@ 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],
@@ -106,26 +95,3 @@ 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."
-    )

tests/agent/test_runner.py

@@ -854,48 +854,6 @@ 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)

tests/utils/test_runtime.py

@@ -1,20 +0,0 @@
-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