# Grid DAG-Based Parallel Execution Specification **Version:** 1.7.x **Status:** Design Specification **Authors:** The Grid Team --- ## Table of Contents 1. [Executive Summary](#executive-summary) 2. [Current State vs DAG State](#current-state-vs-dag-state) 3. [Task Dependency Graph](#task-dependency-graph) 4. [Execution Algorithm](#execution-algorithm) 5. [Speculative Execution](#speculative-execution) 6. [Parallel Spawning](#parallel-spawning) 7. [Out-of-Order Completion](#out-of-order-completion) 8. [Implementation Details](#implementation-details) 9. [State Management](#state-management) 10. [Failure Handling](#failure-handling) 11. [Migration Path](#migration-path) --- ## Executive Summary The Grid currently uses **wave-based sequential execution**: Wave 1 completes entirely before Wave 2 starts. This is simple but leaves parallelism opportunities on the table. **DAG-based execution** enables true parallel execution: any task whose dependencies are satisfied can start immediately, regardless of wave boundaries. This can reduce total execution time by 40-60% for complex missions. ### Key Benefits | Metric | Wave-Based | DAG-Based | Improvement | |--------|-----------|-----------|-------------| | Execution time | 100% (baseline) | 40-60% | 40-60% faster | | Resource utilization | Low (waiting between waves) | High (always executing) | 2-3x efficiency | | Flexibility | Rigid wave boundaries | Dynamic scheduling | Maximum parallelism | | Failure recovery | Restart entire wave | Restart single task | Granular recovery | --- ## Current State vs DAG State ### Current: Wave-Based Execution ``` Wave 1: [task-A, task-B, task-C] ├── All start together ├── Wait for ALL to complete └── Only then start Wave 2 Wave 2: [task-D, task-E] ├── Wait for task-C even though task-D only needs task-A └── Unnecessary blocking ``` **Problem:** Task-D must wait for task-C even if task-D only depends on task-A. ### New: DAG-Based Execution ``` task-A ─────────────────────────→ task-D ──→ task-F \ / task-B ──→ task-C (depends A,B) ──→ task-E ──→ task-G / task-H ──────────────────────────────────────── Timeline: t=0: Start A, B, H (no dependencies) t=1: A completes → Start D immediately t=2: B completes → C can start (A and B done) t=3: H completes t=4: D completes → F can start t=5: C completes → E can start ... ``` **Benefit:** Task-D starts as soon as task-A completes, regardless of task-B or task-C status. --- ## Task Dependency Graph ### Plan YAML Format Planner expresses dependencies explicitly in the plan YAML: ```yaml --- cluster: "E-Commerce Platform" total_tasks: 8 dag_enabled: true tasks: - id: scout-codebase name: "Scout codebase structure" type: scout depends_on: [] # Can start immediately files_read: ["**/*"] estimated_duration: 60 # seconds - id: scout-docs name: "Scout documentation" type: scout depends_on: [] # Can start immediately (parallel with above) files_read: ["*.md", "docs/**/*"] estimated_duration: 45 - id: plan-api name: "Plan API implementation" type: plan depends_on: [scout-codebase, scout-docs] # Waits for both scouts outputs: [".grid/plans/api-plan.md"] estimated_duration: 120 - id: plan-ui name: "Plan UI implementation" type: plan depends_on: [scout-codebase] # Only needs codebase scout outputs: [".grid/plans/ui-plan.md"] estimated_duration: 90 - id: exec-api name: "Execute API implementation" type: execute depends_on: [plan-api] files_modified: ["src/api/**/*"] estimated_duration: 300 - id: exec-ui name: "Execute UI implementation" type: execute depends_on: [plan-ui] # Parallel with exec-api files_modified: ["src/ui/**/*"] estimated_duration: 240 - id: exec-integration name: "Wire API and UI together" type: execute depends_on: [exec-api, exec-ui] # Waits for both files_modified: ["src/app/**/*"] estimated_duration: 180 - id: verify-all name: "Verify entire system" type: verify depends_on: [exec-integration] estimated_duration: 120 # Computed by DAG engine (not manually specified): dag_structure: layers: 0: [scout-codebase, scout-docs] # No dependencies 1: [plan-api, plan-ui] # After layer 0 2: [exec-api, exec-ui] # After respective plans 3: [exec-integration] # After both executions 4: [verify-all] # Final critical_path: [scout-codebase, plan-api, exec-api, exec-integration, verify-all] total_estimated: 780 # seconds on critical path --- ``` ### Dependency Specification Rules 1. **Empty `depends_on`**: Task is immediately executable (root node) 2. **Single dependency**: Task starts when that one task completes 3. **Multiple dependencies**: Task starts when ALL listed tasks complete (AND logic) 4. **No cycles allowed**: DAG validator rejects circular dependencies 5. **Transitive dependencies are implicit**: If A→B→C, don't list A in C's depends_on ### File Conflict Detection Tasks with overlapping `files_modified` cannot run in parallel even if dependency graph allows: ```yaml tasks: - id: exec-api files_modified: ["src/api/routes.ts", "src/shared/types.ts"] - id: exec-ui files_modified: ["src/ui/app.tsx", "src/shared/types.ts"] # CONFLICT! ``` **Resolution:** DAG engine adds implicit dependency: `exec-ui.depends_on.push(exec-api)` --- ## Execution Algorithm ### Core DAG Executor ```python class DAGExecutor: """ DAG-based task executor for The Grid. Executes tasks as soon as their dependencies are satisfied, maximizing parallelism while respecting dependency constraints. """ def __init__(self, plan: dict, max_concurrent: int = 5): self.tasks = {t['id']: t for t in plan['tasks']} self.max_concurrent = max_concurrent # State tracking self.pending = set(self.tasks.keys()) # Not started self.running = {} # id -> Task handle self.completed = {} # id -> result self.failed = {} # id -> error # Dependency graph self.dependents = self._build_dependents() # id -> [tasks that depend on it] self.ready_queue = [] # Tasks ready to execute # File conflict tracking self.locked_files = {} # file -> task_id holding lock def _build_dependents(self) -> dict: """Build reverse dependency map: task -> tasks that depend on it.""" dependents = {tid: [] for tid in self.tasks} for tid, task in self.tasks.items(): for dep in task.get('depends_on', []): dependents[dep].append(tid) return dependents def _is_ready(self, task_id: str) -> bool: """Check if task can start (all dependencies satisfied + no file conflicts).""" task = self.tasks[task_id] # Check all dependencies completed successfully for dep in task.get('depends_on', []): if dep not in self.completed: return False if self.completed[dep].get('status') != 'SUCCESS': return False # Check no file conflicts with running tasks for file in task.get('files_modified', []): if file in self.locked_files: return False return True def _acquire_file_locks(self, task_id: str) -> bool: """Acquire locks on files this task will modify.""" task = self.tasks[task_id] files = task.get('files_modified', []) # Check all files available for file in files: if file in self.locked_files: return False # Acquire all locks for file in files: self.locked_files[file] = task_id return True def _release_file_locks(self, task_id: str): """Release locks when task completes.""" files_to_release = [f for f, t in self.locked_files.items() if t == task_id] for file in files_to_release: del self.locked_files[file] def _find_ready_tasks(self) -> list: """Find all tasks that can start now.""" ready = [] for task_id in self.pending: if self._is_ready(task_id): ready.append(task_id) return ready def _spawn_task(self, task_id: str) -> 'TaskHandle': """Spawn a task as a Grid Program.""" task = self.tasks[task_id] # Acquire file locks if not self._acquire_file_locks(task_id): raise RuntimeError(f"Could not acquire locks for {task_id}") # Gather context from completed dependencies dep_context = self._gather_dependency_context(task_id) # Spawn appropriate Program type if task['type'] == 'scout': handle = self._spawn_scout(task, dep_context) elif task['type'] == 'plan': handle = self._spawn_planner(task, dep_context) elif task['type'] == 'execute': handle = self._spawn_executor(task, dep_context) elif task['type'] == 'verify': handle = self._spawn_recognizer(task, dep_context) else: handle = self._spawn_executor(task, dep_context) # default # Update state self.pending.remove(task_id) self.running[task_id] = handle # Log spawn log_spawn(task_id, task['name'], len(self.running), self.max_concurrent) return handle def _gather_dependency_context(self, task_id: str) -> dict: """Gather outputs and warmth from completed dependencies.""" task = self.tasks[task_id] context = { 'dependency_outputs': {}, 'warmth': { 'codebase_patterns': [], 'gotchas': [], 'user_preferences': [], } } for dep in task.get('depends_on', []): result = self.completed.get(dep, {}) # Collect outputs if 'outputs' in result: context['dependency_outputs'][dep] = result['outputs'] # Merge warmth dep_warmth = result.get('warmth', {}) for key in context['warmth']: context['warmth'][key].extend(dep_warmth.get(key, [])) # Deduplicate warmth for key in context['warmth']: context['warmth'][key] = list(set(context['warmth'][key])) return context def _on_task_complete(self, task_id: str, result: dict): """Handle task completion.""" # Update state del self.running[task_id] self.completed[task_id] = result # Release file locks self._release_file_locks(task_id) # Log completion log_complete(task_id, result.get('status'), len(self.completed), len(self.tasks)) # Check what new tasks are now ready self._schedule_ready_tasks() def _on_task_failure(self, task_id: str, error: dict): """Handle task failure.""" # Update state del self.running[task_id] self.failed[task_id] = error # Release file locks self._release_file_locks(task_id) # Mark all dependents as blocked self._propagate_failure(task_id) # Log failure log_failure(task_id, error) def _propagate_failure(self, failed_task_id: str): """Mark tasks that can no longer complete due to dependency failure.""" to_mark = self.dependents[failed_task_id].copy() while to_mark: task_id = to_mark.pop(0) if task_id in self.pending: self.pending.remove(task_id) self.failed[task_id] = { 'status': 'BLOCKED', 'reason': f'Dependency {failed_task_id} failed' } to_mark.extend(self.dependents[task_id]) def _schedule_ready_tasks(self): """Schedule all tasks that are now ready, up to max_concurrent.""" ready = self._find_ready_tasks() # Sort by priority (critical path first, then estimated duration) ready.sort(key=lambda tid: ( -self.tasks[tid].get('critical_path_priority', 0), self.tasks[tid].get('estimated_duration', 0) )) # Spawn up to capacity slots_available = self.max_concurrent - len(self.running) for task_id in ready[:slots_available]: self._spawn_task(task_id) async def execute(self) -> dict: """ Execute the entire DAG. Returns: dict with completed, failed, and aggregate results """ # Initialize: find all root tasks (no dependencies) self._schedule_ready_tasks() # Main loop: wait for completions, schedule new tasks while self.running or self.pending: # Wait for any running task to complete completed_task_id, result = await self._wait_for_any_completion() if result.get('status') == 'SUCCESS': self._on_task_complete(completed_task_id, result) elif result.get('status') == 'CHECKPOINT': # Task hit checkpoint - pause DAG execution return self._create_checkpoint_result(completed_task_id, result) else: self._on_task_failure(completed_task_id, result) # Schedule newly ready tasks self._schedule_ready_tasks() # All done return { 'status': 'COMPLETE', 'completed': self.completed, 'failed': self.failed, 'total_tasks': len(self.tasks), 'success_count': len([r for r in self.completed.values() if r.get('status') == 'SUCCESS']), 'failure_count': len(self.failed), } ``` ### Task State Machine ``` ┌─────────────────────────────────────────┐ │ │ v │ ┌─────────┐ ┌────────┐ ┌─────────┐ ┌───────────┐ │ │ PENDING │───→│ READY │───→│ RUNNING │───→│ COMPLETED │ │ └─────────┘ └────────┘ └─────────┘ └───────────┘ │ │ │ │ │ │ │ │ ┌──────────┐ │ │ │ └────────→│ FAILED │ │ │ │ └──────────┘ │ │ │ │ │ │ │ ┌─────────────┐ │ │ │ │ │ CHECKPOINT │──┘ │ │ │ └─────────────┘ │ │ │ │ │ │ │ │ (resume) │ │ │ └────────────────────┘ │ │ │ (blocked by │ (file conflict) │ failure) │ v v ┌─────────┐ ┌────────┐ │ BLOCKED │ │ QUEUED │ └─────────┘ └────────┘ State Transitions: - PENDING → READY: All dependencies completed successfully - READY → RUNNING: Spawned, file locks acquired - READY → QUEUED: At max_concurrent, waiting for slot - RUNNING → COMPLETED: Task finished successfully - RUNNING → FAILED: Task failed after retries - RUNNING → CHECKPOINT: Task hit human verification point - CHECKPOINT → RUNNING: User approved, task resumes - PENDING → BLOCKED: Dependency failed, task cannot execute ``` --- ## Speculative Execution ### When to Speculate Speculative execution starts a task BEFORE its dependencies are fully complete, betting that they will succeed. **High-confidence speculation scenarios:** | Scenario | Confidence | Speculation Strategy | |----------|-----------|---------------------| | Planner almost done, Executor ready | 95% | Start Executor with partial plan | | Scout 90% complete, Planner waiting | 85% | Start Planner with partial recon | | All dependency tests passing | 90% | Start dependent task early | | Dependency at "final verification" | 95% | Start next task | **Never speculate when:** - Dependency has shown any failure signals - Dependency is at a decision checkpoint - File conflicts possible - Dependency is < 50% complete ### Speculation Protocol ```python class SpeculativeExecutor(DAGExecutor): """Extended DAG executor with speculation support.""" def __init__(self, plan: dict, speculation_enabled: bool = True, confidence_threshold: float = 0.85): super().__init__(plan) self.speculation_enabled = speculation_enabled self.confidence_threshold = confidence_threshold self.speculative_tasks = {} # task_id -> speculation_state def _assess_speculation_confidence(self, task_id: str) -> float: """ Assess confidence that dependencies will succeed. Returns float 0.0-1.0 representing speculation confidence. """ task = self.tasks[task_id] if not task.get('depends_on'): return 1.0 # No dependencies = full confidence confidences = [] for dep_id in task['depends_on']: dep_confidence = self._get_dependency_confidence(dep_id) confidences.append(dep_confidence) # Overall confidence is product of individual confidences # (all must succeed, so multiply probabilities) overall = 1.0 for c in confidences: overall *= c return overall def _get_dependency_confidence(self, dep_id: str) -> float: """Get confidence that a dependency will succeed.""" if dep_id in self.completed: return 1.0 if self.completed[dep_id].get('status') == 'SUCCESS' else 0.0 if dep_id in self.failed: return 0.0 if dep_id not in self.running: return 0.5 # Unknown, assume moderate # For running tasks, assess based on progress signals handle = self.running[dep_id] progress = handle.get_progress() # 0.0-1.0 health = handle.get_health() # 0.0-1.0 (no errors = 1.0) # Confidence increases with progress and health # At 90% progress with no errors, confidence is ~0.9 return progress * health def _should_speculate(self, task_id: str) -> bool: """Decide whether to speculatively start a task.""" if not self.speculation_enabled: return False task = self.tasks[task_id] # Don't speculate if task has file conflicts for file in task.get('files_modified', []): if file in self.locked_files: return False # Don't speculate if any dependency at checkpoint for dep_id in task.get('depends_on', []): if dep_id in self.running: if self.running[dep_id].at_checkpoint(): return False # Check confidence threshold confidence = self._assess_speculation_confidence(task_id) return confidence >= self.confidence_threshold def _start_speculative(self, task_id: str) -> 'TaskHandle': """Start a task speculatively.""" task = self.tasks[task_id] # Gather partial context from in-progress dependencies partial_context = self._gather_partial_context(task_id) # Spawn with speculation flag handle = self._spawn_task_with_context(task, partial_context, speculative=True) # Track speculation state self.speculative_tasks[task_id] = { 'started_at': now(), 'confidence': self._assess_speculation_confidence(task_id), 'dependencies_complete': False, 'rollback_point': handle.get_rollback_point(), } log_speculation_start(task_id, self.speculative_tasks[task_id]['confidence']) return handle def _on_speculation_validated(self, task_id: str): """Called when speculative task's dependencies all complete successfully.""" spec_state = self.speculative_tasks.get(task_id) if not spec_state: return spec_state['dependencies_complete'] = True # Inject any final context from now-completed dependencies self._inject_final_context(task_id) log_speculation_validated(task_id) def _rollback_speculative(self, task_id: str, reason: str): """Rollback speculative execution when dependency fails.""" spec_state = self.speculative_tasks.get(task_id) if not spec_state: return handle = self.running.get(task_id) if handle: # Cancel the running task handle.cancel() # Rollback any partial changes rollback_point = spec_state['rollback_point'] self._perform_rollback(task_id, rollback_point) # Update state del self.running[task_id] self.pending.add(task_id) # Back to pending (blocked by failure) del self.speculative_tasks[task_id] log_speculation_rollback(task_id, reason) def _perform_rollback(self, task_id: str, rollback_point: dict): """Perform actual rollback of speculative work.""" # Revert any git commits made if rollback_point.get('commit_before'): run_command(f"git reset --hard {rollback_point['commit_before']}") # Delete any created files for file in rollback_point.get('files_created', []): if os.path.exists(file): os.remove(file) # Restore any modified files for file, backup in rollback_point.get('file_backups', {}).items(): write_file(file, backup) ``` ### Speculation Metrics Track speculation effectiveness: ```yaml speculation_metrics: total_speculations: 45 validated: 42 # Dependencies succeeded rolled_back: 3 # Dependencies failed success_rate: 93.3% time_saved: 1847s # Work done while waiting wasted_compute: 89s # Rolled back work net_benefit: 1758s # 29 minutes saved ``` --- ## Parallel Spawning ### True Parallelism via Single Message The key to true parallelism is spawning multiple tasks in a SINGLE message: ```python def spawn_parallel_tasks(task_ids: list[str]) -> list['TaskHandle']: """ Spawn multiple tasks in parallel. CRITICAL: All Task() calls must be in ONE message for true parallelism. """ # Prepare all prompts prompts = [] for task_id in task_ids: task = tasks[task_id] context = gather_dependency_context(task_id) prompt = build_task_prompt(task, context) prompts.append((task_id, prompt)) # SINGLE MESSAGE with multiple Task() calls # This is the syntax that enables parallel execution: handles = [] # In actual implementation, this would be: # Task(prompt=prompts[0][1], description=f"Execute {prompts[0][0]}") # Task(prompt=prompts[1][1], description=f"Execute {prompts[1][0]}") # Task(prompt=prompts[2][1], description=f"Execute {prompts[2][0]}") # ... all in one message block return handles ``` ### Concurrency Limits ```python # Configuration MAX_CONCURRENT_TASKS = 5 # Don't overload system MAX_CONCURRENT_EXECUTORS = 3 # Heavy tasks MAX_CONCURRENT_SCOUTS = 5 # Light tasks MAX_CONCURRENT_RECOGNIZERS = 2 # Verification def get_concurrency_limit(task_type: str) -> int: """Get concurrency limit based on task type.""" limits = { 'scout': MAX_CONCURRENT_SCOUTS, 'plan': 2, # Planners are heavy 'execute': MAX_CONCURRENT_EXECUTORS, 'verify': MAX_CONCURRENT_RECOGNIZERS, } return limits.get(task_type, MAX_CONCURRENT_TASKS) def schedule_ready_tasks_by_type(ready_tasks: list) -> list: """ Schedule tasks respecting per-type concurrency limits. """ scheduled = [] type_counts = count_running_by_type() for task_id in ready_tasks: task_type = tasks[task_id]['type'] limit = get_concurrency_limit(task_type) if type_counts.get(task_type, 0) < limit: scheduled.append(task_id) type_counts[task_type] = type_counts.get(task_type, 0) + 1 if len(scheduled) >= MAX_CONCURRENT_TASKS: break return scheduled ``` ### Load Balancing ```python def prioritize_ready_tasks(ready: list[str]) -> list[str]: """ Sort ready tasks by priority for scheduling. Priority factors (highest to lowest): 1. Critical path membership (speeds up total execution) 2. Number of blocked dependents (unblocks more work) 3. Estimated duration (shorter first for throughput) 4. Task type (prefer scouts to start early) """ def priority_key(task_id: str) -> tuple: task = tasks[task_id] critical_path = 1 if task.get('on_critical_path') else 0 dependent_count = len(dependents.get(task_id, [])) duration = task.get('estimated_duration', 300) type_priority = { 'scout': 0, # Start first 'plan': 1, 'execute': 2, 'verify': 3, }.get(task['type'], 2) return ( -critical_path, # Higher is better -dependent_count, # More dependents = higher priority duration, # Shorter is better type_priority, # Lower is better ) return sorted(ready, key=priority_key) ``` --- ## Out-of-Order Completion ### Handling Async Results Tasks complete in unpredictable order. The DAG executor must handle this gracefully: ```python class AsyncResultCollector: """Collect and process results as they arrive in any order.""" def __init__(self): self.results = {} self.callbacks = {} # task_id -> [callback functions] self.lock = asyncio.Lock() async def register_task(self, task_id: str, handle: 'TaskHandle'): """Register a running task for result collection.""" async with self.lock: self.results[task_id] = { 'status': 'RUNNING', 'handle': handle, 'started_at': now(), } async def on_result(self, task_id: str, result: dict): """Handle result arrival (called from Task completion).""" async with self.lock: self.results[task_id]['status'] = result.get('status', 'UNKNOWN') self.results[task_id]['result'] = result self.results[task_id]['completed_at'] = now() # Execute any registered callbacks for callback in self.callbacks.get(task_id, []): await callback(task_id, result) async def wait_for_any(self, task_ids: list[str]) -> tuple[str, dict]: """ Wait for any of the specified tasks to complete. Returns (task_id, result) of the first to complete. """ while True: async with self.lock: for task_id in task_ids: if task_id in self.results: if self.results[task_id]['status'] != 'RUNNING': return (task_id, self.results[task_id]['result']) await asyncio.sleep(0.1) # Poll interval async def wait_for_all(self, task_ids: list[str]) -> dict[str, dict]: """Wait for all specified tasks to complete.""" results = {} remaining = set(task_ids) while remaining: task_id, result = await self.wait_for_any(list(remaining)) results[task_id] = result remaining.remove(task_id) return results ``` ### Order-Independent State Updates ```python def update_dag_state(task_id: str, result: dict, dag_state: dict): """ Update DAG state when a task completes (in any order). This function is designed to be called for any task completion regardless of order, and will correctly update the state. """ # Record completion dag_state['completed'][task_id] = { 'result': result, 'completed_at': now(), } # Remove from running if task_id in dag_state['running']: del dag_state['running'][task_id] # Check what tasks are now unblocked newly_ready = [] for candidate_id in dag_state['pending']: candidate = dag_state['tasks'][candidate_id] # Check if all dependencies now complete all_deps_done = all( dep_id in dag_state['completed'] for dep_id in candidate.get('depends_on', []) ) # Check if all dependencies succeeded all_deps_succeeded = all( dag_state['completed'][dep_id]['result'].get('status') == 'SUCCESS' for dep_id in candidate.get('depends_on', []) if dep_id in dag_state['completed'] ) if all_deps_done and all_deps_succeeded: newly_ready.append(candidate_id) # Move to ready queue for ready_id in newly_ready: dag_state['pending'].remove(ready_id) dag_state['ready'].append(ready_id) # Update metrics dag_state['metrics']['completed_count'] = len(dag_state['completed']) dag_state['metrics']['last_completion'] = task_id dag_state['metrics']['completion_order'].append(task_id) return newly_ready ``` ### Merge Strategy for Parallel Results When parallel tasks complete, their results need merging: ```yaml # Result merge rules merge_strategy: # Warmth: combine lessons from all parallel tasks warmth: mode: "union" deduplicate: true # Commits: append in completion order commits: mode: "append" order_by: "completion_time" # Files modified: track per-task for conflict detection files_modified: mode: "per_task" # Outputs: namespace by task ID outputs: mode: "namespace" # exec-api outputs in outputs.exec-api.* # exec-ui outputs in outputs.exec-ui.* ``` --- ## Implementation Details ### State File Updates New fields in `.grid/STATE.md` for DAG execution: ```yaml --- # ... existing fields ... # DAG Execution State dag: enabled: true # Task states tasks: scout-codebase: status: completed started_at: "2026-01-24T10:00:00Z" completed_at: "2026-01-24T10:01:00Z" result: success scout-docs: status: completed started_at: "2026-01-24T10:00:00Z" completed_at: "2026-01-24T10:00:45Z" result: success plan-api: status: running started_at: "2026-01-24T10:01:00Z" progress: 0.6 exec-api: status: pending blocked_by: [plan-api] # Execution metrics metrics: total_tasks: 8 completed: 2 running: 1 pending: 5 failed: 0 parallelism_achieved: 2.0 # Average concurrent tasks speculation_success_rate: 1.0 critical_path_progress: 0.25 # Ready queue (tasks that CAN start) ready_queue: - plan-ui # Waiting for concurrency slot # File locks file_locks: "src/api/routes.ts": plan-api --- ``` ### MC Integration Master Control uses DAG executor instead of wave executor: ```python # In MC orchestration def execute_mission(plan: dict): """Execute mission using DAG-based execution.""" # Check if DAG execution enabled if plan.get('dag_enabled', False) and env('GRID_DAG_EXECUTION', 'true') == 'true': executor = DAGExecutor( plan=plan, max_concurrent=int(env('GRID_MAX_CONCURRENT', '5')), speculation_enabled=env('GRID_SPECULATION', 'true') == 'true', ) else: # Fall back to wave-based execution executor = WaveExecutor(plan) # Execute result = await executor.execute() # Handle result if result['status'] == 'COMPLETE': return handle_mission_complete(result) elif result['status'] == 'CHECKPOINT': return handle_checkpoint(result) else: return handle_failure(result) ``` ### Planner Integration Planner calculates DAG metadata during planning: ```python def calculate_dag_metadata(tasks: list[dict]) -> dict: """Calculate DAG structure and critical path.""" # Build dependency graph graph = build_graph(tasks) # Calculate layers (for visualization) layers = calculate_layers(graph) # Find critical path (longest path through DAG) critical_path = find_critical_path(graph, tasks) # Estimate total time (critical path duration) total_estimated = sum( tasks[tid].get('estimated_duration', 0) for tid in critical_path ) # Mark tasks on critical path for task in tasks: task['on_critical_path'] = task['id'] in critical_path task['critical_path_priority'] = ( critical_path.index(task['id']) if task['id'] in critical_path else -1 ) return { 'layers': layers, 'critical_path': critical_path, 'total_estimated': total_estimated, } ``` --- ## State Management ### Checkpoint File for DAG State When checkpointing during DAG execution: ```yaml # .grid/CHECKPOINT.md --- created_at: "2026-01-24T10:15:00Z" reason: "human_verify" session_id: "dag-sess-12345" # DAG-specific checkpoint data dag_state: # Completed tasks (can be skipped on resume) completed_tasks: - id: scout-codebase commit: "abc123" outputs: [".grid/recon/codebase.md"] warmth: codebase_patterns: - "Uses barrel exports" - id: scout-docs commit: "def456" outputs: [".grid/recon/docs.md"] # Running task that hit checkpoint checkpointed_task: id: exec-api progress: 0.6 completed_threads: [1, 2] current_thread: 3 partial_outputs: ["src/api/users.ts"] # Tasks that were running but paused paused_tasks: - id: exec-ui progress: 0.4 can_resume: true # Tasks still pending pending_tasks: - id: exec-integration blocked_by: [exec-api, exec-ui] - id: verify-all blocked_by: [exec-integration] # Resume instructions resume: strategy: "continue" # or "restart_failed" or "rollback_and_restart" start_from: - exec-api (thread 3) - exec-ui (resume) skip: - scout-codebase - scout-docs --- ``` ### Resume from DAG Checkpoint ```python def resume_dag_execution(checkpoint: dict) -> DAGExecutor: """Resume DAG execution from checkpoint.""" # Load original plan plan = load_plan(checkpoint['plan_path']) # Create executor with restored state executor = DAGExecutor(plan) # Restore completed tasks for task_data in checkpoint['dag_state']['completed_tasks']: executor.completed[task_data['id']] = { 'status': 'SUCCESS', 'outputs': task_data['outputs'], 'warmth': task_data['warmth'], 'restored': True, # Mark as restored, not re-executed } # Remove completed from pending for task_id in executor.completed: if task_id in executor.pending: executor.pending.remove(task_id) # Handle checkpointed task checkpointed = checkpoint['dag_state']['checkpointed_task'] # This task needs continuation, not restart executor.continuation_context[checkpointed['id']] = { 'completed_threads': checkpointed['completed_threads'], 'current_thread': checkpointed['current_thread'], 'partial_outputs': checkpointed['partial_outputs'], } # Handle paused tasks for paused in checkpoint['dag_state']['paused_tasks']: if paused['can_resume']: executor.continuation_context[paused['id']] = { 'progress': paused['progress'], } return executor ``` --- ## Failure Handling ### Failure Propagation When a task fails, dependent tasks are blocked: ```python def propagate_failure(failed_task_id: str, dag_state: dict): """ Mark all tasks that depend on failed task as blocked. Uses BFS to find all transitively dependent tasks. """ blocked = set() to_check = [failed_task_id] while to_check: current = to_check.pop(0) for task_id, task in dag_state['tasks'].items(): if current in task.get('depends_on', []): if task_id not in blocked: blocked.add(task_id) to_check.append(task_id) # Update state for task_id in blocked: if task_id in dag_state['pending']: dag_state['pending'].remove(task_id) dag_state['blocked'][task_id] = { 'reason': f'Dependency {failed_task_id} failed', 'blocked_at': now(), } return blocked ``` ### Retry Strategies ```python class RetryStrategy: """Configurable retry strategies for failed tasks.""" IMMEDIATE = "immediate" # Retry right away BACKOFF = "exponential" # Exponential backoff MANUAL = "manual" # Wait for user SKIP = "skip" # Mark as skipped, continue DAG @staticmethod def get_strategy(task: dict, failure: dict) -> str: """Determine retry strategy based on failure type.""" failure_type = failure.get('type') if failure_type == 'timeout': return RetryStrategy.BACKOFF elif failure_type == 'auth_error': return RetryStrategy.MANUAL elif failure_type == 'rate_limit': return RetryStrategy.BACKOFF elif failure_type == 'user_error': return RetryStrategy.MANUAL elif task.get('optional', False): return RetryStrategy.SKIP else: return RetryStrategy.IMMEDIATE def handle_task_failure(task_id: str, failure: dict, dag_state: dict): """Handle task failure with appropriate strategy.""" strategy = RetryStrategy.get_strategy(dag_state['tasks'][task_id], failure) retry_count = dag_state.get('retry_counts', {}).get(task_id, 0) max_retries = dag_state['tasks'][task_id].get('max_retries', 3) if strategy == RetryStrategy.SKIP: # Mark as skipped, don't block dependents dag_state['completed'][task_id] = { 'status': 'SKIPPED', 'reason': failure.get('message'), } return 'skipped' elif retry_count < max_retries: # Retry the task dag_state['retry_counts'][task_id] = retry_count + 1 if strategy == RetryStrategy.BACKOFF: delay = 2 ** retry_count # 1s, 2s, 4s, 8s... schedule_retry(task_id, delay) else: schedule_retry(task_id, 0) return 'retrying' else: # Max retries exceeded dag_state['failed'][task_id] = failure propagate_failure(task_id, dag_state) return 'failed' ``` ### Partial Recovery Allow recovering partial DAG state after session death: ```python def recover_from_session_death(work_dir: str) -> dict: """ Recover DAG state from artifacts on disk. When a session dies unexpectedly, we can reconstruct partial progress from: - Git commits - Created files - STATE.md - SCRATCHPAD.md """ recovery = { 'completed_tasks': [], 'partial_tasks': [], 'unknown_tasks': [], } # Check each task for task_id, task in load_plan()['tasks'].items(): # Look for completion signals outputs = task.get('outputs', []) all_outputs_exist = all(os.path.exists(f) for f in outputs) commits = find_commits_for_task(task_id) has_commits = len(commits) > 0 if all_outputs_exist and has_commits: recovery['completed_tasks'].append({ 'id': task_id, 'commits': commits, 'outputs': outputs, }) elif has_commits or any(os.path.exists(f) for f in outputs): recovery['partial_tasks'].append({ 'id': task_id, 'commits': commits, 'outputs_found': [f for f in outputs if os.path.exists(f)], }) else: recovery['unknown_tasks'].append(task_id) return recovery ``` --- ## Migration Path ### Feature Flag DAG execution is enabled via environment variable: ```bash # Enable DAG execution (default: true in 1.7.x) export GRID_DAG_EXECUTION=true # Disable for fallback to wave-based export GRID_DAG_EXECUTION=false ``` ### Gradual Rollout 1. **Phase 1**: DAG execution opt-in (env var) 2. **Phase 2**: DAG execution default, wave fallback available 3. **Phase 3**: Wave execution deprecated 4. **Phase 4**: Wave execution removed ### Compatibility DAG execution maintains backward compatibility: - Plans without `dag_enabled: true` use wave execution - Plans with `wave` fields still work (converted to DAG) - All existing commands work unchanged --- ## Appendix: Configuration Reference ### Environment Variables | Variable | Default | Description | |----------|---------|-------------| | `GRID_DAG_EXECUTION` | `true` | Enable DAG-based execution | | `GRID_MAX_CONCURRENT` | `5` | Maximum concurrent tasks | | `GRID_SPECULATION` | `true` | Enable speculative execution | | `GRID_SPECULATION_THRESHOLD` | `0.85` | Minimum confidence for speculation | | `GRID_RETRY_MAX` | `3` | Maximum retries per task | ### Config File ```json // .grid/config.json { "dag": { "enabled": true, "max_concurrent": 5, "speculation": { "enabled": true, "confidence_threshold": 0.85 }, "retry": { "max_attempts": 3, "backoff_base": 2 } } } ``` --- *DAG Execution Specification v1.7.x - The Grid Team* *End of Line.*