Ralph Wiggum Loop

---
description: Ralph Wiggum loop pattern for multi-step task completion with Stop hook.
---

# COMMAND: Ralph Wiggum Loop (Multi-Step Task Iteration)

## CONTEXT

The user needs to implement a "Ralph Wiggum Loop" pattern that:

- Keeps Claude iterating until multi-step tasks are complete
- Uses a Stop hook pattern to detect task completion
- Prevents premature termination of complex workflows
- Integrates with the reasoning engine

**Name Origin:** Named after Ralph Wiggum's "I'm winning!" meme - the loop keeps going until it actually wins (completes the task).

## YOUR ROLE

Act as an AI agent architect with expertise in:

- Multi-step task orchestration
- State persistence and tracking
- Loop termination conditions
- Agent hook systems

## OUTPUT STRUCTURE

Create a Ralph Wiggum Loop implementation with:

1. **Stop Hook Pattern** for completion detection
2. **Task State Tracking** with checkpoints
3. **Iteration Logic** with smart stopping
4. **Integration Examples** for reasoning engine
5. **Setup Instructions** for various use cases

## Step 1: Stop Hook Pattern

The Ralph Wiggum Loop uses a "Stop Hook" that checks if work is truly complete.

```python
#!/usr/bin/env python3
"""
Ralph Wiggum Loop - Keep iterating until multi-step tasks are complete

This pattern prevents premature task completion by maintaining state
and checking for actual completion before stopping.
"""

import asyncio
import json
import logging
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("RalphWiggumLoop")


class TaskState:
    """Track the state of a multi-step task."""

    def __init__(self, task_id: str, total_steps: int, state_dir: Path):
        """
        Initialize task state.

        Args:
            task_id: Unique task identifier
            total_steps: Number of steps in the task
            state_dir: Directory to store state files
        """
        self.task_id = task_id
        self.total_steps = total_steps
        self.state_dir = state_dir
        self.state_file = state_dir / f"{task_id}.json"

        self.completed_steps: List[str] = []
        self.current_step: Optional[str] = None
        self.failed_steps: List[str] = []
        self.iterations = 0
        self.max_iterations = total_steps * 3  # Safety limit

        self._load_state()

    def _load_state(self):
        """Load existing state if available."""
        if self.state_file.exists():
            try:
                data = json.loads(self.state_file.read_text())
                self.completed_steps = data.get('completed_steps', [])
                self.current_step = data.get('current_step')
                self.failed_steps = data.get('failed_steps', [])
                self.iterations = data.get('iterations', 0)
            except Exception as e:
                logger.warning(f"Could not load state: {e}")

    def _save_state(self):
        """Save current state to file."""
        self.state_dir.mkdir(parents=True, exist_ok=True)

        data = {
            'task_id': self.task_id,
            'total_steps': self.total_steps,
            'completed_steps': self.completed_steps,
            'current_step': self.current_step,
            'failed_steps': self.failed_steps,
            'iterations': self.iterations,
            'updated_at': datetime.now().isoformat()
        }

        self.state_file.write_text(json.dumps(data, indent=2))

    def start_step(self, step_name: str):
        """Mark a step as started."""
        self.current_step = step_name
        self._save_state()
        logger.info(f"[{self.task_id}] Starting step: {step_name}")

    def complete_step(self, step_name: str):
        """Mark a step as completed."""
        if step_name not in self.completed_steps:
            self.completed_steps.append(step_name)

        if step_name in self.failed_steps:
            self.failed_steps.remove(step_name)

        self.current_step = None
        self._save_state()
        logger.info(f"[{self.task_id}] Completed step: {step_name} ({len(self.completed_steps)}/{self.total_steps})")

    def fail_step(self, step_name: str, error: str):
        """Mark a step as failed."""
        if step_name not in self.failed_steps:
            self.failed_steps.append(step_name)

        self.current_step = None
        self._save_state()
        logger.warning(f"[{self.task_id}] Failed step: {step_name} - {error}")

    def is_complete(self) -> bool:
        """Check if all steps are completed."""
        return len(self.completed_steps) >= self.total_steps

    def should_continue(self) -> bool:
        """Check if loop should continue iterating."""
        self.iterations += 1
        self._save_state()

        if self.is_complete():
            logger.info(f"[{self.task_id}] Task complete! ✓")
            return False

        if self.iterations >= self.max_iterations:
            logger.warning(f"[{self.task_id}] Max iterations reached ({self.max_iterations})")
            return False

        return True

    def get_next_step(self, all_steps: List[str]) -> Optional[str]:
        """Get the next incomplete step."""
        for step in all_steps:
            if step not in self.completed_steps:
                return step
        return None

    def get_progress(self) -> Dict[str, Any]:
        """Get current progress status."""
        return {
            'task_id': self.task_id,
            'progress': len(self.completed_steps) / self.total_steps * 100,
            'completed': len(self.completed_steps),
            'total': self.total_steps,
            'current': self.current_step,
            'failed': len(self.failed_steps),
            'iterations': self.iterations
        }


class RalphWiggumLoop:
    """
    Main loop orchestration for multi-step tasks.

    Implements the Stop Ho