mur_core/workflow/

runner.rs

//! Workflow execution engine — runs workflow steps with constitution checks,
//! audit logging, checkpoints, and failure handling.

use crate::audit::AuditStore;
use crate::constitution::Constitution;
use crate::model::router::ModelRouter;
use crate::types::{
    Action, ActionDecision, ActionType, ExecutionResult, FailureAction, Step, StepResult, StepType,
    Workflow,
};
use crate::workflow::autofix::AutoFixEngine;
use crate::workflow::checkpoint::{save_checkpoint, Checkpoint};
use crate::workflow::parser::resolve_variables;
use anyhow::{Context, Result};
use chrono::Utc;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Instant;
use tokio::process::Command;
use uuid::Uuid;

/// Workflow execution engine.
pub struct WorkflowRunner {
    pub constitution: Arc<Constitution>,
    pub audit: Arc<AuditStore>,
    pub model_router: Option<Arc<ModelRouter>>,
}

impl WorkflowRunner {
    pub fn new(constitution: Arc<Constitution>, audit: Arc<AuditStore>) -> Self {
        Self {
            constitution,
            audit,
            model_router: None,
        }
    }

    /// Create a runner with a model router for AI-powered steps.
    pub fn with_model_router(mut self, router: Arc<ModelRouter>) -> Self {
        self.model_router = Some(router);
        self
    }

    /// Execute a workflow, optionally in shadow (dry-run) mode.
    pub async fn execute(
        &self,
        workflow: &Workflow,
        user_vars: HashMap<String, String>,
        shadow: bool,
    ) -> Result<ExecutionResult> {
        let execution_id = Uuid::new_v4();
        let started_at = Utc::now();
        let timer = Instant::now();

        // Merge user variables with workflow defaults
        let mut variables = workflow.variables.clone();
        variables.extend(user_vars);

        let mut step_results = Vec::new();
        let mut total_cost = 0.0;
        let mut success = true;
        let mut error_msg: Option<String> = None;

        tracing::info!(
            "Starting workflow '{}' (execution={}, shadow={})",
            workflow.name,
            execution_id,
            shadow
        );

        for (idx, step) in workflow.steps.iter().enumerate() {
            // Save checkpoint before each step
            let checkpoint = Checkpoint {
                execution_id,
                step_index: idx,
                variables: variables.clone(),
                completed_steps: step_results.clone(),
                created_at: Utc::now(),
            };
            if let Err(e) = save_checkpoint(&checkpoint).await {
                tracing::warn!("Failed to save checkpoint: {}", e);
            }

            // Resolve variables in action
            let resolved_action = resolve_variables(&step.action, &variables);

            // Constitution check
            let action = step_to_action(step, &resolved_action);
            let decision = self.constitution.check_action(&action);

            match decision {
                ActionDecision::Blocked { reason } => {
                    tracing::warn!("Step '{}' blocked: {}", step.name, reason);
                    self.audit.log_action(&action, "blocked", &reason);
                    step_results.push(StepResult {
                        step_name: step.name.clone(),
                        success: false,
                        output: String::new(),
                        duration_ms: 0,
                        cost: 0.0,
                        model_used: None,
                        error: Some(format!("Blocked by constitution: {}", reason)),
                    });
                    success = false;
                    error_msg = Some(format!("Step '{}' blocked: {}", step.name, reason));
                    break;
                }
                ActionDecision::NeedsApproval { prompt } => {
                    tracing::info!("Step '{}' needs approval: {}", step.name, prompt);
                    // For now, treat as breakpoint — log and skip
                    step_results.push(StepResult {
                        step_name: step.name.clone(),
                        success: false,
                        output: String::new(),
                        duration_ms: 0,
                        cost: 0.0,
                        model_used: None,
                        error: Some(format!("Needs approval: {}", prompt)),
                    });
                    success = false;
                    error_msg = Some(format!("Step '{}' needs approval", step.name));
                    break;
                }
                ActionDecision::Allowed => {}
            }

            // Breakpoint check
            if step.breakpoint {
                let msg = step
                    .breakpoint_message
                    .as_deref()
                    .unwrap_or("Breakpoint reached");
                tracing::info!("⏸ Breakpoint at '{}': {}", step.name, msg);
                if shadow {
                    tracing::info!("  [shadow] Would pause here for confirmation");
                }
                // In non-interactive mode, we log and continue
                // Interactive breakpoints will be added in Phase 3
            }

            // Execute step
            let step_result = if shadow {
                execute_step_shadow(step, &resolved_action)
            } else {
                execute_step(step, &resolved_action, self.model_router.as_ref(), &step_results.iter().map(|r: &StepResult| (r.step_name.clone(), r.output.clone())).collect::<Vec<_>>()).await
            };

            match step_result {
                Ok(mut result) => {
                    total_cost += result.cost;
                    self.audit
                        .log_action(&action, "executed", &result.output);

                    if !result.success {
                        match &step.on_failure {
                            FailureAction::Abort => {
                                error_msg = Some(format!(
                                    "Step '{}' failed: {}",
                                    step.name,
                                    result.error.as_deref().unwrap_or("unknown")
                                ));
                                success = false;
                                step_results.push(result);
                                break;
                            }
                            FailureAction::Skip => {
                                tracing::info!("Skipping failed step '{}'", step.name);
                                result.output =
                                    format!("[skipped] {}", result.error.as_deref().unwrap_or(""));
                                step_results.push(result);
                            }
                            FailureAction::Retry { max } => {
                                let mut retried = false;
                                for attempt in 1..=*max {
                                    tracing::info!(
                                        "Retrying step '{}' (attempt {}/{})",
                                        step.name,
                                        attempt,
                                        max
                                    );
                                    let retry_result = if shadow {
                                        execute_step_shadow(step, &resolved_action)
                                    } else {
                                        execute_step(step, &resolved_action, self.model_router.as_ref(), &step_results.iter().map(|r: &StepResult| (r.step_name.clone(), r.output.clone())).collect::<Vec<_>>()).await
                                    };
                                    match retry_result {
                                        Ok(r) if r.success => {
                                            step_results.push(r);
                                            retried = true;
                                            break;
                                        }
                                        Ok(r) => result = r,
                                        Err(e) => {
                                            result.error = Some(e.to_string());
                                        }
                                    }
                                }
                                if !retried {
                                    error_msg = Some(format!(
                                        "Step '{}' failed after {} retries",
                                        step.name, max
                                    ));
                                    success = false;
                                    step_results.push(result);
                                    break;
                                }
                            }
                            FailureAction::AutoFix => {
                                if let Some(router) = &self.model_router {
                                    let autofix = AutoFixEngine::new(router.clone());
                                    let error_output = result
                                        .error
                                        .as_deref()
                                        .unwrap_or(&result.output);
                                    let max_fix_attempts = 3u32;

                                    tracing::info!(
                                        "AutoFix: analyzing failure for step '{}'",
                                        step.name
                                    );

                                    eprintln!("🔧 Auto-fix: analyzing failure...");
                                    match autofix
                                        .analyze_and_fix(
                                            &step.name,
                                            &resolved_action,
                                            error_output,
                                            max_fix_attempts,
                                        )
                                        .await
                                    {
                                        Ok(fix_result) if fix_result.success => {
                                            eprintln!("🔧 Auto-fix found: {}", fix_result.fix_command);
                                            tracing::info!(
                                                "AutoFix: got fix command after {} attempts: {}",
                                                fix_result.attempts,
                                                fix_result.fix_command
                                            );

                                            // Log the fix attempt to audit
                                            self.audit.log_action(
                                                &action,
                                                "autofix_attempt",
                                                &format!(
                                                    "Fix: {}\nAnalysis: {}",
                                                    fix_result.fix_command,
                                                    fix_result.analysis
                                                ),
                                            );

                                            // Execute the fix command
                                            let fix_step = Step {
                                                name: format!("{}-autofix", step.name),
                                                step_type: StepType::Execute,
                                                action: fix_result.fix_command.clone(),
                                                on_failure: FailureAction::Abort,
                                                breakpoint: false,
                                                breakpoint_message: None,
                                            };
                                            match execute_step(
                                                &fix_step,
                                                &fix_result.fix_command,
                                                self.model_router.as_ref(), &step_results.iter().map(|r: &StepResult| (r.step_name.clone(), r.output.clone())).collect::<Vec<_>>(),
                                            )
                                            .await
                                            {
                                                Ok(fix_exec) if fix_exec.success => {
                                                    eprintln!("🔧 Auto-fix succeeded! Using fix result.");
                                                    total_cost += fix_exec.cost;

                                                    // Use the fix result directly (the fix command
                                                    // replaces the broken original)
                                                    let mut fixed_result = fix_exec;
                                                    fixed_result.step_name = step.name.clone();
                                                    step_results.push(fixed_result);
                                                }
                                                Ok(_fix_exec) => {
                                                    // Fix command itself failed
                                                    success = false;
                                                    error_msg = Some(format!(
                                                        "Step '{}' auto-fix command failed",
                                                        step.name
                                                    ));
                                                    step_results.push(result);
                                                    break;
                                                }
                                                Err(e) => {
                                                    success = false;
                                                    error_msg = Some(format!(
                                                        "Step '{}' fix execution error: {}",
                                                        step.name, e
                                                    ));
                                                    step_results.push(result);
                                                    break;
                                                }
                                            }
                                        }
                                        Ok(_) => {
                                            // AutoFix couldn't generate a fix
                                            success = false;
                                            error_msg = Some(format!(
                                                "Step '{}' failed, auto-fix could not find a solution",
                                                step.name
                                            ));
                                            step_results.push(result);
                                            break;
                                        }
                                        Err(e) => {
                                            // AutoFix errored out (no API key, network, etc.)
                                            tracing::warn!(
                                                "AutoFix error for step '{}': {}",
                                                step.name,
                                                e
                                            );
                                            success = false;
                                            error_msg = Some(format!(
                                                "Step '{}' failed, auto-fix error: {}",
                                                step.name, e
                                            ));
                                            step_results.push(result);
                                            break;
                                        }
                                    }
                                } else {
                                    // No model router — can't auto-fix
                                    tracing::warn!(
                                        "AutoFix requested but no model router configured, aborting at '{}'",
                                        step.name
                                    );
                                    success = false;
                                    error_msg = Some(format!(
                                        "Step '{}' failed, auto-fix requires AI models (run murc init)",
                                        step.name
                                    ));
                                    step_results.push(result);
                                    break;
                                }
                            }
                        }
                    } else {
                        step_results.push(result);
                    }
                }
                Err(e) => {
                    success = false;
                    error_msg = Some(format!("Step '{}' error: {}", step.name, e));
                    step_results.push(StepResult {
                        step_name: step.name.clone(),
                        success: false,
                        output: String::new(),
                        duration_ms: 0,
                        cost: 0.0,
                        model_used: None,
                        error: Some(e.to_string()),
                    });
                    break;
                }
            }
        }

        let finished_at = Utc::now();
        let duration_ms = timer.elapsed().as_millis() as u64;

        Ok(ExecutionResult {
            execution_id,
            workflow_id: workflow.id.clone(),
            steps_completed: step_results.iter().filter(|r| r.success).count(),
            steps_total: workflow.steps.len(),
            success,
            duration_ms,
            total_cost,
            step_results,
            shadow,
            error: error_msg,
            started_at,
            finished_at,
        })
    }
}

/// Execute a step — shell for `execute` type, AI model for analyze/plan/debug/etc.
async fn execute_step(
    step: &Step,
    resolved_action: &str,
    model_router: Option<&Arc<ModelRouter>>,
    prior_outputs: &[(String, String)],
) -> Result<StepResult> {
    let timer = Instant::now();

    let is_ai_step = matches!(
        step.step_type,
        StepType::Analyze
            | StepType::Plan
            | StepType::Debug
            | StepType::Summarize
            | StepType::Code
            | StepType::Refactor
            | StepType::Fix
            | StepType::Search
            | StepType::Classify
            | StepType::SecurityCheck
    );

    // Use AI model if available and step type requires it
    if is_ai_step {
        if let Some(router) = model_router {
            tracing::info!(
                "Step '{}' using AI model (type={:?})",
                step.name,
                step.step_type
            );
            // Build context from previous steps for AI
            let context_prompt = if !prior_outputs.is_empty() {
                let ctx: String = prior_outputs
                    .iter()
                    .map(|(name, output)| format!("## Output from step '{}':\n{}\n", name, output))
                    .collect();
                format!("{}\n\n## Task:\n{}", ctx, resolved_action)
            } else {
                resolved_action.to_string()
            };
            match router.complete_for_step(&step.step_type, &context_prompt).await {
                Ok(response) => {
                    return Ok(StepResult {
                        step_name: step.name.clone(),
                        success: true,
                        output: response.content,
                        duration_ms: timer.elapsed().as_millis() as u64,
                        cost: response.cost,
                        model_used: Some(response.model),
                        error: None,
                    });
                }
                Err(e) => {
                    let err_msg = format!("AI model error: {}", e);
                    tracing::warn!(
                        "AI model failed for step '{}': {}, falling back to shell",
                        step.name,
                        e
                    );
                    // For AI steps, return the error instead of falling back to shell
                    // (shell execution of a prompt makes no sense)
                    return Ok(StepResult {
                        step_name: step.name.clone(),
                        success: false,
                        output: err_msg.clone(),
                        duration_ms: timer.elapsed().as_millis() as u64,
                        cost: 0.0,
                        model_used: None,
                        error: Some(err_msg),
                    });
                }
            }
        } else {
            tracing::debug!(
                "Step '{}' is AI-powered but no model router configured, using shell",
                step.name
            );
        }
    }

    // Shell execution (default for execute type, fallback for AI steps)
    let output = Command::new("sh")
        .arg("-c")
        .arg(resolved_action)
        .output()
        .await
        .with_context(|| format!("Executing step '{}'", step.name))?;

    let stdout = String::from_utf8_lossy(&output.stdout).to_string();
    let stderr = String::from_utf8_lossy(&output.stderr).to_string();
    let combined = if stderr.is_empty() {
        stdout.clone()
    } else {
        format!("{}\n{}", stdout, stderr)
    };

    Ok(StepResult {
        step_name: step.name.clone(),
        success: output.status.success(),
        output: combined,
        duration_ms: timer.elapsed().as_millis() as u64,
        cost: 0.0,
        model_used: None,
        error: if output.status.success() {
            None
        } else {
            Some(format!(
                "Exit code: {}. {}",
                output.status.code().unwrap_or(-1),
                stderr.trim()
            ))
        },
    })
}

/// Shadow mode: log what would happen without executing.
fn execute_step_shadow(step: &Step, resolved_action: &str) -> Result<StepResult> {
    let output = format!(
        "[shadow] Would execute: {}\n  Type: {:?}\n  Failure policy: {:?}{}",
        resolved_action,
        step.step_type,
        step.on_failure,
        if step.breakpoint {
            format!(
                "\n  ⏸ Breakpoint: {}",
                step.breakpoint_message.as_deref().unwrap_or("(none)")
            )
        } else {
            String::new()
        }
    );

    Ok(StepResult {
        step_name: step.name.clone(),
        success: true,
        output,
        duration_ms: 0,
        cost: 0.0,
        model_used: None,
        error: None,
    })
}

/// Convert a Step to an Action for constitution checking.
fn step_to_action(step: &Step, resolved_action: &str) -> Action {
    let action_type = match step.step_type {
        StepType::Execute => ActionType::Execute,
        StepType::Code | StepType::Refactor | StepType::Fix => ActionType::Write,
        StepType::Analyze | StepType::Plan | StepType::Debug | StepType::Summarize => {
            ActionType::ModelInvoke
        }
        StepType::Search | StepType::Classify => ActionType::Read,
        StepType::SecurityCheck => ActionType::Read,
        StepType::Other => ActionType::Other,
    };

    Action {
        id: Uuid::new_v4(),
        action_type,
        description: format!("Workflow step: {}", step.name),
        command: resolved_action.to_string(),
        working_dir: None,
        created_at: Utc::now(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::audit::AuditStore;
    use crate::constitution::Constitution;
    use crate::workflow::parser::parse_workflow_str;

    fn test_constitution() -> Constitution {
        let toml = r#"
[identity]
version = "1.0.0"

[boundaries]
forbidden = ["rm -rf /"]
requires_approval = []
auto_allowed = ["echo", "cat", "curl", "docker", "workflow step"]

[resource_limits]
max_api_cost_per_run = 10.0
max_api_cost_per_day = 50.0
max_execution_time = 3600
max_concurrent_workflows = 5
max_file_write_size = "10MB"
allowed_directories = ["/tmp", "/home"]
blocked_directories = ["/etc", "/sys"]

[model_permissions.thinking_model]
can_execute = false
can_read = true

[model_permissions.coding_model]
can_execute = true
can_read = true

[model_permissions.task_model]
can_execute = true
can_read = true
"#;
        Constitution::from_toml(toml).unwrap()
    }

    #[tokio::test]
    async fn test_shadow_execution() {
        let yaml = r#"
id: test-shadow
name: Shadow Test
description: Test shadow mode
variables:
  name: world
steps:
  - name: greet
    step_type: execute
    action: "echo Hello {{name}}"
  - name: deploy
    step_type: execute
    action: "docker compose up -d"
    breakpoint: true
    breakpoint_message: "Deploy now?"
"#;
        let wf = parse_workflow_str(yaml).unwrap();
        let constitution = Arc::new(test_constitution());
        let audit = Arc::new(AuditStore::new_memory());
        let runner = WorkflowRunner::new(constitution, audit);

        let result = runner
            .execute(&wf, HashMap::new(), true)
            .await
            .unwrap();

        assert!(result.shadow);
        assert!(result.success);
        assert_eq!(result.step_results.len(), 2);
        assert!(result.step_results[0].output.contains("[shadow]"));
        assert!(result.step_results[1].output.contains("Breakpoint"));
    }

    #[tokio::test]
    async fn test_real_execution() {
        let yaml = r#"
id: test-real
name: Real Test
description: Test real execution
variables: {}
steps:
  - name: echo-test
    step_type: execute
    action: "echo hello-from-runner"
"#;
        let wf = parse_workflow_str(yaml).unwrap();
        let constitution = Arc::new(test_constitution());
        let audit = Arc::new(AuditStore::new_memory());
        let runner = WorkflowRunner::new(constitution, audit);

        let result = runner
            .execute(&wf, HashMap::new(), false)
            .await
            .unwrap();

        assert!(!result.shadow);
        assert!(result.success);
        assert!(result.step_results[0]
            .output
            .contains("hello-from-runner"));
    }
}