Loop Guard

Always-on idempotency-aware safety net for tool execution loops

Loop Guard stops broken or misconfigured tools from burning tokens by counting per-turn calls and reacting differently to safe-to-repeat vs state-changing tools.

from praisonaiagents import Agent

agent = Agent(
    name="Safe Coder",
    instructions="Use tools carefully.",
)
agent.start("Fix the failing unit test in src/utils.py")

The user sends a task; Loop Guard tracks tool calls per turn and blocks runaway loops automatically.

Quick Start

1. It's already on

   from praisonaiagents import Agent
   
   agent = Agent(
       instructions="You are a helpful assistant."
   )
   
   # Loop Guard is initialised automatically.
   # No flag needed — every Agent gets one.
   agent.start("Summarise the README")

2. Tune the thresholds

   For power users, you can customize thresholds after agent creation:

   from praisonaiagents import Agent
   from praisonaiagents.escalation.loop_guard import LoopGuard, LoopGuardConfig
   
   agent = Agent(instructions="Polls a slow status endpoint")
   
   agent._loop_guard = LoopGuard(LoopGuardConfig(
       idempotent_warn_threshold=10,
       idempotent_block_threshold=20,
       idempotent_halt_threshold=30,
       max_time_per_turn=300.0,
   ))

   Per-turn guards reuse this guard’s config (created lazily on the turn’s first tool call), so one agent._loop_guard = ... reconfiguration applies to every subsequent turn, streaming or non-streaming, sync or async.

3. Turn it off

   from praisonaiagents import Agent
   from praisonaiagents.escalation.loop_guard import LoopGuard, LoopGuardConfig
   
   agent = Agent(instructions="Agent without safety guard")
   
   agent._loop_guard = LoopGuard(LoopGuardConfig(enabled=False))

How It Works

check() is bracketed by a reset_turn() at _chat_impl / _achat_impl entry, so every turn starts with fresh counts and a fresh time budget. Overlapping concurrent turns on the same Agent each get their own LoopGuard (same config, independent counters), so one caller’s traffic can no longer exhaust another caller’s per-turn limits. agent.execute_tool(...) called outside a chat turn still uses the agent-level guard. Agents with no tools never get a Loop Guard force-created.

User Interaction Flow

Here’s what happens when an agent repeatedly calls the same tool:

Two signals now drive the outcome: how many times the tool has been called this turn, and whether the results are still changing. A busy-but-progressing tool sees at most a WARN; a stuck-with-identical-results tool escalates through WARN to HALT.

Tool Classification

Loop Guard categorizes tools into two buckets with different thresholds:

Idempotent tools (safe to repeat): read_file, list_files, search_files, web_search, get_memory, git_status, git_log, db_query, etc.

Mutating tools (state-changing): write_file, edit_file, delete_file, execute_code, shell, git_commit, git_push, sql_insert, install_package, etc.

Configuration Options

GuardAction values: ALLOW, WARN, BLOCK, HALT

LoopGuardDecision fields: action, code, message, metadata

No-Progress Detection

No-progress detection watches for identical repeated results, not raw tool-call counts.

What counts as progress (resets the streak):

• The result changes for the same tool + same args (e.g. check_status(job=42) returning "IN_PROGRESS" then "COMPLETE").
• A different tool is called (streaks require the same tool, not just the same result value).
• agent._loop_guard.mark_progress("...") is called — marks the current position so only later tool calls can contribute to a future streak.

What does NOT reset the streak:

• The same tool returning the same result over and over — that is the “stuck” pattern the guard exists to catch.
• Falsy repeated results ("", [], {}, 0, False) — these are wrapped internally so they still fingerprint, closing a loophole that used to disable no-progress detection.

Count thresholds are also stuck-aware. The idempotent (5 / 8 / 12) and mutating (3 / 5 / 7) thresholds still count tool calls, but a BLOCK or HALT from those thresholds is downgraded to a WARN when the trailing identical-result run is below no_progress_warn. So a legitimately busy but progressing tool (many calls, changing output) will not be blocked purely on frequency.

Example: polling that used to be halted, now allowed

from praisonaiagents import Agent

agent = Agent(
    name="Deployment Watcher",
    instructions="Poll the deployment status until it finishes.",
)

# check_deploy(job='X') returns 'IN_PROGRESS', 'IN_PROGRESS', 'RUNNING', 'COMPLETE'.
# Each transition changes the result-hash, so the no-progress streak stays at 1.
agent.start("Watch deploy-42 until it is done.")

Example: genuine stall, still halted

from praisonaiagents import Agent

agent = Agent(
    name="Config Reader",
    instructions="Read a config file that never appears.",
)

# read_file('missing.yaml') returns the same error result every time.
# The streak grows to no_progress_halt=8 and the tool is halted with
# ToolExecutionError.
agent.start("Read missing.yaml.")

Manually marking progress

If your agent completes a logical sub-step (a checkpoint, a milestone), call mark_progress() so any future no-progress streak starts fresh from that point:

from praisonaiagents import Agent

agent = Agent(instructions="Multi-step data pipeline")

# ... run some tools ...
agent._loop_guard.mark_progress("stage-1-complete")
# Only tool calls made AFTER this marker can contribute to a future streak.

What happens at each threshold

Sync and Async coverage

Loop Guard applies identically to both execution paths — sync (agent.start() / agent.chat()) and async (agent.achat() / async workflows).

What triggers Loop Guard on the async path:

• Repeated identical async tool calls whose results do not change (no_progress_* streak grows).
• Repeated async tool timeouts — each asyncio.wait_for timeout is now recorded as a failure and counts toward the streak.
• Repeated raised exceptions inside an async tool — each raised call is recorded as a failure the next pre-execution check can act on.

from praisonaiagents import Agent

agent = Agent(instructions="Poll a flaky endpoint.")

# Async path: identical failing calls escalate WARN → HALT and the run stops.
result = await agent.achat("Fetch /status until it returns COMPLETE.")

The sync and async paths share the same LoopGuard instance per Agent, so a mixed workload (sync then async on the same agent) accumulates counts against one turn’s thresholds.

Common Patterns

Async polling with changing results

Polling tools whose output actually transitions state (e.g. pending → running → done) do not need any config change — the no-progress guard sees each transition as progress and never trips. Only the count-based idempotent thresholds (5 / 8 / 12) could otherwise fire on high call volume, and those are now downgraded to WARN while results keep changing.

from praisonaiagents import Agent

agent = Agent(
    name="Job Poller",
    instructions="Call check_job_status until it returns COMPLETE.",
)

# Default LoopGuardConfig is fine — no tuning required.
agent.start("Wait for job-42 to complete.")

If your polling tool paces itself with a wait / sleep tool that returns the same value each call, wrap the pair (check_job_status + wait) with an explicit progress marker between transitions so wait‘s identical returns do not build a stuck streak:

from praisonaiagents import Agent

agent = Agent(instructions="Poll and pace")
# ... after each observed status change ...
agent._loop_guard.mark_progress("status-transition")

Polling a slow status endpoint

from praisonaiagents import Agent
from praisonaiagents.escalation.loop_guard import LoopGuard, LoopGuardConfig

agent = Agent(instructions="Monitor deployment status")

# Increase idempotent thresholds for polling scenarios
agent._loop_guard = LoopGuard(LoopGuardConfig(
    idempotent_warn_threshold=15,
    idempotent_block_threshold=25,
    idempotent_halt_threshold=40,
    max_time_per_turn=300.0,  # 5 minute timeout
))

Strict mode for production database agents

from praisonaiagents import Agent
from praisonaiagents.escalation.loop_guard import LoopGuard, LoopGuardConfig

agent = Agent(instructions="Manage user database")

# Lower mutating thresholds for safety
agent._loop_guard = LoopGuard(LoopGuardConfig(
    mutating_warn_threshold=2,
    mutating_block_threshold=3,
    mutating_halt_threshold=4,
))

Shared agent on a web server / gateway

import threading
from praisonaiagents import Agent

def write_file(path: str) -> str:
    """Write a file (distinct path each call)."""
    return f"wrote {path}"

# One Agent, reused across many overlapping requests
agent = Agent(name="svc", instructions="Write the requested file.", tools=[write_file])

def handle_request(i: int) -> str:
    # Each chat() call is one turn, and each turn gets its own LoopGuard
    return agent.chat(f"Write /tmp/out_{i}.txt")

threads = [threading.Thread(target=handle_request, args=(i,)) for i in range(20)]
for t in threads: t.start()
for t in threads: t.join()

Each chat() / achat() turn, and each streaming generator, binds its own LoopGuard on a ContextVar that is copied into tool worker threads — so twenty overlapping writes to twenty different paths neither warn nor HALT each other.

Long-lived agents

from praisonaiagents import Agent

def lookup_order(q: str) -> str:
    """Look up an order status."""
    return f"order-status-for-{q}"

agent = Agent(
    name="Support Bot",
    instructions="Answer order-status questions.",
    tools=[lookup_order],
)

# Each turn starts clean: no loop-guard warnings from earlier turns,
# and no Turn time limit error after minutes of agent life.
for i in range(16):
    print(agent.chat(f"q{i}"))

Observability with stats

agent = Agent(instructions="Data processing agent")

# Get statistics for dashboards
stats = agent._loop_guard.get_stats()
print(f"Turn elapsed: {stats['turn_elapsed']}s")
print(f"Tool counts: {stats['tool_counts']}")

stats["turn_elapsed"] is the current turn’s elapsed time, not an agent-lifetime value.

Relationship to Loop Detection

Loop Detection is also always-on as of PR #3005 — every Agent runs result-aware name + args + result-hash detection in the same tool-execution path. The two are complementary: Loop Guard counts per-turn tool calls with idempotent-vs-mutating thresholds, while Loop Detection catches repeated identical fingerprints at any frequency and back-fills result hashes so polling with progress is not flagged. Both fire through the same blocked_result path, so trace spans, stream events, and AFTER_TOOL hooks stay consistent regardless of which one triggers.

Best Practices

Related