saddle-runtime 0.3.14

//! 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,
}

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,
        }
    }
}

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 {
    #[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,
        }
    }
}

struct Control<'a, C> {
    deadline: &'a mut ProfuseGwManagedDeadline,
    stop: &'a mut Option<ProfuseGwScopeStop>,
    cancel: Pin<&'a mut C>,
    supervising: 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;
        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,
    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> {
    /// 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,
            control: Mutex::new(Control {
                deadline: &mut self.lease.deadline,
                stop: &mut self.lease.scope_stop,
                cancel: self.cancel.as_mut(),
                supervising: 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,
                },
                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> {
    /// 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 future = Box::pin(future);
        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 catch_unwind(AssertUnwindSafe(|| future.as_mut().poll(cx))) {
                        Ok(Poll::Pending) => Poll::Pending,
                        Ok(Poll::Ready(value)) => Poll::Ready(Ok(value)),
                        Err(_) => 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(future);
        guard.completed = true;
        outcome
    }
}

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,
        };
        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,
                },
                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 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();
        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(),
        });
        let mut scope = lease.into_serial_scope(std::future::pending());
        for expected in [0, 1] {
            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;
        }
        scope.finish_unentered_response(()).ok().unwrap();
        process.finish().unwrap();
        call.succeed();
        observer.flush().await.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() {
        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(BorrowedPending(&mut dropped)));
            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();
    }
}