saddle-service 0.2.0

//! Compile-time closed C4/C5 Service execution candidate.
//!
//! The application generator owns the concrete dispatcher type. This module
//! supplies only fixed payloads, type-derived layout proof, and the linear
//! lookup/execution assembly. It deliberately does not erase handlers or
//! futures.

use std::{
    any::TypeId,
    future::Future,
    marker::PhantomData,
    mem,
    pin::Pin,
    task::{Context, Poll},
};

use saddle_admission::{
    DbRequestPermit, DbRouteCreditDemand, DbRouteResources, ManagedBytes, ManagedResponse,
    ManagedResponseBuilder, RequestMemory,
};
use saddle_db::internal::{
    ManagedWriteResult, QueryOptionalExecution, QueryOptionalInvocation,
    QueryOptionalOperationProof, StaticQueryOptionalOperation, StaticWriteOperation,
    TransactionExecution, TransactionInvocation, TransactionOperationProof, WriteExecution,
    WriteInvocation, WriteOperationProof,
};

use crate::{
    Service,
    c4_entry::{
        EntryCapabilityLimits, EntryFraming, EntryPlanSpec, RegisteredDbDemand, RegisteredEntry,
        RegisteredRouteExecutionProof, RegistryError, ServiceEntryCapability,
        ServiceEntryCapabilityBuilder,
    },
    registry::ServiceDescriptor,
};

const SERVICE_CAPACITY_LEAF_SCHEMA: &[u8] = b"saddle.service-capacity-leaf.v1";

/// Generator/Gate-owned artifact binding consumed by the sole Service leaf
/// assembly. It carries identities only; no resource or capacity array crosses
/// this boundary.
#[doc(hidden)]
pub trait GeneratedServiceCapacityArtifactOwner: Sized {
    fn build_identity(&self) -> [u8; 32];
    fn artifact_identity(&self) -> [u8; 32];
    fn owner_generation(&self) -> u64;
}

#[derive(Debug)]
pub(super) struct ServiceCapacityRouteRecord {
    pub(super) identity: u64,
    pub(super) commitment: usize,
    pub(super) managed_objects_peak: usize,
    pub(super) db_connections: usize,
    pub(super) db_operations: usize,
    pub(super) source_identity: [u8; 32],
}

#[derive(Clone, Copy)]
enum ManagedOwnerContribution {
    RequestBody,
    ResponseReservation,
}

impl ManagedOwnerContribution {
    const fn count(self) -> usize {
        match self {
            Self::RequestBody | Self::ResponseReservation => 1,
        }
    }
}

fn compiled_route_managed_objects_peak() -> Result<usize, ExecutionError> {
    [
        ManagedOwnerContribution::RequestBody,
        ManagedOwnerContribution::ResponseReservation,
    ]
    .into_iter()
    .try_fold(0_usize, |peak, contribution| {
        peak.checked_add(contribution.count())
            .ok_or(ExecutionError::OutputTooLarge)
    })
}

/// Non-replayable Service-owned leaf for Admission's multi-leaf verifier.
/// All fields are derived during the same freeze that creates execution.
#[doc(hidden)]
#[derive(Debug)]
pub(super) struct FrozenServiceCapacitySourceLeaf {
    pub(super) leaf_identity: [u8; 32],
    pub(super) common_identities: [[u8; 32]; 3],
    owner_generation: u64,
    pub(super) route_type_closure_identity: [u8; 32],
    pub(super) routes: Box<[ServiceCapacityRouteRecord]>,
}

/// Borrowed raw capacity view. It observes the Service-owned bundle but is not
/// an independently owned proof or an authorization capability.
#[doc(hidden)]
pub struct FrozenServiceCapacitySourceView<'a> {
    leaf: &'a FrozenServiceCapacitySourceLeaf,
}

impl saddle_admission::ServiceCapacitySourceLeaf for FrozenServiceCapacitySourceView<'_> {
    fn leaf_identity(&self) -> [u8; 32] {
        self.leaf.leaf_identity
    }

    fn common_identities(&self) -> [[u8; 32]; 3] {
        self.leaf.common_identities
    }

    fn owner_generation(&self) -> u64 {
        self.leaf.owner_generation
    }

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

    fn route_count(&self) -> usize {
        self.leaf.routes.len()
    }

    fn route_identity(&self, index: usize) -> Option<u64> {
        self.leaf.routes.get(index).map(|route| route.identity)
    }

    fn managed_commitment_bytes(&self, index: usize) -> Option<usize> {
        self.leaf.routes.get(index).map(|route| route.commitment)
    }

    fn managed_objects_peak(&self, index: usize) -> Option<usize> {
        self.leaf
            .routes
            .get(index)
            .map(|route| route.managed_objects_peak)
    }

    fn db_connections(&self, index: usize) -> Option<usize> {
        self.leaf
            .routes
            .get(index)
            .map(|route| route.db_connections)
    }

    fn db_operations(&self, index: usize) -> Option<usize> {
        self.leaf.routes.get(index).map(|route| route.db_operations)
    }
}

struct ServiceCapacityDigest([u64; 4]);

impl ServiceCapacityDigest {
    fn new() -> Self {
        Self([
            0xcbf2_9ce4_8422_2325,
            0x9e37_79b9_7f4a_7c15,
            0x6a09_e667_f3bc_c909,
            0xbb67_ae85_84ca_a73b,
        ])
    }

    fn write(&mut self, bytes: &[u8]) {
        const PRIMES: [u64; 4] = [
            0x0000_0100_0000_01b3,
            0x9e37_79b1_85eb_ca87,
            0xc2b2_ae3d_27d4_eb4f,
            0x1656_67b1_9e37_79f9,
        ];
        for (index, byte) in bytes.iter().copied().enumerate() {
            for (lane, prime) in self.0.iter_mut().zip(PRIMES) {
                *lane ^= u64::from(byte).wrapping_add(index as u64);
                *lane = lane.wrapping_mul(prime);
                *lane ^= *lane >> 29;
            }
        }
    }

    fn usize(&mut self, value: usize) {
        self.write(&value.to_le_bytes());
    }

    fn finish(self) -> [u8; 32] {
        let mut output = [0; 32];
        for (index, lane) in self.0.into_iter().enumerate() {
            output[index * 8..(index + 1) * 8].copy_from_slice(&lane.to_le_bytes());
        }
        output
    }
}

struct ServiceRouteSourceFacts<'a> {
    route_token: &'a [u8],
    framing: EntryFraming,
    identity: crate::c4_entry::RegisteredEntryIdentity,
    layout_source_identity: [u8; 32],
    commitment: usize,
    managed_objects_peak: usize,
    db_connections: usize,
    db_operations: usize,
}

fn service_route_source_identity(facts: ServiceRouteSourceFacts<'_>) -> [u8; 32] {
    let mut digest = ServiceCapacityDigest::new();
    digest.write(SERVICE_CAPACITY_LEAF_SCHEMA);
    digest.write(b"route-source");
    digest.usize(facts.route_token.len());
    digest.write(facts.route_token);
    digest.write(&facts.framing.source_code());
    digest.write(&facts.identity.contract().opaque().to_le_bytes());
    digest.write(&facts.identity.plan().opaque().to_le_bytes());
    digest.write(&facts.identity.factory().opaque().to_le_bytes());
    digest.write(&facts.layout_source_identity);
    digest.usize(facts.commitment);
    digest.usize(facts.managed_objects_peak);
    digest.usize(facts.db_connections);
    digest.usize(facts.db_operations);
    digest.finish()
}

fn service_route_type_closure_identity(
    artifact: [u8; 32],
    routes: &[ServiceCapacityRouteRecord],
) -> [u8; 32] {
    let mut digest = ServiceCapacityDigest::new();
    digest.write(SERVICE_CAPACITY_LEAF_SCHEMA);
    digest.write(b"route-type-owner-closure");
    digest.write(&artifact);
    digest.write(b"managed-request-body-owner");
    digest.write(b"managed-response-reservation-owner");
    digest.usize(routes.len());
    for route in routes {
        digest.write(&route.source_identity);
        digest.usize(route.managed_objects_peak);
    }
    digest.finish()
}

fn service_route_set_identity(routes: &[ServiceCapacityRouteRecord]) -> [u8; 32] {
    let mut digest = ServiceCapacityDigest::new();
    digest.write(SERVICE_CAPACITY_LEAF_SCHEMA);
    digest.write(b"route-set");
    digest.usize(routes.len());
    for route in routes {
        digest.write(&route.source_identity);
    }
    digest.finish()
}

fn service_leaf_identity(
    build: [u8; 32],
    artifact: [u8; 32],
    route_set: [u8; 32],
    owner_generation: u64,
    route_type_closure: [u8; 32],
    routes: &[ServiceCapacityRouteRecord],
) -> [u8; 32] {
    let mut digest = ServiceCapacityDigest::new();
    digest.write(SERVICE_CAPACITY_LEAF_SCHEMA);
    digest.write(b"leaf");
    digest.write(&build);
    digest.write(&artifact);
    digest.write(&route_set);
    digest.write(&owner_generation.to_le_bytes());
    digest.write(&route_type_closure);
    digest.usize(routes.len());
    for route in routes {
        digest.write(&route.source_identity);
    }
    digest.finish()
}

/// The only assembly that emits a production Service capacity leaf.
#[doc(hidden)]
pub struct CompiledExecutionWithCapacityLeaf<E, C, F, const BODY: usize, const OUTPUT: usize>
where
    E: Fn(usize, C, ManagedBytes, CompiledDbPermit, ManagedResponseBuilder) -> F
        + Send
        + Sync
        + 'static,
    C: Send + Unpin + 'static,
    F: Future<Output = Result<ManagedResponse, ExecutionError>> + Send + 'static,
{
    pub(super) execution: CompiledExecutionCapability<E, C, F, BODY, OUTPUT>,
    pub(super) service_leaf: FrozenServiceCapacitySourceLeaf,
}

impl<E, C, F, const BODY: usize, const OUTPUT: usize>
    CompiledExecutionWithCapacityLeaf<E, C, F, BODY, OUTPUT>
where
    E: Fn(usize, C, ManagedBytes, CompiledDbPermit, ManagedResponseBuilder) -> F
        + Send
        + Sync
        + 'static,
    C: Send + Unpin + 'static,
    F: Future<Output = Result<ManagedResponse, ExecutionError>> + Send + 'static,
{
    /// Borrows the raw capacity facts without separating either owned half.
    pub fn capacity_source(&self) -> FrozenServiceCapacitySourceView<'_> {
        FrozenServiceCapacitySourceView {
            leaf: &self.service_leaf,
        }
    }

    /// Produces deterministic Service-owned machine facts from this frozen
    /// whole. The result is an observation only and carries no authority.
    #[doc(hidden)]
    pub fn production_fact_input(
        &self,
    ) -> Result<
        crate::production_fact::ServiceProductionFactInput,
        crate::production_fact::ServiceProductionFactError,
    > {
        crate::production_fact::service_production_fact_input(self)
    }

    /// Resolves through the registry frozen in this same bundle.
    pub fn lookup(
        &self,
        route_token: &[u8],
        framing: EntryFraming,
        declared_length: Option<u64>,
    ) -> Result<RegisteredRouteExecutionProof, RegistryError> {
        self.execution.lookup(route_token, framing, declared_length)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ExecutionError {
    Registry(RegistryError),
    CompiledRegistry(CompiledRegistryError),
    UnknownExecution,
    InputTooLarge,
    InvalidInput,
    OutputTooLarge,
    RouteResourcesMismatch,
    ResponseReservationFailed,
    Business,
    DependencyUnavailable,
    /// A dependency accepted this request but failed while executing,
    /// decoding its result or finalizing the operation.
    DependencyExecutionFailed,
    MissingCapacityIdentity,
}

impl From<RegistryError> for ExecutionError {
    fn from(error: RegistryError) -> Self {
        Self::Registry(error)
    }
}

impl From<CompiledRegistryError> for ExecutionError {
    fn from(error: CompiledRegistryError) -> Self {
        Self::CompiledRegistry(error)
    }
}

impl ExecutionError {
    /// Exhaustive protocol classification for every generated execution
    /// failure. Adding an execution error cannot silently acquire a response
    /// meaning because this match has no catch-all arm.
    pub const fn response_outcome_class(
        self,
    ) -> saddle_runtime::compiled_route::ResponseOutcomeClass {
        use saddle_runtime::compiled_route::ResponseOutcomeClass;

        match self {
            Self::InputTooLarge | Self::InvalidInput => ResponseOutcomeClass::InvalidRequest,
            Self::Business => ResponseOutcomeClass::BusinessRejected,
            Self::RouteResourcesMismatch | Self::DependencyUnavailable => {
                ResponseOutcomeClass::Unavailable
            }
            Self::Registry(_)
            | Self::CompiledRegistry(_)
            | Self::UnknownExecution
            | Self::OutputTooLarge
            | Self::ResponseReservationFailed
            | Self::DependencyExecutionFailed => ResponseOutcomeClass::Internal,
            Self::MissingCapacityIdentity => ResponseOutcomeClass::Internal,
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CompiledRegistryError {
    InvalidCapacity,
    AllocationFailed,
    CapacityExceeded,
    EmptyRegistry,
    DuplicateContract,
    DuplicateIdentity,
}

struct CompiledServiceEntry {
    contract: TypeId,
    _descriptor: ServiceDescriptor,
}

/// Startup-only marker registry for the closed 0.2 execution graph.
///
/// It contains no implementation, handler, factory, observer, client or
/// dispatcher capability and deliberately does not implement Clone.
pub struct CompiledServiceRegistryBuilder {
    entries: Vec<CompiledServiceEntry>,
    max_services: usize,
}

impl CompiledServiceRegistryBuilder {
    pub fn new(max_services: usize) -> Result<Self, CompiledRegistryError> {
        if max_services == 0 || max_services > u32::MAX as usize {
            return Err(CompiledRegistryError::InvalidCapacity);
        }
        let mut entries = Vec::new();
        entries
            .try_reserve_exact(max_services)
            .map_err(|_| CompiledRegistryError::AllocationFailed)?;
        Ok(Self {
            entries,
            max_services,
        })
    }

    pub fn register<S>(
        &mut self,
        descriptor: ServiceDescriptor,
    ) -> Result<(), CompiledRegistryError>
    where
        S: Service,
    {
        if self.entries.len() == self.max_services {
            return Err(CompiledRegistryError::CapacityExceeded);
        }
        let contract = TypeId::of::<S>();
        if self.entries.iter().any(|entry| entry.contract == contract) {
            return Err(CompiledRegistryError::DuplicateContract);
        }
        if self
            .entries
            .iter()
            .any(|entry| entry._descriptor == descriptor)
        {
            return Err(CompiledRegistryError::DuplicateIdentity);
        }
        self.entries.push(CompiledServiceEntry {
            contract,
            _descriptor: descriptor,
        });
        Ok(())
    }

    pub fn freeze(self) -> Result<CompiledServiceRegistry, CompiledRegistryError> {
        if self.entries.is_empty() {
            return Err(CompiledRegistryError::EmptyRegistry);
        }
        Ok(CompiledServiceRegistry {
            entries: self.entries,
        })
    }
}

/// Linear compiled-only identity owner. No public API exposes its ordinal.
pub struct CompiledServiceRegistry {
    entries: Vec<CompiledServiceEntry>,
}

impl CompiledServiceRegistry {
    pub(super) fn internal_entry_count(&self) -> usize {
        self.entries.len()
    }

    pub(super) fn internal_entry_index<S>(&self) -> Option<usize>
    where
        S: Service,
    {
        self.entries
            .iter()
            .position(|entry| entry.contract == TypeId::of::<S>())
    }
}

mod sealed {
    pub trait ManagedValue {}
    pub trait ManagedCodec {}
}

/// Only Saddle-owned fixed-layout values can enter the generated 0.2 graph.
pub trait ManagedValue: sealed::ManagedValue + Send + 'static {}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct ManagedU64(pub u64);

impl sealed::ManagedValue for ManagedU64 {}
impl ManagedValue for ManagedU64 {}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct ManagedBool(pub bool);

impl sealed::ManagedValue for ManagedBool {}
impl ManagedValue for ManagedBool {}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct ManagedPair<A: ManagedValue, B: ManagedValue>(pub A, pub B);

impl<A: ManagedValue, B: ManagedValue> sealed::ManagedValue for ManagedPair<A, B> {}
impl<A: ManagedValue, B: ManagedValue> ManagedValue for ManagedPair<A, B> {}

/// A codec contribution is sealed and tied to one concrete request/response
/// pair. The generator chooses it; business code cannot implement it.
pub trait ManagedCodec<Request: ManagedValue, Response: ManagedValue>:
    sealed::ManagedCodec + Send + 'static
{
    const WORKSPACE_BYTES: usize;
}

pub struct FixedManagedCodec<Request: ManagedValue, Response: ManagedValue>(
    PhantomData<(Request, Response)>,
);

impl<Request: ManagedValue, Response: ManagedValue> sealed::ManagedCodec
    for FixedManagedCodec<Request, Response>
{
}

impl<Request: ManagedValue, Response: ManagedValue> ManagedCodec<Request, Response>
    for FixedManagedCodec<Request, Response>
{
    const WORKSPACE_BYTES: usize = mem::size_of::<Request>() + mem::size_of::<Response>();
}

/// Saddle-owned, inline byte storage. It never allocates or grows.
#[derive(Debug)]
pub struct FixedBytes<const N: usize> {
    bytes: [u8; N],
    length: usize,
}

impl<const N: usize> FixedBytes<N> {
    pub const fn empty() -> Self {
        Self {
            bytes: [0; N],
            length: 0,
        }
    }

    pub fn try_from_slice(value: &[u8]) -> Result<Self, ExecutionError> {
        if value.len() > N {
            return Err(ExecutionError::InputTooLarge);
        }
        let mut output = Self::empty();
        output.bytes[..value.len()].copy_from_slice(value);
        output.length = value.len();
        Ok(output)
    }

    pub const fn capacity(&self) -> usize {
        N
    }

    pub const fn len(&self) -> usize {
        self.length
    }

    pub const fn is_empty(&self) -> bool {
        self.length == 0
    }

    pub fn as_slice(&self) -> &[u8] {
        &self.bytes[..self.length]
    }

    pub fn try_append(&mut self, value: &[u8]) -> Result<(), ExecutionError> {
        let end = self
            .length
            .checked_add(value.len())
            .ok_or(ExecutionError::OutputTooLarge)?;
        if end > N {
            return Err(ExecutionError::OutputTooLarge);
        }
        self.bytes[self.length..end].copy_from_slice(value);
        self.length = end;
        Ok(())
    }
}

/// Layout proof can only be created from the concrete types named by generated
/// dispatcher code. No numeric constructor is exposed.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ExecutionLayoutProof {
    service: TypeId,
    binding: TypeId,
    body_bytes: usize,
    decode_workspace_bytes: usize,
    request_bytes: usize,
    future_bytes: usize,
    response_bytes: usize,
    encode_workspace_bytes: usize,
    output_bytes: usize,
}

impl ExecutionLayoutProof {
    pub fn bind<
        S,
        const BODY: usize,
        const OUTPUT: usize,
        Request,
        Response,
        Codec,
        HandlerFuture,
    >() -> Result<Self, ExecutionError>
    where
        S: Service<Request = Request, Response = Response>,
        Request: ManagedValue,
        Response: ManagedValue,
        Codec: ManagedCodec<Request, Response>,
        HandlerFuture: Future<Output = Result<Response, ExecutionError>> + Send + 'static,
    {
        let proof = Self {
            service: TypeId::of::<S>(),
            binding: TypeId::of::<(S, Request, Response, Codec, HandlerFuture)>(),
            body_bytes: BODY,
            decode_workspace_bytes: Codec::WORKSPACE_BYTES,
            request_bytes: mem::size_of::<Request>(),
            future_bytes: mem::size_of::<HandlerFuture>(),
            response_bytes: mem::size_of::<Response>(),
            encode_workspace_bytes: Codec::WORKSPACE_BYTES,
            output_bytes: OUTPUT,
        };
        proof.commitment()?;
        Ok(proof)
    }

    pub const fn body_bytes(self) -> usize {
        self.body_bytes
    }

    pub const fn future_bytes(self) -> usize {
        self.future_bytes
    }

    pub fn same_concrete_binding(self, other: Self) -> bool {
        self.binding == other.binding
    }

    pub fn commitment(self) -> Result<usize, ExecutionError> {
        [
            self.body_bytes,
            self.decode_workspace_bytes,
            self.request_bytes,
            self.future_bytes,
            self.response_bytes,
            self.encode_workspace_bytes,
            self.output_bytes,
        ]
        .into_iter()
        .try_fold(0_usize, |total, bytes| {
            total
                .checked_add(bytes)
                .ok_or(ExecutionError::OutputTooLarge)
        })
    }

    fn plan(self) -> Result<EntryPlanSpec, ExecutionError> {
        let body = u64::try_from(self.body_bytes).map_err(|_| ExecutionError::OutputTooLarge)?;
        let decode = self
            .decode_workspace_bytes
            .checked_add(self.request_bytes)
            .ok_or(ExecutionError::OutputTooLarge)?;
        let business = self.future_bytes;
        let response = self.response_bytes;
        let encode = self
            .encode_workspace_bytes
            .checked_add(self.output_bytes)
            .ok_or(ExecutionError::OutputTooLarge)?;
        Ok(EntryPlanSpec::new()
            .body_hard_limit(body)
            .managed_body_peak(body)
            .decode_peak(u64::try_from(decode).map_err(|_| ExecutionError::OutputTooLarge)?)
            .business_working_set_peak(
                u64::try_from(business).map_err(|_| ExecutionError::OutputTooLarge)?,
            )
            .internal_call_peak(0)
            .response_object_peak(
                u64::try_from(response).map_err(|_| ExecutionError::OutputTooLarge)?,
            )
            .encode_destination_peak(
                u64::try_from(encode).map_err(|_| ExecutionError::OutputTooLarge)?,
            )
            .copy_on_success_peak(0))
    }

    /// Explicitly closes a compiled route that has no DB edge.
    pub const fn without_db(self) -> CompiledRouteExecutionProof {
        CompiledRouteExecutionProof {
            layout: self,
            db_connections: 0,
            db_operations: 0,
            db_binding: None,
            db_binding_conflict: false,
        }
    }

    /// Derives route demand from a concrete, validated DB operation proof.
    pub fn with_query_optional<O>(
        self,
        operation: QueryOptionalOperationProof<O>,
    ) -> CompiledRouteExecutionProof
    where
        O: StaticQueryOptionalOperation,
    {
        let credits = operation.layout().credits();
        let binding = CompiledDbBinding {
            kind: CompiledDbOperationKind::QueryOptional,
            operation: TypeId::of::<O>(),
            parameter_bytes: operation.layout().parameter_bytes(),
            result_bytes: operation.layout().optional_row_bytes(),
        };
        CompiledRouteExecutionProof {
            layout: self,
            db_connections: u64::from(credits.connections()),
            db_operations: u64::from(credits.operations()),
            db_binding: Some(binding),
            db_binding_conflict: false,
        }
    }

    /// Derives a standalone-write route from one concrete generated proof.
    pub fn with_write<O>(self, operation: WriteOperationProof<O>) -> CompiledRouteExecutionProof
    where
        O: StaticWriteOperation,
    {
        let layout = operation.layout();
        let credits = layout.credits();
        CompiledRouteExecutionProof {
            layout: self,
            db_connections: u64::from(credits.connections()),
            db_operations: u64::from(credits.operations()),
            db_binding: Some(CompiledDbBinding {
                kind: CompiledDbOperationKind::Write,
                operation: TypeId::of::<O>(),
                parameter_bytes: layout.parameter_bytes(),
                result_bytes: mem::size_of::<ManagedWriteResult>(),
            }),
            db_binding_conflict: false,
        }
    }

    /// Derives one single-level transaction containing one generated write.
    pub fn with_transaction<O>(
        self,
        operation: TransactionOperationProof<O>,
    ) -> CompiledRouteExecutionProof
    where
        O: StaticWriteOperation,
    {
        let layout = operation.layout();
        let credits = layout.credits();
        CompiledRouteExecutionProof {
            layout: self,
            db_connections: u64::from(credits.connections()),
            db_operations: u64::from(credits.operations()),
            db_binding: Some(CompiledDbBinding {
                kind: CompiledDbOperationKind::Transaction,
                operation: TypeId::of::<O>(),
                parameter_bytes: layout.parameter_bytes(),
                result_bytes: mem::size_of::<ManagedWriteResult>(),
            }),
            db_binding_conflict: false,
        }
    }
}

/// Generator-owned closure of execution layout and all DB route edges.
///
/// It has no numeric constructor. Additional operations merge by route
/// maximum from their validated DB proofs.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct CompiledRouteExecutionProof {
    layout: ExecutionLayoutProof,
    db_connections: u64,
    db_operations: u64,
    db_binding: Option<CompiledDbBinding>,
    db_binding_conflict: bool,
}

impl CompiledRouteExecutionProof {
    fn capacity_source_identity(self) -> [u8; 32] {
        let mut digest = ServiceCapacityDigest::new();
        digest.write(SERVICE_CAPACITY_LEAF_SCHEMA);
        digest.write(b"compiled-route-layout");
        for value in [
            self.layout.body_bytes,
            self.layout.decode_workspace_bytes,
            self.layout.request_bytes,
            self.layout.future_bytes,
            self.layout.response_bytes,
            self.layout.encode_workspace_bytes,
            self.layout.output_bytes,
        ] {
            digest.usize(value);
        }
        digest.write(&self.db_connections.to_le_bytes());
        digest.write(&self.db_operations.to_le_bytes());
        match self.db_binding {
            None => digest.write(b"no-db"),
            Some(binding) => {
                digest.write(&[binding.kind as u8]);
                digest.usize(binding.parameter_bytes);
                digest.usize(binding.result_bytes);
            }
        }
        digest.finish()
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[repr(u8)]
enum CompiledDbOperationKind {
    QueryOptional,
    Write,
    Transaction,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct CompiledDbBinding {
    kind: CompiledDbOperationKind,
    operation: TypeId,
    parameter_bytes: usize,
    result_bytes: usize,
}

/// Linear DB resource passed only after matching the admitted permit against
/// the frozen route proof.
pub struct CompiledDbPermit {
    inner: Option<DbRequestPermit>,
    route_demand: RegisteredDbDemand,
    binding: Option<CompiledDbBinding>,
}

impl CompiledDbPermit {
    pub const fn is_none(&self) -> bool {
        self.inner.is_none()
    }

    pub fn demand(&self) -> Option<(usize, usize)> {
        self.inner.as_ref().map(|permit| {
            let demand = permit.demand();
            (demand.connections(), demand.operations())
        })
    }

    /// Consumes the one admitted route permit directly into Database's opaque
    /// typed execution constructor.
    ///
    /// The invocation indivisibly carries the layout produced by its sealed
    /// query proof, while `route_demand` came from the registry proof already
    /// authenticated by the same frozen execution capability. No raw permit
    /// exists at the generated boundary.
    pub fn handoff_query_optional<O>(
        self,
        invocation: QueryOptionalInvocation<O>,
    ) -> Result<QueryOptionalExecution<O>, ExecutionError>
    where
        O: StaticQueryOptionalOperation,
    {
        let layout = invocation.layout();
        let permit = self.into_permit(
            CompiledDbBinding {
                kind: CompiledDbOperationKind::QueryOptional,
                operation: TypeId::of::<O>(),
                parameter_bytes: layout.parameter_bytes(),
                result_bytes: layout.optional_row_bytes(),
            },
            layout.credits().connections(),
            layout.credits().operations(),
        )?;
        Ok(QueryOptionalExecution::from_compiled_handoff(
            permit, invocation,
        ))
    }

    /// Consumes the admitted route permit into one standalone Database write.
    pub fn handoff_write<O>(
        self,
        invocation: WriteInvocation<O>,
    ) -> Result<WriteExecution<O>, ExecutionError>
    where
        O: StaticWriteOperation,
    {
        let layout = invocation.layout();
        let permit = self.into_permit(
            CompiledDbBinding {
                kind: CompiledDbOperationKind::Write,
                operation: TypeId::of::<O>(),
                parameter_bytes: layout.parameter_bytes(),
                result_bytes: mem::size_of::<ManagedWriteResult>(),
            },
            layout.credits().connections(),
            layout.credits().operations(),
        )?;
        Ok(WriteExecution::from_compiled_handoff(permit, invocation))
    }

    /// Consumes the admitted route permit into one single-level transaction.
    pub fn handoff_transaction<O>(
        self,
        invocation: TransactionInvocation<O>,
    ) -> Result<TransactionExecution<O>, ExecutionError>
    where
        O: StaticWriteOperation,
    {
        let layout = invocation.layout();
        let permit = self.into_permit(
            CompiledDbBinding {
                kind: CompiledDbOperationKind::Transaction,
                operation: TypeId::of::<O>(),
                parameter_bytes: layout.parameter_bytes(),
                result_bytes: mem::size_of::<ManagedWriteResult>(),
            },
            layout.credits().connections(),
            layout.credits().operations(),
        )?;
        Ok(TransactionExecution::from_compiled_handoff(
            permit, invocation,
        ))
    }

    fn into_permit(
        self,
        binding: CompiledDbBinding,
        connections: u32,
        operations: u32,
    ) -> Result<DbRequestPermit, ExecutionError> {
        let permit = self.inner.ok_or(ExecutionError::RouteResourcesMismatch)?;
        if self.binding != Some(binding)
            || self.route_demand.connections() != u64::from(connections)
            || self.route_demand.operations() != u64::from(operations)
        {
            permit
                .begin_finalizing()
                .and_then(|claim| claim.complete_after_connection_return())
                .map_err(|_| ExecutionError::DependencyUnavailable)?;
            return Err(ExecutionError::RouteResourcesMismatch);
        }
        Ok(permit)
    }

    #[cfg(test)]
    fn complete_after_test_connection_return(mut self) {
        self.inner
            .take()
            .expect("DB route must carry its admitted permit")
            .begin_finalizing()
            .expect("test connection return must enter finalizing")
            .complete_after_connection_return()
            .expect("test connection return must complete its role");
    }
}

impl CompiledRouteExecutionProof {
    pub fn with_query_optional<O>(mut self, operation: QueryOptionalOperationProof<O>) -> Self
    where
        O: StaticQueryOptionalOperation,
    {
        let layout = operation.layout();
        let credits = layout.credits();
        self.merge_db_binding(CompiledDbBinding {
            kind: CompiledDbOperationKind::QueryOptional,
            operation: TypeId::of::<O>(),
            parameter_bytes: layout.parameter_bytes(),
            result_bytes: layout.optional_row_bytes(),
        });
        self.db_connections = self.db_connections.max(u64::from(credits.connections()));
        self.db_operations = self.db_operations.max(u64::from(credits.operations()));
        self
    }

    pub fn with_write<O>(mut self, operation: WriteOperationProof<O>) -> Self
    where
        O: StaticWriteOperation,
    {
        let layout = operation.layout();
        let credits = layout.credits();
        self.merge_db_binding(CompiledDbBinding {
            kind: CompiledDbOperationKind::Write,
            operation: TypeId::of::<O>(),
            parameter_bytes: layout.parameter_bytes(),
            result_bytes: mem::size_of::<ManagedWriteResult>(),
        });
        self.db_connections = self.db_connections.max(u64::from(credits.connections()));
        self.db_operations = self.db_operations.max(u64::from(credits.operations()));
        self
    }

    pub fn with_transaction<O>(mut self, operation: TransactionOperationProof<O>) -> Self
    where
        O: StaticWriteOperation,
    {
        let layout = operation.layout();
        let credits = layout.credits();
        self.merge_db_binding(CompiledDbBinding {
            kind: CompiledDbOperationKind::Transaction,
            operation: TypeId::of::<O>(),
            parameter_bytes: layout.parameter_bytes(),
            result_bytes: mem::size_of::<ManagedWriteResult>(),
        });
        self.db_connections = self.db_connections.max(u64::from(credits.connections()));
        self.db_operations = self.db_operations.max(u64::from(credits.operations()));
        self
    }

    fn merge_db_binding(&mut self, binding: CompiledDbBinding) {
        if self
            .db_binding
            .is_some_and(|registered| registered != binding)
        {
            self.db_binding_conflict = true;
        } else {
            self.db_binding = Some(binding);
        }
    }
}

/// Generated only for a declared static `Caller -> Callee` edge.
pub struct InternalCallCapability<Caller: Service, Callee: Service> {
    _types: PhantomData<fn(Caller) -> Callee>,
}

/// Linear startup assembly. The same consumed registry produces lookup records
/// and selects generated execution slots.
pub struct CompiledExecutionBuilder<E, L, C, F, const BODY: usize, const OUTPUT: usize>
where
    E: Fn(usize, C, ManagedBytes, CompiledDbPermit, ManagedResponseBuilder) -> F
        + Send
        + Sync
        + 'static,
    L: Fn(usize) -> Option<CompiledRouteExecutionProof> + Send + Sync + 'static,
    C: Send + Unpin + 'static,
    F: Future<Output = Result<ManagedResponse, ExecutionError>> + Send + 'static,
{
    entries: ServiceEntryCapabilityBuilder,
    dispatcher: E,
    layouts: L,
    db_bindings: Vec<Option<CompiledDbBinding>>,
    capacity_routes: Vec<ServiceCapacityRouteRecord>,
    _types: PhantomData<fn(C) -> F>,
}

impl<E, L, C, F, const BODY: usize, const OUTPUT: usize>
    CompiledExecutionBuilder<E, L, C, F, BODY, OUTPUT>
where
    E: Fn(usize, C, ManagedBytes, CompiledDbPermit, ManagedResponseBuilder) -> F
        + Send
        + Sync
        + 'static,
    L: Fn(usize) -> Option<CompiledRouteExecutionProof> + Send + Sync + 'static,
    C: Send + Unpin + 'static,
    F: Future<Output = Result<ManagedResponse, ExecutionError>> + Send + 'static,
{
    pub fn new(
        registry: CompiledServiceRegistry,
        dispatcher: E,
        layouts: L,
        limits: EntryCapabilityLimits,
    ) -> Result<Self, ExecutionError> {
        let entry_count = registry.internal_entry_count();
        let mut db_bindings = Vec::new();
        db_bindings
            .try_reserve_exact(entry_count)
            .map_err(|_| CompiledRegistryError::AllocationFailed)?;
        db_bindings.resize(entry_count, None);
        let mut capacity_routes = Vec::new();
        capacity_routes
            .try_reserve_exact(limits.max_routes())
            .map_err(|_| CompiledRegistryError::AllocationFailed)?;
        Ok(Self {
            entries: ServiceEntryCapabilityBuilder::new_compiled(registry, limits)?,
            dispatcher,
            layouts,
            db_bindings,
            capacity_routes,
            _types: PhantomData,
        })
    }

    /// The plan is derived from the generated dispatcher's concrete types.
    /// Callers cannot supply resource numbers.
    pub fn expose<S>(
        &mut self,
        route_token: &[u8],
        framing: EntryFraming,
    ) -> Result<(), ExecutionError>
    where
        S: Service,
    {
        let slot = self.entries.service_index::<S>()?;
        let proof = (self.layouts)(slot).ok_or(ExecutionError::UnknownExecution)?;
        if proof.layout.service != TypeId::of::<S>()
            || proof.layout.body_bytes != BODY
            || proof.layout.output_bytes != OUTPUT
            || proof.db_binding_conflict
            || (proof.db_binding.is_none()
                != (proof.db_connections == 0 && proof.db_operations == 0))
        {
            return Err(ExecutionError::UnknownExecution);
        }
        if self.db_bindings[slot].is_some() && self.db_bindings[slot] != proof.db_binding {
            return Err(ExecutionError::UnknownExecution);
        }
        self.db_bindings[slot] = proof.db_binding;
        let plan = self.entries.register_compiled_plan(
            proof.layout.plan()?,
            proof.db_connections,
            proof.db_operations,
            u64::try_from(proof.layout.output_bytes).map_err(|_| ExecutionError::OutputTooLarge)?,
        )?;
        let identity = self
            .entries
            .expose_compiled::<S>(route_token, framing, plan)?;
        let commitment = proof.layout.commitment()?;
        let managed_objects_peak = compiled_route_managed_objects_peak()?;
        let db_connections =
            usize::try_from(proof.db_connections).map_err(|_| ExecutionError::OutputTooLarge)?;
        let db_operations =
            usize::try_from(proof.db_operations).map_err(|_| ExecutionError::OutputTooLarge)?;
        self.capacity_routes.push(ServiceCapacityRouteRecord {
            identity: identity.contract().opaque(),
            commitment,
            managed_objects_peak,
            db_connections,
            db_operations,
            source_identity: service_route_source_identity(ServiceRouteSourceFacts {
                route_token,
                framing,
                identity,
                layout_source_identity: proof.capacity_source_identity(),
                commitment,
                managed_objects_peak,
                db_connections,
                db_operations,
            }),
        });
        Ok(())
    }

    pub fn bind_internal_call<Caller, Callee>(
        &self,
    ) -> Result<InternalCallCapability<Caller, Callee>, ExecutionError>
    where
        Caller: Service,
        Callee: Service,
    {
        self.entries.service_index::<Caller>()?;
        self.entries.service_index::<Callee>()?;
        Ok(InternalCallCapability {
            _types: PhantomData,
        })
    }

    pub fn freeze(
        self,
    ) -> Result<CompiledExecutionCapability<E, C, F, BODY, OUTPUT>, ExecutionError> {
        Ok(CompiledExecutionCapability {
            entries: self.entries.freeze()?,
            dispatcher: self.dispatcher,
            db_bindings: self.db_bindings,
            _types: PhantomData,
        })
    }

    /// Freezes execution and the sole Service source leaf from one linear
    /// registry assembly. Artifact identities are observations from the
    /// generated/Gate owner; all route facts remain Service-derived.
    pub fn freeze_with_capacity_leaf<A>(
        self,
        artifact: A,
    ) -> Result<CompiledExecutionWithCapacityLeaf<E, C, F, BODY, OUTPUT>, ExecutionError>
    where
        A: GeneratedServiceCapacityArtifactOwner,
    {
        let build = artifact.build_identity();
        let artifact_identity = artifact.artifact_identity();
        let owner_generation = artifact.owner_generation();
        if build == [0; 32] || artifact_identity == [0; 32] || owner_generation == 0 {
            return Err(ExecutionError::MissingCapacityIdentity);
        }
        if self.capacity_routes.is_empty() {
            return Err(ExecutionError::UnknownExecution);
        }
        let route_set_identity = service_route_set_identity(&self.capacity_routes);
        let route_type_closure_identity =
            service_route_type_closure_identity(artifact_identity, &self.capacity_routes);
        let leaf_identity = service_leaf_identity(
            build,
            artifact_identity,
            route_set_identity,
            owner_generation,
            route_type_closure_identity,
            &self.capacity_routes,
        );
        let service_leaf = FrozenServiceCapacitySourceLeaf {
            leaf_identity,
            common_identities: [build, artifact_identity, route_set_identity],
            owner_generation,
            route_type_closure_identity,
            routes: self.capacity_routes.into_boxed_slice(),
        };
        let execution = CompiledExecutionCapability {
            entries: self.entries.freeze()?,
            dispatcher: self.dispatcher,
            db_bindings: self.db_bindings,
            _types: PhantomData,
        };
        Ok(CompiledExecutionWithCapacityLeaf {
            execution,
            service_leaf,
        })
    }
}

/// Non-Clone owner of both immutable lookup and the sole generated dispatcher.
pub struct CompiledExecutionCapability<E, C, F, const BODY: usize, const OUTPUT: usize>
where
    E: Fn(usize, C, ManagedBytes, CompiledDbPermit, ManagedResponseBuilder) -> F
        + Send
        + Sync
        + 'static,
    C: Send + Unpin + 'static,
    F: Future<Output = Result<ManagedResponse, ExecutionError>> + Send + 'static,
{
    pub(super) entries: ServiceEntryCapability,
    dispatcher: E,
    db_bindings: Vec<Option<CompiledDbBinding>>,
    _types: PhantomData<fn(C) -> F>,
}

impl<E, C, F, const BODY: usize, const OUTPUT: usize>
    CompiledExecutionCapability<E, C, F, BODY, OUTPUT>
where
    E: Fn(usize, C, ManagedBytes, CompiledDbPermit, ManagedResponseBuilder) -> F
        + Send
        + Sync
        + 'static,
    C: Send + Unpin + 'static,
    F: Future<Output = Result<ManagedResponse, ExecutionError>> + Send + 'static,
{
    pub fn lookup(
        &self,
        route_token: &[u8],
        framing: EntryFraming,
        declared_length: Option<u64>,
    ) -> Result<RegisteredRouteExecutionProof, RegistryError> {
        self.entries
            .lookup(route_token, framing, declared_length)
            .map(RegisteredEntry::execution_proof)
    }

    fn execute(
        &self,
        proof: RegisteredRouteExecutionProof,
        context: C,
        body: ManagedBytes,
        db_permit: Option<DbRequestPermit>,
        memory: &RequestMemory,
    ) -> Result<F, ExecutionError> {
        let registered = RegisteredEntry::from_execution_proof(proof);
        let slot = self.entries.execution_slot(registered)?;
        if body.len() > proof.plan().body_hard_limit() as usize {
            return Err(ExecutionError::InputTooLarge);
        }
        let expected = proof.db_demand();
        let actual = db_permit.as_ref().map(DbRequestPermit::demand);
        let resources_match = match actual {
            None => expected.is_none(),
            Some(demand) => {
                u64::try_from(demand.connections()) == Ok(expected.connections())
                    && u64::try_from(demand.operations()) == Ok(expected.operations())
                    && !expected.is_none()
            }
        };
        if !resources_match {
            return Err(ExecutionError::RouteResourcesMismatch);
        }
        let response_capacity = usize::try_from(proof.response_capacity())
            .map_err(|_| ExecutionError::OutputTooLarge)?;
        if response_capacity != OUTPUT {
            return Err(ExecutionError::UnknownExecution);
        }
        let response = memory
            .try_response_builder(response_capacity)
            .map_err(|_| ExecutionError::ResponseReservationFailed)?;
        Ok((self.dispatcher)(
            slot,
            context,
            body,
            CompiledDbPermit {
                inner: db_permit,
                route_demand: expected,
                binding: self.db_bindings[slot],
            },
            response,
        ))
    }
}

pin_project_lite::pin_project! {
    /// Fixed-shape classified successor around the generated concrete Future.
    ///
    /// The current-account error payload is reserved before the business
    /// Future is created and is carried across await without retaining
    /// `RequestMemory`.
    pub struct ClassifiedExecutionFuture<F> {
        #[pin]
        state: ClassifiedExecutionState<F>,
    }
}

pin_project_lite::pin_project! {
    #[project = ClassifiedExecutionStateProj]
    enum ClassifiedExecutionState<F> {
        Running {
            #[pin]
            future: F,
            error_payload: Option<ManagedResponse>,
        },
        Ready {
            outcome: Option<saddle_runtime::compiled_route::CompiledResponseOutcome>,
        },
    }
}

impl<F> ClassifiedExecutionFuture<F> {
    fn running(future: F, error_payload: ManagedResponse) -> Self {
        Self {
            state: ClassifiedExecutionState::Running {
                future,
                error_payload: Some(error_payload),
            },
        }
    }

    fn ready(outcome: saddle_runtime::compiled_route::CompiledResponseOutcome) -> Self {
        Self {
            state: ClassifiedExecutionState::Ready {
                outcome: Some(outcome),
            },
        }
    }
}

fn classified_outcome(
    class: saddle_runtime::compiled_route::ResponseOutcomeClass,
    payload: ManagedResponse,
) -> saddle_runtime::compiled_route::CompiledResponseOutcome {
    use saddle_runtime::compiled_route::{CompiledResponseOutcome, ResponseOutcomeClass};

    match class {
        ResponseOutcomeClass::Success => CompiledResponseOutcome::success(payload),
        ResponseOutcomeClass::InvalidRequest => CompiledResponseOutcome::invalid_request(payload),
        ResponseOutcomeClass::BusinessRejected => {
            CompiledResponseOutcome::business_rejected(payload)
        }
        ResponseOutcomeClass::Unavailable => CompiledResponseOutcome::unavailable(payload),
        ResponseOutcomeClass::Internal => CompiledResponseOutcome::internal(payload),
    }
}

impl<F> Future for ClassifiedExecutionFuture<F>
where
    F: Future<Output = Result<ManagedResponse, ExecutionError>>,
{
    type Output = saddle_runtime::compiled_route::CompiledResponseOutcome;

    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
        match self.project().state.project() {
            ClassifiedExecutionStateProj::Running {
                future,
                error_payload,
            } => match future.poll(context) {
                Poll::Pending => Poll::Pending,
                Poll::Ready(Ok(payload)) => Poll::Ready(
                    saddle_runtime::compiled_route::CompiledResponseOutcome::success(payload),
                ),
                Poll::Ready(Err(error)) => {
                    let payload = error_payload
                        .take()
                        .expect("classified execution future completes only once");
                    Poll::Ready(classified_outcome(error.response_outcome_class(), payload))
                }
            },
            ClassifiedExecutionStateProj::Ready { outcome } => Poll::Ready(
                outcome
                    .take()
                    .expect("classified execution future completes only once"),
            ),
        }
    }
}

#[cfg(all(target_arch = "x86_64", target_os = "linux"))]
impl<E, C, F, const BODY: usize, const OUTPUT: usize>
    saddle_runtime::compiled_route::CompiledRouteAdapter
    for CompiledExecutionCapability<E, C, F, BODY, OUTPUT>
where
    E: Fn(usize, C, ManagedBytes, CompiledDbPermit, ManagedResponseBuilder) -> F
        + Send
        + Sync
        + 'static,
    C: Send + Unpin + 'static,
    F: Future<Output = Result<ManagedResponse, ExecutionError>> + Send + 'static,
{
    type Proof = RegisteredRouteExecutionProof;
    type Context = C;
    type Error = ExecutionError;
    type Future = F;

    fn managed_commitment(&self, proof: Self::Proof) -> Result<usize, Self::Error> {
        self.entries
            .execution_slot(RegisteredEntry::from_execution_proof(proof))?;
        usize::try_from(proof.plan().commitment()).map_err(|_| ExecutionError::OutputTooLarge)
    }

    fn response_capacity(&self, proof: Self::Proof) -> Result<usize, Self::Error> {
        self.entries
            .execution_slot(RegisteredEntry::from_execution_proof(proof))?;
        usize::try_from(proof.response_capacity()).map_err(|_| ExecutionError::OutputTooLarge)
    }

    fn db_resources<'a>(
        &self,
        proof: Self::Proof,
        domain: Option<&'a saddle_admission::DbPermitDomain>,
    ) -> Result<DbRouteResources<'a>, Self::Error> {
        self.entries
            .execution_slot(RegisteredEntry::from_execution_proof(proof))?;
        let demand = proof.db_demand();
        if demand.is_none() {
            return Ok(DbRouteResources::none());
        }
        let domain = domain.ok_or(ExecutionError::RouteResourcesMismatch)?;
        let connections =
            usize::try_from(demand.connections()).map_err(|_| ExecutionError::OutputTooLarge)?;
        let operations =
            usize::try_from(demand.operations()).map_err(|_| ExecutionError::OutputTooLarge)?;
        let demand = DbRouteCreditDemand::new(connections, operations)
            .map_err(|_| ExecutionError::RouteResourcesMismatch)?;
        Ok(DbRouteResources::required(domain, demand))
    }

    fn execute(
        &self,
        proof: Self::Proof,
        context: Self::Context,
        body: ManagedBytes,
        permit: Option<DbRequestPermit>,
        memory: &RequestMemory,
    ) -> Result<Self::Future, Self::Error> {
        CompiledExecutionCapability::execute(self, proof, context, body, permit, memory)
    }
}

#[cfg(all(target_arch = "x86_64", target_os = "linux"))]
impl<E, C, F, const BODY: usize, const OUTPUT: usize>
    saddle_runtime::compiled_route::ClassifiedCompiledRouteAdapter
    for CompiledExecutionCapability<E, C, F, BODY, OUTPUT>
where
    E: Fn(usize, C, ManagedBytes, CompiledDbPermit, ManagedResponseBuilder) -> F
        + Send
        + Sync
        + 'static,
    C: Send + Unpin + 'static,
    F: Future<Output = Result<ManagedResponse, ExecutionError>> + Send + 'static,
{
    type Proof = RegisteredRouteExecutionProof;
    type Context = C;
    type Error = ExecutionError;
    type Future = ClassifiedExecutionFuture<F>;

    fn managed_commitment(&self, proof: Self::Proof) -> Result<usize, Self::Error> {
        saddle_runtime::compiled_route::CompiledRouteAdapter::managed_commitment(self, proof)
    }

    fn response_capacity(&self, proof: Self::Proof) -> Result<usize, Self::Error> {
        saddle_runtime::compiled_route::CompiledRouteAdapter::response_capacity(self, proof)
    }

    fn db_resources<'a>(
        &self,
        proof: Self::Proof,
        domain: Option<&'a saddle_admission::DbPermitDomain>,
    ) -> Result<DbRouteResources<'a>, Self::Error> {
        saddle_runtime::compiled_route::CompiledRouteAdapter::db_resources(self, proof, domain)
    }

    fn execute(
        &self,
        proof: Self::Proof,
        context: Self::Context,
        body: ManagedBytes,
        permit: Option<DbRequestPermit>,
        memory: &RequestMemory,
    ) -> Self::Future {
        let error_payload = memory
            .try_response(&[])
            .expect("zero-byte classified response is inside the current account commitment");
        match CompiledExecutionCapability::execute(self, proof, context, body, permit, memory) {
            Ok(future) => ClassifiedExecutionFuture::running(future, error_payload),
            Err(error) => ClassifiedExecutionFuture::ready(classified_outcome(
                error.response_outcome_class(),
                error_payload,
            )),
        }
    }
}

/// The production adapter is the indivisible same-freeze bundle. Delegation
/// keeps the capacity leaf owned beside execution for the entire handoff.
#[cfg(all(target_arch = "x86_64", target_os = "linux"))]
impl<E, C, F, const BODY: usize, const OUTPUT: usize>
    saddle_runtime::compiled_route::ClassifiedCompiledRouteAdapter
    for CompiledExecutionWithCapacityLeaf<E, C, F, BODY, OUTPUT>
where
    E: Fn(usize, C, ManagedBytes, CompiledDbPermit, ManagedResponseBuilder) -> F
        + Send
        + Sync
        + 'static,
    C: Send + Unpin + 'static,
    F: Future<Output = Result<ManagedResponse, ExecutionError>> + Send + 'static,
{
    type Proof = RegisteredRouteExecutionProof;
    type Context = C;
    type Error = ExecutionError;
    type Future = ClassifiedExecutionFuture<F>;

    fn managed_commitment(&self, proof: Self::Proof) -> Result<usize, Self::Error> {
        saddle_runtime::compiled_route::ClassifiedCompiledRouteAdapter::managed_commitment(
            &self.execution,
            proof,
        )
    }

    fn response_capacity(&self, proof: Self::Proof) -> Result<usize, Self::Error> {
        saddle_runtime::compiled_route::ClassifiedCompiledRouteAdapter::response_capacity(
            &self.execution,
            proof,
        )
    }

    fn db_resources<'a>(
        &self,
        proof: Self::Proof,
        domain: Option<&'a saddle_admission::DbPermitDomain>,
    ) -> Result<DbRouteResources<'a>, Self::Error> {
        saddle_runtime::compiled_route::ClassifiedCompiledRouteAdapter::db_resources(
            &self.execution,
            proof,
            domain,
        )
    }

    fn execute(
        &self,
        proof: Self::Proof,
        context: Self::Context,
        body: ManagedBytes,
        permit: Option<DbRequestPermit>,
        memory: &RequestMemory,
    ) -> Self::Future {
        saddle_runtime::compiled_route::ClassifiedCompiledRouteAdapter::execute(
            &self.execution,
            proof,
            context,
            body,
            permit,
            memory,
        )
    }
}

#[cfg(test)]
mod tests {
    use std::{
        future::Future,
        pin::Pin,
        task::{Context, Poll},
    };

    use saddle_core::Result as SaddleResult;
    use saddle_db::internal::{
        DbU64, ManagedQueryParameters, ManagedQueryRow, QueryOptionalOperationProof,
        StaticQueryOptionalOperation, StaticWriteOperation, TransactionDecision,
        TransactionOperationProof, WriteOperationProof,
    };

    use super::*;
    use crate::{
        ServiceDescriptor,
        c4_entry::{EntryCapabilityLimits, EntryFraming},
    };
    use saddle_admission::{
        AdmissionError, DbCreditProfile, EntryIoAuditPlan, EntryReadPoll,
        OfficialTokioEntryIoAttemptOutcome, OfficialTokioRegistrationProfile,
        ProcessAllocationProfile, ProcessLedger, RequestMemory, ResourceConfig, ResponseWritePoll,
    };
    use saddle_runtime::compiled_route::{CompiledRouteAdapter, ResponseOutcomeClass};

    const BODY: usize = 32;
    const OUTPUT: usize = 64;
    static EXECUTION_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

    fn request_ledger() -> (ProcessLedger, ProcessAllocationProfile) {
        let state = ProcessLedger::minimum_process_state_reserve(1).unwrap();
        let config = ResourceConfig {
            managed_capacity: 512,
            entry_reserve: 64,
            framework_reserve: 4_096,
            task_reserve: 4_096,
            process_state_reserve: state,
            system_estimate: 1_024,
            safety_margin: 512,
            process_limit: 512 + 64 + 4_096 + 4_096 + state + 1_024 + 512,
            max_active_requests: 1,
        };
        let ledger = ProcessLedger::new(config).unwrap();
        let allocation = ledger
            .prepare_process_allocation_profile(usize::MAX)
            .unwrap();
        (ledger, allocation)
    }

    fn entry_plan() -> EntryIoAuditPlan {
        EntryIoAuditPlan::locked_linux_x86_64_tokio_1_53_1(64, 64, &[]).unwrap()
    }

    fn official_ledger() -> (
        ProcessLedger,
        saddle_admission::OfficialTokioDomain,
        ProcessAllocationProfile,
    ) {
        let registration = OfficialTokioRegistrationProfile {
            listener: 1,
            transport_connections: 1,
            runtime_fixed: 1,
        };
        let state = ProcessLedger::minimum_process_state_reserve(1).unwrap()
            + ProcessLedger::official_tokio_state_reserve(registration).unwrap();
        let config = ResourceConfig {
            managed_capacity: 512,
            entry_reserve: 64,
            framework_reserve: 4_096,
            task_reserve: 4_096,
            process_state_reserve: state,
            system_estimate: 1_024,
            safety_margin: 512,
            process_limit: 512 + 64 + 4_096 + 4_096 + state + 1_024 + 512,
            max_active_requests: 1,
        };
        let ledger = ProcessLedger::new(config).unwrap();
        let domain = ledger.prepare_official_tokio_domain(registration).unwrap();
        let allocation = ledger
            .prepare_process_allocation_profile(usize::MAX)
            .unwrap();
        (ledger, domain, allocation)
    }

    struct TestConnection;

    struct UserRead;

    impl EntryReadPoll<TestConnection> for UserRead {
        fn poll_read(
            &mut self,
            _: &mut TestConnection,
            memory: &RequestMemory,
            _: &mut Context<'_>,
        ) -> Poll<Result<ManagedBytes, AdmissionError>> {
            Poll::Ready(memory.try_bytes(&42_u64.to_le_bytes()))
        }
    }

    struct UserWrite;

    impl ResponseWritePoll<TestConnection> for UserWrite {
        fn poll_write(
            &mut self,
            _: &mut TestConnection,
            response: &ManagedResponse,
            _: &mut Context<'_>,
        ) -> Poll<Result<(), AdmissionError>> {
            assert_eq!(
                u64::from_le_bytes(response.as_slice()[..8].try_into().unwrap()),
                45
            );
            assert_eq!(response.as_slice()[8], 1);
            Poll::Ready(Ok(()))
        }
    }

    struct UserLookup;

    impl Service for UserLookup {
        type Request = UserRequest;
        type Response = UserResponse;
    }

    struct OrderCreate;

    impl Service for OrderCreate {
        type Request = OrderRequest;
        type Response = OrderResponse;
    }

    #[derive(Clone, Copy)]
    struct RequestIdentity(u64);

    type UserRequest = ManagedU64;
    type UserResponse = ManagedPair<ManagedU64, ManagedBool>;
    type UserCodec = FixedManagedCodec<UserRequest, UserResponse>;
    type OrderRequest = ManagedPair<ManagedU64, ManagedU64>;
    type OrderResponse = ManagedPair<ManagedU64, ManagedU64>;
    type OrderCodec = FixedManagedCodec<OrderRequest, OrderResponse>;

    struct SelectUser;

    impl StaticQueryOptionalOperation for SelectUser {
        type Parameters = ManagedQueryParameters<DbU64>;
        type Row = ManagedQueryRow<DbU64>;

        const OPERATION: &'static str = "users.lookup";
        const SQL: &'static str = "SELECT id FROM users WHERE id = ?";
    }

    struct SelectOrder;

    impl StaticQueryOptionalOperation for SelectOrder {
        type Parameters = ManagedQueryParameters<DbU64>;
        type Row = ManagedQueryRow<DbU64>;

        const OPERATION: &'static str = "orders.lookup";
        const SQL: &'static str = "SELECT id FROM orders WHERE id = ?";
    }

    struct UpdateUser;

    impl StaticWriteOperation for UpdateUser {
        type Parameters = ManagedQueryParameters<DbU64>;

        const OPERATION: &'static str = "users.update";
        const SQL: &'static str = "UPDATE users SET active = 1 WHERE id = ?";
    }

    struct ReadyOnce<T> {
        value: Option<T>,
        yielded: bool,
    }

    impl<T> ReadyOnce<T> {
        fn new(value: T) -> Self {
            Self {
                value: Some(value),
                yielded: false,
            }
        }
    }

    impl<T: Unpin> Future for ReadyOnce<T> {
        type Output = T;

        fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<T> {
            if !self.yielded {
                self.yielded = true;
                context.waker().wake_by_ref();
                return Poll::Pending;
            }
            Poll::Ready(self.value.take().expect("future completes only once"))
        }
    }

    fn user_handler(
        identity: RequestIdentity,
        request: UserRequest,
    ) -> ReadyOnce<Result<UserResponse, ExecutionError>> {
        ReadyOnce::new(Ok(ManagedPair(
            ManagedU64(request.0 ^ identity.0),
            ManagedBool(true),
        )))
    }

    fn order_handler(
        identity: RequestIdentity,
        request: OrderRequest,
    ) -> ReadyOnce<Result<OrderResponse, ExecutionError>> {
        ReadyOnce::new(Ok(ManagedPair(
            ManagedU64(request.0.0 ^ identity.0),
            request.1,
        )))
    }

    fn generated_layout(slot: usize) -> Option<CompiledRouteExecutionProof> {
        match slot {
            0 => ExecutionLayoutProof::bind::<
                UserLookup,
                BODY,
                OUTPUT,
                UserRequest,
                UserResponse,
                UserCodec,
                ReadyOnce<Result<UserResponse, ExecutionError>>,
            >()
            .ok()
            .map(|layout| {
                layout.with_query_optional(
                    QueryOptionalOperationProof::<SelectUser>::bind()
                        .expect("generated DB operation is valid"),
                )
            }),
            1 => ExecutionLayoutProof::bind::<
                OrderCreate,
                BODY,
                OUTPUT,
                OrderRequest,
                OrderResponse,
                OrderCodec,
                ReadyOnce<Result<OrderResponse, ExecutionError>>,
            >()
            .ok()
            .map(ExecutionLayoutProof::without_db),
            _ => None,
        }
    }

    // Rust 1.85 precise capture is the contract under test; `async fn` would
    // capture all argument lifetimes once Runtime resources are added.
    #[allow(clippy::manual_async_fn)]
    fn generated_dispatch(
        slot: usize,
        identity: RequestIdentity,
        body: ManagedBytes,
        db: CompiledDbPermit,
        mut output: ManagedResponseBuilder,
    ) -> impl Future<Output = Result<ManagedResponse, ExecutionError>> + Send + 'static + use<>
    {
        enum Decoded {
            User(UserRequest),
            Order(OrderRequest),
        }
        let decoded = match slot {
            0 => body
                .as_slice()
                .try_into()
                .map(u64::from_le_bytes)
                .map(ManagedU64)
                .map(Decoded::User)
                .map_err(|_| ExecutionError::InvalidInput),
            1 => body
                .as_slice()
                .try_into()
                .map(|bytes: [u8; 16]| {
                    Decoded::Order(ManagedPair(
                        ManagedU64(u64::from_le_bytes(bytes[..8].try_into().unwrap())),
                        ManagedU64(u64::from_le_bytes(bytes[8..].try_into().unwrap())),
                    ))
                })
                .map_err(|_| ExecutionError::InvalidInput),
            _ => Err(ExecutionError::UnknownExecution),
        };
        drop(body);
        async move {
            match decoded? {
                Decoded::User(request) => {
                    assert_eq!(db.demand(), Some((1, 1)));
                    let response = user_handler(identity, request).await?;
                    // This generated fixture has no physical database connection.
                    // Model its completed return explicitly so the real admission
                    // role is not dropped while still in Query.
                    db.complete_after_test_connection_return();
                    output
                        .try_extend_from_slice(&response.0.0.to_le_bytes())
                        .map_err(|_| ExecutionError::OutputTooLarge)?;
                    output
                        .try_extend_from_slice(&[u8::from(response.1.0)])
                        .map_err(|_| ExecutionError::OutputTooLarge)?;
                }
                Decoded::Order(request) => {
                    assert!(db.is_none());
                    let query = QueryOptionalOperationProof::<SelectUser>::bind()
                        .expect("generated DB operation is valid");
                    assert!(matches!(
                        db.handoff_query_optional(
                            query.invocation(ManagedQueryParameters(DbU64(1)))
                        ),
                        Err(ExecutionError::RouteResourcesMismatch)
                    ));
                    let response = order_handler(identity, request).await?;
                    output
                        .try_extend_from_slice(&response.0.0.to_le_bytes())
                        .map_err(|_| ExecutionError::OutputTooLarge)?;
                    output
                        .try_extend_from_slice(&response.1.0.to_le_bytes())
                        .map_err(|_| ExecutionError::OutputTooLarge)?;
                }
            }
            output
                .finish()
                .map_err(|_| ExecutionError::ResponseReservationFailed)
        }
    }

    fn registry() -> CompiledServiceRegistry {
        let mut builder = CompiledServiceRegistryBuilder::new(2).unwrap();
        builder
            .register::<UserLookup>(ServiceDescriptor::new("users", "user", "lookup"))
            .unwrap();
        builder
            .register::<OrderCreate>(ServiceDescriptor::new("orders", "order", "create"))
            .unwrap();
        builder.freeze().unwrap()
    }

    macro_rules! capability {
        () => {{
            let mut builder = CompiledExecutionBuilder::<_, _, _, _, BODY, OUTPUT>::new(
                registry(),
                generated_dispatch,
                generated_layout,
                EntryCapabilityLimits::new(2, 32, 64, 2, 2),
            )
            .unwrap();
            builder
                .expose::<UserLookup>(
                    b"/users/lookup",
                    EntryFraming::post_managed_content_length(),
                )
                .unwrap();
            builder
                .expose::<OrderCreate>(
                    b"/orders/create",
                    EntryFraming::post_managed_content_length(),
                )
                .unwrap();
            builder.freeze().unwrap()
        }};
    }

    struct TestCapacityArtifact;

    impl GeneratedServiceCapacityArtifactOwner for TestCapacityArtifact {
        fn build_identity(&self) -> [u8; 32] {
            [0xb1; 32]
        }

        fn artifact_identity(&self) -> [u8; 32] {
            [0xa1; 32]
        }

        fn owner_generation(&self) -> u64 {
            7
        }
    }

    #[test]
    fn capacity_leaf_is_frozen_with_the_execution_registry() {
        use saddle_admission::ServiceCapacitySourceLeaf;

        let mut builder = CompiledExecutionBuilder::<_, _, _, _, BODY, OUTPUT>::new(
            registry(),
            generated_dispatch,
            generated_layout,
            EntryCapabilityLimits::new(2, 32, 64, 2, 2),
        )
        .unwrap();
        builder
            .expose::<UserLookup>(
                b"/users/lookup",
                EntryFraming::post_managed_content_length(),
            )
            .unwrap();
        builder
            .expose::<OrderCreate>(
                b"/orders/create",
                EntryFraming::post_managed_content_length(),
            )
            .unwrap();
        let execution = builder
            .freeze_with_capacity_leaf(TestCapacityArtifact)
            .unwrap();
        let leaf = execution.capacity_source();

        assert_eq!(leaf.owner_generation(), 7);
        assert_eq!(leaf.route_count(), 2);
        assert_eq!(leaf.managed_objects_peak(0), Some(2));
        assert_eq!(leaf.managed_objects_peak(1), Some(2));
        assert_ne!(leaf.leaf_identity(), [0; 32]);
        assert_ne!(leaf.route_type_closure_identity(), [0; 32]);
        assert_eq!(leaf.common_identities()[..2], [[0xb1; 32], [0xa1; 32]]);
        assert_ne!(leaf.common_identities()[2], [0; 32]);

        for index in 0..leaf.route_count() {
            let route = leaf.route_identity(index).unwrap();
            assert_ne!(route, 0);
            assert_eq!(route >> 32, leaf.route_identity(0).unwrap() >> 32);
            assert!(leaf.managed_commitment_bytes(index).unwrap() > 0);
        }
        assert_eq!(leaf.db_connections(0), Some(1));
        assert_eq!(leaf.db_operations(0), Some(1));
        assert_eq!(leaf.db_connections(1), Some(0));
        assert_eq!(leaf.db_operations(1), Some(0));

        let query = execution
            .lookup(
                b"/users/lookup",
                EntryFraming::post_managed_content_length(),
                Some(8),
            )
            .unwrap();
        let no_db = execution
            .lookup(
                b"/orders/create",
                EntryFraming::post_managed_content_length(),
                Some(16),
            )
            .unwrap();
        assert_eq!(
            query.identity().contract().opaque(),
            leaf.route_identity(0).unwrap()
        );
        assert_eq!(
            no_db.identity().contract().opaque(),
            leaf.route_identity(1).unwrap()
        );
        assert_eq!(
            query.plan().commitment() as usize,
            leaf.managed_commitment_bytes(0).unwrap()
        );
        assert_eq!(
            no_db.plan().commitment() as usize,
            leaf.managed_commitment_bytes(1).unwrap()
        );

        let first = execution
            .production_fact_input()
            .unwrap()
            .into_canonical_json()
            .unwrap();
        let second = execution
            .production_fact_input()
            .unwrap()
            .into_canonical_json()
            .unwrap();
        assert_eq!(first, second);
        assert!(!first.contains(&b'\n'));
        let document: serde_json::Value = serde_json::from_slice(&first).unwrap();
        assert_eq!(document["authority"], false);
        assert_eq!(document["capacity"]["manifest_leaf"], "service");
        assert_eq!(document["capacity"]["commitment"], "service_capacity");
        assert_eq!(document["capacity"]["routes"][0]["route"], "/orders/create");
        assert_eq!(document["capacity"]["routes"][0]["db_connections"], 0);
        assert_eq!(document["capacity"]["routes"][1]["route"], "/users/lookup");
        assert_eq!(document["capacity"]["routes"][1]["db_connections"], 1);
        assert!(document.get("root_identity").is_none());
        assert!(document.get("owner_generation").is_none());
    }

    #[tokio::test(flavor = "current_thread")]
    async fn factory_reserves_response_and_real_async_path_finishes_after_await() {
        let _serial = EXECUTION_TEST_LOCK.lock().await;
        let capability = capability!();
        let order = capability
            .lookup(
                b"/orders/create",
                EntryFraming::post_managed_content_length(),
                Some(16),
            )
            .unwrap();
        fn assert_runtime_adapter<A: CompiledRouteAdapter>(_: &A) {}
        assert_runtime_adapter(&capability);
        assert_eq!(
            capability.managed_commitment(order).unwrap(),
            usize::try_from(order.plan().commitment()).unwrap()
        );
        assert_eq!(capability.response_capacity(order).unwrap(), OUTPUT);
        assert!(capability.db_resources(order, None).is_ok());
        assert!(order.db_demand().is_none());
        assert_eq!(order.response_capacity(), OUTPUT as u64);
        let (ledger, allocation) = request_ledger();
        let envelope = ledger
            .try_envelope(256, 2_048, |memory| {
                let mut bytes = [0_u8; 16];
                bytes[..8].copy_from_slice(&42_u64.to_le_bytes());
                bytes[8..].copy_from_slice(&1999_u64.to_le_bytes());
                let body = memory.try_bytes(&bytes).unwrap();
                let execution = CompiledRouteAdapter::execute(
                    &capability,
                    order,
                    RequestIdentity(9),
                    body,
                    None,
                    memory,
                )
                .unwrap();
                async move {
                    let response = execution.await.unwrap();
                    assert_eq!(
                        u64::from_le_bytes(response.as_slice()[..8].try_into().unwrap()),
                        35
                    );
                    assert_eq!(
                        u64::from_le_bytes(response.as_slice()[8..].try_into().unwrap()),
                        1999
                    );
                }
            })
            .unwrap();
        let report = envelope.await.unwrap();
        assert_eq!(report.escape_allocations, 0);
        assert_eq!(report.managed_allocations, 2);
        allocation.finish().unwrap();
        ledger.try_shutdown().unwrap();
    }

    #[tokio::test(flavor = "current_thread")]
    async fn sealed_db_permit_and_response_capacity_cross_the_real_async_path() {
        let _serial = EXECUTION_TEST_LOCK.lock().await;
        let capability = capability!();
        let user = capability
            .lookup(
                b"/users/lookup",
                EntryFraming::post_managed_content_length(),
                Some(8),
            )
            .unwrap();
        let proof = user;
        assert_eq!(proof.db_demand().connections(), 1);
        assert_eq!(proof.db_demand().operations(), 1);
        assert_eq!(proof.response_capacity(), OUTPUT as u64);

        let (ledger, domain, allocation) = official_ledger();
        let db = ledger
            .prepare_db_domain(DbCreditProfile {
                connections: 1,
                operations: 1,
            })
            .unwrap();
        let resources = capability.db_resources(proof, Some(&db)).unwrap();
        let envelope = match ledger.attempt_official_tokio_entry_io(
            &domain,
            resources,
            256,
            4_096,
            entry_plan(),
            entry_plan(),
            |_| (TestConnection, UserRead, UserWrite),
            |body, permit, memory| {
                let execution = CompiledRouteAdapter::execute(
                    &capability,
                    user,
                    RequestIdentity(7),
                    body,
                    permit,
                    memory,
                )
                .unwrap();
                fn assert_send_static<T: Send + 'static>(_: &T) {}
                assert_send_static(&execution);
                async move { execution.await.unwrap() }
            },
        ) {
            OfficialTokioEntryIoAttemptOutcome::Ready(envelope) => envelope,
            _ => panic!("sealed route resources must admit"),
        };
        let (envelope, task_slot) = envelope.into_runtime_parts();
        let report = envelope.await.unwrap();
        drop(task_slot);
        assert_eq!(report.escape_allocations, 0);
        assert_eq!(db.snapshot().unwrap().connections_in_use, 0);
        drop(db);
        drop(domain);
        allocation.finish().unwrap();
        ledger.try_shutdown().unwrap();
    }

    #[tokio::test(flavor = "current_thread")]
    async fn missing_db_permit_fails_before_response_reservation_or_dispatch() {
        let _serial = EXECUTION_TEST_LOCK.lock().await;
        let capability = capability!();
        let user = capability
            .lookup(
                b"/users/lookup",
                EntryFraming::post_managed_content_length(),
                Some(8),
            )
            .unwrap();
        let (ledger, allocation) = request_ledger();
        let envelope = ledger
            .try_envelope(256, 2_048, |memory| {
                let body = memory.try_bytes(&42_u64.to_le_bytes()).unwrap();
                assert!(matches!(
                    capability.execute(user, RequestIdentity(7), body, None, memory),
                    Err(ExecutionError::RouteResourcesMismatch)
                ));
                std::future::ready(())
            })
            .unwrap();
        let report = envelope.await.unwrap();
        assert_eq!(report.managed_allocations, 1);
        allocation.finish().unwrap();
        ledger.try_shutdown().unwrap();
    }

    #[tokio::test(flavor = "current_thread")]
    async fn registered_entry_is_the_only_execution_credential() {
        let _serial = EXECUTION_TEST_LOCK.lock().await;
        let first = capability!();
        let second = capability!();
        let foreign = first
            .lookup(
                b"/users/lookup",
                EntryFraming::post_managed_content_length(),
                Some(8),
            )
            .unwrap();
        assert!(matches!(
            CompiledRouteAdapter::managed_commitment(&second, foreign),
            Err(ExecutionError::Registry(RegistryError::UnknownFactory))
        ));
        assert!(matches!(
            second.db_resources(foreign, None),
            Err(ExecutionError::Registry(RegistryError::UnknownFactory))
        ));
        let (ledger, allocation) = request_ledger();
        let envelope = ledger
            .try_envelope(256, 2_048, |memory| {
                let body = memory.try_bytes(&1_u64.to_le_bytes()).unwrap();
                let result = second.execute(foreign, RequestIdentity(1), body, None, memory);
                assert!(matches!(
                    result,
                    Err(ExecutionError::Registry(RegistryError::UnknownFactory))
                ));
                std::future::ready(())
            })
            .unwrap();
        envelope.await.unwrap();
        allocation.finish().unwrap();
        ledger.try_shutdown().unwrap();
    }

    #[test]
    fn proof_is_derived_from_concrete_layouts_and_fixed_payloads_do_not_allocate() {
        type Handler = ReadyOnce<Result<OrderResponse, ExecutionError>>;
        let proof = ExecutionLayoutProof::bind::<
            OrderCreate,
            BODY,
            OUTPUT,
            OrderRequest,
            OrderResponse,
            OrderCodec,
            Handler,
        >()
        .unwrap();
        assert_eq!(proof.body_bytes(), BODY);
        assert_eq!(proof.future_bytes(), mem::size_of::<Handler>());
        assert!(proof.commitment().unwrap() >= BODY + OUTPUT);
        assert_eq!(FixedBytes::<BODY>::empty().capacity(), BODY);
    }

    #[test]
    fn marker_binding_and_typed_internal_edges_are_checked_at_assembly() {
        struct Missing;
        impl Service for Missing {
            type Request = ManagedU64;
            type Response = ManagedU64;
        }

        let mut builder = CompiledExecutionBuilder::<_, _, _, _, BODY, OUTPUT>::new(
            registry(),
            generated_dispatch,
            generated_layout,
            EntryCapabilityLimits::new(2, 32, 64, 2, 2),
        )
        .unwrap();
        let edge = builder
            .bind_internal_call::<UserLookup, OrderCreate>()
            .unwrap();
        let _: InternalCallCapability<UserLookup, OrderCreate> = edge;
        assert!(matches!(
            builder.bind_internal_call::<UserLookup, Missing>(),
            Err(ExecutionError::Registry(RegistryError::UnregisteredService))
        ));

        let order_proof = generated_layout(1).unwrap();
        let mut mismatched = CompiledExecutionBuilder::<_, _, _, _, BODY, OUTPUT>::new(
            registry(),
            generated_dispatch,
            move |_| Some(order_proof),
            EntryCapabilityLimits::new(2, 32, 64, 2, 2),
        )
        .unwrap();
        assert_eq!(
            mismatched
                .expose::<UserLookup>(
                    b"/users/lookup",
                    EntryFraming::post_managed_content_length(),
                )
                .unwrap_err(),
            ExecutionError::UnknownExecution
        );

        builder
            .expose::<UserLookup>(
                b"/users/lookup",
                EntryFraming::post_managed_content_length(),
            )
            .unwrap();
    }

    #[test]
    fn assembly_rejects_foreign_query_proof_for_one_route() {
        let conflicting_layout = |slot| {
            generated_layout(slot).map(|proof| {
                if slot == 0 {
                    proof.with_query_optional(
                        QueryOptionalOperationProof::<SelectOrder>::bind().unwrap(),
                    )
                } else {
                    proof
                }
            })
        };
        let mut builder = CompiledExecutionBuilder::<_, _, _, _, BODY, OUTPUT>::new(
            registry(),
            generated_dispatch,
            conflicting_layout,
            EntryCapabilityLimits::new(2, 32, 64, 2, 2),
        )
        .unwrap();
        assert_eq!(
            builder
                .expose::<UserLookup>(
                    b"/users/lookup",
                    EntryFraming::post_managed_content_length(),
                )
                .unwrap_err(),
            ExecutionError::UnknownExecution
        );
    }

    #[test]
    fn assembly_freezes_standalone_write_and_single_step_transaction_kinds() {
        let write_layout = |slot| {
            if slot != 0 {
                return generated_layout(slot);
            }
            ExecutionLayoutProof::bind::<
                UserLookup,
                BODY,
                OUTPUT,
                UserRequest,
                UserResponse,
                UserCodec,
                ReadyOnce<Result<UserResponse, ExecutionError>>,
            >()
            .ok()
            .map(|layout| layout.with_write(WriteOperationProof::<UpdateUser>::bind().unwrap()))
        };
        let mut write = CompiledExecutionBuilder::<_, _, _, _, BODY, OUTPUT>::new(
            registry(),
            generated_dispatch,
            write_layout,
            EntryCapabilityLimits::new(2, 32, 64, 2, 2),
        )
        .unwrap();
        write
            .expose::<UserLookup>(
                b"/users/update",
                EntryFraming::post_managed_content_length(),
            )
            .unwrap();
        let _write_capability = write.freeze().unwrap();

        let transaction_layout = |slot| {
            if slot != 0 {
                return generated_layout(slot);
            }
            ExecutionLayoutProof::bind::<
                UserLookup,
                BODY,
                OUTPUT,
                UserRequest,
                UserResponse,
                UserCodec,
                ReadyOnce<Result<UserResponse, ExecutionError>>,
            >()
            .ok()
            .map(|layout| {
                layout.with_transaction(TransactionOperationProof::<UpdateUser>::bind().unwrap())
            })
        };
        let mut transaction = CompiledExecutionBuilder::<_, _, _, _, BODY, OUTPUT>::new(
            registry(),
            generated_dispatch,
            transaction_layout,
            EntryCapabilityLimits::new(2, 32, 64, 2, 2),
        )
        .unwrap();
        transaction
            .expose::<UserLookup>(
                b"/users/transaction",
                EntryFraming::post_managed_content_length(),
            )
            .unwrap();
        let _transaction_capability = transaction.freeze().unwrap();
    }

    #[test]
    fn assembly_rejects_write_transaction_value_recombination() {
        let recombined = |slot| {
            if slot != 0 {
                return generated_layout(slot);
            }
            ExecutionLayoutProof::bind::<
                UserLookup,
                BODY,
                OUTPUT,
                UserRequest,
                UserResponse,
                UserCodec,
                ReadyOnce<Result<UserResponse, ExecutionError>>,
            >()
            .ok()
            .map(|layout| {
                layout
                    .with_write(WriteOperationProof::<UpdateUser>::bind().unwrap())
                    .with_transaction(TransactionOperationProof::<UpdateUser>::bind().unwrap())
            })
        };
        let mut builder = CompiledExecutionBuilder::<_, _, _, _, BODY, OUTPUT>::new(
            registry(),
            generated_dispatch,
            recombined,
            EntryCapabilityLimits::new(2, 32, 64, 2, 2),
        )
        .unwrap();
        assert_eq!(
            builder
                .expose::<UserLookup>(
                    b"/users/recombined",
                    EntryFraming::post_managed_content_length(),
                )
                .unwrap_err(),
            ExecutionError::UnknownExecution
        );

        let transaction = TransactionOperationProof::<UpdateUser>::bind()
            .unwrap()
            .invocation(
                ManagedQueryParameters(DbU64(1)),
                TransactionDecision::Commit,
            );
        assert_eq!(
            transaction.layout().parameter_bytes(),
            mem::size_of::<ManagedQueryParameters<DbU64>>()
        );
    }

    #[test]
    fn compiled_registry_is_bounded_and_rejects_duplicate_markers_and_identities() {
        assert!(matches!(
            CompiledServiceRegistryBuilder::new(0),
            Err(CompiledRegistryError::InvalidCapacity)
        ));
        assert!(matches!(
            CompiledServiceRegistryBuilder::new(1).unwrap().freeze(),
            Err(CompiledRegistryError::EmptyRegistry)
        ));

        let mut duplicate_contract = CompiledServiceRegistryBuilder::new(2).unwrap();
        duplicate_contract
            .register::<UserLookup>(ServiceDescriptor::new("users", "user", "lookup"))
            .unwrap();
        assert_eq!(
            duplicate_contract
                .register::<UserLookup>(ServiceDescriptor::new("users", "user", "other"))
                .unwrap_err(),
            CompiledRegistryError::DuplicateContract
        );

        let mut duplicate_identity = CompiledServiceRegistryBuilder::new(2).unwrap();
        duplicate_identity
            .register::<UserLookup>(ServiceDescriptor::new("users", "user", "lookup"))
            .unwrap();
        assert_eq!(
            duplicate_identity
                .register::<OrderCreate>(ServiceDescriptor::new("users", "user", "lookup"))
                .unwrap_err(),
            CompiledRegistryError::DuplicateIdentity
        );

        let mut bounded = CompiledServiceRegistryBuilder::new(1).unwrap();
        bounded
            .register::<UserLookup>(ServiceDescriptor::new("users", "user", "lookup"))
            .unwrap();
        assert_eq!(
            bounded
                .register::<OrderCreate>(ServiceDescriptor::new("orders", "order", "create"))
                .unwrap_err(),
            CompiledRegistryError::CapacityExceeded
        );
    }

    #[test]
    fn execution_errors_exhaustively_map_to_the_closed_response_classes() {
        for error in [ExecutionError::InputTooLarge, ExecutionError::InvalidInput] {
            assert_eq!(
                error.response_outcome_class(),
                ResponseOutcomeClass::InvalidRequest
            );
        }
        assert_eq!(
            ExecutionError::Business.response_outcome_class(),
            ResponseOutcomeClass::BusinessRejected
        );
        assert_eq!(
            ExecutionError::RouteResourcesMismatch.response_outcome_class(),
            ResponseOutcomeClass::Unavailable
        );
        assert_eq!(
            ExecutionError::DependencyUnavailable.response_outcome_class(),
            ResponseOutcomeClass::Unavailable
        );
        assert_eq!(
            ExecutionError::DependencyExecutionFailed.response_outcome_class(),
            ResponseOutcomeClass::Internal
        );
        for error in [
            ExecutionError::Registry(RegistryError::UnknownFactory),
            ExecutionError::CompiledRegistry(CompiledRegistryError::EmptyRegistry),
            ExecutionError::UnknownExecution,
            ExecutionError::OutputTooLarge,
            ExecutionError::ResponseReservationFailed,
        ] {
            assert_eq!(
                error.response_outcome_class(),
                ResponseOutcomeClass::Internal
            );
        }
    }

    #[test]
    fn frozen_capability_is_the_classified_production_adapter() {
        fn assert_classified<A>(_: &A)
        where
            A: saddle_runtime::compiled_route::ClassifiedCompiledRouteAdapter<
                    Proof = RegisteredRouteExecutionProof,
                    Context = RequestIdentity,
                    Error = ExecutionError,
                >,
        {
        }

        let capability = capability!();
        assert_classified(&capability);
    }

    #[allow(dead_code)]
    fn _assert_context_is_owned_and_send(value: RequestIdentity) -> SaddleResult<()> {
        fn assert_send_static<T: Send + 'static>(_: T) {}
        assert_send_static(value);
        Ok(())
    }
}