//! [`SagaBuilder`] — declares the saga's DAG topology.
//!
//! Steps are registered via [`task`] builders that bundle the handler with
//! its options: `.after(...)`, `.on_response(...)` (the await + map),
//! `.timeout(...)`, `.gated_by(...)`, `.terminal_success()` /
//! `.terminal_failure()`. Compensations are registered via [`comp`]
//! builders with `.on_response::<E>()` / `.even_if_active()`.

use std::any::TypeId;
use std::collections::{BTreeMap, BTreeSet};
use std::marker::PhantomData;
use std::sync::Arc;
use std::time::Duration;

use mire::Aggregate;
use serde::Serialize;
use serde::de::DeserializeOwned;

use crate::event_static::EventTypeStatic;
use crate::handler::{
    CompensationHandler, ErasedCompensationHandler, ErasedRequestHandler, ErasedRequestResponse,
    ErasedStepHandler, IntoResponse, RequestHandler, StepHandler, StepKind,
};
use crate::result::AwaitMapError;
use crate::saga::{Saga, Start};
use crate::step::StepName;

/// Mutable topology being assembled. Methods are non-consuming where chainable
/// for ergonomics; `build()` consumes and returns the final compiled topology.
pub struct SagaBuilder<S: Saga> {
    pub(crate) steps: Vec<StepEntry<S>>,
    pub(crate) compensations: BTreeMap<StepName, CompensationEntry<S>>,
    pub(crate) start_correlator: Option<TriggerHook<S>>,
    pub(crate) deps: BTreeMap<TypeId, Arc<dyn std::any::Any + Send + Sync>>,
    pub(crate) signals: Vec<SagaSignal>,
    /// In-process request handlers keyed by the REQUEST event_type string
    /// (the type a step emits). The runner looks one up when a step emits an
    /// event; if present, it invokes the handler and auto-delivers the
    /// returned response back to the saga. If absent, the emit sits in the
    /// log waiting for an external worker to drive it via `deliver_raw`.
    pub(crate) request_handlers: BTreeMap<&'static str, ErasedRequestHandler<S>>,
    /// When true, the runner fails any step that emits an event whose
    /// type has no registered handler and isn't in `allowed_unhandled`.
    /// Surfaces "forgot to register a handler" as a compensation-triggering
    /// failure rather than a silent stall.
    pub(crate) strict_handlers: bool,
    /// Event types explicitly marked as handled by an external worker —
    /// no in-process handler is expected. Only consulted when
    /// `strict_handlers` is true.
    pub(crate) allowed_unhandled: BTreeSet<&'static str>,
    /// Default timeout applied to any awaiting step with no task-level
    /// `.timeout(...)` or per-activation `Emit::timeout`. `None` (the
    /// default) means such steps wait forever.
    pub(crate) default_step_timeout: Option<std::time::Duration>,
}

/// The await + response→output mapping a task declared via
/// [`TaskBuilder::on_response`]. Registered once at build time, so the
/// runner can resume a suspended step without re-running its handler —
/// the phase-2 replacement for the in-handler `Box<dyn Fn>` await map.
pub(crate) struct AwaitSpec {
    pub event_type: &'static str,
    /// JSON-in / JSON-out: deserialise the response event, run the user's
    /// map, serialise the typed output.
    pub map:
        Arc<dyn Fn(serde_json::Value) -> Result<serde_json::Value, AwaitMapError> + Send + Sync>,
    pub output_type_id: TypeId,
    pub output_type_name: &'static str,
}

/// Default `Out` marker for a [`TaskBuilder`] with no `.on_response(...)`:
/// fire-and-forget emit steps complete with `Null` output and register no
/// extractable output type.
pub struct NoResponse;

#[allow(dead_code)] // consumed by the runner
pub struct StepEntry<S: Saga> {
    pub name: StepName,
    pub after: Vec<StepName>,
    pub(crate) terminal: Option<Terminal>,
    pub handler: ErasedStepHandler<S>,
    /// The handler's shape (emit vs value), resolved from its return type.
    pub(crate) kind: StepKind,
    /// The declared await for emit steps. `None` = fire-and-forget (or a
    /// value step, which never awaits).
    pub(crate) await_spec: Option<AwaitSpec>,
    /// Task-level timeout for the await. Overridden per activation by
    /// `Emit::timeout`, overrides the saga's `default_step_timeout`.
    pub(crate) step_timeout: Option<Duration>,
    /// The step's typed output identity: a value handler's return type, or
    /// the `.on_response` map's output type. `None` for fire-and-forget
    /// steps (their `Null` output is not extractable via `Output<T>`).
    pub(crate) output: Option<(TypeId, &'static str)>,
    /// Set when a value (compute) handler was registered with an
    /// `.on_response(...)` declaration — contradictory; surfaced by the
    /// validator as [`TopologyError::ResponseOnComputeStep`].
    pub(crate) response_on_value_step: bool,
    /// Optional gating predicate evaluated before activation. When present
    /// and returning false, the step is `Skipped` rather than activated.
    pub(crate) gate: Option<Gate>,
    /// Optional state-based gating predicate. Both gates (if set) must
    /// return true for the step to activate; otherwise the step is
    /// `Skipped`. Independent of `gate` — combine for "after this
    /// upstream produced X AND saga state is Y".
    pub(crate) state_gate: Option<StateGate>,
    /// How `after` deps combine: `All` (every predecessor must complete)
    /// or `Any` (one is enough). Default `All`.
    pub(crate) merge_kind: MergeKind,
    /// What this step's handler extractors require from the topology.
    /// Collected at registration time via `StepHandler::requirements()`
    /// and consumed by the validator.
    pub(crate) requirements: Vec<crate::extract::ExtractRequirement>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MergeKind {
    All,
    Any,
}

/// Type-erased JSON predicate: decode the input value into the user's
/// type, run their closure, return the verdict. `Err` means the JSON
/// didn't decode (the runner treats that as "skip + log").
pub(crate) type JsonPredicate =
    Box<dyn Fn(&serde_json::Value) -> Result<bool, serde_json::Error> + Send + Sync>;

/// A gating predicate on a step. Reads the named upstream step's typed
/// output (as JSON), runs the user's closure, and either activates or
/// skips the step. Stored erased so the topology can carry many gates
/// without per-type plumbing.
#[allow(dead_code)] // consumed by the runner
pub(crate) struct Gate {
    pub upstream: StepName,
    /// JSON-input predicate. Returns Ok(bool) on success; Err on
    /// deserialise failure (treated by the runner as "skip + log").
    pub predicate: JsonPredicate,
}

/// State-based gate. Evaluated at step activation against the saga's
/// current state (serialised to JSON for type-erased storage). The
/// predicate runs over the user's expected type and the runner skips
/// the step if it returns false.
#[allow(dead_code)] // consumed by the runner
pub(crate) struct StateGate {
    pub predicate: JsonPredicate,
}

#[derive(Debug, Clone, Copy)]
pub(crate) enum Terminal {
    Success,
    Failure,
}

/// A registered compensation: erased handler + the topology requirements
/// its extractors impose. Stored together so the validator can check
/// requirements at build time without re-introspecting the handler.
pub struct CompensationEntry<S: Saga> {
    pub(crate) handler: ErasedCompensationHandler<S>,
    pub(crate) requirements: Vec<crate::extract::ExtractRequirement>,
    /// The response event type that completes this compensation, declared
    /// via [`CompBuilder::on_response`]. `None` = the compensation
    /// completes as soon as its emits are dispatched.
    pub(crate) await_event_type: Option<&'static str>,
    /// When true, the cascade also runs this compensation for a step that
    /// was still **Active/Resolving** (in-flight external work). Only valid
    /// for handlers that don't read the step's own `Output<T>` — a
    /// never-completed step has no output (review SAGA-13). Default false:
    /// in-flight steps are marked `Abandoned` instead.
    pub(crate) even_if_active: bool,
}

pub struct TriggerHook<S: Saga> {
    pub aggregate_category: &'static str,
    /// `TypeId` of `A::Event` from `.start_on::<A, _>(...)`. Used by the
    /// validator to check `Trigger<T>` requirements at build time —
    /// catches the typo where a step reads `Trigger<WrongEvent>` while
    /// `start_on` is bound to a different event type.
    pub trigger_type_id: TypeId,
    /// Display name of the trigger event type. Quoted in validator
    /// diagnostics so the user can see what they registered.
    pub trigger_type_name: &'static str,
    /// Stored as a boxed type-erased closure; the runner reads aggregate
    /// events, deserialises into the aggregate's `Event` type, and calls this
    /// to find the start payload.
    pub correlate: CorrelateFn<S>,
}

/// Type-erased correlator: decode a domain event and decide whether (and
/// how) it starts a fresh saga instance.
pub(crate) type CorrelateFn<S> = Box<dyn Fn(&serde_json::Value) -> Option<Start<S>> + Send + Sync>;

/// Signal registration produced by `builder.signal::<T>()`.
pub struct SagaSignal {
    pub signal_type_name: &'static str,
    pub policy: SignalPolicy,
}

#[derive(Debug, Clone, Copy)]
pub enum SignalPolicy {
    /// Cancel the in-flight step (if any) and trigger compensation.
    TriggersCompensation,
    /// Route to a specific step as that step's await-event payload.
    /// Carries the signal's `EVENT_TYPE` so the validator can check at
    /// build time that the target step actually awaits it.
    RoutesToStep {
        step: StepName,
        event_type: &'static str,
    },
}

impl<S: Saga> SagaBuilder<S> {
    pub(crate) fn new() -> Self {
        Self {
            steps: Vec::new(),
            compensations: BTreeMap::new(),
            start_correlator: None,
            deps: BTreeMap::new(),
            signals: Vec::new(),
            request_handlers: BTreeMap::new(),
            strict_handlers: false,
            allowed_unhandled: BTreeSet::new(),
            default_step_timeout: None,
        }
    }

    /// Configure how a domain event from aggregate `A` triggers a new saga
    /// instance. Called exactly once per saga — calling it twice panics at
    /// build time so silent overwrites can't bury the original correlator.
    ///
    /// # Panics
    ///
    /// Panics if called more than once on the same builder. Build-time
    /// only — no production-path panic risk.
    pub fn start_on<A, F>(mut self, correlate: F) -> Self
    where
        A: Aggregate,
        A::Event: DeserializeOwned + 'static,
        F: Fn(&A::Event) -> Option<Start<S>> + Send + Sync + 'static,
    {
        assert!(
            self.start_correlator.is_none(),
            "saga `{}`: `.start_on(...)` called more than once. Each saga \
             may only declare one trigger aggregate; combine your \
             correlation logic into a single closure if you need to \
             accept multiple event shapes.",
            S::name()
        );
        let category = A::stream_category();
        let correlate_erased: CorrelateFn<S> = Box::new(move |value| {
            let event = serde_json::from_value::<A::Event>(value.clone()).ok()?;
            correlate(&event)
        });
        self.start_correlator = Some(TriggerHook {
            aggregate_category: category,
            trigger_type_id: TypeId::of::<A::Event>(),
            trigger_type_name: std::any::type_name::<A::Event>(),
            correlate: correlate_erased,
        });
        self
    }

    /// Register a step: a named DAG node that runs an async handler with
    /// its options (deps, await declaration, gate, terminal flag) bundled
    /// in via `task()`.
    ///
    /// ```rust,ignore
    /// SagaBuilder::for_saga()
    ///     .step("flight", task(reserve_flight)
    ///         .on_response(|h: FlightHeld| FlightReserved { hold_id: h.hold_id }))
    ///     .step("payment", task(request_payment)
    ///         .after(["flight", "hotel"])
    ///         .on_response(|c: PaymentCaptured| PaymentSettled { payment_id: c.payment_id }))
    ///     .step("join", task(settle).after("payment").terminal_success())
    /// ```
    ///
    /// `task(handler)` keeps the handler and its options visually grouped
    /// inside one `.step(...)` call. The handler's *kind* is inferred from
    /// its return type: [`crate::Emit`] for emit steps (await + output
    /// declared via `.on_response`), or any serialisable value for
    /// immediate-complete compute steps.
    pub fn step<H, Args, Marker, TOut, C, CM, EF, EM>(
        mut self,
        name: StepName,
        task: TaskBuilder<H, TOut, C, EF>,
    ) -> Self
    where
        H: StepHandler<S, Args, Marker> + 'static,
        Args: 'static,
        Marker: 'static,
        TOut: 'static,
        C: RegisterComp<S, CM>,
        EF: RegisterEffect<S, EM>,
    {
        let TaskBuilder {
            handler,
            after,
            merge_kind,
            terminal,
            gate,
            state_gate,
            await_spec,
            timeout,
            comp,
            effect,
            _out,
        } = task;
        let kind = <H as StepHandler<S, Args, Marker>>::kind();
        let requirements = <H as StepHandler<S, Args, Marker>>::requirements();

        // Resolve the step's typed-output identity from its shape; an
        // `.on_response` on a value (compute) handler is contradictory and
        // surfaces at validate() time rather than silently winning.
        let (output, response_on_value_step) = match kind {
            StepKind::Value => (
                <H as StepHandler<S, Args, Marker>>::value_output(),
                await_spec.is_some(),
            ),
            StepKind::Emit => (
                await_spec
                    .as_ref()
                    .map(|spec| (spec.output_type_id, spec.output_type_name)),
                false,
            ),
        };

        // Erase the handler. Extraction errors propagate as Result; the
        // outcome (dispatch list or immediate value) survives erasure.
        let handler = Arc::new(handler);
        let erased: ErasedStepHandler<S> = Arc::new(move |ctx| {
            let h = Arc::clone(&handler);
            Box::pin(async move { h.call(ctx).await })
        });

        let entry = StepEntry {
            name,
            after,
            terminal,
            handler: erased,
            kind,
            await_spec,
            step_timeout: timeout,
            output,
            response_on_value_step,
            gate,
            state_gate,
            merge_kind,
            requirements,
        };
        self.steps.push(entry);
        // Register the task-attached compensation (if any) under this step —
        // including its own release effect — plus the step's effect, both into
        // the shared request-handler map keyed by request event_type.
        comp.register(name, &mut self.compensations, &mut self.request_handlers);
        effect.register(&mut self.request_handlers);
        self
    }

    /// Register a signal type. Returns a [`SignalRegistration`] for policy
    /// configuration (`.triggers_compensation()`, `.routes_to_step(name)`).
    pub fn signal<T: 'static>(self) -> SignalRegistration<S, T> {
        SignalRegistration {
            builder: self,
            _phantom: PhantomData,
        }
    }

    /// Register a shared dependency value. Available to every handler via
    /// `Dep<T>`.
    pub fn dep<T: Send + Sync + 'static>(mut self, value: T) -> Self {
        self.deps.insert(TypeId::of::<T>(), Arc::new(value));
        self
    }

    /// Register a typed in-process **effect** at the builder level — the
    /// handler that fulfils an emitted request, shared across every step that
    /// emits that request type.
    ///
    /// Prefer the co-located [`TaskBuilder::effect`] for the common 1:1 case
    /// (effect belongs to one step); use this builder-level form only when one
    /// handler fulfils a request type emitted by several steps, or to register
    /// a handler with no single owning step.
    ///
    /// Effects have the shape `async fn(Req<R>, …other extractors) -> Resp`.
    /// Because `Req<R>` is the first parameter, the request type `R` and
    /// the response type `Resp` are inferred from the handler's signature
    /// — call this with **no turbofish**:
    ///
    /// ```rust,ignore
    /// async fn handle_flight_hold(Req(r): Req<RequestFlightHold>) -> FlightHeld {
    ///     FlightHeld { booking_id: r.booking_id, hold_id: FlightHoldId::new() }
    /// }
    ///
    /// builder.effect(handle_flight_hold)   // R + Args + Resp all inferred
    /// ```
    ///
    /// When the saga emits an event of type `R`, the runner invokes
    /// `handler` and auto-delivers its typed `Resp` back to the saga.
    /// Sagas can mix registered effects and external responders freely.
    pub fn effect<R, H, Args, Resp>(mut self, handler: H) -> Self
    where
        R: EventTypeStatic + DeserializeOwned + Send + 'static,
        Resp: IntoResponse + 'static,
        H: RequestHandler<S, R, Args, Resp> + 'static,
        Args: 'static,
    {
        let (request_event_type, erased) = erase_request_handler::<S, R, _, Args, Resp>(handler);
        self.request_handlers.insert(request_event_type, erased);
        self
    }

    /// Remove a previously-registered request handler by its request
    /// event_type string. Useful in tests for simulating "external worker
    /// not responding" scenarios.
    pub fn remove_request_handler(&mut self, request_event_type: &str) {
        self.request_handlers.remove(request_event_type);
    }

    /// Opt into strict handler validation: any emit whose type has no
    /// registered `.effect(...)` AND isn't explicitly marked
    /// `.allow_unhandled::<T>()` fails the step immediately with
    /// [`StepError::Failed`], which fires the compensation cascade.
    ///
    /// Surfaces the "I forgot to register a handler" footgun as a loud
    /// failure rather than a silent stall. Off by default; existing
    /// behaviour (emits sit waiting for external delivery) is preserved
    /// when the flag is not set.
    pub fn strict_handlers(mut self) -> Self {
        self.strict_handlers = true;
        self
    }

    /// Declare that emits of type `T` are intentionally handled by an
    /// external worker — no in-process handler is required. Only
    /// consulted when [`strict_handlers`] is active; harmless otherwise.
    pub fn allow_unhandled<T: EventTypeStatic>(mut self) -> Self {
        self.allowed_unhandled
            .insert(<T as EventTypeStatic>::EVENT_TYPE);
        self
    }

    /// Apply `d` as the default timeout for any awaiting step that doesn't
    /// declare one via the task-level `.timeout(...)` or a per-activation
    /// `Emit::timeout`. Without this safety net, steps that forget a
    /// timeout wait forever for their response event.
    pub fn default_step_timeout(mut self, d: std::time::Duration) -> Self {
        self.default_step_timeout = Some(d);
        self
    }
}

/// A configurable step description: a handler bundled with its DAG
/// options (deps, merge kind, terminal flag, gating predicate) and —
/// for emit steps — the await declaration (`.on_response`, `.timeout`).
/// Returned by [`task`] and consumed by [`SagaBuilder::step`].
///
/// The saga's type parameter `S` is *not* on `TaskBuilder` — it gets
/// bound when the task is passed to `.step()`. That lets `task()` be
/// called without specifying the saga. `Out` tracks the step's typed
/// output as declared by `.on_response`; it defaults to [`NoResponse`]
/// (fire-and-forget, or a value handler whose output comes from its
/// return type).
pub struct TaskBuilder<H, Out = NoResponse, C = NoComp, EF = NoEffect> {
    pub(crate) handler: H,
    pub(crate) after: Vec<StepName>,
    pub(crate) merge_kind: MergeKind,
    pub(crate) terminal: Option<Terminal>,
    pub(crate) gate: Option<Gate>,
    pub(crate) state_gate: Option<StateGate>,
    pub(crate) await_spec: Option<AwaitSpec>,
    pub(crate) timeout: Option<Duration>,
    /// The step's compensation, attached via [`TaskBuilder::compensate`]. The
    /// `C` parameter is *inferred* — `NoComp` when absent, `CompBuilder<_>`
    /// when present — so callers never write it. `.step()` registers it under
    /// the step's name via [`RegisterComp`].
    pub(crate) comp: C,
    /// The in-process **effect** that fulfils the request this step emits,
    /// attached via [`TaskBuilder::effect`]. Like `comp`, the `EF` parameter is
    /// *inferred* — `NoEffect` when absent, `EffectBuilder<_>` when present.
    /// `.step()` registers it into the saga's request-handler map (keyed by the
    /// emitted request's `event_type`) via [`RegisterEffect`] — identical
    /// lowering to the builder-level [`SagaBuilder::effect`], just co-located.
    pub(crate) effect: EF,
    pub(crate) _out: PhantomData<Out>,
}

/// The "no compensation" default for [`TaskBuilder`]'s `C` parameter — a step
/// that doesn't roll anything back (terminal steps, pure-compute steps).
pub struct NoComp;

/// The "no effect" default for [`TaskBuilder`]'s / [`CompBuilder`]'s `EF`
/// parameter — a step whose emitted request is fulfilled elsewhere (a
/// builder-level [`SagaBuilder::effect`], or an external responder).
pub struct NoEffect;

/// Carries a typed effect handler attached via [`TaskBuilder::effect`] /
/// [`CompBuilder::effect`] until `.step()` (which has the saga type `S`) can
/// erase and register it. Mirrors how [`CompBuilder`] defers compensation
/// erasure through [`RegisterComp`].
pub struct EffectBuilder<EH> {
    pub(crate) handler: EH,
}

/// Erase a typed request handler into the runner's stored form and return it
/// alongside the request `event_type` it fulfils. Shared by the builder-level
/// [`SagaBuilder::effect`] and the co-located [`EffectBuilder`] registration so
/// both lower identically.
fn erase_request_handler<S, R, H, Args, Resp>(handler: H) -> (&'static str, ErasedRequestHandler<S>)
where
    S: Saga,
    R: EventTypeStatic + DeserializeOwned + Send + 'static,
    Resp: IntoResponse + 'static,
    H: RequestHandler<S, R, Args, Resp> + 'static,
    Args: 'static,
{
    let handler = Arc::new(handler);
    let erased: ErasedRequestHandler<S> = Arc::new(move |ctx| {
        let h = Arc::clone(&handler);
        Box::pin(async move {
            let resp = h.call(ctx).await?;
            let value = resp.into_response();
            Ok(ErasedRequestResponse {
                event_type: value.event_type,
                payload: value.payload,
                is_error: value.is_error,
                error_reason: value.error_reason,
            })
        })
    });
    (<R as EventTypeStatic>::EVENT_TYPE, erased)
}

/// Lets `.step()` register whatever effect a task/compensation carries —
/// `NoEffect` (nothing) or an `EffectBuilder<EH>` (erased + inserted into the
/// request-handler map) — through one code path. `Marker` carries the effect
/// handler's `(R, Args, Resp)` so the impl's type parameters stay constrained
/// and the caller writes no turbofish.
pub trait RegisterEffect<S: Saga, Marker> {
    fn register(self, into: &mut BTreeMap<&'static str, ErasedRequestHandler<S>>);
}

impl<S: Saga> RegisterEffect<S, ()> for NoEffect {
    fn register(self, _: &mut BTreeMap<&'static str, ErasedRequestHandler<S>>) {}
}

impl<S, EH, R, Args, Resp> RegisterEffect<S, (R, Args, Resp)> for EffectBuilder<EH>
where
    S: Saga,
    R: EventTypeStatic + DeserializeOwned + Send + 'static,
    Resp: IntoResponse + 'static,
    EH: RequestHandler<S, R, Args, Resp> + 'static,
    Args: 'static,
{
    fn register(self, into: &mut BTreeMap<&'static str, ErasedRequestHandler<S>>) {
        let (event_type, erased) = erase_request_handler::<S, R, _, Args, Resp>(self.handler);
        into.insert(event_type, erased);
    }
}

/// Lets `.step()` register whatever compensation a task carries — `NoComp`
/// (nothing) or a `CompBuilder<H>` (erased + inserted) — through one code
/// path, so a single `.step` signature handles both. `Marker` carries the
/// compensation handler's extractor args so the impl's type parameters stay
/// constrained (no turbofish for the caller).
pub trait RegisterComp<S: Saga, Marker> {
    fn register(
        self,
        name: StepName,
        comps: &mut BTreeMap<StepName, CompensationEntry<S>>,
        effects: &mut BTreeMap<&'static str, ErasedRequestHandler<S>>,
    );
}

impl<S: Saga> RegisterComp<S, ()> for NoComp {
    fn register(
        self,
        _: StepName,
        _: &mut BTreeMap<StepName, CompensationEntry<S>>,
        _: &mut BTreeMap<&'static str, ErasedRequestHandler<S>>,
    ) {
    }
}

impl<S, H, Args, CEF, EM> RegisterComp<S, (Args, EM)> for CompBuilder<H, CEF>
where
    S: Saga,
    H: CompensationHandler<S, Args> + 'static,
    Args: 'static,
    CEF: RegisterEffect<S, EM>,
{
    fn register(
        self,
        name: StepName,
        comps: &mut BTreeMap<StepName, CompensationEntry<S>>,
        effects: &mut BTreeMap<&'static str, ErasedRequestHandler<S>>,
    ) {
        let CompBuilder {
            handler,
            await_event_type,
            even_if_active,
            effect,
        } = self;
        let requirements = <H as CompensationHandler<S, Args>>::requirements();
        let handler = Arc::new(handler);
        let erased: ErasedCompensationHandler<S> = Arc::new(move |ctx| {
            let h = Arc::clone(&handler);
            Box::pin(async move { h.call(ctx).await })
        });
        comps.insert(
            name,
            CompensationEntry {
                handler: erased,
                requirements,
                await_event_type,
                even_if_active,
            },
        );
        // The compensation's own release effect (if attached via
        // `comp(...).effect(...)`) registers into the shared request-handler map.
        effect.register(effects);
    }
}

/// Wrap a handler in a [`TaskBuilder`] so its options can be chained
/// fluently inside the `.step(name, ...)` call. Keeps options visually
/// nested under the step's argument list rather than spilling onto
/// top-level chain lines.
pub fn task<H>(handler: H) -> TaskBuilder<H, NoResponse, NoComp, NoEffect> {
    TaskBuilder {
        handler,
        after: Vec::new(),
        merge_kind: MergeKind::All,
        terminal: None,
        gate: None,
        state_gate: None,
        await_spec: None,
        timeout: None,
        comp: NoComp,
        effect: NoEffect,
        _out: PhantomData,
    }
}

impl<H, C, EF> TaskBuilder<H, NoResponse, C, EF> {
    /// Declare which response event completes this (emit) step and how
    /// its payload maps to the step's typed output. Registered on the
    /// topology at build time — the durable runner resumes a suspended
    /// step by looking this up, never by re-running the handler.
    ///
    /// ```rust,ignore
    /// task(reserve_flight)
    ///     .on_response(|held: FlightHeld| FlightReserved { hold_id: held.hold_id })
    /// ```
    pub fn on_response<E, O, F>(self, map: F) -> TaskBuilder<H, O, C, EF>
    where
        E: EventTypeStatic + DeserializeOwned + 'static,
        O: Serialize + DeserializeOwned + Send + 'static,
        F: Fn(E) -> O + Send + Sync + 'static,
    {
        let erased: Arc<
            dyn Fn(serde_json::Value) -> Result<serde_json::Value, AwaitMapError> + Send + Sync,
        > = Arc::new(move |value| {
            let e: E = serde_json::from_value(value).map_err(AwaitMapError::DeserialiseEvent)?;
            let o = map(e);
            serde_json::to_value(&o).map_err(AwaitMapError::SerialiseOutput)
        });
        TaskBuilder {
            handler: self.handler,
            after: self.after,
            merge_kind: self.merge_kind,
            terminal: self.terminal,
            gate: self.gate,
            state_gate: self.state_gate,
            await_spec: Some(AwaitSpec {
                event_type: <E as EventTypeStatic>::EVENT_TYPE,
                map: erased,
                output_type_id: TypeId::of::<O>(),
                output_type_name: std::any::type_name::<O>(),
            }),
            timeout: self.timeout,
            comp: self.comp,
            effect: self.effect,
            _out: PhantomData,
        }
    }

    /// Shorthand for [`on_response`](Self::on_response) when the step's
    /// output IS the awaited response event — identity map, no closure,
    /// no per-step output newtype.
    pub fn on_response_event<E>(self) -> TaskBuilder<H, E, C, EF>
    where
        E: EventTypeStatic + Serialize + DeserializeOwned + Send + 'static,
    {
        self.on_response(|e: E| e)
    }
}

impl<H, Out, EF> TaskBuilder<H, Out, NoComp, EF> {
    /// Attach this step's compensation, co-located with the step instead of a
    /// separate `.compensate("name", …)` line. The step's *do* and *undo*
    /// read as one unit; the step name is written once. Optional — steps with
    /// nothing to roll back simply omit it.
    ///
    /// ```rust,ignore
    /// .step("flight", task(reserve_flight)
    ///     .on_response(|h: FlightHeld| FlightReserved { hold_id: h.hold_id })
    ///     .compensate(comp(release_flight).on_response::<FlightReleased>()))
    /// ```
    pub fn compensate<CH, CEF>(
        self,
        comp: CompBuilder<CH, CEF>,
    ) -> TaskBuilder<H, Out, CompBuilder<CH, CEF>, EF> {
        TaskBuilder {
            handler: self.handler,
            after: self.after,
            merge_kind: self.merge_kind,
            terminal: self.terminal,
            gate: self.gate,
            state_gate: self.state_gate,
            await_spec: self.await_spec,
            timeout: self.timeout,
            comp,
            effect: self.effect,
            _out: PhantomData,
        }
    }
}

/// `.effect(...)` — attach the in-process effect that fulfils the request this
/// step emits, co-located on the step's chain. Only available before an effect
/// has been set (the `NoEffect` default).
impl<H, Out, C> TaskBuilder<H, Out, C, NoEffect> {
    /// Register the effect handler that fulfils this step's emitted request,
    /// keyed by the request's `event_type`. Lowers to the *identical*
    /// registration as the builder-level [`SagaBuilder::effect`] — the handler
    /// shape is `async fn(Req<R>, …) -> Resp`, with `R`/`Resp` inferred from the
    /// signature (no turbofish):
    ///
    /// ```rust,ignore
    /// .step("flight", task(reserve_flight)        // sync decide: -> Emit(RequestFlightHold)
    ///     .effect(handle_flight_hold)             // async effect: Req<RequestFlightHold> -> FlightHeld
    ///     .on_response(|h: FlightHeld| FlightReserved { hold_id: h.hold_id }))
    /// ```
    ///
    /// Use this for the common 1:1 case where an effect belongs to one step.
    /// For a handler shared by several steps' emits, register it once at the
    /// builder level with [`SagaBuilder::effect`] instead.
    pub fn effect<EH>(self, handler: EH) -> TaskBuilder<H, Out, C, EffectBuilder<EH>> {
        TaskBuilder {
            handler: self.handler,
            after: self.after,
            merge_kind: self.merge_kind,
            terminal: self.terminal,
            gate: self.gate,
            state_gate: self.state_gate,
            await_spec: self.await_spec,
            timeout: self.timeout,
            comp: self.comp,
            effect: EffectBuilder { handler },
            _out: PhantomData,
        }
    }
}

impl<H, Out, C, EF> TaskBuilder<H, Out, C, EF> {
    /// Bound this step's await with a task-level timeout. Overridden per
    /// activation by [`crate::Emit::timeout`]; overrides the saga's
    /// `default_step_timeout`.
    pub fn timeout(mut self, d: Duration) -> Self {
        self.timeout = Some(d);
        self
    }
    /// All listed predecessors must `Completed` before this step runs
    /// (`MergeKind::All`, the default).
    pub fn after<D: IntoStepNames>(mut self, deps: D) -> Self {
        self.after = deps.into_step_names();
        self.merge_kind = MergeKind::All;
        self
    }

    /// Activate when *any* listed predecessor completes
    /// (`MergeKind::Any`). Skip-cascades when every predecessor ends up
    /// `Skipped`.
    pub fn after_any_of<D: IntoStepNames>(mut self, deps: D) -> Self {
        self.after = deps.into_step_names();
        self.merge_kind = MergeKind::Any;
        self
    }

    /// Mark this step as the saga's terminal-success endpoint.
    pub fn terminal_success(mut self) -> Self {
        self.terminal = Some(Terminal::Success);
        self
    }

    /// Mark this step as the saga's terminal-failure endpoint.
    pub fn terminal_failure(mut self) -> Self {
        self.terminal = Some(Terminal::Failure);
        self
    }

    /// Gate the step on a predicate over an upstream's typed output.
    /// Skipped if the predicate returns false or the upstream is itself
    /// skipped. `T` is inferred from the closure's parameter.
    ///
    /// ```rust,ignore
    /// task(auto_handler)
    ///     .after("amount_check")
    ///     .gated_by("amount_check", |d: &AmountDecision| {
    ///         matches!(d, AmountDecision::AutoOk)
    ///     })
    /// ```
    pub fn gated_by<T, F>(mut self, upstream: StepName, predicate: F) -> Self
    where
        T: DeserializeOwned + 'static,
        F: Fn(&T) -> bool + Send + Sync + 'static,
    {
        let erased: JsonPredicate = Box::new(move |value| {
            let typed: T = serde_json::from_value(value.clone())?;
            Ok(predicate(&typed))
        });
        self.gate = Some(Gate {
            upstream,
            predicate: erased,
        });
        self
    }

    /// Gate the step on a predicate over the saga's current state.
    /// `T` is inferred from the closure parameter and must match the
    /// saga's `State` type (or be deserialisable from it).
    ///
    /// ```rust,ignore
    /// task(manual_review)
    ///     .after("amount_check")
    ///     .gated_by_state(|s: &Booking| s.amount > 10_000)
    /// ```
    ///
    /// Both `gated_by` and `gated_by_state` can be set; the step
    /// activates only if BOTH predicates return true. Either failing
    /// causes a `Skipped` transition (no compensation).
    pub fn gated_by_state<T, F>(mut self, predicate: F) -> Self
    where
        T: DeserializeOwned + 'static,
        F: Fn(&T) -> bool + Send + Sync + 'static,
    {
        let erased: JsonPredicate = Box::new(move |value| {
            let typed: T = serde_json::from_value(value.clone())?;
            Ok(predicate(&typed))
        });
        self.state_gate = Some(StateGate { predicate: erased });
        self
    }
}

/// A configurable compensation description: the handler plus its optional
/// await declaration. Returned by [`comp`] and consumed by
/// [`SagaBuilder::compensate`].
pub struct CompBuilder<H, EF = NoEffect> {
    pub(crate) handler: H,
    pub(crate) await_event_type: Option<&'static str>,
    pub(crate) even_if_active: bool,
    /// The release effect that fulfils the request this compensation emits,
    /// attached via [`CompBuilder::effect`]. `NoEffect` by default. Registered
    /// into the saga's request-handler map by [`RegisterComp`] when `.step()`
    /// wires the compensation in.
    pub(crate) effect: EF,
}

/// Wrap a compensation handler (a sync `fn` returning [`crate::Emit`])
/// so its options can be chained inside the `.compensate(name, ...)`
/// call. Like steps, a compensation is a pure `decide` — its release *effect*
/// is attached separately via [`CompBuilder::effect`].
pub fn comp<H>(handler: H) -> CompBuilder<H, NoEffect> {
    CompBuilder {
        handler,
        await_event_type: None,
        even_if_active: false,
        effect: NoEffect,
    }
}

impl<H, EF> CompBuilder<H, EF> {
    /// Declare which response event marks this compensation complete.
    /// Without it, the compensation completes as soon as its emits are
    /// dispatched (fire-and-forget / no-op compensations).
    pub fn on_response<E: EventTypeStatic>(mut self) -> Self {
        self.await_event_type = Some(<E as EventTypeStatic>::EVENT_TYPE);
        self
    }

    /// Run this compensation in the cascade even when the step was still
    /// **Active/Resolving** — in-flight external work whose response never
    /// arrived (review SAGA-13). Opt in only when the handler does not
    /// read the step's own `Output<T>`: a never-completed step has no
    /// output, so such an extractor fails at runtime and flags
    /// `CompensationFailed`.
    pub fn even_if_active(mut self) -> Self {
        self.even_if_active = true;
        self
    }
}

impl<H> CompBuilder<H, NoEffect> {
    /// Attach the in-process effect that fulfils this compensation's emitted
    /// request (the "undo" effect — release a hold, refund a charge),
    /// co-located on the compensation's chain. Same shape and lowering as
    /// [`TaskBuilder::effect`] / [`SagaBuilder::effect`].
    ///
    /// ```rust,ignore
    /// comp(release_flight)                  // sync decide: -> Emit(ReleaseFlightHold)
    ///     .effect(handle_release_flight)    // async effect: Req<ReleaseFlightHold> -> FlightReleased
    ///     .on_response::<FlightReleased>()
    /// ```
    pub fn effect<EH>(self, handler: EH) -> CompBuilder<H, EffectBuilder<EH>> {
        CompBuilder {
            handler: self.handler,
            await_event_type: self.await_event_type,
            even_if_active: self.even_if_active,
            effect: EffectBuilder { handler },
        }
    }
}

/// Returned by [`SagaBuilder::signal`] for policy chaining.
pub struct SignalRegistration<S: Saga, T> {
    pub(crate) builder: SagaBuilder<S>,
    pub(crate) _phantom: PhantomData<T>,
}

impl<S: Saga, T: 'static> SignalRegistration<S, T> {
    pub fn triggers_compensation(mut self) -> SagaBuilder<S> {
        self.builder.signals.push(SagaSignal {
            signal_type_name: std::any::type_name::<T>(),
            policy: SignalPolicy::TriggersCompensation,
        });
        self.builder
    }
}

impl<S: Saga, T: EventTypeStatic + 'static> SignalRegistration<S, T> {
    /// Route this signal to `name` as that step's awaited event. Requires
    /// `T: EventTypeStatic` so the validator can check at build time that
    /// the target step's `.on_response(...)` awaits the signal's event
    /// type.
    pub fn routes_to_step(mut self, name: StepName) -> SagaBuilder<S> {
        self.builder.signals.push(SagaSignal {
            signal_type_name: std::any::type_name::<T>(),
            policy: SignalPolicy::RoutesToStep {
                step: name,
                event_type: <T as EventTypeStatic>::EVENT_TYPE,
            },
        });
        self.builder
    }
}

// =====================================================================
// IntoStepNames — `.after("flight")` and `.after(["flight", "hotel"])`
// =====================================================================

/// Conversion trait for the [`StepRegistration::after`] argument so both a
/// single step name and a list of names work.
pub trait IntoStepNames {
    fn into_step_names(self) -> Vec<StepName>;
}

impl IntoStepNames for StepName {
    fn into_step_names(self) -> Vec<StepName> {
        vec![self]
    }
}

impl IntoStepNames for Vec<StepName> {
    fn into_step_names(self) -> Vec<StepName> {
        self
    }
}

impl<const N: usize> IntoStepNames for [StepName; N] {
    fn into_step_names(self) -> Vec<StepName> {
        self.to_vec()
    }
}

impl IntoStepNames for &[StepName] {
    fn into_step_names(self) -> Vec<StepName> {
        self.to_vec()
    }
}

impl<const N: usize> IntoStepNames for &[StepName; N] {
    fn into_step_names(self) -> Vec<StepName> {
        self.to_vec()
    }
}

// =====================================================================
// Entry point + validation
// =====================================================================

impl<S: Saga> SagaBuilder<S> {
    pub fn for_saga() -> Self {
        Self::new()
    }

    /// Compile the topology and return the validated form, or a
    /// [`TopologyError`] explaining what's wrong. Surfaced by
    /// [`SagaBuilder::build`] which panics on error; callers can use
    /// [`SagaBuilder::validate`] directly when they want a `Result` instead.
    ///
    /// Checks:
    /// - `start_on` must be present.
    /// - Step names are unique.
    /// - Every `.after(name)` must reference a registered step.
    /// - Every `.compensate(name, ...)` must reference a registered step.
    /// - `.signal::<T>().routes_to_step(name)` must reference a registered step.
    /// - At least one step must be marked terminal.
    pub fn validate(&self) -> Result<(), TopologyError> {
        if self.start_correlator.is_none() {
            return Err(TopologyError::MissingStartOn);
        }

        // Unique step names.
        let mut seen = std::collections::BTreeSet::new();
        for step in &self.steps {
            if !seen.insert(step.name) {
                return Err(TopologyError::DuplicateStepName(step.name));
            }
        }

        // `.after` refs must exist.
        for step in &self.steps {
            for dep in &step.after {
                if !seen.contains(dep) {
                    return Err(TopologyError::UnknownAfterRef {
                        step: step.name,
                        unknown: dep,
                    });
                }
                if *dep == step.name {
                    return Err(TopologyError::SelfAfterRef(step.name));
                }
            }
        }

        // (A compensation can only be attached to a step via
        // `task(..).compensate(..)`, so it always references a registered
        // step — no unknown-compensation-ref check is needed.)

        // `.gated_by` upstream refs must exist.
        for step in &self.steps {
            if let Some(gate) = &step.gate
                && !seen.contains(&gate.upstream)
            {
                return Err(TopologyError::UnknownGateRef {
                    step: step.name,
                    unknown: gate.upstream,
                });
            }
        }

        // Duplicate signal type registrations.
        let mut seen_signals = std::collections::BTreeSet::new();
        for sig in &self.signals {
            if !seen_signals.insert(sig.signal_type_name) {
                return Err(TopologyError::DuplicateSignal(sig.signal_type_name));
            }
        }

        // Output type uniqueness — Output<T> extractor disambiguates
        // upstream outputs by TypeId, so two steps producing the same
        // type would silently collapse. Catch at build time. Steps with
        // no typed output (fire-and-forget) are exempt — nothing can
        // extract them.
        let mut seen_outputs: BTreeMap<TypeId, StepName> = BTreeMap::new();
        for step in &self.steps {
            let Some((type_id, type_name)) = step.output else {
                continue;
            };
            if let Some(prior) = seen_outputs.insert(type_id, step.name) {
                return Err(TopologyError::DuplicateOutputType {
                    type_name,
                    first: prior,
                    second: step.name,
                });
            }
        }

        // A value (compute) handler with an `.on_response(...)` is
        // contradictory: its output comes from its return value and it
        // never suspends.
        for step in &self.steps {
            if step.response_on_value_step {
                return Err(TopologyError::ResponseOnComputeStep { step: step.name });
            }
        }

        // signal `routes_to_step` refs must exist, and the target step
        // must actually await the signal's event type (checkable at build
        // time now that awaits live on the topology).
        for sig in &self.signals {
            if let SignalPolicy::RoutesToStep { step, event_type } = sig.policy {
                if !seen.contains(&step) {
                    return Err(TopologyError::UnknownSignalRoute {
                        signal_type: sig.signal_type_name,
                        unknown: step,
                    });
                }
                let target_awaits = self
                    .steps
                    .iter()
                    .find(|s| s.name == step)
                    .and_then(|s| s.await_spec.as_ref().map(|a| a.event_type));
                if target_awaits != Some(event_type) {
                    return Err(TopologyError::SignalRouteAwaitMismatch {
                        signal_type: sig.signal_type_name,
                        step,
                        signal_event_type: event_type,
                        step_awaits: target_awaits,
                    });
                }
            }
        }

        // At least one terminal step.
        if !self.steps.iter().any(|s| s.terminal.is_some()) {
            return Err(TopologyError::NoTerminalStep);
        }

        // Build-time ambiguous-await check (C7 moved to validate()): two
        // steps that can be concurrently Active and await the same event
        // type are indistinguishable to the router. "Can be concurrently
        // Active" is approximated as: neither is an `after`-ancestor of
        // the other. Gated steps are exempt (mutually-exclusive branches
        // legitimately share a response type); the runtime refusal remains
        // the backstop for those.
        {
            let awaiting: Vec<&StepEntry<S>> = self
                .steps
                .iter()
                .filter(|s| s.await_spec.is_some() && s.gate.is_none() && s.state_gate.is_none())
                .collect();
            for (i, a) in awaiting.iter().enumerate() {
                for b in awaiting.iter().skip(i + 1) {
                    let (Some(spec_a), Some(spec_b)) = (&a.await_spec, &b.await_spec) else {
                        continue;
                    };
                    if spec_a.event_type != spec_b.event_type {
                        continue;
                    }
                    if !self.reaches(a.name, b.name) && !self.reaches(b.name, a.name) {
                        return Err(TopologyError::AmbiguousAwaitTypes {
                            event_type: spec_a.event_type,
                            first: a.name,
                            second: b.name,
                        });
                    }
                }
            }
        }

        // Cycle check via Kahn's algorithm. Distinguishes "you ordered
        // steps strangely" (now harmless — the runner re-evaluates each
        // pass) from "you created an actual cycle" (a hard error with
        // the offending path).
        let mut in_degree: std::collections::BTreeMap<StepName, usize> =
            self.steps.iter().map(|s| (s.name, 0)).collect();
        let mut adj: std::collections::BTreeMap<StepName, Vec<StepName>> =
            self.steps.iter().map(|s| (s.name, Vec::new())).collect();
        for step in &self.steps {
            for dep in &step.after {
                // dep → step (dep must complete before step).
                if let Some(list) = adj.get_mut(dep) {
                    list.push(step.name);
                }
                if let Some(d) = in_degree.get_mut(&step.name) {
                    *d += 1;
                }
            }
        }

        let mut queue: std::collections::VecDeque<StepName> = in_degree
            .iter()
            .filter(|(_, d)| **d == 0)
            .map(|(n, _)| *n)
            .collect();
        let mut visited = 0usize;
        while let Some(node) = queue.pop_front() {
            visited += 1;
            if let Some(next) = adj.get(&node) {
                for &dst in next {
                    if let Some(d) = in_degree.get_mut(&dst) {
                        *d -= 1;
                        if *d == 0 {
                            queue.push_back(dst);
                        }
                    }
                }
            }
        }

        if visited != self.steps.len() {
            // At least one step is part of a cycle. Surface a sample
            // cycle path so the user can see what to break.
            let unresolved: Vec<StepName> = in_degree
                .iter()
                .filter(|(_, d)| **d > 0)
                .map(|(n, _)| *n)
                .collect();
            return Err(TopologyError::CycleDetected { steps: unresolved });
        }

        // Extractor-requirement satisfiability. Walks each step's
        // declared extractor requirements (via the StepHandler tuple
        // impls) and checks them against the DAG. Catches things like
        // `Output<FlightReserved>` with no upstream producer at build
        // time rather than at first activation.
        use crate::extract::ExtractRequirement;
        let outputs_available: BTreeMap<TypeId, (StepName, &'static str)> = self
            .steps
            .iter()
            .filter_map(|s| s.output.map(|(id, name)| (id, (s.name, name))))
            .collect();
        let deps_available: BTreeSet<TypeId> = self.deps.keys().copied().collect();

        for step in &self.steps {
            for req in &step.requirements {
                match req {
                    ExtractRequirement::None => {}
                    ExtractRequirement::Trigger { type_id, type_name } => {
                        // start_on was already required (MissingStartOn). Now
                        // verify the step's `Trigger<T>` matches the
                        // registered trigger type — catches the typo
                        // where `start_on::<Foo, _>` is bound but the
                        // step reads `Trigger<Bar>`.
                        if let Some(hook) = &self.start_correlator
                            && hook.trigger_type_id != *type_id
                        {
                            return Err(TopologyError::TriggerTypeMismatch {
                                step: step.name,
                                requested: type_name,
                                registered: hook.trigger_type_name,
                            });
                        }
                    }
                    ExtractRequirement::UpstreamOutput { type_id, type_name } => {
                        if !outputs_available.contains_key(type_id) {
                            let produced: Vec<(StepName, &'static str)> =
                                outputs_available.values().copied().collect();
                            return Err(TopologyError::UnsatisfiableOutputExtractor {
                                step: step.name,
                                needed_type: type_name,
                                produced,
                            });
                        }
                    }
                    ExtractRequirement::Dep { type_id, type_name } => {
                        if !deps_available.contains(type_id) {
                            return Err(TopologyError::UnsatisfiableDepExtractor {
                                step: step.name,
                                needed_type: type_name,
                            });
                        }
                    }
                    ExtractRequirement::CompensationOnly => {
                        return Err(TopologyError::CauseExtractorInForwardStep { step: step.name });
                    }
                    ExtractRequirement::RequestOnly => {
                        return Err(TopologyError::ReqExtractorInForwardStep { step: step.name });
                    }
                }
            }
        }

        // Compensation handlers — same shape as forward steps, except
        // `Cause` is the *expected* extractor here (not a misuse). The
        // validator allows CompensationOnly and flags everything else
        // that wouldn't work at runtime.
        for (step_name, comp) in &self.compensations {
            for req in &comp.requirements {
                match req {
                    ExtractRequirement::None | ExtractRequirement::CompensationOnly => {}
                    ExtractRequirement::Trigger { .. } => {
                        // Trigger in compensation is unusual but technically
                        // works — the trigger payload survives on the
                        // envelope. Allow.
                    }
                    ExtractRequirement::UpstreamOutput { type_id, type_name } => {
                        if !outputs_available.contains_key(type_id) {
                            let produced: Vec<(StepName, &'static str)> =
                                outputs_available.values().copied().collect();
                            return Err(
                                TopologyError::UnsatisfiableOutputExtractorInCompensation {
                                    step: step_name,
                                    needed_type: type_name,
                                    produced,
                                },
                            );
                        }
                    }
                    ExtractRequirement::Dep { type_id, type_name } => {
                        if !deps_available.contains(type_id) {
                            return Err(TopologyError::UnsatisfiableDepExtractorInCompensation {
                                step: step_name,
                                needed_type: type_name,
                            });
                        }
                    }
                    ExtractRequirement::RequestOnly => {
                        return Err(TopologyError::ReqExtractorInCompensation { step: step_name });
                    }
                }
            }
        }

        Ok(())
    }

    /// Run validation; panic with a clear message if the topology is malformed.
    /// Most users call this implicitly via `Saga::build` → runtime registration.
    pub fn assert_valid(&self) {
        if let Err(e) = self.validate() {
            panic!("invalid saga topology for `{}`: {e}", S::name());
        }
    }

    /// Is `to` reachable from `from` along `after` edges (i.e. is `from`
    /// an ancestor of `to`)? Used by the build-time ambiguous-await check
    /// to decide whether two awaiting steps can be concurrently Active.
    fn reaches(&self, from: StepName, to: StepName) -> bool {
        let mut stack = vec![from];
        let mut visited = BTreeSet::new();
        while let Some(node) = stack.pop() {
            if node == to {
                return true;
            }
            if !visited.insert(node) {
                continue;
            }
            for step in &self.steps {
                if step.after.contains(&node) {
                    stack.push(step.name);
                }
            }
        }
        false
    }

    /// Internal accessor for tests + the runner. Not part of the stable API.
    #[doc(hidden)]
    pub fn step_entries(&self) -> &[StepEntry<S>] {
        &self.steps
    }

    #[doc(hidden)]
    pub fn start_correlator(&self) -> Option<&TriggerHook<S>> {
        self.start_correlator.as_ref()
    }

    #[doc(hidden)]
    pub fn deps_ref(&self) -> &BTreeMap<TypeId, Arc<dyn std::any::Any + Send + Sync>> {
        &self.deps
    }

    #[doc(hidden)]
    pub fn signals_iter(&self) -> impl Iterator<Item = &SagaSignal> {
        self.signals.iter()
    }

    #[doc(hidden)]
    pub fn has_compensation(&self, step: StepName) -> bool {
        self.compensations.contains_key(step)
    }

    /// Whether `step`'s compensation was registered with
    /// `compensate_even_if_active` — i.e. the cascade should run it even
    /// for a step that was still in flight (review SAGA-13).
    #[doc(hidden)]
    pub fn compensates_even_if_active(&self, step: StepName) -> bool {
        self.compensations
            .get(step)
            .map(|c| c.even_if_active)
            .unwrap_or(false)
    }

    #[doc(hidden)]
    pub fn compensation_handler(&self, step: StepName) -> Option<ErasedCompensationHandler<S>> {
        self.compensations.get(step).map(|c| c.handler.clone())
    }

    /// The response event type that completes `step`'s compensation, as
    /// declared via `comp(...).on_response::<E>()`. `None` = the
    /// compensation completes on dispatch.
    #[doc(hidden)]
    pub fn compensation_await_type(&self, step: StepName) -> Option<&'static str> {
        self.compensations
            .get(step)
            .and_then(|c| c.await_event_type)
    }

    /// The response event type `step` awaits, as declared via
    /// `task(...).on_response(...)`. `None` for value steps and
    /// fire-and-forget emits.
    #[doc(hidden)]
    pub fn step_await_event_type(&self, step: StepName) -> Option<&'static str> {
        self.steps
            .iter()
            .find(|s| s.name == step)
            .and_then(|s| s.await_spec.as_ref().map(|a| a.event_type))
    }

    #[doc(hidden)]
    pub fn request_handler(&self, event_type: &str) -> Option<ErasedRequestHandler<S>> {
        self.request_handlers.get(event_type).cloned()
    }

    #[doc(hidden)]
    pub fn is_unhandled_allowed(&self, event_type: &str) -> bool {
        self.allowed_unhandled.contains(event_type)
    }

    /// Render the saga topology as a Mermaid `flowchart TD` diagram.
    ///
    /// The DAG is data — emit it for diff-friendly architecture docs,
    /// PR-attached screenshots, or onboarding material. Steps appear as
    /// nodes, `.after()` deps as edges; terminal-success / -failure
    /// nodes are styled; compensations are rendered as dashed back-edges;
    /// signals show up as floating triggers.
    ///
    /// # Example
    ///
    /// ```text
    /// flowchart TD
    ///   flight["flight"]
    ///   hotel["hotel"]
    ///   payment["payment"]
    ///   confirm(["confirm ✓"])
    ///   flight --> payment
    ///   hotel --> payment
    ///   payment --> confirm
    ///   flight -.compensate.-> flight
    /// ```
    pub fn mermaid(&self) -> String {
        let mut out = String::from("flowchart TD\n");

        for step in &self.steps {
            let label = match step.terminal {
                Some(Terminal::Success) => format!("  {0}([\"{0} ✓\"])\n", step.name),
                Some(Terminal::Failure) => format!("  {0}([\"{0} ✗\"])\n", step.name),
                None => format!("  {0}[\"{0}\"]\n", step.name),
            };
            out.push_str(&label);
        }

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

        for step in &self.steps {
            if self.compensations.contains_key(step.name) {
                out.push_str(&format!("  {0} -.compensate.-> {0}\n", step.name));
            }
        }

        for sig in &self.signals {
            let sig_id = sig.signal_type_name.replace(':', "_");
            match sig.policy {
                SignalPolicy::TriggersCompensation => {
                    out.push_str(&format!(
                        "  {0}{{{{\"signal: {1}\"}}}} -.cancel.-> {0}\n",
                        sig_id, sig.signal_type_name
                    ));
                }
                SignalPolicy::RoutesToStep { step: target, .. } => {
                    out.push_str(&format!(
                        "  {0}{{{{\"signal: {1}\"}}}} ==> {2}\n",
                        sig_id, sig.signal_type_name, target
                    ));
                }
            }
        }

        out
    }
}

fn format_produced(produced: &[(StepName, &'static str)]) -> String {
    if produced.is_empty() {
        return String::from("<none — no step in this saga produces a typed output>");
    }
    produced
        .iter()
        .map(|(step, ty)| format!("`{step}` → {ty}"))
        .collect::<Vec<_>>()
        .join(", ")
}

/// Anything that can go wrong validating a saga topology before the runner
/// starts. Surfaced via `SagaBuilder::validate` (returns Result) or
/// `SagaBuilder::assert_valid` / the runner registration (panics).
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum TopologyError {
    #[error("saga must have a `start_on(...)` registration")]
    MissingStartOn,
    #[error("duplicate step name `{0}`")]
    DuplicateStepName(StepName),
    #[error("step `{step}` declared `.after(`{unknown}`)` but no such step is registered")]
    UnknownAfterRef { step: StepName, unknown: StepName },
    #[error("step `{0}` declared `.after({0})` — a step cannot depend on itself")]
    SelfAfterRef(StepName),
    #[error(
        "cycle detected in saga DAG — steps in the cycle: {steps:?}. \
         Break the cycle by removing one of the `.after(...)` references."
    )]
    CycleDetected { steps: Vec<StepName> },
    #[error(
        "step `{step}` declared `.gated_by(\"{unknown}\", ...)` but no such step is registered"
    )]
    UnknownGateRef { step: StepName, unknown: StepName },
    #[error(
        "steps `{first}` and `{second}` both produce output of type `{type_name}` — \
         `Output<T>` cannot disambiguate. Wrap one of them in a newtype to make their \
         output types distinct."
    )]
    DuplicateOutputType {
        type_name: &'static str,
        first: StepName,
        second: StepName,
    },
    #[error("signal type `{0}` registered more than once")]
    DuplicateSignal(&'static str),
    #[error(
        "step `{step}` reads `Output<{needed_type}>` but no step in this saga \
         produces `{needed_type}`. Steps that DO produce typed outputs: \
         [{}]. Fixes: add a step producing `{needed_type}` (a value handler \
         returning it, or an `.on_response` map yielding it) \
         (and make sure `{step}` declares `.after(<producer>)`), or rename the \
         extractor type to match an existing producer.",
        format_produced(produced)
    )]
    UnsatisfiableOutputExtractor {
        step: StepName,
        needed_type: &'static str,
        produced: Vec<(StepName, &'static str)>,
    },
    #[error(
        "step `{step}` reads `Dep<{needed_type}>` but no dep of that type is \
         registered. Fix: add `.dep(value)` on the builder where `value: \
         {needed_type}`."
    )]
    UnsatisfiableDepExtractor {
        step: StepName,
        needed_type: &'static str,
    },
    #[error(
        "step `{step}` uses the `Cause` extractor, which is only valid inside \
         compensation handlers (where it surfaces the originating step error). \
         Fix: remove `Cause` from this step's signature, or move the logic \
         into the step's `.compensate(comp(handler))`."
    )]
    CauseExtractorInForwardStep { step: StepName },
    #[error(
        "step `{step}` uses the `Req<R>` extractor, which is only valid inside \
         effect handlers registered via `.effect(...)`. Fix: remove `Req<R>` \
         from this step, or register an effect handler with that signature."
    )]
    ReqExtractorInForwardStep { step: StepName },
    #[error(
        "compensation for step `{step}` reads `Output<{needed_type}>` but no \
         step in this saga produces `{needed_type}`. Steps that DO produce \
         typed outputs: [{}]. Fix: produce the type from some step, or \
         change the compensation's extractor to match an existing output.",
        format_produced(produced)
    )]
    UnsatisfiableOutputExtractorInCompensation {
        step: StepName,
        needed_type: &'static str,
        produced: Vec<(StepName, &'static str)>,
    },
    #[error(
        "compensation for step `{step}` reads `Dep<{needed_type}>` but no \
         dep of that type is registered. Fix: add `.dep(value)` on the \
         builder where `value: {needed_type}`."
    )]
    UnsatisfiableDepExtractorInCompensation {
        step: StepName,
        needed_type: &'static str,
    },
    #[error(
        "compensation for step `{step}` uses the `Req<R>` extractor — only \
         valid inside effect handlers registered via `.effect(...)`. Fix: \
         remove `Req<R>` from this compensation handler."
    )]
    ReqExtractorInCompensation { step: StepName },
    #[error(
        "step `{step}` reads `Trigger<{requested}>` but `.start_on(...)` is \
         registered with event type `{registered}` — the runtime would fail \
         to deserialise the trigger payload. Fix: change the step's extractor \
         to `Trigger<{registered}>`, or change `.start_on::<...>(...)` to \
         emit `{requested}`."
    )]
    TriggerTypeMismatch {
        step: StepName,
        requested: &'static str,
        registered: &'static str,
    },
    #[error(
        "`.signal::<{signal_type}>().routes_to_step(\"{unknown}\")` references unknown step `{unknown}`"
    )]
    UnknownSignalRoute {
        signal_type: &'static str,
        unknown: StepName,
    },
    #[error(
        "saga has no terminal step (mark one with `.terminal_success()` or `.terminal_failure()`)"
    )]
    NoTerminalStep,
    #[error(
        "step `{step}` is a compute step (its handler returns a value, not `Emit`) but the task \
         declares `.on_response(...)` — a compute step completes immediately with its return \
         value and never awaits. Fix: drop the `.on_response`, or make the handler return \
         `Emit` and emit the request it awaits a response to."
    )]
    ResponseOnComputeStep { step: StepName },
    #[error(
        "steps `{first}` and `{second}` can be concurrently active and both await event type \
         `{event_type}` — routing is by event type, so their responses are indistinguishable. \
         Give them distinct response event types, or order them with `.after(...)`."
    )]
    AmbiguousAwaitTypes {
        event_type: &'static str,
        first: StepName,
        second: StepName,
    },
    #[error(
        "`.signal::<{signal_type}>().routes_to_step(\"{step}\")` routes event type \
         `{signal_event_type}`, but step `{step}` awaits {}. Fix: declare \
         `.on_response::<{signal_type}>(...)` (or `_event`) on the target task, or route the \
         signal to the step that awaits it.",
        step_awaits.map(|t| format!("`{t}`")).unwrap_or_else(|| "nothing (no `.on_response` declared)".into())
    )]
    SignalRouteAwaitMismatch {
        signal_type: &'static str,
        step: StepName,
        signal_event_type: &'static str,
        step_awaits: Option<&'static str>,
    },
}
