Method execute_plan

core/src/agent/plan_execution.rs:204–728

Execute an execution plan using wave-based dependency-aware scheduling. Steps with no unmet dependencies are grouped into "waves". A wave with a single step executes sequentially (preserving the history chain). A wave with multiple independent steps executes them in parallel via `JoinSet`, then merges their results back into the shared history.

(
        &self,
        history: &[Message],
        plan: &ExecutionPlan,
        session_id: Option<&str>,
        event_tx: Option<mpsc::Sender<AgentEvent>>,
    )

    /// wave with multiple independent steps executes them in parallel via
    /// `JoinSet`, then merges their results back into the shared history.
    pub(super) async fn execute_plan(
        &self,
        history: &[Message],
        plan: &ExecutionPlan,
        session_id: Option<&str>,
        event_tx: Option<mpsc::Sender<AgentEvent>>,
    ) -> Result<AgentResult> {
        let mut plan = plan.clone();
        let task_session_id = session_id.unwrap_or("").to_string();
        let mut current_history = history.to_vec();
        let mut total_usage = TokenUsage::default();
        let mut tool_calls_count = 0;
        let total_steps = plan.steps.len();

        // Add initial user message with the goal
        let steps_text = plan
            .steps
            .iter()
            .enumerate()
            .map(|(i, step)| format!("{}. {}", i + 1, step.content))
            .collect::<Vec<_>>()
            .join("\n");
        current_history.push(Message::user(&crate::prompts::render(
            crate::prompts::PLAN_EXECUTE_GOAL,
            &[("goal", &plan.goal), ("steps", &steps_text)],
        )));
        self.emit_task_updated(&event_tx, &task_session_id, &plan)
            .await;

        loop {
            let ready: Vec<String> = plan
                .get_ready_steps()
                .iter()
                .map(|s| s.id.clone())
                .collect();

            if ready.is_empty() {
                // All done or deadlock
                if plan.has_deadlock() {
                    tracing::warn!(
                        "Plan deadlock detected: {} pending steps with unresolvable dependencies",
                        plan.pending_count()
                    );
                }
                break;
            }

            if ready.len() == 1 {
                // === Single step: sequential execution (preserves history chain) ===
                let step_id = &ready[0];
                let step = plan
                    .steps
                    .iter()
                    .find(|s| s.id == *step_id)
                    .ok_or_else(|| anyhow::anyhow!("step '{}' not found in plan", step_id))?
                    .clone();
                let step_number = plan
                    .steps