Function execute_steps_parallel

core/src/orchestration/executor.rs:158–191

Fan `specs` out across the executor, bounded by its [`concurrency_hint`](AgentExecutor::concurrency_hint), preserving input order. A panicked branch becomes a failed [`StepOutcome`] without dropping the others. This is the barrier (`parallel`) primitive — it awaits every step. Later combinators (pipeline, phases) build on the same seam.

(
    executor: Arc<dyn AgentExecutor>,
    specs: Vec<AgentStepSpec>,
    event_tx: Option<broadcast::Sender<AgentEvent>>,
)

/// This is the barrier (`parallel`) primitive — it awaits every step. Later
/// combinators (pipeline, phases) build on the same seam.
pub async fn execute_steps_parallel(
    executor: Arc<dyn AgentExecutor>,
    specs: Vec<AgentStepSpec>,
    event_tx: Option<broadcast::Sender<AgentEvent>>,
) -> Vec<StepOutcome> {
    let limit = executor.concurrency_hint();
    // Keep (task_id, agent) by index so a panicked branch still yields a
    // correctly-labelled failed outcome (mirrors TaskExecutor's fallback).
    let labels: Vec<(String, String)> = specs
        .iter()
        .map(|s| (s.task_id.clone(), s.agent.clone()))
        .collect();

    let results = run_ordered_parallel_with_limit(specs, limit, move |_idx, spec| {
        let executor = Arc::clone(&executor);
        let event_tx = event_tx.clone();
        async move { executor.execute_step(spec, event_tx).await }
    })
    .await;

    results
        .into_iter()
        .map(|result| match result.output {
            Ok(outcome) => outcome,
            Err(error) => {
                let (task_id, agent) = labels
                    .get(result.index)
                    .cloned()
                    .unwrap_or_else(|| ("unknown".to_string(), "unknown".to_string()));
                StepOutcome::failed(task_id, agent, error.to_string())
            }
        })
        .collect()
}

#[cfg(test)]
mod tests {