Detect and prevent degenerate agent loops.
Features:
1. Hash tracking: SHA-256 of (tool_name, args) blocks after max_identical_calls
2. Ping-pong detection: Sliding window detects A-B-A-B or A-B-C-A-B-C patterns
3. Poll-tool awareness: Tools with spec.metadata["polling"] = True
get relaxed budget
4. Context overflow recovery: 4-stage compression of message history
Source code in src/openjarvis/agents/loop_guard.py
| def __init__(self, config: LoopGuardConfig, *, bus: Optional[EventBus] = None):
self._config = config
self._bus = bus
# Track call hashes and their counts
self._call_counts: dict[str, int] = {}
# Track tool name sequence for pattern detection
self._tool_sequence: deque[str] = deque(maxlen=config.ping_pong_window * 2)
# Track per-tool call counts (for polling budget)
self._per_tool_counts: dict[str, int] = {}
# Track cycle keys that have already been warned (for warn-before-block)
self._warned_cycles: set[str] = set()
try:
from openjarvis._rust_bridge import get_rust_module
_rust = get_rust_module()
self._rust_impl = _rust.LoopGuard(
max_identical=config.max_identical_calls,
max_ping_pong=(
config.ping_pong_window // 2 if config.ping_pong_window > 1 else 2
),
poll_budget=config.poll_tool_budget,
)
except Exception:
self._rust_impl = None
|
Functions
check_call(tool_name: str, arguments: str) -> LoopVerdict
Check whether a tool call should proceed or be blocked.
Source code in src/openjarvis/agents/loop_guard.py
| def check_call(self, tool_name: str, arguments: str) -> LoopVerdict:
"""Check whether a tool call should proceed or be blocked."""
if self._rust_impl is not None:
rust_result = self._rust_impl.check(tool_name, arguments)
# Support both raw Rust return (str | None) and LoopVerdict
if isinstance(rust_result, LoopVerdict):
verdict = rust_result
elif rust_result is not None:
self._emit_triggered("rust_guard", tool_name)
verdict = LoopVerdict(blocked=True, reason=rust_result)
else:
verdict = LoopVerdict()
else:
verdict = self._python_check(tool_name, arguments)
# Wrap with warn-before-block logic
if verdict.blocked and self._config.warn_before_block:
cycle_key = verdict.reason
if cycle_key not in self._warned_cycles:
self._warned_cycles.add(cycle_key)
return LoopVerdict(blocked=False, warned=True, reason=verdict.reason)
return verdict
|
Check whether an agent response indicates a loop. Reserved for future use.
Source code in src/openjarvis/agents/loop_guard.py
| def check_response(self, content: str) -> LoopVerdict:
"""Check whether an agent response indicates a loop. Reserved for future use."""
return LoopVerdict()
|
compress_context(messages: list) -> list
Apply 4-stage context overflow recovery to message list.
Stages:
1. Summarize old tool results (replace content with "[Tool result truncated]")
2. Sliding window — keep only recent messages
3. Drop tool call/result pairs from the middle
4. Truncate to system + last 2 exchanges
Source code in src/openjarvis/agents/loop_guard.py
| def compress_context(self, messages: list) -> list:
"""Apply 4-stage context overflow recovery to message list.
Stages:
1. Summarize old tool results (replace content with "[Tool result truncated]")
2. Sliding window — keep only recent messages
3. Drop tool call/result pairs from the middle
4. Truncate to system + last 2 exchanges
"""
if len(messages) <= self._config.max_context_messages:
return messages
# Stage 1: Truncate old tool result messages
threshold = len(messages) // 2
compressed = []
for i, msg in enumerate(messages):
if i < threshold and self._is_tool(msg):
from openjarvis.core.types import Message, Role
compressed.append(
Message(
role=Role.TOOL,
content="[Tool result truncated]",
tool_call_id=getattr(
msg,
"tool_call_id",
None,
),
name=getattr(msg, "name", None),
)
)
else:
compressed.append(msg)
if len(compressed) <= self._config.max_context_messages:
return compressed
# Stage 2: Sliding window — keep system + recent
system_msgs = [m for m in compressed if self._is_system(m)]
non_system = [m for m in compressed if not self._is_system(m)]
window_size = self._config.max_context_messages - len(system_msgs)
if len(non_system) > window_size:
non_system = non_system[-window_size:]
compressed = system_msgs + non_system
if len(compressed) <= self._config.max_context_messages:
return compressed
# Stage 3: Drop tool call/result pairs from middle
keep_start = max(
len(system_msgs),
len(compressed) // 10,
)
keep_end = len(compressed) // 2
compressed = compressed[:keep_start] + compressed[-keep_end:]
if len(compressed) <= self._config.max_context_messages:
return compressed
# Stage 4: Extreme — system + last 2 exchanges
sys_final = [m for m in compressed if self._is_system(m)]
tail = [m for m in compressed if not self._is_system(m)]
return sys_final + tail[-4:]
|
Reset all tracking state — always via Rust backend.
Source code in src/openjarvis/agents/loop_guard.py
| def reset(self) -> None:
"""Reset all tracking state — always via Rust backend."""
self._call_counts.clear()
self._tool_sequence.clear()
self._per_tool_counts.clear()
self._warned_cycles.clear()
if self._rust_impl is not None:
self._rust_impl.reset()
|