saddle-framework 0.3.25

use std::sync::{Arc, Mutex};
use std::future::Future;

use saddle_core::CallContext;
use saddle_db::internal::{
    ManagedOptionalRow, ManagedWriteResult, QueryOptionalOperationProof, ScopeDatabaseError,
    StaticQueryOptionalOperation, StaticWriteOperation, WriteOperationProof,
};
use saddle_observability::EventContext;
use saddle_runtime::profusegw::{
    ProfuseGwManagedDispatch, ProfuseGwPostDatabaseRequestTerminal,
    finish_profusegw_after_database, finish_profusegw_without_database,
};
use tokio::sync::Notify;

mod reserved_request;

/// Generated named ABI conversion; only generated code may implement this
/// internal surface. It carries no runtime execution or admission authority.
#[doc(hidden)]
pub trait GeneratedNamedQuery: StaticQueryOptionalOperation {
    type NamedParameters: Send;
    type NamedRow: Send;
    const REQUIRES_TRANSACTION: bool;
    fn parameters(value: Self::NamedParameters) -> Self::Parameters;
    fn row(value: ManagedOptionalRow<Self::Row>) -> Option<Self::NamedRow>;
}

#[doc(hidden)]
pub trait GeneratedNamedWrite: StaticWriteOperation {
    type NamedParameters: Send;
    fn parameters(value: Self::NamedParameters) -> Self::Parameters;
}

/// Generated conversion only, not a new execution/decoder authority.
#[doc(hidden)]
pub trait GeneratedOwnedQuery: saddle_db::internal::StaticOwnedQuery {
    type NamedParameters: Send;
    type NamedRow: From<Self::Row> + Send;
    fn parameters(value: Self::NamedParameters) -> Self::Parameters;
    fn restore_parameters(value: Self::Parameters) -> Self::NamedParameters;
    fn validate_parameters(
        _: &Self::NamedParameters,
        _: &DatabaseTarget,
    ) -> Result<(), saddle_db::internal::TimestampTargetError> {
        Ok(())
    }
}
#[doc(hidden)]
pub trait GeneratedOwnedWrite: saddle_db::internal::StaticOwnedWrite {
    type NamedParameters: Send;
    fn parameters(value: Self::NamedParameters) -> Self::Parameters;
    fn restore_parameters(value: Self::Parameters) -> Self::NamedParameters;
    fn validate_parameters(
        _: &Self::NamedParameters,
        _: &DatabaseTarget,
    ) -> Result<(), saddle_db::internal::TimestampTargetError> {
        Ok(())
    }
}

/// Internal borrowed target access, not a range constructor or execution owner.
#[doc(hidden)]
pub struct DatabaseTarget(
    Option<Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>>,
);
impl DatabaseTarget {
    #[cfg(feature = "cf-transaction-construction")]
    pub(crate) fn for_construction(process: Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>) -> Self {
        Self(Some(process))
    }
    pub fn timestamp<const F: usize>(
        &self,
        value: &saddle_db::internal::DbTimestamp<F>,
    ) -> Result<(), saddle_db::internal::TimestampTargetError> {
        self.0
            .as_ref()
            .ok_or(saddle_db::internal::TimestampTargetError::ProcessClosed)?
            .native_timestamp_range()?
            .validate(value)
    }
}

/// A source-rejected argument is returned intact for an explicit caller decision.
pub enum QueryFailure<P> {
    Parameters {
        parameters: P,
    },
    Target {
        parameters: P,
        error: saddle_db::internal::TimestampTargetError,
    },
    Database(ScopeDatabaseError),
}
impl<P> std::fmt::Debug for QueryFailure<P> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Parameters { .. } => f.write_str("Parameters"),
            Self::Target { error, .. } => f.debug_tuple("Target").field(error).finish(),
            Self::Database(e) => f.debug_tuple("Database").field(e).finish(),
        }
    }
}
impl<P> QueryFailure<P> {
    /// Explicitly abandon any returned arguments and retain the technical failure.
    pub fn into_failure(self) -> ScopeDatabaseError {
        match self {
            Self::Parameters { .. } => ScopeDatabaseError::Resource,
            Self::Target { error, .. } => match error {
                saddle_db::internal::TimestampTargetError::OutOfRange => {
                    ScopeDatabaseError::InvalidValue
                }
                saddle_db::internal::TimestampTargetError::UnsupportedTarget
                | saddle_db::internal::TimestampTargetError::ProcessClosed => {
                    ScopeDatabaseError::Unavailable
                }
            },
            Self::Database(e) => e,
        }
    }
}
pub(crate) fn named_failure<O: GeneratedOwnedQuery>(
    error: saddle_db::internal::OwnedExecutionError<O::Parameters>,
) -> QueryFailure<O::NamedParameters> {
    match error {
        saddle_db::internal::OwnedExecutionError::Parameters { parameters, .. } => {
            QueryFailure::Parameters {
                parameters: O::restore_parameters(parameters),
            }
        }
        saddle_db::internal::OwnedExecutionError::Database(e) => QueryFailure::Database(e),
    }
}

#[doc(hidden)]
pub async fn transaction_rows<O: GeneratedOwnedQuery>(
    session: &mut TransactionSession<'_, '_>,
    parameters: O::NamedParameters,
) -> Result<crate::database::Rows<O::Row, O::NamedRow>, QueryFailure<O::NamedParameters>> {
    let proof =
        saddle_db::internal::OwnedQueryProof::<O>::bind().map_err(QueryFailure::Database)?;
    session
        .owned_rows(proof.invocation(O::parameters(parameters)))
        .await
        .map(crate::database_rows::Rows::from_database)
        .map_err(named_failure::<O>)
}
#[doc(hidden)]
pub async fn transaction_optional<O: GeneratedOwnedQuery>(
    session: &mut TransactionSession<'_, '_>,
    parameters: O::NamedParameters,
) -> Result<Option<O::NamedRow>, QueryFailure<O::NamedParameters>> {
    let proof =
        saddle_db::internal::OwnedQueryProof::<O>::bind().map_err(QueryFailure::Database)?;
    session
        .owned_optional(proof.invocation(O::parameters(parameters)))
        .await
        .map(|row| row.map(O::NamedRow::from))
        .map_err(named_failure::<O>)
}
#[doc(hidden)]
pub async fn transaction_probe<O: GeneratedOwnedQuery>(
    session: &mut TransactionSession<'_, '_>,
    parameters: O::NamedParameters,
) -> Result<crate::database::MatchCardinality<O::NamedRow>, QueryFailure<O::NamedParameters>> {
    use crate::database::MatchCardinality;
    use saddle_db::internal::OwnedDbProbe;
    let proof =
        saddle_db::internal::OwnedQueryProof::<O>::bind().map_err(QueryFailure::Database)?;
    session
        .owned_probe(proof.invocation(O::parameters(parameters)))
        .await
        .map(|row| match row {
            OwnedDbProbe::Zero => MatchCardinality::Zero,
            OwnedDbProbe::One(row) => MatchCardinality::One(O::NamedRow::from(row)),
            OwnedDbProbe::Many => MatchCardinality::Many,
        })
        .map_err(named_failure::<O>)
}

#[doc(hidden)]
pub trait GeneratedPageQuery: GeneratedOwnedQuery {
    const IS_OFFSET: bool;
    type Key: Copy + Send + AsRef<[u64]> + AsMut<[u64]>;
    type Cursor: crate::database_rows::CursorWire;
    fn empty_key() -> Self::Key;
    fn filter_fingerprint(value: &Self::NamedParameters) -> [u8; 32];
    fn page_parameters<'a>(
        value: &'a mut Self::NamedParameters,
        after: Option<Self::Key>,
        request: &'a mut DatabaseRequest,
    ) -> std::pin::Pin<
        Box<
            dyn std::future::Future<
                    Output = Result<
                        Self::Parameters,
                        saddle_db::internal::ParameterConstructionError,
                    >,
                > + Send
                + 'a,
        >,
    >;
    fn page_key(row: &Self::Row) -> Self::Key;
}
#[doc(hidden)]
pub trait GeneratedOffsetPageQuery: GeneratedPageQuery {}

enum RequestTerminal {
    Untouched(ProfuseGwManagedDispatch),
    InFlight,
    PostDatabase(ProfuseGwPostDatabaseRequestTerminal),
    Serial(SerialScope),
    Finished,
}

struct LegacyRequestDiagnostics {
    request_scope: saddle_observability::RequestDiagnosticScope<'static>,
    handler: Mutex<Option<saddle_observability::ActiveStage>>,
}

struct RequestState {
    // The formal reserved path owns its root projection/stage through Runtime.
    // Do not charge or allocate a second legacy snapshot and stage on that path.
    legacy: Option<Box<LegacyRequestDiagnostics>>,
    reserved_context: Option<Arc<saddle_runtime::request_task::reserved::ReservedTaskContext>>,
    reserved_failure: Mutex<Option<saddle_runtime::request_task::reserved::ReservedRequestFailure>>,
    encoding_failure: Mutex<Option<saddle_runtime::request_task::reserved::ReservedRequestFailure>>,
    encoding_supervision:Mutex<Option<saddle_core::OperationOutcome>>,
    reserved_observer: Option<saddle_observability::Observer>,
    handler_recorded: std::sync::atomic::AtomicBool,
    outbound_retentions: crate::programming::AttemptRetentions,
    terminal: Mutex<RequestTerminal>,
    cancel: Arc<Notify>,
    completed: Notify,
    diagnostic_handle: Option<saddle_observability::EmergencyDiagnosticHandle>,
}

// Non-authority, type-only input for R0. These are the actual allocations made
// by DatabaseRequest::new, not copies of their structures in a layout fixture.
pub(crate) fn request_storage_layouts()
-> std::result::Result<crate::process::consumer_storage::DatabaseConsumerLayouts, std::alloc::LayoutError>
{
    use crate::process::consumer_storage::{DatabaseConsumerLayouts, shared_layout};
    use std::alloc::Layout;
    Ok(DatabaseConsumerLayouts {
        state: shared_layout::<RequestState>()?,
        legacy_diagnostics: Layout::new::<LegacyRequestDiagnostics>(),
        cancellation_notify: shared_layout::<Notify>()?,
        outbound_retentions: shared_layout::<
            Mutex<Vec<saddle_boundary::request_diagnostics::RequiredBoundaryFailure>>,
        >()?,
        outbound_retention_element: Layout::new::<
            saddle_boundary::request_diagnostics::RequiredBoundaryFailure,
        >(),
        request_handle: Layout::new::<DatabaseRequest>(),
        completion_handle: Layout::new::<DatabaseRequestCompletion>(),
        encoding_handle: Layout::new::<ResponseEncodingContext>(),
    })
}

impl RequestState {
    fn finish_handler(
        &self,
        axes: &saddle_core::DiagnosticOutcomeAxes,
        error: Option<&saddle_core::SaddleError>,
    ) {
        if let (Some(context),Some(observer))=(&self.reserved_context,&self.reserved_observer) {
            if !self.handler_recorded.swap(true,std::sync::atomic::Ordering::AcqRel) {
                let _submission=context.view().ordinary(observer,
                    saddle_observability::root_diagnostic::RootRequestEvent::Handler,
                    saddle_observability::root_diagnostic::RootOutcomeFacts{axes:*axes,..Default::default()});
            }
            return;
        }
        let Some(legacy) = &self.legacy else {
            return;
        };
        let stage = legacy
            .handler
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take();
        if let Some(stage) = stage {
            if let Some(output) = &self.diagnostic_handle {
                let _submission = stage.finish_bounded_reference(output, axes, None);
            } else if let Some(error) = error {
                stage.fail(error);
            } else if matches!(axes.operation, saddle_core::OperationOutcome::Succeeded) {
                stage.succeed();
            } else {
                drop(stage);
            }
        }
    }
}

struct AbortOnLostRequestOwner(bool);
type SerialScope =
    saddle_runtime::profusegw::ProfuseGwSerialScope<tokio::sync::futures::OwnedNotified>;
#[doc(hidden)]
pub type TransactionSession<'session, 'request> = saddle_db::internal::DatabaseTransactionSession<
    'session,
    'request,
    tokio::sync::futures::OwnedNotified,
>;

impl Drop for AbortOnLostRequestOwner {
    fn drop(&mut self) {
        if self.0 {
            std::process::abort();
        }
    }
}

/// Framework-owned request database carrier. Generated capabilities retain
/// only a private reference to this state; business code cannot construct or
/// recover Runtime or Database owners from it.
#[doc(hidden)]
pub struct DatabaseRequest {
    diagnostic_zone:
        Result<saddle_observability::DiagnosticZone, saddle_observability::DiagnosticZoneError>,
    process: Option<Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>>,
    context: CallContext,
    event: EventContext,
    state: Arc<RequestState>,
    diagnostic_handle: Option<saddle_observability::EmergencyDiagnosticHandle>,
}

#[doc(hidden)]
pub struct DatabaseRequestCompletion {
    state: Arc<RequestState>,
}

#[doc(hidden)]
pub struct ResponseEncodingContext {
    diagnostic_zone:
        Result<saddle_observability::DiagnosticZone, saddle_observability::DiagnosticZoneError>,
    state: Arc<RequestState>,
    context: CallContext,
    event: EventContext,
    diagnostic_handle: Option<saddle_observability::EmergencyDiagnosticHandle>,
}
impl ResponseEncodingContext {
    #[doc(hidden)]
    pub fn handler_observation(&self)->Option<HandlerObservation> {
        Some(HandlerObservation{view:self.state.reserved_context.as_ref()?.view(),observer:self.state.reserved_observer.as_ref()?.clone(),output:self.diagnostic_handle.clone()})
    }
    #[doc(hidden)]
    pub fn finish_handler_observation(&self,stage:Option<saddle_runtime::request_task::reserved::ReservedActiveStage<'_>>,axes:&saddle_core::DiagnosticOutcomeAxes) {
        if let Some(stage)=stage {
            if !self.state.handler_recorded.swap(true,std::sync::atomic::Ordering::AcqRel) {
                if stage.finish_nonfailure(saddle_observability::root_diagnostic::RootOutcomeFacts{axes:*axes,..Default::default()}).is_err() {unreachable!("typed response completes handler operation");}
            }
        } else {self.handler_completed(axes);}
    }
    pub fn handler_completed(&self, axes: &saddle_core::DiagnosticOutcomeAxes) {
        // No exception is fabricated for an ordinary business Failure. Output
        // failure must not replace the original response or affect the owner.
        self.state.finish_handler(axes, None);
    }
    pub async fn encode<F>(self, action: F) -> saddle_core::Result<Vec<u8>>
    where
        F: FnOnce() -> Result<Vec<u8>, serde_json::Error> + Send,
    {
        let scope = {
            let mut terminal = self.state.terminal.lock().unwrap();
            match std::mem::replace(&mut *terminal, RequestTerminal::InFlight) {
                RequestTerminal::Serial(scope) => Some(scope),
                other => {
                    *terminal = other;
                    None
                }
            }
        };
        let output = self.diagnostic_handle.as_ref();
        let zone = self.diagnostic_zone;
        let context = (&self.context, &self.event);
        let state=&self.state;
        let encode = async move {
            action().map_err(|error| {
                use saddle_core::{
                    BoundedDiagnostic, BoundedDiagnosticCause, CaptureSite, DiagnosticCategory,
                    DiagnosticCode, DiagnosticOutcomeAxes, DiagnosticStage, OperationOutcome,
                };
                let code = match error.classify() {
                    serde_json::error::Category::Io
                        if error.io_error_kind() == Some(std::io::ErrorKind::FileTooLarge) =>
                    {
                        "saddle.response.payload_limit_exceeded"
                    }
                    serde_json::error::Category::Io => "saddle.response.encoding_io",
                    serde_json::error::Category::Syntax => "saddle.response.encoding_syntax",
                    serde_json::error::Category::Data => "saddle.response.encoding_data",
                    serde_json::error::Category::Eof => "saddle.response.encoding_eof",
                };
                let mut diagnostic = BoundedDiagnostic::capture(
                    DiagnosticCategory::UnexpectedError,
                    CaptureSite::FirstObserved,
                    BoundedDiagnosticCause::new(
                        DiagnosticStage::RequestResponse,
                        DiagnosticCode::new(code).expect("static diagnostic code"),
                    ),
                );
                let axes = DiagnosticOutcomeAxes {
                    operation: OperationOutcome::Failed,
                    ..Default::default()
                };
                if let Some(context)=&state.reserved_context {
                    let code=DiagnosticCode::new(code).expect("closed encoding code");
                    let failure=context.view().source_error_with_facts(&error,diagnostic,code,output,
                        saddle_observability::root_diagnostic::RootRequestEvent::Response,
                        saddle_observability::root_diagnostic::RootOutcomeFacts{axes,..Default::default()});
                    let mut slot=state.encoding_failure.lock().unwrap();
                    assert!(slot.is_none(),"one response encoding per request");
                    *slot=Some(failure);
                    return crate::process::process_dispatch_error();
                }
                let request_scope = match output {
                    Some(output) => saddle_observability::RequestDiagnosticScope::established(
                        output, context.0, context.1,
                    ),
                    None => saddle_observability::RequestDiagnosticScope::output_unavailable(
                        context.0, context.1,
                    ),
                };
                let request_scope = match zone {
                    Ok(zone) => request_scope.with_zone(zone),
                    Err(reason) => {
                        let code = match reason {
                            saddle_observability::DiagnosticZoneError::Empty => {
                                "saddle.diagnostic.zone_empty"
                            }
                            saddle_observability::DiagnosticZoneError::TooLong => {
                                "saddle.diagnostic.zone_too_long"
                            }
                            saddle_observability::DiagnosticZoneError::Unsafe => {
                                "saddle.diagnostic.zone_unsafe"
                            }
                        };
                        diagnostic = diagnostic.wrap(BoundedDiagnosticCause::new(
                            DiagnosticStage::RequestResponse,
                            DiagnosticCode::new(code).expect("closed diagnostic code"),
                        ));
                        request_scope.with_zone_missing(
                            saddle_observability::DiagnosticContextMissing::Unavailable,
                        )
                    }
                };
                let reference = request_scope.capture_required(diagnostic).into_reference();
                if let Some(output) = output {
                    let _boundary_submission = reference.record(output, &axes);
                }
                crate::process::process_dispatch_error()
            })
        };
        if let Some(scope) = scope {
            let mut guard = ConstructionScope {
                scope: Some(scope),
                state: self.state.clone(),
            };
            if let Some(context)=self.state.reserved_context.as_ref() {
                use saddle_runtime::profusegw::{ProfuseGwReservedScopeFailure as E,ProfuseGwScopeStop};
                let construction=guard.scope.as_mut().unwrap().parameter_construction()
                    .map_err(|_|crate::process::process_dispatch_error())?;
                let prepared=construction.prepare_reserved(context);
                let result=match prepared {
                    Ok((supervisor,_observation))=> {
                        let mut encode=std::pin::pin!(encode);
                        let outcome=supervisor.supervise(output,encode.as_mut()).await;
                        let facts=outcome.completion.supervision();
                        if facts.panicked || facts.stop.is_some() {
                            *self.state.encoding_supervision.lock().unwrap()=Some(if facts.panicked {saddle_core::OperationOutcome::Panicked}
                                else if facts.stop==Some(ProfuseGwScopeStop::TimedOut){saddle_core::OperationOutcome::TimedOut}else{saddle_core::OperationOutcome::Cancelled});
                        }
                        outcome.result
                    }
                    Err(error)=>Err(error),
                };
                return match result {
                    Ok(encoded)=>encoded,
                    Err(error)=> {
                        // A completed R invocation retains its original source
                        // until matching join. Only a pre-supervision rejection
                        // has no source and must be captured here, before mapping.
                        if self.state.encoding_supervision.lock().unwrap().is_none() {
                            struct Description<'a>(&'a E);
                            impl std::fmt::Debug for Description<'_> {fn fmt(&self,f:&mut std::fmt::Formatter<'_>)->std::fmt::Result {std::fmt::Display::fmt(self,f)}}
                            impl std::fmt::Display for Description<'_> {fn fmt(&self,f:&mut std::fmt::Formatter<'_>)->std::fmt::Result {match self.0 {E::Preparation(reason)=>write!(f,"Preparation({reason:?})"),E::Execution(reason)=>write!(f,"Execution({reason:?})")}}}
                            let code=saddle_core::DiagnosticCode::new("service.response.preparation").unwrap();
                            let failure=context.view().source_description(&Description(&error),saddle_core::BoundedDiagnostic::capture(
                                saddle_core::DiagnosticCategory::UnexpectedError,saddle_core::CaptureSite::FirstObserved,saddle_core::BoundedDiagnosticCause::new(saddle_core::DiagnosticStage::RequestResponse,code)),code,output,
                                saddle_observability::root_diagnostic::RootRequestEvent::Response,Default::default());
                            *self.state.encoding_failure.lock().unwrap()=Some(failure);
                        }
                        let _original=error;
                        Err(crate::process::process_dispatch_error())
                    }
                };
            }
            guard
                .scope
                .as_mut()
                .unwrap()
                .supervise_between(encode)
                .await
                .map_err(|_| crate::process::process_dispatch_error())?
        } else {
            encode.await
        }
    }
}

#[doc(hidden)]
pub struct HandlerObservation {
    view:saddle_runtime::request_task::reserved::ReservedRequestView,
    observer:saddle_observability::Observer,
    output:Option<saddle_observability::EmergencyDiagnosticHandle>,
}
impl HandlerObservation {
    pub fn start(&self)->saddle_runtime::request_task::reserved::ReservedActiveStage<'_> {
        self.view.start_stage(&self.observer,self.output.as_ref(),saddle_runtime::request_task::reserved::ReservedObservationStage::Handler)
    }
}

// Construction never enters SQL. Restore the same scope even when the borrowing
// future is dropped by the outer request deadline/disconnect select.
struct ConstructionScope {
    scope: Option<SerialScope>,
    state: Arc<RequestState>,
}
impl Drop for ConstructionScope {
    fn drop(&mut self) {
        if let Some(scope) = self.scope.take() {
            *self.state.terminal.lock().unwrap() = RequestTerminal::Serial(scope);
            self.state.completed.notify_one();
        }
    }
}
macro_rules! parameter_constructor {
    ($name:ident,$input:ty,$output:ty,$factory:ident,$convert:expr) => {
        pub async fn $name(
            &mut self,
            source: $input,
        ) -> Result<$output, saddle_db::internal::ParameterConstructionError> {
            use saddle_db::internal::{DatabaseParameterFactory, ParameterConstructionError};
            if self.state.reserved_failure.lock().unwrap().is_some() {
                // The original receipt remains for matching cleanup; do not
                // overwrite it by attempting a second parameter operation.
                return Err(ParameterConstructionError::Scope(saddle_runtime::profusegw::ProfuseGwScopeFailure::ScopeAlreadyEntered));
            }
            let mut scope = self.construction_scope()?;
            let construction = scope
                .scope
                .as_mut()
                .unwrap()
                .parameter_construction()
                .map_err(ParameterConstructionError::Scope)?;
            if let Some(context)=self.state.reserved_context.as_ref() {
                let process=self.process.as_ref().ok_or(ParameterConstructionError::Scope(
                    saddle_runtime::profusegw::ProfuseGwScopeFailure::ScopeAlreadyEntered))?;
                return match process.parameter_factory_reserved(&construction,context).$factory(source).await {
                    Ok(result)=> {
                        if let Some(failure)=result.unretained {
                            let mut slot=self.state.reserved_failure.lock().unwrap();
                            assert!(slot.is_none(),"original parameter failure must be consumed before another operation");
                            *slot=Some(failure);
                        }
                        result.value.map($convert)
                    }
                    Err(saddle_runtime::profusegw::ProfuseGwReservedScopeFailure::Execution(reason))=>Err(ParameterConstructionError::Scope(reason)),
                    Err(saddle_runtime::profusegw::ProfuseGwReservedScopeFailure::Preparation(reason))=> {
                        use saddle_core::{BoundedDiagnostic,BoundedDiagnosticCause,CaptureSite,DiagnosticCategory,DiagnosticCode,DiagnosticStage};
                        #[derive(Debug)]
                        struct Description<'a>(&'a saddle_runtime::request_task::reserved::ReservedContextError);
                        impl std::fmt::Display for Description<'_> {fn fmt(&self,f:&mut std::fmt::Formatter<'_>)->std::fmt::Result {write!(f,"{:?}",self.0)}}
                        let code=DiagnosticCode::new("service.parameters.preparation").unwrap();
                        let failure=context.view().source_description(&Description(&reason),
                            BoundedDiagnostic::capture(DiagnosticCategory::UnexpectedError,CaptureSite::FirstObserved,BoundedDiagnosticCause::new(DiagnosticStage::RequestDb,code)),code,
                            self.diagnostic_handle.as_ref(),saddle_observability::root_diagnostic::RootRequestEvent::Database,Default::default());
                        let mut slot=self.state.reserved_failure.lock().unwrap();
                        assert!(slot.is_none(),"preserve original preparation failure");
                        *slot=Some(failure);
                        // Parameter construction does not enter SQL; the outer
                        // guard restores the original unused scope on every exit.
                        Err(match reason {
                            saddle_runtime::request_task::reserved::ReservedContextError::Storage(error)=>ParameterConstructionError::Memory(error),
                            saddle_runtime::request_task::reserved::ReservedContextError::Context(_)=>ParameterConstructionError::Scope(saddle_runtime::profusegw::ProfuseGwScopeFailure::ScopeAlreadyEntered),
                        })
                    }
                };
            }
            DatabaseParameterFactory::new(&construction)
                .$factory(source)
                .await
                .map($convert)
        }
    };
}
impl DatabaseRequest {
    pub(crate) fn capture_dispatch_error(&self,error:&(dyn std::error::Error+'static)) {
        if let Some(context)=&self.state.reserved_context {
            let source=context.view().source_error(error,self.diagnostic_handle.as_ref(),saddle_core::DiagnosticStage::RequestDecode,
                saddle_observability::root_diagnostic::RootRequestEvent::Ingress,Default::default());
            let mut slot=self.state.encoding_failure.lock().unwrap();
            assert!(slot.is_none(),"one original dispatch failure");
            *slot=Some(source);
        }
    }
    pub(crate) fn shared(&self)->Self {
        Self {diagnostic_zone:self.diagnostic_zone,process:self.process.clone(),context:self.context.clone(),event:self.event.clone(),state:self.state.clone(),diagnostic_handle:self.diagnostic_handle.clone()}
    }
    pub(crate) fn reserved_context(&self)->Option<Arc<saddle_runtime::request_task::reserved::ReservedTaskContext>> {
        self.state.reserved_context.clone()
    }
    pub(crate) fn reserved_observer(&self)->Option<&saddle_observability::Observer> {self.state.reserved_observer.as_ref()}
    pub(crate) async fn supervise_external_concrete<T,F:Future<Output=T>+Send>(&mut self,future:F)->Result<saddle_runtime::profusegw::ProfuseGwReservedScopeOutcome<T>,saddle_runtime::profusegw::ProfuseGwReservedScopeFailure> {
        use saddle_runtime::profusegw::{ProfuseGwReservedScopeFailure as E,ProfuseGwScopeFailure};
        let context=self.reserved_context().expect("formal call carries original context");
        let output=self.diagnostic_handle.clone();
        let mut guard=self.construction_scope().map_err(|_|E::Execution(ProfuseGwScopeFailure::ScopeAlreadyEntered))?;
        let construction=guard.scope.as_mut().unwrap().parameter_construction().map_err(E::Execution)?;
        let (supervisor,_observation)=construction.prepare_reserved(&context)?;
        // The concrete F is already inline in the measured entry body. R owns
        // only a pinned borrow here; its actual pointer heap is reserved by R.
        let mut future=std::pin::pin!(future);
        Ok(supervisor.supervise(output.as_ref(),future.as_mut()).await)
    }
    /// Same ingress/task facts for downstream X integration. This projection
    /// carries no execution owner; the existing output is borrowed separately.
    pub(crate) fn request_diagnostic_scope(
        &self,
    ) -> saddle_observability::RequestDiagnosticScope<'_> {
        match &self.state.legacy {
            Some(legacy) => legacy
                .request_scope
                .reborrow()
                .with_output(self.diagnostic_handle.as_ref()),
            // Compatibility read-only projection, constructed only when an old
            // caller requests it. It is never authority for reserved execution.
            None => saddle_observability::RequestDiagnosticScope::output_unavailable(
                &self.context,
                &self.event,
            )
            .with_output(self.diagnostic_handle.as_ref()),
        }
    }
    pub(crate) fn outbound_retentions(&self) -> crate::programming::AttemptRetentions {
        Arc::clone(&self.state.outbound_retentions)
    }

    pub(crate) fn diagnostic_output(
        &self,
    ) -> Option<saddle_observability::EmergencyDiagnosticHandle> {
        self.diagnostic_handle.clone()
    }
    pub fn parameter_factory(&mut self) -> crate::database_uid::ParameterFactory<'_> {
        crate::database_uid::ParameterFactory { request: self }
    }
    pub fn response_encoding_context(&self) -> ResponseEncodingContext {
        ResponseEncodingContext {
            diagnostic_zone: self.diagnostic_zone,
            state: Arc::clone(&self.state),
            context: self.context.clone(),
            event: self.event.clone(),
            diagnostic_handle: self.diagnostic_handle.clone(),
        }
    }
    fn construction_scope(
        &mut self,
    ) -> Result<ConstructionScope, saddle_db::internal::ParameterConstructionError> {
        use saddle_db::internal::ParameterConstructionError;
        let mut terminal = self.state.terminal.lock().unwrap();
        let scope = match std::mem::replace(&mut *terminal, RequestTerminal::InFlight) {
            RequestTerminal::Untouched(dispatch) => {
                dispatch.into_parameter_scope(Arc::clone(&self.state.cancel).notified_owned())
            }
            RequestTerminal::Serial(scope) => scope,
            other => {
                *terminal = other;
                return Err(ParameterConstructionError::Scope(
                    saddle_runtime::profusegw::ProfuseGwScopeFailure::ScopeAlreadyEntered,
                ));
            }
        };
        Ok(ConstructionScope {
            scope: Some(scope),
            state: Arc::clone(&self.state),
        })
    }
    parameter_constructor!(parameter_text, &str, crate::database::Text, text, |v| v);
    parameter_constructor!(parameter_json, &str, crate::database::Json, json, |v| v);
    parameter_constructor!(
        parameter_i8s,
        &[i8],
        crate::database::List<i8>,
        i8s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_i8s,
        &[Option<i8>],
        crate::database::List<Option<i8>>,
        optional_i8s,
        Into::into
    );
    parameter_constructor!(
        parameter_i16s,
        &[i16],
        crate::database::List<i16>,
        i16s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_i16s,
        &[Option<i16>],
        crate::database::List<Option<i16>>,
        optional_i16s,
        Into::into
    );
    parameter_constructor!(
        parameter_i32s,
        &[i32],
        crate::database::List<i32>,
        i32s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_i32s,
        &[Option<i32>],
        crate::database::List<Option<i32>>,
        optional_i32s,
        Into::into
    );
    parameter_constructor!(
        parameter_i64s,
        &[i64],
        crate::database::List<i64>,
        i64s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_i64s,
        &[Option<i64>],
        crate::database::List<Option<i64>>,
        optional_i64s,
        Into::into
    );
    parameter_constructor!(
        parameter_u8s,
        &[u8],
        crate::database::List<u8>,
        u8s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_u8s,
        &[Option<u8>],
        crate::database::List<Option<u8>>,
        optional_u8s,
        Into::into
    );
    parameter_constructor!(
        parameter_u16s,
        &[u16],
        crate::database::List<u16>,
        u16s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_u16s,
        &[Option<u16>],
        crate::database::List<Option<u16>>,
        optional_u16s,
        Into::into
    );
    parameter_constructor!(
        parameter_u32s,
        &[u32],
        crate::database::List<u32>,
        u32s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_u32s,
        &[Option<u32>],
        crate::database::List<Option<u32>>,
        optional_u32s,
        Into::into
    );
    parameter_constructor!(
        parameter_u64s,
        &[u64],
        crate::database::List<u64>,
        u64s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_u64s,
        &[Option<u64>],
        crate::database::List<Option<u64>>,
        optional_u64s,
        Into::into
    );
    parameter_constructor!(
        parameter_bools,
        &[bool],
        crate::database::List<bool>,
        bools,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_bools,
        &[Option<bool>],
        crate::database::List<Option<bool>>,
        optional_bools,
        Into::into
    );
    parameter_constructor!(parameter_bytes, &[u8], crate::database::Bytes, bytes, |v| v);
    parameter_constructor!(
        parameter_texts,
        &[Option<&str>],
        crate::database::List<Option<crate::database::Text>>,
        texts,
        Into::into
    );
    parameter_constructor!(
        parameter_byte_list,
        &[Option<&[u8]>],
        crate::database::List<Option<crate::database::Bytes>>,
        byte_list,
        Into::into
    );
}

struct QueryTask<O: StaticQueryOptionalOperation> {
    request: SerialScope,
    process: Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>,
    context: CallContext,
    event: EventContext,
    state: Arc<RequestState>,
    proof: QueryOptionalOperationProof<O>,
    parameters: O::Parameters,
    sender: tokio::sync::oneshot::Sender<Result<ManagedOptionalRow<O::Row>, ScopeDatabaseError>>,
}

// Each kind keeps its actual future/result layout and original serial owner.
// A dropped caller cannot drop the physical finalizer task before resume.
macro_rules! owned_query_task {
    (@type $op:ident) => { saddle_db::internal::OwnedQueryInvocation<$op> };
    (@type $op:ident, $size:ident) => { saddle_db::internal::OwnedPageInvocation<$op> };
    (@invocation $proof:ident, $parameters:ident) => { Ok::<_, ScopeDatabaseError>($proof.invocation($parameters)) };
    (@invocation $proof:ident, $parameters:ident, $size:ident) => { match $size.1 { Some(offset)=>$proof.page_offset_invocation($parameters,$size.0,offset), None=>$proof.page_invocation($parameters,$size.0) } };
    ($task:ident, $method:ident, $process_method:ident, $reserved_method:ident, $result:ty, $layout:ident $(, $size:ident)?) => {
        struct $task<O: saddle_db::internal::StaticOwnedQuery> {
            request: SerialScope,
            process: Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>,
            context: CallContext,
            event: EventContext,
            state: Arc<RequestState>,
            invocation: owned_query_task!(@type O $(,$size)?),
            sender: tokio::sync::oneshot::Sender<Result<$result, saddle_db::internal::OwnedExecutionError<O::Parameters>>>,
        }
        impl<O: saddle_db::internal::StaticOwnedQuery> $task<O> {
            async fn run(self) {
                let mut owner_guard = AbortOnLostRequestOwner(true);
                let suspended = self
                    .process
                    .$process_method(self.request, &self.context, self.event, self.invocation)
                    .await;
                DatabaseRequest::publish_ordinary(suspended, &self.state, self.sender).await;
                owner_guard.0 = false;
            }
        }
        pub(crate) fn $layout<O: saddle_db::internal::StaticOwnedQuery, C>()
        -> [std::alloc::Layout; 4] {
            [
                future_layout($task::<O>::run),
                std::alloc::Layout::new::<Result<$result, saddle_db::internal::OwnedExecutionError<O::Parameters>>>(),
                std::alloc::Layout::new::<RequestState>(),
                std::alloc::Layout::new::<C>(),
            ]
        }
        impl DatabaseRequest {
            pub async fn $method<O: saddle_db::internal::StaticOwnedQuery>(
                &mut self,
                parameters: O::Parameters,
                $($size: (usize, Option<u64>),)?
            ) -> Result<$result, saddle_db::internal::OwnedExecutionError<O::Parameters>> {
                let proof = saddle_db::internal::OwnedQueryProof::<O>::bind()?;
                let invocation = owned_query_task!(@invocation proof, parameters $(,$size)?)?;
                let process = self
                    .process
                    .as_ref()
                    .cloned()
                    .ok_or(ScopeDatabaseError::State)?;
                if self.state.reserved_context.is_some() {
                    let (context,mut owner)=self.reserved_owner()?;
                    let result=process.$reserved_method(&mut owner.serial,&context,invocation).await;
                    return owner.resolve(result,&context).await?;
                }
                let request = self.begin_serial()?;
                let (sender, receiver) = tokio::sync::oneshot::channel();
                tokio::spawn(
                    $task {
                        request,
                        process,
                        context: self.context.clone(),
                        event: self.event.clone(),
                        state: Arc::clone(&self.state),
                        invocation,
                        sender,
                    }
                    .run(),
                );
                receiver.await.unwrap_or_else(|_| std::process::abort())
            }
        }
    };
}
owned_query_task!(
    OwnedPageTask,
    owned_page,
    owned_page_scope,
    owned_page_scope_reserved,
    saddle_db::internal::OwnedDbPage<O::Row>,
    owned_page_memory_layout,
    page_size
);
owned_query_task!(
    OwnedRowsTask,
    owned_rows,
    owned_rows_scope,
    owned_rows_scope_reserved,
    saddle_db::internal::OwnedDbRows<O::Row>,
    owned_rows_memory_layout
);
owned_query_task!(
    OwnedProbeTask,
    owned_probe,
    owned_probe_scope,
    owned_probe_scope_reserved,
    saddle_db::internal::OwnedDbProbe<O::Row>,
    owned_probe_memory_layout
);
owned_query_task!(
    OwnedOptionalTask,
    owned_optional,
    owned_optional_scope,
    owned_optional_scope_reserved,
    Option<O::Row>,
    owned_optional_memory_layout
);

struct OwnedWriteTask<O: saddle_db::internal::StaticOwnedWrite> {
    request: SerialScope,
    process: Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>,
    context: CallContext,
    event: EventContext,
    state: Arc<RequestState>,
    invocation: saddle_db::internal::OwnedWriteInvocation<O>,
    sender: tokio::sync::oneshot::Sender<
        Result<ManagedWriteResult, saddle_db::internal::OwnedExecutionError<O::Parameters>>,
    >,
}
impl<O: saddle_db::internal::StaticOwnedWrite> OwnedWriteTask<O> {
    async fn run(self) {
        let mut guard = AbortOnLostRequestOwner(true);
        let suspended = self
            .process
            .owned_write_scope(self.request, &self.context, self.event, self.invocation)
            .await;
        DatabaseRequest::publish_ordinary(suspended, &self.state, self.sender).await;
        guard.0 = false;
    }
}
pub(crate) fn owned_write_memory_layout<O: saddle_db::internal::StaticOwnedWrite, C>()
-> [std::alloc::Layout; 4] {
    [
        future_layout(OwnedWriteTask::<O>::run),
        std::alloc::Layout::new::<
            Result<ManagedWriteResult, saddle_db::internal::OwnedExecutionError<O::Parameters>>,
        >(),
        std::alloc::Layout::new::<RequestState>(),
        std::alloc::Layout::new::<C>(),
    ]
}
#[doc(hidden)]
pub fn named_write_failure<O: GeneratedOwnedWrite>(
    error: saddle_db::internal::OwnedExecutionError<O::Parameters>,
) -> QueryFailure<O::NamedParameters> {
    match error {
        saddle_db::internal::OwnedExecutionError::Parameters { parameters, .. } => {
            QueryFailure::Parameters {
                parameters: O::restore_parameters(parameters),
            }
        }
        saddle_db::internal::OwnedExecutionError::Database(e) => QueryFailure::Database(e),
    }
}
impl DatabaseRequest {
    pub async fn named_write<O: GeneratedOwnedWrite>(
        &mut self,
        parameters: O::NamedParameters,
    ) -> Result<ManagedWriteResult, QueryFailure<O::NamedParameters>> {
        if let Err(error) = O::validate_parameters(&parameters, &self.target()) {
            return Err(QueryFailure::Target { parameters, error });
        }
        let proof =
            saddle_db::internal::OwnedWriteProof::<O>::bind().map_err(QueryFailure::Database)?;
        let process = self
            .process
            .as_ref()
            .cloned()
            .ok_or(QueryFailure::Database(ScopeDatabaseError::State))?;
        if self.state.reserved_context.is_some() {
            let (context,mut owner)=self.reserved_owner().map_err(QueryFailure::Database)?;
            let result=process.owned_write_scope_reserved(&mut owner.serial,&context,proof.invocation(O::parameters(parameters))).await;
            return owner.resolve(result,&context).await.map_err(QueryFailure::Database)?.map_err(named_write_failure::<O>);
        }
        let request = self.begin_serial().map_err(QueryFailure::Database)?;
        let (sender, receiver) = tokio::sync::oneshot::channel();
        tokio::spawn(
            OwnedWriteTask {
                request,
                process,
                context: self.context.clone(),
                event: self.event.clone(),
                state: Arc::clone(&self.state),
                invocation: proof.invocation(O::parameters(parameters)),
                sender,
            }
            .run(),
        );
        receiver
            .await
            .unwrap_or_else(|_| std::process::abort())
            .map_err(named_write_failure::<O>)
    }
}

impl<O: StaticQueryOptionalOperation> QueryTask<O> {
    async fn run(self) {
        let Self {
            request,
            context,
            event,
            state,
            process,
            proof,
            parameters,
            sender,
        } = self;
        let mut owner_guard = AbortOnLostRequestOwner(true);
        let suspended = process
            .query_scope(request, &context, event, proof.invocation(parameters))
            .await;
        DatabaseRequest::publish_ordinary(suspended, &state, sender).await;
        owner_guard.0 = false;
    }
}

struct WriteTask<O: StaticWriteOperation> {
    request: SerialScope,
    process: Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>,
    context: CallContext,
    event: EventContext,
    state: Arc<RequestState>,
    proof: WriteOperationProof<O>,
    parameters: O::Parameters,
    sender: tokio::sync::oneshot::Sender<Result<ManagedWriteResult, ScopeDatabaseError>>,
}

impl<O: StaticWriteOperation> WriteTask<O> {
    async fn run(self) {
        let Self {
            request,
            context,
            event,
            state,
            process,
            proof,
            parameters,
            sender,
        } = self;
        let mut owner_guard = AbortOnLostRequestOwner(true);
        let suspended = process
            .write_scope(request, &context, event, proof.invocation(parameters))
            .await;
        DatabaseRequest::publish_ordinary(suspended, &state, sender).await;
        owner_guard.0 = false;
    }
}

fn future_layout<A, F: std::future::Future>(_: fn(A) -> F) -> std::alloc::Layout {
    // Type inspection only: no fake values, allocation, construction or poll.
    std::alloc::Layout::new::<F>()
}

pub(crate) fn query_memory_layout<O: StaticQueryOptionalOperation, C>() -> [std::alloc::Layout; 4] {
    [
        future_layout(QueryTask::<O>::run),
        std::alloc::Layout::new::<Result<ManagedOptionalRow<O::Row>, ScopeDatabaseError>>(),
        std::alloc::Layout::new::<RequestState>(),
        std::alloc::Layout::new::<(Result<ManagedOptionalRow<O::Row>, ScopeDatabaseError>, C)>(),
    ]
}

pub(crate) fn write_memory_layout<O: StaticWriteOperation, C>() -> [std::alloc::Layout; 4] {
    [
        future_layout(WriteTask::<O>::run),
        std::alloc::Layout::new::<Result<ManagedWriteResult, ScopeDatabaseError>>(),
        std::alloc::Layout::new::<RequestState>(),
        std::alloc::Layout::new::<(Result<ManagedWriteResult, ScopeDatabaseError>, C)>(),
    ]
}

/// Known persistent framework storage for a borrowed transaction route.
/// Not a bound on the generic business callback, spawned Future, result T/E,
/// executor headers or driver allocations. In particular, do not substitute
/// the unrelated standalone QueryTask/WriteTask Future as scope evidence.
pub(crate) fn combined_scope_layout<C>(
    ordinary: [std::alloc::Layout; 4],
) -> [std::alloc::Layout; 4] {
    let scoped = scoped_memory_layout::<C>();
    std::array::from_fn(|index| {
        ordinary[index]
            .extend(scoped[index])
            .expect("concrete execution layouts fit")
            .0
            .pad_to_align()
    })
}

pub(crate) fn scoped_memory_layout<C>() -> [std::alloc::Layout; 4] {
    [
        std::alloc::Layout::new::<SerialScope>(),
        std::alloc::Layout::new::<RequestState>(),
        std::alloc::Layout::new::<DatabaseRequest>(),
        std::alloc::Layout::new::<C>(),
    ]
}

impl DatabaseRequest {
    pub fn target(&self) -> DatabaseTarget {
        DatabaseTarget(self.process.clone())
    }
    pub async fn named_page<O: GeneratedPageQuery>(
        &mut self,
        mut request: crate::database_rows::PageRequest<O>,
    ) -> Result<crate::database_rows::Page<O>, QueryFailure<crate::database_rows::PageRequest<O>>>
    {
        if let Err(error) = O::validate_parameters(&request.parameters, &self.target()) {
            return Err(QueryFailure::Target {
                parameters: request,
                error,
            });
        }
        let parameters =
            match O::page_parameters(&mut request.parameters, request.after, self).await {
                Ok(parameters) => parameters,
                Err(_) => {
                    return Err(QueryFailure::Parameters {
                        parameters: request,
                    });
                }
            };
        let page = match self
            .owned_page::<O>(parameters, (request.size, request.offset))
            .await
        {
            Ok(page) => page,
            Err(saddle_db::internal::OwnedExecutionError::Parameters { .. }) => {
                return Err(QueryFailure::Parameters {
                    parameters: request,
                });
            }
            Err(saddle_db::internal::OwnedExecutionError::Database(e)) => {
                return Err(QueryFailure::Database(e));
            }
        };
        let next = if page.has_more() {
            page.rows().last().map(O::page_key)
        } else {
            None
        };
        let (rows, _) = page.into_parts();
        Ok(crate::database_rows::Page {
            rows: crate::database_rows::Rows::from_database(rows),
            next: next.map(|key| crate::database_rows::PageRequest {
                after: Some(key),
                offset: request.offset.map(|offset| offset + request.size as u64),
                ..request
            }),
        })
    }
    pub async fn named_rows<O: GeneratedOwnedQuery>(
        &mut self,
        parameters: O::NamedParameters,
    ) -> Result<crate::database::Rows<O::Row, O::NamedRow>, QueryFailure<O::NamedParameters>> {
        if let Err(error) = O::validate_parameters(&parameters, &self.target()) {
            return Err(QueryFailure::Target { parameters, error });
        }
        self.owned_rows::<O>(O::parameters(parameters))
            .await
            .map(crate::database_rows::Rows::from_database)
            .map_err(named_failure::<O>)
    }

    pub async fn named_optional<O: GeneratedOwnedQuery>(
        &mut self,
        parameters: O::NamedParameters,
    ) -> Result<Option<O::NamedRow>, QueryFailure<O::NamedParameters>> {
        if let Err(error) = O::validate_parameters(&parameters, &self.target()) {
            return Err(QueryFailure::Target { parameters, error });
        }
        self.owned_optional::<O>(O::parameters(parameters))
            .await
            .map(|row| row.map(O::NamedRow::from))
            .map_err(named_failure::<O>)
    }

    pub async fn named_probe<O: GeneratedOwnedQuery>(
        &mut self,
        parameters: O::NamedParameters,
    ) -> Result<crate::database::MatchCardinality<O::NamedRow>, QueryFailure<O::NamedParameters>>
    {
        use crate::database::MatchCardinality;
        use saddle_db::internal::OwnedDbProbe;
        if let Err(error) = O::validate_parameters(&parameters, &self.target()) {
            return Err(QueryFailure::Target { parameters, error });
        }
        self.owned_probe::<O>(O::parameters(parameters))
            .await
            .map(|result| match result {
                OwnedDbProbe::Zero => MatchCardinality::Zero,
                OwnedDbProbe::One(row) => MatchCardinality::One(O::NamedRow::from(row)),
                OwnedDbProbe::Many => MatchCardinality::Many,
            })
            .map_err(named_failure::<O>)
    }

    pub(crate) fn supervise_external<T: Send + 'static>(
        &self,
        future: std::pin::Pin<
            Box<
                dyn std::future::Future<Output = crate::programming::ExternalFunctionResult<T>>
                    + Send,
            >,
        >,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = crate::programming::ExternalFunctionResult<T>> + Send>,
    > {
        let state = Arc::clone(&self.state);
        let diagnostic_scope = self.request_diagnostic_scope().with_output(None);
        let output = self.diagnostic_output();
        Box::pin(async move {
            use crate::programming::{
                ExecutionCertainty, ExternalFunctionResult, TechnicalFailure, TechnicalFailureCode,
            };
            let stopped = |code, certainty| {
                ExternalFunctionResult::TechnicalFailure(
                    TechnicalFailure::capture(
                        diagnostic_scope.reborrow().with_output(output.as_ref()),
                        code,
                        certainty,
                        "framework.external_supervision_failed",
                    )
                    .finish(output.as_ref()),
                )
            };
            let scope = {
                let mut terminal = state.terminal.lock().unwrap();
                match std::mem::replace(&mut *terminal, RequestTerminal::InFlight) {
                    RequestTerminal::Serial(scope) => Some(scope),
                    other => {
                        let untouched = matches!(other, RequestTerminal::Untouched(_));
                        *terminal = other;
                        if !untouched {
                            return stopped(
                                TechnicalFailureCode::InternalFailure,
                                ExecutionCertainty::NotExecuted,
                            );
                        }
                        None
                    }
                }
            };
            let Some(mut scope) = scope else {
                return future.await;
            };
            let (sender, receiver) = tokio::sync::oneshot::channel();
            tokio::spawn(async move {
                let mut guard = AbortOnLostRequestOwner(true);
                let result = scope.supervise_between(future).await;
                *state.terminal.lock().unwrap() = RequestTerminal::Serial(scope);
                state.completed.notify_one();
                let _ = sender.send(result);
                guard.0 = false;
            });
            match receiver.await {
                Ok(Ok(result)) => result,
                Ok(Err(failure)) => {
                    use saddle_runtime::profusegw::{ProfuseGwScopeFailure, ProfuseGwScopeStop};
                    let code = match failure {
                        ProfuseGwScopeFailure::Stopped(ProfuseGwScopeStop::TimedOut) => {
                            TechnicalFailureCode::DeadlineExceeded
                        }
                        ProfuseGwScopeFailure::Stopped(ProfuseGwScopeStop::Cancelled) => {
                            TechnicalFailureCode::DependencyUnavailable
                        }
                        ProfuseGwScopeFailure::Panicked
                        | ProfuseGwScopeFailure::AlreadySupervised
                        | ProfuseGwScopeFailure::ScopeAlreadyEntered => {
                            TechnicalFailureCode::InternalFailure
                        }
                    };
                    stopped(code, ExecutionCertainty::MayHaveExecuted)
                }
                Err(_) => std::process::abort(),
            }
        })
    }

    pub async fn scoped_transaction<T, E, B>(
        &mut self,
        isolation: saddle_db::internal::TransactionIsolation,
        body: B,
    ) -> saddle_db::internal::ScopeTransactionOutcome<T, E>
    where
        T: Send + 'static,
        E: Send + 'static,
        B: for<'tx, 'session, 'request> FnOnce(
                &'tx mut TransactionSession<'session, 'request>,
            )
                -> saddle_db::internal::ScopeTransactionFuture<
                'tx,
                T,
                E,
            > + Send
            + 'static,
    {
        use saddle_db::internal::{
            ScopeDatabaseError, ScopeTransactionAbort, ScopeTransactionOutcome,
        };
        let rejected = || {
            ScopeTransactionOutcome::Rejected(ScopeTransactionAbort::Technical(
                ScopeDatabaseError::Unavailable,
            ))
        };
        let Some(process) = self.process.as_ref().cloned() else {
            return rejected();
        };
        let scope = {
            let mut state = self.state.terminal.lock().unwrap();
            match std::mem::replace(&mut *state, RequestTerminal::InFlight) {
                RequestTerminal::Untouched(dispatch) => dispatch
                    .into_database_request()
                    .into_serial_scope(Arc::clone(&self.state.cancel).notified_owned()),
                RequestTerminal::Serial(scope) => scope,
                other => {
                    *state = other;
                    return rejected();
                }
            }
        };
        let state = Arc::clone(&self.state);
        let context = self.context.clone();
        let event = self.event.clone();
        let (sender, receiver) = tokio::sync::oneshot::channel();
        // Request-loop cancellation never aborts this owning finalizer task.
        tokio::spawn(async move {
            let mut guard = AbortOnLostRequestOwner(true);
            let mut suspended = process
                .transaction_scope_with_isolation(scope, &context, event, isolation, body)
                .await;
            let (terminal, result) = match suspended.resume().await {
                Ok((next, result)) => (RequestTerminal::Serial(next), result),
                Err(_) => {
                    let (result, terminal) = suspended
                        .into_response_parts()
                        .unwrap_or_else(|_| std::process::abort());
                    (RequestTerminal::PostDatabase(terminal), result)
                }
            };
            *state.terminal.lock().unwrap() = terminal;
            state.completed.notify_one();
            let _ = sender.send(result.outcome);
            guard.0 = false;
        });
        receiver.await.unwrap_or_else(|_| std::process::abort())
    }

    pub(crate) fn new(
        process: Option<Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>>,
        dispatch: ProfuseGwManagedDispatch,
        context: CallContext,
        event: EventContext,
        diagnostic_handle: Option<saddle_observability::EmergencyDiagnosticHandle>,
        handler: Option<saddle_observability::ActiveStage>,
        accepted_zone: &str,
        request_scope: saddle_observability::RequestDiagnosticScope<'static>,
        reserved_context: Option<Arc<saddle_runtime::request_task::reserved::ReservedTaskContext>>,
        reserved_observer: Option<saddle_observability::Observer>,
    ) -> (Self, DatabaseRequestCompletion) {
        let legacy = if reserved_context.is_none() {
            Some(Box::new(LegacyRequestDiagnostics {
                request_scope,
                handler: Mutex::new(handler),
            }))
        } else {
            debug_assert!(
                handler.is_none(),
                "reserved handler stage belongs to the original root"
            );
            None
        };
        let state = Arc::new(RequestState {
            legacy,
            reserved_context,
            reserved_failure: Mutex::new(None),
            encoding_failure:Mutex::new(None),
            encoding_supervision:Mutex::new(None),
            reserved_observer,
            handler_recorded:std::sync::atomic::AtomicBool::new(false),
            outbound_retentions: Arc::new(Mutex::new(Vec::new())),
            terminal: Mutex::new(RequestTerminal::Untouched(dispatch)),
            cancel: Arc::new(Notify::new()),
            completed: Notify::new(),
            diagnostic_handle: diagnostic_handle.clone(),
        });
        (
            Self {
                diagnostic_zone: saddle_observability::DiagnosticZone::from_validated_ingress(
                    accepted_zone,
                ),
                process,
                context,
                event,
                state: Arc::clone(&state),
                diagnostic_handle,
            },
            DatabaseRequestCompletion { state },
        )
    }

    fn begin_serial(&mut self) -> Result<SerialScope, ScopeDatabaseError> {
        if self.process.is_none() {
            return Err(ScopeDatabaseError::Unavailable);
        }
        let mut terminal = self.state.terminal.lock().unwrap();
        let scope = match std::mem::replace(&mut *terminal, RequestTerminal::InFlight) {
            RequestTerminal::Untouched(dispatch) => Ok(dispatch
                .into_database_request()
                .into_serial_scope(Arc::clone(&self.state.cancel).notified_owned())),
            RequestTerminal::Serial(scope) => Ok(scope),
            other => {
                *terminal = other;
                Err(ScopeDatabaseError::State)
            }
        }?;
        Ok(match self.diagnostic_zone {
            Ok(zone) => scope.with_diagnostic_zone(zone),
            Err(_) => scope,
        })
    }

    async fn publish_ordinary<T, E>(
        mut suspended: saddle_runtime::profusegw::ProfuseGwSuspendedScope<
            tokio::sync::futures::OwnedNotified,
            Result<T, E>,
        >,
        state: &RequestState,
        sender: tokio::sync::oneshot::Sender<Result<T, E>>,
    ) {
        let (terminal, value) = match suspended.resume().await {
            Ok((scope, value)) => (RequestTerminal::Serial(scope), value),
            Err(_) => {
                let (value, terminal) = suspended
                    .into_response_parts()
                    .unwrap_or_else(|_| std::process::abort());
                (RequestTerminal::PostDatabase(terminal), value)
            }
        };
        *state.terminal.lock().unwrap() = terminal;
        state.completed.notify_one();
        let _ = sender.send(value);
    }

    pub async fn query_optional<O>(
        &mut self,
        parameters: O::Parameters,
    ) -> Result<ManagedOptionalRow<O::Row>, ScopeDatabaseError>
    where
        O: StaticQueryOptionalOperation,
    {
        let proof =
            QueryOptionalOperationProof::<O>::bind().map_err(|_| ScopeDatabaseError::Mapping)?;
        if self.state.reserved_context.is_some() {
            let process=self.process.clone().ok_or(ScopeDatabaseError::Unavailable)?;
            let (context,mut owner)=self.reserved_owner()?;
            let result=process.query_scope_reserved(&mut owner.serial,&context,proof.invocation(parameters)).await;
            return owner.resolve(result,&context).await?;
        }
        let request = self.begin_serial()?;
        let process = self
            .process
            .as_ref()
            .cloned()
            .unwrap_or_else(|| std::process::abort());
        let context = self.context.clone();
        let event = self.event.clone();
        let state = Arc::clone(&self.state);
        let (sender, receiver) = tokio::sync::oneshot::channel();
        tokio::spawn(
            QueryTask {
                request,
                context,
                event,
                state,
                process,
                proof,
                parameters,
                sender,
            }
            .run(),
        );
        receiver.await.unwrap_or_else(|_| std::process::abort())
    }

    pub async fn write<O>(
        &mut self,
        parameters: O::Parameters,
    ) -> Result<ManagedWriteResult, ScopeDatabaseError>
    where
        O: StaticWriteOperation,
    {
        let proof = WriteOperationProof::<O>::bind().map_err(|_| ScopeDatabaseError::Mapping)?;
        if self.state.reserved_context.is_some() {
            let process=self.process.clone().ok_or(ScopeDatabaseError::Unavailable)?;
            let (context,mut owner)=self.reserved_owner()?;
            let result=process.write_scope_reserved(&mut owner.serial,&context,proof.invocation(parameters)).await;
            return owner.resolve(result,&context).await?;
        }
        let request = self.begin_serial()?;
        let process = self
            .process
            .as_ref()
            .cloned()
            .unwrap_or_else(|| std::process::abort());
        let context = self.context.clone();
        let event = self.event.clone();
        let state = Arc::clone(&self.state);
        let (sender, receiver) = tokio::sync::oneshot::channel();
        tokio::spawn(
            WriteTask {
                request,
                context,
                event,
                state,
                process,
                proof,
                parameters,
                sender,
            }
            .run(),
        );
        receiver.await.unwrap_or_else(|_| std::process::abort())
    }
}

impl DatabaseRequestCompletion {
    pub(crate) fn take_encoding_supervision(&self)->Option<saddle_core::OperationOutcome> {self.state.encoding_supervision.lock().unwrap().take()}
    pub(crate) fn take_encoding_failure(&self)->Option<saddle_runtime::request_task::reserved::ReservedRequestFailure> {
        self.state.encoding_failure.lock().unwrap().take()
    }
    pub(crate) fn take_reserved_failure(&self)->Option<saddle_runtime::request_task::reserved::ReservedRequestFailure> {
        self.state.reserved_failure.lock().unwrap().take()
    }
    pub(crate) fn finish_handler(
        &self,
        outcome: saddle_core::OperationOutcome,
        error: Option<&saddle_core::SaddleError>,
    ) {
        self.state.finish_handler(
            &saddle_core::DiagnosticOutcomeAxes {
                operation: outcome,
                ..Default::default()
            },
            error,
        );
    }
    pub(crate) async fn finish(&self, cancelled: bool) {
        if cancelled {
            self.state.cancel.notify_one();
        }
        loop {
            let terminal = {
                let mut state = self.state.terminal.lock().unwrap();
                match std::mem::replace(&mut *state, RequestTerminal::Finished) {
                    RequestTerminal::InFlight => {
                        *state = RequestTerminal::InFlight;
                        None
                    }
                    terminal => Some(terminal),
                }
            };
            match terminal {
                Some(RequestTerminal::Untouched(dispatch)) => {
                    if cancelled {
                        dispatch.cancel();
                    } else if finish_profusegw_without_database(dispatch, ()).is_err() {
                        std::process::abort();
                    }
                    return;
                }
                Some(RequestTerminal::PostDatabase(terminal)) => {
                    finish_profusegw_after_database(terminal);
                    return;
                }
                Some(RequestTerminal::Serial(scope)) => {
                    if scope.finish_unentered_response(()).is_err() {
                        std::process::abort();
                    }
                    return;
                }
                Some(RequestTerminal::Finished) => return,
                Some(RequestTerminal::InFlight) | None => self.state.completed.notified().await,
            }
        }
    }
}