//! Workflow execution engine — walks the stage graph, dispatches to car-multi
//! patterns and car-engine proposals, manages state flow and saga compensation.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Instant;

use serde_json::Value;
use tracing::{debug, info, warn};

use car_multi::{
    AdversarialReview, AgentRunner, Fleet, MapReduce, Pipeline, SharedInfra, Supervisor, Swarm,
    SwarmMode, Tournament, Vote,
};

use crate::error::WorkflowError;
use crate::result::*;
use crate::types::*;

/// What executing one stage step yields: its typed output, an answer string for
/// edge conditions, and the state deltas to merge into workflow state.
type StepOutcome = (StageOutput, String, HashMap<String, Value>);

/// Workflow execution engine.
pub struct WorkflowEngine {
    runner: Arc<dyn AgentRunner>,
    infra: SharedInfra,
    /// Directory backing `Dedup` steps' persistent seen-sets. `None` defaults to
    /// `~/.car/workflow-dedup` lazily, only when a `Dedup` step actually runs —
    /// pattern/proposal-only workflows touch no disk here.
    dedup_dir: Option<std::path::PathBuf>,
}

impl WorkflowEngine {
    pub fn new(runner: Arc<dyn AgentRunner>, infra: SharedInfra) -> Self {
        Self {
            runner,
            infra,
            dedup_dir: None,
        }
    }

    /// Like [`new`](Self::new) but pins the directory backing `Dedup` steps'
    /// persistent seen-sets (default `~/.car/workflow-dedup`). Use this to
    /// isolate dedup state in tests or relocate it in production.
    pub fn with_dedup_dir(
        runner: Arc<dyn AgentRunner>,
        infra: SharedInfra,
        dedup_dir: impl Into<std::path::PathBuf>,
    ) -> Self {
        Self {
            runner,
            infra,
            dedup_dir: Some(dedup_dir.into()),
        }
    }

    /// Execute a workflow to completion, following the stage graph.
    pub fn run<'a>(
        &'a self,
        workflow: &'a Workflow,
    ) -> futures::future::BoxFuture<'a, Result<WorkflowResult, WorkflowError>> {
        Box::pin(self.run_inner(workflow, HashMap::new()))
    }

    /// Like [`run`](Self::run), but seeds `initial_state` into workflow state
    /// before the run starts — the inter-workflow chaining primitive: hand one
    /// workflow's `final_state` to the next so its edge conditions and stage
    /// templates can read upstream results. The reserved `goal` anchor is
    /// re-pinned from the workflow definition after seeding, so a seeded `goal`
    /// key can never repoint the drift guard. An empty map is exactly `run`.
    pub fn run_with_state<'a>(
        &'a self,
        workflow: &'a Workflow,
        initial_state: HashMap<String, Value>,
    ) -> futures::future::BoxFuture<'a, Result<WorkflowResult, WorkflowError>> {
        Box::pin(self.run_inner(workflow, initial_state))
    }

    async fn run_inner(
        &self,
        workflow: &Workflow,
        initial_state: HashMap<String, Value>,
    ) -> Result<WorkflowResult, WorkflowError> {
        // Validate start stage exists
        if workflow.stage(&workflow.start).is_none() {
            return Err(WorkflowError::NoStartStage);
        }

        // Fail closed on pathologically deep loop/foreach/sub-workflow nesting
        // before recursing into it — guards execution even when the caller
        // skipped `verify_workflow`.
        for stage in &workflow.stages {
            if crate::verify::exceeds_nesting(&stage.step, crate::verify::MAX_STEP_NESTING_DEPTH) {
                return Err(WorkflowError::StageFailed(
                    stage.id.clone(),
                    format!(
                        "loop/foreach/sub-workflow nesting exceeds the limit of {}",
                        crate::verify::MAX_STEP_NESTING_DEPTH
                    ),
                ));
            }
        }

        let mut wf_state = initial_state;
        // Pin the overall goal into state so every stage (and edge condition) can
        // re-anchor on it — a structural guard against goal drift. Pinned AFTER
        // seeding the caller's initial_state, so a seeded `goal` never wins.
        if let Some(goal) = &workflow.goal {
            wf_state.insert("goal".to_string(), Value::String(goal.clone()));
        } else {
            wf_state.remove("goal");
        }
        let cursor = Cursor {
            wf_state,
            stage_results: Vec::new(),
            completed_stage_ids: Vec::new(),
            iterations: 0,
            prior_duration_ms: 0.0,
            current_id: workflow.start.clone(),
        };
        self.drive(workflow, new_run_id(), cursor, &HashMap::new())
            .await
    }

    /// Re-run a workflow, reusing the successful prefix of a `prior` result.
    ///
    /// Every stage that succeeded in `prior` is replayed from its cached result
    /// (instantly, no agent/proposal call); the first stage that did not succeed —
    /// and everything after it — runs live, seeded with `prior.final_state`. Same
    /// workflow + same prior result → the same resume point, deterministically.
    ///
    /// Use this to recover a long run that failed partway, or to continue after
    /// fixing the cause of a failure, without re-paying for completed work. Does
    /// not resume an approval *pause* (use [`resume`](Self::resume) for that),
    /// though it *can* newly pause at an approval gate the prior run never reached.
    ///
    /// Replayed stages keep their original recorded `duration_ms`, so summing
    /// per-stage durations across the original and cached runs would double-count
    /// the replayed prefix.
    pub async fn run_cached(
        &self,
        workflow: &Workflow,
        prior: &WorkflowResult,
    ) -> Result<WorkflowResult, WorkflowError> {
        if workflow.stage(&workflow.start).is_none() {
            return Err(WorkflowError::NoStartStage);
        }

        // Cache only the stages that actually succeeded.
        let cache: HashMap<String, StageResult> = prior
            .stages
            .iter()
            .filter(|s| s.status == StageStatus::Succeeded)
            .map(|s| (s.stage_id.clone(), s.clone()))
            .collect();

        // Seed state from the prior run so cached stages' effects are visible to
        // edge conditions.
        let mut wf_state = prior.final_state.clone();

        // Strip the per-stage bookkeeping (`stage.<id>.{succeeded,error,answer,
        // ...}`) of any stage NOT in the cache. The prior run's failed/partial
        // stages left these behind; carrying them would let an edge that
        // forward-references a not-yet-rerun stage's slot route differently than
        // the original run saw at this point. Genuine data deltas are untouched.
        let stage_ids: Vec<String> = workflow.stages.iter().map(|s| s.id.clone()).collect();
        wf_state.retain(|k, _| {
            match k.strip_prefix("stage.") {
                // Keep keys for cached stages; drop bookkeeping for the rest.
                Some(rest) => {
                    let id = rest.split('.').next().unwrap_or("");
                    cache.contains_key(id) || !stage_ids.iter().any(|s| s == id)
                }
                None => true,
            }
        });

        // Re-pin the goal authoritatively (after seeding, which may carry a stale
        // or clobbered value).
        if let Some(goal) = &workflow.goal {
            wf_state.insert("goal".to_string(), Value::String(goal.clone()));
        } else {
            wf_state.remove("goal");
        }

        let cursor = Cursor {
            wf_state,
            stage_results: Vec::new(),
            completed_stage_ids: Vec::new(),
            iterations: 0,
            prior_duration_ms: 0.0,
            current_id: workflow.start.clone(),
        };
        self.drive(workflow, new_run_id(), cursor, &cache).await
    }

    /// Resume a run that was parked at an approval gate.
    ///
    /// Applies `input` as the approval stage's response (written to its
    /// `output_key` and mirrored to `stage.<id>.answer`), records the stage as
    /// succeeded, then continues the stage graph. Chained approval gates pause
    /// again, returning a fresh [`PausedWorkflow`] each time.
    pub async fn resume(
        &self,
        paused: PausedWorkflow,
        input: HashMap<String, Value>,
    ) -> Result<WorkflowResult, WorkflowError> {
        let PausedWorkflow {
            run_id,
            workflow,
            paused_stage_id,
            mut wf_state,
            mut stage_results,
            mut completed_stage_ids,
            iterations,
            prior_duration_ms,
            ..
        } = paused;

        // The checkpoint is serializable and crosses the FFI/persistence
        // boundary, so treat it as untrusted: re-derive everything we can from
        // the embedded definition rather than the carried copies. Confirm the
        // parked stage really is an approval gate and read its declared
        // `output_key` from the step itself (ignoring the checkpoint's copy).
        let approval = match workflow.stage(&paused_stage_id).map(|s| &s.step) {
            Some(StageStep::Approval(ap)) => ap.clone(),
            Some(_) => {
                return Err(WorkflowError::InvalidResume(format!(
                    "stage '{paused_stage_id}' is not an approval gate"
                )))
            }
            None => {
                return Err(WorkflowError::InvalidResume(format!(
                    "paused stage '{paused_stage_id}' not found in workflow"
                )))
            }
        };
        let output_key = approval.output_key.clone();
        if output_key.trim().is_empty() {
            return Err(WorkflowError::InvalidResume(format!(
                "approval stage '{paused_stage_id}' has empty output_key"
            )));
        }
        if output_key == "goal" {
            return Err(WorkflowError::InvalidResume(
                "approval output_key 'goal' is reserved (it is the drift anchor)".into(),
            ));
        }

        // `goal` is a reserved anchor: re-derive it from the authoritative
        // workflow definition rather than trusting the checkpoint's wf_state
        // (which is treated as untrusted, like output_key above).
        if let Some(goal) = &workflow.goal {
            wf_state.insert("goal".to_string(), Value::String(goal.clone()));
        } else {
            wf_state.remove("goal");
        }

        // Enforce the form contract the human was shown: required fields present,
        // `options` values in range. Lets a caller turn a rejection into a
        // re-prompt instead of silently mis-routing on a failed edge condition.
        validate_approval_input(&approval.fields, &input)?;

        // The human's response, written three ways:
        //   1. the whole object at `output_key`,
        //   2. each top-level field flattened to `output_key.<field>` so edge
        //      conditions can branch on individual scalar answers (approve/revise).
        //      Flattening is top-level only — nested objects stay opaque.
        //   3. `stage.<id>.answer` / `.succeeded` for the standard stage slots.
        let response = Value::Object(input.into_iter().collect());
        if let Value::Object(map) = &response {
            for (k, v) in map {
                wf_state.insert(format!("{output_key}.{k}"), v.clone());
            }
        }
        wf_state.insert(output_key.clone(), response.clone());
        wf_state.insert(
            format!("stage.{paused_stage_id}.succeeded"),
            Value::Bool(true),
        );
        wf_state.insert(
            format!("stage.{paused_stage_id}.answer"),
            Value::String(response.to_string()),
        );

        let stage_name = workflow
            .stage(&paused_stage_id)
            .map(|s| s.name.clone())
            .unwrap_or_else(|| paused_stage_id.clone());
        stage_results.push(StageResult {
            stage_id: paused_stage_id.clone(),
            stage_name,
            status: StageStatus::Succeeded,
            output: StageOutput::Approval { response },
            duration_ms: 0.0,
            error: None,
        });
        completed_stage_ids.push(paused_stage_id.clone());

        info!(stage = %paused_stage_id, run_id = %run_id, "approval resumed");

        // Where does control flow after the gate?
        match next_stage(&workflow, &paused_stage_id, &wf_state) {
            Some(next_id) => {
                let cursor = Cursor {
                    wf_state,
                    stage_results,
                    completed_stage_ids,
                    iterations,
                    prior_duration_ms,
                    current_id: next_id,
                };
                self.drive(&workflow, run_id, cursor, &HashMap::new()).await
            }
            None => {
                // The gate was terminal — the run completes on approval.
                info!(workflow = %workflow.name, "workflow completed (terminal approval gate)");
                Ok(completed_result(
                    &workflow,
                    stage_results,
                    wf_state,
                    prior_duration_ms,
                ))
            }
        }
    }

    /// Drive the stage graph forward from a cursor position until the run
    /// completes, fails, or pauses at an approval gate.
    ///
    /// `cache` maps stage id → a prior successful `StageResult`. A stage found in
    /// the cache is **replayed from cache** (not re-executed) — the prefix of an
    /// already-completed run is reused instantly, and the first uncached stage
    /// (and everything after it) runs live. Pass an empty map for a normal run.
    async fn drive(
        &self,
        workflow: &Workflow,
        run_id: String,
        mut cur: Cursor,
        cache: &HashMap<String, StageResult>,
    ) -> Result<WorkflowResult, WorkflowError> {
        let start = Instant::now();

        loop {
            cur.iterations += 1;
            if cur.iterations > workflow.max_iterations {
                return Err(WorkflowError::CycleLimitReached(workflow.max_iterations));
            }

            let stage = workflow
                .stage(&cur.current_id)
                .ok_or_else(|| WorkflowError::StageNotFound(cur.current_id.clone()))?;

            // Prefix cache: a stage that already succeeded in a prior run is
            // replayed from its cached result rather than re-executed. The
            // seeded wf_state already carries that stage's state, so edge
            // conditions still evaluate correctly. Checked before the approval
            // gate so a previously-answered gate is not re-paused.
            if let Some(cached) = cache.get(&cur.current_id) {
                debug!(stage = %cur.current_id, "replaying cached stage");
                cur.wf_state
                    .insert(format!("stage.{}.succeeded", stage.id), Value::Bool(true));
                cur.stage_results.push(cached.clone());
                cur.completed_stage_ids.push(stage.id.clone());
                match next_stage(workflow, &cur.current_id, &cur.wf_state) {
                    Some(next_id) => {
                        cur.current_id = next_id;
                        continue;
                    }
                    None => {
                        let duration =
                            cur.prior_duration_ms + start.elapsed().as_secs_f64() * 1000.0;
                        return Ok(completed_result(
                            workflow,
                            cur.stage_results,
                            cur.wf_state,
                            duration,
                        ));
                    }
                }
            }

            // Approval gate: pause BEFORE doing any work, snapshot, and return.
            // The run resumes only via `resume()` with the human's response.
            if let StageStep::Approval(ap) = &stage.step {
                // Validated by `verify_workflow`, but guard here too: an empty
                // output_key would silently drop the human's response.
                if ap.output_key.trim().is_empty() {
                    return Err(WorkflowError::StageFailed(
                        stage.id.clone(),
                        "approval stage has empty output_key".into(),
                    ));
                }
                info!(stage = %stage.id, run_id = %run_id, "workflow paused at approval gate");
                // Wall time accumulated across pause/resume segments; the human
                // wait between pause and resume is deliberately not counted.
                let elapsed = cur.prior_duration_ms + start.elapsed().as_secs_f64() * 1000.0;
                let now = chrono::Utc::now();
                let paused = PausedWorkflow {
                    run_id: run_id.clone(),
                    workflow: workflow.clone(),
                    paused_stage_id: stage.id.clone(),
                    prompt: ap.prompt.clone(),
                    fields: ap.fields.clone(),
                    output_key: ap.output_key.clone(),
                    wf_state: cur.wf_state.clone(),
                    stage_results: cur.stage_results.clone(),
                    completed_stage_ids: cur.completed_stage_ids.clone(),
                    iterations: cur.iterations,
                    prior_duration_ms: elapsed,
                    created_at: now,
                };
                return Ok(WorkflowResult {
                    workflow_id: workflow.id.clone(),
                    workflow_name: workflow.name.clone(),
                    status: WorkflowStatus::Paused,
                    stages: cur.stage_results,
                    compensations: vec![],
                    duration_ms: elapsed,
                    timestamp: now,
                    final_state: cur.wf_state,
                    paused: Some(paused),
                });
            }

            debug!(stage = %stage.id, name = %stage.name, iteration = cur.iterations, "executing stage");

            // Execute the stage
            let stage_start = Instant::now();
            // A Deliver stage accumulates per-sink results into a slot owned by
            // THIS frame, not the cancellable step future: a stage timeout
            // cancels the future mid-fan-out, and without the slot the sinks
            // that already fired (non-idempotent external effects) would vanish
            // into `StageOutput::Empty` — invisible partial delivery, and a
            // `run_cached` retry would re-send them all.
            let deliver_partial: Option<DeliverPartial> = match &stage.step {
                StageStep::Deliver(_) => Some(Arc::new(std::sync::Mutex::new(Vec::new()))),
                _ => None,
            };
            let step_fut = async {
                match (&stage.step, &deliver_partial) {
                    (StageStep::Deliver(ds), Some(slot)) => {
                        self.execute_deliver_into(&stage.id, ds, &cur.wf_state, slot)
                            .await
                    }
                    _ => self.execute_step(&stage.id, &stage.step, &cur.wf_state).await,
                }
            };
            let result = if let Some(timeout_ms) = stage.timeout_ms {
                match tokio::time::timeout(
                    std::time::Duration::from_millis(timeout_ms),
                    step_fut,
                )
                .await
                {
                    Ok(r) => r,
                    Err(_) => Err(WorkflowError::Timeout(stage.id.clone(), timeout_ms)),
                }
            } else {
                step_fut.await
            };
            let stage_duration = stage_start.elapsed().as_secs_f64() * 1000.0;

            match result {
                Ok((output, answer, deltas)) => {
                    // Merge the state deltas the step produced (proposal
                    // state_changes, loop/foreach per-iteration slots) FIRST, so
                    // the runtime-reserved slots written below intentionally win
                    // over any colliding delta.
                    for (k, v) in deltas {
                        cur.wf_state.insert(k, v);
                    }
                    // `goal` is a reserved anchor: re-assert it so no stage delta
                    // (a proposal writing a key named "goal", etc.) can repoint
                    // the drift guard mid-run.
                    if let Some(goal) = &workflow.goal {
                        cur.wf_state
                            .insert("goal".to_string(), Value::String(goal.clone()));
                    }
                    cur.wf_state
                        .insert(format!("stage.{}.succeeded", stage.id), Value::Bool(true));
                    cur.wf_state
                        .insert(format!("stage.{}.answer", stage.id), Value::String(answer));

                    cur.stage_results.push(StageResult {
                        stage_id: stage.id.clone(),
                        stage_name: stage.name.clone(),
                        status: StageStatus::Succeeded,
                        output,
                        duration_ms: stage_duration,
                        error: None,
                    });
                    cur.completed_stage_ids.push(stage.id.clone());

                    info!(stage = %stage.id, duration_ms = stage_duration, "stage succeeded");
                }
                Err(e) => {
                    let error_msg = e.to_string();
                    cur.wf_state
                        .insert(format!("stage.{}.succeeded", stage.id), Value::Bool(false));
                    cur.wf_state.insert(
                        format!("stage.{}.error", stage.id),
                        Value::String(error_msg.clone()),
                    );

                    // Drain any per-sink results a cancelled/failed Deliver
                    // fan-out recorded before the error, so partial delivery
                    // stays visible in the failed stage's output.
                    let failed_output = deliver_partial
                        .map(|slot| std::mem::take(&mut *slot.lock().unwrap_or_else(|p| p.into_inner())))
                        .filter(|results| !results.is_empty())
                        .map(|results| StageOutput::Deliver { results })
                        .unwrap_or(StageOutput::Empty);

                    cur.stage_results.push(StageResult {
                        stage_id: stage.id.clone(),
                        stage_name: stage.name.clone(),
                        status: StageStatus::Failed,
                        output: failed_output,
                        duration_ms: stage_duration,
                        error: Some(error_msg.clone()),
                    });

                    warn!(stage = %stage.id, error = %error_msg, "stage failed, running compensation");

                    // Run saga compensation in reverse order
                    let compensations = self.compensate(workflow, &cur.completed_stage_ids).await;

                    let all_compensated = compensations
                        .iter()
                        .all(|c| c.status == StageStatus::Succeeded);
                    let any_compensated = !compensations.is_empty();

                    let status = if any_compensated && all_compensated {
                        WorkflowStatus::Compensated
                    } else if any_compensated {
                        WorkflowStatus::PartiallyCompensated
                    } else {
                        WorkflowStatus::Failed
                    };

                    return Ok(WorkflowResult {
                        workflow_id: workflow.id.clone(),
                        workflow_name: workflow.name.clone(),
                        status,
                        stages: cur.stage_results,
                        compensations,
                        duration_ms: cur.prior_duration_ms + start.elapsed().as_secs_f64() * 1000.0,
                        timestamp: chrono::Utc::now(),
                        final_state: cur.wf_state,
                        paused: None,
                    });
                }
            }

            // Evaluate outgoing edges to find the next stage
            match next_stage(workflow, &cur.current_id, &cur.wf_state) {
                Some(next_id) => {
                    debug!(from = %cur.current_id, to = %next_id, "taking edge");
                    cur.current_id = next_id;
                }
                None => {
                    // Terminal stage — workflow complete
                    info!(
                        workflow = %workflow.name,
                        stages_executed = cur.stage_results.len(),
                        "workflow completed"
                    );
                    let duration = cur.prior_duration_ms + start.elapsed().as_secs_f64() * 1000.0;
                    return Ok(completed_result(
                        workflow,
                        cur.stage_results,
                        cur.wf_state,
                        duration,
                    ));
                }
            }
        }
    }

    /// Execute a single stage step, returning `(output, answer_string,
    /// state_deltas)`. `stage_id` namespaces any state the step writes;
    /// `wf_state` is the read-only state visible to the step. Boxed because
    /// `LoopUntil`/`ForEach` recurse into their inner body step.
    fn execute_step<'a>(
        &'a self,
        stage_id: &'a str,
        step: &'a StageStep,
        wf_state: &'a HashMap<String, Value>,
    ) -> futures::future::BoxFuture<'a, Result<StepOutcome, WorkflowError>> {
        Box::pin(async move {
            match step {
                StageStep::Pattern(ps) => {
                    let (out, answer) = self.execute_pattern(ps, wf_state).await?;
                    // Surface a review verdict as a typed state slot so edges can
                    // branch on `stage.<id>.review_passed` (bool), not prose.
                    let mut deltas = HashMap::new();
                    if let StageOutput::Review {
                        passed,
                        blocker_count,
                        inconclusive,
                        ..
                    } = &out
                    {
                        deltas.insert(
                            format!("stage.{stage_id}.review_passed"),
                            Value::Bool(*passed),
                        );
                        deltas.insert(
                            format!("stage.{stage_id}.review_blockers"),
                            Value::from(*blocker_count),
                        );
                        // Lets edges route "not verified" distinctly from a
                        // real rejection (retry/escalate vs. fix the work).
                        deltas.insert(
                            format!("stage.{stage_id}.review_inconclusive"),
                            Value::Bool(*inconclusive),
                        );
                    }
                    Ok((out, answer, deltas))
                }
                StageStep::Proposal(ps) => {
                    let (out, answer) = self.execute_proposal(ps).await?;
                    // Surface proposal state_changes as deltas so the caller
                    // merges them into workflow state uniformly.
                    let mut deltas = HashMap::new();
                    if let StageOutput::Proposal { ref result } = out {
                        for ar in &result.results {
                            for (k, v) in &ar.state_changes {
                                deltas.insert(k.clone(), v.clone());
                            }
                        }
                    }
                    Ok((out, answer, deltas))
                }
                StageStep::SubWorkflow(sw) => {
                    let (out, answer) = self.execute_sub_workflow(sw).await?;
                    Ok((out, answer, HashMap::new()))
                }
                StageStep::LoopUntil(ls) => self.execute_loop_until(stage_id, ls, wf_state).await,
                StageStep::ForEach(fe) => self.execute_for_each(stage_id, fe, wf_state).await,
                StageStep::Dedup(ds) => self.execute_dedup(stage_id, ds, wf_state).await,
                StageStep::Deliver(ds) => self.execute_deliver(stage_id, ds, wf_state).await,
                // Approval gates are intercepted by `drive`/`resume` before
                // reaching here; this arm only fires if one is nested inside a
                // loop/foreach body or referenced as a compensation handler,
                // neither of which is meaningful.
                StageStep::Approval(_) => Err(WorkflowError::StageFailed(
                    "approval".into(),
                    "an approval gate cannot be executed as a step, loop/foreach body, or compensation".into(),
                )),
            }
        })
    }

    /// Run a `LoopUntil` step: repeat the body until `until` holds or the cap is
    /// hit. Body state is threaded forward across iterations and exposed to
    /// `until` via `stage.<id>.answer` and `stage.<id>.iteration`.
    async fn execute_loop_until(
        &self,
        stage_id: &str,
        ls: &LoopUntilStep,
        wf_state: &HashMap<String, Value>,
    ) -> Result<StepOutcome, WorkflowError> {
        if ls.max_iterations == 0 {
            return Err(WorkflowError::StageFailed(
                stage_id.to_string(),
                "loop_until requires max_iterations >= 1".into(),
            ));
        }

        let body_id = format!("{stage_id}.body");
        let mut working = wf_state.clone();
        let mut accumulated: HashMap<String, Value> = HashMap::new();
        let mut outputs: Vec<Box<StageOutput>> = Vec::new();
        let mut last_answer = String::new();
        let mut satisfied = false;
        let mut ran: u32 = 0;

        for _ in 0..ls.max_iterations {
            let (out, answer, deltas) = self.execute_step(&body_id, &ls.body, &working).await?;
            ran += 1;

            // Thread body deltas + the conventional loop slots forward so the
            // next iteration and the `until` check can observe them.
            for (k, v) in deltas {
                working.insert(k.clone(), v.clone());
                accumulated.insert(k, v);
            }
            let answer_val = Value::String(answer.clone());
            let iter_val = Value::from(ran);
            working.insert(format!("stage.{stage_id}.answer"), answer_val.clone());
            working.insert(format!("stage.{stage_id}.iteration"), iter_val.clone());
            accumulated.insert(format!("stage.{stage_id}.answer"), answer_val);
            accumulated.insert(format!("stage.{stage_id}.iteration"), iter_val);

            last_answer = answer;
            outputs.push(Box::new(out));

            // An empty `until` means "run exactly max_iterations" — never
            // satisfied early.
            if !ls.until.is_empty() && check_conditions(&ls.until, &working) {
                satisfied = true;
                break;
            }
        }

        Ok((
            StageOutput::Loop {
                iterations: ran,
                satisfied,
                iterations_output: outputs,
            },
            last_answer,
            accumulated,
        ))
    }

    /// Run a `ForEach` step: resolve the item list from state at runtime and run
    /// the (item-templated) body once per item, sequentially or with bounded
    /// concurrency. Bodies read the pre-loop state; per-item answers are exposed
    /// under `foreach.<id>.<index>.{item,answer}`.
    async fn execute_for_each(
        &self,
        stage_id: &str,
        fe: &ForEachStep,
        wf_state: &HashMap<String, Value>,
    ) -> Result<StepOutcome, WorkflowError> {
        // Resolve items at runtime. A missing key or non-array is a no-op.
        let items: Vec<Value> = wf_state
            .get(&fe.items_from)
            .and_then(|v| v.as_array().cloned())
            .unwrap_or_default();
        let item_strs: Vec<String> = items.iter().map(render_item).collect();

        // Build a templated body per item up front (pure, cheap) so the async
        // bodies below borrow owned copies.
        let mut templated: Vec<StageStep> = Vec::with_capacity(items.len());
        for (i, item) in item_strs.iter().enumerate() {
            templated.push(template_step(&fe.body, item, i)?);
        }

        let concurrency = fe.max_concurrent.max(1);

        // Build the per-item futures up front (avoids a higher-ranked lifetime
        // snag with `stream::iter(...).map(closure)`).
        let mut futs = Vec::with_capacity(templated.len());
        for (i, body) in templated.iter().enumerate() {
            futs.push(self.run_foreach_body(stage_id, i, body, wf_state));
        }

        // Drive the per-item bodies, preserving item order in the results.
        let results: Vec<Result<StepOutcome, WorkflowError>> = if concurrency <= 1 {
            let mut acc = Vec::with_capacity(futs.len());
            for f in futs {
                acc.push(f.await);
            }
            acc
        } else {
            use futures::stream::StreamExt;
            futures::stream::iter(futs)
                .buffered(concurrency)
                .collect()
                .await
        };

        let mut outputs: Vec<Box<StageOutput>> = Vec::with_capacity(results.len());
        let mut deltas: HashMap<String, Value> = HashMap::new();
        for (i, r) in results.into_iter().enumerate() {
            let (out, answer, body_deltas) = r?;
            deltas.insert(
                format!("foreach.{stage_id}.{i}.item"),
                Value::String(item_strs[i].clone()),
            );
            deltas.insert(
                format!("foreach.{stage_id}.{i}.answer"),
                Value::String(answer),
            );
            // Re-namespace each body's state deltas per item so they don't
            // clobber across concurrent items but stay recoverable downstream.
            for (k, v) in body_deltas {
                deltas.insert(format!("foreach.{stage_id}.{i}.state.{k}"), v);
            }
            outputs.push(Box::new(out));
        }
        deltas.insert(
            format!("foreach.{stage_id}.count"),
            Value::from(item_strs.len()),
        );

        let answer = format!("{} item(s) processed", item_strs.len());
        Ok((
            StageOutput::ForEach {
                items: item_strs,
                outputs,
            },
            answer,
            deltas,
        ))
    }

    /// Run a `Dedup` step: read candidate items from state, drop those already
    /// recorded in the persistent seen-set, commit the survivors, and write the
    /// unseen subset to `into`. Also exposes `stage.<id>.{unseen,total}_count`.
    async fn execute_dedup(
        &self,
        stage_id: &str,
        ds: &DedupStep,
        wf_state: &HashMap<String, Value>,
    ) -> Result<StepOutcome, WorkflowError> {
        let items: Vec<Value> = wf_state
            .get(&ds.items_from)
            .and_then(|v| v.as_array().cloned())
            .unwrap_or_default();
        let total = items.len();

        let dir = self.dedup_dir.clone().unwrap_or_else(default_dedup_dir);
        let store = crate::dedup::DedupStore::open(&dir, &ds.store).map_err(|e| {
            WorkflowError::StageFailed(
                stage_id.to_string(),
                format!("dedup store open failed: {e}"),
            )
        })?;
        // Saturate the cast: a ttl_secs >= 2^63 would wrap to a negative
        // Duration and invert the cutoff into the future, evicting everything.
        // (Such values are absurd — ~292 billion years — but a silent sign flip
        // is worse than a clamp.) `ttl_secs == 0` is rejected by verify_workflow.
        let ttl = ds
            .ttl_secs
            .map(|s| chrono::Duration::seconds(s.min(i64::MAX as u64) as i64));
        let unseen = store
            .filter_unseen(&items, &ds.hash_fields, chrono::Utc::now(), ttl)
            .map_err(|e| {
                WorkflowError::StageFailed(stage_id.to_string(), format!("dedup failed: {e}"))
            })?;
        let unseen_count = unseen.len();

        let mut deltas = HashMap::new();
        deltas.insert(ds.into.clone(), Value::Array(unseen));
        deltas.insert(
            format!("stage.{stage_id}.unseen_count"),
            Value::from(unseen_count),
        );
        deltas.insert(format!("stage.{stage_id}.total_count"), Value::from(total));

        let answer = format!("{unseen_count}/{total} item(s) new");
        Ok((
            StageOutput::Dedup {
                total,
                unseen: unseen_count,
            },
            answer,
            deltas,
        ))
    }

    /// Run a `Deliver` step: fan the workflow's result out to N delivery
    /// sinks, sequentially, best-effort per sink. Each sink is a plain
    /// [`car_ir::ActionProposal`] executed through the same proposal path a
    /// `proposal` step uses (so a messaging sink is "a proposal invoking the
    /// messaging tool", not a car-messaging dependency). A failed sink records
    /// its error and the remaining sinks still run; the stage fails only when
    /// ALL sinks fail. When `payload_key` is set, the value at that state key
    /// is substituted for `{{payload}}` in every string value of each sink
    /// proposal before execution (mirroring ForEach's `{{item}}` templating).
    ///
    /// Used for nested delivers (loop/foreach bodies); a top-level Deliver
    /// stage goes through [`execute_deliver_into`](Self::execute_deliver_into)
    /// with a caller-owned slot so a stage timeout preserves partial results.
    async fn execute_deliver(
        &self,
        stage_id: &str,
        ds: &DeliverStep,
        wf_state: &HashMap<String, Value>,
    ) -> Result<StepOutcome, WorkflowError> {
        let slot: DeliverPartial = Arc::new(std::sync::Mutex::new(Vec::new()));
        self.execute_deliver_into(stage_id, ds, wf_state, &slot)
            .await
    }

    /// [`execute_deliver`](Self::execute_deliver) with the per-sink results
    /// accumulated into a caller-owned `slot` as each sink completes. Because
    /// the slot lives outside this (cancellable) future, a stage timeout that
    /// cancels the fan-out mid-way can still drain the slot and report which
    /// sinks fired — a Deliver's whole purpose is non-idempotent external
    /// effects, so losing that evidence would invite double-delivery on retry.
    ///
    /// There is no per-sink timeout here: bound an individual sink via
    /// `Action.timeout_ms` on the actions inside its proposal; `Stage.timeout_ms`
    /// bounds the whole fan-out.
    async fn execute_deliver_into(
        &self,
        stage_id: &str,
        ds: &DeliverStep,
        wf_state: &HashMap<String, Value>,
        slot: &DeliverPartial,
    ) -> Result<StepOutcome, WorkflowError> {
        // Resolve the payload once, up front. A string renders raw; other JSON
        // renders compact; a missing key renders as the empty string (matching
        // ForEach's leniency toward missing state).
        let payload: Option<String> = ds.payload_key.as_deref().map(|key| {
            wf_state
                .get(key)
                .map(render_item)
                .unwrap_or_default()
        });

        for sink in &ds.sinks {
            // Best-effort per sink includes templating: a sink whose payload
            // substitution fails records that failure and the remaining sinks
            // still run (it must not abort the fan-out or discard prior
            // results).
            let proposal = match &payload {
                Some(p) => match template_proposal_payload(sink, p) {
                    Ok(templated) => Some(templated),
                    Err(e) => {
                        let error = format!("payload templating failed: {e}");
                        warn!(stage = %stage_id, sink = %sink.id, error = %error, "delivery sink failed (continuing fan-out)");
                        push_sink_result(
                            slot,
                            DeliverSinkResult {
                                sink_id: sink.id.clone(),
                                ok: false,
                                result: None,
                                error: Some(error),
                            },
                        );
                        None
                    }
                },
                None => Some(sink.clone()),
            };
            let Some(proposal) = proposal else { continue };
            // Same execution path as a Proposal step: a fresh runtime sharing
            // this infra's state, log, and policies.
            let runtime = self.infra.make_runtime();
            let result = runtime.execute(&proposal).await;
            let ok = result.all_succeeded();
            let error = if ok {
                None
            } else {
                let errors: Vec<String> = result
                    .results
                    .iter()
                    .filter_map(|r| r.error.as_ref())
                    .cloned()
                    .collect();
                Some(errors.join("; "))
            };
            if let Some(e) = &error {
                warn!(stage = %stage_id, sink = %sink.id, error = %e, "delivery sink failed (continuing fan-out)");
            }
            push_sink_result(
                slot,
                DeliverSinkResult {
                    sink_id: sink.id.clone(),
                    ok,
                    result: Some(result),
                    error,
                },
            );
        }

        // Fan-out complete: take the accumulated results (the drain in `drive`
        // only fires on the error path, so no double-report).
        let results: Vec<DeliverSinkResult> =
            std::mem::take(&mut *slot.lock().unwrap_or_else(|p| p.into_inner()));

        let delivered = results.iter().filter(|r| r.ok).count();
        let failed = results.len() - delivered;

        // Best-effort fan-out: the stage fails only when every sink failed.
        if !results.is_empty() && delivered == 0 {
            let errors: Vec<String> = results
                .iter()
                .map(|r| {
                    format!(
                        "{}: {}",
                        r.sink_id,
                        r.error.as_deref().unwrap_or("unknown error")
                    )
                })
                .collect();
            let count = results.len();
            // Put the per-sink evidence back so the caller's error-path drain
            // surfaces it as the failed stage's output rather than Empty.
            *slot.lock().unwrap_or_else(|p| p.into_inner()) = results;
            return Err(WorkflowError::StageFailed(
                stage_id.to_string(),
                format!("all {count} delivery sink(s) failed: {}", errors.join("; ")),
            ));
        }

        let mut deltas = HashMap::new();
        deltas.insert(
            format!("stage.{stage_id}.delivered_count"),
            Value::from(delivered),
        );
        deltas.insert(format!("stage.{stage_id}.failed_count"), Value::from(failed));

        let answer = format!("delivered to {delivered}/{} sink(s)", results.len());
        Ok((StageOutput::Deliver { results }, answer, deltas))
    }

    /// Dispatch to the appropriate car-multi pattern.
    async fn execute_pattern(
        &self,
        step: &PatternStep,
        wf_state: &HashMap<String, Value>,
    ) -> Result<(StageOutput, String), WorkflowError> {
        // Re-anchor the step on the pinned overall goal so a long run can't drift.
        let anchored_task = match wf_state.get("goal").and_then(|g| g.as_str()) {
            Some(goal) if !goal.trim().is_empty() => {
                format!("Overall goal: {goal}\n\nCurrent step: {}", step.task)
            }
            _ => step.task.clone(),
        };
        let task = anchored_task.as_str();
        let runner = &self.runner;
        let infra = &self.infra;

        match step.pattern {
            PatternKind::SwarmParallel => {
                let mut swarm = Swarm::new(step.agents.clone(), SwarmMode::Parallel);
                if let Some(synth) = extract_synthesizer(&step.config, &step.agents) {
                    swarm = swarm.with_synthesizer(synth);
                }
                let r = swarm.run(task, runner, infra).await?;
                Ok((
                    StageOutput::Pattern {
                        outputs: r.outputs,
                        final_answer: r.final_summary.clone(),
                    },
                    r.final_summary,
                ))
            }
            PatternKind::SwarmSequential => {
                let swarm = Swarm::new(step.agents.clone(), SwarmMode::Sequential);
                let r = swarm.run(task, runner, infra).await?;
                Ok((
                    StageOutput::Pattern {
                        outputs: r.outputs,
                        final_answer: r.final_summary.clone(),
                    },
                    r.final_summary,
                ))
            }
            PatternKind::SwarmDebate => {
                let swarm = Swarm::new(step.agents.clone(), SwarmMode::Debate);
                let r = swarm.run(task, runner, infra).await?;
                Ok((
                    StageOutput::Pattern {
                        outputs: r.outputs,
                        final_answer: r.final_summary.clone(),
                    },
                    r.final_summary,
                ))
            }
            PatternKind::Pipeline => {
                let pipeline = Pipeline::new(step.agents.clone());
                let r = pipeline.run(task, runner, infra).await?;
                Ok((
                    StageOutput::Pattern {
                        outputs: r.stages,
                        final_answer: r.final_answer.clone(),
                    },
                    r.final_answer,
                ))
            }
            PatternKind::Supervisor => {
                let max_rounds = step
                    .config
                    .get("max_rounds")
                    .and_then(|v| v.as_u64())
                    .unwrap_or(3) as u32;
                let (supervisor, workers) = split_supervisor_workers(&step.agents, &step.config);
                let r = Supervisor::new(workers, supervisor)
                    .with_max_rounds(max_rounds)
                    .run(task, runner, infra)
                    .await?;
                let all_outputs: Vec<_> = r.rounds.into_iter().flatten().collect();
                Ok((
                    StageOutput::Pattern {
                        outputs: all_outputs,
                        final_answer: r.final_answer.clone(),
                    },
                    r.final_answer,
                ))
            }
            PatternKind::Delegator => {
                let (main_agent, specialists) = split_delegator(&step.agents, &step.config);
                let delegator = car_multi::Delegator::new(main_agent, specialists);
                let r = delegator.run(task, runner, infra).await?;
                Ok((
                    StageOutput::Pattern {
                        outputs: vec![car_multi::AgentOutput {
                            name: "delegator".into(),
                            answer: r.final_answer.clone(),
                            turns: 0,
                            tool_calls: r.delegations.len() as u32,
                            duration_ms: 0.0,
                            error: None,
                            outcome: None,
                            tokens: None,
                            tools_used: Vec::new(),
                        }],
                        final_answer: r.final_answer.clone(),
                    },
                    r.final_answer,
                ))
            }
            PatternKind::MapReduce => {
                let max_concurrent = step
                    .config
                    .get("max_concurrent")
                    .and_then(|v| v.as_u64())
                    .unwrap_or(5) as usize;
                let items: Vec<String> = step
                    .config
                    .get("items")
                    .and_then(|v| serde_json::from_value(v.clone()).ok())
                    .unwrap_or_default();

                if step.agents.len() < 2 {
                    return Err(WorkflowError::StageFailed(
                        "map_reduce".into(),
                        "requires at least 2 agents (mapper + reducer)".into(),
                    ));
                }
                let mapper = step.agents[0].clone();
                let reducer = step.agents[1].clone();

                let mr = MapReduce::new(mapper, reducer).with_max_concurrent(max_concurrent);
                let r = mr
                    .run(
                        task,
                        &items
                            .iter()
                            .map(|s| s.as_str())
                            .collect::<Vec<_>>()
                            .iter()
                            .map(|s| s.to_string())
                            .collect::<Vec<_>>(),
                        runner,
                        infra,
                    )
                    .await?;
                Ok((
                    StageOutput::Pattern {
                        outputs: r.map_outputs,
                        final_answer: r.reduced_answer.clone(),
                    },
                    r.reduced_answer,
                ))
            }
            PatternKind::Vote => {
                let mut vote = Vote::new(step.agents.clone());
                if let Some(synth) = extract_synthesizer(&step.config, &step.agents) {
                    vote = vote.with_synthesizer(synth);
                }
                let r = vote.run(task, runner, infra).await?;
                Ok((
                    StageOutput::Pattern {
                        outputs: r.votes,
                        final_answer: r.winner.clone(),
                    },
                    r.winner,
                ))
            }
            PatternKind::Fleet => {
                let mut fleet = Fleet::new(step.agents.clone());
                if let Some(timeout) = step.config.get("timeout_secs").and_then(|v| v.as_u64()) {
                    fleet = fleet.with_timeout(timeout);
                }
                let r = fleet.run(runner, infra).await?;
                let summary = format!("{} succeeded, {} failed", r.succeeded, r.failed);
                Ok((
                    StageOutput::Pattern {
                        outputs: r.outputs,
                        final_answer: summary.clone(),
                    },
                    summary,
                ))
            }
            PatternKind::AdversarialReview => {
                if step.agents.is_empty() {
                    return Err(WorkflowError::StageFailed(
                        "adversarial_review".into(),
                        "requires a reviewer agent (agents[0])".into(),
                    ));
                }
                let reviewer = step.agents[0].clone();
                let criteria: Vec<String> = step
                    .config
                    .get("criteria")
                    .and_then(|v| serde_json::from_value(v.clone()).ok())
                    .unwrap_or_default();
                // The work to review comes from state (produced by a prior stage).
                let work = step
                    .config
                    .get("review_key")
                    .and_then(|v| v.as_str())
                    .and_then(|key| wf_state.get(key))
                    .map(|v| match v {
                        Value::String(s) => s.clone(),
                        other => other.to_string(),
                    })
                    .unwrap_or_default();

                // Fail closed on empty work: a missing/empty review_key must NOT
                // produce a vacuous PASS (the very "premature done" failure this
                // gate defends against), and skips a wasted reviewer call.
                if work.trim().is_empty() {
                    let answer =
                        "FAIL (no work to review — missing or empty review_key)".to_string();
                    return Ok((
                        StageOutput::Review {
                            passed: false,
                            blocker_count: 1,
                            // A definitive fail (no work), not an unparseable review.
                            inconclusive: false,
                            findings: vec![],
                            reviewer: car_multi::AgentOutput {
                                name: "reviewer".into(),
                                answer: answer.clone(),
                                turns: 0,
                                tool_calls: 0,
                                duration_ms: 0.0,
                                error: None,
                                outcome: None,
                                tokens: None,
                                tools_used: Vec::new(),
                            },
                        },
                        answer,
                    ));
                }

                let r = AdversarialReview::new(reviewer, criteria)
                    .run(&work, runner, infra)
                    .await?;
                // Answer carries a human-readable summary; the typed verdict is
                // surfaced as `stage.<id>.review_passed` by the caller.
                let answer = if r.inconclusive {
                    "INCONCLUSIVE (reviewer output unparseable; not verified)".to_string()
                } else if r.passed {
                    format!("PASS ({} finding(s))", r.findings.len())
                } else {
                    format!("FAIL ({} blocker(s))", r.blocker_count)
                };
                Ok((
                    StageOutput::Review {
                        passed: r.passed,
                        blocker_count: r.blocker_count,
                        inconclusive: r.inconclusive,
                        findings: r.findings,
                        reviewer: r.reviewer_output,
                    },
                    answer,
                ))
            }
            PatternKind::Tournament => {
                if step.agents.len() < 3 {
                    return Err(WorkflowError::StageFailed(
                        "tournament".into(),
                        "requires at least 3 agents (>=2 competitors + 1 judge)".into(),
                    ));
                }
                let judge_idx =
                    step.config
                        .get("judge_index")
                        .and_then(|v| v.as_u64())
                        .unwrap_or(step.agents.len() as u64 - 1) as usize;
                let judge = step
                    .agents
                    .get(judge_idx)
                    .cloned()
                    .unwrap_or_else(|| step.agents.last().unwrap().clone());
                let competitors: Vec<_> = step
                    .agents
                    .iter()
                    .enumerate()
                    .filter(|(i, _)| *i != judge_idx)
                    .map(|(_, a)| a.clone())
                    .collect();

                let r = Tournament::new(competitors, judge)
                    .run(task, runner, infra)
                    .await?;
                Ok((
                    StageOutput::Pattern {
                        outputs: r.candidates,
                        final_answer: r.winner_answer.clone(),
                    },
                    r.winner_answer,
                ))
            }
        }
    }

    /// Run one ForEach body iteration. Returns the body's output, answer, and
    /// state deltas. The deltas are NOT merged into shared state under their own
    /// keys — concurrent items would clobber each other — but the caller
    /// re-namespaces them under `foreach.<id>.<index>.<key>` so per-item
    /// structured output is still recoverable downstream.
    async fn run_foreach_body(
        &self,
        stage_id: &str,
        index: usize,
        body: &StageStep,
        wf_state: &HashMap<String, Value>,
    ) -> Result<StepOutcome, WorkflowError> {
        let child_id = format!("{stage_id}.{index}");
        self.execute_step(&child_id, body, wf_state).await
    }

    /// Execute a proposal step via car-engine.
    async fn execute_proposal(
        &self,
        step: &ProposalStep,
    ) -> Result<(StageOutput, String), WorkflowError> {
        let runtime = self.infra.make_runtime();
        let result = runtime.execute(&step.proposal).await;

        if result.all_succeeded() {
            let answer = result
                .results
                .last()
                .and_then(|r| r.output.as_ref())
                .map(|v| v.to_string())
                .unwrap_or_default();
            Ok((StageOutput::Proposal { result }, answer))
        } else {
            let errors: Vec<String> = result
                .results
                .iter()
                .filter_map(|r| r.error.as_ref())
                .cloned()
                .collect();
            Err(WorkflowError::StageFailed(
                "proposal".into(),
                errors.join("; "),
            ))
        }
    }

    /// Execute a nested sub-workflow.
    async fn execute_sub_workflow(
        &self,
        step: &SubWorkflowStep,
    ) -> Result<(StageOutput, String), WorkflowError> {
        let result = self.run(&step.workflow).await?;

        // Approval gates inside sub-workflows would require nested pause/resume
        // plumbing the parent run cannot yet thread back to the caller. Fail
        // explicitly rather than silently treating a pause as a failure.
        // NOTE: surfacing a nested pause to the parent run would need a
        // checkpoint *stack* (parent position + child checkpoint). Until that
        // exists we fail explicitly. This currently flows as a stage failure,
        // which triggers the parent's saga compensation — i.e. a sub-workflow
        // gate rolls the parent back. Documented limitation.
        if result.is_paused() {
            return Err(WorkflowError::ApprovalInSubWorkflow(
                step.workflow.id.clone(),
            ));
        }

        let answer = result
            .stages
            .last()
            .and_then(|s| match &s.output {
                StageOutput::Pattern { final_answer, .. } => Some(final_answer.clone()),
                StageOutput::Proposal { result } => result
                    .results
                    .last()
                    .and_then(|r| r.output.as_ref())
                    .map(|v| v.to_string()),
                StageOutput::SubWorkflow { result } => Some(format!(
                    "sub-workflow {} {}",
                    result.workflow_name,
                    if result.succeeded() {
                        "completed"
                    } else {
                        "failed"
                    }
                )),
                StageOutput::Approval { response } => Some(response.to_string()),
                StageOutput::Review {
                    passed,
                    blocker_count,
                    inconclusive,
                    ..
                } => Some(format!(
                    "review {}",
                    if *inconclusive {
                        "inconclusive (not verified)".to_string()
                    } else if *passed {
                        "passed".to_string()
                    } else {
                        format!("failed ({blocker_count} blocker(s))")
                    }
                )),
                StageOutput::Loop {
                    iterations,
                    satisfied,
                    ..
                } => Some(format!(
                    "loop ran {} iteration(s), until {}satisfied",
                    iterations,
                    if *satisfied { "" } else { "not " }
                )),
                StageOutput::ForEach { items, .. } => {
                    Some(format!("foreach over {} item(s)", items.len()))
                }
                StageOutput::Dedup { total, unseen } => {
                    Some(format!("dedup: {unseen}/{total} new"))
                }
                StageOutput::Deliver { results } => Some(format!(
                    "delivered to {}/{} sink(s)",
                    results.iter().filter(|r| r.ok).count(),
                    results.len()
                )),
                StageOutput::Empty => None,
            })
            .unwrap_or_default();

        if result.succeeded() {
            Ok((
                StageOutput::SubWorkflow {
                    result: Box::new(result),
                },
                answer,
            ))
        } else {
            Err(WorkflowError::StageFailed(
                "sub_workflow".into(),
                "sub-workflow failed".into(),
            ))
        }
    }

    /// Run saga compensation in reverse order of completed stages.
    async fn compensate(
        &self,
        workflow: &Workflow,
        completed_stage_ids: &[String],
    ) -> Vec<CompensationResult> {
        let mut results = Vec::new();

        for stage_id in completed_stage_ids.iter().rev() {
            let stage = match workflow.stage(stage_id) {
                Some(s) => s,
                None => continue,
            };

            let handler = match &stage.compensation {
                Some(h) => h,
                None => continue,
            };

            debug!(stage = %stage_id, "running compensation");
            let comp_start = Instant::now();

            let comp_result = match handler {
                CompensationHandler::Proposal(ps) => self.execute_proposal(ps).await,
                CompensationHandler::StageRef { stage_id: ref_id } => {
                    if let Some(ref_stage) = workflow.stage(ref_id) {
                        self.execute_step(ref_id, &ref_stage.step, &HashMap::new())
                            .await
                            .map(|(out, answer, _deltas)| (out, answer))
                    } else {
                        Err(WorkflowError::StageNotFound(ref_id.clone()))
                    }
                }
            };

            let duration = comp_start.elapsed().as_secs_f64() * 1000.0;

            match comp_result {
                Ok(_) => {
                    results.push(CompensationResult {
                        for_stage_id: stage_id.clone(),
                        status: StageStatus::Succeeded,
                        duration_ms: duration,
                        error: None,
                    });
                }
                Err(e) => {
                    warn!(stage = %stage_id, error = %e, "compensation failed");
                    results.push(CompensationResult {
                        for_stage_id: stage_id.clone(),
                        status: StageStatus::Failed,
                        duration_ms: duration,
                        error: Some(e.to_string()),
                    });
                }
            }
        }

        results
    }
}

/// Mutable position while driving the stage graph. Captured into a
/// [`PausedWorkflow`] at an approval gate and restored on resume.
struct Cursor {
    wf_state: HashMap<String, Value>,
    stage_results: Vec<StageResult>,
    completed_stage_ids: Vec<String>,
    iterations: u32,
    /// Wall time of executed segments before the current `drive` call (carried
    /// across pause/resume so `duration_ms` reflects total compute, not just the
    /// post-resume segment). Excludes the human wait.
    prior_duration_ms: f64,
    current_id: String,
}

/// Generate a stable run identifier. This is both the resume token and the
/// checkpoint filename, so use the full 122-bit UUID — no truncation.
fn new_run_id() -> String {
    uuid::Uuid::new_v4().simple().to_string()
}

/// Default directory backing `Dedup` seen-sets when the engine wasn't given an
/// explicit one: `~/.car/workflow-dedup` (matching the `~/.car/...` convention
/// used by the scheduler, supervisor, and checkpoint stores).
fn default_dedup_dir() -> std::path::PathBuf {
    match std::env::var_os("HOME").or_else(|| std::env::var_os("USERPROFILE")) {
        Some(home) => std::path::PathBuf::from(home),
        None => {
            // Falling back to cwd makes the seen-set cwd-relative, so the same
            // automation launched from a different directory would reprocess
            // everything. Warn rather than silently regress the cross-run
            // guarantee; pin an explicit dir via `with_dedup_dir` to avoid this.
            warn!("no HOME/USERPROFILE set; dedup seen-set falls back to cwd-relative ./.car/workflow-dedup");
            std::path::PathBuf::from(".")
        }
    }
    .join(".car")
    .join("workflow-dedup")
}

/// Enforce an [`ApprovalStep`]'s form contract against the human's response.
fn validate_approval_input(
    fields: &[ApprovalField],
    input: &HashMap<String, Value>,
) -> Result<(), WorkflowError> {
    for field in fields {
        match input.get(&field.name) {
            None | Some(Value::Null) => {
                if field.required {
                    return Err(WorkflowError::InvalidApprovalInput(format!(
                        "required field '{}' is missing",
                        field.name
                    )));
                }
            }
            Some(value) => {
                if field.field_type == "options" && !field.options.is_empty() {
                    let ok = value
                        .as_str()
                        .map(|s| field.options.iter().any(|o| o == s))
                        .unwrap_or(false);
                    if !ok {
                        return Err(WorkflowError::InvalidApprovalInput(format!(
                            "field '{}' value {} is not one of {:?}",
                            field.name, value, field.options
                        )));
                    }
                }
            }
        }
    }
    Ok(())
}

/// Pick the next stage after `from`: the first outgoing edge whose conditions
/// all pass. `None` means `from` is terminal.
fn next_stage(workflow: &Workflow, from: &str, state: &HashMap<String, Value>) -> Option<String> {
    workflow
        .outgoing_edges(from)
        .iter()
        .find(|e| check_conditions(&e.conditions, state))
        .map(|e| e.to.clone())
}

/// Build a `Completed` workflow result.
fn completed_result(
    workflow: &Workflow,
    stage_results: Vec<StageResult>,
    final_state: HashMap<String, Value>,
    duration_ms: f64,
) -> WorkflowResult {
    WorkflowResult {
        workflow_id: workflow.id.clone(),
        workflow_name: workflow.name.clone(),
        status: WorkflowStatus::Completed,
        stages: stage_results,
        compensations: vec![],
        duration_ms,
        timestamp: chrono::Utc::now(),
        final_state,
        paused: None,
    }
}

// --- ForEach item templating ---

/// Render a ForEach item to the string substituted for `{{item}}`. A JSON
/// string uses its raw value (no quotes); anything else uses its compact JSON.
fn render_item(item: &Value) -> String {
    match item {
        Value::String(s) => s.clone(),
        other => other.to_string(),
    }
}

/// Clone `body` with `{{item}}`/`{{index}}` substituted in every string. Done
/// via a serde round-trip so it covers pattern tasks, agent prompts, and
/// proposal parameters uniformly without per-variant code.
fn template_step(body: &StageStep, item: &str, index: usize) -> Result<StageStep, WorkflowError> {
    let mut v = serde_json::to_value(body).map_err(|e| {
        WorkflowError::StageFailed("foreach".into(), format!("serialize body: {e}"))
    })?;
    substitute_in_value(&mut v, item, index);
    serde_json::from_value(v).map_err(|e| {
        WorkflowError::StageFailed("foreach".into(), format!("rebuild templated body: {e}"))
    })
}

fn substitute_in_value(v: &mut Value, item: &str, index: usize) {
    match v {
        Value::String(s) => {
            if s.contains("{{item}}") || s.contains("{{index}}") {
                *s = s
                    .replace("{{item}}", item)
                    .replace("{{index}}", &index.to_string());
            }
        }
        Value::Array(a) => a
            .iter_mut()
            .for_each(|e| substitute_in_value(e, item, index)),
        Value::Object(m) => m
            .values_mut()
            .for_each(|e| substitute_in_value(e, item, index)),
        _ => {}
    }
}

// --- Deliver payload templating ---

/// Shared accumulator for a Deliver fan-out's per-sink results. Owned by the
/// `drive` frame (not the cancellable step future) so a stage timeout can
/// still drain and report the sinks that already fired.
type DeliverPartial = Arc<std::sync::Mutex<Vec<DeliverSinkResult>>>;

/// Record one sink's outcome into the shared slot. A poisoned lock (a panic
/// in another holder) must not lose delivery evidence — take the inner value.
fn push_sink_result(slot: &DeliverPartial, result: DeliverSinkResult) {
    slot.lock().unwrap_or_else(|p| p.into_inner()).push(result);
}

/// Clone a sink proposal with `{{payload}}` substituted in every string value
/// (action parameters, context, …). Same serde-round-trip technique as
/// [`template_step`], so it covers the whole proposal uniformly. JSON object
/// *keys* are not templated. Errors are plain strings: the caller records
/// them per sink (best-effort fan-out), not as a stage failure.
fn template_proposal_payload(
    proposal: &car_ir::ActionProposal,
    payload: &str,
) -> Result<car_ir::ActionProposal, String> {
    let mut v = serde_json::to_value(proposal)
        .map_err(|e| format!("serialize sink proposal: {e}"))?;
    substitute_payload_in_value(&mut v, payload);
    serde_json::from_value(v).map_err(|e| format!("rebuild templated sink: {e}"))
}

fn substitute_payload_in_value(v: &mut Value, payload: &str) {
    match v {
        Value::String(s) => {
            if s.contains("{{payload}}") {
                *s = s.replace("{{payload}}", payload);
            }
        }
        Value::Array(a) => a
            .iter_mut()
            .for_each(|e| substitute_payload_in_value(e, payload)),
        Value::Object(m) => m
            .values_mut()
            .for_each(|e| substitute_payload_in_value(e, payload)),
        _ => {}
    }
}

// --- Precondition evaluation ---

/// Evaluate edge conditions against workflow state. Returns true if all conditions pass.
fn check_conditions(conditions: &[car_ir::Precondition], state: &HashMap<String, Value>) -> bool {
    conditions
        .iter()
        .all(|cond| evaluate_precondition(cond, state))
}

/// Evaluate a single precondition against a state map.
fn evaluate_precondition(cond: &car_ir::Precondition, state: &HashMap<String, Value>) -> bool {
    let op = cond.operator.as_str();

    match op {
        "exists" => state.contains_key(&cond.key),
        "not_exists" => !state.contains_key(&cond.key),
        _ => {
            let actual = match state.get(&cond.key) {
                Some(v) => v,
                None => return false, // key missing, condition fails
            };
            match op {
                "eq" => actual == &cond.value,
                "neq" => actual != &cond.value,
                "gt" => compare_values(actual, &cond.value)
                    .map_or(false, |o| o == std::cmp::Ordering::Greater),
                "gte" => compare_values(actual, &cond.value)
                    .map_or(false, |o| o != std::cmp::Ordering::Less),
                "lt" => compare_values(actual, &cond.value)
                    .map_or(false, |o| o == std::cmp::Ordering::Less),
                "lte" => compare_values(actual, &cond.value)
                    .map_or(false, |o| o != std::cmp::Ordering::Greater),
                "contains" => {
                    if let (Some(haystack), Some(needle)) = (actual.as_str(), cond.value.as_str()) {
                        haystack.contains(needle)
                    } else {
                        false
                    }
                }
                _ => false,
            }
        }
    }
}

fn compare_values(a: &Value, b: &Value) -> Option<std::cmp::Ordering> {
    match (a.as_f64(), b.as_f64()) {
        (Some(a), Some(b)) => a.partial_cmp(&b),
        _ => match (a.as_str(), b.as_str()) {
            (Some(a), Some(b)) => Some(a.cmp(b)),
            _ => None,
        },
    }
}

// --- Pattern config helpers ---

/// Extract synthesizer agent from config (used by Swarm and Vote).
fn extract_synthesizer(
    config: &HashMap<String, Value>,
    agents: &[car_multi::AgentSpec],
) -> Option<car_multi::AgentSpec> {
    config
        .get("synthesizer_index")
        .and_then(|v| v.as_u64())
        .and_then(|i| agents.get(i as usize))
        .cloned()
}

/// Split agents into supervisor + workers. By default, last agent is supervisor.
fn split_supervisor_workers(
    agents: &[car_multi::AgentSpec],
    config: &HashMap<String, Value>,
) -> (car_multi::AgentSpec, Vec<car_multi::AgentSpec>) {
    let idx = config
        .get("supervisor_index")
        .and_then(|v| v.as_u64())
        .unwrap_or(agents.len().saturating_sub(1) as u64) as usize;

    let supervisor = agents
        .get(idx)
        .cloned()
        .unwrap_or_else(|| agents.last().unwrap().clone());
    let workers: Vec<_> = agents
        .iter()
        .enumerate()
        .filter(|(i, _)| *i != idx)
        .map(|(_, a)| a.clone())
        .collect();
    (supervisor, workers)
}

/// Split agents into main + specialists map for Delegator.
fn split_delegator(
    agents: &[car_multi::AgentSpec],
    _config: &HashMap<String, Value>,
) -> (car_multi::AgentSpec, HashMap<String, car_multi::AgentSpec>) {
    let main = agents
        .first()
        .cloned()
        .unwrap_or_else(|| car_multi::AgentSpec::new("main", ""));
    let specialists: HashMap<String, car_multi::AgentSpec> = agents
        .iter()
        .skip(1)
        .map(|a| (a.name.clone(), a.clone()))
        .collect();
    (main, specialists)
}

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

    #[test]
    fn precondition_eq() {
        let mut state = HashMap::new();
        state.insert("x".into(), Value::Bool(true));

        let cond = car_ir::Precondition {
            key: "x".into(),
            operator: "eq".into(),
            value: Value::Bool(true),
            description: String::new(),
        };
        assert!(evaluate_precondition(&cond, &state));

        let cond_false = car_ir::Precondition {
            key: "x".into(),
            operator: "eq".into(),
            value: Value::Bool(false),
            description: String::new(),
        };
        assert!(!evaluate_precondition(&cond_false, &state));
    }

    #[test]
    fn precondition_exists() {
        let mut state = HashMap::new();
        state.insert("x".into(), Value::Null);

        let exists = car_ir::Precondition {
            key: "x".into(),
            operator: "exists".into(),
            value: Value::Null,
            description: String::new(),
        };
        assert!(evaluate_precondition(&exists, &state));

        let not_exists = car_ir::Precondition {
            key: "y".into(),
            operator: "exists".into(),
            value: Value::Null,
            description: String::new(),
        };
        assert!(!evaluate_precondition(&not_exists, &state));
    }

    #[test]
    fn precondition_numeric_comparison() {
        let mut state = HashMap::new();
        state.insert("count".into(), serde_json::json!(5));

        let gt = car_ir::Precondition {
            key: "count".into(),
            operator: "gt".into(),
            value: serde_json::json!(3),
            description: String::new(),
        };
        assert!(evaluate_precondition(&gt, &state));

        let lt = car_ir::Precondition {
            key: "count".into(),
            operator: "lt".into(),
            value: serde_json::json!(3),
            description: String::new(),
        };
        assert!(!evaluate_precondition(&lt, &state));
    }

    #[test]
    fn empty_conditions_always_pass() {
        let state = HashMap::new();
        assert!(check_conditions(&[], &state));
    }

    // --- Approval gate (HITL pause/resume) ---

    use car_ir::ActionProposal;

    /// A runner that must never be invoked by these tests (approval gates and
    /// empty proposals don't drive agents).
    struct NoopRunner;

    #[async_trait::async_trait]
    impl car_multi::AgentRunner for NoopRunner {
        async fn run(
            &self,
            _spec: &car_multi::AgentSpec,
            _task: &str,
            _runtime: &car_engine::Runtime,
            _mailbox: &car_multi::Mailbox,
        ) -> Result<car_multi::AgentOutput, car_multi::MultiError> {
            Err(car_multi::MultiError::NoOutput)
        }
    }

    fn test_engine() -> WorkflowEngine {
        WorkflowEngine::new(Arc::new(NoopRunner), car_multi::SharedInfra::new())
    }

    fn approval_stage(id: &str, output_key: &str) -> Stage {
        Stage {
            id: id.into(),
            name: id.into(),
            step: StageStep::Approval(ApprovalStep {
                prompt: "approve?".into(),
                fields: vec![],
                output_key: output_key.into(),
            }),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        }
    }

    fn empty_proposal_stage(id: &str) -> Stage {
        Stage {
            id: id.into(),
            name: id.into(),
            step: StageStep::Proposal(ProposalStep {
                proposal: ActionProposal {
                    id: format!("p-{id}"),
                    source: "test".into(),
                    actions: vec![],
                    timestamp: chrono::Utc::now(),
                    context: HashMap::new(),
                },
            }),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        }
    }

    fn edge(from: &str, to: &str, conditions: Vec<car_ir::Precondition>) -> Edge {
        Edge {
            from: from.into(),
            to: to.into(),
            conditions,
            label: String::new(),
        }
    }

    #[tokio::test]
    async fn pauses_at_approval_gate_without_executing_it() {
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "gate".into(),
            goal: None,
            stages: vec![
                approval_stage("gate", "approval"),
                empty_proposal_stage("done"),
            ],
            edges: vec![edge("gate", "done", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };

        let res = test_engine().run(&wf).await.unwrap();

        assert_eq!(res.status, WorkflowStatus::Paused);
        assert!(res.is_paused());
        assert!(
            res.stages.is_empty(),
            "gate body must not run before resume"
        );

        let paused = res.paused.expect("checkpoint present when paused");
        assert_eq!(paused.paused_stage_id, "gate");
        assert_eq!(paused.output_key, "approval");
        assert_eq!(paused.prompt, "approve?");
        assert!(!paused.run_id.is_empty());
    }

    #[tokio::test]
    async fn resume_records_response_and_completes() {
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "gate".into(),
            goal: None,
            stages: vec![
                approval_stage("gate", "approval"),
                empty_proposal_stage("done"),
            ],
            edges: vec![edge("gate", "done", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };
        let eng = test_engine();
        let paused = eng.run(&wf).await.unwrap().paused.unwrap();

        let mut input = HashMap::new();
        input.insert("decision".to_string(), Value::String("approve".into()));
        let res = eng.resume(paused, input).await.unwrap();

        assert_eq!(res.status, WorkflowStatus::Completed);
        assert!(res.succeeded());
        // Both the gate and the downstream stage are recorded, gate first.
        let ran: Vec<&str> = res.stages.iter().map(|s| s.stage_id.as_str()).collect();
        assert_eq!(ran, vec!["gate", "done"]);
        assert!(matches!(res.stages[0].output, StageOutput::Approval { .. }));
        // Response is flattened for edge conditions and mirrored to the answer slot.
        assert_eq!(
            res.final_state.get("approval.decision"),
            Some(&Value::String("approve".into()))
        );
        assert!(res.final_state.contains_key("stage.gate.answer"));
        assert!(res.final_state.contains_key("approval"));
    }

    #[tokio::test]
    async fn resume_branches_on_answer_after_checkpoint_roundtrip() {
        // gate routes to `approved` when approval.decision == "approve",
        // otherwise falls through the unconditional edge to `revise`.
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "gate".into(),
            goal: None,
            stages: vec![
                approval_stage("gate", "approval"),
                empty_proposal_stage("approved"),
                empty_proposal_stage("revise"),
            ],
            edges: vec![
                edge(
                    "gate",
                    "approved",
                    vec![car_ir::Precondition {
                        key: "approval.decision".into(),
                        operator: "eq".into(),
                        value: Value::String("approve".into()),
                        description: String::new(),
                    }],
                ),
                edge("gate", "revise", vec![]),
            ],
            max_iterations: 100,
            metadata: HashMap::new(),
        };
        let eng = test_engine();
        let paused = eng.run(&wf).await.unwrap().paused.unwrap();

        // Simulate a process restart: persist and reload the checkpoint.
        let dir =
            std::env::temp_dir().join(format!("car-wf-resume-{}", uuid::Uuid::new_v4().simple()));
        let store = crate::CheckpointStore::open(&dir).unwrap();
        store.save(&paused).unwrap();
        let run_id = paused.run_id.clone();
        let reloaded = store.load(&run_id).unwrap().expect("checkpoint reloads");

        let mut input = HashMap::new();
        input.insert("decision".to_string(), Value::String("approve".into()));
        let res = eng.resume(reloaded, input).await.unwrap();

        assert_eq!(res.status, WorkflowStatus::Completed);
        let ran: Vec<&str> = res.stages.iter().map(|s| s.stage_id.as_str()).collect();
        assert!(
            ran.contains(&"approved"),
            "approve answer routes to approved"
        );
        assert!(!ran.contains(&"revise"), "revise branch must be skipped");

        store.remove(&run_id).unwrap();
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn iteration_guard_counts_each_stage_once_across_pause() {
        // A (proposal) -> gate. Reaching the gate should cost exactly 2 ticks,
        // not 4: resume advances past the gate before re-driving, so the gate is
        // not double-counted. (Refutes the "gate costs two ticks" concern.)
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "a".into(),
            goal: None,
            stages: vec![
                empty_proposal_stage("a"),
                approval_stage("gate", "approval"),
                empty_proposal_stage("done"),
            ],
            edges: vec![edge("a", "gate", vec![]), edge("gate", "done", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };
        let eng = test_engine();
        let paused = eng.run(&wf).await.unwrap().paused.unwrap();
        // a = tick 1, gate = tick 2.
        assert_eq!(paused.iterations, 2);

        let mut input = HashMap::new();
        input.insert("decision".to_string(), Value::String("ok".into()));
        let res = eng.resume(paused, input).await.unwrap();
        assert_eq!(res.status, WorkflowStatus::Completed);
        // a, gate, done all executed exactly once.
        let ran: Vec<&str> = res.stages.iter().map(|s| s.stage_id.as_str()).collect();
        assert_eq!(ran, vec!["a", "gate", "done"]);
    }

    #[tokio::test]
    async fn resume_rejects_missing_required_field() {
        let mut gate = approval_stage("gate", "approval");
        if let StageStep::Approval(ap) = &mut gate.step {
            ap.fields = vec![ApprovalField {
                name: "decision".into(),
                label: "Decision".into(),
                field_type: "text".into(),
                options: vec![],
                required: true,
            }];
        }
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "gate".into(),
            goal: None,
            stages: vec![gate, empty_proposal_stage("done")],
            edges: vec![edge("gate", "done", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };
        let eng = test_engine();
        let paused = eng.run(&wf).await.unwrap().paused.unwrap();
        // Empty input omits the required field.
        let err = eng.resume(paused, HashMap::new()).await.unwrap_err();
        assert!(matches!(err, WorkflowError::InvalidApprovalInput(_)));
    }

    #[tokio::test]
    async fn resume_rejects_checkpoint_pointing_at_non_approval_stage() {
        // Forge a checkpoint whose paused_stage_id names a proposal stage.
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "work".into(),
            goal: None,
            stages: vec![empty_proposal_stage("work")],
            edges: vec![],
            max_iterations: 100,
            metadata: HashMap::new(),
        };
        let forged = PausedWorkflow {
            run_id: "forged".into(),
            workflow: wf,
            paused_stage_id: "work".into(),
            prompt: String::new(),
            fields: vec![],
            output_key: "x".into(),
            wf_state: HashMap::new(),
            stage_results: vec![],
            completed_stage_ids: vec![],
            iterations: 1,
            prior_duration_ms: 0.0,
            created_at: chrono::Utc::now(),
        };
        let eng = test_engine();
        let err = eng.resume(forged, HashMap::new()).await.unwrap_err();
        assert!(matches!(err, WorkflowError::InvalidResume(_)));
    }

    // --- LoopUntil / ForEach (dynamic workflow IR) ---

    fn state_write_action(key: &str, value: Value) -> car_ir::Action {
        car_ir::Action {
            id: format!("sw-{key}"),
            action_type: car_ir::ActionType::StateWrite,
            tool: None,
            parameters: [
                ("key".to_string(), Value::from(key)),
                ("value".to_string(), value),
            ]
            .into(),
            preconditions: vec![],
            expected_effects: HashMap::new(),
            state_dependencies: vec![],
            read_set: vec![],
            write_set: vec![],
            assumptions: vec![],
            invocation_mode: Default::default(),
            idempotent: false,
            max_retries: 0,
            failure_behavior: car_ir::FailureBehavior::Abort,
            timeout_ms: None,
            metadata: HashMap::new(),
        }
    }

    fn proposal_step_writing(key: &str, value: Value) -> StageStep {
        StageStep::Proposal(ProposalStep {
            proposal: ActionProposal {
                id: format!("p-{key}"),
                source: "test".into(),
                actions: vec![state_write_action(key, value)],
                timestamp: chrono::Utc::now(),
                context: HashMap::new(),
            },
        })
    }

    fn single_stage_wf(id: &str, step: StageStep) -> Workflow {
        Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: id.into(),
            goal: None,
            stages: vec![Stage {
                id: id.into(),
                name: id.into(),
                step,
                compensation: None,
                timeout_ms: None,
                metadata: HashMap::new(),
            }],
            edges: vec![],
            max_iterations: 100,
            metadata: HashMap::new(),
        }
    }

    #[tokio::test]
    async fn loop_until_empty_predicate_runs_to_cap() {
        // Empty `until` => runs exactly max_iterations; satisfied stays false.
        let step = StageStep::LoopUntil(LoopUntilStep {
            body: Box::new(empty_proposal_stage("ignored").step),
            until: vec![],
            max_iterations: 3,
        });
        let wf = single_stage_wf("loop", step);
        let res = test_engine().run(&wf).await.unwrap();

        assert_eq!(res.status, WorkflowStatus::Completed);
        match &res.stages[0].output {
            StageOutput::Loop {
                iterations,
                satisfied,
                iterations_output,
            } => {
                assert_eq!(*iterations, 3);
                assert!(!*satisfied);
                assert_eq!(iterations_output.len(), 3);
            }
            other => panic!("expected Loop output, got {other:?}"),
        }
        assert_eq!(
            res.final_state.get("stage.loop.iteration"),
            Some(&Value::from(3u32))
        );
    }

    #[tokio::test]
    async fn loop_until_stops_when_predicate_satisfied() {
        // Body writes done=true; until checks it => stops after one iteration.
        let step = StageStep::LoopUntil(LoopUntilStep {
            body: Box::new(proposal_step_writing("done", Value::Bool(true))),
            until: vec![car_ir::Precondition {
                key: "done".into(),
                operator: "eq".into(),
                value: Value::Bool(true),
                description: String::new(),
            }],
            max_iterations: 10,
        });
        let wf = single_stage_wf("loop", step);
        let res = test_engine().run(&wf).await.unwrap();

        match &res.stages[0].output {
            StageOutput::Loop {
                iterations,
                satisfied,
                ..
            } => {
                assert_eq!(*iterations, 1, "predicate true after first body run");
                assert!(*satisfied);
            }
            other => panic!("expected Loop output, got {other:?}"),
        }
        // The body's state write is merged into workflow state.
        assert_eq!(res.final_state.get("done"), Some(&Value::Bool(true)));
    }

    #[tokio::test]
    async fn for_each_runs_body_per_runtime_item() {
        // An upstream stage publishes the item list; ForEach fans over it.
        let seed = Stage {
            id: "seed".into(),
            name: "seed".into(),
            step: proposal_step_writing("files", serde_json::json!(["a.rs", "b.rs", "c.rs"])),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        };
        let fan = Stage {
            id: "fan".into(),
            name: "fan".into(),
            step: StageStep::ForEach(ForEachStep {
                items_from: "files".into(),
                // Body writes the item into a per-run key, proving templating.
                body: Box::new(proposal_step_writing(
                    "seen_{{index}}",
                    Value::String("{{item}}".into()),
                )),
                max_concurrent: 2,
            }),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        };
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "seed".into(),
            goal: None,
            stages: vec![seed, fan],
            edges: vec![edge("seed", "fan", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };
        let res = test_engine().run(&wf).await.unwrap();

        assert_eq!(res.status, WorkflowStatus::Completed);
        let fan_out = &res.stages[1].output;
        match fan_out {
            StageOutput::ForEach { items, outputs } => {
                assert_eq!(items, &vec!["a.rs", "b.rs", "c.rs"]);
                assert_eq!(outputs.len(), 3);
            }
            other => panic!("expected ForEach output, got {other:?}"),
        }
        assert_eq!(
            res.final_state.get("foreach.fan.count"),
            Some(&Value::from(3usize))
        );
        // Per-item answer slots are exposed.
        assert_eq!(
            res.final_state.get("foreach.fan.0.item"),
            Some(&Value::String("a.rs".into()))
        );
        // The body's own state delta is recoverable, namespaced per item.
        assert_eq!(
            res.final_state.get("foreach.fan.0.state.seen_0"),
            Some(&Value::String("a.rs".into()))
        );
        assert_eq!(
            res.final_state.get("foreach.fan.2.state.seen_2"),
            Some(&Value::String("c.rs".into()))
        );
    }

    // --- Deliver (result fan-out) ---

    fn failing_tool_action(id: &str) -> car_ir::Action {
        car_ir::Action {
            id: id.into(),
            action_type: car_ir::ActionType::ToolCall,
            tool: Some("definitely_not_a_registered_tool".into()),
            parameters: HashMap::new(),
            preconditions: vec![],
            expected_effects: HashMap::new(),
            state_dependencies: vec![],
            read_set: vec![],
            write_set: vec![],
            assumptions: vec![],
            invocation_mode: Default::default(),
            idempotent: false,
            max_retries: 0,
            failure_behavior: car_ir::FailureBehavior::Abort,
            timeout_ms: None,
            metadata: HashMap::new(),
        }
    }

    fn sink(id: &str, actions: Vec<car_ir::Action>) -> ActionProposal {
        ActionProposal {
            id: id.into(),
            source: "test".into(),
            actions,
            timestamp: chrono::Utc::now(),
            context: HashMap::new(),
        }
    }

    #[tokio::test]
    async fn deliver_fans_out_best_effort_and_templates_payload() {
        // seed writes the digest; deliver fans it out to 2 sinks, one of which
        // fails (unknown tool). The stage still succeeds, both sink results are
        // recorded, and {{payload}} is substituted into the surviving sink.
        let seed = Stage {
            id: "seed".into(),
            name: "seed".into(),
            step: proposal_step_writing("digest", Value::String("hello world".into())),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        };
        let deliver = Stage {
            id: "deliver".into(),
            name: "deliver".into(),
            step: StageStep::Deliver(DeliverStep {
                sinks: vec![
                    sink(
                        "ok-sink",
                        vec![state_write_action(
                            "notify",
                            Value::String("payload: {{payload}}".into()),
                        )],
                    ),
                    sink("bad-sink", vec![failing_tool_action("t1")]),
                ],
                payload_key: Some("digest".into()),
            }),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        };
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "seed".into(),
            goal: None,
            stages: vec![seed, deliver],
            edges: vec![edge("seed", "deliver", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };
        let res = test_engine().run(&wf).await.unwrap();

        assert_eq!(res.status, WorkflowStatus::Completed);
        assert_eq!(res.stages[1].status, StageStatus::Succeeded);
        match &res.stages[1].output {
            StageOutput::Deliver { results } => {
                assert_eq!(results.len(), 2, "both sinks recorded");
                assert_eq!(results[0].sink_id, "ok-sink");
                assert!(results[0].ok);
                assert!(results[0].error.is_none());
                // {{payload}} was substituted from the `digest` state key.
                let pr = results[0].result.as_ref().expect("sink ran");
                assert_eq!(
                    pr.results[0].state_changes.get("notify"),
                    Some(&Value::String("payload: hello world".into()))
                );
                assert_eq!(results[1].sink_id, "bad-sink");
                assert!(!results[1].ok, "unknown tool sink fails");
                assert!(results[1].error.as_deref().unwrap().contains("tool"));
            }
            other => panic!("expected Deliver output, got {other:?}"),
        }
        assert_eq!(
            res.final_state.get("stage.deliver.delivered_count"),
            Some(&Value::from(1usize))
        );
        assert_eq!(
            res.final_state.get("stage.deliver.failed_count"),
            Some(&Value::from(1usize))
        );
        assert_eq!(
            res.final_state.get("stage.deliver.answer"),
            Some(&Value::String("delivered to 1/2 sink(s)".into()))
        );
    }

    #[tokio::test]
    async fn deliver_fails_only_when_all_sinks_fail() {
        let deliver = Stage {
            id: "deliver".into(),
            name: "deliver".into(),
            step: StageStep::Deliver(DeliverStep {
                sinks: vec![
                    sink("bad-1", vec![failing_tool_action("t1")]),
                    sink("bad-2", vec![failing_tool_action("t2")]),
                ],
                payload_key: None,
            }),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        };
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "deliver".into(),
            goal: None,
            stages: vec![deliver],
            edges: vec![],
            max_iterations: 100,
            metadata: HashMap::new(),
        };
        let res = test_engine().run(&wf).await.unwrap();

        assert_eq!(res.status, WorkflowStatus::Failed);
        assert_eq!(res.stages[0].status, StageStatus::Failed);
        let err = res.stages[0].error.as_deref().unwrap();
        assert!(err.contains("all 2 delivery sink(s) failed"), "err: {err}");
        // Both sink ids are named in the aggregate error.
        assert!(err.contains("bad-1") && err.contains("bad-2"));
        // Even on total failure the per-sink evidence is the stage's output,
        // not Empty — a retry decision needs to see what was attempted.
        match &res.stages[0].output {
            StageOutput::Deliver { results } => {
                assert_eq!(results.len(), 2);
                assert!(results.iter().all(|r| !r.ok));
            }
            other => panic!("expected Deliver output with evidence, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn deliver_timeout_preserves_completed_sink_results() {
        // Sink 1 completes instantly; sink 2 grinds through retry backoff
        // (100ms, 200ms, 400ms, ... sleeps) far past the 300ms stage timeout.
        // (An unknown tool would be *rejected* by validation before dispatch —
        // no retries — so the slow sink uses a never-true assertion, which
        // fails at dispatch and is retried with backoff.) The timeout cancels
        // the fan-out mid-way — the completed sink's result must survive into
        // the failed stage's output, otherwise the partial delivery is
        // invisible and a retry double-sends it.
        let mut slow_action = state_write_action("unused", Value::Null);
        slow_action.id = "t-slow".into();
        slow_action.action_type = car_ir::ActionType::Assertion;
        slow_action.parameters = [
            ("key".to_string(), Value::from("never_set")),
            ("expected".to_string(), Value::Bool(true)),
        ]
        .into();
        slow_action.failure_behavior = car_ir::FailureBehavior::Retry;
        slow_action.max_retries = 10;

        let deliver = Stage {
            id: "deliver".into(),
            name: "deliver".into(),
            step: StageStep::Deliver(DeliverStep {
                sinks: vec![
                    sink(
                        "ok-sink",
                        vec![state_write_action("notify", Value::String("sent".into()))],
                    ),
                    sink("slow-sink", vec![slow_action]),
                ],
                payload_key: None,
            }),
            compensation: None,
            timeout_ms: Some(300),
            metadata: HashMap::new(),
        };
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "deliver".into(),
            goal: None,
            stages: vec![deliver],
            edges: vec![],
            max_iterations: 100,
            metadata: HashMap::new(),
        };
        let res = test_engine().run(&wf).await.unwrap();

        assert_eq!(res.status, WorkflowStatus::Failed);
        assert_eq!(res.stages[0].status, StageStatus::Failed);
        let err = res.stages[0].error.as_deref().unwrap();
        assert!(err.contains("timed out"), "err: {err}");
        // The sink that fired before the timeout is reported, not dropped.
        match &res.stages[0].output {
            StageOutput::Deliver { results } => {
                assert_eq!(results.len(), 1, "only the completed sink is recorded");
                assert_eq!(results[0].sink_id, "ok-sink");
                assert!(results[0].ok);
            }
            other => panic!("expected partial Deliver output, got {other:?}"),
        }
    }

    // --- Inter-workflow chaining (initial-state injection) ---

    #[tokio::test]
    async fn run_with_state_threads_prior_final_state_into_next_workflow() {
        let eng = test_engine();

        // Workflow A writes a handoff key into its final state.
        let wf_a = single_stage_wf(
            "a",
            proposal_step_writing("handoff", Value::String("from-a".into())),
        );
        let res_a = eng.run(&wf_a).await.unwrap();
        assert_eq!(res_a.status, WorkflowStatus::Completed);
        assert_eq!(
            res_a.final_state.get("handoff"),
            Some(&Value::String("from-a".into()))
        );

        // Workflow B branches on that key: check -> yes when handoff == "from-a",
        // otherwise the unconditional fallback to no.
        let wf_b = Workflow {
            id: "wf-b".into(),
            name: "WF B".into(),
            start: "check".into(),
            goal: None,
            stages: vec![
                empty_proposal_stage("check"),
                empty_proposal_stage("yes"),
                empty_proposal_stage("no"),
            ],
            edges: vec![
                edge(
                    "check",
                    "yes",
                    vec![car_ir::Precondition {
                        key: "handoff".into(),
                        operator: "eq".into(),
                        value: Value::String("from-a".into()),
                        description: String::new(),
                    }],
                ),
                edge("check", "no", vec![]),
            ],
            max_iterations: 100,
            metadata: HashMap::new(),
        };

        // Chained: B seeded with A's final_state routes down the yes branch.
        let res_b = eng
            .run_with_state(&wf_b, res_a.final_state.clone())
            .await
            .unwrap();
        assert_eq!(res_b.status, WorkflowStatus::Completed);
        let ran: Vec<&str> = res_b.stages.iter().map(|s| s.stage_id.as_str()).collect();
        assert!(ran.contains(&"yes"), "seeded state routes to yes: {ran:?}");
        assert!(!ran.contains(&"no"));

        // Unseeded: the same workflow takes the fallback branch.
        let res_b2 = eng.run(&wf_b).await.unwrap();
        let ran2: Vec<&str> = res_b2.stages.iter().map(|s| s.stage_id.as_str()).collect();
        assert!(ran2.contains(&"no"), "unseeded run falls through: {ran2:?}");
    }

    #[tokio::test]
    async fn run_with_state_cannot_repoint_the_goal_anchor() {
        let eng = test_engine();
        // No goal on the workflow → a seeded `goal` key is stripped.
        let wf = single_stage_wf("a", proposal_step_writing("x", Value::Bool(true)));
        let mut seed_state = HashMap::new();
        seed_state.insert("goal".to_string(), Value::String("injected".into()));
        let res = eng.run_with_state(&wf, seed_state).await.unwrap();
        assert!(res.final_state.get("goal").is_none());

        // A declared goal wins over the seeded one.
        let mut wf_goal = single_stage_wf("a", proposal_step_writing("x", Value::Bool(true)));
        wf_goal.goal = Some("real goal".into());
        let mut seed_state = HashMap::new();
        seed_state.insert("goal".to_string(), Value::String("injected".into()));
        let res = eng.run_with_state(&wf_goal, seed_state).await.unwrap();
        assert_eq!(
            res.final_state.get("goal"),
            Some(&Value::String("real goal".into()))
        );
    }

    #[tokio::test]
    async fn dedup_step_suppresses_seen_items_across_runs() {
        // seed publishes a 3-item list; dedup filters it against a persistent
        // store into `unseen_items`.
        let dir = std::env::temp_dir().join(format!(
            "car-wf-dedup-{}",
            &uuid::Uuid::new_v4().simple().to_string()[..8]
        ));
        let make_wf = || Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "seed".into(),
            goal: None,
            stages: vec![
                Stage {
                    id: "seed".into(),
                    name: "seed".into(),
                    step: proposal_step_writing(
                        "items",
                        serde_json::json!([{"id": 1}, {"id": 2}, {"id": 3}]),
                    ),
                    compensation: None,
                    timeout_ms: None,
                    metadata: HashMap::new(),
                },
                Stage {
                    id: "dedup".into(),
                    name: "dedup".into(),
                    step: StageStep::Dedup(DedupStep {
                        items_from: "items".into(),
                        into: "unseen_items".into(),
                        store: "test-src".into(),
                        hash_fields: vec!["id".into()],
                        ttl_secs: None,
                    }),
                    compensation: None,
                    timeout_ms: None,
                    metadata: HashMap::new(),
                },
            ],
            edges: vec![edge("seed", "dedup", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };

        let engine = WorkflowEngine::with_dedup_dir(
            Arc::new(NoopRunner),
            car_multi::SharedInfra::new(),
            &dir,
        );

        // First run: all 3 are new.
        let first = engine.run(&make_wf()).await.unwrap();
        assert_eq!(first.status, WorkflowStatus::Completed);
        assert_eq!(
            first.final_state.get("stage.dedup.unseen_count"),
            Some(&Value::from(3usize))
        );
        assert_eq!(
            first
                .final_state
                .get("unseen_items")
                .and_then(|v| v.as_array())
                .map(|a| a.len()),
            Some(3)
        );
        match &first.stages[1].output {
            StageOutput::Dedup { total, unseen } => {
                assert_eq!((*total, *unseen), (3, 3));
            }
            other => panic!("expected Dedup output, got {other:?}"),
        }

        // Second run against the same store: the same 3 ids are now suppressed.
        let second = engine.run(&make_wf()).await.unwrap();
        assert_eq!(
            second.final_state.get("stage.dedup.unseen_count"),
            Some(&Value::from(0usize))
        );
        assert_eq!(
            second.final_state.get("stage.dedup.total_count"),
            Some(&Value::from(3usize))
        );

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn for_each_missing_key_is_noop() {
        let step = StageStep::ForEach(ForEachStep {
            items_from: "nonexistent".into(),
            body: Box::new(empty_proposal_stage("b").step),
            max_concurrent: 0,
        });
        let wf = single_stage_wf("fan", step);
        let res = test_engine().run(&wf).await.unwrap();
        assert_eq!(res.status, WorkflowStatus::Completed);
        match &res.stages[0].output {
            StageOutput::ForEach { items, outputs } => {
                assert!(items.is_empty());
                assert!(outputs.is_empty());
            }
            other => panic!("expected ForEach output, got {other:?}"),
        }
    }

    // --- Goal pinning (drift defense) ---

    /// Captures the task string the agent was handed, so we can assert the goal
    /// was anchored into it.
    struct CapturingRunner {
        last_task: std::sync::Arc<std::sync::Mutex<String>>,
    }

    #[async_trait::async_trait]
    impl car_multi::AgentRunner for CapturingRunner {
        async fn run(
            &self,
            spec: &car_multi::AgentSpec,
            task: &str,
            _runtime: &car_engine::Runtime,
            _mailbox: &car_multi::Mailbox,
        ) -> Result<car_multi::AgentOutput, car_multi::MultiError> {
            *self.last_task.lock().unwrap() = task.to_string();
            Ok(car_multi::AgentOutput {
                name: spec.name.clone(),
                answer: "ok".into(),
                turns: 1,
                tool_calls: 0,
                duration_ms: 1.0,
                error: None,
                outcome: None,
                tokens: None,
                tools_used: Vec::new(),
            })
        }
    }

    #[tokio::test]
    async fn goal_is_pinned_and_anchored_into_agent_task() {
        let last_task = std::sync::Arc::new(std::sync::Mutex::new(String::new()));
        let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(CapturingRunner {
            last_task: last_task.clone(),
        });
        let eng = WorkflowEngine::new(runner, car_multi::SharedInfra::new());

        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "s".into(),
            goal: Some("ship the release safely".into()),
            stages: vec![Stage {
                id: "s".into(),
                name: "s".into(),
                step: StageStep::Pattern(PatternStep {
                    pattern: PatternKind::SwarmParallel,
                    task: "draft the notes".into(),
                    agents: vec![car_multi::AgentSpec::new("a", "")],
                    config: HashMap::new(),
                }),
                compensation: None,
                timeout_ms: None,
                metadata: HashMap::new(),
            }],
            edges: vec![],
            max_iterations: 100,
            metadata: HashMap::new(),
        };

        let res = eng.run(&wf).await.unwrap();
        assert_eq!(res.status, WorkflowStatus::Completed);
        // Goal pinned into final state.
        assert_eq!(
            res.final_state.get("goal"),
            Some(&Value::String("ship the release safely".into()))
        );
        // And re-anchored into the agent's task.
        let seen = last_task.lock().unwrap().clone();
        assert!(
            seen.contains("Overall goal: ship the release safely"),
            "got: {seen}"
        );
        assert!(
            seen.contains("Current step: draft the notes"),
            "got: {seen}"
        );
    }

    /// Reviewer returns a structured PASS verdict; other agents echo their name.
    struct ReviewRunner;

    #[async_trait::async_trait]
    impl car_multi::AgentRunner for ReviewRunner {
        async fn run(
            &self,
            spec: &car_multi::AgentSpec,
            _task: &str,
            _runtime: &car_engine::Runtime,
            _mailbox: &car_multi::Mailbox,
        ) -> Result<car_multi::AgentOutput, car_multi::MultiError> {
            let answer = if spec.name.contains("review") {
                r#"{"passed": true, "findings": [{"criterion":"complete","passed":true,"evidence":"all there","severity":"info"}]}"#.to_string()
            } else {
                spec.name.clone()
            };
            Ok(car_multi::AgentOutput {
                name: spec.name.clone(),
                answer,
                turns: 1,
                tool_calls: 0,
                duration_ms: 1.0,
                error: None,
                outcome: None,
                tokens: None,
                tools_used: Vec::new(),
            })
        }
    }

    #[tokio::test]
    async fn adversarial_review_stage_gates_prior_work() {
        let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(ReviewRunner);
        let eng = WorkflowEngine::new(runner, car_multi::SharedInfra::new());

        // Stage 1 writes the "work" to state; stage 2 reviews it with a fresh agent.
        let seed = Stage {
            id: "produce".into(),
            name: "produce".into(),
            step: proposal_step_writing("draft", Value::String("the finished work".into())),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        };
        let review = Stage {
            id: "review".into(),
            name: "review".into(),
            step: StageStep::Pattern(PatternStep {
                pattern: PatternKind::AdversarialReview,
                task: "verify".into(),
                agents: vec![car_multi::AgentSpec::new("reviewer", "be strict")],
                config: [
                    (
                        "criteria".to_string(),
                        serde_json::json!(["work is complete"]),
                    ),
                    ("review_key".to_string(), Value::String("draft".into())),
                ]
                .into(),
            }),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        };
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "produce".into(),
            goal: None,
            stages: vec![seed, review],
            edges: vec![edge("produce", "review", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };

        let res = eng.run(&wf).await.unwrap();
        assert_eq!(res.status, WorkflowStatus::Completed);
        // The typed verdict is the robust branching signal.
        assert_eq!(
            res.final_state.get("stage.review.review_passed"),
            Some(&Value::Bool(true))
        );
        assert!(matches!(
            res.stages[1].output,
            StageOutput::Review { passed: true, .. }
        ));
    }

    #[tokio::test]
    async fn adversarial_review_fails_closed_on_missing_work() {
        let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(ReviewRunner);
        let eng = WorkflowEngine::new(runner, car_multi::SharedInfra::new());
        // review_key points at a key no stage produced => empty work.
        let wf = single_stage_wf(
            "review",
            StageStep::Pattern(PatternStep {
                pattern: PatternKind::AdversarialReview,
                task: "verify".into(),
                agents: vec![car_multi::AgentSpec::new("reviewer", "")],
                config: [("review_key".to_string(), Value::String("nothing".into()))].into(),
            }),
        );
        let res = eng.run(&wf).await.unwrap();
        assert_eq!(res.status, WorkflowStatus::Completed);
        assert_eq!(
            res.final_state.get("stage.review.review_passed"),
            Some(&Value::Bool(false)),
            "empty work must fail closed, not vacuously pass"
        );
    }

    #[tokio::test]
    async fn stage_delta_cannot_clobber_pinned_goal() {
        // A proposal stage that writes a key literally named "goal" must not
        // repoint the drift anchor used by the following pattern stage.
        let last_task = std::sync::Arc::new(std::sync::Mutex::new(String::new()));
        let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(CapturingRunner {
            last_task: last_task.clone(),
        });
        let eng = WorkflowEngine::new(runner, car_multi::SharedInfra::new());

        let hijack = Stage {
            id: "hijack".into(),
            name: "hijack".into(),
            step: proposal_step_writing("goal", Value::String("DO SOMETHING ELSE".into())),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        };
        let work = Stage {
            id: "work".into(),
            name: "work".into(),
            step: StageStep::Pattern(PatternStep {
                pattern: PatternKind::SwarmParallel,
                task: "do the thing".into(),
                agents: vec![car_multi::AgentSpec::new("a", "")],
                config: HashMap::new(),
            }),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        };
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "hijack".into(),
            goal: Some("the real goal".into()),
            stages: vec![hijack, work],
            edges: vec![edge("hijack", "work", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };

        let res = eng.run(&wf).await.unwrap();
        assert_eq!(res.status, WorkflowStatus::Completed);
        // The anchor survived the clobber attempt.
        assert_eq!(
            res.final_state.get("goal"),
            Some(&Value::String("the real goal".into()))
        );
        let seen = last_task.lock().unwrap().clone();
        assert!(seen.contains("Overall goal: the real goal"), "got: {seen}");
    }

    // --- Prefix-cache re-run ---

    #[tokio::test]
    async fn run_cached_replays_prefix_and_runs_rest_live() {
        let last_task = std::sync::Arc::new(std::sync::Mutex::new(String::new()));
        let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(CapturingRunner {
            last_task: last_task.clone(),
        });
        let eng = WorkflowEngine::new(runner, car_multi::SharedInfra::new());

        // A (proposal) -> B (pattern). Prior run completed A but not B.
        let a = Stage {
            id: "a".into(),
            name: "a".into(),
            step: proposal_step_writing("k", Value::String("v".into())),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        };
        let b = Stage {
            id: "b".into(),
            name: "b".into(),
            step: StageStep::Pattern(PatternStep {
                pattern: PatternKind::SwarmParallel,
                task: "do b".into(),
                agents: vec![car_multi::AgentSpec::new("agent", "")],
                config: HashMap::new(),
            }),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        };
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "a".into(),
            goal: None,
            stages: vec![a, b],
            edges: vec![edge("a", "b", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };

        // Prior result: A succeeded with a distinctive cached answer; B absent.
        let mut final_state = HashMap::new();
        final_state.insert("stage.a.succeeded".to_string(), Value::Bool(true));
        final_state.insert("k".to_string(), Value::String("v".into()));
        let prior = WorkflowResult {
            workflow_id: "wf".into(),
            workflow_name: "WF".into(),
            status: WorkflowStatus::Failed,
            stages: vec![StageResult {
                stage_id: "a".into(),
                stage_name: "a".into(),
                status: StageStatus::Succeeded,
                output: StageOutput::Empty,
                duration_ms: 1.0,
                error: None,
            }],
            compensations: vec![],
            duration_ms: 1.0,
            timestamp: chrono::Utc::now(),
            final_state,
            paused: None,
        };

        let res = eng.run_cached(&wf, &prior).await.unwrap();
        assert_eq!(res.status, WorkflowStatus::Completed);
        // A replayed from cache (no re-execution), B ran live.
        let ran: Vec<&str> = res.stages.iter().map(|s| s.stage_id.as_str()).collect();
        assert_eq!(ran, vec!["a", "b"]);
        assert!(
            !last_task.lock().unwrap().is_empty(),
            "stage B should have executed live"
        );
        // Stage A's recorded result is the cached one (Empty output), not a re-run.
        assert!(matches!(res.stages[0].output, StageOutput::Empty));
    }

    #[tokio::test]
    async fn run_cached_strips_stale_uncached_stage_bookkeeping() {
        // Prior state carries a failed (uncached) stage's error key; on re-run
        // that stale bookkeeping must not survive to poison edge routing.
        let eng = test_engine();
        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "a".into(),
            goal: None,
            // "a" is terminal (no edges); "ghost" exists but is unreachable.
            stages: vec![empty_proposal_stage("a"), empty_proposal_stage("ghost")],
            edges: vec![],
            max_iterations: 100,
            metadata: HashMap::new(),
        };
        let mut final_state = HashMap::new();
        final_state.insert("stage.a.succeeded".to_string(), Value::Bool(true));
        final_state.insert(
            "stage.ghost.error".to_string(),
            Value::String("stale".into()),
        );
        final_state.insert("real_data".to_string(), Value::from(42));
        let prior = WorkflowResult {
            workflow_id: "wf".into(),
            workflow_name: "WF".into(),
            status: WorkflowStatus::Failed,
            stages: vec![StageResult {
                stage_id: "a".into(),
                stage_name: "a".into(),
                status: StageStatus::Succeeded,
                output: StageOutput::Empty,
                duration_ms: 1.0,
                error: None,
            }],
            compensations: vec![],
            duration_ms: 1.0,
            timestamp: chrono::Utc::now(),
            final_state,
            paused: None,
        };

        let res = eng.run_cached(&wf, &prior).await.unwrap();
        assert_eq!(res.status, WorkflowStatus::Completed);
        // Stale uncached-stage bookkeeping was stripped...
        assert!(!res.final_state.contains_key("stage.ghost.error"));
        // ...while genuine data and cached-stage keys survive.
        assert_eq!(res.final_state.get("real_data"), Some(&Value::from(42)));
        assert_eq!(
            res.final_state.get("stage.a.succeeded"),
            Some(&Value::Bool(true))
        );
    }

    #[tokio::test]
    async fn run_cached_with_full_cache_executes_nothing() {
        let last_task = std::sync::Arc::new(std::sync::Mutex::new(String::new()));
        let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(CapturingRunner {
            last_task: last_task.clone(),
        });
        let eng = WorkflowEngine::new(runner, car_multi::SharedInfra::new());

        let wf = single_stage_wf(
            "only",
            StageStep::Pattern(PatternStep {
                pattern: PatternKind::SwarmParallel,
                task: "x".into(),
                agents: vec![car_multi::AgentSpec::new("agent", "")],
                config: HashMap::new(),
            }),
        );
        let prior = WorkflowResult {
            workflow_id: "wf".into(),
            workflow_name: "WF".into(),
            status: WorkflowStatus::Completed,
            stages: vec![StageResult {
                stage_id: "only".into(),
                stage_name: "only".into(),
                status: StageStatus::Succeeded,
                output: StageOutput::Empty,
                duration_ms: 1.0,
                error: None,
            }],
            compensations: vec![],
            duration_ms: 1.0,
            timestamp: chrono::Utc::now(),
            final_state: HashMap::new(),
            paused: None,
        };

        let res = eng.run_cached(&wf, &prior).await.unwrap();
        assert_eq!(res.status, WorkflowStatus::Completed);
        assert!(
            last_task.lock().unwrap().is_empty(),
            "fully-cached run must not execute any agent"
        );
    }

    // --- Saga compensation (failure injection) ---
    //
    // These are the only engine tests that drive a forward-stage FAILURE and
    // therefore the `compensate()` reverse/LIFO path. The forward stages and
    // compensations are proposal-based and run through `car_engine::Runtime`
    // with NO executor set, so:
    //   * an empty proposal SUCCEEDS (`all_succeeded()` is vacuously true),
    //   * a `StateWrite` action SUCCEEDS (handled in-runtime, writes shared
    //     state — and crucially records an ordered `StateTransition`),
    //   * a `ToolCall` to an unregistered tool FAILS ("no handler for tool"),
    //     which is how we deterministically fail both a forward stage and a
    //     compensation action.

    /// A car_ir `Action` with all the boilerplate filled in.
    fn comp_action(action_type: car_ir::ActionType) -> car_ir::Action {
        car_ir::Action {
            id: "act".into(),
            action_type,
            tool: None,
            parameters: HashMap::new(),
            preconditions: vec![],
            expected_effects: HashMap::new(),
            state_dependencies: vec![],
            read_set: vec![],
            write_set: vec![],
            assumptions: vec![],
            invocation_mode: Default::default(),
            idempotent: false,
            max_retries: 0,
            failure_behavior: car_ir::FailureBehavior::Abort,
            timeout_ms: None,
            metadata: HashMap::new(),
        }
    }

    fn proposal_of(id: &str, actions: Vec<car_ir::Action>) -> ActionProposal {
        ActionProposal {
            id: format!("p-{id}"),
            source: "test".into(),
            actions,
            timestamp: chrono::Utc::now(),
            context: HashMap::new(),
        }
    }

    /// A proposal step whose `StateWrite` records that `marker_key` ran — this
    /// is what we use to observe compensation order via state transitions.
    fn recording_proposal_step(marker_key: &str) -> ProposalStep {
        let mut action = comp_action(car_ir::ActionType::StateWrite);
        action
            .parameters
            .insert("key".into(), Value::from(marker_key));
        action.parameters.insert("value".into(), Value::Bool(true));
        ProposalStep {
            proposal: proposal_of(marker_key, vec![action]),
        }
    }

    /// A proposal step that always FAILS (ToolCall to an unregistered tool).
    fn failing_proposal_step(id: &str) -> ProposalStep {
        let mut action = comp_action(car_ir::ActionType::ToolCall);
        action.tool = Some("no_such_tool".into());
        ProposalStep {
            proposal: proposal_of(id, vec![action]),
        }
    }

    /// A stage that SUCCEEDS (empty proposal) with an optional compensation
    /// handler recording into `comp.<id>` on rollback.
    fn ok_stage_with_comp(id: &str, comp: Option<CompensationHandler>) -> Stage {
        Stage {
            id: id.into(),
            name: id.into(),
            step: StageStep::Proposal(ProposalStep {
                proposal: proposal_of(id, vec![]),
            }),
            compensation: comp,
            timeout_ms: None,
            metadata: HashMap::new(),
        }
    }

    /// A stage that FAILS (ToolCall to an unregistered tool), triggering the
    /// saga rollback of everything completed before it.
    fn failing_stage(id: &str) -> Stage {
        Stage {
            id: id.into(),
            name: id.into(),
            step: StageStep::Proposal(failing_proposal_step(id)),
            compensation: None,
            timeout_ms: None,
            metadata: HashMap::new(),
        }
    }

    /// Order in which compensation StateWrites landed, derived from the shared
    /// state transition log — strips the `comp.` prefix.
    fn compensation_order(state: &car_state::StateStore) -> Vec<String> {
        state
            .transitions()
            .into_iter()
            .filter(|t| t.key.starts_with("comp."))
            .map(|t| t.key.trim_start_matches("comp.").to_string())
            .collect()
    }

    #[tokio::test]
    async fn compensation_runs_in_reverse_lifo() {
        let infra = SharedInfra::new();
        let state = Arc::clone(&infra.state);
        let eng = WorkflowEngine::new(Arc::new(NoopRunner), infra);

        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "A".into(),
            goal: None,
            stages: vec![
                ok_stage_with_comp(
                    "A",
                    Some(CompensationHandler::Proposal(recording_proposal_step(
                        "comp.A",
                    ))),
                ),
                ok_stage_with_comp(
                    "B",
                    Some(CompensationHandler::Proposal(recording_proposal_step(
                        "comp.B",
                    ))),
                ),
                failing_stage("C"),
            ],
            edges: vec![edge("A", "B", vec![]), edge("B", "C", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };

        let res = eng.run(&wf).await.unwrap();

        assert_eq!(res.status, WorkflowStatus::Compensated);
        // LIFO: B completed last, so it compensates first; C never completed.
        assert_eq!(compensation_order(&state), vec!["B", "A"]);
        let comp_ids: Vec<&str> = res
            .compensations
            .iter()
            .map(|c| c.for_stage_id.as_str())
            .collect();
        assert_eq!(comp_ids, vec!["B", "A"]);
    }

    #[tokio::test]
    async fn compensation_skips_stages_without_handler() {
        let infra = SharedInfra::new();
        let state = Arc::clone(&infra.state);
        let eng = WorkflowEngine::new(Arc::new(NoopRunner), infra);

        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "A".into(),
            goal: None,
            stages: vec![
                ok_stage_with_comp("A", None), // no compensation handler
                ok_stage_with_comp(
                    "B",
                    Some(CompensationHandler::Proposal(recording_proposal_step(
                        "comp.B",
                    ))),
                ),
                failing_stage("C"),
            ],
            edges: vec![edge("A", "B", vec![]), edge("B", "C", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };

        let res = eng.run(&wf).await.unwrap();

        // Only B has a handler, so only B compensates → all (one) succeeded.
        assert_eq!(res.status, WorkflowStatus::Compensated);
        assert_eq!(compensation_order(&state), vec!["B"]);
        let comp_ids: Vec<&str> = res
            .compensations
            .iter()
            .map(|c| c.for_stage_id.as_str())
            .collect();
        assert_eq!(comp_ids, vec!["B"]);
    }

    #[tokio::test]
    async fn compensation_error_is_partial_not_abort() {
        let infra = SharedInfra::new();
        let state = Arc::clone(&infra.state);
        let eng = WorkflowEngine::new(Arc::new(NoopRunner), infra);

        let wf = Workflow {
            id: "wf".into(),
            name: "WF".into(),
            start: "A".into(),
            goal: None,
            stages: vec![
                ok_stage_with_comp(
                    "A",
                    Some(CompensationHandler::Proposal(recording_proposal_step(
                        "comp.A",
                    ))),
                ),
                // B's compensation FAILS — but the saga must continue to A.
                ok_stage_with_comp(
                    "B",
                    Some(CompensationHandler::Proposal(failing_proposal_step(
                        "B-comp",
                    ))),
                ),
                failing_stage("C"),
            ],
            edges: vec![edge("A", "B", vec![]), edge("B", "C", vec![])],
            max_iterations: 100,
            metadata: HashMap::new(),
        };

        let res = eng.run(&wf).await.unwrap();

        // Some compensated (A), some failed (B) → PartiallyCompensated.
        assert_eq!(res.status, WorkflowStatus::PartiallyCompensated);
        // A still compensated even after B's compensation errored.
        assert_eq!(compensation_order(&state), vec!["A"]);

        // The compensation records, in LIFO order, are B (failed) then A (ok).
        assert_eq!(res.compensations.len(), 2);
        assert_eq!(res.compensations[0].for_stage_id, "B");
        assert_eq!(res.compensations[0].status, StageStatus::Failed);
        assert_eq!(res.compensations[1].for_stage_id, "A");
        assert_eq!(res.compensations[1].status, StageStatus::Succeeded);
    }
}
