saddle-runtime 0.3.23

//! Framework composition only. Facade owns the ONE hook and emergency writer.
use saddle_core::{
    CaptureSite, Diagnostic, DiagnosticCategory, DiagnosticCause, DiagnosticCode, DiagnosticStage,
    SaddleError,
};
use saddle_observability::{CallContext, EmergencyDiagnosticHandle, EventContext, Observer};
use std::{
    cell::RefCell,
    future::Future,
    panic::{AssertUnwindSafe, PanicHookInfo, catch_unwind, resume_unwind},
    sync::OnceLock,
};

static OUTPUT: OnceLock<EmergencyDiagnosticHandle> = OnceLock::new();

/// Output observation, NOT a transaction/resource/durable-write receipt.
#[derive(Debug)]
pub struct RuntimeDiagnosticExit {
    pub shutdown: saddle_observability::DiagnosticShutdown,
    pub snapshot: saddle_observability::DiagnosticOutputSnapshot,
}

pub(crate) fn close_output(
    mut owner: saddle_observability::EmergencyDiagnostics,
    deadline: Option<std::time::Instant>,
) -> RuntimeDiagnosticExit {
    let shutdown = loop {
        let status = owner.shutdown();
        if status != saddle_observability::DiagnosticShutdown::Pending {
            break status;
        }
        let remaining = deadline
            .map(|d| d.saturating_duration_since(std::time::Instant::now()))
            .unwrap_or_default();
        if remaining.is_zero() {
            break status;
        }
        // Same absolute shutdown budget. No new async timer, writer join or
        // wait after expiration. Startup failure without a budget is immediate.
        std::thread::sleep(remaining.min(std::time::Duration::from_millis(1)));
    };
    RuntimeDiagnosticExit {
        shutdown,
        snapshot: owner.snapshot(),
    }
}
type RequestContext = (Observer, CallContext, EventContext);
struct Frame {
    stage: DiagnosticStage,
    task: &'static str,
    context: Option<RequestContext>,
    panic: Option<Diagnostic>,
    primary: Option<Diagnostic>,
    request_mode: bool,
    request_panic: Option<RequestCaptured>,
    request_primary: Option<RequestCaptured>,
    cleanup_occurrence: Option<saddle_core::DiagnosticOccurrence>,
    request_scope: Option<saddle_observability::RequestDiagnosticScope<'static>>,
    request_output: Option<EmergencyDiagnosticHandle>,
}

pub(crate) enum RequestCaptured {
    Established(saddle_observability::RequestBoundaryReference<Diagnostic>),
    MissingContext(Diagnostic),
}
impl RequestCaptured {
    pub(crate) fn diagnostic(&self) -> &Diagnostic {
        match self {
            Self::Established(reference) => reference.source_diagnostic(),
            Self::MissingContext(diagnostic) => diagnostic,
        }
    }
    pub(crate) fn occurrence(&self) -> saddle_core::DiagnosticOccurrence {
        self.diagnostic().occurrence()
    }
    pub(crate) fn record(&self, axes: &saddle_core::DiagnosticOutcomeAxes) {
        self.record_with_output(OUTPUT.get(), axes)
    }
    pub(crate) fn record_with_output(&self, output: Option<&EmergencyDiagnosticHandle>, axes: &saddle_core::DiagnosticOutcomeAxes) {
        match self {
            Self::Established(reference) => {
                let _ = reference.record_optional(output, axes);
            }
            Self::MissingContext(_) => boundary(Some(self.occurrence()), axes, None),
        }
    }
}

pub(crate) fn live_request_scope(
    projection: &saddle_core::DbScopeDiagnosticContext<(&CallContext, &EventContext)>,
) -> saddle_observability::RequestDiagnosticScope<'static> {
    saddle_observability::RequestDiagnosticScope::live_db_scope(OUTPUT.get(), projection)
}

fn capture_request(
    diagnostic: Diagnostic,
    context: Option<&RequestContext>,
    checked: Option<&saddle_observability::RequestDiagnosticScope<'static>>,
    output: Option<&EmergencyDiagnosticHandle>,
) -> RequestCaptured {
    if let Some(scope) = checked {
        return RequestCaptured::Established(
            scope.reborrow().with_output(output).capture_existing(diagnostic, context.map(|(observer, _, _)| observer)).into_reference(),
        );
    }
    if let Some((observer, call, event)) = context {
        if let Some(scope) = checked {
            return RequestCaptured::Established(
                scope
                    .capture_existing(diagnostic, Some(observer))
                    .into_reference(),
            );
        }
        let scope = match OUTPUT.get() {
            Some(output) => {
                saddle_observability::RequestDiagnosticScope::established(output, call, event)
            }
            None => saddle_observability::RequestDiagnosticScope::output_unavailable(call, event),
        };
        RequestCaptured::Established(
            scope
                .capture_existing(diagnostic, Some(observer))
                .into_reference(),
        )
    } else {
        // No fabricated context or receipt: this legacy degraded case remains
        // explicitly outside the established-context enforcement claim.
        emit(&diagnostic, None);
        RequestCaptured::MissingContext(diagnostic)
    }
}
pub(crate) fn capture_task_source(
    diagnostic: Diagnostic,
    scope: &saddle_observability::RequestDiagnosticScope<'static>,
    output: Option<&EmergencyDiagnosticHandle>,
) -> RequestCaptured {
    capture_request(diagnostic, None, Some(scope), output)
}
thread_local! { static CURRENT: RefCell<Option<Frame>> = const { RefCell::new(None) }; }

// Called only after the task context verifies the currently executing Tokio ID.
// Updates the same poll's hook frame, not a process-global context registry.
pub(crate) fn refresh_task_scope(scope: saddle_observability::RequestDiagnosticScope<'static>) {
    CURRENT.with(|slot| {
        if let Ok(mut slot) = slot.try_borrow_mut() {
            if let Some(frame) = slot.as_mut() {
                if frame.request_mode && frame.task == "runtime.formal_request_task" {
                    frame.request_scope = Some(scope);
                }
            }
        }
    });
}

/// Called once by Facade with its existing writer handle. No thread/hook is created.
pub fn install_output(handle: EmergencyDiagnosticHandle) -> Result<(), EmergencyDiagnosticHandle> {
    OUTPUT.set(handle)
}

/// Called by Facade's sole panic hook. False means no Runtime poll context;
/// the Facade must capture that unassociated panic itself. Never reads payload.
pub fn capture_current_panic(info: &PanicHookInfo<'_>) -> bool {
    CURRENT.with(|slot| {
        let Ok(mut slot) = slot.try_borrow_mut() else {
            return false;
        };
        let Some(frame) = slot.as_mut() else {
            return false;
        };
        let mut diagnostic =
            Diagnostic::capture_panic(info, frame.stage).with_task(code(frame.task));
        if let Some(primary) = frame.primary.as_ref() {
            diagnostic = diagnostic.during_cleanup_of(primary);
        }
        if let Some(primary) = frame.request_primary.as_ref() {
            diagnostic = diagnostic.during_cleanup_of(primary.diagnostic());
        }
        if let Some(primary) = frame.cleanup_occurrence.as_ref() {
            diagnostic = diagnostic.during_cleanup_of_occurrence(primary);
        }
        if frame.request_mode {
            frame.request_panic = Some(capture_request(
                diagnostic,
                frame.context.as_ref(),
                frame.request_scope.as_ref(),
                frame.request_output.as_ref(),
            ));
            return true;
        }
        emit(&diagnostic, frame.context.as_ref());
        frame.panic = Some(diagnostic);
        true
    })
}

fn code(value: &'static str) -> DiagnosticCode {
    DiagnosticCode::new(value).expect("Runtime diagnostic codes are static schema identifiers")
}

/// Fixed source detail, before a lossy conversion or fail-closed exit. It never
/// formats an error payload, allocates a stack, or changes the execution result.
#[track_caller]
pub(crate) fn bounded_source(
    stage: DiagnosticStage,
    name: &'static str,
    axes: &saddle_core::DiagnosticOutcomeAxes,
) -> (
    saddle_core::BoundedDiagnostic,
    Option<saddle_observability::DiagnosticSubmission>,
) {
    let diagnostic = saddle_core::BoundedDiagnostic::capture(
        DiagnosticCategory::UnexpectedError,
        CaptureSite::FirstObserved,
        saddle_core::BoundedDiagnosticCause::new(stage, code(name)),
    );
    let submission = OUTPUT
        .get()
        .map(|output| output.submit_bounded(Some(&diagnostic), axes, None));
    #[cfg(test)]
    if name == "runtime.driver_ledger_shutdown_failed"
        && std::env::var_os("RUNTIME_LEDGER_SOURCE_TEST").is_some()
    {
        // Fixed test-only observation of submission before the real abort.
        // NOT writer acknowledgement; never available in published builds.
        eprintln!("RUNTIME_LEDGER_SOURCE_SUBMISSION={submission:?}");
    }
    (diagnostic, submission)
}

#[track_caller]
pub(crate) fn request_stop_source(
    timed_out: bool,
    physical_return: bool,
    stage: DiagnosticStage,
    context: Option<(&CallContext, &EventContext)>,
) -> saddle_core::DiagnosticOccurrence {
    let diagnostic = saddle_core::BoundedDiagnostic::capture(
        DiagnosticCategory::ExpectedRejection,
        CaptureSite::FirstObserved,
        saddle_core::BoundedDiagnosticCause::new(
            stage,
            code(if physical_return && timed_out {
                "runtime.physical_return_deadline"
            } else if physical_return {
                "runtime.physical_return_cancelled"
            } else if timed_out {
                "runtime.scope_deadline"
            } else {
                "runtime.scope_cancelled"
            }),
        ),
    );
    if let Some(output) = OUTPUT.get() {
        let _submission = output.submit_bounded(
            Some(&diagnostic),
            &saddle_core::DiagnosticOutcomeAxes {
                operation: if physical_return {
                    saddle_core::OperationOutcome::Unknown
                } else if timed_out {
                    saddle_core::OperationOutcome::TimedOut
                } else {
                    saddle_core::OperationOutcome::Cancelled
                },
                ..Default::default()
            },
            context,
        );
    }
    diagnostic.occurrence()
}

pub(crate) fn boundary(
    occurrence: Option<saddle_core::DiagnosticOccurrence>,
    axes: &saddle_core::DiagnosticOutcomeAxes,
    context: Option<(&CallContext, &EventContext)>,
) {
    if let Some(output) = OUTPUT.get() {
        let _submission = output.submit_boundary(occurrence, axes, context);
    }
}

/// Abort is unchanged. A finite submission is not a written acknowledgement;
/// the existing sink records Full/Closed/encoding degradation when available.
#[track_caller]
pub(crate) fn finalizer_contract_abort(name: &'static str) -> ! {
    let _source = bounded_source(
        DiagnosticStage::FinalizerResource,
        name,
        &saddle_core::DiagnosticOutcomeAxes {
            cleanup: saddle_core::CleanupOutcome::Failed,
            ..Default::default()
        },
    );
    std::process::abort()
}

pub(crate) fn emit(diagnostic: &Diagnostic, context: Option<&RequestContext>) {
    if let Some(output) = OUTPUT.get() {
        if let Some((observer, call, event)) = context {
            let _ = observer.record_diagnostic(diagnostic, output, Some((call, event)));
        } else {
            let _ = output.submit(diagnostic);
        }
    }
    // Absence/Full/Closed is NOT a delivery acknowledgement. Facade retains
    // the writer owner and must inspect its snapshot at the ORIGINAL deadline.
}

#[track_caller]
pub(crate) fn failure(
    stage: DiagnosticStage,
    category: DiagnosticCategory,
    name: &'static str,
) -> Diagnostic {
    Diagnostic::capture(
        category,
        CaptureSite::FirstObserved,
        DiagnosticCause::new(stage, code(name)),
    )
}

/// Invoke at the existing catch boundary, INSIDE Admission's admitted poll.
/// CURRENT is replaced/restored per synchronous poll, never held across await.
pub(crate) fn catching<T>(
    stage: DiagnosticStage,
    task: &'static str,
    context: Option<RequestContext>,
    f: impl FnOnce() -> T,
) -> Result<T, Diagnostic> {
    catching_cleanup(stage, task, context, None, f)
}

/// Cleanup keeps its own origin and links the existing primary BEFORE source
/// submission. No recapture, cloned diagnostic, or change to the business result.
pub(crate) fn catching_cleanup<T>(
    stage: DiagnosticStage,
    task: &'static str,
    context: Option<RequestContext>,
    primary: Option<Diagnostic>,
    f: impl FnOnce() -> T,
) -> Result<T, Diagnostic> {
    let previous = CURRENT.with(|slot| {
        slot.replace(Some(Frame {
            stage,
            task,
            context,
            panic: None,
            primary,
            request_mode: false,
            request_panic: None,
            request_primary: None,
            cleanup_occurrence: None,
            request_scope: None,
            request_output: None,
        }))
    });
    let outcome = catch_unwind(AssertUnwindSafe(f));
    let frame = CURRENT
        .with(|slot| slot.replace(previous))
        .expect("matching synchronous diagnostic frame");
    match outcome {
        Ok(value) => Ok(value),
        Err(payload) => {
            // An inner managed poll has already captured and emitted this
            // occurrence. resume_unwind does not invoke the hook a second time.
            if let Ok(diagnostic) = payload.downcast::<Diagnostic>() {
                return Err(*diagnostic);
            }
            let diagnostic = frame.panic.unwrap_or_else(|| {
                let mut d = failure(
                    stage,
                    DiagnosticCategory::Panic,
                    "runtime.panic_without_hook",
                )
                .with_task(code(task));
                if let Some(primary) = frame.primary.as_ref() {
                    d = d.during_cleanup_of(primary);
                }
                emit(&d, frame.context.as_ref());
                d
            });
            Err(diagnostic)
        }
    }
}

pub(crate) fn catching_request<T>(
    stage: DiagnosticStage,
    task: &'static str,
    context: Option<RequestContext>,
    request_scope: Option<saddle_observability::RequestDiagnosticScope<'static>>,
    primary: Option<RequestCaptured>,
    f: impl FnOnce() -> T,
) -> (Result<T, RequestCaptured>, Option<RequestCaptured>) {
    catching_request_with_output(stage, task, context, request_scope, OUTPUT.get().cloned(), primary, f)
}

pub(crate) fn catching_request_with_output<T>(
    stage: DiagnosticStage,
    task: &'static str,
    context: Option<RequestContext>,
    request_scope: Option<saddle_observability::RequestDiagnosticScope<'static>>,
    request_output: Option<EmergencyDiagnosticHandle>,
    primary: Option<RequestCaptured>,
    f: impl FnOnce() -> T,
) -> (Result<T, RequestCaptured>, Option<RequestCaptured>) {
    let previous = CURRENT.with(|slot| {
        slot.replace(Some(Frame {
            stage,
            task,
            context,
            panic: None,
            primary: None,
            request_mode: true,
            request_panic: None,
            request_primary: primary,
            cleanup_occurrence: None,
            request_scope,
            request_output,
        }))
    });
    let outcome = catch_unwind(AssertUnwindSafe(f));
    let mut frame = CURRENT
        .with(|slot| slot.replace(previous))
        .expect("synchronous request frame");
    let outcome = match outcome {
        Ok(value) => Ok(value),
        Err(payload) => match payload.downcast::<RequestCaptured>() {
            Ok(captured) => Err(*captured),
            Err(payload) => match payload.downcast::<Diagnostic>() {
                // Legacy inner managed poll already emitted this object.
                // Retain it, but never manufacture a submission receipt later.
                Ok(diagnostic) => Err(RequestCaptured::MissingContext(*diagnostic)),
                Err(_) => Err(frame.request_panic.take().unwrap_or_else(|| {
                    let mut diagnostic = failure(
                        stage,
                        DiagnosticCategory::Panic,
                        "runtime.panic_without_hook",
                    )
                    .with_task(code(task));
                    if let Some(primary) = frame.request_primary.as_ref() {
                        diagnostic = diagnostic.during_cleanup_of(primary.diagnostic());
                    }
                    if let Some(primary) = frame.cleanup_occurrence.as_ref() {
                        diagnostic = diagnostic.during_cleanup_of_occurrence(primary);
                    }
                    capture_request(
                        diagnostic,
                        frame.context.as_ref(),
                        frame.request_scope.as_ref(),
                        frame.request_output.as_ref(),
                    )
                })),
            },
        },
    };
    (outcome, frame.request_primary)
}

pub(crate) fn link_task_cleanup(primary: Option<saddle_core::DiagnosticOccurrence>) {
    CURRENT.with(|slot| {
        if let Some(frame) = slot.borrow_mut().as_mut() {
            if frame.request_mode && frame.task == "runtime.formal_request_task_drop" {
                frame.cleanup_occurrence = primary;
            }
        }
    });
}

pub(crate) async fn task<F: Future>(
    future: F,
    stage: DiagnosticStage,
    name: &'static str,
) -> F::Output {
    let mut future = std::pin::pin!(future);
    std::future::poll_fn(|cx| {
        match catching(stage, name, None, || {
            let poll = future.as_mut().poll(cx);
            #[cfg(test)]
            if poll.is_ready()
                && name == "runtime.managed_request_task"
                && std::env::var_os("RUNTIME_DIAGNOSTIC_MANAGER_FAULT").is_some()
            {
                // Private fixed manager failure AFTER the envelope returned,
                // not a handler panic; compiled out of published artifacts.
                panic!("DIAGNOSTIC_PRIVATE_SENTINEL");
            }
            poll
        }) {
            Ok(poll) => poll,
            Err(diagnostic) => resume_unwind(Box::new(diagnostic)),
        }
    })
    .await
}

pub(crate) fn joined(error: tokio::task::JoinError) {
    if error.is_cancelled() {
        boundary(
            None,
            &saddle_core::DiagnosticOutcomeAxes {
                operation: saddle_core::OperationOutcome::Cancelled,
                ..Default::default()
            },
            None,
        );
        return;
    }
    if error.is_panic() {
        let payload = error.into_panic();
        if let Ok(diagnostic) = payload.downcast::<Diagnostic>() {
            boundary(
                Some(diagnostic.occurrence()),
                &saddle_core::DiagnosticOutcomeAxes {
                    operation: saddle_core::OperationOutcome::Panicked,
                    ..Default::default()
                },
                None,
            );
            return;
        } // already emitted at original poll
    }
    let d = failure(
        DiagnosticStage::BackgroundTask,
        DiagnosticCategory::UnexpectedError,
        "runtime.task_join_failed",
    );
    emit(&d, None);
    boundary(
        Some(d.occurrence()),
        &saddle_core::DiagnosticOutcomeAxes {
            operation: saddle_core::OperationOutcome::Panicked,
            ..Default::default()
        },
        None,
    );
}

#[track_caller]
pub(crate) fn attach(
    error: SaddleError,
    stage: DiagnosticStage,
    name: &'static str,
) -> SaddleError {
    if error.diagnostic().is_some() {
        error
    } else {
        error.with_diagnostic(failure(stage, DiagnosticCategory::UnexpectedError, name))
    }
}

pub(crate) fn report(error: &SaddleError) {
    if let Some(d) = error.diagnostic() {
        emit(d, None);
    }
}

pub(crate) fn cleanup(
    error: SaddleError,
    primary: Option<&SaddleError>,
    stage: DiagnosticStage,
) -> SaddleError {
    let mut error = attach(error, stage, "runtime.cleanup_failed");
    if let Some(primary) = primary {
        error = error.during_cleanup_of(primary);
    }
    report(&error);
    error
}