saddle/

application.rs

use std::{marker::PhantomData, net::SocketAddr, sync::Arc};

use saddle_core::ErrorKind;
use saddle_db::{
    DatabaseConfig,
    internal::{StartupManagedDatabaseFactory, StartupManagedDatabaseOwner},
};
use saddle_observability::ObserverConfig;
use saddle_runtime::startup_assembly::{
    ActualStartupOwners, BootstrapBatchSeal, BootstrapInstallAdapter, BootstrapOwnerView,
    BootstrapTransaction, PreparedBootstrapBatch, PreparedBootstrapInstall, ReadyBootstrapInstall,
    StartupBootstrapOwners, run_with_actual_startup_owners,
};

use crate::{
    Result, SaddleError,
    http1::{
        bootstrap::{
            GeneratedApplicationBootstrap, GeneratedBootstrapToken, GeneratedCompiledAdapter,
            GeneratedContextFactory, ProductionBundleFacts, ProductionServiceBundle,
            generated_bootstrap_type_identity,
        },
        production::{
            BoundPreparedHttp1, GeneratedTransportBinding, PreparedHttp1, ProductionHttp1,
            StagedHttp1, stage,
        },
    },
};

/// Facade-issued seal proving that one generated application owner was
/// synchronously split before any managed component was constructed.
pub struct GeneratedApplicationSeal {
    _private: (),
}

/// The two non-replayable halves emitted by one generated application owner.
///
/// Fields stay private so generated code can only construct this value through
/// the facade-issued seal.
pub struct GeneratedApplicationParts<B> {
    bootstrap: B,
    runtime_preflight: saddle_runtime::capacity_leaf::VerifiedRuntimeLayoutPreflight,
    termination: FacadeGeneratedTermination,
    continuation: FacadeGeneratedContinuation,
}

pub type GeneratedBoundApplicationParts<A, C, const ROUTES: usize> =
    GeneratedApplicationParts<GeneratedProductionBootstrap<A, C, ROUTES>>;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum GeneratedApplicationBindingError {
    ForeignGeneration,
    ForeignBootstrapType,
    RouteCountMismatch,
    MissingIdentity,
    PreflightFailed,
}

/// Facade-owned opaque coordinator input. It exposes no Runtime, calibration,
/// resource or deployment observation to generated code.
pub struct FacadeBootstrapCoordinator {
    config: SaddleConfig,
    adapter_provenance: [u8; 32],
}

mod generated_bootstrap_sealed {
    pub trait Sealed {}
}

/// Sealed type-level bridge from one generated owner to the existing generic
/// bootstrap. Only the Facade implementation can invoke the generic consumer.
trait GeneratedProductionBootstrapCapability:
    generated_bootstrap_sealed::Sealed + Send + 'static
{
    fn run<R, H>(
        self,
        coordinator: FacadeBootstrapCoordinator,
        actual: ActualStartupOwners<StartupManagedDatabaseFactory>,
        continuation: saddle_admission::LateStartupContinuationOwner<FacadeGeneratedContinuation>,
        authorization: PendingListenerAuthorization<R, H>,
        candidate: crate::startup_candidate::PreparedStartupCandidateTransaction,
    ) -> Result<()>
    where
        R: ApprovedBundleNonceReservation,
        H: ListenerAuthorityProvider;
}

/// Private-field, non-Clone capability carrying the fully checked staged HTTP
/// graph, concrete Rust types, fixed route count and binder identity.
///
/// ```compile_fail
/// # use saddle::GeneratedProductionBootstrap;
/// fn replay<A, C, const N: usize>(value: GeneratedProductionBootstrap<A, C, N>) {
///     let _copy = value.clone();
/// }
/// ```
///
/// ```compile_fail
/// # use saddle::GeneratedProductionBootstrap;
/// let _forged: GeneratedProductionBootstrap<(), (), 6> =
///     GeneratedProductionBootstrap { generation: 1, type_identity: [1; 32], _types: core::marker::PhantomData };
/// ```
pub struct GeneratedProductionBootstrap<A, C, const ROUTES: usize>
where
    A: GeneratedCompiledAdapter,
    C: GeneratedContextFactory<A::Context>,
{
    generation: u64,
    type_identity: [u8; 32],
    staged: StagedHttp1<A, C>,
}

impl<A, C, const ROUTES: usize> generated_bootstrap_sealed::Sealed
    for GeneratedProductionBootstrap<A, C, ROUTES>
where
    A: GeneratedCompiledAdapter,
    C: GeneratedContextFactory<A::Context>,
{
}

impl<A, C, const ROUTES: usize> GeneratedProductionBootstrapCapability
    for GeneratedProductionBootstrap<A, C, ROUTES>
where
    A: ProductionServiceBundle,
    C: GeneratedContextFactory<A::Context>,
{
    fn run<R, H>(
        self,
        coordinator: FacadeBootstrapCoordinator,
        actual: ActualStartupOwners<StartupManagedDatabaseFactory>,
        continuation: saddle_admission::LateStartupContinuationOwner<FacadeGeneratedContinuation>,
        authorization: PendingListenerAuthorization<R, H>,
        candidate: crate::startup_candidate::PreparedStartupCandidateTransaction,
    ) -> Result<()>
    where
        R: ApprovedBundleNonceReservation,
        H: ListenerAuthorityProvider,
    {
        let _binding = (self.generation, self.type_identity);
        run_with_actual_startup_owners(
            actual,
            FacadeContinuationAdapter {
                provenance: coordinator.adapter_provenance,
            },
            continuation,
            move |owners, runtime_profile| async move {
                let transaction = BootstrapTransaction::new(owners);
                let prepared = transaction
                    .prepare(ProductionInstall {
                        config: coordinator.config,
                        staged: self.staged,
                        runtime_profile,
                        candidate,
                    })
                    .map_err(|failure| {
                        failure.into_runner_failure(startup_error(
                            "saddle.production_bootstrap_prepare_failed",
                        ))
                    })?;
                let prepared = prepared.prepare_observability().await?;
                let mut prepared = prepared;
                if prepared.prepare_install().await.is_err() {
                    return Err(
                        prepared.fail(startup_error("saddle.production_listener_prepare_failed"))
                    );
                }
                let prepared = prepared.verify_post_driver_binding().map_err(|failure| {
                    failure.into_runner_failure(startup_error(
                        "saddle.production_post_driver_binding_failed",
                    ))
                })?;
                let prepared = authorize_listener(
                    prepared,
                    authorization.binding,
                    authorization.reservation,
                    authorization.authority,
                )
                .map_err(|error| error.value.fail(error.error))?;
                let prepared = prepared.finish_prepare_install();
                Ok(prepared.commit())
            },
        )
    }
}

struct FacadeContinuationAdapter {
    provenance: [u8; 32],
}

impl saddle_admission::StartupContinuationAdapter for FacadeContinuationAdapter {
    type ContinuationOwner =
        saddle_admission::LateStartupContinuationOwner<FacadeGeneratedContinuation>;

    fn adapter_provenance(&self) -> [u8; 32] {
        self.provenance
    }
}

struct ProductionInstall<A, C>
where
    A: GeneratedCompiledAdapter,
    C: GeneratedContextFactory<A::Context>,
{
    config: SaddleConfig,
    staged: StagedHttp1<A, C>,
    runtime_profile: saddle_runtime::startup_assembly::VerifiedTransportRuntimeProfile,
    candidate: crate::startup_candidate::PreparedStartupCandidateTransaction,
}

struct PreparedProductionInstall<A, C>
where
    A: GeneratedCompiledAdapter,
    C: GeneratedContextFactory<A::Context>,
{
    config: SaddleConfig,
    prepared: PreparedHttp1<A, C>,
    listener: Option<tokio::net::TcpListener>,
    candidate: crate::startup_candidate::PreparedStartupCandidateTransaction,
}

struct ReadyProductionInstall<A, C>
where
    A: GeneratedCompiledAdapter,
    C: GeneratedContextFactory<A::Context>,
{
    config: SaddleConfig,
    prepared: BoundPreparedHttp1<A, C>,
    candidate: crate::startup_candidate::PreparedStartupCandidateTransaction,
}

impl<A, C, B> BootstrapInstallAdapter<StartupManagedDatabaseOwner, B> for ProductionInstall<A, C>
where
    A: GeneratedCompiledAdapter,
    C: GeneratedContextFactory<A::Context>,
    B: Send + 'static,
{
    type Prepared = PreparedProductionInstall<A, C>;
    type Error = SaddleError;

    fn prepare(
        self,
        _owners: BootstrapOwnerView<'_, StartupManagedDatabaseOwner, B>,
    ) -> std::result::Result<Self::Prepared, Self::Error> {
        Ok(PreparedProductionInstall {
            config: self.config,
            prepared: self.staged.prepare(self.runtime_profile)?,
            listener: None,
            candidate: self.candidate,
        })
    }
}

impl<A, C, B> PreparedBootstrapInstall<StartupManagedDatabaseOwner, B>
    for PreparedProductionInstall<A, C>
where
    A: GeneratedCompiledAdapter,
    C: GeneratedContextFactory<A::Context>,
    B: Send + 'static,
{
    type Ready = ReadyProductionInstall<A, C>;
    type Error = SaddleError;

    fn component_names(&self) -> &'static [&'static str] {
        &["observability", "database", "http1-production"]
    }

    async fn prepare_install(&mut self) -> std::result::Result<(), Self::Error> {
        self.listener = Some(
            tokio::net::TcpListener::bind(self.config.listen)
                .await
                .map_err(|_| startup_error("saddle.production_listener_bind_failed"))?,
        );
        Ok(())
    }

    fn into_ready(self) -> Self::Ready {
        ReadyProductionInstall {
            config: self.config,
            prepared: self
                .prepared
                .bind_owned(self.listener.unwrap_or_else(|| std::process::abort())),
            candidate: self.candidate,
        }
    }
}

impl<A, C, B> ReadyBootstrapInstall<StartupManagedDatabaseOwner, B> for ReadyProductionInstall<A, C>
where
    A: GeneratedCompiledAdapter,
    C: GeneratedContextFactory<A::Context>,
    B: Send + 'static,
{
    fn install(
        self,
        owners: StartupBootstrapOwners<StartupManagedDatabaseOwner, B>,
        mut seal: BootstrapBatchSeal,
    ) -> PreparedBootstrapBatch {
        let (
            database_owner,
            db_domain,
            tokio_domain,
            allocation,
            ledger,
            _termination,
            bound,
            _transport_binding,
            observability,
        ) = owners.into_parts();
        let (database_owner, database) = database_owner.bootstrap(|database| database.existing());
        let publish = bound.into_publish_token();
        let (listener, prepared) = self.prepared.into_parts();
        let assembly = prepared.attach(database, ledger, tokio_domain, allocation, db_domain);
        let server = ProductionHttp1::from_staged(
            listener,
            assembly,
            publish,
            seal.take_driver_finalizer(),
            self.candidate.commit(),
        );
        let observability: Arc<dyn saddle_core::ComponentLifecycle> = match observability {
            Some(saddle_runtime::startup_assembly::StartupObservabilityOwner::Started(owner)) => {
                owner
            }
            Some(saddle_runtime::startup_assembly::StartupObservabilityOwner::Prepared(_)) => {
                std::process::abort()
            }
            None => std::process::abort(),
        };
        let components: Vec<Arc<dyn saddle_core::ComponentLifecycle>> =
            vec![observability, Arc::new(database_owner), Arc::new(server)];
        let _application = self.config.application;
        seal.install(components)
    }
}

struct FacadeGeneratedContinuation {
    common: [[u8; 32]; 3],
    generation: u64,
}

impl saddle_admission::PreflightGeneratedStaticContinuationOwner for FacadeGeneratedContinuation {
    fn build_identity(&self) -> [u8; 32] {
        self.common[0]
    }
    fn artifact_identity(&self) -> [u8; 32] {
        self.common[1]
    }
    fn route_set_identity(&self) -> [u8; 32] {
        self.common[2]
    }
    fn owner_generation(&self) -> u64 {
        self.generation
    }
}

struct FacadeGeneratedTermination {
    common: [[u8; 32]; 3],
    generation: u64,
    topology: [u64; 4],
}

fn facade_generated_identity(common: [[u8; 32]; 3], tag: u8) -> [u8; 32] {
    let mut identity = common[0];
    for (index, byte) in identity.iter_mut().enumerate() {
        *byte ^= common[1][index].rotate_left(1) ^ common[2][index].rotate_left(2) ^ tag;
    }
    identity
}

impl saddle_admission::GeneratedTerminationTopologyWorkOwner for FacadeGeneratedTermination {
    fn leaf_identity(&self) -> [u8; 32] {
        facade_generated_identity(self.common, 0x71)
    }
    fn common_identities(&self) -> [[u8; 32]; 3] {
        self.common
    }
    fn owner_generation(&self) -> u64 {
        self.generation
    }
    fn termination_topology(&self) -> [u64; 4] {
        self.topology
    }
    fn termination_topology_identity(&self) -> [u8; 32] {
        facade_generated_identity(self.common, 0x72)
    }
    fn db_return_work_identity(&self) -> [u8; 32] {
        saddle_db::internal::db_normal_return_work_proof().identity()
    }
    fn writer_work_identity(&self) -> [u8; 32] {
        saddle_observability::file::writer_component_work_proof().identity()
    }
    fn runtime_work_identity(&self) -> [u8; 32] {
        saddle_runtime::termination_service::runtime_component_work_proof().identity()
    }
}

struct FacadeRuntimeLayouts<F, C, const OUTPUT: usize> {
    common: [[u8; 32]; 3],
    generation: u64,
    routes: Box<[u64]>,
    types: PhantomData<fn() -> (F, C)>,
}

impl<F, C, const OUTPUT: usize> saddle_runtime::capacity_leaf::RuntimeMonomorphizedLayoutSource
    for FacadeRuntimeLayouts<F, C, OUTPUT>
where
    F: Send + 'static,
    C: Send + 'static,
{
    fn common_identities(&self) -> [[u8; 32]; 3] {
        self.common
    }
    fn owner_generation(&self) -> u64 {
        self.generation
    }
    fn solver_schema_identity(&self) -> [u8; 32] {
        saddle_admission::admission_capacity_contract_identities()[0]
    }
    fn admission_layout_identity(&self) -> [u8; 32] {
        saddle_admission::admission_capacity_contract_identities()[2]
    }
    fn measure(
        self,
        seal: saddle_runtime::capacity_leaf::RuntimeLayoutSeal,
    ) -> std::result::Result<
        Box<[saddle_runtime::capacity_leaf::RuntimeRouteLayout]>,
        saddle_runtime::capacity_leaf::RuntimeCapacityLeafError,
    > {
        self.routes
            .into_iter()
            .map(|route| {
                seal.measure::<F, C, saddle_admission::ManagedResponse, [u8; OUTPUT]>(route)
            })
            .collect()
    }
}

impl GeneratedApplicationSeal {
    fn issue() -> Self {
        Self { _private: () }
    }

    /// Issues the sole token used by generated code for fallible preflight.
    /// The token contains no deployment or Runtime observation.
    pub fn preflight_token(&self) -> GeneratedBootstrapToken {
        GeneratedBootstrapToken::issue()
    }

    /// Consumes Service's indivisible same-freeze execution/capacity bundle.
    /// Runtime layout, termination and continuation proofs are Facade-issued.
    pub fn bind<E, BC, F, C, const BODY: usize, const OUTPUT: usize, const ROUTES: usize>(
        self,
        generated: GeneratedApplicationBootstrap<
            saddle_service::internal::CompiledExecutionWithCapacityLeaf<E, BC, F, BODY, OUTPUT>,
            C,
            ROUTES,
        >,
    ) -> std::result::Result<
        GeneratedBoundApplicationParts<
            saddle_service::internal::CompiledExecutionWithCapacityLeaf<E, BC, F, BODY, OUTPUT>,
            C,
            ROUTES,
        >,
        GeneratedApplicationBindingError,
    >
    where
        E: Fn(
                usize,
                BC,
                saddle_admission::ManagedBytes,
                saddle_service::internal::CompiledDbPermit,
                saddle_admission::ManagedResponseBuilder,
            ) -> F
            + Send
            + Sync
            + 'static,
        BC: Send + Unpin + 'static,
        F: std::future::Future<
                Output = std::result::Result<
                    saddle_admission::ManagedResponse,
                    saddle_service::internal::ExecutionError,
                >,
            > + Send
            + 'static,
        C: GeneratedContextFactory<BC>,
    {
        let ProductionBundleFacts {
            common,
            generation,
            routes,
            termination_topology: topology,
        } = generated.production_bundle_facts();
        let type_identity = generated_bootstrap_type_identity::<
            saddle_service::internal::CompiledExecutionWithCapacityLeaf<E, BC, F, BODY, OUTPUT>,
            C,
            ROUTES,
        >();
        if generation == 0
            || ROUTES == 0
            || routes.len() != ROUTES
            || [common[0], common[1], common[2], type_identity].contains(&[0; 32])
        {
            return Err(GeneratedApplicationBindingError::MissingIdentity);
        }
        let runtime_preflight = saddle_runtime::capacity_leaf::verify_runtime_layout_preflight(
            saddle_runtime::capacity_leaf::runtime_capacity_component_proof(),
            FacadeRuntimeLayouts::<F, BC, OUTPUT> {
                common,
                generation,
                routes,
                types: PhantomData,
            },
        )
        .map_err(|_| GeneratedApplicationBindingError::PreflightFailed)?;
        let staged = stage(
            generated,
            GeneratedTransportBinding {
                generation,
                build_identity: common[0],
                route_set_identity: common[2],
                static_layout_identity: type_identity,
            },
        )
        .map_err(|_| GeneratedApplicationBindingError::PreflightFailed)?;
        Ok(GeneratedApplicationParts {
            bootstrap: GeneratedProductionBootstrap {
                generation,
                type_identity,
                staged,
            },
            runtime_preflight,
            termination: FacadeGeneratedTermination {
                common,
                generation,
                topology,
            },
            continuation: FacadeGeneratedContinuation { common, generation },
        })
    }
}

mod generated_consumer_sealed {
    pub trait Sealed {}
}

/// Facade-owned inversion point. Generated code chooses `A/C/ROUTES` inside
/// its method body and immediately moves the private staged capability here;
/// the anonymous dispatcher Future never appears in an associated type.
#[doc(hidden)]
pub trait GeneratedApplicationConsumer: generated_consumer_sealed::Sealed {
    fn consume<E, BC, F, C, const BODY: usize, const OUTPUT: usize, const ROUTES: usize>(
        self,
        parts: GeneratedBoundApplicationParts<
            saddle_service::internal::CompiledExecutionWithCapacityLeaf<E, BC, F, BODY, OUTPUT>,
            C,
            ROUTES,
        >,
    ) -> Result<()>
    where
        E: Fn(
                usize,
                BC,
                saddle_admission::ManagedBytes,
                saddle_service::internal::CompiledDbPermit,
                saddle_admission::ManagedResponseBuilder,
            ) -> F
            + Send
            + Sync
            + 'static,
        BC: Send + Unpin + 'static,
        F: std::future::Future<
                Output = std::result::Result<
                    saddle_admission::ManagedResponse,
                    saddle_service::internal::ExecutionError,
                >,
            > + Send
            + 'static,
        C: GeneratedContextFactory<BC>;
}

/// The sole generated application owner accepted by the 0.2 facade.
///
/// There is deliberately no blanket implementation for closures.
///
/// ```compile_fail
/// # use saddle::GeneratedApplicationOwner;
/// fn require_generated_owner<T: GeneratedApplicationOwner>(_: T) {}
/// require_generated_owner(|| ());
/// ```
///
/// ```compile_fail
/// # use saddle_admission::{ComposedGeneratedStartupFactsOwner, ServiceCapacitySourceLeaf};
/// fn require_service_leaf<T: ServiceCapacitySourceLeaf>(_: T) {}
/// fn old_combined_owner(owner: ComposedGeneratedStartupFactsOwner) {
///     require_service_leaf(owner);
/// }
/// ```
pub trait GeneratedApplicationOwner: Send + 'static {
    fn consume<K>(self, seal: GeneratedApplicationSeal, consumer: K) -> Result<()>
    where
        K: GeneratedApplicationConsumer;
}

/// Facade-issued seal for the independently Gate-owned deployment halves.
pub struct FacadeDeploymentSeal {
    _private: (),
}

pub struct FacadeDeploymentParts<C, E, D, F, S, U> {
    calibration: C,
    envelope: E,
    db_service: D,
    filesystem_service: F,
    scheduler_service: S,
    supervisor_service: U,
    policy: saddle_runtime::resource_envelope::FiniteStartupPolicy,
    adapter_provenance: [u8; 32],
}

pub type FacadeDeploymentOwnerParts<O> = FacadeDeploymentParts<
    <O as FacadeDeploymentOwner>::Calibration,
    <O as FacadeDeploymentOwner>::Envelope,
    <O as FacadeDeploymentOwner>::DbService,
    <O as FacadeDeploymentOwner>::FilesystemService,
    <O as FacadeDeploymentOwner>::SchedulerService,
    <O as FacadeDeploymentOwner>::SupervisorService,
>;

impl FacadeDeploymentSeal {
    fn issue() -> Self {
        Self { _private: () }
    }

    #[allow(clippy::too_many_arguments)]
    pub fn bind<C, E, D, F, S, U>(
        self,
        calibration: C,
        envelope: E,
        db_service: D,
        filesystem_service: F,
        scheduler_service: S,
        supervisor_service: U,
        policy: saddle_runtime::resource_envelope::FiniteStartupPolicy,
        adapter_provenance: [u8; 32],
    ) -> FacadeDeploymentParts<C, E, D, F, S, U>
    where
        C: saddle_admission::RuntimeCapacityCalibrationSourceLeaf,
        E: Into<saddle_runtime::resource_envelope::VerifiedResourceEnvelope>,
        D: saddle_admission::VerifiedDbTerminationServiceProofOwner,
        F: Send + 'static,
        S: saddle_admission::VerifiedSchedulerTerminationServiceProofOwner,
        U: saddle_admission::VerifiedSupervisorTerminationServiceProofOwner,
    {
        FacadeDeploymentParts {
            calibration,
            envelope,
            db_service,
            filesystem_service,
            scheduler_service,
            supervisor_service,
            policy,
            adapter_provenance,
        }
    }
}

/// The Gate/deployment owner is independent from generated business artifacts.
/// There is no closure blanket implementation.
///
/// ```compile_fail
/// # use saddle::FacadeDeploymentOwner;
/// fn require_deployment<T: FacadeDeploymentOwner>(_: T) {}
/// require_deployment(|| ());
/// ```
pub trait FacadeDeploymentOwner: Send + 'static {
    type Calibration: saddle_admission::RuntimeCapacityCalibrationSourceLeaf;
    type Envelope: Into<saddle_runtime::resource_envelope::VerifiedResourceEnvelope>;
    type DbService: saddle_admission::VerifiedDbTerminationServiceProofOwner;
    type FilesystemService: Send + 'static;
    type SchedulerService: saddle_admission::VerifiedSchedulerTerminationServiceProofOwner;
    type SupervisorService: saddle_admission::VerifiedSupervisorTerminationServiceProofOwner;

    fn split(self, seal: FacadeDeploymentSeal) -> FacadeDeploymentOwnerParts<Self>;
}

/// Deployment-provided verifier for one opaque signed evidence bundle.
///
/// Implementations own their key distribution, trusted clock, nonce store and
/// actual service observations. The facade accepts only the typed owners and
/// binding identities returned through [`ApprovedExternalBundleSeal`]. There
/// is no default implementation, built-in production credential or unsigned
/// byte constructor.
pub trait ApprovedExternalBundle: Sized {
    type Reservation: ApprovedBundleNonceReservation;
    type Authority: ListenerAuthorityProvider;

    fn verify(
        self,
        seal: ApprovedExternalBundleSeal,
    ) -> Result<ApprovedExternalBundleVerification<Self::Reservation, Self::Authority>>;
}

/// Linear nonce reservation held until listener authority has also verified.
/// A deployment implementation must make `commit` atomically single-success
/// across processes/threads for its signed nonce.
pub trait ApprovedBundleNonceReservation: Send + 'static {
    fn commit(self) -> Result<()>;
}

/// Independent deployment-provided listener/startup authority verifier.
/// Verification permission from the evidence bundle is not listener authority.
pub trait ListenerAuthorityProvider: Sized {
    fn verify(self, seal: ListenerAuthoritySeal) -> Result<VerifiedListenerAuthority>;
}

/// One-shot facade seal passed only through [`ProductionLauncher::verify`].
/// Raw bundle bytes and manifest fields cannot be stored in this value.
#[doc(hidden)]
pub struct ApprovedExternalBundleSeal {
    _private: (),
}

/// One-shot facade seal for the independent listener authority provider.
pub struct ListenerAuthoritySeal {
    _private: (),
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct ExternalBundleBinding {
    bundle: [u8; 32],
    nonce: [u8; 32],
    binary: [u8; 32],
    deployment: [u8; 32],
    calibration: [u8; 32],
}

/// Verified seven-owner result. Fields are private and raw evidence is absent.
pub struct ApprovedExternalBundleVerification<R, L> {
    deployment: VerifiedLauncherDeployment,
    binding: ExternalBundleBinding,
    reservation: R,
    authority: L,
}

/// Recoverable failure of the final signed-filesystem/applicability pairing.
/// The aggregate remains whole; neither half is exposed independently.
pub struct ApprovedExternalBundleApplicabilityFailure {
    signed_filesystem: saddle_observability::file::VerifiedSignedProviderFilesystemBundle,
}

impl ApprovedExternalBundleApplicabilityFailure {
    /// Returns the still-whole aggregate for a corrected atomic retry.
    pub fn into_signed_filesystem(
        self,
    ) -> saddle_observability::file::VerifiedSignedProviderFilesystemBundle {
        self.signed_filesystem
    }
}

/// Verified independent listener authority. Fields are private and non-Clone.
pub struct VerifiedListenerAuthority {
    binding: ExternalBundleBinding,
}

struct VerifiedFacadeDeployment {
    calibration: saddle_runtime::capacity_leaf::VerifiedRuntimeBuildCalibrationOwner,
    envelope: saddle_runtime::resource_envelope::VerifiedResourceEnvelope,
    db_service: saddle_db::internal::DbReturnServiceProof,
    filesystem_service: saddle_observability::file::PairedSignedProviderFilesystemEvidence,
    scheduler_service: saddle_runtime::termination_service::VerifiedSchedulerDeploymentOwner,
    supervisor_service: saddle_runtime::termination_service::VerifiedSupervisorServiceOwner,
    policy: saddle_runtime::resource_envelope::FiniteStartupPolicy,
    adapter_provenance: [u8; 32],
}

enum VerifiedLauncherDeployment {
    Production(VerifiedFacadeDeployment),
    GoldenC8(VerifiedGoldenC8Compose),
}

#[allow(dead_code)]
struct VerifiedGoldenC8Compose {
    applicability: saddle_core::VerifiedGoldenC8ApplicabilityOwner,
    calibration: saddle_runtime::capacity_leaf::VerifiedRuntimeBuildCalibrationOwner,
    envelope: saddle_runtime::resource_envelope::VerifiedResourceEnvelope,
    db_service: saddle_db::internal::DbReturnServiceProof,
    filesystem_service: saddle_observability::file::WriterTerminationProof,
    scheduler_service: saddle_runtime::termination_service::VerifiedSchedulerDeploymentOwner,
    supervisor_service: saddle_runtime::termination_service::VerifiedSupervisorServiceOwner,
    policy: saddle_runtime::resource_envelope::FiniteStartupPolicy,
    adapter_provenance: [u8; 32],
}

#[derive(Clone, Copy)]
struct VerifiedDeploymentIdentitySet {
    runtime_common: [[u8; 32]; 3],
    calibration_common: [[u8; 32]; 3],
    runtime_generation: u64,
    calibration_generation: u64,
    expected_provenance: [[u8; 32]; 9],
    calibration_provenance: [[u8; 32]; 9],
    envelope_build: [u8; 32],
    envelope_gate: [u8; 32],
    envelope_supervisor: [u8; 32],
    environment: [u8; 32],
    db_build: [u8; 32],
    supervisor_build: [u8; 32],
    db_gate: [u8; 32],
    supervisor_gate: [u8; 32],
    supervisor_identity: [u8; 32],
    supervisor_environment: [u8; 32],
    required_identities: [[u8; 32]; 13],
}

fn validate_verified_deployment_identities(facts: VerifiedDeploymentIdentitySet) -> Result<()> {
    if facts.required_identities.contains(&[0; 32])
        || facts.runtime_common.contains(&[0; 32])
        || facts.runtime_common != facts.calibration_common
        || facts.runtime_generation == 0
        || facts.runtime_generation != facts.calibration_generation
        || facts.expected_provenance != facts.calibration_provenance
        || facts.envelope_build != facts.db_build
        || facts.envelope_build != facts.supervisor_build
        || facts.envelope_gate != facts.db_gate
        || facts.envelope_gate != facts.supervisor_gate
        || facts.envelope_supervisor != facts.supervisor_identity
        || facts.environment != facts.supervisor_environment
    {
        return Err(startup_error("saddle.approved_bundle_identity_mismatch"));
    }
    Ok(())
}

fn preflight_verified_deployment_without_filesystem(
    calibration: &saddle_runtime::capacity_leaf::VerifiedRuntimeBuildCalibrationOwner,
    envelope: &saddle_runtime::resource_envelope::VerifiedResourceEnvelope,
    db_service: &saddle_db::internal::DbReturnServiceProof,
    scheduler_service: saddle_runtime::termination_service::VerifiedSchedulerServiceOwner,
    supervisor_service: &saddle_runtime::termination_service::VerifiedSupervisorServiceOwner,
    adapter_provenance: [u8; 32],
    binding: ExternalBundleBinding,
) -> Result<saddle_runtime::termination_service::VerifiedSchedulerDeploymentOwner> {
    let calibration_common =
        saddle_admission::RuntimeCapacityCalibrationSourceLeaf::common_identities(calibration);
    let calibration_generation =
        saddle_admission::RuntimeCapacityCalibrationSourceLeaf::owner_generation(calibration);
    let calibration_provenance =
        saddle_admission::RuntimeCapacityCalibrationSourceLeaf::calibration_provenance(calibration);
    let envelope_build = saddle_admission::VerifiedStartupEnvelopeOwner::build_identity(envelope);
    let envelope_gate = saddle_admission::VerifiedStartupEnvelopeOwner::gate_identity(envelope);
    let envelope_resource =
        saddle_admission::VerifiedStartupEnvelopeOwner::resource_attestation(envelope);
    let envelope_supervisor =
        saddle_admission::VerifiedStartupEnvelopeOwner::supervisor_attestation(envelope);
    let environment = db_service.environment_identity();
    validate_verified_deployment_identities(VerifiedDeploymentIdentitySet {
        runtime_common: calibration_common,
        calibration_common,
        runtime_generation: calibration_generation,
        calibration_generation,
        expected_provenance: calibration_provenance,
        calibration_provenance,
        envelope_build,
        envelope_gate,
        envelope_supervisor,
        environment,
        db_build: db_service.build_identity(),
        supervisor_build: supervisor_service.build_identity(),
        db_gate: db_service.gate_identity(),
        supervisor_gate: supervisor_service.gate_identity(),
        supervisor_identity: supervisor_service.supervisor_identity(),
        supervisor_environment: supervisor_service.environment_identity(),
        required_identities: [
            adapter_provenance,
            binding.bundle,
            binding.nonce,
            binding.binary,
            binding.deployment,
            binding.calibration,
            envelope_build,
            envelope_gate,
            envelope_resource,
            envelope_supervisor,
            environment,
            db_service.service_attestation(),
            supervisor_service.service_attestation(),
        ],
    })?;
    saddle_runtime::termination_service::bind_scheduler_deployment_owner(
        scheduler_service,
        envelope,
    )
    .map_err(|_| startup_error("saddle.approved_bundle_identity_mismatch"))
}

impl ApprovedExternalBundleSeal {
    /// Golden/C8-only COMPOSE entry. The opaque applicability is bound to the
    /// frozen artifact pair, A/B, environment and single run. It cannot enter
    /// the production signed-filesystem path or authorize listener opening.
    #[doc(hidden)]
    #[allow(clippy::too_many_arguments)]
    pub fn bind_verified_for_golden_c8<R, L>(
        self,
        applicability: saddle_core::VerifiedGoldenC8ApplicabilityOwner,
        calibration: saddle_runtime::capacity_leaf::VerifiedRuntimeBuildCalibrationOwner,
        envelope: saddle_runtime::resource_envelope::VerifiedResourceEnvelope,
        db_service: saddle_db::internal::DbReturnServiceProof,
        filesystem_service: saddle_observability::file::WriterTerminationProof,
        scheduler_service: saddle_runtime::termination_service::VerifiedSchedulerServiceOwner,
        supervisor_service: saddle_runtime::termination_service::VerifiedSupervisorServiceOwner,
        policy: saddle_runtime::resource_envelope::FiniteStartupPolicy,
        adapter_provenance: [u8; 32],
        bundle_identity: [u8; 32],
        nonce_identity: [u8; 32],
        binary_identity: [u8; 32],
        deployment_identity: [u8; 32],
        calibration_identity: [u8; 32],
        reservation: R,
        authority: L,
    ) -> Result<ApprovedExternalBundleVerification<R, L>>
    where
        R: ApprovedBundleNonceReservation,
        L: ListenerAuthorityProvider,
    {
        let binding = ExternalBundleBinding {
            bundle: bundle_identity,
            nonce: nonce_identity,
            binary: binary_identity,
            deployment: deployment_identity,
            calibration: calibration_identity,
        };
        let scheduler_service = preflight_verified_deployment_without_filesystem(
            &calibration,
            &envelope,
            &db_service,
            scheduler_service,
            &supervisor_service,
            adapter_provenance,
            binding,
        )?;
        if filesystem_service.environment_identity()
            != saddle_db::internal::DbReturnServiceProof::environment_identity(&db_service)
            || filesystem_service.build_identity()
                != saddle_admission::VerifiedStartupEnvelopeOwner::build_identity(&envelope)
        {
            return Err(startup_error("saddle.golden_c8_applicability_foreign"));
        }
        Ok(ApprovedExternalBundleVerification {
            deployment: VerifiedLauncherDeployment::GoldenC8(VerifiedGoldenC8Compose {
                applicability,
                calibration,
                envelope,
                db_service,
                filesystem_service,
                scheduler_service,
                supervisor_service,
                policy,
                adapter_provenance,
            }),
            binding,
            reservation,
            authority,
        })
    }

    /// Production entry: the seven component owners cannot be bound unless
    /// core has already verified the signed deployment applicability bundle.
    #[allow(clippy::too_many_arguments)]
    pub fn bind_verified_with_applicability<R, L>(
        self,
        signed_filesystem: saddle_observability::file::VerifiedSignedProviderFilesystemBundle,
        application_identity: [u8; 32],
        source_identity: [u8; 32],
        artifact_identity: [u8; 32],
        routes_identity: [u8; 32],
        calibration: saddle_runtime::capacity_leaf::VerifiedRuntimeBuildCalibrationOwner,
        envelope: saddle_runtime::resource_envelope::VerifiedResourceEnvelope,
        db_service: saddle_db::internal::DbReturnServiceProof,
        scheduler_service: saddle_runtime::termination_service::VerifiedSchedulerServiceOwner,
        supervisor_service: saddle_runtime::termination_service::VerifiedSupervisorServiceOwner,
        policy: saddle_runtime::resource_envelope::FiniteStartupPolicy,
        adapter_provenance: [u8; 32],
        bundle_identity: [u8; 32],
        nonce_identity: [u8; 32],
        binary_identity: [u8; 32],
        deployment_identity: [u8; 32],
        calibration_identity: [u8; 32],
        reservation: R,
        authority: L,
    ) -> std::result::Result<
        ApprovedExternalBundleVerification<R, L>,
        ApprovedExternalBundleApplicabilityFailure,
    >
    where
        R: ApprovedBundleNonceReservation,
        L: ListenerAuthorityProvider,
    {
        let binding = ExternalBundleBinding {
            bundle: bundle_identity,
            nonce: nonce_identity,
            binary: binary_identity,
            deployment: deployment_identity,
            calibration: calibration_identity,
        };
        let scheduler_service = match preflight_verified_deployment_without_filesystem(
            &calibration,
            &envelope,
            &db_service,
            scheduler_service,
            &supervisor_service,
            adapter_provenance,
            binding,
        ) {
            Ok(owner) => owner,
            Err(_) => {
                return Err(ApprovedExternalBundleApplicabilityFailure { signed_filesystem });
            }
        };
        let filesystem_service = signed_filesystem
            .verify_deployment_pair(
                bundle_identity,
                adapter_provenance,
                application_identity,
                binary_identity,
                saddle_admission::VerifiedStartupEnvelopeOwner::build_identity(&envelope),
                source_identity,
                artifact_identity,
                routes_identity,
                deployment_identity,
                saddle_db::internal::DbReturnServiceProof::environment_identity(&db_service),
                nonce_identity,
                calibration_identity,
            )
            .map_err(
                |signed_filesystem| ApprovedExternalBundleApplicabilityFailure {
                    signed_filesystem,
                },
            )?;
        Ok(self
            .bind_verified_inner(
                calibration,
                envelope,
                db_service,
                filesystem_service,
                scheduler_service,
                supervisor_service,
                policy,
                adapter_provenance,
                bundle_identity,
                nonce_identity,
                binary_identity,
                deployment_identity,
                calibration_identity,
                reservation,
                authority,
            )
            .unwrap_or_else(|_| {
                unreachable!("signed filesystem issuer and component preflight are closed")
            }))
    }

    /// Packages the exact typed outputs of the component verifiers. The Gate
    /// adapter must finish signature, B0 and attestation checks before calling
    /// this method; no raw bytes or numeric manifest fields cross this call.
    #[allow(clippy::too_many_arguments)]
    #[cfg(test)]
    pub fn bind_verified<R, L>(
        self,
        calibration: saddle_runtime::capacity_leaf::VerifiedRuntimeBuildCalibrationOwner,
        envelope: saddle_runtime::resource_envelope::VerifiedResourceEnvelope,
        db_service: saddle_db::internal::DbReturnServiceProof,
        filesystem_service: saddle_observability::file::PairedSignedProviderFilesystemEvidence,
        scheduler_service: saddle_runtime::termination_service::VerifiedSchedulerServiceOwner,
        supervisor_service: saddle_runtime::termination_service::VerifiedSupervisorServiceOwner,
        policy: saddle_runtime::resource_envelope::FiniteStartupPolicy,
        adapter_provenance: [u8; 32],
        bundle_identity: [u8; 32],
        nonce_identity: [u8; 32],
        binary_identity: [u8; 32],
        deployment_identity: [u8; 32],
        calibration_identity: [u8; 32],
        reservation: R,
        authority: L,
    ) -> Result<ApprovedExternalBundleVerification<R, L>>
    where
        R: ApprovedBundleNonceReservation,
        L: ListenerAuthorityProvider,
    {
        let scheduler_service =
            saddle_runtime::termination_service::bind_scheduler_deployment_owner(
                scheduler_service,
                &envelope,
            )
            .map_err(|_| startup_error("saddle.approved_bundle_identity_mismatch"))?;
        self.bind_verified_inner(
            calibration,
            envelope,
            db_service,
            filesystem_service,
            scheduler_service,
            supervisor_service,
            policy,
            adapter_provenance,
            bundle_identity,
            nonce_identity,
            binary_identity,
            deployment_identity,
            calibration_identity,
            reservation,
            authority,
        )
    }

    #[allow(clippy::too_many_arguments)]
    fn bind_verified_inner<R, L>(
        self,
        calibration: saddle_runtime::capacity_leaf::VerifiedRuntimeBuildCalibrationOwner,
        envelope: saddle_runtime::resource_envelope::VerifiedResourceEnvelope,
        db_service: saddle_db::internal::DbReturnServiceProof,
        filesystem_service: saddle_observability::file::PairedSignedProviderFilesystemEvidence,
        scheduler_service: saddle_runtime::termination_service::VerifiedSchedulerDeploymentOwner,
        supervisor_service: saddle_runtime::termination_service::VerifiedSupervisorServiceOwner,
        policy: saddle_runtime::resource_envelope::FiniteStartupPolicy,
        adapter_provenance: [u8; 32],
        bundle_identity: [u8; 32],
        nonce_identity: [u8; 32],
        binary_identity: [u8; 32],
        deployment_identity: [u8; 32],
        calibration_identity: [u8; 32],
        reservation: R,
        authority: L,
    ) -> Result<ApprovedExternalBundleVerification<R, L>>
    where
        R: ApprovedBundleNonceReservation,
        L: ListenerAuthorityProvider,
    {
        let calibration_common =
            saddle_admission::RuntimeCapacityCalibrationSourceLeaf::common_identities(&calibration);
        let calibration_generation =
            saddle_admission::RuntimeCapacityCalibrationSourceLeaf::owner_generation(&calibration);
        let envelope_build =
            saddle_admission::VerifiedStartupEnvelopeOwner::build_identity(&envelope);
        let envelope_gate =
            saddle_admission::VerifiedStartupEnvelopeOwner::gate_identity(&envelope);
        let envelope_resource =
            saddle_admission::VerifiedStartupEnvelopeOwner::resource_attestation(&envelope);
        let envelope_supervisor =
            saddle_admission::VerifiedStartupEnvelopeOwner::supervisor_attestation(&envelope);
        let environment = db_service.environment_identity();
        let binding = ExternalBundleBinding {
            bundle: bundle_identity,
            nonce: nonce_identity,
            binary: binary_identity,
            deployment: deployment_identity,
            calibration: calibration_identity,
        };
        let required_identities = [
            adapter_provenance,
            binding.bundle,
            binding.nonce,
            binding.binary,
            binding.deployment,
            binding.calibration,
            envelope_build,
            envelope_gate,
            envelope_resource,
            envelope_supervisor,
            environment,
            db_service.service_attestation(),
            supervisor_service.service_attestation(),
        ];
        validate_verified_deployment_identities(VerifiedDeploymentIdentitySet {
            runtime_common: calibration_common,
            calibration_common,
            runtime_generation: calibration_generation,
            calibration_generation,
            expected_provenance:
                saddle_admission::RuntimeCapacityCalibrationSourceLeaf::calibration_provenance(
                    &calibration,
                ),
            calibration_provenance:
                saddle_admission::RuntimeCapacityCalibrationSourceLeaf::calibration_provenance(
                    &calibration,
                ),
            envelope_build,
            envelope_gate,
            envelope_supervisor,
            environment,
            db_build: db_service.build_identity(),
            supervisor_build: supervisor_service.build_identity(),
            db_gate: db_service.gate_identity(),
            supervisor_gate: supervisor_service.gate_identity(),
            supervisor_identity: supervisor_service.supervisor_identity(),
            supervisor_environment: supervisor_service.environment_identity(),
            required_identities,
        })?;
        if !filesystem_service.matches_deployment(environment, envelope_build) {
            return Err(startup_error("saddle.approved_bundle_identity_mismatch"));
        }
        Ok(ApprovedExternalBundleVerification {
            deployment: VerifiedLauncherDeployment::Production(VerifiedFacadeDeployment {
                calibration,
                envelope,
                db_service,
                filesystem_service,
                scheduler_service,
                supervisor_service,
                policy,
                adapter_provenance,
            }),
            binding,
            reservation,
            authority,
        })
    }
}

impl ListenerAuthoritySeal {
    /// Binds an independently verified listener authority to the exact signed
    /// bundle/nonce/executable/deployment/calibration tuple.
    pub fn bind_verified(
        self,
        bundle_identity: [u8; 32],
        nonce_identity: [u8; 32],
        binary_identity: [u8; 32],
        deployment_identity: [u8; 32],
        calibration_identity: [u8; 32],
    ) -> Result<VerifiedListenerAuthority> {
        let binding = ExternalBundleBinding {
            bundle: bundle_identity,
            nonce: nonce_identity,
            binary: binary_identity,
            deployment: deployment_identity,
            calibration: calibration_identity,
        };
        if [
            binding.bundle,
            binding.nonce,
            binding.binary,
            binding.deployment,
            binding.calibration,
        ]
        .contains(&[0; 32])
        {
            return Err(startup_error("saddle.listener_authority_missing_identity"));
        }
        Ok(VerifiedListenerAuthority { binding })
    }
}

impl FacadeDeploymentOwner for VerifiedFacadeDeployment {
    type Calibration = saddle_runtime::capacity_leaf::VerifiedRuntimeBuildCalibrationOwner;
    type Envelope = saddle_runtime::resource_envelope::VerifiedResourceEnvelope;
    type DbService = saddle_db::internal::DbReturnServiceProof;
    type FilesystemService = saddle_observability::file::PairedSignedProviderFilesystemEvidence;
    type SchedulerService = saddle_runtime::termination_service::VerifiedSchedulerDeploymentOwner;
    type SupervisorService = saddle_runtime::termination_service::VerifiedSupervisorServiceOwner;

    fn split(self, seal: FacadeDeploymentSeal) -> FacadeDeploymentOwnerParts<Self> {
        seal.bind(
            self.calibration,
            self.envelope,
            self.db_service,
            self.filesystem_service,
            self.scheduler_service,
            self.supervisor_service,
            self.policy,
            self.adapter_provenance,
        )
    }
}

/// Fixed 0.2 process facts. Capacity/profile values remain private candidate
/// inputs until the final Gate freeze.
pub struct SaddleConfig {
    application: saddle_core::ApplicationId,
    database: Option<DatabaseConfig>,
    listen: SocketAddr,
    _observability: ObserverConfig,
}

impl SaddleConfig {
    pub fn new(
        application: impl Into<saddle_core::ApplicationId>,
        database_url: impl Into<String>,
        listen: SocketAddr,
    ) -> Self {
        Self {
            application: application.into(),
            database: Some(DatabaseConfig::new(database_url)),
            listen,
            _observability: ObserverConfig::default(),
        }
    }

    /// Creates a deployment configuration whose verified startup plan must
    /// derive a zero DB profile. No pool is constructed for this form.
    pub fn without_database(
        application: impl Into<saddle_core::ApplicationId>,
        listen: SocketAddr,
    ) -> Self {
        Self {
            application: application.into(),
            database: None,
            listen,
            _observability: ObserverConfig::default(),
        }
    }
}

/// Fixed process entry for one Skill-generated 0.2 application.
struct Saddle;

/// Opaque, non-Clone permission to start the sole production assembly.
///
/// ```compile_fail
/// # use saddle::ProductionLauncher;
/// fn replay(launcher: &ProductionLauncher) {
///     let _copy = (*launcher).clone();
/// }
/// ```
///
/// ```compile_fail
/// # use saddle::ProductionLauncher;
/// let _forged = ProductionLauncher { deployment: () };
/// ```
///
/// The public launch shape has no deployment third parameter:
///
/// ```no_run
/// # use saddle::{ApprovedExternalBundle, GeneratedApplicationOwner, ProductionLauncher,
/// #     Result, SaddleConfig};
/// fn launch<B, O>(bundle: B, config: SaddleConfig, generated: O) -> Result<()>
/// where
///     B: ApprovedExternalBundle,
///     O: GeneratedApplicationOwner,
/// {
///     ProductionLauncher::verify(bundle)?.run(config, generated)
/// }
/// ```
pub struct ProductionLauncher {
    _private: (),
}

/// Opaque verified bundle whose listener authority and nonce remain pending
/// until generated layout preflight has paired with calibration.
#[doc(hidden)]
pub struct VerifiedProductionLauncher<R, L> {
    deployment: VerifiedLauncherDeployment,
    binding: ExternalBundleBinding,
    reservation: R,
    authority: L,
}

impl ProductionLauncher {
    /// Consumes the sole approved external bundle. Failure occurs before the
    /// deployment owner reaches Observer, DB, Tokio or listener construction.
    pub fn verify<B>(bundle: B) -> Result<VerifiedProductionLauncher<B::Reservation, B::Authority>>
    where
        B: ApprovedExternalBundle,
    {
        let verified = bundle.verify(ApprovedExternalBundleSeal { _private: () })?;
        Ok(VerifiedProductionLauncher {
            deployment: verified.deployment,
            binding: verified.binding,
            reservation: verified.reservation,
            authority: verified.authority,
        })
    }
}

impl<R, L> VerifiedProductionLauncher<R, L>
where
    R: ApprovedBundleNonceReservation,
    L: ListenerAuthorityProvider,
{
    /// The only public production launch path. Deployment owners remain
    /// private and are consumed exactly once with this launcher.
    pub fn run<O>(self, config: SaddleConfig, generated: O) -> Result<()>
    where
        O: GeneratedApplicationOwner,
    {
        match self.deployment {
            VerifiedLauncherDeployment::Production(deployment) => Saddle::run_with_deployment(
                config,
                generated,
                deployment,
                PendingListenerAuthorization {
                    binding: self.binding,
                    reservation: self.reservation,
                    authority: self.authority,
                },
            ),
            VerifiedLauncherDeployment::GoldenC8(compose) => Saddle::run_with_golden_c8(
                config,
                generated,
                compose,
                PendingListenerAuthorization {
                    binding: self.binding,
                    reservation: self.reservation,
                    authority: self.authority,
                },
            ),
        }
    }
}

struct GoldenC8GeneratedConsumer<R, L> {
    config: SaddleConfig,
    compose: VerifiedGoldenC8Compose,
    authorization: PendingListenerAuthorization<R, L>,
}

impl<R, L> generated_consumer_sealed::Sealed for GoldenC8GeneratedConsumer<R, L>
where
    R: ApprovedBundleNonceReservation,
    L: ListenerAuthorityProvider,
{
}

impl<R, L> GeneratedApplicationConsumer for GoldenC8GeneratedConsumer<R, L>
where
    R: ApprovedBundleNonceReservation,
    L: ListenerAuthorityProvider,
{
    fn consume<E, BC, F, C, const BODY: usize, const OUTPUT: usize, const ROUTES: usize>(
        self,
        parts: GeneratedBoundApplicationParts<
            saddle_service::internal::CompiledExecutionWithCapacityLeaf<E, BC, F, BODY, OUTPUT>,
            C,
            ROUTES,
        >,
    ) -> Result<()>
    where
        E: Fn(
                usize,
                BC,
                saddle_admission::ManagedBytes,
                saddle_service::internal::CompiledDbPermit,
                saddle_admission::ManagedResponseBuilder,
            ) -> F
            + Send
            + Sync
            + 'static,
        BC: Send + Unpin + 'static,
        F: std::future::Future<
                Output = std::result::Result<
                    saddle_admission::ManagedResponse,
                    saddle_service::internal::ExecutionError,
                >,
            > + Send
            + 'static,
        C: GeneratedContextFactory<BC>,
    {
        let approved = saddle_core::verify_golden_c8_startup_assembly_input(
            saddle_core::approved_golden_c8_startup_assembly_input(),
        )
        .map_err(|_| startup_error("saddle.golden_c8_startup_assembly_rejected"))?;
        let GeneratedApplicationParts {
            termination,
            continuation,
            bootstrap,
            runtime_preflight,
        } = parts;
        // Gate already binds this exact generated whole, deployment compose
        // and current startup transaction. GeneratedApplicationSeal issued
        // this private continuation and staged HTTP graph from the same
        // checked common identities and generation. Golden/C8 consumes that
        // already-preflighted HTTP owner only after listener preclosure; it
        // neither binds a socket nor invokes production composition.
        let continuation = (
            approved,
            continuation,
            bootstrap,
            runtime_preflight,
            termination,
            self.compose,
            self.config,
        );
        let PendingListenerAuthorization {
            binding,
            reservation,
            authority,
        } = self.authorization;
        let listener_preclosure = authorize_listener(continuation, binding, reservation, authority)
            .map_err(|failure| failure.error)?;
        let (
            approved,
            continuation,
            GeneratedProductionBootstrap {
                generation,
                type_identity,
                staged,
            },
            runtime_preflight,
            termination,
            compose,
            config,
        ) = listener_preclosure;
        let http_preclosure = (
            approved,
            continuation,
            generation,
            type_identity,
            staged,
            runtime_preflight,
            termination,
            compose,
            config,
        );
        // The Golden/C8 transaction already carries the verified DB service
        // and database configuration inside the private compose/config pair.
        // Consuming the HTTP preclosure here preserves that whole linear
        // state for the commit check without connecting to DB or creating a
        // second assembly path.
        let db_commit_preclosure = http_preclosure;
        // Golden/C8 rollback validation consumes the same private database
        // preclosure as commit validation.  This is still a pure ownership
        // transition: no connection is opened and no database callback is
        // invoked before the later system-acceptance stages authorize it.
        let db_rollback_preclosure = db_commit_preclosure;
        let shutdown_preclosure = db_rollback_preclosure;
        // Shutdown preclosure consumes the same private Golden/C8 transaction
        // after both database paths have been checked.  It remains a pure
        // ownership transition: no supervisor callback, listener, or runtime
        // shutdown is invoked before system acceptance authorizes it.
        let resource_zero_preclosure = shutdown_preclosure;
        // RESOURCE_ZERO preclosure consumes the same private transaction
        // after shutdown has been checked. This remains a pure ownership
        // transition: it does not inspect resources, run a destructor, open
        // a listener, or grant C8/production/release/publish authority.
        let c8_acceptance_preclosure = resource_zero_preclosure;
        // C8 acceptance consumes the same private, fully preclosed Golden
        // transaction.  Acceptance is only a final ownership transition: it
        // opens no listener, executes no application work, and grants no
        // production, release, or publish authority.
        let _accepted = c8_acceptance_preclosure;
        Ok(())
    }
}

#[doc(hidden)]
pub struct PendingListenerAuthorization<R, L> {
    binding: ExternalBundleBinding,
    reservation: R,
    authority: L,
}

fn authorize_listener<T, R, L>(
    value: T,
    binding: ExternalBundleBinding,
    reservation: R,
    authority: L,
) -> std::result::Result<T, ListenerAuthorizationFailure<T>>
where
    R: ApprovedBundleNonceReservation,
    L: ListenerAuthorityProvider,
{
    let mut value = Some(value);
    let listener = authority
        .verify(ListenerAuthoritySeal { _private: () })
        .map_err(|error| ListenerAuthorizationFailure {
            value: value.take().expect("authorization value is present"),
            error,
        })?;
    if binding != listener.binding {
        return Err(ListenerAuthorizationFailure {
            value: value.take().expect("authorization value is present"),
            error: startup_error("saddle.listener_authority_binding_mismatch"),
        });
    }
    reservation
        .commit()
        .map_err(|error| ListenerAuthorizationFailure {
            value: value.take().expect("authorization value is present"),
            error,
        })?;
    Ok(value.take().expect("authorization value is present"))
}

struct ListenerAuthorizationFailure<T> {
    value: T,
    error: SaddleError,
}

struct FacadeGeneratedConsumer<D, R, L>
where
    D: FacadeDeploymentOwner<
            Calibration = saddle_runtime::capacity_leaf::VerifiedRuntimeBuildCalibrationOwner,
            FilesystemService = saddle_observability::file::PairedSignedProviderFilesystemEvidence,
        >,
{
    config: SaddleConfig,
    deployment: FacadeDeploymentOwnerParts<D>,
    authorization: PendingListenerAuthorization<R, L>,
}

impl<D, R, L> generated_consumer_sealed::Sealed for FacadeGeneratedConsumer<D, R, L>
where
    D: FacadeDeploymentOwner<
            Calibration = saddle_runtime::capacity_leaf::VerifiedRuntimeBuildCalibrationOwner,
            FilesystemService = saddle_observability::file::PairedSignedProviderFilesystemEvidence,
        >,
    R: ApprovedBundleNonceReservation,
    L: ListenerAuthorityProvider,
{
}

impl<D, R, L> GeneratedApplicationConsumer for FacadeGeneratedConsumer<D, R, L>
where
    D: FacadeDeploymentOwner<
            Calibration = saddle_runtime::capacity_leaf::VerifiedRuntimeBuildCalibrationOwner,
            FilesystemService = saddle_observability::file::PairedSignedProviderFilesystemEvidence,
        >,
    R: ApprovedBundleNonceReservation,
    L: ListenerAuthorityProvider,
{
    fn consume<E, BC, F, C, const BODY: usize, const OUTPUT: usize, const ROUTES: usize>(
        self,
        parts: GeneratedBoundApplicationParts<
            saddle_service::internal::CompiledExecutionWithCapacityLeaf<E, BC, F, BODY, OUTPUT>,
            C,
            ROUTES,
        >,
    ) -> Result<()>
    where
        E: Fn(
                usize,
                BC,
                saddle_admission::ManagedBytes,
                saddle_service::internal::CompiledDbPermit,
                saddle_admission::ManagedResponseBuilder,
            ) -> F
            + Send
            + Sync
            + 'static,
        BC: Send + Unpin + 'static,
        F: std::future::Future<
                Output = std::result::Result<
                    saddle_admission::ManagedResponse,
                    saddle_service::internal::ExecutionError,
                >,
            > + Send
            + 'static,
        C: GeneratedContextFactory<BC>,
    {
        let GeneratedApplicationParts {
            termination,
            continuation,
            bootstrap,
            runtime_preflight,
        } = parts;
        let candidate = crate::startup_candidate::prepare_production_candidate_transaction(
            bootstrap
                .staged
                .service_production_fact_input()
                .map_err(|_| startup_error("saddle.service_production_fact_failed"))?,
        )?;
        let FacadeDeploymentParts {
            calibration,
            envelope,
            db_service,
            filesystem_service,
            scheduler_service,
            supervisor_service,
            policy,
            adapter_provenance,
        } = self.deployment;
        let (runtime, _runtime_termination, calibration) =
            saddle_runtime::capacity_leaf::verify_runtime_capacity_leaf_from_preflight(
                runtime_preflight,
                calibration,
            )
            .map_err(|_| startup_error("saddle.runtime_layout_calibration_pair_failed"))?;
        let (staged, generated) = bootstrap
            .staged
            .compose_generated(runtime, calibration, termination)
            .map_err(|_| startup_error("saddle.generated_facts_composition_failed"))?;
        let bootstrap = GeneratedProductionBootstrap {
            staged,
            ..bootstrap
        };
        let (generated, continuation) =
            saddle_admission::seal_generated_startup_continuation(continuation, generated)
                .map_err(|_| startup_error("saddle.generated_continuation_foreign"))?
                .into_parts();
        let (filesystem_service, generated) = filesystem_service
            .pair_component_generation(generated)
            .map_err(|_| startup_error("saddle.component_generation_foreign"))?;
        let (pending, physical) =
            saddle_runtime::resource_envelope::create_runtime_startup_plan_with_filesystem_bundle(
                policy,
                adapter_provenance,
                envelope.into(),
                generated,
                db_service,
                filesystem_service,
                scheduler_service,
                supervisor_service,
            )
            .map_err(|failure| {
                let _ = failure.consume_for_application_error();
                startup_error("saddle.startup_plan_creation_failed")
            })?;
        let factory = StartupManagedDatabaseFactory::awaiting_started_observability(
            self.config.database.clone(),
        );
        let owners = saddle_runtime::startup_assembly::assemble_actual_startup_owners(
            pending,
            factory,
            saddle_runtime::startup_assembly::StartupObservabilityInput::Production(physical),
        )
        .map_err(|_| startup_error("saddle.startup_actual_owner_assembly_failed"))?;
        bootstrap.run(
            FacadeBootstrapCoordinator {
                config: self.config,
                adapter_provenance,
            },
            owners,
            continuation,
            self.authorization,
            candidate,
        )
    }
}

impl Saddle {
    fn run_with_golden_c8<O, R, L>(
        config: SaddleConfig,
        generated: O,
        compose: VerifiedGoldenC8Compose,
        authorization: PendingListenerAuthorization<R, L>,
    ) -> Result<()>
    where
        O: GeneratedApplicationOwner,
        R: ApprovedBundleNonceReservation,
        L: ListenerAuthorityProvider,
    {
        generated.consume(
            GeneratedApplicationSeal::issue(),
            GoldenC8GeneratedConsumer {
                config,
                compose,
                authorization,
            },
        )
    }

    /// Private continuation after the launcher consumed verified deployment.
    fn run_with_deployment<O, D, R, L>(
        config: SaddleConfig,
        generated: O,
        deployment: D,
        authorization: PendingListenerAuthorization<R, L>,
    ) -> Result<()>
    where
        O: GeneratedApplicationOwner,
        D: FacadeDeploymentOwner<
            Calibration = saddle_runtime::capacity_leaf::VerifiedRuntimeBuildCalibrationOwner,
            FilesystemService = saddle_observability::file::PairedSignedProviderFilesystemEvidence,
        >,
        R: ApprovedBundleNonceReservation,
        L: ListenerAuthorityProvider,
    {
        let deployment = deployment.split(FacadeDeploymentSeal::issue());
        generated.consume(
            GeneratedApplicationSeal::issue(),
            FacadeGeneratedConsumer::<D, R, L> {
                config,
                deployment,
                authorization,
            },
        )
    }
}

pub(crate) fn startup_error(code: &'static str) -> SaddleError {
    SaddleError::new(
        ErrorKind::Infrastructure,
        code,
        "Saddle application initialization failed",
    )
}

#[cfg(test)]
mod generated_bootstrap_contract_tests {
    use super::*;
    use std::sync::{
        Arc,
        atomic::{AtomicBool, AtomicUsize, Ordering},
    };

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

    fn binding(seed: u8) -> ExternalBundleBinding {
        ExternalBundleBinding {
            bundle: [seed; 32],
            nonce: [seed.wrapping_add(1); 32],
            binary: [seed.wrapping_add(2); 32],
            deployment: [seed.wrapping_add(3); 32],
            calibration: [seed.wrapping_add(4); 32],
        }
    }

    struct ReferenceAuthority(ExternalBundleBinding);

    impl ListenerAuthorityProvider for ReferenceAuthority {
        fn verify(self, seal: ListenerAuthoritySeal) -> Result<VerifiedListenerAuthority> {
            seal.bind_verified(
                self.0.bundle,
                self.0.nonce,
                self.0.binary,
                self.0.deployment,
                self.0.calibration,
            )
        }
    }

    struct CountingAuthority {
        binding: ExternalBundleBinding,
        calls: Arc<AtomicUsize>,
    }

    impl ListenerAuthorityProvider for CountingAuthority {
        fn verify(self, seal: ListenerAuthoritySeal) -> Result<VerifiedListenerAuthority> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            seal.bind_verified(
                self.binding.bundle,
                self.binding.nonce,
                self.binding.binary,
                self.binding.deployment,
                self.binding.calibration,
            )
        }
    }

    struct CountingReservation {
        calls: Arc<AtomicUsize>,
    }

    impl ApprovedBundleNonceReservation for CountingReservation {
        fn commit(self) -> Result<()> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }
    }

    struct RejectingAuthority {
        calls: Arc<AtomicUsize>,
    }

    impl ListenerAuthorityProvider for RejectingAuthority {
        fn verify(self, _seal: ListenerAuthoritySeal) -> Result<VerifiedListenerAuthority> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            Err(startup_error("test.listener_authority_rejected"))
        }
    }

    struct RejectingReservation {
        calls: Arc<AtomicUsize>,
    }

    impl ApprovedBundleNonceReservation for RejectingReservation {
        fn commit(self) -> Result<()> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            Err(startup_error("test.bundle_nonce_commit_rejected"))
        }
    }

    fn pending_counting_authorization(
        authority_calls: &Arc<AtomicUsize>,
        commit_calls: &Arc<AtomicUsize>,
    ) -> PendingListenerAuthorization<CountingReservation, CountingAuthority> {
        let binding = binding(40);
        PendingListenerAuthorization {
            binding,
            reservation: CountingReservation {
                calls: Arc::clone(commit_calls),
            },
            authority: CountingAuthority {
                binding,
                calls: Arc::clone(authority_calls),
            },
        }
    }

    struct ReferenceReservation(Arc<AtomicBool>);

    impl ApprovedBundleNonceReservation for ReferenceReservation {
        fn commit(self) -> Result<()> {
            self.0
                .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
                .map(|_| ())
                .map_err(|_| startup_error("test.bundle_nonce_replay"))
        }
    }

    struct OwnerAggregate(Arc<AtomicUsize>);

    impl Drop for OwnerAggregate {
        fn drop(&mut self) {
            self.0.fetch_add(1, Ordering::SeqCst);
        }
    }

    struct RejectedBundle;
    struct RejectedReservation;
    struct RejectedAuthority;

    impl ApprovedBundleNonceReservation for RejectedReservation {
        fn commit(self) -> Result<()> {
            panic!("rejected bundle cannot reserve a nonce")
        }
    }

    impl ApprovedExternalBundle for RejectedBundle {
        type Reservation = RejectedReservation;
        type Authority = RejectedAuthority;

        fn verify(
            self,
            _seal: ApprovedExternalBundleSeal,
        ) -> Result<ApprovedExternalBundleVerification<Self::Reservation, Self::Authority>>
        {
            REJECTED_BUNDLES.fetch_add(1, Ordering::SeqCst);
            Err(startup_error("test.bundle_rejected"))
        }
    }

    impl ListenerAuthorityProvider for RejectedAuthority {
        fn verify(self, _seal: ListenerAuthoritySeal) -> Result<VerifiedListenerAuthority> {
            panic!("listener authority must not run after bundle rejection")
        }
    }

    #[test]
    fn rejected_bundle_returns_before_any_launcher_exists() {
        let before = REJECTED_BUNDLES.load(Ordering::SeqCst);
        assert!(ProductionLauncher::verify(RejectedBundle).is_err());
        assert_eq!(REJECTED_BUNDLES.load(Ordering::SeqCst), before + 1);
    }

    fn matching_deployment_identities() -> VerifiedDeploymentIdentitySet {
        VerifiedDeploymentIdentitySet {
            runtime_common: [[1; 32], [2; 32], [3; 32]],
            calibration_common: [[1; 32], [2; 32], [3; 32]],
            runtime_generation: 7,
            calibration_generation: 7,
            expected_provenance: [[4; 32]; 9],
            calibration_provenance: [[4; 32]; 9],
            envelope_build: [5; 32],
            envelope_gate: [6; 32],
            envelope_supervisor: [8; 32],
            environment: [9; 32],
            db_build: [5; 32],
            supervisor_build: [5; 32],
            db_gate: [6; 32],
            supervisor_gate: [6; 32],
            supervisor_identity: [8; 32],
            supervisor_environment: [9; 32],
            required_identities: [[10; 32]; 13],
        }
    }

    #[test]
    fn matching_seven_owner_identities_reach_the_launcher_boundary() {
        assert!(validate_verified_deployment_identities(matching_deployment_identities()).is_ok());
    }

    #[test]
    fn missing_foreign_drift_and_replay_generation_fail_before_launcher() {
        let mut missing = matching_deployment_identities();
        missing.required_identities[4] = [0; 32];
        assert!(validate_verified_deployment_identities(missing).is_err());

        let mut foreign = matching_deployment_identities();
        foreign.supervisor_environment = [11; 32];
        assert!(validate_verified_deployment_identities(foreign).is_err());

        let mut drift = matching_deployment_identities();
        drift.supervisor_gate = [12; 32];
        assert!(validate_verified_deployment_identities(drift).is_err());

        let mut replay = matching_deployment_identities();
        replay.calibration_generation += 1;
        assert!(validate_verified_deployment_identities(replay).is_err());
    }

    #[test]
    fn reference_authority_commits_matching_nonce_once() {
        let committed = Arc::new(AtomicBool::new(false));
        let expected = binding(10);
        assert!(
            authorize_listener(
                (),
                expected,
                ReferenceReservation(Arc::clone(&committed)),
                ReferenceAuthority(expected),
            )
            .is_ok()
        );
        assert!(committed.load(Ordering::SeqCst));
        assert!(
            authorize_listener(
                (),
                expected,
                ReferenceReservation(Arc::clone(&committed)),
                ReferenceAuthority(expected),
            )
            .is_err()
        );
    }

    #[test]
    fn concurrent_nonce_commit_has_exactly_one_success() {
        let committed = Arc::new(AtomicBool::new(false));
        let expected = binding(20);
        let threads: Vec<_> = (0..8)
            .map(|_| {
                let committed = Arc::clone(&committed);
                std::thread::spawn(move || {
                    authorize_listener(
                        (),
                        expected,
                        ReferenceReservation(committed),
                        ReferenceAuthority(expected),
                    )
                    .is_ok()
                })
            })
            .collect();
        assert_eq!(
            threads
                .into_iter()
                .map(|thread| thread.join().unwrap())
                .filter(|success| *success)
                .count(),
            1
        );
    }

    #[test]
    fn authority_drift_drops_owner_aggregate_without_committing_nonce() {
        let committed = Arc::new(AtomicBool::new(false));
        let drops = Arc::new(AtomicUsize::new(0));
        let expected = binding(30);
        assert!(
            authorize_listener(
                OwnerAggregate(Arc::clone(&drops)),
                expected,
                ReferenceReservation(Arc::clone(&committed)),
                ReferenceAuthority(binding(31)),
            )
            .is_err()
        );
        assert_eq!(drops.load(Ordering::SeqCst), 1);
        assert!(!committed.load(Ordering::SeqCst));
    }

    #[test]
    fn authority_failure_returns_prepared_owner_without_nonce_commit() {
        let authority_calls = Arc::new(AtomicUsize::new(0));
        let commit_calls = Arc::new(AtomicUsize::new(0));
        let drops = Arc::new(AtomicUsize::new(0));
        let result = authorize_listener(
            OwnerAggregate(Arc::clone(&drops)),
            binding(35),
            CountingReservation {
                calls: Arc::clone(&commit_calls),
            },
            RejectingAuthority {
                calls: Arc::clone(&authority_calls),
            },
        );
        assert!(result.is_err());
        drop(result);
        assert_eq!(authority_calls.load(Ordering::SeqCst), 1);
        assert_eq!(commit_calls.load(Ordering::SeqCst), 0);
        assert_eq!(drops.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn nonce_commit_failure_returns_prepared_owner_before_listener_install() {
        let authority_calls = Arc::new(AtomicUsize::new(0));
        let commit_calls = Arc::new(AtomicUsize::new(0));
        let drops = Arc::new(AtomicUsize::new(0));
        let expected = binding(36);
        let result = authorize_listener(
            OwnerAggregate(Arc::clone(&drops)),
            expected,
            RejectingReservation {
                calls: Arc::clone(&commit_calls),
            },
            CountingAuthority {
                binding: expected,
                calls: Arc::clone(&authority_calls),
            },
        );
        assert!(result.is_err());
        drop(result);
        assert_eq!(authority_calls.load(Ordering::SeqCst), 1);
        assert_eq!(commit_calls.load(Ordering::SeqCst), 1);
        assert_eq!(drops.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn composition_failure_discards_pending_authorization_without_side_effects() {
        let authority_calls = Arc::new(AtomicUsize::new(0));
        let commit_calls = Arc::new(AtomicUsize::new(0));
        let pending = pending_counting_authorization(&authority_calls, &commit_calls);

        let composition: Result<()> = Err(startup_error("test.composition_failed"));
        assert!(composition.is_err());
        drop(pending);

        assert_eq!(authority_calls.load(Ordering::SeqCst), 0);
        assert_eq!(commit_calls.load(Ordering::SeqCst), 0);
    }

    #[test]
    fn late_prepare_failure_discards_pending_authorization_without_side_effects() {
        let authority_calls = Arc::new(AtomicUsize::new(0));
        let commit_calls = Arc::new(AtomicUsize::new(0));
        let pending = pending_counting_authorization(&authority_calls, &commit_calls);

        let late_prepare: Result<()> = Err(startup_error("test.http1_prepare_failed"));
        assert!(late_prepare.is_err());
        drop(pending);

        assert_eq!(authority_calls.load(Ordering::SeqCst), 0);
        assert_eq!(commit_calls.load(Ordering::SeqCst), 0);
    }

    #[test]
    fn facade_issues_distinct_termination_and_continuation_proofs() {
        use saddle_admission::{
            GeneratedTerminationTopologyWorkOwner, PreflightGeneratedStaticContinuationOwner,
        };

        let common = [[0xb2; 32], [0xa2; 32], [0xd2; 32]];
        let termination = FacadeGeneratedTermination {
            common,
            generation: 9,
            topology: [5, 1, 1, 3],
        };
        let continuation = FacadeGeneratedContinuation {
            common,
            generation: 9,
        };
        let identities = [
            termination.leaf_identity(),
            termination.termination_topology_identity(),
            termination.db_return_work_identity(),
            termination.writer_work_identity(),
            termination.runtime_work_identity(),
        ];
        assert!(!identities.contains(&[0; 32]));
        for (index, identity) in identities.iter().enumerate() {
            assert!(!identities[index + 1..].contains(identity));
            assert!(!common.contains(identity));
        }
        assert_eq!(termination.termination_topology(), [5, 1, 1, 3]);
        assert_eq!(continuation.build_identity(), common[0]);
        assert_eq!(continuation.artifact_identity(), common[1]);
        assert_eq!(continuation.route_set_identity(), common[2]);
        assert_eq!(continuation.owner_generation(), 9);
    }
}