saddle-runtime 0.3.0-alpha.2

//! Controlled Runtime bridge from deployment attestation to Admission startup.
//!
//! This is a build-internal assembly protocol. It is not a business resource
//! configuration API and exposes no raw-value constructor.

use std::marker::PhantomData;

use saddle_admission::{
    CanonicalObservabilityQueueProfileWitness, ComposedGeneratedStartupFactsOwner,
    ObservabilityQueueNormalizationPermit, PendingStartupPlan, PreparedObservabilityQueueBinding,
    RecoverableStartupCreatorFailure, StartupCreatorError, StartupInputAdapter,
    VerifiedDbTerminationServiceProofOwner, VerifiedFilesystemTerminationServiceProofOwner,
    VerifiedSchedulerTerminationServiceProofOwner, VerifiedStartupEnvelopeOwner,
    VerifiedSupervisorTerminationServiceProofOwner, create_pending_startup_plan_recovering,
    prepare_observability_queue_binding,
};
use saddle_core::VerifiedRuntimeResourceParts;
use saddle_observability::file::{
    GenerationPairedFilesystemEvidence, NormalizedSignedProviderFilesystemRuntimeEvidence,
    SignedProviderFilesystemRuntimeEvidence, normalize_signed_provider_filesystem_evidence,
    production_observability_queue_profile_witness,
};

const CPU_UNITS_PER_CORE: usize = 100;

#[doc(hidden)]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FiniteStartupPolicy {
    ExecutionProtected,
    Balanced,
    BurstProtected,
}

#[doc(hidden)]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ResourceEnvelopeVerificationError {
    InvalidCpu,
    InvalidMemory,
    InvalidCredits,
    InvalidTime,
    InvalidIdentity,
    SizeOverflow,
    AdmissionRejected,
}

/// Exact Runtime-owned resource half. It deliberately does not implement the
/// Admission startup-envelope contract: public time, supervisor and Gate
/// identities are paired by later startup stages rather than invented here.
///
/// ```compile_fail
/// use saddle_runtime::resource_envelope::VerifiedRuntimeResourceEnvelope;
/// fn requires_clone<T: Clone>() {}
/// requires_clone::<VerifiedRuntimeResourceEnvelope>();
/// ```
#[doc(hidden)]
#[allow(dead_code)]
pub struct VerifiedRuntimeResourceEnvelope {
    parts: VerifiedRuntimeResourceParts,
    available: [usize; 7],
}

#[allow(dead_code)]
impl VerifiedRuntimeResourceEnvelope {
    #[doc(hidden)]
    pub(crate) fn available_resources(&self) -> [usize; 7] {
        self.available
    }

    #[doc(hidden)]
    pub(crate) fn into_parts(self) -> VerifiedRuntimeResourceParts {
        self.parts
    }
}

/// Atomic outer result. Neither the resource owner nor the directory whole is
/// independently available after success.
#[doc(hidden)]
pub struct FinalPairedStartupAuthority {
    available: [usize; 7],
    paired: saddle_observability::file::FinalPairedFilesystemCoreOwners,
}

#[doc(hidden)]
pub struct FinalStartupAuthorityPairingFailure {
    resource: VerifiedRuntimeResourceEnvelope,
    directory: saddle_observability::file::VerifiedSignedProviderFilesystemBundle,
    generated: ComposedGeneratedStartupFactsOwner,
    seed: saddle_core::VerifiedDeploymentStartupAuthoritySeed,
    gate_seed: saddle_core::VerifiedDeploymentGateBuildAuthoritySeed,
    scheduler_seed: saddle_core::VerifiedDeploymentSchedulerObservationSeed,
    supervisor_seed: saddle_core::VerifiedDeploymentSupervisorStartupSeed,
}

impl FinalStartupAuthorityPairingFailure {
    fn recoverable(
        resource: VerifiedRuntimeResourceEnvelope,
        directory: saddle_observability::file::VerifiedSignedProviderFilesystemBundle,
        generated: ComposedGeneratedStartupFactsOwner,
        seed: saddle_core::VerifiedDeploymentStartupAuthoritySeed,
        gate_seed: saddle_core::VerifiedDeploymentGateBuildAuthoritySeed,
        scheduler_seed: saddle_core::VerifiedDeploymentSchedulerObservationSeed,
        supervisor_seed: saddle_core::VerifiedDeploymentSupervisorStartupSeed,
    ) -> Self {
        Self {
            resource,
            directory,
            generated,
            seed,
            gate_seed,
            scheduler_seed,
            supervisor_seed,
        }
    }

    #[doc(hidden)]
    #[allow(clippy::result_large_err)]
    pub fn into_recoverable(
        self,
    ) -> (
        VerifiedRuntimeResourceEnvelope,
        saddle_observability::file::VerifiedSignedProviderFilesystemBundle,
        ComposedGeneratedStartupFactsOwner,
        saddle_core::VerifiedDeploymentStartupAuthoritySeed,
        saddle_core::VerifiedDeploymentGateBuildAuthoritySeed,
        saddle_core::VerifiedDeploymentSchedulerObservationSeed,
        saddle_core::VerifiedDeploymentSupervisorStartupSeed,
    ) {
        (
            self.resource,
            self.directory,
            self.generated,
            self.seed,
            self.gate_seed,
            self.scheduler_seed,
            self.supervisor_seed,
        )
    }
}

#[doc(hidden)]
#[allow(clippy::result_large_err)]
pub fn pair_final_startup_authority(
    resource: VerifiedRuntimeResourceEnvelope,
    directory: saddle_observability::file::VerifiedSignedProviderFilesystemBundle,
    generated: ComposedGeneratedStartupFactsOwner,
    seed: saddle_core::VerifiedDeploymentStartupAuthoritySeed,
    gate_seed: saddle_core::VerifiedDeploymentGateBuildAuthoritySeed,
    scheduler_seed: saddle_core::VerifiedDeploymentSchedulerObservationSeed,
    supervisor_seed: saddle_core::VerifiedDeploymentSupervisorStartupSeed,
) -> Result<FinalPairedStartupAuthority, FinalStartupAuthorityPairingFailure> {
    let VerifiedRuntimeResourceEnvelope { parts, available } = resource;
    let paired = match saddle_observability::file::pair_signed_filesystem_bundle_final(
        parts,
        directory,
        generated,
        seed,
        gate_seed,
        scheduler_seed,
        supervisor_seed,
    ) {
        Ok(paired) => paired,
        Err((parts, directory, generated, seed, gate_seed, scheduler_seed, supervisor_seed)) => {
            return Err(FinalStartupAuthorityPairingFailure::recoverable(
                VerifiedRuntimeResourceEnvelope { parts, available },
                directory,
                generated,
                seed,
                gate_seed,
                scheduler_seed,
                supervisor_seed,
            ));
        }
    };
    Ok(FinalPairedStartupAuthority { available, paired })
}

/// The sole production resource-parts verifier. Failure returns the exact
/// same linear owner so the surrounding startup transaction can retry or
/// roll back without re-minting authority.
#[doc(hidden)]
#[allow(clippy::result_large_err)]
pub fn verify_runtime_resource_parts(
    parts: VerifiedRuntimeResourceParts,
) -> Result<
    VerifiedRuntimeResourceEnvelope,
    (
        ResourceEnvelopeVerificationError,
        VerifiedRuntimeResourceParts,
    ),
> {
    let available = match parts.resource_shape() {
        Ok(shape) => shape.into_capacity(),
        Err(_) => {
            return Err((ResourceEnvelopeVerificationError::SizeOverflow, parts));
        }
    };
    Ok(VerifiedRuntimeResourceEnvelope { parts, available })
}

#[doc(hidden)]
pub struct RecoverableRuntimeStartupFailure<G, D, F, S, U> {
    error: ResourceEnvelopeVerificationError,
    envelope: VerifiedResourceEnvelope,
    generated: G,
    db_service: D,
    filesystem_service: F,
    scheduler_service: S,
    supervisor_service: U,
}

impl<G, D, F, S, U> RecoverableRuntimeStartupFailure<G, D, F, S, U> {
    #[doc(hidden)]
    #[allow(clippy::type_complexity)]
    pub fn into_parts(
        self,
    ) -> (
        ResourceEnvelopeVerificationError,
        VerifiedResourceEnvelope,
        G,
        RuntimeTerminationProofOwners<D, F, S, U>,
    ) {
        (
            self.error,
            self.envelope,
            self.generated,
            (
                self.db_service,
                self.filesystem_service,
                self.scheduler_service,
                self.supervisor_service,
            ),
        )
    }
}

#[doc(hidden)]
pub enum RecoverableFilesystemStartupFailure<G, D, S, U> {
    Binding {
        envelope: VerifiedResourceEnvelope,
        generated: G,
        db_service: D,
        filesystem: GenerationPairedFilesystemEvidence,
        scheduler_service: S,
        supervisor_service: U,
    },
    Creator {
        failure: RecoverableRuntimeStartupFailure<
            G,
            D,
            saddle_observability::file::WriterTerminationProof,
            S,
            U,
        >,
        physical: SignedProviderFilesystemRuntimeEvidence,
    },
    Permit {
        pending: PendingStartupPlan,
        physical: SignedProviderFilesystemRuntimeEvidence,
    },
    Prepare {
        pending: PendingStartupPlan,
        permit: ObservabilityQueueNormalizationPermit,
        witness: CanonicalObservabilityQueueProfileWitness,
        physical: SignedProviderFilesystemRuntimeEvidence,
    },
    Normalize {
        pending: PendingStartupPlan,
        binding: PreparedObservabilityQueueBinding,
        physical: SignedProviderFilesystemRuntimeEvidence,
    },
}

impl<G, D, S, U> RecoverableFilesystemStartupFailure<G, D, S, U> {
    #[doc(hidden)]
    pub fn consume_for_application_error(self) -> ResourceEnvelopeVerificationError {
        match self {
            Self::Binding { .. } => ResourceEnvelopeVerificationError::InvalidIdentity,
            Self::Creator { failure, physical } => {
                let (error, envelope, generated, proofs) = failure.into_parts();
                let (_envelope, _generated, _proofs, _physical) =
                    (envelope, generated, proofs, physical);
                error
            }
            Self::Permit { .. } => ResourceEnvelopeVerificationError::AdmissionRejected,
            Self::Prepare { .. } => ResourceEnvelopeVerificationError::AdmissionRejected,
            Self::Normalize { .. } => ResourceEnvelopeVerificationError::AdmissionRejected,
        }
    }
}

/// Gate-owned observation protocol. Production Gate locks one implementation
/// and one assembly; Runtime deliberately provides no closure blanket.
#[doc(hidden)]
pub(crate) trait DeploymentResourceAttestation: Sized {
    fn cpu_quota_us(&self) -> u64;
    fn cpu_period_us(&self) -> u64;
    fn effective_cpuset(&self) -> &[usize];

    fn memory_max_bytes(&self) -> usize;
    fn memory_high_bytes(&self) -> usize;
    fn saddle_logical_memory_bytes(&self) -> usize;
    fn saddle_requested_memory_bytes(&self) -> usize;

    fn registration_credits(&self) -> usize;
    fn event_credits(&self) -> usize;
    fn db_connection_credits(&self) -> usize;
    fn db_operation_credits(&self) -> usize;

    /// Public request timeout and shutdown grace policy.
    fn public_time_policy_ms(&self) -> [u64; 2];

    fn resource_attestation(&self) -> [u8; 32];
    fn public_time_policy_attestation(&self) -> [u8; 32];
    fn environment_identity(&self) -> [u8; 32];
    fn supervisor_attestation(&self) -> [u8; 32];
    fn build_identity(&self) -> [u8; 32];
    fn gate_identity(&self) -> [u8; 32];
}

/// Consuming, non-Clone owner produced only by the synchronous verifier.
#[doc(hidden)]
#[derive(Debug)]
pub struct VerifiedResourceEnvelope {
    available: [usize; 7],
    public_time_policy: [u64; 2],
    pub(crate) resource_attestation: [u8; 32],
    public_time_policy_attestation: [u8; 32],
    pub(crate) environment_identity: [u8; 32],
    pub(crate) supervisor_attestation: [u8; 32],
    pub(crate) build_identity: [u8; 32],
    pub(crate) gate_identity: [u8; 32],
}

struct RuntimeStartupAdapter<D, F, S, U> {
    policy: FiniteStartupPolicy,
    provenance: [u8; 32],
    _owners: PhantomData<RuntimeStartupOwnerTypes<D, F, S, U>>,
}

type RuntimeStartupOwnerTypes<D, F, S, U> = fn() -> (D, F, S, U);
type RuntimeTerminationProofOwners<D, F, S, U> = (D, F, S, U);

impl<D, F, S, U> StartupInputAdapter for RuntimeStartupAdapter<D, F, S, U>
where
    D: VerifiedDbTerminationServiceProofOwner,
    F: VerifiedFilesystemTerminationServiceProofOwner,
    S: VerifiedSchedulerTerminationServiceProofOwner,
    U: VerifiedSupervisorTerminationServiceProofOwner,
{
    type VerifiedEnvelopeOwner = VerifiedResourceEnvelope;
    type GeneratedFactsOwner = ComposedGeneratedStartupFactsOwner;
    type DbServiceProofOwner = D;
    type FilesystemServiceProofOwner = F;
    type SchedulerServiceProofOwner = S;
    type SupervisorServiceProofOwner = U;

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

    fn finite_policy_version(&self) -> u8 {
        match self.policy {
            FiniteStartupPolicy::ExecutionProtected => 1,
            FiniteStartupPolicy::Balanced => 2,
            FiniteStartupPolicy::BurstProtected => 3,
        }
    }

    fn db_return_issuer_identity(&self) -> [u8; 32] {
        crate::db_return_budget::DB_RETURN_BUDGET_ISSUER_IDENTITY
    }
}

impl VerifiedStartupEnvelopeOwner for VerifiedResourceEnvelope {
    fn resource_attestation(&self) -> [u8; 32] {
        self.resource_attestation
    }

    fn available_resources(&self) -> [usize; 7] {
        self.available
    }

    fn public_time_policy_ms(&self) -> [u64; 2] {
        self.public_time_policy
    }

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

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

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

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

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

/// Synchronously verifies and consumes the deployment observation.
#[doc(hidden)]
pub(crate) fn verify_resource_envelope<A: DeploymentResourceAttestation>(
    attestation: A,
) -> Result<VerifiedResourceEnvelope, ResourceEnvelopeVerificationError> {
    let cpuset = attestation.effective_cpuset();
    if attestation.cpu_quota_us() == 0
        || attestation.cpu_period_us() == 0
        || cpuset.is_empty()
        || cpuset.windows(2).any(|pair| pair[0] >= pair[1])
    {
        return Err(ResourceEnvelopeVerificationError::InvalidCpu);
    }
    let quota_units = usize::try_from(attestation.cpu_quota_us())
        .ok()
        .and_then(|quota| quota.checked_mul(CPU_UNITS_PER_CORE))
        .ok_or(ResourceEnvelopeVerificationError::SizeOverflow)?
        / usize::try_from(attestation.cpu_period_us())
            .map_err(|_| ResourceEnvelopeVerificationError::SizeOverflow)?;
    let cpuset_units = cpuset
        .len()
        .checked_mul(CPU_UNITS_PER_CORE)
        .ok_or(ResourceEnvelopeVerificationError::SizeOverflow)?;
    let cpu_units = quota_units.min(cpuset_units);
    if cpu_units < CPU_UNITS_PER_CORE {
        return Err(ResourceEnvelopeVerificationError::InvalidCpu);
    }

    let memory_max = attestation.memory_max_bytes();
    let memory_high = attestation.memory_high_bytes();
    let logical = attestation.saddle_logical_memory_bytes();
    let requested = attestation.saddle_requested_memory_bytes();
    if memory_max == 0
        || memory_high == 0
        || logical == 0
        || requested == 0
        || memory_high > memory_max
        || logical > memory_high
        || requested > memory_high
    {
        return Err(ResourceEnvelopeVerificationError::InvalidMemory);
    }

    let registration = attestation.registration_credits();
    let events = attestation.event_credits();
    let db_connections = attestation.db_connection_credits();
    let db_operations = attestation.db_operation_credits();
    if registration == 0 || events == 0 || ((db_connections == 0) != (db_operations == 0)) {
        return Err(ResourceEnvelopeVerificationError::InvalidCredits);
    }

    let public_time_policy = attestation.public_time_policy_ms();
    if public_time_policy.contains(&0) {
        return Err(ResourceEnvelopeVerificationError::InvalidTime);
    }
    let environment = attestation.environment_identity();
    let resource_attestation = attestation.resource_attestation();
    let public_time_policy_attestation = attestation.public_time_policy_attestation();
    let supervisor_attestation = attestation.supervisor_attestation();
    let build_identity = attestation.build_identity();
    let gate_identity = attestation.gate_identity();
    if [
        environment,
        resource_attestation,
        public_time_policy_attestation,
        supervisor_attestation,
        build_identity,
        gate_identity,
    ]
    .contains(&[0; 32])
    {
        return Err(ResourceEnvelopeVerificationError::InvalidIdentity);
    }

    // Gate's resource attestation is the canonical digest that covers this
    // independently non-zero environment identity. Runtime does not create a
    // second identity or attempt to re-hash Gate's deployment manifest.

    Ok(VerifiedResourceEnvelope {
        available: [
            cpu_units,
            logical,
            requested,
            registration,
            events,
            db_connections,
            db_operations,
        ],
        public_time_policy,
        resource_attestation,
        public_time_policy_attestation,
        environment_identity: environment,
        supervisor_attestation,
        build_identity,
        gate_identity,
    })
}

/// The sole Runtime bridge into Admission's crate-private solver.
#[doc(hidden)]
#[cfg(test)]
pub(crate) fn create_runtime_startup_plan<A, D, F, S, U>(
    policy: FiniteStartupPolicy,
    adapter_provenance: [u8; 32],
    attestation: A,
    generated: ComposedGeneratedStartupFactsOwner,
    termination_proofs: RuntimeTerminationProofOwners<D, F, S, U>,
) -> Result<PendingStartupPlan, ResourceEnvelopeVerificationError>
where
    A: DeploymentResourceAttestation,
    D: VerifiedDbTerminationServiceProofOwner,
    F: VerifiedFilesystemTerminationServiceProofOwner,
    S: VerifiedSchedulerTerminationServiceProofOwner,
    U: VerifiedSupervisorTerminationServiceProofOwner,
{
    let (db_service, filesystem_service, scheduler_service, supervisor_service) =
        termination_proofs;
    if adapter_provenance == [0; 32] {
        return Err(ResourceEnvelopeVerificationError::InvalidIdentity);
    }
    let envelope = verify_resource_envelope(attestation)?;
    create_runtime_startup_plan_from_verified(
        policy,
        adapter_provenance,
        envelope,
        generated,
        (
            db_service,
            filesystem_service,
            scheduler_service,
            supervisor_service,
        ),
    )
}

/// The facade-only successor used after Gate has synchronously verified and
/// consumed the deployment attestation. No raw resource observation crosses
/// this boundary.
#[doc(hidden)]
pub fn create_runtime_startup_plan_from_verified<D, F, S, U>(
    policy: FiniteStartupPolicy,
    adapter_provenance: [u8; 32],
    envelope: VerifiedResourceEnvelope,
    generated: ComposedGeneratedStartupFactsOwner,
    termination_proofs: RuntimeTerminationProofOwners<D, F, S, U>,
) -> Result<PendingStartupPlan, ResourceEnvelopeVerificationError>
where
    D: VerifiedDbTerminationServiceProofOwner,
    F: VerifiedFilesystemTerminationServiceProofOwner,
    S: VerifiedSchedulerTerminationServiceProofOwner,
    U: VerifiedSupervisorTerminationServiceProofOwner,
{
    create_runtime_startup_plan_from_verified_recovering(
        policy,
        adapter_provenance,
        envelope,
        generated,
        termination_proofs,
    )
    .map_err(|failure| failure.into_parts().0)
}

#[doc(hidden)]
#[allow(clippy::result_large_err)]
pub fn create_runtime_startup_plan_from_verified_recovering<D, F, S, U>(
    policy: FiniteStartupPolicy,
    adapter_provenance: [u8; 32],
    envelope: VerifiedResourceEnvelope,
    generated: ComposedGeneratedStartupFactsOwner,
    termination_proofs: RuntimeTerminationProofOwners<D, F, S, U>,
) -> Result<
    PendingStartupPlan,
    RecoverableRuntimeStartupFailure<ComposedGeneratedStartupFactsOwner, D, F, S, U>,
>
where
    D: VerifiedDbTerminationServiceProofOwner,
    F: VerifiedFilesystemTerminationServiceProofOwner,
    S: VerifiedSchedulerTerminationServiceProofOwner,
    U: VerifiedSupervisorTerminationServiceProofOwner,
{
    let (db_service, filesystem_service, scheduler_service, supervisor_service) =
        termination_proofs;
    if adapter_provenance == [0; 32] {
        return Err(RecoverableRuntimeStartupFailure {
            error: ResourceEnvelopeVerificationError::InvalidIdentity,
            envelope,
            generated,
            db_service,
            filesystem_service,
            scheduler_service,
            supervisor_service,
        });
    }
    create_pending_startup_plan_recovering(
        RuntimeStartupAdapter::<D, F, S, U> {
            policy,
            provenance: adapter_provenance,
            _owners: PhantomData,
        },
        envelope,
        generated,
        db_service,
        filesystem_service,
        scheduler_service,
        supervisor_service,
    )
    .map_err(|failure| {
        let RecoverableStartupCreatorFailure {
            error,
            envelope,
            generated,
            db_service,
            filesystem_service,
            scheduler_service,
            supervisor_service,
            ..
        } = failure;
        RecoverableRuntimeStartupFailure {
            error: match error {
                StartupCreatorError::InvalidAdapterFacts => {
                    ResourceEnvelopeVerificationError::InvalidIdentity
                }
                StartupCreatorError::SolverRejected => {
                    ResourceEnvelopeVerificationError::AdmissionRejected
                }
            },
            envelope,
            generated,
            db_service,
            filesystem_service,
            scheduler_service,
            supervisor_service,
        }
    })
}

/// Sole production-shaped filesystem bridge. The provider bundle is checked,
/// split and consumed in this call; no standalone queue profile is accepted.
#[doc(hidden)]
#[allow(clippy::too_many_arguments)]
#[allow(clippy::result_large_err)]
pub fn create_runtime_startup_plan_with_filesystem_bundle<D, S, U>(
    policy: FiniteStartupPolicy,
    adapter_provenance: [u8; 32],
    envelope: VerifiedResourceEnvelope,
    generated: ComposedGeneratedStartupFactsOwner,
    db_service: D,
    filesystem: GenerationPairedFilesystemEvidence,
    scheduler_service: S,
    supervisor_service: U,
) -> Result<
    (
        PendingStartupPlan,
        NormalizedSignedProviderFilesystemRuntimeEvidence,
    ),
    RecoverableFilesystemStartupFailure<ComposedGeneratedStartupFactsOwner, D, S, U>,
>
where
    D: VerifiedDbTerminationServiceProofOwner,
    S: VerifiedSchedulerTerminationServiceProofOwner,
    U: VerifiedSupervisorTerminationServiceProofOwner,
{
    let filesystem = match filesystem
        .consume_runtime_binding(VerifiedStartupEnvelopeOwner::build_identity(&envelope))
    {
        Ok(filesystem) => filesystem,
        Err(filesystem) => {
            return Err(RecoverableFilesystemStartupFailure::Binding {
                envelope,
                generated,
                db_service,
                filesystem,
                scheduler_service,
                supervisor_service,
            });
        }
    };
    let (filesystem_service, physical) = filesystem.into_runtime_evidence();
    let mut pending = match create_runtime_startup_plan_from_verified_recovering(
        policy,
        adapter_provenance,
        envelope,
        generated,
        (
            db_service,
            filesystem_service,
            scheduler_service,
            supervisor_service,
        ),
    ) {
        Ok(pending) => pending,
        Err(failure) => {
            return Err(RecoverableFilesystemStartupFailure::Creator { failure, physical });
        }
    };
    let permit = match pending.issue_observability_queue_normalization_permit() {
        Ok(permit) => permit,
        Err(_) => return Err(RecoverableFilesystemStartupFailure::Permit { pending, physical }),
    };
    let witness = production_observability_queue_profile_witness();
    let (pending, binding) = match prepare_observability_queue_binding(pending, permit, witness) {
        Ok(parts) => parts,
        Err((_, pending, permit, witness)) => {
            return Err(RecoverableFilesystemStartupFailure::Prepare {
                pending,
                permit,
                witness,
                physical,
            });
        }
    };
    let physical = match normalize_signed_provider_filesystem_evidence(binding, physical) {
        Ok(physical) => physical,
        Err((binding, physical)) => {
            return Err(RecoverableFilesystemStartupFailure::Normalize {
                pending,
                binding,
                physical,
            });
        }
    };
    Ok((pending, physical))
}

#[cfg(test)]
pub(crate) mod tests {
    use saddle_admission::{
        GeneratedAllocatorCost, GeneratedBuildCostClass, GeneratedBuildTopologyConstant,
        GeneratedCalibrationConstant, GeneratedResourceDimension, GeneratedSupportLimit,
        GeneratedTerminationTopologyWorkOwner, RuntimeBuildCapacitySourceLeaf,
        RuntimeCapacityCalibrationSourceLeaf, ServiceCapacitySourceLeaf, StartupActualFacts,
        StartupClaimKind, compose_generated_startup_facts_test_fixture,
    };

    use super::*;

    struct Attestation {
        cpu_quota: u64,
        cpuset: Vec<usize>,
        memory_high: usize,
        db: usize,
        public_time_policy: [u64; 2],
        identity: [u8; 32],
    }

    impl DeploymentResourceAttestation for Attestation {
        fn cpu_quota_us(&self) -> u64 {
            self.cpu_quota
        }
        fn cpu_period_us(&self) -> u64 {
            100_000
        }
        fn effective_cpuset(&self) -> &[usize] {
            &self.cpuset
        }
        fn memory_max_bytes(&self) -> usize {
            128_000_000
        }
        fn memory_high_bytes(&self) -> usize {
            self.memory_high
        }
        fn saddle_logical_memory_bytes(&self) -> usize {
            self.memory_high / 2
        }
        fn saddle_requested_memory_bytes(&self) -> usize {
            self.memory_high / 2
        }
        fn registration_credits(&self) -> usize {
            1_024
        }
        fn event_credits(&self) -> usize {
            1_024
        }
        fn db_connection_credits(&self) -> usize {
            self.db
        }
        fn db_operation_credits(&self) -> usize {
            self.db
        }
        fn public_time_policy_ms(&self) -> [u64; 2] {
            self.public_time_policy
        }
        fn resource_attestation(&self) -> [u8; 32] {
            self.identity
        }
        fn public_time_policy_attestation(&self) -> [u8; 32] {
            [2; 32]
        }
        fn environment_identity(&self) -> [u8; 32] {
            self.identity
        }
        fn supervisor_attestation(&self) -> [u8; 32] {
            [3; 32]
        }
        fn build_identity(&self) -> [u8; 32] {
            [4; 32]
        }
        fn gate_identity(&self) -> [u8; 32] {
            [5; 32]
        }
    }

    fn attestation() -> Attestation {
        Attestation {
            cpu_quota: 800_000,
            cpuset: (0..8).collect(),
            memory_high: 64_000_000,
            db: 64,
            public_time_policy: [2_000, 8_000],
            identity: [1; 32],
        }
    }

    const COMMON: [[u8; 32]; 3] = [[0x41; 32], [0x42; 32], [0x43; 32]];
    const PROVENANCE: [[u8; 32]; 9] = [
        [0x61; 32], [0x62; 32], [0x63; 32], [0x64; 32], [0x65; 32], [0x66; 32], [0x67; 32],
        [0x68; 32], [0x69; 32],
    ];

    struct GeneratedLeaf;
    impl ServiceCapacitySourceLeaf for GeneratedLeaf {
        fn leaf_identity(&self) -> [u8; 32] {
            [0x11; 32]
        }
        fn common_identities(&self) -> [[u8; 32]; 3] {
            COMMON
        }
        fn owner_generation(&self) -> u64 {
            7
        }
        fn route_type_closure_identity(&self) -> [u8; 32] {
            [0x51; 32]
        }
        fn route_count(&self) -> usize {
            1
        }
        fn route_identity(&self, index: usize) -> Option<u64> {
            (index == 0).then_some(1)
        }
        fn managed_commitment_bytes(&self, index: usize) -> Option<usize> {
            (index == 0).then_some(1_024)
        }
        fn managed_objects_peak(&self, index: usize) -> Option<usize> {
            (index == 0).then_some(2)
        }
        fn db_connections(&self, index: usize) -> Option<usize> {
            (index == 0).then_some(1)
        }
        fn db_operations(&self, index: usize) -> Option<usize> {
            (index == 0).then_some(1)
        }
    }
    impl RuntimeBuildCapacitySourceLeaf for GeneratedLeaf {
        fn leaf_identity(&self) -> [u8; 32] {
            [0x22; 32]
        }
        fn common_identities(&self) -> [[u8; 32]; 3] {
            COMMON
        }
        fn owner_generation(&self) -> u64 {
            7
        }
        fn component_proof_identity(&self) -> [u8; 32] {
            [0x52; 32]
        }
        fn expected_calibration_provenance(&self) -> [[u8; 32]; 9] {
            PROVENANCE
        }
        fn solver_schema_identity(&self) -> [u8; 32] {
            [0x92; 32]
        }
        fn admission_layout_identity(&self) -> [u8; 32] {
            [0x94; 32]
        }
        fn cost(
            &self,
            class: GeneratedBuildCostClass,
            dimension: GeneratedResourceDimension,
        ) -> Option<usize> {
            Some(
                [
                    [0, 10_000, 10_000, 3, 3, 0, 0],
                    [100, 100, 100, 0, 0, 0, 0],
                    [0, 20_000, 20_000, 1, 1, 0, 0],
                    [0, 1_000, 1_000, 1, 1, 0, 0],
                    [0, 1_000, 1_000, 0, 0, 1, 1],
                ][class as usize][dimension as usize],
            )
        }
        fn support_limit(&self, limit: GeneratedSupportLimit) -> Option<usize> {
            Some([8, 128, 128, 64][limit as usize])
        }
        fn topology_constant(&self, value: GeneratedBuildTopologyConstant) -> Option<usize> {
            Some([1, 2, 3][value as usize])
        }
        fn route_count(&self) -> usize {
            1
        }
        fn route_identity(&self, index: usize) -> Option<u64> {
            (index == 0).then_some(1)
        }
        fn framework_bytes(&self, index: usize) -> Option<usize> {
            (index == 0).then_some(512)
        }
        fn task_storage_bytes(&self, index: usize) -> Option<usize> {
            (index == 0).then_some(4_096)
        }
        fn response_carrier_bytes(&self, index: usize) -> Option<usize> {
            (index == 0).then_some(512)
        }
    }
    impl RuntimeCapacityCalibrationSourceLeaf for GeneratedLeaf {
        fn leaf_identity(&self) -> [u8; 32] {
            [0x69; 32]
        }
        fn common_identities(&self) -> [[u8; 32]; 3] {
            COMMON
        }
        fn owner_generation(&self) -> u64 {
            7
        }
        fn calibration_provenance(&self) -> [[u8; 32]; 9] {
            PROVENANCE
        }
        fn fixed_cost(&self, dimension: GeneratedResourceDimension) -> Option<usize> {
            Some([0, 16_000, 16_000, 3, 3, 0, 0][dimension as usize])
        }
        fn allocator_cost(&self, _cost: GeneratedAllocatorCost) -> Option<usize> {
            Some(1_000)
        }
        fn calibrated_constant(&self, value: GeneratedCalibrationConstant) -> Option<usize> {
            Some([64, 1_024, 512][value as usize])
        }
    }
    impl GeneratedTerminationTopologyWorkOwner for GeneratedLeaf {
        fn leaf_identity(&self) -> [u8; 32] {
            [0x71; 32]
        }
        fn common_identities(&self) -> [[u8; 32]; 3] {
            COMMON
        }
        fn owner_generation(&self) -> u64 {
            7
        }
        fn termination_topology(&self) -> [u64; 4] {
            [1, 1, 1, 2]
        }
        fn termination_topology_identity(&self) -> [u8; 32] {
            [9; 32]
        }
        fn db_return_work_identity(&self) -> [u8; 32] {
            [10; 32]
        }
        fn writer_work_identity(&self) -> [u8; 32] {
            [11; 32]
        }
        fn runtime_work_identity(&self) -> [u8; 32] {
            [12; 32]
        }
    }

    fn generated() -> ComposedGeneratedStartupFactsOwner {
        compose_generated_startup_facts_test_fixture(
            GeneratedLeaf,
            GeneratedLeaf,
            GeneratedLeaf,
            GeneratedLeaf,
        )
        .unwrap()
    }

    struct DbService;
    struct FilesystemService;
    struct SchedulerService;
    struct SupervisorService;
    impl VerifiedDbTerminationServiceProofOwner for DbService {
        fn work_identity(&self) -> [u8; 32] {
            [10; 32]
        }
        fn service_attestation(&self) -> [u8; 32] {
            [13; 32]
        }
        fn max_service_nanos(&self) -> u64 {
            2_000_000
        }
    }
    impl VerifiedFilesystemTerminationServiceProofOwner for FilesystemService {
        fn work_identity(&self) -> [u8; 32] {
            [11; 32]
        }
        fn service_attestation(&self) -> [u8; 32] {
            [14; 32]
        }
        fn max_service_nanos(&self) -> u64 {
            1_000_000
        }
    }
    impl VerifiedSchedulerTerminationServiceProofOwner for SchedulerService {
        fn runtime_work_identity(&self) -> [u8; 32] {
            [12; 32]
        }
        fn service_attestation(&self) -> [u8; 32] {
            [15; 32]
        }
        fn max_delivery_nanos(&self) -> u64 {
            1_000_000
        }
    }
    impl VerifiedSupervisorTerminationServiceProofOwner for SupervisorService {
        fn service_attestation(&self) -> [u8; 32] {
            [16; 32]
        }
        fn shutdown_delivery_nanos(&self) -> u64 {
            1_000_000
        }
        fn delivery_headroom_nanos(&self) -> u64 {
            1_000_000
        }
    }

    pub(crate) fn pending_plan() -> PendingStartupPlan {
        create_runtime_startup_plan(
            FiniteStartupPolicy::Balanced,
            [9; 32],
            attestation(),
            generated(),
            (
                DbService,
                FilesystemService,
                SchedulerService,
                SupervisorService,
            ),
        )
        .unwrap()
    }

    #[test]
    fn verified_owner_enters_the_only_controlled_creator() {
        let mut plan = pending_plan();
        assert_ne!(
            plan.claim(StartupClaimKind::Runtime).unwrap().expected(),
            StartupActualFacts::Runtime {
                worker_threads: 0,
                event_credits: 0,
            }
        );
    }

    #[test]
    fn production_creator_failure_returns_every_owner_for_an_uncharged_retry() {
        let envelope = verify_resource_envelope(attestation()).unwrap();
        let failure = match create_runtime_startup_plan_from_verified_recovering(
            FiniteStartupPolicy::Balanced,
            [0; 32],
            envelope,
            generated(),
            (
                DbService,
                FilesystemService,
                SchedulerService,
                SupervisorService,
            ),
        ) {
            Ok(_) => panic!("zero provenance must fail before a plan can charge resources"),
            Err(failure) => failure,
        };
        let (error, envelope, generated, proofs) = failure.into_parts();
        assert_eq!(error, ResourceEnvelopeVerificationError::InvalidIdentity);

        let mut retried = match create_runtime_startup_plan_from_verified_recovering(
            FiniteStartupPolicy::Balanced,
            [9; 32],
            envelope,
            generated,
            proofs,
        ) {
            Ok(plan) => plan,
            Err(_) => panic!("returned owners must support one clean retry"),
        };
        assert!(retried.claim(StartupClaimKind::Runtime).is_ok());
    }

    #[test]
    fn invalid_resource_and_identity_inputs_fail_before_creator() {
        let mut invalid_cpu = attestation();
        invalid_cpu.cpuset = vec![1, 1];
        assert_eq!(
            verify_resource_envelope(invalid_cpu).unwrap_err(),
            ResourceEnvelopeVerificationError::InvalidCpu
        );
        let mut invalid_memory = attestation();
        invalid_memory.memory_high = 0;
        assert_eq!(
            verify_resource_envelope(invalid_memory).unwrap_err(),
            ResourceEnvelopeVerificationError::InvalidMemory
        );
        let mut invalid_time = attestation();
        invalid_time.public_time_policy[1] = 0;
        assert_eq!(
            verify_resource_envelope(invalid_time).unwrap_err(),
            ResourceEnvelopeVerificationError::InvalidTime
        );
        let mut invalid_identity = attestation();
        invalid_identity.identity = [0; 32];
        assert_eq!(
            verify_resource_envelope(invalid_identity).unwrap_err(),
            ResourceEnvelopeVerificationError::InvalidIdentity
        );
    }
}