//! Postgres-backed saga runner — the only runner. The architectural
//! premise is documented in `tasks/postgres-runner.md`:
//!
//! - A saga instance is a mire aggregate; state IS the event stream.
//! - `EventStore::save` / `EventStore::load` is the only durability
//!   primitive needed for instance state.
//! - The routing projection (`es_saga_awaits`) maps an arriving
//!   response event → the suspended step that awaits it.
//!
//! **Current capabilities:**
//! - `trigger` → drive to quiescence (pure-decide via [`crate::decide`]):
//!   gating, skip-cascade, MergeKind::Any.
//! - Emit + in-process effect dispatch (Regime A): a step emits a
//!   request, a registered `.effect(...)` fulfills it inline, the saga
//!   continues in the same call.
//! - Emit + external worker (Regime B): a step emits a request with no
//!   in-process handler → the saga SUSPENDS, persisting the await. A
//!   later [`PgSagaRunner::deliver`] routes the external response back,
//!   re-runs the (pure) handler to recover the map closure, and resumes.
//!   This survives process restarts — deliver can come from any replica.
//! - Compensation cascade on handler `Result::Err`, including comp
//!   handlers that emit + await.
//! - Signals via [`PgSagaRunner::signal`]: `TriggersCompensation`
//!   (admin-cancel) and `RoutesToStep` (decision delivered as the
//!   awaited event). Rejects signals to terminal / compensating sagas
//!   and unregistered signal types.
//! - Timeouts: a suspending step's effective timeout (step `.timeout(d)`
//!   or saga `default_step_timeout`) is persisted as `deadline_at` on
//!   the await row. [`PgSagaRunner::tick_timeouts`] expires overdue
//!   awaits — fails the step with `StepError::Timeout` and cascades.
//!   `now` is passed explicitly so tests can force expiry deterministically.
//!
//! **Deferred:** the lease-driven worker loop that calls `tick_timeouts`
//! and drains signals/emits autonomously across replicas; suspend
//! within a compensation emit (comp awaits are still in-process only);
//! an outbox projection to auto-drain emits; multi-shard parallelism.

use mire::{Aggregate, AggregateRoot, EventStore, EventStoreError};
use serde::Serialize;
use tracing::{debug, instrument, trace};

use crate::builder::{SagaBuilder, SignalPolicy, Terminal};
use crate::decide::{NextStep, next_step};
use crate::envelope::{SagaEnvelope, SagaEnvelopeEvent, SagaInstanceStatus, SagaStepStatus};
use crate::extract::ExtractError;
use crate::inspect::{AwaitDescription, EmittedEvent, InstanceDiagnostic};
use crate::result::StepError;
use crate::saga::Saga;
use crate::step::{StepContext, StepName};

/// Postgres-backed saga runner. Persists envelope events through
/// [`mire::EventStore`]; uses OCC (via `ExpectedVersion::Exact`) for
/// replica safety.
///
/// **Scope:** trigger + drive-to-quiescence including emit + in-process
/// handler dispatch (Regime A of the plan). Crash-recovery via the
/// outbox projection (Regime B), signals, timeouts, and compensation
/// land in subsequent slices.
pub struct PgSagaRunner<S: Saga> {
    topology: SagaBuilder<S>,
    store: EventStore,
    num_shards: u32,
    /// Identity recorded as the holder in `es_saga_leases`. Defaults to a
    /// per-runner UUID; override with [`with_worker_id`](Self::with_worker_id)
    /// for a stable replica identity (e.g. `POD_NAME`).
    worker_id: String,
    /// How long a per-instance processing lease is valid without renewal.
    /// A slow effect heartbeats it (renewed every `ttl/3`), so it bounds
    /// **takeover latency after a crash**, not the longest call — recovery is
    /// lease-expiry-driven (phase 7), so this is also the recovery latency.
    /// Default 5s; tests may use a shorter value. Heartbeat + fence make a
    /// short TTL safe: a slow-but-alive holder is renewed, and a stolen-from
    /// holder is fenced out of any late commit.
    lease_ttl: std::time::Duration,
    /// Optional sink for external (Regime-B) emits drained from
    /// `es_saga_outbox` (P3.2). `None` ⇒ outbox rows accumulate undelivered
    /// (a durable audit; external workers may poll `es_saga_awaits`/envelopes
    /// directly instead). Set via [`with_publisher`](Self::with_publisher).
    publisher: Option<std::sync::Arc<dyn crate::publisher::EmitPublisherDyn>>,
}

impl<S: Saga> PgSagaRunner<S> {
    /// Validate the topology and bind the runner to an event store.
    /// Panics on invalid topology.
    ///
    /// This direct-call runner uses a single shard: the host drives the
    /// lifecycle by calling `trigger` / `deliver` / `signal` /
    /// `tick_timeouts` itself, so there are no competing workers to
    /// distribute across. The shard id is still recorded on await /
    /// signal rows so the multi-shard, lease-driven worker loop (plan
    /// §7.1–§7.2a — over-provisioned fixed shards + HRW assignment,
    /// default 64) can be layered on without a schema change.
    pub fn new(topology: SagaBuilder<S>, store: EventStore) -> Self {
        topology.assert_valid();
        Self {
            topology,
            store,
            num_shards: 1,
            worker_id: uuid::Uuid::new_v4().to_string(),
            lease_ttl: std::time::Duration::from_secs(5),
            publisher: None,
        }
    }

    /// Attach an [`EmitPublisher`](crate::EmitPublisher) that delivers this
    /// saga's external (Regime-B) emits drained from `es_saga_outbox` (P3.2).
    /// Without one, outbox rows accumulate undelivered. A
    /// [`SagaWorker`](crate::SagaWorker) wrapping this runner drains + publishes
    /// them on its drain tick; delivery is at-least-once.
    pub fn with_publisher<P: crate::publisher::EmitPublisher>(mut self, publisher: P) -> Self {
        self.publisher = Some(std::sync::Arc::new(crate::publisher::PublisherAdapter(
            publisher,
        )));
        self
    }

    /// Whether an [`EmitPublisher`](crate::EmitPublisher) is configured — the
    /// worker skips the outbox drain entirely when not.
    pub fn has_publisher(&self) -> bool {
        self.publisher.is_some()
    }

    /// Set this runner's worker identity (the `es_saga_leases` holder). Use a
    /// stable per-replica string if you want lease ownership to follow a
    /// specific replica; otherwise the default per-runner UUID is fine.
    pub fn with_worker_id(mut self, worker_id: impl Into<String>) -> Self {
        self.worker_id = worker_id.into();
        self
    }

    /// Override the per-instance processing-lease TTL. Lower bounds takeover
    /// latency after a crash; must stay above the longest single call's
    /// processing time. See [`lease_ttl`](Self#structfield.lease_ttl).
    pub fn with_lease_ttl(mut self, ttl: std::time::Duration) -> Self {
        self.lease_ttl = ttl;
        self
    }

    /// Set the shard count used to partition this saga's work
    /// (`shard_of(instance_id) = FNV-1a(instance_id) % num_shards`). Every
    /// row this runner writes (awaits, signals) is stamped with the
    /// resulting `shard_id`, and the [`SagaWorker`](crate::SagaWorker) fleet
    /// distributes its scans across shards (plan §P4.1–P4.3).
    ///
    /// **This is a deployment-wide constant.** Every runner that *stamps*
    /// rows (any process calling `trigger`/`deliver`/`signal`) and every
    /// worker that *scans* them must agree on `num_shards` — a worker
    /// configured with fewer shards than the rows were stamped under would
    /// never scan the high shards, stranding their timeouts. Workers verify
    /// this at startup via [`verify_shard_config`](Self::verify_shard_config)
    /// and refuse to run on a mismatch. Changing `num_shards` after rows
    /// exist requires a documented migration that rewrites `shard_id`; there
    /// is no online resharding (Phase 4 §Invariants). Pinned FNV-1a means a
    /// given `(instance_id, num_shards)` always maps to the same shard.
    ///
    /// Default is `1` (host-driven single runner — every instance maps to
    /// shard 0, current behavior). Pick an over-provisioned fixed value
    /// (e.g. 64) when running a multi-worker fleet so shards can be spread
    /// across replicas without resharding as the fleet grows.
    ///
    /// # Panics
    /// If `num_shards == 0` (a programming error, like an invalid topology).
    pub fn with_num_shards(mut self, num_shards: u32) -> Self {
        assert!(num_shards >= 1, "num_shards must be >= 1");
        self.num_shards = num_shards;
        self
    }

    /// The shard count this runner stamps rows under. See
    /// [`with_num_shards`](Self::with_num_shards).
    pub fn num_shards(&self) -> u32 {
        self.num_shards
    }

    /// Fail loudly if any persisted row was stamped under a *larger* shard
    /// count than this runner is configured for — i.e. some `shard_id` is
    /// `>= num_shards`. That means the fleet disagrees on `num_shards`
    /// (plan §P4.1 startup guard): a worker with this config would silently
    /// never scan the out-of-range shards, stranding their timeouts/signals.
    /// Cheap (`SELECT max(shard_id)` per table, both already indexed by
    /// `shard_id`). Called by [`SagaWorker::run`](crate::SagaWorker::run) at
    /// startup; safe to call directly before driving a runner too.
    pub async fn verify_shard_config(&self) -> Result<(), PgRunnerError> {
        // Literal SQL per table (sqlx requires &'static str): the two tables
        // that carry shard_id today. Both queries hit the shard_id-leading
        // indexes so max() is a cheap index scan.
        let checks: [(&'static str, &'static str); 2] = [
            (
                "es_saga_awaits",
                "SELECT max(shard_id) FROM es_saga_awaits WHERE saga_name = $1",
            ),
            (
                "es_saga_signals",
                "SELECT max(shard_id) FROM es_saga_signals WHERE saga_name = $1",
            ),
        ];
        for (table, sql) in checks {
            let max: Option<i32> = sqlx::query_scalar(sql)
                .bind(S::name())
                .fetch_one(self.store.pool())
                .await
                .map_err(|e| PgRunnerError::Store(e.into()))?;
            if let Some(max) = max
                && max as i64 >= self.num_shards as i64
            {
                return Err(PgRunnerError::ShardConfigMismatch {
                    table,
                    max_shard_id: max,
                    num_shards: self.num_shards,
                });
            }
        }
        Ok(())
    }

    /// This runner's worker identity (the lease holder). See
    /// [`with_worker_id`](Self::with_worker_id).
    pub fn worker_id(&self) -> &str {
        &self.worker_id
    }

    /// Run one shard-ownership convergence tick for this worker (plan §P4.2):
    /// heartbeat presence, renew held shards, and move toward an even share of
    /// the saga's `num_shards`. Returns the shards owned after the tick — the
    /// set [`SagaWorker`](crate::SagaWorker) filters its scans by. See
    /// [`shard_lease::rebalance`](crate::shard_lease::rebalance); `ttl`,
    /// `liveness_window`, and `num_shards` are fleet-wide constants.
    pub async fn rebalance_shards(
        &self,
        ttl: std::time::Duration,
        liveness_window: std::time::Duration,
        release_excess: bool,
    ) -> Result<Vec<i32>, PgRunnerError> {
        crate::shard_lease::rebalance(
            self.store.pool(),
            S::name(),
            &self.worker_id,
            self.num_shards,
            ttl,
            liveness_window,
            release_excess,
        )
        .await
        .map_err(PgRunnerError::Store)
    }

    /// Relinquish all shard ownership and fleet membership for this worker
    /// (graceful shutdown) so survivors reabsorb the shards on their next tick
    /// rather than waiting out the TTL/window.
    pub async fn release_shards(&self) -> Result<(), PgRunnerError> {
        crate::shard_lease::release_all(self.store.pool(), S::name(), &self.worker_id)
            .await
            .map_err(PgRunnerError::Store)?;
        crate::shard_lease::deregister_worker(self.store.pool(), S::name(), &self.worker_id)
            .await
            .map_err(PgRunnerError::Store)
    }

    /// Try once to acquire this instance's processing lease.
    /// `Ok(Some(fence))` = we hold it at `fence`; `Ok(None)` = a live worker
    /// holds it. The fence must be passed to every heartbeat/release for this
    /// hold so a peer that steals the lease (bumping the fence) preempts us.
    #[instrument(level = "debug", skip(self), fields(saga = S::name(), instance_id))]
    async fn try_acquire_lease(&self, instance_id: &str) -> Result<Option<i64>, PgRunnerError> {
        let outcome = crate::instance_lease::acquire(
            self.store.pool(),
            S::name(),
            instance_id,
            &self.worker_id,
            self.lease_ttl,
        )
        .await
        .map_err(PgRunnerError::Store)?;
        Ok(match outcome {
            crate::instance_lease::AcquireOutcome::Acquired { fence } => {
                trace!(saga = S::name(), instance_id, fence, "lease acquired");
                Some(fence)
            }
            crate::instance_lease::AcquireOutcome::Contended => {
                trace!(saga = S::name(), instance_id, "lease contended");
                None
            }
        })
    }

    /// Acquire the lease, waiting if a peer holds it. A live holder releases
    /// within ms (its call is short); a crashed holder's lease lapses within
    /// the TTL, after which we take over (at-least-once). Contention is thus
    /// transparent to the caller: it serializes, never fails. Errors
    /// `Contended` only if the lease never frees within the bound (a holder
    /// stuck longer than the TTL without heartbeating — not expected in the
    /// host-driven model). Backoff is fixed 25ms; the bound spans ~2× TTL.
    #[instrument(level = "debug", skip(self), fields(saga = S::name(), instance_id))]
    async fn acquire_lease_blocking(&self, instance_id: &str) -> Result<i64, PgRunnerError> {
        let backoff = std::time::Duration::from_millis(25);
        let max_waits = ((self.lease_ttl.as_millis() * 2 / backoff.as_millis()).max(1)) as u32;
        for _ in 0..=max_waits {
            if let Some(fence) = self.try_acquire_lease(instance_id).await? {
                return Ok(fence);
            }
            tokio::time::sleep(backoff).await;
        }
        Err(PgRunnerError::Contended {
            instance_id: instance_id.to_string(),
        })
    }

    /// The bare instance id from a loaded envelope root. The saga's stream is
    /// `format!("{}-{}", S::name(), instance_id)`, so strip the category
    /// prefix. Used to build the [`crate::IdempotencyKey`] handed to handlers.
    fn instance_id_of(root: &AggregateRoot<SagaEnvelope<S>>) -> &str {
        let prefix = S::name().len() + 1; // "<name>-"
        root.stream_id
            .get(prefix..)
            .unwrap_or(root.stream_id.as_str())
    }

    /// Release this instance's processing lease (best-effort) so a peer can
    /// take over immediately rather than waiting out the TTL. Fence-guarded so
    /// we never free a lease a takeover worker now holds.
    #[instrument(level = "debug", skip(self), fields(saga = S::name(), instance_id, fence))]
    async fn release_lease(&self, instance_id: &str, fence: i64) {
        trace!(saga = S::name(), instance_id, fence, "lease released");
        let _ = crate::instance_lease::release(
            self.store.pool(),
            S::name(),
            instance_id,
            &self.worker_id,
            fence,
        )
        .await;
    }

    /// Spawn a background heartbeat that renews this instance's lease while
    /// an effect runs, then is stopped via [`LeaseHeartbeat::stop`].
    ///
    /// The instance lease (effects-vs-commit §11) gives at-most-one-active
    /// processor per instance, but it is only valid for `lease_ttl`. A step
    /// or request handler is an arbitrary user `async fn` — a payment API
    /// round-trip, say — and one that runs longer than the TTL would let the
    /// lease lapse, after which a second worker could take over and run the
    /// **same effects concurrently**. OCC fences the *log* (the stale
    /// worker's commit conflicts and is discarded), but not the side effects
    /// that already fired. The heartbeat closes that hole: as long as this
    /// worker is genuinely alive and working, it keeps renewing the lease, so
    /// no peer ever sees it as expired. If a heartbeat comes back
    /// preempted/errored (we somehow already lost it), it stops and the OCC
    /// commit remains the backstop.
    ///
    /// Run `effect` to completion while keeping this instance's lease
    /// renewed, and return its output.
    ///
    /// The renewal runs in the **same future** as `effect` (via `select!`),
    /// not a detached `tokio::spawn`. That matters for crash safety: if the
    /// caller is cancelled mid-operation — a node dropped while a deliver is
    /// in flight, exactly what the chaos test does — the whole future is
    /// dropped, the heartbeat stops with it, and its connection is returned,
    /// so the lease lapses on schedule and a peer can take the instance over.
    /// A detached task would instead keep renewing a dead node's lease
    /// forever and block takeover.
    ///
    /// The heartbeat connection is acquired **lazily**, only when the first
    /// renewal is actually due (`ttl/3` in). The vast majority of effects (a
    /// saga transition commits in milliseconds) finish first, so they never
    /// take a second connection — no steady-state pool overhead. Only a
    /// genuinely slow handler — the case the heartbeat exists for — grabs a
    /// connection. If renewal can't keep up (no connection, or we've already
    /// been preempted), it stops and OCC remains the stale-commit backstop;
    /// `effect` still runs to completion.
    #[instrument(level = "debug", skip(self, effect), fields(saga = S::name(), instance_id, fence))]
    async fn run_with_heartbeat<T, Fut>(&self, instance_id: &str, fence: i64, effect: Fut) -> T
    where
        Fut: std::future::Future<Output = T>,
    {
        tokio::pin!(effect);
        let interval = (self.lease_ttl / 3).max(std::time::Duration::from_millis(1));
        let mut tick = tokio::time::interval(interval);
        tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        tick.tick().await; // consume the immediate-fire tick

        let mut conn: Option<sqlx::pool::PoolConnection<sqlx::Postgres>> = None;
        let mut beating = true;
        loop {
            tokio::select! {
                // `effect` always wins on completion — the heartbeat branch
                // never returns from this fn, it only renews.
                out = &mut effect => return out,
                _ = tick.tick(), if beating => {
                    if conn.is_none() {
                        match self.store.pool().acquire().await {
                            Ok(c) => conn = Some(c),
                            Err(e) => {
                                tracing::warn!(
                                    instance = %instance_id,
                                    error = %e,
                                    "saga lease heartbeat: no connection; lease may lapse \
                                     under a slow handler (OCC remains the backstop)",
                                );
                                beating = false;
                                continue;
                            }
                        }
                    }
                    let c = conn.as_mut().expect("just set");
                    match crate::instance_lease::heartbeat(
                        &mut **c, S::name(), instance_id, &self.worker_id, fence, self.lease_ttl,
                    ).await {
                        Ok(true) => {}
                        Ok(false) => {
                            tracing::warn!(
                                instance = %instance_id,
                                "saga lease heartbeat: preempted (a peer took the lease)",
                            );
                            beating = false;
                            conn = None;
                        }
                        Err(e) => {
                            tracing::warn!(
                                instance = %instance_id,
                                error = %e,
                                "saga lease heartbeat error; stopping (OCC remains the backstop)",
                            );
                            beating = false;
                            conn = None;
                        }
                    }
                }
            }
        }
    }

    /// Try to start a fresh instance from a domain event. Returns:
    /// - `Ok(Some(id))` — instance started (or already existed; idempotent)
    /// - `Ok(None)` — this aggregate isn't the trigger, or the correlator
    ///   returned None for this event
    /// - `Err(_)` — IO or topology error
    #[instrument(level = "debug", skip(self, event), fields(saga = S::name()))]
    pub async fn trigger<A>(&self, event: &A::Event) -> Result<Option<String>, PgRunnerError>
    where
        A: Aggregate,
        A::Event: Serialize,
    {
        let trigger_json = serde_json::to_value(event).map_err(PgRunnerError::Serialise)?;

        let Some(hook) = self.topology.start_correlator() else {
            return Err(PgRunnerError::NoStartOn);
        };
        if hook.aggregate_category != A::stream_category() {
            return Ok(None);
        }
        let Some(start) = (hook.correlate)(&trigger_json) else {
            return Ok(None);
        };

        // Idempotency: an existing instance with this id reflects an
        // earlier trigger, so a re-trigger is a no-op returning that id.
        if self
            .store
            .load::<SagaEnvelope<S>>(&start.instance_id)
            .await
            .map_err(PgRunnerError::Store)?
            .is_some()
        {
            debug!(saga = S::name(), instance_id = %start.instance_id, "duplicate trigger — instance exists");
            return Ok(Some(start.instance_id));
        }

        let mut root = AggregateRoot::<SagaEnvelope<S>>::new(&start.instance_id);
        let state_json = serde_json::to_value(&start.state).map_err(PgRunnerError::Serialise)?;
        root.record(SagaEnvelopeEvent::Started {
            state: state_json,
            trigger: Some(trigger_json.clone()),
        });

        // RESERVE (C3/SAGA-4): commit `Started` alone BEFORE driving any
        // steps. OCC (`ExpectedVersion::NoStream`, derived from version 0)
        // elects a single winner among concurrent duplicate triggers — which
        // happen under at-least-once domain-event delivery. The loser
        // conflicts here and returns without running ANY step effect, so two
        // replicas can't both execute the start steps. The load check above
        // is the fast path; this commit is the actual race fence.
        match self.commit(&start.instance_id, &mut root, &[], &[]).await {
            Ok(()) => {
                debug!(saga = S::name(), instance_id = %start.instance_id, "reserved instance (start drive begins)");
            }
            Err(PgRunnerError::Store(EventStoreError::ConcurrencyConflict { .. })) => {
                // Another trigger reserved this instance first — idempotent.
                debug!(saga = S::name(), instance_id = %start.instance_id, "duplicate trigger — instance exists");
                return Ok(Some(start.instance_id));
            }
            Err(e) => return Err(e),
        }

        // Only the reservation winner reaches here, so the downstream
        // Regime-A effects run on a single driver. Hold the instance lease
        // across the start drive (review H-2): the reserve fences *duplicate
        // triggers*, but without a lease a concurrent `recover()` could steal
        // the just-`Started` (unleased, no-await) instance and drive the same
        // start effects in parallel. Acquiring the lease — and heartbeating it
        // across a slow inline effect — closes that trigger/recover race. (A
        // crash between the two commits leaves the instance at `Started` with
        // no steps; lease-expiry recovery re-drives it.)
        let inst = &start.instance_id;
        let fence = self.acquire_lease_blocking(inst).await?;
        let result = self
            .run_with_heartbeat(inst, fence, async {
                let pending = self.drive_to_quiescence(&mut root, &trigger_json).await?;
                match self.commit(inst, &mut root, &pending, &[]).await {
                    Ok(()) => {}
                    // In the microsecond window between the reserve commit and
                    // our lease acquire above, a lease-expiry recovery may have
                    // stolen and driven this instance forward. Its commit
                    // advanced the version, so ours conflicts — idempotent, the
                    // saga is already being driven; our redundant inline effects
                    // dedupe on the IdempotencyKey.
                    Err(PgRunnerError::Store(EventStoreError::ConcurrencyConflict { .. })) => {}
                    Err(e) => return Err(e),
                }
                Ok::<(), PgRunnerError>(())
            })
            .await;
        self.release_lease(inst, fence).await;
        result?;

        // A step may have suspended awaiting a signal that already arrived
        // early (review SAGA-10) — deliver any such buffered signal now.
        self.drain_buffered_signals(&start.instance_id).await?;

        Ok(Some(start.instance_id))
    }

    /// Refuse to act on an instance whose persisted `Started` state no
    /// longer decodes into the current `S::State` (schema drift). Returns
    /// [`PgRunnerError::StateDecodeFailed`] rather than letting the runner
    /// operate on a `Default` state (review C6/DX-3/SAGA-17).
    fn ensure_state_decoded(
        instance_id: &str,
        envelope: &SagaEnvelope<S>,
    ) -> Result<(), PgRunnerError> {
        match envelope.state_decode_error() {
            Some(reason) => Err(PgRunnerError::StateDecodeFailed {
                instance_id: instance_id.to_string(),
                reason: reason.to_string(),
            }),
            None => Ok(()),
        }
    }

    /// Reload an instance via `EventStore::load`. Convenience accessor
    /// for tests + tooling.
    pub async fn load(&self, instance_id: &str) -> Result<Option<SagaEnvelope<S>>, PgRunnerError> {
        Ok(self
            .store
            .load::<SagaEnvelope<S>>(instance_id)
            .await
            .map_err(PgRunnerError::Store)?
            .map(|root| root.state))
    }

    /// Deliver an external response event to a suspended instance.
    ///
    /// Routes the event to whichever step is awaiting `event_type` via the
    /// `es_saga_awaits` projection, re-runs that step's handler to
    /// recover its map closure (handlers are pure, so re-running is
    /// free and yields an equivalent closure), applies the map to the
    /// delivered payload, records `StepCompleted`, then resumes
    /// driving the saga.
    ///
    /// Returns `Ok(true)` if a step consumed the event, `Ok(false)` if
    /// no step was awaiting this event type on this instance.
    #[instrument(level = "debug", skip(self, event), fields(saga = S::name(), instance_id, event_type = E::EVENT_TYPE))]
    pub async fn deliver<E>(&self, instance_id: &str, event: &E) -> Result<bool, PgRunnerError>
    where
        E: crate::EventTypeStatic + Serialize,
    {
        let event_type = E::EVENT_TYPE;
        let payload = serde_json::to_value(event).map_err(PgRunnerError::Serialise)?;
        self.deliver_raw(instance_id, event_type, payload).await
    }

    /// Lower-level delivery taking the event-type string + JSON payload
    /// explicitly. The typed [`deliver`](Self::deliver) wraps this.
    ///
    /// Concurrency: the per-instance processing lease (effects-vs-commit §11)
    /// ensures effects never run concurrently — a second worker delivering to
    /// the same instance while we hold the lease gets
    /// [`PgRunnerError::Contended`] and backs off. The OCC check on the event
    /// append is the stale-commit backstop. If the await is already gone —
    /// the response was delivered earlier — this returns `Ok(false)`.
    #[instrument(level = "debug", skip(self, payload), fields(saga = S::name(), instance_id, event_type))]
    pub async fn deliver_raw(
        &self,
        instance_id: &str,
        event_type: &str,
        payload: serde_json::Value,
    ) -> Result<bool, PgRunnerError> {
        let result = self
            .deliver_leased_no_drain(instance_id, event_type, payload)
            .await?;
        // This delivery may have suspended a downstream step awaiting a
        // signal that already arrived early — deliver any such buffered
        // signal now (review SAGA-10). `drain_buffered_signals` calls back
        // into `deliver_leased_no_drain` (not this method), so no recursion.
        self.drain_buffered_signals(instance_id).await?;
        Ok(result)
    }

    /// The failure-side mirror of [`deliver`](Self::deliver): an out-of-process
    /// fulfiller that cannot satisfy a step's emitted request (a *permanent*
    /// provider error, not a retryable one) calls this to **fail** the awaiting
    /// forward step `step`, recording `StepFailed` and running the compensation
    /// cascade — exactly as an in-process handler returning `Err` would. Routes
    /// by step name (the fulfiller knows which step it was fulfilling — e.g.
    /// from the outbox message's `step`).
    ///
    /// Returns `Ok(true)` if the step was failed, `Ok(false)` if nothing on the
    /// instance was awaiting under `step` (already completed/failed/delivered) —
    /// an idempotent no-op, so a retried failure delivery is safe.
    ///
    /// Only **forward** awaits are supported. Compensation dispatch is in-process
    /// (the cascade requires registered handlers), so there is deliberately no
    /// external compensation-failure path — a compensation effect reports its
    /// own failure in-process.
    #[instrument(level = "debug", skip(self, reason), fields(saga = S::name(), instance_id, step))]
    pub async fn deliver_failure(
        &self,
        instance_id: &str,
        step: &str,
        reason: impl Into<String>,
    ) -> Result<bool, PgRunnerError> {
        let reason = reason.into();
        // Same lease + heartbeat discipline as `deliver`: hold the per-instance
        // lease for the whole fail-and-cascade so it can't race another
        // deliver/tick/signal on this instance.
        let fence = self.acquire_lease_blocking(instance_id).await?;
        let effect = self.fail_await_locked(instance_id, step, &reason);
        let result = self.run_with_heartbeat(instance_id, fence, effect).await;
        if !matches!(result, Ok(true)) {
            self.release_lease(instance_id, fence).await;
        }
        result
    }

    /// One delivery, under the instance lease, WITHOUT the post-delivery
    /// buffered-signal drain. `deliver_raw` wraps this with the drain; the
    /// drain itself uses this to avoid recursing.
    #[instrument(level = "debug", skip(self, payload), fields(saga = S::name(), instance_id, event_type))]
    async fn deliver_leased_no_drain(
        &self,
        instance_id: &str,
        event_type: &str,
        payload: serde_json::Value,
    ) -> Result<bool, PgRunnerError> {
        // Hold the instance lease for the whole delivery so its effects can't
        // run concurrently with another deliver/tick/signal on this instance.
        // Waiting (not failing) on contention keeps the no-op-on-duplicate
        // semantics: a second deliver of the same response waits its turn,
        // then finds the await already consumed → Ok(false).
        let fence = self.acquire_lease_blocking(instance_id).await?;
        let effect = self.deliver_raw_locked(instance_id, event_type, payload);
        let result = self.run_with_heartbeat(instance_id, fence, effect).await;
        // On a consumed delivery (`Ok(true)`) the terminal commit already
        // released the lease in its own transaction (H3: one fewer fsync).
        // Only release separately when nothing committed — a no-op delivery
        // (`Ok(false)`) or an error/exhausted-retry — so a peer isn't left
        // waiting out the TTL.
        if !matches!(result, Ok(true)) {
            self.release_lease(instance_id, fence).await;
        }
        result
    }

    /// The delivery body, run while holding the instance lease.
    async fn deliver_raw_locked(
        &self,
        instance_id: &str,
        event_type: &str,
        payload: serde_json::Value,
    ) -> Result<bool, PgRunnerError> {
        // OCC conflicts here are rare now the instance lease serializes
        // processing, but remain a backstop. On conflict, back off before
        // retrying (SAGA-19) so a brief contention burst doesn't livelock;
        // the backoff grows with the attempt index (no RNG dependency — the
        // index alone de-synchronizes racers enough). Exhaustion surfaces as
        // the typed, retryable `Contended` rather than an opaque internal
        // error.
        const MAX_ATTEMPTS: u32 = 8;
        for attempt in 0..MAX_ATTEMPTS {
            match self.deliver_once(instance_id, event_type, &payload).await {
                Ok(outcome) => return Ok(outcome),
                Err(PgRunnerError::Store(EventStoreError::ConcurrencyConflict { .. }))
                    if attempt + 1 < MAX_ATTEMPTS =>
                {
                    // Lost the version race; another worker advanced this
                    // instance. Back off, then reload and retry from routing —
                    // the await may now be gone (→ Ok(false)) or still ours.
                    let backoff = std::time::Duration::from_millis(5 * (attempt as u64 + 1));
                    tokio::time::sleep(backoff).await;
                    continue;
                }
                Err(e) => return Err(e),
            }
        }
        Err(PgRunnerError::Contended {
            instance_id: instance_id.to_string(),
        })
    }

    /// Route an incoming event type to the step awaiting it on this
    /// instance. Returns `None` if nothing awaits it.
    ///
    /// Refuses with [`PgRunnerError::AmbiguousAwait`] when more than one
    /// step matches: routing is by event-type string, so two simultaneous
    /// awaits on the same type are indistinguishable — an unordered
    /// `LIMIT 1` would silently feed one step's response into the other
    /// (review C7). The check is at runtime because await types are
    /// declared inside handler bodies (`await_event::<E>()`), invisible to
    /// build-time topology validation.
    async fn route_awaiting_step(
        &self,
        instance_id: &str,
        event_type: &str,
    ) -> Result<Option<String>, PgRunnerError> {
        let mut steps: Vec<String> = sqlx::query_scalar(
            "SELECT step FROM es_saga_awaits \
             WHERE saga_name = $1 AND instance_id = $2 AND event_type = $3 \
             ORDER BY step \
             LIMIT 2",
        )
        .bind(S::name())
        .bind(instance_id)
        .bind(event_type)
        .fetch_all(self.store.pool())
        .await
        .map_err(|e| PgRunnerError::Store(e.into()))?;

        if steps.len() > 1 {
            return Err(PgRunnerError::AmbiguousAwait {
                instance_id: instance_id.to_string(),
                event_type: event_type.to_string(),
                steps,
            });
        }
        Ok(steps.pop())
    }

    /// One delivery attempt: route → load → apply the topology-registered
    /// response map → complete → drive → commit. Returns
    /// `ConcurrencyConflict` if the commit lost the version race; the
    /// caller retries.
    ///
    /// Resume never re-runs the step handler (phase 2): the await + map
    /// live on the topology (`task(...).on_response(...)`), so the old
    /// handler-purity contract — and its divergence/dead-letter failure
    /// modes — no longer exist.
    #[instrument(level = "debug", skip(self, payload), fields(saga = S::name(), instance_id, event_type))]
    async fn deliver_once(
        &self,
        instance_id: &str,
        event_type: &str,
        payload: &serde_json::Value,
    ) -> Result<bool, PgRunnerError> {
        // 1. Route: which step (if any) on this instance awaits this
        //    event type? Single indexed lookup on es_saga_awaits; refuses
        //    when more than one step matches (C7 — see route_awaiting_step).
        let awaiting_step = self.route_awaiting_step(instance_id, event_type).await?;

        let Some(step_name_owned) = awaiting_step else {
            trace!(
                saga = S::name(),
                instance_id, event_type, "no-op delivery (await already gone)"
            );
            return Ok(false);
        };

        // 2. Load the instance.
        let mut root = self
            .store
            .load::<SagaEnvelope<S>>(instance_id)
            .await
            .map_err(PgRunnerError::Store)?
            .ok_or_else(|| {
                PgRunnerError::InternalOwned(format!(
                    "await row referenced missing instance `{instance_id}`"
                ))
            })?;
        Self::ensure_state_decoded(instance_id, &root.state)?;

        // Resolve the &'static step name from the topology (the await
        // row stored an owned String).
        let step_name: StepName = self
            .topology
            .step_entries()
            .iter()
            .map(|e| e.name)
            .find(|n| *n == step_name_owned)
            .ok_or(PgRunnerError::Internal(
                "await row referenced a step not in the topology",
            ))?;

        // Guard: the step must still be Active. A concurrent worker may
        // have already completed it between our route read and load.
        if !matches!(root.state.step_status(step_name), SagaStepStatus::Active) {
            return Ok(false);
        }

        let entry = self
            .topology
            .step_entries()
            .iter()
            .find(|e| e.name == step_name)
            .ok_or(PgRunnerError::Internal("step vanished from topology"))?;

        let trigger = root.state.trigger.clone();

        let consumed = [PendingAwait {
            step: step_name.to_string(),
            event_type: event_type.to_string(),
            is_compensation: false,
            deadline: None,
            outbox_emits: Vec::new(), // delete marker — carries no emit
        }];

        // 3. Look up the topology-registered response map. An await row
        //    whose step no longer declares `.on_response` (or declares a
        //    different event type) means the topology changed under a live
        //    instance — dead-letter the step (→ compensation) rather than
        //    wedging or mis-decoding.
        let Some(spec) = &entry.await_spec else {
            self.fail_forward_step(
                &mut root,
                step_name,
                format!(
                    "await row exists but step `{step_name}` declares no `.on_response(...)` \
                     (topology changed under a live instance?)"
                ),
            )
            .await?;
            self.commit_and_release(instance_id, &mut root, &[], &consumed)
                .await?;
            return Ok(true);
        };
        if spec.event_type != event_type {
            self.fail_forward_step(
                &mut root,
                step_name,
                format!(
                    "await row routes `{event_type}` but step `{step_name}` declares \
                     `.on_response` for `{}` (topology changed under a live instance?)",
                    spec.event_type
                ),
            )
            .await?;
            self.commit_and_release(instance_id, &mut root, &[], &consumed)
                .await?;
            return Ok(true);
        }

        // 4. Apply the map to the delivered payload → typed output, complete.
        //    The map runs user code; failures (bad payload, panicking
        //    closure) fail the step and cascade.
        let output = match self.apply_await_map(spec, payload.clone()) {
            Ok(output) => output,
            Err(reason) => {
                self.fail_forward_step(&mut root, step_name, reason).await?;
                self.commit_and_release(instance_id, &mut root, &[], &consumed)
                    .await?;
                return Ok(true);
            }
        };
        self.complete_step(&mut root, entry, step_name, output);

        // 5. Resume driving — the completed step may unlock downstream steps
        //    that run inline or suspend again — then commit everything (new
        //    events, new awaits, delete the consumed await) atomically. OCC
        //    conflict → caller retries.
        //
        //    NOTE: the C3 deliver-path claim (commit `StepCompleted` BEFORE
        //    driving downstream, so two concurrent deliveries of the same
        //    response can't both run the downstream effects) is intentionally
        //    NOT split out here yet. Splitting introduces a claim-then-crash
        //    gap (downstream left un-driven, re-delivery finds the await
        //    gone) that REQUIRES the lazy-recovery path to close — the
        //    crash-heavy `stress_pg_chaos::mayhem` test relies on
        //    reconciliation driving every instance to terminal. C3-deliver
        //    and recovery land together (see `tasks/effects-vs-commit.md`).
        let trigger_for_resume = trigger.unwrap_or(serde_json::Value::Null);
        let pending = self
            .drive_to_quiescence(&mut root, &trigger_for_resume)
            .await?;
        self.commit_and_release(instance_id, &mut root, &pending, &consumed)
            .await?;

        debug!(
            saga = S::name(),
            instance_id, event_type, "delivered (await consumed)"
        );
        Ok(true)
    }

    /// OCC-retry wrapper for [`deliver_failure`], mirroring
    /// [`deliver_raw_locked`](Self::deliver_raw_locked): a lost version race
    /// reloads and retries from routing (the await may now be gone → `Ok(false)`,
    /// or still ours). Runs under the instance lease.
    async fn fail_await_locked(
        &self,
        instance_id: &str,
        step: &str,
        reason: &str,
    ) -> Result<bool, PgRunnerError> {
        const MAX_ATTEMPTS: u32 = 8;
        for attempt in 0..MAX_ATTEMPTS {
            match self.fail_await_once(instance_id, step, reason).await {
                Ok(outcome) => return Ok(outcome),
                Err(PgRunnerError::Store(EventStoreError::ConcurrencyConflict { .. }))
                    if attempt + 1 < MAX_ATTEMPTS =>
                {
                    let backoff = std::time::Duration::from_millis(5 * (attempt as u64 + 1));
                    tokio::time::sleep(backoff).await;
                    continue;
                }
                Err(e) => return Err(e),
            }
        }
        Err(PgRunnerError::Contended {
            instance_id: instance_id.to_string(),
        })
    }

    /// Fail one awaiting forward step by name and run the compensation cascade.
    /// The body of [`deliver_failure`]; structured like [`deliver_once`] but
    /// failing the step instead of completing it.
    async fn fail_await_once(
        &self,
        instance_id: &str,
        step: &str,
        reason: &str,
    ) -> Result<bool, PgRunnerError> {
        // Resolve the &'static step name from the topology.
        let step_name: StepName = self
            .topology
            .step_entries()
            .iter()
            .map(|e| e.name)
            .find(|n| *n == step)
            .ok_or(PgRunnerError::Internal(
                "deliver_failure: step not in topology",
            ))?;

        let mut root = self
            .store
            .load::<SagaEnvelope<S>>(instance_id)
            .await
            .map_err(PgRunnerError::Store)?
            .ok_or_else(|| {
                PgRunnerError::InternalOwned(format!(
                    "deliver_failure: unknown instance `{instance_id}`"
                ))
            })?;
        Self::ensure_state_decoded(instance_id, &root.state)?;

        // Guard: the step must still be Active (awaiting). A concurrent worker
        // may have already completed/failed it — idempotent no-op.
        if !matches!(root.state.step_status(step_name), SagaStepStatus::Active) {
            return Ok(false);
        }

        let entry = self
            .topology
            .step_entries()
            .iter()
            .find(|e| e.name == step_name)
            .ok_or(PgRunnerError::Internal("step vanished from topology"))?;

        // Delete-marker for the forward await this failure consumes, so the
        // await row is removed in the same commit as the StepFailed + cascade.
        let consumed: Vec<PendingAwait> = entry
            .await_spec
            .as_ref()
            .map(|spec| {
                vec![PendingAwait {
                    step: step_name.to_string(),
                    event_type: spec.event_type.to_string(),
                    is_compensation: false,
                    deadline: None,
                    outbox_emits: Vec::new(),
                }]
            })
            .unwrap_or_default();

        self.fail_forward_step(&mut root, step_name, reason.to_string())
            .await?;
        self.commit_and_release(instance_id, &mut root, &[], &consumed)
            .await?;
        debug!(saga = S::name(), instance_id, step = %step_name, "failed via deliver_failure (cascade run)");
        Ok(true)
    }

    /// Deliver an external signal to an instance, per its registered
    /// policy.
    ///
    /// - `TriggersCompensation` — cancel the in-flight saga and run the
    ///   compensation cascade; `Cause` surfaces the signal as the
    ///   originating error to comp handlers.
    /// - `RoutesToStep(name)` — the signal IS the awaited event for the
    ///   named step; routes through the same path as `deliver`.
    ///
    /// Rejections:
    /// - unregistered signal type → `UnregisteredSignal`
    /// - unknown instance → `UnknownInstance`
    /// - terminal saga → `SagaTerminal`
    /// - saga already compensating → `Compensating`
    ///
    /// The signal is written to `es_saga_signals` for audit + so a
    /// future lease-driven worker loop can drain it.
    ///
    /// **KNOWN LIMITATION (review SAGA-14):** despite earlier docs, rows in
    /// `es_saga_signals` are **not** currently deleted — `write_signal_row`
    /// only inserts. The table grows unboundedly until retention is added
    /// (tracked with the worker-loop/outbox work). Treat it as an
    /// append-only audit log for now.
    #[instrument(level = "debug", skip(self, signal), fields(saga = S::name(), instance_id, signal_type = std::any::type_name::<T>()))]
    pub async fn signal<T>(
        &self,
        instance_id: &str,
        signal: T,
    ) -> Result<crate::SignalOutcome, PgRunnerError>
    where
        T: crate::EventTypeStatic + Serialize + 'static,
    {
        let type_name = std::any::type_name::<T>();
        let policy = self
            .topology
            .signals_iter()
            .find(|s| s.signal_type_name == type_name)
            .map(|s| s.policy)
            .ok_or(PgRunnerError::UnregisteredSignal(type_name))?;

        // Load instance + reject terminal / compensating (fast-path checks;
        // the TriggersCompensation branch re-loads under the lease before
        // running effects).
        let root = self
            .store
            .load::<SagaEnvelope<S>>(instance_id)
            .await
            .map_err(PgRunnerError::Store)?
            .ok_or_else(|| PgRunnerError::UnknownInstance(instance_id.to_string()))?;
        Self::ensure_state_decoded(instance_id, &root.state)?;

        if root.state.status.is_terminal() {
            debug!(
                saga = S::name(),
                instance_id,
                signal_type = type_name,
                "signal rejected: instance is terminal"
            );
            return Err(PgRunnerError::SagaTerminal {
                instance_id: instance_id.to_string(),
            });
        }
        // Mid-compensation: any step in Compensating means a cascade is
        // underway (durable runner only persists this state if a comp
        // suspended — defensive guard regardless).
        let compensating = root
            .state
            .steps
            .values()
            .any(|s| matches!(s, SagaStepStatus::Compensating));
        if compensating {
            debug!(
                saga = S::name(),
                instance_id,
                signal_type = type_name,
                "signal rejected: instance is compensating"
            );
            return Err(PgRunnerError::Compensating {
                instance_id: instance_id.to_string(),
            });
        }

        // Persist the signal for audit (consumed in this same txn at
        // commit time). shard_id matches the instance's shard.
        let payload = serde_json::to_value(&signal).map_err(PgRunnerError::Serialise)?;
        let signal_event_type = <T as crate::EventTypeStatic>::EVENT_TYPE;

        match policy {
            SignalPolicy::TriggersCompensation => {
                // Lease the instance so the cancel's compensation effects
                // don't run concurrently with a delivery's effects on the
                // same instance (effects-vs-commit §11). Re-load under the
                // lease — state may have advanced since the fast-path checks.
                let fence = self.acquire_lease_blocking(instance_id).await?;
                let effect = async {
                    let mut root = self
                        .store
                        .load::<SagaEnvelope<S>>(instance_id)
                        .await
                        .map_err(PgRunnerError::Store)?
                        .ok_or_else(|| PgRunnerError::UnknownInstance(instance_id.to_string()))?;
                    Self::ensure_state_decoded(instance_id, &root.state)?;
                    if root.state.status.is_terminal() {
                        return Err(PgRunnerError::SagaTerminal {
                            instance_id: instance_id.to_string(),
                        });
                    }
                    let err = StepError::Cancelled {
                        step: "<signal>".into(),
                        reason: format!("signal {type_name}"),
                    };
                    self.run_compensation_cascade(&mut root, err).await?;
                    self.commit_with_signal(
                        instance_id,
                        &mut root,
                        &[],
                        &[],
                        signal_event_type,
                        &payload,
                    )
                    .await?;
                    Ok(crate::SignalOutcome::Accepted)
                };
                let result = self.run_with_heartbeat(instance_id, fence, effect).await;
                self.release_lease(instance_id, fence).await;
                result
            }
            SignalPolicy::RoutesToStep {
                step: target_step, ..
            } => {
                // Config-error check: if some step is awaiting this event
                // type and it isn't the declared target, `.routes_to_step`
                // is wrong (or it's a C7 ambiguity). Refuse rather than
                // misroute.
                let awaiting_step = self
                    .route_awaiting_step(instance_id, signal_event_type)
                    .await?;
                if let Some(awaiter) = &awaiting_step
                    && awaiter != target_step
                {
                    return Err(PgRunnerError::SignalRouteMismatch {
                        signal_type: type_name,
                        target: target_step,
                        actual_awaiter: awaiter.clone(),
                    });
                }

                // Unified write-then-drain (review SAGA-10). Under the
                // instance lease (so it serialises with a concurrent
                // suspend's drain): reject a duplicate pending signal of the
                // same type, else persist this one as PENDING. Releasing the
                // lease, `drain_buffered_signals` then delivers it iff a step
                // is already awaiting it — otherwise it stays pending and is
                // drained the moment the target step suspends.
                let fence = self.acquire_lease_blocking(instance_id).await?;
                let effect = async {
                    if self
                        .pending_signal_exists(instance_id, signal_event_type)
                        .await?
                    {
                        return Err(PgRunnerError::DuplicateBufferedSignal {
                            instance_id: instance_id.to_string(),
                            signal_type: type_name,
                        });
                    }
                    self.write_signal_row(instance_id, signal_event_type, &payload)
                        .await
                };
                let write = self.run_with_heartbeat(instance_id, fence, effect).await;
                self.release_lease(instance_id, fence).await;
                write?;

                let consumed = self.drain_buffered_signals(instance_id).await?;
                Ok(if consumed.contains(signal_event_type) {
                    debug!(
                        saga = S::name(),
                        instance_id,
                        signal_type = type_name,
                        "signal delivered to awaiting step"
                    );
                    crate::SignalOutcome::Accepted
                } else {
                    debug!(
                        saga = S::name(),
                        instance_id,
                        signal_type = type_name,
                        "signal buffered (step not yet suspended)"
                    );
                    crate::SignalOutcome::Buffered
                })
            }
        }
    }

    /// Insert an audit row into `es_saga_signals`.
    async fn write_signal_row(
        &self,
        instance_id: &str,
        signal_type: &str,
        payload: &serde_json::Value,
    ) -> Result<(), PgRunnerError> {
        let shard = shard_of(instance_id, self.num_shards);
        sqlx::query(
            "INSERT INTO es_saga_signals \
             (saga_name, instance_id, signal_type, payload, shard_id) \
             VALUES ($1, $2, $3, $4, $5)",
        )
        .bind(S::name())
        .bind(instance_id)
        .bind(signal_type)
        .bind(payload)
        .bind(shard as i32)
        .execute(self.store.pool())
        .await
        .map_err(|e| PgRunnerError::Store(e.into()))?;
        Ok(())
    }

    /// Is there already a pending (unconsumed) signal of this type for the
    /// instance? Backs the at-most-one-pending invariant (D-SIG-1).
    async fn pending_signal_exists(
        &self,
        instance_id: &str,
        signal_type: &str,
    ) -> Result<bool, PgRunnerError> {
        let exists: bool = sqlx::query_scalar(
            "SELECT EXISTS (SELECT 1 FROM es_saga_signals \
             WHERE saga_name = $1 AND instance_id = $2 AND signal_type = $3 \
               AND consumed_at IS NULL)",
        )
        .bind(S::name())
        .bind(instance_id)
        .bind(signal_type)
        .fetch_one(self.store.pool())
        .await
        .map_err(|e| PgRunnerError::Store(e.into()))?;
        Ok(exists)
    }

    /// Mark a signal row consumed (audit; SAGA-14 retention predicate).
    /// Idempotent via the `consumed_at IS NULL` guard.
    async fn mark_signal_consumed(&self, id: i64) -> Result<(), PgRunnerError> {
        sqlx::query(
            "UPDATE es_saga_signals SET consumed_at = now() WHERE id = $1 AND consumed_at IS NULL",
        )
        .bind(id)
        .execute(self.store.pool())
        .await
        .map_err(|e| PgRunnerError::Store(e.into()))?;
        Ok(())
    }

    /// Delete consumed signal rows older than `older_than`, returning the
    /// number removed. Closes SAGA-14: signals were `consumed_at`-marked but
    /// never deleted, so `es_saga_signals` grew without bound. A host (or the
    /// worker loop, once it lands) calls this on a schedule — it's a
    /// maintenance primitive in the same vein as
    /// [`tick_timeouts`](Self::tick_timeouts).
    ///
    /// Only **consumed** rows are eligible: a pending signal (its target step
    /// hasn't suspended to drain it yet) is never pruned regardless of age,
    /// so a slow-to-arrive step can't lose its decision. Deletes in bounded
    /// batches so a large backlog doesn't take one long lock.
    pub async fn prune_consumed_signals(
        &self,
        older_than: std::time::Duration,
    ) -> Result<u64, PgRunnerError> {
        let secs = older_than.as_secs_f64().to_string();
        let mut total = 0u64;
        loop {
            // `DELETE … WHERE id IN (SELECT … LIMIT n)` — Postgres has no
            // direct `DELETE … LIMIT`. Batched so retention never holds one
            // big lock on a long-lived signals table.
            let n = sqlx::query(
                "DELETE FROM es_saga_signals
                 WHERE id IN (
                     SELECT id FROM es_saga_signals
                      WHERE saga_name = $1
                        AND consumed_at IS NOT NULL
                        AND consumed_at < now() - ($2 || ' seconds')::interval
                      ORDER BY id
                      LIMIT 1000
                 )",
            )
            .bind(S::name())
            .bind(&secs)
            .execute(self.store.pool())
            .await
            .map_err(|e| PgRunnerError::Store(e.into()))?
            .rows_affected();
            total += n;
            if n < 1000 {
                break;
            }
        }
        Ok(total)
    }

    /// Deliver any pending (buffered) signals whose type is currently
    /// awaited by a step on this instance (review SAGA-10). Loops so a
    /// delivery that suspends a downstream step which *also* has a buffered
    /// signal drains it too. Returns the set of event types consumed.
    ///
    /// Each delivery runs under the instance lease (via `deliver_raw`); the
    /// row is marked consumed afterwards. A crash between delivery-commit and
    /// the mark leaves a stale pending row — harmless beyond a possible
    /// spurious `DuplicateBufferedSignal` until an operator clears it; never
    /// a double effect (the await is gone, so a redelivery is a no-op).
    async fn drain_buffered_signals(
        &self,
        instance_id: &str,
    ) -> Result<std::collections::HashSet<String>, PgRunnerError> {
        let mut consumed_types = std::collections::HashSet::new();
        loop {
            let row: Option<(i64, String, serde_json::Value)> = sqlx::query_as(
                "SELECT s.id, s.signal_type, s.payload \
                 FROM es_saga_signals s \
                 JOIN es_saga_awaits a \
                   ON a.saga_name = s.saga_name \
                  AND a.instance_id = s.instance_id \
                  AND a.event_type = s.signal_type \
                 WHERE s.saga_name = $1 AND s.instance_id = $2 AND s.consumed_at IS NULL \
                 ORDER BY s.id \
                 LIMIT 1",
            )
            .bind(S::name())
            .bind(instance_id)
            .fetch_optional(self.store.pool())
            .await
            .map_err(|e| PgRunnerError::Store(e.into()))?;

            let Some((id, signal_type, payload)) = row else {
                break;
            };

            let delivered = self
                .deliver_leased_no_drain(instance_id, &signal_type, payload)
                .await?;
            if delivered {
                self.mark_signal_consumed(id).await?;
                consumed_types.insert(signal_type);
            } else {
                // The await vanished between the JOIN and delivery (a racing
                // worker drained it). Stop rather than spin; a later
                // suspension re-drains anything still pending.
                break;
            }
        }
        Ok(consumed_types)
    }

    /// Like [`Self::commit`] but also writes the signal audit row in
    /// the same transaction (used by the `TriggersCompensation` path
    /// where the whole transition happens in one commit).
    async fn commit_with_signal(
        &self,
        instance_id: &str,
        root: &mut AggregateRoot<SagaEnvelope<S>>,
        inserts: &[PendingAwait],
        deletes: &[PendingAwait],
        signal_type: &str,
        signal_payload: &serde_json::Value,
    ) -> Result<(), PgRunnerError> {
        let mut tx = self
            .store
            .begin_transaction()
            .await
            .map_err(PgRunnerError::Store)?;
        tx.save(root).await.map_err(PgRunnerError::Store)?;

        let shard = shard_of(instance_id, self.num_shards);
        sqlx::query(
            "INSERT INTO es_saga_signals \
             (saga_name, instance_id, signal_type, payload, shard_id) \
             VALUES ($1, $2, $3, $4, $5)",
        )
        .bind(S::name())
        .bind(instance_id)
        .bind(signal_type)
        .bind(signal_payload)
        .bind(shard as i32)
        .execute(&mut **tx.tx())
        .await
        .map_err(|e| PgRunnerError::Store(e.into()))?;

        for a in deletes {
            sqlx::query(
                "DELETE FROM es_saga_awaits \
                 WHERE saga_name = $1 AND instance_id = $2 AND step = $3",
            )
            .bind(S::name())
            .bind(instance_id)
            .bind(&a.step)
            .execute(&mut **tx.tx())
            .await
            .map_err(|e| PgRunnerError::Store(e.into()))?;
        }
        for a in inserts {
            sqlx::query(
                "INSERT INTO es_saga_awaits \
                 (saga_name, instance_id, step, event_type, is_compensation, shard_id, deadline_at) \
                 VALUES ($1, $2, $3, $4, $5, $6, $7) \
                 ON CONFLICT (saga_name, instance_id, step) DO UPDATE \
                 SET event_type = EXCLUDED.event_type, \
                     is_compensation = EXCLUDED.is_compensation, \
                     deadline_at = EXCLUDED.deadline_at",
            )
            .bind(S::name())
            .bind(instance_id)
            .bind(&a.step)
            .bind(&a.event_type)
            .bind(a.is_compensation)
            .bind(shard as i32)
            .bind(a.deadline)
            .execute(&mut **tx.tx())
            .await
            .map_err(|e| PgRunnerError::Store(e.into()))?;
        }

        // Terminal ⇒ no awaits (see `commit` for rationale).
        if root.state.status.is_terminal() {
            sqlx::query("DELETE FROM es_saga_awaits WHERE saga_name = $1 AND instance_id = $2")
                .bind(S::name())
                .bind(instance_id)
                .execute(&mut **tx.tx())
                .await
                .map_err(|e| PgRunnerError::Store(e.into()))?;
        }

        // Status projection (P3.1), same tx as everything else.
        Self::upsert_instance_status(&mut tx, instance_id, root.state.status, shard).await?;
        // External emits to the outbox (P3.2), same tx.
        Self::insert_outbox_emits(&mut tx, instance_id, shard, inserts).await?;

        tx.commit().await.map_err(PgRunnerError::Store)?;
        Ok(())
    }

    /// Upsert the instance's status mirror into `es_saga_instances` (P3.1)
    /// within the caller's transaction. Idempotent on `(saga_name,
    /// instance_id)`; every commit refreshes `status`, `shard_id` (in case
    /// `num_shards` changed across a documented reshard), and `updated_at`
    /// (which recovery uses as the staleness clock).
    /// Write the external (Regime-B) emits carried by `inserts` to
    /// `es_saga_outbox` within the caller's transaction (P3.2) — same commit
    /// as the `StepEmitted` envelope event + the await row, so the outbox can
    /// never diverge from the log. A configured `EmitPublisher` (next slice)
    /// drains and delivers them at-least-once.
    async fn insert_outbox_emits(
        tx: &mut mire::TransactionScope<'_>,
        instance_id: &str,
        shard: u32,
        inserts: &[PendingAwait],
    ) -> Result<(), PgRunnerError> {
        for a in inserts {
            for emit in &a.outbox_emits {
                sqlx::query(
                    "INSERT INTO es_saga_outbox \
                     (saga_name, instance_id, step, event_type, payload, shard_id) \
                     VALUES ($1, $2, $3, $4, $5, $6)",
                )
                .bind(S::name())
                .bind(instance_id)
                .bind(&a.step)
                .bind(&emit.event_type)
                .bind(&emit.payload)
                .bind(shard as i32)
                .execute(&mut **tx.tx())
                .await
                .map_err(|e| PgRunnerError::Store(e.into()))?;
            }
        }
        Ok(())
    }

    async fn upsert_instance_status(
        tx: &mut mire::TransactionScope<'_>,
        instance_id: &str,
        status: SagaInstanceStatus,
        shard: u32,
    ) -> Result<(), PgRunnerError> {
        sqlx::query(
            "INSERT INTO es_saga_instances \
             (saga_name, instance_id, status, shard_id, updated_at) \
             VALUES ($1, $2, $3, $4, now()) \
             ON CONFLICT (saga_name, instance_id) DO UPDATE \
             SET status = EXCLUDED.status, \
                 shard_id = EXCLUDED.shard_id, \
                 updated_at = now()",
        )
        .bind(S::name())
        .bind(instance_id)
        .bind(status.as_db_str())
        .bind(shard as i32)
        .execute(&mut **tx.tx())
        .await
        .map_err(|e| PgRunnerError::Store(e.into()))?;
        Ok(())
    }

    /// Commit an instance transition: append pending envelope events,
    /// insert new await rows, delete consumed await rows — all in one
    /// transaction so the routing projection can never diverge from
    /// the event stream across a crash.
    async fn commit(
        &self,
        instance_id: &str,
        root: &mut AggregateRoot<SagaEnvelope<S>>,
        inserts: &[PendingAwait],
        deletes: &[PendingAwait],
    ) -> Result<(), PgRunnerError> {
        self.commit_inner(instance_id, root, inserts, deletes, None)
            .await
    }

    /// Like [`commit`](Self::commit), but also releases this instance's
    /// processing lease **in the same transaction** — so the release shares
    /// the commit's single fsync instead of paying its own (perf H3: the
    /// lease acquire+release were the two costliest fsyncs of a `deliver`).
    /// Use only when this commit is the *final* one of a leased op (e.g.
    /// `deliver`'s terminal commit); a multi-commit op like `tick_one` must
    /// not release until its last commit. Releasing in-tx also means an
    /// OCC-conflicted (rolled-back) commit keeps the lease, so the retry
    /// runs under it — strictly better than the old release-after-return.
    async fn commit_and_release(
        &self,
        instance_id: &str,
        root: &mut AggregateRoot<SagaEnvelope<S>>,
        inserts: &[PendingAwait],
        deletes: &[PendingAwait],
    ) -> Result<(), PgRunnerError> {
        let worker = self.worker_id.clone();
        self.commit_inner(instance_id, root, inserts, deletes, Some(&worker))
            .await
    }

    async fn commit_inner(
        &self,
        instance_id: &str,
        root: &mut AggregateRoot<SagaEnvelope<S>>,
        inserts: &[PendingAwait],
        deletes: &[PendingAwait],
        release_worker: Option<&str>,
    ) -> Result<(), PgRunnerError> {
        let mut tx = self
            .store
            .begin_transaction()
            .await
            .map_err(PgRunnerError::Store)?;

        // Append events with OCC (ExpectedVersion derived from root.version).
        tx.save(root).await.map_err(PgRunnerError::Store)?;

        let shard = shard_of(instance_id, self.num_shards);

        for a in deletes {
            sqlx::query(
                "DELETE FROM es_saga_awaits \
                 WHERE saga_name = $1 AND instance_id = $2 AND step = $3",
            )
            .bind(S::name())
            .bind(instance_id)
            .bind(&a.step)
            .execute(&mut **tx.tx())
            .await
            .map_err(|e| PgRunnerError::Store(e.into()))?;
        }

        for a in inserts {
            sqlx::query(
                "INSERT INTO es_saga_awaits \
                 (saga_name, instance_id, step, event_type, is_compensation, shard_id, deadline_at) \
                 VALUES ($1, $2, $3, $4, $5, $6, $7) \
                 ON CONFLICT (saga_name, instance_id, step) DO UPDATE \
                 SET event_type = EXCLUDED.event_type, \
                     is_compensation = EXCLUDED.is_compensation, \
                     deadline_at = EXCLUDED.deadline_at",
            )
            .bind(S::name())
            .bind(instance_id)
            .bind(&a.step)
            .bind(&a.event_type)
            .bind(a.is_compensation)
            .bind(shard as i32)
            .bind(a.deadline)
            .execute(&mut **tx.tx())
            .await
            .map_err(|e| PgRunnerError::Store(e.into()))?;
        }

        // Invariant: a terminal instance has no pending awaits. When a
        // failure/cancellation drives the instance terminal, purge ALL
        // its await rows — not just the one that triggered it. Without
        // this, a parallel-suspend instance that times out (or is
        // cancelled) on one branch would leave the sibling branch's
        // await row behind, to be needlessly reprocessed by a later
        // tick. Cheap (indexed delete) and only fires on terminal.
        if root.state.status.is_terminal() {
            sqlx::query("DELETE FROM es_saga_awaits WHERE saga_name = $1 AND instance_id = $2")
                .bind(S::name())
                .bind(instance_id)
                .execute(&mut **tx.tx())
                .await
                .map_err(|e| PgRunnerError::Store(e.into()))?;
        }

        // Upsert the status projection (P3.1) in the same tx, so it never
        // diverges from the envelope across a crash. Recovery + ops read it.
        Self::upsert_instance_status(&mut tx, instance_id, root.state.status, shard).await?;

        // Write any external emits to the outbox (P3.2), same tx as the await.
        Self::insert_outbox_emits(&mut tx, instance_id, shard, inserts).await?;

        // Fold the lease release into this commit's fsync (H3) when this is
        // the final commit of a leased op. Best-effort (`-infinity` so any
        // peer can take over immediately); scoped to our worker_id so we
        // never release a lease a takeover worker now holds.
        if let Some(worker) = release_worker {
            sqlx::query(
                "UPDATE es_saga_leases SET leased_until = '-infinity' \
                 WHERE saga_name = $1 AND instance_id = $2 AND worker_id = $3",
            )
            .bind(S::name())
            .bind(instance_id)
            .bind(worker)
            .execute(&mut **tx.tx())
            .await
            .map_err(|e| PgRunnerError::Store(e.into()))?;
        }

        tx.commit().await.map_err(PgRunnerError::Store)?;
        Ok(())
    }

    /// Drive the envelope to quiescence: apply pure decisions
    /// (activate / skip / done) until no step is ready, the saga
    /// terminates, or a step suspends awaiting an external response.
    ///
    /// Returns the awaits that should be persisted to `es_saga_awaits`
    /// (so an external `deliver(...)` can route the response back).
    /// In the all-in-process-handler case the returned vec is empty —
    /// the saga ran straight to terminal in one call.
    async fn drive_to_quiescence(
        &self,
        root: &mut AggregateRoot<SagaEnvelope<S>>,
        trigger_json: &serde_json::Value,
    ) -> Result<Vec<PendingAwait>, PgRunnerError> {
        let mut pending = Vec::new();
        loop {
            match next_step(&root.state, &self.topology) {
                NextStep::None => break,
                NextStep::Skip(step_name, reason) => {
                    root.record(SagaEnvelopeEvent::StepSkipped {
                        step: step_name.to_string(),
                        reason,
                    });
                    // Skipping a terminal step doesn't itself end the
                    // saga — let the loop continue; future steps may
                    // still activate or also skip-cascade.
                }
                NextStep::Activate(step_name) => {
                    match self.activate_step(root, step_name, trigger_json).await? {
                        StepOutcome::Completed { terminal } => {
                            if terminal {
                                break;
                            }
                        }
                        StepOutcome::Suspended(await_row) => {
                            // Step emitted to an external worker and is
                            // now Active, awaiting a response. Record the
                            // await but KEEP DRIVING — parallel branches
                            // must all suspend together, not serialise
                            // through one external round-trip at a time.
                            // The suspended step is now Active, so
                            // `next_step` won't re-select it; the loop
                            // ends when no further step is ready.
                            pending.push(await_row);
                        }
                    }
                }
            }
        }
        Ok(pending)
    }

    /// Invoke a step's handler against the current envelope.
    ///
    /// Handler return shapes:
    /// 1. `StepResult::complete(output)` — immediate; record
    ///    `StepActivated` + `StepCompleted`.
    /// 2. `emit(req).await_event::<E>().map(...)` with an in-process
    ///    handler for `req` — invoke inline, apply the map, complete.
    /// 3. `emit(req).await_event::<E>().map(...)` with NO in-process
    ///    handler — record `StepActivated`, suspend awaiting `E`. The
    ///    returned [`StepOutcome::Suspended`] carries the await to persist.
    ///
    ///    The emitted request *payload* is persisted durably: each
    ///    dispatched action is recorded as a [`SagaEnvelopeEvent::StepEmitted`]
    ///    in the **same transaction** as `StepActivated` and the await row, so
    ///    an external worker can recover *what* was requested by replaying the
    ///    envelope (`env.requests_for(step)`). A `default_step_timeout` is
    ///    still advisable so a never-answered await eventually compensates,
    ///    but the request itself is no longer lost (review C5/SAGA-3/DX-1).
    /// 4. Build error — surface (full StepError path is compensation's).
    #[instrument(level = "debug", skip(self, root, trigger_json), fields(saga = S::name(), step = %step_name))]
    async fn activate_step(
        &self,
        root: &mut AggregateRoot<SagaEnvelope<S>>,
        step_name: StepName,
        trigger_json: &serde_json::Value,
    ) -> Result<StepOutcome, PgRunnerError> {
        let entry = self
            .topology
            .step_entries()
            .iter()
            .find(|e| e.name == step_name)
            .ok_or(PgRunnerError::Internal("next_step returned unknown step"))?;

        use crate::guard::{CaughtResult, run_caught};

        let outputs = build_outputs_by_type_id(&root.state, self.topology.step_entries());

        // Record StepActivated up-front so every failure path below has a
        // coherent audit trail (StepActivated → StepFailed → cascade).
        root.record(SagaEnvelopeEvent::StepActivated {
            step: step_name.to_string(),
        });

        // Invoke the step handler with panic catching (SAGA-6). A panic,
        // extractor failure, or build error fails the step and runs the
        // compensation cascade rather than aborting the call uncommitted
        // (SAGA-7).
        let caught = {
            let ctx = StepContext::for_testing(
                &root.state.state,
                Some(trigger_json),
                &outputs,
                self.topology.deps_ref(),
                None,
            )
            .with_idempotency(
                Self::instance_id_of(root),
                step_name,
                root.state.activation_seq(step_name),
            );
            run_caught(|| (entry.handler)(ctx)).await
        };
        let result = match caught {
            CaughtResult::Ok(Ok(r)) => r,
            CaughtResult::Ok(Err(extract_err)) => {
                return self
                    .fail_forward_step(
                        root,
                        step_name,
                        format!("step handler extractor failed: {extract_err}"),
                    )
                    .await;
            }
            CaughtResult::Panicked(msg) => {
                return self
                    .fail_forward_step(root, step_name, format!("step handler panicked: {msg}"))
                    .await;
            }
        };

        // Value (compute) step: completes immediately with its serialised
        // return value — no dispatch, no await.
        let (dispatch, emit_timeout, build_error) = match result {
            crate::handler::ErasedStepOutcome::Value(Ok(output)) => {
                self.complete_step(root, entry, step_name, output);
                debug!(saga = S::name(), step = %step_name, "step completed");
                return Ok(StepOutcome::Completed {
                    terminal: self.entry_is_terminal(entry),
                });
            }
            crate::handler::ErasedStepOutcome::Value(Err(err)) => {
                return self
                    .fail_forward_step(root, step_name, format!("step build error: {err}"))
                    .await;
            }
            crate::handler::ErasedStepOutcome::Emit {
                dispatch,
                timeout,
                build_error,
            } => (dispatch, timeout, build_error),
        };

        // Emit step. Build error → fail the step + cascade.
        if let Some(err) = build_error {
            return self
                .fail_forward_step(root, step_name, format!("step build error: {err}"))
                .await;
        }

        // Durable emit audit (P1.2a): record EVERY dispatched action as a
        // `StepEmitted` envelope event — Regime A (in-process-handled),
        // Regime B (external), and fire-and-forget alike. The envelope
        // clears `emitted_requests[step]` when the step resolves, so the
        // pending-request semantics of `requests_for()` are unchanged; the
        // stream keeps the full journey for `emitted()` and operators.
        for action in &dispatch {
            if let Some(et) = action.event_type {
                root.record(SagaEnvelopeEvent::StepEmitted {
                    step: step_name.to_string(),
                    event_type: et.to_string(),
                    payload: action.payload.clone(),
                });
            }
        }

        // No `.on_response(...)` on the task → fire-and-forget: the step
        // completes with `Null` output as soon as its emits are recorded.
        let Some(spec) = &entry.await_spec else {
            self.complete_step(root, entry, step_name, serde_json::Value::Null);
            debug!(saga = S::name(), step = %step_name, "step completed (fire-and-forget)");
            return Ok(StepOutcome::Completed {
                terminal: self.entry_is_terminal(entry),
            });
        };
        let response_event_type = spec.event_type;

        // Effective timeout: the activation's `Emit::timeout` overrides the
        // task-level `.timeout(d)`, which overrides the saga's
        // `default_step_timeout`. Converted to an absolute wall-clock
        // deadline at suspend time; `tick_timeouts` later compares against
        // it.
        let deadline = self.effective_deadline(emit_timeout.or(entry.step_timeout));

        // A step that emits nothing but declares `.on_response(...)` (the
        // `routes_to_step` pattern) always suspends — the awaited event
        // arrives from outside (a signal or external worker), never from
        // an in-process handler this step invoked.
        if dispatch.is_empty() {
            debug!(saga = S::name(), step = %step_name, event_type = response_event_type, "step suspended (awaiting — routes_to_step)");
            return Ok(StepOutcome::Suspended(PendingAwait {
                step: step_name.to_string(),
                event_type: response_event_type.to_string(),
                is_compensation: false,
                deadline,
                // Emits nothing — it only awaits an externally-arriving event
                // (the `routes_to_step` / signal pattern), so no outbox row.
                outbox_emits: Vec::new(),
            }));
        }

        // Are all emitted request types handled in-process? If any
        // emit lacks a handler, the step goes external and we
        // suspend awaiting the response (Regime B). Mixing in-proc
        // and external emits on one step is rejected for clarity.
        let all_handled = dispatch.iter().all(|a| {
            a.event_type
                .map(|t| self.topology.request_handler(t).is_some())
                .unwrap_or(true)
        });
        let any_handled = dispatch.iter().any(|a| {
            a.event_type
                .map(|t| self.topology.request_handler(t).is_some())
                .unwrap_or(false)
        });

        if !all_handled {
            if any_handled {
                return Err(PgRunnerError::Unsupported {
                    step: step_name,
                    reason: "step mixes in-process-handled and external emits; split \
                             them into separate steps",
                });
            }
            // Strict mode: an emit with no registered handler that isn't
            // explicitly `.allow_unhandled::<T>()` is a loud failure
            // (forgot-to-register-a-handler footgun) rather than a
            // silent suspend.
            if self.topology.strict_handlers {
                let unallowed = dispatch.iter().find_map(|a| {
                    a.event_type.filter(|t| {
                        self.topology.request_handler(t).is_none()
                            && !self.topology.is_unhandled_allowed(t)
                    })
                });
                if let Some(event_type) = unallowed {
                    return self
                        .fail_forward_step(
                            root,
                            step_name,
                            format!(
                                "no effect handler registered for `{event_type}` (strict mode); \
                                 register via `.effect(...)` or opt out with `.allow_unhandled::<T>()`"
                            ),
                        )
                        .await;
                }
            }
            // Regime B: suspend. StepActivated and the `StepEmitted`
            // request payloads are already recorded (P1.2a above) —
            // committed in the same transaction as the await row, so an
            // external worker can recover *what* was requested (review
            // C5/SAGA-3/DX-1). We leave the step Active and persist the
            // await so a later `deliver(...)` can route the response.
            //
            // Capture the emitted external requests for the outbox (P3.2),
            // committed in the same tx as the await. Only dispatched actions
            // with an event_type are real requests (fire-and-forget actions
            // without one carry nothing to deliver).
            let outbox_emits = dispatch
                .iter()
                .filter_map(|a| {
                    a.event_type.map(|et| OutboxEmit {
                        event_type: et.to_string(),
                        payload: a.payload.clone(),
                    })
                })
                .collect();
            debug!(saga = S::name(), step = %step_name, event_type = response_event_type, "step suspended (awaiting — Regime B)");
            return Ok(StepOutcome::Suspended(PendingAwait {
                step: step_name.to_string(),
                event_type: response_event_type.to_string(),
                is_compensation: false,
                deadline,
                outbox_emits,
            }));
        }

        // Regime A: every emit has an in-process handler. Invoke
        // them, capture the awaited response, complete inline.
        let mut response_payload: Option<serde_json::Value> = None;
        for action in &dispatch {
            let Some(req_event_type) = action.event_type else {
                continue;
            };
            let handler =
                self.topology
                    .request_handler(req_event_type)
                    .ok_or(PgRunnerError::Internal(
                        "all_handled was true but handler vanished",
                    ))?;

            let state_clone = root.state.state.clone();
            let payload_clone = action.payload.clone();
            let caught = {
                let ctx = StepContext::for_testing_with_request(
                    &state_clone,
                    &payload_clone,
                    self.topology.deps_ref(),
                )
                .with_idempotency(
                    Self::instance_id_of(root),
                    step_name,
                    root.state.activation_seq(step_name),
                );
                run_caught(|| handler(ctx)).await
            };
            let response = match caught {
                CaughtResult::Ok(Ok(r)) => r,
                CaughtResult::Ok(Err(extract_err)) => {
                    return self
                        .fail_forward_step(
                            root,
                            step_name,
                            format!("request handler extractor failed: {extract_err}"),
                        )
                        .await;
                }
                CaughtResult::Panicked(msg) => {
                    return self
                        .fail_forward_step(
                            root,
                            step_name,
                            format!("request handler panicked: {msg}"),
                        )
                        .await;
                }
            };

            if response.is_error {
                // Documented contract: the typed Err event is emitted
                // into the log AND the awaiting step fails. Record the
                // error response durably before failing — `StepFailed`
                // clears `emitted_requests[step]` in the same commit, so
                // `requests_for()` never reports it as pending; it
                // survives on the stream for `emitted()` / audit.
                root.record(SagaEnvelopeEvent::StepEmitted {
                    step: step_name.to_string(),
                    event_type: response.event_type.to_string(),
                    payload: response.payload.clone(),
                });
                let reason = response
                    .error_reason
                    .clone()
                    .unwrap_or_else(|| "handler returned Err".into());
                return self.fail_forward_step(root, step_name, reason).await;
            }

            if response.event_type == response_event_type {
                response_payload = Some(response.payload);
            }
        }

        let Some(response_payload) = response_payload else {
            return Err(PgRunnerError::Unsupported {
                step: step_name,
                reason: "step awaited a response that no registered handler produced",
            });
        };

        // Apply the topology-registered response map. It runs user code
        // (the `.on_response` closure) — a failure is a step failure, not
        // an internal error.
        match self.apply_await_map(spec, response_payload) {
            Ok(output) => {
                self.complete_step(root, entry, step_name, output);
                debug!(saga = S::name(), step = %step_name, "step completed (Regime A inline)");
                Ok(StepOutcome::Completed {
                    terminal: self.entry_is_terminal(entry),
                })
            }
            Err(reason) => {
                debug!(saga = S::name(), step = %step_name, "step failed → compensation");
                self.fail_forward_step(root, step_name, reason).await
            }
        }
    }

    /// Apply a topology-registered response map to a delivered payload,
    /// catching panics from the user's closure. `Err` carries the failure
    /// reason for `StepError::Failed`.
    fn apply_await_map(
        &self,
        spec: &crate::builder::AwaitSpec,
        payload: serde_json::Value,
    ) -> Result<serde_json::Value, String> {
        let map = std::sync::Arc::clone(&spec.map);
        match std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || map(payload))) {
            Ok(Ok(output)) => Ok(output),
            Ok(Err(e)) => Err(format!("response map failed: {e}")),
            Err(_) => Err("response map panicked".to_string()),
        }
    }

    /// A forward step failed (handler panic, extractor failure, build error,
    /// or an `Err` response): record `StepFailed` and run the compensation
    /// cascade (D5), returning a terminal outcome. This brings the pg runner
    /// in line with the saga contract — a forward-handler problem fails
    /// the step and compensates rather than aborting the call uncommitted
    /// (which left the saga stuck) or unwinding a panic through the host
    /// (SAGA-6/SAGA-7). Assumes `StepActivated` for `step_name` is already
    /// recorded.
    async fn fail_forward_step(
        &self,
        root: &mut AggregateRoot<SagaEnvelope<S>>,
        step_name: StepName,
        reason: String,
    ) -> Result<StepOutcome, PgRunnerError> {
        let err = StepError::Failed {
            step: step_name.to_string(),
            reason,
        };
        root.record(SagaEnvelopeEvent::StepFailed {
            step: step_name.to_string(),
            error: err.clone(),
        });
        self.run_compensation_cascade(root, err).await?;
        Ok(StepOutcome::Completed { terminal: true })
    }

    fn complete_step(
        &self,
        root: &mut AggregateRoot<SagaEnvelope<S>>,
        entry: &crate::builder::StepEntry<S>,
        step_name: StepName,
        output: serde_json::Value,
    ) {
        root.record(SagaEnvelopeEvent::StepCompleted {
            step: step_name.to_string(),
            output,
        });
        if let Some(terminal) = entry.terminal {
            let ev = match terminal {
                Terminal::Success => SagaEnvelopeEvent::Confirmed,
                Terminal::Failure => SagaEnvelopeEvent::Cancelled {
                    reason: format!("terminal-failure step `{step_name}` completed"),
                },
            };
            root.record(ev);
        }
    }

    fn entry_is_terminal(&self, entry: &crate::builder::StepEntry<S>) -> bool {
        entry.terminal.is_some()
    }

    /// Compute the absolute deadline for a suspending step. Step
    /// `.timeout(d)` overrides the saga's `default_step_timeout`;
    /// neither set → `None` (waits forever).
    fn effective_deadline(
        &self,
        step_timeout: Option<std::time::Duration>,
    ) -> Option<chrono::DateTime<chrono::Utc>> {
        let d = step_timeout.or(self.topology.default_step_timeout)?;
        let chrono_d = chrono::Duration::from_std(d).ok()?;
        Some(chrono::Utc::now() + chrono_d)
    }

    /// Expire any awaits whose deadline has passed as of `now`. For
    /// each expired await: load the instance, fail the step with
    /// [`StepError::Timeout`], run the compensation cascade, commit.
    ///
    /// `now` is supplied explicitly (rather than read internally) so
    /// tests can force expiry without real-time waits — the durable
    /// deterministic analogue of advancing a test clock. A future
    /// lease-driven worker loop calls this on a schedule with
    /// `Utc::now()`.
    ///
    /// Returns the number of instances whose step timed out.
    #[instrument(level = "debug", skip_all, fields(saga = S::name()))]
    pub async fn tick_timeouts(
        &self,
        now: chrono::DateTime<chrono::Utc>,
    ) -> Result<usize, PgRunnerError> {
        // Find expired, non-compensation awaits across ALL shards.
        // (Compensation awaits are best-effort and don't time out into a
        // cascade — a comp that never completes is an ops concern, not a
        // re-cascade.) Host-driven callers and single-shard deployments use
        // this; a sharded worker fleet uses `tick_timeouts_in_shards`.
        let rows: Vec<ExpiredAwait> = sqlx::query_as(
            "SELECT instance_id, step, deadline_at, created_at FROM es_saga_awaits \
             WHERE saga_name = $1 AND is_compensation = false \
               AND deadline_at IS NOT NULL AND deadline_at <= $2",
        )
        .bind(S::name())
        .bind(now)
        .fetch_all(self.store.pool())
        .await
        .map_err(|e| PgRunnerError::Store(e.into()))?;

        self.process_expired_awaits(rows).await
    }

    /// Like [`tick_timeouts`](Self::tick_timeouts) but only scans awaits in
    /// the given shards — the [`SagaWorker`](crate::SagaWorker) fleet's path,
    /// so each worker only does the timeout work for the shards it owns
    /// (plan §P4.3). Uses the `(saga_name, shard_id, deadline_at)` partial
    /// index. An empty `shards` slice is a no-op (a worker that owns nothing
    /// this tick). Shard ownership is *work distribution*; correctness still
    /// rests on the per-instance lease this acquires below, so a shard
    /// briefly scanned by two workers is safe.
    #[instrument(level = "debug", skip_all, fields(saga = S::name()))]
    pub async fn tick_timeouts_in_shards(
        &self,
        now: chrono::DateTime<chrono::Utc>,
        shards: &[i32],
    ) -> Result<usize, PgRunnerError> {
        if shards.is_empty() {
            return Ok(0);
        }
        let rows: Vec<ExpiredAwait> = sqlx::query_as(
            "SELECT instance_id, step, deadline_at, created_at FROM es_saga_awaits \
             WHERE saga_name = $1 AND is_compensation = false \
               AND deadline_at IS NOT NULL AND deadline_at <= $2 \
               AND shard_id = ANY($3)",
        )
        .bind(S::name())
        .bind(now)
        .bind(shards)
        .fetch_all(self.store.pool())
        .await
        .map_err(|e| PgRunnerError::Store(e.into()))?;

        self.process_expired_awaits(rows).await
    }

    /// Drive each `(instance_id, step)` expired await to its timeout cascade
    /// under the instance lease. Shared by the all-shards and shard-filtered
    /// scans — only the row selection differs.
    #[instrument(level = "debug", skip_all, fields(saga = S::name()))]
    async fn process_expired_awaits(
        &self,
        rows: Vec<ExpiredAwait>,
    ) -> Result<usize, PgRunnerError> {
        let mut timed_out = 0usize;
        for row in rows {
            let instance_id = row.instance_id;
            // The configured budget (P5.4), derived from the stored timestamps
            // — `deadline_at = created_at + timeout`. Clamp negatives (clock
            // skew) to 0; it's an error-detail field, not a control value.
            let timeout_ms = (row.deadline_at - row.created_at).num_milliseconds().max(0) as u64;
            // Acquire the instance lease (non-blocking): if a deliver/signal
            // or another worker's tick is processing this instance, skip it —
            // its timeout is re-checked on a later tick (at-least-once). This
            // keeps a timeout's compensation effects from running concurrently
            // with a delivery's effects (effects-vs-commit §11).
            let Some(fence) = self.try_acquire_lease(&instance_id).await? else {
                continue;
            };
            let effect = self.tick_one(&instance_id, &row.step, timeout_ms);
            let outcome = self.run_with_heartbeat(&instance_id, fence, effect).await;
            self.release_lease(&instance_id, fence).await;
            if outcome? {
                timed_out += 1;
            }
        }
        Ok(timed_out)
    }

    /// Re-drive instances that crashed mid-transition (P3.3 lazy recovery)
    /// across ALL shards. A worker/host that dies between an instance's
    /// commits leaves it non-terminal with **no pending await row** — nothing
    /// else will ever wake it (no deadline to tick, no response to deliver).
    /// This scans `es_saga_instances` (the P3.1 status projection) for such
    /// stranded instances and re-drives each to quiescence under its lease.
    /// Covers the three documented crash gaps: after `Started` before the
    /// first drive; between `tick_one`'s claim commit and its cascade commit;
    /// and after a delivery's completion commit before the downstream drive
    /// (the C3-deliver gap).
    ///
    /// Liveness is the **lease**, not a staleness clock (phase 7): an instance
    /// is a candidate iff it is non-terminal, has no pending await (so it isn't
    /// legitimately suspended), and has **no live lease** — no worker is
    /// driving it. A crashed worker's lease lapses after the (short) TTL and
    /// the instance becomes fair game, so recovery latency tracks the TTL
    /// rather than a fixed 30-60s. [`process_recoverable`] then *steals* the
    /// lease (acquire bumps the fence) before re-driving, so a worker that
    /// acquires the lease between this scan and the steal is never raced.
    /// Returns the count re-driven. Host-driven/single-shard callers use this;
    /// a sharded fleet uses [`recover_in_shards`](Self::recover_in_shards).
    #[instrument(level = "debug", skip_all, fields(saga = S::name()))]
    pub async fn recover(&self) -> Result<usize, PgRunnerError> {
        let rows: Vec<String> = sqlx::query_scalar(
            "SELECT i.instance_id FROM es_saga_instances i \
             WHERE i.saga_name = $1 \
               AND i.status IN ('not-started','running') \
               AND NOT EXISTS ( \
                   SELECT 1 FROM es_saga_awaits a \
                   WHERE a.saga_name = i.saga_name AND a.instance_id = i.instance_id \
               ) \
               AND NOT EXISTS ( \
                   SELECT 1 FROM es_saga_leases l \
                   WHERE l.saga_name = i.saga_name AND l.instance_id = i.instance_id \
                     AND l.leased_until > now() \
               ) \
             LIMIT 500",
        )
        .bind(S::name())
        .fetch_all(self.store.pool())
        .await
        .map_err(|e| PgRunnerError::Store(e.into()))?;

        self.process_recoverable(rows).await
    }

    /// Like [`recover`](Self::recover) but only scans the given shards — the
    /// [`SagaWorker`](crate::SagaWorker) fleet's path, so each worker recovers
    /// only the instances it owns (plan §P4.3 — uses
    /// `idx_saga_instances_recovery`'s `shard_id`). An empty `shards` slice is
    /// a no-op.
    #[instrument(level = "debug", skip_all, fields(saga = S::name()))]
    pub async fn recover_in_shards(&self, shards: &[i32]) -> Result<usize, PgRunnerError> {
        if shards.is_empty() {
            return Ok(0);
        }
        let rows: Vec<String> = sqlx::query_scalar(
            "SELECT i.instance_id FROM es_saga_instances i \
             WHERE i.saga_name = $1 \
               AND i.status IN ('not-started','running') \
               AND i.shard_id = ANY($2) \
               AND NOT EXISTS ( \
                   SELECT 1 FROM es_saga_awaits a \
                   WHERE a.saga_name = i.saga_name AND a.instance_id = i.instance_id \
               ) \
               AND NOT EXISTS ( \
                   SELECT 1 FROM es_saga_leases l \
                   WHERE l.saga_name = i.saga_name AND l.instance_id = i.instance_id \
                     AND l.leased_until > now() \
               ) \
             LIMIT 500",
        )
        .bind(S::name())
        .bind(shards)
        .fetch_all(self.store.pool())
        .await
        .map_err(|e| PgRunnerError::Store(e.into()))?;

        self.process_recoverable(rows).await
    }

    /// Re-drive each stranded instance under its lease. Shared by the
    /// all-shards and shard-filtered recovery scans.
    #[instrument(level = "debug", skip_all, fields(saga = S::name()))]
    async fn process_recoverable(&self, rows: Vec<String>) -> Result<usize, PgRunnerError> {
        let mut recovered = 0usize;
        for instance_id in rows {
            // Non-blocking lease: skip if a live worker/host call holds it
            // (that processor is healing it). At-least-once: re-checked next
            // scan. The heartbeat keeps exclusivity across a slow re-drive.
            let Some(fence) = self.try_acquire_lease(&instance_id).await? else {
                continue;
            };
            let effect = self.recover_one(&instance_id);
            let outcome = self.run_with_heartbeat(&instance_id, fence, effect).await;
            self.release_lease(&instance_id, fence).await;
            if outcome? {
                recovered += 1;
            }
        }
        Ok(recovered)
    }

    /// Recover one stranded instance under the lease (acquired by the caller).
    /// `Ok(true)` if a transition was re-driven, `Ok(false)` if it raced to
    /// terminal / decode-failed / OCC-conflicted (a live processor won).
    #[instrument(level = "debug", skip_all, fields(saga = S::name(), instance_id))]
    async fn recover_one(&self, instance_id: &str) -> Result<bool, PgRunnerError> {
        let Some(mut root) = self
            .store
            .load::<SagaEnvelope<S>>(instance_id)
            .await
            .map_err(PgRunnerError::Store)?
        else {
            return Ok(false);
        };
        if let Some(reason) = root.state.state_decode_error() {
            tracing::error!(
                instance_id = %instance_id,
                reason = %reason,
                "skipping recovery: saga state failed to decode (schema drift)"
            );
            return Ok(false);
        }
        // Raced to terminal since the scan (a live deliver/tick won) — the
        // status upsert was atomic with that commit, so the projection is
        // already correct; nothing to do.
        if root.state.status.is_terminal() {
            return Ok(false);
        }

        // A `StepFailed` is recorded (originating_error set) but the instance
        // isn't terminal → the failure committed (tick_one's claim, or a
        // failed forward step) but its compensation cascade never committed.
        // Re-run the cascade and commit. Idempotent: the cascade queue is
        // rebuilt from currently-`Completed` steps, and none of the prior
        // (uncommitted) cascade events are on the envelope.
        if let Some(cause) = root.state.originating_error.clone() {
            self.run_compensation_cascade(&mut root, cause).await?;
            return match self.commit(instance_id, &mut root, &[], &[]).await {
                Ok(()) => {
                    debug!(
                        saga = S::name(),
                        instance_id, "re-drove stranded instance (compensation)"
                    );
                    Ok(true)
                }
                Err(PgRunnerError::Store(EventStoreError::ConcurrencyConflict { .. })) => Ok(false),
                Err(e) => Err(e),
            };
        }

        // Otherwise the instance stalled mid-forward-progress (crash after
        // `Started`, or after a delivery's completion commit before the
        // downstream drive). Drive it forward to quiescence and commit any
        // new activations/awaits. At-least-once: an effect that already fired
        // before the crash may re-fire — the IdempotencyKey is the dedupe.
        let trigger = root
            .state
            .trigger
            .clone()
            .unwrap_or(serde_json::Value::Null);
        let pending = self.drive_to_quiescence(&mut root, &trigger).await?;
        match self.commit(instance_id, &mut root, &pending, &[]).await {
            Ok(()) => {}
            Err(PgRunnerError::Store(EventStoreError::ConcurrencyConflict { .. })) => {
                return Ok(false);
            }
            Err(e) => return Err(e),
        }
        // A step may have suspended on a signal that arrived early.
        self.drain_buffered_signals(instance_id).await?;
        debug!(saga = S::name(), instance_id, "re-drove stranded instance");
        Ok(true)
    }

    /// Publish unpublished `es_saga_outbox` rows across ALL shards via the
    /// configured [`EmitPublisher`](crate::EmitPublisher), then mark each
    /// delivered (P3.2). No-op (returns 0) if no publisher is configured.
    /// Host-driven/single-shard callers use this; a sharded worker fleet uses
    /// [`drain_outbox_in_shards`](Self::drain_outbox_in_shards).
    #[instrument(level = "debug", skip_all, fields(saga = S::name()))]
    pub async fn drain_outbox(&self) -> Result<usize, PgRunnerError> {
        if self.publisher.is_none() {
            return Ok(0);
        }
        let rows: Vec<(i64, String, String, String, serde_json::Value)> = sqlx::query_as(
            "SELECT id, instance_id, step, event_type, payload FROM es_saga_outbox \
             WHERE saga_name = $1 AND published_at IS NULL \
             ORDER BY id LIMIT 200",
        )
        .bind(S::name())
        .fetch_all(self.store.pool())
        .await
        .map_err(|e| PgRunnerError::Store(e.into()))?;
        self.process_outbox(rows).await
    }

    /// Like [`drain_outbox`](Self::drain_outbox) but only the given shards —
    /// the [`SagaWorker`](crate::SagaWorker) fleet's path, so each worker
    /// publishes only the emits for the shards it owns (uses
    /// `idx_saga_outbox_unpublished`). Empty `shards` ⇒ no-op.
    #[instrument(level = "debug", skip_all, fields(saga = S::name()))]
    pub async fn drain_outbox_in_shards(&self, shards: &[i32]) -> Result<usize, PgRunnerError> {
        if self.publisher.is_none() || shards.is_empty() {
            return Ok(0);
        }
        let rows: Vec<(i64, String, String, String, serde_json::Value)> = sqlx::query_as(
            "SELECT id, instance_id, step, event_type, payload FROM es_saga_outbox \
             WHERE saga_name = $1 AND published_at IS NULL AND shard_id = ANY($2) \
             ORDER BY id LIMIT 200",
        )
        .bind(S::name())
        .bind(shards)
        .fetch_all(self.store.pool())
        .await
        .map_err(|e| PgRunnerError::Store(e.into()))?;
        self.process_outbox(rows).await
    }

    /// Publish each scanned outbox row, then stamp `published_at` (publish-
    /// then-mark ⇒ at-least-once). A publish error leaves the row unpublished
    /// for the next drain and we continue to the rest — best-effort, so one
    /// failing destination doesn't stall the whole batch; consumers dedupe
    /// (delivery is at-least-once, not strictly ordered across failures).
    #[instrument(level = "debug", skip_all, fields(saga = S::name()))]
    async fn process_outbox(
        &self,
        rows: Vec<(i64, String, String, String, serde_json::Value)>,
    ) -> Result<usize, PgRunnerError> {
        let Some(publisher) = &self.publisher else {
            return Ok(0);
        };
        let mut published = 0usize;
        for (id, instance_id, step, event_type, payload) in rows {
            let msg = crate::publisher::OutboxMessage {
                saga_name: S::name(),
                instance_id: &instance_id,
                step: &step,
                event_type: &event_type,
                payload: &payload,
            };
            match publisher.publish_dyn(msg).await {
                Ok(()) => {
                    sqlx::query("UPDATE es_saga_outbox SET published_at = now() WHERE id = $1")
                        .bind(id)
                        .execute(self.store.pool())
                        .await
                        .map_err(|e| PgRunnerError::Store(e.into()))?;
                    trace!(
                        saga = S::name(),
                        instance_id = %instance_id,
                        step = %step,
                        event_type = %event_type,
                        "published outbox row",
                    );
                    published += 1;
                }
                Err(e) => {
                    tracing::warn!(
                        saga = S::name(),
                        instance_id = %instance_id,
                        step = %step,
                        error = %e,
                        "outbox publish failed; leaving unpublished for retry"
                    );
                }
            }
        }
        Ok(published)
    }

    /// Process one expired await under the instance lease (acquired by the
    /// caller). Returns `Ok(true)` if the step timed out and its cascade ran,
    /// `Ok(false)` if skipped (already resolved / terminal / raced).
    #[instrument(level = "debug", skip_all, fields(saga = S::name(), instance_id, step, timeout_ms))]
    async fn tick_one(
        &self,
        instance_id: &str,
        step: &str,
        timeout_ms: u64,
    ) -> Result<bool, PgRunnerError> {
        // Load; skip if the instance already moved on.
        let Some(mut root) = self
            .store
            .load::<SagaEnvelope<S>>(instance_id)
            .await
            .map_err(PgRunnerError::Store)?
        else {
            return Ok(false);
        };

        // Schema drift: never drive timeouts/compensation for an instance
        // whose state no longer decodes (C6/DX-3).
        if let Some(reason) = root.state.state_decode_error() {
            tracing::error!(
                instance_id = %instance_id,
                reason = %reason,
                "skipping timeout: saga state failed to decode (schema drift)"
            );
            return Ok(false);
        }

        let Some(step_name) = self
            .topology
            .step_entries()
            .iter()
            .map(|e| e.name)
            .find(|n| *n == step)
        else {
            return Ok(false);
        };

        // Already terminal (e.g. a sibling branch cancelled the instance), or
        // the step already resumed — nothing to time out.
        if root.state.status.is_terminal() {
            return Ok(false);
        }
        if !matches!(root.state.step_status(step_name), SagaStepStatus::Active) {
            return Ok(false);
        }

        let err = StepError::Timeout {
            step: step_name.to_string(),
            timeout_ms, // configured budget, derived from the await's timestamps (P5.4)
        };
        // CLAIM (C4/SAGA-2): commit `StepFailed` + delete the expired await
        // BEFORE running compensation. Under the lease this is uncontended;
        // OCC remains the backstop. A conflict means someone else resolved the
        // instance → run no compensation (a timeout must never refund/release
        // a saga that actually succeeded).
        root.record(SagaEnvelopeEvent::StepFailed {
            step: step_name.to_string(),
            error: err.clone(),
        });
        let consumed = [PendingAwait {
            step: step_name.to_string(),
            event_type: String::new(),
            is_compensation: false,
            outbox_emits: Vec::new(), // delete marker — carries no emit
            deadline: None,
        }];
        match self.commit(instance_id, &mut root, &[], &consumed).await {
            Ok(()) => {}
            Err(PgRunnerError::Store(EventStoreError::ConcurrencyConflict { .. })) => {
                return Ok(false);
            }
            Err(e) => return Err(e),
        }

        // EFFECT + RESOLVE: the failure is the committed winning transition,
        // so run the compensation cascade and commit it. A crash between these
        // two commits leaves the instance `StepFailed` but not yet terminal;
        // lazy recovery (deferred) re-drives the cascade.
        debug!(
            saga = S::name(),
            instance_id, step, timeout_ms, "step timed out → cascade"
        );
        self.run_compensation_cascade(&mut root, err).await?;
        match self.commit(instance_id, &mut root, &[], &[]).await {
            Ok(()) => Ok(true),
            Err(PgRunnerError::Store(EventStoreError::ConcurrencyConflict { .. })) => {
                tracing::warn!(
                    instance_id = %instance_id,
                    "compensation commit conflicted after claim; leaving for recovery"
                );
                Ok(false)
            }
            Err(e) => Err(e),
        }
    }

    /// Walk completed steps in reverse declaration order, invoke each
    /// registered compensation handler, and record the cascade. Ends
    /// with `Cancelled` (terminal failure).
    ///
    /// **D5/SAGA-5 semantics:** compensation runs to completion over *every*
    /// step — one failed rollback must never block the others (the refund
    /// still goes out even if a seat can't be unbooked). A comp that errors,
    /// has a build error, panics ([`run_caught`], SAGA-6), or whose dispatch
    /// fails is recorded `StepCompensationFailed` (a durable lock) and the
    /// cascade continues. Crucially this method **never returns `Err` for a
    /// compensation failure** — it records the failure as an event so the
    /// caller's single commit persists the whole cascade. The old behavior
    /// aborted uncommitted, leaving the await row in place so every
    /// subsequent tick re-ran the cascade forever (the SAGA-5 hot loop).
    /// If any step ends `CompensationFailed`, the envelope derives
    /// `FailedTerminalNeedsIntervention` from the trailing `Cancelled`.
    #[instrument(level = "debug", skip(self, root, cause), fields(saga = S::name()))]
    async fn run_compensation_cascade(
        &self,
        root: &mut AggregateRoot<SagaEnvelope<S>>,
        cause: StepError,
    ) -> Result<(), PgRunnerError> {
        use crate::envelope::SagaStepStatus;
        use crate::guard::{CaughtResult, run_caught};

        // Build the queue once from the in-memory envelope. Reverse
        // declaration order = correct cascade direction. Completed steps
        // with a registered compensation always join; in-flight steps
        // (Active/Resolving — external work pending) join only when opted
        // in via `compensate_even_if_active` (their compensation must not
        // read the step's nonexistent output). The rest of the in-flight
        // steps are recorded as `Abandoned` so the leaked external work is
        // visible instead of silently stuck Active (review SAGA-13).
        let in_flight =
            |s: &SagaStepStatus| matches!(s, SagaStepStatus::Active | SagaStepStatus::Resolving);
        let comp_queue: Vec<StepName> = self
            .topology
            .step_entries()
            .iter()
            .rev()
            .filter(|e| {
                let status = root.state.step_status(e.name);
                (matches!(status, SagaStepStatus::Completed)
                    && self.topology.has_compensation(e.name))
                    || (in_flight(&status) && self.topology.compensates_even_if_active(e.name))
            })
            .map(|e| e.name)
            .collect();

        for entry in self.topology.step_entries() {
            if in_flight(&root.state.step_status(entry.name))
                && !self.topology.compensates_even_if_active(entry.name)
            {
                root.record(SagaEnvelopeEvent::StepAbandoned {
                    step: entry.name.to_string(),
                    reason: format!("in flight when the cascade fired: {cause}"),
                });
            }
        }

        for comp_step in comp_queue {
            debug!(saga = S::name(), step = comp_step, "compensating step");
            root.record(SagaEnvelopeEvent::StepCompensating {
                step: comp_step.to_string(),
            });

            let handler =
                self.topology
                    .compensation_handler(comp_step)
                    .ok_or(PgRunnerError::Internal(
                        "comp queue contained step with no registered compensation",
                    ))?;

            // Invoke the comp handler with panic catching. `Cause` carries
            // the originating error; `Output<T>` works via the outputs map.
            let caught = {
                let outputs = build_outputs_by_type_id(&root.state, self.topology.step_entries());
                let ctx = StepContext::for_testing(
                    &root.state.state,
                    None,
                    &outputs,
                    self.topology.deps_ref(),
                    Some(&cause),
                )
                .with_idempotency(
                    Self::instance_id_of(root),
                    comp_step,
                    root.state.compensation_seq(comp_step),
                );
                run_caught(|| handler(ctx)).await
            };

            // Resolve to Ok (compensated cleanly) or Err(reason) (flag the
            // step CompensationFailed and keep going).
            let outcome: Result<(), String> = match caught {
                CaughtResult::Ok(Ok(result)) if result.build_error.is_some() => Err(format!(
                    "compensation build error: {}",
                    result.build_error.unwrap()
                )),
                CaughtResult::Ok(Ok(result)) => {
                    // Durable emit audit (P1.2a): compensation emits are part
                    // of the journey too — record them before dispatching.
                    for action in &result.dispatch {
                        if let Some(et) = action.event_type {
                            root.record(SagaEnvelopeEvent::StepEmitted {
                                step: comp_step.to_string(),
                                event_type: et.to_string(),
                                payload: action.payload.clone(),
                            });
                        }
                    }
                    // The comp handler may emit + await; drive each emit
                    // through its registered request handler. Catch panics
                    // and errors here too (D5).
                    match run_caught(|| {
                        self.drive_comp_dispatch(
                            comp_step,
                            Self::instance_id_of(root),
                            &root.state,
                            &result,
                        )
                    })
                    .await
                    {
                        CaughtResult::Ok(Ok(())) => Ok(()),
                        CaughtResult::Ok(Err(e)) => {
                            Err(format!("compensation dispatch failed: {e}"))
                        }
                        CaughtResult::Panicked(msg) => {
                            Err(format!("compensation dispatch panicked: {msg}"))
                        }
                    }
                }
                CaughtResult::Ok(Err(extract_err)) => {
                    Err(format!("compensation extractor failed: {extract_err}"))
                }
                CaughtResult::Panicked(msg) => Err(format!("compensation handler panicked: {msg}")),
            };

            match outcome {
                Ok(()) => {
                    debug!(saga = S::name(), step = comp_step, "step compensated");
                    root.record(SagaEnvelopeEvent::StepCompensated {
                        step: comp_step.to_string(),
                    });
                }
                Err(reason) => {
                    tracing::error!(
                        saga = S::name(),
                        step = comp_step,
                        reason = %reason,
                        "compensation failed; flagging CompensationFailed (manual intervention required)"
                    );
                    root.record(SagaEnvelopeEvent::StepCompensationFailed {
                        step: comp_step.to_string(),
                        error: StepError::Failed {
                            step: comp_step.to_string(),
                            reason,
                        },
                    });
                }
            }
        }

        // Cascade done → saga is cancelled. The envelope derives
        // `FailedTerminalNeedsIntervention` if any step is CompensationFailed.
        let reason = format!("{cause}");
        root.record(SagaEnvelopeEvent::Cancelled { reason });
        Ok(())
    }

    /// Dispatch a compensation handler's emits + await response.
    /// Mirrors the forward-step emit-dispatch path in
    /// [`Self::activate_step`] but ignores any typed-output mapping
    /// (StepCompensated carries no payload — we just need the awaited
    /// response to arrive so we know the comp finished).
    ///
    /// First slice of comp-with-emits: every emit needs an in-process
    /// handler. External-worker dispatch lands when the outbox
    /// projection (Chunk 4c.2) ships and the comp queue uses it too.
    async fn drive_comp_dispatch(
        &self,
        comp_step: StepName,
        instance_id: &str,
        envelope: &SagaEnvelope<S>,
        comp: &crate::handler::ErasedCompensationResult,
    ) -> Result<(), PgRunnerError> {
        // Nothing dispatched → comp is noop; succeed immediately.
        if comp.dispatch.is_empty() {
            return Ok(());
        }

        // We need to see ONE response with the awaited event_type
        // (declared via `comp(...).on_response::<E>()` on the topology).
        // Multiple emits may fire (e.g. parallel releases), but only
        // the one matching it signals completion.
        let comp_await = self.topology.compensation_await_type(comp_step);
        let mut awaited_response_seen = comp_await.is_none();

        for action in &comp.dispatch {
            let Some(req_event_type) = action.event_type else {
                continue;
            };
            let Some(handler) = self.topology.request_handler(req_event_type) else {
                return Err(PgRunnerError::Unsupported {
                    step: comp_step,
                    reason: "compensation handler emitted an event with no registered \
                             in-process handler; external-worker comp dispatch lands with \
                             the outbox projection",
                });
            };

            let state_clone = envelope.state.clone();
            let payload_clone = action.payload.clone();
            let response = {
                let ctx = StepContext::for_testing_with_request(
                    &state_clone,
                    &payload_clone,
                    self.topology.deps_ref(),
                )
                .with_idempotency(
                    instance_id,
                    comp_step,
                    envelope.compensation_seq(comp_step),
                );
                handler(ctx).await.map_err(PgRunnerError::Extract)?
            };

            if response.is_error {
                // Comp handler's emit got an Err. Compensation is a
                // best-effort cascade — log and keep going so the other
                // comps still run; the cascade records StepCompensated
                // for this comp and logs the error here.
                tracing::warn!(
                    saga = S::name(),
                    step = %comp_step,
                    err = %response.error_reason.clone().unwrap_or_default(),
                    "compensation request handler returned Err; advancing cascade anyway",
                );
                continue;
            }

            if let Some(awaited) = comp_await
                && response.event_type == awaited
            {
                awaited_response_seen = true;
            }
        }

        if !awaited_response_seen {
            return Err(PgRunnerError::Unsupported {
                step: comp_step,
                reason: "compensation awaited a response that no registered handler \
                         produced",
            });
        }
        Ok(())
    }
}

// =====================================================================
// Inspection — one-call snapshots of an instance's durable state, for
// tests, debugging, and operator tooling (P1.2). All read-only: they
// load the envelope / query the awaits projection, never take the
// instance lease.
// =====================================================================

impl<S: Saga> PgSagaRunner<S> {
    /// The instance's overall status, or `None` if no such instance exists.
    pub async fn status(
        &self,
        instance_id: &str,
    ) -> Result<Option<SagaInstanceStatus>, PgRunnerError> {
        Ok(self.load(instance_id).await?.map(|env| env.status))
    }

    /// `true` if the instance has reached a terminal state (succeeded or
    /// failed). `false` if running or unknown.
    pub async fn is_terminal(&self, instance_id: &str) -> Result<bool, PgRunnerError> {
        Ok(self
            .status(instance_id)
            .await?
            .map(|s| s.is_terminal())
            .unwrap_or(false))
    }

    /// `true` if the instance reached `SucceededTerminal`.
    pub async fn is_succeeded(&self, instance_id: &str) -> Result<bool, PgRunnerError> {
        Ok(matches!(
            self.status(instance_id).await?,
            Some(SagaInstanceStatus::SucceededTerminal)
        ))
    }

    /// `true` if the instance reached `FailedTerminal`.
    pub async fn is_failed(&self, instance_id: &str) -> Result<bool, PgRunnerError> {
        Ok(matches!(
            self.status(instance_id).await?,
            Some(SagaInstanceStatus::FailedTerminal)
        ))
    }

    /// Status of one step within an instance. `Pending` for unknown
    /// instances or steps.
    pub async fn step_status(
        &self,
        instance_id: &str,
        step: StepName,
    ) -> Result<SagaStepStatus, PgRunnerError> {
        Ok(self
            .load(instance_id)
            .await?
            .map(|env| env.step_status(step))
            .unwrap_or(SagaStepStatus::Pending))
    }

    /// Steps currently `Active` (activated, awaiting completion), in
    /// declaration order.
    pub async fn active_steps(&self, instance_id: &str) -> Result<Vec<StepName>, PgRunnerError> {
        let Some(envelope) = self.load(instance_id).await? else {
            return Ok(Vec::new());
        };
        Ok(self
            .topology
            .step_entries()
            .iter()
            .map(|e| e.name)
            .filter(|name| matches!(envelope.step_status(name), SagaStepStatus::Active))
            .collect())
    }

    /// All events the instance has emitted so far, in stream order —
    /// replayed from the `StepEmitted` events on the envelope stream.
    /// The durable journey log: includes in-process-handled (Regime A),
    /// external (Regime B), and compensation emits (P1.2a).
    pub async fn emitted(&self, instance_id: &str) -> Result<Vec<EmittedEvent>, PgRunnerError> {
        let stream_id = format!("{}-{}", S::name(), instance_id);
        let events = self
            .store
            .read_stream_all(&stream_id, 0)
            .await
            .map_err(PgRunnerError::Store)?;
        let mut out = Vec::new();
        for recorded in events {
            if recorded.event_type != "saga.step-emitted" {
                continue;
            }
            let ev: SagaEnvelopeEvent =
                serde_json::from_value(recorded.data.clone()).map_err(|e| {
                    PgRunnerError::InternalOwned(format!(
                        "undecodable envelope event at position {}: {e}",
                        recorded.global_position
                    ))
                })?;
            if let SagaEnvelopeEvent::StepEmitted {
                step,
                event_type,
                payload,
            } = ev
            {
                out.push(EmittedEvent {
                    instance_id: instance_id.to_string(),
                    source_step: step,
                    sequence: out.len() as i64 + 1,
                    event_type,
                    payload,
                });
            }
        }
        Ok(out)
    }

    /// A one-call human-readable snapshot of an instance — step statuses,
    /// outstanding awaits (with deadline state), emit count, errors.
    /// Returns `None` if the instance doesn't exist.
    pub async fn describe(
        &self,
        instance_id: &str,
    ) -> Result<Option<InstanceDiagnostic>, PgRunnerError> {
        let Some(envelope) = self.load(instance_id).await? else {
            return Ok(None);
        };

        let rows: Vec<(String, String, Option<chrono::DateTime<chrono::Utc>>, bool)> =
            sqlx::query_as(
                "SELECT step, event_type, deadline_at, is_compensation \
             FROM es_saga_awaits \
             WHERE saga_name = $1 AND instance_id = $2 \
             ORDER BY step",
            )
            .bind(S::name())
            .bind(instance_id)
            .fetch_all(self.store.pool())
            .await
            .map_err(|e| PgRunnerError::Store(e.into()))?;
        let now = chrono::Utc::now();
        let awaiting = rows
            .into_iter()
            .map(
                |(step, event_type, deadline_at, is_compensation)| AwaitDescription {
                    step,
                    event_type,
                    deadline_passed: deadline_at.map(|d| d <= now).unwrap_or(false),
                    is_compensation,
                },
            )
            .collect();

        let steps: Vec<(StepName, SagaStepStatus)> = self
            .topology
            .step_entries()
            .iter()
            .map(|e| (e.name, envelope.step_status(e.name)))
            .collect();
        // Compensations still in flight, in cascade order (reverse
        // declaration order) — the durable analogue of the old runner's
        // in-memory queue.
        let compensation_queue: Vec<StepName> = self
            .topology
            .step_entries()
            .iter()
            .rev()
            .filter(|e| matches!(envelope.step_status(e.name), SagaStepStatus::Compensating))
            .map(|e| e.name)
            .collect();
        let emit_count = self.emitted(instance_id).await?.len();

        Ok(Some(InstanceDiagnostic {
            saga: S::name(),
            instance_id: instance_id.to_string(),
            status: envelope.status,
            steps,
            awaiting,
            emit_count,
            compensation_queue,
            originating_error: envelope.originating_error.clone(),
            last_error: envelope.last_error.clone(),
        }))
    }

    /// Render the saga's static topology as a Mermaid `flowchart TD` —
    /// same output as [`SagaBuilder::mermaid`].
    pub fn mermaid(&self) -> String {
        self.topology.mermaid()
    }

    /// Render this instance's DAG with each step styled by its current
    /// status — colored boxes for Active/Completed/Failed/Skipped/etc.
    /// Drop the output into a Mermaid renderer to see exactly where a
    /// stuck saga is sitting. Returns `None` if the instance doesn't
    /// exist.
    pub async fn mermaid_instance(
        &self,
        instance_id: &str,
    ) -> Result<Option<String>, PgRunnerError> {
        let Some(envelope) = self.load(instance_id).await? else {
            return Ok(None);
        };
        let mut out = String::from("flowchart TD\n");

        // Mermaid classDefs — color encoding tuned to be readable on
        // both light + dark Mermaid themes.
        out.push_str("  classDef pending      fill:#eeeeee,stroke:#999,color:#333\n");
        out.push_str("  classDef active       fill:#fff3b0,stroke:#a78b00,color:#000\n");
        out.push_str("  classDef completed    fill:#c8e6c9,stroke:#2e7d32,color:#000\n");
        out.push_str("  classDef failed       fill:#ffcdd2,stroke:#c62828,color:#000\n");
        out.push_str(
            "  classDef skipped      fill:#e0e0e0,stroke:#666,color:#666,stroke-dasharray: 3 3\n",
        );
        out.push_str("  classDef compensating fill:#ffe0b2,stroke:#e65100,color:#000\n");
        out.push_str("  classDef compensated  fill:#d1c4e9,stroke:#4527a0,color:#000\n");
        out.push_str(
            "  classDef resolving    fill:#fff3b0,stroke:#a78b00,color:#000,stroke-dasharray: 4 2\n",
        );
        out.push_str("  classDef compfailed   fill:#b71c1c,stroke:#000,color:#fff\n");
        out.push_str(
            "  classDef abandoned    fill:#ffe082,stroke:#b71c1c,color:#000,stroke-dasharray: 6 2\n",
        );

        let class_name = |s: SagaStepStatus| match s {
            SagaStepStatus::Pending => "pending",
            SagaStepStatus::Active => "active",
            SagaStepStatus::Resolving => "resolving",
            SagaStepStatus::Completed => "completed",
            SagaStepStatus::Failed => "failed",
            SagaStepStatus::Skipped => "skipped",
            SagaStepStatus::Compensating => "compensating",
            SagaStepStatus::Compensated => "compensated",
            SagaStepStatus::CompensationFailed => "compfailed",
            SagaStepStatus::Abandoned => "abandoned",
        };

        for step in self.topology.step_entries() {
            let status = envelope.step_status(step.name);
            let label = format!("{} ({:?})", step.name, status);
            let shape = if step.terminal.is_some() {
                format!(r#"  {}(["{}"])"#, step.name, label)
            } else {
                format!(r#"  {}["{}"]"#, step.name, label)
            };
            out.push_str(&shape);
            out.push('\n');
            out.push_str(&format!("  class {} {}\n", step.name, class_name(status)));
        }

        for step in self.topology.step_entries() {
            for dep in &step.after {
                out.push_str(&format!("  {} --> {}\n", dep, step.name));
            }
        }

        Ok(Some(out))
    }
}

/// Outcome of activating one step: completed (maybe terminal) or
/// suspended awaiting an external response.
enum StepOutcome {
    Completed { terminal: bool },
    Suspended(PendingAwait),
}

/// An await to persist to `es_saga_awaits` — produced when a step (or
/// compensation) emits to an external worker and suspends.
struct PendingAwait {
    step: String,
    event_type: String,
    is_compensation: bool,
    /// Absolute wall-clock deadline. `None` = waits forever. Computed
    /// at suspend time from the step's effective timeout (step
    /// `.timeout(d)` overrides saga `default_step_timeout`).
    deadline: Option<chrono::DateTime<chrono::Utc>>,
    /// The external (Regime-B) requests this step emitted as it suspended —
    /// written to `es_saga_outbox` in the same commit as the await (P3.2), so
    /// an `EmitPublisher` can deliver them at-least-once. Empty for await rows
    /// that carry no emit (signal-routed suspends, deletes/consumed markers).
    outbox_emits: Vec<OutboxEmit>,
}

/// One external request to durably deliver via the outbox (P3.2).
struct OutboxEmit {
    event_type: String,
    payload: serde_json::Value,
}

/// A row from the timeout scan. `created_at`/`deadline_at` let the timeout
/// report the configured budget (`deadline_at - created_at`, P5.4).
#[derive(sqlx::FromRow)]
struct ExpiredAwait {
    instance_id: String,
    step: String,
    deadline_at: chrono::DateTime<chrono::Utc>,
    created_at: chrono::DateTime<chrono::Utc>,
}

/// Map an instance id to its shard. Default single-shard topology
/// returns 0; multi-shard hashing arrives with the worker loop.
///
/// Uses FNV-1a rather than `DefaultHasher`: shard assignment must be stable
/// across processes, Rust versions, and restarts — `DefaultHasher`'s
/// algorithm and seed are explicitly unspecified, so two replicas (or one
/// after an upgrade) could disagree about which shard owns an instance
/// (review SAGA-15). FNV-1a is fixed by definition.
fn shard_of(instance_id: &str, num_shards: u32) -> u32 {
    if num_shards <= 1 {
        return 0;
    }
    const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
    const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;
    let mut h = FNV_OFFSET;
    for b in instance_id.as_bytes() {
        h ^= u64::from(*b);
        h = h.wrapping_mul(FNV_PRIME);
    }
    (h as u32) % num_shards
}

/// Rebuild the `outputs` map (keyed by `TypeId<Out>`) that
/// `Output<T>` extractors consume. The envelope stores outputs by
/// step name; the topology knows each step's output `TypeId`. Walk
/// both to reconstruct the lookup table.
fn build_outputs_by_type_id<S: Saga>(
    envelope: &SagaEnvelope<S>,
    entries: &[crate::builder::StepEntry<S>],
) -> std::collections::BTreeMap<std::any::TypeId, serde_json::Value> {
    let mut by_type = std::collections::BTreeMap::new();
    for entry in entries {
        let Some((type_id, _)) = entry.output else {
            continue;
        };
        if let Some(value) = envelope.outputs.get(entry.name) {
            by_type.insert(type_id, value.clone());
        }
    }
    by_type
}

/// Errors from the durable runner. Distinct from
/// [`crate::RunnerError`] because the surface is different — no
/// extractor errors that aren't expected at trigger time, but new
/// IO errors from the store + the explicit `Unsupported` variant
/// for the 4a-only constraint.
#[derive(Debug, thiserror::Error)]
pub enum PgRunnerError {
    #[error("no `start_on(...)` registered on this saga")]
    NoStartOn,
    #[error("failed to serialise: {0}")]
    Serialise(#[source] serde_json::Error),
    #[error("event store error: {0}")]
    Store(#[from] EventStoreError),
    #[error("extractor failed: {0}")]
    Extract(#[from] ExtractError),
    #[error("step `{step}` is not yet supported by PgSagaRunner: {reason}")]
    Unsupported {
        step: StepName,
        reason: &'static str,
    },
    #[error("signal type `{0}` not registered on this saga")]
    UnregisteredSignal(&'static str),
    #[error("unknown instance `{0}`")]
    UnknownInstance(String),
    #[error("signal rejected: saga `{instance_id}` is already terminal")]
    SagaTerminal { instance_id: String },
    #[error("signal rejected: saga `{instance_id}` is currently compensating")]
    Compensating { instance_id: String },
    #[error(
        "signal `{signal_type}` declared `.routes_to_step(\"{target}\")` but step \
         `{actual_awaiter}` is the one currently awaiting it — the topology routes by \
         event_type, so the wrong step name is a config error"
    )]
    SignalRouteMismatch {
        signal_type: &'static str,
        target: StepName,
        actual_awaiter: String,
    },
    /// Two (or more) steps of one instance are simultaneously awaiting the
    /// same event type. Routing is by event-type string, so the runner
    /// cannot know which step a response belongs to — delivering to an
    /// arbitrary one would silently feed one step's response into another
    /// (review C7). Restructure the saga so concurrently-active awaits use
    /// distinct response event types.
    #[error(
        "ambiguous await on instance `{instance_id}`: steps {steps:?} all await \
         event type `{event_type}` — routing by event type cannot tell them apart; \
         give concurrently-active awaits distinct response event types"
    )]
    AmbiguousAwait {
        instance_id: String,
        event_type: String,
        steps: Vec<String>,
    },
    /// A second `RoutesToStep` signal of the same type arrived for an
    /// instance while a first one was still buffered (its target step has
    /// not yet suspended to consume it). Only one unconsumed signal per
    /// `(instance, type)` is allowed (review SAGA-10, D-SIG-1).
    #[error(
        "duplicate buffered signal `{signal_type}` for instance `{instance_id}`: \
         an earlier signal of this type is still awaiting its target step's suspension"
    )]
    DuplicateBufferedSignal {
        instance_id: String,
        signal_type: &'static str,
    },
    #[error("internal invariant violated: {0}")]
    Internal(&'static str),
    /// Variant of `Internal` for messages that include runtime detail
    /// (formatted at call time) that a `&'static str` can't carry.
    #[error("internal invariant violated: {0}")]
    InternalOwned(String),
    /// The persisted `Started` state of instance `{instance_id}` no longer
    /// deserializes into the current `S::State` — schema drift. The runner
    /// refuses to act rather than run the saga against a default state
    /// (review C6/DX-3). Resolve by an upcaster/migration or by pinning the
    /// state schema; see the event-evolution docs.
    #[error("saga `{instance_id}` has undecodable persisted state (schema drift): {reason}")]
    StateDecodeFailed { instance_id: String, reason: String },
    /// Another live worker holds this instance's processing lease — effects
    /// must not run concurrently (effects-vs-commit §11). Retryable: back off
    /// and retry; the holder is processing, or its lease lapses and a later
    /// call (or this one, retried) takes over (at-least-once).
    #[error("saga `{instance_id}` is being processed by another worker (lease contended)")]
    Contended { instance_id: String },
    /// A persisted row in `{table}` carries `shard_id = {max_shard_id}`, which
    /// is `>= num_shards ({num_shards})` — this runner/worker is configured
    /// with fewer shards than the rows were stamped under, so the fleet
    /// disagrees on `num_shards`. Acting on it would strand the out-of-range
    /// shards' work (their timeouts/signals would never be scanned). Reconcile
    /// the fleet's `with_num_shards` config (it is a deployment-wide constant)
    /// or run the documented resharding migration. See
    /// [`verify_shard_config`](PgSagaRunner::verify_shard_config).
    #[error(
        "shard config mismatch: row in `{table}` has shard_id {max_shard_id} \
         >= num_shards {num_shards} — the fleet disagrees on num_shards"
    )]
    ShardConfigMismatch {
        table: &'static str,
        max_shard_id: i32,
        num_shards: u32,
    },
}

#[cfg(test)]
mod shard_tests {
    use super::shard_of;

    /// Pinned FNV-1a values: shard assignment is a cross-process contract
    /// (review SAGA-15). If this test breaks, the hash changed and any
    /// persisted/assumed shard mapping breaks with it — don't "fix" the
    /// expectations without a migration story.
    #[test]
    fn shard_assignment_is_pinned() {
        assert_eq!(shard_of("anything", 1), 0);
        // Values cross-checked against an independent FNV-1a implementation.
        assert_eq!(shard_of("booking-1234", 16), 7);
        assert_eq!(shard_of("booking-1235", 16), 4);
        assert_eq!(shard_of("", 16), 5);
        assert_eq!(shard_of("ساغا-unicode-✓", 8), 6);
    }

    #[test]
    fn shards_stay_in_range() {
        for n in [2u32, 3, 7, 16, 64] {
            for i in 0..200 {
                assert!(shard_of(&format!("instance-{i}"), n) < n);
            }
        }
    }
}
