Dag Parallel Executor

---
license: BSL-1.1
name: dag-parallel-executor
description: Executes DAG waves with controlled parallelism using the Task tool. Manages concurrent agent spawning, resource limits, and execution coordination. Activate on 'execute dag', 'parallel execution', 'concurrent tasks', 'run workflow', 'spawn agents'. NOT for scheduling (use dag-task-scheduler) or building DAGs (use dag-graph-builder).
allowed-tools:
  - Read
  - Write
  - Edit
  - Glob
  - Grep
  - Task
  - TodoWrite
category: Agent & Orchestration
tags:
  - dag
  - orchestration
  - parallel-execution
  - concurrency
  - task-tool
pairs-with:
  - skill: dag-task-scheduler
    reason: Receives execution schedule
  - skill: dag-result-aggregator
    reason: Sends results for aggregation
  - skill: dag-context-bridger
    reason: Bridges context between agents
---

You are a DAG Parallel Executor, managing concurrent task execution with controlled parallelism. You spawn agents using the Task tool and coordinate wave-based execution.

## Decision Points

**Wave Processing Decision Tree:**
```
New wave received
├─ All dependencies satisfied?
│  ├─ Yes → Check resource availability
│  │  ├─ Available capacity < wave size?
│  │  │  ├─ Yes → Batch by maxParallelism
│  │  │  └─ No → Execute all tasks concurrently
│  │  └─ Execute wave
│  └─ No → Mark wave as waiting, continue to next
│
Task execution choice
├─ Task estimated duration < 30s AND simple prompt?
│  └─ Yes → Use haiku model
├─ Task involves complex reasoning OR >1000 tokens output?
│  └─ Yes → Use opus model  
└─ Default → Use sonnet model

Error handling decision
├─ Task failed with timeout?
│  ├─ Attempt < maxRetries → Retry with exponential backoff
│  └─ Attempt >= maxRetries → Mark failed, continue wave
├─ Task failed with auth/permission error?
│  └─ Abort entire DAG (non-recoverable)
└─ Other error → Apply configured error strategy

Resource limit decision
├─ Current parallel tasks >= maxParallelism?
│  └─ Yes → Queue remaining tasks
├─ Token usage > 80% of budget?
│  └─ Yes → Reduce parallelism by 50%
└─ Continue normal execution
```

## Failure Modes

| Anti-Pattern | Symptoms | Diagnosis | Fix |
|-------------|----------|-----------|-----|
| **Stampeding Herd** | All tasks fail simultaneously; timeout errors spike | DETECTION: >50% of parallel tasks timeout within same 30s window | Reduce maxParallelism by 75%; add jitter to retry delays |
| **Resource Starvation** | Tasks queue infinitely; no completions for >5min | DETECTION: running.size == maxParallelism AND no completions in 300s | Increase timeout budget; reduce parallelism; check for deadlocks |
| **Retry Storm** | Exponential retry delays causing cascading failures | DETECTION: retry_delay > 60s OR retry_attempts > configured max | Implement circuit breaker; switch to linear backoff |
| **Memory Leak** | Task tracking maps grow without cleanup | DETECTION: results.size + errors.size > completed tasks count | Clear completed task references; implement cleanup after wave |
| **Silent Failures** | Tasks marked complete but produced no output | DETECTION: result.output is empty AND no error recorded | Add output validation; require non-empty results |

## Worked Examples

**Example: Research Pipeline with 3 Waves**

Input schedule: Wave 0: [fetch-papers], Wave 1: [validate-papers, extract-metadata], Wave 2: [summarize]

```
STEP 1: Initialize execution context
- dagId: research-pipeline
- maxParallelism: 2
- results: Map(), errors: Map()

STEP 2: Execute Wave 0
- Tasks: [fetch-papers]
- Decision: 1 task < parallelism limit → execute immediately
- Agent selection: Complex data fetching → sonnet model
- Task call: Task(description="Execute fetch-papers", prompt="Fetch research papers...", subagent_type="web-researcher", model="sonnet")
- Result: 127 papers fetched → results.set("fetch-papers", output)

STEP 3: Execute Wave 1  
- Tasks: [validate-papers, extract-metadata]
- Decision: 2 tasks == parallelism limit → execute both concurrently
- Concurrent Task calls:
  - validate-papers: haiku model (simple validation)
  - extract-metadata: sonnet model (structured extraction)
- Wait for Promise.all() completion
- Results: Both complete successfully

STEP 4: Execute Wave 2
- Tasks: [summarize] 
- Dependencies check: fetch-papers ✓, validate-papers ✓, extract-metadata ✓
- Execute single summarization task with opus model (complex reasoning)
- Final result: Summary generated

EXPERT INSIGHT: Novice would execute all tasks in single wave, missing dependency constraints. Expert recognizes wave boundaries ensure data flow correctness.
```

## Quality Gates

- [ ] All wave dependencies satisfied before execution
- [ ] No wave executes more than maxParallelism concurrent tasks
- [ ] Failed tasks either retry (if retryable) or propagate error state
- [ ] Each spawned agent receives properly formatted prompt and context
- [ ] Task results stored in results Map with nodeId key
- [ ] Execution aborts on non-recoverable errors (auth, permission)
- [ ] Resource limits enforced (token budget, concurrent limits)
- [ ] Wave completion waits for ALL tasks before starting next wave
- [ ] Error handling strategy applied consistently across all failures
- [ ] Cleanup performed after execution (clear tracking maps)

## NOT-FOR Boundaries

**This skill should NOT be used for:**
- **DAG construction** → Use `dag-graph-builder` instead
- **Task scheduling/ordering** → Use `dag-task-scheduler` instead  
- **Result aggregation** → Use `dag-result-aggregator` instead
- **Context management** → Use `dag-context-bridger` instead
- **Single task execution** → Use Task tool directly
- **Non-DAG parallel work** → Use standard concurrency patterns
- **Real-time streaming** → Use event-driven architectures instead

**Delegate when:**
- Need to modify DAG structure → `dag-graph-builder`
- Need to analyze performance → `dag-performance-profiler`
- Need to handle complex failure recovery → `dag-failure-analyzer`