feat: add opt-in subagent non-convergence guardrail

agent/agent_init.py

@@ -1594,6 +1594,27 @@ def init_agent(
         )
     except Exception as _tlg_err:
         _ra().logger.warning("Tool loop guardrail config ignored: %s", _tlg_err)
+
+    # Opt-in non-convergence tracking is scoped to delegate_task children. The
+    # setting lives under ``delegation`` and must not alter parent/cron agents.
+    agent._progress_tracker = None
+    try:
+        from agent.delegation_context import is_delegated_child_context
+        from agent.progress_tracker import ProgressTracker
+
+        if is_delegated_child_context():
+            _delegation_cfg = _agent_cfg.get("delegation", {})
+            _pt_cfg = (
+                _delegation_cfg.get("progress_tracker", {})
+                if isinstance(_delegation_cfg, dict)
+                else {}
+            )
+            _progress_tracker = ProgressTracker.from_mapping(_pt_cfg)
+            if _progress_tracker.enabled:
+                agent._progress_tracker = _progress_tracker
+    except Exception as _pt_err:
+        _ra().logger.warning("Progress tracker config ignored: %s", _pt_err)
+
     # Cache only the derived auxiliary compression context override that is
     # needed later by the startup feasibility check.  Avoid exposing a
     # broad pseudo-public config object on the agent instance.

agent/conversation_loop.py

@@ -6021,6 +6021,9 @@ def _perform_api_call(next_api_kwargs):
 
                 previous_msg = messages[-1] if messages else None
                 current_interim_visible = agent._interim_assistant_visible_text(assistant_msg)
+                current_interim_has_new_text = (
+                    agent._interim_assistant_has_new_visible_text(assistant_msg)
+                )
                 previous_interim_visible = (
                     agent._interim_assistant_visible_text(previous_msg)
                     if isinstance(previous_msg, dict)
@@ -6055,6 +6058,15 @@ def _perform_api_call(next_api_kwargs):
                         if tc.function.name in agent.valid_tool_names
                     ]
 
+                agent._start_progress_iteration(
+                    messages,
+                    tool_call_count=len(assistant_message.tool_calls or []),
+                    had_text=bool(
+                        current_interim_has_new_text
+                        and not duplicate_previous_interim
+                    ),
+                )
+
                 _tool_turn_persisted = None
                 try:
                     # Persist the assistant tool-call turn before any tool

agent/progress_tracker.py

@@ -0,0 +1,105 @@
+"""Opt-in non-convergence tracking for delegated child agents."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Literal, Mapping
+
+_DEFAULT_WARN_AFTER = 15
+_DEFAULT_HALT_AFTER = 25
+
+
+@dataclass(frozen=True)
+class ProgressDecision:
+    """Decision produced after one model/tool iteration."""
+
+    action: Literal["none", "warn", "halt"] = "none"
+    count: int = 0
+    message: str = ""
+
+
+def _positive_int(value: Any, default: int) -> int:
+    if isinstance(value, bool):
+        return default
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        return default
+    return parsed if parsed > 0 else default
+
+
+def _opt_in_bool(value: Any) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        return value.strip().lower() in {"1", "true", "yes", "on"}
+    return value == 1
+
+
+class ProgressTracker:
+    """Count delegated-child iterations that produce no convergence signal."""
+
+    def __init__(
+        self,
+        *,
+        warn_after: int = _DEFAULT_WARN_AFTER,
+        halt_after: int = _DEFAULT_HALT_AFTER,
+        enabled: Any = False,
+    ) -> None:
+        self.warn_after = _positive_int(warn_after, _DEFAULT_WARN_AFTER)
+        self.halt_after = max(
+            self.warn_after,
+            _positive_int(halt_after, _DEFAULT_HALT_AFTER),
+        )
+        self.enabled = _opt_in_bool(enabled)
+        self._iterations_since_progress = 0
+
+    @classmethod
+    def from_mapping(cls, data: Mapping[str, Any] | None) -> "ProgressTracker":
+        """Build a tracker from ``delegation.progress_tracker`` config."""
+        if not isinstance(data, Mapping):
+            return cls()
+        return cls(
+            enabled=data.get("enabled", False),
+            warn_after=data.get("warn_after", _DEFAULT_WARN_AFTER),
+            halt_after=data.get("halt_after", _DEFAULT_HALT_AFTER),
+        )
+
+    @property
+    def iterations_since_progress(self) -> int:
+        return self._iterations_since_progress
+
+    def finish_iteration(self, *, made_progress: bool) -> ProgressDecision:
+        """Record one complete model/tool round and return its decision."""
+        if not self.enabled:
+            return ProgressDecision()
+        if made_progress:
+            self._iterations_since_progress = 0
+            return ProgressDecision()
+
+        self._iterations_since_progress += 1
+        count = self._iterations_since_progress
+        if count >= self.halt_after:
+            return ProgressDecision(
+                action="halt",
+                count=count,
+                message=(
+                    f"Subagent stopped after {count} iterations without "
+                    "user-visible text or a successful file change."
+                ),
+            )
+        if count >= self.warn_after:
+            return ProgressDecision(
+                action="warn",
+                count=count,
+                message=(
+                    f"[PROGRESS TRACKER: {count} iterations without user-visible "
+                    "text or a successful file change. This subagent is not "
+                    "converging; finish the task or return the useful findings now.]"
+                ),
+            )
+        return ProgressDecision(count=count)
+
+    def reset(self) -> None:
+        """Clear state at the beginning of a new user turn."""
+        self._iterations_since_progress = 0

agent/tool_executor.py

@@ -149,6 +149,9 @@ def _flush_session_db_after_tool_progress(
     transcript survives destructive-but-valid tool calls.
     """
     try:
+        _prepare_progress = getattr(agent, "_prepare_progress_before_tool_flush", None)
+        if callable(_prepare_progress):
+            _prepare_progress(messages)
         persisted = agent._flush_messages_to_session_db(messages) is not False
         if not persisted:
             agent._incremental_persistence_failed = True
@@ -1160,6 +1163,7 @@ def _execute(next_args: dict[str, Any]) -> Any:
             tool_duration = 0.0
         else:
             function_name, function_args, function_result, tool_duration, is_error, blocked, middleware_trace = r
+            mutation_result = function_result
             name = function_name
             args = function_args
             progress_function_name = function_name
@@ -1185,7 +1189,7 @@ def _execute(next_args: dict[str, Any]) -> Any:
             if not blocked:
                 try:
                     agent._record_file_mutation_result(
-                        function_name, function_args, function_result, is_error,
+                        function_name, function_args, mutation_result, is_error,
                     )
                 except Exception as _ver_err:
                     logging.debug("file-mutation verifier record failed: %s", _ver_err)
@@ -1813,6 +1817,7 @@ def _execute(next_args: dict) -> Any:
         # Log tool errors to the persistent error log so [error] tags
         # in the UI always have a corresponding detailed entry on disk.
         _is_error_result, _ = _detect_tool_failure(function_name, function_result)
+        mutation_result = function_result
         # The agent-runtime tools above (todo, session_search, memory,
         # context-engine, memory-manager, clarify, delegate_task) are
         # dispatched inline — they never reach handle_function_call, so the
@@ -1857,7 +1862,7 @@ def _execute(next_args: dict) -> Any:
         if not _execution_blocked:
             try:
                 agent._record_file_mutation_result(
-                    function_name, function_args, function_result, _is_error_result,
+                    function_name, function_args, mutation_result, _is_error_result,
                 )
             except Exception as _ver_err:
                 logging.debug("file-mutation verifier record failed: %s", _ver_err)

agent/turn_context.py

@@ -466,6 +466,9 @@ def build_turn_context(
     agent._unicode_sanitization_passes = 0
     agent._tool_guardrails.reset_for_turn()
     agent._tool_guardrail_halt_decision = None
+    _progress_tracker = getattr(agent, "_progress_tracker", None)
+    if _progress_tracker is not None:
+        _progress_tracker.reset()
     _reset_consol = getattr(agent._memory_store, "reset_consolidation_failures", None)
     if callable(_reset_consol):
         _reset_consol()

cli-config.yaml.example

@@ -1254,6 +1254,14 @@ delegation:
   # provider: "openrouter"                    # Override provider for subagents (empty = inherit parent)
   #                                           # Resolves full credentials (base_url, api_key) automatically.
   #                                           # Supported: openrouter, nous, zai, kimi-coding, minimax
+  # Opt-in non-convergence circuit breaker for delegated children. It counts
+  # complete tool rounds with neither user-visible text nor a successful
+  # write_file/patch result. This complements the iteration budget; it does
+  # not determine whether read-only research is useful.
+  progress_tracker:
+    enabled: false                            # Disabled by default; enable only for runaway children
+    warn_after: 15                            # Add a convergence warning after N stalled rounds
+    halt_after: 25                            # Stop the child through the controlled-halt path
 
 # =============================================================================
 # Honcho Integration (Cross-Session User Modeling)

hermes_cli/config_defaults.py

@@ -1574,6 +1574,13 @@
         "inherit_mcp_toolsets": True,
         "max_iterations": 50,  # per-subagent iteration cap (each subagent gets its own budget,
                                # independent of the parent's max_iterations)
+        # Opt-in circuit breaker for delegated children that keep calling tools
+        # without producing user-visible text or landing a file mutation.
+        "progress_tracker": {
+            "enabled": False,
+            "warn_after": 15,
+            "halt_after": 25,
+        },
         # Subagent summaries return to the parent's context verbatim. A batch
         # fan-out (N children) returns N summaries at once, which can exceed
         # the parent's context window and trigger a compression/429 death

run_agent.py

@@ -3279,6 +3279,8 @@ def _record_file_mutation_result(
             return
         landed = file_mutation_result_landed(tool_name, result)
         if landed:
+            if getattr(self, "_progress_tracker", None) is not None:
+                self._progress_iteration_made_progress = True
             changed = getattr(self, "_turn_file_mutation_paths", None)
             if changed is not None:
                 changed.update(_extract_landed_file_mutation_paths(tool_name, args, result))
@@ -5613,6 +5615,28 @@ def _interim_assistant_visible_text(self, assistant_msg: Dict[str, Any]) -> str:
         content = assistant_msg.get("content")
         return self._strip_think_blocks(flatten_message_text(content)).strip()
 
+    def _interim_assistant_has_new_visible_text(
+        self, assistant_msg: Dict[str, Any]
+    ) -> bool:
+        """Return whether interim emission would surface previously unseen text."""
+        commentary_parts = self._extract_codex_interim_visible_parts(assistant_msg)
+        if commentary_parts:
+            pending_keys: set[str] = set()
+            for part in commentary_parts:
+                key = self._normalize_interim_visible_text(part)
+                if not key or key in pending_keys:
+                    continue
+                pending_keys.add(key)
+                if not self._interim_text_was_delivered(part):
+                    return True
+            return False
+        visible = self._interim_assistant_visible_text(assistant_msg)
+        return bool(
+            visible
+            and visible != "(empty)"
+            and not self._interim_text_was_delivered(visible)
+        )
+
     def _interim_text_was_delivered(self, text: str) -> bool:
         normalized = self._normalize_interim_visible_text(text)
         if not normalized:
@@ -6805,12 +6829,81 @@ def _compress_context(
             if token is not None:
                 reset_conversation_context(token)
 
+    def _start_progress_iteration(
+        self,
+        messages: list,
+        *,
+        tool_call_count: int,
+        had_text: bool,
+    ) -> None:
+        """Initialize one delegated-child tool round before persistence."""
+        if getattr(self, "_progress_tracker", None) is None:
+            return
+        self._progress_iteration_made_progress = False
+        self._progress_iteration_had_text = had_text
+        self._progress_iteration_remaining_results = max(0, int(tool_call_count))
+        self._progress_iteration_completed = False
+        if self._progress_iteration_remaining_results == 0:
+            self._complete_progress_iteration(messages)
+
+    def _prepare_progress_before_tool_flush(self, messages: list) -> None:
+        """Complete tracking immediately before the final tool result is durable."""
+        if getattr(self, "_progress_tracker", None) is None:
+            return
+        remaining = getattr(self, "_progress_iteration_remaining_results", 0)
+        if remaining <= 0:
+            return
+        remaining -= 1
+        self._progress_iteration_remaining_results = remaining
+        if remaining == 0:
+            self._complete_progress_iteration(messages)
+
+    def _complete_progress_iteration(self, messages: list) -> None:
+        """Evaluate the round and attach guidance before its final DB flush."""
+        tracker = getattr(self, "_progress_tracker", None)
+        if tracker is None or getattr(self, "_progress_iteration_completed", False):
+            return
+        self._progress_iteration_completed = True
+        made_progress = bool(
+            getattr(self, "_progress_iteration_had_text", False)
+            or getattr(self, "_progress_iteration_made_progress", False)
+        )
+        decision = tracker.finish_iteration(made_progress=made_progress)
+        if decision.action not in {"warn", "halt"}:
+            return
+
+        for message in reversed(messages):
+            if not isinstance(message, dict) or message.get("role") != "tool":
+                continue
+            content = message.get("content")
+            if isinstance(content, str):
+                message["content"] = f"{content}\n\n{decision.message}"
+            elif isinstance(content, list):
+                content.append({"type": "text", "text": decision.message})
+            break
+
+        if decision.action == "halt":
+            self._set_tool_guardrail_halt(
+                ToolGuardrailDecision(
+                    action="halt",
+                    code="subagent_non_convergence_halt",
+                    message=decision.message,
+                    count=decision.count,
+                )
+            )
+
     def _set_tool_guardrail_halt(self, decision: ToolGuardrailDecision) -> None:
         """Record the first guardrail decision that should stop this turn."""
         if decision.should_halt and self._tool_guardrail_halt_decision is None:
             self._tool_guardrail_halt_decision = decision
 
     def _toolguard_controlled_halt_response(self, decision: ToolGuardrailDecision) -> str:
+        if decision.code == "subagent_non_convergence_halt":
+            return (
+                "I stopped this delegated task because the non-convergence "
+                f"guardrail observed {decision.count} consecutive tool rounds "
+                "without user-visible text or a successful file change."
+            )
         tool = decision.tool_name or "a tool"
         return (
             f"I stopped retrying {tool} because it hit the tool-call guardrail "