Inspired by Claude Code: session-wide runaway-loop caps (web_search + delegate_task)

agent/tool_guardrails.py

@@ -79,6 +79,7 @@ class ToolCallGuardrailConfig:
     no_progress_block_after: int = 5
     idempotent_tools: frozenset[str] = field(default_factory=lambda: IDEMPOTENT_TOOL_NAMES)
     mutating_tools: frozenset[str] = field(default_factory=lambda: MUTATING_TOOL_NAMES)
+    loop_caps: "LoopCapConfig" = field(default_factory=lambda: LoopCapConfig())
 
     @classmethod
     def from_mapping(cls, data: Mapping[str, Any] | None) -> "ToolCallGuardrailConfig":
@@ -121,6 +122,54 @@ def from_mapping(cls, data: Mapping[str, Any] | None) -> "ToolCallGuardrailConfi
                 hard_stop_after.get("idempotent_no_progress", data.get("no_progress_block_after")),
                 defaults.no_progress_block_after,
             ),
+            loop_caps=LoopCapConfig.from_mapping(data.get("loop_caps")),
+        )
+
+
+# Default session-wide caps, matching Claude Code's v2.1.212 runaway-loop
+# Per-turn (per-agent-loop) caps on runaway-prone tool calls. Counts reset at
+# the start of every agent loop (reset_for_turn), so the limit is "within a
+# single turn" rather than cumulative over the whole session. A single loop
+# issuing dozens of web searches or spawning dozens of subagents is already
+# pathological, so the defaults are deliberately low.
+_DEFAULT_MAX_WEB_SEARCHES_PER_TURN = 50
+_DEFAULT_MAX_SUBAGENTS_PER_TURN = 50
+
+
+@dataclass(frozen=True)
+class LoopCapConfig:
+    """Per-turn caps on runaway-prone tool calls.
+
+    Inspired by Claude Code v2.1.212 (Week 29, July 2026), which added caps on
+    WebSearch calls and subagent spawns to stop runaway search / delegation
+    loops. Here the caps count *within a single agent loop* (one turn): the
+    counters reset in ``reset_for_turn`` at the start of every
+    ``run_conversation``, so a legitimate multi-turn session is never starved,
+    but a single turn that spirals into an unbounded search / delegation loop
+    is stopped.
+
+    Semantics differ from the per-turn loop *detector* above (which keys on
+    repeated identical/failing calls): these caps are a hard ceiling on the
+    total count of a tool within the turn and fire regardless of
+    ``hard_stop_enabled``. A value of ``0`` disables the cap (unlimited).
+    """
+
+    max_web_searches: int = _DEFAULT_MAX_WEB_SEARCHES_PER_TURN
+    max_subagents: int = _DEFAULT_MAX_SUBAGENTS_PER_TURN
+
+    @classmethod
+    def from_mapping(cls, data: Mapping[str, Any] | None) -> "LoopCapConfig":
+        """Build config from the ``tool_loop_guardrails.loop_caps`` section."""
+        if not isinstance(data, Mapping):
+            return cls()
+        defaults = cls()
+        return cls(
+            max_web_searches=_non_negative_int(
+                data.get("max_web_searches"), defaults.max_web_searches
+            ),
+            max_subagents=_non_negative_int(
+                data.get("max_subagents"), defaults.max_subagents
+            ),
         )
 
 
@@ -233,13 +282,29 @@ def reset_for_turn(self) -> None:
         self._same_tool_failure_counts: dict[str, int] = {}
         self._no_progress: dict[ToolCallSignature, tuple[str, int]] = {}
         self._halt_decision: ToolGuardrailDecision | None = None
+        # Per-turn runaway-loop cap counters. Reset every turn (this method
+        # runs at the start of each run_conversation), so the caps bound a
+        # single agent loop rather than accumulating across the session.
+        self._turn_web_search_count = 0
+        self._turn_subagent_count = 0
 
     @property
     def halt_decision(self) -> ToolGuardrailDecision | None:
         return self._halt_decision
 
     def before_call(self, tool_name: str, args: Mapping[str, Any] | None) -> ToolGuardrailDecision:
         signature = ToolCallSignature.from_call(tool_name, _coerce_args(args))
+
+        # ── Per-turn runaway-loop caps ──────────────────────────────────
+        # These are hard ceilings on how many times a runaway-prone tool may
+        # be called within a single agent loop (turn). They apply regardless
+        # of hard_stop_enabled (which only governs the per-turn loop detector).
+        # We block BEFORE the call runs once the count is already at the cap,
+        # then increment for an allowed call so the (cap+1)-th is refused.
+        cap_block = self._check_loop_cap(tool_name, _coerce_args(args), signature)
+        if cap_block is not None:
+            return cap_block
+
         if not self.config.hard_stop_enabled:
             return ToolGuardrailDecision(tool_name=tool_name, signature=signature)
 
@@ -379,6 +444,68 @@ def _is_idempotent(self, tool_name: str) -> bool:
             return False
         return tool_name in self.config.idempotent_tools
 
+    def _check_loop_cap(
+        self,
+        tool_name: str,
+        args: Mapping[str, Any],
+        signature: ToolCallSignature,
+    ) -> ToolGuardrailDecision | None:
+        """Enforce and advance the per-turn runaway-loop counters.
+
+        Returns a ``block`` decision when the cap is already reached, otherwise
+        increments the relevant counter for the allowed call and returns
+        ``None``. A cap of 0 disables that limit entirely. Counters reset each
+        turn via ``reset_for_turn``.
+        """
+        caps = self.config.loop_caps
+
+        if tool_name == "web_search":
+            cap = caps.max_web_searches
+            if cap and self._turn_web_search_count >= cap:
+                decision = ToolGuardrailDecision(
+                    action="block",
+                    code="loop_web_search_cap",
+                    message=(
+                        f"Blocked web_search: this turn has already made {cap} "
+                        "web searches, the per-turn limit. This looks like a "
+                        "runaway search loop. Work with the results you already "
+                        "have and give the user your answer."
+                    ),
+                    tool_name=tool_name,
+                    count=self._turn_web_search_count,
+                    signature=signature,
+                )
+                self._halt_decision = decision
+                return decision
+            self._turn_web_search_count += 1
+            return None
+
+        if tool_name == "delegate_task":
+            cap = caps.max_subagents
+            if not cap:
+                return None
+            spawn_count = _subagent_spawn_count(args)
+            if self._turn_subagent_count >= cap:
+                decision = ToolGuardrailDecision(
+                    action="block",
+                    code="loop_subagent_cap",
+                    message=(
+                        f"Blocked delegate_task: this turn has already spawned "
+                        f"{self._turn_subagent_count} subagents (limit {cap}). "
+                        "This looks like a runaway delegation loop. Finish the "
+                        "work with the results you have and answer the user."
+                    ),
+                    tool_name=tool_name,
+                    count=self._turn_subagent_count,
+                    signature=signature,
+                )
+                self._halt_decision = decision
+                return decision
+            self._turn_subagent_count += spawn_count
+            return None
+
+        return None
+
 
 def toolguard_synthetic_result(decision: ToolGuardrailDecision) -> str:
     """Build a synthetic role=tool content string for a blocked tool call."""
@@ -471,6 +598,32 @@ def _positive_int(value: Any, default: int) -> int:
     return parsed if parsed >= 1 else default
 
 
+def _non_negative_int(value: Any, default: int) -> int:
+    """Parse a session-cap value. 0 is a valid (disable) value; negatives and
+    junk fall back to the default."""
+    if value is None:
+        return default
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        return default
+    return parsed if parsed >= 0 else default
+
+
+def _subagent_spawn_count(args: Mapping[str, Any]) -> int:
+    """How many subagents a single delegate_task call spawns.
+
+    delegate_task runs in one of two modes: a batch (``tasks`` is a non-empty
+    list, one child per item) or a single task (``goal``). Count the batch size
+    when present, otherwise 1, so the session subagent cap reflects real spawns
+    rather than delegate_task invocations.
+    """
+    tasks = args.get("tasks") if isinstance(args, Mapping) else None
+    if isinstance(tasks, list) and tasks:
+        return len(tasks)
+    return 1
+
+
 def _sha256(value: str) -> str:
     # surrogatepass: tool results scraped from the web can carry unpaired
     # UTF-16 surrogates (e.g. half of a mathematical-bold pair); a strict

hermes_cli/config.py

@@ -1421,6 +1421,18 @@ def _ensure_hermes_home_managed(home: Path):
             "same_tool_failure": 8,
             "idempotent_no_progress": 5,
         },
+        # Per-turn runaway-loop caps (inspired by Claude Code v2.1.212,
+        # Week 29, July 2026). Hard ceilings on how many times a runaway-prone
+        # tool may be called within a SINGLE agent loop (turn); the counters
+        # reset at the start of every turn, so a legitimate multi-turn session
+        # is never starved. They are always-on and fire regardless of the
+        # warn/hard-stop thresholds above. A single turn issuing dozens of web
+        # searches or spawning dozens of subagents is already pathological, so
+        # the defaults are low. Set either to 0 to disable that cap (unlimited).
+        "loop_caps": {
+            "max_web_searches": 50,   # max web_search calls per turn (0 = unlimited)
+            "max_subagents": 50,      # max subagents spawned per turn (0 = unlimited)
+        },
     },
 
     "compression": {

tests/agent/test_tool_guardrails.py

@@ -277,3 +277,113 @@ def test_after_call_survives_lone_surrogates_in_result_and_args():
     controller.after_call("web_search", {"query": dirty}, '{"error":"\ud835 boom"}', failed=True)
     controller.after_call("web_search", {"query": dirty}, '{"error":"\ud835 boom"}', failed=True)
     assert controller.before_call("web_search", {"query": dirty}).action == "block"
+
+
+# ── Per-turn runaway-loop caps (Claude Code v2.1.212, Week 29) ──────────────
+
+from agent.tool_guardrails import LoopCapConfig  # noqa: E402
+
+
+def test_loop_cap_defaults():
+    caps = ToolCallGuardrailConfig().loop_caps
+    assert caps.max_web_searches == 50
+    assert caps.max_subagents == 50
+
+
+def test_loop_cap_config_parses_nested_section():
+    cfg = ToolCallGuardrailConfig.from_mapping(
+        {"loop_caps": {"max_web_searches": 3, "max_subagents": 0}}
+    )
+    assert cfg.loop_caps.max_web_searches == 3
+    assert cfg.loop_caps.max_subagents == 0
+
+
+def test_loop_cap_zero_disables_and_junk_falls_back():
+    # 0 is a legitimate "unlimited" value; negatives / junk fall back to default.
+    assert LoopCapConfig.from_mapping({"max_web_searches": 0}).max_web_searches == 0
+    assert LoopCapConfig.from_mapping({"max_web_searches": -5}).max_web_searches == 50
+    assert LoopCapConfig.from_mapping({"max_subagents": "nope"}).max_subagents == 50
+
+
+def test_web_search_cap_blocks_after_limit_regardless_of_hard_stop():
+    # Loop caps fire even with hard_stop_enabled=False (the per-turn loop
+    # detector's flag). Each distinct query avoids the loop detector so we know
+    # the block came from the loop cap, not exact-failure repetition.
+    controller = ToolCallGuardrailController(
+        ToolCallGuardrailConfig(
+            hard_stop_enabled=False,
+            loop_caps=LoopCapConfig(max_web_searches=3),
+        )
+    )
+    for i in range(3):
+        assert controller.before_call("web_search", {"query": f"q{i}"}).action == "allow"
+    decision = controller.before_call("web_search", {"query": "q4"})
+    assert decision.action == "block"
+    assert decision.code == "loop_web_search_cap"
+    assert decision.should_halt is True
+
+
+def test_web_search_cap_resets_each_turn():
+    # The cap bounds a single turn: reset_for_turn clears the counter so a
+    # legitimate multi-turn session is never starved.
+    controller = ToolCallGuardrailController(
+        ToolCallGuardrailConfig(loop_caps=LoopCapConfig(max_web_searches=2))
+    )
+    # Turn 1: two searches allowed, the third would block within the turn.
+    assert controller.before_call("web_search", {"query": "a"}).action == "allow"
+    assert controller.before_call("web_search", {"query": "b"}).action == "allow"
+    assert controller.before_call("web_search", {"query": "c"}).action == "block"
+    # New turn: the counter resets, so the budget is fresh again.
+    controller.reset_for_turn()
+    assert controller.before_call("web_search", {"query": "d"}).action == "allow"
+    assert controller.before_call("web_search", {"query": "e"}).action == "allow"
+    assert controller.before_call("web_search", {"query": "f"}).action == "block"
+
+
+def test_subagent_cap_counts_batch_task_spawns():
+    # A single delegate_task batch of N tasks spends N of the subagent budget.
+    controller = ToolCallGuardrailController(
+        ToolCallGuardrailConfig(loop_caps=LoopCapConfig(max_subagents=5))
+    )
+    # First call spawns 3 (batch) → count 3, allowed.
+    assert controller.before_call(
+        "delegate_task", {"tasks": [{"goal": "a"}, {"goal": "b"}, {"goal": "c"}]}
+    ).action == "allow"
+    # Second call spawns 1 (goal) → count 4, allowed.
+    assert controller.before_call("delegate_task", {"goal": "d"}).action == "allow"
+    # Count is 4 (< 5) so this is allowed and bumps to 5.
+    assert controller.before_call("delegate_task", {"goal": "e"}).action == "allow"
+    # Now count is 5 (>= 5) so the next call is blocked.
+    decision = controller.before_call("delegate_task", {"goal": "f"})
+    assert decision.action == "block"
+    assert decision.code == "loop_subagent_cap"
+
+
+def test_subagent_cap_resets_each_turn():
+    controller = ToolCallGuardrailController(
+        ToolCallGuardrailConfig(loop_caps=LoopCapConfig(max_subagents=1))
+    )
+    assert controller.before_call("delegate_task", {"goal": "a"}).action == "allow"
+    assert controller.before_call("delegate_task", {"goal": "b"}).action == "block"
+    controller.reset_for_turn()
+    assert controller.before_call("delegate_task", {"goal": "c"}).action == "allow"
+
+
+def test_loop_caps_disabled_when_zero():
+    controller = ToolCallGuardrailController(
+        ToolCallGuardrailConfig(
+            loop_caps=LoopCapConfig(max_web_searches=0, max_subagents=0)
+        )
+    )
+    for i in range(60):
+        assert controller.before_call("web_search", {"query": f"q{i}"}).action == "allow"
+        assert controller.before_call("delegate_task", {"goal": f"g{i}"}).action == "allow"
+
+
+def test_other_tools_never_touched_by_loop_caps():
+    controller = ToolCallGuardrailController(
+        ToolCallGuardrailConfig(loop_caps=LoopCapConfig(max_web_searches=1))
+    )
+    # read_file / terminal / etc. are unaffected regardless of the web cap.
+    for _ in range(10):
+        assert controller.before_call("read_file", {"path": "/tmp/x"}).action == "allow"

website/docs/user-guide/configuration.md

@@ -1455,10 +1455,21 @@ tool_loop_guardrails:
     exact_failure: 5
     same_tool_failure: 8
     idempotent_no_progress: 5
+  loop_caps:
+    max_web_searches: 50       # max web_search calls per turn (0 = unlimited)
+    max_subagents: 50          # max subagents spawned per turn (0 = unlimited)
 ```
 
 `hard_stop_enabled` defaults to `false` because interactive sessions have a human in the loop. In unattended deployments (gateway, cron, kanban workers) set it to `true` so repeated failures are blocked rather than only warned. See also [Docker / unattended deployments](docker.md).
 
+### Per-turn runaway-loop caps
+
+Separate from the failure-based thresholds above, `loop_caps` sets hard ceilings on how many `web_search` calls and subagent spawns a single agent loop (turn) may make. The counters reset at the start of every turn, so a legitimate multi-turn session is never starved — but a single turn that spirals into an unbounded search or delegation loop is stopped. These are always on and fire regardless of `hard_stop_enabled`. A single turn issuing dozens of web searches or spawning dozens of subagents is already pathological, so the defaults are low. When a cap is reached, the offending tool call is blocked with an explanatory message and the turn stops cleanly instead of burning the rest of the budget. Set either value to `0` to disable that cap entirely.
+
+A single `delegate_task` batch counts each task toward `max_subagents` (a batch of 3 spends 3), so the cap tracks real subagents spawned rather than `delegate_task` invocations.
+
+This mirrors Claude Code's per-session WebSearch and subagent caps (v2.1.212), which also default to 200 and reset on `/clear`.
+
 ## TTS Configuration
 
 ```yaml