Ralph Wiggum Loop

Ralph Wiggum Loop

What is the Ralph Wiggum Loop?

"I'm helping! I'm helping! I'm helping!" — Ralph Wiggum

The Ralph Wiggum loop is a bounded autonomous execution pattern that allows AI agents to execute multi-step plans with:

• Max 10 iterations (prevents infinite loops)
• State persistence (restartable on crash)
• Dependency checking (steps execute in correct order)
• Retry with backoff (handles transient failures)
• Human escalation (blocks on permanent failures)

Named after Ralph Wiggum because the agent enthusiastically attempts tasks repeatedly but has a hard limit to prevent runaway behavior.

Architecture

Plan.md (approved) → Plan Executor (plan_executor.py)
                            ↓
                   Initialize ExecutionState
                   (iterations_remaining=10)
                            ↓
              ┌─────────────┴─────────────┐
              ↓                           ↓
       Check Dependencies            Execute Step
              ↓                           ↓
       All met?                    Success? ───→ Mark [x], decrement iteration
              │                           │
              No                         No (retry 3x)
              ↓                           ↓
        Mark blocked                 Mark [!], escalate
              ↓                           ↓
       Escalate to                vault/Needs_Action/
      vault/Needs_Action/
              ↓
       iterations_remaining == 0? ───→ Escalate (max iterations)
              │
             No
              ↓
         Next Step ──→ Loop back

Quick Start

1. Plan Structure

<!-- vault/Plans/PLAN_client_onboarding_001.md -->
---
plan_id: plan_client_onboarding_001
objective: "Onboard new client - draft intro email, create calendar invite, post LinkedIn announcement"
total_steps: 3
completed_steps: 0
status: awaiting_approval
approval_required: true
iterations_used: 0
max_iterations: 10
---

# Plan: Client Onboarding - Acme Corp

**Objective:** Complete new client onboarding for Acme Corp

## Steps

- [ ] **Step 1:** Draft intro email to client (action_type: mcp_email)
  - Dependencies: none
  - MCP Server: email-mcp
  - Tool: send_email
  - Params: {to: "client@acme.com", subject: "Welcome!", body: "..."}

- [ ] **Step 2:** Create calendar invite for kickoff meeting (action_type: create_file)
  - Dependencies: step_1
  - File: vault/Pending_Approval/Calendar/kickoff_invite.md

- [ ] **Step 3:** Post LinkedIn announcement (action_type: mcp_linkedin)
  - Dependencies: step_1, step_2
  - MCP Server: linkedin-mcp
  - Tool: create_post
  - Params: {text: "Excited to welcome Acme Corp..."}

2. Plan Executor Implementation

# agent_skills/plan_executor.py
from dataclasses import dataclass
from typing import List, Optional
import time

@dataclass
class ExecutionState:
    plan_id: str
    current_step: int
    iterations_remaining: int
    last_action: str
    last_action_timestamp: str
    loop_start_time: str
    status: str  # executing | completed | blocked | escalated

class RalphWiggumLoop:
    """Autonomous multi-step plan executor with bounded iterations"""

    def __init__(self, plan_path: str, max_iterations: int = 10):
        self.plan_path = plan_path
        self.plan = self.load_plan(plan_path)
        self.max_iterations = max_iterations

        # Initialize or load execution state
        state_path = f"vault/In_Progress/{self.plan['plan_id']}/state.md"
        if os.path.exists(state_path):
            self.state = self.load_state(state_path)
        else:
            self.state = ExecutionState(
                plan_id=self.plan['plan_id'],
                current_step=1,
                iterations_remaining=max_iterations,
                last_action="initialized",
                last_action_timestamp=datetime.utcnow().isoformat(),
                loop_start_time=datetime.utcnow().isoformat(),
                status="executing"
            )

    def execute(self):
        """
        Main execution loop

        Returns:
            bool: True if plan completed, False if blocked/escalated
        """
        print(f"🔄 Starting Ralph Wiggum loop for {self.plan['objective']}")
        print(f"   Max iterations: {self.max_iterations}")

        while self.state.iterations_remaining > 0:
            print(f"\n--- Iteration {self.max_iterations - self.state.iterations_remaining + 1}/{self.max_iterations} ---")

            # Get current step
            step = self.plan['steps'][self.state.current_step - 1]

            # Check dependencies
            if not self.check_dependencies(step):
                print(f"⏸️  Step {self.state.current_step} blocked - dependencies not met")
                self.handle_blocked_step(step)
                return False

            # Execute step
            try:
                self.execute_step(step)

                # Mark step complete
                self.mark_step_complete(step)

                # Update state
                self.state.current_step += 1
                self.state.iterations_remaining -= 1
                self.state.last_action = f"completed step {step['step_num']}"
                self.save_state()

                # Check if all steps complete
                if self.state.current_step > self.plan['total_steps']:
                    print("✅ All steps complete!")
                    self.handle_plan_complete()
                    return True

            except Exception as e:
                print(f"❌ Step {self.state.current_step} failed: {e}")

                # Retry with exponential backoff
                if not self.retry_step(step):
                    # All retries failed - escalate
                    self.handle_step_failure(step, e)
                    return False

        # Max iterations reached - escalate
        print(f"⚠️  Max iterations ({self.max_iterations}) reached - escalating")
        self.handle_max_iterations_escalation()
        return False

    def check_dependencies(self, step: dict) -> bool:
        """Check if step dependencies are met"""
        dependencies = step.get('dependencies', [])

        if not dependencies or dependencies == ['none']:
            return True

        # Check each dependency step is completed
        for dep in dependencies:
            dep_num = int(dep.replace('step_', ''))
            dep_step = self.plan['steps'][dep_num - 1]

            if not dep_step.get('completed', False):
                return False

        return True

    def execute_step(self, step: dict):
        """Execute a single plan step"""
        action_type = step['action_type']

        if action_type.startswith('mcp_'):
            # Invoke MCP server
            self.execute_mcp_action(step)

        elif action_type == 'create_file':
            # Create file in vault
            self.execute_file_creation(step)

        elif action_type == 'notify_human':
            # Create notification for human
            self.execute_human_notification(step)

        else:
            raise ValueError(f"Unknown action_type: {action_type}")

    def execute_mcp_action(self, step: dict):
        """Invoke MCP server for step action"""
        from agent_skills.mcp_client import call_mcp_tool

        # Extract MCP details
        server = step['mcp_server']
        tool = step['tool']
        params = step['params']

        # Invoke MCP
        result = call_mcp_tool(server, tool, params)

        # Log MCP action
        log_mcp_action(
            mcp_server=server,
            action=tool,
            outcome="success",
            plan_id=self.plan['plan_id'],
            step_num=step['step_num']
        )

        # Store result in step
        step['mcp_result'] = result

    def execute_file_creation(self, step: dict):
        """Create vault file as step action"""
        file_path = step['file_path']
        content = step.get('content', '')

        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        with open(file_path, 'w') as f:
            f.write(content)

        print(f"📄 Created file: {file_path}")

    def execute_human_notification(self, step: dict):
        """Create notification for human action"""
        notification_path = f"vault/Needs_Action/plan_notification_{self.plan['plan_id']}_step_{step['step_num']}.md"

        content = f"""---
plan_id: {self.plan['plan_id']}
step_num: {step['step_num']}
action_required: true
---

# Human Action Required

**Plan:** {self.plan['objective']}
**Step:** {step['description']}

{step.get('notification_message', 'Manual action required - see plan for details')}
"""

        with open(notification_path, 'w') as f:
            f.write(content)

        print(f"📢 Created notification: {notification_path}")

    def mark_step_complete(self, step: dict):
        """Mark step as [x] in Plan.md"""
        step['completed'] = True

        # Update Plan.md file
        plan_content = open(self.plan_path).read()

        # Replace checkbox: - [ ] → - [x]
        step_pattern = f"- \\[ \\] \\*\\*Step {step['step_num']}:\\*\\*"
        step_replacement = f"- [x] **Step {step['step_num']}:**"

        updated_content = re.sub(step_pattern, step_replacement, plan_content)

        with open(self.plan_path, 'w') as f:
            f.write(updated_content)

        # Update plan completion count
        self.plan['completed_steps'] += 1

    def retry_step(self, step: dict, max_retries: int = 3) -> bool:
        """Retry failed step with exponential backoff"""
        for attempt in range(max_retries):
            try:
                # Wait with exponential backoff: 5s, 10s, 20s
                if attempt > 0:
                    wait_time = 5 * (2 ** attempt)
                    print(f"⏳ Retry {attempt + 1}/{max_retries} in {wait_time}s...")
                    time.sleep(wait_time)

                # Retry execution
                self.execute_step(step)
                print(f"✅ Step succeeded on retry {attempt + 1}")
                return True

            except Exception as e:
                print(f"❌ Retry {attempt + 1} failed: {e}")

        return False

    def handle_step_failure(self, step: dict, error: Exception):
        """Handle permanent step failure - escalate to human"""
        # Mark step as [!] in Plan.md
        step['status'] = 'failed'

        plan_content = open(self.plan_path).read()
        step_pattern = f"- \\[ \\] \\*\\*Step {step['step_num']}:\\*\\*"
        step_replacement = f"- [!] **Step {step['step_num']}:** (FAILED)"

        updated_content = re.sub(step_pattern, step_replacement, plan_content)

        with open(self.plan_path, 'w') as f:
            f.write(updated_content)

        # Create escalation file
        escalation_path = f"vault/Needs_Action/plan_blocked_{self.plan['plan_id']}.md"

        content = f"""---
plan_id: {self.plan['plan_id']}
step_num: {step['step_num']}
severity: high
action_required: true
---

# Plan Blocked - Step Failed

**Plan:** {self.plan['objective']}
**Blocked Step:** {step['description']}

**Error:** {str(error)}

**Recovery Instructions:**
1. Review step details in [[{os.path.basename(self.plan_path)}]]
2. Fix the underlying issue
3. Restart plan execution

**Full Error Details:**

{traceback.format_exc()}

"""

        with open(escalation_path, 'w') as f:
            f.write(content)

        # Update state
        self.state.status = "blocked"
        self.save_state()

        print(f"⚠️  Plan blocked - escalated to: {escalation_path}")

    def handle_max_iterations_escalation(self):
        """Handle max iterations reached - escalate"""
        escalation_path = f"vault/Needs_Action/plan_escalated_{self.plan['plan_id']}.md"

        content = f"""---
plan_id: {self.plan['plan_id']}
severity: medium
action_required: true
max_iterations_reached: true
---

# Plan Escalated - Max Iterations Reached

**Plan:** {self.plan['objective']}
**Iterations Used:** {self.max_iterations}

The plan has reached the maximum iteration limit ({self.max_iterations}) without completing all steps.

**Current Progress:**
- Completed: {self.plan['completed_steps']}/{self.plan['total_steps']} steps
- Current Step: {self.state.current_step}

**Next Actions:**
1. Review plan progress in [[{os.path.basename(self.plan_path)}]]
2. Check for circular dependencies or stuck steps
3. Manually complete remaining steps OR restart with higher iteration limit

**Full Iteration History:**
See vault/Logs/Plan_Execution/{self.plan['plan_id']}.md
"""

        with open(escalation_path, 'w') as f:
            f.write(content)

        # Update state
        self.state.status = "escalated"
        self.save_state()

        print(f"⚠️  Plan escalated - max iterations: {escalation_path}")

    def handle_plan_complete(self):
        """Handle successful plan completion"""
        # Update Plan.md status
        self.update_plan_status("completed")

        # Move to Done
        done_path = f"vault/Done/{os.path.basename(self.plan_path)}"
        shutil.move(self.plan_path, done_path)

        # Clean up execution state
        state_dir = f"vault/In_Progress/{self.plan['plan_id']}/"
        if os.path.exists(state_dir):
            shutil.rmtree(state_dir)

        # Update dashboard
        from agent_skills.dashboard_updater import update_dashboard
        update_dashboard()

        print(f"✅ Plan completed: {self.plan['objective']}")
        print(f"   Iterations used: {self.max_iterations - self.state.iterations_remaining}")

    def save_state(self):
        """Persist execution state to vault/In_Progress/"""
        state_dir = f"vault/In_Progress/{self.plan['plan_id']}/"
        os.makedirs(state_dir, exist_ok=True)

        state_path = os.path.join(state_dir, 'state.md')

        content = f"""---
plan_id: {self.state.plan_id}
current_step: {self.state.current_step}
iterations_remaining: {self.state.iterations_remaining}
last_action: {self.state.last_action}
last_action_timestamp: {self.state.last_action_timestamp}
loop_start_time: {self.state.loop_start_time}
status: {self.state.status}
---

# Execution State: {self.plan['objective']}

**Current Step:** {self.state.current_step}/{self.plan['total_steps']}
**Iterations Remaining:** {self.state.iterations_remaining}/{self.max_iterations}
**Status:** {self.state.status}

**Last Action:** {self.state.last_action}
**Timestamp:** {self.state.last_action_timestamp}
"""

        with open(state_path, 'w') as f:
            f.write(content)

Plan Watcher

# scripts/plan_watcher.py
from agent_skills.plan_executor import RalphWiggumLoop
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class PlanApprovalHandler(FileSystemEventHandler):
    def on_moved(self, event):
        """Detect approved plans"""
        if not event.is_directory and 'Approved/Plans' in event.dest_path:
            # Parse plan
            plan_path = event.dest_path

            # Start Ralph Wiggum loop
            executor = RalphWiggumLoop(plan_path)
            executor.execute()

def watch_approved_plans():
    """Monitor vault/Approved/Plans/ for approved plans"""
    observer = Observer()
    observer.schedule(
        PlanApprovalHandler(),
        path="vault/Pending_Approval/",
        recursive=True
    )
    observer.start()

    print("📋 Plan watcher active")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

Safety Features

Bounded Iterations

# Prevent infinite loops
MAX_ITERATIONS = 10  # Hard limit

# Escalate when limit reached
if iterations_remaining == 0:
    escalate_to_human()

State Persistence

# Restart-safe execution
state_path = f"vault/In_Progress/{plan_id}/state.md"

# On crash, resume from last saved state
if os.path.exists(state_path):
    state = load_state(state_path)
    resume_from_step(state.current_step)

Dependency Validation

# Never execute steps out of order
def check_dependencies(step):
    for dep in step['dependencies']:
        if not is_step_complete(dep):
            return False  # Block execution
    return True

Testing

# tests/integration/test_plan_execution.py
def test_ralph_wiggum_loop():
    """Test multi-step plan execution"""

    # Create test plan
    plan = create_test_plan(steps=[
        {"step_num": 1, "action_type": "mcp_email", "dependencies": []},
        {"step_num": 2, "action_type": "create_file", "dependencies": ["step_1"]},
        {"step_num": 3, "action_type": "mcp_linkedin", "dependencies": ["step_1", "step_2"]}
    ])

    # Mock MCP calls
    with patch('agent_skills.mcp_client.call_mcp_tool') as mock_mcp:
        mock_mcp.return_value = {"success": True}

        # Execute plan
        executor = RalphWiggumLoop(plan_path='test_plan.md')
        result = executor.execute()

        # Verify success
        assert result == True
        assert executor.state.current_step > executor.plan['total_steps']
        assert mock_mcp.call_count == 2  # email + linkedin (file doesn't use MCP)

def test_max_iterations_escalation():
    """Test escalation when max iterations reached"""

    # Create circular dependency plan (infinite loop)
    plan = create_circular_plan()

    executor = RalphWiggumLoop(plan_path='circular_plan.md', max_iterations=5)
    result = executor.execute()

    # Verify escalation
    assert result == False
    assert executor.state.status == "escalated"
    assert os.path.exists(f"vault/Needs_Action/plan_escalated_{executor.plan['plan_id']}.md")

Configuration

# .env
MAX_PLAN_ITERATIONS=10  # Default: 10
PLAN_RETRY_ATTEMPTS=3   # Default: 3

Troubleshooting

Key Files

• agent_skills/plan_executor.py - Ralph Wiggum loop implementation
• scripts/plan_watcher.py - Monitor approved plans
• vault/Plans/ - Plan definitions
• vault/In_Progress/ - Active execution state
• vault/Needs_Action/plan_*.md - Escalations

Production Ready: Bounded iterations, state persistence, dependency validation, retry logic, comprehensive error handling, human escalation patterns.

Remember: Like Ralph Wiggum, the agent is enthusiastic and helpful, but has firm limits to prevent chaos. Max 10 iterations ensures autonomous execution stays safe and bounded.