saddle-runtime 0.3.23

//! Framework composition over an existing Tokio task, not a scheduler or admission issuer.
use crate::{diagnostics, profusegw::ProfuseGwManagedDispatch};
use saddle_core::{CleanupOutcome, DiagnosticOutcomeAxes, DiagnosticStage, OperationOutcome};
use saddle_observability::{
    DiagnosticTaskId, EmergencyDiagnosticHandle, FrameworkRequestFailure, RequestDiagnosticScope,
};
use std::{
    fmt::Write,
    future::Future,
    pin::Pin,
    sync::{Arc, Mutex},
    task::{Context, Poll},
};

struct State {
    scope: RequestDiagnosticScope<'static>,
    output: Option<EmergencyDiagnosticHandle>,
    id: Option<tokio::task::Id>,
    panic: Option<diagnostics::RequestCaptured>,
    cancelled: Option<diagnostics::RequestCaptured>,
    cleanup: Option<diagnostics::RequestCaptured>,
    control_failure: Option<saddle_observability::RequestSourceReceipt>,
    finished: bool,
}

/// Observational context only. No request owner, permit, timer or task issuer.
pub struct RequestTaskContext(Arc<Mutex<State>>);
impl RequestTaskContext {
    /// Retain the actual entry phase for cancellation/panic observed on the next
    /// poll or drop. No execution identity or resource state is changed.
    pub fn set_phase(&self, phase: saddle_observability::DiagnosticRequestPhase) {
        let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
        if !self.valid_poll(&mut state) { return; }
        state.scope = state.scope.reborrow().with_phase(phase);
        let snapshot = state.scope.reborrow();
        drop(state);
        diagnostics::refresh_task_scope(snapshot);
    }

    /// SE supplies only the projection of the validated ingress zone. Keep it
    /// with the original task so join/panic and downstream snapshots agree.
    pub fn bind_ingress_zone(&self, zone: Result<saddle_observability::DiagnosticZone, saddle_observability::DiagnosticZoneError>) {
        let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
        if !self.valid_poll(&mut state) { return; }
        state.scope = match zone {
            Ok(zone) => state.scope.reborrow().with_zone(zone),
            Err(_) => state.scope.reborrow().with_zone_missing(saddle_observability::DiagnosticContextMissing::Unavailable),
        };
        let snapshot = state.scope.reborrow();
        drop(state);
        diagnostics::refresh_task_scope(snapshot);
    }
    fn valid_poll(&self, state: &mut State) -> bool {
        if !state.finished && state.id.is_some() && state.id == tokio::task::try_id() {
            return true;
        }
        if state.control_failure.is_none() {
            state.control_failure = Some(state.scope.reborrow().with_output(state.output.as_ref()).fail(
                (), saddle_core::DiagnosticCategory::UnexpectedError,
                cause("runtime.request_task_context_conflict")));
        }
        false
    }
    pub fn bind_request_identity(&self, request: &saddle_observability::RequestIdentity) {
        let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
        if !self.valid_poll(&mut state) { return; }
        match state.scope.reborrow().bind_request_identity(request) {
            Ok(scope) => state.scope = scope,
            Err(_) => {
                if state.control_failure.is_none() {
                    state.control_failure = Some(state.scope.reborrow().with_output(state.output.as_ref()).fail(
                        (), saddle_core::DiagnosticCategory::UnexpectedError,
                        cause("runtime.request_task_context_conflict")));
                }
            }
        }
        let snapshot = state.scope.reborrow();
        drop(state);
        diagnostics::refresh_task_scope(snapshot);
    }
    /// Snapshot known facts; callers may enrich through OE's checked bind API.
    pub fn scope(&self) -> RequestDiagnosticScope<'static> {
        self.0
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .scope
            .reborrow()
    }
    /// Retain a checked enrichment for subsequent poll/drop and join snapshots.
    /// Task identity is checked again against the actual runtime ID on every poll.
    pub fn bind_established(
        &self,
        call: &saddle_observability::CallContext,
        event: &saddle_observability::EventContext,
    ) -> Result<(), FrameworkRequestFailure<saddle_observability::DiagnosticContextField>> {
        let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
        let next = if state.finished || state.id.is_none() || state.id != tokio::task::try_id() {
            Err(saddle_observability::DiagnosticContextField::Task)
        } else {
            state
                .scope
                .reborrow()
                .bind_established(call, event)
                .map_err(|e| e.field())
        };
        let next = match next {
            Ok(next) => next,
            Err(field) => {
                return Err(state
                    .scope
                    .reborrow()
                    .with_output(state.output.as_ref())
                    .fail(
                        field,
                        saddle_core::DiagnosticCategory::UnexpectedError,
                        cause("runtime.request_task_context_conflict"),
                    ))
            }
        };
        state.scope = next;
        let snapshot = state.scope.reborrow();
        drop(state);
        diagnostics::refresh_task_scope(snapshot);
        Ok(())
    }
    pub fn output(&self) -> Option<EmergencyDiagnosticHandle> {
        self.0
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .output
            .clone()
    }
}

/// Must remain with the corresponding JoinSet entry until its result is consumed.
pub struct RequestTaskJoin(Arc<Mutex<State>>);

/// The original operation outcome and independent destructor failure are retained.
#[must_use]
pub struct RequestTaskOutcome<T, E> {
    result: Option<Result<T, FrameworkRequestFailure<E>>>,
    state: Arc<Mutex<State>>,
    join_failure: Option<RequestTaskJoinFailure>,
}

/// Original Tokio failure and original source stay inseparable at this boundary.
pub struct RequestTaskJoinFailure {
    error: tokio::task::JoinError,
    source: diagnostics::RequestCaptured,
}
impl RequestTaskJoinFailure {
    pub fn error(&self) -> &tokio::task::JoinError {
        &self.error
    }
}

/// Returned to the caller unchanged on a foreign join pairing attempt.
pub struct RequestTaskJoinMismatch<T, E> {
    pub ticket: RequestTaskJoin,
    pub result: Result<RequestTaskOutcome<T, E>, tokio::task::JoinError>,
}

/// A wrapper polled by Service's existing JoinSet. Dropping it catches the body
/// destructor before releasing its diagnostic association; no second task exists.
pub struct RequestTaskFuture<F> {
    body: Option<Pin<Box<F>>>,
    state: Arc<Mutex<State>>,
    completed: bool,
}

fn task_id(id: tokio::task::Id) -> Option<DiagnosticTaskId> {
    struct Buffer {
        bytes: [u8; 64],
        len: usize,
    }
    impl std::fmt::Write for Buffer {
        fn write_str(&mut self, value: &str) -> std::fmt::Result {
            let end = self.len.checked_add(value.len()).ok_or(std::fmt::Error)?;
            if end > self.bytes.len() {
                return Err(std::fmt::Error);
            }
            self.bytes[self.len..end].copy_from_slice(value.as_bytes());
            self.len = end;
            Ok(())
        }
    }
    let mut buffer = Buffer {
        bytes: [0; 64],
        len: 0,
    };
    write!(&mut buffer, "{id}").ok()?;
    DiagnosticTaskId::from_runtime_id(std::str::from_utf8(&buffer.bytes[..buffer.len]).ok()?)
}

fn cause(code: &'static str) -> saddle_core::BoundedDiagnosticCause {
    saddle_core::BoundedDiagnosticCause::new(
        DiagnosticStage::BackgroundTask,
        saddle_core::DiagnosticCode::new(code).expect("static schema code"),
    )
}

fn pair<F: Future>(
    scope: RequestDiagnosticScope<'_>,
    output: Option<EmergencyDiagnosticHandle>,
    make: impl FnOnce(RequestTaskContext) -> F,
) -> (
    RequestTaskFuture<impl Future<Output = F::Output>>,
    RequestTaskJoin,
) {
    let state = Arc::new(Mutex::new(State {
        scope: scope.with_output(None),
        output,
        id: None,
        panic: None,
        cancelled: None,
        cleanup: None,
        control_failure: None,
        finished: false,
    }));
    let mut context = Some(RequestTaskContext(Arc::clone(&state)));
    let mut make = Some(make);
    let mut inner: Option<Pin<Box<F>>> = None;
    // Keep the Ready future alive until the outer cleanup catch: an async
    // `make(context).await` would drop it inside poll and could erase Ready(T).
    let body = std::future::poll_fn(move |cx| {
        if inner.is_none() {
            inner = Some(Box::pin(make.take().expect("factory once")(
                context.take().expect("context once"),
            )));
        }
        inner.as_mut().expect("constructed body").as_mut().poll(cx)
    });
    (
        RequestTaskFuture {
            body: Some(Box::pin(body)),
            state: Arc::clone(&state),
            completed: false,
        },
        RequestTaskJoin(state),
    )
}

/// Borrowed work cannot move its supervisor's owner into the abortable future.
pub type RequestTaskBorrowedFuture<'a, T, E> =
    Pin<Box<dyn Future<Output = Result<T, FrameworkRequestFailure<E>>> + Send + 'a>>;

/// Retained by the original request supervisor, outside the spawned future.
/// No Clone or owner accessor exists before the matching join is consumed.
/// ```compile_fail
/// fn duplicate<O>(ticket: saddle_runtime::request_task::RequestTaskOwnedJoin<O>) {
///     let _ = ticket.clone();
/// }
/// ```
pub struct RequestTaskOwnedJoin<O> {
    join: RequestTaskJoin,
    owner: Arc<tokio::sync::Mutex<O>>,
}

pub struct RequestTaskOwnedJoinMismatch<O, T, E> {
    pub ticket: RequestTaskOwnedJoin<O>,
    pub result: Result<RequestTaskOutcome<T, E>, tokio::task::JoinError>,
}

impl<O> RequestTaskOwnedJoin<O> {
    pub fn bind_spawned(self, id: tokio::task::Id) -> Result<Self, Self> {
        match self.join.bind_spawned(id) {
            Ok(join) => Ok(Self {
                join,
                owner: self.owner,
            }),
            Err(join) => Err(Self {
                join,
                owner: self.owner,
            }),
        }
    }

    /// The join proves the borrowed future (including its destructor) is gone.
    /// This returns resources, NOT a physical disposition or a ZERO assertion.
    pub fn complete<T, E>(
        self,
        result: Result<RequestTaskOutcome<T, E>, tokio::task::JoinError>,
    ) -> Result<(O, RequestTaskOutcome<T, E>), RequestTaskOwnedJoinMismatch<O, T, E>> {
        match self.join.complete(result) {
            Ok(outcome) => {
                // The only other Arc is inside the future; it cannot escape via
                // the borrowed factory and matching join has destroyed it.
                let owner = match Arc::try_unwrap(self.owner) {
                    Ok(owner) => owner.into_inner(),
                    Err(_) => unreachable!("joined task no longer retains owner storage"),
                };
                Ok((owner, outcome))
            }
            Err(failure) => Err(RequestTaskOwnedJoinMismatch {
                ticket: Self {
                    join: failure.ticket,
                    owner: self.owner,
                },
                result: failure.result,
            }),
        }
    }
}

/// Framework composition for a retained request/physical phase owner. The
/// factory only borrows it. A caller-defined phase aggregate must likewise keep
/// its unique cleanup authority in-place across awaits; this is not permission
/// to `take` that authority into a nested abortable future.
pub fn owned_request_task<O: Send + 'static, T: 'static, E: 'static>(
    owner: O,
    scope: RequestDiagnosticScope<'_>,
    output: Option<EmergencyDiagnosticHandle>,
    make: impl for<'a> FnOnce(&'a mut O, RequestTaskContext) -> RequestTaskBorrowedFuture<'a, T, E>
        + Send
        + 'static,
) -> (
    RequestTaskFuture<impl Future<Output = Result<T, FrameworkRequestFailure<E>>> + Send>,
    RequestTaskOwnedJoin<O>,
) {
    let owner = Arc::new(tokio::sync::Mutex::new(owner));
    let body_owner = Arc::clone(&owner);
    let (future, join) = pair(scope, output, move |context| async move {
        // Exactly one internal borrower; no lock acquisition/wait queue is
        // exposed, and the supervisor cannot access the cell before join.
        let mut guard = body_owner
            .try_lock_owned()
            .expect("single task owner borrower");
        let state = Arc::clone(&context.0);
        let mut body = make(&mut *guard, context);
        let result = std::future::poll_fn(|cx| body.as_mut().poll(cx)).await;
        // Keep Ready(T/E) outside the destructor catch. Dropping an awaited
        // future implicitly here would otherwise replace that result on panic.
        let primary = result
            .as_ref()
            .err()
            .map(|failure| failure.source_diagnostic().occurrence());
        drop_task_body(&state, body, primary);
        result
    });
    (future, RequestTaskOwnedJoin { join, owner })
}

/// The admitted owner is retained outside the abortable body until join.
/// ```compile_fail
/// use saddle_runtime::profusegw::ProfuseGwManagedDispatch;
/// use saddle_runtime::request_task::admitted_request_task;
/// use saddle_observability::{RequestDiagnosticScope, FrameworkRequestFailure};
/// fn move_borrowed(owner: ProfuseGwManagedDispatch, scope: RequestDiagnosticScope<'static>) {
///     let _ = admitted_request_task(owner, scope, None, |borrowed, _| Box::pin(async move {
///         borrowed.cancel();
///         Ok::<(), FrameworkRequestFailure<()>>(())
///     }));
/// }
/// ```
pub fn admitted_request_task<T: 'static, E: 'static>(
    owner: ProfuseGwManagedDispatch,
    scope: RequestDiagnosticScope<'_>,
    output: Option<EmergencyDiagnosticHandle>,
    make: impl for<'a> FnOnce(
            &'a mut ProfuseGwManagedDispatch,
            RequestTaskContext,
        ) -> RequestTaskBorrowedFuture<'a, T, E>
        + Send
        + 'static,
) -> (
    RequestTaskFuture<impl Future<Output = Result<T, FrameworkRequestFailure<E>>> + Send>,
    RequestTaskOwnedJoin<ProfuseGwManagedDispatch>,
) {
    owned_request_task(owner, scope, output, make)
}

/// No admitted owner is minted. Service must retain its existing bounded
/// rejection JoinSet limit before constructing/spawning this task.
pub fn rejection_request_task<F: Future>(
    scope: RequestDiagnosticScope<'_>,
    output: Option<EmergencyDiagnosticHandle>,
    make: impl FnOnce(RequestTaskContext) -> F,
) -> (
    RequestTaskFuture<impl Future<Output = F::Output>>,
    RequestTaskJoin,
) {
    pair(scope, output, make)
}

impl<F> Unpin for RequestTaskFuture<F> {}
impl<T, E, F: Future<Output = Result<T, FrameworkRequestFailure<E>>>> Future
    for RequestTaskFuture<F>
{
    type Output = RequestTaskOutcome<T, E>;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        let scope = {
            let mut state = this.state.lock().unwrap_or_else(|p| p.into_inner());
            let actual = tokio::task::try_id();
            let next = actual
                .filter(|id| state.id.is_none_or(|bound| bound == *id))
                .and_then(task_id)
                .and_then(|id| state.scope.reborrow().with_task(id).ok());
            if let Some(next) = next {
                state.id = actual;
                state.scope = next;
            } else {
                state.control_failure = Some(
                    state
                        .scope
                        .reborrow()
                        .with_output(state.output.as_ref())
                        .fail(
                            (),
                            saddle_core::DiagnosticCategory::UnexpectedError,
                            cause("runtime.request_task_identity_unavailable_or_foreign"),
                        ),
                );
                drop(state);
                this.drop_body();
                this.completed = true;
                return Poll::Ready(RequestTaskOutcome {
                    result: None,
                    state: Arc::clone(&this.state),
                    join_failure: None,
                });
            }
            state.scope.reborrow()
        };
        let output = this
            .state
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .output
            .clone();
        // The actual poll catch remains synchronous and uses the sole Facade hook.
        // No lock is held while business code polls or updates its context.
        let (result, _) = diagnostics::catching_request_with_output(
            DiagnosticStage::RequestDb,
            "runtime.formal_request_task",
            None,
            Some(scope),
            output.clone(),
            None,
            || {
                this.body
                    .as_mut()
                    .expect("task not polled after Ready")
                    .as_mut()
                    .poll(cx)
            },
        );
        let result = match result {
            Ok(Poll::Pending) => return Poll::Pending,
            Ok(Poll::Ready(value)) => Some(value),
            Err(panic) => {
                this.state.lock().unwrap_or_else(|p| p.into_inner()).panic = Some(panic);
                None
            }
        };
        drop(output);
        let primary = result
            .as_ref()
            .and_then(|result| result.as_ref().err())
            .map(|failure| failure.source_diagnostic().occurrence());
        this.drop_body_with_primary(primary);
        this.completed = true;
        Poll::Ready(RequestTaskOutcome {
            result,
            state: Arc::clone(&this.state),
            join_failure: None,
        })
    }
}

impl<F> RequestTaskFuture<F> {
    fn drop_body(&mut self) {
        self.drop_body_with_primary(None);
    }
    fn drop_body_with_primary(&mut self, primary: Option<saddle_core::DiagnosticOccurrence>) {
        let Some(body) = self.body.take() else { return };
        drop_task_body(&self.state, body, primary);
    }
}

fn drop_task_body<B>(
    task_state: &Arc<Mutex<State>>,
    body: B,
    occurrence: Option<saddle_core::DiagnosticOccurrence>,
) {
    let (scope, output, primary, was_cancelled) = {
        let mut state = task_state.lock().unwrap_or_else(|p| p.into_inner());
        let was_cancelled = state.panic.is_none() && state.cancelled.is_some();
        (
            state.scope.reborrow(),
            state.output.clone(),
            state.panic.take().or_else(|| state.cancelled.take()),
            was_cancelled,
        )
    };
    let (cleanup, primary) = diagnostics::catching_request_with_output(
        DiagnosticStage::FinalizerResource,
        "runtime.formal_request_task_drop",
        None,
        Some(scope),
        output,
        primary,
        || {
            diagnostics::link_task_cleanup(occurrence);
            drop(body)
        },
    );
    let mut state = task_state.lock().unwrap_or_else(|p| p.into_inner());
    if was_cancelled {
        state.cancelled = primary;
    } else {
        state.panic = primary;
    }
    if let Err(cleanup) = cleanup {
        state.cleanup = Some(cleanup);
    }
}
impl<F> Drop for RequestTaskFuture<F> {
    fn drop(&mut self) {
        if !self.completed && self.body.is_some() {
            let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
            let diagnostic = diagnostics::failure(
                DiagnosticStage::BackgroundTask,
                saddle_core::DiagnosticCategory::ExpectedRejection,
                "runtime.request_task_cancelled",
            );
            state.cancelled = Some(diagnostics::capture_task_source(
                diagnostic,
                &state.scope,
                state.output.as_ref(),
            ));
        }
        self.drop_body();
    }
}

impl RequestTaskJoin {
    /// Bind the ID returned by the SAME JoinSet::spawn AbortHandle before polling
    /// any join result. No decimal/string/task-kind issuer is exposed here.
    pub fn bind_spawned(self, id: tokio::task::Id) -> Result<Self, Self> {
        {
            let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
            if state.id.is_some_and(|existing| existing != id) {
                drop(state);
                return Err(self);
            }
            let Some(diagnostic_id) = task_id(id) else {
                drop(state);
                return Err(self);
            };
            let scope = match state.scope.reborrow().with_task(diagnostic_id) {
                Ok(scope) => scope,
                Err(_) => {
                    drop(state);
                    return Err(self);
                }
            };
            state.id = Some(id);
            state.scope = scope;
        }
        Ok(self)
    }
    /// Consumes only this task's outcome. A foreign result returns both inputs.
    pub fn complete<T, E>(
        self,
        result: Result<RequestTaskOutcome<T, E>, tokio::task::JoinError>,
    ) -> Result<RequestTaskOutcome<T, E>, RequestTaskJoinMismatch<T, E>> {
        let matches = match &result {
            Ok(outcome) => Arc::ptr_eq(&self.0, &outcome.state),
            Err(error) => self.0.lock().unwrap_or_else(|p| p.into_inner()).id == Some(error.id()),
        };
        if !matches {
            return Err(RequestTaskJoinMismatch {
                ticket: self,
                result,
            });
        }
        match result {
            Ok(outcome) => Ok(outcome),
            Err(error) => {
                let source = {
                    let mut state = self.0.lock().unwrap_or_else(|p| p.into_inner());
                    state.cancelled.take().unwrap_or_else(|| {
                        let diagnostic = diagnostics::failure(
                            DiagnosticStage::BackgroundTask,
                            saddle_core::DiagnosticCategory::UnexpectedError,
                            "runtime.request_task_join_failed",
                        );
                        diagnostics::capture_task_source(
                            diagnostic,
                            &state.scope,
                            state.output.as_ref(),
                        )
                    })
                };
                Ok(RequestTaskOutcome {
                    result: None,
                    state: self.0,
                    join_failure: Some(RequestTaskJoinFailure { error, source }),
                })
            }
        }
    }
}

impl<T, E> RequestTaskOutcome<T, E> {
    /// Borrow original business/technical result without stripping its receipt.
    pub fn result(&self) -> Option<&Result<T, FrameworkRequestFailure<E>>> {
        self.result.as_ref()
    }
    pub fn join_failure(&self) -> Option<&RequestTaskJoinFailure> {
        self.join_failure.as_ref()
    }
    pub fn panicked(&self) -> bool {
        self.state
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .panic
            .is_some()
    }
    pub fn cleanup_failed(&self) -> bool {
        self.state
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .cleanup
            .is_some()
    }
    pub fn control_failed(&self) -> bool {
        self.state
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .control_failure
            .is_some()
    }
    /// A terminal consumes this carrier; returned carrier still holds all facts.
    pub fn finish(self) -> RequestTaskTerminal<T, E> {
        let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
        state.finished = true;
        let axes = DiagnosticOutcomeAxes {
            operation: if state.panic.is_some() {
                OperationOutcome::Panicked
            } else if let Some(source) = &self.join_failure {
                if source.error().is_cancelled() {
                    OperationOutcome::Cancelled
                } else {
                    OperationOutcome::Failed
                }
            } else if state.control_failure.is_some() || matches!(&self.result, Some(Err(_))) {
                OperationOutcome::Failed
            } else {
                OperationOutcome::Succeeded
            },
            cleanup: if state.cleanup.is_some() {
                CleanupOutcome::Failed
            } else {
                CleanupOutcome::Succeeded
            },
            ..Default::default()
        };
        if let Some(Err(source)) = &self.result {
            let _ = source.record_boundary_optional(state.output.as_ref(), &axes);
        }
        if let Some(source) = &self.join_failure {
            source
                .source
                .record_with_output(state.output.as_ref(), &axes);
        }
        if let Some(source) = &state.control_failure {
            let _ = source.record_boundary_optional(state.output.as_ref(), &axes);
        }
        if let Some(source) = &state.panic {
            source.record_with_output(state.output.as_ref(), &axes);
        }
        if let Some(source) = &state.cleanup {
            source.record_with_output(
                state.output.as_ref(),
                &DiagnosticOutcomeAxes {
                    cleanup: CleanupOutcome::Failed,
                    ..Default::default()
                },
            );
        }
        drop(state);
        RequestTaskTerminal(self)
    }
}
/// No second finish method: retain the terminal with the enclosing task report.
#[must_use]
pub struct RequestTaskTerminal<T, E>(RequestTaskOutcome<T, E>);
impl<T, E> RequestTaskTerminal<T, E> {
    pub fn outcome(&self) -> &RequestTaskOutcome<T, E> {
        &self.0
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use saddle_observability::{
        EarlyRequestContext, EmergencyDiagnostics, FileLoggingConfig, Rotation,
    };
    fn early() -> RequestDiagnosticScope<'static> {
        RequestDiagnosticScope::early(None, EarlyRequestContext::socket_accepted("app"))
    }
    type TaskResult = Result<u32, FrameworkRequestFailure<u32>>;

    #[tokio::test]
    async fn owned_task_storage_released_after_terminal() {
        let (future,ticket)=owned_request_task(41u32,early(),None,|value,_| Box::pin(async move {
            Ok::<u32,FrameworkRequestFailure<()>>(*value)
        }));
        let state=Arc::downgrade(&ticket.join.0);
        let storage=Arc::downgrade(&ticket.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!(owner,41);
        assert!(storage.upgrade().is_none(),"owner cell freed at join recovery");
        assert!(state.upgrade().is_some(),"terminal still retains source facts");
        let terminal=outcome.finish();drop(terminal);
        assert!(state.upgrade().is_none(),"last diagnostic state released");
    }

    #[test]
    fn task_storage_payload_layout() {
        // Payload layouts only; allocator/Arc headers and caller future size
        // are additional. This is not a reservation or an allocation oracle.
        eprintln!("RP_STORAGE state={} mutex_state={} dispatch={} owner_mutex={} join={} owned_join={} old_task_payload={}",
            std::mem::size_of::<State>(), std::mem::size_of::<Mutex<State>>(),
            std::mem::size_of::<ProfuseGwManagedDispatch>(),
            std::mem::size_of::<tokio::sync::Mutex<ProfuseGwManagedDispatch>>(),
            std::mem::size_of::<RequestTaskJoin>(),
            std::mem::size_of::<RequestTaskOwnedJoin<ProfuseGwManagedDispatch>>(),
            std::mem::size_of::<std::time::Instant>() + std::mem::size_of::<saddle_admission::ProfuseGwCapacitySnapshot>());
        assert!(std::mem::size_of::<State>() > std::mem::size_of::<std::time::Instant>()
            + std::mem::size_of::<saddle_admission::ProfuseGwCapacitySnapshot>());
    }

    #[tokio::test]
    async fn real_tasks_retain_identity_errors_and_cleanup() {
        const CHILD: &str = "RP_TASK_DIAGNOSTIC_CHILD";
        let Some(path) = std::env::var_os(CHILD) else {
            let path = std::env::temp_dir().join(format!("saddle-rp-tasks-{}", std::process::id()));
            std::fs::create_dir(&path).unwrap();
            let mut child = std::process::Command::new(std::env::current_exe().unwrap())
                .args([
                    "--exact",
                    "request_task::tests::real_tasks_retain_identity_errors_and_cleanup",
                    "--nocapture",
                ])
                .env(CHILD, &path)
                .spawn()
                .unwrap();
            let started = std::time::Instant::now();
            let status = loop {
                if let Some(status) = child.try_wait().unwrap() {
                    break status;
                }
                if started.elapsed() > std::time::Duration::from_secs(10) {
                    child.kill().unwrap();
                    child.wait().unwrap();
                    panic!("task child timeout")
                }
                std::thread::sleep(std::time::Duration::from_millis(10));
            };
            assert!(status.success());
            std::fs::remove_file(path.join("saddle.emergency.log")).unwrap();
            std::fs::remove_dir(path).unwrap();
            return;
        };
        let output = EmergencyDiagnostics::start(&FileLoggingConfig::new(
            std::path::PathBuf::from(&path),
            Rotation::Daily,
        ))
        .unwrap();
        std::panic::set_hook(Box::new(|info| {
            let _ = diagnostics::capture_current_panic(info);
        }));
        struct Body {
            panic_poll: bool,
            panic_drop: bool,
        }
        impl Future for Body {
            type Output = TaskResult;
            fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<TaskResult> {
                if self.panic_poll {
                    panic!("PRIVATE_TASK_PANIC")
                };
                Poll::Ready(Ok(41))
            }
        }
        impl Drop for Body {
            fn drop(&mut self) {
                if self.panic_drop {
                    panic!("PRIVATE_TASK_DROP")
                }
            }
        }
        let mut identities = Vec::new();
        for (panic_poll, panic_drop) in [(false, false), (true, false), (false, true), (true, true)]
        {
            let (future, ticket) =
                rejection_request_task(early(), Some(output.handle()), move |_| Body {
                    panic_poll,
                    panic_drop,
                });
            let mut tasks = tokio::task::JoinSet::new();
            let abort = tasks.spawn(future);
            identities.push(abort.id().to_string());
            let ticket = ticket.bind_spawned(abort.id()).ok().unwrap();
            let terminal = ticket
                .complete(tasks.join_next().await.unwrap())
                .ok()
                .unwrap()
                .finish();
            assert_eq!(terminal.outcome().panicked(), panic_poll);
            assert_eq!(terminal.outcome().cleanup_failed(), panic_drop);
            if !panic_poll {
                assert_eq!(
                    *terminal.outcome().result().unwrap().as_ref().ok().unwrap(),
                    41
                );
            }
        }
        // Same-poll enrichment reaches the original hook, not only the next poll.
        let (future, ticket) =
            rejection_request_task(early(), Some(output.handle()), move |context| async move {
                let observer = saddle_observability::Observer::with_writer(
                    Default::default(),
                    std::io::sink(),
                )
                .unwrap();
                let (call, _) = observer
                    .start_external_call_checked(
                        "app",
                        "module",
                        "service",
                        "route",
                        Some("original-trace"),
                    )
                    .unwrap();
                let event = saddle_observability::EventContext::new(
                    saddle_observability::RequestIdentity::new("request").unwrap(),
                    saddle_observability::RouteIdentity::new("route").unwrap(),
                    1,
                )
                .unwrap();
                assert!(context.bind_established(call.context(), &event).is_ok());
                panic!("PRIVATE_ENRICHED_PANIC");
                #[allow(unreachable_code)]
                Ok::<u32, FrameworkRequestFailure<u32>>(0)
            });
        let task = tokio::spawn(future);
        let ticket = ticket.bind_spawned(task.id()).ok().unwrap();
        assert!(ticket
            .complete(task.await)
            .ok()
            .unwrap()
            .finish()
            .outcome()
            .panicked());

        // Cancel BEFORE first poll still has the exact spawn identity at join.
        let (future, ticket) = rejection_request_task(early(), Some(output.handle()), |_| {
            std::future::pending::<TaskResult>()
        });
        let task = tokio::spawn(future);
        let ticket = ticket.bind_spawned(task.id()).ok().unwrap();
        task.abort();
        let terminal = ticket.complete(task.await).ok().unwrap().finish();
        assert!(terminal
            .outcome()
            .join_failure()
            .unwrap()
            .error()
            .is_cancelled());

        // Pending task abort: same cancellation is recorded before its actual
        // destructor panic, then both original sources survive join.
        struct PendingDrop;
        impl Future for PendingDrop {
            type Output = TaskResult;
            fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<TaskResult> {
                Poll::Pending
            }
        }
        impl Drop for PendingDrop {
            fn drop(&mut self) {
                panic!("PRIVATE_PENDING_DROP")
            }
        }
        let (sent, received) = tokio::sync::oneshot::channel();
        let (future, ticket) = rejection_request_task(early(), Some(output.handle()), move |_| {
            let _ = sent.send(());
            PendingDrop
        });
        let task = tokio::spawn(future);
        let ticket = ticket.bind_spawned(task.id()).ok().unwrap();
        received.await.unwrap();
        task.abort();
        let terminal = ticket.complete(task.await).ok().unwrap().finish();
        assert!(terminal
            .outcome()
            .join_failure()
            .unwrap()
            .error()
            .is_cancelled());
        assert!(terminal.outcome().cleanup_failed());

        struct FailureDrop(Option<FrameworkRequestFailure<u32>>);
        impl Future for FailureDrop {
            type Output = TaskResult;
            fn poll(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<TaskResult> {
                Poll::Ready(Err(self.0.take().unwrap()))
            }
        }
        impl Drop for FailureDrop {
            fn drop(&mut self) {
                panic!("PRIVATE_BOUNDED_CLEANUP")
            }
        }
        let (future, ticket) =
            owned_request_task((), early(), Some(output.handle()), |_, context| {
                let handle = context.output();
                let primary = context.scope().with_output(handle.as_ref()).fail(
                    73,
                    saddle_core::DiagnosticCategory::UnexpectedError,
                    cause("runtime.task_primary_test"),
                );
                Box::pin(FailureDrop(Some(primary)))
            });
        let task = tokio::spawn(future);
        let ticket = ticket.bind_spawned(task.id()).ok().unwrap();
        let ((), outcome) = ticket.complete(task.await).ok().unwrap();
        let primary_id = outcome
            .result()
            .unwrap()
            .as_ref()
            .err()
            .unwrap()
            .source_diagnostic()
            .id();
        assert_eq!(
            *outcome.result().unwrap().as_ref().err().unwrap().error(),
            73
        );
        assert!(outcome.cleanup_failed());
        drop(outcome.finish());

        // Wrong ticket returns original inputs; both can then complete correctly.
        let (a, ta) = rejection_request_task(early(), None, |_| {
            std::future::ready(Ok::<u32, FrameworkRequestFailure<u32>>(1))
        });
        let (b, tb) = rejection_request_task(early(), None, |_| {
            std::future::ready(Ok::<u32, FrameworkRequestFailure<u32>>(2))
        });
        let a = tokio::spawn(a);
        let b = tokio::spawn(b);
        let ta = ta.bind_spawned(a.id()).ok().unwrap();
        let tb = tb.bind_spawned(b.id()).ok().unwrap();
        let ra = a.await;
        let rb = b.await;
        let mismatch = ta.complete(rb).err().unwrap();
        assert!(mismatch.ticket.complete(ra).is_ok());
        assert!(tb.complete(mismatch.result).is_ok());

        let exit = diagnostics::close_output(
            output,
            Some(std::time::Instant::now() + std::time::Duration::from_secs(2)),
        );
        assert_eq!(exit.snapshot.enqueued, exit.snapshot.written);
        assert_eq!(exit.snapshot.dropped, 0);
        let log =
            std::fs::read_to_string(std::path::PathBuf::from(path).join("saddle.emergency.log"))
                .unwrap();
        assert!(!log.contains("PRIVATE_"));
        let rows: Vec<serde_json::Value> = log
            .lines()
            .map(|line| serde_json::from_str(line).unwrap())
            .collect();
        let sources: Vec<_> = rows
            .iter()
            .filter(|row| !row["diagnostic"].is_null())
            .collect();
        assert_eq!(sources.len(), 10); // includes bounded primary + caught cleanup
        let linked_cleanup = sources
            .iter()
            .find(|source| source["diagnostic"]["primary_diagnostic_id"] == primary_id)
            .unwrap();
        assert_ne!(linked_cleanup["diagnostic"]["diagnostic_id"], primary_id);
        let enriched = sources
            .iter()
            .find(|r| r["context"]["trace_id"]["value"] == "original-trace")
            .unwrap();
        assert_eq!(
            enriched["diagnostic"]["stack_status"],
            "unavailable_deferred"
        );
        for source in sources {
            assert_eq!(source["context"]["task"]["state"], "present");
            let id = &source["diagnostic"]["diagnostic_id"];
            assert!(rows
                .iter()
                .any(|r| r["diagnostic"].is_null()
                    && &r["diagnostic_reference"]["diagnostic_id"] == id));
        }
    }
}