Inspired by Claude Code: session-wide runaway-loop caps for web_search and delegate_task

Add per-session lifetime caps on web_search calls and subagent spawns
(defaults 200/200, matching Claude Code v2.1.212). Unlike the existing
per-turn tool-loop guardrails, these count over the whole session and
reset only when a fresh agent is built (/new, /clear). Hitting a cap
blocks the offending call and halts the turn cleanly.

- agent/tool_guardrails.py: SessionCapConfig + session counters on the
  controller (in __init__, not reset_for_turn, so they persist across
  turns). before_call() enforces caps first, independent of
  hard_stop_enabled. delegate_task batches count each task.
- hermes_cli/config.py: tool_loop_guardrails.session_caps defaults.
- docs + tests (unit + E2E validated against a real AIAgent).

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)
+    session_caps: "SessionCapConfig" = field(default_factory=lambda: SessionCapConfig())
 
     @classmethod
     def from_mapping(cls, data: Mapping[str, Any] | None) -> "ToolCallGuardrailConfig":
@@ -121,6 +122,51 @@ class ToolCallGuardrailConfig:
                 hard_stop_after.get("idempotent_no_progress", data.get("no_progress_block_after")),
                 defaults.no_progress_block_after,
             ),
+            session_caps=SessionCapConfig.from_mapping(data.get("session_caps")),
+        )
+
+
+# Default session-wide caps, matching Claude Code's v2.1.212 runaway-loop
+# limits (CLAUDE_CODE_MAX_WEB_SEARCHES_PER_SESSION /
+# CLAUDE_CODE_MAX_SUBAGENTS_PER_SESSION, both default 200, reset on /clear).
+_DEFAULT_MAX_WEB_SEARCHES_PER_SESSION = 200
+_DEFAULT_MAX_SUBAGENTS_PER_SESSION = 200
+
+
+@dataclass(frozen=True)
+class SessionCapConfig:
+    """Session-wide lifetime caps on runaway-prone tool calls.
+
+    Inspired by Claude Code v2.1.212 (Week 29, July 2026), which added a
+    per-session cap on WebSearch calls and subagent spawns to stop runaway
+    search / delegation loops. These caps count over the whole session (not
+    per turn like the loop-detection guardrails above) and reset only when a
+    fresh agent is created — i.e. on ``/new`` or ``/clear`` — because the
+    counters live on the ``AIAgent`` instance, not in ``reset_for_turn``.
+
+    Semantics differ from the per-turn loop guardrails: session caps are a
+    hard safety ceiling and fire regardless of ``hard_stop_enabled``. A value
+    of ``0`` disables the cap (unlimited). Defaults match Claude Code (200);
+    a legitimate session almost never issues 200 web searches or spawns 200
+    subagents, so hitting the cap is a strong runaway-loop signal.
+    """
+
+    max_web_searches: int = _DEFAULT_MAX_WEB_SEARCHES_PER_SESSION
+    max_subagents: int = _DEFAULT_MAX_SUBAGENTS_PER_SESSION
+
+    @classmethod
+    def from_mapping(cls, data: Mapping[str, Any] | None) -> "SessionCapConfig":
+        """Build config from the ``tool_loop_guardrails.session_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
+            ),
         )
 
 
@@ -226,6 +272,12 @@ class ToolCallGuardrailController:
 
     def __init__(self, config: ToolCallGuardrailConfig | None = None):
         self.config = config or ToolCallGuardrailConfig()
+        # Session-wide counters persist for the life of this controller (i.e.
+        # the life of the AIAgent instance). They are deliberately NOT cleared
+        # in reset_for_turn — a runaway loop that spans many turns must still be
+        # caught. They reset only when a fresh agent is built (/new, /clear).
+        self._session_web_search_count = 0
+        self._session_subagent_count = 0
         self.reset_for_turn()
 
     def reset_for_turn(self) -> None:
@@ -240,6 +292,16 @@ class ToolCallGuardrailController:
 
     def before_call(self, tool_name: str, args: Mapping[str, Any] | None) -> ToolGuardrailDecision:
         signature = ToolCallSignature.from_call(tool_name, _coerce_args(args))
+
+        # ── Session-wide runaway-loop caps ──────────────────────────────
+        # These are hard safety ceilings that 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_session_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 +441,68 @@ class ToolCallGuardrailController:
             return False
         return tool_name in self.config.idempotent_tools
 
+    def _check_session_cap(
+        self,
+        tool_name: str,
+        args: Mapping[str, Any],
+        signature: ToolCallSignature,
+    ) -> ToolGuardrailDecision | None:
+        """Enforce and advance the session-wide 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.
+        """
+        caps = self.config.session_caps
+
+        if tool_name == "web_search":
+            cap = caps.max_web_searches
+            if cap and self._session_web_search_count >= cap:
+                decision = ToolGuardrailDecision(
+                    action="block",
+                    code="session_web_search_cap",
+                    message=(
+                        f"Blocked web_search: this session has already made {cap} "
+                        "web searches, the per-session limit. This looks like a "
+                        "runaway search loop. Work with the results you already "
+                        "have, or start a fresh session (/new) to reset the budget."
+                    ),
+                    tool_name=tool_name,
+                    count=self._session_web_search_count,
+                    signature=signature,
+                )
+                self._halt_decision = decision
+                return decision
+            self._session_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._session_subagent_count >= cap:
+                decision = ToolGuardrailDecision(
+                    action="block",
+                    code="session_subagent_cap",
+                    message=(
+                        f"Blocked delegate_task: this session has already spawned "
+                        f"{self._session_subagent_count} subagents (limit {cap}). "
+                        "This looks like a runaway delegation loop. Finish the work "
+                        "with the results you have, or start a fresh session (/new) "
+                        "to reset the budget."
+                    ),
+                    tool_name=tool_name,
+                    count=self._session_subagent_count,
+                    signature=signature,
+                )
+                self._halt_decision = decision
+                return decision
+            self._session_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 +595,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,17 @@ DEFAULT_CONFIG = {
             "same_tool_failure": 8,
             "idempotent_no_progress": 5,
         },
+        # Session-wide runaway-loop caps (inspired by Claude Code v2.1.212,
+        # Week 29, July 2026). Unlike the per-turn warn/hard-stop thresholds
+        # above, these count over the WHOLE session and reset only when a fresh
+        # session starts (/new or /clear). They are always-on hard ceilings — a
+        # legitimate session almost never issues 200 web searches or spawns 200
+        # subagents, so hitting one is a strong runaway-loop signal. Set either
+        # to 0 to disable that cap (unlimited).
+        "session_caps": {
+            "max_web_searches": 200,   # max web_search calls per session (0 = unlimited)
+            "max_subagents": 200,      # max subagents spawned per session (0 = unlimited)
+        },
     },
 
     "compression": {

tests/agent/test_tool_guardrails.py

@@ -277,3 +277,97 @@ 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"
+
+
+# ── Session-wide runaway-loop caps (Claude Code v2.1.212, Week 29) ──────────
+
+from agent.tool_guardrails import SessionCapConfig  # noqa: E402
+
+
+def test_session_cap_defaults_match_claude_code():
+    caps = ToolCallGuardrailConfig().session_caps
+    assert caps.max_web_searches == 200
+    assert caps.max_subagents == 200
+
+
+def test_session_cap_config_parses_nested_section():
+    cfg = ToolCallGuardrailConfig.from_mapping(
+        {"session_caps": {"max_web_searches": 3, "max_subagents": 0}}
+    )
+    assert cfg.session_caps.max_web_searches == 3
+    assert cfg.session_caps.max_subagents == 0
+
+
+def test_session_cap_zero_disables_and_junk_falls_back():
+    # 0 is a legitimate "unlimited" value; negatives / junk fall back to default.
+    assert SessionCapConfig.from_mapping({"max_web_searches": 0}).max_web_searches == 0
+    assert SessionCapConfig.from_mapping({"max_web_searches": -5}).max_web_searches == 200
+    assert SessionCapConfig.from_mapping({"max_subagents": "nope"}).max_subagents == 200
+
+
+def test_web_search_cap_blocks_after_limit_regardless_of_hard_stop():
+    # Session 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 session cap, not exact-failure repetition.
+    controller = ToolCallGuardrailController(
+        ToolCallGuardrailConfig(
+            hard_stop_enabled=False,
+            session_caps=SessionCapConfig(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 == "session_web_search_cap"
+    assert decision.should_halt is True
+
+
+def test_web_search_cap_persists_across_turn_resets():
+    controller = ToolCallGuardrailController(
+        ToolCallGuardrailConfig(session_caps=SessionCapConfig(max_web_searches=2))
+    )
+    assert controller.before_call("web_search", {"query": "a"}).action == "allow"
+    controller.reset_for_turn()  # a per-turn reset must NOT clear the session count
+    assert controller.before_call("web_search", {"query": "b"}).action == "allow"
+    controller.reset_for_turn()
+    assert controller.before_call("web_search", {"query": "c"}).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(session_caps=SessionCapConfig(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 == "session_subagent_cap"
+
+
+def test_session_caps_disabled_when_zero():
+    controller = ToolCallGuardrailController(
+        ToolCallGuardrailConfig(
+            session_caps=SessionCapConfig(max_web_searches=0, max_subagents=0)
+        )
+    )
+    for i in range(50):
+        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_session_caps():
+    controller = ToolCallGuardrailController(
+        ToolCallGuardrailConfig(session_caps=SessionCapConfig(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
+  session_caps:
+    max_web_searches: 200      # max web_search calls per session (0 = unlimited)
+    max_subagents: 200         # max subagents spawned per session (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).
 
+### Session-wide runaway-loop caps
+
+Separate from the per-turn thresholds above, `session_caps` sets hard ceilings on the total number of `web_search` calls and subagent spawns over the **whole session** (not per turn). These are always on — a legitimate session almost never issues 200 web searches or spawns 200 subagents, so hitting a cap is a strong runaway-loop signal. 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. The counters reset only when a fresh session starts (`/new` or `/clear`). 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