dag-parallel-executor

interface ExecutionContext {
  dagId: DAGId;
  schedule: ScheduledWave[];
  results: Map<NodeId, TaskResult>;
  errors: Map<NodeId, TaskError>;
  config: ExecutorConfig;
}

async function executeDAG(
  schedule: ScheduledWave[],
  config: ExecutorConfig
): Promise<ExecutionResult> {
  const context: ExecutionContext = {
    dagId: schedule[0]?.dagId,
    schedule,
    results: new Map(),
    errors: new Map(),
    config,
  };

  for (const wave of schedule) {
    await executeWave(wave, context);

    // Check for fatal errors
    if (shouldAbortExecution(context)) {
      break;
    }
  }

  return buildExecutionResult(context);
}

async function executeWave(
  wave: ScheduledWave,
  context: ExecutionContext
): Promise<void> {
  const { maxParallelism } = context.config;
  const tasks = wave.tasks;

  // Execute in batches respecting parallelism limit
  for (let i = 0; i < tasks.length; i += maxParallelism) {
    const batch = tasks.slice(i, i + maxParallelism);

    // Execute batch concurrently
    const promises = batch.map(task =>
      executeTask(task, context)
    );

    await Promise.all(promises);
  }
}

async function executeTask(
  task: ScheduledTask,
  context: ExecutionContext
): Promise<void> {
  const node = getNodeFromTask(task, context);

  // Build Task tool parameters
  const taskParams = {
    description: `Execute ${node.skillId}: ${task.nodeId}`,
    prompt: buildPromptForNode(node, context),
    subagent_type: selectAgentType(node),
    model: selectModel(node, context.config),
  };

  try {
    // Use Task tool to spawn agent
    const result = await spawnAgent(taskParams);
    context.results.set(task.nodeId, {
      output: result,
      completedAt: new Date(),
    });
  } catch (error) {
    handleTaskError(task, error, context);
  }
}

function selectAgentType(node: DAGNode): string {
  // Map node types to appropriate agent types
  switch (node.type) {
    case 'skill':
      return node.skillId;  // Use skill as agent type
    case 'agent':
      return node.agentDefinition.type;
    case 'mcp-tool':
      return 'general-purpose';
    default:
      return 'general-purpose';
  }
}

function selectModel(
  node: DAGNode,
  config: ExecutorConfig
): 'haiku' | 'sonnet' | 'opus' {
  // Select model based on task complexity
  const complexity = estimateComplexity(node);

  if (complexity === 'simple' && config.allowHaiku) {
    return 'haiku';
  } else if (complexity === 'complex' && config.allowOpus) {
    return 'opus';
  }
  return 'sonnet';
}

// Execute multiple independent tasks in single message
function buildParallelTaskCalls(
  tasks: ScheduledTask[],
  context: ExecutionContext
): TaskToolCall[] {
  return tasks.map(task => ({
    tool: 'Task',
    params: {
      description: `Node: ${task.nodeId}`,
      prompt: buildPromptForNode(
        getNodeFromTask(task, context),
        context
      ),
      subagent_type: selectAgentType(
        getNodeFromTask(task, context)
      ),
    },
  }));
}

async function executeWithRetry(
  task: ScheduledTask,
  context: ExecutionContext
): Promise<TaskResult> {
  const { maxRetries, retryDelayMs, exponentialBackoff } =
    task.config;

  let lastError: Error;

  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await executeTask(task, context);
    } catch (error) {
      lastError = error;

      if (attempt < maxRetries) {
        const delay = exponentialBackoff
          ? retryDelayMs * Math.pow(2, attempt)
          : retryDelayMs;
        await sleep(delay);
      }
    }
  }

  throw lastError;
}

function handleTaskError(
  task: ScheduledTask,
  error: Error,
  context: ExecutionContext
): void {
  context.errors.set(task.nodeId, {
    message: error.message,
    code: classifyError(error),
    recoverable: isRecoverable(error),
  });

  switch (context.config.errorHandling) {
    case 'stop-on-failure':
      context.aborted = true;
      break;

    case 'continue-on-failure':
      // Mark dependent nodes as skipped
      markDependentsSkipped(task.nodeId, context);
      break;

    case 'retry-then-skip':
      // Already retried, now skip
      markDependentsSkipped(task.nodeId, context);
      break;
  }
}

executionState:
  dagId: research-pipeline
  status: running
  startedAt: "2024-01-15T10:00:00Z"

  waves:
    - wave: 0
      status: completed
      duration: 28500ms
      tasks:
        - nodeId: gather-sources
          status: completed
          duration: 28500ms
          tokensUsed: 4500

    - wave: 1
      status: running
      tasks:
        - nodeId: validate-sources
          status: running
          startedAt: "2024-01-15T10:00:30Z"
        - nodeId: extract-metadata
          status: running
          startedAt: "2024-01-15T10:00:30Z"

  progress:
    completedNodes: 1
    runningNodes: 2
    pendingNodes: 3
    failedNodes: 0

  resources:
    tokensUsed: 4500
    estimatedCost: 0.05

function optimizeBatching(
  wave: ScheduledWave,
  config: ExecutorConfig
): ScheduledTask[][] {
  const tasks = wave.tasks;
  const maxParallel = config.maxParallelism;

  // Sort by estimated duration (shortest first)
  // This improves overall throughput
  tasks.sort((a, b) =>
    a.estimatedDuration - b.estimatedDuration
  );

  // Create balanced batches
  const batches: ScheduledTask[][] = [];
  for (let i = 0; i < tasks.length; i += maxParallel) {
    batches.push(tasks.slice(i, i + maxParallel));
  }

  return batches;
}

async function executeWaveWithEarlyCompletion(
  wave: ScheduledWave,
  context: ExecutionContext
): Promise<void> {
  const pending = new Set(wave.tasks.map(t => t.nodeId));
  const running = new Map<NodeId, Promise<void>>();

  while (pending.size > 0 || running.size > 0) {
    // Start new tasks up to parallelism limit
    while (
      pending.size > 0 &&
      running.size < context.config.maxParallelism
    ) {
      const task = pending.values().next().value;
      pending.delete(task);

      const promise = executeTask(task, context)
        .finally(() => running.delete(task));
      running.set(task, promise);
    }

    // Wait for any task to complete
    if (running.size > 0) {
      await Promise.race(running.values());
    }
  }
}