agentq 0.2.1

use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use crate::{
    Accepted, ClaimResult, DurableStore, Event, Job, JobHandle, JobLost, JobResult, Outcome,
    Priority, PushError, Queue, RetryPolicy, StepDef, StepFunc, StepStatus, StoreError, Workflow,
};

type SharedStepFunc =
    Arc<dyn Fn() -> Pin<Box<dyn Future<Output = JobResult> + Send>> + Send + Sync>;

/// Marker error returned from a step body to enter [`StepStatus::Waiting`].
///
/// The worker turns this into a failed [`Outcome`]; the engine recognizes the
/// encoding and records [`Event::StepWaiting`] instead of treating it as a
/// hard failure.
#[derive(Debug)]
pub struct WaitForInput(pub String);

impl std::fmt::Display for WaitForInput {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{WAIT_PREFIX}{}", self.0)
    }
}

impl std::error::Error for WaitForInput {}

const WAIT_PREFIX: &str = "agentq:wait:";

fn parse_waiting_reason(reason: &str) -> Option<&str> {
    reason.strip_prefix(WAIT_PREFIX)
}

/// Marker error returned from a step body to fail the workflow without retrying.
///
/// The worker turns this into a failed [`Outcome`]; the engine recognizes the
/// encoding and records [`Event::WorkflowFailed`] immediately, skipping
/// [`Event::RetryScheduled`] even when [`RetryPolicy::max_attempts`] would
/// allow more tries.
#[derive(Debug)]
pub struct NonRetryable(pub String);

impl std::fmt::Display for NonRetryable {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{NON_RETRYABLE_PREFIX}{}", self.0)
    }
}

impl std::error::Error for NonRetryable {}

const NON_RETRYABLE_PREFIX: &str = "agentq:non-retryable:";

fn parse_non_retryable_reason(reason: &str) -> Option<&str> {
    reason.strip_prefix(NON_RETRYABLE_PREFIX)
}

/// Drives workflow steps through the V1 [`Queue`] and records V2 [`Event`]s.
pub struct WorkflowEngine<S: DurableStore> {
    queue: Queue,
    store: S,
    worker_id: String,
    lease_ttl: Duration,
    priority: Priority,
    in_flight: Mutex<HashMap<String, RunningWorkflow>>,
    terminal_write: Mutex<()>,
}

struct RunningWorkflow {
    steps: Vec<StepDef>,
    bodies: Vec<SharedStepFunc>,
}

/// Whether a step run finished the step or left it blocked on external input.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StepOutcome {
    Completed,
    Waiting,
}

#[derive(Debug)]
pub enum EngineError {
    Store(StoreError),
    Push(PushError),
    JobLost(JobLost),
    WorkflowFailed { reason: String },
}

impl std::fmt::Display for EngineError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            EngineError::Store(e) => write!(f, "{e}"),
            EngineError::Push(e) => write!(f, "{e}"),
            EngineError::JobLost(e) => write!(f, "{e}"),
            EngineError::WorkflowFailed { reason } => write!(f, "workflow failed: {reason}"),
        }
    }
}

impl std::error::Error for EngineError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            EngineError::Store(e) => Some(e),
            EngineError::Push(e) => Some(e),
            EngineError::JobLost(e) => Some(e),
            EngineError::WorkflowFailed { .. } => None,
        }
    }
}

impl From<StoreError> for EngineError {
    fn from(err: StoreError) -> EngineError {
        EngineError::Store(err)
    }
}

impl<S: DurableStore> WorkflowEngine<S> {
    pub fn new(
        queue: Queue,
        store: S,
        worker_id: String,
        lease_ttl: Duration,
        priority: Priority,
    ) -> Self {
        WorkflowEngine {
            queue,
            store,
            worker_id,
            lease_ttl,
            priority,
            in_flight: Mutex::new(HashMap::new()),
            terminal_write: Mutex::new(()),
        }
    }

    /// Registers a workflow's step bodies without running anything, so a
    /// later [`recover`](crate::recover) call (after a process restart) has
    /// something to re-admit an interrupted step into.
    pub fn register_workflow(
        &self,
        workflow: Workflow,
        bodies: Vec<StepFunc>,
    ) -> Result<(), EngineError> {
        if bodies.len() != workflow.steps.len() {
            return Err(EngineError::Store(StoreError::Backend(
                "step body count does not match workflow definition".into(),
            )));
        }

        let shared_bodies: Vec<SharedStepFunc> = bodies
            .into_iter()
            .map(|body| Arc::from(body) as SharedStepFunc)
            .collect();

        self.in_flight
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .insert(
                workflow.id.clone(),
                RunningWorkflow {
                    steps: workflow.steps,
                    bodies: shared_bodies,
                },
            );
        Ok(())
    }

    /// Runs all steps in order, recording events and honoring per-step retry policy.
    ///
    /// Stops (without error) at the first step that enters
    /// [`StepStatus::Waiting`] — the workflow stays registered so a later
    /// [`Self::resume`] can continue it from that point.
    pub async fn run(&self, workflow: Workflow, bodies: Vec<StepFunc>) -> Result<(), EngineError> {
        let workflow_id = workflow.id.clone();
        self.register_workflow(workflow, bodies)?;

        self.store.append_event(&Event::WorkflowStarted {
            workflow_id: workflow_id.clone(),
        })?;

        let (steps, shared_bodies) = self.mounted_steps_and_bodies(&workflow_id)?;
        self.drive_from(&workflow_id, &steps, &shared_bodies, 0)
            .await
    }

    /// Unblocks a step in [`StepStatus::Waiting`], supplying opaque `input`
    /// for its next run, then continues driving the remaining steps in
    /// order — stopping again if a later step also enters `Waiting`.
    /// Records [`Event::WorkflowCancelled`] and un-registers the workflow so the
    /// engine will not claim or schedule further steps.
    ///
    /// Idempotent: if the workflow already has a terminal event
    /// ([`Event::WorkflowCompleted`], [`Event::WorkflowFailed`], or
    /// [`Event::WorkflowCancelled`]), returns `Ok(())` without appending
    /// anything.
    ///
    /// Cancellation does not abort a step attempt already pushed to the V1
    /// [`Queue`] and executing in a worker — V1 has no in-flight abort. Side
    /// effects from that run may still occur. While driving steps, cancellation
    /// is observed at the next step boundary (and between retry attempts for
    /// the current step); a body already running on the queue is not stopped.
    pub async fn cancel(&self, workflow_id: &str, reason: String) -> Result<(), EngineError> {
        let _terminal_guard = self
            .terminal_write
            .lock()
            .unwrap_or_else(|e| e.into_inner());

        let events = self.store.load_events(workflow_id)?;
        if workflow_has_terminal_event(&events) {
            return Ok(());
        }

        self.store.append_event(&Event::WorkflowCancelled {
            workflow_id: workflow_id.to_string(),
            reason,
        })?;

        self.in_flight
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .remove(workflow_id);
        Ok(())
    }

    pub async fn resume(
        &self,
        workflow_id: &str,
        step_index: usize,
        input: String,
    ) -> Result<(), EngineError> {
        let events = self.store.load_events(workflow_id)?;
        if workflow_is_cancelled(&events) {
            return Err(EngineError::Store(StoreError::Backend(
                "workflow is cancelled".into(),
            )));
        }
        let status = step_status_from_events(&events, step_index);
        match status {
            StepStatus::Waiting { .. } => {}
            _ => {
                return Err(EngineError::Store(StoreError::Backend(
                    "step is not waiting".into(),
                )));
            }
        }

        let (step, body) = {
            let guard = self.in_flight.lock().unwrap_or_else(|e| e.into_inner());
            let running = guard.get(workflow_id).ok_or(StoreError::NotFound)?;
            if step_index >= running.steps.len() {
                return Err(EngineError::Store(StoreError::NotFound));
            }
            (
                running.steps[step_index].clone(),
                running.bodies[step_index].clone(),
            )
        };

        self.store.append_event(&Event::StepResumed {
            workflow_id: workflow_id.to_string(),
            step_index,
            input,
        })?;
        let attempt = current_attempt_from_events(&events, step_index);

        match self
            .run_step(workflow_id, step_index, &step, body, attempt)
            .await
        {
            Ok(StepOutcome::Waiting) => Ok(()),
            Ok(StepOutcome::Completed) => {
                let (steps, bodies) = self.mounted_steps_and_bodies(workflow_id)?;
                self.drive_from(workflow_id, &steps, &bodies, step_index + 1)
                    .await
            }
            Err(err) => {
                self.drop_from_in_flight(workflow_id);
                Err(err)
            }
        }
    }

    fn mounted_steps_and_bodies(
        &self,
        workflow_id: &str,
    ) -> Result<(Vec<StepDef>, Vec<SharedStepFunc>), EngineError> {
        let guard = self.in_flight.lock().unwrap_or_else(|e| e.into_inner());
        let running = guard
            .get(workflow_id)
            .ok_or(EngineError::Store(StoreError::NotFound))?;
        Ok((running.steps.clone(), running.bodies.clone()))
    }

    fn drop_from_in_flight(&self, workflow_id: &str) {
        self.in_flight
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .remove(workflow_id);
    }

    /// Runs steps `start_index..` in order. Returns `Ok(())` whether the
    /// workflow ran to completion or stopped early at a `Waiting` step —
    /// callers distinguish the two by checking the store, not this return
    /// value, since both are "no error" outcomes for the caller of `run`/
    /// `resume`. On full completion, records `WorkflowCompleted` and
    /// un-registers the workflow; on `Waiting`, leaves it registered.
    async fn drive_from(
        &self,
        workflow_id: &str,
        steps: &[StepDef],
        bodies: &[SharedStepFunc],
        start_index: usize,
    ) -> Result<(), EngineError> {
        for step_index in start_index..steps.len() {
            let events = self.store.load_events(workflow_id)?;
            if workflow_is_cancelled(&events) {
                return Ok(());
            }

            match self
                .run_step(
                    workflow_id,
                    step_index,
                    &steps[step_index],
                    bodies[step_index].clone(),
                    0,
                )
                .await
            {
                Ok(StepOutcome::Completed) => continue,
                Ok(StepOutcome::Waiting) => return Ok(()),
                Err(err) => {
                    self.drop_from_in_flight(workflow_id);
                    return Err(err);
                }
            }
        }

        {
            let _terminal_guard = self
                .terminal_write
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let events = self.store.load_events(workflow_id)?;
            if !workflow_has_terminal_event(&events) {
                self.store.append_event(&Event::WorkflowCompleted {
                    workflow_id: workflow_id.to_string(),
                })?;
            }
        }
        self.in_flight
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .remove(workflow_id);
        Ok(())
    }

    /// Opaque input from the most recent [`Event::StepResumed`] for this step in
    /// durable storage, if any.
    pub fn resume_input(&self, workflow_id: &str, step_index: usize) -> Option<String> {
        let events = match self.store.load_events(workflow_id) {
            Ok(events) => events,
            Err(_) => return None,
        };
        let mut last = None;
        for event in &events {
            if let Event::StepResumed {
                step_index: si,
                input,
                ..
            } = event
            {
                if *si == step_index {
                    last = Some(input.clone());
                }
            }
        }
        last
    }

    pub(crate) fn store(&self) -> &S {
        &self.store
    }

    /// Clears an expired lease in the store and re-runs the step when bodies
    /// are mounted (via [`Self::register_workflow`] or a prior [`Self::run`]
    /// in this same process) — continuing through the remaining steps if
    /// that step completes. If the workflow isn't mounted (nothing has
    /// re-registered it since a process restart), this only records
    /// [`Event::WorkerRecovered`] and does nothing further; the caller is
    /// responsible for calling [`Self::register_workflow`] first.
    pub(crate) async fn readmit_after_worker_recovery(
        &self,
        workflow_id: &str,
        step_index: usize,
        attempt: u32,
    ) -> Result<(), EngineError> {
        self.store.append_event(&Event::WorkerRecovered {
            workflow_id: workflow_id.to_string(),
            step_index,
        })?;

        let (step, body) = {
            let guard = self.in_flight.lock().unwrap_or_else(|e| e.into_inner());
            let Some(running) = guard.get(workflow_id) else {
                return Ok(());
            };
            if step_index >= running.steps.len() {
                return Ok(());
            }
            (
                running.steps[step_index].clone(),
                running.bodies[step_index].clone(),
            )
        };

        match self
            .run_step(workflow_id, step_index, &step, body, attempt)
            .await
        {
            Ok(StepOutcome::Waiting) => Ok(()),
            Ok(StepOutcome::Completed) => {
                let (steps, bodies) = self.mounted_steps_and_bodies(workflow_id)?;
                self.drive_from(workflow_id, &steps, &bodies, step_index + 1)
                    .await
            }
            Err(err) => {
                self.drop_from_in_flight(workflow_id);
                Err(err)
            }
        }
    }

    /// Runs one step attempt loop: claim, push to the queue, await the job, record
    /// outcome, and retry per [`StepDef::retry_policy`] until success or exhaustion.
    ///
    /// When [`StepDef::timeout`] is set, the wait budget starts after the job is
    /// pushed and its handle is obtained (not while claiming). The engine races that
    /// handle against the timeout; if the timeout wins, it treats the attempt as
    /// [`Outcome::Failed`] like any other failure.
    ///
    /// Step bodies run on a separate `tokio` task in the worker. Dropping the
    /// [`JobHandle`] after a timeout does not cancel that task — the original body
    /// keeps running to completion in the background. Its eventual outcome is written
    /// into V1 state under the timed-out attempt's job key, but nothing durable
    /// reads it once the engine retries under a new key. A timeout can therefore
    /// leave an orphaned execution whose side effects still occur with no event
    /// recorded for that run.
    async fn await_job_renewing_lease(
        &self,
        workflow_id: &str,
        step_index: usize,
        handle: JobHandle,
        step_timeout: Option<Duration>,
    ) -> Result<Outcome, EngineError> {
        let renewal_interval = {
            let every = self.lease_ttl / 3;
            if every.is_zero() {
                Duration::from_millis(1)
            } else {
                every
            }
        };
        let (renew_tx, mut renew_rx) = tokio::sync::mpsc::unbounded_channel();
        let renewal_handle = tokio::spawn(async move {
            let mut interval = tokio::time::interval(renewal_interval);
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
            loop {
                interval.tick().await;
                if renew_tx.send(()).is_err() {
                    break;
                }
            }
        });
        let renewal_abort = renewal_handle.abort_handle();

        let outcome = if let Some(step_timeout) = step_timeout {
            tokio::pin!(handle);
            let deadline = tokio::time::sleep(step_timeout);
            tokio::pin!(deadline);
            loop {
                tokio::select! {
                    biased;
                    res = handle.as_mut() => {
                        renewal_abort.abort();
                        break match res {
                            Ok(outcome) => outcome,
                            Err(err) => return Err(EngineError::JobLost(err)),
                        };
                    }
                    () = &mut deadline => {
                        renewal_abort.abort();
                        break Outcome::Failed {
                            reason: format!("step timed out after {:?}", step_timeout),
                        };
                    }
                    Some(()) = renew_rx.recv() => {
                        if let Err(err) = self.store.renew_lease(
                            workflow_id,
                            step_index,
                            &self.worker_id,
                            self.lease_ttl,
                        ) {
                            tracing::warn!(
                                workflow_id,
                                step_index,
                                worker_id = %self.worker_id,
                                error = %err,
                                "step lease renewal failed"
                            );
                        }
                    }
                }
            }
        } else {
            tokio::pin!(handle);
            loop {
                tokio::select! {
                    biased;
                    res = handle.as_mut() => {
                        renewal_abort.abort();
                        break match res {
                            Ok(outcome) => outcome,
                            Err(err) => return Err(EngineError::JobLost(err)),
                        };
                    }
                    Some(()) = renew_rx.recv() => {
                        if let Err(err) = self.store.renew_lease(
                            workflow_id,
                            step_index,
                            &self.worker_id,
                            self.lease_ttl,
                        ) {
                            tracing::warn!(
                                workflow_id,
                                step_index,
                                worker_id = %self.worker_id,
                                error = %err,
                                "step lease renewal failed"
                            );
                        }
                    }
                }
            }
        };

        Ok(outcome)
    }

    async fn run_step(
        &self,
        workflow_id: &str,
        step_index: usize,
        step: &StepDef,
        body: SharedStepFunc,
        start_attempt: u32,
    ) -> Result<StepOutcome, EngineError> {
        let mut attempt = start_attempt;

        loop {
            match self
                .store
                .claim_step(workflow_id, step_index, &self.worker_id, self.lease_ttl)?
            {
                ClaimResult::AlreadyCompleted { .. } => return Ok(StepOutcome::Completed),
                ClaimResult::HeldByOther => {
                    return Err(EngineError::Store(StoreError::Backend(
                        "step is held by another worker".into(),
                    )));
                }
                ClaimResult::AlreadyWaiting => return Ok(StepOutcome::Waiting),
                ClaimResult::Claimed => {}
            }

            self.store.append_event(&Event::StepStarted {
                workflow_id: workflow_id.to_string(),
                step_index,
                attempt,
            })?;

            let key = format!("{workflow_id}:{step_index}:{attempt}");
            let func = body.clone();
            let job = Job::new(key, self.priority, Box::new(move || func()));

            let handle = match self.queue.push(job).await {
                Ok(accepted) => match accepted {
                    Accepted::Queued(handle) | Accepted::InFlight(handle) => handle,
                    Accepted::Cached { output } => {
                        self.store.append_event(&Event::StepCompleted {
                            workflow_id: workflow_id.to_string(),
                            step_index,
                            output: output.clone(),
                        })?;
                        return Ok(StepOutcome::Completed);
                    }
                },
                Err(err) => return Err(EngineError::Push(err)),
            };

            let outcome = self
                .await_job_renewing_lease(workflow_id, step_index, handle, step.timeout)
                .await?;

            match outcome {
                Outcome::Completed { output } => {
                    self.store.append_event(&Event::StepCompleted {
                        workflow_id: workflow_id.to_string(),
                        step_index,
                        output,
                    })?;
                    return Ok(StepOutcome::Completed);
                }
                Outcome::Failed { reason } => {
                    if let Some(wait_reason) = parse_waiting_reason(&reason) {
                        self.store.append_event(&Event::StepWaiting {
                            workflow_id: workflow_id.to_string(),
                            step_index,
                            reason: wait_reason.to_string(),
                        })?;
                        return Ok(StepOutcome::Waiting);
                    }

                    if let Some(non_retryable_reason) = parse_non_retryable_reason(&reason) {
                        let stripped_reason = non_retryable_reason.to_string();
                        self.store.append_event(&Event::StepFailed {
                            workflow_id: workflow_id.to_string(),
                            step_index,
                            reason: stripped_reason.clone(),
                        })?;
                        self.store.append_event(&Event::WorkflowFailed {
                            workflow_id: workflow_id.to_string(),
                            reason: stripped_reason.clone(),
                        })?;
                        return Err(EngineError::WorkflowFailed {
                            reason: stripped_reason,
                        });
                    }

                    self.store.append_event(&Event::StepFailed {
                        workflow_id: workflow_id.to_string(),
                        step_index,
                        reason: reason.clone(),
                    })?;

                    let next_attempt = attempt + 1;
                    if next_attempt < step.retry_policy.max_attempts {
                        self.store.append_event(&Event::RetryScheduled {
                            workflow_id: workflow_id.to_string(),
                            step_index,
                            attempt: next_attempt,
                        })?;
                        let delay = backoff_duration(&step.retry_policy, attempt);
                        tokio::time::sleep(delay).await;
                        if workflow_is_cancelled(&self.store.load_events(workflow_id)?) {
                            return Ok(StepOutcome::Waiting);
                        }
                        attempt = next_attempt;
                        continue;
                    }

                    self.store.append_event(&Event::WorkflowFailed {
                        workflow_id: workflow_id.to_string(),
                        reason: reason.clone(),
                    })?;
                    return Err(EngineError::WorkflowFailed { reason });
                }
            }
        }
    }
}

fn workflow_has_terminal_event(events: &[Event]) -> bool {
    events.iter().any(|e| {
        matches!(
            e,
            Event::WorkflowCompleted { .. }
                | Event::WorkflowFailed { .. }
                | Event::WorkflowCancelled { .. }
        )
    })
}

fn workflow_is_cancelled(events: &[Event]) -> bool {
    events
        .iter()
        .any(|e| matches!(e, Event::WorkflowCancelled { .. }))
}

fn backoff_duration(policy: &RetryPolicy, failed_attempt: u32) -> Duration {
    match policy.backoff {
        crate::Backoff::Fixed(d) => d,
        crate::Backoff::Exponential { base, max } => {
            let mut duration = base;
            for _ in 0..failed_attempt {
                duration = duration.saturating_mul(2);
                if duration >= max {
                    return max;
                }
            }
            duration.min(max)
        }
    }
}

fn current_attempt_from_events(events: &[Event], step_index: usize) -> u32 {
    let mut attempt = 0u32;
    for event in events {
        match event {
            Event::StepStarted {
                step_index: si,
                attempt: a,
                ..
            } if *si == step_index => attempt = *a,
            Event::RetryScheduled {
                step_index: si,
                attempt: a,
                ..
            } if *si == step_index => attempt = *a,
            _ => {}
        }
    }
    attempt
}

fn step_status_from_events(events: &[Event], step_index: usize) -> StepStatus {
    let mut status = StepStatus::Pending;
    let mut attempt = 0u32;

    for event in events {
        match event {
            Event::StepStarted {
                step_index: si,
                attempt: a,
                ..
            } if *si == step_index => {
                attempt = *a;
                status = StepStatus::Pending;
            }
            Event::StepCompleted {
                step_index: si,
                output,
                ..
            } if *si == step_index => {
                status = StepStatus::Completed {
                    output: output.clone(),
                };
            }
            Event::StepFailed {
                step_index: si,
                reason,
                ..
            } if *si == step_index => {
                status = StepStatus::Failed {
                    reason: reason.clone(),
                    attempt,
                };
            }
            Event::StepWaiting {
                step_index: si,
                reason,
                ..
            } if *si == step_index => {
                status = StepStatus::Waiting {
                    reason: reason.clone(),
                };
            }
            Event::RetryScheduled {
                step_index: si,
                attempt: a,
                ..
            } if *si == step_index => {
                attempt = *a;
                status = StepStatus::Pending;
            }
            Event::StepResumed { step_index: si, .. } if *si == step_index => {
                status = StepStatus::Pending;
            }
            Event::WorkerRecovered { step_index: si, .. } if *si == step_index => {
                status = StepStatus::Pending;
            }
            _ => {}
        }
    }

    status
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{Backoff, Execution};
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[derive(Clone)]
    struct MemStore {
        events: Arc<Mutex<Vec<Event>>>,
    }

    impl MemStore {
        fn new() -> Self {
            MemStore {
                events: Arc::new(Mutex::new(Vec::new())),
            }
        }
    }

    impl DurableStore for MemStore {
        fn append_event(&self, event: &Event) -> Result<(), StoreError> {
            self.events
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .push(event.clone());
            Ok(())
        }

        fn load_events(&self, workflow_id: &str) -> Result<Vec<Event>, StoreError> {
            let events = self.events.lock().unwrap_or_else(|e| e.into_inner());
            Ok(events
                .iter()
                .filter(|e| event_workflow_id(e) == workflow_id)
                .cloned()
                .collect())
        }

        fn claim_step(
            &self,
            _workflow_id: &str,
            _step_index: usize,
            _worker_id: &str,
            _lease_ttl: Duration,
        ) -> Result<ClaimResult, StoreError> {
            Ok(ClaimResult::Claimed)
        }

        fn renew_lease(
            &self,
            _workflow_id: &str,
            _step_index: usize,
            _worker_id: &str,
            _lease_ttl: Duration,
        ) -> Result<(), StoreError> {
            Ok(())
        }

        fn expired_leases(&self) -> Result<Vec<Execution>, StoreError> {
            Ok(Vec::new())
        }
    }

    fn event_workflow_id(event: &Event) -> &str {
        match event {
            Event::WorkflowStarted { workflow_id } => workflow_id,
            Event::StepStarted { workflow_id, .. } => workflow_id,
            Event::StepCompleted { workflow_id, .. } => workflow_id,
            Event::StepFailed { workflow_id, .. } => workflow_id,
            Event::RetryScheduled { workflow_id, .. } => workflow_id,
            Event::StepWaiting { workflow_id, .. } => workflow_id,
            Event::StepResumed { workflow_id, .. } => workflow_id,
            Event::WorkflowCompleted { workflow_id } => workflow_id,
            Event::WorkflowFailed { workflow_id, .. } => workflow_id,
            Event::WorkflowCancelled { workflow_id, .. } => workflow_id,
            Event::WorkerRecovered { workflow_id, .. } => workflow_id,
        }
    }

    fn workflow_completed(events: &[Event]) -> bool {
        events
            .iter()
            .any(|e| matches!(e, Event::WorkflowCompleted { .. }))
    }

    fn filter_step_lifecycle(events: &[Event]) -> Vec<Event> {
        events
            .iter()
            .filter(|e| {
                matches!(
                    e,
                    Event::StepFailed { .. }
                        | Event::RetryScheduled { .. }
                        | Event::StepCompleted { .. }
                )
            })
            .cloned()
            .collect()
    }

    #[tokio::test]
    async fn retry_after_failure_then_completes() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let attempts = AtomicUsize::new(0);
        let body: StepFunc = Box::new(move || {
            let n = attempts.fetch_add(1, Ordering::SeqCst);
            Box::pin(async move {
                if n == 0 {
                    Err("first try failed".into())
                } else {
                    Ok("ok".to_string())
                }
            })
        });

        let workflow = Workflow {
            id: "wf-retry".to_string(),
            steps: vec![StepDef {
                name: "only".to_string(),
                retry_policy: RetryPolicy {
                    max_attempts: 2,
                    backoff: Backoff::Fixed(Duration::from_millis(1)),
                },
                timeout: None,
            }],
        };

        engine
            .run(workflow, vec![body])
            .await
            .expect("workflow should complete");

        let events = store.load_events("wf-retry").expect("load events");
        assert!(workflow_completed(&events));

        let lifecycle = filter_step_lifecycle(&events);
        assert_eq!(lifecycle.len(), 3);
        assert!(matches!(lifecycle[0], Event::StepFailed { .. }));
        assert!(matches!(lifecycle[1], Event::RetryScheduled { .. }));
        assert!(matches!(lifecycle[2], Event::StepCompleted { .. }));
    }

    #[tokio::test]
    async fn non_retryable_failure_fails_workflow_without_scheduling_retry() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let body: StepFunc = Box::new(|| {
            Box::pin(async {
                Err(Box::new(NonRetryable("bad request".to_string()))
                    as Box<dyn std::error::Error + Send + Sync>)
            })
        });

        let workflow = Workflow {
            id: "wf-non-retryable".to_string(),
            steps: vec![StepDef {
                name: "only".to_string(),
                retry_policy: RetryPolicy {
                    max_attempts: 5,
                    backoff: Backoff::Fixed(Duration::from_millis(1)),
                },
                timeout: None,
            }],
        };

        let result = engine.run(workflow, vec![body]).await;
        assert!(
            matches!(result, Err(EngineError::WorkflowFailed { .. })),
            "non-retryable step failure should fail the workflow"
        );

        let events = store.load_events("wf-non-retryable").expect("load events");
        assert!(
            events
                .iter()
                .any(|e| matches!(e, Event::WorkflowFailed { .. }))
        );
        assert!(
            !events
                .iter()
                .any(|e| matches!(e, Event::RetryScheduled { .. })),
            "non-retryable failure must not schedule a retry"
        );
    }

    #[tokio::test]
    async fn step_timeout_retries_then_workflow_fails_when_attempts_exhausted() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let body: StepFunc = Box::new(|| Box::pin(std::future::pending()));

        let workflow = Workflow {
            id: "wf-timeout-fail".to_string(),
            steps: vec![StepDef {
                name: "hang".to_string(),
                retry_policy: RetryPolicy {
                    max_attempts: 2,
                    backoff: Backoff::Fixed(Duration::from_millis(1)),
                },
                timeout: Some(Duration::from_millis(20)),
            }],
        };

        let result = engine.run(workflow, vec![body]).await;
        assert!(
            matches!(result, Err(EngineError::WorkflowFailed { .. })),
            "expected workflow failure after timeouts exhaust retries"
        );

        let events = store.load_events("wf-timeout-fail").expect("load events");
        assert!(!workflow_completed(&events));
        assert!(
            events
                .iter()
                .any(|e| matches!(e, Event::WorkflowFailed { .. }))
        );

        let retry_count = events
            .iter()
            .filter(|e| matches!(e, Event::RetryScheduled { .. }))
            .count();
        assert_eq!(retry_count, 1, "one retry between two timed-out attempts");
    }

    #[tokio::test]
    async fn step_timeout_on_first_attempt_then_succeeds_on_retry() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let attempts = AtomicUsize::new(0);
        let body: StepFunc = Box::new(move || {
            let n = attempts.fetch_add(1, Ordering::SeqCst);
            Box::pin(async move {
                if n == 0 {
                    std::future::pending::<JobResult>().await
                } else {
                    Ok("ok".to_string())
                }
            })
        });

        let workflow = Workflow {
            id: "wf-timeout-retry-ok".to_string(),
            steps: vec![StepDef {
                name: "recover".to_string(),
                retry_policy: RetryPolicy {
                    max_attempts: 2,
                    backoff: Backoff::Fixed(Duration::from_millis(1)),
                },
                timeout: Some(Duration::from_millis(20)),
            }],
        };

        engine
            .run(workflow, vec![body])
            .await
            .expect("workflow should complete after retry succeeds");

        let events = store
            .load_events("wf-timeout-retry-ok")
            .expect("load events");
        assert!(workflow_completed(&events));

        let lifecycle = filter_step_lifecycle(&events);
        assert_eq!(lifecycle.len(), 3);
        assert!(matches!(lifecycle[0], Event::StepFailed { .. }));
        assert!(matches!(lifecycle[1], Event::RetryScheduled { .. }));
        assert!(matches!(lifecycle[2], Event::StepCompleted { .. }));
    }

    #[tokio::test]
    async fn waiting_step_pauses_workflow_until_resumed_then_continues() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let step2_ran = Arc::new(AtomicUsize::new(0));
        let step2_ran_clone = step2_ran.clone();

        let step0: StepFunc = Box::new(|| Box::pin(async { Ok("first".to_string()) }));

        let step1_attempts = AtomicUsize::new(0);
        let step1: StepFunc = Box::new(move || {
            let n = step1_attempts.fetch_add(1, Ordering::SeqCst);
            Box::pin(async move {
                if n == 0 {
                    Err(Box::new(WaitForInput("need approval".to_string()))
                        as Box<dyn std::error::Error + Send + Sync>)
                } else {
                    Ok("approved".to_string())
                }
            })
        });

        let step2: StepFunc = Box::new(move || {
            step2_ran_clone.fetch_add(1, Ordering::SeqCst);
            Box::pin(async { Ok("done".to_string()) })
        });

        let no_retry = RetryPolicy {
            max_attempts: 1,
            backoff: Backoff::Fixed(Duration::from_millis(1)),
        };
        let workflow = Workflow {
            id: "wf-wait".to_string(),
            steps: vec![
                StepDef {
                    name: "first".into(),
                    retry_policy: no_retry.clone(),
                    timeout: None,
                },
                StepDef {
                    name: "approval".into(),
                    retry_policy: no_retry.clone(),
                    timeout: None,
                },
                StepDef {
                    name: "last".into(),
                    retry_policy: no_retry,
                    timeout: None,
                },
            ],
        };

        engine
            .run(workflow, vec![step0, step1, step2])
            .await
            .expect("run should not error while a step is waiting");

        assert_eq!(
            step2_ran.load(Ordering::SeqCst),
            0,
            "step after a Waiting step must not run before resume"
        );
        let events = store.load_events("wf-wait").expect("load events");
        assert!(
            events
                .iter()
                .any(|e| matches!(e, Event::StepWaiting { step_index: 1, .. }))
        );
        assert!(!workflow_completed(&events));

        engine
            .resume("wf-wait", 1, "approved by human".to_string())
            .await
            .expect("resume should drive the workflow to completion");

        assert_eq!(
            step2_ran.load(Ordering::SeqCst),
            1,
            "step after the resumed step should run exactly once after resume"
        );
        let events = store.load_events("wf-wait").expect("load events");
        assert!(workflow_completed(&events));
    }

    /// Two concurrent `run()` calls on a fresh workflow should execute the step
    /// body once, coalesced by V1 job-key dedup at attempt 0.
    ///
    /// This does not prove that concurrent calls remain safe once one call chain
    /// has retried a step and the other has not (diverging job keys).
    #[tokio::test]
    async fn concurrent_run_on_fresh_workflow_executes_step_once() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store,
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let counter = Arc::new(AtomicUsize::new(0));

        let no_retry = RetryPolicy {
            max_attempts: 1,
            backoff: Backoff::Fixed(Duration::from_millis(1)),
        };
        let workflow = Workflow {
            id: "wf-concurrent-fresh".to_string(),
            steps: vec![StepDef {
                name: "only".into(),
                retry_policy: no_retry,
                timeout: None,
            }],
        };

        let bodies_a: Vec<StepFunc> = {
            let counter = counter.clone();
            vec![Box::new(move || {
                counter.fetch_add(1, Ordering::SeqCst);
                Box::pin(async { Ok("done".to_string()) })
            })]
        };
        let bodies_b: Vec<StepFunc> = {
            let counter = counter.clone();
            vec![Box::new(move || {
                counter.fetch_add(1, Ordering::SeqCst);
                Box::pin(async { Ok("done".to_string()) })
            })]
        };

        let (result_a, result_b) = tokio::join!(
            engine.run(workflow.clone(), bodies_a),
            engine.run(workflow.clone(), bodies_b),
        );

        result_a.expect("first concurrent run should succeed");
        result_b.expect("second concurrent run should succeed");
        assert_eq!(
            counter.load(Ordering::SeqCst),
            1,
            "job-key dedup should admit only one concurrent step execution"
        );
    }

    fn count_workflow_completed(events: &[Event]) -> usize {
        events
            .iter()
            .filter(|e| matches!(e, Event::WorkflowCompleted { .. }))
            .count()
    }

    fn count_workflow_cancelled(events: &[Event]) -> usize {
        events
            .iter()
            .filter(|e| matches!(e, Event::WorkflowCancelled { .. }))
            .count()
    }

    #[cfg(feature = "sqlite")]
    fn sqlite_engine(store: crate::SqliteStore) -> WorkflowEngine<crate::SqliteStore> {
        WorkflowEngine::new(
            Queue::builder().start(),
            store,
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        )
    }

    #[cfg(feature = "sqlite")]
    fn sqlite_engine_with_lease(
        store: crate::SqliteStore,
        worker_id: &str,
        lease_ttl: Duration,
    ) -> WorkflowEngine<crate::SqliteStore> {
        WorkflowEngine::new(
            Queue::builder().start(),
            store,
            worker_id.to_string(),
            lease_ttl,
            Priority::High,
        )
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn live_worker_renews_lease_so_other_worker_cannot_claim() {
        use std::sync::Arc;

        let store = crate::SqliteStore::new(":memory:").expect("open db");
        let lease_ttl = Duration::from_millis(50);
        let engine = Arc::new(sqlite_engine_with_lease(store, "worker-live", lease_ttl));

        let step_started = Arc::new(tokio::sync::Notify::new());
        let notify = step_started.clone();
        let body: StepFunc = Box::new(move || {
            let notify = notify.clone();
            Box::pin(async move {
                notify.notify_waiters();
                tokio::time::sleep(Duration::from_millis(200)).await;
                Ok("done".to_string())
            })
        });

        let no_retry = RetryPolicy {
            max_attempts: 1,
            backoff: Backoff::Fixed(Duration::from_millis(1)),
        };
        let workflow = Workflow {
            id: "wf-lease-renew".to_string(),
            steps: vec![StepDef {
                name: "slow".into(),
                retry_policy: no_retry,
                timeout: None,
            }],
        };

        let run_handle = tokio::spawn({
            let engine = Arc::clone(&engine);
            let workflow = workflow.clone();
            async move { engine.run(workflow, vec![body]).await }
        });

        tokio::time::timeout(Duration::from_secs(5), step_started.notified())
            .await
            .expect("step should start within deadline");
        tokio::time::sleep(Duration::from_millis(80)).await;

        let steal = engine
            .store()
            .claim_step("wf-lease-renew", 0, "worker-other", Duration::from_secs(60))
            .expect("claim attempt");
        assert!(
            matches!(steal, ClaimResult::HeldByOther),
            "renewing worker must keep the lease while the step is still running"
        );

        run_handle
            .await
            .expect("join workflow task")
            .expect("workflow should complete");
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn resume_input_survives_simulated_restart_from_sqlite_log() {
        let path = std::env::temp_dir().join(format!(
            "agentq-resume-input-{}-{}.db",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .expect("clock")
                .as_nanos()
        ));
        let path_str = path
            .to_str()
            .expect("temp path must be valid UTF-8 for sqlite");

        let store = crate::SqliteStore::new(path_str).expect("open db");
        let engine = sqlite_engine(store);

        let step1_attempts = AtomicUsize::new(0);
        let step0: StepFunc = Box::new(|| Box::pin(async { Ok("first".to_string()) }));
        let step1: StepFunc = Box::new(move || {
            let n = step1_attempts.fetch_add(1, Ordering::SeqCst);
            Box::pin(async move {
                if n == 0 {
                    Err(Box::new(WaitForInput("need approval".to_string()))
                        as Box<dyn std::error::Error + Send + Sync>)
                } else {
                    Ok("approved".to_string())
                }
            })
        });

        let no_retry = RetryPolicy {
            max_attempts: 1,
            backoff: Backoff::Fixed(Duration::from_millis(1)),
        };
        let workflow = Workflow {
            id: "wf-resume-input-restart".to_string(),
            steps: vec![
                StepDef {
                    name: "first".into(),
                    retry_policy: no_retry.clone(),
                    timeout: None,
                },
                StepDef {
                    name: "approval".into(),
                    retry_policy: no_retry,
                    timeout: None,
                },
            ],
        };

        engine
            .run(workflow, vec![step0, step1])
            .await
            .expect("run should pause at waiting step");

        let resume_text = "approved by human after restart";
        engine
            .resume("wf-resume-input-restart", 1, resume_text.to_string())
            .await
            .expect("resume should complete workflow");

        drop(engine);

        let store_after_restart =
            crate::SqliteStore::new(path_str).expect("reopen db after simulated restart");
        let engine_after_restart = sqlite_engine(store_after_restart);

        assert_eq!(
            engine_after_restart.resume_input("wf-resume-input-restart", 1),
            Some(resume_text.to_string()),
            "resume input must be read from durable events, not process memory"
        );

        let _ = std::fs::remove_file(path);
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn waiting_step_cannot_be_re_run_into_duplicate_side_effect() {
        use std::sync::Arc;

        let store = crate::SqliteStore::new(":memory:").expect("open db");
        let engine = sqlite_engine(store);
        let approval_runs = Arc::new(AtomicUsize::new(0));

        let no_retry = RetryPolicy {
            max_attempts: 1,
            backoff: Backoff::Fixed(Duration::from_millis(1)),
        };
        let workflow = Workflow {
            id: "wf-wait-rerun".to_string(),
            steps: vec![
                StepDef {
                    name: "first".into(),
                    retry_policy: no_retry.clone(),
                    timeout: None,
                },
                StepDef {
                    name: "approval".into(),
                    retry_policy: no_retry.clone(),
                    timeout: None,
                },
            ],
        };

        let make_bodies = |counter: Arc<AtomicUsize>| {
            let step0: StepFunc = Box::new(|| Box::pin(async { Ok("first".to_string()) }));
            let counter = counter.clone();
            let step1: StepFunc = Box::new(move || {
                let n = counter.fetch_add(1, Ordering::SeqCst);
                Box::pin(async move {
                    if n == 0 {
                        Err(Box::new(WaitForInput("needs approval".to_string()))
                            as Box<dyn std::error::Error + Send + Sync>)
                    } else {
                        Ok("approved".to_string())
                    }
                })
            });
            vec![step0, step1]
        };

        engine
            .run(workflow.clone(), make_bodies(approval_runs.clone()))
            .await
            .expect("first run should pause at waiting step");

        assert_eq!(approval_runs.load(Ordering::SeqCst), 1);
        let events = engine
            .store()
            .load_events("wf-wait-rerun")
            .expect("load events");
        assert!(
            events
                .iter()
                .any(|e| matches!(e, Event::StepWaiting { step_index: 1, .. }))
        );
        assert_eq!(count_workflow_completed(&events), 0);

        engine
            .run(workflow.clone(), make_bodies(approval_runs.clone()))
            .await
            .expect("second run should not re-execute waiting step");

        assert_eq!(
            approval_runs.load(Ordering::SeqCst),
            1,
            "approval step must not run again while waiting"
        );

        engine
            .resume("wf-wait-rerun", 1, "approved".to_string())
            .await
            .expect("resume should complete workflow");

        assert_eq!(approval_runs.load(Ordering::SeqCst), 2);
        let events = engine
            .store()
            .load_events("wf-wait-rerun")
            .expect("load events");
        assert_eq!(count_workflow_completed(&events), 1);
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn completed_workflow_cannot_accumulate_duplicate_completion_events() {
        let store = crate::SqliteStore::new(":memory:").expect("open db");
        let engine = sqlite_engine(store);

        let no_retry = RetryPolicy {
            max_attempts: 1,
            backoff: Backoff::Fixed(Duration::from_millis(1)),
        };
        let workflow = Workflow {
            id: "wf-once-complete".to_string(),
            steps: vec![
                StepDef {
                    name: "a".into(),
                    retry_policy: no_retry.clone(),
                    timeout: None,
                },
                StepDef {
                    name: "b".into(),
                    retry_policy: no_retry,
                    timeout: None,
                },
            ],
        };

        let bodies: Vec<StepFunc> = vec![
            Box::new(|| Box::pin(async { Ok("a".to_string()) })),
            Box::new(|| Box::pin(async { Ok("b".to_string()) })),
        ];

        engine
            .run(workflow.clone(), bodies)
            .await
            .expect("workflow should complete");

        let events = engine
            .store()
            .load_events("wf-once-complete")
            .expect("load events");
        assert_eq!(count_workflow_completed(&events), 1);

        let bodies_again: Vec<StepFunc> = vec![
            Box::new(|| Box::pin(async { Ok("a".to_string()) })),
            Box::new(|| Box::pin(async { Ok("b".to_string()) })),
        ];

        engine
            .run(workflow.clone(), bodies_again)
            .await
            .expect("re-run on completed workflow should succeed");

        let events = engine
            .store()
            .load_events("wf-once-complete")
            .expect("load events");
        assert_eq!(count_workflow_completed(&events), 1);
    }

    #[tokio::test]
    async fn cancel_never_run_workflow_appends_one_cancelled_event() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let no_retry = RetryPolicy {
            max_attempts: 1,
            backoff: Backoff::Fixed(Duration::from_millis(1)),
        };
        let workflow = Workflow {
            id: "wf-cancel-fresh".to_string(),
            steps: vec![StepDef {
                name: "only".into(),
                retry_policy: no_retry,
                timeout: None,
            }],
        };

        engine
            .register_workflow(
                workflow,
                vec![Box::new(|| Box::pin(async { Ok("x".into()) }))],
            )
            .expect("register");

        engine
            .cancel("wf-cancel-fresh", "user requested".to_string())
            .await
            .expect("cancel");

        let events = store.load_events("wf-cancel-fresh").expect("load");
        assert_eq!(count_workflow_cancelled(&events), 1);
    }

    #[tokio::test]
    async fn cancel_between_steps_stops_before_next_step_and_no_completion() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let step1_ran = Arc::new(AtomicUsize::new(0));
        let step1_ran_clone = step1_ran.clone();
        let step0_done = Arc::new(tokio::sync::Notify::new());
        let step0_done_clone = step0_done.clone();

        let no_retry = RetryPolicy {
            max_attempts: 1,
            backoff: Backoff::Fixed(Duration::from_millis(1)),
        };
        let workflow = Workflow {
            id: "wf-cancel-mid".to_string(),
            steps: vec![
                StepDef {
                    name: "first".into(),
                    retry_policy: no_retry.clone(),
                    timeout: None,
                },
                StepDef {
                    name: "second".into(),
                    retry_policy: no_retry,
                    timeout: None,
                },
            ],
        };

        let step0: StepFunc = Box::new(move || {
            let notify = step0_done_clone.clone();
            Box::pin(async move {
                notify.notify_one();
                tokio::task::yield_now().await;
                Ok("first".to_string())
            })
        });
        let step1: StepFunc = Box::new(move || {
            let counter = step1_ran_clone.clone();
            Box::pin(async move {
                counter.fetch_add(1, Ordering::SeqCst);
                Ok("second".to_string())
            })
        });

        let engine = Arc::new(engine);
        let run_engine = engine.clone();
        let run_handle = tokio::spawn(async move {
            run_engine
                .run(workflow, vec![step0, step1])
                .await
                .expect("run");
        });

        step0_done.notified().await;
        engine
            .cancel("wf-cancel-mid", "stop".to_string())
            .await
            .expect("cancel");
        run_handle.await.expect("run task");

        assert_eq!(
            step1_ran.load(Ordering::SeqCst),
            0,
            "step after cancellation must not run"
        );
        let events = store.load_events("wf-cancel-mid").expect("load events");
        assert_eq!(count_workflow_cancelled(&events), 1);
        assert_eq!(count_workflow_completed(&events), 0);
    }

    #[tokio::test]
    async fn cancel_during_retry_backoff_stops_before_next_attempt() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let attempts = Arc::new(AtomicUsize::new(0));
        let attempts_clone = attempts.clone();

        let workflow = Workflow {
            id: "wf-cancel-during-backoff".to_string(),
            steps: vec![StepDef {
                name: "flaky".into(),
                retry_policy: RetryPolicy {
                    max_attempts: 3,
                    backoff: Backoff::Fixed(Duration::from_millis(300)),
                },
                timeout: None,
            }],
        };

        let step: StepFunc = Box::new(move || {
            attempts_clone.fetch_add(1, Ordering::SeqCst);
            Box::pin(async { Err("transient failure".into()) })
        });

        let engine = Arc::new(engine);
        let run_engine = engine.clone();
        let run_handle = tokio::spawn(async move {
            run_engine.run(workflow, vec![step]).await.expect("run");
        });

        tokio::time::sleep(Duration::from_millis(100)).await;
        engine
            .cancel("wf-cancel-during-backoff", "stop mid-backoff".to_string())
            .await
            .expect("cancel");
        run_handle.await.expect("run task");

        assert_eq!(
            attempts.load(Ordering::SeqCst),
            1,
            "cancelling during retry backoff must prevent the next attempt from running"
        );
        let events = store
            .load_events("wf-cancel-during-backoff")
            .expect("load events");
        assert_eq!(count_workflow_cancelled(&events), 1);
        assert_eq!(count_workflow_completed(&events), 0);
        assert!(
            !events
                .iter()
                .any(|e| matches!(e, Event::WorkflowFailed { .. })),
            "cancellation during backoff must not also record a WorkflowFailed"
        );
    }

    #[tokio::test]
    async fn cancel_while_waiting_blocks_resume() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let no_retry = RetryPolicy {
            max_attempts: 1,
            backoff: Backoff::Fixed(Duration::from_millis(1)),
        };
        let workflow = Workflow {
            id: "wf-cancel-wait".to_string(),
            steps: vec![
                StepDef {
                    name: "first".into(),
                    retry_policy: no_retry.clone(),
                    timeout: None,
                },
                StepDef {
                    name: "approval".into(),
                    retry_policy: no_retry,
                    timeout: None,
                },
            ],
        };

        let step0: StepFunc = Box::new(|| Box::pin(async { Ok("first".to_string()) }));
        let step1: StepFunc = Box::new(|| {
            Box::pin(async {
                Err(Box::new(WaitForInput("need approval".to_string()))
                    as Box<dyn std::error::Error + Send + Sync>)
            })
        });

        engine
            .run(workflow, vec![step0, step1])
            .await
            .expect("run pauses at waiting");

        engine
            .cancel("wf-cancel-wait", "no longer needed".to_string())
            .await
            .expect("cancel");

        let result = engine
            .resume("wf-cancel-wait", 1, "too late".to_string())
            .await;
        assert!(
            matches!(result, Err(EngineError::Store(StoreError::Backend(_)))),
            "resume on cancelled workflow should error"
        );
    }

    #[tokio::test]
    async fn cancel_twice_appends_one_cancelled_event() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let no_retry = RetryPolicy {
            max_attempts: 1,
            backoff: Backoff::Fixed(Duration::from_millis(1)),
        };
        let workflow = Workflow {
            id: "wf-cancel-twice".to_string(),
            steps: vec![StepDef {
                name: "only".into(),
                retry_policy: no_retry,
                timeout: None,
            }],
        };

        engine
            .register_workflow(
                workflow,
                vec![Box::new(|| Box::pin(async { Ok("x".into()) }))],
            )
            .expect("register");

        engine
            .cancel("wf-cancel-twice", "once".to_string())
            .await
            .expect("first cancel");
        engine
            .cancel("wf-cancel-twice", "again".to_string())
            .await
            .expect("second cancel");

        let events = store.load_events("wf-cancel-twice").expect("load");
        assert_eq!(count_workflow_cancelled(&events), 1);
    }

    #[tokio::test]
    async fn cancel_completed_workflow_does_not_add_cancelled_event() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let no_retry = RetryPolicy {
            max_attempts: 1,
            backoff: Backoff::Fixed(Duration::from_millis(1)),
        };
        let workflow = Workflow {
            id: "wf-cancel-done".to_string(),
            steps: vec![StepDef {
                name: "only".into(),
                retry_policy: no_retry,
                timeout: None,
            }],
        };

        engine
            .run(
                workflow,
                vec![Box::new(|| Box::pin(async { Ok("done".to_string()) }))],
            )
            .await
            .expect("run completes");

        engine
            .cancel("wf-cancel-done", "too late".to_string())
            .await
            .expect("cancel is no-op");

        let events = store.load_events("wf-cancel-done").expect("load");
        assert_eq!(count_workflow_cancelled(&events), 0);
        assert_eq!(count_workflow_completed(&events), 1);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
    async fn concurrent_cancel_and_completion_yield_one_terminal_event() {
        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = Arc::new(WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        ));

        let mut handles = Vec::new();
        for i in 0..300 {
            let workflow_id = format!("wf-race-{i}");
            let no_retry = RetryPolicy {
                max_attempts: 1,
                backoff: Backoff::Fixed(Duration::from_millis(1)),
            };
            let workflow = Workflow {
                id: workflow_id.clone(),
                steps: vec![StepDef {
                    name: "only".into(),
                    retry_policy: no_retry,
                    timeout: None,
                }],
            };

            let run_engine = engine.clone();
            handles.push(tokio::spawn(async move {
                let _ = run_engine
                    .run(
                        workflow,
                        vec![Box::new(|| Box::pin(async { Ok("done".to_string()) }))],
                    )
                    .await;
            }));

            let cancel_engine = engine.clone();
            handles.push(tokio::spawn(async move {
                let _ = cancel_engine
                    .cancel(&workflow_id, "racing cancel".to_string())
                    .await;
            }));
        }

        for handle in handles {
            let _ = handle.await;
        }

        for i in 0..300 {
            let workflow_id = format!("wf-race-{i}");
            let events = store.load_events(&workflow_id).expect("load events");
            let terminal_count = events
                .iter()
                .filter(|e| {
                    matches!(
                        e,
                        Event::WorkflowCompleted { .. }
                            | Event::WorkflowCancelled { .. }
                            | Event::WorkflowFailed { .. }
                    )
                })
                .count();
            assert!(
                terminal_count <= 1,
                "workflow {workflow_id} got {terminal_count} terminal events, expected at most 1: {events:?}"
            );
        }
    }

    #[tokio::test]
    async fn resume_failure_cleans_up_in_flight_workflow() {
        struct DropFlag(Arc<std::sync::atomic::AtomicBool>);
        impl Drop for DropFlag {
            fn drop(&mut self) {
                self.0.store(true, Ordering::SeqCst);
            }
        }

        let queue = Queue::builder().start();
        let store = MemStore::new();
        let engine = WorkflowEngine::new(
            queue,
            store.clone(),
            "worker-1".to_string(),
            Duration::from_secs(60),
            Priority::High,
        );

        let dropped = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let guard = DropFlag(dropped.clone());

        let attempts = AtomicUsize::new(0);
        let body: StepFunc = Box::new(move || {
            let _keep_alive = &guard;
            let n = attempts.fetch_add(1, Ordering::SeqCst);
            Box::pin(async move {
                if n == 0 {
                    Err(Box::new(WaitForInput("needs approval".to_string()))
                        as Box<dyn std::error::Error + Send + Sync>)
                } else {
                    Err("boom".into())
                }
            })
        });

        let workflow = Workflow {
            id: "wf-resume-cleanup".to_string(),
            steps: vec![StepDef {
                name: "only".to_string(),
                retry_policy: RetryPolicy {
                    max_attempts: 1,
                    backoff: Backoff::Fixed(Duration::from_millis(1)),
                },
                timeout: None,
            }],
        };

        engine
            .run(workflow, vec![body])
            .await
            .expect("run should pause at waiting step");
        assert!(
            !dropped.load(Ordering::SeqCst),
            "step body must still be retained while the workflow is waiting"
        );

        let result = engine
            .resume("wf-resume-cleanup", 0, "approved".to_string())
            .await;
        assert!(
            matches!(result, Err(EngineError::WorkflowFailed { .. })),
            "resumed attempt should exhaust its retry policy and fail the workflow"
        );

        assert!(
            dropped.load(Ordering::SeqCst),
            "workflow must be removed from in_flight after resume fails permanently"
        );
    }
}