cf-gears-toolkit 0.10.0

//! Host Runtime - orchestrates the full `ToolKit` lifecycle
//!
//! This gear contains the `HostRuntime` type that owns and coordinates
//! the execution of all lifecycle phases.
//!
//! High-level phase order:
//! - `pre_init` (system gears only)
//! - DB migrations (gears with DB capability)
//! - `init` (all gears)
//! - proxy-wiring (`#[toolkit::consumes]` clients; feature-gated)
//! - `post_init` (system gears only; runs after *all* `init` complete)
//! - REST wiring (gears with REST capability; requires a single REST host)
//! - gRPC registration (gears with gRPC capability; requires a single gRPC hub)
//! - start/stop (stateful gears)
//! - `OoP` spawn / wait / stop (host-only orchestration)
//!
//! Both lifecycle paths — in-process (`run_phases_internal`) and `OoP`
//! (`run_oop_serving`) — run these in the same relative order. Proxy-wiring in
//! particular must stay before `start`: a gear resolving a consumed contract
//! during its own `start` has to find the client in the `ClientHub` regardless
//! of profile. The `OoP` path additionally has no REST-wiring, directory-register
//! or spawn phases (`oop_serve` owns those concerns).

use axum::Router;
use std::collections::HashSet;
use std::sync::Arc;

use tokio_util::sync::CancellationToken;
use uuid::Uuid;

use crate::backends::OopSpawnConfig;
use crate::client_hub::ClientHub;
use crate::config::ConfigProvider;
use crate::context::GearContextBuilder;
use crate::registry::{
    ApiGatewayCap, GearEntry, GearRegistry, GrpcHubCap, RegistryError, RestApiCap, RunnableCap,
    SystemCap,
};
use crate::runtime::{GearManager, GrpcInstallerStore, OopSpawnOptions, SystemContext};

#[cfg(feature = "db")]
use crate::registry::DatabaseCap;

/// How the runtime should provide DBs to gears.
#[derive(Clone)]
pub enum DbOptions {
    /// No database integration. `GearCtx::db()` will be `None`, `db_required()` will error.
    None,
    /// Use a `DbManager` to handle database connections with Figment-based configuration.
    #[cfg(feature = "db")]
    Manager(Arc<toolkit_db::DbManager>),
}

/// Runtime execution mode that determines which phases to run.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RunMode {
    /// Run all phases and wait for shutdown signal (normal application mode).
    Full,
    /// Run only pre-init and DB migration phases, then exit (for cloud deployments).
    MigrateOnly,
}

/// Environment variable name for passing directory endpoint to `OoP` gears.
pub const TOOLKIT_DIRECTORY_ENDPOINT_ENV: &str = "TOOLKIT_DIRECTORY_ENDPOINT";

/// Environment variable name for passing rendered gear config to `OoP` gears.
pub const TOOLKIT_MODULE_CONFIG_ENV: &str = "TOOLKIT_MODULE_CONFIG";

/// Default shutdown deadline for graceful gear stop (35 seconds).
///
/// This is intentionally 5 seconds longer than `WithLifecycle::stop_timeout` (30s default)
/// to ensure deterministic behavior: the lifecycle's internal timeout fires first,
/// and the runtime deadline acts as a hard backstop.
pub const DEFAULT_SHUTDOWN_DEADLINE: std::time::Duration = std::time::Duration::from_secs(35);

/// `HostRuntime` owns the lifecycle orchestration for `ToolKit`.
///
/// It encapsulates all runtime state and drives gears through the full lifecycle (see gear docs).
/// Read a consumer's ADR-0004 static-endpoint override for `dep_gear` from
/// `gears.<owner_gear>.config.consumer_wiring.<dep_gear>` (a base endpoint URI
/// string). This is the dev/test escape hatch that bypasses service discovery;
/// returns `None` when unset.
fn static_endpoint_override(
    cfg: &dyn ConfigProvider,
    owner_gear: &str,
    dep_gear: &str,
) -> Option<String> {
    cfg.get_gear_config(owner_gear)?
        .get("config")?
        .get("consumer_wiring")?
        .get(dep_gear)?
        .as_str()
        .map(str::to_owned)
}

pub struct HostRuntime {
    registry: GearRegistry,
    ctx_builder: GearContextBuilder,
    instance_id: Uuid,
    gear_manager: Arc<GearManager>,
    grpc_installers: Arc<GrpcInstallerStore>,
    client_hub: Arc<ClientHub>,
    /// Per-gear config, retained for the proxy-wiring phase to read a consumer's
    /// static-endpoint override (dev/test escape hatch, ADR-0004).
    gears_cfg: Arc<dyn ConfigProvider>,
    /// Process-level dependency-resolution + draining signal, published in
    /// `client_hub` for the `/readyz` probe and updated by the proxy-wiring
    /// readiness loop + draining watcher. In-process, an
    /// [`ReadinessHealthcheck`](super::readiness::ReadinessHealthcheck) leaf
    /// bridges it into the gateway's healthcheck registry.
    dep_checker: Arc<super::readiness::DependencyChecker>,
    /// Set once the in-process directory-register phase has advertised this
    /// process's REST providers, so shutdown deregisters them exactly once — and
    /// only in the in-process host path. In `OoP` serving, presence + deregister
    /// is owned by `oop_serve`, so this stays `false` and avoids a double
    /// deregister.
    rest_providers_registered: std::sync::atomic::AtomicBool,
    cancel: CancellationToken,
    #[allow(dead_code)]
    db_options: DbOptions,
    /// `OoP` gear spawn configuration and backend
    oop_options: Option<OopSpawnOptions>,
    /// Maximum time allowed for graceful shutdown before hard-stop signal is sent.
    shutdown_deadline: std::time::Duration,
}

impl HostRuntime {
    /// Create a new `HostRuntime` instance.
    ///
    /// This prepares all runtime components but does not start any lifecycle phases.
    pub fn new(
        registry: GearRegistry,
        gears_cfg: Arc<dyn ConfigProvider>,
        db_options: DbOptions,
        client_hub: Arc<ClientHub>,
        cancel: CancellationToken,
        instance_id: Uuid,
        oop_options: Option<OopSpawnOptions>,
    ) -> Self {
        // Create runtime-owned components for system gears
        let gear_manager = Arc::new(GearManager::new());
        let grpc_installers = Arc::new(GrpcInstallerStore::new());

        // Process-level dependency/draining signal, published so the gateway's
        // /readyz handler can fetch it (concrete-type key). Created before any
        // phase runs.
        let dep_checker = Arc::new(super::readiness::DependencyChecker::new());
        client_hub.register::<super::readiness::DependencyChecker>(dep_checker.clone());

        // Build the context builder that will resolve per-gear DbHandles
        let ctx_builder = GearContextBuilder::new(
            instance_id,
            gears_cfg.clone(),
            client_hub.clone(),
            cancel.clone(),
        );
        #[cfg(feature = "db")]
        let ctx_builder = match &db_options {
            DbOptions::Manager(mgr) => ctx_builder.with_db_manager(mgr.clone()),
            DbOptions::None => ctx_builder,
        };

        Self {
            registry,
            ctx_builder,
            instance_id,
            gear_manager,
            grpc_installers,
            client_hub,
            gears_cfg,
            dep_checker,
            rest_providers_registered: std::sync::atomic::AtomicBool::new(false),
            cancel,
            db_options,
            oop_options,
            shutdown_deadline: DEFAULT_SHUTDOWN_DEADLINE,
        }
    }

    /// Set a custom shutdown deadline for graceful gear stop.
    ///
    /// This is the maximum time the runtime will wait for each gear to stop gracefully
    /// before sending the hard-stop signal (cancelling the deadline token).
    ///
    /// # Relationship with `WithLifecycle::stop_timeout`
    ///
    /// When using `WithLifecycle`, its `stop_timeout` (default 30s) races against this
    /// `shutdown_deadline` (default [`DEFAULT_SHUTDOWN_DEADLINE`], 35s). To ensure
    /// deterministic behavior:
    ///
    /// - `WithLifecycle::stop_timeout` should be **less than** `shutdown_deadline`
    /// - This allows the lifecycle's internal timeout to trigger first for graceful cleanup
    /// - The runtime's `deadline_token` then acts as a hard backstop
    ///
    /// Example: `stop_timeout = 30s`, `shutdown_deadline = 35s`
    #[must_use]
    pub fn with_shutdown_deadline(mut self, deadline: std::time::Duration) -> Self {
        self.shutdown_deadline = deadline;
        self
    }

    /// Set the process-wide platform-plane credential source, applied to every
    /// [`GearCtx`](crate::context::GearCtx) this runtime builds so
    /// `#[toolkit::provides]`-generated clients attach `X-ToolKit-Internal-Token`
    /// on platform-plane methods. `None` (Profile 1 / in-process, or no
    /// credential configured) attaches nothing.
    #[must_use]
    pub fn with_internal_token_provider(
        mut self,
        provider: Option<toolkit_contract::runtime::config::InternalTokenProvider>,
    ) -> Self {
        self.ctx_builder = self.ctx_builder.with_internal_token_provider(provider);
        self
    }

    /// `PRE_INIT` phase: wire runtime internals into system gears.
    ///
    /// This phase runs before init and only for gears with the "system" capability.
    ///
    /// # Errors
    /// Returns `RegistryError` if system wiring fails.
    pub fn run_pre_init_phase(&self) -> Result<(), RegistryError> {
        tracing::info!("Phase: pre_init");

        let sys_ctx = SystemContext::new(
            self.instance_id,
            Arc::clone(&self.gear_manager),
            Arc::clone(&self.grpc_installers),
        );

        for entry in self.registry.gears() {
            // Check for cancellation before processing each gear
            if self.cancel.is_cancelled() {
                tracing::warn!("Pre-init phase cancelled by signal");
                return Err(RegistryError::Cancelled);
            }

            if let Some(sys_mod) = entry.caps.query::<SystemCap>() {
                tracing::debug!(gear = entry.name, "Running system pre_init");
                sys_mod
                    .pre_init(&sys_ctx)
                    .map_err(|e| RegistryError::PreInit {
                        gear: entry.name,
                        source: e,
                    })?;
            }
        }

        Ok(())
    }

    /// Helper: resolve context for a gear with error mapping.
    #[cfg(feature = "db")]
    async fn gear_context(
        &self,
        gear_name: &'static str,
    ) -> Result<crate::context::GearCtx, RegistryError> {
        self.ctx_builder
            .for_gear(gear_name)
            .await
            .map_err(|e| RegistryError::DbMigrate {
                gear: gear_name,
                source: e,
            })
    }

    /// Helper: extract DB handle and gear if both exist.
    #[cfg(feature = "db")]
    async fn db_migration_target(
        &self,
        gear_name: &'static str,
        ctx: &crate::context::GearCtx,
        db_gear: Option<Arc<dyn crate::contracts::DatabaseCapability>>,
    ) -> Result<
        Option<(
            toolkit_db::Db,
            Arc<dyn crate::contracts::DatabaseCapability>,
        )>,
        RegistryError,
    > {
        let Some(dbm) = db_gear else {
            return Ok(None);
        };

        // Important: DB migrations require access to the underlying `Db`, not just `DBProvider`.
        // `GearCtx` intentionally exposes only `DBProvider` for better DX and to reduce mistakes.
        // So the runtime resolves the `Db` directly from its `DbManager`.
        let db = match &self.db_options {
            DbOptions::None => None,
            #[cfg(feature = "db")]
            DbOptions::Manager(mgr) => {
                mgr.get(gear_name)
                    .await
                    .map_err(|e| RegistryError::DbMigrate {
                        gear: gear_name,
                        source: e.into(),
                    })?
            }
        };

        _ = ctx; // ctx is kept for parity/error context; DB is resolved from manager above.
        Ok(db.map(|db| (db, dbm)))
    }

    /// Helper: run migrations for a single gear using the new migration runner.
    ///
    /// This collects migrations from the gear and executes them via the
    /// runtime's privileged connection. Gears never see the raw connection.
    #[cfg(feature = "db")]
    async fn migrate_gear(
        gear_name: &'static str,
        db: &toolkit_db::Db,
        db_gear: Arc<dyn crate::contracts::DatabaseCapability>,
    ) -> Result<(), RegistryError> {
        // Collect migrations from the gear
        let migrations = db_gear.migrations();

        if migrations.is_empty() {
            tracing::debug!(gear = gear_name, "No migrations to run");
            return Ok(());
        }

        tracing::debug!(
            gear = gear_name,
            count = migrations.len(),
            "Running DB migrations"
        );

        // Execute migrations using the migration runner
        let result =
            toolkit_db::migration_runner::run_migrations_for_gear(db, gear_name, migrations)
                .await
                .map_err(|e| RegistryError::DbMigrate {
                    gear: gear_name,
                    source: anyhow::Error::new(e),
                })?;

        tracing::info!(
            gear = gear_name,
            applied = result.applied,
            skipped = result.skipped,
            "DB migrations completed"
        );

        Ok(())
    }

    /// DB MIGRATION phase: run migrations for all gears with DB capability.
    ///
    /// Runs before init, with system gears processed first.
    ///
    /// Gears provide migrations via `DatabaseCapability::migrations()`.
    /// The runtime executes them with a privileged connection that gears
    /// never receive directly. Each gear gets a separate migration history
    /// table, preventing cross-gear interference.
    #[cfg(feature = "db")]
    async fn run_db_phase(&self) -> Result<(), RegistryError> {
        tracing::info!("Phase: db (before init)");

        for entry in self.registry.gears_by_system_priority() {
            // Check for cancellation before processing each gear
            if self.cancel.is_cancelled() {
                tracing::warn!("DB migration phase cancelled by signal");
                return Err(RegistryError::Cancelled);
            }

            let ctx = self.gear_context(entry.name).await?;
            let db_gear = entry.caps.query::<DatabaseCap>();

            match self
                .db_migration_target(entry.name, &ctx, db_gear.clone())
                .await?
            {
                Some((db, dbm)) => {
                    Self::migrate_gear(entry.name, &db, dbm).await?;
                }
                None if db_gear.is_some() => {
                    tracing::debug!(
                        gear = entry.name,
                        "Gear has DbGear trait but no DB handle (no config)"
                    );
                }
                None => {}
            }
        }

        Ok(())
    }

    /// INIT phase: initialize all gears in topological order.
    ///
    /// System gears initialize first, followed by user gears.
    async fn run_init_phase(&self) -> Result<(), RegistryError> {
        tracing::info!("Phase: init");

        for entry in self.registry.gears_by_system_priority() {
            let ctx =
                self.ctx_builder
                    .for_gear(entry.name)
                    .await
                    .map_err(|e| RegistryError::Init {
                        gear: entry.name,
                        source: e,
                    })?;
            tracing::info!(gear = entry.name, "Initializing a gear...");
            entry
                .core
                .init(&ctx)
                .await
                .map_err(|e| RegistryError::Init {
                    gear: entry.name,
                    source: e,
                })?;
            tracing::info!(gear = entry.name, "Initialized a gear.");
        }

        Ok(())
    }

    /// `POST_INIT` phase: optional hook after ALL gears completed `init()`.
    ///
    /// Consumer proxy-wiring phase (eventual readiness).
    ///
    /// Runs after init (compile-time / local registrations) and before
    /// post-init. Replays each `ConsumerRegistration` emitted by
    /// `#[toolkit::consumes]`: if a compile-time impl is already in the
    /// `ClientHub` it wins (the wiring closure short-circuits); otherwise a
    /// directory-resolving REST client is registered under the contract trait.
    ///
    /// Non-blocking: endpoint discovery is lazy/per-call inside the resolving
    /// client, so this phase never waits on provider availability (ADR-0007).
    /// A no-op when no consumer is registered, preserving the phase-order
    /// invariants relied on by existing tests.
    #[allow(
        clippy::unused_async,
        reason = "kept async for symmetry with the other `run_*_phase` steps awaited in sequence by `run_gear_phases`; the awaited work runs in a spawned readiness-probe task"
    )]
    async fn run_proxy_wiring_phase(&self) -> Result<(), RegistryError> {
        use crate::discovery::{
            ConsumerRegistration, DirectoryEndpointResolver, NullEndpointResolver,
        };
        use toolkit_contract::runtime::resolving::EndpointResolver;

        let regs: Vec<&ConsumerRegistration> = inventory::iter::<ConsumerRegistration>
            .into_iter()
            .collect();
        if regs.is_empty() {
            return Ok(());
        }
        tracing::info!(
            count = regs.len(),
            "Phase: proxy-wiring (consumer discovery)"
        );

        // Without a DirectoryClient we cannot resolve *remote* providers, but we
        // must NOT silently skip wiring: co-located consumers still short-circuit
        // to their local impl, and remote consumers must register as unresolved
        // readiness gates so `/readyz` stays 503 (a misconfigured consumer must
        // not report Ready). A null resolver makes every remote lookup `Ok(None)`
        // so the loop below classifies deps correctly without a directory.
        let (resolver, have_directory): (Arc<dyn EndpointResolver>, bool) =
            if let Ok(dir) = self.client_hub.get::<dyn crate::DirectoryClient>() {
                (Arc::new(DirectoryEndpointResolver::new(dir)), true)
            } else {
                tracing::error!(
                    consumers = regs.len(),
                    "proxy-wiring: no DirectoryClient in ClientHub; remote consumer \
                     dependencies cannot be resolved and will gate /readyz (503). \
                     Co-located (local) dependencies are unaffected."
                );
                (Arc::new(NullEndpointResolver), false)
            };

        // Wire each consumer contract. The outcome distinguishes a co-located
        // local impl (hub short-circuit) from a directory-resolving REST client.
        // Every dep is registered as a readiness gate; local ones are marked
        // resolved immediately, and only remote ones gate readiness + get the
        // background directory-resolve loop (ADR-0007: startup-gating + sticky).
        // `owner_gear` is derived by `#[toolkit::consumes]` from the struct
        // ident, while `#[toolkit::gear(name = ...)]` sets the registry name
        // independently. When they diverge, wiring still works (the loop below
        // does not filter on owner) but the static-override config key silently
        // resolves to nothing. Say so rather than leaving the operator to wonder
        // why `consumer_wiring` is ignored.
        let known_gears: std::collections::HashSet<&str> =
            self.registry.gears().iter().map(GearEntry::name).collect();
        for reg in &regs {
            if !known_gears.contains(reg.owner_gear) {
                tracing::warn!(
                    owner = reg.owner_gear,
                    dep = reg.dep_gear,
                    "proxy-wiring: consumer's owner gear name does not match any registered gear; \
                     the `gears.{}.config.consumer_wiring.{}` static override will never resolve. \
                     Rename the gear to the kebab-case of its struct ident.",
                    reg.owner_gear,
                    reg.dep_gear,
                );
            }
        }

        let mut remote_deps: Vec<String> = Vec::new();
        for reg in &regs {
            // ADR-0004 static-endpoint override (dev/test escape hatch): if the
            // consumer's config declares a fixed endpoint for this dep, wire it
            // directly (bypassing discovery) via a `StaticEndpointResolver`. A
            // fixed endpoint needs no probe loop, so it is readiness-resolved
            // immediately. Emitted at `warn!` — it must not be used in production.
            let static_override =
                static_endpoint_override(self.gears_cfg.as_ref(), reg.owner_gear, reg.dep_gear);
            let (reg_resolver, is_static): (Arc<dyn EndpointResolver>, bool) =
                if let Some(endpoint) = &static_override {
                    tracing::warn!(
                        owner = reg.owner_gear,
                        dep = reg.dep_gear,
                        endpoint = %endpoint,
                        "proxy-wiring: STATIC endpoint override in use (ADR-0004 dev/test \
                         escape hatch) - bypasses service discovery; MUST NOT be used in \
                         production"
                    );
                    (
                        Arc::new(crate::discovery::StaticEndpointResolver::new(
                            endpoint.clone(),
                        )),
                        true,
                    )
                } else {
                    (Arc::clone(&resolver), false)
                };

            // Thread the process's platform-plane credential onto the wired
            // (directory-resolving) client so its platform-plane methods attach
            // `X-ToolKit-Internal-Token` (`cpt-cf-adr-two-plane-auth`). This is
            // the genuine remote inter-gear path (Profile 2/3); a co-located
            // local impl short-circuits before the credential is used.
            let outcome = (reg.wire)(
                &self.client_hub,
                reg_resolver,
                self.ctx_builder.internal_token_provider(),
            )
            .map_err(|source| RegistryError::ProxyWiring {
                gear: reg.owner_gear,
                source,
            })?;
            self.dep_checker.register_dep(reg.dep_gear.to_owned());
            match outcome {
                // Local impl won the hub short-circuit — resolved.
                crate::discovery::WireOutcome::Local => {
                    self.dep_checker.mark_resolved(reg.dep_gear);
                }
                // Static override → fixed endpoint, always resolvable — no probe.
                crate::discovery::WireOutcome::Remote if is_static => {
                    self.dep_checker.mark_resolved(reg.dep_gear);
                }
                // Directory-resolved remote → gate readiness + background probe.
                crate::discovery::WireOutcome::Remote => remote_deps.push(reg.dep_gear.to_owned()),
            }
            tracing::debug!(
                owner = reg.owner_gear,
                dep = reg.dep_gear,
                outcome = ?outcome,
                static_override = is_static,
                "wired consumer contract"
            );
        }

        // Only remote deps need directory resolution; local wins are already
        // resolved above. Without a directory the null resolver can never resolve
        // them, so skip the probe entirely and leave those deps gating /readyz.
        if !have_directory || remote_deps.is_empty() {
            return Ok(());
        }

        let readiness = Arc::clone(&self.dep_checker);
        let cancel = self.cancel.clone();
        tokio::spawn(async move {
            const BASE: std::time::Duration = std::time::Duration::from_millis(100);
            const MAX: std::time::Duration = std::time::Duration::from_secs(30);
            let mut pending = remote_deps;
            let mut backoff = BASE;
            while !pending.is_empty() {
                let mut still_pending = Vec::new();
                for dep in pending {
                    match resolver.resolve_endpoint(&dep).await {
                        Ok(Some(_)) => {
                            readiness.mark_resolved(&dep);
                            tracing::info!(dep = %dep, "readiness: dependency resolved");
                        }
                        // No live instance yet — expected during startup churn.
                        Ok(None) => still_pending.push(dep),
                        // Genuine directory-backend failure: surface it (a stuck
                        // Starting / 503 pod otherwise has no diagnostic trail).
                        Err(e) => {
                            tracing::warn!(dep = %dep, error = %e, "readiness: directory lookup failed");
                            still_pending.push(dep);
                        }
                    }
                }
                pending = still_pending;
                if pending.is_empty() {
                    break;
                }
                tokio::select! {
                    () = cancel.cancelled() => break,
                    () = tokio::time::sleep(backoff) => {}
                }
                backoff = (backoff * 2).min(MAX);
            }
        });

        Ok(())
    }

    /// This provides a global barrier between initialization-time registration
    /// and subsequent phases that may rely on a fully-populated runtime registry.
    /// `init` -> proxy-wiring -> `post_init`, the segment both lifecycle paths
    /// share.
    ///
    /// Extracted so the two paths cannot disagree on it. They did once: the
    /// `OoP` path ran proxy-wiring *after* `start`, because the call was dropped
    /// where a retired untyped `resolve_deps` stopgap used to sit rather than
    /// chosen deliberately. A gear resolving a consumed contract during its own
    /// `start` then found nothing in the `ClientHub` under Profile 2/3 while
    /// working fine in Profile 1.
    ///
    /// Everything after this segment legitimately differs — the in-process path
    /// composes a REST router, the `OoP` path leaves that to `oop_serve`.
    ///
    /// Keeping this a single private method is what enforces the invariant: if
    /// a future change inlines the three calls back into both paths, this method
    /// becomes dead code and the workspace's `-D warnings` build fails.
    async fn run_init_wiring_post_init(&self) -> Result<(), RegistryError> {
        self.run_init_phase().await?;

        self.run_proxy_wiring_phase().await?;

        self.run_post_init_phase().await
    }

    ///
    /// System gears run first, followed by user gears, preserving topo order.
    async fn run_post_init_phase(&self) -> Result<(), RegistryError> {
        tracing::info!("Phase: post_init");

        let sys_ctx = SystemContext::new(
            self.instance_id,
            Arc::clone(&self.gear_manager),
            Arc::clone(&self.grpc_installers),
        );

        for entry in self.registry.gears_by_system_priority() {
            if let Some(sys_mod) = entry.caps.query::<SystemCap>() {
                sys_mod
                    .post_init(&sys_ctx)
                    .await
                    .map_err(|e| RegistryError::PostInit {
                        gear: entry.name,
                        source: e,
                    })?;
            }
        }

        Ok(())
    }

    /// REST phase: compose the router against the REST host.
    ///
    /// This is a synchronous phase that builds the final Router by:
    /// 1. Preparing the host gear
    /// 2. Registering all REST providers
    /// 3. Finalizing with `OpenAPI` endpoints
    async fn run_rest_phase(&self) -> Result<Router, RegistryError> {
        tracing::info!("Phase: rest (sync)");

        let mut router = Router::new();

        // Find host(s) and whether any rest gears exist
        let host_count = self
            .registry
            .gears()
            .iter()
            .filter(|e| e.caps.has::<ApiGatewayCap>())
            .count();

        match host_count {
            0 => {
                return if self
                    .registry
                    .gears()
                    .iter()
                    .any(|e| e.caps.has::<RestApiCap>())
                {
                    Err(RegistryError::RestRequiresHost)
                } else {
                    Ok(router)
                };
            }
            1 => { /* proceed */ }
            _ => return Err(RegistryError::MultipleRestHosts),
        }

        // Resolve the single host entry and its gear context
        let host_idx = self
            .registry
            .gears()
            .iter()
            .position(|e| e.caps.has::<ApiGatewayCap>())
            .ok_or(RegistryError::RestHostNotFoundAfterValidation)?;
        let host_entry = &self.registry.gears()[host_idx];
        let Some(host) = host_entry.caps.query::<ApiGatewayCap>() else {
            return Err(RegistryError::RestHostMissingFromEntry);
        };
        let host_ctx = self
            .ctx_builder
            .for_gear(host_entry.name)
            .await
            .map_err(|e| RegistryError::RestPrepare {
                gear: host_entry.name,
                source: e,
            })?;

        // use host as the registry
        let registry: &dyn crate::contracts::OpenApiRegistry = host.as_registry();

        // Healthcheck registry, passed explicitly to the REST host and providers below
        // (not via ClientHub). Seeded with the host's shutdown token so in-flight checks
        // are aborted on shutdown.
        let hc_registry = Arc::new(
            crate::healthcheck::RestHealthcheckRegistry::with_cancellation(
                host_ctx.cancellation_token().clone(),
            ),
        );

        // Bridge the process-level eventual-readiness state into the served
        // probe: while any consumed dependency is unresolved (or the process is
        // draining) this synthetic check reports Unhealthy, so the gateway's
        // `/readyz` returns 503 until all `#[toolkit::consumes]` deps are wired.
        // (`/healthz` is a static liveness handler and is unaffected.)
        hc_registry.register(
            "readiness",
            Arc::new(super::readiness::ReadinessHealthcheck::new(
                self.dep_checker.clone(),
            )),
        );

        // 1) Host prepare: base Router / global middlewares / basic OAS meta
        router = host
            .rest_prepare(&host_ctx, router, hc_registry.clone())
            .map_err(|source| RegistryError::RestPrepare {
                gear: host_entry.name,
                source,
            })?;

        // 2) Register all REST providers (in the current discovery order)
        for e in self.registry.gears() {
            if let Some(rest) = e.caps.query::<RestApiCap>() {
                let ctx = self.ctx_builder.for_gear(e.name).await.map_err(|err| {
                    RegistryError::RestRegister {
                        gear: e.name,
                        source: err,
                    }
                })?;

                router = rest
                    .register_rest(&ctx, router, registry)
                    .map_err(|source| RegistryError::RestRegister {
                        gear: e.name,
                        source,
                    })?;

                // Register the gear's readiness healthcheck after successful route registration.
                if let Some(hc) = rest.healthcheck(&ctx) {
                    hc_registry.register(e.name, hc);
                }
            }
        }

        // 3) Host finalize: attach /openapi.json and /docs, persist Router if needed (no server start)
        router = host
            .rest_finalize(&host_ctx, router, hc_registry)
            .map_err(|source| RegistryError::RestFinalize {
                gear: host_entry.name,
                source,
            })?;

        Ok(router)
    }

    /// gRPC registration phase: collect services from all grpc gears.
    ///
    /// Services are stored in the installer store for the `grpc-hub` to consume during start.
    async fn run_grpc_phase(&self) -> Result<(), RegistryError> {
        tracing::info!("Phase: grpc (registration)");

        // If no grpc_hub and no grpc_services, skip the phase
        if self.registry.grpc_hub.is_none() && self.registry.grpc_services.is_empty() {
            return Ok(());
        }

        // If there are grpc_services but no hub, that's an error
        if self.registry.grpc_hub.is_none() && !self.registry.grpc_services.is_empty() {
            return Err(RegistryError::GrpcRequiresHub);
        }

        // If there's a hub, collect all services grouped by gear and hand them off to the installer store
        if let Some(hub_name) = &self.registry.grpc_hub {
            let mut gears_data = Vec::new();
            let mut seen = HashSet::new();
            // Compiled-in service_name -> owning-gear map, installed as
            // authoritative ownership below (see
            // `GearManager::merge_authoritative_grpc_service_owners`).
            let mut owners: std::collections::HashMap<String, String> =
                std::collections::HashMap::new();

            // Collect services from all grpc gears
            for (gear_name, service_gear) in &self.registry.grpc_services {
                let ctx = self.ctx_builder.for_gear(gear_name).await.map_err(|err| {
                    RegistryError::GrpcRegister {
                        gear: gear_name.clone(),
                        source: err,
                    }
                })?;

                let installers = service_gear
                    .get_grpc_services(&ctx)
                    .await
                    .map_err(|source| RegistryError::GrpcRegister {
                        gear: gear_name.clone(),
                        source,
                    })?;

                for reg in &installers {
                    if !seen.insert(reg.service_name) {
                        return Err(RegistryError::GrpcRegister {
                            gear: gear_name.clone(),
                            source: anyhow::anyhow!(
                                "Duplicate gRPC service name: {}",
                                reg.service_name
                            ),
                        });
                    }
                    owners.insert(reg.service_name.to_owned(), gear_name.clone());
                }

                gears_data.push(crate::runtime::GearInstallers {
                    gear_name: gear_name.clone(),
                    installers,
                });
            }

            // Ordering matters: this runs after the orchestrator's init seeded
            // operator config and before the directory-register / OoP-spawn
            // phases, so ownership is pinned before the first self-registration.
            self.gear_manager
                .merge_authoritative_grpc_service_owners(owners);

            self.grpc_installers
                .set(crate::runtime::GrpcInstallerData { gears: gears_data })
                .map_err(|source| RegistryError::GrpcRegister {
                    gear: hub_name.clone(),
                    source,
                })?;
        }

        Ok(())
    }

    /// START phase: start all stateful gears.
    ///
    /// System gears start first, followed by user gears.
    async fn run_start_phase(&self) -> Result<(), RegistryError> {
        tracing::info!("Phase: start");

        for e in self.registry.gears_by_system_priority() {
            if let Some(s) = e.caps.query::<RunnableCap>() {
                tracing::debug!(
                    gear = e.name,
                    is_system = e.caps.has::<SystemCap>(),
                    "Starting stateful gear"
                );
                s.start(self.cancel.clone())
                    .await
                    .map_err(|source| RegistryError::Start {
                        gear: e.name,
                        source,
                    })?;
                tracing::info!(gear = e.name, "Started gear");
            }
        }

        Ok(())
    }

    /// Stop a single gear, logging errors but continuing execution.
    async fn stop_one_gear(entry: &GearEntry, cancel: CancellationToken) {
        if let Some(s) = entry.caps.query::<RunnableCap>() {
            match s.stop(cancel).await {
                Err(err) => {
                    tracing::warn!(gear =  entry.name, error = %err, "Failed to stop gear");
                }
                _ => {
                    tracing::info!(gear = entry.name, "Stopped gear");
                }
            }
        }
    }

    /// STOP phase: stop all stateful gears in reverse order.
    ///
    /// # Two-Phase Shutdown Contract
    ///
    /// This phase implements a proper two-phase shutdown for **each gear**:
    ///
    /// 1. **Graceful stop request**: Each gear's `stop(deadline_token)` is called with a
    ///    *fresh* cancellation token (not the already-cancelled root token). Gears should
    ///    interpret this as "please stop gracefully".
    ///
    /// 2. **Hard-stop deadline**: After `shutdown_deadline` expires **for that gear**,
    ///    its `deadline_token` is cancelled. Gears should interpret this as "abort immediately".
    ///
    /// Each gear gets its own independent deadline — if gear A takes 25s to stop,
    /// gear B still gets the full `shutdown_deadline` for its graceful shutdown.
    ///
    /// This allows gears to implement real graceful shutdown:
    /// - Request cooperative shutdown of child tasks
    /// - Wait for them to finish gracefully
    /// - If `deadline_token` fires, switch to hard-abort mode
    ///
    /// Errors are logged but do not fail the shutdown process.
    /// Note: `OoP` gears are stopped automatically by the backend when the
    /// cancellation token is triggered.
    async fn run_stop_phase(&self) -> Result<(), RegistryError> {
        tracing::info!("Phase: stop");

        // Drop our REST providers from the directory first, so consumers stop
        // resolving an endpoint that is about to disappear.
        self.deregister_rest_providers().await;

        let deadline = self.shutdown_deadline;

        // Stop all gears in reverse order, each with its own independent deadline
        for e in self.registry.gears().iter().rev() {
            let gear_name = e.name;

            // Create a fresh deadline token for THIS gear
            // Each gear gets the full shutdown_deadline independently
            let deadline_token = CancellationToken::new();
            let deadline_token_for_timeout = deadline_token.clone();

            // Spawn a task to cancel this gear's deadline token after shutdown_deadline
            let deadline_task = tokio::spawn(async move {
                tokio::time::sleep(deadline).await;
                tracing::warn!(
                    gear = gear_name,
                    deadline_secs = deadline.as_secs(),
                    "Gear shutdown deadline reached, sending hard-stop signal"
                );
                deadline_token_for_timeout.cancel();
            });

            // Stop this gear with its own deadline token
            // The gear can observe the token transition from uncancelled→cancelled
            Self::stop_one_gear(e, deadline_token).await;

            // Cancel the deadline task and await it to ensure full cleanup
            deadline_task.abort();
            #[allow(clippy::let_underscore_must_use)]
            let _ = deadline_task.await;
        }

        Ok(())
    }

    /// Run the stop phase with a watchdog that force-exits the process if the
    /// stop phase hangs on a blocking syscall. The watchdog is disarmed whether
    /// the stop phase succeeds or fails, so a failing stop phase does not leave
    /// the watchdog active and eventually force-exit the process.
    async fn run_stop_phase_guarded(&self) -> Result<(), RegistryError> {
        let gear_count = u32::try_from(self.registry.gears().len().max(1)).unwrap_or(1);
        let stop_timeout = self
            .shutdown_deadline
            .checked_mul(gear_count)
            .and_then(|d| d.checked_add(std::time::Duration::from_secs(5)))
            .unwrap_or(self.shutdown_deadline);

        // Use a channel to arm/disarm the watchdog. If the lifecycle future is
        // dropped before the stop phase finishes (e.g. an outer timeout), the
        // sender is dropped and the watchdog exits without killing the process.
        let (disarm_tx, disarm_rx) = std::sync::mpsc::channel::<()>();
        std::thread::spawn(move || {
            match disarm_rx.recv_timeout(stop_timeout) {
                Ok(()) | Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
                    // Stop phase completed, or the lifecycle future was cancelled.
                }
                Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
                    tracing::warn!(
                        timeout_secs = stop_timeout.as_secs(),
                        "shutdown: stop phase timed out, force exiting"
                    );
                    std::process::exit(1);
                }
            }
        });

        let stop_result = self.run_stop_phase().await;
        // Disarm the watchdog before propagating the stop-phase result. This runs
        // for both success and failure so a failing stop phase does not leave the
        // watchdog armed and eventually force-exit the process.
        let _ = disarm_tx.send(()).ok();

        stop_result
    }

    /// `OoP` SPAWN phase: spawn out-of-process gears after start phase.
    ///
    /// This phase runs after `grpc-hub` is already listening, so we can pass
    /// the real directory endpoint to `OoP` gears.
    async fn run_oop_spawn_phase(&self) -> Result<(), RegistryError> {
        let oop_opts = match &self.oop_options {
            Some(opts) if !opts.gears.is_empty() => opts,
            _ => return Ok(()),
        };

        tracing::info!("Phase: oop_spawn");

        // Wait for grpc_hub to publish its endpoint (it runs async in start phase)
        let directory_endpoint = self.wait_for_grpc_hub_endpoint().await;

        for gear_cfg in &oop_opts.gears {
            // Build environment with directory endpoint and rendered config
            // Note: User controls --config via execution.args in master config
            let mut env = gear_cfg.env.clone();
            env.insert(
                TOOLKIT_MODULE_CONFIG_ENV.to_owned(),
                gear_cfg.rendered_config_json.clone(),
            );
            if let Some(ref endpoint) = directory_endpoint {
                env.insert(TOOLKIT_DIRECTORY_ENDPOINT_ENV.to_owned(), endpoint.clone());
            }

            // Use args from execution config as-is (user controls --config via args)
            let args = gear_cfg.args.clone();

            let spawn_config = OopSpawnConfig {
                gear_name: gear_cfg.gear_name.clone(),
                binary: gear_cfg.binary.clone(),
                args,
                env,
                working_directory: gear_cfg.working_directory.clone(),
            };

            oop_opts
                .backend
                .spawn(spawn_config)
                .await
                .map_err(|e| RegistryError::OopSpawn {
                    gear: gear_cfg.gear_name.clone(),
                    source: e,
                })?;

            tracing::info!(
                gear =  %gear_cfg.gear_name,
                directory_endpoint = ?directory_endpoint,
                "Spawned OoP gear via backend"
            );
        }

        Ok(())
    }

    /// Wait for `grpc-hub` to publish its bound endpoint.
    ///
    /// Polls the `GrpcHubGear::bound_endpoint()` with a short interval until available or timeout.
    /// Returns None if no `grpc-hub` is running or if it times out.
    async fn wait_for_grpc_hub_endpoint(&self) -> Option<String> {
        const POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(10);
        const MAX_WAIT: std::time::Duration = std::time::Duration::from_secs(5);

        // Find grpc_hub in registry
        let grpc_hub = self
            .registry
            .gears()
            .iter()
            .find_map(|e| e.caps.query::<GrpcHubCap>());

        let Some(hub) = grpc_hub else {
            return None; // No grpc_hub registered
        };

        let start = std::time::Instant::now();

        loop {
            if let Some(endpoint) = hub.bound_endpoint() {
                tracing::debug!(
                    endpoint = %endpoint,
                    elapsed_ms = start.elapsed().as_millis(),
                    "gRPC hub endpoint available"
                );
                return Some(endpoint);
            }

            if start.elapsed() > MAX_WAIT {
                tracing::warn!("Timed out waiting for gRPC hub to bind");
                return None;
            }

            tokio::time::sleep(POLL_INTERVAL).await;
        }
    }

    /// Wait for the REST host gateway to publish its bound endpoint.
    ///
    /// The gateway binds its listener asynchronously in the start phase, so the
    /// bound endpoint may not be set the instant the start phase returns; poll
    /// with a short interval until available or timeout.
    async fn wait_for_rest_endpoint(
        &self,
        host: &Arc<dyn crate::contracts::ApiGatewayCapability>,
    ) -> Option<String> {
        const POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(10);
        const MAX_WAIT: std::time::Duration = std::time::Duration::from_secs(5);

        let start = std::time::Instant::now();
        loop {
            if let Some(endpoint) = host.bound_endpoint() {
                return Some(endpoint);
            }
            if start.elapsed() > MAX_WAIT {
                tracing::warn!("Timed out waiting for REST host to bind");
                return None;
            }
            tokio::time::sleep(POLL_INTERVAL).await;
        }
    }

    /// Directory-register phase (eventual readiness, provider side).
    ///
    /// After the REST server has bound, advertise every in-process REST provider
    /// gear in the service directory under its own gear name, pointing at the
    /// shared gateway endpoint. Consumers resolving a provider gear name then
    /// receive this endpoint and the gateway routes to the provider's handlers.
    ///
    /// No-op when there is no REST host or no REST provider gears, so non-REST
    /// deployments and existing tests are unaffected. Registers through the
    /// `DirectoryClient` in the `ClientHub`, which uniformly targets the
    /// in-process directory (`LocalDirectoryClient`) or the central directory
    /// (`DirectoryGrpcClient` for `OoP`) depending on what the host wired.
    async fn run_directory_register_phase(&self) -> Result<(), RegistryError> {
        let rest_gears = self.rest_provider_gears();
        if rest_gears.is_empty() {
            return Ok(());
        }

        let Some(host) = self
            .registry
            .gears()
            .iter()
            .find_map(|e| e.caps.query::<ApiGatewayCap>())
        else {
            return Ok(()); // no REST host serving the routes
        };

        let Some(endpoint) = self.wait_for_rest_endpoint(&host).await else {
            tracing::warn!(
                "directory-register: REST host endpoint unavailable; skipping REST provider registration"
            );
            return Ok(());
        };

        let Ok(dir) = self.client_hub.get::<dyn crate::DirectoryClient>() else {
            tracing::debug!(
                "directory-register: no DirectoryClient in ClientHub; skipping REST provider registration"
            );
            return Ok(());
        };

        let instance_id = self.instance_id.to_string();
        for gear in rest_gears {
            // The directory keys instances by (gear, instance_id) and replaces
            // wholesale. grpc-hub may have already registered this same
            // (gear, instance_id) with gRPC services during the start phase, so
            // carry the grpc services and version forward instead of clobbering
            // them to empty — adding the REST endpoint must augment, not
            // replace, the entry.
            //
            // Labels are deliberately NOT read-and-rewritten here. Carrying them
            // through would make this a cross-process read-modify-write with no
            // compare-and-set: any label change committed between the read and
            // the write would be silently reverted. Instead we register with an
            // empty label set, which `GearInstance::with_metadata_of` treats as
            // "preserve the stored labels" — an atomic no-op on labels.
            let (grpc_services, version) = match dir.list_instances(gear).await {
                Ok(insts) => insts
                    .into_iter()
                    .find(|i| i.instance_id == instance_id)
                    .map(|i| (i.grpc_services, i.version))
                    .unwrap_or_default(),
                Err(e) => {
                    // A failed directory read must not silently drop the
                    // carried-forward metadata: log it, then fall back to an
                    // empty augmentation so REST registration still proceeds.
                    tracing::warn!(
                        gear,
                        error = %e,
                        "directory-register: failed to read existing registration; \
                         re-registering with empty grpc_services/version"
                    );
                    (Vec::new(), None)
                }
            };
            // OpenAPI spec is published separately (grpc-hub start phase); the
            // REST-augmentation registration does not carry it. Labels are
            // omitted so the store preserves the stored set (see above).
            let mut info = crate::RegisterInstanceInfo::new(gear.to_owned(), instance_id.clone())
                .with_grpc_services(grpc_services)
                .with_rest_endpoint(crate::ServiceEndpoint::new(endpoint.clone()));
            if let Some(version) = version {
                info = info.with_version(version);
            }
            match dir.register_instance(info).await {
                Ok(()) => {
                    tracing::info!(gear, endpoint = %endpoint, "registered REST provider in directory");
                }
                Err(e) => {
                    tracing::warn!(gear, error = %e, "directory-register: failed to register REST provider");
                }
            }
        }
        // Mark that this (in-process host) process advertised its REST providers,
        // so the stop phase deregisters them exactly once. `OoP` serving never
        // runs this phase, so its deregister is owned solely by `oop_serve`.
        self.rest_providers_registered
            .store(true, std::sync::atomic::Ordering::SeqCst);
        Ok(())
    }

    /// Names of all gears that provide a REST API (have `RestApiCap`), excluding
    /// the REST host gateway itself (`ApiGatewayCap`) — the gateway is the
    /// transport, not a contract provider, so it must not be advertised in the
    /// directory under its own gear name.
    fn rest_provider_gears(&self) -> Vec<&'static str> {
        self.registry
            .gears()
            .iter()
            .filter(|e| e.caps.has::<RestApiCap>() && !e.caps.has::<ApiGatewayCap>())
            .map(|e| e.name)
            .collect()
    }

    /// Deregister this process's REST providers from the directory on shutdown,
    /// so consumers stop resolving an endpoint that is going away. Best-effort.
    async fn deregister_rest_providers(&self) {
        // Only the in-process host path registers REST providers in the directory
        // (via `run_directory_register_phase`). In `OoP` serving, `oop_serve`
        // owns presence + deregister, so skip here to avoid a double deregister.
        if !self
            .rest_providers_registered
            .load(std::sync::atomic::Ordering::SeqCst)
        {
            return;
        }
        let rest_gears = self.rest_provider_gears();
        if rest_gears.is_empty() {
            return;
        }
        let Ok(dir) = self.client_hub.get::<dyn crate::DirectoryClient>() else {
            return;
        };
        let instance_id = self.instance_id.to_string();
        for gear in rest_gears {
            if let Err(e) = dir.deregister_instance(gear, &instance_id).await {
                tracing::warn!(gear, error = %e, "directory-deregister: failed to deregister REST provider");
            }
        }
    }

    /// Run the full gear lifecycle (all phases).
    ///
    /// This is the standard entry point for normal application execution.
    /// It runs all phases from pre-init through shutdown.
    ///
    /// # Errors
    ///
    /// Returns an error if any gear phase fails during execution.
    pub async fn run_gear_phases(self) -> anyhow::Result<()> {
        self.run_phases_internal(RunMode::Full).await
    }

    /// Run only the migration phases (pre-init + DB migration).
    ///
    /// This is designed for cloud deployment workflows where database migrations
    /// need to run as a separate step before starting the application.
    /// The process exits after migrations complete.
    ///
    /// # Errors
    ///
    /// Returns an error if pre-init or migration phases fail.
    pub async fn run_migration_phases(self) -> anyhow::Result<()> {
        self.run_phases_internal(RunMode::MigrateOnly).await
    }

    /// Internal implementation that runs gear phases based on the mode.
    ///
    /// This private method contains the actual phase execution logic and is called
    /// by both `run_gear_phases()` and `run_migration_phases()`.
    ///
    /// # Modes
    ///
    /// - `RunMode::Full`: Executes all phases and waits for shutdown signal
    /// - `RunMode::MigrateOnly`: Executes only pre-init and DB migration phases, then exits
    ///
    /// # Phases (Full Mode)
    ///
    /// 1. Pre-init (system gears only)
    /// 2. DB migration (all gears with database capability)
    /// 3. Init (all gears)
    /// 4. Post-init (system gears only)
    /// 5. REST (gears with REST capability)
    /// 6. gRPC (gears with gRPC capability)
    /// 7. Start (runnable gears)
    /// 8. `OoP` spawn (out-of-process gears)
    /// 9. Wait for cancellation
    /// 10. Stop (runnable gears in reverse order)
    async fn run_phases_internal(self, mode: RunMode) -> anyhow::Result<()> {
        // Log execution mode
        match mode {
            RunMode::Full => {
                tracing::info!("Running full lifecycle (all phases)");
            }
            RunMode::MigrateOnly => {
                tracing::info!("Running in migration mode (pre-init + db phases only)");
            }
        }

        // 1. Pre-init phase (before init, only for system gears)
        self.run_pre_init_phase()?;

        // 2. DB migration phase (system gears first)
        #[cfg(feature = "db")]
        {
            self.run_db_phase().await?;
        }
        #[cfg(not(feature = "db"))]
        {
            // No DB integration in this build.
        }

        // Exit early if running in migration-only mode
        if mode == RunMode::MigrateOnly {
            tracing::info!("Migration phases completed successfully");
            return Ok(());
        }

        // 3. Init -> proxy-wiring -> post-init (shared with the OoP path)
        self.run_init_wiring_post_init().await?;

        // 5. REST phase (synchronous router composition)
        let _router = self.run_rest_phase().await?;

        // 6. gRPC registration phase
        self.run_grpc_phase().await?;

        // 7. Start phase
        self.run_start_phase().await?;

        // Draining watcher: flip readiness to Draining the moment shutdown
        // begins so /readyz reports 503 and the orchestrator drains the pod out
        // of the load balancer before the stop phase tears gears down.
        {
            let readiness = Arc::clone(&self.dep_checker);
            let cancel = self.cancel.clone();
            tokio::spawn(async move {
                cancel.cancelled().await;
                readiness.set_draining(true);
            });
        }

        // 7b. Directory-register phase: advertise in-process REST providers in
        //     the directory once the gateway has bound its listener.
        self.run_directory_register_phase().await?;

        // 8. OoP spawn phase (after grpc_hub is running)
        self.run_oop_spawn_phase().await?;

        // 9. Wait for cancellation
        self.cancel.cancelled().await;

        // 10. Stop phase with hard timeout.
        //     Blocking stop implementations are guarded by a watchdog thread so
        //     a hang cannot block shutdown, whether in the in-process or OoP path.
        self.run_stop_phase_guarded().await?;
        Ok(())
    }
}

/// Out-of-process HTTP serving lifecycle (`cpt-cf-component-oop-bootstrap`).
#[cfg(feature = "bootstrap")]
impl HostRuntime {
    /// Compose a **host-less** REST router from all `RestApiCap` gears, plus the
    /// gear's generated `OpenAPI` document (serialized JSON).
    ///
    /// Unlike [`run_rest_phase`](Self::run_rest_phase), this does not require an
    /// `ApiGatewayCap` host: `OoP` gears serve their own routes directly.
    async fn compose_oop_router(
        &self,
        options: &crate::runtime::OopServeOptions,
        hc_registry: &Arc<crate::healthcheck::RestHealthcheckRegistry>,
    ) -> anyhow::Result<(Router, String)> {
        use crate::api::{OpenApiInfo, OpenApiRegistryImpl};
        use anyhow::Context as _;

        let registry = OpenApiRegistryImpl::new();
        let mut router = Router::new();

        for entry in self.registry.gears() {
            if let Some(rest) = entry.caps.query::<RestApiCap>() {
                let ctx = self
                    .ctx_builder
                    .for_gear(entry.name)
                    .await
                    .with_context(|| format!("OoP router: build context for '{}'", entry.name))?;
                router = rest
                    .register_rest(&ctx, router, &registry)
                    .with_context(|| format!("OoP router: register_rest for '{}'", entry.name))?;

                // Register the gear's readiness healthcheck (the same mechanism
                // the api-gateway host uses), so /readyz reflects it identically
                // whether the gear runs in-process or OoP.
                if let Some(hc) = rest.healthcheck(&ctx) {
                    hc_registry.register(entry.name, hc);
                }
            }
        }

        let info = OpenApiInfo {
            title: options.gear_name.clone(),
            version: options
                .version
                .clone()
                .unwrap_or_else(|| "0.0.0".to_owned()),
            description: None,
            servers: vec![],
        };
        let openapi = registry
            .build_openapi(&info)
            .context("OoP router: build OpenAPI document")?;
        let json = serde_json::to_string(&openapi).context("OoP router: serialize OpenAPI")?;

        Ok((router, json))
    }

    /// Run the full `OoP` gear lifecycle: phases (`pre_init` … `start`), then
    /// serve the composed router with framework probes, background
    /// self-registration, dependency resolution, and graceful drain, then the
    /// `stop` phase.
    ///
    /// # Errors
    /// Returns an error if any lifecycle phase or the HTTP server fails.
    pub async fn run_oop_serving(
        self,
        options: crate::runtime::OopServeOptions,
    ) -> anyhow::Result<()> {
        use crate::runtime::ReadinessState;

        tracing::info!("Running OoP serving lifecycle");

        // Make the directory client reachable to the proxy-wiring phase, which
        // resolves remote `#[toolkit::consumes]` providers through the
        // `ClientHub`. The `OoP` presence loop uses `options.directory`
        // directly; consumer wiring reads it here.
        if self.client_hub.get::<dyn crate::DirectoryClient>().is_err() {
            self.client_hub
                .register::<dyn crate::DirectoryClient>(Arc::clone(&options.directory));
        }

        // Shared gear healthcheck registry (same mechanism as the api-gateway
        // host path). Seeded with the root cancellation token so in-flight
        // checks are aborted on shutdown. Populated during router composition.
        let hc_registry = Arc::new(
            crate::healthcheck::RestHealthcheckRegistry::with_cancellation(self.cancel.clone()),
        );

        // Readiness gates on `#[toolkit::consumes]` dependencies only — the same
        // policy as the in-process host path — via the shared `DependencyChecker`
        // fed by the proxy-wiring phase below. `deps = [...]`-only declarations
        // remain for topo-sort ordering but do NOT gate `/readyz`. The
        // healthcheck registry supplies the per-gear readiness dimension; both
        // feed the `/readyz` aggregate.
        let readiness = ReadinessState::from_checker(
            Arc::clone(&self.dep_checker),
            Arc::clone(&hc_registry),
            options.healthcheck_timeout,
        );

        // Bind the HTTP server and serve probes BEFORE the (possibly slow)
        // lifecycle phases, so the kubelet's liveness probe (`/healthz`) passes
        // immediately instead of getting connection-refused during `start()`.
        // Gear routes reply `503 starting` until attached below.
        let mut server = super::oop_serve::OopHttpServer::start(
            Arc::clone(&readiness),
            options,
            self.cancel.clone(),
        )
        .await?;

        // Lifecycle phases up to start, then wire consumers (typed
        // directory-resolving clients feed the shared `DependencyChecker`),
        // then compose the host-less REST router + OpenAPI spec (collecting each
        // gear's healthcheck into the shared registry). Grouped so a failure
        // tears the probe server down cleanly.
        let mut started = false;
        let composed: anyhow::Result<(Router, String)> = async {
            self.run_pre_init_phase()?;
            #[cfg(feature = "db")]
            self.run_db_phase().await?;
            // Init -> proxy-wiring -> post-init, shared with the in-process path
            // so a gear resolving a consumed contract during `start` finds the
            // client in the hub under both profiles.
            self.run_init_wiring_post_init().await?;
            self.run_grpc_phase().await?;
            self.run_start_phase().await?;
            started = true;
            // The gear lifecycle has populated the ClientHub. If an in-process
            // authn stack (e.g. a linked authn-resolver gear) registered a
            // DynBearerAuthenticator bridge, install the tenant plane now —
            // before `attach` layers security_context_middleware. (The platform
            // plane is built eagerly at bootstrap from `oop_http.internal_auth`.)
            server.resolve_bearer_authenticator(&self.client_hub);
            self.compose_oop_router(server.options(), &hc_registry)
                .await
        }
        .await;

        let serve_result = match composed {
            Ok((gear_router, openapi_json)) => {
                // Publish gear routes (they go live) + start directory presence.
                // Dependency resolution already ran in the proxy-wiring phase.
                server.attach(gear_router, openapi_json);
                // Serve until cancelled, drain, then deregister.
                server.join().await
            }
            Err(e) => {
                tracing::error!(error = %e, "OoP startup failed before serving gear routes");
                // Tear down the probe server that is already bound.
                self.cancel.cancel();
                if let Err(join_err) = server.join().await {
                    tracing::warn!(error = %join_err, "OoP probe server teardown after startup failure errored");
                }
                Err(e)
            }
        };

        // Stop phase runs only if start completed successfully. Errors are logged
        // but do not fail the shutdown process (same contract as run_stop_phase).
        if started && let Err(e) = self.run_stop_phase_guarded().await {
            tracing::warn!(error = %e, "OoP stop phase reported an error");
        }

        serve_result
    }
}

#[cfg(test)]
#[cfg(feature = "bootstrap")]
#[cfg_attr(coverage_nightly, coverage(off))]
#[path = "host_runtime_oop_tests.rs"]
mod host_runtime_oop_tests;

#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests {
    use super::*;
    use crate::context::GearCtx;
    use crate::contracts::{Gear, RunnableCapability, SystemCapability};
    use crate::registry::RegistryBuilder;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use tokio::sync::Mutex;

    #[derive(Default)]
    #[allow(dead_code)]
    struct DummyCore;
    #[async_trait::async_trait]
    impl Gear for DummyCore {
        async fn init(&self, _ctx: &GearCtx) -> anyhow::Result<()> {
            Ok(())
        }
    }

    struct StopOrderTracker {
        my_order: usize,
        stop_order: Arc<AtomicUsize>,
    }

    impl StopOrderTracker {
        fn new(counter: &Arc<AtomicUsize>, stop_order: Arc<AtomicUsize>) -> Self {
            let my_order = counter.fetch_add(1, Ordering::SeqCst);
            Self {
                my_order,
                stop_order,
            }
        }
    }

    #[async_trait::async_trait]
    impl Gear for StopOrderTracker {
        async fn init(&self, _ctx: &GearCtx) -> anyhow::Result<()> {
            Ok(())
        }
    }

    #[async_trait::async_trait]
    impl RunnableCapability for StopOrderTracker {
        async fn start(&self, _cancel: CancellationToken) -> anyhow::Result<()> {
            Ok(())
        }
        async fn stop(&self, _cancel: CancellationToken) -> anyhow::Result<()> {
            let order = self.stop_order.fetch_add(1, Ordering::SeqCst);
            tracing::info!(my_order = self.my_order, stop_order = order, "Gear stopped");
            Ok(())
        }
    }

    #[tokio::test]
    async fn test_stop_phase_reverse_order() {
        let counter = Arc::new(AtomicUsize::new(0));
        let stop_order = Arc::new(AtomicUsize::new(0));

        let gear_a = Arc::new(StopOrderTracker::new(&counter, stop_order.clone()));
        let gear_b = Arc::new(StopOrderTracker::new(&counter, stop_order.clone()));
        let gear_c = Arc::new(StopOrderTracker::new(&counter, stop_order.clone()));

        let mut builder = RegistryBuilder::default();
        builder.register_core_with_meta("a", &[], gear_a.clone() as Arc<dyn Gear>);
        builder.register_core_with_meta("b", &["a"], gear_b.clone() as Arc<dyn Gear>);
        builder.register_core_with_meta("c", &["b"], gear_c.clone() as Arc<dyn Gear>);

        builder.register_stateful_with_meta("a", gear_a.clone() as Arc<dyn RunnableCapability>);
        builder.register_stateful_with_meta("b", gear_b.clone() as Arc<dyn RunnableCapability>);
        builder.register_stateful_with_meta("c", gear_c.clone() as Arc<dyn RunnableCapability>);

        let registry = builder.build_topo_sorted().unwrap();

        // Verify gear order is a -> b -> c
        let gear_names: Vec<_> = registry.gears().iter().map(|m| m.name).collect();
        assert_eq!(gear_names, vec!["a", "b", "c"]);

        let client_hub = Arc::new(ClientHub::new());
        let cancel = CancellationToken::new();
        let config_provider: Arc<dyn ConfigProvider> = Arc::new(EmptyConfigProvider);

        let runtime = HostRuntime::new(
            registry,
            config_provider,
            DbOptions::None,
            client_hub,
            cancel.clone(),
            Uuid::new_v4(),
            None,
        );

        // Run stop phase
        runtime.run_stop_phase().await.unwrap();

        // Verify gears stopped in reverse order: c (stop_order=0), b (stop_order=1), a (stop_order=2)
        // Gear order is: a=0, b=1, c=2
        // Stop order should be: c=0, b=1, a=2
        assert_eq!(stop_order.load(Ordering::SeqCst), 3);
    }

    #[tokio::test]
    async fn test_stop_phase_continues_on_error() {
        struct FailingGear {
            should_fail: bool,
            stopped: Arc<AtomicUsize>,
        }

        #[async_trait::async_trait]
        impl Gear for FailingGear {
            async fn init(&self, _ctx: &GearCtx) -> anyhow::Result<()> {
                Ok(())
            }
        }

        #[async_trait::async_trait]
        impl RunnableCapability for FailingGear {
            async fn start(&self, _cancel: CancellationToken) -> anyhow::Result<()> {
                Ok(())
            }
            async fn stop(&self, _cancel: CancellationToken) -> anyhow::Result<()> {
                self.stopped.fetch_add(1, Ordering::SeqCst);
                if self.should_fail {
                    anyhow::bail!("Intentional failure")
                }
                Ok(())
            }
        }

        let stopped = Arc::new(AtomicUsize::new(0));
        let gear_a = Arc::new(FailingGear {
            should_fail: false,
            stopped: stopped.clone(),
        });
        let gear_b = Arc::new(FailingGear {
            should_fail: true,
            stopped: stopped.clone(),
        });
        let gear_c = Arc::new(FailingGear {
            should_fail: false,
            stopped: stopped.clone(),
        });

        let mut builder = RegistryBuilder::default();
        builder.register_core_with_meta("a", &[], gear_a.clone() as Arc<dyn Gear>);
        builder.register_core_with_meta("b", &["a"], gear_b.clone() as Arc<dyn Gear>);
        builder.register_core_with_meta("c", &["b"], gear_c.clone() as Arc<dyn Gear>);

        builder.register_stateful_with_meta("a", gear_a.clone() as Arc<dyn RunnableCapability>);
        builder.register_stateful_with_meta("b", gear_b.clone() as Arc<dyn RunnableCapability>);
        builder.register_stateful_with_meta("c", gear_c.clone() as Arc<dyn RunnableCapability>);

        let registry = builder.build_topo_sorted().unwrap();

        let client_hub = Arc::new(ClientHub::new());
        let cancel = CancellationToken::new();
        let config_provider: Arc<dyn ConfigProvider> = Arc::new(EmptyConfigProvider);

        let runtime = HostRuntime::new(
            registry,
            config_provider,
            DbOptions::None,
            client_hub,
            cancel.clone(),
            Uuid::new_v4(),
            None,
        );

        // Run stop phase - should not fail even though gear_b fails
        runtime.run_stop_phase().await.unwrap();

        // All gears should have attempted to stop
        assert_eq!(stopped.load(Ordering::SeqCst), 3);
    }

    struct EmptyConfigProvider;
    impl ConfigProvider for EmptyConfigProvider {
        fn get_gear_config(&self, _gear_name: &str) -> Option<&serde_json::Value> {
            None
        }
    }

    #[test]
    fn static_endpoint_override_reads_nested_consumer_wiring_key() {
        struct MapCfg(std::collections::HashMap<String, serde_json::Value>);
        impl ConfigProvider for MapCfg {
            fn get_gear_config(&self, gear: &str) -> Option<&serde_json::Value> {
                self.0.get(gear)
            }
        }
        let mut map = std::collections::HashMap::new();
        map.insert(
            "orders".to_owned(),
            serde_json::json!({
                "config": { "consumer_wiring": { "billing": "http://localhost:8081" } }
            }),
        );
        let cfg = MapCfg(map);

        // Present override is read from `config.consumer_wiring.<dep>`.
        assert_eq!(
            super::static_endpoint_override(&cfg, "orders", "billing").as_deref(),
            Some("http://localhost:8081")
        );
        // Absent dep / owner → None (falls through to directory resolution).
        assert_eq!(
            super::static_endpoint_override(&cfg, "orders", "inventory"),
            None
        );
        assert_eq!(
            super::static_endpoint_override(&cfg, "warehouse", "billing"),
            None
        );
        assert_eq!(
            super::static_endpoint_override(&EmptyConfigProvider, "orders", "billing"),
            None
        );
    }

    /// The override is keyed by the *gear name* (kebab), which is what
    /// `#[toolkit::consumes]` puts in `ConsumerRegistration::owner_gear`.
    /// Regression guard: the macro used to emit `stringify!(StructIdent)`, so
    /// the lookup asked for `gears.ApiContractsConsumer` — a key no config
    /// declares — and the escape hatch could never fire.
    #[test]
    fn static_endpoint_override_is_keyed_by_kebab_gear_name() {
        struct MapCfg(std::collections::HashMap<String, serde_json::Value>);
        impl ConfigProvider for MapCfg {
            fn get_gear_config(&self, gear: &str) -> Option<&serde_json::Value> {
                self.0.get(gear)
            }
        }
        let mut map = std::collections::HashMap::new();
        map.insert(
            "api-contracts-consumer".to_owned(),
            serde_json::json!({
                "config": { "consumer_wiring": { "api-contracts": "http://localhost:9099" } }
            }),
        );
        let cfg = MapCfg(map);

        assert_eq!(
            super::static_endpoint_override(&cfg, "api-contracts-consumer", "api-contracts")
                .as_deref(),
            Some("http://localhost:9099"),
        );
        // The pre-fix value — the Rust struct ident — must NOT resolve.
        assert_eq!(
            super::static_endpoint_override(&cfg, "ApiContractsConsumer", "api-contracts"),
            None,
        );
    }

    #[tokio::test]
    async fn test_post_init_runs_after_all_init_and_system_first() {
        #[derive(Clone)]
        struct TrackHooks {
            name: &'static str,
            events: Arc<Mutex<Vec<String>>>,
        }

        #[async_trait::async_trait]
        impl Gear for TrackHooks {
            async fn init(&self, _ctx: &GearCtx) -> anyhow::Result<()> {
                self.events.lock().await.push(format!("init:{}", self.name));
                Ok(())
            }
        }

        #[async_trait::async_trait]
        impl SystemCapability for TrackHooks {
            fn pre_init(&self, _sys: &crate::runtime::SystemContext) -> anyhow::Result<()> {
                Ok(())
            }

            async fn post_init(&self, _sys: &crate::runtime::SystemContext) -> anyhow::Result<()> {
                self.events
                    .lock()
                    .await
                    .push(format!("post_init:{}", self.name));
                Ok(())
            }
        }

        let events = Arc::new(Mutex::new(Vec::<String>::new()));
        let sys_a = Arc::new(TrackHooks {
            name: "sys_a",
            events: events.clone(),
        });
        let user_b = Arc::new(TrackHooks {
            name: "user_b",
            events: events.clone(),
        });
        let user_c = Arc::new(TrackHooks {
            name: "user_c",
            events: events.clone(),
        });

        let mut builder = RegistryBuilder::default();
        builder.register_core_with_meta("sys_a", &[], sys_a.clone() as Arc<dyn Gear>);
        builder.register_core_with_meta("user_b", &["sys_a"], user_b.clone() as Arc<dyn Gear>);
        builder.register_core_with_meta("user_c", &["user_b"], user_c.clone() as Arc<dyn Gear>);
        builder.register_system_with_meta("sys_a", sys_a.clone() as Arc<dyn SystemCapability>);

        let registry = builder.build_topo_sorted().unwrap();

        let client_hub = Arc::new(ClientHub::new());
        let cancel = CancellationToken::new();
        let config_provider: Arc<dyn ConfigProvider> = Arc::new(EmptyConfigProvider);

        let runtime = HostRuntime::new(
            registry,
            config_provider,
            DbOptions::None,
            client_hub,
            cancel,
            Uuid::new_v4(),
            None,
        );

        // Run init phase for all gears, then post_init as a separate barrier phase.
        runtime.run_init_phase().await.unwrap();
        runtime.run_post_init_phase().await.unwrap();

        let events = events.lock().await.clone();
        let first_post_init = events
            .iter()
            .position(|e| e.starts_with("post_init:"))
            .expect("expected post_init events");
        assert!(
            events[..first_post_init]
                .iter()
                .all(|e| e.starts_with("init:")),
            "expected all init events before post_init, got: {events:?}"
        );

        // system-first order within each phase
        assert_eq!(
            events,
            vec![
                "init:sys_a",
                "init:user_b",
                "init:user_c",
                "post_init:sys_a",
            ]
        );
    }

    /// The in-process and `OoP` paths must agree on where proxy-wiring sits.
    ///
    /// They did not: the `OoP` path ran it *after* `start`, so a gear resolving
    /// a consumed contract during its own `start` found nothing in the
    /// `ClientHub` under Profile 2/3 while working under Profile 1. Both paths
    /// now go through `run_init_wiring_post_init`, which pins the order.
    ///
    /// This test covers the two observable endpoints of that segment. The
    /// wiring step between them is a no-op here (no `#[toolkit::consumes]`
    /// registration is linked into this test binary), so what actually stops
    /// the paths diverging again is the shared method itself — inlining it back
    /// into both call sites makes it dead code and fails the `-D warnings`
    /// build.
    #[tokio::test]
    async fn init_wiring_post_init_runs_as_one_ordered_segment() {
        #[derive(Clone)]
        struct TrackHooks {
            events: Arc<Mutex<Vec<String>>>,
        }

        #[async_trait::async_trait]
        impl Gear for TrackHooks {
            async fn init(&self, _ctx: &GearCtx) -> anyhow::Result<()> {
                self.events.lock().await.push("init".to_owned());
                Ok(())
            }
        }

        #[async_trait::async_trait]
        impl SystemCapability for TrackHooks {
            fn pre_init(&self, _sys: &crate::runtime::SystemContext) -> anyhow::Result<()> {
                Ok(())
            }

            async fn post_init(&self, _sys: &crate::runtime::SystemContext) -> anyhow::Result<()> {
                self.events.lock().await.push("post_init".to_owned());
                Ok(())
            }
        }

        let events = Arc::new(Mutex::new(Vec::<String>::new()));
        let gear = Arc::new(TrackHooks {
            events: events.clone(),
        });

        let mut builder = RegistryBuilder::default();
        builder.register_core_with_meta("sys", &[], gear.clone() as Arc<dyn Gear>);
        builder.register_system_with_meta("sys", gear.clone() as Arc<dyn SystemCapability>);
        let registry = builder.build_topo_sorted().unwrap();

        let runtime = HostRuntime::new(
            registry,
            Arc::new(EmptyConfigProvider) as Arc<dyn ConfigProvider>,
            DbOptions::None,
            Arc::new(ClientHub::new()),
            CancellationToken::new(),
            Uuid::new_v4(),
            None,
        );

        runtime.run_init_wiring_post_init().await.unwrap();

        assert_eq!(events.lock().await.clone(), vec!["init", "post_init"]);
    }

    #[tokio::test]
    async fn test_stop_phase_provides_fresh_deadline_token() {
        use std::sync::atomic::AtomicBool;

        struct TokenCheckGear {
            stop_was_called: AtomicBool,
            token_was_cancelled_on_entry: AtomicBool,
        }

        #[async_trait::async_trait]
        impl Gear for TokenCheckGear {
            async fn init(&self, _ctx: &GearCtx) -> anyhow::Result<()> {
                Ok(())
            }
        }

        #[async_trait::async_trait]
        impl RunnableCapability for TokenCheckGear {
            async fn start(&self, _cancel: CancellationToken) -> anyhow::Result<()> {
                Ok(())
            }
            async fn stop(&self, deadline_token: CancellationToken) -> anyhow::Result<()> {
                // Record that stop() was called
                self.stop_was_called.store(true, Ordering::SeqCst);
                // Record whether the token was already cancelled when stop() was called
                self.token_was_cancelled_on_entry
                    .store(deadline_token.is_cancelled(), Ordering::SeqCst);
                Ok(())
            }
        }

        let gear = Arc::new(TokenCheckGear {
            stop_was_called: AtomicBool::new(false),
            // Default to true to detect if stop() was never called
            token_was_cancelled_on_entry: AtomicBool::new(true),
        });

        let mut builder = RegistryBuilder::default();
        builder.register_core_with_meta("test", &[], gear.clone() as Arc<dyn Gear>);
        builder.register_stateful_with_meta("test", gear.clone() as Arc<dyn RunnableCapability>);

        let registry = builder.build_topo_sorted().unwrap();
        let client_hub = Arc::new(ClientHub::new());
        let cancel = CancellationToken::new();
        let config_provider: Arc<dyn ConfigProvider> = Arc::new(EmptyConfigProvider);

        let runtime = HostRuntime::new(
            registry,
            config_provider,
            DbOptions::None,
            client_hub,
            cancel.clone(),
            Uuid::new_v4(),
            None,
        );

        // Run stop phase - the deadline token should NOT be cancelled
        runtime.run_stop_phase().await.unwrap();

        // First, verify stop() was actually called (guards against silent registration failures)
        assert!(
            gear.stop_was_called.load(Ordering::SeqCst),
            "stop() was never called - gear may not have been registered correctly"
        );

        // The token should NOT have been cancelled when stop() was called
        // This is the key fix: gears get a fresh token, not the already-cancelled root token
        assert!(
            !gear.token_was_cancelled_on_entry.load(Ordering::SeqCst),
            "deadline_token should NOT be cancelled when stop() is called - this enables graceful shutdown"
        );
    }

    #[tokio::test]
    async fn test_stop_phase_graceful_shutdown_completes_before_deadline() {
        use std::sync::atomic::AtomicBool;
        use std::time::Duration;

        struct GracefulGear {
            graceful_completed: AtomicBool,
            deadline_fired: AtomicBool,
        }

        #[async_trait::async_trait]
        impl Gear for GracefulGear {
            async fn init(&self, _ctx: &GearCtx) -> anyhow::Result<()> {
                Ok(())
            }
        }

        #[async_trait::async_trait]
        impl RunnableCapability for GracefulGear {
            async fn start(&self, _cancel: CancellationToken) -> anyhow::Result<()> {
                Ok(())
            }
            async fn stop(&self, deadline_token: CancellationToken) -> anyhow::Result<()> {
                // Simulate graceful shutdown that completes quickly (10ms)
                tokio::select! {
                    () = tokio::time::sleep(Duration::from_millis(10)) => {
                        self.graceful_completed.store(true, Ordering::SeqCst);
                    }
                    () = deadline_token.cancelled() => {
                        self.deadline_fired.store(true, Ordering::SeqCst);
                    }
                }
                Ok(())
            }
        }

        let gear = Arc::new(GracefulGear {
            graceful_completed: AtomicBool::new(false),
            deadline_fired: AtomicBool::new(false),
        });

        let mut builder = RegistryBuilder::default();
        builder.register_core_with_meta("test", &[], gear.clone() as Arc<dyn Gear>);
        builder.register_stateful_with_meta("test", gear.clone() as Arc<dyn RunnableCapability>);

        let registry = builder.build_topo_sorted().unwrap();
        let client_hub = Arc::new(ClientHub::new());
        let cancel = CancellationToken::new();
        let config_provider: Arc<dyn ConfigProvider> = Arc::new(EmptyConfigProvider);

        // Use a long deadline (5s) - gear should complete gracefully before this
        let runtime = HostRuntime::new(
            registry,
            config_provider,
            DbOptions::None,
            client_hub,
            cancel.clone(),
            Uuid::new_v4(),
            None,
        )
        .with_shutdown_deadline(Duration::from_secs(5));

        runtime.run_stop_phase().await.unwrap();

        // Graceful shutdown should have completed
        assert!(
            gear.graceful_completed.load(Ordering::SeqCst),
            "graceful shutdown should complete"
        );
        // Deadline should NOT have fired (gear finished before deadline)
        assert!(
            !gear.deadline_fired.load(Ordering::SeqCst),
            "deadline should not fire when graceful shutdown completes quickly"
        );
    }

    #[tokio::test]
    async fn test_stop_phase_deadline_fires_for_slow_gear() {
        use std::sync::atomic::AtomicBool;
        use std::time::Duration;

        struct SlowGear {
            graceful_completed: AtomicBool,
            deadline_fired: AtomicBool,
        }

        #[async_trait::async_trait]
        impl Gear for SlowGear {
            async fn init(&self, _ctx: &GearCtx) -> anyhow::Result<()> {
                Ok(())
            }
        }

        #[async_trait::async_trait]
        impl RunnableCapability for SlowGear {
            async fn start(&self, _cancel: CancellationToken) -> anyhow::Result<()> {
                Ok(())
            }
            async fn stop(&self, deadline_token: CancellationToken) -> anyhow::Result<()> {
                // Simulate slow graceful shutdown (would take 10s, but deadline is 100ms)
                tokio::select! {
                    () = tokio::time::sleep(Duration::from_secs(10)) => {
                        self.graceful_completed.store(true, Ordering::SeqCst);
                    }
                    () = deadline_token.cancelled() => {
                        self.deadline_fired.store(true, Ordering::SeqCst);
                    }
                }
                Ok(())
            }
        }

        let gear = Arc::new(SlowGear {
            graceful_completed: AtomicBool::new(false),
            deadline_fired: AtomicBool::new(false),
        });

        let mut builder = RegistryBuilder::default();
        builder.register_core_with_meta("test", &[], gear.clone() as Arc<dyn Gear>);
        builder.register_stateful_with_meta("test", gear.clone() as Arc<dyn RunnableCapability>);

        let registry = builder.build_topo_sorted().unwrap();
        let client_hub = Arc::new(ClientHub::new());
        let cancel = CancellationToken::new();
        let config_provider: Arc<dyn ConfigProvider> = Arc::new(EmptyConfigProvider);

        // Use a short deadline (100ms) - gear should be interrupted by deadline
        let runtime = HostRuntime::new(
            registry,
            config_provider,
            DbOptions::None,
            client_hub,
            cancel.clone(),
            Uuid::new_v4(),
            None,
        )
        .with_shutdown_deadline(Duration::from_millis(100));

        runtime.run_stop_phase().await.unwrap();

        // Graceful shutdown should NOT have completed (deadline fired first)
        assert!(
            !gear.graceful_completed.load(Ordering::SeqCst),
            "graceful shutdown should not complete when deadline fires first"
        );
        // Deadline should have fired
        assert!(
            gear.deadline_fired.load(Ordering::SeqCst),
            "deadline should fire for slow gears"
        );
    }
}