aion/engine/

builder.rs

//! `EngineBuilder` and build wiring.

use std::{num::NonZeroUsize, path::PathBuf, sync::Arc, time::Duration};

use chrono::Utc;

use aion_core::SearchAttributeSchema;
use aion_package::{ExtractionLimits, Package};
use aion_store::visibility::VisibilityStore;
use aion_store::{EventStore, InMemoryStore};

use crate::{
    EngineError, Registry, RuntimeConfig, RuntimeHandle, SignalDeliveryConfig, SupervisionTree,
    WorkflowCatalog,
    activity::bridge::ActivityDispatcher,
    durability::ActiveWorkflowRecoverySeam,
    runtime::{
        ChildNifBridge, ChildNifBridgeParts, NifEntry, NifRegistration, install_child_nif_bridge,
        install_nif_runtime_context, install_query_bridge, install_signal_nif_bridge,
        nif_determinism::{NifContextSource, install_nif_context_source},
    },
    signal::SignalResumeHandoff,
};

use super::api::{Engine, EngineComponents};
use super::delegated::{DelegatedSeams, EventPublisher, QueryService, SignalRouter};
use super::seams::{
    SeamAssembly, SignalRouterFactory, assemble_delegated_seams, wrap_event_streaming,
};
use super::startup::{
    StartupRecoveryContext, recover_active_workflows_on_startup, recover_timers_on_startup,
};

/// Source for a workflow package collected before `build()` performs fallible
/// loading and runtime registration.
#[derive(Clone, Debug)]
pub enum WorkflowPackageSource {
    /// Load a package from this `.aion` archive path during `build()`.
    Path(PathBuf),
    /// Use an already-loaded package value.
    Package(Box<Package>),
}

/// Install the engine-scoped NIF seams that are available before delegated
/// seams exist: runtime context, timer bridge, deterministic context source,
/// query bridge, and the optional activity dispatcher.
///
/// Returns the query mailbox engine handle installed in the query bridge, so
/// `build()` can wire the concrete query-dispatch seam over the same
/// delivery path the NIF-side `dispatch_query` uses.
fn install_engine_nif_seams(
    nif_state: &Arc<crate::runtime::EngineNifState>,
    registry: &Arc<Registry>,
    store: &Arc<dyn EventStore>,
    runtime: &Arc<RuntimeHandle>,
    activity_dispatcher: Option<Arc<dyn ActivityDispatcher>>,
    query_timeout: Option<Duration>,
) -> Arc<dyn crate::engine_seam::EngineHandle> {
    install_nif_runtime_context(
        nif_state,
        Arc::clone(registry),
        Arc::clone(runtime),
        tokio::runtime::Handle::current(),
    );
    crate::runtime::nif_timer_bridge::install_timer_nif_bridge(
        nif_state,
        Arc::clone(registry),
        Arc::clone(store),
        tokio::runtime::Handle::current(),
        runtime.signal_delivery(),
    );
    install_nif_context_source(
        nif_state,
        Arc::new(NifContextSource::new(
            Arc::clone(registry),
            tokio::runtime::Handle::current(),
            Arc::clone(store),
            runtime.signal_delivery(),
        )),
    );
    let query_mailbox_engine = install_query_bridge(
        nif_state,
        Arc::clone(registry),
        runtime,
        tokio::runtime::Handle::current(),
        query_timeout,
    );
    if let Some(dispatcher) = activity_dispatcher {
        nif_state.set_activity_dispatcher(dispatcher);
    }
    query_mailbox_engine
}

/// Assemble the startup catalog: persisted runtime deploys reload first
/// (with their persisted route pointers), then explicit operator-supplied
/// sources load on top.
///
/// The order is the routing-intent precedence: a package named explicitly at
/// THIS boot (`--workflow-package` / builder source) is the operator's newest
/// instruction and wins the route for its type, while every persisted deploy
/// still reloads so startup recovery — which runs after this and resolves
/// each run's recorded pinned version — finds every version it needs.
/// Operator-file sources are not persisted; only the runtime deploy seam
/// writes package rows.
async fn assemble_startup_catalog(
    runtime: &RuntimeHandle,
    store: &dyn EventStore,
    sources: Vec<WorkflowPackageSource>,
) -> Result<Arc<WorkflowCatalog>, EngineError> {
    let catalog = Arc::new(WorkflowCatalog::new());
    crate::loader::persistence::reload_persisted_packages(runtime, catalog.as_ref(), store).await?;
    for source in sources {
        let package = package_from_source(source)?;
        let outcome = catalog.load_package(runtime, &package).await?;
        tracing::info!(
            workflow_type = outcome.record.workflow_type(),
            content_hash = %outcome.record.version(),
            freshly_loaded = outcome.freshly_loaded,
            "loaded workflow package {}",
            outcome.record.workflow_type()
        );
    }
    Ok(catalog)
}

impl From<Package> for WorkflowPackageSource {
    fn from(package: Package) -> Self {
        Self::Package(Box::new(package))
    }
}

fn spawn_visibility_reconciliation_task(
    interval: Duration,
    store: Arc<dyn EventStore>,
    visibility_store: Arc<dyn VisibilityStore>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        loop {
            tokio::time::sleep(interval).await;
            if let Err(error) = crate::lifecycle::visibility::reconcile_visibility(
                Arc::clone(&store),
                Arc::clone(&visibility_store),
            )
            .await
            {
                tracing::warn!(
                    error = %error,
                    "periodic visibility reconciliation failed; crash-consistency window may remain until a later reconciliation repairs visibility"
                );
            }
        }
    })
}

/// Apply owned-shard scoping to the store BEFORE any recovery or enumeration
/// reads it, so a multi-shard node recovers only its shards.
///
/// `None` leaves the store untouched — the single-node default, where the store
/// owns ALL shards and boot is byte-identical to today (the scoping hook is
/// never called). `Some(set)` forwards through any store decorator to the
/// sharded backend; a single-shard backend ignores it.
fn apply_owned_shards(store: &dyn EventStore, owned_shards: Option<&[usize]>) {
    if let Some(shards) = owned_shards {
        store.set_owned_shards(Some(shards));
    }
}

/// Win the per-shard election and become the live owner of each owned shard
/// BEFORE startup recovery reads them (SS-2).
///
/// Ordering matters: this runs after [`apply_owned_shards`] (so the store is
/// already scoped to this node's shards) and BEFORE
/// [`recover_active_workflows_on_startup`], so a distributed backend's
/// `become_live` union-merge has made every committed write on those shards
/// locally present before recovery enumerates them. The election is driven
/// through the type-erased [`ReadableEventStore::acquire_owned_shards`] seam,
/// whose distributed implementation runs the blocking coordinator on a bare
/// off-runtime thread — so calling it from this async `build()` honours
/// haematite's no-blocking-election-inside-an-async-context constraint.
///
/// `None` (the single-node default) skips election entirely, and the seam is a
/// no-op for every non-distributed backend even when a shard set is configured,
/// so boot stays byte-identical to today.
fn acquire_owned_shards(
    store: &dyn EventStore,
    owned_shards: Option<&[usize]>,
) -> Result<(), EngineError> {
    if let Some(shards) = owned_shards {
        store.acquire_owned_shards(shards)?;
    }
    Ok(())
}

/// Spawn the periodic visibility reconciliation task when an interval is
/// configured, returning its join handle; otherwise return `None`.
fn maybe_spawn_visibility_reconciliation(
    interval: Option<Duration>,
    store: &Arc<dyn EventStore>,
    visibility_store: &Arc<dyn VisibilityStore>,
) -> Option<tokio::task::JoinHandle<()>> {
    interval.map(|interval| {
        spawn_visibility_reconciliation_task(
            interval,
            Arc::clone(store),
            Arc::clone(visibility_store),
        )
    })
}

impl From<PathBuf> for WorkflowPackageSource {
    fn from(path: PathBuf) -> Self {
        Self::Path(path)
    }
}

impl From<&std::path::Path> for WorkflowPackageSource {
    fn from(path: &std::path::Path) -> Self {
        Self::Path(path.to_path_buf())
    }
}

impl From<&str> for WorkflowPackageSource {
    fn from(path: &str) -> Self {
        Self::Path(PathBuf::from(path))
    }
}

impl From<String> for WorkflowPackageSource {
    fn from(path: String) -> Self {
        Self::Path(PathBuf::from(path))
    }
}

/// Tracks which optional engine seams the caller explicitly overrode, so
/// `build()` can detect mutually-exclusive configuration (e.g. an event
/// publisher set both directly and via event streaming). Grouped so the builder
/// keeps its boolean configuration flags few and named.
#[derive(Default)]
struct SeamOverrides {
    /// The caller installed an explicit event-publisher seam.
    event_publisher: bool,
    /// The caller installed an explicit query-service seam.
    query_service: bool,
}

/// Builder for the embedded, transport-agnostic workflow engine.
pub struct EngineBuilder {
    store: Option<Arc<dyn EventStore>>,
    visibility_store: Option<Arc<dyn VisibilityStore>>,
    scheduler_threads: Option<usize>,
    signal_delivery: SignalDeliveryConfig,
    outbox_enabled: bool,
    bootstrap_schedule_coordinator: bool,
    owned_shards: Option<Vec<usize>>,
    workflow_sources: Vec<WorkflowPackageSource>,
    host_nifs: Vec<NifEntry>,
    recovery: Option<Arc<dyn ActiveWorkflowRecoverySeam>>,
    delegated: DelegatedSeams,
    signal_router_factory: Option<SignalRouterFactory>,
    activity_dispatcher: Option<Arc<dyn ActivityDispatcher>>,
    active_registry: Option<Arc<Registry>>,
    visibility_reconciliation_interval: Option<Duration>,
    search_attribute_schema: SearchAttributeSchema,
    event_streaming_capacity: Option<NonZeroUsize>,
    query_timeout: Option<Duration>,
    seam_overrides: SeamOverrides,
}

impl Default for EngineBuilder {
    fn default() -> Self {
        Self::new()
    }
}

impl EngineBuilder {
    /// Create a builder with no store, no scheduler-thread override, no loaded
    /// workflows, and no host NIFs.
    #[must_use]
    pub fn new() -> Self {
        Self {
            store: None,
            visibility_store: None,
            scheduler_threads: None,
            signal_delivery: SignalDeliveryConfig::default(),
            outbox_enabled: false,
            // The only field that defaults true: single-node engines seed the
            // schedule coordinator. A multi-node deployment disables it on nodes
            // that do not own the coordinator's shard (see the builder method).
            bootstrap_schedule_coordinator: true,
            // No shard restriction by default: the store owns ALL shards, which
            // is byte-identical to single-node behaviour. `build()` only ever
            // touches owned-shard scoping when a deployment sets this.
            owned_shards: None,
            workflow_sources: Vec::new(),
            host_nifs: Vec::new(),
            recovery: None,
            delegated: DelegatedSeams::default(),
            signal_router_factory: None,
            activity_dispatcher: None,
            active_registry: None,
            visibility_reconciliation_interval: None,
            search_attribute_schema: SearchAttributeSchema::new(),
            event_streaming_capacity: None,
            query_timeout: None,
            seam_overrides: SeamOverrides::default(),
        }
    }

    /// Record the caller-supplied workflow query reply timeout.
    ///
    /// Setting a timeout installs the concrete query-dispatch seam during
    /// `build()` (unless [`Self::query_service`] overrides it) and enables
    /// the in-engine `dispatch_query` NIF. There is no default: without this
    /// call the query seam stays deferred and `Engine::query` fails typed
    /// with its "not configured" error.
    #[must_use]
    pub const fn query_timeout(mut self, timeout: Duration) -> Self {
        self.query_timeout = Some(timeout);
        self
    }

    /// Inspect the configured workflow query reply timeout.
    #[must_use]
    pub const fn configured_query_timeout(&self) -> Option<Duration> {
        self.query_timeout
    }

    /// Opt in to live event streaming with a caller-provided broadcast capacity.
    ///
    /// `build()` wraps the configured store in a
    /// [`PublishingEventStore`](crate::publish::PublishingEventStore) before any
    /// recorder, recovery, or NIF bridge captures the store — so every
    /// successful append publishes — and installs the matching
    /// [`BroadcastEventPublisher`](crate::publish::BroadcastEventPublisher) as
    /// the event-publisher seam behind [`Engine::subscribe`]. Without this call
    /// the deferred publisher remains installed and subscriptions are empty.
    #[must_use]
    pub const fn event_streaming(mut self, capacity: NonZeroUsize) -> Self {
        self.event_streaming_capacity = Some(capacity);
        self
    }

    /// Supply the search attribute schema validating every recorded attribute.
    ///
    /// The default schema is empty, which rejects all search attributes: a
    /// deployment must declare each attribute name and type before workflows
    /// can record values for it.
    #[must_use]
    pub fn search_attribute_schema(mut self, schema: SearchAttributeSchema) -> Self {
        self.search_attribute_schema = schema;
        self
    }

    /// Supply the event store used by the engine.
    #[must_use]
    pub fn store<S>(mut self, store: S) -> Self
    where
        S: EventStore,
    {
        self.store = Some(Arc::new(store));
        self
    }

    /// Supply an already type-erased event store.
    #[must_use]
    pub fn store_arc(mut self, store: Arc<dyn EventStore>) -> Self {
        self.store = Some(store);
        self
    }

    /// Supply the visibility store used by the engine for workflow projections.
    #[must_use]
    pub fn visibility_store<S>(mut self, visibility_store: S) -> Self
    where
        S: VisibilityStore,
    {
        self.visibility_store = Some(Arc::new(visibility_store));
        self
    }

    /// Supply an already type-erased visibility store.
    #[must_use]
    pub fn visibility_store_arc(mut self, visibility_store: Arc<dyn VisibilityStore>) -> Self {
        self.visibility_store = Some(visibility_store);
        self
    }

    /// Explicitly opt in to an ephemeral in-memory visibility store.
    ///
    /// This is intended for tests and local scenarios that do not need durable
    /// visibility projections. Visibility data stored this way does not survive
    /// process restarts.
    #[must_use]
    pub fn in_memory_visibility(mut self) -> Self {
        self.visibility_store = Some(Arc::new(InMemoryStore::default()));
        self
    }

    /// Record the caller-supplied scheduler thread count.
    ///
    /// If this setter is never called, `None` is passed through to beamr.
    #[must_use]
    pub const fn scheduler_threads(mut self, threads: usize) -> Self {
        self.scheduler_threads = Some(threads);
        self
    }

    /// Record the caller-supplied periodic visibility reconciliation interval.
    ///
    /// If this setter is never called, no periodic background reconciliation task is spawned.
    #[must_use]
    pub const fn visibility_reconciliation_interval(mut self, interval: Duration) -> Self {
        self.visibility_reconciliation_interval = Some(interval);
        self
    }

    /// Record the caller-supplied signal delivery readiness and retry policy.
    #[must_use]
    pub const fn signal_delivery(mut self, signal_delivery: SignalDeliveryConfig) -> Self {
        self.signal_delivery = signal_delivery;
        self
    }

    /// Record whether the durable-outbox fan-out dispatch path is enabled.
    #[must_use]
    pub fn outbox_enabled(mut self, enabled: bool) -> Self {
        self.outbox_enabled = enabled;
        self
    }

    /// Control whether `build()` seeds the schedule-coordinator history.
    ///
    /// Default `true` (single-node). Under multi-shard active-active the
    /// coordinator stream is owned by exactly ONE shard; a deployment sets this
    /// `false` on every node that does NOT own that shard, so only the owner
    /// seeds (and serves) it — a non-owner would otherwise try to write the
    /// coordinator stream and race or fence the real owner.
    #[must_use]
    pub const fn bootstrap_schedule_coordinator(mut self, enabled: bool) -> Self {
        self.bootstrap_schedule_coordinator = enabled;
        self
    }

    /// Restrict this engine's store to the distribution shards this node owns.
    ///
    /// Under multi-shard active-active a node serves only a SUBSET of the
    /// cluster's shards. `build()` calls
    /// [`ReadableEventStore::set_owned_shards`](aion_store::ReadableEventStore::set_owned_shards)
    /// with this set BEFORE startup recovery, so the node recovers and
    /// enumerates only the workflows / timers / outbox rows that live on its
    /// shards. The set is deduplicated and ordered by the store.
    ///
    /// Not calling this leaves the store owning ALL shards — the single-node
    /// default, which is byte-identical to today's behaviour (`build()` never
    /// touches the scoping hook). Single-shard backends (in-memory, libSQL)
    /// ignore the call regardless, since they own everything unconditionally.
    #[must_use]
    pub fn owned_shards(mut self, shards: impl IntoIterator<Item = usize>) -> Self {
        self.owned_shards = Some(shards.into_iter().collect());
        self
    }

    /// Inspect the configured owned-shard set (`None` = own all shards).
    #[must_use]
    pub fn configured_owned_shards(&self) -> Option<&[usize]> {
        self.owned_shards.as_deref()
    }

    /// Add one workflow package source to load during `build()`.
    #[must_use]
    pub fn load_workflows(mut self, source: impl Into<WorkflowPackageSource>) -> Self {
        self.workflow_sources.push(source.into());
        self
    }

    /// Add many workflow package sources to load during `build()`.
    #[must_use]
    pub fn load_workflow_sources<I, S>(mut self, sources: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<WorkflowPackageSource>,
    {
        self.workflow_sources
            .extend(sources.into_iter().map(Into::into));
        self
    }

    /// Collect host-supplied NIF entries to install before workflow modules load.
    #[must_use]
    pub fn register_nifs(mut self, entries: impl IntoIterator<Item = NifEntry>) -> Self {
        self.host_nifs.extend(entries);
        self
    }

    /// Override the AD recovery seam used while repopulating active workflows.
    #[must_use]
    pub fn recovery_seam(mut self, recovery: Arc<dyn ActiveWorkflowRecoverySeam>) -> Self {
        self.recovery = Some(recovery);
        self
    }

    /// Use the production AD recovery seam created after runtime/package loading.
    #[must_use]
    pub fn production_recovery_seam(mut self) -> Self {
        self.recovery = None;
        self
    }

    /// Override the AT signal-routing seam.
    #[must_use]
    pub fn signal_router(mut self, signal_router: Arc<dyn SignalRouter>) -> Self {
        self.signal_router_factory = None;
        self.delegated = DelegatedSeams::new(
            signal_router,
            self.delegated.query_service_arc(),
            self.delegated.event_publisher_arc(),
        );
        self
    }

    /// Override the AT signal-routing seam after the runtime is assembled.
    #[must_use]
    pub fn signal_router_factory<F>(mut self, factory: F) -> Self
    where
        F: Fn(Arc<RuntimeHandle>, Arc<SignalResumeHandoff>) -> Arc<dyn SignalRouter>
            + Send
            + Sync
            + 'static,
    {
        self.signal_router_factory = Some(Arc::new(factory));
        self
    }

    /// Override the AT query-dispatch seam.
    ///
    /// An explicit override wins over the concrete service that
    /// [`Self::query_timeout`] would otherwise install during `build()`.
    #[must_use]
    pub fn query_service(mut self, query_service: Arc<dyn QueryService>) -> Self {
        self.seam_overrides.query_service = true;
        self.delegated = DelegatedSeams::new(
            self.delegated.signal_router_arc(),
            query_service,
            self.delegated.event_publisher_arc(),
        );
        self
    }

    /// Override the AD/AT live event-publisher seam.
    ///
    /// Mutually exclusive with [`Self::event_streaming`], which installs the
    /// broadcast publisher itself; configuring both fails `build()`.
    #[must_use]
    pub fn event_publisher(mut self, event_publisher: Arc<dyn EventPublisher>) -> Self {
        self.seam_overrides.event_publisher = true;
        self.delegated = DelegatedSeams::new(
            self.delegated.signal_router_arc(),
            self.delegated.query_service_arc(),
            event_publisher,
        );
        self
    }

    /// Supply the activity dispatcher that backs activity dispatch NIFs.
    ///
    /// When set, the dispatcher is installed in the global bridge before
    /// workflow modules are loaded. Without a dispatcher, `dispatch_activity`
    /// returns an error to workflow code instead of crashing the process.
    #[must_use]
    pub fn activity_dispatcher(mut self, dispatcher: Arc<dyn ActivityDispatcher>) -> Self {
        self.activity_dispatcher = Some(dispatcher);
        self
    }

    /// Supply the active workflow registry used by the built engine.
    ///
    /// Server-owned dispatchers that run behind raw NIFs use this to correlate a
    /// calling BEAM pid to the same workflow handle the engine registers.
    #[must_use]
    pub fn active_registry(mut self, registry: Arc<Registry>) -> Self {
        self.active_registry = Some(registry);
        self
    }

    /// Inspect the configured scheduler thread count.
    #[must_use]
    pub const fn scheduler_thread_count(&self) -> Option<usize> {
        self.scheduler_threads
    }

    /// Inspect the configured periodic visibility reconciliation interval.
    #[must_use]
    pub const fn configured_visibility_reconciliation_interval(&self) -> Option<Duration> {
        self.visibility_reconciliation_interval
    }

    /// Assemble the runtime configuration from the builder-supplied scheduler,
    /// signal delivery, and outbox knobs.
    fn runtime_config(&self) -> RuntimeConfig {
        RuntimeConfig::new(self.scheduler_threads)
            .with_signal_delivery(self.signal_delivery)
            .with_outbox_enabled(self.outbox_enabled)
    }

    /// Construct the live engine.
    ///
    /// # Errors
    ///
    /// Returns typed [`EngineError`] variants for missing store, runtime startup,
    /// NIF registration, package loading, store reads, registry/supervision lock
    /// poison, or deferred AD recovery failures for active histories.
    pub async fn build(self) -> Result<Engine, EngineError> {
        let runtime_config = self.runtime_config();
        let (store, streaming_publisher) = wrap_event_streaming(
            self.store.ok_or(EngineError::MissingStore)?,
            self.event_streaming_capacity,
            self.seam_overrides.event_publisher,
        )?;
        let visibility_store = self
            .visibility_store
            .ok_or(EngineError::MissingVisibilityStore)?;

        apply_owned_shards(store.as_ref(), self.owned_shards.as_deref());
        // SS-2: become the fenced live owner of this node's shards BEFORE any
        // recovery enumerates them, so a distributed backend's `become_live`
        // union-merge has landed every committed write locally first. No-op for
        // single-node / non-distributed backends, so default boot is unchanged.
        acquire_owned_shards(store.as_ref(), self.owned_shards.as_deref())?;

        let runtime = Arc::new(RuntimeHandle::new(runtime_config)?);

        let mut nifs = NifRegistration::new();
        nifs.add_engine_nifs().add_host_nifs(self.host_nifs);
        runtime.install_nifs(nifs)?;

        // Persisted runtime deploys must be resident before startup recovery
        // below resolves any run's recorded pinned version — this is the
        // restart half of the deploy durability promise.
        let catalog =
            assemble_startup_catalog(runtime.as_ref(), store.as_ref(), self.workflow_sources)
                .await?;

        let registry = self
            .active_registry
            .unwrap_or_else(|| Arc::new(Registry::default()));
        let nif_state = Arc::clone(runtime.nif_state());
        let query_mailbox_engine = install_engine_nif_seams(
            &nif_state,
            &registry,
            &store,
            &runtime,
            self.activity_dispatcher,
            self.query_timeout,
        );
        let supervision = Arc::new(SupervisionTree::new());
        let search_attribute_schema = Arc::new(self.search_attribute_schema);
        let signal_handoff = Arc::new(SignalResumeHandoff::new());

        let delegated = assemble_delegated_seams(SeamAssembly {
            configured: self.delegated,
            signal_router_factory: self.signal_router_factory,
            runtime: Arc::clone(&runtime),
            signal_handoff: Arc::clone(&signal_handoff),
            streaming_publisher,
            query_mailbox_engine,
            query_timeout: self.query_timeout,
            query_service_overridden: self.seam_overrides.query_service,
        });

        install_signal_nif_bridge(
            &nif_state,
            Arc::new(crate::runtime::SignalNifBridge::new(
                Arc::clone(&registry),
                Arc::clone(&runtime),
                tokio::runtime::Handle::current(),
                delegated.signal_router_arc(),
            )),
        );
        install_configured_child_nif_bridge(&ChildBridgeAssembly {
            nif_state: &nif_state,
            store: &store,
            visibility_store: &visibility_store,
            runtime: &runtime,
            catalog: &catalog,
            registry: &registry,
            supervision: &supervision,
            signal_handoff: &signal_handoff,
            search_attribute_schema: &search_attribute_schema,
            watch_backoff: self.signal_delivery,
        })?;

        // Startup recovery re-spawns active workflow processes, and those
        // processes begin replaying on scheduler threads immediately. Replay
        // re-executes workflow code through the engine NIFs, so every NIF
        // bridge (signal, child) must be installed before the first recovered
        // process can run, or an early replayed spawn_child/receive_signal
        // call fails with a missing-bridge error.
        recover_active_workflows_on_startup(StartupRecoveryContext {
            store: Arc::clone(&store),
            visibility_store: Arc::clone(&visibility_store),
            runtime: Arc::clone(&runtime),
            catalog: Arc::clone(&catalog),
            registry: Arc::clone(&registry),
            supervision: Arc::clone(&supervision),
            recovery: self.recovery,
            search_attribute_schema: Arc::clone(&search_attribute_schema),
            bootstrap_schedule_coordinator: self.bootstrap_schedule_coordinator,
        })
        .await?;
        recover_timers_on_startup(&nif_state, Arc::clone(&store)).await?;

        let visibility_reconciliation_task = maybe_spawn_visibility_reconciliation(
            self.visibility_reconciliation_interval,
            &store,
            &visibility_store,
        );

        let engine = Engine::new(EngineComponents {
            store,
            visibility_store,
            runtime,
            catalog,
            registry,
            supervision,
            delegated,
            signal_handoff,
            search_attribute_schema,
            visibility_reconciliation_task,
        });
        engine.catchup_schedule_coordinator().await?;
        engine.recover_schedules_on_startup(Utc::now()).await?;
        Ok(engine)
    }
}

/// Borrowed engine components assembled into the child NIF bridge.
struct ChildBridgeAssembly<'a> {
    nif_state: &'a Arc<crate::runtime::EngineNifState>,
    store: &'a Arc<dyn EventStore>,
    visibility_store: &'a Arc<dyn VisibilityStore>,
    runtime: &'a Arc<RuntimeHandle>,
    catalog: &'a Arc<WorkflowCatalog>,
    registry: &'a Arc<Registry>,
    supervision: &'a Arc<SupervisionTree>,
    signal_handoff: &'a Arc<SignalResumeHandoff>,
    search_attribute_schema: &'a Arc<aion_core::SearchAttributeSchema>,
    /// The child-terminal watcher reuses the builder's delivery retry
    /// policy for its registry-miss backoff windows.
    watch_backoff: SignalDeliveryConfig,
}

/// Register the `WorkflowDeadlineHandler` on the timer bridge.
///
/// The handler holds the runtime weakly so the runtime → nif-state → bridge →
/// handler chain never cycles back into the runtime (the documented
/// cycle-avoidance the timer bridge observes with its `Weak<EngineNifState>`).
///
/// # Errors
///
/// Returns [`EngineError::Runtime`] when no timer bridge is installed.
fn register_workflow_deadline_handler(
    nif_state: &crate::runtime::EngineNifState,
    runtime: &Arc<RuntimeHandle>,
    store: &Arc<dyn EventStore>,
    visibility_store: &Arc<dyn VisibilityStore>,
    registry: &Arc<Registry>,
) -> Result<(), EngineError> {
    crate::runtime::nif_timer_bridge::register_deadline_handler(
        nif_state,
        Arc::new(crate::lifecycle::deadline::WorkflowDeadlineHandler::new(
            Arc::downgrade(runtime),
            Arc::clone(store),
            Arc::clone(visibility_store),
            Arc::clone(registry),
        )),
    )
    .map_err(|error| EngineError::Runtime {
        reason: format!("failed to register workflow deadline handler: {error}"),
    })
}

fn install_configured_child_nif_bridge(
    assembly: &ChildBridgeAssembly<'_>,
) -> Result<(), EngineError> {
    install_child_nif_bridge(
        assembly.nif_state,
        Arc::new(ChildNifBridge::new(ChildNifBridgeParts {
            store: Arc::clone(assembly.store),
            visibility_store: Arc::clone(assembly.visibility_store),
            runtime: Arc::clone(assembly.runtime),
            catalog: Arc::clone(assembly.catalog),
            registry: Arc::clone(assembly.registry),
            supervision: Arc::clone(assembly.supervision),
            signal_handoff: Arc::clone(assembly.signal_handoff),
            search_attribute_schema: Arc::clone(assembly.search_attribute_schema),
            tokio_handle: tokio::runtime::Handle::current(),
            watch_backoff: assembly.watch_backoff,
        })?),
    );
    // Register the workflow-deadline handler here too: it needs the same
    // teardown deps this assembly carries, and this runs before startup timer
    // recovery, so an already-due `deadline:{run_id}` swept at boot routes to
    // the engine rather than failing as an unhandled reserved fire.
    register_workflow_deadline_handler(
        assembly.nif_state,
        assembly.runtime,
        assembly.store,
        assembly.visibility_store,
        assembly.registry,
    )
}

pub(super) fn package_from_source(source: WorkflowPackageSource) -> Result<Package, EngineError> {
    match source {
        WorkflowPackageSource::Path(path) => {
            // Operator-local startup packages from config/CLI are trusted
            // input; only the network deploy path extracts bounded.
            Package::load_from_path(&path, ExtractionLimits::unbounded()).map_err(|error| {
                EngineError::Load {
                    reason: format!(
                        "failed to load workflow package `{}`: {error}",
                        path.display()
                    ),
                }
            })
        }
        WorkflowPackageSource::Package(package) => Ok(*package),
    }
}

#[cfg(test)]
mod tests {
    use std::{num::NonZeroUsize, path::PathBuf, process::Command, sync::Arc, time::Duration};

    use aion_core::{Event, EventEnvelope, Payload, WorkflowId, WorkflowStatus};
    use aion_package::{
        BeamModule, BeamSet, CURRENT_FORMAT_VERSION, DeclaredActivity, ExtractionLimits, Manifest,
        ManifestVersion, Package, PackageBuilder,
    };
    use aion_store::visibility::{ListWorkflowsFilter, VisibilityStore};
    use aion_store::{InMemoryStore, ReadableEventStore, WritableEventStore, WriteToken};
    use chrono::Utc;
    use futures::StreamExt;
    use serde_json::json;

    use crate::engine::api_schedule::{
        schedule_coordinator_run_id, schedule_coordinator_workflow_id,
        schedule_coordinator_workflow_type,
    };
    use crate::runtime::{Mfa, NifEntry};

    use super::EngineBuilder;
    use crate::EngineError;

    fn payload() -> Result<Payload, aion_core::PayloadError> {
        Payload::from_json(&json!({ "input": true }))
    }

    fn started(
        workflow_id: &WorkflowId,
        workflow_type: &str,
    ) -> Result<Event, aion_core::PayloadError> {
        Ok(Event::WorkflowStarted {
            envelope: EventEnvelope {
                seq: 1,
                recorded_at: Utc::now(),
                workflow_id: workflow_id.clone(),
            },
            workflow_type: workflow_type.to_owned(),
            input: payload()?,
            run_id: aion_core::RunId::new(uuid::Uuid::from_u128(1)),
            parent_run_id: None,
            package_version: aion_core::PackageVersion::new("a".repeat(64)),
        })
    }

    fn completed(workflow_id: &WorkflowId) -> Result<Event, aion_core::PayloadError> {
        Ok(Event::WorkflowCompleted {
            envelope: EventEnvelope {
                seq: 2,
                recorded_at: Utc::now(),
                workflow_id: workflow_id.clone(),
            },
            result: payload()?,
        })
    }

    fn package_manifest() -> Manifest {
        Manifest {
            entry_module: "counter".to_owned(),
            entry_function: "version".to_owned(),
            input_schema: json!({ "type": "object" }),
            output_schema: json!({ "type": "integer" }),
            timeout: Some(Duration::from_secs(30)),
            activities: vec![DeclaredActivity {
                activity_type: "activity/test".to_owned(),
            }],
            version: ManifestVersion::new("test"),
            format_version: CURRENT_FORMAT_VERSION,
            additional_workflows: Vec::new(),
        }
    }

    fn compile_counter_beam() -> Result<Vec<u8>, Box<dyn std::error::Error>> {
        let temp_dir =
            std::env::temp_dir().join(format!("aion-engine-builder-{}", uuid::Uuid::new_v4()));
        std::fs::create_dir(&temp_dir)?;
        let source_path = temp_dir.join("counter.erl");
        let beam_path = temp_dir.join("counter.beam");
        std::fs::write(
            &source_path,
            "-module(counter).\n-export([version/0]).\nversion() -> 1.\n",
        )?;
        let status = Command::new("erlc")
            .arg("-o")
            .arg(&temp_dir)
            .arg(&source_path)
            .status()?;
        if !status.success() {
            let cleanup_result = std::fs::remove_dir_all(&temp_dir);
            drop(cleanup_result);
            return Err(format!("erlc failed with status {status}").into());
        }
        let bytes = std::fs::read(beam_path)?;
        std::fs::remove_dir_all(temp_dir)?;
        Ok(bytes)
    }

    fn fixture_package() -> Result<Package, Box<dyn std::error::Error>> {
        let beams = BeamSet::new(vec![BeamModule::new("counter", compile_counter_beam()?)])?;
        let archive = PackageBuilder::new(package_manifest(), beams).write_to_bytes()?;
        Ok(Package::load_from_bytes(
            archive,
            ExtractionLimits::unbounded(),
        )?)
    }

    fn write_fixture_package(package: &Package) -> Result<PathBuf, Box<dyn std::error::Error>> {
        let path =
            std::env::temp_dir().join(format!("aion-engine-builder-{}.aion", uuid::Uuid::new_v4()));
        PackageBuilder::new(package.manifest().clone(), package.beams().clone())
            .write_to_path(&path)?;
        Ok(path)
    }

    #[tokio::test]
    async fn build_without_store_returns_missing_store() {
        let error = EngineBuilder::new().build().await.err();

        assert!(matches!(error, Some(EngineError::MissingStore)));
    }

    #[tokio::test]
    async fn build_without_visibility_store_returns_missing_visibility_store() {
        let error = EngineBuilder::new()
            .store(InMemoryStore::default())
            .build()
            .await
            .err();

        assert!(matches!(error, Some(EngineError::MissingVisibilityStore)));
    }

    #[tokio::test]
    async fn in_memory_visibility_allows_build_without_visibility_store() -> Result<(), EngineError>
    {
        let engine = EngineBuilder::new()
            .store(InMemoryStore::default())
            .in_memory_visibility()
            .build()
            .await?;

        engine.shutdown()?;
        Ok(())
    }

    fn capacity(value: usize) -> Result<NonZeroUsize, Box<dyn std::error::Error>> {
        NonZeroUsize::new(value).ok_or_else(|| "capacity must be non-zero".into())
    }

    #[tokio::test]
    async fn event_streaming_delivers_recorder_appends_through_engine_subscribe()
    -> Result<(), Box<dyn std::error::Error>> {
        let engine = EngineBuilder::new()
            .store(InMemoryStore::default())
            .in_memory_visibility()
            .event_streaming(capacity(8)?)
            .build()
            .await?;
        let workflow_id = WorkflowId::new_v4();
        let mut subscription = engine.subscribe(crate::EventFilter {
            workflow_id: Some(workflow_id.clone()),
            run: None,
            family: None,
        });

        // The production append path: a Recorder over the engine's store,
        // which `event_streaming` wrapped before any recorder existed.
        let mut recorder = crate::durability::Recorder::new(workflow_id.clone(), engine.store());
        recorder
            .record_workflow_started(
                Utc::now(),
                crate::durability::WorkflowStartRecord {
                    workflow_type: "checkout".to_owned(),
                    input: payload()?,
                    run_id: aion_core::RunId::new(uuid::Uuid::from_u128(7)),
                    parent_run_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;

        let item = tokio::time::timeout(Duration::from_secs(2), subscription.next())
            .await?
            .ok_or("subscription ended without delivering the appended event")?;
        let event = item?;
        assert_eq!(event.workflow_id(), &workflow_id);
        assert_eq!(event.seq(), 1);
        assert!(matches!(event, Event::WorkflowStarted { .. }));
        engine.shutdown()?;
        Ok(())
    }

    #[tokio::test]
    async fn without_event_streaming_subscriptions_stay_on_deferred_empty_stream()
    -> Result<(), Box<dyn std::error::Error>> {
        let engine = EngineBuilder::new()
            .store(InMemoryStore::default())
            .in_memory_visibility()
            .build()
            .await?;

        let mut subscription = engine.subscribe(crate::EventFilter::default());
        let item = tokio::time::timeout(Duration::from_secs(2), subscription.next()).await?;

        assert!(item.is_none(), "deferred publisher streams must be empty");
        engine.shutdown()?;
        Ok(())
    }

    #[tokio::test]
    async fn event_streaming_conflicts_with_explicit_event_publisher()
    -> Result<(), Box<dyn std::error::Error>> {
        let error = EngineBuilder::new()
            .store(InMemoryStore::default())
            .in_memory_visibility()
            .event_publisher(Arc::new(crate::DeferredEventPublisher))
            .event_streaming(capacity(8)?)
            .build()
            .await
            .err();

        assert!(matches!(
            error,
            Some(EngineError::ConflictingEventPublisher)
        ));
        Ok(())
    }

    #[test]
    fn query_timeout_is_only_set_by_caller() {
        assert_eq!(EngineBuilder::new().configured_query_timeout(), None);
        assert_eq!(
            EngineBuilder::new()
                .query_timeout(Duration::from_secs(3))
                .configured_query_timeout(),
            Some(Duration::from_secs(3))
        );
    }

    async fn insert_running_workflow(
        engine: &crate::Engine,
    ) -> Result<(WorkflowId, aion_core::RunId), Box<dyn std::error::Error>> {
        let workflow_id = WorkflowId::new_v4();
        let run_id = aion_core::RunId::new_v4();
        let mut recorder = crate::durability::Recorder::new(workflow_id.clone(), engine.store());
        recorder
            .record_workflow_started(
                Utc::now(),
                crate::durability::WorkflowStartRecord {
                    workflow_type: "checkout".to_owned(),
                    input: payload()?,
                    run_id: run_id.clone(),
                    parent_run_id: None,
                    package_version: aion_core::PackageVersion::new("a".repeat(64)),
                },
            )
            .await?;
        let handle = crate::registry::WorkflowHandle::new(crate::registry::WorkflowHandleParts {
            workflow_id: workflow_id.clone(),
            run_id: run_id.clone(),
            pid: engine.runtime().spawn_test_process_with_trap_exit(true)?,
            workflow_type: "checkout".to_owned(),
            namespace: String::from("default"),
            loaded_version: aion_package::ContentHash::from_bytes([2; 32]),
            cached_status: WorkflowStatus::Running,
            residency: crate::registry::HandleResidency::Resident,
            recorder,
            completion: crate::registry::CompletionNotifier::new(),
        });
        engine
            .registry()
            .insert((workflow_id.clone(), run_id.clone()), handle)?;
        Ok((workflow_id, run_id))
    }

    #[tokio::test]
    async fn query_timeout_installs_the_concrete_query_seam()
    -> Result<(), Box<dyn std::error::Error>> {
        let engine = EngineBuilder::new()
            .store(InMemoryStore::default())
            .in_memory_visibility()
            .query_timeout(Duration::from_millis(250))
            .build()
            .await?;
        let (workflow_id, run_id) = insert_running_workflow(&engine).await?;

        // The concrete seam reached the query mailbox engine, which answers
        // an unregistered name with a typed UnknownQuery — the deferred seam
        // would have failed with its "not configured" runtime error instead.
        let result = engine.query(&workflow_id, &run_id, "state").await;

        assert!(matches!(
            result,
            Err(crate::EngineError::Query(crate::QueryError::UnknownQuery(name))) if name == "state"
        ));
        engine.shutdown()?;
        Ok(())
    }

    #[tokio::test]
    async fn without_query_timeout_the_query_seam_stays_deferred()
    -> Result<(), Box<dyn std::error::Error>> {
        let engine = EngineBuilder::new()
            .store(InMemoryStore::default())
            .in_memory_visibility()
            .build()
            .await?;
        let (workflow_id, run_id) = insert_running_workflow(&engine).await?;

        let result = engine.query(&workflow_id, &run_id, "state").await;

        assert!(matches!(
            result,
            Err(crate::EngineError::Runtime { reason }) if reason.contains("not configured")
        ));
        engine.shutdown()?;
        Ok(())
    }

    #[test]
    fn owned_shards_are_only_set_by_caller() {
        // The default builder configures NO shard restriction: `build()` never
        // touches the store's scoping hook, so single-node boot owns ALL shards
        // and is byte-identical to today.
        assert_eq!(EngineBuilder::new().configured_owned_shards(), None);
        assert_eq!(
            EngineBuilder::new()
                .owned_shards([2, 0, 2, 1])
                .configured_owned_shards(),
            Some([2, 0, 2, 1].as_slice())
        );
    }

    #[test]
    fn scheduler_threads_are_only_set_by_caller() {
        assert_eq!(EngineBuilder::new().scheduler_thread_count(), None);
        assert_eq!(
            EngineBuilder::new()
                .scheduler_threads(4)
                .scheduler_thread_count(),
            Some(4)
        );
    }

    #[test]
    fn visibility_reconciliation_interval_is_only_set_by_caller() {
        let interval = Duration::from_millis(250);

        assert_eq!(
            EngineBuilder::new().configured_visibility_reconciliation_interval(),
            None
        );
        assert_eq!(
            EngineBuilder::new()
                .visibility_reconciliation_interval(interval)
                .configured_visibility_reconciliation_interval(),
            Some(interval)
        );
    }

    #[tokio::test]
    async fn duplicate_host_nif_mfa_returns_typed_error() {
        let mfa = Mfa::new("host", "zero", 0);
        let error = EngineBuilder::new()
            .store(InMemoryStore::default())
            .in_memory_visibility()
            .register_nifs([
                NifEntry::new(mfa.clone(), crate::runtime::nif::test_native_zero),
                NifEntry::dirty(mfa, crate::runtime::nif::test_native_zero),
            ])
            .build()
            .await
            .err();

        assert!(matches!(
            error,
            Some(EngineError::NifRegistration { reason }) if reason.contains("host:zero/0")
        ));
    }

    #[tokio::test]
    async fn empty_store_builds_coordinator_history_without_registry_or_supervision()
    -> Result<(), EngineError> {
        let store = Arc::new(InMemoryStore::default());
        let engine = EngineBuilder::new()
            .store_arc(store.clone())
            .in_memory_visibility()
            .build()
            .await?;

        assert!(engine.registry().list()?.is_empty());
        assert_eq!(engine.supervision().type_supervisor_count()?, 1);
        assert_eq!(engine.workflow_catalog().workflows()?.len(), 0);

        let coordinator_id = schedule_coordinator_workflow_id();
        let active = store.list_active().await?;
        assert_eq!(active, vec![coordinator_id.clone()]);
        let history = store.read_history(&coordinator_id).await?;
        let [started] = history.as_slice() else {
            return Err(EngineError::Load {
                reason: format!(
                    "expected exactly one coordinator event, found {}",
                    history.len()
                ),
            });
        };
        match started {
            Event::WorkflowStarted {
                workflow_type,
                input,
                run_id,
                parent_run_id,
                ..
            } => {
                assert_eq!(workflow_type, schedule_coordinator_workflow_type());
                assert_eq!(
                    input,
                    &Payload::from_json(&json!({})).map_err(|error| {
                        EngineError::Load {
                            reason: format!("failed to build expected payload: {error}"),
                        }
                    })?
                );
                assert_eq!(run_id, &schedule_coordinator_run_id());
                assert!(parent_run_id.is_none());
            }
            other => {
                return Err(EngineError::Load {
                    reason: format!("expected coordinator WorkflowStarted, found {other:?}"),
                });
            }
        }

        engine.shutdown()?;
        let rebuilt = EngineBuilder::new()
            .store_arc(store.clone())
            .in_memory_visibility()
            .build()
            .await?;
        let rebuilt_history = store.read_history(&coordinator_id).await?;
        assert_eq!(rebuilt_history.len(), 1);
        rebuilt.shutdown()?;

        Ok(())
    }

    #[tokio::test]
    async fn build_loads_already_loaded_package() -> Result<(), Box<dyn std::error::Error>> {
        let package = fixture_package()?;
        let version = package.content_hash().clone();
        let deployed_entry_module = package.deployed_entry_module();

        let engine = EngineBuilder::new()
            .store(InMemoryStore::default())
            .in_memory_visibility()
            .load_workflows(package)
            .build()
            .await?;

        let loaded = engine
            .workflow_catalog()
            .get("counter", &version)?
            .ok_or("loaded package record missing")?;
        assert_eq!(loaded.deployed_entry_module(), deployed_entry_module);
        assert!(
            engine
                .runtime()
                .has_registered_module(&deployed_entry_module)
        );
        Ok(())
    }

    #[tokio::test]
    async fn startup_reconciliation_backfills_completed_visibility()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = Arc::new(InMemoryStore::default());
        let completed_id = WorkflowId::new_v4();

        store
            .append(
                WriteToken::recorder(),
                &completed_id,
                &[
                    started(&completed_id, "billing")?,
                    completed(&completed_id)?,
                ],
                0,
            )
            .await?;

        let engine = EngineBuilder::new()
            .store_arc(store.clone())
            .visibility_store_arc(store.clone())
            .build()
            .await?;

        let summaries = store.list_workflows(ListWorkflowsFilter::default()).await?;
        let completed_summary = summaries
            .iter()
            .find(|summary| summary.workflow_id == completed_id)
            .ok_or("completed workflow missing from visibility")?;

        assert_eq!(completed_summary.status, WorkflowStatus::Completed);
        assert!(completed_summary.close_time.is_some());
        engine.shutdown()?;
        Ok(())
    }

    #[tokio::test]
    async fn periodic_visibility_reconciliation_repairs_gap_after_startup()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = Arc::new(InMemoryStore::default());
        let engine = EngineBuilder::new()
            .store_arc(store.clone())
            .visibility_store_arc(store.clone())
            .visibility_reconciliation_interval(Duration::from_millis(25))
            .build()
            .await?;
        let workflow_id = WorkflowId::new_v4();

        store
            .append(
                WriteToken::recorder(),
                &workflow_id,
                &[started(&workflow_id, "checkout")?],
                0,
            )
            .await?;

        tokio::time::timeout(Duration::from_secs(2), async {
            loop {
                let summaries = store.list_workflows(ListWorkflowsFilter::default()).await?;
                if summaries.iter().any(|summary| {
                    summary.workflow_id == workflow_id && summary.status == WorkflowStatus::Running
                }) {
                    return Ok::<(), aion_store::StoreError>(());
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await??;

        engine.shutdown()?;
        Ok(())
    }

    #[tokio::test]
    async fn build_loads_package_from_path() -> Result<(), Box<dyn std::error::Error>> {
        let package = fixture_package()?;
        let version = package.content_hash().clone();
        let path = write_fixture_package(&package)?;

        let engine = EngineBuilder::new()
            .store(InMemoryStore::default())
            .in_memory_visibility()
            .load_workflows(path.as_path())
            .build()
            .await?;
        std::fs::remove_file(path)?;

        assert!(
            engine
                .workflow_catalog()
                .get("counter", &version)?
                .is_some()
        );
        Ok(())
    }
}