taquba-workflow 0.13.0

use std::collections::HashMap;
use std::future::Future;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use futures_util::TryStreamExt;
use taquba::object_store::ObjectStore;
use taquba::{
    Clock, EnqueueOptions, EnqueueRequest, EnqueueResult, JobRecord, JobStatus, Queue,
    SettlementEffects, WaitOutcome, WorkerHandle,
};
use tokio_util::sync::CancellationToken;
use tracing::{debug, instrument, warn};

use crate::durable::{
    self, DurableCurrentStep, DurableErrorKind, DurableRunOutcome, DurableRunRecord,
    DurableRunResult, DurableStepOutcome, DurableStepOutcomeRecord, DurableTermination,
};
use crate::effects::StagedEffects;
use crate::error::{Error, Result};
use crate::group::{GroupStore, Membership, RunGroup, pending_member, terminated_member};
use crate::keys::{
    DEDUP_PREFIX, GROUP_TERMINAL_KV_PREFIX, HEADER_RUN_ID, HEADER_STEP, HEADER_TERMINAL,
    RESERVED_HEADER_PREFIX, RESERVED_KV_PREFIX, RunId, TERMINAL_KV_PREFIX, hash_input,
    outcome_kv_key, run_kv_key, step_kv_key,
};
use crate::memo::{MemoStore, RUN_RESULT_MEMO_KEY};
use crate::runner::{StepErrorKind, StepOutcome, StepRunner, Trigger};
use crate::sweep::{Clearable, Sweep, run_periodically};
use crate::terminal::{RunOutcome, TerminalHook, TerminalStatus};
use crate::view::WorkflowView;
use crate::worker::{ClaimedStep, StepWorker};

/// The encoded current-step pointer for `job_id` at `step_number`.
fn current_step_bytes(step_number: u32, job_id: &str) -> Vec<u8> {
    durable::encode(&DurableCurrentStep {
        step_number,
        job_id: job_id.to_string(),
    })
}

/// The portion of `delay` still ahead of `now_ms`, measured from
/// `stored_at_ms`. Saturates to the full delay if the clock reads
/// earlier than the stored timestamp.
fn remaining_delay(stored_at_ms: u64, now_ms: u64, delay: Duration) -> Duration {
    let elapsed = Duration::from_millis(now_ms.saturating_sub(stored_at_ms));
    delay.saturating_sub(elapsed)
}

/// Per-step enqueue options the runtime forwards through to Taquba. The
/// runtime always owns `headers` (it injects [`HEADER_RUN_ID`] and
/// [`HEADER_STEP`]) and `dedup_key` (it derives one from
/// `(run_id, step_number)`), so callers only pick the three fields below.
#[derive(Debug, Default)]
pub(crate) struct StepEnqueueOpts {
    /// Earliest claimable time for the step. `None` means immediate.
    pub(crate) run_at: Option<SystemTime>,
    /// Per-step priority override.
    pub(crate) priority: Option<u32>,
    /// Per-step `max_attempts` override.
    pub(crate) max_attempts: Option<u32>,
    /// Additional runtime-owned reserved headers for the step job.
    pub(crate) reserved_headers: Vec<(&'static str, String)>,
}

/// The settings of a run's steps, applied by [`WorkflowRuntime::submit`]
/// through [`RunSpec::options`], by [`RunGroup::submit`] and
/// [`RunGroup::resume`] to every member of a group and by
/// [`jobs::JobRunner::submit_with`](crate::jobs::JobRunner::submit_with)
/// to a job.
#[derive(Debug, Clone, Default)]
pub struct RunOptions {
    /// Submitter-supplied metadata, threaded through every step of the
    /// run and surfaced to the terminal hook. Reserved `workflow.*` keys
    /// are rejected at submission with [`Error::ReservedHeaderInSubmit`].
    pub headers: HashMap<String, String>,
    /// Priority of every step; the queue's default when `None`.
    pub priority: Option<u32>,
    /// Attempt limit of every step; the queue's `max_attempts` when
    /// `None`.
    pub max_attempts_per_step: Option<u32>,
    /// Earliest time the first step may run. The step-0 job waits in
    /// the queue's scheduled state until the queue's clock passes this
    /// time; `None` makes it claimable at once.
    pub run_at: Option<SystemTime>,
}

/// Spec passed to [`WorkflowRuntime::submit`].
#[derive(Debug, Clone, Default)]
pub struct RunSpec {
    /// Caller-supplied run identifier. If `None`, the runtime generates
    /// a ULID. The dedup key for the first step job is `run:{run_id}:0`, so
    /// re-submitting the same `run_id` while the run is active returns the
    /// existing job rather than creating a duplicate.
    ///
    /// A terminated run releases its id for re-submission. The second
    /// run shares the first run's memo and step-output entries, which is
    /// what makes a re-submission resume from them, and under
    /// [`WorkflowRuntimeBuilder::memo_retention`] the first run's marker
    /// expires against those shared entries even while the second run is
    /// executing. The second run then re-executes the affected steps.
    pub run_id: Option<RunId>,
    /// Bytes handed to the runner as the first step's payload.
    pub input: Vec<u8>,
    /// The settings of the run's steps.
    pub options: RunOptions,
    /// Effects applied in the same transaction as the step-0 enqueue, as
    /// [`taquba::Queue::enqueue_with_effects`] applies them: the enqueues, the
    /// KV writes, the KV deletes and the expiry entries. Applied only when the
    /// submission is new: a duplicate submission's effects are dropped, and the
    /// effects do not participate in the duplicate-submission input check. A KV
    /// key written or deleted must not start with the reserved `workflow/`
    /// prefix ([`RESERVED_KV_PREFIX`]). Values are capped at
    /// [`taquba::MAX_KV_VALUE_SIZE`].
    pub effects: SettlementEffects,
}

/// Outcome of [`WorkflowRuntime::submit`].
///
/// `submit` is idempotent on `run_id`: re-submitting an active run is a
/// no-op and the returned `SubmitOutcome` carries `newly_submitted = false`.
#[derive(Debug, Clone)]
pub struct SubmitOutcome {
    /// The run's identifier (generated if the spec didn't carry one).
    pub run_id: RunId,
    /// `true` if this call enqueued a new run; `false` if a run with this
    /// id was already active (its durable run record exists) and this
    /// call was a no-op. Call
    /// [`WorkflowRuntime::status`] for the run's current state when
    /// needed.
    pub newly_submitted: bool,
    /// The id of the queue job currently representing the run: its
    /// first step for a new submission, and the step the run has reached
    /// for a duplicate, read from the run's durable current-step pointer.
    pub job_id: String,
}

/// Status snapshot of a run, read from its durable state by
/// [`WorkflowView::status`].
#[derive(Debug, Clone)]
pub struct RunStatus {
    /// The run's identifier.
    pub run_id: RunId,
    /// Lifecycle state of the run's current step, or its termination.
    pub state: RunState,
    /// Step number of the run's current step; the final step of a
    /// terminated run.
    pub current_step: u32,
}

/// Lifecycle state tracked in [`RunStatus::state`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RunState {
    /// A step job exists in the queue but has not yet been claimed.
    Pending,
    /// A step is currently being processed by a worker.
    Running,
    /// [`WorkflowRuntime::cancel`] was called for this run and the run
    /// has not yet terminated. Reported until the in-flight step
    /// returns and the runtime settles the run as
    /// [`crate::TerminalStatus::Cancelled`]; after that,
    /// [`WorkflowRuntime::status`] returns `None`.
    ///
    /// Only set by external cancellation. A pure runner-issued
    /// [`crate::StepOutcome::Cancel`] (with no external `cancel()`
    /// call) terminates as `Cancelled` without ever transitioning
    /// through `Cancelling`: a runner-issued cancel is observed when
    /// `run_step` returns, and the run terminates at that point.
    Cancelling,
    /// The run reached a terminal state. Reported from the run's terminal
    /// record, written with the terminating settlement and removed with
    /// the run's memo entries by the memo sweep under
    /// [`WorkflowRuntimeBuilder::memo_retention`].
    Terminated(RunTermination),
}

/// A run result record as read back: the committed outcome and the
/// termination the record belongs to.
#[derive(Debug, Clone)]
pub(crate) struct RunResult {
    pub(crate) termination: RunTermination,
    pub(crate) outcome: RunOutcome,
}

/// The committed terminal outcome of a run, as
/// [`RunState::Terminated`] reports it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunTermination {
    /// How the run ended.
    pub status: TerminalStatus,
    /// The failure reason, or the runner's reason for a
    /// [`StepOutcome::Cancel`]; `None` for a success and for an external
    /// cancellation.
    pub error: Option<String>,
    /// The classification of a failure: the kind of the
    /// [`StepError`](crate::StepError) that dead-lettered the run, or
    /// [`StepErrorKind::Permanent`] for a [`StepOutcome::Fail`] verdict.
    /// `None` for a success, a cancellation and a termination outside
    /// the worker.
    pub error_kind: Option<StepErrorKind>,
    /// The number of the step whose settlement terminated the run.
    pub final_step: u32,
    /// The runtime clock's time at the terminating settlement, in
    /// milliseconds since the Unix epoch.
    pub terminated_at_ms: u64,
}

impl From<DurableTermination> for RunTermination {
    fn from(record: DurableTermination) -> Self {
        Self {
            status: record.status.into(),
            error: record.error,
            error_kind: record.error_kind.map(Into::into),
            final_step: record.final_step,
            terminated_at_ms: record.terminated_at_ms,
        }
    }
}

/// The end of a run, as [`WorkflowRuntime::wait`] reports it.
#[derive(Debug, Clone)]
pub struct RunEnd {
    /// The run's termination, from its terminal record.
    pub termination: RunTermination,
    /// The committed outcome, when the worker that terminated the run
    /// recorded one; see [`WorkflowRuntime::outcome`].
    pub outcome: Option<RunOutcome>,
}

/// Builder for [`WorkflowRuntime`].
///
/// Construct via [`WorkflowRuntime::builder`].
pub struct WorkflowRuntimeBuilder<R, H> {
    queue: Arc<Queue>,
    object_store: Arc<dyn ObjectStore>,
    queue_name: String,
    memo_prefix: Option<String>,
    runner: R,
    terminal_hook: H,
    max_concurrent_steps: usize,
    poll_interval: Duration,
    memo_retention: Option<Duration>,
    group_retention: Option<Duration>,
    step_output_replay: bool,
    clock: Arc<dyn Clock>,
}

impl<R: StepRunner, H: TerminalHook> WorkflowRuntimeBuilder<R, H> {
    /// The Taquba queue name that step jobs are enqueued onto. Defaults to
    /// `"workflow-steps"`. Multiple runtimes can share a `Queue` handle by
    /// using distinct queue names.
    pub fn queue_name(mut self, name: impl Into<String>) -> Self {
        self.queue_name = name.into();
        self
    }

    /// The object-store path prefix of the [`Delivery::memo`](crate::Delivery::memo)
    /// entries. Defaults to `"{queue_name}-memo"`, so runtimes with
    /// distinct queue names on one object store have distinct memo
    /// namespaces.
    pub fn memo_prefix(mut self, prefix: impl Into<String>) -> Self {
        self.memo_prefix = Some(prefix.into());
        self
    }

    /// Maximum number of steps processed concurrently in [`WorkflowRuntime::run`].
    /// Defaults to 16.
    pub fn max_concurrent_steps(mut self, n: usize) -> Self {
        assert!(n > 0, "max_concurrent_steps must be at least 1");
        self.max_concurrent_steps = n;
        self
    }

    /// Maximum time the worker loop waits on an empty queue before re-checking.
    /// Defaults to 250ms. The retention sweeps run at this interval.
    pub fn poll_interval(mut self, interval: Duration) -> Self {
        self.poll_interval = interval;
        self
    }

    /// Enable memo retention with the given window. When set, the
    /// runtime writes a terminal marker for every run that reaches a
    /// terminal state, and the in-process sweeper clears that run's memo
    /// entries and terminal record `retention` after termination. When
    /// unset (default), no marker is written and memo entries and
    /// terminal records are retained indefinitely.
    pub fn memo_retention(mut self, retention: Duration) -> Self {
        self.memo_retention = Some(retention);
        self
    }

    /// Enable content-addressed replay of runner-returned step outcomes.
    ///
    /// When enabled, the runtime writes every [`StepOutcome`] the runner
    /// returns, including `Fail` and `Cancel`, to object storage before
    /// applying it. Step errors ([`StepError`](crate::StepError)) are not
    /// recorded, so retries still invoke the runner. The replay key is
    /// scoped to `(run_id, step_number, SHA-256(step payload))`. If the
    /// same step is delivered again after a crash before ack, the stored
    /// outcome is replayed without invoking the runner again. The record
    /// includes the effects staged through
    /// [`Delivery::effects`](crate::Delivery::effects), so a
    /// replayed outcome applies them as well. A replayed
    /// [`StepOutcome::Continue`] with a [`Trigger::After`] delay reduces
    /// the delay by the time already elapsed since the outcome was
    /// stored, preserving the original schedule.
    ///
    /// This is disabled by default because it adds one object-store read
    /// per step delivery (the replay lookup) plus one write per recorded
    /// outcome, and makes that write part of step settlement.
    pub fn step_output_replay(mut self) -> Self {
        self.step_output_replay = true;
        self
    }

    /// Override the [`Clock`] the runtime reads its timestamps from.
    /// Defaults to the same clock the [`Queue`] was opened with (via
    /// [`Queue::clock`]).
    pub fn clock(mut self, clock: Arc<dyn Clock>) -> Self {
        self.clock = clock;
        self
    }

    /// Remove a [`RunGroup`]'s state (its manifest, member records and
    /// the memo entries and terminal records of its members) `retention` after a
    /// [`RunGroup::results`] consumer observed the last member's
    /// termination, through a sweep when the worker starts and at every
    /// poll interval after that. When unset (default), no group
    /// terminal marker is written. A group whose results are never
    /// consumed is retained until [`RunGroup::forget`] in either case.
    pub fn group_retention(mut self, retention: Duration) -> Self {
        self.group_retention = Some(retention);
        self
    }

    /// Finalize the builder.
    pub fn build(self) -> WorkflowRuntime<R, H>
    where
        H: 'static,
    {
        let terminal_hook = Arc::new(self.terminal_hook);
        let observes: Arc<dyn Fn(&RunOutcome) -> bool + Send + Sync> = {
            let hook = terminal_hook.clone();
            Arc::new(move |outcome| hook.observes(outcome))
        };
        let memo_prefix = self
            .memo_prefix
            .unwrap_or_else(|| format!("{}-memo", self.queue_name));
        let memo_store = MemoStore::new(self.object_store.clone(), memo_prefix.clone());
        let group_store = GroupStore::new(
            self.object_store,
            memo_prefix,
            memo_store.clone(),
            self.queue.clone(),
        );
        let memo_sweep = self.memo_retention.map(|retention| {
            Arc::new(Sweep::new(
                TERMINAL_KV_PREFIX,
                retention,
                RunStore {
                    memo_store: memo_store.clone(),
                },
            ))
        });
        let group_sweep = self.group_retention.map(|retention| {
            Arc::new(Sweep::new(
                GROUP_TERMINAL_KV_PREFIX,
                retention,
                group_store.clone(),
            ))
        });
        let view = WorkflowView::new(self.queue.view().clone(), memo_store.clone());
        let core = RuntimeCore {
            queue: self.queue,
            view,
            queue_name: self.queue_name,
            max_concurrent_steps: self.max_concurrent_steps,
            poll_interval: self.poll_interval,
            memo_store,
            group_store,
            memo_sweep,
            group_sweep,
            step_output_replay: self.step_output_replay,
            clock: self.clock,
            observes,
        };
        let inner = RuntimeInner {
            runner: self.runner,
            terminal_hook,
            core: Arc::new(core),
        };
        WorkflowRuntime {
            inner: Arc::new(inner),
        }
    }
}

/// The retained state of a terminated run: its memo and step-output
/// entries in the object store and its terminal record in the queue's
/// KV namespace, removed together by the memo sweep.
struct RunStore {
    memo_store: MemoStore,
}

impl Clearable for RunStore {
    type Error = Error;

    async fn clear(&self, run_id: &RunId) -> Result<Vec<Vec<u8>>> {
        self.memo_store.clear_memos_for_run(run_id).await?;
        Ok(vec![outcome_kv_key(run_id)])
    }
}

/// Durable runtime for workflow runs. Cheap to clone (internally `Arc`).
pub struct WorkflowRuntime<R, H> {
    pub(crate) inner: Arc<RuntimeInner<R, H>>,
}

impl<R, H> Clone for WorkflowRuntime<R, H> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
        }
    }
}

/// A runtime's [`StepRunner`] and [`TerminalHook`], with the shared
/// [`RuntimeCore`] they operate on: the executing half of a runtime,
/// held by the worker.
pub(crate) struct RuntimeInner<R, H> {
    pub(crate) runner: R,
    pub(crate) terminal_hook: Arc<H>,
    pub(crate) core: Arc<RuntimeCore>,
}

/// The control half of a runtime: the queue, the stores and the settings,
/// with every operation that reads or settles run state without invoking the
/// runner or the hook. It is not generic over the runner and hook types, so the
/// worker, [`WorkflowRuntime`] and [`RunGroup`](crate::RunGroup) share it.
pub(crate) struct RuntimeCore {
    pub(crate) queue: Arc<Queue>,
    pub(crate) view: WorkflowView,
    queue_name: String,
    max_concurrent_steps: usize,
    poll_interval: Duration,
    pub(crate) memo_store: MemoStore,
    pub(crate) group_store: GroupStore,
    /// The sweep that removes the memo entries and the terminal record
    /// of a run a window after its termination, when
    /// [`WorkflowRuntimeBuilder::memo_retention`] is set. Without it no
    /// terminal marker is written.
    pub(crate) memo_sweep: Option<Arc<Sweep>>,
    /// The sweep that removes the state of a group a window after its
    /// last member terminated, when
    /// [`WorkflowRuntimeBuilder::group_retention`] is set. Without it no
    /// group marker is written.
    pub(crate) group_sweep: Option<Arc<Sweep>>,
    /// Whether runner-returned step outcomes are persisted and replayed
    /// by `(run_id, step_number, SHA-256(step payload))`.
    pub(crate) step_output_replay: bool,
    /// Time source. Defaults to the queue's clock; tests can substitute
    /// a [`MockClock`](taquba::MockClock) to virtualise time.
    pub(crate) clock: Arc<dyn Clock>,
    /// [`TerminalHook::observes`] of the runtime's hook, which decides
    /// whether a termination enqueues a notification job.
    observes: Arc<dyn Fn(&RunOutcome) -> bool + Send + Sync>,
}

impl<R: StepRunner, H: TerminalHook> WorkflowRuntime<R, H> {
    /// Start configuring a runtime. Takes the four required dependencies
    /// (Taquba queue, object store, [`StepRunner`], [`TerminalHook`]); optional
    /// fields are set via [`WorkflowRuntimeBuilder`] methods before [`build`].
    ///
    /// The object store backs [`Delivery::memo`]; it does **not** need to be
    /// the same store the [`Queue`] was opened with, though sharing one store
    /// is the common case (just clone the `Arc`). Use a distinct
    /// [`WorkflowRuntimeBuilder::memo_prefix`] when multiple runtimes share one
    /// store.
    ///
    /// Use [`crate::NoopTerminalHook`] if you don't need terminal callbacks.
    ///
    /// [`Delivery::memo`]: crate::Delivery::memo
    /// [`build`]: WorkflowRuntimeBuilder::build
    pub fn builder(
        queue: Arc<Queue>,
        object_store: Arc<dyn ObjectStore>,
        runner: R,
        terminal_hook: H,
    ) -> WorkflowRuntimeBuilder<R, H> {
        let clock = queue.clock();
        WorkflowRuntimeBuilder {
            queue,
            object_store,
            queue_name: "workflow-steps".to_string(),
            memo_prefix: None,
            runner,
            terminal_hook,
            max_concurrent_steps: 16,
            poll_interval: Duration::from_millis(250),
            memo_retention: None,
            group_retention: None,
            step_output_replay: false,
            clock,
        }
    }

    /// Submit a new run. Enqueues step 0 with payload `spec.input`.
    ///
    /// Idempotent on `(run_id, spec.input)`: if a run with the same id is
    /// already active (its durable run record in Taquba's user KV
    /// namespace exists, whichever process submitted it) and
    /// `spec.input` matches the original submission, this
    /// call is a no-op and the returned [`SubmitOutcome`] has
    /// `newly_submitted = false`. A re-submission of an active `run_id`
    /// with a *different* input is rejected with [`Error::InputMismatch`];
    /// pick a fresh `run_id` for a new run.
    pub async fn submit(&self, spec: RunSpec) -> Result<SubmitOutcome> {
        self.inner.core.submit(spec).await
    }

    /// The runtime's view of its store: the read-only queries over the queue
    /// and the memo store the runtime writes to.
    pub fn view(&self) -> &WorkflowView {
        &self.inner.core.view
    }

    /// [`WorkflowView::status`] through the runtime's view, so the status is
    /// available after a restart and from any runtime over the same queue.
    pub async fn status(&self, run_id: &RunId) -> Result<Option<RunStatus>> {
        self.inner.core.view.status(run_id).await
    }

    /// [`WorkflowView::outcome`] through the runtime's view.
    pub async fn outcome(&self, run_id: &RunId) -> Result<Option<RunOutcome>> {
        self.inner.core.view.outcome(run_id).await
    }

    /// Wait until the run `run_id` terminates and report its end. The
    /// wait follows the run's current step across its steps, so it
    /// answers for a run of any length, after a restart and from any
    /// runtime over the same queue; a run already terminated is
    /// reported at once from its records. A step dead-lettered outside
    /// the worker is terminated by the worker's dead-step
    /// reconciliation, which the wait polls for at the poll interval.
    ///
    /// Returns [`Error::RunNotFound`] for a run the runtime has no
    /// record of: never submitted, or terminated and swept.
    pub async fn wait(&self, run_id: &RunId) -> Result<RunEnd> {
        self.inner.core.wait(run_id).await
    }

    /// [`Self::wait`] bounded by `timeout`; `Ok(None)` when the timeout
    /// elapses first.
    pub async fn wait_timeout(&self, run_id: &RunId, timeout: Duration) -> Result<Option<RunEnd>> {
        self.inner.core.wait_timeout(run_id, timeout).await
    }

    /// Request cancellation of an active run.
    ///
    /// Returns `Ok(true)` once the request is recorded on the run's
    /// durable record, or `Ok(false)` if the run is unknown or already
    /// terminal, including a run whose current step the queue
    /// dead-lettered outside the worker, which the worker's dead-step
    /// reconciliation terminates as failed. The request reaches a run
    /// after a restart and from any runtime over the same queue.
    ///
    /// The run terminates as [`TerminalStatus::Cancelled`](crate::TerminalStatus::Cancelled) and its
    /// notification job is enqueued for the terminal hook:
    ///
    /// - **Pending / scheduled step**: the queued step job is removed
    ///   and the notification enqueued in one transaction before this
    ///   call returns; the hook runs from a worker afterwards.
    /// - **Running step**: cancellation is delivered to the runner via
    ///   [`Delivery::cancel_token`](crate::Delivery::cancel_token); runners
    ///   that watch the token short-circuit immediately. Runners that ignore
    ///   the token are allowed to run to completion (futures cannot be safely
    ///   aborted mid-step). In both cases the runner's [`StepOutcome`] /
    ///   [`StepError`](crate::StepError) is discarded and the worker settles
    ///   the run once the step returns, with any pending transient retry
    ///   suppressed and the step acked rather than nacked.
    /// - A step claimed after the request is settled as cancelled
    ///   without running.
    ///
    /// Cancellation is best-effort: a run whose terminal step settles while the
    /// request is recorded keeps the outcome it committed. A request applies
    /// only to the run it is recorded on, and a later submission of the same
    /// run id starts without it.
    pub async fn cancel(&self, run_id: &RunId) -> Result<bool> {
        self.inner.core.cancel(run_id).await
    }

    /// The group named `id`.
    pub fn group(&self, id: RunId) -> RunGroup {
        RunGroup::new(self.inner.core.clone(), id)
    }

    /// A group with a generated id.
    pub fn new_group(&self) -> RunGroup {
        RunGroup::new(self.inner.core.clone(), RunId::generate())
    }

    /// Spawn [`Self::run`] as a Tokio task and return a handle for
    /// graceful shutdown. The worker runs until `shutdown` resolves or
    /// [`RunnerHandle::shutdown`] is called; in-flight steps finish
    /// either way.
    pub fn spawn<F>(&self, shutdown: F) -> RunnerHandle
    where
        F: Future<Output = ()> + Send + 'static,
        R: 'static,
        H: 'static,
    {
        let runtime = self.clone();
        WorkerHandle::spawn(shutdown, |stop| async move { runtime.run_with(stop).await })
    }

    /// Drive the step worker loop until `shutdown` resolves. Spawns up
    /// to `max_concurrent_steps` step processors, the dead-step
    /// reconciliation that terminates runs whose step the queue
    /// dead-lettered outside the worker and, when
    /// [`WorkflowRuntimeBuilder::memo_retention`] is set, a
    /// memo-retention sweeper, all running in parallel. All halt cleanly
    /// when `shutdown` resolves or the worker errors.
    pub async fn run<F>(&self, shutdown: F) -> Result<()>
    where
        F: Future<Output = ()>,
        R: 'static,
        H: 'static,
    {
        let stop = CancellationToken::new();
        let mut worker = std::pin::pin!(self.run_with(stop.clone()));
        tokio::select! {
            res = &mut worker => res,
            () = shutdown => {
                stop.cancel();
                worker.await
            }
        }
    }

    /// [`Self::run`] over a token: the worker and the background loops
    /// stop when `stop` is cancelled, and `stop` is cancelled when the
    /// worker returns on its own, so the background loops halt with it.
    async fn run_with(&self, stop: CancellationToken) -> Result<()>
    where
        R: 'static,
        H: 'static,
    {
        let mut background: Vec<_> = self
            .inner
            .core
            .sweeps()
            .map(|sweep| {
                let sweep = sweep.clone();
                let core = self.inner.core.clone();
                let token = stop.clone();
                tokio::spawn(async move {
                    sweep
                        .run(&core.queue, &*core.clock, core.poll_interval, token)
                        .await;
                })
            })
            .collect();
        background.push({
            let core = self.inner.core.clone();
            let token = stop.clone();
            tokio::spawn(async move { core.run_dead_step_reconciliation(token).await })
        });

        let worker = Arc::new(StepWorker {
            inner: self.inner.clone(),
        });
        let result = taquba::run_worker_concurrent(
            &self.inner.core.queue,
            &self.inner.core.queue_name,
            worker,
            self.inner.core.max_concurrent_steps,
            self.inner.core.poll_interval,
            stop.clone().cancelled_owned(),
        )
        .await;
        stop.cancel();

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

        result?;
        Ok(())
    }
}

/// A handle to a worker task spawned by [`WorkflowRuntime::spawn`].
///
/// Dropping a `RunnerHandle` does not stop the worker: the task
/// continues until the `shutdown` future passed to `spawn` resolves.
/// Call [`shutdown`](WorkerHandle::shutdown) or
/// [`wait`](WorkerHandle::wait) to stop or join the worker explicitly.
pub type RunnerHandle = WorkerHandle<Result<()>>;

impl RuntimeCore {
    /// [`WorkflowRuntime::submit`].
    #[instrument(skip(self, spec), fields(run_id))]
    pub(crate) async fn submit(&self, spec: RunSpec) -> Result<SubmitOutcome> {
        let run_id = Self::validate_spec(&spec)?;
        tracing::Span::current().record("run_id", run_id.as_str());
        self.enqueue_run(&run_id, spec, None).await
    }

    /// [`Self::submit`] of a member of a group: the run's step jobs
    /// hold the membership and its member record is written with the
    /// enqueue.
    pub(crate) async fn submit_member(
        &self,
        membership: &Membership,
        spec: RunSpec,
    ) -> Result<SubmitOutcome> {
        let run_id = Self::validate_spec(&spec)?;
        self.enqueue_run(&run_id, spec, Some(membership)).await
    }

    /// Check `spec`'s headers and KV keys, and return the run id,
    /// generated when the spec names none.
    fn validate_spec(spec: &RunSpec) -> Result<RunId> {
        for k in spec.options.headers.keys() {
            if k.starts_with(RESERVED_HEADER_PREFIX) {
                return Err(Error::ReservedHeaderInSubmit(k.clone()));
            }
        }
        let deletes = spec.effects.kv_deletes.iter();
        for key in spec.effects.kv_writes.keys().chain(deletes) {
            if key.starts_with(RESERVED_KV_PREFIX.as_bytes()) {
                return Err(Error::ReservedKvKey(
                    String::from_utf8_lossy(key).into_owned(),
                ));
            }
        }
        Ok(spec.run_id.clone().unwrap_or_else(RunId::generate))
    }

    /// Enqueue step 0 of `run_id` unless the run is active. The run
    /// record is written with the step-0 enqueue and deleted with the
    /// termination, so it identifies an active run whichever process
    /// submitted it, and the step's dedup key serialises two
    /// submissions that both find no record. A `membership` is set on
    /// the step job and its pending member record is written with the
    /// enqueue.
    async fn enqueue_run(
        &self,
        run_id: &RunId,
        spec: RunSpec,
        membership: Option<&Membership>,
    ) -> Result<SubmitOutcome> {
        let input_hash = hash_input(&spec.input);
        let duplicate = |job_id: String| SubmitOutcome {
            run_id: run_id.clone(),
            newly_submitted: false,
            job_id,
        };
        let check_input = |existing: DurableRunRecord| {
            if existing.input_hash == input_hash {
                Ok(())
            } else {
                Err(Error::InputMismatch(run_id.clone()))
            }
        };

        if let Some(existing) = self.view.run_record(run_id).await? {
            check_input(existing)?;
            let current = self.current_step(run_id).await?;
            return Ok(duplicate(current.job_id));
        }

        let opts = StepEnqueueOpts {
            run_at: spec.options.run_at,
            priority: spec.options.priority,
            max_attempts: spec.options.max_attempts_per_step,
            reserved_headers: membership
                .map(Membership::reserved_headers)
                .unwrap_or_default(),
        };
        let (request, job_id) =
            self.step_enqueue_request(run_id, 0, spec.input, &spec.options.headers, opts);

        let record_bytes = durable::encode(&DurableRunRecord {
            run_id: run_id.clone(),
            submitted_at_ms: self.clock.now_ms(),
            input_hash,
            cancel_requested: false,
        });
        let mut effects = spec
            .effects
            .kv_put(run_kv_key(run_id), record_bytes)
            .kv_put(step_kv_key(run_id), current_step_bytes(0, &job_id));
        if let Some(membership) = membership {
            effects = effects.kv_put(
                membership.kv_key(),
                durable::encode(&pending_member(run_id)),
            );
        }

        let job_id = match self
            .queue
            .enqueue_with_effects(&request.queue, request.payload, request.options, effects)
            .await?
            .0
        {
            EnqueueResult::New(id) => id,
            // A concurrent submission committed first; its record holds
            // the input to check against. A dedup hit without a record
            // is a store this runtime did not write, reported as a
            // duplicate.
            EnqueueResult::AlreadyEnqueued(existing) => {
                if let Some(record) = self.view.run_record(run_id).await? {
                    check_input(record)?;
                }
                return Ok(duplicate(existing));
            }
        };

        debug!(run_id = %run_id, job_id = %job_id, "run submitted");
        Ok(SubmitOutcome {
            run_id: run_id.clone(),
            newly_submitted: true,
            job_id,
        })
    }

    /// [`WorkflowRuntime::wait`].
    pub(crate) async fn wait(&self, run_id: &RunId) -> Result<RunEnd> {
        self.wait_run(run_id)
            .await?
            .ok_or_else(|| Error::RunNotFound(run_id.clone()))
    }

    /// [`WorkflowRuntime::wait_timeout`].
    pub(crate) async fn wait_timeout(
        &self,
        run_id: &RunId,
        timeout: Duration,
    ) -> Result<Option<RunEnd>> {
        match tokio::time::timeout(timeout, self.wait(run_id)).await {
            Ok(end) => end.map(Some),
            Err(_) => Ok(None),
        }
    }

    /// [`WorkflowRuntime::cancel`].
    pub(crate) async fn cancel(&self, run_id: &RunId) -> Result<bool> {
        let Some(input_hash) = self.request_cancel(run_id).await? else {
            return Ok(false);
        };
        // Settle the current step now: remove it while it is queued, or
        // fire the claim's cancellation token, the parent of
        // `Delivery::cancel_token`, while it runs. A step that settles in
        // between is followed to its successor; a step claimed after the
        // request terminates the run on its own.
        loop {
            let Some((_, job)) = self.view.current_job(run_id).await? else {
                // Terminated on its own after the request was recorded.
                return Ok(false);
            };
            if job.status == JobStatus::Dead {
                // Dead-lettered by the queue outside the worker;
                // reconciliation terminates the run and the request is
                // not honoured.
                return Ok(false);
            }
            let claimed = ClaimedStep::parse(&job)?;
            // `error` is `None`: external cancellation supplies no reason
            // at the API level. The effects are built before the outcome
            // is known; the queue applies them only on `Removed`.
            let outcome = claimed.cancelled(None);
            let termination = self.termination(&outcome, None, input_hash);
            let effects = self.terminate_collecting_effects(&outcome, &claimed, termination);
            match self.queue.cancel_with(&job.id, effects).await?.0 {
                taquba::CancelOutcome::Removed | taquba::CancelOutcome::Requested => {
                    return Ok(true);
                }
                taquba::CancelOutcome::NotFound => continue,
            }
        }
    }

    /// Settle a run into its terminal state: return the deletes of the
    /// durable run record and the current-step pointer, the writes of
    /// the terminal record and the terminal marker (when memo retention
    /// is enabled), the write of the member record (when the run is a
    /// group member) and the terminal-notification enqueue
    /// (when the hook observes this outcome) as [`SettlementEffects`]
    /// for the settlement transaction. The notification job's payload
    /// is the committed outcome and the configured [`TerminalHook`]
    /// runs as its worker; `terminal_step` is the step that produced
    /// the outcome, or the pending step a cancellation removes, and
    /// `termination` the record of the outcome the settlement writes.
    ///
    /// The effects are pure: nothing is written and no state is
    /// mutated here, so a caller that builds them and then commits a
    /// non-terminal outcome leaves no trace. A settlement that fails
    /// redelivers the step, which re-terminates and rebuilds the same
    /// effects.
    pub(crate) fn terminate_collecting_effects(
        &self,
        outcome: &RunOutcome,
        terminal_step: &ClaimedStep<'_>,
        termination: DurableTermination,
    ) -> SettlementEffects {
        let terminated_at_ms = termination.terminated_at_ms;
        let kv_deletes = vec![run_kv_key(&outcome.run_id), step_kv_key(&outcome.run_id)];
        let mut kv_writes = HashMap::new();
        kv_writes.insert(
            outcome_kv_key(&outcome.run_id),
            durable::encode(&termination),
        );
        if let Some(membership) = &terminal_step.membership {
            kv_writes.insert(
                membership.kv_key(),
                durable::encode(&terminated_member(&outcome.run_id, termination)),
            );
        }
        let enqueues = if (self.observes)(outcome) {
            vec![self.notification_enqueue_request(outcome, Some(terminal_step.job))]
        } else {
            Vec::new()
        };
        let effects = SettlementEffects::default()
            .enqueues(enqueues)
            .kv_writes(kv_writes)
            .kv_deletes(kv_deletes);
        match &self.memo_sweep {
            Some(sweep) => sweep.mark(effects, &outcome.run_id, terminated_at_ms),
            None => effects,
        }
    }

    /// Terminate every run whose step job the queue dead-lettered
    /// outside the worker path: a lease that expired past the attempt
    /// limit, or a claim dead-lettered by crash recovery when the queue
    /// was opened. Such a settlement runs no workflow code, so the run
    /// record and the current-step pointer survive it and no
    /// notification is enqueued. A dead step job that the run's
    /// current-step pointer names identifies the case exactly, because
    /// every worker-path dead-letter deletes the pointer in its own
    /// transaction and a re-submission of the run id names a new job.
    /// The run terminates as [`TerminalStatus::Failed`] with the queue record's
    /// last error, through the same effects as a worker-path
    /// termination committed as one transaction with no transition of
    /// their own; the runner's failure writes cannot apply, since no
    /// runner returned. Returns the number of runs terminated.
    pub(crate) async fn reconcile_dead_steps(&self) -> Result<usize> {
        const PAGE: usize = 256;
        let mut terminated = 0usize;
        let mut dead = std::pin::pin!(self.queue.view().jobs(
            &self.queue_name,
            JobStatus::Dead,
            PAGE
        ));
        while let Some(job) = dead.try_next().await? {
            if job.headers.contains_key(HEADER_TERMINAL) {
                continue;
            }
            let Ok(claimed) = ClaimedStep::parse(&job) else {
                continue;
            };
            let run_id = &claimed.run_id;
            let current = self.view.current_step_if_active(run_id).await?;
            if current.is_none_or(|current| current.job_id != job.id) {
                continue;
            }
            // The record is written and deleted with the pointer; a
            // pointer without one is a store the runtime did not write,
            // left for the worker to report.
            let Some(record) = self.view.run_record(run_id).await? else {
                warn!(run_id = %run_id, job_id = %job.id, "dead step has a current-step pointer but no run record");
                continue;
            };
            let error = job
                .last_error
                .clone()
                .unwrap_or_else(|| "step dead-lettered outside the worker".to_string());
            let outcome = claimed.failed(error);
            let termination = self.termination(&outcome, None, record.input_hash);
            let effects = self.terminate_collecting_effects(&outcome, &claimed, termination);
            self.queue.commit_effects(effects).await?;
            warn!(run_id = %run_id, step_number = claimed.step_number, job_id = %job.id, "terminated a run whose step was dead-lettered outside the worker");
            terminated += 1;
        }
        Ok(terminated)
    }

    /// The reconciliation loop: a pass when the worker starts, then a
    /// pass whenever the queue's dead count has changed since the last
    /// successful pass, checked every poll interval, until `stop` is
    /// cancelled.
    async fn run_dead_step_reconciliation(&self, stop: CancellationToken) {
        run_periodically(
            self.poll_interval,
            &stop,
            None,
            |reconciled_at: Option<i64>| async move {
                match self.queue.view().stats(&self.queue_name).await {
                    Ok(stats) if reconciled_at != Some(stats.dead) => {
                        match self.reconcile_dead_steps().await {
                            Ok(_) => Some(stats.dead),
                            Err(err) => {
                                warn!("dead-step reconciliation failed: {err}");
                                reconciled_at
                            }
                        }
                    }
                    Ok(_) => reconciled_at,
                    Err(err) => {
                        warn!("dead-step reconciliation could not read queue stats: {err}");
                        reconciled_at
                    }
                }
            },
        )
        .await;
    }

    /// The retention sweeps [`WorkflowRuntime::run`] runs.
    fn sweeps(&self) -> impl Iterator<Item = &Arc<Sweep>> {
        self.memo_sweep.iter().chain(self.group_sweep.iter())
    }

    /// One pass of every retention sweep. Returns the number of markers
    /// removed.
    #[cfg(test)]
    pub(crate) async fn sweep_once(&self) -> Result<usize> {
        let mut removed = 0;
        for sweep in self.sweeps() {
            removed += sweep.pass(&self.queue, &*self.clock).await?;
        }
        Ok(removed)
    }

    /// The current-step pointer of a run whose durable record exists.
    pub(crate) async fn current_step(&self, run_id: &RunId) -> Result<DurableCurrentStep> {
        self.view
            .current_step_if_active(run_id)
            .await?
            .ok_or_else(|| Error::InconsistentRunState(run_id.clone()))
    }

    /// Wait until `run_id` terminates, following its current step
    /// across steps, and report its end; `None` for a run with no
    /// current step and no record. A current step the queue
    /// dead-lettered outside the worker is polled at the poll interval
    /// until reconciliation terminates the run.
    pub(crate) async fn wait_run(&self, run_id: &RunId) -> Result<Option<RunEnd>> {
        loop {
            let Some((current, _)) = self.view.current_job(run_id).await? else {
                return self.run_end(run_id).await;
            };
            match self.queue.wait_for_completion(&current.job_id).await? {
                // `NotFound`: the job settled between the two reads; the
                // next read follows the pointer.
                WaitOutcome::Done(_) | WaitOutcome::Cancelled | WaitOutcome::NotFound => {}
                WaitOutcome::Dead(_) => {
                    // A worker-path dead-letter deletes the pointer with
                    // its settlement; a pointer that still names the dead
                    // job identifies a dead-letter outside the worker,
                    // which reconciliation terminates.
                    let unreconciled = self
                        .view
                        .current_step_if_active(run_id)
                        .await?
                        .is_some_and(|step| step.job_id == current.job_id);
                    if unreconciled {
                        tokio::time::sleep(self.poll_interval).await;
                    }
                }
            }
        }
    }

    /// The end of the terminated run `run_id` from its terminal record
    /// and the run result record of that termination; `None` when no
    /// terminal record remains.
    async fn run_end(&self, run_id: &RunId) -> Result<Option<RunEnd>> {
        let Some(termination) = self.view.terminal_record(run_id).await? else {
            return Ok(None);
        };
        let termination = RunTermination::from(termination);
        let outcome = self
            .view
            .run_result_of(run_id, &termination)
            .await?
            .map(|result| result.outcome);
        Ok(Some(RunEnd {
            termination,
            outcome,
        }))
    }

    /// The termination of `outcome`'s run at the clock's current time;
    /// `input_hash` is the run record's.
    pub(crate) fn termination(
        &self,
        outcome: &RunOutcome,
        error_kind: Option<StepErrorKind>,
        input_hash: [u8; 32],
    ) -> DurableTermination {
        DurableTermination {
            status: outcome.status.into(),
            error: outcome.error.clone(),
            error_kind: error_kind.map(DurableErrorKind::from),
            final_step: outcome.final_step,
            terminated_at_ms: self.clock.now_ms(),
            input_hash,
        }
    }

    /// Write the run result record of `outcome`'s run, belonging to
    /// `termination`.
    pub(crate) async fn store_run_result(
        &self,
        outcome: &RunOutcome,
        termination: &DurableTermination,
    ) -> Result<()> {
        let record = DurableRunResult {
            termination: termination.clone(),
            outcome: DurableRunOutcome::from(outcome),
        };
        self.memo_store
            .new_run_memo(&outcome.run_id)
            .put(RUN_RESULT_MEMO_KEY, &durable::encode(&record))
            .await
    }

    /// Record a cancellation request on the run record of `run_id`.
    /// Returns the record's input hash when the run is active, `None`
    /// otherwise; a request already recorded counts as recorded again.
    async fn request_cancel(&self, run_id: &RunId) -> Result<Option<[u8; 32]>> {
        let key = run_kv_key(run_id);
        loop {
            let Some(current) = self.queue.view().kv_get(&key).await? else {
                return Ok(None);
            };
            let mut record: DurableRunRecord = durable::decode(&current)?;
            if record.cancel_requested {
                return Ok(Some(record.input_hash));
            }
            record.cancel_requested = true;
            if self
                .queue
                .kv_compare_put(&key, Some(&current), &durable::encode(&record))
                .await?
            {
                return Ok(Some(record.input_hash));
            }
        }
    }

    /// Build the enqueue request for one step of a run, with a
    /// pre-assigned job id so the current-step pointer written with
    /// the enqueue can name it. Returns the request and the assigned
    /// id.
    fn step_enqueue_request(
        &self,
        run_id: &RunId,
        step_number: u32,
        payload: Vec<u8>,
        user_headers: &HashMap<String, String>,
        opts: StepEnqueueOpts,
    ) -> (EnqueueRequest, String) {
        let job_id = self.queue.next_job_id();
        let mut headers = user_headers.clone();
        headers.insert(HEADER_RUN_ID.to_string(), run_id.to_string());
        headers.insert(HEADER_STEP.to_string(), step_number.to_string());
        for (key, value) in &opts.reserved_headers {
            headers.insert((*key).to_string(), value.clone());
        }

        let request = EnqueueRequest {
            queue: self.queue_name.clone(),
            payload,
            options: EnqueueOptions::default()
                .headers(headers)
                .run_at(opts.run_at)
                .priority(opts.priority)
                .max_attempts(opts.max_attempts)
                .dedup_key(Some(format!("{DEDUP_PREFIX}{run_id}:{step_number}")))
                .id_override(Some(job_id.clone())),
        };
        (request, job_id)
    }

    /// Build the enqueue request for a run's terminal-notification job.
    /// The priority and attempt limit are inherited from `terminal_step`,
    /// the job of the step that produced the outcome, when one exists.
    fn notification_enqueue_request(
        &self,
        outcome: &RunOutcome,
        terminal_step: Option<&JobRecord>,
    ) -> EnqueueRequest {
        let payload = durable::encode(&DurableRunOutcome::from(outcome));
        let mut headers = HashMap::new();
        headers.insert(HEADER_RUN_ID.to_string(), outcome.run_id.to_string());
        headers.insert(HEADER_TERMINAL.to_string(), "1".to_string());
        EnqueueRequest {
            queue: self.queue_name.clone(),
            payload,
            options: EnqueueOptions::default()
                .headers(headers)
                .priority(terminal_step.map(|job| job.priority))
                .max_attempts(terminal_step.map(|job| job.max_attempts))
                .dedup_key(Some(format!("{DEDUP_PREFIX}{}:terminal", outcome.run_id))),
        }
    }

    /// The instant `delay` after the runtime's clock now, as a
    /// [`SystemTime`] for an enqueue's `run_at`.
    pub(crate) fn run_at_after(&self, delay: Duration) -> SystemTime {
        UNIX_EPOCH + Duration::from_millis(self.clock.now_ms()) + delay
    }

    pub(crate) async fn load_step_output(
        &self,
        run_id: &RunId,
        step_number: u32,
        step_payload: &[u8],
    ) -> Result<Option<(StepOutcome, StagedEffects)>> {
        let Some(bytes) = self
            .memo_store
            .get_step_output(run_id, step_number, step_payload)
            .await?
        else {
            return Ok(None);
        };
        let Some(record) = durable::decode_or_absent::<DurableStepOutcomeRecord>(
            &bytes,
            "step-output replay record",
            &format_args!("{run_id}/{step_number}"),
        ) else {
            return Ok(None);
        };
        let mut outcome = StepOutcome::from(record.outcome);
        match &mut outcome {
            StepOutcome::Continue {
                when: Trigger::After(delay),
                ..
            } => {
                *delay = remaining_delay(record.stored_at_ms, self.clock.now_ms(), *delay);
            }
            StepOutcome::Continue {
                when: Trigger::OnSignal { timeout, .. },
                ..
            } => {
                *timeout = remaining_delay(record.stored_at_ms, self.clock.now_ms(), *timeout);
            }
            _ => {}
        }
        Ok(Some((outcome, record.effects)))
    }

    pub(crate) async fn store_step_output(
        &self,
        run_id: &RunId,
        step_number: u32,
        step_payload: &[u8],
        outcome: &StepOutcome,
        effects: &StagedEffects,
    ) -> Result<()> {
        let record = DurableStepOutcomeRecord {
            stored_at_ms: self.clock.now_ms(),
            outcome: DurableStepOutcome::from(outcome),
            effects: effects.clone(),
        };
        let bytes = rmp_serde::to_vec_named(&record)?;
        self.memo_store
            .put_step_output(run_id, step_number, step_payload, &bytes)
            .await
    }

    /// Build the effects that advance the run of `claimed` to its next
    /// step: the next step's enqueue joins the current step's
    /// acknowledgement transaction, so the transition is atomic.
    pub(crate) async fn advance(
        &self,
        claimed: &ClaimedStep<'_>,
        payload: Vec<u8>,
        opts: StepEnqueueOpts,
    ) -> SettlementEffects {
        self.advance_with_kv(claimed, payload, opts, |_| HashMap::new())
            .await
    }

    /// [`Self::advance`] with caller KV writes joined to the same
    /// acknowledgement transaction. `kv_writes` receives the next step's
    /// pre-assigned job id so the writes can reference it.
    pub(crate) async fn advance_with_kv(
        &self,
        claimed: &ClaimedStep<'_>,
        payload: Vec<u8>,
        opts: StepEnqueueOpts,
        kv_writes: impl FnOnce(&str) -> HashMap<Vec<u8>, Vec<u8>>,
    ) -> SettlementEffects {
        let run_id = &claimed.run_id;
        let next_step = claimed.step_number + 1;
        let (request, next_job_id) =
            self.step_enqueue_request(run_id, next_step, payload, &claimed.headers, opts);
        let mut kv_writes = kv_writes(&next_job_id);
        kv_writes.insert(
            step_kv_key(run_id),
            current_step_bytes(next_step, &next_job_id),
        );
        SettlementEffects::default()
            .enqueues(vec![request])
            .kv_writes(kv_writes)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::durable::DurableMember;
    use crate::effects::{EffectsHandle, TerminalEffects};
    use crate::group::GroupMember;
    use crate::keys::group_member_kv_key;
    use crate::keys::{TERMINAL_KV_PREFIX, signal_buf_kv_key, signal_wait_kv_key};
    use crate::runner::{Step, StepError};
    use crate::signal::SignalOutcome;
    use crate::terminal::NoopTerminalHook;
    use crate::terminal::TerminalStatus;
    use crate::test_util::{
        advance, fast_options, open_queue, open_queue_at, open_queue_at_with, open_queue_with, rid,
    };
    use crate::view::WorkflowView;
    use std::sync::Mutex as StdMutex;
    use std::sync::atomic::{AtomicU32, Ordering};
    use taquba::object_store::ObjectStoreExt;
    use taquba::object_store::memory::InMemory;
    use taquba::{Expired, ExpiryIndex};
    use taquba::{LeaseHandle, MockClock, OpenOptions, QueueConfig, QueueReader};
    use tokio::sync::oneshot;

    /// Recording terminal hook backed by an mpsc channel.
    struct ChannelHook {
        tx: tokio::sync::mpsc::UnboundedSender<RunOutcome>,
    }

    impl TerminalHook for ChannelHook {
        async fn on_termination(
            &self,
            outcome: &RunOutcome,
            _effects: &TerminalEffects,
        ) -> std::result::Result<(), StepError> {
            let _ = self.tx.send(outcome.clone());
            Ok(())
        }
    }

    /// Runner that executes a fixed list of step outcomes in order.
    struct ScriptedRunner {
        script: Arc<StdMutex<Vec<StepOutcome>>>,
    }

    impl ScriptedRunner {
        fn new(steps: Vec<StepOutcome>) -> Self {
            Self {
                script: Arc::new(StdMutex::new(steps)),
            }
        }
    }

    impl StepRunner for ScriptedRunner {
        async fn run_step(&self, _step: &Step) -> std::result::Result<StepOutcome, StepError> {
            let next = self.script.lock().unwrap().remove(0);
            Ok(next)
        }
    }

    /// Runner that returns a clone of `result` on every step and counts
    /// its calls.
    struct FixedRunner {
        result: std::result::Result<StepOutcome, StepError>,
        calls: Arc<AtomicU32>,
    }

    impl FixedRunner {
        fn new(result: std::result::Result<StepOutcome, StepError>) -> Self {
            Self {
                result,
                calls: Arc::new(AtomicU32::new(0)),
            }
        }
    }

    impl StepRunner for FixedRunner {
        async fn run_step(&self, _step: &Step) -> std::result::Result<StepOutcome, StepError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            self.result.clone()
        }
    }

    /// Runner whose step never returns.
    struct PauseRunner;

    impl StepRunner for PauseRunner {
        async fn run_step(&self, _step: &Step) -> std::result::Result<StepOutcome, StepError> {
            std::future::pending().await
        }
    }

    /// Runner whose step must not be claimed.
    struct UnreachableRunner;

    impl StepRunner for UnreachableRunner {
        async fn run_step(&self, _step: &Step) -> std::result::Result<StepOutcome, StepError> {
            unreachable!("worker must not claim the step");
        }
    }

    /// Runner that reports the claim of its step, holds the step until
    /// its [`Gate`] is released and then returns a clone of `result`.
    struct GatedRunner {
        claimed: Arc<tokio::sync::Notify>,
        release: tokio::sync::Mutex<Option<oneshot::Receiver<()>>>,
        result: std::result::Result<StepOutcome, StepError>,
        calls: Arc<AtomicU32>,
    }

    /// The test's side of a [`GatedRunner`].
    struct Gate {
        claimed: Arc<tokio::sync::Notify>,
        release: StdMutex<Option<oneshot::Sender<()>>>,
        calls: Arc<AtomicU32>,
    }

    impl GatedRunner {
        fn new(result: std::result::Result<StepOutcome, StepError>) -> (Self, Gate) {
            let claimed = Arc::new(tokio::sync::Notify::new());
            let calls = Arc::new(AtomicU32::new(0));
            let (release_tx, release_rx) = oneshot::channel();
            let runner = Self {
                claimed: claimed.clone(),
                release: tokio::sync::Mutex::new(Some(release_rx)),
                result,
                calls: calls.clone(),
            };
            let gate = Gate {
                claimed,
                release: StdMutex::new(Some(release_tx)),
                calls,
            };
            (runner, gate)
        }
    }

    impl StepRunner for GatedRunner {
        async fn run_step(&self, _step: &Step) -> std::result::Result<StepOutcome, StepError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            self.claimed.notify_one();
            let rx = self
                .release
                .lock()
                .await
                .take()
                .expect("gate consumed twice");
            let _ = rx.await;
            self.result.clone()
        }
    }

    impl Gate {
        /// Wait until the runner holds the step.
        async fn claimed(&self) {
            tokio::time::timeout(Duration::from_secs(2), self.claimed.notified())
                .await
                .expect("runner reached gate");
        }

        fn release(&self) {
            if let Some(tx) = self.release.lock().unwrap().take() {
                let _ = tx.send(());
            }
        }
    }

    /// Every terminal marker in the queue's KV namespace, as
    /// `(run_id, terminal_at_ms)` pairs in key order (oldest first).
    async fn terminal_markers(queue: &Queue) -> Vec<(RunId, u64)> {
        let page = queue
            .view()
            .kv_scan(TERMINAL_KV_PREFIX, .., 1_000)
            .await
            .unwrap();
        let index = ExpiryIndex::new(TERMINAL_KV_PREFIX);
        page.entries
            .iter()
            .map(|(key, _)| {
                let (at_ms, suffix) = index.parse(key).expect("well-formed marker key");
                let id = std::str::from_utf8(suffix).expect("run id");
                (RunId::new(id).expect("run id"), at_ms)
            })
            .collect()
    }

    /// The terminal status `status` reports for `run_id`; `None` while
    /// the run is active or unknown.
    async fn terminal_status_of<R: StepRunner, H: TerminalHook>(
        runtime: &WorkflowRuntime<R, H>,
        run_id: &RunId,
    ) -> Option<TerminalStatus> {
        match runtime.status(run_id).await.unwrap().map(|s| s.state) {
            Some(RunState::Terminated(termination)) => Some(termination.status),
            _ => None,
        }
    }

    fn spawn_runtime<R, H>(runtime: WorkflowRuntime<R, H>) -> oneshot::Sender<()>
    where
        R: StepRunner + 'static,
        H: TerminalHook + 'static,
    {
        let (tx, rx) = oneshot::channel::<()>();
        tokio::spawn(async move {
            let _ = runtime
                .run(async move {
                    let _ = rx.await;
                })
                .await;
        });
        tx
    }

    /// A runner that extends its lease through the step and reports the
    /// resulting expiry.
    struct RenewingRunner {
        queue: Arc<Queue>,
        tx: tokio::sync::mpsc::UnboundedSender<u64>,
    }

    impl StepRunner for RenewingRunner {
        async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
            step.lease
                .ensure_at_least(Duration::from_secs(600))
                .map_err(|e| StepError::transient(e.to_string()))?;
            let expiry = self
                .queue
                .lease_expiry("workflow-steps", &step.job_id)
                .expect("a running step holds a lease");
            let _ = self.tx.send(expiry);
            Ok(StepOutcome::Succeed { result: Vec::new() })
        }
    }

    #[tokio::test(start_paused = true)]
    async fn a_step_runner_extends_its_lease_through_the_step() {
        let base = 1_700_000_000_000;
        let (queue, store, _clock) = open_queue_at(base).await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let (hook_tx, mut hook_rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            RenewingRunner { queue, tx },
            ChannelHook { tx: hook_tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        let expiry = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert!(
            expiry >= base + 600_000,
            "the extension must reach the lease registry",
        );
        let outcome = tokio::time::timeout(Duration::from_secs(2), hook_rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.status, TerminalStatus::Succeeded);

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn single_step_succeeds_and_writes_no_marker_without_retention() {
        let (queue, store) = open_queue().await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue,
            store.clone(),
            ScriptedRunner::new(vec![StepOutcome::Succeed {
                result: b"done".to_vec(),
            }]),
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: b"in".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        assert_eq!(outcome.run_id, handle.run_id);
        assert_eq!(outcome.status, TerminalStatus::Succeeded);
        assert_eq!(outcome.result.as_deref(), Some(b"done".as_slice()));
        assert_eq!(outcome.final_step, 0);
        assert_eq!(
            terminal_status_of(&runtime, &handle.run_id).await,
            Some(TerminalStatus::Succeeded),
            "the terminal record is written without retention",
        );
        assert!(terminal_markers(&runtime.inner.core.queue).await.is_empty());

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn multi_step_run_advances_through_continue_with_its_headers() {
        let (queue, store) = open_queue().await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue,
            store.clone(),
            ScriptedRunner::new(vec![
                StepOutcome::continue_now(b"step1".to_vec()),
                StepOutcome::continue_now(b"step2".to_vec()),
                StepOutcome::Succeed {
                    result: b"final".to_vec(),
                },
            ]),
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: b"start".to_vec(),
                options: RunOptions {
                    headers: HashMap::from([
                        ("trace_id".to_string(), "abc-123".to_string()),
                        ("tenant".to_string(), "acme".to_string()),
                    ]),
                    ..Default::default()
                },
                ..Default::default()
            })
            .await
            .unwrap();
        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        assert_eq!(outcome.run_id, handle.run_id);
        assert_eq!(outcome.final_step, 2);
        assert_eq!(outcome.status, TerminalStatus::Succeeded);
        assert_eq!(outcome.result.as_deref(), Some(b"final".as_slice()));
        assert_eq!(outcome.headers.get("trace_id").unwrap(), "abc-123");
        assert_eq!(outcome.headers.get("tenant").unwrap(), "acme");
        assert!(!outcome.headers.contains_key(HEADER_RUN_ID));
        assert!(!outcome.headers.contains_key(HEADER_STEP));

        let recorded =
            runtime.outcome(&handle.run_id).await.unwrap().expect(
                "the worker writes the run result record before the terminating settlement",
            );
        assert_eq!(recorded.status, TerminalStatus::Succeeded);
        assert_eq!(recorded.final_step, 2);
        assert_eq!(recorded.result.as_deref(), Some(b"final".as_slice()));
        assert_eq!(recorded.headers, outcome.headers);

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn continue_after_delays_next_step_until_promotion() {
        let initial = 1_700_000_000_000u64;
        let (queue, store, clock) = open_queue_at(initial).await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            ScriptedRunner::new(vec![
                StepOutcome::continue_after(b"step1".to_vec(), Duration::from_secs(60)),
                StepOutcome::Succeed {
                    result: b"final".to_vec(),
                },
            ]),
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: b"start".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();

        // The delayed step is held in `scheduled` and the run must not
        // terminate while the delay is pending.
        assert!(
            tokio::time::timeout(Duration::from_millis(500), rx.recv())
                .await
                .is_err()
        );
        let stats = queue.view().stats("workflow-steps").await.unwrap();
        assert_eq!(stats.scheduled, 1);

        advance(&clock, Duration::from_secs(61)).await;
        queue.promote_scheduled_now().await.unwrap();

        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.run_id, handle.run_id);
        assert_eq!(outcome.final_step, 1);
        assert_eq!(outcome.status, TerminalStatus::Succeeded);

        let _ = shutdown.send(());
    }

    type ObservedSignals = Arc<StdMutex<Vec<Option<Vec<u8>>>>>;

    /// Two-step runner: step 0 continues with a signal wait on
    /// `correlation_key`, step 1 records [`Step::signal`] and succeeds.
    struct SignalProbe {
        correlation_key: String,
        timeout: Duration,
        observed: ObservedSignals,
    }

    impl StepRunner for SignalProbe {
        async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
            if step.step_number == 0 {
                Ok(StepOutcome::continue_on_signal(
                    Vec::new(),
                    self.correlation_key.clone(),
                    self.timeout,
                ))
            } else {
                self.observed.lock().unwrap().push(step.signal.clone());
                Ok(StepOutcome::Succeed { result: Vec::new() })
            }
        }
    }

    async fn wait_for_scheduled(queue: &Queue, count: i64) {
        for _ in 0..200 {
            if queue
                .view()
                .stats("workflow-steps")
                .await
                .unwrap()
                .scheduled
                == count
            {
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        panic!("scheduled count never reached {count}");
    }

    fn signal_probe_runtime(
        queue: Arc<Queue>,
        store: Arc<dyn taquba::object_store::ObjectStore>,
        correlation_key: &str,
        timeout: Duration,
    ) -> (
        WorkflowRuntime<SignalProbe, ChannelHook>,
        ObservedSignals,
        tokio::sync::mpsc::UnboundedReceiver<RunOutcome>,
    ) {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        let observed = Arc::new(StdMutex::new(Vec::new()));
        let runtime = WorkflowRuntime::builder(
            queue,
            store,
            SignalProbe {
                correlation_key: correlation_key.to_string(),
                timeout,
                observed: observed.clone(),
            },
            ChannelHook { tx },
        )
        .build();
        (runtime, observed, rx)
    }

    #[tokio::test(start_paused = true)]
    async fn signal_wakes_waiting_run_early_with_payload() {
        let (queue, store, _clock) = open_queue_at(1_700_000_000_000).await;
        let (runtime, observed, mut rx) =
            signal_probe_runtime(queue.clone(), store, "order-1", Duration::from_secs(3600));
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        wait_for_scheduled(&queue, 1).await;

        let outcome = runtime.signal("order-1", b"paid".to_vec()).await.unwrap();
        assert_eq!(outcome, SignalOutcome::Delivered);

        // The run completes without the timeout elapsing on the mock clock.
        let terminal = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(terminal.status, TerminalStatus::Succeeded);
        assert_eq!(
            observed.lock().unwrap().as_slice(),
            &[Some(b"paid".to_vec())]
        );

        // Both durable signal entries are consumed.
        assert!(
            queue
                .view()
                .kv_get(&signal_wait_kv_key("order-1"))
                .await
                .unwrap()
                .is_none()
        );
        assert!(
            queue
                .view()
                .kv_get(&signal_buf_kv_key("order-1"))
                .await
                .unwrap()
                .is_none()
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_run_waiting_on_a_signal_survives_a_close_and_reopen() {
        let store: Arc<dyn taquba::object_store::ObjectStore> = Arc::new(InMemory::new());
        let open = |store: Arc<dyn taquba::object_store::ObjectStore>| async move {
            Arc::new(
                Queue::open_with_options(
                    store,
                    "test",
                    OpenOptions::default().clock(Arc::new(MockClock::new(1_700_000_000_000))),
                )
                .await
                .unwrap(),
            )
        };

        let queue = open(store.clone()).await;
        let (runtime, _observed, _rx) = signal_probe_runtime(
            queue.clone(),
            store.clone(),
            "approval",
            Duration::from_secs(3600),
        );
        let (stop_tx, stop_rx) = oneshot::channel::<()>();
        let worker = tokio::spawn({
            let runtime = runtime.clone();
            async move {
                runtime
                    .run(async move {
                        let _ = stop_rx.await;
                    })
                    .await
            }
        });
        runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        wait_for_scheduled(&queue, 1).await;
        let _ = stop_tx.send(());
        worker.await.unwrap().unwrap();
        drop(runtime);
        Arc::into_inner(queue)
            .expect("no other queue references at close")
            .close()
            .await
            .unwrap();

        let queue = open(store.clone()).await;
        assert_eq!(
            queue
                .view()
                .stats("workflow-steps")
                .await
                .unwrap()
                .scheduled,
            1
        );

        let (runtime, observed, mut rx) =
            signal_probe_runtime(queue.clone(), store, "approval", Duration::from_secs(3600));
        let shutdown = spawn_runtime(runtime.clone());

        let delivery = runtime
            .signal("approval", b"approved".to_vec())
            .await
            .unwrap();
        assert_eq!(delivery, SignalOutcome::Delivered);

        let terminal = tokio::time::timeout(Duration::from_secs(5), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(terminal.status, TerminalStatus::Succeeded);
        assert_eq!(
            observed.lock().unwrap().as_slice(),
            &[Some(b"approved".to_vec())]
        );
        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn signal_timeout_delivers_none() {
        let (queue, store, clock) = open_queue_at(1_700_000_000_000).await;
        let (runtime, observed, mut rx) =
            signal_probe_runtime(queue.clone(), store, "order-2", Duration::from_secs(60));
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        wait_for_scheduled(&queue, 1).await;

        advance(&clock, Duration::from_secs(61)).await;
        queue.promote_scheduled_now().await.unwrap();

        let terminal = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(terminal.status, TerminalStatus::Succeeded);
        assert_eq!(observed.lock().unwrap().as_slice(), &[None]);

        assert!(
            queue
                .view()
                .kv_get(&signal_wait_kv_key("order-2"))
                .await
                .unwrap()
                .is_none()
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_buffered_signal_is_consumed_at_registration_and_a_later_one_replaces_it() {
        let (queue, store, _clock) = open_queue_at(1_700_000_000_000).await;
        let (runtime, observed, mut rx) =
            signal_probe_runtime(queue.clone(), store, "order-3", Duration::from_secs(3600));
        let shutdown = spawn_runtime(runtime.clone());

        assert_eq!(
            runtime.signal("order-3", b"first".to_vec()).await.unwrap(),
            SignalOutcome::Buffered
        );
        assert_eq!(
            runtime.signal("order-3", b"second".to_vec()).await.unwrap(),
            SignalOutcome::Buffered
        );

        runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();

        // The buffered signal is consumed at registration: the run
        // completes without any waiting and without the timeout elapsing.
        let terminal = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(terminal.status, TerminalStatus::Succeeded);
        assert_eq!(
            observed.lock().unwrap().as_slice(),
            &[Some(b"second".to_vec())]
        );

        assert!(
            queue
                .view()
                .kv_get(&signal_buf_kv_key("order-3"))
                .await
                .unwrap()
                .is_none()
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn clear_signal_discards_buffered_signal() {
        let (queue, store, _clock) = open_queue_at(1_700_000_000_000).await;
        let (runtime, _observed, _rx) =
            signal_probe_runtime(queue.clone(), store, "order-5", Duration::from_secs(60));

        assert_eq!(
            runtime.signal("order-5", b"stale".to_vec()).await.unwrap(),
            SignalOutcome::Buffered
        );
        assert!(runtime.clear_signal("order-5").await.unwrap());
        assert!(!runtime.clear_signal("order-5").await.unwrap());
        assert!(
            queue
                .view()
                .kv_get(&signal_buf_kv_key("order-5"))
                .await
                .unwrap()
                .is_none()
        );
    }

    #[tokio::test(start_paused = true)]
    async fn duplicate_waiter_registration_fails_the_run() {
        let (queue, store, _clock) = open_queue_at(1_700_000_000_000).await;
        let (runtime, _observed, mut rx) =
            signal_probe_runtime(queue.clone(), store, "order-6", Duration::from_secs(3600));
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                run_id: Some(rid("run-a")),
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        wait_for_scheduled(&queue, 1).await;

        runtime
            .submit(RunSpec {
                run_id: Some(rid("run-b")),
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();

        let terminal = tokio::time::timeout(Duration::from_secs(5), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(terminal.run_id, "run-b");
        assert_eq!(terminal.status, TerminalStatus::Failed);
        assert!(
            terminal
                .error
                .as_deref()
                .is_some_and(|e| e.contains("already registered"))
        );
        assert_eq!(
            queue.view().stats("workflow-steps").await.unwrap().dead,
            1,
            "the rejected registration dead-letters run-b's step",
        );
        assert!(
            queue
                .view()
                .kv_get(&run_kv_key(&rid("run-b")))
                .await
                .unwrap()
                .is_none(),
            "the run record delete rides the dead-letter",
        );
        assert!(
            queue
                .view()
                .kv_get(&run_kv_key(&rid("run-a")))
                .await
                .unwrap()
                .is_some(),
            "the waiting run keeps its record",
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn buffered_signal_missed_by_the_wake_is_delivered_at_timeout() {
        let (queue, store, clock) = open_queue_at(1_700_000_000_000).await;
        let (runtime, observed, mut rx) =
            signal_probe_runtime(queue.clone(), store, "order-7", Duration::from_secs(60));
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        wait_for_scheduled(&queue, 1).await;

        // A signal that was buffered without winning the wake (written
        // directly to simulate the settling-registration window).
        queue
            .kv_put(&signal_buf_kv_key("order-7"), b"late")
            .await
            .unwrap();

        advance(&clock, Duration::from_secs(61)).await;
        queue.promote_scheduled_now().await.unwrap();

        let terminal = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(terminal.status, TerminalStatus::Succeeded);
        assert_eq!(
            observed.lock().unwrap().as_slice(),
            &[Some(b"late".to_vec())]
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn cancelled_waiter_leaves_no_live_index_for_the_next_signal() {
        let (queue, store, _clock) = open_queue_at(1_700_000_000_000).await;
        let (runtime, _observed, mut rx) =
            signal_probe_runtime(queue.clone(), store, "order-8", Duration::from_secs(3600));
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        wait_for_scheduled(&queue, 1).await;

        assert!(runtime.cancel(&handle.run_id).await.unwrap());
        let terminal = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(terminal.status, TerminalStatus::Cancelled);

        // The signal finds only a stale index entry, cleans it and buffers.
        assert_eq!(
            runtime.signal("order-8", b"orphan".to_vec()).await.unwrap(),
            SignalOutcome::Buffered
        );
        assert!(
            queue
                .view()
                .kv_get(&signal_wait_kv_key("order-8"))
                .await
                .unwrap()
                .is_none()
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_failure_notification_inherits_the_step_limits() {
        let (queue, store) = open_queue().await;
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            FixedRunner::new(Err(StepError::permanent("nope"))),
            ChannelHook { tx },
        )
        .build();
        runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                options: RunOptions {
                    priority: Some(3),
                    max_attempts_per_step: Some(5),
                    ..Default::default()
                },
                ..Default::default()
            })
            .await
            .unwrap();

        let job = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .unwrap();
        let err = runtime
            .inner
            .process_step(&job, &LeaseHandle::detached())
            .await
            .unwrap_err();
        let failure = err
            .downcast_ref::<taquba::FailWith>()
            .expect("a terminating failure carries its effects");
        let notification = &failure.effects.enqueues[0];
        assert_eq!(notification.options.priority, Some(3));
        assert_eq!(notification.options.max_attempts, Some(5));
    }

    #[tokio::test(start_paused = true)]
    async fn a_duplicate_submit_is_idempotent_drops_its_effects_and_rejects_a_changed_input() {
        let (queue, store) = open_queue().await;
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            ScriptedRunner::new(vec![]),
            NoopTerminalHook,
        )
        .build();
        // No worker loop runs, so the step stays queued and the run is
        // active for every later submit.
        let spec = |input: &[u8], key: &[u8]| RunSpec {
            run_id: Some(rid("fixed-id")),
            input: input.to_vec(),
            effects: SettlementEffects::default().kv_put(key, b"1"),
            ..Default::default()
        };

        let first = runtime.submit(spec(b"x", b"app/first")).await.unwrap();
        assert!(first.newly_submitted);
        assert!(runtime.status(&rid("fixed-id")).await.unwrap().is_some());
        assert_eq!(
            queue.view().kv_get(b"app/first").await.unwrap().as_deref(),
            Some(b"1".as_slice())
        );

        let duplicate = runtime.submit(spec(b"x", b"app/second")).await.unwrap();
        assert_eq!(duplicate.run_id, "fixed-id");
        assert!(!duplicate.newly_submitted);
        assert_eq!(duplicate.job_id, first.job_id);
        assert!(queue.view().kv_get(b"app/second").await.unwrap().is_none());

        let err = runtime.submit(spec(b"y", b"app/third")).await.unwrap_err();
        assert!(matches!(&err, Error::InputMismatch(id) if id == "fixed-id"));
        assert!(err.is_permanent());
        assert!(queue.view().kv_get(b"app/third").await.unwrap().is_none());
    }

    #[tokio::test(start_paused = true)]
    async fn a_duplicate_known_only_from_the_durable_record_reports_the_current_job() {
        let (queue, store) = open_queue().await;
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        // Step 0 continues into a step scheduled an hour out, so the run
        // rests at step 1 with a job the second runtime never saw.
        let first = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            ScriptedRunner::new(vec![StepOutcome::continue_after(
                b"next".to_vec(),
                Duration::from_secs(3600),
            )]),
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(first.clone());
        let submitted = first
            .submit(RunSpec {
                run_id: Some(rid("durable")),
                input: b"x".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        for _ in 0..200 {
            if queue
                .view()
                .stats("workflow-steps")
                .await
                .unwrap()
                .scheduled
                == 1
            {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        let _ = shutdown.send(());

        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let second = WorkflowRuntime::builder(
            queue.clone(),
            store,
            ScriptedRunner::new(vec![]),
            ChannelHook { tx },
        )
        .build();
        let duplicate = second
            .submit(RunSpec {
                run_id: Some(rid("durable")),
                input: b"x".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        assert!(!duplicate.newly_submitted);
        assert_ne!(
            duplicate.job_id, submitted.job_id,
            "the pointer moved to step 1"
        );
        let step_1 = queue
            .view()
            .get_job(&duplicate.job_id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(step_1.status, taquba::JobStatus::Scheduled);
        assert_eq!(
            step_1.headers.get(HEADER_STEP).map(String::as_str),
            Some("1")
        );
    }

    #[tokio::test(start_paused = true)]
    async fn a_run_submitted_with_run_at_stays_scheduled_until_then() {
        struct Echo;
        impl StepRunner for Echo {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                Ok(StepOutcome::Succeed {
                    result: step.payload.clone(),
                })
            }
        }

        let t0 = 1_700_000_000_000;
        let (queue, store, clock) = open_queue_at(t0).await;
        let runtime =
            WorkflowRuntime::builder(queue.clone(), store, Echo, NoopTerminalHook).build();
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                options: RunOptions {
                    run_at: Some(UNIX_EPOCH + Duration::from_millis(t0 + 60_000)),
                    ..Default::default()
                },
                ..Default::default()
            })
            .await
            .unwrap();
        let scheduled = queue
            .view()
            .list_jobs("workflow-steps", taquba::JobStatus::Scheduled, None, 10)
            .await
            .unwrap()
            .jobs;
        assert_eq!(scheduled.len(), 1);
        let job_id = scheduled[0].id.clone();

        let waiter = tokio::spawn({
            let queue = queue.clone();
            async move { queue.wait_for_completion(&job_id).await }
        });
        advance(&clock, Duration::from_secs(120)).await;
        assert!(matches!(
            waiter.await.unwrap().unwrap(),
            taquba::WaitOutcome::Done(_),
        ));

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_step_reports_its_attempt_limit() {
        struct Recording(Arc<std::sync::Mutex<Option<u32>>>);
        impl StepRunner for Recording {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                *self.0.lock().unwrap() = Some(step.max_attempts);
                Ok(StepOutcome::Succeed { result: Vec::new() })
            }
        }

        let seen = Arc::new(std::sync::Mutex::new(None));
        let (queue, store) = open_queue().await;
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            Recording(seen.clone()),
            NoopTerminalHook,
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let outcome = runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                options: RunOptions {
                    max_attempts_per_step: Some(7),
                    ..Default::default()
                },
                ..Default::default()
            })
            .await
            .unwrap();
        let job_id = outcome.job_id;
        queue.wait_for_completion(&job_id).await.unwrap();
        assert_eq!(*seen.lock().unwrap(), Some(7));

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn concurrent_submits_of_one_run_admit_one_and_reject_a_changed_input() {
        let (queue, store) = open_queue().await;
        let runtime =
            WorkflowRuntime::builder(queue, store.clone(), PauseRunner, NoopTerminalHook).build();
        let spec = |input: &[u8]| RunSpec {
            run_id: Some(rid("raced")),
            input: input.to_vec(),
            ..Default::default()
        };

        let (first, same, changed) = tokio::join!(
            runtime.submit(spec(b"x")),
            runtime.submit(spec(b"x")),
            runtime.submit(spec(b"y")),
        );
        let first = first.unwrap();
        let same = same.unwrap();
        assert!(first.newly_submitted);
        assert!(!same.newly_submitted);
        assert_eq!(same.job_id, first.job_id);
        assert!(matches!(changed, Err(Error::InputMismatch(id)) if id == "raced"));
    }

    #[tokio::test(start_paused = true)]
    async fn restart_resumes_at_next_step() {
        // Headline durability test: after step 0 has acked and step 1 is in
        // the queue, kill runtime A entirely and spawn runtime B on the same
        // Queue handle. B should claim and complete step 1 without re-running
        // step 0.
        //
        // To make this race-free we gate step 0's runner: the test holds the
        // gate while signalling shutdown to A so A enters drain mode without
        // ever claiming step 1. Then the gate is opened, A's spawned step-0
        // task finishes (enqueueing step 1 + acking step 0) and A exits.
        struct CompleteOnStep1;
        impl StepRunner for CompleteOnStep1 {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                assert_eq!(step.step_number, 1, "runtime B should only ever see step 1");
                assert_eq!(step.payload.as_slice(), b"step1-payload");
                Ok(StepOutcome::Succeed {
                    result: b"resumed".to_vec(),
                })
            }
        }

        let (queue, store) = open_queue().await;

        let (runner, gate) =
            GatedRunner::new(Ok(StepOutcome::continue_now(b"step1-payload".to_vec())));
        let runtime_a =
            WorkflowRuntime::builder(queue.clone(), store.clone(), runner, NoopTerminalHook)
                .max_concurrent_steps(1)
                .build();

        let (shutdown_a_tx, shutdown_a_rx) = oneshot::channel::<()>();
        let worker_a = {
            let runtime_a = runtime_a.clone();
            tokio::spawn(async move {
                let _ = runtime_a
                    .run(async move {
                        let _ = shutdown_a_rx.await;
                    })
                    .await;
            })
        };

        let handle = runtime_a
            .submit(RunSpec {
                input: b"input".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();

        gate.claimed().await;
        let s = runtime_a
            .status(&handle.run_id)
            .await
            .unwrap()
            .expect("status");
        assert_eq!(s.state, RunState::Running);
        assert_eq!(s.current_step, 0);

        // A's worker is in the at-capacity select-loop. Signal shutdown
        // first, then open the gate so step 0 finishes processing inside
        // drain mode (A will not claim step 1).
        let _ = shutdown_a_tx.send(());
        gate.release();

        worker_a.await.expect("runtime A drained cleanly");

        // Bring up runtime B on the same Queue handle. It should pick up
        // step 1 from where A left off.
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime_b =
            WorkflowRuntime::builder(queue, store.clone(), CompleteOnStep1, ChannelHook { tx })
                .build();
        let shutdown_b = spawn_runtime(runtime_b.clone());

        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("hook fired in time")
            .expect("hook channel open");

        assert_eq!(outcome.run_id, handle.run_id);
        assert_eq!(outcome.status, TerminalStatus::Succeeded);
        assert_eq!(outcome.result.as_deref(), Some(b"resumed".as_slice()));
        assert_eq!(outcome.final_step, 1);

        let _ = shutdown_b.send(());
    }

    #[test]
    fn remaining_delay_measures_from_stored_timestamp() {
        let delay = Duration::from_secs(10);
        assert_eq!(remaining_delay(1_000, 4_000, delay), Duration::from_secs(7));
        assert_eq!(remaining_delay(1_000, 20_000, delay), Duration::ZERO);
        assert_eq!(remaining_delay(5_000, 1_000, delay), delay);
    }

    #[tokio::test(start_paused = true)]
    async fn step_output_replay_skips_runner_after_crash_before_ack() {
        let (queue, store) = open_queue().await;
        let calls = Arc::new(AtomicU32::new(0));
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            FixedRunner {
                result: Ok(StepOutcome::continue_now(b"step1-payload".to_vec())),
                calls: calls.clone(),
            },
            NoopTerminalHook,
        )
        .step_output_replay()
        .build();

        runtime
            .submit(RunSpec {
                run_id: Some(rid("replay-run")),
                input: b"input".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();

        let job = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .unwrap();

        // Simulate a crash after the step output was stored but before
        // the settlement committed: discard the effects of the first
        // delivery so nothing is enqueued.
        let _ = runtime
            .inner
            .process_step(&job, &LeaseHandle::detached())
            .await
            .unwrap();
        assert_eq!(calls.load(Ordering::SeqCst), 1);

        // Re-processing the same claimed record replays the stored step
        // outcome without invoking the runner a second time.
        let effects = runtime
            .inner
            .process_step(&job, &LeaseHandle::detached())
            .await
            .unwrap();
        assert_eq!(calls.load(Ordering::SeqCst), 1);
        queue.ack_with(&job, effects).await.unwrap();

        let next = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(next.payload.as_slice(), b"step1-payload");
        assert_eq!(next.headers.get(HEADER_RUN_ID).unwrap(), "replay-run");
        assert_eq!(next.headers.get(HEADER_STEP).unwrap(), "1");
        assert!(
            queue
                .claim("workflow-steps", Duration::from_secs(30))
                .await
                .unwrap()
                .is_none(),
            "the replayed continue must enqueue step 1 exactly once",
        );
    }

    #[tokio::test(start_paused = true)]
    async fn corrupt_step_output_replay_entry_falls_back_to_runner() {
        let (queue, store) = open_queue().await;
        let calls = Arc::new(AtomicU32::new(0));
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            FixedRunner {
                result: Ok(StepOutcome::continue_now(b"step1-payload".to_vec())),
                calls: calls.clone(),
            },
            NoopTerminalHook,
        )
        .step_output_replay()
        .build();

        runtime
            .submit(RunSpec {
                run_id: Some(rid("corrupt-run")),
                input: b"input".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();

        let job = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .unwrap();
        runtime
            .inner
            .core
            .memo_store
            .put_step_output(&rid("corrupt-run"), 0, &job.payload, b"not msgpack")
            .await
            .unwrap();

        runtime
            .inner
            .process_step(&job, &LeaseHandle::detached())
            .await
            .unwrap();
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "corrupt entry is treated as a miss",
        );

        // The recomputed outcome overwrites the corrupt entry, so a
        // second delivery replays it without invoking the runner again.
        runtime
            .inner
            .process_step(&job, &LeaseHandle::detached())
            .await
            .unwrap();
        assert_eq!(calls.load(Ordering::SeqCst), 1);
    }

    #[tokio::test(start_paused = true)]
    async fn step_output_replay_of_terminal_outcome_skips_runner() {
        let (queue, store) = open_queue().await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let calls = Arc::new(AtomicU32::new(0));
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            FixedRunner {
                result: Ok(StepOutcome::Succeed {
                    result: b"final".to_vec(),
                }),
                calls: calls.clone(),
            },
            ChannelHook { tx },
        )
        .step_output_replay()
        .build();

        runtime
            .submit(RunSpec {
                run_id: Some(rid("terminal-replay")),
                input: b"input".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        let job = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .unwrap();

        runtime
            .inner
            .process_step(&job, &LeaseHandle::detached())
            .await
            .unwrap();
        assert_eq!(calls.load(Ordering::SeqCst), 1);

        // Redelivery after a crash before ack: the stored outcome
        // settles the run without invoking the runner again.
        let effects = runtime
            .inner
            .process_step(&job, &LeaseHandle::detached())
            .await
            .unwrap();
        assert_eq!(calls.load(Ordering::SeqCst), 1);
        queue.ack_with(&job, effects).await.unwrap();

        // The committed settlement enqueued one notification; the hook
        // observes the replayed outcome when it is processed.
        let notification = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .unwrap();
        let effects = runtime
            .inner
            .process_step(&notification, &LeaseHandle::detached())
            .await
            .unwrap();
        queue.ack_with(&notification, effects).await.unwrap();
        let outcome = rx.recv().await.unwrap();
        assert_eq!(outcome.status, TerminalStatus::Succeeded);
        assert_eq!(outcome.result.as_deref(), Some(b"final".as_slice()));
        assert_eq!(calls.load(Ordering::SeqCst), 1);
    }

    /// Submits a run whose runner always returns
    /// [`StepError::transient`], capped at `max_attempts`. Asserts the
    /// runner is invoked exactly `max_attempts` times (per-step max-attempts
    /// propagation) and that the terminal hook fires Failed exactly once on
    /// the final attempt (fire-once-on-last-attempt logic).
    async fn assert_transient_retries_until_max(max_attempts: u32) {
        let (queue, store) = open_queue_with(fast_options()).await;
        let calls = Arc::new(AtomicU32::new(0));
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            FixedRunner {
                result: Err(StepError::transient("flaky")),
                calls: calls.clone(),
            },
            ChannelHook { tx },
        )
        .memo_retention(Duration::from_secs(60))
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                options: RunOptions {
                    max_attempts_per_step: Some(max_attempts),
                    ..Default::default()
                },
                ..Default::default()
            })
            .await
            .unwrap();

        let outcome = tokio::time::timeout(Duration::from_secs(3), rx.recv())
            .await
            .expect("hook fired in time")
            .expect("hook channel open");

        assert_eq!(outcome.status, TerminalStatus::Failed);
        assert_eq!(outcome.error.as_deref(), Some("flaky"));
        assert_eq!(
            calls.load(Ordering::SeqCst),
            max_attempts,
            "runner called once per attempt up to max_attempts"
        );

        // Settle window: assert no duplicate hook fires after the terminal one.
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert!(rx.try_recv().is_err(), "hook fired more than once");

        // The notification job was enqueued with the exhausted nack, so
        // its effects are committed once the hook fires.
        assert_eq!(queue.view().stats("workflow-steps").await.unwrap().dead, 1);
        assert!(
            queue
                .view()
                .kv_get(&run_kv_key(&handle.run_id))
                .await
                .unwrap()
                .is_none(),
            "the run record delete rides the exhausted nack",
        );
        assert_eq!(
            terminal_markers(&queue)
                .await
                .iter()
                .filter(|(run_id, _)| *run_id == handle.run_id)
                .count(),
            1,
            "the terminal marker rides the exhausted nack",
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_cancellation_survives_a_restart() {
        // Models a restart: the request recorded on the run record and
        // the job's persisted `cancel_requested` survive while a fresh
        // runtime starts with no process state. The runner returns
        // Succeed, so a Cancelled outcome shows the request was read.
        let (queue, store, _clock) = open_queue_at(10_000).await;
        let (tx_a, _rx_a) = tokio::sync::mpsc::unbounded_channel();
        let before = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            ScriptedRunner::new(vec![StepOutcome::Succeed {
                result: b"done".to_vec(),
            }]),
            ChannelHook { tx: tx_a },
        )
        .build();

        let handle = before
            .submit(RunSpec {
                input: b"x".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        let claim = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .expect("step 0 is claimable");
        assert!(before.cancel(&handle.run_id).await.unwrap());

        let (tx_b, mut rx_b) = tokio::sync::mpsc::unbounded_channel();
        let after = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            ScriptedRunner::new(vec![StepOutcome::Succeed {
                result: b"done".to_vec(),
            }]),
            ChannelHook { tx: tx_b },
        )
        .build();
        assert_eq!(
            after.status(&handle.run_id).await.unwrap().map(|s| s.state),
            Some(RunState::Cancelling),
            "the fresh runtime reads the request from the run record",
        );

        let effects = after
            .inner
            .process_step(&claim, &queue.lease_handle(&claim))
            .await
            .unwrap();
        queue.ack_with(&claim, effects).await.unwrap();

        let notification = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .expect("the terminal notification is claimable");
        let effects = after
            .inner
            .process_step(&notification, &LeaseHandle::detached())
            .await
            .unwrap();
        queue.ack_with(&notification, effects).await.unwrap();

        let outcome = rx_b.recv().await.unwrap();
        assert_eq!(outcome.status, TerminalStatus::Cancelled);
        assert!(outcome.result.is_none(), "the succeed payload is discarded");
    }

    #[tokio::test(start_paused = true)]
    async fn a_cancellation_after_the_settlement_read_reaches_the_next_step() {
        // The worker reads the claim's token once the runner has returned.
        // A request recorded after that read does not affect the
        // advancing settlement; it is read from the run record when the
        // next step is claimed, which is then settled without running.
        let (queue, store, _clock) = open_queue_at(10_000).await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            ScriptedRunner::new(vec![
                StepOutcome::Continue {
                    payload: b"next".to_vec(),
                    when: Trigger::Immediate,
                },
                StepOutcome::Succeed {
                    result: b"done".to_vec(),
                },
            ]),
            ChannelHook { tx },
        )
        .build();

        let handle = runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        let step0 = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .expect("step 0 is claimable");
        let effects = runtime
            .inner
            .process_step(&step0, &queue.lease_handle(&step0))
            .await
            .unwrap();
        assert!(runtime.cancel(&handle.run_id).await.unwrap());
        queue.ack_with(&step0, effects).await.unwrap();

        let step1 = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .expect("step 1 is claimable");
        let effects = runtime
            .inner
            .process_step(&step1, &queue.lease_handle(&step1))
            .await
            .unwrap();
        queue.ack_with(&step1, effects).await.unwrap();

        let notification = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .expect("the terminal notification is claimable");
        let effects = runtime
            .inner
            .process_step(&notification, &LeaseHandle::detached())
            .await
            .unwrap();
        queue.ack_with(&notification, effects).await.unwrap();

        let outcome = rx.recv().await.unwrap();
        assert_eq!(outcome.status, TerminalStatus::Cancelled);
        assert_eq!(outcome.final_step, 1);
        assert_eq!(
            terminal_status_of(&runtime, &handle.run_id).await,
            Some(outcome.status)
        );
    }

    #[tokio::test(start_paused = true)]
    async fn cancelling_a_pending_run_commits_its_marker_and_fires_the_hook_once() {
        // Pending case: a run sits in the queue, we call `cancel()` before
        // any worker claims it. `cancel` removes the step job and enqueues
        // the notification before returning.

        let (queue, store, _clock) = open_queue_at(10_000).await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            UnreachableRunner,
            ChannelHook { tx },
        )
        .memo_retention(Duration::from_secs(60))
        .build();
        // Note: deliberately do NOT spawn the worker loop, so the submitted
        // step stays Pending in the queue while we cancel it.

        let mut headers = HashMap::new();
        headers.insert("tenant".to_string(), "acme".to_string());

        let handle = runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                options: RunOptions {
                    headers,
                    ..Default::default()
                },
                ..Default::default()
            })
            .await
            .unwrap();
        let status = runtime
            .status(&handle.run_id)
            .await
            .unwrap()
            .expect("active");
        assert_eq!(status.state, RunState::Pending);

        let was_cancelled = runtime.cancel(&handle.run_id).await.unwrap();
        assert!(was_cancelled);
        let status = runtime.status(&handle.run_id).await.unwrap().unwrap();
        assert_eq!(
            status.state,
            RunState::Terminated(RunTermination {
                status: TerminalStatus::Cancelled,
                error: None,
                error_kind: None,
                final_step: 0,
                terminated_at_ms: 10_000,
            }),
            "the terminal record commits with the removal",
        );
        assert_eq!(status.current_step, 0);
        assert!(
            runtime.outcome(&handle.run_id).await.unwrap().is_none(),
            "no worker terminated the run, so no run result record exists",
        );
        assert!(
            !runtime.cancel(&handle.run_id).await.unwrap(),
            "a second cancel finds no run record",
        );

        // The marker and the run record's delete commit with the removal,
        // before the notification is processed.
        let markers = terminal_markers(&queue).await;
        assert_eq!(markers, vec![(handle.run_id.clone(), 10_000)]);
        assert_eq!(
            queue
                .view()
                .kv_get(&run_kv_key(&handle.run_id))
                .await
                .unwrap(),
            None,
        );

        // The step job is gone; the one claimable job is the notification.
        let notification = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .unwrap();
        let effects = runtime
            .inner
            .process_step(&notification, &LeaseHandle::detached())
            .await
            .unwrap();
        queue.ack_with(&notification, effects).await.unwrap();
        let outcome = rx.recv().await.unwrap();
        assert_eq!(outcome.run_id, handle.run_id);
        assert_eq!(outcome.status, TerminalStatus::Cancelled);
        // External cancellation carries no reason: `error` is `None`.
        assert!(outcome.error.is_none());
        assert_eq!(outcome.headers.get("tenant").unwrap(), "acme");
        assert!(
            queue
                .claim("workflow-steps", Duration::from_secs(30))
                .await
                .unwrap()
                .is_none(),
            "the cancel enqueues one notification",
        );
        assert!(rx.try_recv().is_err());

        let stats = queue.view().stats("workflow-steps").await.unwrap();
        assert_eq!(stats.dead, 0, "cancel must not dead-letter");
        assert_eq!(stats.pending, 0, "cancelled job must be removed");
    }

    #[tokio::test(start_paused = true)]
    async fn a_view_reads_the_same_state_through_the_runtime_and_a_reader() {
        let (queue, store, _clock) = open_queue_at(10_000).await;
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            UnreachableRunner,
            NoopTerminalHook,
        )
        .memo_prefix("memo")
        .build();
        // The worker loop is not spawned, so the step stays pending.
        let handle = runtime.submit(RunSpec::default()).await.unwrap();

        let reader = QueueReader::open(store.clone(), "test").await.unwrap();
        let view = WorkflowView::new(reader.view().clone(), MemoStore::new(store.clone(), "memo"));
        let through_reader = view.status(&handle.run_id).await.unwrap().expect("active");
        let through_runtime = runtime.status(&handle.run_id).await.unwrap().unwrap();
        assert_eq!(through_reader.run_id, handle.run_id);
        assert_eq!(through_reader.state, RunState::Pending);
        assert_eq!(through_reader.state, through_runtime.state);
        assert_eq!(through_reader.current_step, through_runtime.current_step);
        assert!(view.outcome(&handle.run_id).await.unwrap().is_none());
        assert!(view.status(&rid("unknown")).await.unwrap().is_none());

        assert!(runtime.cancel(&handle.run_id).await.unwrap());
        let reader = QueueReader::open(store.clone(), "test").await.unwrap();
        let view = WorkflowView::new(reader.view().clone(), MemoStore::new(store, "memo"));
        let status = view
            .status(&handle.run_id)
            .await
            .unwrap()
            .expect("terminal record");
        assert_eq!(
            status.state,
            RunState::Terminated(RunTermination {
                status: TerminalStatus::Cancelled,
                error: None,
                error_kind: None,
                final_step: 0,
                terminated_at_ms: 10_000,
            }),
        );
    }

    #[tokio::test(start_paused = true)]
    async fn the_status_of_a_run_does_not_read_its_offloaded_step_payload() {
        let payloads: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
        let (queue, store, _clock) = open_queue_at_with(
            10_000,
            OpenOptions::default()
                .payload_offload_threshold(64)
                .payload_store(payloads.clone()),
        )
        .await;
        let runtime = WorkflowRuntime::builder(queue, store, UnreachableRunner, NoopTerminalHook)
            .memo_prefix("memo")
            .build();
        let handle = runtime
            .submit(RunSpec {
                input: vec![7u8; 512],
                ..Default::default()
            })
            .await
            .unwrap();

        // A read that fetches the step's payload object fails once the object
        // is deleted.
        let objects: Vec<_> = payloads.list(None).try_collect().await.unwrap();
        assert!(!objects.is_empty(), "the step payload is offloaded");
        for object in objects {
            payloads.delete(&object.location).await.unwrap();
        }
        let status = runtime.status(&handle.run_id).await.unwrap().unwrap();
        assert_eq!(status.state, RunState::Pending);
    }

    /// Drive a single step that blocks on a gate, calls `cancel(run_id)`
    /// while the step is in-flight, and then has the runner return the
    /// supplied error. Asserts that external cancellation suppresses the
    /// error path entirely: the hook fires `Cancelled` (not `Failed`),
    /// no dead-letter is produced regardless of `permanent`/`transient`,
    /// and the worker returns `Ok` (no retry, no PermanentFailure
    /// propagation).
    async fn assert_cancel_suppresses_runner_error(error: StepError) {
        let (queue, store) = open_queue_with(fast_options()).await;
        let (runner, gate) = GatedRunner::new(Err(error));
        let (hook_tx, mut hook_rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            runner,
            ChannelHook { tx: hook_tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        gate.claimed().await;

        let was_cancelled = runtime.cancel(&handle.run_id).await.unwrap();
        assert!(was_cancelled);

        // Release the runner. It returns Err; without cancellation this
        // would either dead-letter (permanent) or nack for retry
        // (transient). Cancellation must suppress both.
        gate.release();

        let outcome = tokio::time::timeout(Duration::from_secs(2), hook_rx.recv())
            .await
            .expect("hook fired")
            .expect("hook channel open");
        assert_eq!(outcome.status, TerminalStatus::Cancelled);
        assert!(
            outcome.error.is_none(),
            "external cancel must carry no reason (Some(_) would imply runner-issued StepOutcome::Cancel)",
        );
        assert_eq!(
            terminal_status_of(&runtime, &handle.run_id).await,
            Some(outcome.status)
        );

        // Settle window: assert no retry attempt and no dead-letter or
        // duplicate hook fires after the terminal one.
        tokio::time::sleep(Duration::from_millis(100)).await;
        assert_eq!(
            gate.calls.load(Ordering::SeqCst),
            1,
            "cancellation must suppress retries",
        );
        let stats = queue.view().stats("workflow-steps").await.unwrap();
        assert_eq!(stats.dead, 0, "cancellation must suppress dead-letter");
        assert!(
            hook_rx.try_recv().is_err(),
            "hook must fire exactly once for the cancelled run",
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn cancel_suppresses_a_runner_error() {
        // Without cancellation, `StepError::permanent` dead-letters the
        // step and `StepError::transient` nacks for retry. With an
        // external cancel in flight, the worker must ack and fire
        // `Cancelled` instead, without re-invoking the runner.
        assert_cancel_suppresses_runner_error(StepError::permanent("would-dead-letter")).await;
        assert_cancel_suppresses_runner_error(StepError::transient("would-retry")).await;
    }

    #[tokio::test(start_paused = true)]
    async fn cancel_signals_step_token_for_cooperative_short_circuit() {
        // A runner that watches `step.cancel_token` should short-circuit
        // long after-claim work as soon as `WorkflowRuntime::cancel` is
        // called. Without the token, cancellation latency is bounded by
        // step duration; with it, the runner returns essentially
        // immediately. The test pins this by using a step that would
        // otherwise sleep for 30 seconds; if the token didn't fire, the
        // test would time out.
        struct CooperativeRunner {
            claimed: Arc<tokio::sync::Notify>,
        }
        impl StepRunner for CooperativeRunner {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                self.claimed.notify_one();
                tokio::select! {
                    _ = tokio::time::sleep(Duration::from_secs(30)) => {
                        Ok(StepOutcome::Succeed { result: b"slow".to_vec() })
                    }
                    _ = step.cancel_token.cancelled() => {
                        Ok(StepOutcome::Cancel { reason: "cooperative".to_string() })
                    }
                }
            }
        }

        let (queue, store) = open_queue().await;
        let claimed = Arc::new(tokio::sync::Notify::new());
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            CooperativeRunner {
                claimed: claimed.clone(),
            },
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        tokio::time::timeout(Duration::from_secs(2), claimed.notified())
            .await
            .expect("runner observed token");

        let was_cancelled = runtime.cancel(&handle.run_id).await.unwrap();
        assert!(was_cancelled);

        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("hook fired well before the 30s sleep would have")
            .expect("hook channel open");

        assert_eq!(outcome.status, TerminalStatus::Cancelled);
        // Runner-issued Cancel wins precedence over external cancel, so
        // the runner's reason surfaces.
        assert_eq!(outcome.error.as_deref(), Some("cooperative"));
        assert_eq!(
            terminal_status_of(&runtime, &handle.run_id).await,
            Some(outcome.status)
        );

        let stats = queue.view().stats("workflow-steps").await.unwrap();
        assert_eq!(stats.dead, 0);

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn cancel_returns_false_for_a_terminated_or_unknown_run() {
        // Submit a run that succeeds normally, wait for the terminal
        // hook, then call `cancel`. The run record was deleted with the
        // success, so `cancel` must report `Ok(false)` and must not fire
        // a second hook.
        let (queue, store) = open_queue().await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue,
            store.clone(),
            ScriptedRunner::new(vec![StepOutcome::Succeed {
                result: b"done".to_vec(),
            }]),
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();

        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("Succeeded hook fired")
            .expect("hook channel open");
        assert_eq!(outcome.status, TerminalStatus::Succeeded);
        assert_eq!(
            terminal_status_of(&runtime, &handle.run_id).await,
            Some(outcome.status)
        );

        let was_cancelled = runtime.cancel(&handle.run_id).await.unwrap();
        assert!(
            !was_cancelled,
            "cancel on an already-terminated run must report Ok(false)",
        );

        tokio::time::sleep(Duration::from_millis(50)).await;
        assert!(
            rx.try_recv().is_err(),
            "no Cancelled hook may fire after the run already terminated as Succeeded",
        );

        assert!(!runtime.cancel(&rid("never-submitted")).await.unwrap());

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn transient_retries_until_max_attempts() {
        assert_transient_retries_until_max(1).await;
        assert_transient_retries_until_max(3).await;
    }

    #[tokio::test(start_paused = true)]
    async fn step_memo_survives_across_attempts_of_the_same_step() {
        // First attempt writes the memo entry and returns a transient
        // error so the runtime retries the step. The second attempt
        // reads the same key back and succeeds. This exercises the
        // central use case: at-least-once retries of one step should
        // short-circuit work the prior attempt already did.
        struct MemoRetryRunner;
        impl StepRunner for MemoRetryRunner {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                if step.attempts == 1 {
                    step.memo
                        .put("cached", b"first-attempt-value")
                        .await
                        .map_err(|e| StepError::transient(e.to_string()))?;
                    return Err(StepError::transient("force a retry"));
                }
                let got = step
                    .memo
                    .get("cached")
                    .await
                    .map_err(|e| StepError::transient(e.to_string()))?;
                assert_eq!(got, Some(b"first-attempt-value".to_vec()));
                Ok(StepOutcome::Succeed {
                    result: got.unwrap_or_default(),
                })
            }
        }

        let (queue, store) = open_queue_with(fast_options()).await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime =
            WorkflowRuntime::builder(queue, store, MemoRetryRunner, ChannelHook { tx }).build();
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                input: b"start".to_vec(),
                options: RunOptions {
                    max_attempts_per_step: Some(3),
                    ..Default::default()
                },
                ..Default::default()
            })
            .await
            .unwrap();
        let outcome = tokio::time::timeout(Duration::from_secs(3), rx.recv())
            .await
            .expect("hook fired in time")
            .expect("hook channel open");
        assert_eq!(outcome.status, TerminalStatus::Succeeded);
        assert_eq!(
            outcome.result.as_deref(),
            Some(b"first-attempt-value".as_slice())
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn terminal_marker_is_written_at_the_runtime_clock() {
        // The queue's MockClock is shared into the runtime by default
        // (via Queue::clock()), so a `clock.advance` between submit and
        // terminate is visible in the marker's terminal_at_ms.
        let (queue, store, clock) = open_queue_at(10_000).await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue,
            store.clone(),
            ScriptedRunner::new(vec![StepOutcome::Succeed {
                result: b"done".to_vec(),
            }]),
            ChannelHook { tx },
        )
        .memo_retention(Duration::from_secs(60))
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: b"in".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        advance(&clock, Duration::from_secs(30)).await;
        let _ = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        let markers = terminal_markers(&runtime.inner.core.queue).await;
        assert_eq!(markers.len(), 1);
        assert_eq!(markers[0].0, handle.run_id);
        // MockClock only moves on explicit advance/set, so the value the
        // effects builder reads is exactly the post-advance clock.
        assert_eq!(markers[0].1, 10_000 + 30_000);

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn submit_rejects_reserved_headers_and_reserved_kv_keys() {
        let (queue, store, _clock) = open_queue_at(10_000).await;
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue,
            store,
            ScriptedRunner::new(vec![]),
            ChannelHook { tx },
        )
        .build();

        let err = runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                options: RunOptions {
                    headers: HashMap::from([("workflow.run_id".to_string(), "evil".to_string())]),
                    ..Default::default()
                },
                ..Default::default()
            })
            .await
            .unwrap_err();
        assert!(
            matches!(&err, Error::ReservedHeaderInSubmit(k) if k == "workflow.run_id"),
            "got: {err:?}"
        );

        let err = runtime
            .submit(RunSpec {
                input: Vec::new(),
                effects: SettlementEffects::default().kv_put(b"workflow/x", b"v"),
                ..Default::default()
            })
            .await
            .unwrap_err();
        assert!(matches!(err, Error::ReservedKvKey(_)));

        let err = runtime
            .submit(RunSpec {
                input: Vec::new(),
                effects: SettlementEffects::default().kv_delete(b"workflow/x"),
                ..Default::default()
            })
            .await
            .unwrap_err();
        assert!(matches!(err, Error::ReservedKvKey(_)));
    }

    #[tokio::test(start_paused = true)]
    async fn submit_applies_the_deletes_and_expiry_entries_of_the_spec() {
        let (queue, store, _clock) = open_queue_at(10_000).await;
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            ScriptedRunner::new(vec![]),
            ChannelHook { tx },
        )
        .build();
        let index = ExpiryIndex::new(b"app/expiry/".to_vec());
        queue.kv_put(b"app/stale", b"1").await.unwrap();
        // An entry at 9_000 sets the bound of the index, so a pass reads the
        // entry at 8_000 only if the submit records it.
        queue
            .commit_effects(SettlementEffects::default().expiry_entry(&index, 9_000, b"later"))
            .await
            .unwrap();

        runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                effects: SettlementEffects::default()
                    .kv_delete(b"app/stale")
                    .expiry_entry(&index, 8_000, b"run"),
                ..Default::default()
            })
            .await
            .unwrap();

        assert!(queue.view().kv_get(b"app/stale").await.unwrap().is_none());
        let mut seen = Vec::new();
        index
            .pass(&queue, 10_000, Duration::ZERO, |at_ms, suffix| {
                seen.push((at_ms, suffix));
                std::future::ready(Expired::Delete(SettlementEffects::default()))
            })
            .await
            .unwrap();
        assert_eq!(seen, [(8_000, b"run".to_vec()), (9_000, b"later".to_vec())]);
    }

    #[tokio::test(start_paused = true)]
    async fn a_malformed_terminal_marker_is_deleted_without_clearing_memos() {
        let (queue, store, clock) = open_queue_at(10_000).await;
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            ScriptedRunner::new(vec![]),
            ChannelHook { tx },
        )
        .memo_retention(Duration::from_secs(60))
        .build();

        let memos = MemoStore::new(store, "workflow-steps-memo");
        memos
            .new_memo(&rid("bystander"), 0)
            .put("k", b"expensive")
            .await
            .unwrap();
        // A marker with an empty id, which `RunId` rejects.
        let marker = ExpiryIndex::new(TERMINAL_KV_PREFIX).entry_key(0, b"");
        queue.kv_put(&marker, b"").await.unwrap();
        // A key too short for a time.
        let unparseable = [TERMINAL_KV_PREFIX, b"short"].concat();
        queue.kv_put(&unparseable, b"").await.unwrap();

        advance(&clock, Duration::from_secs(3_600)).await;
        runtime.inner.core.sweep_once().await.unwrap();
        assert_eq!(
            memos.new_memo(&rid("bystander"), 0).get("k").await.unwrap(),
            Some(b"expensive".to_vec()),
            "an unrelated run's memo entries must survive",
        );
        assert!(
            queue.view().kv_get(&marker).await.unwrap().is_none(),
            "the marker is removed and not retried on every sweep",
        );
        assert!(queue.view().kv_get(&unparseable).await.unwrap().is_none());
    }

    #[tokio::test(start_paused = true)]
    async fn cancelling_a_running_step_overrides_its_outcome_and_writes_its_marker_at_settlement() {
        // Between the request and the settlement the run reports
        // Cancelling and holds its record with no marker (the queue
        // discards the effects on the `Requested` arm); the settlement
        // then commits Cancelled in place of the runner's outcome.
        let (queue, store, _clock) = open_queue_at(10_000).await;
        let (runner, gate) = GatedRunner::new(Ok(StepOutcome::Succeed {
            result: b"would-have-succeeded".to_vec(),
        }));
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime =
            WorkflowRuntime::builder(queue.clone(), store.clone(), runner, ChannelHook { tx })
                .memo_retention(Duration::from_secs(60))
                .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        gate.claimed().await;
        assert_eq!(
            runtime
                .status(&handle.run_id)
                .await
                .unwrap()
                .expect("active")
                .state,
            RunState::Running
        );

        assert!(runtime.cancel(&handle.run_id).await.unwrap());
        assert_eq!(
            runtime
                .status(&handle.run_id)
                .await
                .unwrap()
                .expect("entry retained while termination is in flight")
                .state,
            RunState::Cancelling
        );
        assert!(
            terminal_markers(&queue).await.is_empty(),
            "a run still executing its step must have no terminal marker",
        );
        assert!(
            queue
                .view()
                .kv_get(&run_kv_key(&handle.run_id))
                .await
                .unwrap()
                .is_some(),
            "the run record must survive a cancel the worker has to finish",
        );

        gate.release();
        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("hook fired")
            .expect("hook channel open");
        assert_eq!(outcome.status, TerminalStatus::Cancelled);
        assert!(
            outcome.result.is_none(),
            "succeed payload must be discarded"
        );
        let status = runtime.status(&handle.run_id).await.unwrap().unwrap();
        assert!(matches!(
            status.state,
            RunState::Terminated(RunTermination {
                status: TerminalStatus::Cancelled,
                error: None,
                ..
            })
        ));
        assert_eq!(
            terminal_markers(&queue).await,
            vec![(handle.run_id.clone(), 10_000)],
            "the worker's settlement writes it",
        );
        assert_eq!(queue.view().stats("workflow-steps").await.unwrap().dead, 0);

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_permanent_step_error_dead_letters_with_its_marker_and_no_staged_effects() {
        let (queue, store, _clock) = open_queue_at(10_000).await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            EffectStagingRunner::new(vec![Err(StepError::permanent("nope"))]),
            ChannelHook { tx },
        )
        .memo_retention(Duration::from_secs(60))
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.run_id, handle.run_id);
        assert_eq!(outcome.status, TerminalStatus::Failed);
        assert_eq!(outcome.error.as_deref(), Some("nope"));
        let status = runtime.status(&handle.run_id).await.unwrap().unwrap();
        assert!(
            matches!(
                status.state,
                RunState::Terminated(RunTermination {
                    status: TerminalStatus::Failed,
                    error: Some(ref error),
                    error_kind: Some(StepErrorKind::Permanent),
                    ..
                }) if error == "nope"
            ),
            "the terminal record commits with the dead-letter and carries the error kind",
        );
        let recorded = runtime.outcome(&handle.run_id).await.unwrap().unwrap();
        assert_eq!(recorded.status, TerminalStatus::Failed);
        assert_eq!(recorded.error.as_deref(), Some("nope"));

        // The notification was enqueued by the dead-letter transaction, so
        // the dead job and the marker are already visible, and the staged
        // effect was discarded with the failure.
        assert_eq!(queue.view().stats("workflow-steps").await.unwrap().dead, 1);
        assert_eq!(
            terminal_markers(&queue).await,
            vec![(handle.run_id.clone(), 10_000)],
        );
        assert!(queue.view().kv_get(b"app/step-0").await.unwrap().is_none());

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_retrying_step_error_commits_no_terminal_marker() {
        // A transient failure with attempts left nacks for retry, and
        // `nack_with` discards the effects on that branch.
        struct FlakyRunner {
            attempts: Arc<std::sync::atomic::AtomicUsize>,
            clock: MockClock,
        }
        impl StepRunner for FlakyRunner {
            async fn run_step(&self, _step: &Step) -> std::result::Result<StepOutcome, StepError> {
                if self
                    .attempts
                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
                    == 0
                {
                    return Err(StepError::transient("flaky"));
                }
                // Separate the two settlements in clock time before the
                // succeeding one: a marker wrongly written for the retry
                // would otherwise share this one's key and go unobserved.
                self.clock.advance(Duration::from_secs(1));
                Ok(StepOutcome::Succeed {
                    result: b"done".to_vec(),
                })
            }
        }

        let (queue, store, clock) = open_queue_at_with(10_000, fast_options()).await;
        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            FlakyRunner {
                attempts: attempts.clone(),
                clock: clock.clone(),
            },
            ChannelHook { tx },
        )
        .memo_retention(Duration::from_secs(60))
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: b"x".to_vec(),
                options: RunOptions {
                    max_attempts_per_step: Some(3),
                    ..Default::default()
                },
                ..Default::default()
            })
            .await
            .unwrap();
        let outcome = tokio::time::timeout(Duration::from_secs(5), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.status, TerminalStatus::Succeeded);
        assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 2);

        // Exactly one marker, stamped after the retry's clock advance:
        // the retry produced none, the success did.
        let markers = terminal_markers(&queue).await;
        assert_eq!(markers.len(), 1);
        assert_eq!(markers[0].0, handle.run_id);
        assert_eq!(markers[0].1, 11_000);

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn the_sweep_clears_only_markers_older_than_the_cutoff() {
        // Markers sort by timestamp, so the sweep scans from the start
        // of the range and returns at the first unexpired marker.
        let (queue, store, _clock) = open_queue_at(10_000).await;
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            ScriptedRunner::new(vec![]),
            ChannelHook { tx },
        )
        .memo_retention(Duration::from_secs(1))
        .build();

        let memos = MemoStore::new(store, "workflow-steps-memo");
        let sweep = runtime.inner.core.memo_sweep.as_ref().unwrap();
        for (run_id, at_ms) in [("old", 1_000u64), ("young", 9_500u64)] {
            let run_id = rid(run_id);
            memos.new_memo(&run_id, 0).put("k", b"v").await.unwrap();
            queue
                .commit_effects(sweep.mark(SettlementEffects::default(), &run_id, at_ms))
                .await
                .unwrap();
        }

        // Clock at 10_000 with 1s retention: cutoff 9_000.
        let cleared = runtime.inner.core.sweep_once().await.unwrap();
        assert_eq!(cleared, 1);

        let remaining = terminal_markers(&queue).await;
        assert_eq!(remaining.len(), 1);
        assert_eq!(remaining[0].0, "young");
        assert_eq!(memos.new_memo(&rid("old"), 0).get("k").await.unwrap(), None);
        assert_eq!(
            memos.new_memo(&rid("young"), 0).get("k").await.unwrap(),
            Some(b"v".to_vec()),
        );
    }

    #[tokio::test(start_paused = true)]
    async fn a_re_submitted_run_shares_its_entries_until_the_first_run_expires() {
        let (queue, store, clock) = open_queue_at(10_000).await;
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store.clone(),
            FixedRunner::new(Ok(StepOutcome::Succeed {
                result: b"done".to_vec(),
            })),
            NoopTerminalHook,
        )
        .memo_retention(Duration::from_secs(1))
        .build();
        let spec = RunSpec {
            run_id: Some(rid("shared")),
            input: b"x".to_vec(),
            ..Default::default()
        };
        runtime.submit(spec.clone()).await.unwrap();
        let claim = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .unwrap();
        let effects = runtime
            .inner
            .process_step(&claim, &LeaseHandle::detached())
            .await
            .unwrap();
        queue.ack_with(&claim, effects).await.unwrap();
        let memos = MemoStore::new(store, "workflow-steps-memo");
        memos
            .new_memo(&rid("shared"), 0)
            .put("k", b"v")
            .await
            .unwrap();

        assert!(runtime.submit(spec).await.unwrap().newly_submitted);
        assert_eq!(
            memos.new_memo(&rid("shared"), 0).get("k").await.unwrap(),
            Some(b"v".to_vec()),
            "the second run reads the first run's entry",
        );

        // The first run's marker expires while the second run is active.
        clock.advance(Duration::from_secs(2));
        assert_eq!(runtime.inner.core.sweep_once().await.unwrap(), 1);
        assert_eq!(
            memos.new_memo(&rid("shared"), 0).get("k").await.unwrap(),
            None
        );
        assert_eq!(
            runtime
                .status(&rid("shared"))
                .await
                .unwrap()
                .map(|s| s.state),
            Some(RunState::Pending),
            "the second run is still active",
        );
    }

    /// Yield up to `iters` times waiting for `cond` to become true.
    /// Used in sweeper tests to let the spawned sweep task make
    /// progress between `tokio::time::advance` and the assertion;
    /// returns true if the condition held within the budget.
    async fn yield_until<F, Fut>(iters: usize, mut cond: F) -> bool
    where
        F: FnMut() -> Fut,
        Fut: Future<Output = bool>,
    {
        for _ in 0..iters {
            if cond().await {
                return true;
            }
            tokio::task::yield_now().await;
        }
        false
    }

    #[tokio::test(start_paused = true)]
    async fn the_sweeper_clears_a_marker_only_after_retention_elapses() {
        // Retention 200ms and a 10ms poll interval. A pass 199ms after
        // the marker is written leaves it. Within a poll interval of
        // the boundary the sweep loop clears the marker and the run's
        // memo entries.
        let (queue, store, clock) = open_queue_at(10_000).await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue,
            store.clone(),
            ScriptedRunner::new(vec![StepOutcome::Succeed {
                result: b"done".to_vec(),
            }]),
            ChannelHook { tx },
        )
        .memo_retention(Duration::from_millis(200))
        .poll_interval(Duration::from_millis(10))
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: b"in".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        let _ = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        let memos = MemoStore::new(store.clone(), "workflow-steps-memo");
        memos
            .new_memo(&handle.run_id, 0)
            .put("k", b"cached")
            .await
            .unwrap();

        advance(&clock, Duration::from_millis(199)).await;
        assert_eq!(runtime.inner.core.sweep_once().await.unwrap(), 0);
        let markers = terminal_markers(&runtime.inner.core.queue).await;
        assert_eq!(
            markers.len(),
            1,
            "a marker within the window must not be swept"
        );

        advance(&clock, Duration::from_millis(1)).await;
        advance(&clock, Duration::from_millis(10)).await;
        let cleared = yield_until(50, || async {
            terminal_markers(&runtime.inner.core.queue).await.is_empty()
        })
        .await;
        assert!(cleared, "sweeper did not clear the expired marker");
        assert_eq!(
            memos.new_memo(&handle.run_id, 0).get("k").await.unwrap(),
            None,
            "sweeper did not clear the run's memo entries",
        );
        assert!(
            runtime.status(&handle.run_id).await.unwrap().is_none(),
            "sweeper did not clear the run's terminal record",
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn sweeper_keeps_memos_of_runs_without_a_terminal_marker() {
        // A run gets a terminal marker only once it terminates, and a
        // terminated run never resumes. The sweep is keyed on those
        // markers, so an in-flight run's memo entries are never deleted
        // out from under a resume, even past the retention window. Here
        // a memo entry exists for a run with no terminal marker;
        // advancing well past retention must leave it in place.
        let (queue, store, clock) = open_queue_at(10_000).await;
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue,
            store.clone(),
            ScriptedRunner::new(vec![]),
            ChannelHook { tx },
        )
        .memo_retention(Duration::from_millis(100))
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let memos = MemoStore::new(store.clone(), "workflow-steps-memo");
        memos
            .new_memo(&rid("in-flight-run"), 0)
            .put("k", b"cached")
            .await
            .unwrap();

        advance(&clock, Duration::from_millis(500)).await;
        // Give the sweeper several ticks to run against the advanced clock.
        for _ in 0..50 {
            tokio::task::yield_now().await;
        }

        assert_eq!(
            memos
                .new_memo(&rid("in-flight-run"), 0)
                .get("k")
                .await
                .unwrap(),
            Some(b"cached".to_vec()),
            "sweep must not remove memos of a run with no terminal marker",
        );

        let _ = shutdown.send(());
    }

    async fn wait_for_kv(queue: &Queue, key: &[u8]) -> Vec<u8> {
        for _ in 0..200 {
            if let Some(v) = queue.view().kv_get(key).await.unwrap() {
                return v.to_vec();
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        panic!(
            "kv key `{}` was never written",
            String::from_utf8_lossy(key)
        );
    }

    async fn wait_for_drained(queue: &Queue) {
        for _ in 0..200 {
            let stats = queue.view().stats("workflow-steps").await.unwrap();
            if stats.pending == 0 && stats.claimed == 0 && stats.scheduled == 0 {
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        panic!("the queue never drained");
    }

    /// Runner that stages one `app/step-{n}` write per step and returns
    /// the next scripted result.
    struct EffectStagingRunner {
        script: Arc<StdMutex<Vec<std::result::Result<StepOutcome, StepError>>>>,
    }

    impl EffectStagingRunner {
        fn new(script: Vec<std::result::Result<StepOutcome, StepError>>) -> Self {
            Self {
                script: Arc::new(StdMutex::new(script)),
            }
        }
    }

    impl StepRunner for EffectStagingRunner {
        async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
            step.effects
                .put(format!("app/step-{}", step.step_number), b"done".to_vec())
                .map_err(|e| StepError::permanent(e.to_string()))?;
            self.script.lock().unwrap().remove(0)
        }
    }

    #[tokio::test(start_paused = true)]
    async fn step_effects_commit_with_the_acking_settlement() {
        let (queue, store) = open_queue().await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            EffectStagingRunner::new(vec![
                Ok(StepOutcome::continue_now(b"next".to_vec())),
                Ok(StepOutcome::Succeed { result: Vec::new() }),
            ]),
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.status, TerminalStatus::Succeeded);
        assert_eq!(wait_for_kv(&queue, b"app/step-0").await, b"done");
        assert_eq!(wait_for_kv(&queue, b"app/step-1").await, b"done");

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_step_effect_is_readable_by_the_next_step() {
        struct ReadingRunner {
            read_under_staging: Arc<StdMutex<Option<Option<Vec<u8>>>>>,
        }
        impl StepRunner for ReadingRunner {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                let read = step
                    .kv
                    .get(b"app/marker")
                    .await
                    .map_err(|e| StepError::permanent(e.to_string()))?
                    .map(|b| b.to_vec());
                if step.step_number == 0 {
                    step.effects
                        .put("app/marker", b"v".to_vec())
                        .map_err(|e| StepError::permanent(e.to_string()))?;
                    let staged_read = step
                        .kv
                        .get(b"app/marker")
                        .await
                        .map_err(|e| StepError::permanent(e.to_string()))?
                        .map(|b| b.to_vec());
                    *self.read_under_staging.lock().unwrap() = Some(staged_read);
                    return Ok(StepOutcome::continue_now(Vec::new()));
                }
                Ok(StepOutcome::Succeed {
                    result: read.unwrap_or_default(),
                })
            }
        }

        let (queue, store) = open_queue().await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let read_under_staging = Arc::new(StdMutex::new(None));
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            ReadingRunner {
                read_under_staging: read_under_staging.clone(),
            },
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.status, TerminalStatus::Succeeded);
        assert_eq!(outcome.result.as_deref(), Some(b"v".as_slice()));
        assert_eq!(*read_under_staging.lock().unwrap(), Some(None));

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_run_memo_written_in_one_step_is_readable_in_the_next() {
        struct JournalRunner;
        impl StepRunner for JournalRunner {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                if step.step_number == 0 {
                    step.run_memo.put("journal", b"entry").await?;
                    return Ok(StepOutcome::continue_now(Vec::new()));
                }
                let value = step.run_memo.get("journal").await?;
                Ok(StepOutcome::Succeed {
                    result: value.unwrap_or_default(),
                })
            }
        }

        let (queue, store) = open_queue().await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime =
            WorkflowRuntime::builder(queue, store, JournalRunner, ChannelHook { tx }).build();
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.status, TerminalStatus::Succeeded);
        assert_eq!(outcome.result.as_deref(), Some(b"entry".as_slice()));

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_fail_verdict_acks_with_its_effects_and_no_dead_letter() {
        let (queue, store) = open_queue().await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            EffectStagingRunner::new(vec![Ok(StepOutcome::Fail {
                reason: "denied".to_string(),
            })]),
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.run_id, handle.run_id);
        assert_eq!(outcome.status, TerminalStatus::Failed);
        assert_eq!(outcome.error.as_deref(), Some("denied"));
        assert_eq!(
            terminal_status_of(&runtime, &handle.run_id).await,
            Some(outcome.status)
        );
        assert_eq!(wait_for_kv(&queue, b"app/step-0").await, b"done");
        assert_eq!(
            queue.view().stats("workflow-steps").await.unwrap().dead,
            0,
            "a Fail verdict must not dead-letter"
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_runner_cancelling_its_own_token_is_not_an_external_cancel() {
        // The runner receives a child of the claim's token, so firing it
        // leaves the parent uncancelled and the step's staged effects are
        // applied.
        struct SelfCancellingRunner;
        impl StepRunner for SelfCancellingRunner {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                step.cancel_token.cancel();
                step.effects
                    .put("app/step-0", b"done")
                    .map_err(|e| StepError::permanent(e.to_string()))?;
                Ok(StepOutcome::Succeed {
                    result: b"finished".to_vec(),
                })
            }
        }

        let (queue, store, _clock) = open_queue_at(10_000).await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            SelfCancellingRunner,
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.status, TerminalStatus::Succeeded);
        assert_eq!(outcome.result.as_deref(), Some(b"finished".as_slice()));
        assert_eq!(wait_for_kv(&queue, b"app/step-0").await, b"done");

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_cancel_verdict_acks_with_its_effects_and_no_dead_letter() {
        let (queue, store) = open_queue().await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            EffectStagingRunner::new(vec![Ok(StepOutcome::Cancel {
                reason: "obsolete".to_string(),
            })]),
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.run_id, handle.run_id);
        assert_eq!(outcome.status, TerminalStatus::Cancelled);
        assert_eq!(outcome.error.as_deref(), Some("obsolete"));
        assert_eq!(
            terminal_status_of(&runtime, &handle.run_id).await,
            Some(outcome.status)
        );
        assert_eq!(wait_for_kv(&queue, b"app/step-0").await, b"done");
        assert_eq!(
            queue.view().stats("workflow-steps").await.unwrap().dead,
            0,
            "a Cancel verdict must not dead-letter"
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn an_external_cancel_discards_staged_effects() {
        struct StageThenAwaitCancel {
            started: tokio::sync::mpsc::UnboundedSender<()>,
        }

        impl StepRunner for StageThenAwaitCancel {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                step.effects
                    .put(b"app/override".to_vec(), b"staged".to_vec())
                    .map_err(|e| StepError::permanent(e.to_string()))?;
                let _ = self.started.send(());
                step.cancel_token.cancelled().await;
                Ok(StepOutcome::continue_now(Vec::new()))
            }
        }

        let (queue, store) = open_queue().await;
        let (started_tx, mut started_rx) = tokio::sync::mpsc::unbounded_channel();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            StageThenAwaitCancel {
                started: started_tx,
            },
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let handle = runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        tokio::time::timeout(Duration::from_secs(2), started_rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert!(runtime.cancel(&handle.run_id).await.unwrap());

        let outcome = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.status, TerminalStatus::Cancelled);
        assert_eq!(outcome.error, None);

        wait_for_drained(&queue).await;
        assert!(
            queue
                .view()
                .kv_get(b"app/override")
                .await
                .unwrap()
                .is_none()
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_run_dead_lettered_by_the_reaper_is_terminated_by_reconciliation() {
        let (queue, store, clock) = open_queue_at_with(
            1_700_000_000_000,
            fast_options().default_queue_config(
                QueueConfig::default()
                    .retry_backoff_base(Duration::ZERO)
                    .lease_duration(Duration::from_secs(1)),
            ),
        )
        .await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime =
            WorkflowRuntime::builder(queue.clone(), store, PauseRunner, ChannelHook { tx })
                .poll_interval(Duration::from_millis(10))
                .build();
        let shutdown = spawn_runtime(runtime.clone());

        let submitted = runtime
            .submit(RunSpec {
                run_id: Some(rid("hung")),
                input: Vec::new(),
                options: RunOptions {
                    max_attempts_per_step: Some(1),
                    ..Default::default()
                },
                ..Default::default()
            })
            .await
            .unwrap();
        for _ in 0..200 {
            if queue.view().stats("workflow-steps").await.unwrap().claimed == 1 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert_eq!(
            runtime.status(&rid("hung")).await.unwrap().map(|s| s.state),
            Some(RunState::Running)
        );

        // The lease expires past the attempt limit: the reaper dead-letters
        // the step outside the worker, with no worker in the loop.
        advance(&clock, Duration::from_secs(2)).await;
        let outcome = tokio::time::timeout(Duration::from_secs(5), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.run_id, "hung");
        assert_eq!(outcome.status, TerminalStatus::Failed);
        assert_eq!(outcome.final_step, 0);
        assert_eq!(
            queue
                .view()
                .get_job(&submitted.job_id)
                .await
                .unwrap()
                .unwrap()
                .status,
            JobStatus::Dead
        );
        assert!(
            queue
                .view()
                .kv_get(&run_kv_key(&rid("hung")))
                .await
                .unwrap()
                .is_none()
        );
        assert!(
            queue
                .view()
                .kv_get(&step_kv_key(&rid("hung")))
                .await
                .unwrap()
                .is_none()
        );
        assert_eq!(
            terminal_status_of(&runtime, &rid("hung")).await,
            Some(TerminalStatus::Failed)
        );
        let _ = shutdown.send(());

        // The run id is submitted again while the dead job is retained:
        // the pass identifies a dead-letter outside the worker by the
        // current step's job, so the new run is left alone.
        let again = runtime
            .submit(RunSpec {
                run_id: Some(rid("hung")),
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        assert!(again.newly_submitted);
        assert_eq!(runtime.inner.core.reconcile_dead_steps().await.unwrap(), 0);
        assert!(
            runtime.status(&rid("hung")).await.unwrap().is_some(),
            "the re-submitted run is active"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn a_wait_follows_the_run_across_its_steps() {
        let (queue, store) = open_queue().await;
        let runtime = WorkflowRuntime::builder(
            queue,
            store,
            ScriptedRunner::new(vec![
                StepOutcome::continue_now(b"next".to_vec()),
                StepOutcome::Succeed {
                    result: b"done".to_vec(),
                },
            ]),
            NoopTerminalHook,
        )
        .build();
        assert!(matches!(
            runtime.wait(&rid("absent")).await,
            Err(Error::RunNotFound(id)) if id == "absent"
        ));
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                run_id: Some(rid("two")),
                input: b"x".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();
        let end = tokio::time::timeout(Duration::from_secs(5), runtime.wait(&rid("two")))
            .await
            .expect("the wait resolved")
            .unwrap();
        let outcome = end.outcome.expect("the worker recorded the outcome");
        assert_eq!(
            (outcome.final_step, outcome.result.as_deref()),
            (1, Some(b"done".as_slice()))
        );
        assert_eq!(
            (end.termination.status, end.termination.final_step),
            (TerminalStatus::Succeeded, 1)
        );

        // A run already terminated is reported at once.
        let again = runtime.wait(&rid("two")).await.unwrap();
        assert_eq!(again.termination, end.termination);
        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_result_record_of_an_earlier_run_is_not_reported_for_a_re_submitted_run_id() {
        let (queue, store, clock) = open_queue_at(10_000).await;
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            FixedRunner::new(Ok(StepOutcome::Succeed {
                result: b"done".to_vec(),
            })),
            NoopTerminalHook,
        )
        .build();
        let spec = RunSpec {
            run_id: Some(rid("again")),
            input: b"x".to_vec(),
            ..Default::default()
        };
        runtime.submit(spec.clone()).await.unwrap();
        let claim = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .unwrap();
        let effects = runtime
            .inner
            .process_step(&claim, &LeaseHandle::detached())
            .await
            .unwrap();
        queue.ack_with(&claim, effects).await.unwrap();
        let first = runtime.wait(&rid("again")).await.unwrap();
        assert_eq!(first.termination.status, TerminalStatus::Succeeded);
        assert!(first.outcome.is_some());

        // The run id is submitted again: the earlier run's records
        // remain until the new run's termination overwrites them.
        clock.advance(Duration::from_secs(1));
        assert!(runtime.submit(spec).await.unwrap().newly_submitted);
        assert!(
            runtime.outcome(&rid("again")).await.unwrap().is_none(),
            "the run is active"
        );
        assert!(runtime.cancel(&rid("again")).await.unwrap());
        let end = runtime.wait(&rid("again")).await.unwrap();
        assert_eq!(
            (end.termination.status, end.termination.terminated_at_ms),
            (TerminalStatus::Cancelled, 11_000)
        );
        assert!(
            end.outcome.is_none(),
            "no worker terminated the new run, so the earlier run's record is not its outcome"
        );
        assert!(runtime.outcome(&rid("again")).await.unwrap().is_none());
    }

    #[tokio::test(start_paused = true)]
    async fn a_cancel_request_does_not_reach_a_re_submission_of_the_run_id() {
        let (queue, store, _clock) = open_queue_at(10_000).await;
        let runtime =
            WorkflowRuntime::builder(queue.clone(), store, UnreachableRunner, NoopTerminalHook)
                .build();
        let spec = RunSpec {
            run_id: Some(rid("again")),
            input: b"x".to_vec(),
            ..Default::default()
        };
        runtime.submit(spec.clone()).await.unwrap();
        assert!(runtime.cancel(&rid("again")).await.unwrap());
        let end = runtime.wait(&rid("again")).await.unwrap();
        assert_eq!(end.termination.status, TerminalStatus::Cancelled);

        assert!(runtime.submit(spec).await.unwrap().newly_submitted);
        let record = runtime
            .view()
            .run_record(&rid("again"))
            .await
            .unwrap()
            .unwrap();
        assert!(!record.cancel_requested);
        let status = runtime.status(&rid("again")).await.unwrap().unwrap();
        assert_eq!(status.state, RunState::Pending);
    }

    #[tokio::test(start_paused = true)]
    async fn a_step_dead_lettered_outside_the_worker_is_waited_for_until_reconciliation() {
        let (queue, store, _clock) = open_queue_at(10_000).await;
        let runtime =
            WorkflowRuntime::builder(queue.clone(), store, UnreachableRunner, NoopTerminalHook)
                .poll_interval(Duration::from_millis(10))
                .build();
        runtime
            .submit(RunSpec {
                run_id: Some(rid("hung")),
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        let claim = queue
            .claim("workflow-steps", Duration::from_secs(60))
            .await
            .unwrap()
            .unwrap();
        queue.dead_letter(&claim, "hung").await.unwrap();

        let waiting = tokio::spawn({
            let runtime = runtime.clone();
            async move { runtime.wait(&rid("hung")).await }
        });
        assert!(
            !runtime.cancel(&rid("hung")).await.unwrap(),
            "the request is not honoured"
        );
        assert!(
            runtime
                .wait_timeout(&rid("hung"), Duration::from_secs(1))
                .await
                .unwrap()
                .is_none(),
            "nothing terminates the run without a worker"
        );
        assert_eq!(runtime.inner.core.reconcile_dead_steps().await.unwrap(), 1);
        let end = tokio::time::timeout(Duration::from_secs(5), waiting)
            .await
            .expect("the wait resolved")
            .unwrap()
            .unwrap();
        assert_eq!(
            end.termination,
            RunTermination {
                status: TerminalStatus::Failed,
                error: Some("hung".into()),
                error_kind: None,
                final_step: 0,
                terminated_at_ms: 10_000,
            },
            "the termination is read from the terminal record",
        );
        assert!(end.outcome.is_none());
        assert_eq!(
            runtime.wait(&rid("hung")).await.unwrap().termination,
            end.termination,
            "the record is retained without memo retention",
        );
    }

    #[tokio::test(start_paused = true)]
    async fn a_pointer_over_a_missing_job_is_an_inconsistent_run_state() {
        let (queue, store) = open_queue().await;
        let runtime =
            WorkflowRuntime::builder(queue.clone(), store, UnreachableRunner, NoopTerminalHook)
                .build();
        let submitted = runtime
            .submit(RunSpec {
                run_id: Some(rid("torn")),
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        // The step job is removed without the runtime, so the pointer
        // outlives it.
        queue.cancel(&submitted.job_id).await.unwrap();
        assert!(matches!(
            runtime.status(&rid("torn")).await,
            Err(Error::InconsistentRunState(id)) if id == "torn"
        ));
        assert!(matches!(
            runtime.wait(&rid("torn")).await,
            Err(Error::InconsistentRunState(id)) if id == "torn"
        ));
        assert!(matches!(
            runtime.cancel(&rid("torn")).await,
            Err(Error::InconsistentRunState(id)) if id == "torn"
        ));
    }

    #[tokio::test(start_paused = true)]
    async fn a_member_record_is_rewritten_only_by_the_terminating_settlement() {
        struct RecordReadingRunner {
            pending_seen: Arc<StdMutex<Vec<bool>>>,
        }

        impl StepRunner for RecordReadingRunner {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                let record = step
                    .kv
                    .get(&group_member_kv_key(&rid("g"), "m"))
                    .await?
                    .expect("the member record is written with the submission");
                let member: DurableMember = rmp_serde::from_slice(&record).unwrap();
                self.pending_seen
                    .lock()
                    .unwrap()
                    .push(member.terminated.is_none());
                Err(StepError::transient("still failing"))
            }
        }

        let (queue, store) = open_queue_with(fast_options()).await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let pending_seen = Arc::new(StdMutex::new(Vec::new()));
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            RecordReadingRunner {
                pending_seen: pending_seen.clone(),
            },
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        let group = runtime.group(rid("g"));
        group
            .submit(
                vec![GroupMember {
                    key: "m".to_string(),
                    input: Vec::new(),
                }],
                &RunOptions {
                    max_attempts_per_step: Some(2),
                    ..RunOptions::default()
                },
            )
            .await
            .unwrap();
        let outcome = tokio::time::timeout(Duration::from_secs(5), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(outcome.status, TerminalStatus::Failed);
        for _ in 0..200 {
            if queue.view().stats("workflow-steps").await.unwrap().dead == 1 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert_eq!(queue.view().stats("workflow-steps").await.unwrap().dead, 1);

        // The retried first attempt left the record pending; the
        // exhausted second attempt's termination commits with the
        // dead-letter.
        assert_eq!(*pending_seen.lock().unwrap(), vec![true, true]);
        let members = group.members().await.unwrap();
        assert_eq!(members.len(), 1);
        assert_eq!(members[0].key, "m");
        assert_eq!(members[0].status(), Some(TerminalStatus::Failed));
        assert_eq!(
            members[0]
                .record
                .terminated
                .as_ref()
                .unwrap()
                .error
                .as_deref(),
            Some("still failing")
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_replayed_step_outcome_restores_its_staged_effects() {
        struct StagingContinueRunner {
            calls: Arc<AtomicU32>,
        }

        impl StepRunner for StagingContinueRunner {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                self.calls.fetch_add(1, Ordering::SeqCst);
                step.effects
                    .put(b"app/replayed".to_vec(), b"v".to_vec())
                    .map_err(|e| StepError::transient(e.to_string()))?;
                step.effects
                    .delete(b"app/stale".to_vec())
                    .map_err(|e| StepError::transient(e.to_string()))?;
                Ok(StepOutcome::continue_now(b"step1".to_vec()))
            }
        }

        let (queue, store) = open_queue().await;
        let calls = Arc::new(AtomicU32::new(0));
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            StagingContinueRunner {
                calls: calls.clone(),
            },
            NoopTerminalHook,
        )
        .step_output_replay()
        .build();

        queue.kv_put(b"app/stale", b"old").await.unwrap();
        runtime
            .submit(RunSpec {
                run_id: Some(rid("replay-effects")),
                input: b"input".to_vec(),
                ..Default::default()
            })
            .await
            .unwrap();

        let job = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .unwrap();

        // First delivery: the returned effects are dropped, simulating a
        // crash between the replay-record write and the settlement.
        let _ = runtime
            .inner
            .process_step(&job, &LeaseHandle::detached())
            .await
            .unwrap();
        assert_eq!(calls.load(Ordering::SeqCst), 1);
        assert!(
            queue
                .view()
                .kv_get(b"app/replayed")
                .await
                .unwrap()
                .is_none()
        );

        // Redelivery replays the stored outcome and restores the staged
        // effects into the settlement without invoking the runner.
        let effects = runtime
            .inner
            .process_step(&job, &LeaseHandle::detached())
            .await
            .unwrap();
        assert_eq!(calls.load(Ordering::SeqCst), 1);
        assert_eq!(
            effects.kv_writes.get(b"app/replayed".as_slice()),
            Some(&b"v".to_vec())
        );
        assert!(effects.kv_deletes.contains(&b"app/stale".to_vec()));
        queue.ack_with(&job, effects).await.unwrap();
        assert_eq!(
            queue
                .view()
                .kv_get(b"app/replayed")
                .await
                .unwrap()
                .as_deref(),
            Some(b"v".as_slice())
        );
        assert!(queue.view().kv_get(b"app/stale").await.unwrap().is_none());
    }

    #[tokio::test(start_paused = true)]
    async fn only_the_committed_outcome_produces_a_notification() {
        struct GatedSecondAttempt {
            calls: Arc<AtomicU32>,
            running: tokio::sync::mpsc::UnboundedSender<()>,
        }

        impl StepRunner for GatedSecondAttempt {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                if self.calls.fetch_add(1, Ordering::SeqCst) == 0 {
                    return Ok(StepOutcome::Succeed {
                        result: b"done".to_vec(),
                    });
                }
                let _ = self.running.send(());
                step.cancel_token.cancelled().await;
                Ok(StepOutcome::Succeed {
                    result: b"done".to_vec(),
                })
            }
        }

        let (queue, store) = open_queue().await;
        let (running_tx, mut running_rx) = tokio::sync::mpsc::unbounded_channel();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            GatedSecondAttempt {
                calls: Arc::new(AtomicU32::new(0)),
                running: running_tx,
            },
            ChannelHook { tx },
        )
        .build();

        runtime
            .submit(RunSpec {
                run_id: Some(rid("phantom")),
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        let job = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .unwrap();

        // First settlement attempt: the runner succeeds, but the effects
        // are dropped, as when the settlement loses the claim. The
        // Succeeded notification is dropped with them.
        let _ = runtime
            .inner
            .process_step(&job, &queue.lease_handle(&job))
            .await
            .unwrap();

        // The redelivered attempt observes an external cancel and
        // commits Cancelled.
        let worker = {
            let inner = runtime.inner.clone();
            let queue = queue.clone();
            tokio::spawn(async move {
                let effects = inner
                    .process_step(&job, &queue.lease_handle(&job))
                    .await
                    .unwrap();
                queue.ack_with(&job, effects).await.unwrap();
            })
        };
        running_rx.recv().await.unwrap();
        assert!(runtime.cancel(&rid("phantom")).await.unwrap());
        worker.await.unwrap();

        let notification = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .expect("the committed settlement enqueued its notification");
        let effects = runtime
            .inner
            .process_step(&notification, &LeaseHandle::detached())
            .await
            .unwrap();
        queue.ack_with(&notification, effects).await.unwrap();
        let outcome = rx.recv().await.unwrap();
        assert_eq!(outcome.status, TerminalStatus::Cancelled);
        assert!(
            queue
                .claim("workflow-steps", Duration::from_secs(30))
                .await
                .unwrap()
                .is_none(),
            "the outcome that never committed must produce no notification",
        );
        assert!(rx.try_recv().is_err());
    }

    #[tokio::test(start_paused = true)]
    async fn hook_effects_commit_with_the_notification_ack() {
        struct EffectHook;

        impl TerminalHook for EffectHook {
            async fn on_termination(
                &self,
                outcome: &RunOutcome,
                effects: &TerminalEffects,
            ) -> std::result::Result<(), StepError> {
                effects
                    .put(
                        format!("app/outcomes/{}", outcome.run_id),
                        outcome.status.as_str(),
                    )
                    .map_err(|e| StepError::permanent(e.to_string()))?;
                effects
                    .enqueue(EnqueueRequest {
                        queue: "side-effects".to_string(),
                        payload: outcome.run_id.to_string().into_bytes(),
                        options: EnqueueOptions::default(),
                    })
                    .map_err(|e| StepError::permanent(e.to_string()))?;
                Ok(())
            }
        }

        let (queue, store) = open_queue().await;
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            ScriptedRunner::new(vec![StepOutcome::Succeed { result: Vec::new() }]),
            EffectHook,
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                run_id: Some(rid("hooked")),
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();

        assert_eq!(
            wait_for_kv(&queue, b"app/outcomes/hooked").await,
            b"succeeded"
        );
        let side = queue
            .claim("side-effects", Duration::from_secs(30))
            .await
            .unwrap()
            .expect("the staged enqueue committed with the notification ack");
        assert_eq!(side.payload.as_slice(), b"hooked");

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_transiently_failing_hook_retries_the_notification() {
        struct FlakyHook {
            calls: Arc<AtomicU32>,
        }

        impl TerminalHook for FlakyHook {
            async fn on_termination(
                &self,
                outcome: &RunOutcome,
                effects: &TerminalEffects,
            ) -> std::result::Result<(), StepError> {
                if self.calls.fetch_add(1, Ordering::SeqCst) == 0 {
                    return Err(StepError::transient("first attempt fails"));
                }
                effects
                    .put(format!("app/notified/{}", outcome.run_id), b"1".to_vec())
                    .map_err(|e| StepError::permanent(e.to_string()))?;
                Ok(())
            }
        }

        let (queue, store) = open_queue_with(fast_options()).await;
        let calls = Arc::new(AtomicU32::new(0));
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            ScriptedRunner::new(vec![StepOutcome::Succeed { result: Vec::new() }]),
            FlakyHook {
                calls: calls.clone(),
            },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                run_id: Some(rid("flaky")),
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();

        assert_eq!(wait_for_kv(&queue, b"app/notified/flaky").await, b"1");
        assert_eq!(calls.load(Ordering::SeqCst), 2);

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_noop_hook_enqueues_no_notification() {
        let (queue, store) = open_queue().await;
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            ScriptedRunner::new(vec![StepOutcome::Succeed { result: Vec::new() }]),
            NoopTerminalHook,
        )
        .build();
        runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        let job = queue
            .claim("workflow-steps", Duration::from_secs(30))
            .await
            .unwrap()
            .unwrap();
        let effects = runtime
            .inner
            .process_step(&job, &LeaseHandle::detached())
            .await
            .unwrap();
        assert!(effects.enqueues.is_empty());
        queue.ack_with(&job, effects).await.unwrap();
        assert!(
            queue
                .claim("workflow-steps", Duration::from_secs(30))
                .await
                .unwrap()
                .is_none()
        );
    }

    #[cfg(feature = "webhooks")]
    #[tokio::test(start_paused = true)]
    async fn the_webhook_hook_stages_its_delivery_as_a_notification_effect() {
        use crate::terminal::WebhookTerminalHook;

        let (queue, store) = open_queue().await;
        let runtime = WorkflowRuntime::builder(
            queue.clone(),
            store,
            ScriptedRunner::new(vec![
                StepOutcome::Succeed {
                    result: b"payload".to_vec(),
                },
                StepOutcome::Succeed { result: Vec::new() },
            ]),
            WebhookTerminalHook::new("callbacks"),
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                run_id: Some(rid("with-callback")),
                input: Vec::new(),
                options: RunOptions {
                    headers: HashMap::from([(
                        "callback_url".to_string(),
                        "https://example.com/done".to_string(),
                    )]),
                    ..Default::default()
                },
                ..Default::default()
            })
            .await
            .unwrap();

        let webhook = loop {
            if let Some(job) = queue
                .claim("callbacks", Duration::from_secs(30))
                .await
                .unwrap()
            {
                break job;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        };
        assert_eq!(webhook.payload.as_slice(), b"payload");
        assert_eq!(
            webhook.headers.get("webhook.url").unwrap(),
            "https://example.com/done"
        );
        assert_eq!(
            webhook.headers.get("http.Workflow-Run-Status").unwrap(),
            "succeeded"
        );

        // A run without a callback header enqueues no notification.
        runtime
            .submit(RunSpec {
                run_id: Some(rid("without-callback")),
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        wait_for_drained(&queue).await;
        assert!(
            queue
                .claim("callbacks", Duration::from_secs(30))
                .await
                .unwrap()
                .is_none()
        );

        let _ = shutdown.send(());
    }

    #[tokio::test(start_paused = true)]
    async fn a_late_write_through_an_escaped_handle_is_refused() {
        struct EscapingRunner {
            escaped: Arc<StdMutex<Option<EffectsHandle>>>,
        }

        impl StepRunner for EscapingRunner {
            async fn run_step(&self, step: &Step) -> std::result::Result<StepOutcome, StepError> {
                *self.escaped.lock().unwrap() = Some(step.effects.clone());
                Ok(StepOutcome::Succeed { result: Vec::new() })
            }
        }

        let (queue, store) = open_queue().await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let escaped = Arc::new(StdMutex::new(None));
        let runtime = WorkflowRuntime::builder(
            queue,
            store,
            EscapingRunner {
                escaped: escaped.clone(),
            },
            ChannelHook { tx },
        )
        .build();
        let shutdown = spawn_runtime(runtime.clone());

        runtime
            .submit(RunSpec {
                input: Vec::new(),
                ..Default::default()
            })
            .await
            .unwrap();
        tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        let handle = escaped.lock().unwrap().take().unwrap();
        assert!(matches!(
            handle.put("app/late", "v"),
            Err(Error::EffectsSealed)
        ));

        let _ = shutdown.send(());
    }
}