Parallel Dispatch Protocol
Engine for executing DAG tasks in parallel using worker agents.
Core Concepts
Parallel Groups
Tasks from the DAG that can execute simultaneously:
- Same topological level (no dependencies between them)
- Either conflict-free (no shared files) or with coordination strategy
Worker Pool
Maximum 5 concurrent workers to prevent context overflow:
- Each worker is an ephemeral agent created by agent-factory
- Workers coordinate via message bus
- 30-minute timeout with heartbeat monitoring
Dispatch Protocol
1. Parse DAG for Parallel Groups
def get_executable_parallel_groups(dag: dict, completed: set) -> list:
"""
Get parallel groups that are ready to execute.
A group is ready if all dependencies are completed.
"""
ready_groups = []
for pg in dag.get("parallel_groups", []):
# Check if all tasks in group have met dependencies
all_ready = True
for task_id in pg["tasks"]:
task = next((n for n in dag["nodes"] if n["id"] == task_id), None)
if not task:
continue
# Check if all dependencies completed
for dep in task.get("depends_on", []):
if dep not in completed:
all_ready = False
break
if not all_ready:
break
if all_ready:
# Check no tasks in group are already completed
if not any(t in completed for t in pg["tasks"]):
ready_groups.append(pg)
return ready_groups2. Create Workers for Parallel Group
def dispatch_parallel_group(
parallel_group: dict,
dag: dict,
track_id: str,
bus_path: str
) -> list:
"""
Dispatch all workers for a parallel group.
Returns list of dispatched worker handles.
"""
from agent_factory import create_workers_for_parallel_group, dispatch_workers
# 1. Create worker agents
workers = create_workers_for_parallel_group(
parallel_group, dag, track_id, bus_path
)
# 2. Check pool capacity
active_workers = count_active_workers(bus_path)
if active_workers + len(workers) > 5:
# Split into batches
batch_size = 5 - active_workers
workers = workers[:batch_size]
# 3. Dispatch workers via parallel Task calls
handles = dispatch_workers(workers)
# 4. Log dispatch
for worker in workers:
post_message(bus_path, "WORKER_DISPATCHED", "orchestrator", {
"worker_id": worker["worker_id"],
"task_id": worker["task_id"],
"parallel_group": parallel_group["id"]
})
return handles3. Monitor Worker Progress
async def monitor_parallel_group(
parallel_group: dict,
workers: list,
bus_path: str,
timeout_minutes: int = 60
) -> dict:
"""
Monitor workers until all complete or fail.
Returns aggregated results.
"""
import asyncio
from datetime import datetime, timedelta
start_time = datetime.utcnow()
timeout = timedelta(minutes=timeout_minutes)
pending_tasks = set(pg["tasks"] for pg in [parallel_group])
completed_tasks = set()
failed_tasks = {}
while pending_tasks and (datetime.utcnow() - start_time) < timeout:
# Check for completions
for task_id in list(pending_tasks):
event_file = f"{bus_path}/events/TASK_COMPLETE_{task_id}.event"
if os.path.exists(event_file):
pending_tasks.remove(task_id)
completed_tasks.add(task_id)
# Get completion details
msgs = read_messages(bus_path, msg_type="TASK_COMPLETE")
for msg in msgs:
if msg["payload"]["task_id"] == task_id:
# Log success
break
# Check for failures
for task_id in list(pending_tasks):
event_file = f"{bus_path}/events/TASK_FAILED_{task_id}.event"
if os.path.exists(event_file):
pending_tasks.remove(task_id)
# Get failure details
msgs = read_messages(bus_path, msg_type="TASK_FAILED")
for msg in msgs:
if msg["payload"]["task_id"] == task_id:
failed_tasks[task_id] = msg["payload"]["error"]
break
# Check for stale workers (no heartbeat)
stale = check_stale_workers(bus_path, threshold_minutes=10)
for stale_worker in stale:
task_id = stale_worker["task_id"]
if task_id in pending_tasks:
failed_tasks[task_id] = f"Worker stale: no heartbeat for {stale_worker['minutes_stale']} min"
pending_tasks.remove(task_id)
# Check for deadlocks
deadlock_cycle = detect_deadlock(bus_path)
if deadlock_cycle:
for worker_id in deadlock_cycle:
# Find task for this worker
status = get_worker_status(bus_path, worker_id)
if status and status["task_id"] in pending_tasks:
failed_tasks[status["task_id"]] = f"Deadlock detected in cycle: {deadlock_cycle}"
pending_tasks.remove(status["task_id"])
await asyncio.sleep(5)
# Handle timeout
for task_id in pending_tasks:
failed_tasks[task_id] = "Timeout: task did not complete within time limit"
return {
"completed": list(completed_tasks),
"failed": failed_tasks,
"success": len(failed_tasks) == 0
}Failure Handling
Failure Isolation
When one worker fails, isolate the failure:
def handle_worker_failure(
failed_task_id: str,
dag: dict,
bus_path: str
) -> dict:
"""
Handle a failed worker. Isolate failure and continue with independent tasks.
Returns impact analysis.
"""
# 1. Find tasks that depend on the failed task
blocked_tasks = []
for node in dag["nodes"]:
if failed_task_id in node.get("depends_on", []):
blocked_tasks.append(node["id"])
# 2. Recursively find all downstream tasks
def find_all_downstream(task_id, visited=None):
if visited is None:
visited = set()
if task_id in visited:
return []
visited.add(task_id)
downstream = []
for node in dag["nodes"]:
if task_id in node.get("depends_on", []):
downstream.append(node["id"])
downstream.extend(find_all_downstream(node["id"], visited))
return downstream
all_blocked = set(blocked_tasks)
for task in blocked_tasks:
all_blocked.update(find_all_downstream(task))
# 3. Mark blocked tasks
for task_id in all_blocked:
post_message(bus_path, "TASK_BLOCKED", "orchestrator", {
"task_id": task_id,
"blocked_by": failed_task_id,
"reason": "Upstream task failed"
})
# 4. Find tasks that can still proceed
all_tasks = set(n["id"] for n in dag["nodes"])
can_proceed = all_tasks - all_blocked - {failed_task_id}
return {
"failed_task": failed_task_id,
"blocked_tasks": list(all_blocked),
"can_proceed": list(can_proceed),
"needs_fix": True
}Recovery Strategy
def attempt_recovery(
failure_result: dict,
dag: dict,
track_id: str,
bus_path: str,
max_retries: int = 2
) -> dict:
"""
Attempt to recover from failure.
"""
failed_task = failure_result["failed_task"]
# 1. Check retry count
retry_key = f"retry_{failed_task}"
retries = get_coordination_log_count(bus_path, retry_key)
if retries >= max_retries:
return {
"action": "ESCALATE",
"reason": f"Task {failed_task} failed {retries} times, needs manual intervention"
}
# 2. Log retry attempt
log_coordination(bus_pa