ironflow_engine/

engine.rs

//! The core [`Engine`] -- orchestrates workflow execution and persistence.
//!
//! The engine ties together a `RunStore` for persistence, an [`AgentProvider`]
//! for AI operations, and a registry of [`WorkflowHandler`]s.
//!
//! Handlers are Rust-native: steps can reference previous outputs, use native
//! `if`/`else`/`match` for conditional branching, and execute in parallel.

use std::collections::{HashMap, HashSet};
use std::fmt;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use chrono::{DateTime, TimeDelta, Utc};
use rust_decimal::Decimal;
use serde_json::{Value, to_value};
use tokio::spawn;
use tokio::sync::{Mutex as AsyncMutex, OwnedMutexGuard};
use tokio::time::timeout;
use tracing::{error, info, warn};
use uuid::Uuid;

use ironflow_core::error::OperationError;
#[cfg(feature = "prometheus")]
use ironflow_core::metric_names::{
    RUN_BUDGET_EXCEEDED_TOTAL, RUN_COST_USD, RUN_DURATION_SECONDS, RUNS_ACTIVE, RUNS_TOTAL,
};
use ironflow_core::provider::{AgentProvider, LABEL_ROOT_RUN_ID};
use ironflow_store::error::StoreError;
use ironflow_store::models::{
    ConcurrencyLimit, LeaseUpdate, NewRun, NewSignal, ProviderKind, Run, RunActor, RunCreation,
    RunFilter, RunStatus, RunUpdate, SignalInsert, SignalStepResolution, StepKind, StepStatus,
    StepUpdate, TriggerKind, normalize_worker_tags, validate_concurrency_limits, validate_priority,
    validate_worker_tags,
};
use ironflow_store::store::Store;
#[cfg(feature = "prometheus")]
use metrics::{counter, gauge, histogram};

use crate::artifact::ArtifactSink;
use crate::budget::{BudgetConfig, month_start};
use crate::chain::ChainClaim;
use crate::context::{PARENT_RUN_ID_LABEL, WorkflowContext, interrupt_running_steps};
use crate::error::EngineError;
use crate::executor::{StepInterceptor, StepOutput, StepResult};
use crate::guard::{WorkflowGuardConfig, WorkflowGuardState};
use crate::handler::{WorkflowHandler, WorkflowInfo, clamp_priority};
use crate::log_sender::LogSender;
use crate::notify::{
    ApprovalRequestedEvent, Event, EventBusReleaser, EventPublisher, EventSubscriber,
    RunBudgetExceededEvent, RunFailedEvent, RunStatusChangedEvent, SignalAwaitedEvent,
    SignalReceivedEvent, WorkflowEventBus,
};
use crate::plan::{
    ExecutionPlan, PlanOptions, PlanRecorder, SharedPlanRecorder, estimate_durations, lock_plan,
};
use crate::retry_policy::{backoff_for_retry, is_run_retryable};
use crate::schedule::CronSchedule;
use crate::signal::{
    Signal, SignalDelivery, SignalOrigin, SignalRejected, SignalResumed, SignalSendFuture,
    SignalSender, received_output, validate_signal, validate_step_payload,
};
use ironflow_core::decision::DecisionProvider;

/// Page size used by [`Engine::list_all_runs`].
const LIST_RUNS_PAGE_SIZE: u32 = 100;

/// Result of a workflow execution, carrying the final [`Run`] and per-step
/// metrics collected during execution.
///
/// Returned by [`Engine::run_handler`], [`Engine::execute_handler_run`],
/// [`Engine::execute_run`], and [`Engine::resume_run`].
///
/// # Examples
///
/// ```no_run
/// use ironflow_engine::engine::WorkflowResult;
///
/// # fn example(result: WorkflowResult) {
/// println!("run {} finished with {} steps", result.run.id, result.steps.len());
/// for step in &result.steps {
///     println!("  {} ({:?}): {}ms", step.name, step.status, step.duration_ms);
/// }
/// # }
/// ```
#[derive(Debug, Clone)]
pub struct WorkflowResult {
    /// The finalized run record.
    pub run: Run,
    /// Per-step results in execution order.
    pub steps: Vec<StepResult>,
}

/// Optional settings for [`Engine::enqueue_handler_with_options`].
///
/// All fields fall back to handler or server defaults when left at their
/// [`Default`] value.
///
/// # Examples
///
/// ```
/// use ironflow_engine::engine::EnqueueOptions;
/// use rust_decimal::Decimal;
///
/// let options = EnqueueOptions {
///     max_retries: 3,
///     max_cost_usd: Some(Decimal::new(50, 2)),
///     ..Default::default()
/// };
/// assert_eq!(options.max_retries, 3);
/// ```
#[derive(Debug, Clone, Default)]
pub struct EnqueueOptions {
    /// Number of automatic retries granted to the run.
    pub max_retries: u32,
    /// Labels merged on top of the handler's default labels.
    pub labels: HashMap<String, String>,
    /// Defer execution until this instant instead of running as soon as a
    /// worker picks the run up.
    pub scheduled_at: Option<DateTime<Utc>>,
    /// Cost cap for the run. Overrides both the handler default and the server
    /// default. `None` falls back to
    /// [`BudgetConfig::resolve_run_cap`](crate::budget::BudgetConfig::resolve_run_cap).
    pub max_cost_usd: Option<Decimal>,
    /// Authenticated principal that triggered the run. `None` for cron,
    /// webhook, and programmatic triggers.
    pub created_by: Option<RunActor>,
    /// Idempotency key binding this enqueue to a single run.
    ///
    /// When set and already bound to a run created within
    /// [`IDEMPOTENCY_WINDOW`](ironflow_store::entities::IDEMPOTENCY_WINDOW),
    /// nothing is enqueued and the original run is replayed.
    pub idempotency_key: Option<String>,
    /// Concurrency key making the run exclusive.
    ///
    /// When set, the run is refused with [`EngineError::ConcurrencyConflict`]
    /// while another non-terminal run holds the same key. The key is released
    /// once the holder reaches a terminal state.
    pub concurrency_key: Option<String>,
    /// Concurrency groups the run belongs to, each with the maximum number of
    /// root runs of that group allowed to execute at once.
    ///
    /// Unlike [`concurrency_key`](Self::concurrency_key), the run is always
    /// created: it stays pending until every group is under its limit. Empty
    /// means no limit. Invalid limits are refused with
    /// [`EngineError::InvalidConcurrencyLimit`].
    pub concurrency_limits: Vec<ConcurrencyLimit>,
    /// Queue priority of the run, overriding
    /// [`WorkflowHandler::priority`].
    ///
    /// Workers pick the pending run with the highest priority first, then the
    /// oldest among equal priorities. `None` falls back to the handler's
    /// priority, clamped to the accepted range. An explicit value outside
    /// [`MIN_PRIORITY`](ironflow_store::entities::MIN_PRIORITY)`..=`[`MAX_PRIORITY`](ironflow_store::entities::MAX_PRIORITY)
    /// is refused with [`EngineError::InvalidPriority`].
    pub priority: Option<i16>,
    /// Worker tags the run requires, merged with the handler's
    /// [`required_worker_tags`](WorkflowHandler::required_worker_tags).
    ///
    /// Only a worker carrying every tag picks the run. Invalid tags are
    /// refused with [`EngineError::InvalidWorkerTag`].
    pub worker_tags: Vec<String>,
}

/// Where a run resumes once an approval, a human input or an escalation
/// resolves the gate it was suspended on.
///
/// # Examples
///
/// ```
/// use ironflow_engine::engine::ExecutionMode;
///
/// assert_eq!(ExecutionMode::default(), ExecutionMode::Local);
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum ExecutionMode {
    /// Resume in-process via [`Engine::resume_run`]. Used by
    /// [`crate::testing::TestEngine`] and single-process deployments where
    /// the API also registers the handlers.
    #[default]
    Local,
    /// Requeue the run to `Pending` instead. A worker's `pick_next_pending`
    /// claims it and finishes it via [`Engine::execute_handler_run`].
    Workers,
}

/// The workflow orchestration engine.
///
/// Holds references to the store, agent provider, and a registry of
/// [`WorkflowHandler`]s.
///
/// # Examples
///
/// ```no_run
/// use std::sync::Arc;
/// use ironflow_engine::engine::Engine;
/// use ironflow_engine::config::ShellConfig;
/// use ironflow_engine::handler::{WorkflowHandler, HandlerFuture, WorkflowInfo};
/// use ironflow_engine::context::WorkflowContext;
/// use ironflow_store::memory::InMemoryStore;
/// use ironflow_store::models::TriggerKind;
/// use ironflow_core::providers::claude::ClaudeCodeProvider;
/// use serde_json::json;
///
/// struct CiWorkflow;
/// impl WorkflowHandler for CiWorkflow {
///     fn name(&self) -> &str { "ci" }
///     fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
///         Box::pin(async move {
///             ctx.shell("test", ShellConfig::new("cargo test")).await?;
///             Ok(())
///         })
///     }
/// }
///
/// # async fn example() -> Result<(), ironflow_engine::error::EngineError> {
/// let store = Arc::new(InMemoryStore::new());
/// let provider = Arc::new(ClaudeCodeProvider::new());
/// let mut engine = Engine::new(store, provider);
/// engine.register(CiWorkflow)?;
///
/// let result = engine.run_handler("ci", TriggerKind::Manual, json!({})).await?;
/// tracing::info!(run_id = %result.run.id, status = ?result.run.status, steps = result.steps.len(), "run completed");
/// # Ok(())
/// # }
/// ```
pub struct Engine {
    store: Arc<dyn Store>,
    provider: Arc<dyn AgentProvider>,
    handlers: HashMap<String, Arc<dyn WorkflowHandler>>,
    event_publisher: EventPublisher,
    log_sender: Option<LogSender>,
    budget: BudgetConfig,
    artifact_sink: Option<Arc<dyn ArtifactSink>>,
    guard_config: Option<WorkflowGuardConfig>,
    event_bus: Option<WorkflowEventBus>,
    decision_provider: Option<Arc<dyn DecisionProvider>>,
    step_interceptor: Option<Arc<dyn StepInterceptor>>,
    execution_mode: ExecutionMode,
    worker_tags: Option<Arc<Vec<String>>>,
    /// Where `ctx.send_signal` delivers. `None` delivers in this process.
    signal_sender: Option<Arc<dyn SignalSender>>,
    active_runs: ActiveRuns,
    local_resumes: ActiveRuns,
}

/// Per-run locks of this process, keyed by run id.
type ActiveRuns = Arc<Mutex<HashMap<Uuid, Arc<AsyncMutex<()>>>>>;

/// Marks a run as executing in this process, or as being resumed locally,
/// until it is dropped.
///
/// Holds the run's lock, so a resume of the same run waits for the execution
/// to end ([`Engine::wait_until_idle`]) instead of replaying it in parallel.
struct ActiveRunGuard {
    runs: ActiveRuns,
    run_id: Uuid,
    entry: Arc<AsyncMutex<()>>,
    lock: Option<OwnedMutexGuard<()>>,
}

impl Drop for ActiveRunGuard {
    fn drop(&mut self) {
        self.lock.take();
        prune_active_run(&self.runs, self.run_id, &self.entry);
    }
}

/// Take the lock of `run_id` in `runs`, waiting for its current holder.
async fn lock_run(runs: &ActiveRuns, run_id: Uuid) -> ActiveRunGuard {
    let entry = {
        let mut runs = runs.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        Arc::clone(runs.entry(run_id).or_default())
    };
    let lock = Arc::clone(&entry).lock_owned().await;
    ActiveRunGuard {
        runs: Arc::clone(runs),
        run_id,
        entry,
        lock: Some(lock),
    }
}

/// Remove the lock of `run_id` from `runs` once nobody holds or awaits it.
///
/// `entry` is the caller's own clone: the map plus that clone make two
/// references, any more is a waiter that still needs the lock in the map.
fn prune_active_run(runs: &ActiveRuns, run_id: Uuid, entry: &Arc<AsyncMutex<()>>) {
    let mut runs = runs.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
    let unused = runs
        .get(&run_id)
        .is_some_and(|current| Arc::ptr_eq(current, entry))
        && Arc::strong_count(entry) == 2;
    if unused {
        runs.remove(&run_id);
    }
}

/// Delivers the signals of `ctx.send_signal` with an engine of this process.
struct InProcessSignalSender(Arc<Engine>);

impl SignalSender for InProcessSignalSender {
    fn send(&self, signal: NewSignal, origin: SignalOrigin) -> SignalSendFuture<'_> {
        Box::pin(self.0.deliver_signal_from(signal, Some(origin)))
    }
}

/// Validate a workflow category path.
///
/// A category is a `/`-separated list of non-empty segments. This function
/// rejects empty paths, leading or trailing `/`, consecutive `/`, and
/// segments containing only whitespace.
///
/// # Errors
///
/// Returns [`EngineError::InvalidWorkflow`] when the category is malformed.
fn validate_category(handler_name: &str, category: &str) -> Result<(), EngineError> {
    let reject = |reason: &str| {
        Err(EngineError::InvalidWorkflow(format!(
            "handler '{handler_name}' has invalid category '{category}': {reason}"
        )))
    };

    if category.is_empty() {
        return reject("empty category");
    }
    if category.starts_with('/') {
        return reject("leading '/'");
    }
    if category.ends_with('/') {
        return reject("trailing '/'");
    }
    for segment in category.split('/') {
        if segment.is_empty() {
            return reject("empty segment (double '/')");
        }
        if segment.trim().is_empty() {
            return reject("whitespace-only segment");
        }
    }
    Ok(())
}

/// Read a run id from the label `key` of a sub-workflow child run.
///
/// `None` when the run was not started by a `Workflow` step, when the label
/// is missing or invalid, or when it points back at the run itself.
fn chain_label(run: &Run, key: &str) -> Option<Uuid> {
    if !matches!(run.trigger, TriggerKind::Workflow) {
        return None;
    }
    let id = Uuid::parse_str(run.labels.get(key)?).ok()?;
    (id != run.id).then_some(id)
}

/// The root run of the chain a sub-workflow child run belongs to.
///
/// Resuming a child resumes this root instead: the root replays its steps
/// and re-enters the same child run through its open `Workflow` step. The
/// worker uses it to follow its lease from a child to the root it resumed.
///
/// `None` when `run` is not a sub-workflow child run.
///
/// # Examples
///
/// ```no_run
/// use ironflow_engine::engine::chain_root;
/// use ironflow_store::entities::Run;
/// use uuid::Uuid;
///
/// fn lease_target(run: &Run) -> Uuid {
///     chain_root(run).unwrap_or(run.id)
/// }
/// ```
pub fn chain_root(run: &Run) -> Option<Uuid> {
    chain_label(run, LABEL_ROOT_RUN_ID)
}

/// The run whose `Workflow` step started this sub-workflow child run.
pub(crate) fn chain_parent(run: &Run) -> Option<Uuid> {
    chain_label(run, PARENT_RUN_ID_LABEL)
}

impl Engine {
    /// Create a new engine with the given store and agent provider.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_store::memory::InMemoryStore;
    /// use ironflow_core::providers::claude::ClaudeCodeProvider;
    ///
    /// let engine = Engine::new(
    ///     Arc::new(InMemoryStore::new()),
    ///     Arc::new(ClaudeCodeProvider::new()),
    /// );
    /// ```
    pub fn new(store: Arc<dyn Store>, provider: Arc<dyn AgentProvider>) -> Self {
        Self {
            store,
            provider,
            handlers: HashMap::new(),
            event_publisher: EventPublisher::new(),
            log_sender: None,
            budget: BudgetConfig::new(),
            artifact_sink: None,
            guard_config: None,
            event_bus: None,
            decision_provider: None,
            step_interceptor: None,
            execution_mode: ExecutionMode::default(),
            worker_tags: None,
            signal_sender: None,
            active_runs: Arc::new(Mutex::new(HashMap::new())),
            local_resumes: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    /// Mark `run_id` as executing in this process, waiting for a previous
    /// execution of the same run to end first.
    async fn track_execution(&self, run_id: Uuid) -> ActiveRunGuard {
        lock_run(&self.active_runs, run_id).await
    }

    /// Wait until no execution of `run_id` is running in this process.
    async fn wait_until_idle(&self, run_id: Uuid) {
        let entry = {
            let runs = self
                .active_runs
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            runs.get(&run_id).map(Arc::clone)
        };
        if let Some(entry) = entry {
            drop(Arc::clone(&entry).lock_owned().await);
            prune_active_run(&self.active_runs, run_id, &entry);
        }
    }

    /// Whether an execution of `run_id` is running in this process.
    pub(crate) fn is_executing(&self, run_id: Uuid) -> bool {
        let runs = self
            .active_runs
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        runs.get(&run_id)
            .is_some_and(|entry| entry.try_lock().is_err())
    }

    /// Deliver the signals of `ctx.send_signal` through `sender` instead of
    /// this engine's store. A worker, whose store cannot deliver, sends them
    /// to the API.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::signal::SignalSender;
    /// # use ironflow_core::providers::claude::ClaudeCodeProvider;
    /// # use ironflow_store::memory::InMemoryStore;
    ///
    /// # fn example(sender: Arc<dyn SignalSender>) {
    /// let engine = Engine::new(Arc::new(InMemoryStore::new()), Arc::new(ClaudeCodeProvider::new()))
    ///     .with_signal_sender(sender);
    /// # }
    /// ```
    pub fn with_signal_sender(mut self, sender: Arc<dyn SignalSender>) -> Self {
        self.signal_sender = Some(sender);
        self
    }

    /// Wire a [`DecisionProvider`] backend for `ctx.decision(...)` steps.
    ///
    /// Without this, a workflow that reaches a decision step fails with
    /// [`EngineError::NoDecisionProvider`].
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_core::providers::claude::ClaudeCodeProvider;
    /// use ironflow_core::providers::record_replay_decision::RecordReplayDecisionProvider;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_store::memory::InMemoryStore;
    ///
    /// let engine = Engine::new(
    ///     Arc::new(InMemoryStore::new()),
    ///     Arc::new(ClaudeCodeProvider::new()),
    /// )
    /// .with_decision_provider(Arc::new(RecordReplayDecisionProvider::replay("tests/fixtures")));
    /// ```
    pub fn with_decision_provider(mut self, provider: Arc<dyn DecisionProvider>) -> Self {
        self.decision_provider = Some(provider);
        self
    }

    /// Wire a [`StepInterceptor`] that resolves steps without executing them.
    ///
    /// Used by [`crate::testing::TestEngine`] to mock shell, HTTP and approval
    /// steps. `None` (the default) executes every step for real.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    ///
    /// use ironflow_core::providers::claude::ClaudeCodeProvider;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::executor::StepInterceptor;
    /// use ironflow_store::memory::InMemoryStore;
    ///
    /// # fn example(interceptor: Arc<dyn StepInterceptor>) {
    /// let engine = Engine::new(
    ///     Arc::new(InMemoryStore::new()),
    ///     Arc::new(ClaudeCodeProvider::new()),
    /// )
    /// .with_step_interceptor(interceptor);
    /// # }
    /// ```
    pub fn with_step_interceptor(mut self, interceptor: Arc<dyn StepInterceptor>) -> Self {
        self.step_interceptor = Some(interceptor);
        self
    }

    /// The step interceptor wired into this engine, if any.
    pub fn step_interceptor(&self) -> Option<&Arc<dyn StepInterceptor>> {
        self.step_interceptor.as_ref()
    }

    /// Apply cost guardrails to this engine.
    ///
    /// Without this, both the per-run cap default and the monthly quota are
    /// disabled and the engine behaves exactly as before.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_core::providers::claude::ClaudeCodeProvider;
    /// use ironflow_engine::budget::BudgetConfig;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_store::memory::InMemoryStore;
    ///
    /// let engine = Engine::new(
    ///     Arc::new(InMemoryStore::new()),
    ///     Arc::new(ClaudeCodeProvider::new()),
    /// )
    /// .with_budget_config(BudgetConfig::from_env());
    /// ```
    pub fn with_budget_config(mut self, budget: BudgetConfig) -> Self {
        self.budget = budget;
        self
    }

    /// Returns the cost guardrails applied by this engine.
    pub fn budget_config(&self) -> &BudgetConfig {
        &self.budget
    }

    /// Apply workflow guard configuration to this engine.
    ///
    /// When set, every workflow run created by this engine is protected
    /// by the guard. Handlers can override this via
    /// [`WorkflowHandler::guard_config`].
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_core::providers::claude::ClaudeCodeProvider;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::guard::WorkflowGuardConfig;
    /// use ironflow_store::memory::InMemoryStore;
    ///
    /// let engine = Engine::new(
    ///     Arc::new(InMemoryStore::new()),
    ///     Arc::new(ClaudeCodeProvider::new()),
    /// )
    /// .with_guard_config(WorkflowGuardConfig::default());
    /// ```
    pub fn with_guard_config(mut self, config: WorkflowGuardConfig) -> Self {
        self.guard_config = Some(config);
        self
    }

    /// Returns the workflow guard configuration, if any.
    pub fn guard_config(&self) -> Option<&WorkflowGuardConfig> {
        self.guard_config.as_ref()
    }

    /// Choose where a run resumes once its approval, human input or
    /// escalation is resolved.
    ///
    /// Defaults to [`ExecutionMode::Local`]. Set [`ExecutionMode::Workers`]
    /// on an API process that delegates execution to workers.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_core::providers::claude::ClaudeCodeProvider;
    /// use ironflow_engine::engine::{Engine, ExecutionMode};
    /// use ironflow_store::memory::InMemoryStore;
    ///
    /// let engine = Engine::new(
    ///     Arc::new(InMemoryStore::new()),
    ///     Arc::new(ClaudeCodeProvider::new()),
    /// )
    /// .with_execution_mode(ExecutionMode::Workers);
    /// ```
    pub fn with_execution_mode(mut self, mode: ExecutionMode) -> Self {
        self.execution_mode = mode;
        self
    }

    /// Returns where a run resumes once its gate is resolved.
    pub fn execution_mode(&self) -> ExecutionMode {
        self.execution_mode
    }

    /// Declare the worker tags carried by the process running this engine.
    ///
    /// Set by a worker at build time. A `ctx.workflow(..)` step then refuses
    /// a child whose [`required_worker_tags`](WorkflowHandler::required_worker_tags)
    /// are not all carried here, instead of running it on the wrong host.
    /// Unset by default (API or local mode): no check is made.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_core::providers::claude::ClaudeCodeProvider;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_store::memory::InMemoryStore;
    ///
    /// let mut engine = Engine::new(
    ///     Arc::new(InMemoryStore::new()),
    ///     Arc::new(ClaudeCodeProvider::new()),
    /// );
    /// engine.set_worker_tags(vec!["gpu".to_string()]);
    /// assert_eq!(engine.worker_tags(), Some(&["gpu".to_string()][..]));
    /// ```
    pub fn set_worker_tags(&mut self, tags: Vec<String>) {
        self.worker_tags = Some(Arc::new(tags));
    }

    /// Returns the worker tags set by [`set_worker_tags`](Self::set_worker_tags),
    /// or `None` when the engine does not run inside a tagged worker.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_core::providers::claude::ClaudeCodeProvider;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_store::memory::InMemoryStore;
    ///
    /// let engine = Engine::new(
    ///     Arc::new(InMemoryStore::new()),
    ///     Arc::new(ClaudeCodeProvider::new()),
    /// );
    /// assert!(engine.worker_tags().is_none());
    /// ```
    pub fn worker_tags(&self) -> Option<&[String]> {
        self.worker_tags.as_deref().map(Vec::as_slice)
    }

    /// Attach a log sender for real-time step output streaming.
    ///
    /// When set, all workflow contexts created by this engine will forward
    /// step output (shell stdout/stderr, agent system messages) to the
    /// given sender.
    pub fn set_log_sender(&mut self, sender: LogSender) {
        self.log_sender = Some(sender);
    }

    /// Attach the backend that stores and serves artifact bytes.
    ///
    /// Every context this engine builds inherits it. Without one, steps that
    /// declare artifacts fail explicitly and every other step is unaffected.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    ///
    /// use ironflow_engine::artifact::ArtifactSink;
    /// use ironflow_engine::engine::Engine;
    ///
    /// # fn example(engine: &mut Engine, sink: Arc<dyn ArtifactSink>) {
    /// engine.set_artifact_sink(sink);
    /// # }
    /// ```
    pub fn set_artifact_sink(&mut self, sink: Arc<dyn ArtifactSink>) {
        self.artifact_sink = Some(sink);
    }

    /// The artifact backend attached to this engine, if any.
    pub fn artifact_sink(&self) -> Option<&Arc<dyn ArtifactSink>> {
        self.artifact_sink.as_ref()
    }

    /// Attach a [`WorkflowEventBus`] for per-run real-time monitoring.
    ///
    /// When set, all workflow contexts created by this engine will publish
    /// step-level events (`StepStarted`, `StepCompleted`, `StepFailed`)
    /// to the bus.
    ///
    /// The bus channel of a run is released when the run reaches a terminal
    /// status: subscribers drain the buffered events, then their receiver
    /// closes. Call this once per engine; each call registers a releaser.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::notify::WorkflowEventBus;
    ///
    /// # fn example(engine: &mut Engine) {
    /// engine.set_event_bus(WorkflowEventBus::new());
    /// # }
    /// ```
    pub fn set_event_bus(&mut self, bus: WorkflowEventBus) {
        self.event_bus = Some(bus.clone());
        self.event_publisher
            .subscribe(EventBusReleaser::new(bus), &[Event::RUN_STATUS_CHANGED]);
    }

    /// The workflow event bus attached to this engine, if any.
    pub fn event_bus(&self) -> Option<&WorkflowEventBus> {
        self.event_bus.as_ref()
    }

    /// Returns a reference to the backing store.
    pub fn store(&self) -> &Arc<dyn Store> {
        &self.store
    }

    /// Returns a reference to the agent provider.
    pub fn provider(&self) -> &Arc<dyn AgentProvider> {
        &self.provider
    }

    /// Build a [`WorkflowContext`] with access to the handler registry.
    ///
    /// The context is seeded with the run's attempt number and the cost and
    /// duration already accumulated by previous attempts, so a retried run
    /// reports the total it really consumed rather than only its last attempt.
    ///
    /// The run's persisted `max_cost_usd` becomes the context cost cap; `None`
    /// disables the per-run budget check for that context.
    fn build_context(&self, run: &Run) -> WorkflowContext {
        let handlers = self.handlers.clone();
        let resolver: crate::context::HandlerResolver =
            Arc::new(move |name: &str| handlers.get(name).cloned());
        let mut ctx = WorkflowContext::with_handler_resolver(
            run.id,
            run.workflow_name.clone(),
            self.store.clone(),
            self.provider.clone(),
            resolver,
        );
        ctx.carry_over_run_totals(run.retry_count + 1, run.cost_usd, run.duration_ms);
        ctx.set_max_cost_usd(run.max_cost_usd);
        ctx.set_run_created_at(run.created_at);
        if let Some(ref sender) = self.log_sender {
            ctx.set_log_sender(sender.clone());
        }
        if let Some(ref sink) = self.artifact_sink {
            ctx.set_artifact_sink(sink.clone());
        }
        if let Some(ref bus) = self.event_bus {
            ctx.set_event_bus(bus.clone());
        }
        if let Some(ref provider) = self.decision_provider {
            ctx.set_decision_provider(provider.clone());
        }
        if let Some(ref interceptor) = self.step_interceptor {
            ctx.set_step_interceptor(interceptor.clone());
        }
        if let Some(ref tags) = self.worker_tags {
            ctx.set_worker_tags(tags.clone());
        }
        // Delivering in this process wakes runs on a background task, which
        // needs an owned engine: `&self` is all a context build gets.
        let sender = self
            .signal_sender
            .clone()
            .unwrap_or_else(|| Arc::new(InProcessSignalSender(Arc::new(self.shared_copy()))));
        ctx.set_signal_sender(sender);
        ctx
    }

    /// A copy sharing the store, the subscribers and the in-process run locks,
    /// with the handlers registered so far.
    fn shared_copy(&self) -> Engine {
        Engine {
            store: self.store.clone(),
            provider: self.provider.clone(),
            handlers: self.handlers.clone(),
            event_publisher: self.event_publisher.clone(),
            log_sender: self.log_sender.clone(),
            budget: self.budget.clone(),
            artifact_sink: self.artifact_sink.clone(),
            guard_config: self.guard_config.clone(),
            event_bus: self.event_bus.clone(),
            decision_provider: self.decision_provider.clone(),
            step_interceptor: self.step_interceptor.clone(),
            execution_mode: self.execution_mode,
            worker_tags: self.worker_tags.clone(),
            signal_sender: self.signal_sender.clone(),
            active_runs: self.active_runs.clone(),
            local_resumes: self.local_resumes.clone(),
        }
    }

    /// The guard configuration that applies to `handler`: its own
    /// `guard_config()` if present, otherwise the engine's global one.
    pub(crate) fn effective_guard_config(
        &self,
        handler: &dyn WorkflowHandler,
    ) -> Option<WorkflowGuardConfig> {
        handler.guard_config().or_else(|| self.guard_config.clone())
    }

    /// Build a context with the guard attached.
    ///
    /// Uses [`effective_guard_config`](Self::effective_guard_config). The
    /// shared state carries the tokens and the active time spent by earlier
    /// executions of the run, plus the run deadline, so a resume, a requeue
    /// or a retry never starts the limits from zero.
    ///
    /// Seeding from `run` is correct because `run` is always the root of its
    /// chain: `execute_handler_run` and `resume_run` send a child to
    /// `resume_chain`, which resumes the root, and child contexts share the
    /// root's state through `execute_child_workflow`. So `run.id` is the
    /// `LABEL_ROOT_RUN_ID` of the whole tree and `run.created_at` is the
    /// creation of the root.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Store`] if the steps of the run tree cannot be
    /// read.
    async fn build_context_with_guard(
        &self,
        run: &Run,
        handler: &dyn WorkflowHandler,
    ) -> Result<WorkflowContext, EngineError> {
        let mut ctx = self.build_context(run);
        if let Some(config) = self.effective_guard_config(handler) {
            let mut state = WorkflowGuardState::new()
                .with_tokens_used(self.tokens_used_by_run_tree(run).await?)
                .with_elapsed(Duration::from_millis(run.duration_ms));
            if let Some(at) = config.deadline_for(run.created_at) {
                state = state.with_deadline_at(at);
            }
            ctx.set_guard(config, Arc::new(Mutex::new(state)));
        }
        Ok(ctx)
    }

    /// Record `err`, raised after the run was claimed but before its handler
    /// ran, as the run's failure, then hand it back.
    ///
    /// Without it the run stays `Running` until its lease expires, and only
    /// shows "worker lease expired" instead of the real cause.
    async fn fail_before_execution(&self, run_id: Uuid, err: EngineError) -> EngineError {
        let retryable = is_run_retryable(&err);
        if let Err(store_err) = self
            .fail_or_schedule_retry(run_id, &err.to_string(), retryable, None, None)
            .await
        {
            error!(run_id = %run_id, store_error = %store_err, "failed to persist run failure");
        }
        err
    }

    /// List every run matching `filter`, page after page.
    ///
    /// Every page is read before the caller acts on a run: a run that leaves
    /// the filter would shift the next page.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Store`] if a page cannot be read.
    pub(crate) async fn list_all_runs(&self, filter: RunFilter) -> Result<Vec<Run>, EngineError> {
        let mut runs = Vec::new();
        let mut page = 1;
        loop {
            let result = self
                .store
                .list_runs(filter.clone(), page, LIST_RUNS_PAGE_SIZE)
                .await?;
            if result.items.is_empty() {
                break;
            }
            runs.extend(result.items);
            if runs.len() as u64 >= result.total {
                break;
            }
            page += 1;
        }
        Ok(runs)
    }

    /// Tokens used by the agent steps of `root` and of every run of its tree
    /// (all statuses, all attempts).
    ///
    /// Steps replayed from an earlier execution never record their tokens
    /// again, so the persisted rows are the exact total and nothing is
    /// counted twice. The steps of earlier retry attempts are not replayed:
    /// their rows count once here, and the steps that run again in the new
    /// attempt are counted live as new rows.
    async fn tokens_used_by_run_tree(&self, root: &Run) -> Result<u64, EngineError> {
        let tree_label = HashMap::from([(LABEL_ROOT_RUN_ID.to_string(), root.id.to_string())]);
        let filter = RunFilter {
            labels: Some(tree_label),
            ..RunFilter::default()
        };
        let mut run_ids = vec![root.id];
        run_ids.extend(
            self.list_all_runs(filter)
                .await?
                .into_iter()
                .map(|run| run.id),
        );

        let mut tokens: u64 = 0;
        for run_id in run_ids {
            for step in self.store.list_steps(run_id).await? {
                if step.kind == StepKind::Agent {
                    let step_tokens = StepOutput::from(&step).total_tokens();
                    tokens = tokens.saturating_add(step_tokens);
                }
            }
        }
        Ok(tokens)
    }

    /// Reject the creation of a new run when the monthly quota is exhausted.
    ///
    /// The window is the current calendar month in UTC. Runs already in flight
    /// are never interrupted -- only creation is refused.
    ///
    /// The enqueue helpers call it themselves. The API also calls it before
    /// retrying or replaying a run, as those create runs without going through
    /// the enqueue helpers.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::MonthlyBudgetExceeded`] when the accumulated cost
    /// of the month has reached the configured quota. Returns
    /// [`EngineError::Store`] when the aggregate query fails.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ironflow_engine::engine::Engine;
    ///
    /// # use ironflow_engine::error::EngineError;
    /// # async fn example(engine: &Engine) -> Result<(), EngineError> {
    /// engine.check_monthly_quota("deploy").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn check_monthly_quota(&self, workflow_name: &str) -> Result<(), EngineError> {
        let Some(limit) = self.budget.monthly_cost_limit_usd else {
            return Ok(());
        };

        let stats = self
            .store
            .get_stats(RunFilter {
                created_after: Some(month_start(Utc::now())),
                ..RunFilter::default()
            })
            .await?;

        if stats.total_cost_usd < limit {
            return Ok(());
        }

        warn!(
            workflow = %workflow_name,
            limit_usd = %limit,
            spent_usd = %stats.total_cost_usd,
            "monthly cost quota exhausted, refusing new run"
        );

        #[cfg(feature = "prometheus")]
        counter!(
            RUN_BUDGET_EXCEEDED_TOTAL,
            "workflow" => workflow_name.to_string(),
            "scope" => "monthly",
        )
        .increment(1);

        Err(EngineError::MonthlyBudgetExceeded {
            limit_usd: limit,
            spent_usd: stats.total_cost_usd,
        })
    }

    // -----------------------------------------------------------------------
    // Handler registration
    // -----------------------------------------------------------------------

    /// Register a [`WorkflowHandler`] for dynamic workflow execution.
    ///
    /// The handler is looked up by [`WorkflowHandler::name`] when executing
    /// or enqueuing.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::InvalidWorkflow`] if a handler with the same
    /// name is already registered.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::handler::{WorkflowHandler, HandlerFuture};
    /// use ironflow_engine::context::WorkflowContext;
    /// use ironflow_engine::config::ShellConfig;
    /// use ironflow_store::memory::InMemoryStore;
    /// use ironflow_core::providers::claude::ClaudeCodeProvider;
    ///
    /// struct MyWorkflow;
    /// impl WorkflowHandler for MyWorkflow {
    ///     fn name(&self) -> &str { "my-workflow" }
    ///     fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
    ///         Box::pin(async move {
    ///             ctx.shell("step1", ShellConfig::new("echo done")).await?;
    ///             Ok(())
    ///         })
    ///     }
    /// }
    ///
    /// let mut engine = Engine::new(
    ///     Arc::new(InMemoryStore::new()),
    ///     Arc::new(ClaudeCodeProvider::new()),
    /// );
    /// engine.register(MyWorkflow)?;
    /// # Ok::<(), ironflow_engine::error::EngineError>(())
    /// ```
    pub fn register(&mut self, handler: impl WorkflowHandler + 'static) -> Result<(), EngineError> {
        let name = handler.name().to_string();
        if self.handlers.contains_key(&name) {
            return Err(EngineError::InvalidWorkflow(format!(
                "handler '{}' already registered",
                name
            )));
        }
        if let Some(category) = handler.category() {
            validate_category(&name, category)?;
        }
        self.handlers.insert(name, Arc::new(handler));
        Ok(())
    }

    /// Register a pre-boxed workflow handler.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::InvalidWorkflow`] if a handler with the same
    /// name is already registered or if its category is invalid.
    pub fn register_boxed(&mut self, handler: Box<dyn WorkflowHandler>) -> Result<(), EngineError> {
        let name = handler.name().to_string();
        if self.handlers.contains_key(&name) {
            return Err(EngineError::InvalidWorkflow(format!(
                "handler '{}' already registered",
                name
            )));
        }
        if let Some(category) = handler.category() {
            validate_category(&name, category)?;
        }
        self.handlers.insert(name, Arc::from(handler));
        Ok(())
    }

    /// Get a registered handler by name.
    pub fn get_handler(&self, name: &str) -> Option<&Arc<dyn WorkflowHandler>> {
        self.handlers.get(name)
    }

    /// List registered handler names.
    pub fn handler_names(&self) -> Vec<&str> {
        self.handlers.keys().map(|s| s.as_str()).collect()
    }

    /// Get detailed info about a registered workflow handler.
    pub fn handler_info(&self, name: &str) -> Option<WorkflowInfo> {
        self.handlers.get(name).map(|h| h.describe())
    }

    /// List handlers that have a cron schedule configured.
    ///
    /// Returns pairs of `(workflow_name, cron_expression)` for all handlers
    /// where [`WorkflowHandler::schedule`] returns `Some`.
    ///
    /// Use this to wire scheduled handlers into a scheduler
    /// (e.g. the schedule ticker in `ironflow-api`).
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use std::sync::Arc;
    /// # use ironflow_engine::engine::Engine;
    /// # use ironflow_store::memory::InMemoryStore;
    /// # use ironflow_core::providers::claude::ClaudeCodeProvider;
    /// let engine = Engine::new(
    ///     Arc::new(InMemoryStore::new()),
    ///     Arc::new(ClaudeCodeProvider::new()),
    /// );
    /// for (name, schedule) in engine.scheduled_handlers() {
    ///     tracing::info!("{name} runs on schedule: {schedule}");
    /// }
    /// ```
    pub fn scheduled_handlers(&self) -> Vec<(&str, &CronSchedule)> {
        self.handlers
            .iter()
            .filter_map(|(name, handler)| handler.schedule().map(|sched| (name.as_str(), sched)))
            .collect()
    }

    /// Register an event subscriber for domain events.
    ///
    /// The subscriber is called only for events whose type is in
    /// `event_types`. Pass [`Event::ALL`] to receive every event.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::notify::{Event, WebhookSubscriber};
    /// use ironflow_store::memory::InMemoryStore;
    /// use ironflow_core::providers::claude::ClaudeCodeProvider;
    /// use std::sync::Arc;
    ///
    /// let mut engine = Engine::new(
    ///     Arc::new(InMemoryStore::new()),
    ///     Arc::new(ClaudeCodeProvider::new()),
    /// );
    ///
    /// engine.subscribe(
    ///     WebhookSubscriber::new("https://hooks.example.com/events"),
    ///     &[Event::RUN_STATUS_CHANGED, Event::STEP_FAILED],
    /// );
    /// ```
    pub fn subscribe(
        &mut self,
        subscriber: impl EventSubscriber + 'static,
        event_types: &[&'static str],
    ) {
        self.event_publisher.subscribe(subscriber, event_types);
    }

    /// Returns a reference to the event publisher.
    ///
    /// Useful for publishing events from outside the engine (e.g. auth
    /// routes in the API layer).
    pub fn event_publisher(&self) -> &EventPublisher {
        &self.event_publisher
    }

    // -----------------------------------------------------------------------
    // Dynamic workflow execution (WorkflowHandler)
    // -----------------------------------------------------------------------

    /// Execute a registered handler inline.
    ///
    /// Creates a run, builds a [`WorkflowContext`], calls the handler's
    /// [`execute`](WorkflowHandler::execute), and finalizes the run.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::InvalidWorkflow`] if no handler is registered
    /// with that name. Returns [`EngineError`] if execution fails.
    ///
    /// When the workflow is paused (see [`Engine::pause_workflow`]) the run is
    /// created and left `Pending`, with no steps: the returned
    /// `WorkflowResult.run.status.state` is `Pending`. The hold applies in
    /// every execution mode; [`Engine::resume_workflow`] runs the held run
    /// under [`ExecutionMode::Local`].
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_store::memory::InMemoryStore;
    /// use ironflow_store::models::TriggerKind;
    /// use ironflow_core::providers::claude::ClaudeCodeProvider;
    /// use serde_json::json;
    ///
    /// # async fn example(engine: &Engine) -> Result<(), ironflow_engine::error::EngineError> {
    /// let run = engine.run_handler("deploy", TriggerKind::Manual, json!({})).await?;
    /// # Ok(())
    /// # }
    /// ```
    #[tracing::instrument(name = "engine.run_handler", skip_all, fields(workflow = %handler_name))]
    pub async fn run_handler(
        &self,
        handler_name: &str,
        trigger: TriggerKind,
        payload: Value,
    ) -> Result<WorkflowResult, EngineError> {
        let handler = self
            .handlers
            .get(handler_name)
            .ok_or_else(|| {
                EngineError::InvalidWorkflow(format!("no handler registered: {handler_name}"))
            })?
            .clone();

        self.check_monthly_quota(handler_name).await?;

        let handler_version = handler.version().map(str::to_string);
        let max_cost_usd = self
            .budget
            .resolve_run_cap(None, handler.default_max_cost_usd());
        let run = self
            .store
            .create_run(NewRun {
                created_by: None,
                workflow_name: handler_name.to_string(),
                trigger,
                payload,
                max_retries: 0,
                handler_version,
                labels: handler.default_labels(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: clamp_priority(handler.priority()),
                concurrency_limits: Vec::new(),
                max_cost_usd,
                worker_tags: normalize_worker_tags(handler.required_worker_tags()),
            })
            .await?
            .into_run();

        let run_id = run.id;
        info!(run_id = %run_id, handler_version = run.handler_version.as_deref().unwrap_or(""), "run created");

        if self.is_workflow_paused(handler_name).await? {
            info!(run_id = %run_id, workflow = %handler_name, "workflow paused, run left pending");
            return Ok(WorkflowResult {
                run,
                steps: Vec::new(),
            });
        }

        // Built before the run turns `Running`: a store error while seeding the
        // guard leaves the run `Pending`, not `Running` with no execution and
        // no finalization behind it.
        let mut ctx = self
            .build_context_with_guard(&run, handler.as_ref())
            .await?;

        self.store
            .update_run_status(run_id, RunStatus::Running)
            .await?;

        #[cfg(feature = "prometheus")]
        gauge!(RUNS_ACTIVE, "workflow" => handler_name.to_string()).increment(1.0);

        let run_start = Instant::now();
        let result = handler.execute(&mut ctx).await;
        self.finalize_run(run_id, handler_name, result, &ctx, run_start, run.labels)
            .await
    }

    /// Build the execution plan for a registered handler without running it.
    ///
    /// Executes the handler with every step method in recording mode: no
    /// command is spawned, no HTTP request is sent, no agent is called,
    /// nothing is persisted. Conditions declared with
    /// [`WorkflowContext::when`](crate::context::WorkflowContext::when) are
    /// evaluated against `payload`; those declared with
    /// [`WorkflowContext::when_dynamic`](crate::context::WorkflowContext::when_dynamic)
    /// are reported as unevaluable.
    ///
    /// Step outputs are synthetic and success-shaped, so the plan follows the
    /// nominal branch. A handler that unwraps a decision answer, or that
    /// deserializes `ctx.input::<T>()` against a payload it does not match,
    /// aborts the plan: the partial plan is returned with
    /// [`ExecutionPlan::incomplete_reason`] set.
    ///
    /// The handler is bounded twice: by
    /// [`MAX_PLANNED_STEPS`](crate::plan::MAX_PLANNED_STEPS) on every step
    /// kind, and by [`PlanOptions::timeout`]. Either one
    /// returns the partial plan, with [`ExecutionPlan::truncated`] and
    /// [`ExecutionPlan::incomplete_reason`] set, not an error. `ctx.store()`
    /// and `ctx.secrets()` return an error while planning.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::InvalidWorkflow`] when no handler is registered
    /// under `handler_name` or when `options.max_depth` is zero. Returns
    /// [`EngineError::Store`] when the duration-history query fails. A handler
    /// that errors mid-plan does **not** fail this call.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::error::EngineError;
    /// use ironflow_engine::plan::PlanOptions;
    /// use serde_json::json;
    ///
    /// # async fn example(engine: &Engine) -> Result<(), EngineError> {
    /// let plan = engine
    ///     .plan_handler("deploy", json!({"env": "prod"}), PlanOptions::default())
    ///     .await?;
    /// for step in &plan.steps {
    ///     println!("{} ({:?})", step.name, step.kind);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[tracing::instrument(name = "engine.plan_handler", skip_all, fields(workflow = %handler_name))]
    pub async fn plan_handler(
        &self,
        handler_name: &str,
        payload: Value,
        options: PlanOptions,
    ) -> Result<ExecutionPlan, EngineError> {
        if options.max_depth == 0 {
            return Err(EngineError::InvalidWorkflow(
                "max_depth must be at least 1".to_string(),
            ));
        }

        let handler = self
            .handlers
            .get(handler_name)
            .ok_or_else(|| {
                EngineError::InvalidWorkflow(format!("no handler registered: {handler_name}"))
            })?
            .clone();

        let estimates = if options.estimate_durations {
            estimate_durations(&self.store, handler_name, options.sample_runs).await?
        } else {
            HashMap::new()
        };

        let shared: SharedPlanRecorder = Arc::new(Mutex::new(PlanRecorder::new(
            handler_name.to_string(),
            payload,
            options.max_depth,
            estimates,
        )));

        // Deliberately bare: no guard, no event bus, no log sender, no artifact
        // sink and no budget. Planning produces no side effect to report.
        let handlers = self.handlers.clone();
        let resolver: crate::context::HandlerResolver =
            Arc::new(move |name: &str| handlers.get(name).cloned());
        let mut ctx = WorkflowContext::with_handler_resolver(
            Uuid::now_v7(),
            handler_name.to_string(),
            self.store.clone(),
            self.provider.clone(),
            resolver,
        );
        ctx.set_plan(shared.clone());

        match timeout(options.timeout, handler.execute(&mut ctx)).await {
            Ok(Ok(())) => {}
            Ok(Err(err)) => lock_plan(&shared).fail(err.to_string()),
            Err(_) => lock_plan(&shared).fail(format!(
                "plan timed out after {} s",
                options.timeout.as_secs_f64()
            )),
        }
        drop(ctx);

        let plan = match Arc::try_unwrap(shared) {
            Ok(mutex) => mutex
                .into_inner()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .into_plan(),
            Err(shared) => lock_plan(&shared).snapshot(),
        };

        info!(
            workflow = %handler_name,
            steps = plan.steps.len(),
            truncated = plan.truncated,
            "execution plan built"
        );

        Ok(plan)
    }

    /// Enqueue a handler-based workflow for worker execution.
    ///
    /// The workflow name is stored in the run. The worker looks up the
    /// handler by name when executing.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::InvalidWorkflow`] if no handler is registered.
    /// Returns [`EngineError::MonthlyBudgetExceeded`] if the monthly cost quota
    /// is exhausted.
    #[tracing::instrument(name = "engine.enqueue_handler", skip_all, fields(workflow = %handler_name))]
    pub async fn enqueue_handler(
        &self,
        handler_name: &str,
        trigger: TriggerKind,
        payload: Value,
        max_retries: u32,
    ) -> Result<Run, EngineError> {
        self.enqueue_handler_with_options(
            handler_name,
            trigger,
            payload,
            EnqueueOptions {
                max_retries,
                ..Default::default()
            },
        )
        .await
        .map(RunCreation::into_run)
    }

    /// Enqueue a handler-based workflow with labels, deferred scheduling, an
    /// optional cost cap, an optional author, and an optional idempotency key.
    ///
    /// See [`EnqueueOptions`] for the individual settings.
    ///
    /// When [`EnqueueOptions::idempotency_key`] is set and already bound to a run
    /// created within
    /// [`IDEMPOTENCY_WINDOW`](ironflow_store::entities::IDEMPOTENCY_WINDOW), nothing
    /// is enqueued and the original run is returned as [`RunCreation::Existing`].
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::InvalidWorkflow`] if no handler is registered.
    /// Returns [`EngineError::MonthlyBudgetExceeded`] if the monthly cost quota
    /// is exhausted. Returns [`EngineError::ConcurrencyConflict`] if
    /// [`EnqueueOptions::concurrency_key`] is held by another non-terminal run.
    /// Returns [`EngineError::InvalidConcurrencyLimit`] if
    /// [`EnqueueOptions::concurrency_limits`] is invalid, before any other check.
    /// Returns [`EngineError::InvalidPriority`] if [`EnqueueOptions::priority`]
    /// is out of range, before any other check.
    /// Returns [`EngineError::InvalidWorkerTag`] if [`EnqueueOptions::worker_tags`]
    /// holds an invalid tag, checked right after the concurrency limits.
    /// Returns [`EngineError::Store`] if the run cannot be persisted.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ironflow_engine::engine::{Engine, EnqueueOptions};
    /// use ironflow_store::models::TriggerKind;
    /// use serde_json::json;
    ///
    /// # async fn example(engine: &Engine) -> Result<(), ironflow_engine::error::EngineError> {
    /// let creation = engine
    ///     .enqueue_handler_with_options(
    ///         "deploy",
    ///         TriggerKind::Api,
    ///         json!({"env": "prod"}),
    ///         EnqueueOptions {
    ///             max_retries: 3,
    ///             idempotency_key: Some("github:abc-123".to_string()),
    ///             ..Default::default()
    ///         },
    ///     )
    ///     .await?;
    ///
    /// if creation.is_created() {
    ///     println!("enqueued {}", creation.run().id);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[tracing::instrument(name = "engine.enqueue_handler_with_options", skip_all, fields(workflow = %handler_name))]
    pub async fn enqueue_handler_with_options(
        &self,
        handler_name: &str,
        trigger: TriggerKind,
        payload: Value,
        options: EnqueueOptions,
    ) -> Result<RunCreation, EngineError> {
        let EnqueueOptions {
            max_retries,
            labels,
            scheduled_at,
            max_cost_usd,
            created_by,
            idempotency_key,
            concurrency_key,
            concurrency_limits,
            priority,
            worker_tags,
        } = options;

        // Checked first: a malformed request is the caller's error, whatever
        // the handler or the quota.
        validate_concurrency_limits(&concurrency_limits)
            .map_err(EngineError::InvalidConcurrencyLimit)?;
        if let Some(priority) = priority {
            validate_priority(priority).map_err(EngineError::InvalidPriority)?;
        }
        validate_worker_tags(&worker_tags).map_err(EngineError::InvalidWorkerTag)?;

        let handler = self.handlers.get(handler_name).ok_or_else(|| {
            EngineError::InvalidWorkflow(format!("no handler registered: {handler_name}"))
        })?;

        self.check_monthly_quota(handler_name).await?;

        let handler_version = handler.version().map(str::to_string);
        let mut merged_labels = handler.default_labels();
        merged_labels.extend(labels);
        let resolved_cap = self
            .budget
            .resolve_run_cap(max_cost_usd, handler.default_max_cost_usd());
        let priority = priority.unwrap_or_else(|| clamp_priority(handler.priority()));
        let required_tags = normalize_worker_tags(
            handler
                .required_worker_tags()
                .into_iter()
                .chain(worker_tags),
        );

        let creation = self
            .store
            .create_run(NewRun {
                workflow_name: handler_name.to_string(),
                trigger,
                payload,
                max_retries,
                handler_version,
                labels: merged_labels,
                scheduled_at,
                created_by,
                idempotency_key,
                concurrency_key,
                priority,
                concurrency_limits,
                max_cost_usd: resolved_cap,
                worker_tags: required_tags,
            })
            .await?;

        match &creation {
            RunCreation::Created(run) => info!(
                run_id = %run.id,
                workflow = %handler_name,
                max_cost_usd = ?resolved_cap,
                "handler run enqueued"
            ),
            RunCreation::Existing(run) => info!(
                run_id = %run.id,
                workflow = %handler_name,
                "idempotent replay, nothing enqueued"
            ),
        }

        Ok(creation)
    }

    /// Execute a handler-based run (used by the worker after pick_next_pending).
    ///
    /// Looks up the handler by the run's `workflow_name` and executes it
    /// with a fresh [`WorkflowContext`], after
    /// [`AgentProvider::release_run`] has stopped whatever a previous
    /// execution of the run left running.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::InvalidWorkflow`] if no handler matches. A
    /// failed release fails the execution with [`EngineError::Operation`],
    /// replayed while the run has retries left. Returns
    /// [`EngineError::HandlerVersionMismatch`] when the handler's current
    /// version is incompatible with the run's `handler_version` -- checked
    /// before any step is replayed.
    ///
    /// A suspended child run of a sub-workflow is never executed on its own:
    /// its root run is resumed instead, and re-enters the child (see
    /// [`resume_run`](Self::resume_run)).
    #[tracing::instrument(name = "engine.execute_handler_run", skip_all, fields(run_id = %run_id))]
    pub async fn execute_handler_run(&self, run_id: Uuid) -> Result<WorkflowResult, EngineError> {
        let run = self
            .store
            .get_run(run_id)
            .await?
            .ok_or(EngineError::Store(StoreError::RunNotFound(run_id)))?;

        if let Some(root_run_id) = chain_root(&run) {
            return self.resume_chain(run, root_run_id).await;
        }

        let _active = self.track_execution(run_id).await;

        let handler = self
            .handlers
            .get(&run.workflow_name)
            .ok_or_else(|| {
                EngineError::InvalidWorkflow(format!(
                    "no handler registered: {}",
                    run.workflow_name
                ))
            })?
            .clone();

        #[cfg(feature = "prometheus")]
        gauge!(RUNS_ACTIVE, "workflow" => run.workflow_name.clone()).increment(1.0);

        let run_start = Instant::now();
        let mut ctx = match self.build_context_with_guard(&run, handler.as_ref()).await {
            Ok(ctx) => ctx,
            Err(err) => return Err(self.fail_before_execution(run_id, err).await),
        };

        // Replay the steps already persisted for this run: on a retry, an approval
        // a human already granted must not be asked again, and a run requeued to
        // `Pending` after its approval, human input or escalation resolved
        // (`ExecutionMode::Workers`, `retry_count` unchanged) must not re-run
        // completed steps. Neither must a run requeued by the reaper after its
        // worker lost the lease, which stays in the same attempt too. A
        // brand-new run has no steps, so this is a no-op.
        //
        // The handler version is checked first: replaying an incompatible
        // handler's steps risks serving one step's cached output to another
        // (`EngineError::ReplayDivergence`), so no step is replayed at all
        // when the handler changed incompatibly since the run was created.
        let result = if handler.is_version_compatible(run.handler_version.as_deref()) {
            if let Err(err) = ctx.load_replay_steps().await {
                return Err(self.fail_before_execution(run_id, err).await);
            }
            self.release_then_execute(run_id, handler.as_ref(), &mut ctx)
                .await
        } else {
            Err(EngineError::HandlerVersionMismatch {
                run_id,
                workflow_name: run.workflow_name.clone(),
                run_version: run
                    .handler_version
                    .clone()
                    .unwrap_or_else(|| "unknown".to_string()),
                current_version: handler
                    .version()
                    .map(str::to_string)
                    .unwrap_or_else(|| "unknown".to_string()),
            })
        };

        self.finalize_run(
            run_id,
            &run.workflow_name,
            result,
            &ctx,
            run_start,
            run.labels,
        )
        .await
    }

    /// Execute a run by its ID (used by the worker after pick_next_pending).
    ///
    /// Delegates to [`execute_handler_run`](Self::execute_handler_run).
    ///
    /// # Errors
    ///
    /// Returns [`EngineError`] if the run is not found or execution fails.
    #[tracing::instrument(name = "engine.execute_run", skip_all, fields(run_id = %run_id))]
    pub async fn execute_run(&self, run_id: Uuid) -> Result<WorkflowResult, EngineError> {
        self.execute_handler_run(run_id).await
    }

    /// Resume a run after human approval.
    ///
    /// Re-executes the handler with step replay: completed steps return
    /// cached output, approved approval steps are skipped, and execution
    /// continues from the first unexecuted step.
    ///
    /// Supports multiple approval gates -- each resume replays all prior
    /// steps and stops at the next approval (or completes the run).
    ///
    /// When `run_id` is a child run of a sub-workflow, the root run of its
    /// chain is moved back to `Running` and resumed instead: it replays, and
    /// its open `Workflow` step re-enters the same child run. The returned
    /// result is the root run's.
    ///
    /// Like [`execute_handler_run`](Self::execute_handler_run), the handler
    /// only starts once [`AgentProvider::release_run`] has stopped whatever
    /// a previous execution of the run left running.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::InvalidWorkflow`] if no handler matches, or when
    /// the root run of a child cannot be resumed (it is running or finished:
    /// the child is then failed).
    /// Returns [`EngineError`] if execution fails or hits another approval.
    /// A failed release fails the execution with [`EngineError::Operation`].
    /// Returns [`EngineError::HandlerVersionMismatch`] when the handler's
    /// current version is incompatible with the run's `handler_version` --
    /// checked before any step is replayed.
    #[tracing::instrument(name = "engine.resume_run", skip_all, fields(run_id = %run_id))]
    pub async fn resume_run(&self, run_id: Uuid) -> Result<WorkflowResult, EngineError> {
        let run = self
            .store
            .get_run(run_id)
            .await?
            .ok_or(EngineError::Store(StoreError::RunNotFound(run_id)))?;

        if let Some(root_run_id) = chain_root(&run) {
            return self.resume_chain(run, root_run_id).await;
        }

        self.resume_loaded_run(run).await
    }

    /// Resume the root run of a suspended child run.
    ///
    /// The root waits without `scheduled_at` while its child is suspended, so
    /// nothing but this path ever wakes it. It is moved to `Running` and
    /// resumed; its replay re-enters the child. A root already running, resumed
    /// by a sibling of the child in a wave, is waited for: if it re-entered the
    /// child meanwhile, the root is returned as it stands and nothing else
    /// runs. A finished root cannot take the child back: the child is failed
    /// (see [`claim_chain_root`](Self::claim_chain_root)).
    ///
    /// When the child holds a worker lease (it was picked by a worker), the
    /// lease is transferred to the root in the same update that moves it to
    /// `Running`, then released on the child: the worker keeps renewing the
    /// root it now executes, and the reaper recovers the root if that worker
    /// dies. A child without a lease (inline execution, API-side resume)
    /// leaves the root without one, as before.
    async fn resume_chain(
        &self,
        child: Run,
        root_run_id: Uuid,
    ) -> Result<WorkflowResult, EngineError> {
        let child_run_id = child.id;
        let lease = child.worker_id.zip(child.lease_expires_at);
        match self
            .claim_chain_root(child_run_id, root_run_id, lease.as_ref())
            .await?
        {
            ChainClaim::Claimed => {}
            ChainClaim::Settled(root) => {
                return Ok(WorkflowResult {
                    run: *root,
                    steps: Vec::new(),
                });
            }
        }

        if lease.is_some() {
            self.store
                .update_run(
                    child_run_id,
                    RunUpdate {
                        lease: Some(LeaseUpdate::Release),
                        ..RunUpdate::default()
                    },
                )
                .await?;
        }

        info!(
            run_id = %child_run_id,
            root_run_id = %root_run_id,
            lease_transferred = lease.is_some(),
            "child run resumed through its root run"
        );

        let root = self
            .store
            .get_run(root_run_id)
            .await?
            .ok_or(EngineError::Store(StoreError::RunNotFound(root_run_id)))?;
        self.resume_loaded_run(root).await
    }

    /// Move the suspended root run of a chain to `Running`.
    ///
    /// With `lease`, the root takes the worker lease of the child that
    /// resumes it, atomically with the transition, so it is never `Running`
    /// without an owner.
    pub(crate) async fn move_root_to_running(
        &self,
        root_run_id: Uuid,
        lease: Option<&(String, DateTime<Utc>)>,
    ) -> Result<(), EngineError> {
        match lease {
            Some((worker_id, expires_at)) => {
                self.store
                    .update_run(
                        root_run_id,
                        RunUpdate {
                            status: Some(RunStatus::Running),
                            lease: Some(LeaseUpdate::Set {
                                worker_id: worker_id.clone(),
                                expires_at: *expires_at,
                            }),
                            ..RunUpdate::default()
                        },
                    )
                    .await?;
            }
            None => {
                self.store
                    .update_run_status(root_run_id, RunStatus::Running)
                    .await?;
            }
        }
        Ok(())
    }

    /// Resume `run`, already loaded and already `Running`.
    async fn resume_loaded_run(&self, run: Run) -> Result<WorkflowResult, EngineError> {
        let run_id = run.id;
        let _active = self.track_execution(run_id).await;
        let handler = self
            .handlers
            .get(&run.workflow_name)
            .ok_or_else(|| {
                EngineError::InvalidWorkflow(format!(
                    "no handler registered: {}",
                    run.workflow_name
                ))
            })?
            .clone();

        info!(run_id = %run_id, workflow = %run.workflow_name, "resuming run after approval");

        let run_start = Instant::now();
        let mut ctx = match self.build_context_with_guard(&run, handler.as_ref()).await {
            Ok(ctx) => ctx,
            Err(err) => return Err(self.fail_before_execution(run_id, err).await),
        };

        let result = if handler.is_version_compatible(run.handler_version.as_deref()) {
            if let Err(err) = ctx.load_replay_steps().await {
                return Err(self.fail_before_execution(run_id, err).await);
            }
            self.release_then_execute(run_id, handler.as_ref(), &mut ctx)
                .await
        } else {
            Err(EngineError::HandlerVersionMismatch {
                run_id,
                workflow_name: run.workflow_name.clone(),
                run_version: run
                    .handler_version
                    .clone()
                    .unwrap_or_else(|| "unknown".to_string()),
                current_version: handler
                    .version()
                    .map(str::to_string)
                    .unwrap_or_else(|| "unknown".to_string()),
            })
        };

        self.finalize_run(
            run_id,
            &run.workflow_name,
            result,
            &ctx,
            run_start,
            run.labels,
        )
        .await
    }

    /// Store a signal and resolve every step waiting for its `(name, key)`.
    ///
    /// Each waiting step validates the payload against the JSON schema it
    /// stored when it opened. A matching step is completed with the payload
    /// and its run, if `Sleeping`, goes back to `Pending`: under
    /// [`ExecutionMode::Local`] it resumes in a background task, under
    /// [`ExecutionMode::Workers`] a worker picks it up. A step whose schema
    /// the payload does not match keeps waiting and is listed in
    /// [`SignalDelivery::rejected`].
    ///
    /// The signal is stored even when nobody waits for it: a run opening its
    /// wait step later still finds it. A signal whose `idempotency_id` was
    /// already used is not stored nor delivered again, and comes back with
    /// [`SignalDelivery::duplicate`] set.
    ///
    /// Publishes [`Event::SignalReceived`] for every stored signal.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::InvalidSignal`] when `name` or `key` is empty,
    /// or when `name`, `key` or `idempotency_id` is longer than
    /// [`MAX_SIGNAL_FIELD_LEN`](crate::signal::MAX_SIGNAL_FIELD_LEN), and
    /// [`EngineError::Store`] when the signal cannot be stored or its
    /// waiters cannot be listed.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::error::EngineError;
    /// use ironflow_store::entities::NewSignal;
    /// use serde_json::json;
    ///
    /// # async fn example(engine: Arc<Engine>) -> Result<(), EngineError> {
    /// let delivery = engine
    ///     .deliver_signal(NewSignal {
    ///         name: "ci.pipeline_finished".to_string(),
    ///         key: "4f2a9c1".to_string(),
    ///         payload: json!({"status": "success"}),
    ///         idempotency_id: Some("delivery-42".to_string()),
    ///     })
    ///     .await?;
    /// println!("{} runs resumed", delivery.resumed.len());
    /// # Ok(())
    /// # }
    /// ```
    pub async fn deliver_signal(
        self: &Arc<Self>,
        signal: NewSignal,
    ) -> Result<SignalDelivery, EngineError> {
        self.deliver_signal_from(signal, None).await
    }

    /// [`deliver_signal`](Self::deliver_signal), recording the workflow step
    /// that sent the signal on [`Event::SignalReceived`]: the audit log entry
    /// then belongs to that run and step.
    ///
    /// # Errors
    ///
    /// Same as [`deliver_signal`](Self::deliver_signal).
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::error::EngineError;
    /// use ironflow_engine::signal::SignalOrigin;
    /// use ironflow_store::entities::NewSignal;
    /// use serde_json::json;
    ///
    /// # async fn example(engine: Arc<Engine>, origin: SignalOrigin) -> Result<(), EngineError> {
    /// let signal = NewSignal {
    ///     name: "gitlab.human_replied".to_string(),
    ///     key: "discussion-7".to_string(),
    ///     payload: json!({"body": "ship it"}),
    ///     idempotency_id: None,
    /// };
    /// engine.deliver_signal_from(signal, Some(origin)).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn deliver_signal_from(
        self: &Arc<Self>,
        signal: NewSignal,
        origin: Option<SignalOrigin>,
    ) -> Result<SignalDelivery, EngineError> {
        validate_signal(&signal)?;

        let stored = match self.store.insert_signal(signal).await? {
            SignalInsert::Created(stored) => stored,
            SignalInsert::Duplicate(existing) => {
                info!(
                    signal_id = %existing.id,
                    signal = %existing.name,
                    key = %existing.key,
                    "duplicate signal ignored"
                );
                return Ok(SignalDelivery {
                    signal_id: existing.id,
                    duplicate: true,
                    resumed: Vec::new(),
                    rejected: Vec::new(),
                });
            }
        };

        let waiters = self
            .store
            .list_signal_waiters(&stored.name, &stored.key)
            .await?;
        let mut resumed = Vec::new();
        let mut rejected = Vec::new();

        for step in waiters {
            if let Err(error) = validate_step_payload(step.input.as_ref(), &stored.payload) {
                rejected.push(SignalRejected {
                    run_id: step.run_id,
                    step_id: step.id,
                    error,
                });
                continue;
            }

            match self
                .store
                .resolve_signal_step(step.id, received_output(&stored))
                .await
            {
                Ok(SignalStepResolution::Resolved {
                    run_id,
                    run_resumed,
                }) => {
                    resumed.push(SignalResumed {
                        run_id,
                        step_id: step.id,
                    });
                    if run_resumed && self.execution_mode == ExecutionMode::Local {
                        self.spawn_local_resume(run_id);
                    }
                }
                // A concurrent delivery or the timeout resolved it first.
                Ok(SignalStepResolution::NotWaiting { .. }) => {}
                Err(err) => {
                    error!(
                        run_id = %step.run_id,
                        step_id = %step.id,
                        error = %err,
                        "failed to resolve a waiting signal step"
                    );
                    rejected.push(SignalRejected {
                        run_id: step.run_id,
                        step_id: step.id,
                        error: err.to_string(),
                    });
                }
            }
        }

        info!(
            signal_id = %stored.id,
            signal = %stored.name,
            key = %stored.key,
            resumed = resumed.len(),
            rejected = rejected.len(),
            "signal received"
        );
        self.event_publisher
            .publish(Event::SignalReceived(SignalReceivedEvent {
                signal_id: stored.id,
                name: stored.name.clone(),
                key: stored.key.clone(),
                resumed_runs: resumed.iter().map(|r| r.run_id).collect(),
                sender_run_id: origin.map(|o| o.run_id),
                sender_step_id: origin.map(|o| o.step_id),
                at: stored.received_at,
            }));

        Ok(SignalDelivery {
            signal_id: stored.id,
            duplicate: false,
            resumed,
            rejected,
        })
    }

    /// Send a typed signal: shorthand for [`deliver_signal`](Self::deliver_signal)
    /// with `S::NAME` as the name and `signal` as the payload.
    ///
    /// `key` identifies the occurrence (a commit SHA, an order ID).
    /// `idempotency_id`, when set, makes a redelivery of the same event (a
    /// webhook retried by its sender) a no-op.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Serialization`] when `signal` cannot be
    /// serialized, and every error of [`deliver_signal`](Self::deliver_signal).
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::error::EngineError;
    /// use ironflow_engine::signal::Signal;
    /// use schemars::JsonSchema;
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Serialize, Deserialize, JsonSchema)]
    /// struct PipelineFinished {
    ///     status: String,
    /// }
    ///
    /// impl Signal for PipelineFinished {
    ///     const NAME: &'static str = "ci.pipeline_finished";
    /// }
    ///
    /// # async fn example(engine: Arc<Engine>) -> Result<(), EngineError> {
    /// let finished = PipelineFinished { status: "success".to_string() };
    /// engine.send_signal(&finished, "4f2a9c1", Some("delivery-42")).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn send_signal<S: Signal>(
        self: &Arc<Self>,
        signal: &S,
        key: &str,
        idempotency_id: Option<&str>,
    ) -> Result<SignalDelivery, EngineError> {
        let payload = to_value(signal)?;
        self.deliver_signal(NewSignal {
            name: S::NAME.to_string(),
            key: key.to_string(),
            payload,
            idempotency_id: idempotency_id.map(str::to_string),
        })
        .await
    }

    /// Resume a run requeued to `Pending` in a background task.
    ///
    /// Used under [`ExecutionMode::Local`], where no worker would pick the
    /// run up. The state change already happened, so a failed resume is
    /// logged, not rolled back.
    ///
    /// The local resumes of a run are serialized: a task reads the run only
    /// once the previous local resume of it has returned, so two tasks never
    /// both act on the same read. The task then waits for the execution of
    /// the run still in flight in this process, if any: a run paused then
    /// resumed while a step runs would otherwise be replayed next to the
    /// execution that has not noticed the pause yet. Once it is gone, the run
    /// is read again:
    ///
    /// - `Pending` is claimed with the `Pending -> Running` transition, a
    ///   compare-and-set in every store, the one a worker performs when it
    ///   picks a run. Only a successful claim resumes it here: a run another
    ///   executor claimed first is left to it.
    /// - `Running` with a worker lease is owned by that worker and left alone.
    /// - `Running` without a lease was left unfinished by the previous
    ///   execution in this process and is resumed as is, unless an execution
    ///   of it is still running here.
    /// - any other status means the previous execution finished the run or it
    ///   was taken elsewhere, and nothing is done.
    pub(crate) fn spawn_local_resume(self: &Arc<Self>, run_id: Uuid) {
        let engine = Arc::clone(self);
        spawn(async move {
            // One local resume of a run at a time, held until its execution
            // returns: a sibling task reads the run only once this one is done,
            // so it never sees the claim of a resume that has not started yet.
            let _resuming = lock_run(&engine.local_resumes, run_id).await;
            engine.wait_until_idle(run_id).await;
            let run = match engine.store.get_run(run_id).await {
                Ok(Some(run)) => run,
                Ok(None) => {
                    error!(run_id = %run_id, "run to restart not found");
                    return;
                }
                Err(err) => {
                    error!(run_id = %run_id, error = %err, "failed to load a run to restart");
                    return;
                }
            };
            match run.status.state {
                RunStatus::Pending => {
                    match engine
                        .store
                        .update_run_status(run_id, RunStatus::Running)
                        .await
                    {
                        Ok(()) => {}
                        Err(StoreError::InvalidTransition { from, .. }) => {
                            info!(
                                run_id = %run_id,
                                status = %from,
                                "run claimed elsewhere, left to its worker"
                            );
                            return;
                        }
                        Err(err) => {
                            error!(run_id = %run_id, error = %err, "failed to restart a woken run");
                            return;
                        }
                    }
                }
                RunStatus::Running if run.worker_id.is_some() => {
                    info!(
                        run_id = %run_id,
                        worker_id = ?run.worker_id,
                        "run owned by a worker lease, local resume skipped"
                    );
                    return;
                }
                RunStatus::Running => {
                    if engine.is_executing(run_id) {
                        info!(run_id = %run_id, "run already executing in this process, resume skipped");
                        return;
                    }
                }
                status => {
                    info!(
                        run_id = %run_id,
                        status = %status,
                        "run no longer waiting to restart, resume skipped"
                    );
                    return;
                }
            }
            if let Err(err) = engine.resume_run(run_id).await {
                error!(run_id = %run_id, error = %err, "failed to resume a woken run");
            }
        });
    }

    /// Record a run failure, replaying the run later when retries remain.
    ///
    /// This is the single place where a failed run's fate is decided. When
    /// `retryable` is true and the run has not exhausted `max_retries`, the run
    /// moves to [`RunStatus::Retrying`] with `scheduled_at` set to
    /// `now + backoff` -- [`pick_next_pending`](ironflow_store::store::RunStore::pick_next_pending)
    /// picks it up again once that time has passed. Otherwise the run moves to
    /// [`RunStatus::Failed`].
    ///
    /// Either way, steps left non-terminal by the failed attempt are closed via
    /// [`fail_orphaned_steps`](Self::fail_orphaned_steps) so they are never
    /// confused with the next attempt's steps, and the sub-workflow runs it
    /// left non-terminal are cancelled by
    /// [`cancel_descendants`](Self::cancel_descendants) (a failure there is
    /// logged, not returned).
    ///
    /// Callers pass `retryable` explicitly rather than an error value, because
    /// the worker classifies failures it observes from the outside (a timeout, a
    /// panicked task) that never produce an [`EngineError`]. Use
    /// [`is_run_retryable`] to classify an
    /// [`EngineError`].
    ///
    /// `cost_usd` and `duration_ms` are `None` when the caller does not know the
    /// totals (a timeout or a panic observed from outside the handler); the
    /// values already stored on the run are then left untouched.
    ///
    /// A run an operator paused meanwhile is left untouched and
    /// [`RunStatus::Paused`] is returned: the resume replays the attempt.
    ///
    /// Returns the status the run was moved to.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Store`] if the run does not exist or the update
    /// cannot be persisted.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::error::EngineError;
    /// use uuid::Uuid;
    ///
    /// # async fn example(engine: &Engine, run_id: Uuid) -> Result<(), EngineError> {
    /// let status = engine
    ///     .fail_or_schedule_retry(run_id, "run timed out after 300s", true, None, None)
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn fail_or_schedule_retry(
        &self,
        run_id: Uuid,
        error: &str,
        retryable: bool,
        cost_usd: Option<Decimal>,
        duration_ms: Option<u64>,
    ) -> Result<RunStatus, EngineError> {
        let run = self
            .store
            .get_run(run_id)
            .await?
            .ok_or(EngineError::Store(StoreError::RunNotFound(run_id)))?;

        // An operator paused the run: its fate is decided by the resume or
        // the cancellation, not by the attempt that stopped.
        if run.status.state == RunStatus::Paused {
            info!(run_id = %run_id, error = %error, "run paused, failure not recorded");
            return Ok(RunStatus::Paused);
        }

        let has_attempts_left = run.retry_count < run.max_retries;
        let update = if retryable && has_attempts_left {
            let backoff = backoff_for_retry(run.retry_count);
            let scheduled_at = Utc::now() + TimeDelta::milliseconds(backoff.as_millis() as i64);

            info!(
                run_id = %run_id,
                workflow = %run.workflow_name,
                attempt = run.retry_count + 1,
                max_retries = run.max_retries,
                backoff_secs = backoff.as_secs(),
                scheduled_at = %scheduled_at,
                "run failed, scheduling retry"
            );

            RunUpdate {
                status: Some(RunStatus::Retrying),
                error: Some(error.to_string()),
                increment_retry: true,
                cost_usd,
                duration_ms,
                scheduled_at: Some(scheduled_at),
                ..RunUpdate::default()
            }
        } else {
            RunUpdate {
                status: Some(RunStatus::Failed),
                error: Some(error.to_string()),
                cost_usd,
                duration_ms,
                completed_at: Some(Utc::now()),
                ..RunUpdate::default()
            }
        };

        let status = update.status.unwrap_or(RunStatus::Failed);
        self.store.update_run(run_id, update).await?;
        self.fail_orphaned_steps(run_id, error).await?;
        // The attempt is over: a retry starts new children, and nothing drives
        // those this attempt left running.
        self.cancel_descendants_of_stopped_run(run_id, error).await;

        Ok(status)
    }

    /// Mark the steps of a run requeued after a lost worker lease as interrupted.
    ///
    /// Every `Running` step is marked `Failed` with
    /// [`STEP_INTERRUPTED_ERROR`](ironflow_store::store::STEP_INTERRUPTED_ERROR).
    /// The run keeps its attempt number, so when it is picked up again its
    /// finished steps are replayed and each interrupted step is executed again
    /// at the same position; an interrupted `Workflow` step re-enters the same
    /// child run. `Pending` and `AwaitingApproval` steps are left as they are,
    /// unlike [`fail_orphaned_steps`](Self::fail_orphaned_steps), which ends
    /// the run's steps for good.
    ///
    /// Errors from individual step updates are logged but do not abort the cleanup.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError`] if listing steps fails.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::error::EngineError;
    /// use uuid::Uuid;
    ///
    /// # async fn example(engine: &Engine, run_id: Uuid) -> Result<(), EngineError> {
    /// engine.interrupt_running_steps(run_id).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn interrupt_running_steps(&self, run_id: Uuid) -> Result<(), EngineError> {
        interrupt_running_steps(self.store.as_ref(), run_id).await
    }

    /// Fail all non-terminal steps for a run.
    ///
    /// Called after a run is marked as failed (timeout, error, panic) to clean up
    /// orphaned steps that are still in `Running`, `Pending`, or `AwaitingApproval`.
    ///
    /// - `Running` / `AwaitingApproval` steps are marked `Failed`.
    /// - `Pending` steps are marked `Skipped` (FSM does not allow Pending -> Failed).
    ///
    /// Errors from individual step updates are logged but do not abort the cleanup.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError`] if listing steps fails.
    pub async fn fail_orphaned_steps(
        &self,
        run_id: Uuid,
        error_message: &str,
    ) -> Result<(), EngineError> {
        let steps = self.store.list_steps(run_id).await?;
        let now = Utc::now();

        for step in steps {
            if step.status.state.is_terminal() {
                continue;
            }

            let (target_status, error) = match step.status.state {
                StepStatus::Running | StepStatus::AwaitingApproval => {
                    let err = if step.error.is_some() {
                        None
                    } else {
                        Some(error_message.to_string())
                    };
                    (StepStatus::Failed, err)
                }
                StepStatus::Pending => (StepStatus::Skipped, None),
                _ => continue,
            };

            if let Err(e) = self
                .store
                .update_step(
                    step.id,
                    StepUpdate {
                        status: Some(target_status),
                        error,
                        completed_at: Some(now),
                        ..StepUpdate::default()
                    },
                )
                .await
            {
                warn!(
                    run_id = %run_id,
                    step_id = %step.id,
                    step_name = %step.name,
                    error = %e,
                    "failed to cleanup orphaned step"
                );
            } else {
                info!(
                    run_id = %run_id,
                    step_id = %step.id,
                    step_name = %step.name,
                    from = %step.status.state,
                    to = %target_status,
                    "cleaned up orphaned step"
                );
            }
        }

        Ok(())
    }

    /// Execute `handler` once [`AgentProvider::release_run`] has stopped
    /// whatever a previous execution of the run left running (an agent pod
    /// writing to a shared worktree). A failed release fails the execution
    /// before its first step, with [`EngineError::Operation`].
    async fn release_then_execute(
        &self,
        run_id: Uuid,
        handler: &dyn WorkflowHandler,
        ctx: &mut WorkflowContext,
    ) -> Result<(), EngineError> {
        match self.provider.release_run(&run_id.to_string()).await {
            Ok(()) => handler.execute(ctx).await,
            Err(e) => Err(EngineError::Operation(OperationError::Agent(e))),
        }
    }

    /// Finalize a run with the given result and context.
    ///
    /// On success: updates run to Completed with cost, duration, and completed_at.
    /// On failure: updates run to Failed with error, cost, duration, and completed_at.
    /// Always: fetches and returns the final Run.
    async fn finalize_run(
        &self,
        run_id: Uuid,
        workflow_name: &str,
        result: Result<(), EngineError>,
        ctx: &WorkflowContext,
        run_start: Instant,
        run_labels: HashMap<String, String>,
    ) -> Result<WorkflowResult, EngineError> {
        // Covers the whole run: previous attempts plus this one, so a retried
        // run reports the time it really consumed.
        let total_duration = ctx.carried_duration_ms() + run_start.elapsed().as_millis() as u64;
        let completed_at = Utc::now();

        // Paused while it ran: whatever the handler returned, the run stays
        // paused for the operator. Only what this execution spent is kept;
        // the resume replays from the first step that did not complete.
        if let Some(run) = self.store.get_run(run_id).await?
            && run.status.state == RunStatus::Paused
        {
            let run = self
                .store
                .update_run_returning(
                    run_id,
                    RunUpdate {
                        cost_usd: Some(ctx.total_cost_usd()),
                        duration_ms: Some(total_duration),
                        ..RunUpdate::default()
                    },
                )
                .await?;
            info!(
                run_id = %run_id,
                outcome = ?result.err().map(|err| err.to_string()),
                "run paused, execution stopped"
            );
            return Ok(WorkflowResult {
                run,
                steps: ctx.step_results().to_vec(),
            });
        }

        // The pause stopped this execution but the run was resumed since: the
        // resume continues or restarts it, so nothing is recorded here.
        if matches!(result, Err(EngineError::RunPaused { .. })) {
            let run = self
                .store
                .get_run(run_id)
                .await?
                .ok_or(EngineError::Store(StoreError::RunNotFound(run_id)))?;
            info!(
                run_id = %run_id,
                status = %run.status.state,
                "run resumed after the pause, execution stopped"
            );
            return Ok(WorkflowResult {
                run,
                steps: ctx.step_results().to_vec(),
            });
        }

        let final_status;
        let final_run;

        match result {
            Ok(()) => {
                final_status = if ctx.has_allowed_failure() {
                    RunStatus::Warning
                } else {
                    RunStatus::Completed
                };
                final_run = self
                    .store
                    .update_run_returning(
                        run_id,
                        RunUpdate {
                            status: Some(final_status),
                            cost_usd: Some(ctx.total_cost_usd()),
                            duration_ms: Some(total_duration),
                            completed_at: Some(completed_at),
                            output: ctx.output().cloned(),
                            ..RunUpdate::default()
                        },
                    )
                    .await?;

                info!(
                    run_id = %run_id,
                    status = %final_status,
                    cost_usd = %ctx.total_cost_usd(),
                    duration_ms = total_duration,
                    "run completed"
                );
            }
            Err(EngineError::ApprovalRequired {
                run_id: approval_run_id,
                step_id,
                ref message,
            }) => {
                final_status = RunStatus::AwaitingApproval;
                final_run = self
                    .store
                    .update_run_returning(
                        run_id,
                        RunUpdate {
                            status: Some(RunStatus::AwaitingApproval),
                            cost_usd: Some(ctx.total_cost_usd()),
                            duration_ms: Some(total_duration),
                            ..RunUpdate::default()
                        },
                    )
                    .await?;

                info!(
                    run_id = %approval_run_id,
                    step_id = %step_id,
                    message = %message,
                    "run awaiting approval"
                );

                self.publish_approval_requested(approval_run_id, step_id, message)
                    .await?;
            }
            Err(EngineError::ChildSuspended {
                run_id: child_run_id,
                ref cause,
            }) => {
                final_status = cause.suspension_status();
                // No `scheduled_at`: the suspended descendant owns the wake-up
                // and resumes this run through `resume_chain`. A wake-up armed
                // here too would resume the chain twice.
                final_run = self
                    .store
                    .update_run_returning(
                        run_id,
                        RunUpdate {
                            status: Some(final_status),
                            cost_usd: Some(ctx.total_cost_usd()),
                            duration_ms: Some(total_duration),
                            ..RunUpdate::default()
                        },
                    )
                    .await?;

                let leaf = cause.suspension_leaf();
                info!(
                    run_id = %run_id,
                    child_run_id = %child_run_id,
                    status = %final_status,
                    cause = %leaf,
                    "run suspended with its child run"
                );

                self.publish_suspension(leaf).await?;
                // A wave reports its first suspended child: the others wait too.
                for sibling in ctx.sibling_suspensions() {
                    self.publish_suspension(sibling.suspension_leaf()).await?;
                }
            }
            Err(EngineError::HumanInputRequired {
                run_id: input_run_id,
                step_id,
                ref message,
            }) => {
                final_status = RunStatus::AwaitingApproval;
                final_run = self
                    .store
                    .update_run_returning(
                        run_id,
                        RunUpdate {
                            status: Some(RunStatus::AwaitingApproval),
                            cost_usd: Some(ctx.total_cost_usd()),
                            duration_ms: Some(total_duration),
                            ..RunUpdate::default()
                        },
                    )
                    .await?;

                // No `ApprovalRequested` event: a human input is not an approval.
                info!(
                    run_id = %input_run_id,
                    step_id = %step_id,
                    message = %message,
                    "run awaiting human input"
                );
            }
            Err(EngineError::DelaySleeping {
                run_id: delay_run_id,
                step_id,
                wake_at,
            }) => {
                final_status = RunStatus::Sleeping;
                final_run = self
                    .store
                    .update_run_returning(
                        run_id,
                        RunUpdate {
                            status: Some(RunStatus::Sleeping),
                            cost_usd: Some(ctx.total_cost_usd()),
                            duration_ms: Some(total_duration),
                            scheduled_at: Some(wake_at),
                            ..RunUpdate::default()
                        },
                    )
                    .await?;

                info!(
                    run_id = %delay_run_id,
                    step_id = %step_id,
                    wake_at = %wake_at,
                    "run sleeping until delay elapses"
                );
            }
            Err(EngineError::CapacitySleeping {
                run_id: capacity_run_id,
                step_id,
                ref kind,
                wake_at,
            }) => {
                final_status = RunStatus::Sleeping;
                final_run = self
                    .store
                    .update_run_returning(
                        run_id,
                        RunUpdate {
                            status: Some(RunStatus::Sleeping),
                            cost_usd: Some(ctx.total_cost_usd()),
                            duration_ms: Some(total_duration),
                            scheduled_at: Some(wake_at),
                            capacity_wait_kind: Some(ProviderKind::new(kind.as_str())),
                            ..RunUpdate::default()
                        },
                    )
                    .await?;

                info!(
                    run_id = %capacity_run_id,
                    step_id = %step_id,
                    kind = %kind,
                    wake_at = %wake_at,
                    "run sleeping until provider capacity returns"
                );
            }
            Err(EngineError::SignalWaiting {
                run_id: wait_run_id,
                step_id,
                ref step_name,
                ref name,
                ref key,
                deadline_at,
            }) => {
                final_status = RunStatus::Sleeping;
                // Atomic with the step lock: a signal delivered since the step
                // opened leaves the run due right away instead of until the
                // deadline.
                let waiting = self
                    .store
                    .suspend_run_on_signal(run_id, step_id, deadline_at)
                    .await?;
                final_run = self
                    .store
                    .update_run_returning(
                        run_id,
                        RunUpdate {
                            cost_usd: Some(ctx.total_cost_usd()),
                            duration_ms: Some(total_duration),
                            ..RunUpdate::default()
                        },
                    )
                    .await?;

                if waiting {
                    self.event_publisher
                        .publish(Event::SignalAwaited(SignalAwaitedEvent {
                            run_id: wait_run_id,
                            step_id,
                            step_name: step_name.clone(),
                            name: name.clone(),
                            key: key.clone(),
                            deadline_at,
                            at: Utc::now(),
                        }));
                }

                info!(
                    run_id = %wait_run_id,
                    step_id = %step_id,
                    signal = %name,
                    key = %key,
                    deadline_at = %deadline_at,
                    waiting,
                    "run sleeping until a signal arrives"
                );
            }
            Err(err) => {
                // A guardrail stop (budget or workflow guard) is deliberate,
                // not a breakage: the run is cancelled, never failed and
                // never replayed.
                let guardrail_stop = matches!(
                    err,
                    EngineError::RunBudgetExceeded { .. } | EngineError::WorkflowGuardRejected(_)
                );

                final_status = if guardrail_stop {
                    if let Err(store_err) = self
                        .store
                        .update_run(
                            run_id,
                            RunUpdate {
                                status: Some(RunStatus::Cancelled),
                                error: Some(err.to_string()),
                                cost_usd: Some(ctx.total_cost_usd()),
                                duration_ms: Some(total_duration),
                                completed_at: Some(completed_at),
                                output: ctx.output().cloned(),
                                ..RunUpdate::default()
                            },
                        )
                        .await
                    {
                        error!(run_id = %run_id, store_error = %store_err, "failed to persist run cancellation");
                    }
                    if let Err(cleanup_err) = self
                        .fail_orphaned_steps(run_id, "run stopped: guardrail limit reached")
                        .await
                    {
                        error!(run_id = %run_id, store_error = %cleanup_err, "failed to cleanup orphaned steps");
                    }
                    RunStatus::Cancelled
                } else {
                    // Written before the failure so a failed run keeps the
                    // verdict the handler set before returning its error.
                    if let Some(output) = ctx.output()
                        && let Err(store_err) = self
                            .store
                            .update_run(
                                run_id,
                                RunUpdate {
                                    output: Some(output.clone()),
                                    ..RunUpdate::default()
                                },
                            )
                            .await
                    {
                        error!(run_id = %run_id, store_error = %store_err, "failed to persist run output");
                    }
                    self.fail_or_schedule_retry(
                        run_id,
                        &err.to_string(),
                        is_run_retryable(&err),
                        Some(ctx.total_cost_usd()),
                        Some(total_duration),
                    )
                    .await
                    .unwrap_or_else(|store_err| {
                        error!(run_id = %run_id, store_error = %store_err, "failed to persist run failure");
                        RunStatus::Failed
                    })
                };

                if matches!(err, EngineError::RunBudgetExceeded { .. }) {
                    self.on_run_budget_exceeded(workflow_name, run_id, &err);
                }

                error!(run_id = %run_id, status = %final_status, error = %err, "run stopped");

                self.publish_run_status_changed(
                    workflow_name,
                    run_id,
                    final_status,
                    Some(err.to_string()),
                    ctx,
                    total_duration,
                    run_labels,
                );

                #[cfg(feature = "prometheus")]
                self.emit_run_metrics(workflow_name, final_status, total_duration, ctx);

                return Err(err);
            }
        }

        self.publish_run_status_changed(
            workflow_name,
            run_id,
            final_status,
            None,
            ctx,
            total_duration,
            run_labels,
        );

        #[cfg(feature = "prometheus")]
        self.emit_run_metrics(workflow_name, final_status, total_duration, ctx);

        Ok(WorkflowResult {
            run: final_run,
            steps: ctx.step_results().to_vec(),
        })
    }

    /// Publish [`Event::ApprovalRequested`] for an approval gate that opened
    /// on `run_id`, with the requirement recorded when the gate opened.
    pub(crate) async fn publish_approval_requested(
        &self,
        run_id: Uuid,
        step_id: Uuid,
        message: &str,
    ) -> Result<(), EngineError> {
        let requirement = self
            .store
            .get_step(step_id)
            .await?
            .and_then(|s| s.approval_requirement);
        self.event_publisher
            .publish(Event::ApprovalRequested(ApprovalRequestedEvent {
                run_id,
                step_id,
                message: message.to_string(),
                requirement,
                at: Utc::now(),
            }));
        Ok(())
    }

    /// Fail every ancestor of a child run, closest first.
    ///
    /// Used when a gate inside a sub-workflow is rejected: the child run
    /// fails, and the runs suspended with it (its parent, up to the root)
    /// must not stay suspended on a child that will never resume. Each
    /// ancestor goes through [`fail_or_schedule_retry`](Self::fail_or_schedule_retry)
    /// without a retry, which also fails its open `Workflow` step. A run that
    /// is not a child of a sub-workflow has no ancestor: nothing happens.
    ///
    /// A paused ancestor is not failed: the root of the chain is set to resume
    /// to `Pending`, so its replay observes the failed child once an operator
    /// resumes it.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Store`] if a run of the chain does not exist or
    /// its failure cannot be persisted.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ironflow_engine::engine::Engine;
    /// use ironflow_engine::error::EngineError;
    /// use uuid::Uuid;
    ///
    /// # async fn example(engine: &Engine, child_run_id: Uuid) -> Result<(), EngineError> {
    /// engine
    ///     .fail_ancestors(child_run_id, "approval rejected in a sub-workflow")
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn fail_ancestors(&self, run_id: Uuid, reason: &str) -> Result<(), EngineError> {
        let mut current = self
            .store
            .get_run(run_id)
            .await?
            .ok_or(EngineError::Store(StoreError::RunNotFound(run_id)))?;
        // Labels are data: a chain that loops back on itself stops there.
        let mut visited = HashSet::from([run_id]);

        while let Some(parent_id) = chain_parent(&current) {
            if !visited.insert(parent_id) {
                break;
            }
            let status = self
                .fail_or_schedule_retry(parent_id, reason, false, None, None)
                .await?;
            // A paused chain waits for its operator: the root replays on
            // resume and observes the failed child, like a cancelled one.
            if status == RunStatus::Paused {
                self.requeue_paused_root(&current).await?;
                break;
            }
            info!(
                run_id = %run_id,
                ancestor_run_id = %parent_id,
                status = %status,
                "ancestor run failed with its child"
            );
            current = self
                .store
                .get_run(parent_id)
                .await?
                .ok_or(EngineError::Store(StoreError::RunNotFound(parent_id)))?;
        }

        Ok(())
    }

    /// Emit Prometheus metrics for a completed run.
    #[cfg(feature = "prometheus")]
    fn emit_run_metrics(
        &self,
        workflow_name: &str,
        status: RunStatus,
        duration_ms: u64,
        ctx: &WorkflowContext,
    ) {
        let status_str = status.to_string();
        let wf = workflow_name.to_string();

        counter!(RUNS_TOTAL, "workflow" => wf.clone(), "status" => status_str.clone()).increment(1);
        histogram!(RUN_DURATION_SECONDS, "workflow" => wf.clone(), "status" => status_str)
            .record(duration_ms as f64 / 1000.0);
        histogram!(RUN_COST_USD, "workflow" => wf.clone()).record(
            ctx.total_cost_usd()
                .to_string()
                .parse::<f64>()
                .unwrap_or(0.0),
        );
        gauge!(RUNS_ACTIVE, "workflow" => wf).decrement(1.0);
    }

    /// Record the metric and publish the audit event for a run that hit its
    /// cost cap.
    ///
    /// A non-[`RunBudgetExceeded`](EngineError::RunBudgetExceeded) error is
    /// ignored, so callers can pass the error unconditionally.
    fn on_run_budget_exceeded(&self, workflow_name: &str, run_id: Uuid, err: &EngineError) {
        let EngineError::RunBudgetExceeded {
            limit_usd,
            spent_usd,
            step_budget_usd,
            ..
        } = err
        else {
            return;
        };

        #[cfg(feature = "prometheus")]
        counter!(
            RUN_BUDGET_EXCEEDED_TOTAL,
            "workflow" => workflow_name.to_string(),
            "scope" => "run",
        )
        .increment(1);

        self.event_publisher
            .publish(Event::RunBudgetExceeded(RunBudgetExceededEvent {
                run_id,
                workflow_name: workflow_name.to_string(),
                limit_usd: *limit_usd,
                spent_usd: *spent_usd,
                step_budget_usd: *step_budget_usd,
                at: Utc::now(),
            }));
    }

    /// Publish a run status changed event to all registered subscribers.
    ///
    /// `from` is always `Running` because `finalize_run` is only called
    /// from a running state.
    #[allow(clippy::too_many_arguments)]
    fn publish_run_status_changed(
        &self,
        workflow_name: &str,
        run_id: Uuid,
        to: RunStatus,
        error: Option<String>,
        ctx: &WorkflowContext,
        duration_ms: u64,
        labels: HashMap<String, String>,
    ) {
        let now = Utc::now();
        let cost_usd = ctx.total_cost_usd();
        let wf = workflow_name.to_string();

        self.event_publisher
            .publish(Event::RunStatusChanged(RunStatusChangedEvent {
                run_id,
                workflow_name: wf.clone(),
                from: RunStatus::Running,
                to,
                error: error.clone(),
                cost_usd,
                duration_ms,
                labels: labels.clone(),
                at: now,
            }));

        if to == RunStatus::Failed {
            self.event_publisher
                .publish(Event::RunFailed(RunFailedEvent {
                    run_id,
                    workflow_name: wf,
                    error,
                    cost_usd,
                    duration_ms,
                    labels,
                    at: now,
                }));
        }
    }
}

impl fmt::Debug for Engine {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Engine")
            .field("handlers", &self.handlers.keys().collect::<Vec<_>>())
            .finish_non_exhaustive()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::ShellConfig;
    use crate::handler::{HandlerFuture, WorkflowHandler};
    use ironflow_core::providers::claude::ClaudeCodeProvider;
    use ironflow_core::providers::record_replay::RecordReplayProvider;
    use ironflow_store::memory::InMemoryStore;
    use ironflow_store::models::{MAX_PRIORITY, MIN_PRIORITY, StepApproval, StepStatus};
    use serde_json::json;

    // Test handler that echoes a message via shell
    struct EchoWorkflow;

    impl WorkflowHandler for EchoWorkflow {
        fn name(&self) -> &str {
            "echo-workflow"
        }

        fn describe(&self) -> WorkflowInfo {
            WorkflowInfo {
                description: "A simple workflow that echoes hello".to_string(),
                source_code: None,
                sub_workflows: Vec::new(),
                category: None,
                version: self.version().map(str::to_string),
                compatible_versions: Vec::new(),
                input_schema: None,
                default_labels: HashMap::new(),
                schedule: self.schedule().cloned(),
                default_max_cost_usd: self.default_max_cost_usd(),
                priority: self.priority(),
            }
        }

        fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
            Box::pin(async move {
                ctx.shell("greet", ShellConfig::new("echo hello")).await?;
                Ok(())
            })
        }
    }

    // Test handler that fails
    struct FailingWorkflow;

    impl WorkflowHandler for FailingWorkflow {
        fn name(&self) -> &str {
            "failing-workflow"
        }

        fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
            Box::pin(async move {
                ctx.shell("fail", ShellConfig::new("exit 1")).await?;
                Ok(())
            })
        }
    }

    fn create_test_engine() -> Engine {
        let store = Arc::new(InMemoryStore::new());
        let inner = ClaudeCodeProvider::new();
        let provider: Arc<dyn AgentProvider> = Arc::new(RecordReplayProvider::replay(
            inner,
            "/tmp/ironflow-fixtures",
        ));
        Engine::new(store, provider)
    }

    #[test]
    fn engine_new_creates_instance() {
        let engine = create_test_engine();
        assert_eq!(engine.handler_names().len(), 0);
    }

    #[test]
    fn execution_mode_defaults_to_local() {
        let engine = create_test_engine();
        assert_eq!(engine.execution_mode(), ExecutionMode::Local);
    }

    #[test]
    fn with_execution_mode_overrides_the_default() {
        let engine = create_test_engine().with_execution_mode(ExecutionMode::Workers);
        assert_eq!(engine.execution_mode(), ExecutionMode::Workers);
    }

    #[test]
    fn engine_register_handler() {
        let mut engine = create_test_engine();
        let result = engine.register(EchoWorkflow);
        assert!(result.is_ok());
        assert_eq!(engine.handler_names().len(), 1);
        assert!(engine.handler_names().contains(&"echo-workflow"));
    }

    #[test]
    fn engine_register_duplicate_returns_error() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();
        let result = engine.register(EchoWorkflow);
        assert!(result.is_err());
    }

    #[test]
    fn engine_get_handler_found() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();
        let handler = engine.get_handler("echo-workflow");
        assert!(handler.is_some());
    }

    #[test]
    fn engine_get_handler_not_found() {
        let engine = create_test_engine();
        let handler = engine.get_handler("nonexistent");
        assert!(handler.is_none());
    }

    #[test]
    fn engine_handler_names_lists_all() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();
        engine.register(FailingWorkflow).unwrap();
        let names = engine.handler_names();
        assert_eq!(names.len(), 2);
        assert!(names.contains(&"echo-workflow"));
        assert!(names.contains(&"failing-workflow"));
    }

    #[test]
    fn engine_handler_info_returns_description() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();
        let info = engine.handler_info("echo-workflow");
        assert!(info.is_some());
        let info = info.unwrap();
        assert_eq!(info.description, "A simple workflow that echoes hello");
    }

    struct CategorizedWorkflow;

    impl WorkflowHandler for CategorizedWorkflow {
        fn name(&self) -> &str {
            "categorized"
        }
        fn category(&self) -> Option<&str> {
            Some("data/etl")
        }
        fn execute<'a>(
            &'a self,
            _ctx: &'a mut WorkflowContext,
        ) -> crate::handler::HandlerFuture<'a> {
            Box::pin(async move { Ok(()) })
        }
    }

    #[test]
    fn engine_default_describe_propagates_category() {
        let mut engine = create_test_engine();
        engine.register(CategorizedWorkflow).unwrap();
        let info = engine.handler_info("categorized").unwrap();
        assert_eq!(info.category.as_deref(), Some("data/etl"));
    }

    #[test]
    fn engine_default_describe_without_category() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();
        let info = engine.handler_info("echo-workflow").unwrap();
        assert!(info.category.is_none());
    }

    // -----------------------------------------------------------------------
    // Schedule tests
    // -----------------------------------------------------------------------

    struct ScheduledWorkflow {
        schedule: CronSchedule,
    }

    impl ScheduledWorkflow {
        fn new() -> Self {
            Self {
                schedule: CronSchedule::new("0 0 * * * *").unwrap(),
            }
        }
    }

    impl WorkflowHandler for ScheduledWorkflow {
        fn name(&self) -> &str {
            "scheduled"
        }
        fn schedule(&self) -> Option<&CronSchedule> {
            Some(&self.schedule)
        }
        fn execute<'a>(
            &'a self,
            _ctx: &'a mut WorkflowContext,
        ) -> crate::handler::HandlerFuture<'a> {
            Box::pin(async move { Ok(()) })
        }
    }

    #[test]
    fn engine_default_describe_propagates_schedule() {
        let mut engine = create_test_engine();
        engine.register(ScheduledWorkflow::new()).unwrap();
        let info = engine.handler_info("scheduled").unwrap();
        assert_eq!(
            info.schedule.as_ref().map(|s| s.as_str()),
            Some("0 0 * * * *")
        );
    }

    #[test]
    fn engine_default_describe_without_schedule() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();
        let info = engine.handler_info("echo-workflow").unwrap();
        assert!(info.schedule.is_none());
    }

    #[test]
    fn scheduled_handlers_returns_only_scheduled() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();
        engine.register(ScheduledWorkflow::new()).unwrap();
        engine.register(FailingWorkflow).unwrap();

        let scheduled = engine.scheduled_handlers();
        assert_eq!(scheduled.len(), 1);
        assert_eq!(scheduled[0].0, "scheduled");
        assert_eq!(scheduled[0].1.as_str(), "0 0 * * * *");
    }

    #[test]
    fn scheduled_handlers_empty_when_none_scheduled() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();
        engine.register(FailingWorkflow).unwrap();

        let scheduled = engine.scheduled_handlers();
        assert!(scheduled.is_empty());
    }

    struct BadCategoryWorkflow(&'static str);

    impl WorkflowHandler for BadCategoryWorkflow {
        fn name(&self) -> &str {
            "bad-category"
        }
        fn category(&self) -> Option<&str> {
            Some(self.0)
        }
        fn execute<'a>(
            &'a self,
            _ctx: &'a mut WorkflowContext,
        ) -> crate::handler::HandlerFuture<'a> {
            Box::pin(async move { Ok(()) })
        }
    }

    #[test]
    fn engine_register_rejects_empty_category() {
        let mut engine = create_test_engine();
        let err = engine.register(BadCategoryWorkflow("")).unwrap_err();
        match err {
            EngineError::InvalidWorkflow(msg) => assert!(msg.contains("empty category")),
            other => panic!("expected InvalidWorkflow, got {other:?}"),
        }
    }

    #[test]
    fn engine_register_rejects_leading_slash_category() {
        let mut engine = create_test_engine();
        let err = engine
            .register(BadCategoryWorkflow("/data/etl"))
            .unwrap_err();
        match err {
            EngineError::InvalidWorkflow(msg) => assert!(msg.contains("leading '/'")),
            other => panic!("expected InvalidWorkflow, got {other:?}"),
        }
    }

    #[test]
    fn engine_register_rejects_trailing_slash_category() {
        let mut engine = create_test_engine();
        let err = engine
            .register(BadCategoryWorkflow("data/etl/"))
            .unwrap_err();
        match err {
            EngineError::InvalidWorkflow(msg) => assert!(msg.contains("trailing '/'")),
            other => panic!("expected InvalidWorkflow, got {other:?}"),
        }
    }

    #[test]
    fn engine_register_rejects_double_slash_category() {
        let mut engine = create_test_engine();
        let err = engine
            .register(BadCategoryWorkflow("data//etl"))
            .unwrap_err();
        match err {
            EngineError::InvalidWorkflow(msg) => assert!(msg.contains("empty segment")),
            other => panic!("expected InvalidWorkflow, got {other:?}"),
        }
    }

    #[test]
    fn engine_register_rejects_whitespace_only_segment_category() {
        let mut engine = create_test_engine();
        let err = engine
            .register(BadCategoryWorkflow("data/ /etl"))
            .unwrap_err();
        match err {
            EngineError::InvalidWorkflow(msg) => assert!(msg.contains("whitespace-only segment")),
            other => panic!("expected InvalidWorkflow, got {other:?}"),
        }
    }

    #[test]
    fn engine_register_accepts_valid_nested_category() {
        let mut engine = create_test_engine();
        assert!(engine.register(CategorizedWorkflow).is_ok());
    }

    #[tokio::test]
    async fn engine_unknown_workflow_returns_error() {
        let engine = create_test_engine();
        let result = engine
            .run_handler("unknown", TriggerKind::Manual, json!({}))
            .await;
        assert!(result.is_err());
        match result {
            Err(EngineError::InvalidWorkflow(msg)) => {
                assert!(msg.contains("no handler registered"));
            }
            _ => panic!("expected InvalidWorkflow error"),
        }
    }

    #[tokio::test]
    async fn engine_enqueue_handler_creates_pending_run() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();

        let run = engine
            .enqueue_handler("echo-workflow", TriggerKind::Manual, json!({}), 0)
            .await
            .unwrap();
        assert_eq!(run.status.state, RunStatus::Pending);
        assert_eq!(run.workflow_name, "echo-workflow");
    }

    #[tokio::test]
    async fn enqueue_handler_leaves_the_run_unattributed() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();

        let run = engine
            .enqueue_handler("echo-workflow", TriggerKind::Manual, json!({}), 0)
            .await
            .unwrap();

        assert!(run.created_by.is_none());
    }

    #[tokio::test]
    async fn enqueue_handler_with_options_records_the_author() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();
        let actor = RunActor::User {
            user_id: Uuid::now_v7(),
        };

        let run = engine
            .enqueue_handler_with_options(
                "echo-workflow",
                TriggerKind::Api,
                json!({}),
                EnqueueOptions {
                    created_by: Some(actor.clone()),
                    ..Default::default()
                },
            )
            .await
            .unwrap()
            .into_run();

        assert_eq!(run.created_by, Some(actor));
    }

    #[tokio::test]
    async fn enqueue_handler_with_options_accepts_no_author() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();

        let run = engine
            .enqueue_handler_with_options(
                "echo-workflow",
                TriggerKind::Cron {
                    schedule: "0 * * * * *".to_string(),
                    schedule_id: None,
                    scheduled_for: None,
                },
                json!({}),
                EnqueueOptions::default(),
            )
            .await
            .unwrap()
            .into_run();

        assert!(run.created_by.is_none());
    }

    #[tokio::test]
    async fn enqueue_handler_with_options_stores_concurrency_limits() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();
        let limits = vec![
            ConcurrencyLimit::new("repo:acme", 2),
            ConcurrencyLimit::new("tenant:42", 5),
        ];

        let run = engine
            .enqueue_handler_with_options(
                "echo-workflow",
                TriggerKind::Api,
                json!({}),
                EnqueueOptions {
                    concurrency_limits: limits.clone(),
                    ..Default::default()
                },
            )
            .await
            .unwrap()
            .into_run();

        assert_eq!(run.concurrency_limits, limits);
    }

    #[tokio::test]
    async fn enqueue_rejects_invalid_concurrency_limits() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();

        let invalid = [
            vec![ConcurrencyLimit::new("repo:acme", 0)],
            vec![ConcurrencyLimit::new("", 1)],
            vec![
                ConcurrencyLimit::new("repo:acme", 1),
                ConcurrencyLimit::new("repo:acme", 2),
            ],
        ];
        for concurrency_limits in invalid {
            let err = engine
                .enqueue_handler_with_options(
                    "echo-workflow",
                    TriggerKind::Api,
                    json!({}),
                    EnqueueOptions {
                        concurrency_limits,
                        ..Default::default()
                    },
                )
                .await
                .unwrap_err();
            assert!(
                matches!(err, EngineError::InvalidConcurrencyLimit(_)),
                "{err:?}"
            );
        }

        // Validated before the handler lookup.
        let err = engine
            .enqueue_handler_with_options(
                "not-registered",
                TriggerKind::Api,
                json!({}),
                EnqueueOptions {
                    concurrency_limits: vec![ConcurrencyLimit::new("repo:acme", 0)],
                    ..Default::default()
                },
            )
            .await
            .unwrap_err();
        assert!(
            matches!(err, EngineError::InvalidConcurrencyLimit(_)),
            "{err:?}"
        );

        let page = engine
            .store()
            .list_runs(RunFilter::default(), 1, 10)
            .await
            .unwrap();
        assert_eq!(page.total, 0, "no run may be created");
    }

    struct UrgentWorkflow;

    impl WorkflowHandler for UrgentWorkflow {
        fn name(&self) -> &str {
            "urgent-workflow"
        }

        fn priority(&self) -> i16 {
            60
        }

        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
            Box::pin(async { Ok(()) })
        }
    }

    #[tokio::test]
    async fn enqueue_priority_defaults_to_the_handler_priority() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();
        engine.register(UrgentWorkflow).unwrap();

        let echo = engine
            .enqueue_handler_with_options(
                "echo-workflow",
                TriggerKind::Api,
                json!({}),
                EnqueueOptions::default(),
            )
            .await
            .unwrap()
            .into_run();
        assert_eq!(echo.priority, 0);

        let urgent = engine
            .enqueue_handler_with_options(
                "urgent-workflow",
                TriggerKind::Api,
                json!({}),
                EnqueueOptions::default(),
            )
            .await
            .unwrap()
            .into_run();
        assert_eq!(urgent.priority, 60);
    }

    #[tokio::test]
    async fn enqueue_priority_explicit_value_overrides_the_handler() {
        let mut engine = create_test_engine();
        engine.register(UrgentWorkflow).unwrap();

        let run = engine
            .enqueue_handler_with_options(
                "urgent-workflow",
                TriggerKind::Api,
                json!({}),
                EnqueueOptions {
                    priority: Some(-20),
                    ..Default::default()
                },
            )
            .await
            .unwrap()
            .into_run();
        assert_eq!(run.priority, -20);

        let stored = engine.store().get_run(run.id).await.unwrap().unwrap();
        assert_eq!(stored.priority, -20);
    }

    #[tokio::test]
    async fn enqueue_priority_out_of_range_is_rejected() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();

        for priority in [MAX_PRIORITY + 1, MIN_PRIORITY - 1] {
            let err = engine
                .enqueue_handler_with_options(
                    "echo-workflow",
                    TriggerKind::Api,
                    json!({}),
                    EnqueueOptions {
                        priority: Some(priority),
                        ..Default::default()
                    },
                )
                .await
                .unwrap_err();
            assert!(matches!(err, EngineError::InvalidPriority(_)), "{err:?}");
        }

        // Validated before the handler lookup.
        let err = engine
            .enqueue_handler_with_options(
                "not-registered",
                TriggerKind::Api,
                json!({}),
                EnqueueOptions {
                    priority: Some(MAX_PRIORITY + 1),
                    ..Default::default()
                },
            )
            .await
            .unwrap_err();
        assert!(matches!(err, EngineError::InvalidPriority(_)), "{err:?}");

        let page = engine
            .store()
            .list_runs(RunFilter::default(), 1, 10)
            .await
            .unwrap();
        assert_eq!(page.total, 0, "no run may be created");
    }

    #[tokio::test]
    async fn enqueue_priority_bounds_are_accepted() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();

        for priority in [MIN_PRIORITY, MAX_PRIORITY] {
            let run = engine
                .enqueue_handler_with_options(
                    "echo-workflow",
                    TriggerKind::Api,
                    json!({}),
                    EnqueueOptions {
                        priority: Some(priority),
                        ..Default::default()
                    },
                )
                .await
                .unwrap()
                .into_run();
            assert_eq!(run.priority, priority);
        }
    }

    struct GpuWorkflow;

    impl WorkflowHandler for GpuWorkflow {
        fn name(&self) -> &str {
            "gpu-workflow"
        }

        fn required_worker_tags(&self) -> Vec<String> {
            vec!["gpu".to_string()]
        }

        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
            Box::pin(async move { Ok(()) })
        }
    }

    #[tokio::test]
    async fn enqueue_merges_handler_and_request_worker_tags() {
        let mut engine = create_test_engine();
        engine.register(GpuWorkflow).unwrap();

        let run = engine
            .enqueue_handler_with_options(
                "gpu-workflow",
                TriggerKind::Api,
                json!({}),
                EnqueueOptions {
                    worker_tags: vec!["region:eu".to_string(), "gpu".to_string()],
                    ..Default::default()
                },
            )
            .await
            .unwrap()
            .into_run();

        assert_eq!(
            run.worker_tags,
            vec!["gpu".to_string(), "region:eu".to_string()]
        );
    }

    #[tokio::test]
    async fn enqueue_without_worker_tags_keeps_handler_tags() {
        let mut engine = create_test_engine();
        engine.register(GpuWorkflow).unwrap();
        engine.register(EchoWorkflow).unwrap();

        let gpu = engine
            .enqueue_handler("gpu-workflow", TriggerKind::Api, json!({}), 0)
            .await
            .unwrap();
        assert_eq!(gpu.worker_tags, vec!["gpu".to_string()]);

        let echo = engine
            .enqueue_handler("echo-workflow", TriggerKind::Api, json!({}), 0)
            .await
            .unwrap();
        assert!(echo.worker_tags.is_empty());
    }

    #[tokio::test]
    async fn enqueue_rejects_invalid_worker_tags() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();

        for worker_tags in [
            vec!["bad,tag".to_string()],
            vec![" ".to_string()],
            vec!["x".repeat(65)],
        ] {
            let err = engine
                .enqueue_handler_with_options(
                    "echo-workflow",
                    TriggerKind::Api,
                    json!({}),
                    EnqueueOptions {
                        worker_tags,
                        ..Default::default()
                    },
                )
                .await
                .unwrap_err();
            assert!(matches!(err, EngineError::InvalidWorkerTag(_)), "{err:?}");
        }

        // Validated before the handler lookup.
        let err = engine
            .enqueue_handler_with_options(
                "not-registered",
                TriggerKind::Api,
                json!({}),
                EnqueueOptions {
                    worker_tags: vec!["bad,tag".to_string()],
                    ..Default::default()
                },
            )
            .await
            .unwrap_err();
        assert!(matches!(err, EngineError::InvalidWorkerTag(_)), "{err:?}");
    }

    #[test]
    fn worker_tags_are_unset_by_default() {
        let engine = create_test_engine();
        assert!(engine.worker_tags().is_none());
    }

    #[test]
    fn set_worker_tags_stores_the_tags() {
        let mut engine = create_test_engine();
        engine.set_worker_tags(vec!["gpu".to_string()]);
        assert_eq!(engine.worker_tags(), Some(&["gpu".to_string()][..]));

        engine.set_worker_tags(Vec::new());
        assert_eq!(engine.worker_tags(), Some(&[][..]));
    }

    #[tokio::test]
    async fn run_handler_records_handler_worker_tags() {
        let mut engine = create_test_engine();
        engine.register(GpuWorkflow).unwrap();

        let result = engine
            .run_handler("gpu-workflow", TriggerKind::Manual, json!({}))
            .await
            .unwrap();
        assert_eq!(result.run.worker_tags, vec!["gpu".to_string()]);
    }

    #[tokio::test]
    async fn run_handler_leaves_the_run_unattributed() {
        let mut engine = create_test_engine();
        engine.register(EchoWorkflow).unwrap();

        let run = engine
            .run_handler("echo-workflow", TriggerKind::Manual, json!({}))
            .await
            .unwrap()
            .run;

        assert!(run.created_by.is_none());
    }

    #[tokio::test]
    async fn run_handler_priority_comes_from_the_handler() {
        let mut engine = create_test_engine();
        engine.register(UrgentWorkflow).unwrap();

        let run = engine
            .run_handler("urgent-workflow", TriggerKind::Manual, json!({}))
            .await
            .unwrap()
            .run;

        assert_eq!(run.priority, 60);
    }

    #[tokio::test]
    async fn engine_register_boxed() {
        let mut engine = create_test_engine();
        let handler: Box<dyn WorkflowHandler> = Box::new(EchoWorkflow);
        let result = engine.register_boxed(handler);
        assert!(result.is_ok());
        assert_eq!(engine.handler_names().len(), 1);
    }

    #[tokio::test]
    async fn engine_store_and_provider_accessors() {
        let store = Arc::new(InMemoryStore::new());
        let inner = ClaudeCodeProvider::new();
        let provider: Arc<dyn AgentProvider> = Arc::new(RecordReplayProvider::replay(
            inner,
            "/tmp/ironflow-fixtures",
        ));
        let engine = Engine::new(store.clone(), provider.clone());

        // Verify accessors return references
        let _ = engine.store();
        let _ = engine.provider();
    }

    // -----------------------------------------------------------------------
    // Operation trait tests
    // -----------------------------------------------------------------------

    use crate::operation::{Operation, OperationContext};
    use async_trait::async_trait;
    use ironflow_core::error::OperationError;
    use ironflow_store::models::StepKind;

    struct FakeGitlabOp {
        project_id: u64,
        title: String,
    }

    #[async_trait]
    impl Operation for FakeGitlabOp {
        fn kind(&self) -> &str {
            "gitlab"
        }

        async fn execute(&self, _ctx: &OperationContext) -> Result<Value, OperationError> {
            Ok(json!({
                "issue_id": 42,
                "project_id": self.project_id,
                "title": self.title,
            }))
        }

        fn input(&self) -> Option<Value> {
            Some(json!({
                "project_id": self.project_id,
                "title": self.title,
            }))
        }
    }

    struct FailingOp;

    #[async_trait]
    impl Operation for FailingOp {
        fn kind(&self) -> &str {
            "broken-service"
        }

        async fn execute(&self, _ctx: &OperationContext) -> Result<Value, OperationError> {
            Err(OperationError::Http {
                status: None,
                message: "service unavailable".to_string(),
            })
        }
    }

    struct OperationWorkflow;

    impl WorkflowHandler for OperationWorkflow {
        fn name(&self) -> &str {
            "operation-workflow"
        }

        fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
            Box::pin(async move {
                let op = FakeGitlabOp {
                    project_id: 123,
                    title: "Bug report".to_string(),
                };
                ctx.operation("create-issue", &op).await?;
                Ok(())
            })
        }
    }

    struct FailingOperationWorkflow;

    impl WorkflowHandler for FailingOperationWorkflow {
        fn name(&self) -> &str {
            "failing-operation-workflow"
        }

        fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
            Box::pin(async move {
                ctx.operation("broken-call", &FailingOp).await?;
                Ok(())
            })
        }
    }

    struct MixedWorkflow;

    impl WorkflowHandler for MixedWorkflow {
        fn name(&self) -> &str {
            "mixed-workflow"
        }

        fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
            Box::pin(async move {
                ctx.shell("build", ShellConfig::new("echo built")).await?;
                let op = FakeGitlabOp {
                    project_id: 456,
                    title: "Deploy done".to_string(),
                };
                let result = ctx.operation("notify-gitlab", &op).await?;
                assert_eq!(result.output["issue_id"], 42);
                Ok(())
            })
        }
    }

    #[tokio::test]
    async fn operation_step_happy_path() {
        let mut engine = create_test_engine();
        engine.register(OperationWorkflow).unwrap();

        let run = engine
            .run_handler("operation-workflow", TriggerKind::Manual, json!({}))
            .await
            .unwrap()
            .run;

        assert_eq!(run.status.state, RunStatus::Completed);

        let steps = engine.store().list_steps(run.id).await.unwrap();

        assert_eq!(steps.len(), 1);
        assert_eq!(steps[0].name, "create-issue");
        assert_eq!(steps[0].kind, StepKind::Custom("gitlab".to_string()));
        assert_eq!(
            steps[0].status.state,
            ironflow_store::models::StepStatus::Completed
        );

        let output = steps[0].output.as_ref().unwrap();
        assert_eq!(output["issue_id"], 42);
        assert_eq!(output["project_id"], 123);

        let input = steps[0].input.as_ref().unwrap();
        assert_eq!(input["project_id"], 123);
        assert_eq!(input["title"], "Bug report");
    }

    #[tokio::test]
    async fn operation_step_failure_marks_run_failed() {
        let mut engine = create_test_engine();
        engine.register(FailingOperationWorkflow).unwrap();

        let result = engine
            .run_handler("failing-operation-workflow", TriggerKind::Manual, json!({}))
            .await;

        assert!(result.is_err());
    }

    #[tokio::test]
    async fn operation_mixed_with_shell_steps() {
        let mut engine = create_test_engine();
        engine.register(MixedWorkflow).unwrap();

        let run = engine
            .run_handler("mixed-workflow", TriggerKind::Manual, json!({}))
            .await
            .unwrap()
            .run;

        assert_eq!(run.status.state, RunStatus::Completed);

        let steps = engine.store().list_steps(run.id).await.unwrap();

        assert_eq!(steps.len(), 2);
        assert_eq!(steps[0].kind, StepKind::Shell);
        assert_eq!(steps[1].kind, StepKind::Custom("gitlab".to_string()));
        assert_eq!(steps[0].position, 0);
        assert_eq!(steps[1].position, 1);
    }

    // -----------------------------------------------------------------------
    // Approval + resume tests
    // -----------------------------------------------------------------------

    use crate::config::ApprovalConfig;

    struct SingleApprovalWorkflow;

    impl WorkflowHandler for SingleApprovalWorkflow {
        fn name(&self) -> &str {
            "single-approval"
        }

        fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
            Box::pin(async move {
                ctx.shell("build", ShellConfig::new("echo built")).await?;
                ctx.approval("gate", ApprovalConfig::new("OK?")).await?;
                ctx.shell("deploy", ShellConfig::new("echo deployed"))
                    .await?;
                Ok(())
            })
        }
    }

    struct DoubleApprovalWorkflow;

    impl WorkflowHandler for DoubleApprovalWorkflow {
        fn name(&self) -> &str {
            "double-approval"
        }

        fn execute<'a>(&'a self, ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
            Box::pin(async move {
                ctx.shell("build", ShellConfig::new("echo built")).await?;
                ctx.approval("staging-gate", ApprovalConfig::new("Deploy staging?"))
                    .await?;
                ctx.shell("deploy-staging", ShellConfig::new("echo staging"))
                    .await?;
                ctx.approval("prod-gate", ApprovalConfig::new("Deploy prod?"))
                    .await?;
                ctx.shell("deploy-prod", ShellConfig::new("echo prod"))
                    .await?;
                Ok(())
            })
        }
    }

    #[tokio::test]
    async fn approval_pauses_run() {
        let mut engine = create_test_engine();
        engine.register(SingleApprovalWorkflow).unwrap();

        let run = engine
            .run_handler("single-approval", TriggerKind::Manual, json!({}))
            .await
            .unwrap()
            .run;

        assert_eq!(run.status.state, RunStatus::AwaitingApproval);

        let steps = engine.store().list_steps(run.id).await.unwrap();
        assert_eq!(steps.len(), 2); // build + approval gate
        assert_eq!(steps[0].kind, StepKind::Shell);
        assert_eq!(steps[0].status.state, StepStatus::Completed);
        assert_eq!(steps[1].kind, StepKind::Approval);
        assert_eq!(steps[1].status.state, StepStatus::AwaitingApproval);
    }

    /// Record one vote on the open approval gate of `run_id`, then move the
    /// run to `Running`, the way the approve route does.
    async fn approve_open_gate(store: &Arc<dyn Store>, run_id: Uuid) {
        let gate = store
            .list_steps(run_id)
            .await
            .unwrap()
            .into_iter()
            .find(|step| step.status.state == StepStatus::AwaitingApproval)
            .expect("an open approval gate");
        store
            .record_step_approval(
                gate.id,
                StepApproval {
                    user_id: Uuid::now_v7(),
                    approved_by: "tester".to_string(),
                    at: Utc::now(),
                },
            )
            .await
            .unwrap();
        store
            .update_run_status(run_id, RunStatus::Running)
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn approval_resume_completes_run() {
        let mut engine = create_test_engine();
        engine.register(SingleApprovalWorkflow).unwrap();

        // First execution: pauses at approval
        let run = engine
            .run_handler("single-approval", TriggerKind::Manual, json!({}))
            .await
            .unwrap()
            .run;
        assert_eq!(run.status.state, RunStatus::AwaitingApproval);

        // Simulate approval: one vote, then transition to Running
        approve_open_gate(engine.store(), run.id).await;

        // Resume: replays build, skips approval, executes deploy
        let resumed = engine.resume_run(run.id).await.unwrap().run;
        assert_eq!(resumed.status.state, RunStatus::Completed);

        let steps = engine.store().list_steps(run.id).await.unwrap();
        assert_eq!(steps.len(), 3); // build + approval + deploy
        assert_eq!(steps[0].name, "build");
        assert_eq!(steps[0].status.state, StepStatus::Completed);
        assert_eq!(steps[1].name, "gate");
        assert_eq!(steps[1].kind, StepKind::Approval);
        assert_eq!(steps[1].status.state, StepStatus::Completed);
        assert_eq!(steps[2].name, "deploy");
        assert_eq!(steps[2].status.state, StepStatus::Completed);
    }

    #[tokio::test]
    async fn double_approval_two_resumes() {
        let mut engine = create_test_engine();
        engine.register(DoubleApprovalWorkflow).unwrap();

        // First execution: pauses at staging-gate
        let run = engine
            .run_handler("double-approval", TriggerKind::Manual, json!({}))
            .await
            .unwrap()
            .run;
        assert_eq!(run.status.state, RunStatus::AwaitingApproval);

        let steps = engine.store().list_steps(run.id).await.unwrap();
        assert_eq!(steps.len(), 2); // build + staging-gate

        // First approval
        approve_open_gate(engine.store(), run.id).await;

        let resumed = engine.resume_run(run.id).await.unwrap().run;
        assert_eq!(resumed.status.state, RunStatus::AwaitingApproval);

        let steps = engine.store().list_steps(run.id).await.unwrap();
        assert_eq!(steps.len(), 4); // build + staging-gate + deploy-staging + prod-gate

        // Second approval
        approve_open_gate(engine.store(), run.id).await;

        let final_run = engine.resume_run(run.id).await.unwrap().run;
        assert_eq!(final_run.status.state, RunStatus::Completed);

        let steps = engine.store().list_steps(run.id).await.unwrap();
        assert_eq!(steps.len(), 5);
        assert_eq!(steps[0].name, "build");
        assert_eq!(steps[1].name, "staging-gate");
        assert_eq!(steps[2].name, "deploy-staging");
        assert_eq!(steps[3].name, "prod-gate");
        assert_eq!(steps[4].name, "deploy-prod");

        for step in &steps {
            assert_eq!(step.status.state, StepStatus::Completed);
        }
    }

    // -----------------------------------------------------------------------
    // fail_orphaned_steps tests
    // -----------------------------------------------------------------------

    use ironflow_store::models::{NewStep, StepUpdate, step_trace_id};

    async fn create_step_with_status(
        store: &Arc<dyn Store>,
        run_id: Uuid,
        name: &str,
        position: u32,
        status: StepStatus,
    ) -> ironflow_store::models::Step {
        let step = store
            .create_step(NewStep {
                run_id,
                trace_id: step_trace_id(run_id, name, position),
                name: name.to_string(),
                kind: StepKind::Shell,
                position,
                input: None,
                is_error_handler: false,
            })
            .await
            .unwrap();

        match status {
            StepStatus::Pending => {}
            StepStatus::Running => {
                store
                    .update_step(
                        step.id,
                        StepUpdate {
                            status: Some(StepStatus::Running),
                            ..StepUpdate::default()
                        },
                    )
                    .await
                    .unwrap();
            }
            StepStatus::Completed => {
                store
                    .update_step(
                        step.id,
                        StepUpdate {
                            status: Some(StepStatus::Running),
                            ..StepUpdate::default()
                        },
                    )
                    .await
                    .unwrap();
                store
                    .update_step(
                        step.id,
                        StepUpdate {
                            status: Some(StepStatus::Completed),
                            ..StepUpdate::default()
                        },
                    )
                    .await
                    .unwrap();
            }
            StepStatus::AwaitingApproval => {
                store
                    .update_step(
                        step.id,
                        StepUpdate {
                            status: Some(StepStatus::Running),
                            ..StepUpdate::default()
                        },
                    )
                    .await
                    .unwrap();
                store
                    .update_step(
                        step.id,
                        StepUpdate {
                            status: Some(StepStatus::AwaitingApproval),
                            ..StepUpdate::default()
                        },
                    )
                    .await
                    .unwrap();
            }
            _ => panic!("unsupported status for test helper: {status}"),
        }

        store.get_step(step.id).await.unwrap().unwrap()
    }

    #[tokio::test]
    async fn fail_orphaned_steps_marks_running_as_failed() {
        let engine = create_test_engine();
        let run = engine
            .store()
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: 0,
                concurrency_limits: Vec::new(),
                max_cost_usd: None,
                worker_tags: Vec::new(),
            })
            .await
            .unwrap()
            .into_run();

        let step = create_step_with_status(
            engine.store(),
            run.id,
            "running-step",
            0,
            StepStatus::Running,
        )
        .await;

        engine
            .fail_orphaned_steps(run.id, "parent run timed out")
            .await
            .unwrap();

        let updated = engine.store().get_step(step.id).await.unwrap().unwrap();
        assert_eq!(updated.status.state, StepStatus::Failed);
        assert_eq!(updated.error.as_deref(), Some("parent run timed out"));
        assert!(updated.completed_at.is_some());
    }

    #[tokio::test]
    async fn fail_orphaned_steps_marks_pending_as_skipped() {
        let engine = create_test_engine();
        let run = engine
            .store()
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: 0,
                concurrency_limits: Vec::new(),
                max_cost_usd: None,
                worker_tags: Vec::new(),
            })
            .await
            .unwrap()
            .into_run();

        let step = create_step_with_status(
            engine.store(),
            run.id,
            "pending-step",
            0,
            StepStatus::Pending,
        )
        .await;

        engine
            .fail_orphaned_steps(run.id, "parent run timed out")
            .await
            .unwrap();

        let updated = engine.store().get_step(step.id).await.unwrap().unwrap();
        assert_eq!(updated.status.state, StepStatus::Skipped);
        assert!(updated.error.is_none());
        assert!(updated.completed_at.is_some());
    }

    #[tokio::test]
    async fn fail_orphaned_steps_marks_awaiting_approval_as_failed() {
        let engine = create_test_engine();
        let run = engine
            .store()
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: 0,
                concurrency_limits: Vec::new(),
                max_cost_usd: None,
                worker_tags: Vec::new(),
            })
            .await
            .unwrap()
            .into_run();

        let step = create_step_with_status(
            engine.store(),
            run.id,
            "approval-step",
            0,
            StepStatus::AwaitingApproval,
        )
        .await;

        engine
            .fail_orphaned_steps(run.id, "parent run timed out")
            .await
            .unwrap();

        let updated = engine.store().get_step(step.id).await.unwrap().unwrap();
        assert_eq!(updated.status.state, StepStatus::Failed);
        assert_eq!(updated.error.as_deref(), Some("parent run timed out"));
        assert!(updated.completed_at.is_some());
    }

    #[tokio::test]
    async fn fail_orphaned_steps_skips_terminal_steps() {
        let engine = create_test_engine();
        let run = engine
            .store()
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: 0,
                concurrency_limits: Vec::new(),
                max_cost_usd: None,
                worker_tags: Vec::new(),
            })
            .await
            .unwrap()
            .into_run();

        let completed_step =
            create_step_with_status(engine.store(), run.id, "done", 0, StepStatus::Completed).await;
        let running_step =
            create_step_with_status(engine.store(), run.id, "in-flight", 1, StepStatus::Running)
                .await;

        engine
            .fail_orphaned_steps(run.id, "parent run timed out")
            .await
            .unwrap();

        let completed = engine
            .store()
            .get_step(completed_step.id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(completed.status.state, StepStatus::Completed);

        let failed = engine
            .store()
            .get_step(running_step.id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(failed.status.state, StepStatus::Failed);
    }

    #[tokio::test]
    async fn fail_orphaned_steps_mixed_states() {
        let engine = create_test_engine();
        let run = engine
            .store()
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: 0,
                concurrency_limits: Vec::new(),
                max_cost_usd: None,
                worker_tags: Vec::new(),
            })
            .await
            .unwrap()
            .into_run();

        let s_completed =
            create_step_with_status(engine.store(), run.id, "step-1", 0, StepStatus::Completed)
                .await;
        let s_running =
            create_step_with_status(engine.store(), run.id, "step-2", 1, StepStatus::Running).await;
        let s_pending =
            create_step_with_status(engine.store(), run.id, "step-3", 2, StepStatus::Pending).await;

        engine.fail_orphaned_steps(run.id, "timeout").await.unwrap();

        let r_completed = engine
            .store()
            .get_step(s_completed.id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(r_completed.status.state, StepStatus::Completed);

        let r_running = engine
            .store()
            .get_step(s_running.id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(r_running.status.state, StepStatus::Failed);
        assert_eq!(r_running.error.as_deref(), Some("timeout"));

        let r_pending = engine
            .store()
            .get_step(s_pending.id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(r_pending.status.state, StepStatus::Skipped);
        assert!(r_pending.error.is_none());
    }

    #[tokio::test]
    async fn fail_orphaned_steps_no_steps_is_noop() {
        let engine = create_test_engine();
        let run = engine
            .store()
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: 0,
                concurrency_limits: Vec::new(),
                max_cost_usd: None,
                worker_tags: Vec::new(),
            })
            .await
            .unwrap()
            .into_run();

        let result = engine.fail_orphaned_steps(run.id, "timeout").await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn fail_orphaned_steps_preserves_existing_error() {
        let engine = create_test_engine();
        let run = engine
            .store()
            .create_run(NewRun {
                created_by: None,
                workflow_name: "test".to_string(),
                trigger: TriggerKind::Manual,
                payload: json!({}),
                max_retries: 0,
                handler_version: None,
                labels: HashMap::new(),
                scheduled_at: None,
                idempotency_key: None,
                concurrency_key: None,
                priority: 0,
                concurrency_limits: Vec::new(),
                max_cost_usd: None,
                worker_tags: Vec::new(),
            })
            .await
            .unwrap()
            .into_run();

        let step_with_error = create_step_with_status(
            engine.store(),
            run.id,
            "already-errored",
            0,
            StepStatus::Running,
        )
        .await;

        engine
            .store()
            .update_step(
                step_with_error.id,
                StepUpdate {
                    error: Some("real error from provider".to_string()),
                    ..StepUpdate::default()
                },
            )
            .await
            .unwrap();

        let step_no_error = create_step_with_status(
            engine.store(),
            run.id,
            "no-error-yet",
            1,
            StepStatus::Running,
        )
        .await;

        engine
            .fail_orphaned_steps(run.id, "parent run failed")
            .await
            .unwrap();

        let updated_with = engine
            .store()
            .get_step(step_with_error.id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(updated_with.status.state, StepStatus::Failed);
        assert_eq!(
            updated_with.error.as_deref(),
            Some("real error from provider"),
        );

        let updated_without = engine
            .store()
            .get_step(step_no_error.id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(updated_without.status.state, StepStatus::Failed);
        assert_eq!(updated_without.error.as_deref(), Some("parent run failed"),);
    }

    // -----------------------------------------------------------------------
    // spawn_local_resume tests
    // -----------------------------------------------------------------------

    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::time::Duration;

    use ironflow_store::models::LeaseRequest;
    use tokio::time::sleep;

    /// Counts its executions, then completes.
    struct CountingWorkflow {
        executions: Arc<AtomicUsize>,
    }

    impl WorkflowHandler for CountingWorkflow {
        fn name(&self) -> &str {
            "local-resume-counter"
        }

        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
            Box::pin(async move {
                self.executions.fetch_add(1, Ordering::SeqCst);
                Ok(())
            })
        }
    }

    /// An engine running [`CountingWorkflow`], with one `Pending` run of it.
    async fn engine_with_pending_counter() -> (Arc<Engine>, Arc<AtomicUsize>, Uuid) {
        let executions = Arc::new(AtomicUsize::new(0));
        let mut engine = create_test_engine();
        engine
            .register(CountingWorkflow {
                executions: executions.clone(),
            })
            .unwrap();
        let run = engine
            .enqueue_handler("local-resume-counter", TriggerKind::Manual, json!({}), 0)
            .await
            .unwrap();
        (Arc::new(engine), executions, run.id)
    }

    #[tokio::test]
    async fn local_resume_leaves_a_run_claimed_by_a_worker_alone() {
        let (engine, executions, run_id) = engine_with_pending_counter().await;
        let lease = LeaseRequest {
            worker_id: "worker-1".to_string(),
            ttl: Duration::from_secs(30),
        };
        let picked = engine
            .store()
            .pick_next_pending_for(Some(lease), None)
            .await
            .unwrap()
            .expect("the run is due");
        assert_eq!(picked.id, run_id);

        engine.spawn_local_resume(run_id);
        sleep(Duration::from_millis(200)).await;

        assert_eq!(executions.load(Ordering::SeqCst), 0);
        let run = engine.store().get_run(run_id).await.unwrap().unwrap();
        assert_eq!(run.status.state, RunStatus::Running);
        assert_eq!(run.worker_id.as_deref(), Some("worker-1"));
    }

    #[tokio::test]
    async fn local_resume_claims_a_pending_run_once() {
        let (engine, executions, run_id) = engine_with_pending_counter().await;

        engine.spawn_local_resume(run_id);
        engine.spawn_local_resume(run_id);
        sleep(Duration::from_millis(200)).await;

        assert_eq!(executions.load(Ordering::SeqCst), 1);
        let run = engine.store().get_run(run_id).await.unwrap().unwrap();
        assert_eq!(run.status.state, RunStatus::Completed);
    }

    #[tokio::test]
    async fn local_resume_of_a_running_run_executes_it_once() {
        let (engine, executions, run_id) = engine_with_pending_counter().await;
        engine
            .store()
            .update_run_status(run_id, RunStatus::Running)
            .await
            .unwrap();

        engine.spawn_local_resume(run_id);
        engine.spawn_local_resume(run_id);
        sleep(Duration::from_millis(200)).await;

        assert_eq!(executions.load(Ordering::SeqCst), 1);
        let run = engine.store().get_run(run_id).await.unwrap().unwrap();
        assert_eq!(run.status.state, RunStatus::Completed);
    }
}