Bot Loop Protection

Break runaway bot-to-bot reply loops with a sliding-window pair budget

BotLoopGuard breaks runaway bot-to-bot reply loops — two bots answering each other forever — by budgeting how many exchanges a pair may have inside a sliding window. Once a channel opts in with allow_bots: true, the guard is auto-instantiated per adapter — no code to wire.

# gateway.yaml
channels:
  telegram:
    token: ${TELEGRAM_BOT_TOKEN}
    allow_bots: true          # accept bot-authored inbound messages
    bot_loop_protection:      # break runaway A<->B reply loops
      max_events_per_window: 20

That’s it — the BotLoopGuard is auto-instantiated per adapter, and any A↔B pair that exceeds 20 exchanges in a 60s window is suppressed for a 60s cooldown.

Two bots ping-pong replies; the guard counts each exchange and, once a pair passes its budget, drops the turn and opens a cooldown.

Quick Start

1. Enable with allow_bots: true

   Flip the opt-in on the channel — everything else uses the default budget (20 exchanges / 60s window / 60s cooldown).

   # gateway.yaml
   channels:
     telegram:
       token: ${TELEGRAM_BOT_TOKEN}
       allow_bots: true

2. Tune the budget

   Add bot_loop_protection keys to change how quickly a loop trips and how long it stays quiet.

   # gateway.yaml
   channels:
     telegram:
       token: ${TELEGRAM_BOT_TOKEN}
       allow_bots: true
       bot_loop_protection:
         max_events_per_window: 10
         window_seconds: 30
         cooldown_seconds: 120

   The same two fields exist on BotConfig for the Python constructor path:

   from praisonaiagents import Agent
   from praisonaiagents.bots import BotConfig
   
   agent = Agent(
       name="Group Helper",
       instructions="Answer questions posted in the room.",
   )
   
   config = BotConfig(
       allow_bots=True,
       bot_loop_protection={
           "max_events_per_window": 10,
           "window_seconds": 30,
           "cooldown_seconds": 120,
       },
   )

3. Advanced — wire the primitive yourself

   Custom adapters that do not inherit MessageHookMixin can call BotLoopGuard.observe(...) directly — this is the escape hatch, not the recommended path.

   from praisonaiagents.bots import BotLoopGuard, BotLoopPolicy
   
   guard = BotLoopGuard(BotLoopPolicy(max_events_per_window=20, window_seconds=60))
   
   # In your bot inbound handler, before dispatching to the agent:
   if sender_is_bot and not guard.observe(
       self_bot_id="mybot",
       sender_bot_id=inbound_sender_id,
   ):
       return  # drop the turn — pair is over budget / in cooldown

How It Works

The guard tracks each bot pair in a sliding window; exchange #21 trips the budget and both bots go quiet for the cooldown.

A → B and B → A collapse to the same pair key, so a ping-pong loop accrues one budget instead of two independent streams.

How adapters wire it

Every adapter inherits MessageHookMixin.bot_loop_allows(sender, self_bot_id=...), the shared helper that folds the allow_bots opt-in and the per-pair budget into one verdict — a human sender always passes, a bot sender passes only when allow_bots: true and the pair is under budget.

When to Enable

BotLoopGuard only matters when your gateway accepts bot-authored inbound messages and multiple bots can share a room.

Configuration Options

Two BotConfig fields turn the guard on and shape its budget. Both match praisonaiagents/bots/config.py.

bot_loop_protection accepts the same keys as BotLoopPolicy, coerced via BotLoopPolicy.from_dict(...). Every field matches praisonaiagents/bots/silence.py.

BotLoopPolicy.from_dict(data) builds a policy from a mapping; None returns the enabled default and unknown keys are ignored.

BotLoopPolicy.from_dict tolerates malformed numeric values — a typo like max_events_per_window: "abc" falls back to the field default instead of crashing the inbound path on the first bot message.

The runtime BotLoopGuard exposes this surface:

Common Patterns

Once allow_bots: true is set the guard runs automatically; these primitive-level patterns cover custom adapters and tests.

Group room with multiple assistants — use one guard per gateway and call observe() only when the sender is a bot.

from praisonaiagents.bots import BotLoopGuard

guard = BotLoopGuard()

def on_inbound(msg, sender_is_bot):
    if sender_is_bot and not guard.observe(
        self_bot_id="mybot", sender_bot_id=msg.sender_id
    ):
        return  # loop budget hit — drop silently
    handle(msg)

Testing with an injected clock — pass explicit now= values for deterministic time-dependent tests.

from praisonaiagents.bots import BotLoopGuard, BotLoopPolicy

guard = BotLoopGuard(BotLoopPolicy(max_events_per_window=2, window_seconds=60.0))
assert guard.observe(self_bot_id="a", sender_bot_id="b", now=0.0)
assert guard.observe(self_bot_id="a", sender_bot_id="b", now=1.0)
assert not guard.observe(self_bot_id="a", sender_bot_id="b", now=2.0)

Resetting on gateway restart — call reset() after redeploy so leftover cooldowns from an earlier process don’t leak.

guard.reset()  # clear tracked pairs and cooldowns

When It Fires vs Doesn’t

Only bot-authored inbound turns reach the guard; humans and single-bot deployments never call observe().

Best Practices

Related