fix: hard stop tool loops on non-interactive platforms

This commit is contained in:
João Vitor Cunha
2026-06-19 14:44:04 -03:00
committed by Teknium
parent 6879a621b1
commit ee2147f9e6
7 changed files with 82 additions and 10 deletions
+2 -1
View File
@@ -1882,7 +1882,8 @@ def init_agent(
try:
agent._tool_guardrails = ToolCallGuardrailController(
ToolCallGuardrailConfig.from_mapping(
_agent_cfg.get("tool_loop_guardrails", {})
_agent_cfg.get("tool_loop_guardrails", {}),
platform=platform,
)
)
except Exception as _tlg_err:
+23 -5
View File
@@ -110,12 +110,14 @@ class ToolCallGuardrailConfig:
"""Thresholds for per-turn tool-call loop detection.
Warnings are enabled by default and never prevent tool execution. Hard stops
are explicit opt-in so interactive CLI/TUI sessions get a gentle nudge unless
the user enables circuit-breaker behavior in config.yaml.
stay opt-in for interactive CLI/TUI sessions, but default on for
non-interactive gateway/cron platforms where nobody is present to interrupt
a model that ignores loop warnings.
"""
warnings_enabled: bool = True
hard_stop_enabled: bool = False
non_interactive_hard_stop_enabled: bool = True
exact_failure_warn_after: int = 2
exact_failure_block_after: int = 5
same_tool_failure_warn_after: int = 3
@@ -127,10 +129,10 @@ class ToolCallGuardrailConfig:
loop_caps: "LoopCapConfig" = field(default_factory=lambda: LoopCapConfig())
@classmethod
def from_mapping(cls, data: Mapping[str, Any] | None) -> "ToolCallGuardrailConfig":
def from_mapping(cls, data: Mapping[str, Any] | None, *, platform: str | None = None) -> "ToolCallGuardrailConfig":
"""Build config from the `tool_loop_guardrails` config.yaml section."""
if not isinstance(data, Mapping):
return cls()
data = {}
warn_after = data.get("warn_after")
if not isinstance(warn_after, Mapping):
@@ -140,9 +142,18 @@ class ToolCallGuardrailConfig:
hard_stop_after = {}
defaults = cls()
hard_stop_enabled = _as_bool(data.get("hard_stop_enabled"), defaults.hard_stop_enabled)
non_interactive_hard_stop_enabled = _as_bool(
data.get("non_interactive_hard_stop_enabled"),
defaults.non_interactive_hard_stop_enabled,
)
if _is_non_interactive_platform(platform) and non_interactive_hard_stop_enabled:
hard_stop_enabled = True
return cls(
warnings_enabled=_as_bool(data.get("warnings_enabled"), defaults.warnings_enabled),
hard_stop_enabled=_as_bool(data.get("hard_stop_enabled"), defaults.hard_stop_enabled),
hard_stop_enabled=hard_stop_enabled,
non_interactive_hard_stop_enabled=non_interactive_hard_stop_enabled,
exact_failure_warn_after=_positive_int(
warn_after.get("exact_failure", data.get("exact_failure_warn_after")),
defaults.exact_failure_warn_after,
@@ -218,6 +229,13 @@ class LoopCapConfig:
)
def _is_non_interactive_platform(platform: str | None) -> bool:
"""Return true for gateway/cron sessions where tool loops are unattended."""
if not isinstance(platform, str) or not platform.strip():
return False
return platform.strip().lower() not in {"cli", "tui"}
@dataclass(frozen=True)
class IdenticalCallObservation:
"""Outcome of observing one completed tool call for the stall guards.
+5 -3
View File
@@ -533,12 +533,14 @@ browser:
# Tool Loop Guardrails
# =============================================================================
# Soft warnings are enabled by default. They append guidance to repeated failed
# or non-progressing tool results but still let the tool execute. Hard stops are
# opt-in circuit breakers for autonomous/cron sessions where stopping a loop is
# preferable to spending the full iteration budget.
# or non-progressing tool results but still let the tool execute. Hard stops stay
# opt-in for interactive CLI/TUI sessions, but default on for non-interactive
# gateway/cron sessions where nobody is present to interrupt a model that
# ignores loop warnings.
tool_loop_guardrails:
warnings_enabled: true
hard_stop_enabled: false
non_interactive_hard_stop_enabled: true
warn_after:
exact_failure: 2
same_tool_failure: 3
+1
View File
@@ -1100,6 +1100,7 @@ def _ensure_hermes_home_managed(home: Path):
from hermes_cli.config_defaults import DEFAULT_CONFIG, OPTIONAL_ENV_VARS # noqa: F401
# =============================================================================
# Config Migration System
# =============================================================================
+1
View File
@@ -320,6 +320,7 @@ LEGACY_AUTHOR_MAP = {
"pinkiilqwq@users.noreply.github.com": "PINKIIILQWQ", # PR #45035 salvage (resume-to-tip; #38763)
"pink@PinkdeMacBook-Air.local": "PINKIIILQWQ", # PR #45035 local git identity (resume-to-tip; #38763)
"ailang323@163.com": "ailang323", # PR #48682 salvage (compression-tip predicate; #38763)
"59806492+sitkarev@users.noreply.github.com": "sitkarev",
"zheng@omegasys.eu": "omegazheng",
"220877172+james47kjv@users.noreply.github.com": "james47kjv",
+32
View File
@@ -33,6 +33,18 @@ def test_tool_call_signature_hashes_canonical_nested_unicode_args_without_exposi
assert "☤" not in json.dumps(metadata)
def test_default_config_is_soft_warning_only_with_hard_stop_disabled():
cfg = ToolCallGuardrailConfig()
assert cfg.warnings_enabled is True
assert cfg.hard_stop_enabled is False
assert cfg.non_interactive_hard_stop_enabled is True
assert cfg.exact_failure_warn_after == 2
assert cfg.same_tool_failure_warn_after == 3
assert cfg.no_progress_warn_after == 2
assert cfg.exact_failure_block_after == 5
assert cfg.same_tool_failure_halt_after == 8
assert cfg.no_progress_block_after == 5
def test_config_parses_nested_warn_and_hard_stop_thresholds():
@@ -63,6 +75,26 @@ def test_config_parses_nested_warn_and_hard_stop_thresholds():
assert cfg.no_progress_block_after == 8
def test_gateway_platform_defaults_to_hard_stop_without_changing_cli_default():
cli_cfg = ToolCallGuardrailConfig.from_mapping({}, platform="cli")
telegram_cfg = ToolCallGuardrailConfig.from_mapping({}, platform="telegram")
cron_cfg = ToolCallGuardrailConfig.from_mapping({}, platform="cron")
assert cli_cfg.hard_stop_enabled is False
assert telegram_cfg.hard_stop_enabled is True
assert cron_cfg.hard_stop_enabled is True
def test_non_interactive_hard_stop_can_be_disabled_explicitly():
cfg = ToolCallGuardrailConfig.from_mapping(
{"non_interactive_hard_stop_enabled": False},
platform="telegram",
)
assert cfg.hard_stop_enabled is False
assert cfg.non_interactive_hard_stop_enabled is False
def test_default_repeated_identical_failed_call_warns_without_blocking():
controller = ToolCallGuardrailController()
args = {"query": "same"}
@@ -36,7 +36,12 @@ def _mock_response(content="Hello", finish_reason="stop", tool_calls=None):
return SimpleNamespace(choices=[choice], model="test/model", usage=None)
def _make_agent(*tool_names: str, max_iterations: int = 10, config: dict | None = None) -> AIAgent:
def _make_agent(
*tool_names: str,
max_iterations: int = 10,
config: dict | None = None,
platform: str | None = None,
) -> AIAgent:
with (
patch("run_agent.get_tool_definitions", return_value=_make_tool_defs(*tool_names)),
patch("run_agent.check_toolset_requirements", return_value={}),
@@ -51,6 +56,7 @@ def _make_agent(*tool_names: str, max_iterations: int = 10, config: dict | None
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
platform=platform or "cli",
)
agent.client = MagicMock()
agent._cached_system_prompt = "You are helpful."
@@ -86,6 +92,17 @@ def _hard_stop_config(**overrides) -> dict:
return cfg
def test_gateway_platform_uses_hard_stop_default_without_cli_opt_in():
agent = _make_agent("web_search", platform="telegram")
args = {"query": "same"}
_seed_exact_failures(agent, "web_search", args, count=5)
decision = getattr(agent, "_tool_guardrails").before_call("web_search", args)
assert decision.action == "block"
assert decision.code == "repeated_exact_failure_block"
def test_default_sequential_path_warns_repeated_exact_failure_without_blocking_execution():
agent = _make_agent("web_search")
args = {"query": "same"}