saddle-runtime 0.3.23

//! Framework/Database composition only. No business-facing authority issuer.
use super::*;

pub(super) enum NextScope {
    Continuation(saddle_core::DbScopeContinuation),
    Pair(DbPhysicalRequestHalf, DbPhysicalExecutionHalf),
}

/// Same request and pinned cancellation source across all serial scopes.
/// No Clone, raw constructor, deadline reset or new request admission.
#[doc(hidden)]
pub struct ProfuseGwSerialScope<C> {
    lease: ProfuseGwConcreteDbRequestLease,
    cancel: Pin<Box<C>>,
    entry: ScopeEntry,
    diagnostic_zone: Option<saddle_observability::DiagnosticZone>,
}

enum ScopeEntry {
    Entered,
    ResumedUnused,
}

/// Construction-only borrow. No conversion to the SQL driver or physical
/// execution half is exposed. The request loop retains the unique whole.
///
/// ```compile_fail
/// fn cannot_enter<C>(borrow: saddle_runtime::profusegw::ProfuseGwParameterConstruction<'_, C>) {
///     let driver = borrow.driver;
/// }
/// ```
/// ```compile_fail
/// fn cannot_downgrade<C: std::future::Future<Output = ()>>(
///     lease: saddle_runtime::profusegw::ProfuseGwConcreteDbRequestLease, cancel: C,
/// ) { let unused = lease.into_parameter_scope(cancel); }
/// ```
#[doc(hidden)]
pub struct ProfuseGwParameterConstruction<'a, C> {
    driver: ProfuseGwScopeDriver<'a, C>,
}

impl ProfuseGwManagedDispatch {
    /// Before the first SQL, retain untouched authority while building owned
    /// parameters. A previously entered DB lease has no equivalent transition.
    #[doc(hidden)]
    pub fn into_parameter_scope<C: Future<Output = ()>>(
        self,
        cancel: C,
    ) -> ProfuseGwSerialScope<C> {
        ProfuseGwSerialScope {
            lease: self.into_database_request(),
            cancel: Box::pin(cancel),
            entry: ScopeEntry::ResumedUnused,
            diagnostic_zone: None,
        }
    }
}

impl<C: Future<Output = ()>> ProfuseGwParameterConstruction<'_, C> {
    #[doc(hidden)]
    pub async fn database_memory(
        &self,
    ) -> Result<saddle_admission::RequestMemory, ProfuseGwScopeStop> {
        self.driver.database_memory().await
    }

    /// Poll the DB-owned constructor under the original admitted allocation
    /// context. This borrow grants no connection or operation capability.
    #[doc(hidden)]
    pub async fn supervise<F: Future>(
        &self,
        constructor: F,
    ) -> Result<F::Output, ProfuseGwScopeFailure> {
        self.driver.supervise(constructor).await
    }
}

/// DB retains this beside its owning phase, not inside the borrowed body.
/// Observability's consumer can spell the exact Core type without Runtime.
#[doc(hidden)]
pub type ProfuseGwTransactionObservation =
    saddle_core::DbScopeObservation<(Observer, CallContext, EventContext)>;

#[derive(Debug, PartialEq, Eq)]
#[doc(hidden)]
pub enum ProfuseGwTransactionObservationError {
    MissingContext,
    Unavailable,
}

/// ```compile_fail
/// use saddle_runtime::profusegw::ProfuseGwSerialScope;
/// fn duplicate<C>(owner: ProfuseGwSerialScope<C>) { let _ = owner.clone(); }
/// ```
/// ```compile_fail
/// use saddle_runtime::profusegw::ProfuseGwSerialScope;
/// fn forge<C>() -> ProfuseGwSerialScope<C> { ProfuseGwSerialScope {} }
/// ```
/// ```compile_fail
/// use saddle_runtime::profusegw::ProfuseGwSerialScope;
/// use std::future::Future;
/// fn replay<C: Future<Output=()>>(owner: ProfuseGwSerialScope<C>) {
///     let _ = owner.into_physical_finalization();
///     let _ = owner.into_physical_finalization();
/// }
/// ```
impl<C> ProfuseGwSerialScope<C> {}

impl ProfuseGwConcreteDbRequestLease {
    /// Project the original request context and checked physical scope without
    /// entering an operation or consuming the transaction observation allowance.
    #[doc(hidden)]
    pub fn request_diagnostic_scope<'a>(
        &self,
        output: Option<&'a saddle_observability::EmergencyDiagnosticHandle>,
    ) -> Result<
        saddle_observability::RequestDiagnosticScope<'a>,
        ProfuseGwTransactionObservationError,
    > {
        let observation = self
            .observation
            .as_ref()
            .ok_or(ProfuseGwTransactionObservationError::MissingContext)?;
        let projection = self
            .db_request
            .project_diagnostic_context(
                &self.db_execution,
                (&observation.context, &observation.event_context),
            )
            .map_err(|_| ProfuseGwTransactionObservationError::Unavailable)?;
        Ok(saddle_observability::RequestDiagnosticScope::live_db_scope(
            output,
            &projection,
        ))
    }

    #[doc(hidden)]
    pub fn into_serial_scope<C: Future<Output = ()>>(self, cancel: C) -> ProfuseGwSerialScope<C> {
        ProfuseGwSerialScope {
            lease: self,
            cancel: Box::pin(cancel),
            entry: ScopeEntry::Entered,
            diagnostic_zone: None,
        }
    }
}

struct Control<'a, C> {
    deadline: &'a mut ProfuseGwManagedDeadline,
    stop: &'a mut Option<ProfuseGwScopeStop>,
    cancel: Pin<&'a mut C>,
    supervising: bool,
    observation: &'a Option<ProfuseGwRequestObservation>,
    physical_return: bool,
}

impl<C: Future<Output = ()>> Control<'_, C> {
    fn check(&mut self, cx: &mut Context<'_>) -> Result<(), ProfuseGwScopeStop> {
        if let Some(stop) = *self.stop {
            return Err(stop);
        }
        let stop = if self.cancel.as_mut().poll(cx).is_ready() {
            Some(ProfuseGwScopeStop::Cancelled)
        } else if tokio::time::Instant::now() >= self.deadline.timer.deadline()
            || self.deadline.timer.as_mut().poll(cx).is_ready()
        {
            Some(ProfuseGwScopeStop::TimedOut)
        } else {
            None
        };
        *self.stop = stop;
        if let Some(stop) = stop {
            record_request_stop(
                self.observation.as_ref(),
                stop,
                self.physical_return,
                if self.physical_return {
                    saddle_core::DiagnosticStage::FinalizerResource
                } else {
                    saddle_core::DiagnosticStage::RequestDb
                },
            );
        }
        stop.map_or(Ok(()), Err)
    }
}

/// Borrowed supervisor and step checkpoint share one control, not a second
/// timer. A DB session may borrow this driver while its body is supervised.
#[doc(hidden)]
pub struct ProfuseGwScopeDriver<'a, C> {
    execution: &'a ProfuseGwLightweightExecutionOwner,
    db_request: &'a DbPhysicalRequestHalf,
    db_execution: &'a DbPhysicalExecutionHalf,
    diagnostic_zone: Option<saddle_observability::DiagnosticZone>,
    observation: &'a Option<ProfuseGwRequestObservation>,
    control: Mutex<Control<'a, C>>,
}

#[derive(Debug, PartialEq, Eq)]
#[doc(hidden)]
pub enum ProfuseGwScopeFailure {
    Stopped(ProfuseGwScopeStop),
    Panicked,
    AlreadySupervised,
    ScopeAlreadyEntered,
}

impl<C: Future<Output = ()>> ProfuseGwSerialScope<C> {
    /// Facade supplies the same accepted request's safe projection before each
    /// DB call, including a resumed serial scope. No execution state changes.
    #[doc(hidden)]
    pub fn with_diagnostic_zone(mut self, zone: saddle_observability::DiagnosticZone) -> Self {
        self.diagnostic_zone = Some(zone);
        self
    }
    /// Borrow only the original context and checked pair. Does not enter SQL
    /// or consume the one transaction-observation allowance.
    #[doc(hidden)]
    pub fn request_diagnostic_scope<'a>(
        &self,
        output: Option<&'a saddle_observability::EmergencyDiagnosticHandle>,
    ) -> Result<
        saddle_observability::RequestDiagnosticScope<'a>,
        ProfuseGwTransactionObservationError,
    > {
        let context = self.lease.request_diagnostic_scope(output)?;
        Ok(match self.diagnostic_zone {
            Some(zone) => context.with_zone(zone),
            None => context,
        })
    }

    /// No entry side effect. Failure returns with the whole still in place.
    #[doc(hidden)]
    pub fn parameter_construction(
        &mut self,
    ) -> Result<ProfuseGwParameterConstruction<'_, C>, ProfuseGwScopeFailure> {
        if matches!(self.entry, ScopeEntry::Entered) {
            return Err(ProfuseGwScopeFailure::ScopeAlreadyEntered);
        }
        Ok(ProfuseGwParameterConstruction {
            driver: self.borrow_driver(),
        })
    }
    /// Scope entry only: clones safe log context, never request/physical
    /// authority. Failure does not alter the timer, credits or physical path.
    #[doc(hidden)]
    pub fn take_transaction_observation(
        &mut self,
    ) -> Result<ProfuseGwTransactionObservation, ProfuseGwTransactionObservationError> {
        let observation = self
            .lease
            .observation
            .as_ref()
            .ok_or(ProfuseGwTransactionObservationError::MissingContext)?;
        let context = (
            observation.observer.clone(),
            observation.context.clone(),
            observation.event_context.clone(),
        );
        let correlation = self
            .lease
            .db_request
            .take_scope_observation(&self.lease.db_execution, context)
            .map_err(|_| ProfuseGwTransactionObservationError::Unavailable)?;
        self.entry = ScopeEntry::Entered;
        Ok(correlation)
    }

    #[doc(hidden)]
    pub fn driver(&mut self) -> ProfuseGwScopeDriver<'_, C> {
        self.entry = ScopeEntry::Entered;
        self.borrow_driver()
    }

    fn borrow_driver(&mut self) -> ProfuseGwScopeDriver<'_, C> {
        ProfuseGwScopeDriver {
            execution: &self.lease.execution,
            db_request: &self.lease.db_request,
            db_execution: &self.lease.db_execution,
            diagnostic_zone: self.diagnostic_zone,
            observation: &self.lease.observation,
            control: Mutex::new(Control {
                deadline: &mut self.lease.deadline,
                stop: &mut self.lease.scope_stop,
                cancel: self.cancel.as_mut(),
                supervising: false,
                observation: &self.lease.observation,
                physical_return: false,
            }),
        }
    }

    /// Between scopes, outbound/business awaits are still admitted and timed.
    /// This does not enter another DB scope or acquire a connection.
    #[doc(hidden)]
    pub async fn supervise_between<F: Future>(
        &mut self,
        future: F,
    ) -> Result<F::Output, ProfuseGwScopeFailure> {
        if matches!(self.entry, ScopeEntry::Entered) {
            return Err(ProfuseGwScopeFailure::ScopeAlreadyEntered);
        }
        self.borrow_driver().supervise(future).await
    }

    /// Final response before first SQL, or after resuming without entering
    /// another scope. The reserved scope is NotUsed; prior physical disposition/T is not
    /// replayed. Once driver() is requested, DB physical finalization is required.
    #[doc(hidden)]
    #[allow(
        clippy::result_large_err,
        reason = "return the exact linear whole and value without a new allocation"
    )]
    pub fn finish_unentered_response<T>(self, value: T) -> Result<T, (Self, T)> {
        if matches!(self.entry, ScopeEntry::Entered) {
            return Err((self, value));
        }
        let ProfuseGwConcreteDbRequestLease {
            execution,
            deadline,
            db_request,
            db_execution,
            observation,
            scope_stop,
        } = self.lease;
        match seal_db_request_not_used(db_request, db_execution, value) {
            Ok(receipt) => {
                record_terminal(observation, execution.cancel_observed());
                drop(deadline);
                Ok(receipt.into_value())
            }
            Err((db_request, db_execution, value)) => Err((
                Self {
                    lease: ProfuseGwConcreteDbRequestLease {
                        execution,
                        deadline,
                        db_request,
                        db_execution,
                        observation,
                        scope_stop,
                    },
                    cancel: self.cancel,
                    entry: self.entry,
                    diagnostic_zone: self.diagnostic_zone,
                },
                value,
            )),
        }
    }

    /// Only after all borrowed futures/driver are destroyed may DB transfer
    /// its owning phase/connection into physical finalization.
    #[doc(hidden)]
    pub fn into_physical_finalization(
        self,
    ) -> (ProfuseGwSerialCompletion<C>, DbPhysicalExecutionHalf) {
        let (completion, execution) = self
            .lease
            .into_physical_finalization()
            .into_database_execution();
        (
            ProfuseGwSerialCompletion {
                completion,
                cancel: self.cancel,
            },
            execution,
        )
    }
}

impl<C: Future<Output = ()>> ProfuseGwScopeDriver<'_, C> {
    fn diagnostic_scope(&self) -> Option<saddle_observability::RequestDiagnosticScope<'static>> {
        let observation = self.observation.as_ref()?;
        let projection = self
            .db_request
            .project_diagnostic_context(
                self.db_execution,
                (&observation.context, &observation.event_context),
            )
            .ok()?;
        let scope = crate::diagnostics::live_request_scope(&projection);
        Some(match self.diagnostic_zone {
            Some(zone) => scope.with_zone(zone),
            None => scope,
        })
    }
    /// Required at business entry, SQL entry/exit and immediately before COMMIT.
    #[doc(hidden)]
    pub async fn checkpoint(&self) -> Result<(), ProfuseGwScopeStop> {
        let mut yielded = false;
        poll_fn(|cx| {
            let mut control = self.control.lock().unwrap_or_else(|p| p.into_inner());
            if let Err(stop) = control.check(cx) {
                return Poll::Ready(Err(stop));
            }
            if !yielded {
                yielded = true;
                cx.waker().wake_by_ref();
                Poll::Pending
            } else {
                Poll::Ready(Ok(()))
            }
        })
        .await
    }

    /// Framework/DB container construction only. Shares this execution's
    /// account; it neither admits work nor grants another scope. Existing
    /// allocated values keep their charge after this borrowed driver ends.
    /// Callers must not expose this storage capability to business code.
    #[doc(hidden)]
    pub async fn database_memory(
        &self,
    ) -> Result<saddle_admission::RequestMemory, ProfuseGwScopeStop> {
        self.checkpoint().await?;
        Ok(self.execution.request_memory())
    }

    /// Governs the ENTIRE body, including non-DB Pending awaits. Never hold
    /// the control lock while polling the body (steps borrow it recursively).
    /// Body is destroyed before returning stop/panic to the owning DB phase.
    #[doc(hidden)]
    pub async fn supervise<F: Future>(
        &self,
        future: F,
    ) -> Result<F::Output, ProfuseGwScopeFailure> {
        {
            let mut control = self.control.lock().unwrap_or_else(|p| p.into_inner());
            if control.supervising {
                return Err(ProfuseGwScopeFailure::AlreadySupervised);
            }
            control.supervising = true;
        }
        let mut guard = SupervisionGuard {
            control: &self.control,
            completed: false,
        };
        let mut body = SupervisedBody {
            driver: self,
            future: Some(Box::pin(future)),
            primary: None,
        };
        let outcome = poll_fn(|cx| {
            {
                let mut control = self.control.lock().unwrap_or_else(|p| p.into_inner());
                if let Err(stop) = control.check(cx) {
                    return Poll::Ready(Err(ProfuseGwScopeFailure::Stopped(stop)));
                }
            }
            // Catch BUSINESS unwind inside the admitted poll boundary. Catching
            // outside would unwind the allocator TLS guard and permanently
            // poison the account before physical cleanup could run.
            let mut caught = poll_fn(|cx| {
                match crate::diagnostics::catching_request(
                    saddle_core::DiagnosticStage::RequestDb,
                    "runtime.request_scope",
                    self.observation.as_ref().map(|o| {
                        (
                            o.observer.clone(),
                            o.context.clone(),
                            o.event_context.clone(),
                        )
                    }),
                    self.diagnostic_scope(),
                    None,
                    || {
                        body.future
                            .as_mut()
                            .expect("body lives until supervision ends")
                            .as_mut()
                            .poll(cx)
                    },
                )
                .0
                {
                    Ok(Poll::Pending) => Poll::Pending,
                    Ok(Poll::Ready(value)) => Poll::Ready(Ok(value)),
                    Err(diagnostic) => {
                        let occurrence = diagnostic.occurrence();
                        if let Some(observation) = self.observation {
                            observation
                                .diagnostic
                                .lock()
                                .unwrap_or_else(|e| e.into_inner())
                                .get_or_insert(occurrence);
                        }
                        diagnostic.record(&saddle_core::DiagnosticOutcomeAxes {
                            operation: saddle_core::OperationOutcome::Panicked,
                            ..Default::default()
                        });
                        body.primary = Some(diagnostic);
                        Poll::Ready(Err(()))
                    }
                }
            });
            match self
                .execution
                .poll_database_query(Pin::new(&mut caught), cx)
            {
                Poll::Pending => Poll::Pending,
                Poll::Ready(Ok(value)) => {
                    // Preserve acknowledgement/business T even if its poll
                    // crossed the deadline. Sticky stop bars the next step;
                    // DB still owns commit certainty (never rewrite it).
                    let mut control = self.control.lock().unwrap_or_else(|p| p.into_inner());
                    let _ = control.check(cx);
                    Poll::Ready(Ok(value))
                }
                Poll::Ready(Err(())) => {
                    let mut control = self.control.lock().unwrap_or_else(|p| p.into_inner());
                    control.stop.get_or_insert(ProfuseGwScopeStop::Cancelled);
                    Poll::Ready(Err(ProfuseGwScopeFailure::Panicked))
                }
            }
        })
        .await;
        drop(body);
        guard.completed = true;
        outcome
    }

    fn drop_body<F: Future>(
        &self,
        future: Pin<Box<F>>,
        primary_diagnostic: Option<crate::diagnostics::RequestCaptured>,
    ) {
        let (cleanup, primary) = crate::diagnostics::catching_request(
            saddle_core::DiagnosticStage::FinalizerResource,
            "runtime.request_scope_drop",
            self.observation.as_ref().map(|o| {
                (
                    o.observer.clone(),
                    o.context.clone(),
                    o.event_context.clone(),
                )
            }),
            self.diagnostic_scope(),
            primary_diagnostic,
            || drop(future),
        );
        let mut retained = primary;
        if let Err(diagnostic) = cleanup {
            let occurrence = diagnostic.occurrence();
            if let Some(observation) = self.observation {
                observation
                    .cleanup_failed
                    .store(true, std::sync::atomic::Ordering::Relaxed);
                observation
                    .diagnostic
                    .lock()
                    .unwrap_or_else(|e| e.into_inner())
                    .get_or_insert(occurrence);
            }
            // No more execution after failed cleanup, but do not rewrite the
            // original Ready(T), including commit certainty, as an operation error.
            self.control
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .stop
                .get_or_insert(ProfuseGwScopeStop::Cancelled);
            diagnostic.record(&saddle_core::DiagnosticOutcomeAxes {
                cleanup: saddle_core::CleanupOutcome::Failed,
                ..Default::default()
            });
            if retained.is_none() {
                retained = Some(diagnostic);
            }
        }
        if let (Some(observation), Some(captured)) = (self.observation.as_ref(), retained) {
            observation
                .request_diagnostic
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .get_or_insert(captured);
        }
    }
}

// Own the body until either normal completion or cancellation of supervise.
// Drop runs before SupervisionGuard releases the borrowed control domain.
struct SupervisedBody<'a, 'b, F: Future, C: Future<Output = ()>> {
    driver: &'a ProfuseGwScopeDriver<'b, C>,
    future: Option<Pin<Box<F>>>,
    primary: Option<crate::diagnostics::RequestCaptured>,
}

impl<F: Future, C: Future<Output = ()>> Drop for SupervisedBody<'_, '_, F, C> {
    fn drop(&mut self) {
        if let Some(future) = self.future.take() {
            self.driver.drop_body(future, self.primary.take());
        }
    }
}

struct SupervisionGuard<'a, 'b, C> {
    control: &'a Mutex<Control<'b, C>>,
    completed: bool,
}
impl<C> Drop for SupervisionGuard<'_, '_, C> {
    fn drop(&mut self) {
        let mut control = self.control.lock().unwrap_or_else(|p| p.into_inner());
        if !self.completed {
            control.stop.get_or_insert(ProfuseGwScopeStop::Cancelled);
        }
        control.supervising = false;
    }
}

/// Keeps the original cancellation source alongside physical completion.
#[doc(hidden)]
pub struct ProfuseGwSerialCompletion<C> {
    completion: ProfuseGwDatabaseFinalizationCompletion,
    cancel: Pin<Box<C>>,
}

impl<C: Future<Output = ()>> ProfuseGwSerialCompletion<C> {
    /// Ready means stop physical return and discard; it is NOT a disposition.
    #[doc(hidden)]
    pub fn poll_physical_stop(&mut self, cx: &mut Context<'_>) -> Poll<()> {
        let mut control = Control {
            deadline: &mut self.completion.deadline,
            stop: &mut self.completion.scope_stop,
            cancel: self.cancel.as_mut(),
            supervising: false,
            observation: &self.completion.observation,
            physical_return: true,
        };
        if control.check(cx).is_err() {
            Poll::Ready(())
        } else {
            Poll::Pending
        }
    }

    #[doc(hidden)]
    #[allow(
        clippy::result_large_err,
        reason = "foreign receipt must return both complete owners"
    )]
    pub fn complete<T>(
        self,
        physical: DbPhysicalDispositionOwner<T>,
    ) -> Result<ProfuseGwSuspendedScope<C, T>, (Self, DbPhysicalDispositionOwner<T>)> {
        match finish_profusegw_database_disposition(self.completion, physical) {
            Ok(owner) => Ok(ProfuseGwSuspendedScope {
                inner: Some((owner, self.cancel)),
            }),
            Err(failure) => Err((
                Self {
                    completion: failure.completion,
                    cancel: self.cancel,
                },
                failure.physical,
            )),
        }
    }
}

/// Inert whole: Pending/drop of resume retains every owner. No execution
/// capability is exposed until the same time/cancel checkpoint succeeds.
#[doc(hidden)]
pub struct ProfuseGwSuspendedScope<C, T> {
    inner: Option<(ProfuseGwPostDatabaseManagedOwner<T>, Pin<Box<C>>)>,
}

#[derive(Debug)]
#[doc(hidden)]
pub enum ProfuseGwScopeResumeError {
    Consumed,
    Stopped(ProfuseGwScopeStop),
    ScopeExhausted,
    Admission(AdmissionError),
}

impl<C: Future<Output = ()>, T> ProfuseGwSuspendedScope<C, T> {
    #[doc(hidden)]
    pub async fn resume(
        &mut self,
    ) -> Result<(ProfuseGwSerialScope<C>, T), ProfuseGwScopeResumeError> {
        let (owner, cancel) = self
            .inner
            .as_mut()
            .ok_or(ProfuseGwScopeResumeError::Consumed)?;
        scope_checkpoint(
            &mut owner.terminal.deadline,
            &mut owner.terminal.scope_stop,
            cancel.as_mut(),
        )
        .await
        .map_err(ProfuseGwScopeResumeError::Stopped)?;
        // No await/callback from here until owner transfer is complete.
        let Some((owner, cancel)) = self.inner.take() else {
            return Err(ProfuseGwScopeResumeError::Consumed);
        };
        let ProfuseGwPostDatabaseManagedOwner { terminal, value } = owner;
        let ProfuseGwPostDatabaseRequestTerminal {
            admission,
            deadline,
            observation,
            scope_stop,
            next_scope,
        } = terminal;
        let pair = match next_scope {
            NextScope::Pair(request, execution) => (request, execution),
            NextScope::Continuation(continuation) => match continuation.into_next_scope() {
                Ok(pair) => pair,
                Err(continuation) => {
                    self.inner = Some((
                        ProfuseGwPostDatabaseManagedOwner {
                            terminal: ProfuseGwPostDatabaseRequestTerminal {
                                admission,
                                deadline,
                                observation,
                                scope_stop,
                                next_scope: NextScope::Continuation(continuation),
                            },
                            value,
                        },
                        cancel,
                    ));
                    return Err(ProfuseGwScopeResumeError::ScopeExhausted);
                }
            },
        };
        match admission.resume_after_runtime_checkpoint() {
            Ok(execution) => Ok((
                ProfuseGwSerialScope {
                    lease: ProfuseGwConcreteDbRequestLease {
                        execution,
                        deadline,
                        db_request: pair.0,
                        db_execution: pair.1,
                        observation,
                        scope_stop,
                    },
                    cancel,
                    entry: ScopeEntry::ResumedUnused,
                    diagnostic_zone: None,
                },
                value,
            )),
            Err((admission, error)) => {
                self.inner = Some((
                    ProfuseGwPostDatabaseManagedOwner {
                        terminal: ProfuseGwPostDatabaseRequestTerminal {
                            admission,
                            deadline,
                            observation,
                            scope_stop,
                            next_scope: NextScope::Pair(pair.0, pair.1),
                        },
                        value,
                    },
                    cancel,
                ));
                Err(ProfuseGwScopeResumeError::Admission(error))
            }
        }
    }

    /// Business value may be used to construct the final response; the terminal
    /// still holds all Admission resources until writeback/disconnect finish.
    #[doc(hidden)]
    #[allow(
        clippy::result_large_err,
        reason = "preserve the inert whole on repeated consumption"
    )]
    pub fn into_response_parts(
        mut self,
    ) -> Result<(T, ProfuseGwPostDatabaseRequestTerminal), Self> {
        match self.inner.take() {
            Some((owner, _cancel)) => Ok(owner.into_response_parts()),
            None => Err(self),
        }
    }
}

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

    fn process() -> ProfuseGwRuntimeProcess {
        process_capacity(1)
    }

    fn process_capacity(capacity: u32) -> ProfuseGwRuntimeProcess {
        let pending = saddle_admission::freeze_deployment_resource_budget(
            capacity, 32, 5_000, capacity, capacity, 1_000_000, 1_000_000, 1_000_000, 1_000_000,
        )
        .unwrap();
        let (application, listener) = saddle_core::BootstrapRendezvousIssuer::issue()
            .freeze_application(saddle_core::GeneratedApplicationFreezeSource::new(
                "app",
                b"descriptor",
                &["route"],
            ))
            .unwrap();
        let listener = listener
            .freeze_listener(saddle_core::ListenerStartupFreezeSource::new(
                "app",
                "127.0.0.1:8000".parse().unwrap(),
                "127.0.0.1:9000".parse().unwrap(),
                Duration::from_millis(5_000),
            ))
            .ok()
            .unwrap();
        let (whole, receipt) = saddle_core::pair_bootstrap_rendezvous(application, listener)
            .ok()
            .unwrap();
        let budget =
            saddle_admission::bind_deployment_resource_budget_bootstrap(pending, whole, receipt)
                .ok()
                .unwrap();
        ProfuseGwRuntimeProcess::new(prepare_profusegw_lightweight_profile(budget).ok().unwrap())
    }

    fn admit(process: &ProfuseGwRuntimeProcess) -> ProfuseGwConcreteDbRequestLease {
        match process.try_admit() {
            ProfuseGwCoordinatorAdmissionOutcome::Ready(dispatch, _) => {
                dispatch.into_database_request()
            }
            _ => panic!("admission must succeed"),
        }
    }

    fn full(process: &ProfuseGwRuntimeProcess) {
        assert!(matches!(
            process.try_admit(),
            ProfuseGwCoordinatorAdmissionOutcome::CapacityRejected(_)
        ));
    }

    #[tokio::test]
    async fn scope_observation_original_context_serial_and_zero() {
        let directory = std::env::temp_dir().join(format!("saddle-scope-zone-{}", std::process::id()));
        let output = saddle_observability::EmergencyDiagnostics::start(
            &saddle_observability::FileLoggingConfig::new(directory.clone(), saddle_observability::Rotation::Daily),
        ).unwrap();
        let handle = output.handle();
        let mut process = process();
        let physical = process.db_startup.take().unwrap().into_process_capability();
        let mut lease = admit(&process);
        let original_deadline = lease.deadline.timer.deadline();
        assert!(matches!(
            lease.request_diagnostic_scope(None),
            Err(ProfuseGwTransactionObservationError::MissingContext)
        ));
        let observer = Observer::with_writer(
            saddle_observability::ObserverConfig::default(),
            std::io::sink(),
        )
        .unwrap();
        let (call, _) = observer
            .start_external_call_checked("app", "module", "service", "route", Some("safe-trace"))
            .unwrap();
        let context = call.context().clone();
        let event = EventContext::new(
            saddle_observability::RequestIdentity::new("request").unwrap(),
            saddle_observability::RouteIdentity::new("route").unwrap(),
            1,
        )
        .unwrap();
        lease.observation = Some(ProfuseGwRequestObservation {
            observer: observer.clone(),
            context: context.clone(),
            event_context: event.clone(),
            diagnostic: Mutex::new(None),
            request_diagnostic: Mutex::new(None),
            cleanup_failed: std::sync::atomic::AtomicBool::new(false),
        });
        for _ in 0..2 {
            assert!(lease.request_diagnostic_scope(None).is_ok());
        }
        let mut scope = lease.into_serial_scope(std::future::pending());
        for expected in [0, 1] {
            // The Facade reattaches its original validated zone before each DB call.
            scope = scope.with_diagnostic_zone(
                saddle_observability::DiagnosticZone::from_validated_ingress("local/区域").unwrap(),
            );
            let reference = scope.request_diagnostic_scope(Some(&handle)).unwrap()
                .fail((), saddle_core::DiagnosticCategory::ExpectedRejection,
                    saddle_core::BoundedDiagnosticCause::new(saddle_core::DiagnosticStage::RequestDb,
                        saddle_core::DiagnosticCode::new("test.zone").unwrap()))
                .into_reference();
            reference.record(&handle, &saddle_core::DiagnosticOutcomeAxes::default());
            let was_entered = matches!(scope.entry, ScopeEntry::Entered);
            for _ in 0..2 {
                let diagnostic = scope.request_diagnostic_scope(None).unwrap();
                let receipt = diagnostic.fail(
                    41_u32,
                    saddle_core::DiagnosticCategory::ExpectedRejection,
                    saddle_core::BoundedDiagnosticCause::new(
                        saddle_core::DiagnosticStage::RequestDb,
                        saddle_core::DiagnosticCode::new("test.projection").unwrap(),
                    ),
                );
                assert_eq!(*receipt.error(), 41);
                assert_eq!(
                    receipt.submission(),
                    saddle_observability::DiagnosticSubmission::OutputUnavailable
                );
                assert_eq!(matches!(scope.entry, ScopeEntry::Entered), was_entered);
            }
            let saved = scope.lease.observation.take().unwrap();
            assert!(matches!(
                scope.take_transaction_observation(),
                Err(ProfuseGwTransactionObservationError::MissingContext)
            ));
            scope.lease.observation = Some(saved);
            let token = scope.take_transaction_observation().unwrap();
            assert!(matches!(
                scope.take_transaction_observation(),
                Err(ProfuseGwTransactionObservationError::Unavailable)
            ));
            assert_eq!(scope.lease.deadline.timer.deadline(), original_deadline);
            let (completion, execution) = scope.into_physical_finalization();
            let checked = token.bind_terminal(&execution).ok().unwrap();
            let ((_observer, actual_context, actual_event), fields) = checked.into_log_parts();
            assert_eq!(actual_context, context);
            assert_eq!(actual_event, event);
            assert_eq!(
                serde_json::to_value(fields).unwrap(),
                serde_json::json!({"transaction_scope":expected})
            );
            let proof = physical.connection_returned(execution, ()).ok().unwrap();
            let mut suspended = completion.complete(proof).ok().unwrap();
            full(&process);
            scope = suspended.resume().await.unwrap().0;
        }
        assert!(scope.request_diagnostic_scope(None).is_ok());
        scope.finish_unentered_response(()).ok().unwrap();
        process.finish().unwrap();
        call.succeed();
        observer.flush().await.unwrap();
        let exit = crate::diagnostics::close_output(output,
            Some(std::time::Instant::now() + Duration::from_secs(2)));
        assert_eq!(exit.snapshot.written, 4);
        let file = directory.join("saddle.emergency.log");
        let records: Vec<serde_json::Value> = std::fs::read_to_string(&file).unwrap()
            .lines().map(|line| serde_json::from_str(line).unwrap()).collect();
        for (index, pair) in records.chunks_exact(2).enumerate() {
            assert_eq!(pair[0]["context"], pair[1]["context"]);
            assert_eq!(pair[0]["context"]["zone"], serde_json::json!({"state":"present","value":"local/区域"}));
            assert_eq!(pair[0]["context"]["scope"]["value"]["transaction_scope"], index);
        }
        std::fs::remove_file(file).unwrap();
        std::fs::remove_dir(directory).unwrap();
    }

    #[tokio::test]
    async fn public_task_ready_drop_panic_keeps_value_and_owner() {
        struct Fault<'a>(&'a mut ProfuseGwManagedDispatch);
        impl Future for Fault<'_> {
            type Output=Result<u32,saddle_observability::FrameworkRequestFailure<()>>;
            fn poll(self:Pin<&mut Self>,_:&mut Context<'_>)->Poll<Self::Output> {
                let _=self.0.deadline_unix_ms();Poll::Ready(Ok(41))
            }
        }
        impl Drop for Fault<'_> {fn drop(&mut self){panic!("owned borrowed cleanup test");}}
        let mut process=process();
        let _physical=process.db_startup.take().unwrap().into_process_capability();
        let owner=match process.try_admit(){ProfuseGwCoordinatorAdmissionOutcome::Ready(owner,_)=>owner,_=>panic!("admit")};
        let context=saddle_observability::RequestDiagnosticScope::early(None,
            saddle_observability::EarlyRequestContext::socket_accepted("app"));
        let (future,ticket)=crate::request_task::admitted_request_task(owner,context,None,|owner,_|Box::pin(Fault(owner)));
        let task=tokio::spawn(future);let ticket=ticket.bind_spawned(task.id()).ok().unwrap();
        let (owner,outcome)=ticket.complete(task.await).ok().unwrap();
        assert_eq!(outcome.result().unwrap().as_ref().ok(),Some(&41));
        assert!(outcome.cleanup_failed());
        let terminal=outcome.finish();drop(terminal);
        owner.cancel();process.finish().unwrap();
    }

    #[tokio::test]
    async fn public_task_entered_scope_abort_preserves_physical_handoff() {
        let mut process = process();
        let physical = process.db_startup.take().unwrap().into_process_capability();
        let scope = admit(&process).into_serial_scope(std::future::pending::<()>());
        let deadline = scope.lease.deadline.timer.deadline();
        let context=saddle_observability::RequestDiagnosticScope::early(None,
            saddle_observability::EarlyRequestContext::socket_accepted("app"));
        let (sent,started)=tokio::sync::oneshot::channel();
        let (future,ticket)=crate::request_task::owned_request_task(scope,context,None,move |scope,_|Box::pin(async move {
            let driver=scope.driver();
            let _=driver.supervise(async move {
                let _=sent.send(());
                std::future::pending::<()>().await;
            }).await;
            Ok::<_,saddle_observability::FrameworkRequestFailure<()>>(())
        }));
        let task=tokio::spawn(future);let ticket=ticket.bind_spawned(task.id()).ok().unwrap();
        started.await.unwrap();task.abort();
        let (scope,outcome)=ticket.complete(task.await).ok().unwrap();
        assert_eq!(scope.lease.deadline.timer.deadline(),deadline);
        assert_eq!(scope.lease.scope_stop,Some(ProfuseGwScopeStop::Cancelled));
        let (scope,())=scope.finish_unentered_response(()).err().unwrap();
        full(&process);
        let terminal=outcome.finish();
        assert!(terminal.outcome().join_failure().unwrap().error().is_cancelled());
        drop(terminal);
        // This is a simulated physical disposition using the existing Core
        // test capability, not a real SQL/connection proof. Runtime cannot sign it.
        let (completion,execution)=scope.into_physical_finalization();
        let receipt=physical.connection_discarded(execution,41).ok().unwrap();
        let mut suspended=completion.complete(receipt).ok().unwrap();
        assert!(matches!(suspended.resume().await,Err(ProfuseGwScopeResumeError::Stopped(_))));
        let (value,terminal)=suspended.into_response_parts().ok().unwrap();
        assert_eq!(value,41);
        finish_profusegw_after_database(terminal);
        process.finish().unwrap();
    }

    #[tokio::test]
    async fn public_task_pending_owner_drop_recovers_resources() {
        let mut process=process();
        let _physical=process.db_startup.take().unwrap().into_process_capability();
        let owner=match process.try_admit(){ProfuseGwCoordinatorAdmissionOutcome::Ready(owner,_)=>owner,_=>panic!("admit")};
        let scope=saddle_observability::RequestDiagnosticScope::early(None,
            saddle_observability::EarlyRequestContext::socket_accepted("app"));
        let (sent,started)=tokio::sync::oneshot::channel();
        let (future,ticket)=crate::request_task::admitted_request_task(owner,scope,None,move |owner,_|Box::pin(async move {
            let _=sent.send(());
            let _ = owner.deadline_unix_ms();
            std::future::pending::<()>().await;
            Ok::<_,saddle_observability::FrameworkRequestFailure<()>>(())
        }));
        let task=tokio::spawn(future);let ticket=ticket.bind_spawned(task.id()).ok().unwrap();
        started.await.unwrap();task.abort();
        let (owner,outcome)=ticket.complete(task.await).ok().unwrap();
        let terminal=outcome.finish();
        assert!(terminal.outcome().join_failure().unwrap().error().is_cancelled());
        drop(terminal);
        owner.cancel();
        match process.try_admit() {
            ProfuseGwCoordinatorAdmissionOutcome::Ready(owner,_)=>owner.cancel(),
            _=>panic!("same process must remain healthy and reusable after cancellation"),
        }
        assert_eq!(process.finish(),Ok(()));
    }

    #[tokio::test]
    async fn public_task_admitted_owner_cancel_and_zero() {
        for abort_before_poll in [false,true] {
            let mut process=process();
            let _physical=process.db_startup.take().unwrap().into_process_capability();
            let owner=match process.try_admit(){ProfuseGwCoordinatorAdmissionOutcome::Ready(owner,_)=>owner,_=>panic!("admit")};
            let scope=saddle_observability::RequestDiagnosticScope::early(None,
                saddle_observability::EarlyRequestContext::socket_accepted("app"));
            let (future,ticket)=crate::request_task::admitted_request_task(owner,scope,None,|owner,_|Box::pin(async move {
                let _ = owner.deadline_unix_ms();Ok::<_,saddle_observability::FrameworkRequestFailure<()>>(41)
            }));
            let task=tokio::spawn(future);let ticket=ticket.bind_spawned(task.id()).ok().unwrap();
            if abort_before_poll{task.abort();}
            let (owner,outcome)=ticket.complete(task.await).ok().unwrap();
            let terminal=outcome.finish();
            if abort_before_poll{assert!(terminal.outcome().join_failure().unwrap().error().is_cancelled());}
            else {assert_eq!(*terminal.outcome().result().unwrap().as_ref().ok().unwrap(),41);}
            owner.cancel();process.finish().unwrap();
        }
    }

    #[tokio::test]
    async fn physical_stop_diagnostic_keeps_value_and_boundary_axes() {
        const CHILD: &str = "RUNTIME_PHYSICAL_DIAGNOSTIC_CHILD";
        if let Some(path) = std::env::var_os(CHILD) {
            let panic_case = std::env::var_os("RUNTIME_SCOPE_PANIC_CASE").is_some();
            let drop_case = std::env::var_os("RUNTIME_SCOPE_DROP_CASE").is_some();
            let output = saddle_observability::EmergencyDiagnostics::start(
                &saddle_observability::FileLoggingConfig::new(
                    std::path::PathBuf::from(&path),
                    saddle_observability::Rotation::Daily,
                ),
            )
            .unwrap();
            assert!(crate::diagnostics::install_output(output.handle()).is_ok());
            std::panic::set_hook(Box::new(|info| {
                let _ = crate::diagnostics::capture_current_panic(info);
            }));
            let mut process = process();
            let physical = process.db_startup.take().unwrap().into_process_capability();
            let mut lease = admit(&process);
            let observer = Observer::with_writer(
                saddle_observability::ObserverConfig::default(),
                std::io::sink(),
            )
            .unwrap();
            let (call, _) = observer
                .start_external_call_checked(
                    "app",
                    "module",
                    "service",
                    "route",
                    Some("safe-trace"),
                )
                .unwrap();
            lease.observation = Some(ProfuseGwRequestObservation {
                observer: observer.clone(),
                context: call.context().clone(),
                event_context: EventContext::new(
                    saddle_observability::RequestIdentity::new("request").unwrap(),
                    saddle_observability::RouteIdentity::new("route").unwrap(),
                    1,
                )
                .unwrap(),
                diagnostic: Mutex::new(None),
                request_diagnostic: Mutex::new(None),
                cleanup_failed: std::sync::atomic::AtomicBool::new(false),
            });
            let mut scope = lease.into_serial_scope(std::future::pending());
            scope = scope.with_diagnostic_zone(
                saddle_observability::DiagnosticZone::from_validated_ingress("original-zone")
                    .unwrap(),
            );
            struct FaultFuture {
                panic_case: bool,
                drop_case: bool,
            }
            impl Future for FaultFuture {
                type Output = u32;
                fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<u32> {
                    if self.panic_case {
                        panic!("PRIVATE_SCOPE_PANIC_SENTINEL");
                    }
                    Poll::Ready(41)
                }
            }
            impl Drop for FaultFuture {
                fn drop(&mut self) {
                    if self.drop_case {
                        panic!("PRIVATE_SCOPE_DROP_SENTINEL");
                    }
                }
            }
            let value = scope
                .driver()
                .supervise(FaultFuture {
                    panic_case,
                    drop_case,
                })
                .await;
            let (mut completion, execution) = scope.into_physical_finalization();
            // Test-only expiry of the original timer AFTER operation Ready.
            completion
                .completion
                .deadline
                .timer
                .as_mut()
                .reset(tokio::time::Instant::now());
            poll_fn(|cx| {
                assert!(completion.poll_physical_stop(cx).is_ready());
                assert!(completion.poll_physical_stop(cx).is_ready());
                Poll::Ready(())
            })
            .await;
            let receipt = physical
                .connection_discarded(execution, value)
                .ok()
                .unwrap();
            let suspended = completion.complete(receipt).ok().unwrap();
            let (value, terminal) = suspended.into_response_parts().ok().unwrap();
            if panic_case {
                assert_eq!(value, Err(ProfuseGwScopeFailure::Panicked));
            } else {
                assert_eq!(value, Ok(41));
            }
            finish_profusegw_after_database(terminal);
            process.finish().unwrap();
            if panic_case {
                drop(call);
            } else {
                call.succeed();
            }
            observer.flush().await.unwrap();
            let exit = crate::diagnostics::close_output(
                output,
                Some(std::time::Instant::now() + Duration::from_secs(2)),
            );
            assert_eq!(exit.snapshot.dropped, 0);
            assert_eq!(exit.snapshot.enqueued, exit.snapshot.written);
            let log = std::fs::read_to_string(
                std::path::PathBuf::from(path).join("saddle.emergency.log"),
            )
            .unwrap();
            let rows: Vec<serde_json::Value> = log
                .lines()
                .map(|s| serde_json::from_str(s).unwrap())
                .collect();
            let sources: Vec<_> = rows.iter().filter(|v| !v["diagnostic"].is_null()).collect();
            assert_eq!(
                sources.len(),
                if panic_case && drop_case { 2 } else { 1 },
                "source submits once per actual occurrence"
            );
            assert!(!log.contains("PRIVATE_SCOPE_PANIC_SENTINEL"));
            assert!(!log.contains("PRIVATE_SCOPE_DROP_SENTINEL"));
            if panic_case || drop_case {
                for source in &sources {
                    assert_eq!(source["context"]["scope"]["state"], "present");
                    assert_eq!(source["context"]["scope"]["value"]["transaction_scope"], 0);
                    assert_eq!(source["context"]["zone"]["value"], "original-zone");
                    assert_eq!(source["diagnostic"]["stack_status"], "unavailable_deferred");
                }
            }
            if panic_case {
                assert_eq!(sources[0]["diagnostic"]["category"], "panic");
                assert_eq!(sources[0]["diagnostic"]["capture_site"], "origin");
                let boundary = rows
                    .iter()
                    .find(|v| v["axes"]["operation"] == "panicked")
                    .unwrap();
                assert!(boundary["diagnostic"].is_null());
                assert_eq!(
                    boundary["diagnostic_reference"]["diagnostic_id"],
                    sources[0]["diagnostic"]["diagnostic_id"]
                );
            } else if !drop_case {
                assert_eq!(sources[0]["axes"]["operation"], "unknown");
                assert_eq!(
                    sources[0]["diagnostic"]["causes"][0]["code"],
                    "runtime.physical_return_deadline"
                );
            }
            if drop_case {
                let cleanup = sources
                    .iter()
                    .find(|v| v["diagnostic"]["task"] == "runtime.request_scope_drop")
                    .unwrap();
                if panic_case {
                    assert_eq!(
                        cleanup["diagnostic"]["primary_diagnostic_id"],
                        sources[0]["diagnostic"]["diagnostic_id"]
                    );
                }
                let boundary = rows
                    .iter()
                    .find(|v| v["axes"]["cleanup"] == "failed")
                    .unwrap();
                assert!(boundary["diagnostic"].is_null());
                assert_eq!(
                    boundary["diagnostic_reference"]["diagnostic_id"],
                    cleanup["diagnostic"]["diagnostic_id"]
                );
                assert_eq!(boundary["axes"]["operation"], "unknown");
            }
            let terminal = rows
                .iter()
                .find(|v| v["axes"]["physical"] == "discarded")
                .unwrap();
            assert!(terminal["diagnostic"].is_null());
            assert_eq!(
                terminal["diagnostic_reference"]["diagnostic_id"],
                sources[0]["diagnostic"]["diagnostic_id"]
            );
            assert_eq!(terminal["axes"]["business"], "unknown");
            assert_eq!(
                terminal["axes"]["cleanup"],
                if drop_case { "failed" } else { "succeeded" }
            );
            assert_eq!(terminal["axes"]["delivery"], "unknown");
            if panic_case || drop_case {
                assert_eq!(terminal["context"]["trace_id"]["value"], "safe-trace");
                assert_eq!(terminal["context"], sources[0]["context"]);
                assert_eq!(terminal["context"]["attempt"]["value"], 1);
                assert_eq!(terminal["context"]["scope"]["state"], "present");
                assert_eq!(
                    terminal["context"]["scope"]["value"]["transaction_scope"],
                    0
                );
            } else {
                assert_eq!(terminal["trace_id"], "safe-trace");
            }
            return;
        }
        for (panic_case, drop_case) in [(false, false), (true, false), (false, true), (true, true)]
        {
            let path = std::env::temp_dir().join(format!(
                "saddle-physical-diagnostic-{}-{panic_case}-{drop_case}",
                std::process::id()
            ));
            std::fs::create_dir(&path).unwrap();
            let mut command = std::process::Command::new(std::env::current_exe().unwrap());
            command.args(["--exact", "profusegw::serial_scope::tests::physical_stop_diagnostic_keeps_value_and_boundary_axes"])
            .env(CHILD, &path);
            if panic_case {
                command.env("RUNTIME_SCOPE_PANIC_CASE", "1");
            }
            if drop_case {
                command.env("RUNTIME_SCOPE_DROP_CASE", "1");
            }
            let mut child = command.spawn().unwrap();
            let started = std::time::Instant::now();
            let status = loop {
                if let Some(status) = child.try_wait().unwrap() {
                    break status;
                }
                if started.elapsed() > Duration::from_secs(10) {
                    child.kill().unwrap();
                    child.wait().unwrap();
                    panic!("physical diagnostic watchdog");
                }
                std::thread::sleep(Duration::from_millis(10));
            };
            assert!(status.success());
            std::fs::remove_file(path.join("saddle.emergency.log")).unwrap();
            std::fs::remove_dir(path).unwrap();
        }
    }

    struct Cancel {
        requested: Arc<AtomicBool>,
        completed: bool,
    }
    impl Future for Cancel {
        type Output = ();
        fn poll(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<()> {
            assert!(
                !self.completed,
                "completed cancellation future polled twice"
            );
            if self.requested.load(Ordering::SeqCst) {
                self.completed = true;
                Poll::Ready(())
            } else {
                Poll::Pending
            }
        }
    }
    struct BorrowedPending<'a>(&'a mut bool);
    impl Future for BorrowedPending<'_> {
        type Output = ();
        fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<()> {
            Poll::Pending
        }
    }
    impl Drop for BorrowedPending<'_> {
        fn drop(&mut self) {
            *self.0 = true;
        }
    }

    // Borrowing-only adapter matching DB SessionKernel::run_body at 8aad0787.
    // Not a SQL/kernel replacement: it tests the two lifetimes and Drop order.
    struct BorrowingSession<'driver, 'request, C> {
        driver: &'driver ProfuseGwScopeDriver<'request, C>,
        dropped: bool,
    }
    impl<C: Future<Output = ()> + Send> BorrowingSession<'_, '_, C> {
        async fn run_body<T>(
            &mut self,
            body: impl for<'tx> FnOnce(&'tx mut Self) -> Pin<Box<dyn Future<Output = T> + Send + 'tx>>,
        ) -> T {
            body(self).await
        }
    }

    #[tokio::test]
    async fn parameter_construction_first_request_not_used_and_stop() {
        for mode in 0..4 {
            let mut process = process();
            let _physical = process.db_startup.take().unwrap().into_process_capability();
            let dispatch = match process.try_admit() {
                ProfuseGwCoordinatorAdmissionOutcome::Ready(dispatch, _) => dispatch,
                _ => panic!("admission"),
            };
            let original_deadline = dispatch.deadline.timer.deadline();
            let mut scope = dispatch.into_parameter_scope(async move {
                if mode != 2 {
                    std::future::pending::<()>().await;
                }
            });
            if mode == 3 {
                scope
                    .lease
                    .deadline
                    .timer
                    .as_mut()
                    .reset(tokio::time::Instant::now());
            }
            let input = "original HTTP input";
            {
                let construction = scope.parameter_construction().unwrap();
                let result = construction
                    .supervise(async {
                        let memory = construction.database_memory().await.unwrap();
                        let held = memory.try_bytes(input.as_bytes()).unwrap();
                        if mode == 1 {
                            static TOO_LARGE: [u8; 32 * 1024 * 1024] = [0; 32 * 1024 * 1024];
                            assert!(matches!(
                                memory.try_bytes(&TOO_LARGE),
                                Err(AdmissionError::BudgetExceeded { .. })
                            ));
                            assert_eq!(held.as_slice(), input.as_bytes());
                        }
                        held
                    })
                    .await;
                match mode {
                    2 => assert!(matches!(
                        result,
                        Err(ProfuseGwScopeFailure::Stopped(
                            ProfuseGwScopeStop::Cancelled
                        ))
                    )),
                    3 => assert!(matches!(
                        result,
                        Err(ProfuseGwScopeFailure::Stopped(ProfuseGwScopeStop::TimedOut))
                    )),
                    _ => assert_eq!(result.unwrap().as_slice(), input.as_bytes()),
                }
            }
            assert!(matches!(scope.entry, ScopeEntry::ResumedUnused));
            if mode != 3 {
                assert_eq!(scope.lease.deadline.timer.deadline(), original_deadline);
            }
            assert_eq!(input, "original HTTP input");
            full(&process);
            scope.finish_unentered_response(()).ok().unwrap();
            process.finish().unwrap();
        }
    }

    #[tokio::test]
    async fn parameter_construction_then_sql_cannot_return_to_not_used() {
        let mut process = process();
        let physical = process.db_startup.take().unwrap().into_process_capability();
        let dispatch = match process.try_admit() {
            ProfuseGwCoordinatorAdmissionOutcome::Ready(dispatch, _) => dispatch,
            _ => panic!("admission"),
        };
        let mut scope = dispatch.into_parameter_scope(std::future::pending::<()>());
        let deadline = scope.lease.deadline.timer.deadline();
        let held = {
            let construction = scope.parameter_construction().unwrap();
            construction
                .supervise(async {
                    construction
                        .database_memory()
                        .await
                        .unwrap()
                        .try_bytes(b"parameter")
                        .unwrap()
                })
                .await
                .unwrap()
        };
        {
            let driver = scope.driver();
            driver
                .supervise(async {
                    assert_eq!(held.as_slice(), b"parameter");
                })
                .await
                .unwrap();
        }
        assert!(matches!(
            scope.parameter_construction(),
            Err(ProfuseGwScopeFailure::ScopeAlreadyEntered)
        ));
        let (scope, ()) = scope.finish_unentered_response(()).err().unwrap();
        let (completion, execution) = scope.into_physical_finalization();
        let receipt = physical.connection_returned(execution, held).ok().unwrap();
        let mut suspended = completion.complete(receipt).ok().unwrap();
        let (mut scope, held) = suspended.resume().await.unwrap();
        assert_eq!(scope.lease.deadline.timer.deadline(), deadline);
        {
            let construction = scope.parameter_construction().unwrap();
            construction
                .supervise(async {
                    let second = construction
                        .database_memory()
                        .await
                        .unwrap()
                        .try_bytes(b"second")
                        .unwrap();
                    assert_eq!(held.as_slice(), b"parameter");
                    drop(second);
                })
                .await
                .unwrap();
        }
        drop(held);
        scope.finish_unentered_response(()).ok().unwrap();
        process.finish().unwrap();
    }

    #[tokio::test]
    async fn composed_execution_reuses_request_across_physical_returns_and_outbound() {
        // Runtime-only physical boundary simulation, not SQL or a transaction.
        // The helper borrows a driver and application data; it owns no permit.
        async fn module<C: Future<Output = ()>>(
            driver: &ProfuseGwScopeDriver<'_, C>,
            values: &mut Vec<u32>,
        ) -> usize {
            driver.checkpoint().await.unwrap();
            values.push(7);
            values.len()
        }
        let mut process = process();
        let physical = process.db_startup.take().unwrap().into_process_capability();
        let lease = admit(&process);
        let deadline = lease.deadline.timer.deadline();
        let requested = Arc::new(AtomicBool::new(false));
        let mut scope = lease.into_serial_scope(Cancel {
            requested: requested.clone(),
            completed: false,
        });
        let mut values = Vec::new();
        for count in 1..=3 {
            let value = {
                let driver = scope.driver();
                driver
                    .supervise(module(&driver, &mut values))
                    .await
                    .unwrap()
            };
            assert_eq!(value, count);
            let (completion, execution) = scope.into_physical_finalization();
            let receipt = physical.connection_returned(execution, value).ok().unwrap();
            let mut suspended = completion.complete(receipt).ok().unwrap();
            full(&process);
            let (mut next, held) = suspended.resume().await.unwrap();
            assert_eq!(held, count);
            assert_eq!(next.lease.deadline.timer.deadline(), deadline);
            // Ordinary outbound/business work between DB acquisitions retains
            // the same request commitment, without entering another DB scope.
            assert_eq!(
                next.supervise_between(async { values.len() })
                    .await
                    .unwrap(),
                count
            );
            full(&process);
            scope = next;
        }
        requested.store(true, Ordering::SeqCst);
        let mut polled = false;
        assert_eq!(
            scope.supervise_between(async { polled = true }).await,
            Err(ProfuseGwScopeFailure::Stopped(
                ProfuseGwScopeStop::Cancelled
            ))
        );
        assert!(!polled);
        assert_eq!(
            scope.supervise_between(async {}).await,
            Err(ProfuseGwScopeFailure::Stopped(
                ProfuseGwScopeStop::Cancelled
            ))
        );
        // Cancellation was consumed once; only final response cleanup remains.
        full(&process);
        assert_eq!(
            scope.finish_unentered_response(values).ok().unwrap(),
            vec![7; 3]
        );
        process.finish().unwrap();
    }

    #[tokio::test]
    async fn serial_scope_same_request_keeps_credit_timer_and_value() {
        let mut process = process();
        let physical = process.db_startup.take().unwrap().into_process_capability();
        let lease = admit(&process);
        let deadline = lease.deadline.timer.deadline();
        let mut scope = lease.into_serial_scope(std::future::pending());
        for value in [11, 22, 33] {
            {
                let driver = scope.driver();
                let body = async {
                    driver.checkpoint().await.unwrap();
                    driver.checkpoint().await.unwrap();
                    value
                };
                assert_eq!(driver.supervise(body).await.unwrap(), value);
            }
            let (completion, execution) = scope.into_physical_finalization();
            let disposition = physical.connection_returned(execution, value).ok().unwrap();
            let mut suspended = completion.complete(disposition).ok().unwrap();
            full(&process);
            let (next, held_value) = suspended.resume().await.unwrap();
            assert_eq!(held_value, value);
            assert_eq!(next.lease.deadline.timer.deadline(), deadline);
            assert!(matches!(
                suspended.resume().await,
                Err(ProfuseGwScopeResumeError::Consumed)
            ));
            scope = next;
        }
        let (completion, execution) = scope.into_physical_finalization();
        let disposition = physical
            .connection_discarded(execution, "Unknown preserved")
            .ok()
            .unwrap();
        let suspended = completion.complete(disposition).ok().unwrap();
        let (value, terminal) = suspended.into_response_parts().ok().unwrap();
        assert_eq!(value, "Unknown preserved");
        full(&process);
        finish_profusegw_after_database(terminal);
        process.finish().unwrap();
    }

    #[tokio::test]
    async fn serial_scope_non_db_pending_stop_drops_before_finalization() {
        for cancelled in [false, true] {
            let mut process = process();
            let physical = process.db_startup.take().unwrap().into_process_capability();
            let lease = admit(&process);
            let requested = Arc::new(AtomicBool::new(false));
            let mut scope = lease.into_serial_scope(Cancel {
                requested: requested.clone(),
                completed: false,
            });
            let dropped;
            {
                let driver = scope.driver();
                let mut session = BorrowingSession {
                    driver: &driver,
                    dropped: false,
                };
                let supervised = driver.supervise(session.run_body(|session| {
                    Box::pin(async move {
                        session.driver.checkpoint().await.unwrap();
                        BorrowedPending(&mut session.dropped).await;
                    })
                }));
                // A generated boxed Send callback can borrow this same driver.
                fn require_send<T: Send>(_: &T) {}
                require_send(&supervised);
                {
                    tokio::pin!(supervised);
                    poll_fn(|cx| {
                        assert!(supervised.as_mut().poll(cx).is_pending());
                        Poll::Ready(())
                    })
                    .await;
                    // Second poll enters the non-DB pending body after checkpoint yield.
                    poll_fn(|cx| {
                        assert!(supervised.as_mut().poll(cx).is_pending());
                        Poll::Ready(())
                    })
                    .await;
                    if cancelled {
                        requested.store(true, Ordering::SeqCst);
                    } else {
                        // Deterministically expire the SAME timer only after
                        // observing the actual non-DB body Pending. Test-only.
                        driver
                            .control
                            .lock()
                            .unwrap()
                            .deadline
                            .timer
                            .as_mut()
                            .reset(tokio::time::Instant::now());
                    }
                    assert!(matches!(
                        supervised.await,
                        Err(ProfuseGwScopeFailure::Stopped(_))
                    ));
                }
                dropped = session.dropped;
            }
            assert!(
                dropped,
                "borrowed business future must die before phase finalizer"
            );
            let (mut completion, execution) = scope.into_physical_finalization();
            poll_fn(|cx| {
                assert!(completion.poll_physical_stop(cx).is_ready());
                Poll::Ready(())
            })
            .await;
            let disposition = physical.connection_discarded(execution, ()).ok().unwrap();
            let mut suspended = completion.complete(disposition).ok().unwrap();
            assert!(matches!(
                suspended.resume().await,
                Err(ProfuseGwScopeResumeError::Stopped(_))
            ));
            assert!(matches!(
                suspended.resume().await,
                Err(ProfuseGwScopeResumeError::Stopped(_))
            ));
            let (_, terminal) = suspended.into_response_parts().ok().unwrap();
            finish_profusegw_after_database(terminal);
            process.finish().unwrap();
        }
    }

    #[tokio::test]
    async fn serial_scope_dropped_supervision_is_sticky() {
        struct PendingDrop<'a> {
            dropped: &'a mut bool,
            panic_drop: bool,
        }
        impl Future for PendingDrop<'_> {
            type Output = ();
            fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<()> {
                Poll::Pending
            }
        }
        impl Drop for PendingDrop<'_> {
            fn drop(&mut self) {
                *self.dropped = true;
                if self.panic_drop {
                    panic!("request pending body cleanup failure");
                }
            }
        }
        for panic_drop in [false, true] {
            let mut process = process();
            let physical = process.db_startup.take().unwrap().into_process_capability();
            let mut scope = admit(&process).into_serial_scope(std::future::pending());
            let mut dropped = false;
            {
                let driver = scope.driver();
                let mut future = Box::pin(driver.supervise(PendingDrop {
                    dropped: &mut dropped,
                    panic_drop,
                }));
                poll_fn(|cx| {
                    assert!(future.as_mut().poll(cx).is_pending());
                    Poll::Ready(())
                })
                .await;
                drop(future);
                assert_eq!(
                    driver.checkpoint().await,
                    Err(ProfuseGwScopeStop::Cancelled)
                );
            }
            assert!(dropped);
            let (completion, execution) = scope.into_physical_finalization();
            let disposition = physical.connection_discarded(execution, ()).ok().unwrap();
            let mut suspended = completion.complete(disposition).ok().unwrap();
            assert!(matches!(
                suspended.resume().await,
                Err(ProfuseGwScopeResumeError::Stopped(_))
            ));
            let (_, terminal) = suspended.into_response_parts().ok().unwrap();
            finish_profusegw_after_database(terminal);
            process.finish().unwrap();
        }
    }

    #[tokio::test]
    async fn serial_scope_resume_pending_drop_and_foreign_return_owners() {
        let mut a = process_capacity(2);
        let pa = a.db_startup.take().unwrap().into_process_capability();
        let (ca, ea) = admit(&a)
            .into_serial_scope(std::future::pending())
            .into_physical_finalization();
        let (cb, eb) = admit(&a)
            .into_serial_scope(std::future::pending())
            .into_physical_finalization();
        let ra = pa.connection_returned(ea, 1).ok().unwrap();
        let rb = pa.connection_discarded(eb, 2).ok().unwrap();
        let (ca, rb) = ca.complete(rb).err().unwrap();
        let (cb, ra) = cb.complete(ra).err().unwrap();
        full(&a);
        let mut suspended = ca.complete(ra).ok().unwrap();
        {
            let mut attempt = Box::pin(suspended.resume());
            poll_fn(|cx| {
                assert!(attempt.as_mut().poll(cx).is_pending());
                Poll::Ready(())
            })
            .await;
        }
        full(&a);
        let (mut scope, value) = suspended.resume().await.unwrap();
        assert_eq!(value, 1);
        assert_eq!(scope.supervise_between(async { 7 }).await.unwrap(), 7);
        full(&a);
        assert_eq!(scope.finish_unentered_response(value).ok().unwrap(), 1);
        let (_, terminal) = cb
            .complete(rb)
            .ok()
            .unwrap()
            .into_response_parts()
            .ok()
            .unwrap();
        finish_profusegw_after_database(terminal);
        a.finish().unwrap();
    }

    #[tokio::test]
    async fn serial_scope_ready_loop_and_panic_keep_terminal() {
        for panic_body in [false, true] {
            let mut process = process();
            let physical = process.db_startup.take().unwrap().into_process_capability();
            let mut lease = admit(&process);
            lease
                .deadline
                .timer
                .as_mut()
                .reset(tokio::time::Instant::now() + Duration::from_millis(5));
            let mut scope = lease.into_serial_scope(std::future::pending());
            {
                let driver = scope.driver();
                let result = driver
                    .supervise(async {
                        assert!(!panic_body, "injected business unwind");
                        loop {
                            driver.checkpoint().await?;
                        }
                        #[allow(unreachable_code)]
                        Ok::<(), ProfuseGwScopeStop>(())
                    })
                    .await;
                if panic_body {
                    assert!(matches!(result, Err(ProfuseGwScopeFailure::Panicked)));
                } else {
                    assert!(matches!(
                        result,
                        Err(ProfuseGwScopeFailure::Stopped(ProfuseGwScopeStop::TimedOut))
                            | Ok(Err(ProfuseGwScopeStop::TimedOut))
                    ));
                }
            }
            let (completion, execution) = scope.into_physical_finalization();
            let physical = physical.connection_discarded(execution, ()).ok().unwrap();
            let mut suspended = completion.complete(physical).ok().unwrap();
            assert!(matches!(
                suspended.resume().await,
                Err(ProfuseGwScopeResumeError::Stopped(_))
            ));
            let (_, terminal) = suspended.into_response_parts().ok().unwrap();
            finish_profusegw_after_database(terminal);
            process.finish().unwrap();
        }
    }

    #[tokio::test]
    async fn serial_scope_physical_expiry_and_restore_keep_stop() {
        let mut process = process();
        let physical = process.db_startup.take().unwrap().into_process_capability();
        let mut lease = admit(&process);
        let original_deadline = lease.deadline.timer.deadline();
        lease.scope_stop = Some(ProfuseGwScopeStop::Cancelled);
        let lease = lease.restore_dispatch().into_database_request();
        assert_eq!(lease.scope_stop, Some(ProfuseGwScopeStop::Cancelled));
        assert_eq!(lease.deadline.timer.deadline(), original_deadline);
        let scope = lease.into_serial_scope(std::future::poll_fn(|_| -> Poll<()> {
            panic!("sticky stop must not poll cancellation again")
        }));
        let (mut completion, execution) = scope.into_physical_finalization();
        poll_fn(|cx| {
            assert!(completion.poll_physical_stop(cx).is_ready());
            Poll::Ready(())
        })
        .await;
        let receipt = physical.connection_discarded(execution, ()).ok().unwrap();
        let mut suspended = completion.complete(receipt).ok().unwrap();
        assert!(matches!(
            suspended.resume().await,
            Err(ProfuseGwScopeResumeError::Stopped(
                ProfuseGwScopeStop::Cancelled
            ))
        ));
        let (_, terminal) = suspended.into_response_parts().ok().unwrap();
        finish_profusegw_after_database(terminal);
        process.finish().unwrap();

        let mut process = self::process();
        let physical = process.db_startup.take().unwrap().into_process_capability();
        let scope = admit(&process).into_serial_scope(std::future::pending());
        let (mut completion, execution) = scope.into_physical_finalization();
        completion
            .completion
            .deadline
            .timer
            .as_mut()
            .reset(tokio::time::Instant::now());
        poll_fn(|cx| {
            assert!(completion.poll_physical_stop(cx).is_ready());
            Poll::Ready(())
        })
        .await;
        let receipt = physical
            .connection_returned(execution, "committed fact remains")
            .ok()
            .unwrap();
        let mut suspended = completion.complete(receipt).ok().unwrap();
        assert!(matches!(
            suspended.resume().await,
            Err(ProfuseGwScopeResumeError::Stopped(
                ProfuseGwScopeStop::TimedOut
            ))
        ));
        let (value, terminal) = suspended.into_response_parts().ok().unwrap();
        assert_eq!(value, "committed fact remains");
        finish_profusegw_after_database(terminal);
        process.finish().unwrap();
    }
}