prodigy 0.4.4

//! Execution pipeline for orchestrating workflow execution
//!
//! This module contains the logic for executing workflows, managing session state,
//! and handling signal interrupts during execution.
//!
//! ## Pure Helper Functions
//!
//! This module includes several pure, testable helper functions that extract
//! business logic from the main execution pipeline:
//!
//! - `build_variable_map`: Transforms command outputs into variable mappings
//! - `build_command_string`: Constructs command strings from name and arguments
//! - `should_store_outputs`: Determines if command outputs should be persisted
//! - `build_step_description`: Creates human-readable step descriptions
//!
//! These functions are designed to be:
//! - Pure (no side effects)
//! - Independently testable
//! - Easy to reason about
//! - Focused on single responsibilities

use crate::abstractions::git::GitOperations;
use crate::cook::execution::claude::ClaudeExecutor;
use crate::cook::interaction::UserInteraction;
use crate::cook::orchestrator::{CookConfig, ExecutionEnvironment};
use crate::cook::session::{SessionManager, SessionState, SessionStatus, SessionUpdate};
use crate::cook::workflow::ExtendedWorkflowConfig;
use crate::subprocess::SubprocessManager;
use crate::worktree::{WorktreeManager, WorktreeStatus};
use anyhow::{anyhow, Context, Result};
use log::debug;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::task::JoinHandle;

/// Represents the outcome of a workflow execution
#[derive(Debug, Clone, PartialEq)]
enum ExecutionOutcome {
    Success,
    Interrupted,
    Failed(String),
}

/// Classify the execution result based on result and session status
fn classify_execution_result(
    result: &Result<()>,
    session_status: SessionStatus,
) -> ExecutionOutcome {
    match result {
        Ok(_) => ExecutionOutcome::Success,
        Err(e) => {
            if session_status == SessionStatus::Interrupted {
                ExecutionOutcome::Interrupted
            } else {
                ExecutionOutcome::Failed(e.to_string())
            }
        }
    }
}

/// Determine if a checkpoint should be saved based on the outcome
///
/// Checkpoints are saved for both interrupted and failed workflows to enable resume.
#[allow(dead_code)] // Reserved for future checkpoint optimization
fn should_save_checkpoint(outcome: &ExecutionOutcome) -> bool {
    matches!(
        outcome,
        ExecutionOutcome::Interrupted | ExecutionOutcome::Failed(_)
    )
}

/// Generate the resume message for the user
#[allow(dead_code)] // Reserved for future message centralization
fn determine_resume_message(
    session_id: &str,
    playbook_path: &str,
    outcome: &ExecutionOutcome,
) -> Option<String> {
    match outcome {
        ExecutionOutcome::Interrupted => Some(format!(
            "\nSession interrupted. Resume with: prodigy run {} --resume {}",
            playbook_path, session_id
        )),
        ExecutionOutcome::Failed(_) => Some(format!(
            "\n💡 To resume from last checkpoint, run: prodigy resume {}",
            session_id
        )),
        ExecutionOutcome::Success => None,
    }
}

/// Execution pipeline for coordinating workflow execution
pub struct ExecutionPipeline {
    session_manager: Arc<dyn SessionManager>,
    user_interaction: Arc<dyn UserInteraction>,
    claude_executor: Arc<dyn ClaudeExecutor>,
    #[allow(dead_code)]
    git_operations: Arc<dyn GitOperations>,
    subprocess: SubprocessManager,
    session_ops: super::session_ops::SessionOperations,
    workflow_executor: super::workflow_execution::WorkflowExecutor,
}

impl ExecutionPipeline {
    /// Create a new execution pipeline
    pub fn new(
        session_manager: Arc<dyn SessionManager>,
        user_interaction: Arc<dyn UserInteraction>,
        claude_executor: Arc<dyn ClaudeExecutor>,
        git_operations: Arc<dyn GitOperations>,
        subprocess: SubprocessManager,
        session_ops: super::session_ops::SessionOperations,
        workflow_executor: super::workflow_execution::WorkflowExecutor,
    ) -> Self {
        Self {
            session_manager,
            user_interaction,
            claude_executor,
            git_operations,
            subprocess,
            session_ops,
            workflow_executor,
        }
    }

    /// Initialize session metadata with workflow hash and type
    pub async fn initialize_session_metadata(
        &self,
        session_id: &str,
        config: &CookConfig,
    ) -> Result<()> {
        debug!("About to start session");
        self.session_manager.start_session(session_id).await?;
        debug!("Session started successfully");
        self.user_interaction
            .display_info(&format!("Starting session: {}", session_id));
        debug!("Session message displayed");

        // Calculate and store workflow hash
        debug!("Calculating workflow hash");
        let workflow_hash =
            super::session_ops::SessionOperations::calculate_workflow_hash(&config.workflow);
        debug!("Workflow hash calculated: {}", workflow_hash);

        debug!("Classifying workflow type");
        let workflow_type = super::core::DefaultCookOrchestrator::classify_workflow_type(config);
        debug!("Workflow type classified: {:?}", workflow_type);

        // Update session with workflow metadata
        debug!("Updating session with workflow hash");
        debug!("About to call update_session");
        let result = self
            .session_manager
            .update_session(SessionUpdate::SetWorkflowHash(workflow_hash))
            .await;
        debug!("update_session call returned");
        result?;
        debug!("Workflow hash updated");

        debug!("Updating session with workflow type");
        self.session_manager
            .update_session(SessionUpdate::SetWorkflowType(workflow_type.into()))
            .await?;
        debug!("Workflow type updated");

        Ok(())
    }

    /// Create a WorktreeManager from the config
    fn create_worktree_manager(&self, config: &CookConfig) -> Result<WorktreeManager> {
        // Get merge config from workflow or mapreduce config
        let merge_config = config.workflow.merge.clone().or_else(|| {
            config
                .mapreduce_config
                .as_ref()
                .and_then(|m| m.merge.clone())
        });

        // Get workflow environment variables
        let workflow_env = config.workflow.env.clone().unwrap_or_default();

        WorktreeManager::with_config(
            config.project_path.to_path_buf(),
            self.subprocess.clone(),
            config.command.verbosity,
            merge_config,
            workflow_env,
        )
    }

    /// Update worktree state to mark as interrupted
    fn update_worktree_interrupted_state(
        worktree_manager: &WorktreeManager,
        worktree_name: &str,
    ) -> Result<()> {
        worktree_manager.update_session_state(worktree_name, |state| {
            state.status = WorktreeStatus::Interrupted;
            state.interrupted_at = Some(chrono::Utc::now());
            state.interruption_type = Some(crate::worktree::InterruptionType::Unknown);
            state.resumable = true;
        })
    }

    /// Handle successful session completion
    async fn handle_session_success(&self) -> Result<()> {
        self.session_manager
            .update_session(SessionUpdate::UpdateStatus(SessionStatus::Completed))
            .await?;
        self.user_interaction
            .display_success("Cook session completed successfully!");
        Ok(())
    }

    /// Handle session interruption
    async fn handle_session_interruption(
        &self,
        session_id: &str,
        config: &CookConfig,
        env: &ExecutionEnvironment,
    ) -> Result<()> {
        // Display resume message
        let playbook_path = config
            .workflow
            .commands
            .first()
            .map(|_| config.command.playbook.display().to_string())
            .unwrap_or_else(|| "<workflow>".to_string());

        self.user_interaction.display_warning(&format!(
            "\nSession interrupted. Resume with: prodigy run {} --resume {}",
            playbook_path, session_id
        ));

        // Save checkpoint for resume
        let checkpoint_path = env.working_dir.join(".prodigy").join("session_state.json");
        self.session_manager.save_state(&checkpoint_path).await?;

        // Update worktree state if using a worktree
        if let Some(ref name) = env.worktree_name {
            let worktree_manager = self.create_worktree_manager(config)?;
            Self::update_worktree_interrupted_state(&worktree_manager, name.as_ref())?;
        }

        Ok(())
    }

    /// Handle session failure
    async fn handle_session_failure(
        &self,
        error: &anyhow::Error,
        session_id: &str,
        env: &ExecutionEnvironment,
    ) -> Result<()> {
        self.session_manager
            .update_session(SessionUpdate::UpdateStatus(SessionStatus::Failed))
            .await?;
        self.session_manager
            .update_session(SessionUpdate::AddError(error.to_string()))
            .await?;
        self.user_interaction
            .display_error(&format!("Session failed: {error}"));

        // Get current session state
        let state = self
            .session_manager
            .get_state()
            .context("Failed to get session state for checkpoint")?;

        // Save checkpoint for resume (Spec 186 FR1)
        if state.workflow_state.is_some() {
            // Save checkpoint to enable resume
            self.session_manager
                .save_checkpoint(&state)
                .await
                .context("Failed to save checkpoint on failure")?;

            // Also save to checkpoint path for CLI resume command
            let checkpoint_path = env.working_dir.join(".prodigy").join("session_state.json");
            self.session_manager
                .save_state(&checkpoint_path)
                .await
                .context("Failed to save session state on failure")?;

            self.user_interaction.display_info(&format!(
                "\n💡 Checkpoint saved. Resume with: prodigy resume {}",
                session_id
            ));
        }

        Ok(())
    }

    /// Execute cleanup and complete the session
    async fn execute_cleanup_and_completion(
        &self,
        cleanup_fn: impl std::future::Future<Output = Result<()>>,
    ) -> Result<super::super::session::SessionSummary> {
        // Run cleanup
        cleanup_fn.await?;

        // Complete session
        self.session_manager.complete_session().await
    }

    /// Display session completion summary
    async fn display_completion_summary(
        &self,
        summary: &super::super::session::SessionSummary,
        config: &CookConfig,
    ) -> Result<()> {
        // Don't display session stats in dry-run mode
        if !config.command.dry_run {
            self.user_interaction.display_info(&format!(
                "Session complete: {} iterations, {} files changed",
                summary.iterations, summary.files_changed
            ));
        }

        Ok(())
    }

    /// Setup signal handlers for graceful interruption
    pub fn setup_signal_handlers(
        &self,
        config: &CookConfig,
        session_id: &str,
        worktree_name: Option<Arc<str>>,
    ) -> Result<JoinHandle<()>> {
        log::debug!("Setting up signal handlers");

        let worktree_manager = Arc::new(self.create_worktree_manager(config)?);

        crate::cook::signal_handler::setup_interrupt_handlers(
            worktree_manager,
            session_id.to_string(),
        )?;

        log::debug!("Signal handlers set up successfully");

        let session_manager = self.session_manager.clone();
        let worktree_name = worktree_name.clone();
        let project_path = Arc::clone(&config.project_path);
        let subprocess = self.subprocess.clone();

        let interrupt_handler = tokio::spawn(async move {
            tokio::signal::ctrl_c().await.ok();
            // Mark session as interrupted when Ctrl+C is pressed
            session_manager
                .update_session(SessionUpdate::MarkInterrupted)
                .await
                .ok();

            // Also update worktree state if using a worktree
            if let Some(ref name) = worktree_name {
                if let Ok(worktree_manager) =
                    WorktreeManager::new(project_path.to_path_buf(), subprocess.clone())
                {
                    let _ = worktree_manager.update_session_state(name.as_ref(), |state| {
                        state.status = WorktreeStatus::Interrupted;
                        state.interrupted_at = Some(chrono::Utc::now());
                        state.interruption_type =
                            Some(crate::worktree::InterruptionType::UserInterrupt);
                        state.resumable = true;
                    });
                }
            }
        });

        Ok(interrupt_handler)
    }

    /// Finalize session with appropriate status and messaging
    pub async fn finalize_session(
        &self,
        env: &ExecutionEnvironment,
        config: &CookConfig,
        execution_result: Result<(), anyhow::Error>,
        cleanup_fn: impl std::future::Future<Output = Result<()>>,
    ) -> Result<()> {
        // Classify the execution outcome
        let session_status = if execution_result.is_err() {
            self.session_manager
                .get_state()
                .context("Failed to get session state after cook error")?
                .status
        } else {
            SessionStatus::InProgress
        };

        let outcome = classify_execution_result(&execution_result, session_status);

        // Route to appropriate handler
        match outcome {
            ExecutionOutcome::Success => {
                self.handle_session_success().await?;
                let summary = self.execute_cleanup_and_completion(cleanup_fn).await?;
                self.display_completion_summary(&summary, config).await?;
                Ok(())
            }
            ExecutionOutcome::Interrupted => {
                self.handle_session_interruption(&env.session_id, config, env)
                    .await?;
                execution_result
            }
            ExecutionOutcome::Failed(_) => {
                self.handle_session_failure(
                    execution_result.as_ref().unwrap_err(),
                    &env.session_id,
                    env,
                )
                .await?;
                execution_result
            }
        }
    }

    /// Validate that a session is in a resumable state
    ///
    /// Checks if the session status allows for resumption.
    fn validate_session_resumable(&self, session_id: &str, state: &SessionState) -> Result<()> {
        if !state.is_resumable() {
            return Err(anyhow!(
                "Session {} is not resumable (status: {:?})",
                session_id,
                state.status
            ));
        }
        Ok(())
    }

    /// Validate that the workflow hasn't been modified since the session was interrupted
    ///
    /// Compares the stored workflow hash with the current workflow hash.
    fn validate_workflow_unchanged(&self, state: &SessionState, config: &CookConfig) -> Result<()> {
        if let Some(ref stored_hash) = state.workflow_hash {
            let current_hash =
                super::session_ops::SessionOperations::calculate_workflow_hash(&config.workflow);
            if current_hash != *stored_hash {
                return Err(anyhow!(
                    "Workflow has been modified since interruption. \
                     Use --force to override or start a new session."
                ));
            }
        }
        Ok(())
    }

    /// Restore config from saved workflow state
    ///
    /// Updates the config with saved arguments and map patterns from the workflow state.
    fn restore_config_from_workflow_state(
        &self,
        config: &mut CookConfig,
        workflow_state: &super::super::session::WorkflowState,
    ) {
        config.command.args = workflow_state.input_args.clone();
        config.command.map = workflow_state.map_patterns.clone();
    }

    /// Display session completion summary
    ///
    /// Shows session statistics unless in dry-run mode.
    fn display_session_completion(
        &self,
        summary: &super::super::session::SessionSummary,
        is_dry_run: bool,
    ) {
        if !is_dry_run {
            self.user_interaction.display_info(&format!(
                "Session complete: {} iterations, {} files changed",
                summary.iterations, summary.files_changed
            ));
        }
    }

    /// Handle the result of a resumed workflow execution
    ///
    /// Processes success, interruption, and failure cases appropriately.
    async fn handle_resume_result(
        &self,
        result: Result<()>,
        session_file: &std::path::Path,
        session_id: &str,
        playbook_path: &std::path::Path,
    ) -> Result<()> {
        match result {
            Ok(_) => {
                self.session_manager
                    .update_session(SessionUpdate::UpdateStatus(SessionStatus::Completed))
                    .await?;
                self.user_interaction
                    .display_success("Resumed session completed successfully!");
                Ok(())
            }
            Err(e) => {
                // Check if session was interrupted again
                let current_state = self
                    .session_manager
                    .get_state()
                    .context("Failed to get session state after resume error")?;
                if current_state.status == SessionStatus::Interrupted {
                    self.user_interaction.display_warning(&format!(
                        "\nSession interrupted again. Resume with: prodigy run {} --resume {}",
                        playbook_path.display(),
                        session_id
                    ));
                    // Save updated checkpoint
                    self.session_manager.save_state(session_file).await?;
                } else {
                    self.session_manager
                        .update_session(SessionUpdate::UpdateStatus(SessionStatus::Failed))
                        .await?;
                    self.session_manager
                        .update_session(SessionUpdate::AddError(e.to_string()))
                        .await?;
                    self.user_interaction
                        .display_error(&format!("Resumed session failed: {e}"));
                }
                Err(e)
            }
        }
    }

    /// Load session state with fallback to worktree session file
    ///
    /// This function attempts to load the session state from UnifiedSessionManager first.
    /// If not found, it falls back to loading from the worktree session file.
    async fn load_session_with_fallback(
        &self,
        session_id: &str,
        config: &CookConfig,
    ) -> Result<SessionState> {
        // Try to load the session state from UnifiedSessionManager
        let state_result = self.session_manager.load_session(session_id).await;

        match state_result {
            Ok(s) => Ok(s),
            Err(_) => {
                // Session not found in unified storage, try loading from worktree
                // The config.project_path should already be the worktree path when resuming
                let session_file = config
                    .project_path
                    .join(".prodigy")
                    .join("session_state.json");

                if !session_file.exists() {
                    return Err(anyhow!(
                        "Session not found: {}\nTried:\n  - Unified session storage\n  - Worktree session file: {}",
                        session_id,
                        session_file.display()
                    ));
                }

                // Load from worktree session file
                self.session_manager.load_state(&session_file).await?;
                self.session_manager.get_state()
            }
        }
    }

    /// Resume a workflow from a previously interrupted session
    pub async fn resume_workflow(&self, session_id: &str, mut config: CookConfig) -> Result<()> {
        // Load session state with fallback to worktree
        let state = self.load_session_with_fallback(session_id, &config).await?;

        // Validate the session is resumable
        self.validate_session_resumable(session_id, &state)?;

        // Validate workflow hasn't changed
        self.validate_workflow_unchanged(&state, &config)?;

        // Display resume information
        self.user_interaction.display_info(&format!(
            "🔄 Resuming session: {} from {}",
            session_id,
            state
                .get_resume_info()
                .unwrap_or_else(|| "unknown state".to_string())
        ));

        // Restore the environment
        let env = self.restore_environment(&state, &config).await?;

        // Update the session manager with the loaded state
        // Use the working directory from the restored environment
        let session_file = env.working_dir.join(".prodigy").join("session_state.json");
        self.session_manager.load_state(&session_file).await?;

        // Transition session status to InProgress (from Failed, Interrupted, etc.)
        self.session_manager
            .update_session(SessionUpdate::UpdateStatus(SessionStatus::InProgress))
            .await?;

        // Resume the workflow execution from the saved state
        if let Some(ref workflow_state) = state.workflow_state {
            // Restore config from saved workflow state
            self.restore_config_from_workflow_state(&mut config, workflow_state);

            // Restore execution context if available
            if let Some(ref exec_context) = state.execution_context {
                // This context would need to be passed to the workflow executor
                // For now, we'll just log that it was restored
                self.user_interaction.display_info(&format!(
                    "Restored {} variables and {} step outputs",
                    exec_context.variables.len(),
                    exec_context.step_outputs.len()
                ));
            }

            // Execute the workflow starting from the saved position
            let result = self
                .resume_workflow_execution(
                    &env,
                    &config,
                    workflow_state.current_iteration,
                    workflow_state.current_step,
                )
                .await;

            // Handle the result (success, interruption, or failure)
            self.handle_resume_result(result, &session_file, session_id, &config.command.playbook)
                .await?;

            // Cleanup - need to call the cleanup function from the orchestrator
            // For now, we'll just complete the session without cleanup
            // TODO: Pass cleanup function as a parameter or make it available

            // Complete session and display summary
            let summary = self.session_manager.complete_session().await?;
            self.display_session_completion(&summary, config.command.dry_run);
        } else {
            return Err(anyhow!(
                "Session {} has no workflow state to resume",
                session_id
            ));
        }

        Ok(())
    }

    /// Restore the execution environment from saved state
    async fn restore_environment(
        &self,
        state: &SessionState,
        config: &CookConfig,
    ) -> Result<ExecutionEnvironment> {
        self.session_ops.restore_environment(state, config).await
    }

    /// Resume workflow execution from a specific point
    async fn resume_workflow_execution(
        &self,
        env: &ExecutionEnvironment,
        config: &CookConfig,
        start_iteration: usize,
        start_step: usize,
    ) -> Result<()> {
        use super::core::WorkflowType;

        self.user_interaction.display_info(&format!(
            "Resuming from iteration {} step {}",
            start_iteration + 1,
            start_step + 1
        ));

        // Load existing completed steps from session state
        let existing_state = self
            .session_manager
            .get_state()
            .context("Failed to get session state before workflow execution")?;
        let completed_steps = existing_state
            .workflow_state
            .as_ref()
            .map(|ws| ws.completed_steps.clone())
            .unwrap_or_default();

        // Create workflow state for checkpointing
        let workflow_state = crate::cook::session::WorkflowState {
            current_iteration: start_iteration,
            current_step: start_step,
            completed_steps,
            workflow_path: config.command.playbook.clone(),
            input_args: config.command.args.clone(),
            map_patterns: config.command.map.clone(),
            using_worktree: true,
        };

        // Update session with workflow state
        self.session_manager
            .update_session(SessionUpdate::UpdateWorkflowState(workflow_state))
            .await?;

        // Determine workflow type and route to appropriate resume handler
        let workflow_type = super::core::DefaultCookOrchestrator::classify_workflow_type(config);

        // For MapReduce workflows, use specialized resume mechanism
        if workflow_type == WorkflowType::MapReduce {
            // Check if there's an existing MapReduce job to resume
            if let Some(_mapreduce_config) = &config.mapreduce_config {
                // MapReduce workflows need to be executed through the orchestrator
                // which has access to the MapReduce execution logic
                return Err(anyhow!(
                    "MapReduce resume requires orchestrator-level execution"
                ));
            }
        }

        // Execute the workflow based on type, but skip completed steps
        match workflow_type {
            WorkflowType::MapReduce => {
                // MapReduce workflows have their own resume mechanism
                Err(anyhow!(
                    "MapReduce workflow requires mapreduce configuration"
                ))
            }
            WorkflowType::StructuredWithOutputs => {
                self.workflow_executor
                    .execute_structured_workflow_from(env, config, start_iteration, start_step)
                    .await
            }
            WorkflowType::WithArguments => {
                self.workflow_executor
                    .execute_iterative_workflow_from(env, config, start_iteration, start_step)
                    .await
            }
            WorkflowType::Standard => {
                self.workflow_executor
                    .execute_standard_workflow_from(env, config, start_iteration, start_step)
                    .await
            }
        }
    }

    /// Execute a structured workflow with outputs
    pub async fn execute_structured_workflow(
        &self,
        env: &ExecutionEnvironment,
        config: &CookConfig,
    ) -> Result<()> {
        // Analysis will be run per-command as needed based on their configuration

        // Track outputs from previous commands
        let mut command_outputs: HashMap<String, HashMap<String, String>> = HashMap::new();

        // Execute iterations if configured
        let max_iterations = config.command.max_iterations;
        for iteration in 1..=max_iterations {
            if iteration > 1 {
                self.user_interaction
                    .display_progress(&format!("Starting iteration {iteration}/{max_iterations}"));
            }

            // Increment iteration counter once per iteration, not per command
            self.session_manager
                .update_session(SessionUpdate::IncrementIteration)
                .await?;

            // Execute each command in sequence
            for (step_index, cmd) in config.workflow.commands.iter().enumerate() {
                let mut command = cmd.to_command();
                // Apply defaults from the command registry
                crate::config::apply_command_defaults(&mut command);

                // Display step start with description
                let step_description = build_step_description(&command.name, &command.args);
                self.user_interaction.step_start(
                    (step_index + 1) as u32,
                    config.workflow.commands.len() as u32,
                    &step_description,
                );

                // Analysis functionality has been removed in v0.3.0

                // Resolve variables from command outputs for use in variable expansion
                let resolved_variables = build_variable_map(&command_outputs);

                // Build final command string with resolved arguments
                let final_command =
                    build_command_string(&command.name, &command.args, &resolved_variables);

                self.user_interaction
                    .display_action(&format!("Executing command: {final_command}"));

                // Execute the command
                let mut env_vars = HashMap::new();
                env_vars.insert("PRODIGY_AUTOMATION".to_string(), "true".to_string());

                let result = self
                    .claude_executor
                    .execute_claude_command(&final_command, &env.working_dir, env_vars)
                    .await?;

                if !result.success {
                    anyhow::bail!(
                        "Command '{}' failed with exit code {:?}. Error: {}",
                        command.name,
                        result.exit_code,
                        result.stderr
                    );
                } else {
                    // Track file changes when command succeeds
                    self.session_manager
                        .update_session(SessionUpdate::AddFilesChanged(1))
                        .await?;
                }

                // Handle outputs if specified
                if let Some(ref outputs) = command.outputs {
                    let mut cmd_output_map = HashMap::new();

                    for (output_name, output_decl) in outputs {
                        self.user_interaction.display_info(&format!(
                            "🔍 Looking for output '{}' with pattern: {}",
                            output_name, output_decl.file_pattern
                        ));

                        // Find files matching the pattern in git commits
                        let pattern_result = self
                            .find_files_matching_pattern(
                                &output_decl.file_pattern,
                                &env.working_dir,
                            )
                            .await;

                        match pattern_result {
                            Ok(file_path) => {
                                self.user_interaction
                                    .display_success(&format!("Found output file: {file_path}"));
                                cmd_output_map.insert(output_name.clone(), file_path);
                            }
                            Err(e) => {
                                self.user_interaction.display_warning(&format!(
                                    "Failed to find output '{output_name}': {e}"
                                ));
                                return Err(e);
                            }
                        }
                    }

                    // Store outputs for this command if it has an ID
                    if should_store_outputs(&command.id) {
                        if let Some(ref id) = command.id {
                            command_outputs.insert(id.clone(), cmd_output_map);
                            self.user_interaction
                                .display_success(&format!("💾 Stored outputs for command '{id}'"));
                        }
                    }
                }
            }

            // Check if we should continue iterations
            if iteration < max_iterations {
                // Could add logic here to check if improvements were made
                // For now, continue with all iterations as requested
            }
        }

        Ok(())
    }

    /// Find files matching a pattern in the last git commit
    pub async fn find_files_matching_pattern(
        &self,
        pattern: &str,
        working_dir: &std::path::Path,
    ) -> Result<String> {
        use tokio::process::Command;

        self.user_interaction.display_info(&format!(
            "🔎 Searching for files matching '{pattern}' in last commit"
        ));

        // Get list of files changed in the last commit
        let output = Command::new("git")
            .args(["diff", "--name-only", "HEAD~1", "HEAD"])
            .current_dir(working_dir)
            .output()
            .await?;

        if !output.status.success() {
            return Err(anyhow!(
                "Failed to get git diff: {}",
                String::from_utf8_lossy(&output.stderr)
            ));
        }

        let files = String::from_utf8(output.stdout)?;

        // Check each file in the diff against the pattern
        for file in files.lines() {
            let file = file.trim();
            if file.is_empty() {
                continue;
            }

            // Match based on pattern type
            let matches = if let Some(suffix) = pattern.strip_prefix('*') {
                // Wildcard pattern - match suffix
                file.ends_with(suffix)
            } else if pattern.contains('*') {
                // Glob-style pattern
                self.matches_glob_pattern(file, pattern)
            } else {
                // Simple substring match - just check if filename contains pattern
                file.split('/')
                    .next_back()
                    .unwrap_or(file)
                    .contains(pattern)
            };

            if matches {
                let full_path = working_dir.join(file);
                return Ok(full_path.to_string_lossy().to_string());
            }
        }

        Err(anyhow!(
            "No files found matching pattern '{}' in last commit",
            pattern
        ))
    }

    /// Helper to match glob-style patterns
    pub fn matches_glob_pattern(&self, file: &str, pattern: &str) -> bool {
        super::workflow_classifier::matches_glob_pattern(file, pattern)
    }

    /// Execute a MapReduce workflow with a pre-configured executor
    pub async fn execute_mapreduce_workflow_with_executor(
        &self,
        env: &ExecutionEnvironment,
        config: &CookConfig,
        mapreduce_config: &crate::config::MapReduceWorkflowConfig,
        mut executor: crate::cook::workflow::WorkflowExecutorImpl,
    ) -> Result<()> {
        // Display MapReduce-specific message
        self.user_interaction.display_info(&format!(
            "Executing MapReduce workflow: {}",
            mapreduce_config.name
        ));

        // Convert MapReduce config to ExtendedWorkflowConfig
        // Extract setup commands if they exist
        let setup_steps = mapreduce_config
            .setup
            .as_ref()
            .map(|setup| setup.commands.clone())
            .unwrap_or_default();

        // Use pure functions for environment variable interpolation
        use crate::cook::environment::EnvValue;
        use crate::cook::execution::mapreduce::env_interpolation::{
            env_values_to_plain_map, interpolate_workflow_env_with_positional_args,
            positional_args_as_env_vars,
        };

        // Interpolate workflow env variables with positional args BEFORE creating phases
        // This ensures that env vars like "BLOG_POST: $1" get resolved to actual values
        let mut interpolated_env = interpolate_workflow_env_with_positional_args(
            mapreduce_config.env.as_ref(),
            &config.command.args,
        )?;

        // Also add positional args themselves as environment variables (ARG_1, ARG_2, etc.)
        let positional_env = positional_args_as_env_vars(&config.command.args);
        interpolated_env.extend(positional_env);

        // Add auto-merge/auto-confirm flags to workflow environment if -y flag is provided.
        // This passes the flags through the workflow environment instead of using global
        // env mutation, ensuring child processes receive these values via their env map.
        if config.command.auto_accept {
            interpolated_env.insert(
                "PRODIGY_AUTO_MERGE".to_string(),
                EnvValue::Static("true".to_string()),
            );
            interpolated_env.insert(
                "PRODIGY_AUTO_CONFIRM".to_string(),
                EnvValue::Static("true".to_string()),
            );
        }

        // Convert interpolated EnvValue map to plain HashMap<String, String> for MapPhase
        let workflow_env_plain = env_values_to_plain_map(&interpolated_env);

        // Create MapPhase with interpolated env
        let mut map_phase = mapreduce_config.to_map_phase().context(
            "Failed to resolve MapReduce configuration. Check that environment variables are properly defined."
        )?;
        // Override the workflow_env with interpolated values
        map_phase.workflow_env = workflow_env_plain;

        let extended_workflow = ExtendedWorkflowConfig {
            name: mapreduce_config.name.clone(),
            mode: crate::cook::workflow::WorkflowMode::MapReduce,
            steps: setup_steps,
            setup_phase: mapreduce_config.to_setup_phase().context(
                "Failed to resolve setup phase configuration. Check that environment variables are properly defined."
            )?,
            map_phase: Some(map_phase),
            reduce_phase: mapreduce_config.to_reduce_phase(),
            max_iterations: 1, // MapReduce runs once
            iterate: false,
            retry_defaults: None,
            environment: None,
            // collect_metrics removed - MMM focuses on orchestration
        };

        // Set global environment configuration if present in MapReduce workflow
        if !interpolated_env.is_empty()
            || mapreduce_config.secrets.is_some()
            || mapreduce_config.env_files.is_some()
            || mapreduce_config.profiles.is_some()
        {
            let global_env_config = crate::cook::environment::EnvironmentConfig {
                global_env: interpolated_env,
                secrets: mapreduce_config.secrets.clone().unwrap_or_default(),
                env_files: mapreduce_config.env_files.clone().unwrap_or_default(),
                inherit: true,
                profiles: mapreduce_config.profiles.clone().unwrap_or_default(),
                active_profile: None,
            };
            executor = executor.with_environment_config(global_env_config)?;
        }
        // Also check standard workflow env (for backward compatibility with workflows that use both)
        else if config.workflow.env.is_some()
            || config.workflow.secrets.is_some()
            || config.workflow.env_files.is_some()
            || config.workflow.profiles.is_some()
        {
            // Interpolate standard workflow env vars too
            let interpolated_standard_env = interpolate_workflow_env_with_positional_args(
                config.workflow.env.as_ref(),
                &config.command.args,
            )?;

            let global_env_config = crate::cook::environment::EnvironmentConfig {
                global_env: interpolated_standard_env,
                secrets: config.workflow.secrets.clone().unwrap_or_default(),
                env_files: config.workflow.env_files.clone().unwrap_or_default(),
                inherit: true,
                profiles: config.workflow.profiles.clone().unwrap_or_default(),
                active_profile: None,
            };
            executor = executor.with_environment_config(global_env_config)?;
        }

        // Execute the MapReduce workflow
        executor.execute(&extended_workflow, env).await
    }
}

/// Interpolate workflow environment variables with positional arguments
///
/// Resolves variable references like "$1", "${ARG_1}", etc. in workflow env values
/// with actual positional argument values. This enables workflows to reference
/// command-line arguments in their environment variable definitions.
///
/// # Arguments
/// * `workflow_env` - Optional workflow environment variable map from YAML
/// * `positional_args` - Positional arguments from command line (--args)
///
/// # Returns
/// HashMap of fully interpolated environment variables ready for use
///
/// # Example
/// ```text
/// // Workflow YAML has: env: { BLOG_POST: "$1", OUTPUT: "${ARG_2}" }
/// // Command: prodigy run workflow.yml --args blog.md output/
/// // Result: { BLOG_POST: "blog.md", OUTPUT: "output/" }
/// ```
/// Build a variable map from command outputs
///
/// This pure function takes command outputs and transforms them into a flat
/// map of variables with names in the format "command_id.output_name".
///
/// # Arguments
/// * `command_outputs` - Map of command IDs to their output maps
///
/// # Returns
/// A flat HashMap where keys are "command_id.output_name" and values are the output values
fn build_variable_map(
    command_outputs: &HashMap<String, HashMap<String, String>>,
) -> HashMap<String, String> {
    let mut resolved_variables = HashMap::new();

    // Collect all available outputs as variables
    for (cmd_id, outputs) in command_outputs {
        for (output_name, value) in outputs {
            let var_name = format!("{cmd_id}.{output_name}");
            resolved_variables.insert(var_name, value.clone());
        }
    }

    resolved_variables
}

/// Build a command string from command name and arguments
///
/// This pure function constructs a complete command string by combining the
/// command name with resolved arguments, filtering out empty arguments.
///
/// # Arguments
/// * `command_name` - The name of the command (without leading '/')
/// * `args` - The command arguments to resolve
/// * `variables` - Map of variables for argument resolution
///
/// # Returns
/// A complete command string with format "/<command_name> \<arg1\> \<arg2\> ..."
fn build_command_string(
    command_name: &str,
    args: &[crate::config::command::CommandArg],
    variables: &HashMap<String, String>,
) -> String {
    let mut cmd_parts = vec![format!("/{command_name}")];
    for arg in args {
        let resolved_arg = arg.resolve(variables);
        if !resolved_arg.is_empty() {
            cmd_parts.push(resolved_arg);
        }
    }
    cmd_parts.join(" ")
}

/// Determine if command outputs should be stored
///
/// This pure function checks if a command has an ID, which determines whether
/// its outputs should be stored for later reference by other commands.
///
/// # Arguments
/// * `command_id` - Optional command ID
///
/// # Returns
/// true if outputs should be stored, false otherwise
fn should_store_outputs(command_id: &Option<String>) -> bool {
    command_id.is_some()
}

/// Build a step description from command name and arguments
///
/// This pure function constructs a human-readable step description by
/// combining the command name with resolved arguments, filtering out empty ones.
///
/// # Arguments
/// * `command_name` - The name of the command
/// * `args` - The command arguments to resolve
///
/// # Returns
/// A formatted string like "command_name: arg1 arg2 arg3"
fn build_step_description(
    command_name: &str,
    args: &[crate::config::command::CommandArg],
) -> String {
    let args_str = args
        .iter()
        .map(|a| a.resolve(&HashMap::new()))
        .filter(|s| !s.is_empty())
        .collect::<Vec<_>>()
        .join(" ");

    if args_str.is_empty() {
        command_name.to_string()
    } else {
        format!("{command_name}: {args_str}")
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_classify_execution_result_success() {
        let result: Result<()> = Ok(());
        let outcome = classify_execution_result(&result, SessionStatus::InProgress);
        assert_eq!(outcome, ExecutionOutcome::Success);
    }

    #[test]
    fn test_classify_execution_result_interrupted() {
        let result: Result<()> = Err(anyhow!("interrupted"));
        let outcome = classify_execution_result(&result, SessionStatus::Interrupted);
        assert_eq!(outcome, ExecutionOutcome::Interrupted);
    }

    #[test]
    fn test_classify_execution_result_failed() {
        let result: Result<()> = Err(anyhow!("test error"));
        let outcome = classify_execution_result(&result, SessionStatus::InProgress);
        match outcome {
            ExecutionOutcome::Failed(msg) => assert!(msg.contains("test error")),
            _ => panic!("Expected Failed outcome"),
        }
    }

    #[test]
    fn test_should_save_checkpoint_on_interrupted() {
        let outcome = ExecutionOutcome::Interrupted;
        assert!(should_save_checkpoint(&outcome));
    }

    #[test]
    fn test_should_not_save_checkpoint_on_success() {
        let outcome = ExecutionOutcome::Success;
        assert!(!should_save_checkpoint(&outcome));
    }

    #[test]
    fn test_should_save_checkpoint_on_failed() {
        // Spec 186: Checkpoints should be saved on failure to enable resume
        let outcome = ExecutionOutcome::Failed("error".to_string());
        assert!(should_save_checkpoint(&outcome));
    }

    #[test]
    fn test_determine_resume_message_interrupted() {
        let outcome = ExecutionOutcome::Interrupted;
        let message = determine_resume_message("session-123", "workflow.yml", &outcome);
        assert!(message.is_some());
        assert!(message
            .unwrap()
            .contains("prodigy run workflow.yml --resume session-123"));
    }

    #[test]
    fn test_determine_resume_message_failed() {
        let outcome = ExecutionOutcome::Failed("error".to_string());
        let message = determine_resume_message("session-123", "workflow.yml", &outcome);
        assert!(message.is_some());
        assert!(message.unwrap().contains("prodigy resume session-123"));
    }

    #[test]
    fn test_determine_resume_message_success() {
        let outcome = ExecutionOutcome::Success;
        let message = determine_resume_message("session-123", "workflow.yml", &outcome);
        assert!(message.is_none());
    }

    #[test]
    fn test_build_variable_map_empty() {
        let command_outputs = HashMap::new();
        let result = build_variable_map(&command_outputs);
        assert!(result.is_empty());
    }

    #[test]
    fn test_build_variable_map_single_command_single_output() {
        let mut command_outputs = HashMap::new();
        let mut outputs = HashMap::new();
        outputs.insert("result".to_string(), "value1".to_string());
        command_outputs.insert("cmd1".to_string(), outputs);

        let result = build_variable_map(&command_outputs);

        assert_eq!(result.len(), 1);
        assert_eq!(result.get("cmd1.result"), Some(&"value1".to_string()));
    }

    #[test]
    fn test_build_variable_map_multiple_commands_multiple_outputs() {
        let mut command_outputs = HashMap::new();

        let mut outputs1 = HashMap::new();
        outputs1.insert("result".to_string(), "value1".to_string());
        outputs1.insert("status".to_string(), "ok".to_string());
        command_outputs.insert("cmd1".to_string(), outputs1);

        let mut outputs2 = HashMap::new();
        outputs2.insert("output".to_string(), "data".to_string());
        command_outputs.insert("cmd2".to_string(), outputs2);

        let result = build_variable_map(&command_outputs);

        assert_eq!(result.len(), 3);
        assert_eq!(result.get("cmd1.result"), Some(&"value1".to_string()));
        assert_eq!(result.get("cmd1.status"), Some(&"ok".to_string()));
        assert_eq!(result.get("cmd2.output"), Some(&"data".to_string()));
    }

    #[test]
    fn test_build_command_string_no_args() {
        use crate::config::command::CommandArg;
        let args: Vec<CommandArg> = vec![];
        let variables = HashMap::new();

        let result = build_command_string("test-cmd", &args, &variables);

        assert_eq!(result, "/test-cmd");
    }

    #[test]
    fn test_build_command_string_with_static_args() {
        use crate::config::command::CommandArg;
        let args = vec![
            CommandArg::Literal("arg1".to_string()),
            CommandArg::Literal("arg2".to_string()),
        ];
        let variables = HashMap::new();

        let result = build_command_string("test-cmd", &args, &variables);

        assert_eq!(result, "/test-cmd arg1 arg2");
    }

    #[test]
    fn test_build_command_string_with_variable_references() {
        use crate::config::command::CommandArg;
        let args = vec![
            CommandArg::Variable("FILE".to_string()),
            CommandArg::Literal("--flag".to_string()),
        ];
        let mut variables = HashMap::new();
        variables.insert("FILE".to_string(), "test.txt".to_string());

        let result = build_command_string("test-cmd", &args, &variables);

        assert_eq!(result, "/test-cmd test.txt --flag");
    }

    #[test]
    fn test_build_command_string_filters_empty_args() {
        use crate::config::command::CommandArg;
        let args = vec![
            CommandArg::Literal("arg1".to_string()),
            CommandArg::Literal("".to_string()), // Empty literal will be filtered
            CommandArg::Literal("arg2".to_string()),
        ];
        let variables = HashMap::new();

        let result = build_command_string("test-cmd", &args, &variables);

        assert_eq!(result, "/test-cmd arg1 arg2");
    }

    #[test]
    fn test_should_store_outputs_with_id() {
        let command_id = Some("cmd1".to_string());
        assert!(should_store_outputs(&command_id));
    }

    #[test]
    fn test_should_store_outputs_without_id() {
        let command_id: Option<String> = None;
        assert!(!should_store_outputs(&command_id));
    }

    #[test]
    fn test_should_store_outputs_with_empty_id() {
        // Even an empty string ID should allow storage
        let command_id = Some("".to_string());
        assert!(should_store_outputs(&command_id));
    }

    #[test]
    fn test_build_step_description_no_args() {
        use crate::config::command::CommandArg;
        let args: Vec<CommandArg> = vec![];

        let result = build_step_description("test-cmd", &args);

        assert_eq!(result, "test-cmd");
    }

    #[test]
    fn test_build_step_description_with_args() {
        use crate::config::command::CommandArg;
        let args = vec![
            CommandArg::Literal("arg1".to_string()),
            CommandArg::Literal("arg2".to_string()),
        ];

        let result = build_step_description("test-cmd", &args);

        assert_eq!(result, "test-cmd: arg1 arg2");
    }

    #[test]
    fn test_build_step_description_filters_empty() {
        use crate::config::command::CommandArg;
        let args = vec![
            CommandArg::Literal("arg1".to_string()),
            CommandArg::Literal("".to_string()), // Empty arg
            CommandArg::Literal("arg2".to_string()),
        ];

        let result = build_step_description("test-cmd", &args);

        assert_eq!(result, "test-cmd: arg1 arg2");
    }
}