IDEMPOTENT_TOOLS = {
"read_file", "list_dir", "web_search", "web_fetch",
"browser_snapshot", "grep", "glob", "get_file_info",
"search_files", "view_image", "memory_recall", "get_env",
}
import hashlib, json
from dataclasses import dataclass
from typing import Literal
Action = Literal["allow", "warn", "block", "halt"]
@dataclass
class LoopGuardConfig:
warnings_enabled: bool = True
hard_stop_enabled: bool = False
exact_warn_after: int = 2
exact_block_after: int = 5
tool_warn_after: int = 3
tool_halt_after: int = 8
noprog_warn_after: int = 2
noprog_block_after: int = 5
@dataclass
class LoopGuardDecision:
action: Action
message: str = ""
def _hash_args(args: dict) -> str:
return hashlib.sha256(
json.dumps(args, sort_keys=True, ensure_ascii=False).encode()
).hexdigest()[:16]
class ToolLoopGuard:
def __init__(self, cfg: LoopGuardConfig | None = None):
self._cfg = cfg or LoopGuardConfig()
self._exact_fails: dict[str, int] = {}
self._tool_fails: dict[str, int] = {}
self._noprog: dict[str, tuple[int, str]] = {}
def before_call(self, tool: str, args: dict) -> LoopGuardDecision:
key = f"{tool}:{_hash_args(args)}"
if (self._exact_fails.get(key, 0) >= self._cfg.exact_block_after
and self._cfg.hard_stop_enabled):
n = self._exact_fails[key]
return LoopGuardDecision("block",
f"{tool} failed {n}x with identical args — hard stop")
if (self._tool_fails.get(tool, 0) >= self._cfg.tool_halt_after
and self._cfg.hard_stop_enabled):
n = self._tool_fails[tool]
return LoopGuardDecision("halt", f"{tool} failed {n}x this turn — halting")
return LoopGuardDecision("allow")
def after_call(self, tool: str, args: dict,
result: str, success: bool) -> LoopGuardDecision:
key = f"{tool}:{_hash_args(args)}"
if not success:
self._exact_fails[key] = self._exact_fails.get(key, 0) + 1
self._tool_fails[tool] = self._tool_fails.get(tool, 0) + 1
n = self._exact_fails[key]
if n >= self._cfg.exact_warn_after and self._cfg.warnings_enabled:
return LoopGuardDecision("warn",
f"{tool} failed {n}x with same args — try different approach")
return LoopGuardDecision("allow")
if tool in IDEMPOTENT_TOOLS:
rh = hashlib.sha256(result.encode()).hexdigest()[:16]
prev_n, prev_h = self._noprog.get(key, (0, ""))
if rh == prev_h:
new_n = prev_n + 1
self._noprog[key] = (new_n, rh)
if new_n >= self._cfg.noprog_block_after and self._cfg.hard_stop_enabled:
return LoopGuardDecision("block",
f"{tool} returned identical result {new_n}x — no progress")
if new_n >= self._cfg.noprog_warn_after and self._cfg.warnings_enabled:
return LoopGuardDecision("warn",
f"{tool} returned same result {new_n}x — change query")
else:
self._noprog[key] = (0, rh)
return LoopGuardDecision("allow")
def reset(self) -> None:
"""Call once at turn start — not per tool call."""
self._exact_fails.clear()
self._tool_fails.clear()
self._noprog.clear()
guard = ToolLoopGuard(LoopGuardConfig(hard_stop_enabled=True))
guard.reset()
async def dispatch_tool(tool: str, args: dict) -> str:
pre = guard.before_call(tool, args)
if pre.action in ("block", "halt"):
return f"[LOOP-GUARD] {pre.message}"
if pre.action == "warn":
print(f"[WARN] {pre.message}")
try:
result, success = await run_tool(tool, args), True
except Exception as e:
result, success = str(e), False
post = guard.after_call(tool, args, result, success)
if post.action == "warn":
print(f"[WARN] {post.message}")
if post.action in ("block", "halt") and guard._cfg.hard_stop_enabled:
return f"[LOOP-GUARD] {post.message}"
return result
❌ reset() called per-call instead of per-turn — counts never accumulate
❌ Mutating tools in IDEMPOTENT_TOOLS — write_file returning same result is normal
❌ hard_stop_enabled=True in interactive mode — blocks user without escape
❌ Arg hash on unsorted dict — same args hash differently each call
❌ Result hash on truncated output — large identical results bypass dedup
