loop-operator

🔄 Loop Operator — Autonomous Execution with Safety

Philosophy: Autonomous loops are powerful but dangerous. Every loop must have a clear exit condition, a safety budget, and a checkpoint mechanism. Unbounded loops are bugs, not features.

1. When to Use This Skill

• Running multi-step agentic workflows autonomously
• Processing batches of tasks (files, tests, migrations)
• Implementing retry logic with exponential backoff
• Building CI/CD pipeline stages
• Any iterative process that needs bounded execution

2. The Autonomous Loop Contract

Every autonomous loop MUST have these 5 properties:

Property	Description	Example
Exit Condition	When does the loop stop successfully?	All tests pass, all files processed
Max Iterations	Hard cap on loop cycles	max_iterations=50
Max Duration	Hard cap on wall-clock time	max_hours=2
Failure Budget	How many consecutive failures before abort	early_stop_fails=3
Checkpoint	State saved after each iteration for resume	checkpoint_dir=.agent/loops/

The Bounded Loop Template

import time
import json
from pathlib import Path

class BoundedLoop:
    def __init__(
        self,
        max_iterations: int = 50,
        max_duration_seconds: int = 7200,
        early_stop_fails: int = 3,
        checkpoint_dir: str = ".agent/loops",
    ):
        self.max_iterations = max_iterations
        self.max_duration_seconds = max_duration_seconds
        self.early_stop_fails = early_stop_fails
        self.checkpoint_dir = Path(checkpoint_dir)
        self.checkpoint_dir.mkdir(parents=True, exist_ok=True)
        
        self.iteration = 0
        self.consecutive_failures = 0
        self.start_time = time.time()
        self.results = []
    
    def should_continue(self) -> bool:
        """Check all exit conditions."""
        if self.iteration >= self.max_iterations:
            print(f"⛔ Max iterations reached ({self.max_iterations})")
            return False
        
        elapsed = time.time() - self.start_time
        if elapsed >= self.max_duration_seconds:
            print(f"⛔ Max duration reached ({self.max_duration_seconds}s)")
            return False
        
        if self.consecutive_failures >= self.early_stop_fails:
            print(f"⛔ Too many consecutive failures ({self.early_stop_fails})")
            return False
        
        return True
    
    def checkpoint(self, state: dict):
        """Save state for resume capability."""
        checkpoint = {
            "iteration": self.iteration,
            "elapsed_seconds": time.time() - self.start_time,
            "consecutive_failures": self.consecutive_failures,
            "state": state,
            "results": self.results,
        }
        path = self.checkpoint_dir / "latest.json"
        path.write_text(json.dumps(checkpoint, indent=2))
    
    def record_result(self, success: bool, details: dict):
        """Track iteration outcomes."""
        if success:
            self.consecutive_failures = 0
        else:
            self.consecutive_failures += 1
        
        self.results.append({
            "iteration": self.iteration,
            "success": success,
            "details": details,
        })
        self.iteration += 1