Long-Running Agents
Two-Agent Architecture
4 min read
When tasks exceed what a single session can handle, split responsibilities. The Initializer + Worker pattern, documented by Anthropic for Claude-based systems, elegantly solves the long-running session problem.
The Pattern
Two specialized agents with distinct roles:
┌─────────────────┐ ┌─────────────────┐
│ INITIALIZER │────▶│ WORKER │
│ │ │ │
│ • Analyzes task │ │ • Executes work │
│ • Creates plan │ │ • Uses tools │
│ • Sets up state │ │ • Checkpoints │
│ • Fresh context │ │ • Can restart │
└─────────────────┘ └─────────────────┘
The Initializer Agent
Runs once at the start. Fresh context, full attention.
class InitializerAgent:
"""Analyzes the task and creates execution plan."""
def __init__(self, llm_client):
self.llm = llm_client
def initialize(self, task: str, codebase_path: str) -> dict:
# Step 1: Analyze the codebase
structure = self.analyze_codebase(codebase_path)
# Step 2: Break down the task
plan = self.llm.generate(f"""
Task: {task}
Codebase structure: {structure}
Create a detailed execution plan:
1. List all subtasks in order
2. Identify dependencies between subtasks
3. Note which files each subtask affects
4. Estimate complexity of each subtask
""")
# Step 3: Save initial state
state = {
"task": task,
"plan": plan,
"subtasks": self.parse_subtasks(plan),
"completed": [],
"current_index": 0,
"created_at": datetime.now().isoformat()
}
return state
The Worker Agent
Runs in a loop, picking up where it left off.
class WorkerAgent:
"""Executes subtasks with checkpointing."""
def __init__(self, llm_client, state_file: str):
self.llm = llm_client
self.state = self.load_state(state_file)
def run(self):
while self.state["current_index"] < len(self.state["subtasks"]):
subtask = self.state["subtasks"][self.state["current_index"]]
# Build focused context for this subtask
context = self.build_context(subtask)
# Execute with minimal context
result = self.execute_subtask(subtask, context)
# Checkpoint immediately
self.checkpoint(subtask, result)
# Move to next
self.state["current_index"] += 1
self.save_state()
def build_context(self, subtask: dict) -> str:
"""Only load what's needed for this subtask."""
relevant_files = subtask.get("files", [])
recent_changes = self.state["completed"][-3:] # Last 3 only
return f"""
Current subtask: {subtask['description']}
Relevant files: {self.read_files(relevant_files)}
Recent changes: {json.dumps(recent_changes, indent=2)}
"""
Why This Works
| Benefit | Explanation |
|---|---|
| Fresh starts | Worker can restart with clean context |
| Focused context | Each subtask gets only relevant info |
| Resumable | Crashes don't lose progress |
| Debuggable | Clear checkpoint trail |
| Cost efficient | No context bloat from old turns |
The Handoff
def run_long_task(task: str, codebase: str):
state_file = f"state_{hash(task)}.json"
# Check for existing state
if os.path.exists(state_file):
print("Resuming from checkpoint...")
worker = WorkerAgent(llm, state_file)
else:
print("Initializing new task...")
initializer = InitializerAgent(llm)
state = initializer.initialize(task, codebase)
save_state(state_file, state)
worker = WorkerAgent(llm, state_file)
# Run worker (can be interrupted and resumed)
worker.run()
Real-World Usage
This pattern powers tools like:
- Claude Code: Uses initializer for planning, worker for execution
- Cursor: Background agents that survive IDE restarts
- Devin: Multi-hour coding sessions with checkpointing
Nerd Note: The initializer should be aggressive about breaking down tasks. Better to have 20 small subtasks than 5 big ones—easier to checkpoint and resume.
Next: Tracking progress across sessions. :::