o7 0.1.1

#![allow(non_snake_case)]

//! O7 Execution Engine — Core implementation.
//!
//! Executes parsed O7 workflow ASTs sequentially, dispatching exec blocks
//! to an injected harness function. Collects ExecutionEvents during the run.

use std::collections::HashMap;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

use crate::harness::check::{parse_check_result, parse_match_result};
use crate::harness::types::{CheckOutput, MatchOutput};
use crate::parser::ast::*;

use super::types::*;

/// Generate a unique run ID.
fn generate_run_id() -> RunId {
    format!(
        "run-{}-{}",
        chrono::Utc::now().timestamp_millis(),
        &uuid::Uuid::new_v4().to_string()[..8]
    )
}

fn count_exec_step_starts(events: &[ExecutionEvent]) -> usize {
    events
        .iter()
        .filter(|event| {
            matches!(
                event,
                ExecutionEvent::StepStarted { stepPath, .. }
                    if stepPath.last().is_some_and(|segment| segment.starts_with("exec:"))
            )
        })
        .count()
}

/// Internal error type for engine control flow.
#[derive(Debug)]
#[allow(dead_code)]
enum EngineError {
    /// A step failed (stop-on-failure semantics).
    StepFailure {
        message: String,
        step_path: StepPath,
    },
    /// The engine was paused by an external request.
    Paused { step_path: StepPath },
    /// A branch was cancelled (fail-fast in par-and).
    BranchCancelled,
}

// ========================================================
// Executor trait — shared execution logic for engine and branch contexts
// ========================================================

/// Common workflow execution behavior shared between `ExecutionEngine` (top-level)
/// and `BranchExecutionContext` (parallel branch isolation).
///
/// Required methods provide access to shared state and hook points for behavioral
/// differences. Default methods implement the core execute_* logic that both
/// contexts share.
trait Executor {
    // --- Required accessors ---
    fn workflow_map(&self) -> &HashMap<String, WorkflowDecl>;
    fn harness_dispatch(&self) -> &HarnessDispatchFn;
    fn on_event_ref(&self) -> Option<&OnEventCallback>;
    fn last_exec_stdout_ref(&self) -> &str;
    fn set_last_exec_stdout(&mut self, s: String);
    fn allocate_exec_ordinal(&self) -> usize;

    // --- Hooks for behavioral differences (no-ops by default for branch context) ---

    /// Called before a step starts. Engine emits SafeBoundary + tracks InProgress.
    fn before_step(
        &mut self,
        _run_id: &str,
        _step_path: &StepPath,
        _events: &mut Vec<ExecutionEvent>,
    ) {
    }

    /// Called after a step completes successfully. Engine tracks Completed + emits SafeBoundary.
    fn after_step_complete(
        &mut self,
        _run_id: &str,
        _step_path: &StepPath,
        _events: &mut Vec<ExecutionEvent>,
    ) {
    }

    /// Called on step failure. Engine tracks Failed.
    fn on_step_fail(&mut self, _step_path: &StepPath) {}

    /// Called before each statement in workflow/conditional/loop. Engine checks pause.
    fn check_pause_hook<'a>(
        &'a mut self,
        _run_id: &'a str,
        _call_stack: &'a StepPath,
        _events: &'a mut Vec<ExecutionEvent>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), EngineError>> + 'a>> {
        Box::pin(async { Ok(()) })
    }

    /// Called around conditional body. Engine emits SafeBoundary.
    fn emit_safe_boundary_hook(
        &mut self,
        _run_id: &str,
        _boundary_type: &str,
        _step_path: &StepPath,
        _events: &mut Vec<ExecutionEvent>,
    ) {
    }

    /// Handle a par-and statement. Returns None if not supported (branch context),
    /// which the default execute_statement maps to an error.
    fn handle_par_and<'a>(
        &'a mut self,
        _run_id: &'a str,
        _stmt: &'a ParAndStatement,
        _call_stack: &'a StepPath,
        _events: &'a mut Vec<ExecutionEvent>,
    ) -> Option<Pin<Box<dyn std::future::Future<Output = Result<(), EngineError>> + 'a>>> {
        None
    }

    // --- Default implementations (shared logic) ---

    /// Push event to the events vec and fire the on_event callback.
    fn emit(&self, event: ExecutionEvent, events: &mut Vec<ExecutionEvent>) {
        events.push(event.clone());
        if let Some(callback) = self.on_event_ref() {
            callback(&event);
        }
    }

    /// Emit StepFailed event and return the EngineError.
    fn emit_step_failure(
        &self,
        run_id: &str,
        step_path: StepPath,
        error: String,
        events: &mut Vec<ExecutionEvent>,
    ) -> EngineError {
        self.emit(
            ExecutionEvent::StepFailed {
                runId: run_id.to_string(),
                stepPath: step_path.clone(),
                error: error.clone(),
            },
            events,
        );
        EngineError::StepFailure {
            message: error,
            step_path,
        }
    }

    /// Record a step failure: hook on_step_fail + emit StepFailed + return EngineError.
    fn fail_step(
        &mut self,
        run_id: &str,
        step_path: StepPath,
        error: String,
        events: &mut Vec<ExecutionEvent>,
    ) -> EngineError {
        self.on_step_fail(&step_path);
        self.emit(
            ExecutionEvent::StepFailed {
                runId: run_id.to_string(),
                stepPath: step_path.clone(),
                error: error.clone(),
            },
            events,
        );
        EngineError::StepFailure {
            message: error,
            step_path,
        }
    }

    /// Execute a workflow's body statements sequentially, starting from `start_index`.
    fn execute_workflow<'a>(
        &'a mut self,
        run_id: &'a str,
        wf: &'a WorkflowDecl,
        call_stack: &'a StepPath,
        events: &'a mut Vec<ExecutionEvent>,
        start_index: usize,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), EngineError>> + 'a>> {
        Box::pin(async move {
            for i in start_index..wf.body.len() {
                self.check_pause_hook(run_id, call_stack, events).await?;
                self.execute_statement(run_id, &wf.body[i], call_stack, events)
                    .await?;
            }
            Ok(())
        })
    }

    /// Dispatch a single statement.
    fn execute_statement<'a>(
        &'a mut self,
        run_id: &'a str,
        stmt: &'a Statement,
        call_stack: &'a StepPath,
        events: &'a mut Vec<ExecutionEvent>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), EngineError>> + 'a>> {
        Box::pin(async move {
            match stmt {
                Statement::Run(s) => {
                    self.execute_run(run_id, &s.workflow_name, call_stack, events)
                        .await
                }
                Statement::Exec(e) => self.execute_exec(run_id, e, call_stack, events).await,
                Statement::If(s) => {
                    self.execute_conditional(
                        run_id,
                        &s.check_name,
                        &s.body,
                        false,
                        call_stack,
                        events,
                    )
                    .await
                }
                Statement::IfNot(s) => {
                    self.execute_conditional(
                        run_id,
                        &s.check_name,
                        &s.body,
                        true,
                        call_stack,
                        events,
                    )
                    .await
                }
                Statement::While(s) => {
                    self.execute_loop(run_id, &s.check_name, &s.body, false, call_stack, events)
                        .await
                }
                Statement::WhileNot(s) => {
                    self.execute_loop(run_id, &s.check_name, &s.body, true, call_stack, events)
                        .await
                }
                Statement::ParAnd(s) => {
                    if let Some(fut) = self.handle_par_and(run_id, s, call_stack, events) {
                        fut.await
                    } else {
                        Err(EngineError::StepFailure {
                            message: "Nested par-and is not supported inside parallel branches"
                                .to_string(),
                            step_path: call_stack.clone(),
                        })
                    }
                }
                Statement::Match(s) => self.execute_match(run_id, s, call_stack, events).await,
            }
        })
    }

    /// Execute a `run <workflow>` statement.
    fn execute_run<'a>(
        &'a mut self,
        run_id: &'a str,
        workflow_name: &'a str,
        call_stack: &'a StepPath,
        events: &'a mut Vec<ExecutionEvent>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), EngineError>> + 'a>> {
        Box::pin(async move {
            let mut step_path = call_stack.clone();
            step_path.push(workflow_name.to_string());

            self.before_step(run_id, &step_path, events);
            self.emit(
                ExecutionEvent::StepStarted {
                    runId: run_id.to_string(),
                    stepPath: step_path.clone(),
                },
                events,
            );

            // Look up target workflow
            let target_wf = self.workflow_map().get(workflow_name).cloned();
            if target_wf.is_none() {
                let error = format!("Unknown workflow reference: {}", workflow_name);
                return Err(self.fail_step(run_id, step_path, error, events));
            }

            // Check for circular references
            if call_stack.contains(&workflow_name.to_string()) {
                let mut chain = call_stack.clone();
                chain.push(workflow_name.to_string());
                let error = format!("Circular reference detected: {}", chain.join(" -> "));
                return Err(self.fail_step(run_id, step_path, error, events));
            }

            let target_wf = target_wf.unwrap();
            match self
                .execute_workflow(run_id, &target_wf, &step_path, events, 0)
                .await
            {
                Ok(()) => {
                    self.emit(
                        ExecutionEvent::StepCompleted {
                            runId: run_id.to_string(),
                            stepPath: step_path.clone(),
                        },
                        events,
                    );
                    self.after_step_complete(run_id, &step_path, events);
                    Ok(())
                }
                Err(EngineError::StepFailure { message, .. }) => {
                    Err(self.fail_step(run_id, step_path, message, events))
                }
                Err(other) => Err(other),
            }
        })
    }

    /// Execute an `exec` block by dispatching to the harness.
    fn execute_exec<'a>(
        &'a mut self,
        run_id: &'a str,
        exec_block: &'a ExecBlock,
        call_stack: &'a StepPath,
        events: &'a mut Vec<ExecutionEvent>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), EngineError>> + 'a>> {
        Box::pin(async move {
            let mut step_path = call_stack.clone();
            step_path.push(format!("exec:{}", exec_block.harness));

            self.before_step(run_id, &step_path, events);
            self.emit(
                ExecutionEvent::StepStarted {
                    runId: run_id.to_string(),
                    stepPath: step_path.clone(),
                },
                events,
            );

            let dispatch_result = (self.harness_dispatch())(
                exec_block,
                HarnessExecContext {
                    run_id: run_id.to_string(),
                    step_path: step_path.clone(),
                    exec_ordinal: self.allocate_exec_ordinal(),
                },
            )
            .await;

            match dispatch_result {
                Ok(result) => {
                    self.set_last_exec_stdout(result.stdout.clone());
                    if result.exit_code != 0 {
                        let error = format!(
                            "Exec '{}' failed with exit code {}",
                            exec_block.harness, result.exit_code
                        );
                        return Err(self.fail_step(run_id, step_path, error, events));
                    }
                    self.emit(
                        ExecutionEvent::StepCompleted {
                            runId: run_id.to_string(),
                            stepPath: step_path.clone(),
                        },
                        events,
                    );
                    self.after_step_complete(run_id, &step_path, events);
                    Ok(())
                }
                Err(err_msg) => Err(self.fail_step(run_id, step_path, err_msg, events)),
            }
        })
    }

    /// Execute a check workflow and parse its result.
    fn evaluate_check<'a>(
        &'a mut self,
        run_id: &'a str,
        check_name: &'a str,
        call_stack: &'a StepPath,
        events: &'a mut Vec<ExecutionEvent>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<CheckOutput, EngineError>> + 'a>> {
        Box::pin(async move {
            let check_wf = self.workflow_map().get(check_name).cloned();
            if check_wf.is_none() {
                let error = format!("Unknown check workflow: {}", check_name);
                let mut step_path = call_stack.clone();
                step_path.push(check_name.to_string());
                return Err(self.emit_step_failure(run_id, step_path, error, events));
            }

            let check_wf = check_wf.unwrap();
            self.set_last_exec_stdout(String::new());
            let mut check_call_stack = call_stack.clone();
            check_call_stack.push(check_name.to_string());
            self.execute_workflow(run_id, &check_wf, &check_call_stack, events, 0)
                .await?;

            match parse_check_result(self.last_exec_stdout_ref()) {
                Ok(check_output) => {
                    self.emit(
                        ExecutionEvent::CheckEvaluated {
                            runId: run_id.to_string(),
                            checkName: check_name.to_string(),
                            result: check_output.result,
                            reason: check_output.reason.clone(),
                        },
                        events,
                    );
                    Ok(check_output)
                }
                Err(parse_err) => {
                    let error = format!(
                        "Check '{}' returned invalid output: {}",
                        check_name, parse_err
                    );
                    let mut step_path = call_stack.clone();
                    step_path.push(check_name.to_string());
                    Err(self.emit_step_failure(run_id, step_path, error, events))
                }
            }
        })
    }

    /// Execute a match check workflow and parse its variant result.
    fn evaluate_match_check<'a>(
        &'a mut self,
        run_id: &'a str,
        check_name: &'a str,
        call_stack: &'a StepPath,
        events: &'a mut Vec<ExecutionEvent>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<MatchOutput, EngineError>> + 'a>> {
        Box::pin(async move {
            let check_wf = self.workflow_map().get(check_name).cloned();
            if check_wf.is_none() {
                let error = format!("Unknown check workflow: {}", check_name);
                let mut step_path = call_stack.clone();
                step_path.push(check_name.to_string());
                return Err(self.emit_step_failure(run_id, step_path, error, events));
            }

            let check_wf = check_wf.unwrap();
            self.set_last_exec_stdout(String::new());
            let mut check_call_stack = call_stack.clone();
            check_call_stack.push(check_name.to_string());
            self.execute_workflow(run_id, &check_wf, &check_call_stack, events, 0)
                .await?;

            match parse_match_result(self.last_exec_stdout_ref()) {
                Ok(match_output) => Ok(match_output),
                Err(parse_err) => {
                    let error = format!(
                        "Match check '{}' returned invalid output: {}",
                        check_name, parse_err
                    );
                    let mut step_path = call_stack.clone();
                    step_path.push(check_name.to_string());
                    Err(self.emit_step_failure(run_id, step_path, error, events))
                }
            }
        })
    }

    /// Execute a `match` statement.
    fn execute_match<'a>(
        &'a mut self,
        run_id: &'a str,
        stmt: &'a MatchStatement,
        call_stack: &'a StepPath,
        events: &'a mut Vec<ExecutionEvent>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), EngineError>> + 'a>> {
        Box::pin(async move {
            let match_output = self
                .evaluate_match_check(run_id, &stmt.check_name, call_stack, events)
                .await?;

            // Find matching arm
            let matched_arm = stmt
                .arms
                .iter()
                .enumerate()
                .find(|(_, arm)| arm.variant == match_output.variant);

            let (body, arm_index): (&[Statement], Option<i64>) = if let Some((idx, arm)) =
                matched_arm
            {
                (&arm.body, Some(idx as i64))
            } else if let Some(ref else_body) = stmt.else_body {
                (else_body.as_slice(), None)
            } else {
                // No matching arm and no else — error
                let arm_names: Vec<&str> = stmt.arms.iter().map(|a| a.variant.as_str()).collect();
                let error = format!(
                    "match on '{}' returned unrecognized variant '{}'; expected one of: {} (add an 'else' arm to handle unexpected variants)",
                    stmt.check_name,
                    match_output.variant,
                    arm_names.join(", ")
                );
                let mut step_path = call_stack.clone();
                step_path.push(stmt.check_name.clone());
                return Err(self.emit_step_failure(run_id, step_path, error, events));
            };

            // Emit MatchEvaluated event
            self.emit(
                ExecutionEvent::MatchEvaluated {
                    runId: run_id.to_string(),
                    checkName: stmt.check_name.clone(),
                    variant: match_output.variant.clone(),
                    reason: match_output.reason.clone(),
                    armIndex: arm_index,
                },
                events,
            );

            // Emit before-match-arm safe boundary
            self.emit_safe_boundary_hook(
                run_id,
                safe_boundary_types::BEFORE_MATCH_ARM,
                call_stack,
                events,
            );

            // Execute each statement in the selected body
            for stmt_item in body {
                self.check_pause_hook(run_id, call_stack, events).await?;
                self.execute_statement(run_id, stmt_item, call_stack, events)
                    .await?;
            }

            // Emit after-match-arm safe boundary
            self.emit_safe_boundary_hook(
                run_id,
                safe_boundary_types::AFTER_MATCH_ARM,
                call_stack,
                events,
            );

            Ok(())
        })
    }

    /// Execute an if/if-not conditional.
    fn execute_conditional<'a>(
        &'a mut self,
        run_id: &'a str,
        check_name: &'a str,
        body: &'a [Statement],
        negated: bool,
        call_stack: &'a StepPath,
        events: &'a mut Vec<ExecutionEvent>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), EngineError>> + 'a>> {
        Box::pin(async move {
            let check_output = self
                .evaluate_check(run_id, check_name, call_stack, events)
                .await?;

            let should_execute = if negated {
                !check_output.result
            } else {
                check_output.result
            };

            if should_execute {
                self.emit_safe_boundary_hook(
                    run_id,
                    safe_boundary_types::BEFORE_CONDITIONAL_BODY,
                    call_stack,
                    events,
                );
                for stmt in body {
                    self.check_pause_hook(run_id, call_stack, events).await?;
                    self.execute_statement(run_id, stmt, call_stack, events)
                        .await?;
                }
                self.emit_safe_boundary_hook(
                    run_id,
                    safe_boundary_types::AFTER_CONDITIONAL_BODY,
                    call_stack,
                    events,
                );
            }

            Ok(())
        })
    }

    /// Execute a while/while-not loop.
    fn execute_loop<'a>(
        &'a mut self,
        run_id: &'a str,
        check_name: &'a str,
        body: &'a [Statement],
        negated: bool,
        call_stack: &'a StepPath,
        events: &'a mut Vec<ExecutionEvent>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), EngineError>> + 'a>> {
        Box::pin(async move {
            loop {
                let check_output = self
                    .evaluate_check(run_id, check_name, call_stack, events)
                    .await?;

                let should_continue = if negated {
                    !check_output.result
                } else {
                    check_output.result
                };

                if !should_continue {
                    break;
                }

                self.emit_safe_boundary_hook(
                    run_id,
                    safe_boundary_types::BEFORE_LOOP_ITERATION,
                    call_stack,
                    events,
                );
                for stmt in body {
                    self.check_pause_hook(run_id, call_stack, events).await?;
                    self.execute_statement(run_id, stmt, call_stack, events)
                        .await?;
                }
                self.emit_safe_boundary_hook(
                    run_id,
                    safe_boundary_types::AFTER_LOOP_ITERATION,
                    call_stack,
                    events,
                );
            }

            Ok(())
        })
    }
}

/// The core O7 execution engine.
///
/// Takes parsed workflows and a harness dispatch function, then executes
/// statements sequentially with stop-on-failure semantics. Supports
/// pause/resume/reset, safe boundaries, and parallel (par-and) execution.
pub struct ExecutionEngine {
    workflow_map: HashMap<String, WorkflowDecl>,
    harness_dispatch: HarnessDispatchFn,
    on_event: Option<OnEventCallback>,
    on_save: Option<OnSaveCallback>,
    /// Captured stdout from the most recent exec block (used by check evaluation).
    last_exec_stdout: String,
    /// Pause flag — set by pause_run, checked between steps.
    pub(crate) pause_requested: Arc<AtomicBool>,
    /// Notify signal for waking the engine after a TUI pause.
    pub pause_notify: Arc<tokio::sync::Notify>,
    /// When true, check_pause blocks in-place instead of unwinding.
    /// Set to true for TUI mode; false (default) for CLI mode.
    pub suspend_on_pause: bool,
    /// The current run's state tracking object.
    run_state: Option<RunState>,
    /// Branch outputs from the most recent par-and.
    branch_outputs: Vec<BranchOutput>,
    exec_ordinal_ref: Arc<AtomicUsize>,
}

impl ExecutionEngine {
    /// Create a new ExecutionEngine.
    pub fn new(
        workflows: Vec<WorkflowDecl>,
        harness_dispatch: HarnessDispatchFn,
        on_event: Option<OnEventCallback>,
        on_save: Option<OnSaveCallback>,
    ) -> Self {
        let mut workflow_map = HashMap::new();
        for wf in workflows {
            workflow_map.insert(wf.name.clone(), wf);
        }
        ExecutionEngine {
            workflow_map,
            harness_dispatch,
            on_event,
            on_save,
            last_exec_stdout: String::new(),
            pause_requested: Arc::new(AtomicBool::new(false)),
            pause_notify: Arc::new(tokio::sync::Notify::new()),
            suspend_on_pause: false,
            run_state: None,
            branch_outputs: Vec::new(),
            exec_ordinal_ref: Arc::new(AtomicUsize::new(0)),
        }
    }

    /// Start executing a workflow by name.
    /// Returns the run ID and all collected events.
    pub async fn start_run(&mut self, root_workflow: &str) -> (RunId, Vec<ExecutionEvent>) {
        let run_id = generate_run_id();
        let mut events: Vec<ExecutionEvent> = Vec::new();

        self.pause_requested.store(false, Ordering::SeqCst);
        self.exec_ordinal_ref.store(0, Ordering::SeqCst);
        self.run_state = Some(RunState {
            runId: run_id.clone(),
            rootWorkflow: root_workflow.to_string(),
            status: RunStatus::Running,
            steps: Vec::new(),
            events: Vec::new(),
            safeBoundaries: Vec::new(),
            lastSafeBoundaryIndex: -1,
        });

        let root_wf = self.workflow_map.get(root_workflow).cloned();
        if root_wf.is_none() {
            self.emit(
                ExecutionEvent::RunFailed {
                    runId: run_id.clone(),
                    position: vec![root_workflow.to_string()],
                    error: format!("Unknown workflow: {}", root_workflow),
                },
                &mut events,
            );
            if let Some(ref mut state) = self.run_state {
                state.status = RunStatus::Failed;
                state
                    .events
                    .extend_from_slice(&events[state.events.len()..]);
            }
            self.save_state();
            return (run_id, events);
        }

        let root_wf = root_wf.unwrap();
        let call_stack = vec![root_workflow.to_string()];

        let result = self
            .execute_workflow(&run_id.clone(), &root_wf, &call_stack, &mut events, 0)
            .await;
        self.finalize_run(&run_id, result, &mut events);

        (run_id, events)
    }

    /// Request pausing the current run. The engine will stop before the next step.
    pub fn pause_run(&self, _run_id: &str) {
        self.pause_requested.store(true, Ordering::SeqCst);
    }

    /// Resume a paused run from persisted state.
    /// Re-executes from the last safe boundary (ADR-21).
    pub async fn resume_run(&mut self, state: RunState) -> (RunId, Vec<ExecutionEvent>) {
        let run_id = state.runId.clone();
        let mut events: Vec<ExecutionEvent> = state.events.clone();

        self.pause_requested.store(false, Ordering::SeqCst);
        self.exec_ordinal_ref
            .store(count_exec_step_starts(&state.events), Ordering::SeqCst);
        self.run_state = Some(RunState {
            runId: run_id.clone(),
            rootWorkflow: state.rootWorkflow.clone(),
            status: RunStatus::Running,
            steps: state.steps.clone(),
            events: Vec::new(), // will be populated via emit
            safeBoundaries: state.safeBoundaries.clone(),
            lastSafeBoundaryIndex: state.lastSafeBoundaryIndex,
        });

        let root_wf = self.workflow_map.get(&state.rootWorkflow).cloned();
        if root_wf.is_none() {
            self.emit(
                ExecutionEvent::RunFailed {
                    runId: run_id.clone(),
                    position: vec![state.rootWorkflow.clone()],
                    error: format!("Unknown workflow: {}", state.rootWorkflow),
                },
                &mut events,
            );
            if let Some(ref mut st) = self.run_state {
                st.status = RunStatus::Failed;
                st.events.extend_from_slice(&events[st.events.len()..]);
            }
            self.save_state();
            return (run_id, events);
        }

        let root_wf = root_wf.unwrap();
        let resume_from_index = self.find_resume_step_index(&state);
        let call_stack = vec![state.rootWorkflow.clone()];

        let result = self
            .execute_workflow(
                &run_id.clone(),
                &root_wf,
                &call_stack,
                &mut events,
                resume_from_index,
            )
            .await;
        self.finalize_run(&run_id, result, &mut events);

        (run_id, events)
    }

    /// Reset a run to a target safe boundary.
    /// Returns a new RunState that can be resumed.
    pub fn reset_run(&mut self, run_id: &str, target_boundary: usize) -> Result<RunState, String> {
        let state = self
            .run_state
            .as_ref()
            .ok_or_else(|| format!("No run state found for runId: {}", run_id))?;

        if state.runId != run_id {
            return Err(format!("No run state found for runId: {}", run_id));
        }

        // Validate the target boundary
        if !state.safeBoundaries.contains(&target_boundary) {
            return Err(format!(
                "Invalid reset target: {} is not a valid safe boundary index. Valid boundaries: {:?}",
                target_boundary, state.safeBoundaries
            ));
        }

        // Truncate events to the target boundary (inclusive)
        let truncated_events: Vec<ExecutionEvent> = state.events[..=target_boundary].to_vec();

        // Truncate safe boundaries
        let truncated_boundaries: Vec<usize> = state
            .safeBoundaries
            .iter()
            .filter(|&&b| b <= target_boundary)
            .cloned()
            .collect();

        // Truncate steps — keep only steps whose paths appear in the truncated events
        let truncated_steps = Self::truncate_steps_to_events(&state.steps, &truncated_events);

        let new_state = RunState {
            runId: run_id.to_string(),
            rootWorkflow: state.rootWorkflow.clone(),
            status: RunStatus::Paused,
            steps: truncated_steps,
            events: truncated_events,
            safeBoundaries: truncated_boundaries,
            lastSafeBoundaryIndex: target_boundary as i64,
        };

        self.run_state = Some(new_state.clone());
        Ok(new_state)
    }

    /// Get the current RunState (for inspection/testing).
    pub fn get_run_state(&self) -> Option<&RunState> {
        self.run_state.as_ref()
    }

    /// Get the branch outputs from the most recent par-and execution.
    pub fn get_branch_outputs(&self) -> &[BranchOutput] {
        &self.branch_outputs
    }

    /// Execute a `par-and` statement — launch branches sequentially (v1),
    /// collect results, then run the join workflow.
    ///
    /// Note: In v1, branches execute sequentially rather than truly in parallel.
    /// This is consistent with the TS implementation being single-threaded (JavaScript).
    /// Fail-fast is checked between branches.
    async fn execute_par_and(
        &mut self,
        run_id: &str,
        stmt: &ParAndStatement,
        call_stack: &StepPath,
        events: &mut Vec<ExecutionEvent>,
    ) -> Result<(), EngineError> {
        let fail_policy = stmt.fail_policy.clone().unwrap_or(FailPolicy::FailFast);

        let mut any_failed = false;
        let mut first_error = String::new();
        let mut first_error_path: StepPath = Vec::new();
        let mut branch_results: Vec<Option<BranchOutput>> = vec![None; stmt.branches.len()];

        for (branch_index, branch) in stmt.branches.iter().enumerate() {
            let branch_path: StepPath = {
                let mut p = call_stack.clone();
                p.push(format!("par-and:{}", branch.workflow_name));
                p
            };

            // Check if cancelled before starting (fail-fast)
            if any_failed && fail_policy == FailPolicy::FailFast {
                // Skip remaining branches
                continue;
            }

            self.emit(
                ExecutionEvent::BranchStarted {
                    runId: run_id.to_string(),
                    branchPath: branch_path.clone(),
                },
                events,
            );

            // Emit SafeBoundary after branch transition
            self.emit_safe_boundary(
                run_id,
                safe_boundary_types::AFTER_BRANCH_TRANSITION,
                &branch_path,
                events,
            );

            let branch_wf = self.workflow_map.get(&branch.workflow_name).cloned();
            if branch_wf.is_none() {
                let error = format!("Unknown workflow reference: {}", branch.workflow_name);
                self.emit(
                    ExecutionEvent::BranchFailed {
                        runId: run_id.to_string(),
                        branchPath: branch_path.clone(),
                        error: error.clone(),
                    },
                    events,
                );
                any_failed = true;
                if first_error.is_empty() {
                    first_error = error;
                    first_error_path = branch_path;
                }
                continue;
            }

            let branch_wf = branch_wf.unwrap();

            // Check for nested par-and (not supported inside branches)
            if Self::workflow_contains_par_and(&branch_wf) {
                let error = "Nested par-and is not supported inside parallel branches".to_string();
                self.emit(
                    ExecutionEvent::BranchFailed {
                        runId: run_id.to_string(),
                        branchPath: branch_path.clone(),
                        error: error.clone(),
                    },
                    events,
                );
                any_failed = true;
                if first_error.is_empty() {
                    first_error = error;
                    first_error_path = branch_path;
                }
                continue;
            }

            // Execute branch using a BranchExecutionContext for isolated state
            let mut branch_ctx = BranchExecutionContext::new(
                &self.workflow_map,
                self.harness_dispatch.clone(),
                self.on_event.as_ref(),
                self.exec_ordinal_ref.clone(),
            );

            match branch_ctx
                .execute_workflow(run_id, &branch_wf, &branch_path, events, 0)
                .await
            {
                Ok(()) => {
                    let output_str = branch_ctx.get_last_exec_stdout();
                    let output_val = if output_str.is_empty() {
                        None
                    } else {
                        serde_json::from_str::<serde_json::Value>(&output_str).ok()
                    };
                    self.emit(
                        ExecutionEvent::BranchCompleted {
                            runId: run_id.to_string(),
                            branchPath: branch_path.clone(),
                            output: output_val,
                        },
                        events,
                    );
                    branch_results[branch_index] = Some(BranchOutput {
                        workflow: branch.workflow_name.clone(),
                        output: if output_str.is_empty() {
                            None
                        } else {
                            Some(output_str)
                        },
                    });
                }
                Err(EngineError::StepFailure { message, .. }) => {
                    self.emit(
                        ExecutionEvent::BranchFailed {
                            runId: run_id.to_string(),
                            branchPath: branch_path.clone(),
                            error: message.clone(),
                        },
                        events,
                    );
                    any_failed = true;
                    if first_error.is_empty() {
                        first_error = message;
                        first_error_path = branch_path;
                    }
                }
                Err(EngineError::BranchCancelled) => {
                    // Branch was cancelled
                    continue;
                }
                Err(EngineError::Paused { step_path }) => {
                    return Err(EngineError::Paused { step_path });
                }
            }
        }

        // Build ordered branch_outputs (only successful branches)
        let branch_outputs: Vec<BranchOutput> = branch_results.into_iter().flatten().collect();

        if any_failed {
            let error = if first_error.is_empty() {
                "One or more branches failed".to_string()
            } else {
                first_error.clone()
            };
            let step_path = if first_error_path.is_empty() {
                call_stack.clone()
            } else {
                first_error_path
            };
            return Err(EngineError::StepFailure {
                message: error,
                step_path,
            });
        }

        // All branches succeeded — run the join workflow
        let join_wf_name = &stmt.join_workflow_name;
        let mut join_path = call_stack.clone();
        join_path.push(format!("join:{}", join_wf_name));

        // Emit SafeBoundary before join
        self.emit_safe_boundary(run_id, safe_boundary_types::BEFORE_JOIN, &join_path, events);

        self.emit(
            ExecutionEvent::JoinStarted {
                runId: run_id.to_string(),
                joinWorkflow: join_wf_name.clone(),
            },
            events,
        );

        let join_wf = self.workflow_map.get(join_wf_name).cloned();
        if join_wf.is_none() {
            let error = format!("Unknown join workflow: {}", join_wf_name);
            return Err(EngineError::StepFailure {
                message: error,
                step_path: join_path,
            });
        }

        // Store branch outputs so the join workflow can access them
        self.branch_outputs = branch_outputs;

        let join_wf = join_wf.unwrap();
        self.execute_workflow(run_id, &join_wf, &join_path, events, 0)
            .await?;

        // Emit SafeBoundary after join
        self.emit_safe_boundary(run_id, safe_boundary_types::AFTER_JOIN, &join_path, events);

        Ok(())
    }

    // ========================================================
    // Helper methods
    // ========================================================

    /// Finalize a run by emitting the appropriate terminal event, updating run state, and saving.
    fn finalize_run(
        &mut self,
        run_id: &RunId,
        result: Result<(), EngineError>,
        events: &mut Vec<ExecutionEvent>,
    ) {
        match result {
            Ok(()) => {
                self.emit(
                    ExecutionEvent::RunCompleted {
                        runId: run_id.clone(),
                    },
                    events,
                );
                if let Some(ref mut state) = self.run_state {
                    state.status = RunStatus::Completed;
                    state
                        .events
                        .extend_from_slice(&events[state.events.len()..]);
                }
                self.save_state();
            }
            Err(EngineError::Paused { step_path }) => {
                self.emit(
                    ExecutionEvent::RunPaused {
                        runId: run_id.clone(),
                        position: step_path,
                    },
                    events,
                );
                if let Some(ref mut state) = self.run_state {
                    state.status = RunStatus::Paused;
                    state
                        .events
                        .extend_from_slice(&events[state.events.len()..]);
                }
                self.save_state();
            }
            Err(EngineError::StepFailure { message, step_path }) => {
                self.emit(
                    ExecutionEvent::RunFailed {
                        runId: run_id.clone(),
                        position: step_path,
                        error: message,
                    },
                    events,
                );
                if let Some(ref mut state) = self.run_state {
                    state.status = RunStatus::Failed;
                    state
                        .events
                        .extend_from_slice(&events[state.events.len()..]);
                }
                self.save_state();
            }
            Err(EngineError::BranchCancelled) => {
                // Should not reach here at top level
            }
        }
    }

    /// Emit a safe boundary event and update run state tracking.
    fn emit_safe_boundary(
        &mut self,
        run_id: &str,
        boundary_type: &str,
        step_path: &StepPath,
        events: &mut Vec<ExecutionEvent>,
    ) {
        let event = ExecutionEvent::SafeBoundary {
            runId: run_id.to_string(),
            boundaryType: boundary_type.to_string(),
            stepPath: step_path.clone(),
        };
        self.emit(event, events);

        if let Some(ref mut state) = self.run_state {
            let idx = events.len() - 1;
            state.safeBoundaries.push(idx);
            state.lastSafeBoundaryIndex = idx as i64;
            // Sync only new events since last save (avoids O(n²) full clone)
            state
                .events
                .extend_from_slice(&events[state.events.len()..]);
        }
        self.save_state();
    }

    /// Update or insert a step's status in run_state.
    fn track_step(&mut self, step_path: &StepPath, status: StepStatus) {
        if let Some(ref mut state) = self.run_state {
            let existing = state.steps.iter_mut().find(|s| &s.stepPath == step_path);

            if let Some(entry) = existing {
                entry.status = status;
            } else {
                state.steps.push(StepState {
                    stepPath: step_path.clone(),
                    status,
                });
            }
        }
    }

    /// Check if pause has been requested.
    ///
    /// In `suspend_on_pause` mode (TUI): emits RunPaused, saves state, then blocks
    /// until the pause flag is cleared and the notify is signalled. This preserves
    /// the full call stack (including nested loops/conditionals) across pause/resume.
    ///
    /// In normal mode (CLI): returns Err(Paused) to unwind the call stack.
    async fn check_pause(
        &mut self,
        run_id: &str,
        call_stack: &StepPath,
        events: &mut Vec<ExecutionEvent>,
    ) -> Result<(), EngineError> {
        if !self.pause_requested.load(Ordering::SeqCst) {
            return Ok(());
        }

        if self.suspend_on_pause {
            // Emit RunPaused event and persist state for crash recovery
            self.emit(
                ExecutionEvent::RunPaused {
                    runId: run_id.to_string(),
                    position: call_stack.clone(),
                },
                events,
            );
            if let Some(ref mut state) = self.run_state {
                state.status = RunStatus::Paused;
                state
                    .events
                    .extend_from_slice(&events[state.events.len()..]);
            }
            self.save_state();

            // Block until resumed — the TUI clears pause_requested and signals notify
            let notify = self.pause_notify.clone();
            while self.pause_requested.load(Ordering::SeqCst) {
                notify.notified().await;
            }

            // Resumed — update state back to running
            if let Some(ref mut state) = self.run_state {
                state.status = RunStatus::Running;
            }
            Ok(())
        } else {
            // CLI mode: unwind the call stack
            Err(EngineError::Paused {
                step_path: call_stack.clone(),
            })
        }
    }

    /// Persist run state via the on_save callback if present.
    fn save_state(&self) {
        if let (Some(callback), Some(state)) = (&self.on_save, &self.run_state) {
            callback(state);
        }
    }

    // ========================================================
    // Resume helpers
    // ========================================================

    /// Find the step index to resume from based on the last safe boundary.
    fn find_resume_step_index(&self, state: &RunState) -> usize {
        if state.lastSafeBoundaryIndex < 0 {
            return 0;
        }

        let boundary_idx = state.lastSafeBoundaryIndex as usize;
        if boundary_idx >= state.events.len() {
            return 0;
        }

        // Look at the SafeBoundary event to determine position
        let boundary_event = &state.events[boundary_idx];
        let (boundary_type, boundary_step_path) = match boundary_event {
            ExecutionEvent::SafeBoundary {
                boundaryType,
                stepPath,
                ..
            } => (boundaryType.as_str(), stepPath),
            _ => return 0,
        };

        // Find which step index this corresponds to in the root workflow
        let root_wf = match self.workflow_map.get(&state.rootWorkflow) {
            Some(wf) => wf,
            None => return 0,
        };

        let step_name = if boundary_step_path.len() > 1 {
            &boundary_step_path[1]
        } else {
            return 0;
        };

        for (i, stmt) in root_wf.body.iter().enumerate() {
            match stmt {
                Statement::Run(s) if s.workflow_name == *step_name => {
                    if boundary_type == safe_boundary_types::BEFORE_STEP_START {
                        return i;
                    }
                    if boundary_type == safe_boundary_types::AFTER_STEP_COMPLETE {
                        return i + 1;
                    }
                }
                Statement::Exec(e) if format!("exec:{}", e.harness) == *step_name => {
                    if boundary_type == safe_boundary_types::BEFORE_STEP_START {
                        return i;
                    }
                    if boundary_type == safe_boundary_types::AFTER_STEP_COMPLETE {
                        return i + 1;
                    }
                }
                _ => {}
            }
        }

        0
    }

    /// Truncate steps to only those whose paths appear in the given events.
    fn truncate_steps_to_events(steps: &[StepState], events: &[ExecutionEvent]) -> Vec<StepState> {
        let valid_paths: std::collections::HashSet<&Vec<String>> = events
            .iter()
            .filter_map(|event| match event {
                ExecutionEvent::StepStarted { stepPath, .. }
                | ExecutionEvent::StepCompleted { stepPath, .. }
                | ExecutionEvent::StepFailed { stepPath, .. }
                | ExecutionEvent::SafeBoundary { stepPath, .. } => Some(stepPath),
                _ => None,
            })
            .collect();

        steps
            .iter()
            .filter(|s| valid_paths.contains(&s.stepPath))
            .cloned()
            .collect()
    }

    /// Check if a workflow (or any nested workflow) contains a par-and statement.
    fn workflow_contains_par_and(wf: &WorkflowDecl) -> bool {
        Self::stmts_contain_par_and(&wf.body)
    }

    /// Recursively check if any statements contain par-and.
    fn stmts_contain_par_and(stmts: &[Statement]) -> bool {
        for stmt in stmts {
            if matches!(stmt, Statement::ParAnd(_)) {
                return true;
            }
            if stmt
                .body()
                .is_some_and(|body| Self::stmts_contain_par_and(body))
            {
                return true;
            }
        }
        false
    }
}

impl Executor for ExecutionEngine {
    fn workflow_map(&self) -> &HashMap<String, WorkflowDecl> {
        &self.workflow_map
    }

    fn harness_dispatch(&self) -> &HarnessDispatchFn {
        &self.harness_dispatch
    }

    fn on_event_ref(&self) -> Option<&OnEventCallback> {
        self.on_event.as_ref()
    }

    fn last_exec_stdout_ref(&self) -> &str {
        &self.last_exec_stdout
    }

    fn set_last_exec_stdout(&mut self, s: String) {
        self.last_exec_stdout = s;
    }

    fn allocate_exec_ordinal(&self) -> usize {
        self.exec_ordinal_ref.fetch_add(1, Ordering::SeqCst)
    }

    fn before_step(
        &mut self,
        run_id: &str,
        step_path: &StepPath,
        events: &mut Vec<ExecutionEvent>,
    ) {
        self.emit_safe_boundary(
            run_id,
            safe_boundary_types::BEFORE_STEP_START,
            step_path,
            events,
        );
        self.track_step(step_path, StepStatus::InProgress);
    }

    fn after_step_complete(
        &mut self,
        run_id: &str,
        step_path: &StepPath,
        events: &mut Vec<ExecutionEvent>,
    ) {
        self.track_step(step_path, StepStatus::Completed);
        self.emit_safe_boundary(
            run_id,
            safe_boundary_types::AFTER_STEP_COMPLETE,
            step_path,
            events,
        );
    }

    fn on_step_fail(&mut self, step_path: &StepPath) {
        self.track_step(step_path, StepStatus::Failed);
    }

    fn check_pause_hook<'a>(
        &'a mut self,
        run_id: &'a str,
        call_stack: &'a StepPath,
        events: &'a mut Vec<ExecutionEvent>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), EngineError>> + 'a>> {
        Box::pin(async move { self.check_pause(run_id, call_stack, events).await })
    }

    fn emit_safe_boundary_hook(
        &mut self,
        run_id: &str,
        boundary_type: &str,
        step_path: &StepPath,
        events: &mut Vec<ExecutionEvent>,
    ) {
        self.emit_safe_boundary(run_id, boundary_type, step_path, events);
    }

    fn handle_par_and<'a>(
        &'a mut self,
        run_id: &'a str,
        stmt: &'a ParAndStatement,
        call_stack: &'a StepPath,
        events: &'a mut Vec<ExecutionEvent>,
    ) -> Option<Pin<Box<dyn std::future::Future<Output = Result<(), EngineError>> + 'a>>> {
        Some(Box::pin(async move {
            self.execute_par_and(run_id, stmt, call_stack, events).await
        }))
    }
}

// ========================================================
// BranchExecutionContext — isolated branch execution
// ========================================================

/// Branch-local execution context for parallel branches.
/// Shares workflow map and harness dispatch, but has its own stdout tracking.
/// Does NOT support nested par-and (uses default rejection from Executor trait).
struct BranchExecutionContext<'a> {
    workflow_map: &'a HashMap<String, WorkflowDecl>,
    harness_dispatch: HarnessDispatchFn,
    on_event: Option<&'a OnEventCallback>,
    last_exec_stdout: String,
    exec_ordinal_ref: Arc<AtomicUsize>,
}

impl<'a> BranchExecutionContext<'a> {
    fn new(
        workflow_map: &'a HashMap<String, WorkflowDecl>,
        harness_dispatch: HarnessDispatchFn,
        on_event: Option<&'a OnEventCallback>,
        exec_ordinal_ref: Arc<AtomicUsize>,
    ) -> Self {
        BranchExecutionContext {
            workflow_map,
            harness_dispatch,
            on_event,
            last_exec_stdout: String::new(),
            exec_ordinal_ref,
        }
    }

    fn get_last_exec_stdout(&self) -> String {
        self.last_exec_stdout.clone()
    }
}

impl<'a> Executor for BranchExecutionContext<'a> {
    fn workflow_map(&self) -> &HashMap<String, WorkflowDecl> {
        self.workflow_map
    }

    fn harness_dispatch(&self) -> &HarnessDispatchFn {
        &self.harness_dispatch
    }

    fn on_event_ref(&self) -> Option<&OnEventCallback> {
        self.on_event.map(|cb| cb as &OnEventCallback)
    }

    fn last_exec_stdout_ref(&self) -> &str {
        &self.last_exec_stdout
    }

    fn set_last_exec_stdout(&mut self, s: String) {
        self.last_exec_stdout = s;
    }

    fn allocate_exec_ordinal(&self) -> usize {
        self.exec_ordinal_ref.fetch_add(1, Ordering::SeqCst)
    }
    // All hooks use default no-op implementations — no tracking, no pause, no safe boundaries, no par-and.
}

// ========================================================
// Unit tests
// ========================================================

#[cfg(test)]
mod tests {
    use super::*;
    use crate::harness::types::ExecResult;

    fn make_echo_harness() -> HarnessDispatchFn {
        Arc::new(|exec_block: &ExecBlock, _ctx: HarnessExecContext| {
            let prompt = exec_block.prompt.clone().unwrap_or_default();
            Box::pin(async move {
                Ok(ExecResult {
                    exit_code: 0,
                    stdout: prompt,
                    stderr: String::new(),
                    harness_events: Vec::new(),
                })
            })
        })
    }

    fn make_failing_harness() -> HarnessDispatchFn {
        Arc::new(|exec_block: &ExecBlock, _ctx: HarnessExecContext| {
            let harness = exec_block.harness.clone();
            Box::pin(async move {
                Ok(ExecResult {
                    exit_code: 1,
                    stdout: String::new(),
                    stderr: format!("{} failed", harness),
                    harness_events: Vec::new(),
                })
            })
        })
    }

    fn parse_workflows(source: &str) -> Vec<WorkflowDecl> {
        let result = crate::parser::parse_file(source, "test.7");
        result.workflows().unwrap().clone()
    }

    #[tokio::test]
    async fn test_simple_exec() {
        let wfs = parse_workflows(
            "version 1\nworkflow main\n  exec\n    harness: echo\n    prompt: \"hello\"\n",
        );
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("main").await;
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunCompleted { .. }))
        );
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::StepStarted { .. }))
        );
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::StepCompleted { .. }))
        );
    }

    #[tokio::test]
    async fn test_simple_run() {
        let wfs = parse_workflows(
            "version 1\nworkflow greet\n  exec\n    harness: echo\n    prompt: \"hello\"\nworkflow main\n  run greet\n",
        );
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("main").await;
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunCompleted { .. }))
        );
    }

    #[tokio::test]
    async fn test_unknown_workflow() {
        let wfs = parse_workflows(
            "version 1\nworkflow main\n  exec\n    harness: echo\n    prompt: \"hi\"\n",
        );
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("nonexistent").await;
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunFailed { .. }))
        );
    }

    #[tokio::test]
    async fn test_unknown_workflow_reference() {
        let wfs = parse_workflows("version 1\nworkflow main\n  run nonexistent\n");
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("main").await;
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::StepFailed { .. }))
        );
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunFailed { .. }))
        );
    }

    #[tokio::test]
    async fn test_exec_failure() {
        let wfs = parse_workflows(
            "version 1\nworkflow main\n  exec\n    harness: fail\n    prompt: \"boom\"\n",
        );
        let mut engine = ExecutionEngine::new(wfs, make_failing_harness(), None, None);
        let (_, events) = engine.start_run("main").await;
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::StepFailed { .. }))
        );
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunFailed { .. }))
        );
    }

    #[tokio::test]
    async fn test_if_true_executes_body() {
        let source = r#"version 1
workflow check-true
  exec
    harness: echo
    prompt: "{\"result\": true, \"reason\": \"yes\"}"
workflow do-action
  exec
    harness: echo
    prompt: "action done"
workflow main
  if check-true
    run do-action
"#;
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("main").await;
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::CheckEvaluated { result: true, .. }))
        );
        // do-action should have been executed
        assert!(events.iter().any(|e| matches!(
            e,
            ExecutionEvent::StepStarted { stepPath, .. }
            if stepPath.last().map_or(false, |s| s.contains("do-action"))
        )));
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunCompleted { .. }))
        );
    }

    #[tokio::test]
    async fn test_if_false_skips_body() {
        let source = r#"version 1
workflow check-false
  exec
    harness: echo
    prompt: "{\"result\": false}"
workflow do-action
  exec
    harness: echo
    prompt: "action done"
workflow main
  if check-false
    run do-action
"#;
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("main").await;
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::CheckEvaluated { result: false, .. }))
        );
        // do-action should NOT have been executed
        assert!(!events.iter().any(|e| matches!(
            e,
            ExecutionEvent::StepStarted { stepPath, .. }
            if stepPath.last().map_or(false, |s| s.contains("do-action"))
        )));
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunCompleted { .. }))
        );
    }

    #[tokio::test]
    async fn test_if_not_true_skips_body() {
        let source = r#"version 1
workflow check-true
  exec
    harness: echo
    prompt: "{\"result\": true}"
workflow do-action
  exec
    harness: echo
    prompt: "action done"
workflow main
  if not check-true
    run do-action
"#;
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("main").await;
        // if-not true => should NOT execute body
        assert!(!events.iter().any(|e| matches!(
            e,
            ExecutionEvent::StepStarted { stepPath, .. }
            if stepPath.last().map_or(false, |s| s.contains("do-action"))
        )));
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunCompleted { .. }))
        );
    }

    #[tokio::test]
    async fn test_if_not_false_executes_body() {
        let source = r#"version 1
workflow check-false
  exec
    harness: echo
    prompt: "{\"result\": false}"
workflow do-action
  exec
    harness: echo
    prompt: "action done"
workflow main
  if not check-false
    run do-action
"#;
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("main").await;
        // if-not false => should execute body
        assert!(events.iter().any(|e| matches!(
            e,
            ExecutionEvent::StepStarted { stepPath, .. }
            if stepPath.last().map_or(false, |s| s.contains("do-action"))
        )));
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunCompleted { .. }))
        );
    }

    #[tokio::test]
    async fn test_while_loop() {
        // Use a counter via harness that changes output each call
        use std::sync::atomic::AtomicUsize;
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_clone = counter.clone();

        let harness: HarnessDispatchFn =
            Arc::new(move |exec_block: &ExecBlock, _ctx: HarnessExecContext| {
                let prompt = exec_block.prompt.clone().unwrap_or_default();
                let counter = counter_clone.clone();
                Box::pin(async move {
                    if prompt.contains("check") {
                        let count = counter.fetch_add(1, Ordering::SeqCst);
                        // Return true for first 2 calls, false on 3rd
                        let result = count < 2;
                        Ok(ExecResult {
                            exit_code: 0,
                            stdout: format!("{{\"result\": {}}}", result),
                            stderr: String::new(),
                            harness_events: Vec::new(),
                        })
                    } else {
                        Ok(ExecResult {
                            exit_code: 0,
                            stdout: prompt,
                            stderr: String::new(),
                            harness_events: Vec::new(),
                        })
                    }
                })
            });

        let source = r#"version 1
workflow check-continue
  exec
    harness: echo
    prompt: "check"
workflow do-work
  exec
    harness: echo
    prompt: "working"
workflow main
  while check-continue
    run do-work
"#;
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, harness, None, None);
        let (_, events) = engine.start_run("main").await;

        // Should have 3 CheckEvaluated events (2 true, 1 false)
        let check_events: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, ExecutionEvent::CheckEvaluated { .. }))
            .collect();
        assert_eq!(check_events.len(), 3);

        // Should have 2 iterations of do-work
        let work_starts: Vec<_> = events
            .iter()
            .filter(|e| {
                matches!(
                    e,
                    ExecutionEvent::StepStarted { stepPath, .. }
                    if stepPath.last().map_or(false, |s| s.contains("do-work"))
                )
            })
            .collect();
        assert_eq!(work_starts.len(), 2);

        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunCompleted { .. }))
        );
    }

    #[tokio::test]
    async fn test_pause() {
        let source = "version 1\nworkflow a\n  exec\n    harness: echo\n    prompt: \"a\"\nworkflow b\n  exec\n    harness: echo\n    prompt: \"b\"\nworkflow main\n  run a\n  run b\n";
        let wfs = parse_workflows(source);

        // Create a harness that triggers pause after the first exec
        let pause_flag = Arc::new(AtomicBool::new(false));
        let pause_flag_clone = pause_flag.clone();
        let call_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let call_count_clone = call_count.clone();

        let harness: HarnessDispatchFn =
            Arc::new(move |exec_block: &ExecBlock, _ctx: HarnessExecContext| {
                let prompt = exec_block.prompt.clone().unwrap_or_default();
                let pause_flag = pause_flag_clone.clone();
                let call_count = call_count_clone.clone();
                Box::pin(async move {
                    let count = call_count.fetch_add(1, Ordering::SeqCst);
                    // After first exec completes, request pause
                    if count == 0 {
                        pause_flag.store(true, Ordering::SeqCst);
                    }
                    Ok(ExecResult {
                        exit_code: 0,
                        stdout: prompt,
                        stderr: String::new(),
                        harness_events: Vec::new(),
                    })
                })
            });

        let mut engine = ExecutionEngine::new(wfs, harness, None, None);
        // Share the pause flag with the engine
        engine.pause_requested = pause_flag;
        let (_, events) = engine.start_run("main").await;
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunPaused { .. }))
        );
    }

    #[tokio::test]
    async fn test_safe_boundaries_emitted() {
        let source = "version 1\nworkflow greet\n  exec\n    harness: echo\n    prompt: \"hi\"\nworkflow main\n  run greet\n";
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("main").await;
        let boundary_count = events
            .iter()
            .filter(|e| matches!(e, ExecutionEvent::SafeBoundary { .. }))
            .count();
        assert!(
            boundary_count >= 2,
            "Expected at least 2 safe boundaries, got {}",
            boundary_count
        );
    }

    #[tokio::test]
    async fn test_run_state_tracking() {
        let source = "version 1\nworkflow greet\n  exec\n    harness: echo\n    prompt: \"hi\"\nworkflow main\n  run greet\n";
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let _ = engine.start_run("main").await;

        let state = engine.get_run_state().unwrap();
        assert_eq!(state.status, RunStatus::Completed);
        assert_eq!(state.rootWorkflow, "main");
        assert!(!state.steps.is_empty());
        assert!(!state.safeBoundaries.is_empty());
    }

    #[tokio::test]
    async fn test_on_event_callback() {
        use std::sync::Mutex;
        let collected = Arc::new(Mutex::new(Vec::<String>::new()));
        let collected_clone = collected.clone();

        let on_event: OnEventCallback = Box::new(move |event: &ExecutionEvent| {
            collected_clone
                .lock()
                .unwrap()
                .push(event.event_type().to_string());
        });

        let source = "version 1\nworkflow main\n  exec\n    harness: echo\n    prompt: \"hi\"\n";
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), Some(on_event), None);
        let _ = engine.start_run("main").await;

        let events = collected.lock().unwrap();
        assert!(events.contains(&"StepStarted".to_string()));
        assert!(events.contains(&"StepCompleted".to_string()));
        assert!(events.contains(&"RunCompleted".to_string()));
    }

    #[tokio::test]
    async fn test_on_save_callback() {
        use std::sync::Mutex;
        let save_count = Arc::new(Mutex::new(0usize));
        let save_count_clone = save_count.clone();

        let on_save: OnSaveCallback = Box::new(move |_state: &RunState| {
            *save_count_clone.lock().unwrap() += 1;
        });

        let source = "version 1\nworkflow greet\n  exec\n    harness: echo\n    prompt: \"hi\"\nworkflow main\n  run greet\n";
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, Some(on_save));
        let _ = engine.start_run("main").await;

        let count = *save_count.lock().unwrap();
        assert!(count > 0, "on_save should have been called at least once");
    }

    #[tokio::test]
    async fn test_reset_run() {
        let source = "version 1\nworkflow a\n  exec\n    harness: echo\n    prompt: \"a\"\nworkflow b\n  exec\n    harness: echo\n    prompt: \"b\"\nworkflow main\n  run a\n  run b\n";
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (run_id, _events) = engine.start_run("main").await;

        let state = engine.get_run_state().unwrap();
        assert!(!state.safeBoundaries.is_empty());

        // Reset to the first safe boundary
        let first_boundary = state.safeBoundaries[0];
        let new_state = engine.reset_run(&run_id, first_boundary).unwrap();
        assert_eq!(new_state.status, RunStatus::Paused);
        assert_eq!(new_state.lastSafeBoundaryIndex, first_boundary as i64);
    }

    #[tokio::test]
    async fn test_reset_run_invalid_boundary() {
        let source = "version 1\nworkflow main\n  exec\n    harness: echo\n    prompt: \"hi\"\n";
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (run_id, _) = engine.start_run("main").await;

        let result = engine.reset_run(&run_id, 99999);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("Invalid reset target"));
    }

    #[tokio::test]
    async fn test_circular_reference_detected() {
        // Create a circular reference: main -> a -> main
        let source = "version 1\nworkflow a\n  run main\nworkflow main\n  run a\n";
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("main").await;
        assert!(events.iter().any(|e| match e {
            ExecutionEvent::StepFailed { error, .. } => error.contains("Circular reference"),
            _ => false,
        }));
    }

    #[tokio::test]
    async fn test_par_and_all_succeed() {
        let source = r#"version 1
workflow branch-a
  exec
    harness: echo
    prompt: "output-a"
workflow branch-b
  exec
    harness: echo
    prompt: "output-b"
workflow merge
  exec
    harness: echo
    prompt: "merged"
workflow main
  par-and merge
    run branch-a
    run branch-b
"#;
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("main").await;

        // Both branches should have started and completed
        let branch_started_count = events
            .iter()
            .filter(|e| matches!(e, ExecutionEvent::BranchStarted { .. }))
            .count();
        assert_eq!(branch_started_count, 2);

        let branch_completed_count = events
            .iter()
            .filter(|e| matches!(e, ExecutionEvent::BranchCompleted { .. }))
            .count();
        assert_eq!(branch_completed_count, 2);

        // Join should have started
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::JoinStarted { .. }))
        );

        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunCompleted { .. }))
        );

        // Branch outputs should be available
        let outputs = engine.get_branch_outputs();
        assert_eq!(outputs.len(), 2);
    }

    #[tokio::test]
    async fn test_par_and_branch_failure() {
        // One branch fails: use a harness that fails for branch-b
        let harness: HarnessDispatchFn =
            Arc::new(|exec_block: &ExecBlock, _ctx: HarnessExecContext| {
                let prompt = exec_block.prompt.clone().unwrap_or_default();
                Box::pin(async move {
                    if prompt.contains("fail") {
                        Ok(ExecResult {
                            exit_code: 1,
                            stdout: String::new(),
                            stderr: "intentional failure".to_string(),
                            harness_events: Vec::new(),
                        })
                    } else {
                        Ok(ExecResult {
                            exit_code: 0,
                            stdout: prompt,
                            stderr: String::new(),
                            harness_events: Vec::new(),
                        })
                    }
                })
            });

        let source = r#"version 1
workflow branch-a
  exec
    harness: echo
    prompt: "output-a"
workflow branch-b
  exec
    harness: echo
    prompt: "fail"
workflow merge
  exec
    harness: echo
    prompt: "merged"
workflow main
  par-and merge
    run branch-a
    run branch-b
"#;
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, harness, None, None);
        let (_, events) = engine.start_run("main").await;

        // Should have a branch failure
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::BranchFailed { .. }))
        );

        // Should have a run failure (since a branch failed)
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunFailed { .. }))
        );
    }

    #[tokio::test]
    async fn test_generate_run_id() {
        let id = generate_run_id();
        assert!(id.starts_with("run-"));
        assert!(id.len() > 10); // Reasonable length
    }

    #[tokio::test]
    async fn test_step_path_construction() {
        let source = "version 1\nworkflow greet\n  exec\n    harness: echo\n    prompt: \"hi\"\nworkflow main\n  run greet\n";
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("main").await;

        // Check that step paths are correctly constructed
        let step_started_events: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                ExecutionEvent::StepStarted { stepPath, .. } => Some(stepPath.clone()),
                _ => None,
            })
            .collect();

        // Should have: ["main", "greet"] and ["main", "greet", "exec:echo"]
        assert!(
            step_started_events
                .iter()
                .any(|p| p == &vec!["main".to_string(), "greet".to_string()])
        );
        assert!(step_started_events.iter().any(|p| p
            == &vec![
                "main".to_string(),
                "greet".to_string(),
                "exec:echo".to_string()
            ]));
    }

    #[tokio::test]
    async fn test_safe_boundary_types_in_events() {
        let source = "version 1\nworkflow greet\n  exec\n    harness: echo\n    prompt: \"hi\"\nworkflow main\n  run greet\n";
        let wfs = parse_workflows(source);
        let mut engine = ExecutionEngine::new(wfs, make_echo_harness(), None, None);
        let (_, events) = engine.start_run("main").await;

        let boundary_types: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                ExecutionEvent::SafeBoundary { boundaryType, .. } => Some(boundaryType.clone()),
                _ => None,
            })
            .collect();

        assert!(boundary_types.contains(&safe_boundary_types::BEFORE_STEP_START.to_string()));
        assert!(boundary_types.contains(&safe_boundary_types::AFTER_STEP_COMPLETE.to_string()));
    }

    // ================================================================
    // Match execution tests
    // ================================================================

    /// Helper: build a harness that returns predefined stdout based on prompt content.
    fn make_match_harness(classify_stdout: &str) -> HarnessDispatchFn {
        let classify_output = classify_stdout.to_string();
        Arc::new(move |exec_block: &ExecBlock, _ctx: HarnessExecContext| {
            let prompt = exec_block.prompt.clone().unwrap_or_default();
            let classify_output = classify_output.clone();
            Box::pin(async move {
                let stdout = if prompt.contains("classify") {
                    classify_output
                } else {
                    prompt
                };
                Ok(ExecResult {
                    exit_code: 0,
                    stdout,
                    stderr: String::new(),
                    harness_events: Vec::new(),
                })
            })
        })
    }

    #[tokio::test]
    async fn test_match_routes_to_matching_variant_arm() {
        let source = r#"version 1
workflow classify
  exec
    harness: mock
    prompt: "classify"
workflow handle-small
  exec
    harness: mock
    prompt: "small-action"
workflow handle-large
  exec
    harness: mock
    prompt: "large-action"
workflow main
  match classify
    small -> run handle-small
    large -> run handle-large
"#;
        let wfs = parse_workflows(source);
        let harness = make_match_harness(r#"{"variant":"small"}"#);
        let mut engine = ExecutionEngine::new(wfs, harness, None, None);
        let (_, events) = engine.start_run("main").await;

        // Run should complete
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunCompleted { .. }))
        );

        // handle-small should have been executed
        assert!(events.iter().any(|e| matches!(
            e,
            ExecutionEvent::StepStarted { stepPath, .. }
            if stepPath.iter().any(|s| s.contains("handle-small"))
        )));

        // handle-large should NOT have been executed
        assert!(!events.iter().any(|e| matches!(
            e,
            ExecutionEvent::StepStarted { stepPath, .. }
            if stepPath.iter().any(|s| s.contains("handle-large"))
        )));
    }

    #[tokio::test]
    async fn test_match_routes_to_else_arm_when_no_variant_matches() {
        let source = r#"version 1
workflow classify
  exec
    harness: mock
    prompt: "classify"
workflow handle-small
  exec
    harness: mock
    prompt: "small-action"
workflow handle-large
  exec
    harness: mock
    prompt: "large-action"
workflow handle-fallback
  exec
    harness: mock
    prompt: "fallback-action"
workflow main
  match classify
    small -> run handle-small
    large -> run handle-large
    else -> run handle-fallback
"#;
        let wfs = parse_workflows(source);
        let harness = make_match_harness(r#"{"variant":"unknown"}"#);
        let mut engine = ExecutionEngine::new(wfs, harness, None, None);
        let (_, events) = engine.start_run("main").await;

        // Run should complete
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunCompleted { .. }))
        );

        // handle-fallback (else body) should have been executed
        assert!(events.iter().any(|e| matches!(
            e,
            ExecutionEvent::StepStarted { stepPath, .. }
            if stepPath.iter().any(|s| s.contains("handle-fallback"))
        )));

        // small and large should NOT have been executed
        assert!(!events.iter().any(|e| matches!(
            e,
            ExecutionEvent::StepStarted { stepPath, .. }
            if stepPath.iter().any(|s| s.contains("handle-small"))
        )));
        assert!(!events.iter().any(|e| matches!(
            e,
            ExecutionEvent::StepStarted { stepPath, .. }
            if stepPath.iter().any(|s| s.contains("handle-large"))
        )));

        // MatchEvaluated event should have armIndex: None (null) for else arm
        let match_event = events
            .iter()
            .find(|e| matches!(e, ExecutionEvent::MatchEvaluated { .. }));
        assert!(
            match_event.is_some(),
            "MatchEvaluated event should be emitted"
        );
        if let ExecutionEvent::MatchEvaluated { armIndex, .. } = match_event.unwrap() {
            assert!(
                armIndex.is_none(),
                "armIndex should be None for else arm, got {:?}",
                armIndex
            );
        }
    }

    #[tokio::test]
    async fn test_match_throws_runtime_error_when_no_match_and_no_else() {
        let source = r#"version 1
workflow classify
  exec
    harness: mock
    prompt: "classify"
workflow handle-small
  exec
    harness: mock
    prompt: "small-action"
workflow handle-large
  exec
    harness: mock
    prompt: "large-action"
workflow main
  match classify
    small -> run handle-small
    large -> run handle-large
"#;
        let wfs = parse_workflows(source);
        let harness = make_match_harness(r#"{"variant":"unknown"}"#);
        let mut engine = ExecutionEngine::new(wfs, harness, None, None);
        let (_, events) = engine.start_run("main").await;

        // Should fail
        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunFailed { .. }))
        );
        assert!(
            !events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunCompleted { .. }))
        );

        // Error should mention unrecognized variant and available arms
        let step_failed = events
            .iter()
            .find(|e| matches!(e, ExecutionEvent::StepFailed { .. }));
        assert!(step_failed.is_some());
        if let ExecutionEvent::StepFailed { error, .. } = step_failed.unwrap() {
            assert!(
                error.contains("unrecognized variant"),
                "error should contain 'unrecognized variant': {}",
                error
            );
            assert!(
                error.contains("unknown"),
                "error should mention variant: {}",
                error
            );
            assert!(
                error.contains("small"),
                "error should list expected arms: {}",
                error
            );
            assert!(
                error.contains("large"),
                "error should list expected arms: {}",
                error
            );
        }
    }

    #[tokio::test]
    async fn test_match_emits_before_and_after_match_arm_boundaries() {
        let source = r#"version 1
workflow classify
  exec
    harness: mock
    prompt: "classify"
workflow handle-small
  exec
    harness: mock
    prompt: "small-action"
workflow main
  match classify
    small -> run handle-small
"#;
        let wfs = parse_workflows(source);
        let harness = make_match_harness(r#"{"variant":"small"}"#);
        let mut engine = ExecutionEngine::new(wfs, harness, None, None);
        let (_, events) = engine.start_run("main").await;

        assert!(
            events
                .iter()
                .any(|e| matches!(e, ExecutionEvent::RunCompleted { .. }))
        );

        let boundary_types: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                ExecutionEvent::SafeBoundary { boundaryType, .. } => Some(boundaryType.clone()),
                _ => None,
            })
            .collect();

        assert!(
            boundary_types.contains(&safe_boundary_types::BEFORE_MATCH_ARM.to_string()),
            "should emit before-match-arm boundary"
        );
        assert!(
            boundary_types.contains(&safe_boundary_types::AFTER_MATCH_ARM.to_string()),
            "should emit after-match-arm boundary"
        );

        // Verify ordering: before comes before after
        let before_idx = events.iter().position(|e| {
            matches!(
                e,
                ExecutionEvent::SafeBoundary { boundaryType, .. }
                if boundaryType == safe_boundary_types::BEFORE_MATCH_ARM
            )
        });
        let after_idx = events.iter().position(|e| {
            matches!(
                e,
                ExecutionEvent::SafeBoundary { boundaryType, .. }
                if boundaryType == safe_boundary_types::AFTER_MATCH_ARM
            )
        });
        assert!(before_idx.unwrap() < after_idx.unwrap());
    }

    #[tokio::test]
    async fn test_match_passes_variant_reason_through_match_evaluated_event() {
        let source = r#"version 1
workflow classify
  exec
    harness: mock
    prompt: "classify"
workflow handle-small
  exec
    harness: mock
    prompt: "small-action"
workflow main
  match classify
    small -> run handle-small
"#;
        let wfs = parse_workflows(source);
        let harness = make_match_harness(r#"{"variant":"small","reason":"under 50 lines"}"#);
        let mut engine = ExecutionEngine::new(wfs, harness, None, None);
        let (_, events) = engine.start_run("main").await;

        let match_events: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, ExecutionEvent::MatchEvaluated { .. }))
            .collect();
        assert_eq!(match_events.len(), 1);

        if let ExecutionEvent::MatchEvaluated {
            variant,
            reason,
            checkName,
            armIndex,
            ..
        } = &match_events[0]
        {
            assert_eq!(variant, "small");
            assert_eq!(reason.as_deref(), Some("under 50 lines"));
            assert_eq!(checkName, "classify");
            assert_eq!(*armIndex, Some(0));
        } else {
            panic!("Expected MatchEvaluated event");
        }
    }

    #[tokio::test]
    async fn test_match_omits_reason_from_match_evaluated_when_not_present() {
        let source = r#"version 1
workflow classify
  exec
    harness: mock
    prompt: "classify"
workflow handle-small
  exec
    harness: mock
    prompt: "small-action"
workflow main
  match classify
    small -> run handle-small
"#;
        let wfs = parse_workflows(source);
        let harness = make_match_harness(r#"{"variant":"small"}"#);
        let mut engine = ExecutionEngine::new(wfs, harness, None, None);
        let (_, events) = engine.start_run("main").await;

        let match_events: Vec<_> = events
            .iter()
            .filter(|e| matches!(e, ExecutionEvent::MatchEvaluated { .. }))
            .collect();
        assert_eq!(match_events.len(), 1);

        if let ExecutionEvent::MatchEvaluated {
            variant,
            reason,
            armIndex,
            ..
        } = &match_events[0]
        {
            assert_eq!(variant, "small");
            assert!(reason.is_none(), "reason should be None when not in output");
            assert_eq!(*armIndex, Some(0));
        } else {
            panic!("Expected MatchEvaluated event");
        }
    }
}