Escalation Doom-Loop Detection

Session-level doom-loop detection that records actions and progress markers and recommends a recovery action

The escalation DoomLoopDetector watches a long-running agent session for stuck states — repeated actions, repeated failures, no progress, resource exhaustion, or repeated output — and recommends a RecoveryAction so the agent can break out instead of burning its whole iteration budget.

from praisonaiagents.escalation import DoomLoopDetector

detector = DoomLoopDetector()
detector.start_session()

detector.record_action("read_file", {"path": "foo.py"}, "content", success=True)
detector.record_action("read_file", {"path": "foo.py"}, "content", success=True)
detector.record_action("read_file", {"path": "foo.py"}, "content", success=True)

if detector.is_doom_loop():
    print(detector.get_recovery_action())  # e.g. RecoveryAction.RETRY_DIFFERENT

import { DoomLoopDetector, RecoveryAction } from 'praisonai';

const detector = new DoomLoopDetector();
detector.startSession();

detector.recordAction('read_file', { path: 'foo.py' }, 'content', true);
detector.recordAction('read_file', { path: 'foo.py' }, 'content', true);
detector.recordAction('read_file', { path: 'foo.py' }, 'content', true);

if (detector.isDoomLoop()) {
  console.log(detector.getRecoveryAction()); // e.g. RecoveryAction.RETRY_DIFFERENT
}

The detector consumes the actions and progress markers you feed it, classifies the stuck state into a DoomLoopType, and maps that to a RecoveryAction on the recovery ladder.

Quick Start

1. Construct with defaults

   from praisonaiagents.escalation import DoomLoopDetector
   
   detector = DoomLoopDetector()
   detector.start_session()

2. Record actions and progress markers

   detector.record_action("search", {"q": "config"}, "no results", success=False)
   detector.mark_progress("wrote initial draft")  # call on real progress

3. Check for a loop and get the recovery action

   if detector.is_doom_loop():
       event = detector.get_loop_event()
       print(event.loop_type, event.description)
       print(detector.get_recovery_action())

How It Works

The detector keeps an ordered history of record_action(...) calls plus a separate list of mark_progress(...) markers. After each recorded action it re-runs every check; is_doom_loop() returns True as soon as any check fires, and get_loop_type() reports which one.

Loop types

Recovery ladder

get_recovery_action() walks a ladder based on how many recovery attempts have already been made:

Configuration Options

Pass a DoomLoopConfig to tune thresholds:

from praisonaiagents.escalation import DoomLoopDetector, DoomLoopConfig

detector = DoomLoopDetector(
    DoomLoopConfig(
        max_no_progress_steps=3,
        max_consecutive_failures=2,
        escalate_on_loop=True,
    )
)

How progress markers work

mark_progress() records a (marker, timestamp) tuple. When checking NO_PROGRESS, the detector only counts markers within the current no-progress window — markers older than the window boundary (the timestamp of the action immediately preceding the window) are filtered out.

from praisonaiagents.escalation import DoomLoopDetector, DoomLoopConfig

detector = DoomLoopDetector(DoomLoopConfig(max_no_progress_steps=5))
detector.start_session()

detector.mark_progress("read config.yaml")  # early success

# ... many later steps that repeat the same unproductive result ...
for _ in range(5):
    detector.record_action("retry", {"n": 1}, "same error", success=False)

detector.is_doom_loop()  # True — the early marker no longer suppresses NO_PROGRESS

Before this recency filter (PR #3960), one early mark_progress(...) disabled NO_PROGRESS detection for the rest of the run, no matter how long the agent then spun on unproductive work. Now NO_PROGRESS fires correctly on long runs where an early success previously suppressed it forever.

Common Patterns

Drive recovery from the detected action

from praisonaiagents.escalation import DoomLoopDetector
from praisonaiagents.escalation.doom_loop import RecoveryAction

detector = DoomLoopDetector()
detector.start_session()

# ... record actions during the run ...

if detector.is_doom_loop():
    action = detector.get_recovery_action()
    if action == RecoveryAction.RETRY_DIFFERENT:
        detector.increment_recovery()
        # switch strategy and continue
    elif action == RecoveryAction.ESCALATE_MODEL:
        detector.increment_recovery()
        # swap in a stronger model
    elif action in (RecoveryAction.REQUEST_HELP, RecoveryAction.ABORT):
        # stop and surface to the user
        pass

Inspect session statistics

stats = detector.get_stats()
print(stats["total_actions"], stats["failed_actions"], stats["progress_markers"])

Best Practices

Related