acton-service 0.30.0

//! Type-safe service builder that enforces API versioning and best practices
//!
//! This module provides a compile-time enforced pattern for building microservices
//! that CANNOT have unversioned routes. The type system makes it impossible to
//! bypass versioning.
//!
//! ## Design Principles
//!
//! 1. **Impossible to bypass versioning**: Only `VersionedRoutes` can be used
//! 2. **Batteries-included**: Health and readiness endpoints are automatic
//! 3. **Type-state pattern**: Compiler enforces configuration order
//! 4. **Opaque types**: Internal Router cannot be accessed directly
//!
//! ## Example
//!
//! ```rust,ignore
//! use acton_service::prelude::*;
//!
//! #[tokio::main]
//! async fn main() -> Result<()> {
//!     // Create versioned routes (ONLY way to create routes)
//!     let routes = VersionedApiBuilder::new()
//!         .with_base_path("/api")
//!         .add_version(ApiVersion::V1, |router| {
//!             router.route("/users", get(list_users))
//!         })
//!         .build_routes();  // Returns VersionedRoutes (not Router!)
//!
//!     // Build service with type-safe builder
//!     // Config loading and tracing initialization happen automatically
//!     let service = ServiceBuilder::new()
//!         .with_routes(routes)  // Only accepts VersionedRoutes
//!         .build();  // Automatically loads config and initializes tracing
//!
//!     // Health and readiness endpoints are automatically included
//!     service.serve().await?;
//!
//!     Ok(())
//! }
//! ```

use crate::config::Config;
use crate::middleware::{request_id_layer, request_id_propagation_layer, sensitive_headers_layer};
use crate::state::AppState;
use axum::Router;
use serde::{de::DeserializeOwned, Serialize};
use std::time::Duration;
use tower_http::{
    catch_panic::CatchPanicLayer,
    compression::CompressionLayer,
    cors::CorsLayer,
    limit::RequestBodyLimitLayer,
    timeout::TimeoutLayer,
    trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer},
};

/// Opaque wrapper around versioned routes with batteries-included health/readiness
///
/// This type can ONLY be created by `VersionedApiBuilder::build_routes()`.
/// It cannot be constructed manually, ensuring all routes are versioned.
///
/// Uses an enum to support both stateless routes (`Router<()>`) and stateful
/// routes (`Router<AppState<T>>`)
#[derive(Debug)]
pub enum VersionedRoutes<T = ()>
where
    T: Serialize + DeserializeOwned + Clone + Default + Send + Sync + 'static,
{
    /// Routes without state (typical versioned API routes)
    WithoutState(Router<()>),
    /// Routes with AppState (includes health/readiness endpoints)
    WithState(Router<AppState<T>>),
}

impl<T> VersionedRoutes<T>
where
    T: Serialize + DeserializeOwned + Clone + Default + Send + Sync + 'static,
{
    /// Create from a stateless router (crate-private, only accessible to VersionedApiBuilder)
    #[allow(dead_code)]
    pub(crate) fn from_router(router: Router<()>) -> Self {
        Self::WithoutState(router)
    }

    /// Create from a stateful router (crate-private)
    pub(crate) fn from_router_with_state(router: Router<AppState<T>>) -> Self {
        Self::WithState(router)
    }
}

impl<T> Default for VersionedRoutes<T>
where
    T: Serialize + DeserializeOwned + Clone + Default + Send + Sync + 'static,
{
    /// Default routes with health and readiness endpoints
    fn default() -> Self {
        use axum::routing::get;

        let health_router: Router<AppState<T>> = Router::new()
            .route("/health", get(crate::health::health::<T>))
            .route("/ready", get(crate::health::readiness::<T>));

        Self::WithState(health_router)
    }
}

/// Simplified service builder with sensible defaults
///
/// Generic parameter `T` allows custom config extensions.
/// Use `ServiceBuilder<()>` (the default) for no custom config.
///
/// All fields are optional with defaults:
/// - config: Uses `Config::default()`
/// - routes: Uses `VersionedRoutes::default()` (health + readiness only)
/// - state: Uses `AppState::default()`
/// - grpc_services: None (gRPC server disabled by default)
/// - cedar: None (auto-configures from config.cedar if enabled)
/// - agent_runtime: None (agent-based reactive components disabled by default)
///
/// Health and readiness endpoints are ALWAYS included (automatically added by ServiceBuilder).
pub struct ServiceBuilder<T = ()>
where
    T: Serialize + DeserializeOwned + Clone + Default + Send + Sync + 'static,
{
    config: Option<Config<T>>,
    routes: Option<VersionedRoutes<T>>,
    state: Option<AppState<T>>,
    #[cfg(feature = "grpc")]
    grpc_services: Option<tonic::service::Routes>,
    #[cfg(feature = "cedar-authz")]
    cedar: Option<crate::middleware::cedar::CedarAuthz>,
    #[cfg(feature = "cedar-authz")]
    cedar_path_normalizer: Option<fn(&str) -> String>,
    #[cfg(feature = "graphql")]
    graphql: Option<crate::graphql::VersionedGraphQL>,
    agent_runtime: Option<acton_reactive::prelude::ActorRuntime>,
    actor_extensions: Vec<Box<dyn crate::extensions::ActorExtensionSpawner>>,
    /// Caller-supplied HTTP TLS credentials. When set, `build()` uses them
    /// verbatim and never reads `[tls]` cert/key files.
    #[cfg(feature = "tls")]
    tls_config_override: Option<crate::tls::TlsConfigSource>,
    /// Caller-supplied gRPC TLS credentials. When set, `build()` uses them
    /// verbatim and never reads `[grpc.tls]` cert/key files.
    #[cfg(all(feature = "grpc", feature = "tls"))]
    grpc_tls_config_override: Option<crate::tls::TlsConfigSource>,
    /// Caller-supplied callback handed the resolved reloadable TLS sources at
    /// serve time. See [`ServiceBuilder::with_tls_reload`].
    #[cfg(feature = "tls")]
    tls_reload_hook: Option<Box<dyn FnOnce(crate::tls::TlsReloadHandle) + Send>>,
}

impl<T> ServiceBuilder<T>
where
    T: Serialize + DeserializeOwned + Clone + Default + Send + Sync + 'static,
{
    /// Create a new service builder with defaults
    pub fn new() -> Self {
        Self {
            config: None,
            routes: None,
            state: None,
            #[cfg(feature = "grpc")]
            grpc_services: None,
            #[cfg(feature = "cedar-authz")]
            cedar: None,
            #[cfg(feature = "cedar-authz")]
            cedar_path_normalizer: None,
            #[cfg(feature = "graphql")]
            graphql: None,
            agent_runtime: None,
            actor_extensions: Vec::new(),
            #[cfg(feature = "tls")]
            tls_config_override: None,
            #[cfg(all(feature = "grpc", feature = "tls"))]
            grpc_tls_config_override: None,
            #[cfg(feature = "tls")]
            tls_reload_hook: None,
        }
    }

    /// Supply a pre-built HTTP TLS configuration.
    ///
    /// When set, `build()` uses this config verbatim for the HTTP listener and
    /// never reads the `[tls]` `cert_path` / `key_path` files. Use this when the
    /// application has already loaded and validated the key material: it removes
    /// the second read that would otherwise sit between your check and the
    /// listener coming up, closing that time-of-check/time-of-use window.
    ///
    /// The override applies whenever it is set, regardless of what `[tls]` says.
    ///
    /// The credentials supplied here are **static**: the listener serves this
    /// exact config for its whole life. To rotate certificates without a
    /// restart, use [`with_tls_config_source`](Self::with_tls_config_source).
    #[cfg(feature = "tls")]
    pub fn with_tls_config(
        mut self,
        config: std::sync::Arc<tokio_rustls::rustls::ServerConfig>,
    ) -> Self {
        self.tls_config_override = Some(crate::tls::TlsConfigSource::from_server_config(config));
        self
    }

    /// Supply reloadable HTTP TLS credentials.
    ///
    /// The rotatable form of [`with_tls_config`](Self::with_tls_config): it
    /// carries the same time-of-check/time-of-use guarantee, and additionally
    /// lets the application replace the certificate while the listener runs.
    /// Keep a clone of the source and call
    /// [`TlsConfigSource::reload`](crate::tls::TlsConfigSource::reload) after
    /// the credential files change; the next handshake picks up the new config.
    /// A failed reload is fail-closed and keeps the previous credentials
    /// serving, so rotation can never take the listener down.
    ///
    /// The override applies whenever it is set, regardless of what `[tls]` says.
    #[cfg(feature = "tls")]
    pub fn with_tls_config_source(mut self, source: crate::tls::TlsConfigSource) -> Self {
        self.tls_config_override = Some(source);
        self
    }

    /// Supply a pre-built TLS configuration for the separate-port gRPC listener.
    ///
    /// The gRPC twin of [`with_tls_config`](Self::with_tls_config), carrying the
    /// same time-of-check/time-of-use guarantee. Applies only to the dual-port
    /// gRPC listener; in single-port mode the HTTP TLS config terminates both.
    ///
    /// The override applies whenever it is set, regardless of what `[grpc.tls]`
    /// says, and suppresses the fallback to the shared `[tls]` config.
    ///
    /// The credentials supplied here are **static**. To rotate them without a
    /// restart, use
    /// [`with_grpc_tls_config_source`](Self::with_grpc_tls_config_source).
    #[cfg(all(feature = "grpc", feature = "tls"))]
    pub fn with_grpc_tls_config(
        mut self,
        config: std::sync::Arc<tokio_rustls::rustls::ServerConfig>,
    ) -> Self {
        self.grpc_tls_config_override =
            Some(crate::tls::TlsConfigSource::from_server_config(config));
        self
    }

    /// Supply reloadable TLS credentials for the separate-port gRPC listener.
    ///
    /// The gRPC twin of
    /// [`with_tls_config_source`](Self::with_tls_config_source), with the same
    /// rotation and fail-closed semantics. Applies only to the dual-port gRPC
    /// listener; in single-port mode the HTTP TLS source terminates both.
    ///
    /// The override applies whenever it is set, regardless of what `[grpc.tls]`
    /// says, and suppresses the fallback to the shared `[tls]` config.
    #[cfg(all(feature = "grpc", feature = "tls"))]
    pub fn with_grpc_tls_config_source(mut self, source: crate::tls::TlsConfigSource) -> Self {
        self.grpc_tls_config_override = Some(source);
        self
    }

    /// Drive TLS credential rotation from your own trigger.
    ///
    /// `f` is called once by [`ActonService::serve`], as the listeners come up,
    /// with a [`TlsReloadHandle`](crate::tls::TlsReloadHandle) over every
    /// reloadable source the service resolved. Spawn whatever watches your
    /// rotation source — a Vault lease renewal, a Kubernetes secret watch, an
    /// admin endpoint, a message queue — and call
    /// [`TlsReloadHandle::reload_all`](crate::tls::TlsReloadHandle::reload_all)
    /// when the credential files change.
    ///
    /// This is the preferred way to reach the credentials. The alternative,
    /// [`ActonService::tls_config_source`], requires cloning the handle out
    /// before `serve()` consumes the service; a hook registered here is invoked
    /// by `serve()` itself, so that ordering cannot be got wrong.
    ///
    /// For the common cases — poll the files on an interval, or reload on
    /// `SIGHUP` — prefer the built-in triggers: set `reload_interval_secs` or
    /// `reload_on_sighup` on the `[tls]` config section and write no code at
    /// all. This hook is for triggers the framework does not model.
    ///
    /// # When it is not called
    ///
    /// The hook is skipped, with a `WARN` explaining why, when no reloadable
    /// source resolved: TLS is disabled entirely, or every source was injected
    /// as an already-loaded `ServerConfig`
    /// ([`with_tls_config`](Self::with_tls_config)) and so has no files to
    /// reread. Registering a hook and silently never calling it would leave a
    /// service that believes it can rotate and cannot.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let service = ServiceBuilder::new()
    ///     .with_config(config)
    ///     .with_routes(routes)
    ///     .with_tls_reload(|handle| {
    ///         tokio::spawn(async move {
    ///             let mut events = watch_secret_store().await;
    ///             while events.next().await.is_some() {
    ///                 for (listener, result) in handle.reload_all() {
    ///                     if let Err(e) = result {
    ///                         tracing::error!("{listener} TLS reload failed: {e}");
    ///                     }
    ///                 }
    ///             }
    ///         });
    ///     })
    ///     .build();
    ///
    /// service.serve().await?;
    /// ```
    #[cfg(feature = "tls")]
    pub fn with_tls_reload<F>(mut self, f: F) -> Self
    where
        F: FnOnce(crate::tls::TlsReloadHandle) + Send + 'static,
    {
        self.tls_reload_hook = Some(Box::new(f));
        self
    }

    /// Register a custom actor extension.
    ///
    /// The actor will be spawned under a framework-managed supervisor during
    /// [`build()`](Self::build). It must implement [`ActorExtension`](crate::extensions::ActorExtension),
    /// which requires configuring message handlers via the `configure` method.
    ///
    /// Access the actor's handle in request handlers via [`AppState::actor::<A>()`](crate::state::AppState::actor).
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// ServiceBuilder::new()
    ///     .with_actor::<MyCache>()
    ///     .with_routes(routes)
    ///     .build()
    ///     .serve()
    ///     .await?;
    /// ```
    pub fn with_actor<A: crate::extensions::ActorExtension>(mut self) -> Self {
        self.actor_extensions
            .push(Box::new(crate::extensions::ActorExtensionEntry::<A>(
                std::marker::PhantomData,
            )));
        self
    }

    /// Set the service configuration (optional, defaults to Config::default())
    pub fn with_config(mut self, config: Config<T>) -> Self {
        self.config = Some(config);
        self
    }

    /// Add versioned routes to the service
    ///
    /// **IMPORTANT**: This method ONLY accepts `VersionedRoutes`, which can
    /// only be created by `VersionedApiBuilder::build_routes()`.
    /// This makes it impossible to add unversioned routes.
    ///
    /// If not provided, defaults to VersionedRoutes::default() (empty routes).
    pub fn with_routes(mut self, routes: VersionedRoutes<T>) -> Self {
        self.routes = Some(routes);
        self
    }

    /// Set the application state (optional, defaults to AppState::default())
    pub fn with_state(mut self, state: AppState<T>) -> Self {
        self.state = Some(state);
        self
    }

    /// Add gRPC services to the service (optional, requires "grpc" feature)
    ///
    /// When gRPC services are provided, the server will support both HTTP and gRPC
    /// protocols on the same port (by default) or separate ports (if configured).
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use acton_service::prelude::*;
    /// use acton_service::grpc::server::GrpcServicesBuilder;
    ///
    /// let grpc_services = GrpcServicesBuilder::new()
    ///     .add_service(UserServiceServer::new(user_service))
    ///     .build()
    ///     .expect("At least one gRPC service must be added");
    ///
    /// let service = ServiceBuilder::new()
    ///     .with_routes(http_routes)
    ///     .with_grpc_services(grpc_services)
    ///     .build();
    /// ```
    #[cfg(feature = "grpc")]
    pub fn with_grpc_services(mut self, services: tonic::service::Routes) -> Self {
        self.grpc_services = Some(services);
        self
    }

    /// Register a versioned GraphQL schema collection (requires "graphql" feature).
    ///
    /// Each schema is mounted at `/{base_path}/{version}/graphql` inside the
    /// existing versioned Axum router, so it inherits the same middleware
    /// stack (auth, tracing, rate limiting, CORS, ...) that protects REST
    /// endpoints.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use acton_service::prelude::*;
    /// use acton_service::graphql::VersionedGraphQLBuilder;
    /// use async_graphql::{Object, Schema, EmptyMutation, EmptySubscription};
    ///
    /// struct Query;
    /// #[Object] impl Query { async fn hello(&self) -> &'static str { "world" } }
    ///
    /// let schema = Schema::build(Query, EmptyMutation, EmptySubscription).finish();
    /// let graphql = VersionedGraphQLBuilder::new()
    ///     .with_base_path("/api")
    ///     .add_version(ApiVersion::V1, schema)
    ///     .build();
    ///
    /// let service = ServiceBuilder::new()
    ///     .with_routes(routes)
    ///     .with_versioned_graphql(graphql)
    ///     .build();
    /// ```
    #[cfg(feature = "graphql")]
    pub fn with_versioned_graphql(mut self, graphql: crate::graphql::VersionedGraphQL) -> Self {
        self.graphql = Some(graphql);
        self
    }

    /// Set Cedar authorization with explicit configuration
    ///
    /// This allows full control over Cedar initialization. Use this when you need:
    /// - Custom path normalization
    /// - Policy caching
    /// - Other advanced Cedar customization
    ///
    /// For simple cases, just use `.with_config()` and Cedar will auto-configure.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use acton_service::prelude::*;
    /// use acton_service::middleware::cedar::CedarAuthz;
    ///
    /// let cedar = CedarAuthz::builder(config.cedar.unwrap())
    ///     .with_path_normalizer(normalize_fn)
    ///     .with_cache(redis_cache)
    ///     .build()
    ///     .await?;
    ///
    /// let service = ServiceBuilder::new()
    ///     .with_config(config)
    ///     .with_cedar(cedar)  // Explicit Cedar instance
    ///     .with_routes(routes)
    ///     .build();
    /// ```
    #[cfg(feature = "cedar-authz")]
    pub fn with_cedar(mut self, cedar: crate::middleware::cedar::CedarAuthz) -> Self {
        self.cedar = Some(cedar);
        self
    }

    /// Set ONLY a custom path normalizer for Cedar (convenience method)
    ///
    /// This is the recommended way for most users who just need custom path normalization.
    /// Cedar will auto-configure from config.cedar with your custom normalizer.
    ///
    /// By default, Cedar uses a generic path normalizer that replaces UUIDs and numeric IDs
    /// with `{id}` placeholders. Use this method to provide custom normalization logic for
    /// your application's specific path patterns.
    ///
    /// This is only needed when:
    /// - You have alphanumeric IDs (like "user123", "doc1") that aren't UUIDs or numeric
    /// - You have slug-based routes (like "/articles/my-article-title")
    /// - Complex path patterns not handled by the default normalizer
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use acton_service::prelude::*;
    ///
    /// // Define a custom normalizer for alphanumeric document IDs
    /// fn normalize_document_paths(path: &str) -> String {
    ///     // Handles: /api/v1/documents/user123/doc1 -> /api/v1/documents/{user_id}/{doc_id}
    ///     let doc_pattern = regex::Regex::new(
    ///         r"^(/api/v[0-9]+/documents/)([a-zA-Z0-9_-]+)/([a-zA-Z0-9_-]+)$"
    ///     ).unwrap();
    ///
    ///     if let Some(caps) = doc_pattern.captures(path) {
    ///         return format!("{}{{user_id}}/{{doc_id}}", &caps[1]);
    ///     }
    ///     path.to_string()
    /// }
    ///
    /// let service = ServiceBuilder::new()
    ///     .with_config(config)
    ///     .with_routes(routes)
    ///     .with_cedar_path_normalizer(normalize_document_paths)
    ///     .build();
    /// ```
    #[cfg(feature = "cedar-authz")]
    pub fn with_cedar_path_normalizer(mut self, normalizer: fn(&str) -> String) -> Self {
        self.cedar_path_normalizer = Some(normalizer);
        self
    }

    /// Initialize the agent runtime (internal use only)
    ///
    /// Returns a mutable reference to the `ActorRuntime` for spawning agents.
    /// Called automatically by `build()` when connection pools are configured.
    fn init_agent_runtime(&mut self) -> &mut acton_reactive::prelude::ActorRuntime {
        // Note: agent_runtime should already be initialized in the async block
        // before this is called
        self.agent_runtime
            .as_mut()
            .expect("Agent runtime not initialized")
    }

    /// Get the agent broker handle (internal use only)
    fn broker(&self) -> Option<acton_reactive::prelude::ActorHandle> {
        self.agent_runtime.as_ref().map(|r| r.broker())
    }

    /// Build the service
    ///
    /// Automatically handles:
    /// - **Config loading**: Calls `Config::load()` if not provided (falls back to `Config::default()` on error)
    /// - **Tracing initialization**: Initializes tracing with the loaded config
    /// - **Pool agent spawning**: Spawns internal agents for database/redis/nats when configured
    /// - **Health endpoints**: Always includes `/health` and `/ready` endpoints
    ///
    /// Uses defaults for any fields not set:
    /// - config: `Config::load()` → `Config::default()` if load fails
    /// - routes: `VersionedRoutes::default()` (health + readiness only)
    /// - state: `AppState::default()` with agent-managed pools
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// // Minimal - everything is automatic
    /// let service = ServiceBuilder::new().build();
    /// // → Loads config, initializes tracing, spawns pool agents, adds health endpoints
    ///
    /// // With custom routes (most common)
    /// let service = ServiceBuilder::new()
    ///     .with_routes(versioned_routes)
    ///     .build();
    /// // → Pool agents automatically manage database/redis/nats connections
    ///
    /// // Override config (e.g., for testing)
    /// let custom_config = Config { /* ... */ };
    /// let service = ServiceBuilder::new()
    ///     .with_config(custom_config)
    ///     .with_routes(routes)
    ///     .build();
    /// // → Uses your config, spawns appropriate pool agents
    /// ```
    pub fn build(mut self) -> ActonService<T> {
        // Collects the first fatal misconfiguration found while building.
        // Surfaced by `try_build()`, or by `serve()` before any socket binds.
        // Security-relevant subsystems (TLS, token auth) record here instead of
        // degrading to a weaker posture than the operator configured.
        let mut startup_error: Option<crate::error::Error> = None;

        // Load config if not provided
        let config = self.config.take().unwrap_or_else(|| {
            Config::<T>::load().unwrap_or_else(|e| {
                eprintln!("Warning: Failed to load config: {}, using defaults", e);
                Config::<T>::default()
            })
        });

        // Initialize tracing with the loaded config
        if let Err(e) = crate::observability::init_tracing(&config) {
            eprintln!("Warning: Failed to initialize tracing: {}", e);
        }

        // Initialize the meter provider (OTLP push and/or Prometheus pull readers)
        // before middleware is layered, so the HTTP metrics layer can obtain a meter.
        #[cfg(feature = "_metrics")]
        if let Err(e) = crate::observability::init_meter_provider(&config) {
            eprintln!("Warning: Failed to initialize metrics: {}", e);
        }

        // Determine if we need to spawn pool agents
        #[cfg(feature = "database")]
        let needs_db_agent = config.database.is_some();

        #[cfg(feature = "cache")]
        let needs_redis_agent = config.redis.is_some();

        #[cfg(feature = "events")]
        let needs_nats_agent = config.nats.is_some();

        #[cfg(feature = "turso")]
        let needs_turso_agent = config.turso.is_some();

        #[cfg(feature = "surrealdb")]
        let needs_surrealdb_agent = config.surrealdb.is_some();

        #[cfg(feature = "clickhouse")]
        let needs_clickhouse_agent = config.clickhouse.is_some();

        #[cfg(any(
            feature = "database",
            feature = "cache",
            feature = "events",
            feature = "turso",
            feature = "surrealdb",
            feature = "clickhouse"
        ))]
        let needs_agents = {
            #[cfg(feature = "database")]
            let db = needs_db_agent;
            #[cfg(not(feature = "database"))]
            let db = false;

            #[cfg(feature = "cache")]
            let redis = needs_redis_agent;
            #[cfg(not(feature = "cache"))]
            let redis = false;

            #[cfg(feature = "events")]
            let nats = needs_nats_agent;
            #[cfg(not(feature = "events"))]
            let nats = false;

            #[cfg(feature = "turso")]
            let turso = needs_turso_agent;
            #[cfg(not(feature = "turso"))]
            let turso = false;

            #[cfg(feature = "surrealdb")]
            let surrealdb = needs_surrealdb_agent;
            #[cfg(not(feature = "surrealdb"))]
            let surrealdb = false;

            #[cfg(feature = "clickhouse")]
            let clickhouse = needs_clickhouse_agent;
            #[cfg(not(feature = "clickhouse"))]
            let clickhouse = false;

            db || redis || nats || turso || surrealdb || clickhouse
        };

        // Initialize agent runtime and spawn pool agents if needed
        #[cfg(feature = "database")]
        let shared_db_pool: Option<crate::agents::SharedDbPool> = if needs_db_agent {
            Some(std::sync::Arc::new(tokio::sync::RwLock::new(None)))
        } else {
            None
        };

        #[cfg(feature = "cache")]
        let shared_redis_pool: Option<crate::agents::SharedRedisPool> = if needs_redis_agent {
            Some(std::sync::Arc::new(tokio::sync::RwLock::new(None)))
        } else {
            None
        };

        #[cfg(feature = "events")]
        let shared_nats_client: Option<crate::agents::SharedNatsClient> = if needs_nats_agent {
            Some(std::sync::Arc::new(tokio::sync::RwLock::new(None)))
        } else {
            None
        };

        #[cfg(feature = "turso")]
        let shared_turso_db: Option<crate::agents::SharedTursoDb> = if needs_turso_agent {
            Some(std::sync::Arc::new(tokio::sync::RwLock::new(None)))
        } else {
            None
        };

        #[cfg(feature = "surrealdb")]
        let shared_surrealdb_client: Option<crate::agents::SharedSurrealDb> =
            if needs_surrealdb_agent {
                Some(std::sync::Arc::new(tokio::sync::RwLock::new(None)))
            } else {
                None
            };

        #[cfg(feature = "clickhouse")]
        let shared_clickhouse_client: Option<crate::agents::SharedClickHouseClient> =
            if needs_clickhouse_agent {
                Some(std::sync::Arc::new(tokio::sync::RwLock::new(None)))
            } else {
                None
            };

        // Agent handles for AppState
        #[cfg(feature = "database")]
        let mut db_agent_handle: Option<acton_reactive::prelude::ActorHandle> = None;
        #[cfg(feature = "cache")]
        let mut redis_agent_handle: Option<acton_reactive::prelude::ActorHandle> = None;
        #[cfg(feature = "events")]
        let mut nats_agent_handle: Option<acton_reactive::prelude::ActorHandle> = None;
        #[cfg(feature = "turso")]
        let mut turso_agent_handle: Option<acton_reactive::prelude::ActorHandle> = None;
        #[cfg(feature = "surrealdb")]
        let mut surrealdb_agent_handle: Option<acton_reactive::prelude::ActorHandle> = None;
        #[cfg(feature = "clickhouse")]
        let mut clickhouse_agent_handle: Option<acton_reactive::prelude::ActorHandle> = None;

        #[cfg(any(
            feature = "database",
            feature = "cache",
            feature = "events",
            feature = "turso",
            feature = "surrealdb",
            feature = "clickhouse"
        ))]
        let mut broker_handle = if needs_agents {
            // Use block_in_place to run async code in sync context
            // Initialize agent runtime inside the async block using launch_async()
            if runtime_supports_block_in_place() == Some(true) {
                tokio::task::block_in_place(|| {
                    tokio::runtime::Handle::current().block_on(async {
                        // Initialize the agent runtime using launch_async() for async context
                        tracing::debug!("Initializing acton-reactive agent runtime");
                        self.agent_runtime =
                            Some(acton_reactive::prelude::ActonApp::launch_async().await);
                        let runtime = self.init_agent_runtime();

                        // Spawn database pool agent
                        #[cfg(feature = "database")]
                        if let Some(ref db_config) = config.database {
                            match crate::agents::DatabasePoolAgent::spawn(
                                runtime,
                                db_config.clone(),
                                shared_db_pool.clone(),
                            )
                            .await
                            {
                                Ok(handle) => {
                                    tracing::info!("Database pool agent spawned");
                                    db_agent_handle = Some(handle);
                                }
                                Err(e) => {
                                    tracing::warn!("Failed to spawn database pool agent: {}", e);
                                }
                            }
                        }

                        // Spawn Redis pool agent
                        #[cfg(feature = "cache")]
                        if let Some(ref redis_config) = config.redis {
                            match crate::agents::RedisPoolAgent::spawn(
                                runtime,
                                redis_config.clone(),
                                shared_redis_pool.clone(),
                            )
                            .await
                            {
                                Ok(handle) => {
                                    tracing::info!("Redis pool agent spawned");
                                    redis_agent_handle = Some(handle);
                                }
                                Err(e) => {
                                    tracing::warn!("Failed to spawn Redis pool agent: {}", e);
                                }
                            }
                        }

                        // Spawn NATS pool agent
                        #[cfg(feature = "events")]
                        if let Some(ref nats_config) = config.nats {
                            match crate::agents::NatsPoolAgent::spawn(
                                runtime,
                                nats_config.clone(),
                                shared_nats_client.clone(),
                            )
                            .await
                            {
                                Ok(handle) => {
                                    tracing::info!("NATS pool agent spawned");
                                    nats_agent_handle = Some(handle);
                                }
                                Err(e) => {
                                    tracing::warn!("Failed to spawn NATS pool agent: {}", e);
                                }
                            }
                        }

                        // Spawn Turso database agent
                        #[cfg(feature = "turso")]
                        if let Some(ref turso_config) = config.turso {
                            match crate::agents::TursoDbAgent::spawn(
                                runtime,
                                turso_config.clone(),
                                shared_turso_db.clone(),
                            )
                            .await
                            {
                                Ok(handle) => {
                                    tracing::info!("Turso database agent spawned");
                                    turso_agent_handle = Some(handle);
                                }
                                Err(e) => {
                                    tracing::warn!("Failed to spawn Turso database agent: {}", e);
                                }
                            }
                        }

                        // Spawn SurrealDB agent
                        #[cfg(feature = "surrealdb")]
                        if let Some(ref surrealdb_config) = config.surrealdb {
                            match crate::agents::SurrealDbAgent::spawn(
                                runtime,
                                surrealdb_config.clone(),
                                shared_surrealdb_client.clone(),
                            )
                            .await
                            {
                                Ok(handle) => {
                                    tracing::info!("SurrealDB agent spawned");
                                    surrealdb_agent_handle = Some(handle);
                                }
                                Err(e) => {
                                    tracing::warn!("Failed to spawn SurrealDB agent: {}", e);
                                }
                            }
                        }

                        // Spawn ClickHouse agent
                        #[cfg(feature = "clickhouse")]
                        if let Some(ref ch_config) = config.clickhouse {
                            match crate::agents::ClickHousePoolAgent::spawn(
                                runtime,
                                ch_config.clone(),
                                shared_clickhouse_client.clone(),
                            )
                            .await
                            {
                                Ok(handle) => {
                                    tracing::info!("ClickHouse pool agent spawned");
                                    clickhouse_agent_handle = Some(handle);
                                }
                                Err(e) => {
                                    tracing::warn!("Failed to spawn ClickHouse pool agent: {}", e);
                                }
                            }
                        }
                    });
                });
            } else if runtime_supports_block_in_place() == Some(false) {
                let err = crate::error::Error::Internal(
                    "database/cache/events agents are configured but the tokio runtime is \
                     single-threaded, so they cannot be spawned from ServiceBuilder::build(); \
                     use a multi-threaded runtime (#[tokio::main], or \
                     #[tokio::test(flavor = \"multi_thread\")] in tests)"
                        .to_string(),
                );
                tracing::error!("{}", err);
                record_startup_error(&mut startup_error, err);
            }

            self.broker()
        } else {
            None
        };

        // Spawn audit agent if configured
        #[cfg(feature = "audit")]
        let audit_logger: Option<crate::audit::AuditLogger> = {
            let audit_config = config.audit.clone().unwrap_or_default();
            if audit_config.enabled {
                let service_name = config.service.name.clone();

                // Attach DB-backed storage for whichever database feature is active.
                // Pool agents connect asynchronously, so the selected backend resolves
                // lazily: it holds the shared pool handle and builds the concrete
                // storage (running its append-only DDL) on first use, once connected.
                // Until then the agent runs on the in-memory chain + syslog.
                let storage: Option<std::sync::Arc<dyn crate::audit::AuditStorage>> =
                    crate::audit::wiring::select_audit_storage(
                        &crate::audit::wiring::AuditStorageHandles {
                            #[cfg(feature = "database")]
                            db_pool: shared_db_pool.clone(),
                            #[cfg(feature = "turso")]
                            turso_db: shared_turso_db.clone(),
                            #[cfg(feature = "surrealdb")]
                            surrealdb_client: shared_surrealdb_client.clone(),
                            #[cfg(feature = "clickhouse")]
                            clickhouse_client: shared_clickhouse_client.clone(),
                        },
                    );

                match runtime_supports_block_in_place() {
                    Some(true) => tokio::task::block_in_place(|| {
                        tokio::runtime::Handle::current().block_on(async {
                            // Initialize agent runtime if not already done
                            if self.agent_runtime.is_none() {
                                self.agent_runtime =
                                    Some(acton_reactive::prelude::ActonApp::launch_async().await);
                            }

                            if let Some(ref mut runtime) = self.agent_runtime {
                                let logger_config = audit_config.clone();
                                match crate::audit::AuditAgent::spawn(
                                    runtime,
                                    audit_config,
                                    storage,
                                    service_name.clone(),
                                )
                                .await
                                {
                                    Ok(handle) => {
                                        tracing::info!("Audit agent spawned");
                                        Some(crate::audit::AuditLogger::new(
                                            handle,
                                            service_name,
                                            logger_config,
                                        ))
                                    }
                                    Err(e) => {
                                        tracing::error!("Failed to spawn audit agent: {}", e);
                                        None
                                    }
                                }
                            } else {
                                tracing::warn!("No agent runtime available for audit agent");
                                None
                            }
                        })
                    }),
                    Some(false) => {
                        let err = crate::error::Error::Internal(
                            "audit logging is enabled (the default) but the tokio runtime is \
                             single-threaded, so the audit agent cannot be spawned from \
                             ServiceBuilder::build(); use a multi-threaded runtime \
                             (#[tokio::main], or #[tokio::test(flavor = \"multi_thread\")] in \
                             tests) or set [audit] enabled = false"
                                .to_string(),
                        );
                        tracing::error!("{}", err);
                        record_startup_error(&mut startup_error, err);
                        None
                    }
                    None => {
                        tracing::warn!("No tokio runtime available for audit agent");
                        None
                    }
                }
            } else {
                None
            }
        };

        // Config fingerprinting and ConfigLoaded audit event (NIST CM-3)
        #[cfg(feature = "audit")]
        let config_fingerprint: Option<String> = {
            if let Some(ref logger) = audit_logger {
                let audit_cfg = logger.config();
                if audit_cfg.audit_config_events {
                    match serde_json::to_value(&config) {
                        Ok(config_value) => {
                            let redacted = crate::audit::config_audit::redact_config(&config_value);
                            let fingerprint =
                                crate::audit::config_audit::compute_config_fingerprint(&redacted);

                            let event = crate::audit::config_audit::build_config_loaded_event(
                                &config.service.name,
                                &fingerprint,
                                &redacted,
                                &config.service.environment,
                            );

                            let logger_clone = logger.clone();
                            // Transitively guarded: an `AuditLogger` only exists
                            // when the audit-agent spawn above took its
                            // `Some(true)` (multi-threaded) branch, so the
                            // runtime flavor is already known to be safe here.
                            tokio::task::block_in_place(|| {
                                tokio::runtime::Handle::current()
                                    .block_on(async { logger_clone.log(event).await })
                            });

                            tracing::info!(
                                config_hash = %fingerprint,
                                "ConfigLoaded audit event emitted (CM-3)"
                            );

                            Some(fingerprint)
                        }
                        Err(e) => {
                            tracing::warn!(
                                "Failed to serialize config for audit fingerprint: {}",
                                e
                            );
                            None
                        }
                    }
                } else {
                    None
                }
            } else {
                None
            }
        };

        // Spawn key rotation agent if configured (requires auth + a DB backend)
        #[cfg(feature = "auth")]
        let key_manager: Option<std::sync::Arc<crate::auth::key_rotation::KeyManager>> = {
            let kr_enabled = config
                .auth
                .as_ref()
                .and_then(|a| a.key_rotation.as_ref())
                .is_some_and(|kr| kr.enabled);

            if kr_enabled {
                Self::init_key_rotation(
                    &mut self,
                    &config,
                    #[cfg(feature = "database")]
                    &shared_db_pool,
                    #[cfg(feature = "turso")]
                    &shared_turso_db,
                    #[cfg(feature = "surrealdb")]
                    &shared_surrealdb_client,
                    #[cfg(feature = "audit")]
                    &audit_logger,
                    &mut startup_error,
                )
            } else {
                None
            }
        };

        // Spawn background worker agent if configured
        let background_worker: Option<crate::agents::BackgroundWorker> = {
            let bw_config = config.background_worker.clone().unwrap_or_default();
            if bw_config.enabled {
                match runtime_supports_block_in_place() {
                    Some(true) => tokio::task::block_in_place(|| {
                        tokio::runtime::Handle::current().block_on(async {
                            if self.agent_runtime.is_none() {
                                self.agent_runtime =
                                    Some(acton_reactive::prelude::ActonApp::launch_async().await);
                            }
                            if let Some(ref mut runtime) = self.agent_runtime {
                                match crate::agents::BackgroundWorker::spawn(runtime, &bw_config)
                                    .await
                                {
                                    Ok(worker) => {
                                        tracing::info!("Background worker agent spawned");
                                        Some(worker)
                                    }
                                    Err(e) => {
                                        tracing::error!("Failed to spawn background worker: {}", e);
                                        None
                                    }
                                }
                            } else {
                                tracing::warn!("No agent runtime available for background worker");
                                None
                            }
                        })
                    }),
                    Some(false) => {
                        let err = crate::error::Error::Internal(
                            "the background worker is enabled but the tokio runtime is \
                             single-threaded, so its agent cannot be spawned from \
                             ServiceBuilder::build(); use a multi-threaded runtime \
                             (#[tokio::main], or #[tokio::test(flavor = \"multi_thread\")] in \
                             tests) or set [background_worker] enabled = false"
                                .to_string(),
                        );
                        tracing::error!("{}", err);
                        record_startup_error(&mut startup_error, err);
                        None
                    }
                    None => {
                        tracing::warn!("No tokio runtime available for background worker");
                        None
                    }
                }
            } else {
                None
            }
        };

        // Spawn actor extensions under a supervisor if any were registered
        let pending_extensions = std::mem::take(&mut self.actor_extensions);
        let actor_extensions = if !pending_extensions.is_empty() {
            match runtime_supports_block_in_place() {
                Some(true) => tokio::task::block_in_place(|| {
                    tokio::runtime::Handle::current().block_on(async {
                        // Initialize agent runtime if not already done (e.g., no pool features enabled)
                        if self.agent_runtime.is_none() {
                            tracing::debug!(
                                "Initializing acton-reactive agent runtime for actor extensions"
                            );
                            self.agent_runtime =
                                Some(acton_reactive::prelude::ActonApp::launch_async().await);
                        }

                        let runtime = self.init_agent_runtime();

                        // Spawn the extensions supervisor
                        let supervisor =
                            runtime.new_actor::<crate::extensions::ExtensionsSupervisorState>();
                        let supervisor_handle = supervisor.start().await;
                        tracing::info!("Extensions supervisor spawned");

                        // Spawn each user actor under supervision
                        let mut handles = std::collections::HashMap::new();
                        for ext in &pending_extensions {
                            match ext.spawn(&supervisor_handle, runtime).await {
                                Ok((type_id, handle)) => {
                                    handles.insert(type_id, handle);
                                }
                                Err(e) => {
                                    tracing::error!("Failed to spawn actor extension: {}", e);
                                }
                            }
                        }

                        crate::extensions::ActorExtensions::from(handles)
                    })
                }),
                Some(false) => {
                    let err = crate::error::Error::Internal(
                        "actor extensions are registered but the tokio runtime is \
                         single-threaded, so they cannot be spawned from \
                         ServiceBuilder::build(); use a multi-threaded runtime \
                         (#[tokio::main], or #[tokio::test(flavor = \"multi_thread\")] in \
                         tests) or stop registering actor extensions"
                            .to_string(),
                    );
                    tracing::error!("{}", err);
                    record_startup_error(&mut startup_error, err);
                    crate::extensions::ActorExtensions::default()
                }
                None => {
                    tracing::warn!("No tokio runtime available, skipping actor extension spawning");
                    crate::extensions::ActorExtensions::default()
                }
            }
        } else {
            crate::extensions::ActorExtensions::default()
        };

        // When no pool features are enabled, the runtime may still have been
        // initialized for actor extensions. Provide the broker if available.
        #[cfg(not(any(
            feature = "database",
            feature = "cache",
            feature = "events",
            feature = "turso",
            feature = "surrealdb",
            feature = "clickhouse"
        )))]
        let broker_handle: Option<acton_reactive::prelude::ActorHandle> = self.broker();

        // Pool features enabled, but no pool agents needed (e.g. only actor
        // extensions). The actor-extension spawn block above may have
        // initialized the runtime; pick up the broker now if it is available.
        #[cfg(any(
            feature = "database",
            feature = "cache",
            feature = "events",
            feature = "turso",
            feature = "surrealdb",
            feature = "clickhouse"
        ))]
        if broker_handle.is_none() {
            broker_handle = self.broker();
        }

        let routes = self.routes.unwrap_or_default();

        // Build AppState with agent-managed pools
        let state = if let Some(provided_state) = self.state {
            provided_state
        } else {
            let mut state = AppState::new(config.clone());

            // Set broker handle for event broadcasting
            if let Some(broker) = broker_handle {
                state.set_broker(broker);
            }

            // Set shared pool storage (agents will update these when connected)
            #[cfg(feature = "database")]
            if let Some(pool) = shared_db_pool {
                state.set_db_pool_storage(pool);
            }

            #[cfg(feature = "cache")]
            if let Some(pool) = shared_redis_pool {
                state.set_redis_pool_storage(pool);
            }

            #[cfg(feature = "events")]
            if let Some(client) = shared_nats_client {
                state.set_nats_client_storage(client);
            }

            #[cfg(feature = "turso")]
            if let Some(db) = shared_turso_db {
                state.set_turso_db_storage(db);
            }

            #[cfg(feature = "surrealdb")]
            if let Some(client) = shared_surrealdb_client {
                state.set_surrealdb_client_storage(client);
            }

            #[cfg(feature = "clickhouse")]
            if let Some(client) = shared_clickhouse_client {
                state.set_clickhouse_client_storage(client);
            }

            #[cfg(feature = "audit")]
            if let Some(ref logger) = audit_logger {
                state.set_audit_logger(logger.clone());
            }

            #[cfg(feature = "audit")]
            if let Some(ref fp) = config_fingerprint {
                state.set_config_fingerprint(fp.clone());
            }

            #[cfg(feature = "auth")]
            if let Some(ref km) = key_manager {
                state.set_key_manager(km.clone());
            }

            if let Some(ref worker) = background_worker {
                state.set_background_worker(worker.clone());
            }

            if !actor_extensions.is_empty() {
                state.set_actor_extensions(actor_extensions);
            }

            state
        };

        // Clone state before it's consumed by Router::with_state().
        // AppState uses Arc internally, so this is a cheap reference-count bump.
        let state_clone = state.clone();

        // Handle both types of versioned routes
        let app = match routes {
            VersionedRoutes::WithState(router) => {
                // Health routes already added, just attach state
                #[cfg(feature = "audit")]
                let router = router.route(
                    "/admin/config/drift",
                    axum::routing::get(crate::audit::config_audit::drift_check_handler::<T>),
                );
                #[cfg(feature = "prometheus-metrics")]
                let router = router.route(
                    "/metrics",
                    axum::routing::get(crate::observability::metrics_handler),
                );
                router.with_state(state)
            }
            VersionedRoutes::WithoutState(router) => {
                // Add health routes and attach state
                use axum::routing::get;
                let health_router: Router<AppState<T>> = Router::new()
                    .route("/health", get(crate::health::health))
                    .route("/ready", get(crate::health::readiness));

                #[cfg(feature = "audit")]
                let health_router = health_router.route(
                    "/admin/config/drift",
                    get(crate::audit::config_audit::drift_check_handler::<T>),
                );

                #[cfg(feature = "prometheus-metrics")]
                let health_router =
                    health_router.route("/metrics", get(crate::observability::metrics_handler));

                // Use fallback_service to include the versioned routes
                let router_with_health = health_router.fallback_service(router);
                router_with_health.with_state(state)
            }
        };

        // Resolve Cedar (if configured) up front so the GraphQL transport can
        // share the same instance via resolver-level checks. We move it out of
        // `self` here; the HTTP middleware section below consumes this local.
        #[cfg(feature = "cedar-authz")]
        let cedar_authz: Option<crate::middleware::cedar::CedarAuthz> = {
            if let Some(cedar) = self.cedar.take() {
                tracing::debug!("Using explicit Cedar authorization middleware");
                Some(cedar)
            } else if let Some(ref cedar_config) = config.cedar {
                if cedar_config.enabled {
                    match runtime_supports_block_in_place() {
                        Some(true) => {
                            let cedar_path_normalizer = self.cedar_path_normalizer;
                            match tokio::task::block_in_place(|| {
                                tokio::runtime::Handle::current().block_on(async {
                                    let mut builder = crate::middleware::cedar::CedarAuthz::builder(
                                        cedar_config.clone(),
                                    );
                                    if let Some(normalizer) = cedar_path_normalizer {
                                        builder = builder.with_path_normalizer(normalizer);
                                    }
                                    builder.build().await
                                })
                            }) {
                                Ok(cedar) => {
                                    if cedar_path_normalizer.is_some() {
                                        tracing::debug!("Auto-configured Cedar authorization middleware with custom path normalizer");
                                    } else {
                                        tracing::debug!("Auto-configured Cedar authorization middleware with default path normalizer");
                                    }
                                    Some(cedar)
                                }
                                Err(e) => {
                                    let err = crate::error::Error::Internal(format!(
                                        "Cedar authorization is enabled but its initialization \
                                         failed; refusing to start rather than serving routes \
                                         without policy enforcement: {e}"
                                    ));
                                    tracing::error!("{}", err);
                                    record_startup_error(&mut startup_error, err);
                                    None
                                }
                            }
                        }
                        Some(false) => {
                            let err = crate::error::Error::Internal(
                                "Cedar authorization is enabled but the tokio runtime is \
                                 single-threaded, so its policy set cannot be loaded from \
                                 ServiceBuilder::build(); use a multi-threaded runtime \
                                 (#[tokio::main], or #[tokio::test(flavor = \"multi_thread\")] \
                                 in tests) or set [cedar] enabled = false. Refusing to start \
                                 rather than serving routes without policy enforcement"
                                    .to_string(),
                            );
                            tracing::error!("{}", err);
                            record_startup_error(&mut startup_error, err);
                            None
                        }
                        None => {
                            let err = crate::error::Error::Internal(
                                "Cedar authorization is enabled but no tokio runtime is \
                                 available to initialize it; refusing to start rather than \
                                 serving routes without policy enforcement"
                                    .to_string(),
                            );
                            tracing::error!("{}", err);
                            record_startup_error(&mut startup_error, err);
                            None
                        }
                    }
                } else {
                    None
                }
            } else {
                None
            }
        };

        // Keep a Cedar handle for the gRPC routes; the HTTP middleware
        // section below consumes `cedar_authz` itself.
        #[cfg(all(feature = "grpc", feature = "cedar-authz"))]
        let grpc_cedar = cedar_authz.clone();

        // Merge GraphQL routes (if any) BEFORE general middleware so GraphQL
        // endpoints inherit auth, tracing, CORS, etc.
        #[cfg(feature = "graphql")]
        let app = if let Some(graphql) = self.graphql.take() {
            let graphql_enabled = config.graphql.as_ref().map(|g| g.enabled).unwrap_or(true);
            if graphql_enabled {
                match crate::graphql::mount::build_router(
                    graphql,
                    config.graphql.as_ref(),
                    #[cfg(feature = "graphql-cedar")]
                    cedar_authz.clone(),
                ) {
                    Some(graphql_router) => {
                        tracing::debug!("Merging GraphQL transport routes");
                        app.merge(graphql_router)
                    }
                    None => app,
                }
            } else {
                tracing::debug!("GraphQL transport configured but disabled by config");
                app
            }
        } else {
            app
        };

        // Apply general middleware stack (CORS, compression, timeout, TraceLayer, etc.)
        // Layers are applied in reverse order (bottom layer is innermost/first)
        // Whether the HTTP listener will terminate TLS. An injected config
        // activates TLS independently of the `[tls]` section, so both sources
        // count — otherwise HSTS would be dropped on encrypted connections.
        #[cfg(feature = "tls")]
        let tls_active =
            self.tls_config_override.is_some() || config.tls.as_ref().is_some_and(|t| t.enabled);
        #[cfg(not(feature = "tls"))]
        let tls_active = false;

        let mut app = Self::apply_middleware(app, &config, tls_active);

        // Apply session middleware if configured
        // Session runs after general middleware, before JWT/Cedar
        // This allows session-based auth and JWT auth to coexist
        #[cfg(feature = "session")]
        if let Some(ref session_config) = config.session {
            use crate::session::SessionStorage;

            match session_config.storage {
                #[cfg(feature = "session-memory")]
                SessionStorage::Memory => {
                    use crate::session::create_memory_session_layer;
                    tracing::info!("Initializing in-memory session store");
                    let session_layer = create_memory_session_layer(session_config);
                    app = app.layer(session_layer);
                }
                #[cfg(feature = "session-redis")]
                SessionStorage::Redis => {
                    use crate::session::create_redis_session_layer;
                    if let Some(ref redis_url) = session_config.redis_url {
                        tracing::info!("Initializing Redis session store");
                        // Checked before any connection attempt, so a
                        // single-threaded runtime fails fast rather than
                        // reaching Redis and then panicking in tokio.
                        match runtime_supports_block_in_place() {
                            Some(true) => {
                                let session_config_clone = session_config.clone();
                                let redis_url_clone = redis_url.clone();
                                match tokio::task::block_in_place(|| {
                                    tokio::runtime::Handle::current().block_on(async {
                                        create_redis_session_layer(
                                            &session_config_clone,
                                            &redis_url_clone,
                                        )
                                        .await
                                    })
                                }) {
                                    Ok(session_layer) => {
                                        app = app.layer(session_layer);
                                    }
                                    Err(e) => {
                                        tracing::error!(
                                            "Failed to create Redis session store: {}",
                                            e
                                        );
                                    }
                                }
                            }
                            Some(false) => {
                                let err = crate::error::Error::Internal(
                                    "Redis-backed sessions are enabled but the tokio runtime \
                                     is single-threaded, so the session store cannot be \
                                     initialized from ServiceBuilder::build(); use a \
                                     multi-threaded runtime (#[tokio::main], or \
                                     #[tokio::test(flavor = \"multi_thread\")] in tests) or \
                                     set [session] storage = \"memory\""
                                        .to_string(),
                                );
                                tracing::error!("{}", err);
                                record_startup_error(&mut startup_error, err);
                            }
                            None => {
                                tracing::error!(
                                    "No tokio runtime available for Redis session initialization"
                                );
                            }
                        }
                    } else {
                        tracing::error!(
                            "Redis session storage configured but redis_url is missing"
                        );
                    }
                }
                #[cfg(not(feature = "session-memory"))]
                SessionStorage::Memory => {
                    tracing::error!("Memory session storage requested but 'session-memory' feature is not enabled");
                }
                #[cfg(not(feature = "session-redis"))]
                SessionStorage::Redis => {
                    tracing::error!("Redis session storage requested but 'session-redis' feature is not enabled");
                }
            }
        }

        // Auto-apply Cedar middleware if available (resolved earlier so the
        // GraphQL transport can share the same instance for resolver checks).
        // NOTE: Cedar must be applied BEFORE JWT because Axum layers run in reverse order
        // This ensures the execution order is: Request → General Middleware → JWT → Cedar → Handler
        #[cfg(feature = "cedar-authz")]
        if let Some(cedar) = cedar_authz {
            app = app.layer(axum::middleware::from_fn_with_state(
                cedar,
                crate::middleware::cedar::CedarAuthz::middleware,
            ));
        }

        // Apply audit middleware if configured
        // NOTE: Applied BEFORE token auth in layer order, so it runs AFTER token auth.
        // Execution order: Request → General MW → Token Auth → Audit MW → Cedar → Handler
        // This ensures Claims are available when the audit middleware runs.
        #[cfg(feature = "audit")]
        if let Some(ref logger) = audit_logger {
            app = app.layer(axum::middleware::from_fn_with_state(
                logger.clone(),
                crate::audit::middleware::audit_middleware,
            ));
        }

        // Auto-apply governor rate-limit middleware if enabled.
        //
        // Layer order rationale: governor is applied here, AFTER cedar (above)
        // and BEFORE token auth (below) in source order. Because axum applies
        // layers in reverse, the runtime order becomes:
        //
        //     Request -> General MW -> Token Auth -> Audit -> Cedar -> Governor -> Handler
        //
        // This ensures Claims (set by token auth) are visible to the governor
        // middleware. The layer is attached to the OUTER router, so
        // `request.uri().path()` sees the full pre-nest path -- route keys like
        // "POST /api/v1/uploads" match as documented.
        #[cfg(feature = "governor")]
        if config.rate_limit.auto_apply {
            let gov =
                crate::middleware::governor::GovernorRateLimit::new(config.rate_limit.clone());
            tracing::debug!("Auto-applying governor rate-limit middleware");
            app = app.layer(axum::middleware::from_fn_with_state(
                gov,
                crate::middleware::governor::GovernorRateLimit::middleware,
            ));
        }

        // Auto-apply token authentication middleware if configured
        // NOTE: Token auth must be applied AFTER Cedar because Axum layers run in reverse order
        // This ensures the execution order is: Request → General Middleware → Token Auth → Cedar → Handler
        //
        // The constructed validators are also kept for the gRPC routes below,
        // so both listeners enforce the same identity configuration.
        #[cfg(feature = "grpc")]
        let mut grpc_paseto: Option<crate::middleware::paseto::PasetoAuth> = None;
        #[cfg(all(feature = "grpc", feature = "jwt"))]
        let mut grpc_jwt: Option<crate::middleware::jwt::JwtAuth> = None;
        if let Some(token_config) = &config.token {
            match token_config {
                crate::config::TokenConfig::Paseto(paseto_config) => {
                    match crate::middleware::paseto::PasetoAuth::new(paseto_config) {
                        Ok(paseto_auth) => {
                            #[cfg(feature = "auth")]
                            let paseto_auth = if let Some(ref km) = key_manager {
                                paseto_auth.with_key_manager(km.clone())
                            } else {
                                paseto_auth
                            };
                            #[cfg(feature = "grpc")]
                            {
                                grpc_paseto = Some(paseto_auth.clone());
                            }
                            tracing::debug!("Auto-applying PASETO authentication middleware");
                            app = app.layer(axum::middleware::from_fn_with_state(
                                paseto_auth,
                                crate::middleware::paseto::PasetoAuth::middleware,
                            ));
                        }
                        Err(e) => {
                            let err = crate::error::Error::Internal(format!(
                                "PASETO authentication is configured but its configuration is \
                                 invalid; refusing to start rather than serving authenticated \
                                 routes without authentication: {e}"
                            ));
                            tracing::error!("{}", err);
                            record_startup_error(&mut startup_error, err);
                        }
                    }
                }
                #[cfg(feature = "jwt")]
                crate::config::TokenConfig::Jwt(jwt_config) => {
                    match crate::middleware::jwt::JwtAuth::new(jwt_config) {
                        Ok(jwt_auth) => {
                            #[cfg(feature = "auth")]
                            let jwt_auth = if let Some(ref km) = key_manager {
                                jwt_auth.with_key_manager(km.clone())
                            } else {
                                jwt_auth
                            };
                            #[cfg(feature = "grpc")]
                            {
                                grpc_jwt = Some(jwt_auth.clone());
                            }
                            tracing::debug!("Auto-applying JWT authentication middleware");
                            app = app.layer(axum::middleware::from_fn_with_state(
                                jwt_auth,
                                crate::middleware::jwt::JwtAuth::middleware,
                            ));
                        }
                        Err(e) => {
                            let err = crate::error::Error::Internal(format!(
                                "JWT authentication is configured but its configuration is \
                                 invalid; refusing to start rather than serving authenticated \
                                 routes without authentication: {e}"
                            ));
                            tracing::error!("{}", err);
                            record_startup_error(&mut startup_error, err);
                        }
                    }
                }
            }
        }

        // Inject AuditLogger as a request extension so auth middleware can access it.
        // Applied last in layer order (runs first in execution), making it available
        // to all subsequent middleware including token auth.
        #[cfg(feature = "audit")]
        if let Some(ref logger) = audit_logger {
            app = app.layer(axum::Extension(logger.clone()));
        }

        let listener_addr = std::net::SocketAddr::new(config.service.bind, config.service.port);

        // Resolve the HTTP TLS config. A `[tls]` section with `enabled = true`
        // is the operator's explicit statement of intended posture, so a load
        // failure is fatal: it is recorded in `startup_error` and surfaced by
        // `try_build()` / `serve()` before any listener binds. Serving plaintext
        // where TLS was configured would silently invert the fail-safe direction.
        #[cfg(feature = "tls")]
        let tls_config = {
            if let Some(source) = self.tls_config_override.take() {
                tracing::info!("TLS configured from caller-supplied credentials");
                Some(source)
            } else if let Some(ref tls_cfg) = config.tls {
                if tls_cfg.enabled {
                    // A config-driven source remembers its paths, so the service
                    // can rotate these credentials later without a restart.
                    match crate::tls::TlsConfigSource::from_tls_config(tls_cfg) {
                        Ok(sc) => {
                            tracing::info!(
                                "TLS configured (cert: {}, key: {})",
                                tls_cfg.cert_path.display(),
                                tls_cfg.key_path.display()
                            );
                            Some(sc)
                        }
                        Err(e) => {
                            let err = crate::error::Error::Internal(format!(
                                "TLS is enabled but its configuration failed to load; \
                                 refusing to start a plaintext listener: {e}"
                            ));
                            tracing::error!("{}", err);
                            record_startup_error(&mut startup_error, err);
                            None
                        }
                    }
                } else {
                    None
                }
            } else {
                None
            }
        };

        // Resolve per-listener TLS for the separate-port gRPC listener.
        // A present `[grpc.tls]` section is authoritative for the gRPC
        // listener — `enabled = false` means plaintext gRPC even when the
        // shared `[tls]` is active. Only an absent section inherits it.
        // As with HTTP above, a failed load of an enabled section is fatal
        // rather than a silent downgrade to plaintext.
        #[cfg(all(feature = "grpc", feature = "tls"))]
        let grpc_tls_config = if let Some(source) = self.grpc_tls_config_override.take() {
            tracing::info!("gRPC TLS configured from caller-supplied credentials");
            Some(source)
        } else {
            match config.grpc.as_ref().and_then(|g| g.tls.as_ref()) {
                Some(grpc_tls) if grpc_tls.enabled => {
                    match crate::tls::TlsConfigSource::from_tls_config(grpc_tls) {
                        Ok(sc) => {
                            tracing::info!(
                                "gRPC TLS configured (cert: {}, key: {})",
                                grpc_tls.cert_path.display(),
                                grpc_tls.key_path.display()
                            );
                            Some(sc)
                        }
                        Err(e) => {
                            let err = crate::error::Error::Internal(format!(
                                "gRPC TLS is enabled but its configuration failed to load; \
                                 refusing to start a plaintext gRPC listener: {e}"
                            ));
                            tracing::error!("{}", err);
                            record_startup_error(&mut startup_error, err);
                            None
                        }
                    }
                }
                Some(_) => {
                    tracing::info!(
                        "gRPC TLS explicitly disabled; separate-port gRPC listener serves plaintext"
                    );
                    None
                }
                None => tls_config.clone(),
            }
        };

        // Resolve the built-in credential-rotation triggers. Validation happens
        // here so an interval that would spin the runtime is reported by
        // `try_build()` / `serve()` before anything binds, in the same way a
        // failed TLS load is.
        #[cfg(feature = "tls")]
        let tls_reload_interval = match config.tls.as_ref().filter(|t| t.enabled) {
            Some(tls_cfg) => match crate::tls::validate_reload_interval(tls_cfg, "[tls]") {
                Ok(interval) => {
                    crate::tls::warn_if_reload_config_is_unusable(
                        tls_config.as_ref(),
                        tls_cfg,
                        "[tls]",
                    );
                    interval
                }
                Err(err) => {
                    tracing::error!("{}", err);
                    record_startup_error(&mut startup_error, err);
                    None
                }
            },
            None => None,
        };

        #[cfg(all(feature = "grpc", feature = "tls"))]
        let grpc_tls_reload_interval = match config
            .grpc
            .as_ref()
            .and_then(|g| g.tls.as_ref())
            .filter(|t| t.enabled)
        {
            Some(grpc_tls) => match crate::tls::validate_reload_interval(grpc_tls, "[grpc.tls]") {
                Ok(interval) => {
                    crate::tls::warn_if_reload_config_is_unusable(
                        grpc_tls_config.as_ref(),
                        grpc_tls,
                        "[grpc.tls]",
                    );
                    interval
                }
                Err(err) => {
                    tracing::error!("{}", err);
                    record_startup_error(&mut startup_error, err);
                    None
                }
            },
            None => None,
        };

        // Either section asking for SIGHUP installs the one handler that
        // reloads every listener; see `crate::tls::spawn_sighup_reload`.
        #[cfg(feature = "tls")]
        let tls_reload_on_sighup = {
            let http = config
                .tls
                .as_ref()
                .is_some_and(|t| t.enabled && t.reload_on_sighup);
            #[cfg(feature = "grpc")]
            let grpc = config
                .grpc
                .as_ref()
                .and_then(|g| g.tls.as_ref())
                .is_some_and(|t| t.enabled && t.reload_on_sighup);
            #[cfg(not(feature = "grpc"))]
            let grpc = false;
            http || grpc
        };

        // Convert gRPC routes to an axum router and apply their own auth and
        // authorization stack. Merged routers do not inherit each other's
        // layers, so without this the gRPC listener would serve every method
        // unauthenticated regardless of what `[token]` and `[cedar]` declare.
        // Token auth (outermost) validates the `authorization` metadata and
        // injects Claims; Cedar (inner) then authorizes each method. Health
        // and reflection services are exempt, as are `public_paths` prefixes.
        // Registering gRPC services that no listener will ever expose is a
        // misconfiguration, not a preference. Caught here, before the routes
        // are consumed, so `try_build()` / `serve()` report it without binding.
        #[cfg(feature = "grpc")]
        if let Some(err) =
            grpc_registration_error(self.grpc_services.is_some(), config.grpc.as_ref())
        {
            tracing::error!("{}", err);
            record_startup_error(&mut startup_error, err);
        }

        #[cfg(feature = "grpc")]
        let grpc_router: Option<axum::Router> = match self.grpc_services {
            Some(routes) => {
                let mut grpc_app = routes.into_axum_router();

                #[cfg(feature = "cedar-authz")]
                if let Some(cedar) = grpc_cedar {
                    tracing::debug!("Auto-applying Cedar authorization to gRPC routes");
                    grpc_app =
                        grpc_app.layer(crate::middleware::cedar::CedarAuthzLayer::new(cedar));
                }

                if let Some(paseto) = grpc_paseto {
                    tracing::debug!("Auto-applying PASETO authentication to gRPC routes");
                    let public_paths = match &config.token {
                        Some(crate::config::TokenConfig::Paseto(c)) => c.public_paths.clone(),
                        _ => Vec::new(),
                    };
                    grpc_app = grpc_app.layer(
                        crate::grpc::middleware::GrpcTokenAuthLayer::new(paseto)
                            .with_public_paths(public_paths),
                    );
                }
                #[cfg(feature = "jwt")]
                if let Some(jwt) = grpc_jwt {
                    tracing::debug!("Auto-applying JWT authentication to gRPC routes");
                    let public_paths = match &config.token {
                        Some(crate::config::TokenConfig::Jwt(c)) => c.public_paths.clone(),
                        _ => Vec::new(),
                    };
                    grpc_app = grpc_app.layer(
                        crate::grpc::middleware::GrpcTokenAuthLayer::new(jwt)
                            .with_public_paths(public_paths),
                    );
                }

                // Make the audit logger visible to the Cedar layer for denial
                // events. Outermost, so it is inserted before auth/Cedar run.
                #[cfg(feature = "audit")]
                if let Some(ref logger) = audit_logger {
                    grpc_app = grpc_app.layer(axum::Extension(logger.clone()));
                }

                Some(grpc_app)
            }
            None => None,
        };

        ActonService {
            config,
            state: state_clone,
            listener_addr,
            app,
            #[cfg(feature = "grpc")]
            grpc_routes: grpc_router,
            #[cfg(feature = "tls")]
            tls_config,
            #[cfg(all(feature = "grpc", feature = "tls"))]
            grpc_tls_config,
            #[cfg(feature = "tls")]
            tls_reload_hook: self.tls_reload_hook.take(),
            #[cfg(feature = "tls")]
            tls_reload_interval,
            #[cfg(all(feature = "grpc", feature = "tls"))]
            grpc_tls_reload_interval,
            #[cfg(feature = "tls")]
            tls_reload_on_sighup,
            agent_runtime: self.agent_runtime,
            startup_error,
        }
    }

    /// Build the service, returning an error on fatal misconfiguration.
    ///
    /// Identical to [`build`](Self::build) except that a security-relevant
    /// misconfiguration is reported here rather than at [`ActonService::serve`].
    /// Prefer this when the caller wants to observe the failure before the
    /// service is handed off. The conditions are the same either way — a
    /// configured-but-unloadable TLS section, or invalid token-auth config —
    /// and in neither case does a listener bind.
    ///
    /// # Errors
    ///
    /// Returns the first fatal misconfiguration encountered during the build.
    pub fn try_build(self) -> crate::error::Result<ActonService<T>> {
        let mut service = self.build();
        match service.startup_error.take() {
            Some(e) => Err(e),
            None => Ok(service),
        }
    }

    /// Initialize key rotation subsystem if enabled and a DB backend is available
    ///
    /// Creates the appropriate storage backend, initializes the KeyManager,
    /// runs storage initialization (table creation), refreshes the key cache,
    /// and spawns the KeyRotationAgent for periodic rotation checks.
    ///
    /// Returns `Some(Arc<KeyManager>)` on success, `None` if no DB backend is
    /// available or initialization fails.
    #[cfg(feature = "auth")]
    #[allow(unused_variables)]
    fn init_key_rotation(
        builder: &mut Self,
        config: &Config<T>,
        #[cfg(feature = "database")] shared_db_pool: &Option<crate::agents::SharedDbPool>,
        #[cfg(feature = "turso")] shared_turso_db: &Option<crate::agents::SharedTursoDb>,
        #[cfg(feature = "surrealdb")] shared_surrealdb_client: &Option<
            crate::agents::SharedSurrealDb,
        >,
        #[cfg(feature = "audit")] audit_logger: &Option<crate::audit::AuditLogger>,
        startup_error: &mut Option<crate::error::Error>,
    ) -> Option<std::sync::Arc<crate::auth::key_rotation::KeyManager>> {
        let kr_config = config
            .auth
            .as_ref()
            .and_then(|a| a.key_rotation.as_ref())
            .cloned()?;

        // Checked up front, before storage resolution, so an unusable runtime
        // is reported even when no database backend has connected yet.
        match runtime_supports_block_in_place() {
            Some(true) => {}
            Some(false) => {
                let err = crate::error::Error::Internal(
                    "key rotation is enabled but the tokio runtime is single-threaded, so its \
                     storage cannot be initialized and its agent cannot be spawned from \
                     ServiceBuilder::build(); use a multi-threaded runtime (#[tokio::main], or \
                     #[tokio::test(flavor = \"multi_thread\")] in tests) or set \
                     [auth.key_rotation] enabled = false"
                        .to_string(),
                );
                tracing::error!("{}", err);
                record_startup_error(startup_error, err);
                return None;
            }
            None => {
                tracing::warn!("No tokio runtime available for key rotation initialization");
                return None;
            }
        }

        let service_name = config.service.name.clone();

        // Try to create a storage backend from available DB pools.
        // The pool may not be connected yet (agents connect asynchronously),
        // so we attempt to read the shared lock. If the pool is not available
        // yet we log a warning and skip key rotation initialization -- the
        // KeyRotationAgent's periodic tick will retry once pools are ready.
        let storage: Option<std::sync::Arc<dyn crate::auth::key_rotation::KeyRotationStorage>> = {
            #[allow(unused_mut)]
            let mut s: Option<
                std::sync::Arc<dyn crate::auth::key_rotation::KeyRotationStorage>,
            > = None;

            #[cfg(feature = "database")]
            if s.is_none() {
                if let Some(ref pool_lock) = shared_db_pool {
                    if let Ok(guard) = pool_lock.try_read() {
                        if let Some(ref pool) = *guard {
                            s = Some(std::sync::Arc::new(
                                crate::auth::key_rotation::PgKeyRotationStorage::new(pool.clone()),
                            ));
                            tracing::debug!("Key rotation using PostgreSQL storage");
                        }
                    }
                }
            }

            #[cfg(feature = "turso")]
            if s.is_none() {
                if let Some(ref db_lock) = shared_turso_db {
                    if let Ok(guard) = db_lock.try_read() {
                        if let Some(ref db) = *guard {
                            s = Some(std::sync::Arc::new(
                                crate::auth::key_rotation::TursoKeyRotationStorage::new(db.clone()),
                            ));
                            tracing::debug!("Key rotation using Turso storage");
                        }
                    }
                }
            }

            #[cfg(feature = "surrealdb")]
            if s.is_none() {
                if let Some(ref client_lock) = shared_surrealdb_client {
                    if let Ok(guard) = client_lock.try_read() {
                        if let Some(ref client) = *guard {
                            s = Some(std::sync::Arc::new(
                                crate::auth::key_rotation::SurrealKeyRotationStorage::new(
                                    client.clone(),
                                ),
                            ));
                            tracing::debug!("Key rotation using SurrealDB storage");
                        }
                    }
                }
            }

            s
        };

        let storage = match storage {
            Some(s) => s,
            None => {
                tracing::warn!(
                    "Key rotation enabled but no database pool is available yet. \
                     Key rotation will not start until a database backend connects."
                );
                return None;
            }
        };

        // Create key manager and initialize storage + cache
        let key_manager = crate::auth::key_rotation::KeyManager::new(
            storage,
            service_name.clone(),
            kr_config.clone(),
        );

        // The runtime flavor was verified above, so `block_in_place` is safe here.
        let init_result = tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                // Initialize storage (create tables/indexes)
                key_manager.storage().initialize().await?;

                // Populate the in-memory cache from storage
                key_manager.refresh_cache().await?;

                Ok::<(), crate::error::Error>(())
            })
        });

        if let Err(e) = init_result {
            tracing::error!("Failed to initialize key rotation storage: {}", e);
            return None;
        }

        let km_arc = std::sync::Arc::new(key_manager.clone());

        // Spawn the KeyRotationAgent. The runtime flavor was verified above.
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                // Initialize agent runtime if not already done
                if builder.agent_runtime.is_none() {
                    builder.agent_runtime =
                        Some(acton_reactive::prelude::ActonApp::launch_async().await);
                }

                if let Some(ref mut runtime) = builder.agent_runtime {
                    match crate::auth::key_rotation::KeyRotationAgent::spawn(
                        runtime,
                        key_manager,
                        kr_config,
                        #[cfg(feature = "audit")]
                        audit_logger.clone(),
                    )
                    .await
                    {
                        Ok(_handle) => {
                            tracing::info!("Key rotation agent spawned");
                        }
                        Err(e) => {
                            tracing::error!("Failed to spawn key rotation agent: {}", e);
                        }
                    }
                }
            });
        });

        Some(km_arc)
    }

    /// Apply middleware stack based on configuration
    ///
    /// Applies middleware in the correct order to ensure proper request handling
    ///
    /// `tls_active` reports whether the HTTP listener will actually terminate
    /// TLS. It is passed in rather than re-derived from `config.tls` because a
    /// caller-supplied config via [`with_tls_config`](Self::with_tls_config)
    /// enables TLS regardless of what the `[tls]` section says; deriving it here
    /// would drop HSTS from connections that are in fact encrypted.
    fn apply_middleware(app: Router, config: &Config<T>, tls_active: bool) -> Router {
        let body_limit = config.middleware.body_limit_mb * 1024 * 1024;

        let mut app = app;

        // CORS (outermost layer) - configurable
        let cors_layer = match config.middleware.cors_mode.as_str() {
            "permissive" => CorsLayer::permissive(),
            "restrictive" => CorsLayer::new(),
            "disabled" => CorsLayer::new(),
            _ => {
                tracing::warn!(
                    "Unknown CORS mode: {}, defaulting to permissive",
                    config.middleware.cors_mode
                );
                CorsLayer::permissive()
            }
        };
        app = app.layer(cors_layer);

        // Security headers (after CORS, before compression)
        app = crate::middleware::security_headers::apply_security_headers(
            app,
            &config.middleware.security_headers,
            tls_active,
        );

        // Compression - configurable
        if config.middleware.compression {
            app = app.layer(CompressionLayer::new());
        }

        // Request timeout
        app = app.layer(TimeoutLayer::with_status_code(
            http::StatusCode::REQUEST_TIMEOUT,
            Duration::from_secs(config.service.timeout_secs),
        ));

        // Request body size limit - configurable
        app = app.layer(RequestBodyLimitLayer::new(body_limit));

        // Tracing (HTTP request/response logging) - always enabled
        app = app.layer(
            TraceLayer::new_for_http()
                .make_span_with(DefaultMakeSpan::new().include_headers(true))
                .on_response(DefaultOnResponse::new().include_headers(true)),
        );

        // HTTP metrics layer (OpenTelemetry) - records into the global meter that
        // feeds the OTLP push and/or Prometheus pull readers. Requires the meter
        // provider to have been initialized in `build()` beforehand.
        #[cfg(feature = "_metrics")]
        if let Some(metrics_cfg) = &config.middleware.metrics {
            if metrics_cfg.enabled {
                let mw_config = crate::middleware::metrics::MetricsConfig::new()
                    .with_service_name(config.service.name.clone())
                    .with_include_path(metrics_cfg.include_path)
                    .with_include_method(metrics_cfg.include_method)
                    .with_include_status(metrics_cfg.include_status)
                    .with_latency_buckets(metrics_cfg.latency_buckets_ms.clone());
                if let Some(layer) = crate::middleware::metrics::create_metrics_layer(&mw_config) {
                    app = app.layer(layer);
                }
            }
        }

        // Request context - resolves client IP, request ID, and user agent once for
        // every downstream consumer. Added here so it executes immediately after the
        // request-tracking layers below (later-added layer = outer = runs first) and
        // before auth/audit, which would otherwise see a request ID that has not been
        // generated yet and no ConnectInfo-derived IP (issue #17).
        app = app.layer(axum::middleware::from_fn(
            crate::middleware::request_context::request_context_middleware,
        ));

        // Request tracking layers - based on config
        if config.middleware.request_tracking.mask_sensitive_headers {
            app = app.layer(sensitive_headers_layer());
        }
        if config.middleware.request_tracking.propagate_headers {
            app = app.layer(request_id_propagation_layer());
        }
        if config.middleware.request_tracking.request_id_enabled {
            app = app.layer(request_id_layer());
        }

        // Panic recovery (innermost layer) - configurable
        if config.middleware.catch_panic {
            app = app.layer(CatchPanicLayer::new());
        }

        app
    }
}

/// Whether the entered tokio runtime supports `tokio::task::block_in_place`.
///
/// `None` when no runtime is entered at all. `Some(false)` on a current-thread
/// runtime, where `block_in_place` panics instead of blocking — callers must
/// check this up front rather than discovering the flavor as a panic inside
/// tokio during `build()`.
///
/// Ungated: the background-worker and actor-extension call sites carry no
/// feature gate of their own, so this is reachable in every build.
fn runtime_supports_block_in_place() -> Option<bool> {
    tokio::runtime::Handle::try_current()
        .ok()
        .map(|handle| handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread)
}

/// Record a fatal build-time misconfiguration, keeping the first one seen.
///
/// The build continues after a failure so that every problem gets logged in one
/// pass rather than one per restart, but only the first error is returned to the
/// caller — it is the one that stopped the intended posture from being honoured.
fn record_startup_error(slot: &mut Option<crate::error::Error>, error: crate::error::Error) {
    if slot.is_none() {
        *slot = Some(error);
    }
}

/// Detect gRPC services registered against a build that can never serve them.
///
/// `with_grpc_services` is an explicit statement that this process serves gRPC.
/// If `[grpc]` is absent or `enabled = false`, those services are unreachable:
/// the listener only ever speaks HTTP, and every RPC the caller registered
/// fails at the client with a transport error that points nowhere near the
/// cause. Dropping them silently is the same fail-unsafe direction as serving
/// plaintext where TLS was configured, so it is fatal at build time instead.
///
/// Pure: derives the verdict solely from its two arguments.
#[cfg(feature = "grpc")]
fn grpc_registration_error(
    services_registered: bool,
    grpc: Option<&crate::config::GrpcConfig>,
) -> Option<crate::error::Error> {
    let grpc_enabled = grpc.is_some_and(|g| g.enabled);
    if !services_registered || grpc_enabled {
        return None;
    }

    let cause = match grpc {
        Some(_) => "the `[grpc]` section sets `enabled = false`",
        None => "no `[grpc]` section is configured",
    };

    Some(crate::error::Error::Internal(format!(
        "gRPC services were registered with `with_grpc_services`, but gRPC is not enabled: \
         {cause}. Those services would be silently discarded and every RPC would fail at the \
         client, so the build is refused. To fix, either set `enabled = true` under `[grpc]` in \
         your config file, or supply a config whose `grpc` field is \
         `Some(GrpcConfig {{ enabled: true, ..Default::default() }})`. If gRPC is not wanted, \
         remove the `with_grpc_services` call."
    )))
}

impl<T> Default for ServiceBuilder<T>
where
    T: Serialize + DeserializeOwned + Clone + Default + Send + Sync + 'static,
{
    fn default() -> Self {
        Self::new()
    }
}

/// Opaque service wrapper
///
/// This type wraps the final Router and Config. It cannot be manipulated
/// directly - the only way to use it is to call `serve()`.
///
/// This prevents developers from:
/// - Adding unversioned routes after construction
/// - Bypassing the type-safe builder
/// - Accessing the internal Router
pub struct ActonService<T = ()>
where
    T: Serialize + DeserializeOwned + Clone + Default + Send + Sync + 'static,
{
    config: Config<T>,
    state: AppState<T>,
    listener_addr: std::net::SocketAddr,
    app: Router,
    #[cfg(feature = "grpc")]
    grpc_routes: Option<axum::Router>,
    #[cfg(feature = "tls")]
    tls_config: Option<crate::tls::TlsConfigSource>,
    /// Per-listener TLS credentials for the separate-port gRPC listener. Falls
    /// back to `tls_config` (the HTTP TLS) when `[grpc.tls]` is not configured.
    #[cfg(all(feature = "grpc", feature = "tls"))]
    grpc_tls_config: Option<crate::tls::TlsConfigSource>,
    /// Caller-registered rotation hook, invoked once by `serve()`.
    #[cfg(feature = "tls")]
    tls_reload_hook: Option<Box<dyn FnOnce(crate::tls::TlsReloadHandle) + Send>>,
    /// Validated poll period for the HTTP credentials, from `[tls]`.
    #[cfg(feature = "tls")]
    tls_reload_interval: Option<std::time::Duration>,
    /// Validated poll period for the gRPC credentials, from `[grpc.tls]`.
    #[cfg(all(feature = "grpc", feature = "tls"))]
    grpc_tls_reload_interval: Option<std::time::Duration>,
    /// Whether either TLS section asked for `SIGHUP`-driven reloading.
    #[cfg(feature = "tls")]
    tls_reload_on_sighup: bool,
    agent_runtime: Option<acton_reactive::prelude::ActorRuntime>,
    /// Fatal misconfiguration recorded during `build()`. `serve()` returns this
    /// before binding any listener, so a service that could not honour its
    /// configured security posture never accepts a connection.
    startup_error: Option<crate::error::Error>,
}

impl<T> ActonService<T>
where
    T: Serialize + DeserializeOwned + Clone + Default + Send + Sync + 'static,
{
    /// The HTTP listener's TLS credentials, if TLS resolved.
    ///
    /// **Clone this before calling [`serve`](Self::serve).** `serve` consumes
    /// the service, so a handle not taken beforehand cannot be taken at all.
    /// Move the clone into whatever task drives rotation:
    ///
    /// ```rust,ignore
    /// let service = builder.build();
    /// let tls = service.tls_config_source();          // before serve()
    /// tokio::spawn(async move {
    ///     if let Some(tls) = tls {
    ///         watch_for_new_certs().await;
    ///         let _ = tls.reload();
    ///     }
    /// });
    /// service.serve().await?;                          // consumes the service
    /// ```
    ///
    /// Prefer [`ServiceBuilder::with_tls_reload`], which hands the same handles
    /// to a callback at serve time and so has no ordering to get wrong, or the
    /// `reload_interval_secs` / `reload_on_sighup` config triggers, which need
    /// no code at all.
    ///
    /// `None` means no TLS resolved for the HTTP listener: no `[tls]` section,
    /// `enabled = false`, and no injected override.
    ///
    /// A returned source is not necessarily reloadable. One injected via
    /// [`ServiceBuilder::with_tls_config`] holds an already-loaded
    /// `ServerConfig` with no files behind it, and
    /// [`reload`](crate::tls::TlsConfigSource::reload) on it returns an error;
    /// check [`is_reloadable`](crate::tls::TlsConfigSource::is_reloadable)
    /// first if the source could be either.
    #[cfg(feature = "tls")]
    #[must_use]
    pub fn tls_config_source(&self) -> Option<crate::tls::TlsConfigSource> {
        self.tls_config.clone()
    }

    /// The separate-port gRPC listener's TLS credentials, if TLS resolved.
    ///
    /// The gRPC twin of [`tls_config_source`](Self::tls_config_source), with
    /// the same ordering constraint: clone it out **before**
    /// [`serve`](Self::serve) consumes the service.
    ///
    /// When `[grpc.tls]` is absent the gRPC listener inherits the `[tls]`
    /// credentials, and this returns the same source
    /// [`tls_config_source`](Self::tls_config_source) does — reloading either
    /// rotates both. Use
    /// [`TlsConfigSource::ptr_eq`](crate::tls::TlsConfigSource::ptr_eq) to tell
    /// that case from two genuinely separate sets of credentials.
    ///
    /// `None` means the gRPC listener resolved no TLS: it serves plaintext, or
    /// single-port mode is in use and the HTTP listener terminates TLS for both.
    #[cfg(all(feature = "grpc", feature = "tls"))]
    #[must_use]
    pub fn grpc_tls_config_source(&self) -> Option<crate::tls::TlsConfigSource> {
        self.grpc_tls_config.clone()
    }

    /// Start the configured credential-rotation triggers.
    ///
    /// Called once from [`serve`](Self::serve), before any listener binds, so
    /// every serve path — dual-port gRPC, single-port hybrid, and HTTP-only —
    /// gets the same triggers without each having to remember to start them.
    ///
    /// The returned guard owns the spawned tasks and aborts them when `serve`
    /// returns.
    #[cfg(feature = "tls")]
    fn install_tls_reload_triggers(&mut self) -> crate::tls::TlsReloadTasks {
        let http_source = self.tls_config.clone();
        #[cfg(feature = "grpc")]
        let grpc_source = self.grpc_tls_config.clone();
        #[cfg(not(feature = "grpc"))]
        let grpc_source: Option<crate::tls::TlsConfigSource> = None;
        #[cfg(feature = "grpc")]
        let grpc_interval = self.grpc_tls_reload_interval;
        #[cfg(not(feature = "grpc"))]
        let grpc_interval = None;

        let tls_resolved = http_source.is_some() || grpc_source.is_some();
        // `new` keeps only sources a reload could act on, and drops a gRPC
        // source that is really the HTTP one inherited.
        let handle = crate::tls::TlsReloadHandle::new(http_source, grpc_source);

        // The config-driven triggers are shared with `Server::serve`, so the
        // same config file produces the same rotation behaviour on both paths.
        let tasks = crate::tls::install_reload_triggers(
            &handle,
            self.tls_reload_interval,
            grpc_interval,
            self.tls_reload_on_sighup,
        );

        // Caller-registered hook, invoked last so the built-in triggers are
        // already running when it takes over.
        if let Some(hook) = self.tls_reload_hook.take() {
            if handle.is_empty() {
                if tls_resolved {
                    tracing::warn!(
                        "a with_tls_reload hook is registered, but every resolved TLS \
                         source was supplied as an already-loaded ServerConfig and has \
                         no files to reread. The hook will not be called. Use \
                         with_tls_config_source or the [tls] section so the credentials \
                         can be reloaded."
                    );
                } else {
                    tracing::warn!(
                        "a with_tls_reload hook is registered, but no TLS is configured, \
                         so there are no credentials to reload and the hook will not be \
                         called"
                    );
                }
            } else {
                hook(handle);
            }
        }

        tasks
    }

    /// Serve the application
    ///
    /// This runs the HTTP server (and optionally gRPC server) with graceful shutdown support.
    ///
    /// If gRPC services are configured:
    /// - Single-port mode (default): Both HTTP and gRPC on same port, routed by content-type
    /// - Dual-port mode: HTTP on configured port, gRPC on separate port
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let service = ServiceBuilder::new()
    ///     .with_config(config)
    ///     .with_routes(routes)
    ///     .with_state(state)
    ///     .build();
    ///
    /// service.serve().await?;
    /// ```
    #[cfg_attr(not(feature = "grpc"), allow(unused_mut))]
    pub async fn serve(mut self) -> crate::error::Result<()> {
        use tokio::net::TcpListener;
        use tokio::signal;

        // Fail before any socket binds. A misconfiguration that would force a
        // weaker posture than configured (TLS that could not load, invalid token
        // auth) must never reach the point of accepting connections.
        if let Some(e) = self.startup_error.take() {
            return Err(e);
        }

        // Start credential rotation before binding, so a certificate that
        // rotates during startup is picked up rather than missed. The guard is
        // held for the whole of `serve`; dropping it on any return path aborts
        // the trigger tasks, which would otherwise outlive the listeners they
        // rotate. Every serve path below is covered by this one call.
        #[cfg(feature = "tls")]
        let _tls_reload_tasks = self.install_tls_reload_triggers();

        // Graceful shutdown signal
        async fn shutdown_signal() {
            let ctrl_c = async {
                signal::ctrl_c()
                    .await
                    .expect("failed to install Ctrl+C handler");
            };

            #[cfg(unix)]
            let terminate = async {
                signal::unix::signal(signal::unix::SignalKind::terminate())
                    .expect("failed to install signal handler")
                    .recv()
                    .await;
            };

            #[cfg(not(unix))]
            let terminate = std::future::pending::<()>();

            tokio::select! {
                _ = ctrl_c => {},
                _ = terminate => {},
            }
        }

        #[cfg(feature = "grpc")]
        {
            // Check if gRPC is enabled and services are provided
            if let Some(ref grpc_config) = self.config.grpc {
                if grpc_config.enabled && self.grpc_routes.is_some() {
                    let grpc_routes = self.grpc_routes.take().unwrap();

                    if grpc_config.use_separate_port {
                        // Dual-port mode: HTTP and gRPC on separate ports.
                        // The gRPC listener uses its own bind when set, else the
                        // service-level bind.
                        let grpc_port = grpc_config.port;
                        let grpc_bind = grpc_config.effective_bind(self.config.service.bind);
                        let grpc_addr = std::net::SocketAddr::new(grpc_bind, grpc_port);

                        tracing::info!("Starting HTTP service on {}", self.listener_addr);
                        tracing::info!("Starting gRPC service on {}", grpc_addr);

                        let http_listener = TcpListener::bind(&self.listener_addr).await?;
                        let grpc_listener = TcpListener::bind(&grpc_addr).await?;

                        // The gRPC axum router (auth and Cedar layers were
                        // applied at build time)
                        let grpc_app = grpc_routes;

                        // Spawn gRPC server on separate task (with optional
                        // per-listener TLS, falling back to the HTTP TLS config).
                        #[cfg(feature = "tls")]
                        let grpc_tls_config = self.grpc_tls_config.clone();

                        let grpc_handle = tokio::spawn(async move {
                            #[cfg(feature = "tls")]
                            if let Some(ref tls_source) = grpc_tls_config {
                                let tls_listener = crate::tls::TlsListener::with_config_source(
                                    grpc_listener,
                                    tls_source.clone(),
                                );
                                return axum::serve(
                                    tls_listener,
                                    grpc_app.into_make_service_with_connect_info::<
                                        crate::tls::TlsConnectInfo,
                                    >(),
                                )
                                    .with_graceful_shutdown(shutdown_signal())
                                    .await;
                            }

                            axum::serve(grpc_listener, grpc_app)
                                .with_graceful_shutdown(shutdown_signal())
                                .await
                        });

                        // Run HTTP server (with optional TLS). The TLS listener
                        // exposes `TlsConnectInfo` (remote address plus any
                        // verified client certificate) as connect-info.
                        #[cfg(feature = "tls")]
                        if let Some(ref tls_source) = self.tls_config {
                            let tls_listener = crate::tls::TlsListener::with_config_source(
                                http_listener,
                                tls_source.clone(),
                            );
                            tracing::info!("TLS enabled (HTTPS) for both HTTP and gRPC");
                            let http_result = axum::serve(
                                tls_listener,
                                self.app.into_make_service_with_connect_info::<
                                    crate::tls::TlsConnectInfo,
                                >(),
                            )
                                .with_graceful_shutdown(shutdown_signal())
                                .await;
                            let _ = grpc_handle.await;
                            http_result?;

                            tracing::info!("Server shutdown complete");
                            if let Some(mut runtime) = self.agent_runtime {
                                tracing::info!("Shutting down agent runtime...");
                                if let Err(e) = runtime.shutdown_all().await {
                                    tracing::error!("Agent runtime shutdown error: {}", e);
                                }
                                tracing::info!("Agent runtime shutdown complete");
                            }
                            return Ok(());
                        }

                        // Run HTTP server (plain TCP)
                        let http_result = axum::serve(
                            http_listener,
                            self.app
                                .into_make_service_with_connect_info::<std::net::SocketAddr>(),
                        )
                        .with_graceful_shutdown(shutdown_signal())
                        .await;

                        // Wait for gRPC server
                        let _ = grpc_handle.await;

                        http_result?;
                    } else {
                        // Single-port mode: Hybrid HTTP + gRPC on same port
                        tracing::info!(
                            "Starting hybrid HTTP+gRPC service on {}",
                            self.listener_addr
                        );

                        let listener = TcpListener::bind(&self.listener_addr).await?;

                        // Merge HTTP and gRPC services (the gRPC router keeps
                        // its own auth and Cedar layers through the merge)
                        let hybrid_service = grpc_routes.merge(self.app);

                        // The TLS listener exposes `TlsConnectInfo` (remote
                        // address plus any verified client certificate) as
                        // connect-info.
                        #[cfg(feature = "tls")]
                        if let Some(ref tls_source) = self.tls_config {
                            let tls_listener = crate::tls::TlsListener::with_config_source(
                                listener,
                                tls_source.clone(),
                            );
                            tracing::info!("TLS enabled (HTTPS) for hybrid HTTP+gRPC");
                            axum::serve(
                                tls_listener,
                                hybrid_service.into_make_service_with_connect_info::<
                                    crate::tls::TlsConnectInfo,
                                >(),
                            )
                                .with_graceful_shutdown(shutdown_signal())
                                .await?;

                            tracing::info!("Server shutdown complete");
                            if let Some(mut runtime) = self.agent_runtime {
                                tracing::info!("Shutting down agent runtime...");
                                if let Err(e) = runtime.shutdown_all().await {
                                    tracing::error!("Agent runtime shutdown error: {}", e);
                                }
                                tracing::info!("Agent runtime shutdown complete");
                            }
                            return Ok(());
                        }

                        axum::serve(
                            listener,
                            hybrid_service
                                .into_make_service_with_connect_info::<std::net::SocketAddr>(),
                        )
                        .with_graceful_shutdown(shutdown_signal())
                        .await?;
                    }

                    tracing::info!("Server shutdown complete");

                    // Shutdown agent runtime after server stops (gRPC path)
                    if let Some(mut runtime) = self.agent_runtime {
                        tracing::info!("Shutting down agent runtime...");
                        if let Err(e) = runtime.shutdown_all().await {
                            tracing::error!("Agent runtime shutdown error: {}", e);
                        }
                        tracing::info!("Agent runtime shutdown complete");
                    }

                    return Ok(());
                }
            }
        }

        // HTTP-only mode (no gRPC or gRPC disabled)
        tracing::info!("Starting HTTP service on {}", self.listener_addr);

        let listener = TcpListener::bind(&self.listener_addr).await?;

        // The TLS listener exposes `TlsConnectInfo` (remote address plus any
        // verified client certificate) as connect-info.
        #[cfg(feature = "tls")]
        if let Some(ref tls_source) = self.tls_config {
            let tls_listener =
                crate::tls::TlsListener::with_config_source(listener, tls_source.clone());
            tracing::info!("TLS enabled (HTTPS)");
            axum::serve(
                tls_listener,
                self.app
                    .into_make_service_with_connect_info::<crate::tls::TlsConnectInfo>(),
            )
            .with_graceful_shutdown(shutdown_signal())
            .await?;

            tracing::info!("Server shutdown complete");

            if let Some(mut runtime) = self.agent_runtime {
                tracing::info!("Shutting down agent runtime...");
                if let Err(e) = runtime.shutdown_all().await {
                    tracing::error!("Agent runtime shutdown error: {}", e);
                }
                tracing::info!("Agent runtime shutdown complete");
            }

            return Ok(());
        }

        axum::serve(
            listener,
            self.app
                .into_make_service_with_connect_info::<std::net::SocketAddr>(),
        )
        .with_graceful_shutdown(shutdown_signal())
        .await?;

        tracing::info!("Server shutdown complete");

        // Shutdown agent runtime after server stops (HTTP-only path)
        if let Some(mut runtime) = self.agent_runtime {
            tracing::info!("Shutting down agent runtime...");
            if let Err(e) = runtime.shutdown_all().await {
                tracing::error!("Agent runtime shutdown error: {}", e);
            }
            tracing::info!("Agent runtime shutdown complete");
        }

        Ok(())
    }

    /// Get a reference to the service configuration
    pub fn config(&self) -> &Config<T> {
        &self.config
    }

    /// Get a reference to the application state
    ///
    /// This is useful for accessing services (e.g., `BackgroundWorker`) between
    /// `build()` and `serve()` — for example, to submit startup tasks like
    /// cache warming or data synchronization.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let service = ServiceBuilder::new()
    ///     .with_config(config)
    ///     .with_routes(routes)
    ///     .build();
    ///
    /// // Submit a startup task before serving
    /// if let Some(worker) = service.state().background_worker() {
    ///     worker.submit("cache-warm", || async {
    ///         // warm caches...
    ///         Ok(())
    ///     }).await;
    /// }
    ///
    /// service.serve().await?;
    /// ```
    pub fn state(&self) -> &AppState<T> {
        &self.state
    }
}

#[cfg(test)]
mod tests {
    // This test verifies the type-state pattern at compile time
    #[test]
    fn test_service_builder_states_compile() {
        // This should compile - correct order
        // let _service = ServiceBuilder::new()
        //     .with_config(config)
        //     .with_routes(routes)
        //     .with_state(state)
        //     .build();

        // These should NOT compile (commented out to prevent compilation errors):

        // ❌ Cannot build without config
        // let _service = ServiceBuilder::new()
        //     .build();

        // ❌ Cannot skip routes
        // let _service = ServiceBuilder::new()
        //     .with_config(config)
        //     .with_state(state)
        //     .build();

        // ❌ Cannot call with_routes on wrong state
        // let _service = ServiceBuilder::new()
        //     .with_routes(routes);

        // ❌ Cannot call with_state on wrong state
        // let _service = ServiceBuilder::new()
        //     .with_config(config)
        //     .with_state(state);
    }

    #[test]
    fn test_acton_service_exposes_state() {
        use crate::config::Config;
        use crate::prelude::ServiceBuilder;

        let config = Config::<()>::default();
        let service = ServiceBuilder::new().with_config(config).build();

        // state() should be accessible and background_worker() returns None
        // when not configured
        assert!(service.state().background_worker().is_none());
    }

    #[test]
    fn test_versioned_routes_cannot_be_constructed_manually() {
        // This should NOT compile (VersionedRoutes has private fields):
        // let routes = VersionedRoutes { router: Router::new() };

        // The ONLY way to create VersionedRoutes is through VersionedApiBuilder
    }

    /// A `[cedar]` section whose policy file cannot be loaded must be a hard
    /// failure. Serving every route without policy enforcement where the
    /// operator configured Cedar is the same silent downgrade class as
    /// serving plaintext where TLS was configured.
    #[cfg(feature = "cedar-authz")]
    #[tokio::test(flavor = "multi_thread")]
    async fn cedar_init_failure_is_fatal_not_an_unenforced_downgrade() {
        use crate::config::{CedarConfig, Config};
        use crate::prelude::ServiceBuilder;

        let config = Config::<()> {
            cedar: Some(CedarConfig {
                enabled: true,
                policy_path: "/nonexistent/policies.cedar".into(),
                hot_reload: false,
                hot_reload_interval_secs: 60,
                cache_enabled: false,
                cache_ttl_secs: 60,
                fail_open: false,
            }),
            ..Default::default()
        };

        let error = ServiceBuilder::new()
            .with_config(config)
            .try_build()
            .err()
            .expect("an unloadable Cedar policy set must fail the build, not skip authorization");

        let message = error.to_string();
        assert!(
            message.contains(
                "refusing to start rather than serving routes without policy enforcement"
            ),
            "error must explain the refusal, got: {message}"
        );
    }

    /// A disabled `[cedar]` section must not be fatal — only an enabled one
    /// that cannot be initialized.
    #[cfg(feature = "cedar-authz")]
    #[tokio::test(flavor = "multi_thread")]
    async fn disabled_cedar_section_builds_successfully() {
        use crate::config::{CedarConfig, Config};
        use crate::prelude::ServiceBuilder;

        let config = Config::<()> {
            cedar: Some(CedarConfig {
                enabled: false,
                policy_path: "/nonexistent/policies.cedar".into(),
                hot_reload: false,
                hot_reload_interval_secs: 60,
                cache_enabled: false,
                cache_ttl_secs: 60,
                fail_open: false,
            }),
            ..Default::default()
        };

        ServiceBuilder::new()
            .with_config(config)
            .try_build()
            .expect("a disabled cedar section must not fail the build");
    }

    /// Build a `[tls]` section over real, loadable credentials in `dir`.
    #[cfg(feature = "tls")]
    fn loadable_tls_section(dir: &std::path::Path) -> crate::config::TlsConfig {
        let certified = rcgen::generate_simple_self_signed(vec!["localhost".to_string()])
            .expect("self-signed cert generation");
        let cert_path = dir.join("server.pem");
        let key_path = dir.join("server.key");
        std::fs::write(&cert_path, certified.cert.pem()).expect("write cert");
        std::fs::write(&key_path, certified.signing_key.serialize_pem()).expect("write key");

        crate::config::TlsConfig {
            enabled: true,
            cert_path,
            key_path,
            client_ca_path: None,
            client_auth_optional: false,
            reload_interval_secs: None,
            reload_on_sighup: false,
        }
    }

    /// `reload_interval_secs = 0` cannot mean "never" — omitting the field does
    /// — and taken literally it polls the disk without pause. It must be caught
    /// at build time, not discovered as a hot task in production.
    #[cfg(feature = "tls")]
    #[test]
    fn zero_reload_interval_is_rejected_at_build() {
        use crate::config::Config;
        use crate::prelude::ServiceBuilder;

        let dir = tempfile::tempdir().expect("temp dir");
        let mut tls = loadable_tls_section(dir.path());
        tls.reload_interval_secs = Some(0);

        let error = ServiceBuilder::new()
            .with_config(Config::<()> {
                tls: Some(tls),
                ..Default::default()
            })
            .try_build()
            .err()
            .expect("a zero poll interval must fail the build");

        let message = error.to_string();
        assert!(
            message.contains("reload_interval_secs = 0"),
            "the error must name the offending setting, got: {message}"
        );
        assert!(
            message.contains("[tls]"),
            "the error must name the section to fix, got: {message}"
        );
    }

    /// The same validation on the gRPC section, which carries its own interval.
    #[cfg(all(feature = "grpc", feature = "tls"))]
    #[test]
    fn zero_grpc_reload_interval_is_rejected_at_build() {
        use crate::config::{Config, GrpcConfig};
        use crate::prelude::ServiceBuilder;

        let dir = tempfile::tempdir().expect("temp dir");
        let mut tls = loadable_tls_section(dir.path());
        tls.reload_interval_secs = Some(0);

        let error = ServiceBuilder::new()
            .with_config(Config::<()> {
                grpc: Some(GrpcConfig {
                    enabled: true,
                    use_separate_port: true,
                    bind: None,
                    tls: Some(tls),
                    port: 50051,
                    reflection_enabled: false,
                    health_check_enabled: false,
                    max_message_size_mb: 4,
                    connection_timeout_secs: 10,
                    timeout_secs: 30,
                    proto: Default::default(),
                }),
                ..Default::default()
            })
            .try_build()
            .err()
            .expect("a zero gRPC poll interval must fail the build");

        assert!(
            error.to_string().contains("[grpc.tls]"),
            "the error must name the gRPC section, got: {error}"
        );
    }

    /// A positive interval builds, so the validation rejects only the value
    /// that cannot work rather than the feature.
    #[cfg(feature = "tls")]
    #[test]
    fn a_positive_reload_interval_builds() {
        use crate::config::Config;
        use crate::prelude::ServiceBuilder;

        let dir = tempfile::tempdir().expect("temp dir");
        let mut tls = loadable_tls_section(dir.path());
        tls.reload_interval_secs = Some(30);
        tls.reload_on_sighup = true;

        ServiceBuilder::new()
            .with_config(Config::<()> {
                tls: Some(tls),
                ..Default::default()
            })
            .try_build()
            .expect("a positive interval and SIGHUP reloading must build");
    }

    /// The hook exists so callers need not clone handles out before `serve()`
    /// consumes the service. It must actually receive the config-driven source.
    #[cfg(feature = "tls")]
    #[tokio::test(flavor = "multi_thread")]
    async fn tls_reload_hook_receives_the_config_driven_source() {
        use crate::config::Config;
        use crate::prelude::ServiceBuilder;
        use std::sync::mpsc;

        let dir = tempfile::tempdir().expect("temp dir");
        let tls = loadable_tls_section(dir.path());
        let (tx, rx) = mpsc::channel();

        let mut service = ServiceBuilder::new()
            .with_config(Config::<()> {
                tls: Some(tls),
                ..Default::default()
            })
            .with_tls_reload(move |handle| {
                let _ = tx.send(handle);
            })
            .try_build()
            .expect("build");

        // What `serve()` does with the triggers, without binding a socket.
        let _tasks = service.install_tls_reload_triggers();

        let handle = rx
            .try_recv()
            .expect("the hook must be called when a reloadable source resolved");
        assert!(
            handle.http().is_some(),
            "the hook must receive the HTTP listener's credentials"
        );
        assert!(
            handle
                .http()
                .is_some_and(crate::tls::TlsConfigSource::is_reloadable),
            "the source handed over must be one a reload can act on"
        );
    }

    /// A hook over credentials that can never be reloaded would leave the
    /// caller believing rotation is wired up. It is skipped, and the skip is
    /// audible.
    #[cfg(feature = "tls")]
    #[tokio::test(flavor = "multi_thread")]
    async fn tls_reload_hook_is_skipped_when_only_static_sources_exist() {
        use crate::config::Config;
        use crate::prelude::ServiceBuilder;
        use std::sync::mpsc;

        let (tx, rx) = mpsc::channel();

        let mut service = ServiceBuilder::new()
            .with_config(Config::<()>::default())
            // An injected `ServerConfig` has no files behind it.
            .with_tls_config(test_server_config())
            .with_tls_reload(move |handle| {
                let _ = tx.send(handle);
            })
            .try_build()
            .expect("build");

        let _tasks = service.install_tls_reload_triggers();

        assert!(
            rx.try_recv().is_err(),
            "a hook that could only ever fail to reload must not be called"
        );
    }

    /// No TLS at all is the other way the hook has nothing to work with.
    #[cfg(feature = "tls")]
    #[tokio::test(flavor = "multi_thread")]
    async fn tls_reload_hook_is_skipped_when_no_tls_is_configured() {
        use crate::config::Config;
        use crate::prelude::ServiceBuilder;
        use std::sync::mpsc;

        let (tx, rx) = mpsc::channel();

        let mut service = ServiceBuilder::new()
            .with_config(Config::<()>::default())
            .with_tls_reload(move |handle| {
                let _ = tx.send(handle);
            })
            .try_build()
            .expect("build");

        let _tasks = service.install_tls_reload_triggers();

        assert!(
            rx.try_recv().is_err(),
            "with no TLS resolved there is nothing to reload and the hook must be skipped"
        );
    }

    /// The accessor is the escape hatch for callers not using the hook. It must
    /// hand back the same credentials the listener will serve.
    #[cfg(feature = "tls")]
    #[test]
    fn tls_config_source_accessor_exposes_the_resolved_credentials() {
        use crate::config::Config;
        use crate::prelude::ServiceBuilder;

        let dir = tempfile::tempdir().expect("temp dir");
        let tls = loadable_tls_section(dir.path());

        let service = ServiceBuilder::new()
            .with_config(Config::<()> {
                tls: Some(tls),
                ..Default::default()
            })
            .try_build()
            .expect("build");

        let source = service
            .tls_config_source()
            .expect("an enabled [tls] section must expose a source");
        assert!(
            source.is_reloadable(),
            "credentials loaded from the config must be rotatable"
        );

        // Clones observe one another, which is what makes grabbing the handle
        // before `serve()` useful at all.
        let second = service.tls_config_source().expect("source");
        assert!(
            source.ptr_eq(&second),
            "the accessor must not clone the state"
        );
    }

    #[cfg(feature = "tls")]
    #[test]
    fn tls_config_source_accessor_is_none_without_tls() {
        use crate::config::Config;
        use crate::prelude::ServiceBuilder;

        let service = ServiceBuilder::new()
            .with_config(Config::<()>::default())
            .try_build()
            .expect("build");

        assert!(
            service.tls_config_source().is_none(),
            "no TLS resolved must read as None, not as an unusable source"
        );
    }

    /// A `[tls]` section pointing at cert material that cannot be loaded must be
    /// a hard failure. Serving plaintext where the operator configured TLS is a
    /// silent downgrade of the configured security posture.
    #[cfg(feature = "tls")]
    #[test]
    fn tls_load_failure_is_fatal_not_a_plaintext_downgrade() {
        use crate::config::{Config, TlsConfig};
        use crate::prelude::ServiceBuilder;

        let config = Config::<()> {
            tls: Some(TlsConfig {
                enabled: true,
                cert_path: "/nonexistent/cert.pem".into(),
                key_path: "/nonexistent/key.pem".into(),
                client_ca_path: None,
                client_auth_optional: false,
                reload_interval_secs: None,
                reload_on_sighup: false,
            }),
            ..Default::default()
        };

        let error = ServiceBuilder::new()
            .with_config(config)
            .try_build()
            .err()
            .expect("unloadable TLS material must fail the build, not degrade to plaintext");

        // The returned error must stand on its own. An operator who sees only
        // this (a crash message, a supervisor log) should not have to go find
        // the tracing line to learn why startup was refused.
        let message = error.to_string();
        assert!(
            message.contains("refusing to start a plaintext listener"),
            "error must explain the refusal, got: {message}"
        );
        assert!(
            message.contains("/nonexistent/cert.pem"),
            "error must name the material that could not be loaded, got: {message}"
        );
    }

    /// The gRPC twin of the above: `[grpc.tls]` is authoritative for the
    /// separate-port listener, so a failed load there must not leave that
    /// listener serving plaintext on a possibly non-loopback bind.
    #[cfg(all(feature = "grpc", feature = "tls"))]
    #[test]
    fn grpc_tls_load_failure_is_fatal_not_a_plaintext_downgrade() {
        use crate::config::{Config, GrpcConfig, TlsConfig};
        use crate::prelude::ServiceBuilder;

        let grpc = Some(GrpcConfig {
            enabled: true,
            use_separate_port: true,
            bind: None,
            tls: Some(TlsConfig {
                enabled: true,
                cert_path: "/nonexistent/cert.pem".into(),
                key_path: "/nonexistent/key.pem".into(),
                client_ca_path: None,
                client_auth_optional: false,
                reload_interval_secs: None,
                reload_on_sighup: false,
            }),
            port: 50051,
            reflection_enabled: false,
            health_check_enabled: false,
            max_message_size_mb: 4,
            connection_timeout_secs: 10,
            timeout_secs: 30,
            proto: Default::default(),
        });
        let config = Config::<()> {
            grpc,
            ..Default::default()
        };

        let result = ServiceBuilder::new().with_config(config).try_build();

        assert!(
            result.is_err(),
            "unloadable gRPC TLS material must fail the build, not degrade to plaintext"
        );
    }

    /// TLS that is present but explicitly disabled is a deliberate operator
    /// choice, not a failure — it must still build.
    #[cfg(feature = "tls")]
    #[test]
    fn disabled_tls_section_builds_successfully() {
        use crate::config::{Config, TlsConfig};
        use crate::prelude::ServiceBuilder;

        let config = Config::<()> {
            tls: Some(TlsConfig {
                enabled: false,
                cert_path: "/nonexistent/cert.pem".into(),
                key_path: "/nonexistent/key.pem".into(),
                client_ca_path: None,
                client_auth_optional: false,
                reload_interval_secs: None,
                reload_on_sighup: false,
            }),
            ..Default::default()
        };

        assert!(
            ServiceBuilder::new()
                .with_config(config)
                .try_build()
                .is_ok(),
            "an explicitly disabled TLS section must not block startup"
        );
    }

    /// An injected config must be used verbatim, with no read of the configured
    /// cert paths — that second read is the time-of-check/time-of-use window
    /// `with_tls_config` exists to close. Paths that would fail to load prove
    /// they were never touched.
    #[cfg(feature = "tls")]
    #[test]
    fn injected_tls_config_bypasses_the_file_load() {
        use crate::config::{Config, TlsConfig};
        use crate::prelude::ServiceBuilder;

        let config = Config::<()> {
            tls: Some(TlsConfig {
                enabled: true,
                cert_path: "/nonexistent/cert.pem".into(),
                key_path: "/nonexistent/key.pem".into(),
                client_ca_path: None,
                client_auth_optional: false,
                reload_interval_secs: None,
                reload_on_sighup: false,
            }),
            ..Default::default()
        };

        let result = ServiceBuilder::new()
            .with_config(config)
            .with_tls_config(test_server_config())
            .try_build();

        assert!(
            result.is_ok(),
            "an injected ServerConfig must be used without reading the configured paths"
        );
    }

    /// The audit agent (enabled by default) cannot be spawned from a
    /// current-thread runtime; `build()` must record that as a startup error
    /// that `serve()` surfaces, instead of letting `block_in_place` panic with
    /// an unactionable message deep inside tokio. This test deliberately stays
    /// on the default current-thread test runtime.
    #[cfg(feature = "audit")]
    #[tokio::test]
    async fn serve_refuses_when_audit_cannot_spawn_on_current_thread_runtime() {
        use crate::config::Config;
        use crate::prelude::ServiceBuilder;

        let base = Config::<()>::default();
        let config = Config::<()> {
            service: crate::config::ServiceConfig {
                port: 0,
                ..base.service.clone()
            },
            ..base
        };

        let service = ServiceBuilder::new().with_config(config).build();

        let err = service
            .serve()
            .await
            .expect_err("serve() must refuse when the audit agent could not be spawned");
        assert!(
            err.to_string().contains("audit"),
            "error must name the audit subsystem: {err}"
        );
    }

    /// Base config for the current-thread guard tests below.
    ///
    /// Audit is enabled by default and its own guard fires first, so it is
    /// switched off here — otherwise the audit startup error would mask the
    /// subsystem actually under test.
    #[cfg(test)]
    fn config_without_audit() -> crate::config::Config<()> {
        let base = crate::config::Config::<()>::default();
        crate::config::Config::<()> {
            service: crate::config::ServiceConfig {
                port: 0,
                ..base.service.clone()
            },
            #[cfg(feature = "audit")]
            audit: Some(crate::audit::AuditConfig {
                enabled: false,
                ..Default::default()
            }),
            ..base
        }
    }

    /// A configured background worker cannot spawn its agent on a
    /// current-thread runtime; `try_build()` must say so instead of letting
    /// `block_in_place` panic inside tokio.
    #[tokio::test]
    async fn build_refuses_background_worker_on_current_thread_runtime() {
        use crate::prelude::ServiceBuilder;

        let config = crate::config::Config::<()> {
            background_worker: Some(crate::agents::BackgroundWorkerConfig {
                enabled: true,
                ..Default::default()
            }),
            ..config_without_audit()
        };

        let Err(err) = ServiceBuilder::new().with_config(config).try_build() else {
            panic!("try_build() must refuse a background worker it cannot spawn");
        };
        assert!(
            err.to_string().contains("background worker"),
            "error must name the background worker subsystem: {err}"
        );
    }

    /// Registered actor extensions cannot be spawned on a current-thread
    /// runtime, so the build must fail with a message naming them.
    #[tokio::test]
    async fn build_refuses_actor_extensions_on_current_thread_runtime() {
        use crate::extensions::ActorExtension;
        use crate::prelude::ServiceBuilder;
        use acton_reactive::prelude::{Idle, ManagedActor};

        #[derive(Debug, Default, Clone)]
        struct NoopActor;

        impl ActorExtension for NoopActor {
            fn configure(_actor: &mut ManagedActor<Idle, Self>) {}
        }

        let Err(err) = ServiceBuilder::new()
            .with_config(config_without_audit())
            .with_actor::<NoopActor>()
            .try_build()
        else {
            panic!("try_build() must refuse actor extensions it cannot spawn");
        };
        assert!(
            err.to_string().contains("actor extensions"),
            "error must name the actor-extension subsystem: {err}"
        );
    }

    /// Cedar cannot load its policy set on a current-thread runtime. Failing
    /// closed matters most here: silently continuing would serve routes with
    /// no policy enforcement at all.
    #[cfg(feature = "cedar-authz")]
    #[tokio::test]
    async fn build_refuses_cedar_on_current_thread_runtime() {
        use crate::prelude::ServiceBuilder;

        let config = crate::config::Config::<()> {
            cedar: Some(crate::config::CedarConfig {
                enabled: true,
                // Never read: the flavor guard fires before any policy load.
                policy_path: "/nonexistent/policies.cedar".into(),
                hot_reload: false,
                hot_reload_interval_secs: 30,
                cache_enabled: false,
                cache_ttl_secs: 60,
                fail_open: false,
            }),
            ..config_without_audit()
        };

        let Err(err) = ServiceBuilder::new().with_config(config).try_build() else {
            panic!("try_build() must refuse Cedar it cannot initialize");
        };
        assert!(
            err.to_string().contains("Cedar"),
            "error must name the Cedar subsystem: {err}"
        );
    }

    /// Redis-backed sessions must fail on the runtime flavor *before* any
    /// connection is attempted, so this test needs no Redis instance — the
    /// URL below is never dialed.
    #[cfg(feature = "session-redis")]
    #[tokio::test]
    async fn build_refuses_redis_sessions_on_current_thread_runtime() {
        use crate::prelude::ServiceBuilder;

        let config = crate::config::Config::<()> {
            session: Some(crate::session::SessionConfig {
                storage: crate::session::SessionStorage::Redis,
                // Unreachable on purpose: the guard must fire before dialing.
                redis_url: Some("redis://127.0.0.1:1/".to_string()),
                ..Default::default()
            }),
            ..config_without_audit()
        };

        let Err(err) = ServiceBuilder::new().with_config(config).try_build() else {
            panic!("try_build() must refuse Redis sessions it cannot initialize");
        };
        assert!(
            err.to_string().contains("Redis-backed sessions"),
            "error must name the Redis session subsystem: {err}"
        );
    }

    /// Key rotation checks the runtime flavor before it resolves a storage
    /// backend, so the guard fires even with no database configured and the
    /// test needs no infrastructure.
    #[cfg(feature = "auth")]
    #[tokio::test]
    async fn build_refuses_key_rotation_on_current_thread_runtime() {
        use crate::prelude::ServiceBuilder;

        let config = crate::config::Config::<()> {
            auth: Some(crate::auth::AuthConfig {
                key_rotation: Some(crate::auth::key_rotation::KeyRotationConfig {
                    enabled: true,
                    ..Default::default()
                }),
                ..Default::default()
            }),
            ..config_without_audit()
        };

        let Err(err) = ServiceBuilder::new().with_config(config).try_build() else {
            panic!("try_build() must refuse key rotation it cannot initialize");
        };
        assert!(
            err.to_string().contains("key rotation"),
            "error must name the key rotation subsystem: {err}"
        );
    }

    /// `serve()` must reject a degraded build before it binds any listener, so
    /// callers using the infallible `build()` are protected too.
    #[cfg(feature = "tls")]
    // Multi-threaded: `build()` spawns the audit agent (enabled by default),
    // which requires a runtime where `block_in_place` can block.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn serve_refuses_to_bind_after_a_failed_tls_load() {
        use crate::config::{Config, TlsConfig};
        use crate::prelude::ServiceBuilder;

        let base = Config::<()>::default();
        let config = Config::<()> {
            // Port 0 binds successfully, so any error can only come from the
            // startup gate rather than from an unavailable port.
            service: crate::config::ServiceConfig {
                port: 0,
                ..base.service.clone()
            },
            tls: Some(TlsConfig {
                enabled: true,
                cert_path: "/nonexistent/cert.pem".into(),
                key_path: "/nonexistent/key.pem".into(),
                client_ca_path: None,
                client_auth_optional: false,
                reload_interval_secs: None,
                reload_on_sighup: false,
            }),
            ..base
        };

        let service = ServiceBuilder::new().with_config(config).build();

        assert!(
            service.serve().await.is_err(),
            "serve() must surface the failed TLS load instead of listening in plaintext"
        );
    }

    /// An injected TLS config activates TLS independently of the `[tls]`
    /// section, so HSTS must still be sent. Deriving the flag from
    /// `config.tls.enabled` alone would drop the header on connections that are
    /// in fact encrypted.
    #[cfg(feature = "tls")]
    // Multi-threaded: `build()` spawns the audit agent (enabled by default),
    // which requires a runtime where `block_in_place` can block.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn injected_tls_config_still_sends_hsts() {
        use crate::config::Config;
        use crate::prelude::ServiceBuilder;
        use axum::body::Body;
        use axum::http::Request;
        use tower::ServiceExt;

        // No `[tls]` section at all — TLS comes purely from the injected config.
        let config = Config::<()>::default();
        assert!(config.tls.is_none(), "test premise: no [tls] section");

        let service = ServiceBuilder::new()
            .with_config(config)
            .with_tls_config(test_server_config())
            .try_build()
            .expect("build with injected TLS config");

        let response = service
            .app
            .oneshot(
                Request::builder()
                    .uri("/health")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .expect("health request");

        assert!(
            response.headers().contains_key("strict-transport-security"),
            "HSTS must be sent when an injected config puts the listener on TLS"
        );
    }

    /// Build a usable self-signed `ServerConfig` for injection tests.
    #[cfg(feature = "tls")]
    fn test_server_config() -> std::sync::Arc<tokio_rustls::rustls::ServerConfig> {
        use tokio_rustls::rustls::pki_types::PrivatePkcs8KeyDer;
        use tokio_rustls::rustls::ServerConfig;

        crate::crypto::ensure_default_crypto_provider();

        let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_string()])
            .expect("self-signed cert generation");
        let key = PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der());

        std::sync::Arc::new(
            ServerConfig::builder()
                .with_no_client_auth()
                .with_single_cert(vec![cert.cert.der().clone()], key.into())
                .expect("server config"),
        )
    }

    /// `GrpcConfig::default()` must not stand up a gRPC surface on its own.
    /// Callers reach for it to fill in ports and timeouts, not to opt in.
    #[cfg(feature = "grpc")]
    #[test]
    fn grpc_config_default_is_disabled() {
        use crate::config::GrpcConfig;

        let config = GrpcConfig::default();

        assert!(
            !config.enabled,
            "GrpcConfig::default() must be disabled so it cannot enable gRPC by accident"
        );
        assert_eq!(
            config.port, 9090,
            "the Default impl must agree with the serde default for `port`"
        );
    }

    /// Registering gRPC services with no `[grpc]` section at all must fail the
    /// build rather than start an HTTP-only service that drops every RPC.
    #[cfg(feature = "grpc")]
    #[tokio::test(flavor = "multi_thread")]
    async fn grpc_services_without_grpc_config_is_fatal() {
        use crate::config::Config;
        use crate::prelude::ServiceBuilder;

        let config = Config::<()> {
            grpc: None,
            ..Default::default()
        };

        let error = ServiceBuilder::new()
            .with_config(config)
            .with_grpc_services(tonic::service::Routes::default())
            .try_build()
            .err()
            .expect("registering gRPC services with no [grpc] section must fail the build");

        let message = error.to_string();
        assert!(
            message.contains("no `[grpc]` section is configured"),
            "error must name the cause, got: {message}"
        );
        assert!(
            message.contains("enabled = true"),
            "error must state the fix, got: {message}"
        );
    }

    /// An explicitly disabled `[grpc]` section is equally fatal when services
    /// were registered — the services still have nowhere to be served.
    #[cfg(feature = "grpc")]
    #[tokio::test(flavor = "multi_thread")]
    async fn grpc_services_with_disabled_grpc_config_is_fatal() {
        use crate::config::{Config, GrpcConfig};
        use crate::prelude::ServiceBuilder;

        let config = Config::<()> {
            grpc: Some(GrpcConfig {
                enabled: false,
                ..Default::default()
            }),
            ..Default::default()
        };

        let error = ServiceBuilder::new()
            .with_config(config)
            .with_grpc_services(tonic::service::Routes::default())
            .try_build()
            .err()
            .expect("registering gRPC services against `enabled = false` must fail the build");

        assert!(
            error
                .to_string()
                .contains("the `[grpc]` section sets `enabled = false`"),
            "error must name the cause, got: {error}"
        );
    }

    /// The happy path: services registered against an enabled `[grpc]` section
    /// builds, so the new check cannot regress working configurations.
    #[cfg(feature = "grpc")]
    #[tokio::test(flavor = "multi_thread")]
    async fn grpc_services_with_enabled_grpc_config_builds() {
        use crate::config::{Config, GrpcConfig};
        use crate::prelude::ServiceBuilder;

        let config = Config::<()> {
            grpc: Some(GrpcConfig {
                enabled: true,
                ..Default::default()
            }),
            ..Default::default()
        };

        assert!(
            ServiceBuilder::new()
                .with_config(config)
                .with_grpc_services(tonic::service::Routes::default())
                .try_build()
                .is_ok(),
            "an enabled [grpc] section with registered services must build"
        );
    }

    /// No services registered means nothing to lose, whatever `[grpc]` says.
    /// Guards against the check firing on plain HTTP services.
    #[cfg(feature = "grpc")]
    #[tokio::test(flavor = "multi_thread")]
    async fn no_grpc_services_without_grpc_config_builds() {
        use crate::config::Config;
        use crate::prelude::ServiceBuilder;

        let config = Config::<()> {
            grpc: None,
            ..Default::default()
        };

        assert!(
            ServiceBuilder::new()
                .with_config(config)
                .try_build()
                .is_ok(),
            "an HTTP-only service must not be affected by the gRPC registration check"
        );
    }
}