aves 0.5.0

use std::any::Any;
use std::fmt;
use std::future::Future;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use iroh::Endpoint;
use tokio::sync::{broadcast, mpsc};
use tokio::task::{JoinHandle, JoinSet};
use uuid::Uuid;

use crate::auth::{
    AuthAdmissionRegistry, AuthBootOptions, AuthCapability, AuthDomain, AuthSudoService, AvesAuthError,
    MachineIssuerId, PairingCoordinator, PairingOwner, messages::IssuePeerJwtInput,
};
use crate::bus::{
    AuthAdmissionConfiguration, BusControlError, BusHandle, BusRpcContext, BusSudoService, EventBus, IrohSubstrate,
    OpaqueRequestEnvelope, PeerConnectivity, RemoteOperationError, RemoteOperationInstaller, RemoteOperationReply,
    RemoteOperationRequest, RunningIrohSubstrate, TransportBindings, call,
};
use crate::config::{ConfigError, ConfigRegistrationSudoService, SealRegistrationInput};
use crate::flow::{
    FlowActivationAcknowledgement, FlowDomain, FlowError, FlowMobilityCommitAcknowledgement, FlowMobilityDestination,
    FlowMobilityService, FlowMobilitySnapshot, FlowSyncServices, FlowYieldAcknowledgement, WorkspaceError,
};

use super::context::{ConstructionFacade, ConstructionInputs, ServiceReader, ServiceSlotRegistry};
use super::domains::DomainDescriptor;

/// Associates an empty domain marker with the service-provider type set
/// generated by that domain's `domain!` declaration.
pub trait DomainProviderMetadata {
    /// Type-level set of sudo-service slots published by this domain.
    type ProviderSet;
}

/// Pre-resolved immutable inputs for a Aves runtime.
pub struct RuntimeOptions {
    /// Application-owned service namespace used by Aves Auth's keyring.
    pub keyring_service_namespace: String,

    /// Opaque, explicitly requested inputs needed during domain construction.
    pub construction_inputs: Box<dyn Any + Send + Sync>,

    /// Application-provided transport factories and route metadata.
    pub transport_bindings: TransportBindings,

    /// Application data root used by Auth and the native Iroh substrate.
    pub data_root: PathBuf,

    /// Application-generated public Bus operation installer.
    pub remote_operation_installer: Option<RemoteOperationInstaller>,
}

/// Reusable runtime supervisor for an installed set of domains.
pub struct Runtime<RootBroadcast>
where
    RootBroadcast: Clone + Send + 'static,
{
    /// Installed domain descriptors.
    domains: Vec<DomainDescriptor<EventBus<RootBroadcast>>>,

    /// Registry for service slots used by the runtime.
    services: ServiceSlotRegistry,

    /// Handle to the runtime's bus.
    bus: BusHandle<RootBroadcast>,

    /// Handle to the auth service.
    auth_service: Arc<dyn AuthSudoService>,

    /// Task that handles auth admission.
    auth_task: JoinHandle<()>,

    /// Registry for auth admission.
    auth_admission_registry: AuthAdmissionRegistry,

    /// Handle to the pairing coordinator.
    pairing_owner: PairingOwner,

    /// Handle to the running Iroh substrate.
    iroh: RunningIrohSubstrate,

    /// Flag indicating whether shutdown has been started.
    shutdown_started: AtomicBool,
}

impl<RootBroadcast> Runtime<RootBroadcast>
where
    RootBroadcast: Clone + Send + 'static,
{
    /// Creates a runtime from opaque construction inputs and installed domains.
    pub async fn new(
        options: RuntimeOptions,
        domains: Vec<DomainDescriptor<EventBus<RootBroadcast>>>,
    ) -> Result<Self, BusControlError> {
        let RuntimeOptions {
            keyring_service_namespace,
            mut construction_inputs,
            transport_bindings,
            data_root,
            remote_operation_installer,
        } = options;
        if keyring_service_namespace.is_empty() {
            return Err(BusControlError::EmptyKeyringServiceNamespace);
        }

        // first we constrcut the bus and the core services registry
        let mut bus = EventBus::new();
        let mut services = ServiceSlotRegistry::new();
        let bus_publisher = services.publisher::<dyn BusSudoService>();

        // next we construct the auth domain and its handoff
        let mut auth_construction = None;
        let auth_options = AuthBootOptions {
            namespace: keyring_service_namespace,
            audience: Uuid::new_v4(),
            peer_config_path: data_root.join("auth").join("peers.json"),
        };
        let auth_domain = DomainDescriptor::for_domain::<AuthDomain>("auth");
        {
            let mut facade = ConstructionFacade::new_with_auth(
                &mut bus,
                &mut services,
                &*construction_inputs,
                &auth_options,
                &mut auth_construction,
            );
            auth_domain.construct(&mut facade).await;
        }

        // The Auth actor supplies the machine key only; public admission is installed
        // after the native substrate and bilateral pairing coordinator exist.
        let auth_construction = auth_construction
            .expect("built-in Auth construction must provide its handoff")
            .map_err(BusControlError::from)?;
        let auth_task = auth_construction.actor_task;
        let registry = auth_construction.admission_registry.clone();
        let peer_connectivity = PeerConnectivity::new(false);
        let substrate = IrohSubstrate::prepare(
            &data_root.join("flows"),
            &auth_construction.machine_secret_key,
            peer_connectivity.clone(),
        )
        .await
        .map_err(|error| BusControlError::Serve {
            protocol: "iroh",
            source: Box::new(error),
        })?;

        let mobility_service: Arc<dyn FlowMobilityService> = Arc::new(RuntimeFlowMobilityService {
            services: substrate.flow_services(),
            pairing_owner: auth_construction.pairing_owner.clone(),
            auth_service: auth_construction.service.clone(),
            source_issuer: auth_construction.machine_secret_key.public().into(),
        });

        if let Some(construction_inputs) = construction_inputs.downcast_mut::<ConstructionInputs>() {
            construction_inputs.insert(substrate.flow_services());
            construction_inputs.insert(mobility_service);
            construction_inputs.insert(peer_connectivity.clone());
        }
        let auth_verifier = auth_construction.service.clone();
        let auth_service = auth_construction.service.clone();

        // once we have auth configured, add it to the buss
        bus.install_auth(AuthAdmissionConfiguration {
            registry: registry.clone(),
            verifier: auth_verifier,
            audience: auth_options.audience,
        });

        // now install the root jwt and context
        let root_jwt = auth_construction.root_jwt;
        let root_context = BusRpcContext::authenticated(bus.input_sender(), root_jwt.clone());

        // once we have the bus and it's auth then we install the normal domains
        let mut installed_domains = Vec::with_capacity(domains.len() + 1);
        installed_domains.push(auth_domain);
        installed_domains.extend(domains);

        // Construction is intentionally sequential: every callback receives
        // the same mutable registration borrow, and a construction panic is
        // fatal without Aves rollback or compensation. The facade contains
        // only the marker-declared typed slots; reads remain late-bound.
        for domain in installed_domains.iter().skip(1) {
            let mut facade = ConstructionFacade::new_with_root_context(
                &mut bus,
                &mut services,
                &*construction_inputs,
                root_context.clone(),
            );
            domain.construct(&mut facade).await;
        }

        // The sole remote dispatcher is installed only after all generated-operation
        // receivers are registered. This keeps startup from admitting a request into
        // a partially composed Poseidon runtime.
        let remote_handler = remote_operation_installer.map(|installer| installer.install(bus.input_sender()));

        let iroh = substrate.start(
            remote_handler,
            auth_construction.first_contact_handler.clone(),
            auth_construction.pairing_owner.clone(),
        );
        let bus = BusHandle::new(bus, transport_bindings, bus_publisher, root_jwt, peer_connectivity).await?;
        Ok(Self {
            domains: installed_domains,
            services,
            bus,
            auth_admission_registry: registry,
            auth_task,
            iroh,
            pairing_owner: auth_construction.pairing_owner,
            auth_service,
            shutdown_started: AtomicBool::new(false),
        })
    }

    /// Calls one generated public operation on a confirmed remote machine.
    pub async fn call_remote_operation(
        &self,
        target: MachineIssuerId,
        qualified_operation: String,
        input: serde_json::Value,
    ) -> Result<serde_json::Value, RemoteOperationError> {
        call_remote_operation_with(
            &self.iroh.remote_endpoint().await?,
            &self.pairing_owner,
            &self.auth_service,
            target,
            qualified_operation,
            input,
        )
        .await
    }

    /// Returns the shared registry used to admit locally and remotely issued JWTs.
    pub fn auth_admission_registry(&self) -> AuthAdmissionRegistry {
        self.auth_admission_registry.clone()
    }

    /// Returns a reader for one compile-time-selected service slot.
    pub fn service_reader<Service>(&self) -> ServiceReader<Service>
    where
        Service: ?Sized + Send + Sync + 'static,
    {
        self.services.declared_reader()
    }

    /// Subscribes to root broadcasts.
    pub fn subscribe_broadcasts(&self) -> broadcast::Receiver<RootBroadcast> {
        self.bus.subscribe_broadcasts()
    }

    /// Returns a clone of the opaque request ingress sender.
    pub fn request_sender(&self) -> mpsc::Sender<OpaqueRequestEnvelope> {
        self.bus.input_sender()
    }

    /// Returns the authenticated normal RPC context, or an unavailable context.
    pub fn rpc_context(&self) -> BusRpcContext {
        self.bus.rpc_context()
    }

    /// Starts every installed domain concurrently, like `Promise.all` over the
    /// descriptor callbacks.
    pub async fn start(&self) -> Result<(), ConfigError> {
        if let Some(config) = self
            .services
            .declared_reader::<dyn ConfigRegistrationSudoService>()
            .get()
        {
            config.seal_registration(SealRegistrationInput {}).await?;
        }
        let mut starts = JoinSet::new();
        for domain in &self.domains {
            starts.spawn(domain.start(self.bus.rpc_context()));
        }

        while starts.join_next().await.is_some() {}
        Ok(())
    }

    /// Stops and drains Sluice network ingress, then stops every installed domain through
    /// live bus ingress before shutting down the native Iroh handlers and bus. Domain
    /// callbacks contain and report their own failures.
    pub async fn shutdown(&self) -> Result<(), BusControlError> {
        if self.shutdown_started.swap(true, Ordering::AcqRel) {
            return Ok(());
        }

        let lifecycle_context = self.bus.rpc_context();
        let _ = tokio::time::timeout(Duration::from_secs(30), async {
            for domain in self
                .domains
                .iter()
                .filter(|domain| domain.name() == FlowDomain::RUNTIME_DOMAIN_NAME)
            {
                domain.stop(lifecycle_context.clone()).await;
            }
            let mut stops = JoinSet::new();
            for domain in self
                .domains
                .iter()
                .filter(|domain| domain.name() != FlowDomain::RUNTIME_DOMAIN_NAME)
            {
                stops.spawn(domain.stop(lifecycle_context.clone()));
            }
            while stops.join_next().await.is_some() {}
        })
        .await;

        let iroh_result = self.iroh.shutdown_router().await;
        self.iroh.close_endpoint().await;
        let bus_result = self.bus.shutdown().await;
        match (iroh_result, bus_result) {
            (Err(error), Ok(())) => {
                self.auth_task.abort();
                Err(BusControlError::Serve {
                    protocol: "iroh",
                    source: Box::new(error),
                })
            }
            (Ok(()), Ok(())) => {
                self.auth_task.abort();
                Ok(())
            }
            (_, Err(error)) => {
                self.shutdown_started.store(false, Ordering::Release);
                Err(error)
            }
        }
    }
}

/// Runtime implementation of the flow mobility service.
struct RuntimeFlowMobilityService {
    /// The flow sync services.
    services: FlowSyncServices,

    /// The auth service.
    auth_service: Arc<dyn AuthSudoService>,

    /// The source issuer.
    source_issuer: MachineIssuerId,

    /// The pairing owner.
    pairing_owner: PairingOwner,
}

impl FlowMobilityService for RuntimeFlowMobilityService {
    fn import_snapshot(
        &self,
        destination: FlowMobilityDestination,
        snapshot: FlowMobilitySnapshot,
    ) -> Pin<Box<dyn Future<Output = Result<FlowMobilityCommitAcknowledgement, FlowError>> + Send + '_>> {
        Box::pin(async move {
            let input = serde_json::json!({ "snapshot": snapshot, "sourcePeerKey": self.source_issuer });

            let output = self
                .call(destination.peer_key, "flow.importMobilitySnapshot", input)
                .await?;

            #[derive(serde::Deserialize)]
            struct CommitOutput {
                acknowledgement: FlowMobilityCommitAcknowledgement,
            }

            serde_json::from_value::<CommitOutput>(output)
                .map(|output| output.acknowledgement)
                .map_err(|error| FlowError::MobilityPeer {
                    message: error.to_string(),
                })
        })
    }

    fn promote_replica(
        &self,
        destination: FlowMobilityDestination,
        acknowledgement: FlowYieldAcknowledgement,
    ) -> Pin<Box<dyn Future<Output = Result<FlowActivationAcknowledgement, FlowError>> + Send + '_>> {
        Box::pin(async move {
            let input = serde_json::json!({ "flowId": acknowledgement.flow_id, "acknowledgement": acknowledgement });

            let output = self
                .call(destination.peer_key, "flow.promoteFlowReplica", input)
                .await?;

            #[derive(serde::Deserialize)]
            struct PromotionOutput {
                acknowledgement: FlowActivationAcknowledgement,
            }

            serde_json::from_value::<PromotionOutput>(output)
                .map(|output| output.acknowledgement)
                .map_err(|error| FlowError::MobilityPeer {
                    message: error.to_string(),
                })
        })
    }
}

impl RuntimeFlowMobilityService {
    async fn call(
        &self,
        target: MachineIssuerId,
        qualified_operation: &str,
        input: serde_json::Value,
    ) -> Result<serde_json::Value, FlowError> {
        let endpoint = self.services.native.read().await.endpoint.clone();
        if !self.services.connectivity.is_enabled() {
            return Err(WorkspaceError::ProviderUnavailable.into());
        }
        call_remote_operation_with(
            &endpoint,
            &self.pairing_owner,
            &self.auth_service,
            target,
            qualified_operation.to_owned(),
            input,
        )
        .await
        .map_err(|error| FlowError::MobilityPeer {
            message: error.to_string(),
        })
    }
}

async fn call_remote_operation_with(
    endpoint: &Endpoint,
    pairing_owner: &PairingOwner,
    auth_service: &Arc<dyn AuthSudoService>,
    target: MachineIssuerId,
    qualified_operation: String,
    input: serde_json::Value,
) -> Result<serde_json::Value, RemoteOperationError> {
    let (peer_endpoint, audience) = PairingCoordinator::refresh_audience(pairing_owner, &target)
        .await
        .map_err(|error| RemoteOperationError::Auth(Box::new(error)))?;

    let output = auth_service
        .issue_peer_jwt(IssuePeerJwtInput {
            target,
            audience,
            sub: "remote-operation".to_owned(),
            cap: vec![AuthCapability::AllBus],
        })
        .await
        .map_err(|error| RemoteOperationError::Auth(Box::new(error)))?;

    let request = RemoteOperationRequest {
        qualified_operation,
        input,
        jwt: output.token,
    };

    match call::<_, RemoteOperationReply<serde_json::Value>>(endpoint, peer_endpoint, &request).await? {
        RemoteOperationReply::Success(output) => Ok(output),
        RemoteOperationReply::DomainError(error) => Err(RemoteOperationError::Domain(error)),
        RemoteOperationReply::Rejected(error) => Err(RemoteOperationError::Rejected(error)),
    }
}

/// Declares one installed domain publication set.
///
/// The declaration emits lifecycle descriptors, one global RSPC router,
/// normal operation descriptors, and separate per-domain MCP publications
/// from the same ordered entries. Each marker must implement
/// [`DomainProviderMetadata`], which is generated by its `domain!` registry.
/// Provider metadata and duplicate-provider mappings are checked at compile
/// time. Each entry also declares the concrete broadcast type and root enum
/// variant used for its broadcast projection.
#[macro_export]
macro_rules! runtime_domains {
    (
        pub $name:ident {
            context: $context:ty,
            root_error: $root_error:ty,
            root_broadcast: $root_broadcast:ty,
            entries: [
                $(
                    $domain:ty => $rpc_router:path,
                    normal: $normal_operations:expr, mcp: $mcp_operations:expr,
                    $server_name:literal, $route_slug:literal,
                    broadcast: $broadcast:ty => $root_variant:ident
                ),* $(,)?
            ] $(,)?
        }
    ) => {
        trait __RuntimeDomainsUniqueProviderSet {}
        $(
            impl __RuntimeDomainsUniqueProviderSet
                for (
                    <$domain as $crate::DomainProviderMetadata>::ProviderSet,
                    $domain,
                )
            {}
        )*
        $(
            const _: fn() = || {
                fn assert_provider_metadata<T: $crate::DomainProviderMetadata>() {}
                assert_provider_metadata::<$domain>();
            };
        )*
        $(
            impl ::std::convert::From<$broadcast> for $root_broadcast {
                fn from(value: $broadcast) -> Self {
                    Self::$root_variant(value)
                }
            }
        )*

        pub struct $name;

        impl $name {
            pub fn lifecycle() -> ::std::vec::Vec<$crate::DomainDescriptor<$crate::bus::EventBus<$root_broadcast>>>
            where
                $(
                    $domain: $crate::runtime::DomainDescriptorFactory<$crate::bus::EventBus<$root_broadcast>>,
                )*
            {
                vec![
                    $(
                        $crate::DomainDescriptor::for_domain::<$domain>(stringify!($domain)),
                    )*
                ]
            }

            pub fn rpc_router() -> ::rspc::Router<$context> {
                ::rspc::Router::<$context>::new()
                    $(.nest($route_slug, $rpc_router()))*
            }

            pub fn normal_operation_descriptors() -> ::std::vec::Vec<
                &'static [$crate::bus::RequestOperationDescriptor<$context, $root_error>],
            > {
                vec![
                    $($normal_operations,)*
                ]
            }

            pub fn mcp_publications() -> ::std::vec::Vec<$crate::DomainPublication<$context, $root_error>> {
                vec![
                    $(
                        $crate::DomainPublication {
                            server_name: $server_name,
                            route_slug: $route_slug,
                            operations: $mcp_operations,
                        },
                    )*
                ]
            }
        }
    };
}

/// Error returned by a generated in-process bus client.
#[derive(Debug)]
pub enum GeneratedCallError<DomainError> {
    /// The request input channel is closed.
    InputClosed,
    /// The request handler dropped its reply channel.
    ReplyDropped,
    /// The domain handler returned its typed error.
    DomainError(DomainError),
    /// Authentication rejected the request before domain routing.
    Authentication(AvesAuthError),
    /// The reply did not contain the response variant declared by the operation.
    UnexpectedResponse { expected: String, actual: String },
}

impl<DomainError> fmt::Display for GeneratedCallError<DomainError>
where
    DomainError: fmt::Display,
{
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InputClosed => formatter.write_str("bus input is closed"),
            Self::ReplyDropped => formatter.write_str("bus reply was dropped"),
            Self::DomainError(error) => error.fmt(formatter),
            Self::Authentication(error) => error.fmt(formatter),
            Self::UnexpectedResponse { expected, actual } => {
                write!(formatter, "expected {expected}, got {actual}")
            }
        }
    }
}

impl<DomainError> std::error::Error for GeneratedCallError<DomainError>
where
    DomainError: std::error::Error + 'static,
{
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::DomainError(error) => Some(error),
            Self::Authentication(error) => Some(error),
            Self::InputClosed | Self::ReplyDropped | Self::UnexpectedResponse { .. } => None,
        }
    }
}

/// Error produced when a generated response cannot be projected to its variant.
#[derive(Debug, thiserror::Error)]
#[error("expected {expected}, got {actual}")]
pub struct ResponseProjectionError {
    /// The response variant requested by the operation.
    pub expected: String,
    /// Debug representation of the actual response.
    pub actual: String,
}

impl<DomainError> From<ResponseProjectionError> for GeneratedCallError<DomainError> {
    fn from(error: ResponseProjectionError) -> Self {
        Self::UnexpectedResponse {
            expected: error.expected,
            actual: error.actual,
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicUsize, Ordering};

    use serde::Serialize;
    use specta::Type;
    use thiserror::Error;

    use crate::bus::BusRpcContext;

    use super::super::context::ConstructionFacade;
    use super::super::domains::{DomainCallbacks, DomainDescriptorFactory, DomainFuture, dispatch_lifecycle_request};
    use super::*;

    #[cfg(feature = "keyring")]
    use std::sync::atomic::AtomicBool;

    #[cfg(feature = "keyring")]
    use tokio::sync::oneshot;
    #[cfg(feature = "keyring")]
    use uuid::Uuid;

    #[cfg(feature = "keyring")]
    use super::super::context::ConstructionInputs;
    #[cfg(feature = "keyring")]
    use crate::bus::{EventBus, OpaqueRequestEnvelope, TransportBindings};

    #[cfg(feature = "keyring")]
    use super::{DomainDescriptor, Runtime, RuntimeOptions};

    pub struct Alpha;
    pub struct Beta;

    static STOP_COUNT: AtomicUsize = AtomicUsize::new(0);

    #[derive(Debug, Error, Serialize, Type)]
    #[error("test root error")]
    pub struct TestRootError;

    impl ::rspc::Error for TestRootError {
        fn into_procedure_error(self) -> ::rspc::ProcedureError {
            ::rspc::ProcedureError::Resolver(::rspc::ResolverError::new(self, None::<Self>))
        }
    }

    impl DomainProviderMetadata for Alpha {
        type ProviderSet = ();
    }

    impl DomainProviderMetadata for Beta {
        type ProviderSet = ();
    }

    #[derive(Debug, Clone, PartialEq)]
    pub enum AlphaBroadcast {
        Started,
    }

    #[derive(Debug, Clone, PartialEq)]
    pub enum BetaBroadcast {
        Finished,
    }

    #[derive(Debug, Clone, PartialEq)]
    pub enum TestRootBroadcast {
        Alpha(AlphaBroadcast),
        Beta(BetaBroadcast),
    }

    impl<Bus: ?Sized> DomainDescriptorFactory<Bus> for Alpha {
        fn callbacks() -> DomainCallbacks<Bus> {
            fn construct<'facade, 'bus, Bus: ?Sized>(
                _: &'facade mut ConstructionFacade<'bus, Bus>,
            ) -> DomainFuture<'facade> {
                Box::pin(async {})
            }
            fn start(_: BusRpcContext) -> DomainFuture<'static> {
                Box::pin(async {})
            }
            fn stop(_: BusRpcContext) -> DomainFuture<'static> {
                STOP_COUNT.fetch_add(1, Ordering::SeqCst);
                Box::pin(async {})
            }
            DomainCallbacks {
                new: construct::<Bus>,
                start,
                stop,
            }
        }
    }

    impl<Bus: ?Sized> DomainDescriptorFactory<Bus> for Beta {
        fn callbacks() -> DomainCallbacks<Bus> {
            fn construct<'facade, 'bus, Bus: ?Sized>(
                _: &'facade mut ConstructionFacade<'bus, Bus>,
            ) -> DomainFuture<'facade> {
                Box::pin(async {})
            }
            fn start(_: BusRpcContext) -> DomainFuture<'static> {
                Box::pin(async {})
            }
            fn stop(_: BusRpcContext) -> DomainFuture<'static> {
                STOP_COUNT.fetch_add(1, Ordering::SeqCst);
                Box::pin(async {})
            }
            DomainCallbacks {
                new: construct::<Bus>,
                start,
                stop,
            }
        }
    }

    #[cfg(feature = "keyring")]
    struct RegistrationMarker;
    #[cfg(feature = "keyring")]
    struct RegistrationRequest {
        reply_to: oneshot::Sender<u8>,
    }

    #[cfg(feature = "keyring")]
    struct ShutdownMarker;
    #[cfg(feature = "keyring")]
    struct ShutdownRequest {
        reply_to: oneshot::Sender<Result<(), ()>>,
    }
    #[cfg(feature = "keyring")]
    static SHUTDOWN_TEST_STOP_COUNT: AtomicUsize = AtomicUsize::new(0);
    #[cfg(feature = "keyring")]
    static SHUTDOWN_TEST_DELIVERED: AtomicBool = AtomicBool::new(false);

    #[cfg(feature = "keyring")]
    fn registration_domain<'facade, 'bus>(
        facade: &'facade mut ConstructionFacade<'bus, EventBus<TestRootBroadcast>>,
    ) -> DomainFuture<'facade> {
        let channels = facade
            .bus_mut()
            .register::<RegistrationMarker, RegistrationRequest, AlphaBroadcast>();
        let mut requests_rx = channels.requests_rx;
        tokio::spawn(async move {
            if let Some(request) = requests_rx.recv().await {
                let _ = request.0.reply_to.send(7);
            }
        });
        Box::pin(async {})
    }

    #[cfg(feature = "keyring")]
    fn runtime_test_bindings() -> TransportBindings {
        TransportBindings::new(
            |_, _, shutdown| {
                Box::pin(async move {
                    let _ = shutdown.await;
                    Ok(())
                })
            },
            |_, _, shutdown| {
                Box::pin(async move {
                    let _ = shutdown.await;
                    Ok(())
                })
            },
            Vec::new(),
        )
    }

    #[cfg(feature = "keyring")]
    fn runtime_test_options() -> RuntimeOptions {
        RuntimeOptions {
            keyring_service_namespace: "aves-test".to_owned(),
            construction_inputs: Box::new(ConstructionInputs::new()),
            transport_bindings: runtime_test_bindings(),
            data_root: std::env::temp_dir().join(format!("aves-runtime-test-{}", Uuid::new_v4())),
            remote_operation_installer: None,
        }
    }

    #[cfg(feature = "keyring")]
    fn shutdown_test_construct<'facade, 'bus>(
        facade: &'facade mut ConstructionFacade<'bus, EventBus<TestRootBroadcast>>,
    ) -> DomainFuture<'facade> {
        let channels = facade
            .bus_mut()
            .register::<ShutdownMarker, ShutdownRequest, AlphaBroadcast>();
        let mut requests_rx = channels.requests_rx;
        tokio::spawn(async move {
            while let Some(request) = requests_rx.recv().await {
                SHUTDOWN_TEST_DELIVERED.store(true, Ordering::SeqCst);
                let _ = request.0.reply_to.send(Ok(()));
            }
        });
        Box::pin(async {})
    }

    #[cfg(feature = "keyring")]
    fn shutdown_test_stop(context: BusRpcContext) -> DomainFuture<'static> {
        Box::pin(async move {
            let result = dispatch_lifecycle_request::<ShutdownMarker, ShutdownRequest, (), ()>(context, |reply_to| {
                ShutdownRequest { reply_to }
            })
            .await;
            assert_eq!(result, Some(Ok(())));
            SHUTDOWN_TEST_STOP_COUNT.fetch_add(1, Ordering::SeqCst);
        })
    }

    #[cfg(feature = "keyring")]
    fn shutdown_test_noop_construct<'facade, 'bus>(
        _: &'facade mut ConstructionFacade<'bus, EventBus<TestRootBroadcast>>,
    ) -> DomainFuture<'facade> {
        Box::pin(async {})
    }

    #[tokio::test]
    #[cfg(feature = "keyring")]
    async fn runtime_finalizes_bus_after_registration_and_exposes_only_projections() {
        let descriptor = DomainDescriptor::from_callbacks(
            "Registration",
            DomainCallbacks {
                new: registration_domain,
                start: |_| Box::pin(async {}),
                stop: |_| Box::pin(async {}),
            },
        );
        let runtime = Runtime::<TestRootBroadcast>::new(runtime_test_options(), vec![descriptor])
            .await
            .expect("runtime bus finalization should succeed");

        let _request_sender: tokio::sync::mpsc::Sender<OpaqueRequestEnvelope> = runtime.request_sender();
        let _broadcasts = runtime.subscribe_broadcasts();
        let (reply_to, reply) = oneshot::channel();
        runtime
            .request_sender()
            .send(OpaqueRequestEnvelope::new::<RegistrationMarker, _>(
                runtime
                    .rpc_context()
                    .jwt()
                    .expect("root JWT should be available"),
                RegistrationRequest { reply_to },
            ))
            .await
            .expect("finalized bus ingress should remain live");
        assert_eq!(
            reply
                .await
                .expect("registered request handler should reply"),
            7
        );
        assert!(runtime.rpc_context().admit().is_ok());
        assert!(
            runtime
                .service_reader::<dyn crate::bus::BusSudoService>()
                .get()
                .is_some()
        );
        runtime
            .shutdown()
            .await
            .expect("runtime shutdown should succeed");
    }

    #[tokio::test]
    #[cfg(feature = "keyring")]
    async fn runtime_shutdown_stops_domains_before_bus_and_is_idempotent() {
        SHUTDOWN_TEST_STOP_COUNT.store(0, Ordering::SeqCst);
        SHUTDOWN_TEST_DELIVERED.store(false, Ordering::SeqCst);
        let runtime = Runtime::<TestRootBroadcast>::new(
            runtime_test_options(),
            vec![
                DomainDescriptor::from_callbacks(
                    "ShutdownRegistration",
                    DomainCallbacks {
                        new: shutdown_test_construct,
                        start: |_| Box::pin(async {}),
                        stop: shutdown_test_stop,
                    },
                ),
                DomainDescriptor::from_callbacks(
                    "ShutdownSecond",
                    DomainCallbacks {
                        new: shutdown_test_noop_construct,
                        start: |_| Box::pin(async {}),
                        stop: shutdown_test_stop,
                    },
                ),
            ],
        )
        .await
        .expect("runtime bus finalization should succeed");

        runtime
            .shutdown()
            .await
            .expect("runtime shutdown should succeed");
        runtime
            .shutdown()
            .await
            .expect("second shutdown should be idempotent");
        assert_eq!(SHUTDOWN_TEST_STOP_COUNT.load(Ordering::SeqCst), 2);
        assert!(SHUTDOWN_TEST_DELIVERED.load(Ordering::SeqCst));
    }

    mod alpha {
        pub fn rpc_router() -> ::rspc::Router<()> {
            ::rspc::Router::<()>::new().procedure(
                "alphaProcedure",
                ::rspc::Procedure::builder::<super::TestRootError>()
                    .query(|_, _: ()| async { Ok::<(), super::TestRootError>(()) }),
            )
        }
    }

    mod beta {
        pub fn rpc_router() -> ::rspc::Router<()> {
            ::rspc::Router::<()>::new().procedure(
                "betaProcedure",
                ::rspc::Procedure::builder::<super::TestRootError>()
                    .query(|_, _: ()| async { Ok::<(), super::TestRootError>(()) }),
            )
        }
    }

    static ALPHA_OPERATIONS: &[crate::bus::RequestOperationDescriptor<(), TestRootError>] = &[];
    static BETA_OPERATIONS: &[crate::bus::RequestOperationDescriptor<(), TestRootError>] = &[];

    aves::runtime_domains! {
        pub TestDomains {
            context: (),
            root_error: TestRootError,
            root_broadcast: TestRootBroadcast,
            entries: [
                Alpha => alpha::rpc_router, normal: ALPHA_OPERATIONS, mcp: ALPHA_OPERATIONS, "Alpha", "alpha", broadcast: AlphaBroadcast => Alpha,
                Beta => beta::rpc_router, normal: BETA_OPERATIONS, mcp: BETA_OPERATIONS, "Beta", "beta", broadcast: BetaBroadcast => Beta,
            ]
        }
    }

    #[tokio::test]
    async fn one_declaration_emits_ordered_lifecycle_router_and_mcp_publications() {
        let lifecycle = TestDomains::lifecycle();
        assert_eq!(lifecycle.len(), 2);
        assert_eq!(lifecycle[0].name(), "Alpha");
        assert_eq!(lifecycle[1].name(), "Beta");

        let global_router = TestDomains::rpc_router();
        let paths = (&global_router)
            .into_iter()
            .map(|(path, _)| {
                path.iter()
                    .map(ToString::to_string)
                    .collect::<Vec<_>>()
                    .join("/")
            })
            .collect::<std::collections::BTreeSet<_>>();
        assert!(paths.contains("alpha/alphaProcedure"));
        assert!(paths.contains("beta/betaProcedure"));

        let normal_operations = TestDomains::normal_operation_descriptors();
        assert_eq!(normal_operations.len(), 2);
        assert!(std::ptr::eq(normal_operations[0], ALPHA_OPERATIONS));
        assert!(std::ptr::eq(normal_operations[1], BETA_OPERATIONS));

        let publications = TestDomains::mcp_publications();
        assert_eq!(publications.len(), 2);
        assert_eq!(publications[0].route_slug, "alpha");
        assert_eq!(publications[0].server_name, "Alpha");
        assert_eq!(publications[1].route_slug, "beta");
        assert_eq!(publications[1].server_name, "Beta");
        assert!(std::ptr::eq(publications[0].operations, ALPHA_OPERATIONS));
        assert!(std::ptr::eq(publications[1].operations, BETA_OPERATIONS));
    }

    #[test]
    fn declaration_projects_concrete_broadcasts_into_root_variants() {
        assert_eq!(
            TestRootBroadcast::from(AlphaBroadcast::Started),
            TestRootBroadcast::Alpha(AlphaBroadcast::Started)
        );
        assert_eq!(
            TestRootBroadcast::from(BetaBroadcast::Finished),
            TestRootBroadcast::Beta(BetaBroadcast::Finished)
        );
    }
}