Implementing the Manager-Worker Pattern — Cross-Session Orchestration
When running multiple sessions in Claude Control, I needed a way for one session to delegate work to others and aggregate the results. I implemented this as a manager-worker pattern. The manager session plans the overall work; the worker sessions carry it out.
Manager-Worker Architecture
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
class TaskStatus(str, Enum):
PENDING = "pending"
ASSIGNED = "assigned"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class Task:
id: str
description: str
assigned_to: Optional[str] = None
status: TaskStatus = TaskStatus.PENDING
result: Optional[str] = None
priority: int = 0
class ManagerSession:
def __init__(self, session_id: str, session_manager, redis_store):
self.session_id = session_id
self.manager = session_manager
self.store = redis_store
self.task_queue: List[Task] = []
self.workers: Dict[str, str] = {} # worker_id -> session_id
async def decompose_task(self, main_task: str) -> List[Task]:
prompt = f"""다음 작업을 독립적인 하위 태스크로 분해하세요:
{main_task}
각 태스크는 병렬로 실행 가능해야 합니다.
JSON 배열로 출력하세요."""
response = await self.manager.send_prompt(self.session_id, prompt)
subtasks = json.loads(response)
return [Task(id=f"task_{i}", description=t["description"],
priority=t.get("priority", 0))
for i, t in enumerate(subtasks)]
Task Distribution and Execution
async def distribute_tasks(self):
available_workers = await self._get_available_workers()
pending_tasks = sorted(
[t for t in self.task_queue if t.status == TaskStatus.PENDING],
key=lambda t: -t.priority,
)
for task, worker_id in zip(pending_tasks, available_workers):
task.assigned_to = worker_id
task.status = TaskStatus.ASSIGNED
await self._assign_task_to_worker(worker_id, task)
async def _assign_task_to_worker(self, worker_id: str, task: Task):
worker_session = self.workers[worker_id]
await self.store.redis.publish("claude:tasks", json.dumps({
"type": "task_assigned",
"worker_id": worker_id,
"task": {"id": task.id, "description": task.description},
}))
# Send prompt to the worker
await self.manager.send_prompt(worker_session, task.description)
task.status = TaskStatus.IN_PROGRESS
Worker Session Management Tools
I implemented a tool set for the manager to administrate its workers.
class ManagerTools:
@staticmethod
async def list_workers(manager: ManagerSession) -> List[dict]:
return [
{"worker_id": wid, "session_id": sid, "status": "active"}
for wid, sid in manager.workers.items()
]
@staticmethod
async def create_worker(manager: ManagerSession, workdir: str) -> str:
worker_id = f"worker_{len(manager.workers)}"
session_id = f"session_{worker_id}"
await manager.manager.create_session(session_id, workdir)
manager.workers[worker_id] = session_id
return worker_id
@staticmethod
async def collect_results(manager: ManagerSession) -> dict:
results = {}
for task in manager.task_queue:
if task.status == TaskStatus.COMPLETED:
results[task.id] = task.result
return results
Real-World Example
This pattern really shone during a large-scale refactor that touched the frontend, backend, and infrastructure simultaneously. The manager drew up the change plan, each worker handled its own domain, and the manager reviewed the results — a task that would have taken half a day manually was done in an hour.
Retrospective
The manager-worker pattern has become the most powerful feature in Claude Control. That said, handling tasks with inter-worker dependencies is still a work in progress. DAG-based task scheduling should solve that.