saddle_observability/

diagnostic.rs

//! Independent emergency output. Ordinary submission never performs filesystem IO;
//! a confirmed source may synchronously take over after the worker exits.
use crate::FileLoggingConfig;
use saddle_core::{CallContext, Diagnostic};
use serde::Serialize;
use std::{
    fmt,
    fs::{self, File, OpenOptions},
    io::{self, Write},
    path::PathBuf,
    sync::{
        Arc, Condvar, Mutex, OnceLock,
        atomic::{AtomicBool, AtomicU64, Ordering},
        mpsc::{self, SyncSender, TrySendError},
    },
    thread::{self, JoinHandle},
    time::Duration,
};

pub const EMERGENCY_FILE_NAME: &str = "saddle.emergency.log";
pub const BOOTSTRAP_FILE_NAME: &str = "saddle.bootstrap.log";
const QUEUE_CAPACITY: usize = 64;
const RECORD_BYTES: usize = 32 * 1024;
const TARGET_PATH_BYTES: usize = 4160;
static ACTIVE: AtomicBool = AtomicBool::new(false);

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DiagnosticSubmission {
    /// No configured output exists. A light receipt retains association/status,
    /// not source detail; legacy carriers may separately retain their detail.
    OutputUnavailable,
    Enqueued,
    /// Every segment of this source, including its terminal marker, was written.
    Written,
    Full,
    Closed,
    EncodingFailed,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum DiagnosticShutdown {
    Pending,
    Finished,
    WorkerPanicked,
}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct DiagnosticOutputSnapshot {
    pub initialized: bool,
    pub first_failure: Option<DiagnosticIoFailure>,
    pub enqueued: u64,
    pub written: u64,
    pub dropped: u64,
    pub output_failed: u64,
    pub stderr_failed: u64,
    pub stderr_suppressed: u64,
    pub truncated: u64,
    pub closed: bool,
}
#[derive(Default)]
struct Status {
    owns_active: AtomicBool,
    initialized: AtomicBool,
    failure: AtomicU64,
    enqueued: AtomicU64,
    written: AtomicU64,
    dropped: AtomicU64,
    output_failed: AtomicU64,
    stderr_failed: AtomicU64,
    stderr_suppressed: AtomicU64,
    truncated: AtomicU64,
    closed: AtomicBool,
    submitting: AtomicU64,
    worker_stopped: AtomicBool,
    fallback_path: OnceLock<BoundedTargetPath>,
    fallback_file: Mutex<Option<File>>,
    confirmation: Mutex<Confirmation>,
    confirmation_ready: Condvar,
}
impl Drop for Status {
    fn drop(&mut self) {
        if self.owns_active.load(Ordering::Acquire) {
            drop(self.fallback_file.get_mut().unwrap_or_else(|e| e.into_inner()).take());
            ACTIVE.store(false, Ordering::Release);
        }
    }
}
#[derive(Default)]
struct Confirmation {
    next: u64,
    active: Option<u64>,
    completed: Option<(u64, io::Result<()>)>,
}
struct BoundedTargetPath {
    bytes: [u8; TARGET_PATH_BYTES],
    len: usize,
}
impl BoundedTargetPath {
    fn new(path: &std::path::Path) -> io::Result<Self> {
        use std::os::unix::ffi::OsStrExt;
        let source = path.as_os_str().as_bytes();
        if source.len() > TARGET_PATH_BYTES {
            return Err(io::Error::new(io::ErrorKind::InvalidInput, "emergency path exceeds bound"));
        }
        let mut bytes = [0; TARGET_PATH_BYTES];
        bytes[..source.len()].copy_from_slice(source);
        Ok(Self { bytes, len: source.len() })
    }
    fn as_path(&self) -> &std::path::Path {
        use std::os::unix::ffi::OsStrExt;
        std::path::Path::new(std::ffi::OsStr::from_bytes(&self.bytes[..self.len]))
    }
}
struct ConfirmationLease<'a> {
    status: &'a Status,
    sequence: u64,
}
impl Status {
    fn acquire_confirmation(&self) -> ConfirmationLease<'_> {
        let mut state = self.confirmation.lock().unwrap_or_else(|e| e.into_inner());
        while state.active.is_some() {
            state = self.confirmation_ready.wait(state).unwrap_or_else(|e| e.into_inner());
        }
        state.next = state.next.wrapping_add(1);
        let sequence = state.next;
        state.active = Some(sequence);
        ConfirmationLease { status: self, sequence }
    }
    fn acknowledge(&self, sequence: u64, delivery: io::Result<()>) {
        let mut state = self.confirmation.lock().unwrap_or_else(|e| e.into_inner());
        if state.active == Some(sequence) {
            state.completed = Some((sequence, delivery));
            self.confirmation_ready.notify_all();
        }
    }
}
impl ConfirmationLease<'_> {
    fn wait(&self) -> io::Result<()> {
        let mut state = self.status.confirmation.lock().unwrap_or_else(|e| e.into_inner());
        loop {
            if let Some((sequence, delivery)) = state.completed.take() {
                debug_assert_eq!(sequence, self.sequence);
                return delivery;
            }
            if self.status.worker_stopped.load(Ordering::Acquire) {
                return Err(io::Error::new(io::ErrorKind::BrokenPipe, "diagnostic writer stopped"));
            }
            state = self.status.confirmation_ready.wait(state).unwrap_or_else(|e| e.into_inner());
        }
    }
}
impl Drop for ConfirmationLease<'_> {
    fn drop(&mut self) {
        let mut state = self.status.confirmation.lock().unwrap_or_else(|e| e.into_inner());
        if state.active == Some(self.sequence) {
            state.active = None;
            state.completed = None;
            self.status.confirmation_ready.notify_all();
        }
    }
}
impl Status {
    fn fail(&self, stage: u64, error: &io::Error) {
        let kind = match error.kind() {
            io::ErrorKind::PermissionDenied => 1,
            io::ErrorKind::NotFound => 2,
            io::ErrorKind::NotADirectory => 3,
            io::ErrorKind::InvalidInput => 4,
            io::ErrorKind::StorageFull => 5,
            io::ErrorKind::BrokenPipe => 6,
            io::ErrorKind::WouldBlock => 7,
            _ => 8,
        };
        let os = error.raw_os_error();
        let packed = (stage << 56)
            | (kind << 48)
            | (u64::from(os.is_some()) << 40)
            | u64::from(os.unwrap_or(0) as u32);
        let _ = self
            .failure
            .compare_exchange(0, packed, Ordering::AcqRel, Ordering::Acquire);
    }
    fn snapshot(&self) -> DiagnosticOutputSnapshot {
        let failure = self.failure.load(Ordering::Acquire);
        DiagnosticOutputSnapshot {
            initialized: self.initialized.load(Ordering::Acquire),
            first_failure: (failure != 0).then(|| DiagnosticIoFailure {
                stage: if failure >> 56 == 1 {
                    "initialize"
                } else {
                    "write"
                },
                kind: [
                    "none",
                    "permission_denied",
                    "not_found",
                    "not_a_directory",
                    "invalid_input",
                    "storage_full",
                    "broken_pipe",
                    "would_block",
                    "other",
                ][((failure >> 48) & 255) as usize],
                os_code: (failure & (1 << 40) != 0).then_some(failure as u32 as i32),
            }),
            enqueued: self.enqueued.load(Ordering::Acquire),
            written: self.written.load(Ordering::Acquire),
            dropped: self.dropped.load(Ordering::Acquire),
            output_failed: self.output_failed.load(Ordering::Acquire),
            stderr_failed: self.stderr_failed.load(Ordering::Acquire),
            stderr_suppressed: self.stderr_suppressed.load(Ordering::Acquire),
            truncated: self.truncated.load(Ordering::Acquire),
            closed: self.closed.load(Ordering::Acquire),
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
pub struct DiagnosticIoFailure {
    pub stage: &'static str,
    pub kind: &'static str,
    pub os_code: Option<i32>,
}

/// Startup owner: at most one independent writer per process. No join on Drop.
pub struct EmergencyDiagnostics {
    handle: EmergencyDiagnosticHandle,
    worker: Option<JoinHandle<()>>,
    target: PathBuf,
    source_target_preopened: bool,
}
/// The selected runtime original output was opened before the application
/// could begin. Its private field prevents constructing a formal output from
/// an arbitrary queue handle; individual writes still require confirmation.
///
/// ```compile_fail
/// use saddle_observability::{EmergencyDiagnosticHandle, SourceOutput};
/// fn forge(handle: EmergencyDiagnosticHandle) -> SourceOutput { SourceOutput(handle) }
/// ```
#[derive(Clone)]
pub struct SourceOutput(EmergencyDiagnosticHandle);
impl SourceOutput {
    pub fn handle(&self) -> &EmergencyDiagnosticHandle { &self.0 }
}
/// Synchronous, configuration-independent startup original owner. It uses
/// the same bounded record encoding and verified append open as the runtime
/// writer, without claiming the process writer slot or spawning a thread.
#[derive(Clone)]
pub struct BootstrapDiagnostics {
    inner: Arc<BootstrapInner>,
}
struct BootstrapInner {
    file: Mutex<File>,
    target: PathBuf,
}
static BOOTSTRAP_ACTIVE: OnceLock<Mutex<std::sync::Weak<BootstrapInner>>> = OnceLock::new();

impl BootstrapDiagnostics {
    pub fn start(directory: &std::path::Path) -> Result<Self, EmergencyInitError> {
        validate_emergency_directory(directory)?;
        let target = directory.join(BOOTSTRAP_FILE_NAME);
        let registry = BOOTSTRAP_ACTIVE.get_or_init(|| Mutex::new(std::sync::Weak::new()));
        let mut active = registry.lock().unwrap_or_else(|error| error.into_inner());
        if let Some(inner) = active.upgrade() {
            return if inner.target == target {
                Ok(Self { inner })
            } else {
                Err(EmergencyInitError::AlreadyActive)
            };
        }
        let file = open_emergency_file(&target, true)
            .map_err(|error| EmergencyInitError::Spawn(error.kind()))?;
        let inner = Arc::new(BootstrapInner { file: Mutex::new(file), target });
        *active = Arc::downgrade(&inner);
        Ok(Self { inner })
    }

    pub fn target(&self) -> &std::path::Path { &self.inner.target }

    pub(crate) fn write_original_record<T: Serialize>(&self, record: &T) -> io::Result<()> {
        let mut bytes = FixedDiagnosticBytes { bytes: [0; 8192], len: 0 };
        serde_json::to_writer(&mut bytes, record).map_err(io::Error::other)?;
        bytes.bytes[bytes.len] = b'\n';
        bytes.len += 1;
        self.inner.file.lock().unwrap_or_else(|error| error.into_inner())
            .write_all(&bytes.bytes[..bytes.len])
    }
}
#[derive(Clone)]
pub struct EmergencyDiagnosticHandle {
    sender: SyncSender<DiagnosticPacket>,
    status: Arc<Status>,
}

/// Startup-instance discriminator shared by finalization records in this
/// process. It is observational and never participates in request deadlines.
pub(crate) fn finalization_process_instance() -> u128 {
    static INSTANCE: std::sync::OnceLock<u128> = std::sync::OnceLock::new();
    *INSTANCE.get_or_init(|| std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH).map_or(0, |v| v.as_nanos()))
}
struct DetachedObjectLifetime {
    context: saddle_core::request_context::RequestContextSnapshot,
    output: EmergencyDiagnosticHandle,
    process_instance: u128,
    kind: &'static str,
}
impl DetachedObjectLifetime {
    fn emit(&self, fact: saddle_admission::ObjectLifetimeFact, created: bool) {
        static SEQUENCE: AtomicU64 = AtomicU64::new(1);
        static CLOCK: OnceLock<std::time::Instant> = OnceLock::new();
        #[derive(Serialize)]
        struct Record<'a> {
            schema: &'static str, pid: u32, process_instance: u128, local_request: u64,
            sequence: u64, monotonic_ns: u128, kind: &'static str, phase: &'static str,
            object_id: u64, ledger_instance: u64, account: saddle_admission::AccountIdentity,
            storage_bytes: usize, storage_allocated: bool, ledger_charge_applied: bool,
            storage_destroyed: bool, ledger_charge_released: bool,
            context: &'a saddle_core::request_context::RequestContextSnapshot,
        }
        let _ = self.output.submit_fixed_record(&Record {
            schema: "request-object-lifetime/1", pid: std::process::id(),
            process_instance: self.process_instance, local_request: self.context.local_request_id(),
            sequence: SEQUENCE.fetch_add(1, Ordering::Relaxed),
            monotonic_ns: CLOCK.get_or_init(std::time::Instant::now).elapsed().as_nanos().saturating_add(1),
            kind: self.kind, phase: if created { "created" } else { "released" },
            object_id: fact.object_id, ledger_instance: fact.ledger_instance, account: fact.account,
            storage_bytes: fact.storage_bytes, storage_allocated: created, ledger_charge_applied: created,
            storage_destroyed: !created, ledger_charge_released: !created, context: &self.context,
        });
    }
}
impl saddle_admission::ObjectLifetimeObserver for DetachedObjectLifetime {
    fn created(&self, fact: saddle_admission::ObjectLifetimeFact) { self.emit(fact, true); }
    fn released(&self, fact: saddle_admission::ObjectLifetimeFact) { self.emit(fact, false); }
}
fn rpc_lifecycle_stamp() -> (u64, u128) {
    static SEQUENCE: AtomicU64 = AtomicU64::new(1);
    static CLOCK: OnceLock<std::time::Instant> = OnceLock::new();
    (SEQUENCE.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| n.checked_add(1)).unwrap_or(0),
        CLOCK.get_or_init(std::time::Instant::now).elapsed().as_nanos().saturating_add(1))
}
/// Inline detached metadata; the actual enclosing Shared layout prepays it.
#[doc(hidden)]
#[derive(Clone)]
pub struct RpcPhysicalOwnerReceipt {
    context: saddle_core::request_context::RequestContextSnapshot,
    output: EmergencyDiagnosticHandle,
    process_instance: u128,
    call_id: [u8; 256],
    call_id_len: usize,
}
impl RpcPhysicalOwnerReceipt {
    #[doc(hidden)]
    pub fn driver_phase(&self, stage: saddle_admission::RpcStageReleaseFact,
        driver_id: u64, phase: &'static str) {
        #[derive(Serialize)]
        struct Record<'a> {
            schema: &'static str, event: &'static str, pid: u32, process_instance: u128,
            local_request: u64, call_id: &'a str, driver_id: u64, phase: &'static str,
            sequence: u64, monotonic_ns: u128, stage: saddle_admission::RpcStageReleaseFact,
            context: &'a saddle_core::request_context::RequestContextSnapshot,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        let _ = self.output.submit_fixed_record(&Record {
            schema: "request-rpc-driver/1", event: "rpc_driver_lifetime", pid: std::process::id(),
            process_instance: self.process_instance, local_request: self.context.local_request_id(),
            call_id: std::str::from_utf8(&self.call_id[..self.call_id_len]).unwrap(), driver_id, phase,
            sequence, monotonic_ns, stage, context: &self.context,
        });
    }
    #[doc(hidden)]
    pub fn released(self, stage: saddle_admission::RpcStageReleaseFact) {
        #[derive(Serialize)]
        struct Record<'a> {
            schema: &'static str, event: &'static str, pid: u32,
            process_instance: u128, local_request: u64, call_id: &'a str,
            sequence: u64, monotonic_ns: u128,
            stage: saddle_admission::RpcStageReleaseFact,
            context: &'a saddle_core::request_context::RequestContextSnapshot,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        let _ = self.output.submit_fixed_record(&Record {
            schema: "request-rpc-physical-owner/1", event: "rpc_shared_allocation_released",
            pid: std::process::id(), process_instance: self.process_instance,
            local_request: self.context.local_request_id(),
            call_id: std::str::from_utf8(&self.call_id[..self.call_id_len]).unwrap(),
            sequence, monotonic_ns, stage, context: &self.context,
        });
    }
}
struct DetachedRpcStageRelease {
    context: saddle_core::request_context::RequestContextSnapshot,
    output: EmergencyDiagnosticHandle,
    process_instance: u128,
    call_id: [u8; 256],
    call_id_len: usize,
}
impl saddle_admission::RpcStageReleaseObserver for DetachedRpcStageRelease {
    fn released(&self, fact: saddle_admission::RpcStageReleaseFact) {
        #[derive(Serialize)]
        struct Record<'a> {
            schema: &'static str, event: &'static str, pid: u32,
            process_instance: u128, local_request: u64, call_id: &'a str,
            sequence: u64, monotonic_ns: u128,
            stage: saddle_admission::RpcStageReleaseFact,
            context: &'a saddle_core::request_context::RequestContextSnapshot,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        let call_id = std::str::from_utf8(&self.call_id[..self.call_id_len]).unwrap();
        let _ = self.output.submit_fixed_record(&Record {
            schema: "request-rpc-stage/1", event: "rpc_stage_credit_released",
            pid: std::process::id(), process_instance: self.process_instance,
            local_request: self.context.local_request_id(), call_id,
            sequence, monotonic_ns,
            stage: fact, context: &self.context,
        });
    }
}
struct DetachedAccountRelease {
    context: saddle_core::request_context::RequestContextSnapshot,
    output: EmergencyDiagnosticHandle,
    process_instance: u128,
}
impl saddle_admission::AccountReleaseObserver for DetachedAccountRelease {
    fn released(&self, fact: saddle_admission::AccountReleaseFact) {
        #[derive(Serialize)]
        struct Record<'a> {
            schema: &'static str, event: &'static str, process_id: u32,
            process_instance: u128, completion: &'static str,
            account: saddle_admission::AccountReleaseFact,
            context: &'a saddle_core::request_context::RequestContextSnapshot,
        }
        let _ = self.output.submit_fixed_record(&Record {
            schema: "request-finalization/1", event: "request_account_released",
            process_id: std::process::id(), process_instance: self.process_instance,
            completion: "physically_released", account: fact, context: &self.context,
        });
    }
}
impl EmergencyDiagnosticHandle {
    #[doc(hidden)]
    pub fn record_request_stop_decision(&self,
        context: saddle_core::request_context::RequestContextSnapshot, cause: &'static str) {
        if !matches!(cause, "original_deadline" | "peer_disconnected") { return; }
        #[derive(Serialize)]
        struct Record {
            schema: &'static str, event: &'static str, decision_point: &'static str,
            pid: u32, process_instance: u128, local_request: u64,
            sequence: u64, monotonic_ns: u128, cause: &'static str,
            context: saddle_core::request_context::RequestContextSnapshot,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        let _ = self.submit_fixed_record(&Record {
            schema: "request-stop/1", event: "request_stop_decided",
            decision_point: "handler_select_before_dispatch_drop", pid: std::process::id(),
            process_instance: finalization_process_instance(), local_request: context.local_request_id(),
            sequence, monotonic_ns, cause, context,
        });
    }
    /// Low-level bounded projection. Root callers transfer source identity via
    /// the consuming runtime ReservedSupervisionFailure adapter, not capture.
    #[doc(hidden)]
    pub fn record_encoding_selection_association(&self,
        context: saddle_core::request_context::RequestContextSnapshot,
        original_deadline_unix_ms: i64,
        source: Option<(saddle_core::DiagnosticOccurrence, saddle_core::RequestExecutionView)>) -> DiagnosticSubmission {
        #[derive(Serialize)]
        struct Record<'a> {
            schema: &'static str, event: &'static str, decision_point: &'static str,
            pid: u32, process_instance: u128, local_request: u64,
            sequence: u64, monotonic_ns: u128, original_deadline_unix_ms: i64,
            source_association: &'static str,
            source_occurrence: Option<saddle_core::DiagnosticOccurrence>,
            source_context: Option<saddle_core::request_context::UnifiedContext<'a>>,
            context: saddle_core::request_context::RequestContextSnapshot,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        crate::root_diagnostic::original::write_encoding_selection(self, &Record {
            schema: "encoding-terminal-source/1", event: "encoding_terminal_selected",
            decision_point: "encoding_response_selection", pid: std::process::id(),
            process_instance: finalization_process_instance(), local_request: context.local_request_id(),
            sequence, monotonic_ns, original_deadline_unix_ms,
            source_association: if source.is_some() { "original_timeout" } else { "unavailable" },
            source_occurrence: source.as_ref().map(|value| value.0),
            source_context: source.as_ref().map(|value| value.1.unified_context()), context,
        })
    }
    #[doc(hidden)]
    pub fn rpc_physical_owner_receipt(&self,
        context: saddle_core::request_context::RequestContextSnapshot, call_id: &str)
        -> Option<RpcPhysicalOwnerReceipt> {
        if call_id.is_empty() || call_id.len() > 256 { return None; }
        let mut copied = [0; 256];
        copied[..call_id.len()].copy_from_slice(call_id.as_bytes());
        Some(RpcPhysicalOwnerReceipt { context, output: self.clone(),
            process_instance: finalization_process_instance(), call_id: copied, call_id_len: call_id.len() })
    }
    #[doc(hidden)]
    pub fn prepare_rpc_stage_release(&self,
        context: saddle_core::request_context::RequestContextSnapshot,
        storage: &saddle_admission::ProcessLogStorage, call_id: &str)
        -> Option<saddle_admission::RpcStageReleaseObservation> {
        // Reject observation only; never truncate identity or change RPC behavior.
        if call_id.is_empty() || call_id.len() > 256 { return None; }
        let mut copied = [0u8; 256];
        copied[..call_id.len()].copy_from_slice(call_id.as_bytes());
        saddle_admission::RpcStageReleaseObservation::prepare(storage, DetachedRpcStageRelease {
            context, output: self.clone(), process_instance: finalization_process_instance(),
            call_id: copied, call_id_len: call_id.len(),
        }).ok()
    }
    #[doc(hidden)]
    pub fn record_shutdown_signal_received(&self, application: saddle_core::ContextLabel, kind: &'static str, received: bool) {
        #[derive(Serialize)]
        struct Record {
            schema: &'static str, pid: u32, process_instance: u128, application: saddle_core::ContextLabel,
            signal: &'static str, received: bool, sequence: u64, monotonic_ns: u128,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        let _ = self.submit_fixed_record(&Record {
            schema: "process-signal-receipt/1", pid: std::process::id(),
            process_instance: finalization_process_instance(), application,
            signal: kind, received, sequence, monotonic_ns,
        });
    }
    #[doc(hidden)]
    pub fn record_request_task_first_poll(&self,
        context: saddle_core::request_context::RequestContextSnapshot) {
        #[derive(Serialize)]
        struct Record {
            schema: &'static str, event: &'static str, pid: u32, process_instance: u128,
            local_request: u64, sequence: u64, monotonic_ns: u128,
            context: saddle_core::request_context::RequestContextSnapshot,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        let _ = self.submit_fixed_record(&Record {
            schema: "request-task-execution/1", event: "before_first_body_poll",
            pid: std::process::id(), process_instance: finalization_process_instance(),
            local_request: context.local_request_id(), sequence, monotonic_ns, context,
        });
    }
    #[doc(hidden)]
    pub fn record_request_lifecycle(&self, application: &str, event: &'static str,
        observed_phase: u8, active_claims: usize) {
        #[derive(Serialize)]
        struct Record<'a> {
            schema: &'static str, pid: u32, process_instance: u128,
            application: &'a str, event: &'static str, observed_phase: u8,
            active_claims: usize, sequence: u64, monotonic_ns: u128,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        let _ = self.submit_fixed_record(&Record {
            schema: "request-lifecycle-boundary/1", pid: std::process::id(),
            process_instance: finalization_process_instance(), application, event,
            observed_phase, active_claims, sequence, monotonic_ns,
        });
    }
    #[doc(hidden)]
    pub fn record_delivery_attempt(&self, context: saddle_core::request_context::RequestContextSnapshot,
        attempt_id: u64, phase: &'static str, write_starts: u64, write_completions: u64,
        bytes_written: u64, delivery: saddle_core::ResponseDelivery) {
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        if sequence == 0 { return; }
        #[derive(Serialize)]
        struct Record {
            schema: &'static str, pid: u32, process_instance: u128, local_request: u64,
            sequence: u64, monotonic_ns: u128,
            attempt_id: u64, phase: &'static str, write_starts: u64, write_completions: u64,
            bytes_written: u64, delivery: saddle_core::ResponseDelivery,
            context: saddle_core::request_context::RequestContextSnapshot,
        }
        let _ = self.submit_fixed_record(&Record {
            schema: "request-delivery-attempt/1", pid: std::process::id(),
            process_instance: finalization_process_instance(), local_request: context.local_request_id(),
            sequence, monotonic_ns, attempt_id, phase, write_starts, write_completions, bytes_written, delivery, context,
        });
    }
    #[doc(hidden)]
    pub fn record_request_database_admission(&self,
        context: saddle_core::request_context::RequestContextSnapshot,
        ledger_instance: u64, account: saddle_admission::AccountIdentity,
        accepted: bool, reason: &'static str) {
        #[derive(Serialize)]
        struct Record {
            schema: &'static str, event: &'static str, pid: u32, process_instance: u128,
            local_request: u64, ledger_instance: u64, account: saddle_admission::AccountIdentity,
            accepted: bool, reason: &'static str, sequence: u64, monotonic_ns: u128,
            context: saddle_core::request_context::RequestContextSnapshot,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        let _ = self.submit_fixed_record(&Record {
            schema: "request-database-admission/1", event: "database_stage_attempt",
            pid: std::process::id(), process_instance: finalization_process_instance(),
            local_request: context.local_request_id(), ledger_instance, account, accepted, reason,
            sequence, monotonic_ns, context,
        });
    }
    #[doc(hidden)]
    pub fn record_request_task_owner(&self,
        context: saddle_core::request_context::RequestContextSnapshot,
        ledger_instance: u64, account: saddle_admission::AccountIdentity, task_role: &'static str) {
        #[derive(Serialize)]
        struct Record {
            schema: &'static str, event: &'static str, pid: u32, process_instance: u128,
            local_request: u64, ledger_instance: u64, account: saddle_admission::AccountIdentity,
            task_role: &'static str, sequence: u64, monotonic_ns: u128,
            context: saddle_core::request_context::RequestContextSnapshot,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        let _ = self.submit_fixed_record(&Record {
            schema: "request-task-roster/2", event: "task_owner_bound", pid: std::process::id(),
            process_instance: finalization_process_instance(), local_request: context.local_request_id(),
            ledger_instance, account, task_role, sequence, monotonic_ns, context,
        });
    }
    #[doc(hidden)]
    pub fn record_request_read_owner(&self,
        context: saddle_core::request_context::RequestContextSnapshot,
        ledger_instance: u64, account: saddle_admission::AccountIdentity,
        peer: std::net::SocketAddrV4) {
        #[derive(Serialize)]
        struct Record {
            schema: &'static str, event: &'static str, pid: u32, process_instance: u128,
            local_request: u64, ledger_instance: u64, account: saddle_admission::AccountIdentity,
            peer_address: [u8; 4], peer_port: u16, sequence: u64, monotonic_ns: u128,
            context: saddle_core::request_context::RequestContextSnapshot,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        let _ = self.submit_fixed_record(&Record {
            schema: "request-read-owner/1", event: "recognition_read_head_entered",
            pid: std::process::id(), process_instance: finalization_process_instance(),
            local_request: context.local_request_id(), ledger_instance, account,
            peer_address: peer.ip().octets(), peer_port: peer.port(), sequence, monotonic_ns, context,
        });
    }
    #[doc(hidden)]
    pub fn record_request_account_identity(&self, view: &saddle_core::RequestExecutionView,
        permit: &saddle_admission::StoragePermit) {
        let Some(account) = permit.release_account_identity() else { return; };
        let context = view.diagnostic_snapshot();
        #[derive(Serialize)]
        struct Record<'a> {
            schema: &'static str, event: &'static str, pid: u32, process_instance: u128,
            local_request: u64, ledger_instance: u64, account: saddle_admission::AccountIdentity,
            context: &'a saddle_core::request_context::RequestContextSnapshot,
        }
        let _ = self.submit_fixed_record(&Record {
            schema: "request-account-identity/1", event: "request_account_bound",
            pid: std::process::id(), process_instance: finalization_process_instance(),
            local_request: context.local_request_id(), ledger_instance: permit.diagnostic_ledger_instance(),
            account, context: &context,
        });
    }
    #[doc(hidden)]
    pub fn prepare_object_lifetime(&self, view: &saddle_core::RequestExecutionView,
        storage: &saddle_admission::ProcessLogStorage, kind: &'static str)
        -> Option<saddle_admission::ObjectLifetimeObservation> {
        if !matches!(kind, "raw_body" | "typed_input" | "db_parameter_text") { return None; }
        saddle_admission::ObjectLifetimeObservation::prepare(storage, DetachedObjectLifetime {
            context: view.diagnostic_snapshot(), output: self.clone(),
            process_instance: finalization_process_instance(), kind,
        }).ok()
    }
    /// The copy contains only encoded identity, not a root/account reference.
    /// Its box is prepaid independently from the account being observed.
    #[doc(hidden)]
    pub fn prepare_account_release(&self, view: &saddle_core::RequestExecutionView,
        storage: &saddle_admission::ProcessLogStorage)
        -> Option<saddle_admission::AccountReleaseObservation> {
        // Reuse the bounded, detached identity snapshot. Parsing encoded JSON
        // here would allocate after the observed account's audit identity ends.
        let context = view.diagnostic_snapshot();
        saddle_admission::AccountReleaseObservation::prepare(storage, DetachedAccountRelease {
            context, output: self.clone(), process_instance: finalization_process_instance(),
        }).ok()
    }
}

// One existing queue/writer; inline admitted records allocate no payload blocks.
#[allow(clippy::large_enum_variant)] // Boxing here would allocate on admitted submission.
enum DiagnosticPacket {
    Startup(Vec<u8>, Option<saddle_core::DeferredDiagnosticStack>),
    Admitted(FixedDiagnosticBytes),
    Original(FixedDiagnosticBytes),
    ConfirmedOriginal(FixedDiagnosticBytes, u64),
}
impl DiagnosticPacket {
    fn bytes(&self) -> &[u8] {
        match self {
            Self::Startup(v, _) => v,
            Self::Admitted(v) | Self::Original(v) | Self::ConfirmedOriginal(v, _) => &v.bytes[..v.len],
        }
    }
    fn resolve_on_worker(&mut self, status: &Status) {
        let Self::Startup(bytes, stack) = self else {
            return;
        };
        let Some(stack) = stack.take() else {
            return;
        };
        let resolved = stack.resolve_on_output_worker();
        let Ok(mut record) = serde_json::from_slice::<serde_json::Value>(bytes) else {
            status.dropped.fetch_add(1, Ordering::Relaxed);
            return;
        };
        record["diagnostic"]["stack_status"] = resolved.stack_status.into();
        record["diagnostic"]["stack"] = resolved.stack.into();
        record["diagnostic"]["stack_truncated"] = resolved.stack_truncated.into();
        if resolved.stack_truncated {
            status.truncated.fetch_add(1, Ordering::Relaxed);
        }
        let mut encoded = LimitedBytes(Vec::with_capacity(RECORD_BYTES));
        if serde_json::to_writer(&mut encoded, &record).is_err() {
            encoded.0.clear();
            record["diagnostic"]["stack"] = "".into();
            record["diagnostic"]["stack_status"] = "omitted_output_limit".into();
            record["diagnostic"]["stack_truncated"] = true.into();
            record["encoding_truncated"] = true.into();
            status.truncated.fetch_add(1, Ordering::Relaxed);
            if serde_json::to_writer(&mut encoded, &record).is_err() {
                // Original bounded row remains truthful: captured_unresolved.
                return;
            }
        }
        encoded.0.push(b'\n');
        *bytes = encoded.0;
    }
}
#[derive(Clone)]
pub(crate) struct FixedDiagnosticBytes {
    pub(crate) bytes: [u8; 8192],
    pub(crate) len: usize,
}
pub(crate) fn root_frame_layouts() -> (std::alloc::Layout, std::alloc::Layout, usize) {
    (
        std::alloc::Layout::new::<FixedDiagnosticBytes>(),
        std::alloc::Layout::new::<DiagnosticPacket>(),
        QUEUE_CAPACITY,
    )
}
/// One process-owned allocation, shared by every handle and confirmation.
/// Includes the Arc counters, confirmation slot and bounded takeover target.
pub(crate) fn root_status_owner_layout() -> std::alloc::Layout {
    // Mirrors alloc::sync::ArcInner in the pinned 1.95.0 toolchain.
    #[repr(C, align(2))]
    struct ArcOwner<T> { strong: std::sync::atomic::AtomicUsize, weak: std::sync::atomic::AtomicUsize, value: T }
    std::alloc::Layout::new::<ArcOwner<Status>>()
}
impl Write for FixedDiagnosticBytes {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        if bytes.len() > self.bytes.len().saturating_sub(self.len + 1) {
            return Err(io::ErrorKind::WriteZero.into());
        }
        self.bytes[self.len..self.len + bytes.len()].copy_from_slice(bytes);
        self.len += bytes.len();
        Ok(bytes.len())
    }
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

#[derive(Debug)]
pub enum EmergencyInitError {
    InvalidDirectory,
    AlreadyActive,
    Open(io::Error),
    Spawn(io::ErrorKind),
}
impl fmt::Display for EmergencyInitError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "emergency diagnostic initialization failed: {self:?}")
    }
}
impl std::error::Error for EmergencyInitError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self { Self::Open(error) => Some(error), _ => None }
    }
}

struct LimitedBytes(Vec<u8>);
impl Write for LimitedBytes {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        if self.0.len().saturating_add(bytes.len()) > RECORD_BYTES - 1 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "diagnostic encoding bound",
            ));
        }
        self.0.extend_from_slice(bytes);
        Ok(bytes.len())
    }
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}
#[derive(Serialize)]
struct Envelope<'a> {
    event: &'static str,
    timestamp_unix_ms: u128,
    diagnostic: &'a Diagnostic,
    trace_id: Option<&'a str>,
    rpc_id: Option<String>,
    span_id: Option<String>,
    operation: Option<String>,
    request: Option<&'a str>,
    route: Option<&'a str>,
}

impl<'a> Envelope<'a> {
    fn new(
        diagnostic: &'a Diagnostic,
        context: Option<(&'a CallContext, &'a crate::EventContext)>,
    ) -> Self {
        Self {
            event: "framework.diagnostic",
            timestamp_unix_ms: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_millis(),
            diagnostic,
            trace_id: context.map(|(call, _)| call.trace_correlation_id().as_str()),
            rpc_id: context
                .and_then(|(call, _)| call.rpc_correlation_id().map(|rpc| rpc.as_str().to_owned())),
            span_id: context.map(|(call, _)| call.span_id().to_string()),
            operation: context.map(|(call, _)| call.operation().to_string()),
            request: context.map(|(_, event)| event.diagnostic_request()),
            route: context.map(|(_, event)| event.diagnostic_route()),
        }
    }
}

impl EmergencyDiagnosticHandle {
    /// Allocation-free admitted-path encoding and try-submit. No source callback,
    /// stack capture, filesystem IO, or modification of the caller's Result.
    /// Diagnostic and axes remain with the caller on every outcome.
    pub fn submit_bounded(
        &self,
        diagnostic: Option<&saddle_core::BoundedDiagnostic>,
        axes: &saddle_core::DiagnosticOutcomeAxes,
        context: Option<(&CallContext, &crate::EventContext)>,
    ) -> DiagnosticSubmission {
        self.submit_bounded_parts(
            diagnostic,
            axes,
            context.map(|(c, e)| (c, e.diagnostic_request(), e.diagnostic_route())),
            None,
        )
    }
    pub(crate) fn submit_bounded_parts(
        &self,
        diagnostic: Option<&saddle_core::BoundedDiagnostic>,
        axes: &saddle_core::DiagnosticOutcomeAxes,
        context: Option<(&CallContext, &str, &str)>,
        stage: Option<(&'static str, u64)>,
    ) -> DiagnosticSubmission {
        self.encode_bounded(
            if stage.is_some() { None } else { diagnostic },
            diagnostic.map(|d| d.occurrence()),
            axes,
            context,
            stage,
            stage.is_some(),
        )
    }
    /// Boundary outcome only; the source owns and submits the detailed diagnostic.
    /// An occurrence does not assert that the source record reached storage.
    pub fn submit_boundary(
        &self,
        occurrence: Option<saddle_core::DiagnosticOccurrence>,
        axes: &saddle_core::DiagnosticOutcomeAxes,
        context: Option<(&CallContext, &crate::EventContext)>,
    ) -> DiagnosticSubmission {
        self.encode_bounded(
            None,
            occurrence,
            axes,
            context.map(|(c, e)| (c, e.diagnostic_request(), e.diagnostic_route())),
            None,
            true,
        )
    }
    pub(crate) fn submit_boundary_parts(
        &self,
        occurrence: Option<saddle_core::DiagnosticOccurrence>,
        axes: &saddle_core::DiagnosticOutcomeAxes,
        context: Option<(&CallContext, &str, &str)>,
        stage: Option<(&'static str, u64)>,
    ) -> DiagnosticSubmission {
        self.encode_bounded(None, occurrence, axes, context, stage, true)
    }
    #[allow(clippy::too_many_arguments)]
    fn encode_bounded(
        &self,
        diagnostic: Option<&saddle_core::BoundedDiagnostic>,
        occurrence: Option<saddle_core::DiagnosticOccurrence>,
        axes: &saddle_core::DiagnosticOutcomeAxes,
        context: Option<(&CallContext, &str, &str)>,
        stage: Option<(&'static str, u64)>,
        boundary: bool,
    ) -> DiagnosticSubmission {
        self.status.submitting.fetch_add(1, Ordering::AcqRel);
        struct Exit<'a>(&'a AtomicU64);
        impl Drop for Exit<'_> {
            fn drop(&mut self) {
                self.0.fetch_sub(1, Ordering::Release);
            }
        }
        let _exit = Exit(&self.status.submitting);
        // Fixed scan/encoding work even if a caller supplied an oversized ID.
        if context.is_some_and(|(c, request, route)| {
            c.operation().as_str().len() > 256 || request.len() > 128 || route.len() > 192
        }) {
            self.status.dropped.fetch_add(1, Ordering::Relaxed);
            return DiagnosticSubmission::EncodingFailed;
        }
        if self.status.closed.load(Ordering::Acquire) {
            self.status.dropped.fetch_add(1, Ordering::Relaxed);
            return DiagnosticSubmission::Closed;
        }
        #[derive(Serialize)]
        struct Record<'a> {
            event: &'static str,
            stage: Option<&'static str>,
            elapsed_ms: Option<u64>,
            timestamp_unix_ms: u128,
            diagnostic: Option<&'a saddle_core::BoundedDiagnostic>,
            diagnostic_reference: Option<saddle_core::DiagnosticOccurrence>,
            axes: &'a saddle_core::DiagnosticOutcomeAxes,
            trace_id: Option<&'a str>,
            rpc_id: Option<&'a str>,
            span_id: Option<&'a str>,
            operation: Option<&'a str>,
            request: Option<&'a str>,
            route: Option<&'a str>,
        }
        let mut span = [b'0'; 16];
        if let Some((call, _, _)) = context {
            let value = call.span_id().as_u64();
            for (i, byte) in span.iter_mut().enumerate() {
                *byte = b"0123456789abcdef"[((value >> ((15 - i) * 4)) & 15) as usize];
            }
        }
        let record = Record {
            event: if stage.is_some() {
                "framework.stage.finished"
            } else if boundary {
                "framework.boundary.outcome"
            } else {
                "framework.diagnostic"
            },
            stage: stage.map(|s| s.0),
            elapsed_ms: stage.map(|s| s.1),
            timestamp_unix_ms: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_millis(),
            diagnostic,
            diagnostic_reference: occurrence,
            axes,
            trace_id: context.map(|(c, _, _)| c.trace_correlation_id().as_str()),
            rpc_id: context.and_then(|(c, _, _)| c.rpc_correlation_id().map(|r| r.as_str())),
            span_id: context.and_then(|_| std::str::from_utf8(&span).ok()),
            operation: context.map(|(c, _, _)| c.operation().as_str()),
            request: context.map(|(_, request, _)| request),
            route: context.map(|(_, _, route)| route),
        };
        self.encode_and_queue(&record)
    }
    /// Internal shared bounded encoder. The request adapter owns its sealed receipt.
    pub(crate) fn submit_fixed_record<T: Serialize>(&self, record: &T) -> DiagnosticSubmission {
        self.status.submitting.fetch_add(1, Ordering::AcqRel);
        let result = if self.status.closed.load(Ordering::Acquire) {
            self.status.dropped.fetch_add(1, Ordering::Relaxed);
            DiagnosticSubmission::Closed
        } else {
            self.encode_and_queue(record)
        };
        self.status.submitting.fetch_sub(1, Ordering::Release);
        result
    }
    fn encode_and_queue<T: Serialize>(&self, record: &T) -> DiagnosticSubmission {
        self.encode_and_queue_fallback(record, None, false)
    }
    pub(crate) fn submit_original_record<T: Serialize>(&self, record: &T) -> DiagnosticSubmission {
        self.status.submitting.fetch_add(1, Ordering::AcqRel);
        let result = if self.status.closed.load(Ordering::Acquire) {
            self.status.dropped.fetch_add(1, Ordering::Relaxed);
            DiagnosticSubmission::Closed
        } else {
            self.encode_and_queue_fallback(record, None, true)
        };
        self.status.submitting.fetch_sub(1, Ordering::Release);
        result
    }
    /// A source segment is acknowledged only after the writer has written it.
    /// This is deliberately separate from the legacy, lossy original path.
    pub(crate) fn write_original_record<T: Serialize>(&self, record: &T) -> io::Result<()> {
        self.status.submitting.fetch_add(1, Ordering::AcqRel);
        struct Exit<'a>(&'a AtomicU64);
        impl Drop for Exit<'_> {
            fn drop(&mut self) { self.0.fetch_sub(1, Ordering::Release); }
        }
        let _exit = Exit(&self.status.submitting);
        let mut bytes = FixedDiagnosticBytes { bytes: [0; 8192], len: 0 };
        serde_json::to_writer(&mut bytes, record).map_err(io::Error::other)?;
        bytes.bytes[bytes.len] = b'\n';
        bytes.len += 1;
        let lease = self.status.acquire_confirmation();
        let retry = bytes.clone();
        match self.sender.send(DiagnosticPacket::ConfirmedOriginal(bytes, lease.sequence)) {
            Ok(()) => {
                self.status.enqueued.fetch_add(1, Ordering::Relaxed);
                match lease.wait() {
                    Err(error) if error.kind() == io::ErrorKind::BrokenPipe
                        && self.status.worker_stopped.load(Ordering::Acquire) => {
                        self.write_original_after_worker(&retry, false)
                    }
                    result => result,
                }
            }
            Err(_) => self.write_original_after_worker(&retry, true),
        }
    }
    /// The worker has exited. Use its pinned append target while the borrowed
    /// source still exists; only a failed initial open needs a fresh open.
    fn write_original_after_worker(&self, bytes: &FixedDiagnosticBytes, count_admission: bool) -> io::Result<()> {
        if !self.status.worker_stopped.load(Ordering::Acquire) {
            return Err(io::Error::new(io::ErrorKind::BrokenPipe, "diagnostic writer unavailable"));
        }
        let path = self.status.fallback_path.get().ok_or_else(||
            io::Error::new(io::ErrorKind::NotFound, "diagnostic output target unavailable"))?;
        let mut fallback = self.status.fallback_file.lock().unwrap_or_else(|e| e.into_inner());
        if fallback.is_none() {
            *fallback = Some(open_emergency_file(path.as_path(), false)?);
        }
        fallback.as_mut().expect("fallback file opened").write_all(&bytes.bytes[..bytes.len])?;
        if count_admission { self.status.enqueued.fetch_add(1, Ordering::Relaxed); }
        self.status.written.fetch_add(1, Ordering::Release);
        Ok(())
    }
    pub(crate) fn submit_fixed_with_fallback<T: Serialize>(
        &self,
        record: &T,
        fallback: &T,
    ) -> DiagnosticSubmission {
        self.status.submitting.fetch_add(1, Ordering::AcqRel);
        let result = if self.status.closed.load(Ordering::Acquire) {
            self.status.dropped.fetch_add(1, Ordering::Relaxed);
            DiagnosticSubmission::Closed
        } else {
            self.encode_and_queue_fallback(record, Some(fallback), false)
        };
        self.status.submitting.fetch_sub(1, Ordering::Release);
        result
    }
    fn encode_and_queue_fallback<T: Serialize>(
        &self,
        record: &T,
        fallback: Option<&T>,
        original: bool,
    ) -> DiagnosticSubmission {
        let mut bytes = FixedDiagnosticBytes {
            bytes: [0; 8192],
            len: 0,
        };
        if serde_json::to_writer(&mut bytes, record).is_err() {
            bytes.len = 0;
            let recovered = fallback
                .is_some_and(|fallback| serde_json::to_writer(&mut bytes, fallback).is_ok());
            if !recovered {
                self.status.dropped.fetch_add(1, Ordering::Relaxed);
                return DiagnosticSubmission::EncodingFailed;
            }
            self.status.truncated.fetch_add(1, Ordering::Relaxed);
        }
        bytes.bytes[bytes.len] = b'\n';
        bytes.len += 1;
        let packet = if original {
            DiagnosticPacket::Original(bytes)
        } else {
            DiagnosticPacket::Admitted(bytes)
        };
        match self.sender.try_send(packet) {
            Ok(()) => {
                self.status.enqueued.fetch_add(1, Ordering::Relaxed);
                DiagnosticSubmission::Enqueued
            }
            Err(TrySendError::Full(_)) => {
                self.status.dropped.fetch_add(1, Ordering::Relaxed);
                DiagnosticSubmission::Full
            }
            Err(TrySendError::Disconnected(_)) => {
                self.status.dropped.fetch_add(1, Ordering::Relaxed);
                DiagnosticSubmission::Closed
            }
        }
    }
    pub fn snapshot(&self) -> DiagnosticOutputSnapshot {
        self.status.snapshot()
    }
    /// Enqueued means accepted in memory, NOT written or durable.
    pub fn submit(&self, diagnostic: &Diagnostic) -> DiagnosticSubmission {
        self.submit_context(diagnostic, None)
    }
    pub fn submit_context(
        &self,
        diagnostic: &Diagnostic,
        context: Option<(&CallContext, &crate::EventContext)>,
    ) -> DiagnosticSubmission {
        self.submit_existing_record(
            &Envelope::new(diagnostic, context),
            diagnostic.deferred_stack(),
        )
    }
    // Retains the existing 32KiB legacy encoding/truncation behavior. This is
    // NOT the allocation-free admitted bounded encoder.
    pub(crate) fn submit_existing_record<T: Serialize>(
        &self,
        record: &T,
        stack: Option<saddle_core::DeferredDiagnosticStack>,
    ) -> DiagnosticSubmission {
        let active = self.status.submitting.fetch_add(1, Ordering::AcqRel);
        struct Exit<'a>(&'a AtomicU64);
        impl Drop for Exit<'_> {
            fn drop(&mut self) {
                self.0.fetch_sub(1, Ordering::AcqRel);
            }
        }
        let _exit = Exit(&self.status.submitting);
        if active >= QUEUE_CAPACITY as u64 {
            self.status.dropped.fetch_add(1, Ordering::Relaxed);
            return DiagnosticSubmission::Full;
        }
        if self.status.closed.load(Ordering::Acquire) {
            self.status.dropped.fetch_add(1, Ordering::Relaxed);
            return DiagnosticSubmission::Closed;
        }
        let mut bytes = LimitedBytes(Vec::with_capacity(RECORD_BYTES));
        if serde_json::to_writer(&mut bytes, record).is_err() {
            bytes.0.clear();
            let mut reduced = match serde_json::to_value(record) {
                Ok(value) => value,
                Err(_) => {
                    self.status.dropped.fetch_add(1, Ordering::Relaxed);
                    return DiagnosticSubmission::EncodingFailed;
                }
            };
            reduced["diagnostic"]["stack"] = serde_json::Value::String(String::new());
            reduced["diagnostic"]["stack_truncated"] = true.into();
            reduced["encoding_truncated"] = true.into();
            self.status.truncated.fetch_add(1, Ordering::Relaxed);
            if serde_json::to_writer(&mut bytes, &reduced).is_err() {
                self.status.dropped.fetch_add(1, Ordering::Relaxed);
                return DiagnosticSubmission::EncodingFailed;
            }
        }
        bytes.0.push(b'\n');
        match self
            .sender
            .try_send(DiagnosticPacket::Startup(bytes.0, stack))
        {
            Ok(()) => {
                self.status.enqueued.fetch_add(1, Ordering::Relaxed);
                DiagnosticSubmission::Enqueued
            }
            Err(TrySendError::Full(_)) => {
                self.status.dropped.fetch_add(1, Ordering::Relaxed);
                DiagnosticSubmission::Full
            }
            Err(TrySendError::Disconnected(_)) => {
                self.status.dropped.fetch_add(1, Ordering::Relaxed);
                DiagnosticSubmission::Closed
            }
        }
    }
}

impl crate::Observer {
    /// Mirrors one safe diagnostic into independent emergency output and the
    /// existing logger. The result acknowledges only emergency queue admission.
    pub fn record_diagnostic(
        &self,
        diagnostic: &Diagnostic,
        emergency: &EmergencyDiagnosticHandle,
        context: Option<(&CallContext, &crate::EventContext)>,
    ) -> DiagnosticSubmission {
        let result = emergency.submit_context(diagnostic, context);
        self.mirror_existing_record(&Envelope::new(diagnostic, context), diagnostic.category());
        result
    }
    pub(crate) fn mirror_existing_record<T: Serialize>(
        &self,
        record: &T,
        category: saddle_core::DiagnosticCategory,
    ) {
        if let Ok(serde_json::Value::Object(fields)) = serde_json::to_value(record) {
            let level = if matches!(category, saddle_core::DiagnosticCategory::ExpectedRejection) {
                crate::EventLevel::Info
            } else {
                crate::EventLevel::Error
            };
            let mut record = crate::logger::LogRecord::new(level, "framework.diagnostic");
            record.data = fields;
            record.data.remove("event");
            record.data.remove("timestamp_unix_ms");
            self.emit(record);
        }
    }
}
fn open_emergency_file(path: &std::path::Path, create_parent: bool) -> io::Result<File> {
    if create_parent {
        if let Some(parent) = path.parent() { fs::create_dir_all(parent)?; }
    }
    // The worker and its bounded takeover use the same verified target.
    use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
    let file = OpenOptions::new()
        .create(true)
        .append(true)
        .mode(0o644)
        .custom_flags(
            rustix::fs::OFlags::NOFOLLOW.bits() as i32
                | rustix::fs::OFlags::NONBLOCK.bits() as i32,
        )
        .open(path)?;
    if !file.metadata()?.is_file() {
        return Err(io::Error::new(io::ErrorKind::InvalidInput, "non-regular emergency file"));
    }
    rustix::fs::flock(&file, rustix::fs::FlockOperation::NonBlockingLockExclusive)
        .map_err(io::Error::from)?;
    file.set_permissions(fs::Permissions::from_mode(0o644))?;
    Ok(file)
}

fn validate_emergency_directory(directory: &std::path::Path) -> Result<(), EmergencyInitError> {
    let directory = directory.to_str().ok_or(EmergencyInitError::InvalidDirectory)?;
    if directory.is_empty()
        || directory.len() > 4096
        || directory.chars().any(char::is_control)
        || directory.contains("://")
        || directory.contains(['@', '?', '#'])
    {
        return Err(EmergencyInitError::InvalidDirectory);
    }
    Ok(())
}

impl EmergencyDiagnostics {
    /// Directory setup is performed by the worker, never the submitting request.
    /// Callers must check snapshot/written, not equate successful start with IO readiness.
    pub fn start(config: &FileLoggingConfig) -> Result<Self, EmergencyInitError> {
        Self::start_inner(config, false)
    }

    /// Startup preopens the verified target while its raw I/O error can still
    /// be written to the independent bootstrap original file.
    pub fn start_checked(config: &FileLoggingConfig) -> Result<Self, EmergencyInitError> {
        Self::start_inner(config, true)
    }

    fn start_inner(config: &FileLoggingConfig, preopen: bool) -> Result<Self, EmergencyInitError> {
        validate_emergency_directory(config.directory())?;
        let target = config.directory().join(EMERGENCY_FILE_NAME);
        let fallback_path = BoundedTargetPath::new(&target)
            .map_err(|_| EmergencyInitError::InvalidDirectory)?;
        if ACTIVE
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return Err(EmergencyInitError::AlreadyActive);
        }
        let initial_file = if preopen {
            match open_emergency_file(&target, true) {
                Ok(file) => Some(file),
                Err(error) => {
                    ACTIVE.store(false, Ordering::Release);
                    return Err(EmergencyInitError::Open(error));
                }
            }
        } else { None };
        let path = target.clone();
        let (sender, receiver) = mpsc::sync_channel::<DiagnosticPacket>(QUEUE_CAPACITY);
        let status = Arc::new(Status::default());
        status.owns_active.store(true, Ordering::Release);
        assert!(status.fallback_path.set(fallback_path).is_ok(), "new diagnostic status");
        let worker_status = status.clone();
        let worker = thread::Builder::new()
            .name("saddle-emergency-writer".into())
            .spawn(move || {
                struct ActiveGuard(Arc<Status>);
                impl Drop for ActiveGuard {
                    fn drop(&mut self) {
                        let _state = self.0.confirmation.lock().unwrap_or_else(|e| e.into_inner());
                        self.0.worker_stopped.store(true, Ordering::Release);
                        self.0.confirmation_ready.notify_all();
                    }
                }
                let _active = ActiveGuard(worker_status.clone());
                let file = initial_file.map(Ok).unwrap_or_else(|| open_emergency_file(&path, true)).and_then(|file| {
                    // Keep the worker's locked inode for a bounded source
                    // takeover. Rotation cannot split one original across
                    // the old and new path between acknowledged segments.
                    let takeover = file.try_clone()?;
                    *worker_status.fallback_file.lock().unwrap_or_else(|e| e.into_inner()) = Some(takeover);
                    Ok(file)
                });
                if let Err(error) = &file {
                    worker_status.fail(1, error);
                }
                worker_status.initialized.store(true, Ordering::Release);
                let mut output = file;
                let mut stderr_attempted = false;
                loop {
                    match receiver.recv_timeout(Duration::from_millis(10)) {
                        Ok(mut bytes) => {
                            bytes.resolve_on_worker(&worker_status);
                            let delivery = deliver_with_disclosure(
                                &mut output,
                                &mut io::stderr(),
                                &path,
                                bytes.bytes(),
                                &worker_status,
                                &mut stderr_attempted,
                                !matches!(bytes, DiagnosticPacket::Original(_) | DiagnosticPacket::ConfirmedOriginal(_, _)),
                            );
                            if let DiagnosticPacket::ConfirmedOriginal(_, sequence) = bytes {
                                worker_status.acknowledge(sequence, delivery);
                            }
                        }
                        Err(mpsc::RecvTimeoutError::Timeout) => {
                            if worker_status.closed.load(Ordering::Acquire)
                                && worker_status.submitting.load(Ordering::Acquire) == 0
                            {
                                break;
                            }
                        }
                        Err(mpsc::RecvTimeoutError::Disconnected) => break,
                    }
                }
            })
            .map_err(|error| {
                status.owns_active.store(false, Ordering::Release);
                ACTIVE.store(false, Ordering::Release);
                EmergencyInitError::Spawn(error.kind())
            })?;
        Ok(Self {
            handle: EmergencyDiagnosticHandle { sender, status },
            worker: Some(worker),
            target,
            source_target_preopened: preopen,
        })
    }
    /// Issue the formal output capability only after checked startup opened
    /// the selected controlled file. A later writer failure is reported by
    /// the synchronous original write rather than by this capability.
    pub fn source_output(&self) -> Option<SourceOutput> {
        self.source_target_preopened.then(|| SourceOutput(self.handle()))
    }
    pub fn handle(&self) -> EmergencyDiagnosticHandle {
        self.handle.clone()
    }
    pub fn target(&self) -> &std::path::Path {
        &self.target
    }
    pub fn snapshot(&self) -> DiagnosticOutputSnapshot {
        self.handle.snapshot()
    }
    /// Call using the EXISTING lifecycle deadline; Pending is not a new budget.
    pub fn shutdown(&mut self) -> DiagnosticShutdown {
        self.handle.status.closed.store(true, Ordering::Release);
        if self
            .worker
            .as_ref()
            .is_some_and(|worker| !worker.is_finished())
        {
            return DiagnosticShutdown::Pending;
        }
        match self.worker.take().map(JoinHandle::join) {
            Some(Err(_)) => DiagnosticShutdown::WorkerPanicked,
            _ => DiagnosticShutdown::Finished,
        }
    }
}
impl Drop for EmergencyDiagnostics {
    fn drop(&mut self) {
        self.handle.status.closed.store(true, Ordering::Release);
    }
}

#[cfg(test)]
fn deliver<W: Write, E: Write>(
    output: &mut io::Result<W>,
    stderr: &mut E,
    target: &std::path::Path,
    bytes: &[u8],
    status: &Status,
    stderr_attempted: &mut bool,
) {
    let _observed_result = deliver_with_disclosure(
        output,
        stderr,
        target,
        bytes,
        status,
        stderr_attempted,
        true,
    );
}

#[allow(clippy::too_many_arguments)] // Original disclosure is independent of output reliability.
fn deliver_with_disclosure<W: Write, E: Write>(
    output: &mut io::Result<W>,
    stderr: &mut E,
    target: &std::path::Path,
    bytes: &[u8],
    status: &Status,
    stderr_attempted: &mut bool,
    allow_stderr_content: bool,
) -> io::Result<()> {
    let result = match output {
        Ok(writer) => writer.write_all(bytes),
        Err(error) => Err(error
            .raw_os_error()
            .map(io::Error::from_raw_os_error)
            .unwrap_or_else(|| io::Error::from(error.kind()))),
    };
    if let Err(ref error) = result {
        status.fail(2, &error);
        status.output_failed.fetch_add(1, Ordering::Relaxed);
        // Only the independent worker can block here. No recursive logger call.
        if !*stderr_attempted {
            *stderr_attempted = true;
            let failure = serde_json::json!({"event":"framework.diagnostic.output_failed", "target":target.to_string_lossy(), "io_kind":format!("{:?}", error.kind()), "os_code":error.raw_os_error()});
            if serde_json::to_writer(&mut *stderr, &failure)
                .and_then(|_| stderr.write_all(b"\n").map_err(serde_json::Error::io))
                .is_err()
                || (allow_stderr_content && stderr.write_all(bytes).is_err())
            {
                status.stderr_failed.fetch_add(1, Ordering::Relaxed);
            }
        } else {
            status.stderr_suppressed.fetch_add(1, Ordering::Relaxed);
        }
    } else {
        status.written.fetch_add(1, Ordering::Release);
    }
    result
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn finalization_maximum_identity_frame_and_queue_failures_are_bounded() {
        use saddle_core::request_context::{ContextFact, ContextLabel, RequestIdentityGroup,
            RequestLocalFacts, RequestRootPublisher, RequestViewPhase};
        let label = "a".repeat(256);
        let request = "\\".repeat(256);
        let call = saddle_core::CallContext::new(label.clone().into(), label.clone().into(),
            label.clone().into(), label.clone().into(), saddle_core::TraceId::from_u128(1),
            saddle_core::SpanId::from_u64(1));
        let mut publisher = RequestRootPublisher::create(ContextLabel::checked(&label).unwrap(), ContextFact::NotEstablished).unwrap();
        publisher.publish(RequestIdentityGroup::from_validated(&call, &request, &label, u32::MAX,
            ContextFact::Present(ContextLabel::checked(&label).unwrap())).unwrap()).ok().unwrap();
        let view = publisher.reference().view(RequestLocalFacts::new(RequestViewPhase::Reading));
        let mut context = FixedDiagnosticBytes { bytes: [0; 8192], len: 0 };
        serde_json::to_writer(&mut context, &view).unwrap();
        let (sender, receiver) = mpsc::sync_channel(1);
        let status = Arc::new(Status::default());
        let observer = DetachedAccountRelease { context: view.diagnostic_snapshot(), output: EmergencyDiagnosticHandle {
            sender, status: status.clone(),
        }, process_instance: u128::MAX };
        let fact = saddle_admission::AccountReleaseFact { account_slot: usize::MAX,
            account_generation: u64::MAX, audit_generation: u64::MAX, healthy: true };
        saddle_admission::AccountReleaseObserver::released(&observer, fact);
        assert_eq!(status.enqueued.load(Ordering::Relaxed), 1);
        // One queue slot remains occupied. Failure cannot fabricate Written.
        saddle_admission::AccountReleaseObserver::released(&observer, fact);
        assert!(status.dropped.load(Ordering::Relaxed) > 0);
        let DiagnosticPacket::Admitted(bytes) = receiver.recv().unwrap() else { panic!("fixed admitted frame"); };
        assert!(bytes.len <= 8192);
        let row: serde_json::Value = serde_json::from_slice(&bytes.bytes[..bytes.len]).unwrap();
        assert_eq!(row["context"]["request"]["value"], request);
        assert_eq!(row["account"]["account_generation"], u64::MAX);
        status.closed.store(true, Ordering::Release);
        saddle_admission::AccountReleaseObserver::released(&observer, fact);
        assert!(receiver.try_recv().is_err());
        assert_eq!(status.written.load(Ordering::Relaxed), 0);
    }

    #[test]
    fn object_lifetime_native_frame_preserves_identity_and_queue_loss() {
        use saddle_core::request_context::{ContextFact, ContextLabel, RequestLocalFacts,
            RequestRootPublisher, RequestViewPhase};
        let publisher = RequestRootPublisher::create(ContextLabel::checked("objects").unwrap(),
            ContextFact::NotEstablished).unwrap();
        let view = publisher.reference().view(RequestLocalFacts::new(RequestViewPhase::Reading));
        let (sender, receiver) = mpsc::sync_channel(1);
        let status = Arc::new(Status::default());
        let observer = DetachedObjectLifetime {
            context: view.diagnostic_snapshot(), output: EmergencyDiagnosticHandle {sender, status: status.clone()},
            process_instance: 77, kind: "raw_body",
        };
        let fact = saddle_admission::ObjectLifetimeFact {
            object_id: 19, ledger_instance: 23, storage_bytes: 4096,
            account: saddle_admission::AccountIdentity {account_slot: 0, account_generation: 7, audit_generation: 11},
        };
        observer.emit(fact, true);
        observer.emit(fact, false);
        assert_eq!(status.dropped.load(Ordering::Relaxed), 1);
        let DiagnosticPacket::Admitted(bytes) = receiver.recv().unwrap() else { panic!("fixed native frame") };
        let row: serde_json::Value = serde_json::from_slice(&bytes.bytes[..bytes.len]).unwrap();
        assert_eq!(row["schema"], "request-object-lifetime/1");
        assert_eq!(row["local_request"], row["context"]["local_request"]);
        assert_eq!(row["ledger_instance"], 23);
        assert_eq!(row["account"]["audit_generation"], 11);
        assert_eq!(row["storage_allocated"], true);
        assert_eq!(row["storage_destroyed"], false);
        observer.emit(fact, false);
        let DiagnosticPacket::Admitted(bytes) = receiver.recv().unwrap() else { panic!("fixed native frame") };
        let released: serde_json::Value = serde_json::from_slice(&bytes.bytes[..bytes.len]).unwrap();
        assert_eq!(released["phase"], "released");
        assert_eq!(released["storage_destroyed"], true);
        assert!(released["sequence"].as_u64().unwrap() > row["sequence"].as_u64().unwrap());
        assert!(released["monotonic_ns"].as_u64().unwrap() >= row["monotonic_ns"].as_u64().unwrap());
        assert_eq!(status.written.load(Ordering::Relaxed), 0);
    }

    #[test]
    fn source_output_requires_checked_preopen_and_keeps_selected_target() {
        let directory = std::env::temp_dir().join(format!(
            "saddle-source-output-{}-{}", std::process::id(),
            std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)
                .unwrap().as_nanos()));
        std::fs::create_dir(&directory).unwrap();
        let config = FileLoggingConfig::new(directory.clone(), crate::Rotation::Daily);
        let mut unchecked = EmergencyDiagnostics::start(&config).unwrap();
        assert!(unchecked.source_output().is_none());
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        while unchecked.shutdown() == DiagnosticShutdown::Pending
            && std::time::Instant::now() < deadline { std::thread::yield_now(); }
        assert_eq!(unchecked.shutdown(), DiagnosticShutdown::Finished);
        drop(unchecked);

        let mut checked = EmergencyDiagnostics::start_checked(&config).unwrap();
        let selected = checked.source_output().expect("checked output");
        selected.handle().write_original_record(&"selected source 3701").unwrap();
        assert!(std::fs::read_to_string(checked.target()).unwrap()
            .contains("selected source 3701"));
        drop(selected);
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        while checked.shutdown() == DiagnosticShutdown::Pending
            && std::time::Instant::now() < deadline { std::thread::yield_now(); }
        assert_eq!(checked.shutdown(), DiagnosticShutdown::Finished);
        drop(checked);
        std::fs::remove_dir_all(directory).unwrap();
    }

    #[test]
    fn confirmed_original_waits_for_writer_acknowledgement() {
        let (sender, receiver) = mpsc::sync_channel(1);
        let status = Arc::new(Status::default());
        let handle = EmergencyDiagnosticHandle { sender, status: status.clone() };
        // Shutdown may set closed while the worker still owns the output.
        // Confirmed source packets must continue through that live writer.
        status.closed.store(true, Ordering::Release);
        let (done, completed) = mpsc::sync_channel(1);
        let worker = thread::spawn(move || {
            done.send(handle.write_original_record(&"confirmed source")).unwrap();
        });
        let packet = receiver.recv().unwrap();
        assert!(matches!(completed.try_recv(), Err(mpsc::TryRecvError::Empty)));
        let DiagnosticPacket::ConfirmedOriginal(bytes, sequence) = packet else { panic!("confirmed packet"); };
        assert!(std::str::from_utf8(&bytes.bytes[..bytes.len]).unwrap().contains("confirmed source"));
        status.acknowledge(sequence, Ok(()));
        completed.recv().unwrap().unwrap();
        worker.join().unwrap();

        let (sender, receiver) = mpsc::sync_channel(1);
        let status = Arc::new(Status::default());
        let handle = EmergencyDiagnosticHandle { sender, status: status.clone() };
        let worker = thread::spawn(move || handle.write_original_record(&"failed source"));
        let DiagnosticPacket::ConfirmedOriginal(_, sequence) = receiver.recv().unwrap() else { panic!("confirmed packet"); };
        status.acknowledge(sequence, Err(io::ErrorKind::PermissionDenied.into()));
        assert_eq!(worker.join().unwrap().unwrap_err().kind(), io::ErrorKind::PermissionDenied);
    }

    #[test]
    fn confirmed_original_serializes_concurrent_sources_and_shutdown() {
        let (sender, receiver) = mpsc::sync_channel(2);
        let status = Arc::new(Status::default());
        let handle = EmergencyDiagnosticHandle { sender, status: status.clone() };
        let first = handle.clone();
        let second = handle.clone();
        let a = thread::spawn(move || first.write_original_record(&"source A"));
        let DiagnosticPacket::ConfirmedOriginal(_, first_sequence) = receiver.recv().unwrap() else { panic!("first packet"); };
        let b = thread::spawn(move || second.write_original_record(&"source B"));
        // The second caller cannot enqueue a segment until the first writer
        // acknowledgement releases the one process-owned confirmation slot.
        assert!(matches!(receiver.try_recv(), Err(mpsc::TryRecvError::Empty)));
        status.acknowledge(first_sequence, Ok(()));
        assert!(a.join().unwrap().is_ok());
        let DiagnosticPacket::ConfirmedOriginal(_, second_sequence) = receiver.recv_timeout(Duration::from_secs(1)).unwrap() else { panic!("second packet"); };
        status.acknowledge(second_sequence, Ok(()));
        assert!(b.join().unwrap().is_ok());
        status.closed.store(true, Ordering::Release);
        drop(receiver);
        status.worker_stopped.store(true, Ordering::Release);
        assert_eq!(handle.write_original_record(&"without a configured target").unwrap_err().kind(), io::ErrorKind::NotFound);
        assert_eq!(status.submitting.load(Ordering::Acquire), 0);
    }

    #[test]
    fn cancelled_confirmation_cannot_acknowledge_next_source() {
        let status = Status::default();
        let cancelled = status.acquire_confirmation();
        let old_sequence = cancelled.sequence;
        drop(cancelled);
        let current = status.acquire_confirmation();
        status.acknowledge(old_sequence, Ok(()));
        assert!(status.confirmation.lock().unwrap().completed.is_none());
        status.acknowledge(current.sequence, Err(io::ErrorKind::PermissionDenied.into()));
        assert_eq!(current.wait().unwrap_err().kind(), io::ErrorKind::PermissionDenied);
    }

    #[test]
    fn stopped_writer_releases_waiter_without_receipt() {
        let (sender, receiver) = mpsc::sync_channel(1);
        let status = Arc::new(Status::default());
        let weak_owner = Arc::downgrade(&status);
        let handle = EmergencyDiagnosticHandle { sender, status: status.clone() };
        let waiter = thread::spawn(move || handle.write_original_record(&"unwritten source"));
        let _packet = receiver.recv_timeout(Duration::from_secs(1)).unwrap();
        {
            let _state = status.confirmation.lock().unwrap();
            status.worker_stopped.store(true, Ordering::Release);
            status.confirmation_ready.notify_all();
        }
        // No selected target exists. The waiter wakes on BrokenPipe, but the
        // takeover cannot confirm a write and reports the missing target.
        assert_eq!(waiter.join().unwrap().unwrap_err().kind(), io::ErrorKind::NotFound);
        assert_eq!(status.snapshot().written, 0);
        assert_eq!(status.submitting.load(Ordering::Acquire), 0);
        drop(receiver);
        drop(status);
        assert!(weak_owner.upgrade().is_none(), "last owner releases the physical confirmation allocation");
    }

    #[test]
    fn stopped_writer_with_preopened_target_writes_before_waiter_returns() {
        let _serial = FILE_TEST.lock().unwrap();
        let directory = std::env::temp_dir().join(format!(
            "saddle-stopped-writer-target-{}-{}", std::process::id(), sample().id()
        ));
        fs::create_dir(&directory).unwrap();
        let target = directory.join(EMERGENCY_FILE_NAME);
        let file = open_emergency_file(&target, true).unwrap();
        let (sender, receiver) = mpsc::sync_channel(1);
        let status = Arc::new(Status::default());
        status.fallback_path.set(BoundedTargetPath::new(&target).unwrap()).ok().unwrap();
        *status.fallback_file.lock().unwrap() = Some(file);
        let handle = EmergencyDiagnosticHandle { sender, status: status.clone() };
        let (done, completed) = mpsc::sync_channel(1);
        let waiter = thread::spawn(move || {
            done.send(handle.write_original_record(&"source after stopped writer 7814")).unwrap();
        });
        let DiagnosticPacket::ConfirmedOriginal(_, _) = receiver.recv_timeout(Duration::from_secs(1)).unwrap()
            else { panic!("confirmed packet"); };
        assert!(matches!(completed.try_recv(), Err(mpsc::TryRecvError::Empty)));
        {
            let _state = status.confirmation.lock().unwrap();
            status.worker_stopped.store(true, Ordering::Release);
            status.confirmation_ready.notify_all();
        }
        completed.recv_timeout(Duration::from_secs(2)).unwrap().unwrap();
        waiter.join().unwrap();
        assert_eq!(fs::read_to_string(&target).unwrap(), "\"source after stopped writer 7814\"\n");
        assert_eq!(status.snapshot().written, 1);
        assert_eq!(status.submitting.load(Ordering::Acquire), 0);
        drop(receiver);
        drop(status);
        fs::remove_file(target).unwrap();
        fs::remove_dir(directory).unwrap();
    }

    #[test]
    fn stopped_writer_without_target_keeps_owned_original_unconfirmed() {
        use crate::root_diagnostic::{ComponentSourceKind, RecordedSaddleError};
        use saddle_core::{ContextLabel, ErrorKind};
        #[derive(Debug)]
        struct OriginalChain { cause: io::Error }
        impl fmt::Display for OriginalChain {
            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
                formatter.write_str("unconfigured top source 7817")
            }
        }
        impl std::error::Error for OriginalChain {
            fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { Some(&self.cause) }
        }
        let (sender, receiver) = mpsc::sync_channel(1);
        drop(receiver);
        let status = Arc::new(Status::default());
        status.worker_stopped.store(true, Ordering::Release);
        let output = SourceOutput(EmergencyDiagnosticHandle { sender, status: status.clone() });
        let application = ContextLabel::checked("missing-target-test").unwrap();
        let result = RecordedSaddleError::component_result::<(), _>(
            Err(OriginalChain { cause: io::Error::other("unconfigured nested source 7818") }),
            &application, &output, ComponentSourceKind::Cleanup, None,
            ErrorKind::Infrastructure, "test.missing_target", "safe failure",
        );
        let failure = result.unwrap_err();
        assert_eq!(failure.safe().code(), "test.missing_target");
        let original = failure.original_if_unconfirmed().expect("no target cannot sign Written");
        assert_eq!(original.to_string(), "unconfigured top source 7817");
        assert_eq!(original.source().unwrap().to_string(), "unconfigured nested source 7818");
        assert_eq!(status.snapshot().written, 0);
        assert_eq!(status.submitting.load(Ordering::Acquire), 0);
    }

    #[test]
    fn failed_writer_confirmation_retains_component_original_and_cause() {
        use crate::root_diagnostic::{ComponentSourceKind, RecordedSaddleError};
        use saddle_core::{ContextLabel, ErrorKind};
        #[derive(Debug)]
        struct OriginalChain { cause: io::Error }
        impl fmt::Display for OriginalChain {
            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
                formatter.write_str("top source 7815")
            }
        }
        impl std::error::Error for OriginalChain {
            fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { Some(&self.cause) }
        }
        let (sender, receiver) = mpsc::sync_channel(1);
        let status = Arc::new(Status::default());
        let output = SourceOutput(EmergencyDiagnosticHandle { sender, status: status.clone() });
        let (done, completed) = mpsc::sync_channel(1);
        let waiter = thread::spawn(move || {
            let original = OriginalChain { cause: io::Error::other("nested source 7816") };
            let application = ContextLabel::checked("confirmation-failure-test").unwrap();
            let result = RecordedSaddleError::component_result::<(), _>(
                Err(original), &application, &output, ComponentSourceKind::Cleanup, None,
                ErrorKind::Infrastructure, "test.confirmation_failed", "safe failure",
            );
            done.send(result).unwrap();
        });
        let DiagnosticPacket::ConfirmedOriginal(bytes, sequence) = receiver.recv_timeout(Duration::from_secs(2)).unwrap()
            else { panic!("confirmed packet"); };
        assert!(matches!(completed.try_recv(), Err(mpsc::TryRecvError::Empty)));
        // Exercise the same delivery and acknowledgement path as the worker:
        // an actual Write error must reach the lease, not a fabricated Ok.
        let mut output: io::Result<Failed> = Ok(Failed);
        let mut stderr = Vec::new();
        let mut stderr_attempted = false;
        let delivery = deliver_with_disclosure(&mut output, &mut stderr,
            std::path::Path::new("/failed/controlled/source"), &bytes.bytes[..bytes.len],
            &status, &mut stderr_attempted, false);
        assert_eq!(delivery.as_ref().unwrap_err().raw_os_error(), Some(28));
        status.acknowledge(sequence, delivery);
        let failure = completed.recv_timeout(Duration::from_secs(2)).unwrap().unwrap_err();
        waiter.join().unwrap();
        assert_eq!(failure.safe().code(), "test.confirmation_failed");
        let original = failure.original_if_unconfirmed().expect("failed confirmation keeps owned source");
        assert_eq!(original.to_string(), "top source 7815");
        assert_eq!(original.source().unwrap().to_string(), "nested source 7816");
        assert_eq!(status.snapshot().written, 0);
        assert_eq!(status.snapshot().output_failed, 1);
        assert_eq!(status.submitting.load(Ordering::Acquire), 0);
    }

    #[test]
    fn confirmed_writer_releases_process_owner_after_shutdown() {
        let _serial = FILE_TEST.lock().unwrap();
        let directory = std::env::temp_dir().join(format!(
            "saddle-confirm-owner-{}-{}", std::process::id(), sample().id()
        ));
        let mut owner = EmergencyDiagnostics::start(&FileLoggingConfig::new(&directory, crate::Rotation::Daily)).unwrap();
        let handle = owner.handle();
        let weak_owner = Arc::downgrade(&handle.status);
        handle.write_original_record(&"complete source").unwrap();
        let text = fs::read_to_string(owner.target()).unwrap();
        assert!(text.contains("complete source"));
        let end = std::time::Instant::now() + Duration::from_secs(3);
        while owner.shutdown() == DiagnosticShutdown::Pending && std::time::Instant::now() < end {
            thread::sleep(Duration::from_millis(1));
        }
        assert_eq!(owner.shutdown(), DiagnosticShutdown::Finished);
        let path = owner.target().to_path_buf();
        let archived = directory.join("saddle.emergency.archived.log");
        fs::rename(&path, &archived).unwrap();
        handle.write_original_record(&"after shutdown A").unwrap();
        handle.write_original_record(&"after shutdown B").unwrap();
        let archived_text = fs::read_to_string(&archived).unwrap();
        assert!(archived_text.contains("after shutdown A") && archived_text.contains("after shutdown B"),
            "the takeover pins one inode across external rotation");
        assert!(matches!(EmergencyDiagnostics::start(&FileLoggingConfig::new(&directory, crate::Rotation::Daily)),
            Err(EmergencyInitError::AlreadyActive)), "a live source handle owns the takeover target");
        drop(handle);
        drop(owner);
        assert!(weak_owner.upgrade().is_none());
        let mut restarted = EmergencyDiagnostics::start(&FileLoggingConfig::new(&directory, crate::Rotation::Daily)).unwrap();
        restarted.handle().write_original_record(&"new owner").unwrap();
        let restarted_end = std::time::Instant::now() + Duration::from_secs(3);
        while restarted.shutdown() == DiagnosticShutdown::Pending && std::time::Instant::now() < restarted_end {
            thread::sleep(Duration::from_millis(1));
        }
        assert_eq!(restarted.shutdown(), DiagnosticShutdown::Finished);
        drop(restarted);
        assert!(fs::read_to_string(&path).unwrap().contains("new owner"));
        fs::remove_file(path).unwrap();
        fs::remove_file(archived).unwrap();
        fs::remove_dir(directory).unwrap();
    }

    #[test]
    fn ae_paused_queue_partial_full_disconnect_and_admitted_encoding() {
        use crate::{
            AdmissionCapacityFacts, AdmissionConstruction, AdmissionEventContext,
            CapacityDimension, Observer, ObserverConfig,
        };
        use saddle_admission::*;
        use saddle_core::*;
        use std::{
            future::Future,
            task::{Context, Poll, Waker},
        };
        // Real observation source, established outside the request envelope.
        let pending = freeze_deployment_resource_budget(
            2, 128, 5000, 1, 3, 1_000_000, 1_000_000, 1_000_000, 1_000_000,
        )
        .unwrap()
        .reserve_known_database_memory(0, 0, 0)
        .ok()
        .unwrap();
        let (application, listener) = BootstrapRendezvousIssuer::issue()
            .freeze_application(GeneratedApplicationFreezeSource::new(
                "app",
                b"descriptor",
                &["route"],
            ))
            .unwrap();
        let listener = listener
            .freeze_listener(ListenerStartupFreezeSource::new(
                "app",
                "127.0.0.1:8000".parse().unwrap(),
                "127.0.0.1:9000".parse().unwrap(),
                Duration::from_millis(5000),
            ))
            .ok()
            .unwrap();
        let (whole, receipt) = pair_bootstrap_rendezvous(application, listener)
            .ok()
            .unwrap();
        let budget = bind_deployment_resource_budget_bootstrap(pending, whole, receipt)
            .ok()
            .unwrap();
        let process = prepare_profusegw_lightweight_profile(budget).ok().unwrap();
        let (permit, original) = match process.verified_profile().try_admit_observed() {
            ProfuseGwLightweightObservedAdmissionOutcome::Ready(permit, receipt) => {
                (permit, receipt)
            }
            _ => panic!("fixture admission failed"),
        };
        let facts = AdmissionCapacityFacts::from_receipt(&original);
        let observer = Observer::with_writer(ObserverConfig::default(), io::sink()).unwrap();
        let app = ContextLabel::checked("app").unwrap();
        let context = AdmissionEventContext::Unrooted {
            application: &app,
            lifecycle: RequestViewPhase::Reading,
        };
        let (sender, receiver) = mpsc::sync_channel(QUEUE_CAPACITY);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        for _ in 0..63 {
            assert_eq!(
                handle.submit_fixed_record(&0),
                DiagnosticSubmission::Enqueued
            );
        }
        let ledger = ProcessLedger::new(ResourceConfig {
            managed_capacity: 1024 * 1024,
            entry_reserve: 1024 * 1024,
            framework_reserve: 1024 * 1024,
            task_reserve: 1024 * 1024,
            process_state_reserve: 1024 * 1024,
            system_estimate: 1024 * 1024,
            safety_margin: 1024 * 1024,
            process_limit: 8 * 1024 * 1024,
            max_active_requests: 1,
        })
        .unwrap();
        let future = ledger
            .try_envelope(65536, 65536, |_| async {
                let result = observer.record_admission_capacity(
                    Some(&handle),
                    context,
                    &facts,
                    AdmissionConstruction::Cancelled,
                );
                assert_eq!(result.cpu, DiagnosticSubmission::Enqueued);
                assert_eq!(result.memory, DiagnosticSubmission::Full);
                assert_eq!(result.database, DiagnosticSubmission::Full);
                assert_eq!(result.profuse_contract, DiagnosticSubmission::Full);
            })
            .unwrap();
        let mut future = std::pin::pin!(future);
        assert!(
            matches!(
                future
                    .as_mut()
                    .poll(&mut Context::from_waker(Waker::noop())),
                Poll::Ready(Ok(_))
            ),
            "actual unrooted encode must not escape the request allocation envelope"
        );
        for _ in 0..63 {
            receiver.try_recv().unwrap();
        }
        let row: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        assert_eq!(row["capacity_dimension"], "cpu");
        assert_eq!(row["decision"], "accepted");
        assert_eq!(row["resource_source"], "admission_ledger_same_lock");
        assert!(row["resource"]["managed_available_bytes"].as_u64().is_some());
        assert!(row["resource"]["framework_demand_bytes"].as_u64().is_some());
        assert_eq!(row["construction"]["status"], "cancelled");
        assert_eq!(handle.snapshot().enqueued, 64);
        assert_eq!(handle.snapshot().dropped, 3);
        // Existing encoder failure is tested independently; valid closed AE
        // fields cannot supply an arbitrary overlength string to force this.
        assert_eq!(
            handle.submit_fixed_record(&"x".repeat(8192)),
            DiagnosticSubmission::EncodingFailed
        );
        assert!(receiver.try_recv().is_err());
        drop(receiver);
        let closed = observer.record_admission_capacity(
            Some(&handle),
            context,
            &facts,
            AdmissionConstruction::Ready,
        );
        assert_eq!(
            [
                closed.cpu,
                closed.memory,
                closed.database,
                closed.profuse_contract
            ],
            [DiagnosticSubmission::Closed; 4]
        );
        for dimension in [
            CapacityDimension::Cpu,
            CapacityDimension::Memory,
            CapacityDimension::Database,
            CapacityDimension::ProfuseContract,
        ] {
            assert_eq!(
                observer
                    .metrics_snapshot()
                    .capacity_accepted(Some(dimension)),
                2
            );
        }
        assert_eq!(original.decision(), ProfuseGwCapacityDecision::Accepted);
        permit.into_execution().cancel();
        process.finish().unwrap();
    }

    #[test]
    fn root_original_partial_cycle_full_and_disclosure_are_explicit() {
        use crate::root_diagnostic::OriginalCaptureState;
        use crate::{RootDiagnosticScope, RootOutcomeFacts, RootRequestEvent};
        use saddle_core::*;
        #[derive(Debug)]
        struct BadFormat;
        impl std::fmt::Display for BadFormat {
            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                f.write_str("exact-before-format-error")?;
                Err(std::fmt::Error)
            }
        }
        impl std::error::Error for BadFormat {}
        #[derive(Debug)]
        struct Cycle;
        impl std::fmt::Display for Cycle {
            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                f.write_str("cycle-original")
            }
        }
        impl std::error::Error for Cycle {
            fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
                Some(self)
            }
        }
        let root = RequestRootPublisher::create(
            ContextLabel::checked("test").unwrap(),
            ContextFact::NotEstablished,
        )
        .unwrap();
        let view = root
            .reference()
            .view(RequestLocalFacts::new(RequestViewPhase::Handler));
        let (sender, receiver) = mpsc::sync_channel(64);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let source = |error: &(dyn std::error::Error + 'static), output| {
            RootDiagnosticScope::new(&view, output).source_error(
                error,
                DiagnosticStage::RequestHandler,
                RootRequestEvent::Handler,
                RootOutcomeFacts::default(),
            )
        };
        let missing = source(&BadFormat, None);
        assert_eq!(
            missing.original_capture(),
            OriginalCaptureState::OutputUnavailable
        );
        let partial = source(&BadFormat, Some(&handle));
        assert_eq!(
            partial.original_capture(),
            OriginalCaptureState::FormattingFailed
        );
        assert_eq!(partial.submission(), DiagnosticSubmission::EncodingFailed);
        let mut rows: Vec<serde_json::Value> = receiver
            .try_iter()
            .map(|p| serde_json::from_slice(p.bytes()).unwrap())
            .collect();
        assert_eq!(rows.last().unwrap()["payload"], "exact-before-format-error");
        assert_eq!(rows.last().unwrap()["state"], "formatting_failed");
        assert!(!rows.iter().any(|r| r["state"] == "exposed_chain_complete"));
        let cycle = source(&Cycle, Some(&handle));
        assert_eq!(cycle.original_capture(), OriginalCaptureState::CauseCycle);
        rows = receiver
            .try_iter()
            .map(|p| serde_json::from_slice(p.bytes()).unwrap())
            .collect();
        assert_eq!(rows.last().unwrap()["state"], "cause_cycle");
        // Same production queue, no retry/fallback or fabricated completion.
        for _ in 0..64 {
            assert_eq!(
                handle.submit_original_record(&0),
                DiagnosticSubmission::Enqueued
            );
        }
        let full = source(&Cycle, Some(&handle));
        assert_eq!(
            full.original_capture(),
            OriginalCaptureState::OutputRejected
        );
        assert_eq!(full.submission(), DiagnosticSubmission::Full);
        let _ = receiver.try_iter().count();
        handle.status.closed.store(true, Ordering::Release);
        assert_eq!(
            source(&Cycle, Some(&handle)).submission(),
            DiagnosticSubmission::Closed
        );
        // Raw original content is restricted to the existing protected file, not
        // mirrored into stderr if that file fails. Only output status may escape.
        let mut file: io::Result<Vec<u8>> = Err(io::ErrorKind::PermissionDenied.into());
        let mut stderr = Vec::new();
        let result = deliver_with_disclosure(
            &mut file,
            &mut stderr,
            std::path::Path::new("saddle.emergency.log"),
            b"CT_ARTIFICIAL_SECRET_NOT_A_CREDENTIAL",
            &Status::default(),
            &mut false,
            false,
        );
        assert_eq!(result.unwrap_err().kind(), io::ErrorKind::PermissionDenied);
        let stderr = String::from_utf8(stderr).unwrap();
        assert!(stderr.contains("output_failed"));
        assert!(!stderr.contains("CT_ARTIFICIAL_SECRET"));
    }

    #[test]
    fn root_original_admitted_and_existing_panic_keep_identity() {
        use crate::root_diagnostic::OriginalCaptureState;
        use crate::{RootDiagnosticScope, RootOutcomeFacts, RootRequestEvent};
        use saddle_core::*;
        use std::{
            future::Future,
            task::{Context, Poll, Waker},
        };
        let root = RequestRootPublisher::create(
            ContextLabel::checked("test").unwrap(),
            ContextFact::Unavailable,
        )
        .unwrap();
        let view = root
            .reference()
            .view(RequestLocalFacts::new(RequestViewPhase::Handler));
        let (sender, receiver) = mpsc::sync_channel(64);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let diagnostic = Diagnostic::capture(
            DiagnosticCategory::Panic,
            CaptureSite::Origin,
            DiagnosticCause::new(
                DiagnosticStage::RequestHandler,
                DiagnosticCode::new("panic.caught").unwrap(),
            ),
        );
        let id = diagnostic.id();
        let error = io::Error::from(io::ErrorKind::TimedOut);
        let ledger = saddle_admission::ProcessLedger::new(saddle_admission::ResourceConfig {
            managed_capacity: 1024 * 1024,
            entry_reserve: 1024 * 1024,
            framework_reserve: 1024 * 1024,
            task_reserve: 1024 * 1024,
            process_state_reserve: 1024 * 1024,
            system_estimate: 1024 * 1024,
            safety_margin: 1024 * 1024,
            process_limit: 8 * 1024 * 1024,
            max_active_requests: 1,
        })
        .unwrap();
        let future = ledger
            .try_envelope(65536, 65536, |_| async {
                let scope = RootDiagnosticScope::new(&view, Some(&handle));
                let failure = scope.source_error(
                    &error,
                    DiagnosticStage::RequestDb,
                    RootRequestEvent::Database,
                    RootOutcomeFacts::default(),
                );
                assert_eq!(
                    failure.original_capture(),
                    OriginalCaptureState::CompleteEnqueued
                );
                let panic = scope.source_existing_description(
                    &"raw panic: CT_ARTIFICIAL_SECRET_NOT_A_CREDENTIAL",
                    diagnostic,
                    DiagnosticCode::new("task.failed").unwrap(),
                    RootRequestEvent::Supervision,
                    RootOutcomeFacts::default(),
                );
                assert_eq!(
                    panic.original_capture(),
                    OriginalCaptureState::CompleteEnqueued
                );
                panic
                    .finish(&view, None, RootOutcomeFacts::default())
                    .ok()
                    .unwrap()
            })
            .unwrap();
        let mut future = std::pin::pin!(future);
        assert!(
            matches!(
                future
                    .as_mut()
                    .poll(&mut Context::from_waker(Waker::noop())),
                Poll::Ready(Ok(_))
            ),
            "source formatter/queue must not escape admission"
        );
        let rows: Vec<serde_json::Value> = receiver
            .try_iter()
            .map(|p| serde_json::from_slice(p.bytes()).unwrap())
            .collect();
        let panic: Vec<_> = rows
            .iter()
            .filter(|r| r["occurrence"]["diagnostic_id"] == id)
            .collect();
        assert_eq!(panic.len(), 4);
        let header: serde_json::Value =
            serde_json::from_str(panic[0]["payload"].as_str().unwrap()).unwrap();
        assert_eq!(header["facts"]["diagnostic_id"], id);
        assert_eq!(header["facts"]["category"], "panic");
        assert_eq!(
            header["source_interface"],
            "unavailable_description_debug_only"
        );
        assert_eq!(
            panic[1]["payload"],
            "raw panic: CT_ARTIFICIAL_SECRET_NOT_A_CREDENTIAL"
        );
        assert_eq!(panic[2]["channel"], "debug");
        assert_eq!(panic[3]["state"], "description_complete_source_unavailable");
    }
    use saddle_core::{
        CaptureSite, DiagnosticCategory, DiagnosticCause, DiagnosticCode, DiagnosticStage,
    };
    #[test]
    fn root_contract_oversize_detail_and_basic_encoding_failure_are_explicit() {
        use crate::{RootDiagnosticScope, RootOutcomeFacts, RootRequestEvent};
        use saddle_core::*;
        let (sender, receiver) = mpsc::sync_channel(4);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let text = "m".repeat(256);
        let protocol = "\\\"".repeat(128);
        let call = CallContext::new(
            ApplicationId::new(text.clone()),
            ModuleId::new(text.clone()),
            ServiceId::new(text.clone()),
            OperationId::new(text.clone()),
            TraceId::from_u128(1),
            SpanId::from_u64(1),
        )
        .with_trace_correlation_id(TraceCorrelationId::new(protocol.clone()).unwrap())
        .with_rpc_correlation_id(RpcCorrelationId::new(&protocol));
        let mut root = RequestRootPublisher::create(
            ContextLabel::checked(&text).unwrap(),
            ContextFact::NotEstablished,
        )
        .unwrap();
        assert!(
            root.publish(
                RequestIdentityGroup::from_validated(
                    &call,
                    &protocol,
                    &text,
                    1,
                    ContextFact::Present(ContextLabel::checked(&text).unwrap())
                )
                .unwrap()
            )
            .is_ok()
        );
        let view = root
            .reference()
            .view(RequestLocalFacts::new(RequestViewPhase::Database));
        let metadata = "T".repeat(192);
        let cause = BoundedDiagnosticCause::new(
            DiagnosticStage::RequestDb,
            DiagnosticCode::new("driver.type").unwrap(),
        )
        .with_object(InlineDiagnosticText::metadata(&metadata))
        .with_types(
            Some(InlineDiagnosticText::metadata(&metadata)),
            Some(InlineDiagnosticText::metadata(&metadata)),
            None,
        );
        let detail = BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            CaptureSite::Origin,
            cause,
        )
        .wrap(cause)
        .wrap(cause)
        .wrap(cause);
        let id = detail.id();
        let receipt = RootDiagnosticScope::new(&view, Some(&handle)).source(
            detail,
            DiagnosticCode::new("driver.type").unwrap(),
            RootRequestEvent::Database,
            RootOutcomeFacts::default(),
        );
        assert_eq!(receipt.submission(), DiagnosticSubmission::Enqueued);
        let row: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        assert_eq!(row["detail_status"], "omitted_encoding_capacity");
        assert_eq!(row["occurrence"]["diagnostic_id"], id);
        assert!(row["diagnostic"].is_null());
        assert_eq!(handle.snapshot().truncated, 1);
        let (receipt, result) = receipt
            .boundary(
                &view,
                Some(&handle),
                RootRequestEvent::Database,
                RootOutcomeFacts::default(),
            )
            .ok()
            .unwrap();
        assert_eq!(result, DiagnosticSubmission::EncodingFailed);
        assert!(receiver.try_recv().is_err());
        assert_eq!(receipt.submission(), DiagnosticSubmission::Enqueued);
        assert_eq!(handle.snapshot().dropped, 1);
    }

    #[test]
    fn root_contract_stage_terminal_is_single_and_admitted() {
        use crate::{
            Observer, ObserverConfig, RootDiagnosticScope, RootOutcomeFacts, RootRequestEvent,
            Stage, StageOutcome,
        };
        use saddle_core::*;
        use std::{
            future::Future,
            task::{Context, Poll, Waker},
        };
        let observer = Observer::with_writer(ObserverConfig::default(), std::io::sink()).unwrap();
        let root = RequestRootPublisher::create(
            ContextLabel::checked("app").unwrap(),
            ContextFact::NotEstablished,
        )
        .unwrap();
        let view = root
            .reference()
            .view(RequestLocalFacts::new(RequestViewPhase::Handler));
        let foreign = RequestRootPublisher::create(
            ContextLabel::checked("app").unwrap(),
            ContextFact::NotEstablished,
        )
        .unwrap();
        let foreign_view = foreign
            .reference()
            .view(RequestLocalFacts::new(RequestViewPhase::Handler));
        let ledger = saddle_admission::ProcessLedger::new(saddle_admission::ResourceConfig {
            managed_capacity: 1024 * 1024,
            entry_reserve: 1024 * 1024,
            framework_reserve: 1024 * 1024,
            task_reserve: 1024 * 1024,
            process_state_reserve: 1024 * 1024,
            system_estimate: 1024 * 1024,
            safety_margin: 1024 * 1024,
            process_limit: 8 * 1024 * 1024,
            max_active_requests: 1,
        })
        .unwrap();
        let future = ledger
            .try_envelope(65536, 65536, |_| async {
                let facts = |operation| RootOutcomeFacts {
                    axes: DiagnosticOutcomeAxes {
                        operation,
                        ..Default::default()
                    },
                    ..Default::default()
                };
                let stage = RootDiagnosticScope::new(&view, None)
                    .start_stage(&observer, RootRequestEvent::Response);
                // A technical failure cannot take the receipt-free completion path.
                let stage = stage
                    .finish_nonfailure(facts(OperationOutcome::Failed))
                    .err()
                    .unwrap();
                assert_eq!(
                    stage
                        .finish_nonfailure(facts(OperationOutcome::Succeeded))
                        .ok()
                        .unwrap(),
                    DiagnosticSubmission::OutputUnavailable
                );
                let source = RootDiagnosticScope::new(&view, None).source(
                    BoundedDiagnostic::capture(
                        DiagnosticCategory::UnexpectedError,
                        CaptureSite::Origin,
                        BoundedDiagnosticCause::new(
                            DiagnosticStage::RequestHandler,
                            DiagnosticCode::new("handler.failed").unwrap(),
                        ),
                    ),
                    DiagnosticCode::new("handler.failed").unwrap(),
                    RootRequestEvent::Handler,
                    facts(OperationOutcome::Failed),
                );
                let foreign_stage = RootDiagnosticScope::new(&foreign_view, None)
                    .start_stage(&observer, RootRequestEvent::Response);
                let (foreign_stage, source) = foreign_stage
                    .finish_failure(source, facts(OperationOutcome::Failed))
                    .err()
                    .unwrap();
                assert_eq!(
                    foreign_stage
                        .finish_nonfailure(facts(OperationOutcome::Rejected))
                        .ok()
                        .unwrap(),
                    DiagnosticSubmission::OutputUnavailable
                );
                let stage = RootDiagnosticScope::new(&view, None)
                    .start_stage(&observer, RootRequestEvent::Response);
                let (source, status) = stage
                    .finish_failure(source, facts(OperationOutcome::Failed))
                    .ok()
                    .unwrap();
                assert_eq!(status, DiagnosticSubmission::OutputUnavailable);
                source
                    .finish(&view, None, facts(OperationOutcome::Failed))
                    .ok()
                    .unwrap()
            })
            .unwrap();
        let mut future = std::pin::pin!(future);
        assert!(matches!(
            future
                .as_mut()
                .poll(&mut Context::from_waker(Waker::noop())),
            Poll::Ready(Ok(_))
        ));
        let metrics = observer.metrics_snapshot();
        assert_eq!(metrics.requests(StageOutcome::Success), 1);
        assert_eq!(metrics.requests(StageOutcome::Rejected), 1);
        assert_eq!(metrics.requests(StageOutcome::Failure), 1);
        assert_eq!(metrics.requests(StageOutcome::Cancelled), 0);
        assert_eq!(
            metrics.stage_latency(Stage::Response).iter().sum::<u64>(),
            3
        );
    }

    #[test]
    fn root_contract_existing_source_cleanup_and_admitted_submission() {
        use crate::{RootDiagnosticScope, RootOutcomeFacts, RootRequestEvent};
        use saddle_core::*;
        use std::{
            future::Future,
            task::{Context, Poll, Waker},
        };
        let (sender, receiver) = mpsc::sync_channel(4);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let root = RequestRootPublisher::create(
            ContextLabel::checked("app").unwrap(),
            ContextFact::Unavailable,
        )
        .unwrap();
        let view = root.reference().view(
            RequestLocalFacts::new(RequestViewPhase::Handler).with_task(ContextFact::Present(17)),
        );
        let diagnostic = Diagnostic::capture(
            DiagnosticCategory::Panic,
            CaptureSite::Origin,
            DiagnosticCause::new(
                DiagnosticStage::RequestHandler,
                DiagnosticCode::new("panic.caught").unwrap(),
            ),
        );
        let id = diagnostic.id();
        let first = RootDiagnosticScope::new(&view, Some(&handle)).source_existing(
            diagnostic,
            DiagnosticCode::new("task.failed").unwrap(),
            RootRequestEvent::Supervision,
            RootOutcomeFacts::default(),
        );
        let occurrence = first.occurrence();
        let ledger = saddle_admission::ProcessLedger::new(saddle_admission::ResourceConfig {
            managed_capacity: 1024 * 1024,
            entry_reserve: 1024 * 1024,
            framework_reserve: 1024 * 1024,
            task_reserve: 1024 * 1024,
            process_state_reserve: 1024 * 1024,
            system_estimate: 1024 * 1024,
            safety_margin: 1024 * 1024,
            process_limit: 8 * 1024 * 1024,
            max_active_requests: 1,
        })
        .unwrap();
        let future = ledger
            .try_envelope(65536, 65536, |_| async {
                let cleanup = BoundedDiagnostic::capture(
                    DiagnosticCategory::UnexpectedError,
                    CaptureSite::Origin,
                    BoundedDiagnosticCause::new(
                        DiagnosticStage::FinalizerResource,
                        DiagnosticCode::new("cleanup.failed").unwrap(),
                    ),
                )
                .during_cleanup_occurrence(occurrence);
                let second = RootDiagnosticScope::new(&view, Some(&handle)).source(
                    cleanup,
                    DiagnosticCode::new("cleanup.failed").unwrap(),
                    RootRequestEvent::Finalization,
                    RootOutcomeFacts::default(),
                );
                assert_eq!(second.submission(), DiagnosticSubmission::Enqueued);
                first
                    .finish(&view, None, RootOutcomeFacts::default())
                    .ok()
                    .unwrap()
            })
            .unwrap();
        let mut future = std::pin::pin!(future);
        let result = future
            .as_mut()
            .poll(&mut Context::from_waker(Waker::noop()));
        assert!(
            matches!(result, Poll::Ready(Ok(_))),
            "bounded receipt path must not allocate outside admitted contract: {result:?}"
        );
        let primary: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        let cleanup: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        assert_eq!(primary["diagnostic"]["diagnostic_id"], id);
        assert_eq!(primary["category"], "panic");
        assert_eq!(
            primary["diagnostic"]["stack_status"],
            "unavailable_deferred"
        );
        assert_eq!(cleanup["occurrence"]["primary_diagnostic_id"], id);
        assert_ne!(cleanup["occurrence"]["diagnostic_id"], id);
        assert_eq!(cleanup["context"], primary["context"]);
    }

    #[test]
    fn root_contract_real_file_readback_preserves_safe_facts_without_secret() {
        use crate::{RootDiagnosticScope, RootOutcomeFacts, RootRequestEvent};
        use saddle_core::*;
        let (sender, receiver) = mpsc::sync_channel(1);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let root = RequestRootPublisher::create(
            ContextLabel::checked("app").unwrap(),
            ContextFact::NotEstablished,
        )
        .unwrap();
        let view = root
            .reference()
            .view(RequestLocalFacts::new(RequestViewPhase::Reading));
        let cause = BoundedDiagnosticCause::new(
            DiagnosticStage::RequestDb,
            DiagnosticCode::new("db.type").unwrap(),
        )
        .with_object(InlineDiagnosticText::metadata(
            "mysql://SECRET_SENTINEL@host",
        ))
        .with_sqlstate("22003")
        .with_database_code(1264)
        .with_column(Some(3), Some(4))
        .with_types(
            Some(InlineDiagnosticText::metadata("i32")),
            Some(InlineDiagnosticText::metadata("BIGINT")),
            None,
        );
        let failure = RootDiagnosticScope::new(&view, Some(&handle)).source(
            BoundedDiagnostic::capture(
                DiagnosticCategory::UnexpectedError,
                CaptureSite::Origin,
                cause,
            ),
            DiagnosticCode::new("db.type").unwrap(),
            RootRequestEvent::Database,
            RootOutcomeFacts::default(),
        );
        let public = failure
            .finish(&view, None, RootOutcomeFacts::default())
            .ok()
            .unwrap();
        drop((view, root));
        let path = std::env::temp_dir().join(format!(
            "saddle-c-readback-{}-{}.jsonl",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let file = std::fs::OpenOptions::new()
            .create_new(true)
            .write(true)
            .open(&path)
            .unwrap();
        let packet = receiver.try_recv().unwrap();
        let mut output = Ok(file);
        deliver(
            &mut output,
            &mut io::sink(),
            &path,
            packet.bytes(),
            &handle.status,
            &mut false,
        );
        output.as_mut().unwrap().flush().unwrap();
        let raw = std::fs::read(&path).unwrap();
        std::fs::remove_file(&path).unwrap();
        assert!(!String::from_utf8_lossy(&raw).contains("SECRET_SENTINEL"));
        let row: serde_json::Value = serde_json::from_slice(&raw).unwrap();
        assert_eq!(row["diagnostic"]["causes"][0]["object"]["redacted"], true);
        assert_eq!(row["diagnostic"]["causes"][0]["sqlstate"]["value"], "22003");
        assert_eq!(row["diagnostic"]["causes"][0]["column_index"], 3);
        assert_eq!(handle.snapshot().written, 1);
        assert_eq!(public.source_submission(), DiagnosticSubmission::Enqueued);
        assert_eq!(
            public.terminal_submission(),
            DiagnosticSubmission::OutputUnavailable
        );
    }
    #[test]
    fn root_contract_source_boundary_supervision_and_queue_degradation() {
        use crate::{
            RootDiagnosticScope, RootOutcomeFacts, RootRequestEvent,
        };
        use saddle_core::{
            BoundedDiagnostic, BoundedDiagnosticCause, ContextFact, ContextLabel,
            RequestLocalFacts, RequestRootPublisher, RequestViewPhase,
        };
        let (sender, receiver) = mpsc::sync_channel(2);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let root = RequestRootPublisher::create(
            ContextLabel::checked("app").unwrap(),
            ContextFact::NotEstablished,
        )
        .unwrap();
        let view = root
            .reference()
            .view(RequestLocalFacts::new(RequestViewPhase::Reading));
        let make = |output| {
            RootDiagnosticScope::new(&view, output).source(
                BoundedDiagnostic::capture(
                    DiagnosticCategory::UnexpectedError,
                    CaptureSite::Origin,
                    BoundedDiagnosticCause::new(
                        DiagnosticStage::RequestDecode,
                        DiagnosticCode::new("read.failed").unwrap(),
                    ),
                ),
                DiagnosticCode::new("decode.failed").unwrap(),
                RootRequestEvent::Ingress,
                RootOutcomeFacts::default(),
            )
        };
        let missing = make(None);
        assert_eq!(
            missing.submission(),
            DiagnosticSubmission::OutputUnavailable
        );
        let receipt = make(Some(&handle));
        assert_eq!(receipt.submission(), DiagnosticSubmission::Enqueued);
        let foreign = RequestRootPublisher::create(
            ContextLabel::checked("app").unwrap(),
            ContextFact::NotEstablished,
        )
        .unwrap()
        .reference()
        .view(RequestLocalFacts::new(RequestViewPhase::Reading));
        let receipt = receipt
            .boundary(
                &foreign,
                Some(&handle),
                RootRequestEvent::Supervision,
                RootOutcomeFacts::default(),
            )
            .err()
            .unwrap();
        let public = receipt
            .finish(&view, Some(&handle), RootOutcomeFacts::default())
            .ok()
            .unwrap();
        assert_eq!(make(Some(&handle)).submission(), DiagnosticSubmission::Full);
        handle.status.closed.store(true, Ordering::Release);
        assert_eq!(
            make(Some(&handle)).submission(),
            DiagnosticSubmission::Closed
        );
        drop((missing, foreign, view, root));
        // These are the actual existing queue packets, still unread until all
        // request owners above ended. Public error is Copy and root-free.
        let source: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        let boundary: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        let public = serde_json::to_value(public).unwrap();
        assert_eq!(source["occurrence"], boundary["occurrence"]);
        assert_eq!(source["occurrence"], public["occurrence"]);
        assert_eq!(source["context"], boundary["source_context"]);
        assert_eq!(source["context"]["trace_id"]["state"], "not_established");
        assert!(source["diagnostic"].is_object());
        assert!(boundary["diagnostic"].is_null());
        assert!(public.get("context").is_none());
        assert_eq!(handle.snapshot().written, 0);
        assert_eq!(handle.snapshot().enqueued, 2);
        assert_eq!(handle.snapshot().dropped, 2);
    }
    #[test]
    fn outbound_child_queue_source_boundary_full_closed() {
        use crate::{EventContext, RequestDiagnosticScope, RequestIdentity, RouteIdentity};
        use saddle_core::{
            BoundedDiagnosticCause, CallContext, DiagnosticOutcomeAxes, RpcCorrelationId, SpanId,
            TraceId,
        };
        let (sender, receiver) = mpsc::sync_channel(2);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let call = |rpc: &str, span| {
            CallContext::new(
                "app".into(),
                "module".into(),
                "service".into(),
                "operation".into(),
                TraceId::from_u128(1),
                SpanId::from_u64(span),
            )
            .with_rpc_correlation_id(RpcCorrelationId::new(rpc))
        };
        let parent_call = call("0", 1);
        let child_call = call("0.1", 2);
        let event = EventContext::new(
            RequestIdentity::new("request-1").unwrap(),
            RouteIdentity::new("route").unwrap(),
            1,
        )
        .unwrap();
        let parent = RequestDiagnosticScope::established(&handle, &parent_call, &event);
        let child = parent.derive_outbound_child(&child_call, &event).unwrap();
        let fail = || {
            child.fail(
                7,
                DiagnosticCategory::UnexpectedError,
                BoundedDiagnosticCause::new(
                    DiagnosticStage::RequestDecode,
                    DiagnosticCode::new("outbound.failed").unwrap(),
                ),
            )
        };
        let source = fail();
        assert_eq!(source.submission(), DiagnosticSubmission::Enqueued);
        let (retained, delivery) =
            source.finish_boundary_retained(Some(&handle), &DiagnosticOutcomeAxes::default());
        assert_eq!(*retained.error(), 7);
        assert_eq!(
            delivery.boundary_submission(),
            DiagnosticSubmission::Enqueued
        );
        assert_eq!(fail().submission(), DiagnosticSubmission::Full);
        handle.status.closed.store(true, Ordering::Release);
        assert_eq!(fail().submission(), DiagnosticSubmission::Closed);
        let source: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        let boundary: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        assert_eq!(source["context"], boundary["context"]);
        assert_eq!(
            source["diagnostic_reference"],
            boundary["diagnostic_reference"]
        );
        assert_eq!(source["context"]["rpc_id"]["value"], "0.1");
        assert_eq!(source["context"]["span_id"]["value"], "0000000000000002");
        assert!(source["diagnostic"].is_object());
        assert!(boundary["diagnostic"].is_null());
        assert_eq!(handle.snapshot().written, 0);
    }
    #[test]
    fn early_request_source_boundary_admitted_full_closed_and_missing_output() {
        use crate::{DiagnosticRequestPhase, EarlyRequestContext, RequestDiagnosticScope};
        use saddle_core::{BoundedDiagnosticCause, DiagnosticOutcomeAxes, OperationOutcome};
        use std::{
            future::Future,
            task::{Context, Poll, Waker},
        };
        let (sender, receiver) = mpsc::sync_channel(2);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let known_call = saddle_core::CallContext::new(
            "app".into(),
            "module".into(),
            "service".into(),
            "operation".into(),
            saddle_core::TraceId::from_u128(1),
            saddle_core::SpanId::from_u64(2),
        );
        let known_event = crate::EventContext::new(
            crate::RequestIdentity::new("real-request").unwrap(),
            crate::RouteIdentity::new("/route").unwrap(),
            1,
        )
        .unwrap();
        let ledger = saddle_admission::ProcessLedger::new(saddle_admission::ResourceConfig {
            managed_capacity: 1024 * 1024,
            entry_reserve: 1024 * 1024,
            framework_reserve: 1024 * 1024,
            task_reserve: 1024 * 1024,
            process_state_reserve: 1024 * 1024,
            system_estimate: 1024 * 1024,
            safety_margin: 1024 * 1024,
            process_limit: 8 * 1024 * 1024,
            max_active_requests: 1,
        })
        .unwrap();
        let future = ledger
            .try_envelope(65536, 65536, |_| async {
                let axes = DiagnosticOutcomeAxes {
                    operation: OperationOutcome::Failed,
                    ..Default::default()
                };
                let cause = || {
                    BoundedDiagnosticCause::new(
                        DiagnosticStage::RequestDecode,
                        DiagnosticCode::new("request.read_failed").unwrap(),
                    )
                };
                let enriched = RequestDiagnosticScope::early(
                    None,
                    EarlyRequestContext::socket_accepted("app"),
                )
                .with_task(crate::DiagnosticTaskId::from_runtime_id("42").unwrap())
                .unwrap_or_else(|_| panic!("task conflict"))
                .bind_established(&known_call, &known_event)
                .unwrap_or_else(|_| panic!("context conflict"));
                let unchanged =
                    enriched
                        .reborrow()
                        .fail(22u32, DiagnosticCategory::UnexpectedError, cause());
                assert_eq!(
                    unchanged.submission(),
                    DiagnosticSubmission::OutputUnavailable
                );
                let scope = RequestDiagnosticScope::early(
                    Some(&handle),
                    EarlyRequestContext::socket_accepted("app"),
                )
                .with_phase(DiagnosticRequestPhase::ReadingHead);
                let failure = scope.fail(23u32, DiagnosticCategory::UnexpectedError, cause());
                let id = failure.source_diagnostic().id();
                assert_eq!(failure.submission(), DiagnosticSubmission::Enqueued);
                let (failure, delivered) = failure.finish_boundary_retained(Some(&handle), &axes);
                assert_eq!(
                    delivered.boundary_submission(),
                    DiagnosticSubmission::Enqueued
                );
                assert_eq!(*failure.error(), 23);
                let full = scope.fail(24u32, DiagnosticCategory::UnexpectedError, cause());
                assert_eq!(full.submission(), DiagnosticSubmission::Full);
                let (full, delivered) = full.finish_boundary_retained(None, &axes);
                assert_eq!(*full.error(), 24);
                assert_eq!(delivered.source_submission(), DiagnosticSubmission::Full);
                assert_eq!(
                    delivered.boundary_submission(),
                    DiagnosticSubmission::OutputUnavailable
                );
                handle.status.closed.store(true, Ordering::Release);
                let closed = scope.fail(25u32, DiagnosticCategory::UnexpectedError, cause());
                assert_eq!(closed.submission(), DiagnosticSubmission::Closed);
                let no_output = scope.with_output(None).fail(
                    26u32,
                    DiagnosticCategory::UnexpectedError,
                    cause(),
                );
                assert_eq!(
                    no_output.submission(),
                    DiagnosticSubmission::OutputUnavailable
                );
                assert_eq!(*no_output.error(), 26);
                id
            })
            .unwrap();
        let mut future = std::pin::pin!(future);
        let result = future
            .as_mut()
            .poll(&mut Context::from_waker(Waker::noop()));
        assert!(
            matches!(result, Poll::Ready(Ok(_))),
            "early path escaped: {result:?}"
        );
        let source: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        let boundary: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        assert_eq!(source["context"], boundary["context"]);
        assert_eq!(
            source["diagnostic_reference"],
            boundary["diagnostic_reference"]
        );
        assert_eq!(source["context"]["trace_id"]["state"], "not_established");
        assert_eq!(
            source["context"]["lifecycle"]["value"]["value"],
            "reading_head"
        );
        assert!(source["diagnostic"].is_object());
        assert!(boundary["diagnostic"].is_null());
        assert_eq!(boundary["axes"]["operation"], "failed");
        assert_eq!(handle.snapshot().written, 0); // queue submission is not delivery
    }
    use std::time::Instant;
    #[test]
    fn early_retained_reference_late_output_keeps_original_submission_and_context() {
        use crate::{EarlyRequestContext, RequestDiagnosticScope};
        use saddle_core::{BoundedDiagnostic, BoundedDiagnosticCause, DiagnosticOutcomeAxes};
        let (sender, receiver) = mpsc::sync_channel(2);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let scope =
            RequestDiagnosticScope::early(None, EarlyRequestContext::socket_accepted("app"));
        let diagnostic = BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            BoundedDiagnosticCause::new(
                DiagnosticStage::RequestDecode,
                DiagnosticCode::new("request.read_failed").unwrap(),
            ),
        );
        let source = serde_json::to_value(&diagnostic).unwrap();
        let reference = scope.capture_required(diagnostic).into_reference();
        let (retained, delivery) =
            reference.finish_retained(Some(&handle), &DiagnosticOutcomeAxes::default());
        assert_eq!(
            delivery.source_submission(),
            DiagnosticSubmission::OutputUnavailable
        );
        assert_eq!(
            delivery.boundary_submission(),
            DiagnosticSubmission::Enqueued
        );
        assert_eq!(
            serde_json::to_value(retained.source_diagnostic()).unwrap(),
            source
        );
        let packet = receiver.try_recv().unwrap();
        let row: serde_json::Value = serde_json::from_slice(packet.bytes()).unwrap();
        assert_eq!(row["source_submission"], "output_unavailable");
        assert_eq!(row["context"]["trace_id"]["state"], "not_established");
        assert!(row["diagnostic"].is_null()); // late boundary is not a fabricated source retry
        assert!(receiver.try_recv().is_err());
        assert_eq!(handle.snapshot().written, 0);
    }
    static FILE_TEST: std::sync::Mutex<()> = std::sync::Mutex::new(());
    #[test]
    fn zone_projection_exact_missing_and_unsafe_are_explicit() {
        use crate::{
            DiagnosticContextMissing as Missing, DiagnosticZone, EventContext,
            RequestDiagnosticScope, RequestIdentity, RouteIdentity,
        };
        let (sender, receiver) = mpsc::sync_channel(2);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let call = CallContext::new(
            "app".into(),
            "module".into(),
            "service".into(),
            "query".into(),
            saddle_core::TraceId::from_u128(1),
            saddle_core::SpanId::from_u64(2),
        );
        let event = EventContext::new(
            RequestIdentity::new("r1").unwrap(),
            RouteIdentity::new("/query").unwrap(),
            1,
        )
        .unwrap();
        let cause = saddle_core::BoundedDiagnosticCause::new(
            DiagnosticStage::RequestDb,
            DiagnosticCode::new("db.decode.type").unwrap(),
        );
        for zone in ["zone-a", "杭州-zone/蓝"] {
            let scope = RequestDiagnosticScope::established(&handle, &call, &event)
                .with_zone(DiagnosticZone::from_validated_ingress(zone).unwrap())
                .with_zone_missing(Missing::Unavailable);
            let source = scope
                .fail((), DiagnosticCategory::UnexpectedError, cause)
                .into_reference();
            source.record(&handle, &saddle_core::DiagnosticOutcomeAxes::default());
            for _ in 0..2 {
                let row: serde_json::Value =
                    serde_json::from_slice(receiver.recv().unwrap().bytes()).unwrap();
                assert_eq!(
                    row["context"]["zone"],
                    serde_json::json!({"state":"present","value":zone})
                );
            }
        }
        for (reason, expected) in [
            (Missing::NotApplicable, "not_applicable"),
            (Missing::NotEstablished, "not_established"),
            (Missing::Unavailable, "unavailable"),
        ] {
            RequestDiagnosticScope::established(&handle, &call, &event)
                .with_zone_missing(reason)
                .fail((), DiagnosticCategory::UnexpectedError, cause);
            let row: serde_json::Value =
                serde_json::from_slice(receiver.recv().unwrap().bytes()).unwrap();
            assert_eq!(
                row["context"]["zone"],
                serde_json::json!({"state":expected})
            );
        }
        for invalid in [
            "",
            " ",
            "https://user:SECRET@host",
            "zone\nSECRET",
            "zone?token=SECRET",
        ] {
            assert!(DiagnosticZone::from_validated_ingress(invalid).is_err());
        }
        assert!(DiagnosticZone::from_validated_ingress(&"a".repeat(257)).is_err());
        assert!(DiagnosticZone::from_validated_ingress(&"a".repeat(256)).is_ok());
    }
    #[test]
    fn existing_source_receipt_keeps_stack_and_submits_once() {
        use crate::{EventContext, RequestDiagnosticScope, RequestIdentity, RouteIdentity};
        let (sender, receiver) = mpsc::sync_channel(2);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let call = CallContext::new(
            "app".into(),
            "module".into(),
            "service".into(),
            "query".into(),
            saddle_core::TraceId::from_u128(1),
            saddle_core::SpanId::from_u64(2),
        );
        let event = EventContext::new(
            RequestIdentity::new("request-1").unwrap(),
            RouteIdentity::new("/query").unwrap(),
            1,
        )
        .unwrap();
        let diagnostic = Diagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            DiagnosticCause::new(
                DiagnosticStage::RequestDb,
                DiagnosticCode::new("db.decode.type").unwrap(),
            ),
        );
        let before = serde_json::to_value(&diagnostic).unwrap();
        let receipt = RequestDiagnosticScope::established(&handle, &call, &event)
            .capture_existing(diagnostic, None);
        assert_eq!(receipt.submission(), DiagnosticSubmission::Enqueued);
        let reference = receipt.into_reference();
        assert_eq!(
            serde_json::to_value(reference.source_diagnostic()).unwrap(),
            before
        );
        let source: serde_json::Value =
            serde_json::from_slice(receiver.recv().unwrap().bytes()).unwrap();
        assert_eq!(source["diagnostic"], before);
        assert!(receiver.try_recv().is_err());
        assert_eq!(handle.snapshot().enqueued, 1);
        assert_eq!(
            reference.record(&handle, &saddle_core::DiagnosticOutcomeAxes::default()),
            DiagnosticSubmission::Enqueued
        );
        let boundary: serde_json::Value =
            serde_json::from_slice(receiver.recv().unwrap().bytes()).unwrap();
        assert!(boundary["diagnostic"].is_null());
        assert_eq!(
            source["diagnostic_reference"],
            boundary["diagnostic_reference"]
        );
        assert_eq!(source["context"], boundary["context"]);
        assert_eq!(handle.snapshot().enqueued, 2);
        assert_eq!(handle.snapshot().written, 0);
    }
    #[test]
    fn controlled_request_source_preserves_error_projection_and_degradation() {
        use crate::{EventContext, RequestDiagnosticScope, RequestIdentity, RouteIdentity};
        use saddle_core::*;
        let (sender, receiver) = mpsc::sync_channel(2);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let call = CallContext::new(
            "app".into(),
            "module".into(),
            "service".into(),
            "query".into(),
            TraceId::from_u128(1),
            SpanId::from_u64(2),
        );
        let event = EventContext::new(
            RequestIdentity::new("req-1").unwrap(),
            RouteIdentity::new("/agent/{code}").unwrap(),
            2,
        )
        .unwrap();
        let scope = RequestDiagnosticScope::established(&handle, &call, &event);
        let cause = BoundedDiagnosticCause::new(
            DiagnosticStage::RequestDb,
            DiagnosticCode::new("db.decode.type").unwrap(),
        );
        let failure = scope.fail(17u32, DiagnosticCategory::UnexpectedError, cause);
        assert_eq!(failure.submission(), DiagnosticSubmission::Enqueued);
        let failure = failure.map_error(|error| (error, false));
        assert_eq!(failure.error(), &(17, false));
        assert_eq!(
            failure.record_boundary(&handle, &DiagnosticOutcomeAxes::default()),
            DiagnosticSubmission::Enqueued
        );
        let full = scope.fail(19, DiagnosticCategory::UnexpectedError, cause);
        assert_eq!(full.submission(), DiagnosticSubmission::Full);
        assert_eq!(full.error(), &19);
        let source: serde_json::Value =
            serde_json::from_slice(receiver.recv().unwrap().bytes()).unwrap();
        let boundary: serde_json::Value =
            serde_json::from_slice(receiver.recv().unwrap().bytes()).unwrap();
        assert_eq!(source["context"], boundary["context"]);
        assert_eq!(
            source["diagnostic_reference"],
            boundary["diagnostic_reference"]
        );
        assert!(boundary["diagnostic"].is_null());
        assert!(source["diagnostic"].is_object());
        let c = &source["context"];
        for (key, expected) in [
            ("application", "app"),
            ("module", "module"),
            ("service", "service"),
            ("operation", "query"),
            ("request", "req-1"),
            ("route", "/agent/{code}"),
        ] {
            assert_eq!(c[key]["value"]["value"], expected);
        }
        assert_eq!(c["attempt"]["value"], 2);
        assert_eq!(c["span_id"]["value"], "0000000000000002");
        assert_eq!(c["rpc_id"]["state"], "unavailable");
        for field in ["scope", "task", "lifecycle", "zone", "target"] {
            assert_eq!(c[field]["state"], "unavailable");
        }
        handle.status.closed.store(true, Ordering::Release);
        let closed = scope.fail(21, DiagnosticCategory::UnexpectedError, cause);
        assert_eq!(closed.submission(), DiagnosticSubmission::Closed);
        assert_eq!(closed.error(), &21);
        assert_eq!(handle.snapshot().written, 0);
        let unavailable = RequestDiagnosticScope::output_unavailable(&call, &event);
        let original = BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            cause,
        );
        let original_id = original.id();
        let reference = unavailable.capture_required(original).into_reference();
        assert_eq!(
            reference.source_submission(),
            DiagnosticSubmission::OutputUnavailable
        );
        assert_eq!(reference.source_diagnostic().id(), original_id);
        assert_eq!(
            reference.record(&handle, &DiagnosticOutcomeAxes::default()),
            DiagnosticSubmission::Closed
        );
    }
    #[test]
    fn source_detail_and_boundary_reference_are_separate() {
        use saddle_core::*;
        let (sender, receiver) = mpsc::sync_channel(4);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let cause = BoundedDiagnosticCause::new(
            DiagnosticStage::RequestDb,
            DiagnosticCode::new("db.decode.type").unwrap(),
        );
        let primary = BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            cause,
        );
        let cleanup = BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            cause,
        )
        .during_cleanup_of(&primary);
        let reference = cleanup.occurrence();
        let axes = DiagnosticOutcomeAxes::default();
        assert_eq!(
            handle.submit_bounded(Some(&cleanup), &axes, None),
            DiagnosticSubmission::Enqueued
        );
        drop(cleanup);
        assert_eq!(
            handle.submit_boundary(Some(reference), &axes, None),
            DiagnosticSubmission::Enqueued
        );
        assert_eq!(
            handle.submit_boundary_parts(Some(reference), &axes, None, Some(("database", 1))),
            DiagnosticSubmission::Enqueued
        );
        let source: serde_json::Value =
            serde_json::from_slice(receiver.recv().unwrap().bytes()).unwrap();
        assert!(source["diagnostic"].is_object());
        for event in ["framework.boundary.outcome", "framework.stage.finished"] {
            let boundary: serde_json::Value =
                serde_json::from_slice(receiver.recv().unwrap().bytes()).unwrap();
            assert_eq!(boundary["event"], event);
            assert!(boundary["diagnostic"].is_null());
            assert_eq!(
                boundary["diagnostic_reference"],
                source["diagnostic_reference"]
            );
            assert_eq!(
                boundary["diagnostic_reference"]["primary_diagnostic_id"],
                primary.id()
            );
            assert!(!boundary.to_string().contains("db.decode.type"));
        }
        assert_eq!(handle.snapshot().written, 0);
    }
    #[test]
    fn deferred_unavailable_preserves_source_through_output_worker() {
        use saddle_core::*;
        let (sender, receiver) = mpsc::sync_channel(1);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let diagnostic = Diagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            DiagnosticCause::new(
                DiagnosticStage::RequestDb,
                DiagnosticCode::new("db.decode.type").unwrap(),
            ),
        );
        let original = serde_json::to_value(&diagnostic).unwrap();
        assert_eq!(original["stack_status"], "unavailable_deferred");
        assert!(diagnostic.deferred_stack().is_none());
        assert!(original["stack"].as_str().unwrap().is_empty());
        assert_eq!(handle.submit(&diagnostic), DiagnosticSubmission::Enqueued);
        let packet = receiver.try_recv().unwrap();
        let queued: serde_json::Value = serde_json::from_slice(packet.bytes()).unwrap();
        assert_eq!(queued["diagnostic"], original);
        let status = handle.status.clone();
        let row = std::thread::spawn(move || {
            let mut packet = packet;
            packet.resolve_on_worker(&status);
            serde_json::from_slice::<serde_json::Value>(packet.bytes()).unwrap()
        })
        .join()
        .unwrap();
        for key in ["diagnostic_id", "primary_diagnostic_id", "origin", "causes"] {
            assert_eq!(row["diagnostic"][key], original[key]);
        }
        assert_eq!(row["diagnostic"]["stack_status"], "unavailable_deferred");
        assert!(row["diagnostic"]["stack"].as_str().unwrap().is_empty());
        assert_eq!(serde_json::to_value(&diagnostic).unwrap(), original);
    }
    #[test]
    fn request_stage_finishes_once_with_original_receipt_context() {
        use saddle_core::*;
        let (sender, receiver) = mpsc::sync_channel(4);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let observer =
            crate::Observer::with_writer(crate::ObserverConfig::default(), io::sink()).unwrap();
        let (call, _) = observer
            .start_external_call_checked("app", "module", "service", "query", None)
            .unwrap();
        let event = crate::EventContext::new(
            crate::RequestIdentity::new("original-request").unwrap(),
            crate::RouteIdentity::new("/query").unwrap(),
            1,
        )
        .unwrap();
        let mut scope = crate::RequestDiagnosticScope::established(&handle, call.context(), &event)
            .with_zone(crate::DiagnosticZone::from_validated_ingress("zone-a").unwrap())
            .with_db_operation(
                crate::DiagnosticDbOperation::from_registered("AgentByCode").unwrap(),
            )
            .with_db_operation_missing(crate::DiagnosticContextMissing::Unavailable);
        let reference = scope
            .capture_required(BoundedDiagnostic::capture(
                DiagnosticCategory::UnexpectedError,
                CaptureSite::FirstObserved,
                BoundedDiagnosticCause::new(
                    DiagnosticStage::RequestDb,
                    DiagnosticCode::new("db.decode.type").unwrap(),
                ),
            ))
            .into_reference();
        scope.set_db_operation(
            crate::DiagnosticDbOperation::from_registered("NextOperation").unwrap(),
        );
        let stage = observer.start_stage(call.context(), crate::Stage::Database, event.clone());
        let axes = DiagnosticOutcomeAxes {
            operation: OperationOutcome::Failed,
            physical: PhysicalDispositionFact::Returned,
            ..Default::default()
        };
        let delivery = stage.finish_request_failure(Some(&handle), &axes, &reference);
        assert_eq!(
            delivery.boundary_submission(),
            DiagnosticSubmission::Enqueued
        );
        let source: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        let boundary: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        assert_eq!(source["context"], boundary["context"]);
        assert_eq!(source["context"]["db_operation"]["value"], "AgentByCode");
        assert_eq!(source["context"]["operation"]["value"]["value"], "query");
        let next = scope.fail(
            (),
            DiagnosticCategory::ExpectedRejection,
            BoundedDiagnosticCause::new(
                DiagnosticStage::RequestDb,
                DiagnosticCode::new("db.next").unwrap(),
            ),
        );
        assert_eq!(next.submission(), DiagnosticSubmission::Enqueued);
        let next: serde_json::Value =
            serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
        assert_eq!(next["context"]["db_operation"]["value"], "NextOperation");
        for field in ["operation", "request", "scope", "zone"] {
            assert_eq!(next["context"][field], source["context"][field]);
        }
        assert!(crate::DiagnosticDbOperation::from_registered("SELECT * FROM private").is_none());
        assert!(crate::DiagnosticDbOperation::from_registered("https://secret@host").is_none());
        assert_eq!(
            source["diagnostic_reference"],
            boundary["diagnostic_reference"]
        );
        assert_eq!(boundary["source_submission"], "enqueued");
        assert_eq!(boundary["axes"]["operation"], "failed");
        assert!(boundary["diagnostic"].is_null());
        assert!(boundary["elapsed_ms"].is_u64());
        assert!(receiver.try_recv().is_err());
        let stage = observer.start_stage(call.context(), crate::Stage::Database, event);
        assert_eq!(
            stage
                .finish_request_failure(None, &axes, &reference)
                .boundary_submission(),
            DiagnosticSubmission::OutputUnavailable
        );
        assert!(receiver.try_recv().is_err());
        assert_eq!(
            observer
                .metrics_snapshot()
                .stage_latency(crate::Stage::Database)
                .iter()
                .sum::<u64>(),
            2
        );
        call.succeed();
        let _ = observer.shutdown();
    }
    #[test]
    fn nonfailure_finish_optional_output_and_no_failed_bypass() {
        use saddle_core::*;
        let (sender, receiver) = mpsc::sync_channel(2);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let observer =
            crate::Observer::with_writer(crate::ObserverConfig::default(), io::sink()).unwrap();
        let (call, _) = observer
            .start_external_call_checked("app", "module", "service", "query", None)
            .unwrap();
        let event = crate::EventContext::new(
            crate::RequestIdentity::new("r").unwrap(),
            crate::RouteIdentity::new("/query").unwrap(),
            1,
        )
        .unwrap();
        for operation in [OperationOutcome::Succeeded, OperationOutcome::Rejected] {
            let stage = observer.start_stage(call.context(), crate::Stage::Database, event.clone());
            let axes = DiagnosticOutcomeAxes {
                operation,
                ..Default::default()
            };
            assert!(matches!(
                stage.finish_nonfailure_bounded(Some(&handle), &axes),
                Ok(DiagnosticSubmission::Enqueued)
            ));
            let row: serde_json::Value =
                serde_json::from_slice(receiver.try_recv().unwrap().bytes()).unwrap();
            assert!(row["diagnostic_reference"].is_null());
            assert!(row["diagnostic"].is_null());
            assert!(receiver.try_recv().is_err());
            let stage = observer.start_stage(call.context(), crate::Stage::Database, event.clone());
            assert!(matches!(
                stage.finish_nonfailure_bounded(None, &axes),
                Ok(DiagnosticSubmission::OutputUnavailable)
            ));
        }
        for operation in [
            OperationOutcome::Failed,
            OperationOutcome::Panicked,
            OperationOutcome::Unknown,
            OperationOutcome::TimedOut,
            OperationOutcome::Cancelled,
        ] {
            let stage = observer.start_stage(call.context(), crate::Stage::Database, event.clone());
            let axes = DiagnosticOutcomeAxes {
                operation,
                ..Default::default()
            };
            let stage = match stage.finish_nonfailure_bounded(None, &axes) {
                Err(stage) => stage,
                Ok(_) => panic!("technical axis bypass"),
            };
            assert!(receiver.try_recv().is_err());
            let reference =
                crate::RequestDiagnosticScope::output_unavailable(call.context(), &event)
                    .capture_required(BoundedDiagnostic::capture(
                        DiagnosticCategory::UnexpectedError,
                        CaptureSite::FirstObserved,
                        BoundedDiagnosticCause::new(
                            DiagnosticStage::RequestDb,
                            DiagnosticCode::new("db.decode.type").unwrap(),
                        ),
                    ))
                    .into_reference();
            assert_eq!(
                stage
                    .finish_request_failure(None, &axes, &reference)
                    .boundary_submission(),
                DiagnosticSubmission::OutputUnavailable
            );
        }
        assert_eq!(
            observer
                .metrics_snapshot()
                .stage_latency(crate::Stage::Database)
                .iter()
                .sum::<u64>(),
            8
        );
        call.succeed();
        let _ = observer.shutdown();
    }
    #[test]
    fn bounded_diagnostic_admitted_no_escape_full_closed() {
        use saddle_core::*;
        use std::{
            future::Future,
            task::{Context, Poll, Waker},
        };
        let (sender, receiver) = mpsc::sync_channel(1);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let observer =
            crate::Observer::with_writer(crate::ObserverConfig::default(), io::sink()).unwrap();
        let (call, _) = observer
            .start_external_call_checked("app", "module", "service", "query", None)
            .unwrap();
        let stage = observer.start_stage(
            call.context(),
            crate::Stage::Database,
            crate::EventContext::new(
                crate::RequestIdentity::new("r1").unwrap(),
                crate::RouteIdentity::new("query").unwrap(),
                1,
            )
            .unwrap(),
        );
        let diagnostic_event = crate::EventContext::new(
            crate::RequestIdentity::new("r1").unwrap(),
            crate::RouteIdentity::new("/query").unwrap(),
            1,
        )
        .unwrap();
        let diagnostic_call = call.context().clone();
        let success_stage = observer.start_stage(
            call.context(),
            crate::Stage::Database,
            diagnostic_event.clone(),
        );
        let receipt_stage = observer.start_stage(
            call.context(),
            crate::Stage::Database,
            diagnostic_event.clone(),
        );
        let unavailable_stage = observer.start_stage(
            call.context(),
            crate::Stage::Database,
            diagnostic_event.clone(),
        );
        let ledger = saddle_admission::ProcessLedger::new(saddle_admission::ResourceConfig {
            managed_capacity: 1024 * 1024,
            entry_reserve: 1024 * 1024,
            framework_reserve: 1024 * 1024,
            task_reserve: 1024 * 1024,
            process_state_reserve: 1024 * 1024,
            system_estimate: 1024 * 1024,
            safety_margin: 1024 * 1024,
            process_limit: 8 * 1024 * 1024,
            max_active_requests: 1,
        })
        .unwrap();
        let future = ledger
            .try_envelope(65536, 65536, |_| async {
                assert!(matches!(
                    success_stage.finish_nonfailure_bounded(
                        None,
                        &DiagnosticOutcomeAxes {
                            operation: OperationOutcome::Succeeded,
                            ..Default::default()
                        }
                    ),
                    Ok(DiagnosticSubmission::OutputUnavailable)
                ));
                observer.observe_database_disposition(crate::DatabaseDisposition::Returned);
                let cause = BoundedDiagnosticCause::new(
                    DiagnosticStage::RequestDb,
                    DiagnosticCode::new("db.decode.type").unwrap(),
                )
                .with_column(Some(0), Some(3))
                .with_types(
                    Some(InlineDiagnosticText::metadata("Option<i64>")),
                    Some(InlineDiagnosticText::metadata("BIGINT")),
                    None,
                );
                let primary = BoundedDiagnostic::capture(
                    DiagnosticCategory::UnexpectedError,
                    CaptureSite::FirstObserved,
                    cause,
                );
                let axes = DiagnosticOutcomeAxes {
                    operation: OperationOutcome::Failed,
                    physical: PhysicalDispositionFact::Returned,
                    ..Default::default()
                };
                assert_eq!(
                    handle.submit_bounded(Some(&primary), &axes, None),
                    DiagnosticSubmission::Enqueued
                );
                assert_eq!(
                    handle.submit_bounded(Some(&primary), &axes, None),
                    DiagnosticSubmission::Full
                );
                assert_eq!(
                    stage.finish_bounded(&handle, &axes, Some(&primary)),
                    DiagnosticSubmission::Full
                );
                let controlled = crate::RequestDiagnosticScope::established(
                    &handle,
                    &diagnostic_call,
                    &diagnostic_event,
                )
                .with_zone(crate::DiagnosticZone::from_validated_ingress("zone-a").unwrap())
                .with_db_operation(
                    crate::DiagnosticDbOperation::from_registered("AgentByCode").unwrap(),
                );
                let failure = controlled.fail(17u32, DiagnosticCategory::UnexpectedError, cause);
                assert_eq!(failure.submission(), DiagnosticSubmission::Full);
                let (original, delivery) = failure.finish_boundary(&handle, &axes);
                assert_eq!(original, 17);
                assert_eq!(delivery.boundary_submission(), DiagnosticSubmission::Full);
                let reference = controlled.capture_required(primary).into_reference();
                let delivery =
                    receipt_stage.finish_request_failure(Some(&handle), &axes, &reference);
                assert_eq!(delivery.source_submission(), DiagnosticSubmission::Full);
                assert_eq!(delivery.boundary_submission(), DiagnosticSubmission::Full);
                let delivery = unavailable_stage.finish_request_failure(None, &axes, &reference);
                assert_eq!(
                    delivery.boundary_submission(),
                    DiagnosticSubmission::OutputUnavailable
                );
                handle.status.closed.store(true, Ordering::Release);
                assert_eq!(
                    handle.submit_bounded(Some(reference.source_diagnostic()), &axes, None),
                    DiagnosticSubmission::Closed
                );
                reference.source_diagnostic().id()
            })
            .unwrap();
        let mut future = std::pin::pin!(future);
        let result = future
            .as_mut()
            .poll(&mut Context::from_waker(Waker::noop()));
        assert!(
            matches!(result, Poll::Ready(Ok(_))),
            "admitted capture/submit must not escape: {result:?}"
        );
        let packet = receiver.try_recv().unwrap();
        let row: serde_json::Value = serde_json::from_slice(packet.bytes()).unwrap();
        assert_eq!(row["axes"]["operation"], "failed");
        assert_eq!(row["axes"]["physical"], "returned");
        assert_eq!(handle.snapshot().dropped, 7);
        assert_eq!(
            observer
                .metrics_snapshot()
                .database(crate::DatabaseDisposition::Returned),
            1
        );
        assert_eq!(
            observer
                .metrics_snapshot()
                .database(crate::DatabaseDisposition::Discarded),
            0
        );
        assert_eq!(
            observer
                .metrics_snapshot()
                .database(crate::DatabaseDisposition::NotUsed),
            0
        );
        assert_eq!(
            observer
                .metrics_snapshot()
                .stage_latency(crate::Stage::Database)
                .iter()
                .sum::<u64>(),
            4
        );
        call.succeed();
        let _ = observer.shutdown();
    }
    #[test]
    fn bounded_disconnected_and_oversized_context_are_explicit() {
        let (sender, receiver) = mpsc::sync_channel(1);
        drop(receiver);
        let handle = EmergencyDiagnosticHandle {
            sender,
            status: Arc::new(Status::default()),
        };
        let axes = saddle_core::DiagnosticOutcomeAxes::default();
        assert_eq!(
            handle.submit_bounded(None, &axes, None),
            DiagnosticSubmission::Closed
        );
        let observer =
            crate::Observer::with_writer(crate::ObserverConfig::default(), io::sink()).unwrap();
        let (call, _) = observer
            .start_external_call_checked("app", "module", "service", "x".repeat(257), None)
            .unwrap();
        assert_eq!(
            handle.submit_bounded_parts(
                None,
                &axes,
                Some((call.context(), "request", "route")),
                None
            ),
            DiagnosticSubmission::EncodingFailed
        );
        assert_eq!(handle.snapshot().written, 0);
        assert_eq!(handle.snapshot().dropped, 2);
        call.succeed();
        let _ = observer.shutdown();
    }
    #[test]
    fn diagnostic_protocol_rpc_is_not_internal_span() {
        let observer =
            crate::Observer::with_writer(crate::ObserverConfig::default(), io::sink()).unwrap();
        let (call, _) = observer
            .start_external_call_with_rpc(
                "test",
                "test",
                "test",
                "query",
                Some("opaque-trace"),
                saddle_core::RpcCorrelationId::new("0.7").unwrap(),
            )
            .unwrap();
        let event = crate::EventContext::new(
            crate::RequestIdentity::new("request-7").unwrap(),
            crate::RouteIdentity::new("bill.query").unwrap(),
            1,
        )
        .unwrap();
        let diagnostic = sample();
        let row = serde_json::to_value(Envelope::new(&diagnostic, Some((call.context(), &event))))
            .unwrap();
        assert_eq!(row["rpc_id"], "0.7");
        assert_eq!(row["span_id"], call.context().span_id().to_string());
        assert_ne!(row["rpc_id"], row["span_id"]);
        assert_eq!(row["trace_id"], "opaque-trace");
        assert_eq!(row["operation"], "query");
        assert_eq!(row["request"], "request-7");
        assert!(!row.to_string().contains("DO_NOT_LOG_SECRET"));
        let missing = serde_json::to_value(Envelope::new(&diagnostic, None)).unwrap();
        for key in [
            "trace_id",
            "rpc_id",
            "span_id",
            "operation",
            "request",
            "route",
        ] {
            assert!(missing[key].is_null());
        }
        call.succeed();
        let _ = observer.shutdown();
    }
    fn sample() -> Diagnostic {
        Diagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            DiagnosticCause::new(
                DiagnosticStage::RequestDb,
                DiagnosticCode::new("db.connection_failed").unwrap(),
            )
            .with_io(&io::Error::other("DO_NOT_LOG_SECRET"))
            .with_driver_details(saddle_core::DiagnosticDriverDetails::new(
                Some(0),
                Some(4),
                saddle_core::DiagnosticTypeName::from_metadata("core::option::Option<i64>"),
                saddle_core::DiagnosticTypeName::from_metadata("BIGINT"),
            ))
            .with_input_location(
                saddle_core::DiagnosticInputLocation::new(Some(17), Some(3))
                    .with_file(
                        saddle_core::DiagnosticLocator::from_projection(
                            "mapping.json",
                            false,
                            false,
                        )
                        .unwrap(),
                    )
                    .with_column(
                        saddle_core::DiagnosticLocator::from_projection(
                            "DO_NOT_LOG_SECRET",
                            true,
                            true,
                        )
                        .unwrap(),
                    ),
            ),
        )
    }
    struct Failed;
    impl Write for Failed {
        fn write(&mut self, _: &[u8]) -> io::Result<usize> {
            Err(io::Error::from_raw_os_error(28))
        }
        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }
    #[test]
    fn diagnostic_independent_file_after_main_failure() {
        let _serial = FILE_TEST.lock().unwrap();
        let directory = std::env::temp_dir().join(format!(
            "saddle-emergency-{}-{}",
            std::process::id(),
            sample().id()
        ));
        let config = FileLoggingConfig::new(&directory, crate::Rotation::Daily);
        let mut emergency = EmergencyDiagnostics::start(&config).unwrap();
        struct MarkFailure(Arc<AtomicU64>);
        impl Write for MarkFailure {
            fn write(&mut self, _: &[u8]) -> io::Result<usize> {
                self.0.fetch_add(1, Ordering::Release);
                Err(io::Error::from_raw_os_error(28))
            }
            fn flush(&mut self) -> io::Result<()> {
                Ok(())
            }
        }
        let main_failed = Arc::new(AtomicU64::new(0));
        let observer = crate::Observer::with_writer(
            crate::ObserverConfig::default(),
            MarkFailure(main_failed.clone()),
        )
        .unwrap();
        let diagnostic = sample();
        let (call, _) = observer
            .start_external_call_checked("test", "test", "test", "operation", Some("safe-trace"))
            .unwrap();
        let event = crate::EventContext::new(
            crate::RequestIdentity::new("request-1").unwrap(),
            crate::RouteIdentity::new("route.test").unwrap(),
            1,
        )
        .unwrap();
        let end = Instant::now() + Duration::from_secs(3);
        while main_failed.load(Ordering::Acquire) == 0 && Instant::now() < end {
            thread::sleep(Duration::from_millis(1));
        }
        assert!(
            main_failed.load(Ordering::Acquire) > 0,
            "primary writer failed before diagnostic submission"
        );
        assert_eq!(
            observer.record_diagnostic(
                &diagnostic,
                &emergency.handle(),
                Some((call.context(), &event))
            ),
            DiagnosticSubmission::Enqueued
        );
        let end = Instant::now() + Duration::from_secs(3);
        while emergency.snapshot().written != 1 && Instant::now() < end {
            thread::sleep(Duration::from_millis(1));
        }
        assert_eq!(emergency.snapshot().written, 1);
        let text = fs::read_to_string(emergency.target()).unwrap();
        let row: serde_json::Value = serde_json::from_str(text.trim()).unwrap();
        assert_eq!(row["trace_id"], "safe-trace");
        assert_eq!(row["request"], "request-1");
        assert!(row["rpc_id"].is_null());
        assert_eq!(row["span_id"], call.context().span_id().to_string());
        assert_eq!(row["operation"], "operation");
        assert_eq!(row["diagnostic"]["diagnostic_id"], diagnostic.id());
        let driver = &row["diagnostic"]["causes"][0]["driver_details"];
        assert_eq!(driver["column_index"], 0);
        assert_eq!(driver["column_count"], 4);
        assert_eq!(
            driver["target_rust_type"]["value"],
            "core::option::Option<i64>"
        );
        assert_eq!(driver["actual_db_type"]["value"], "BIGINT");
        let input = &row["diagnostic"]["causes"][0]["input_location"];
        assert_eq!(input["json_line"], 17);
        assert_eq!(input["json_column"], 3);
        assert_eq!(input["file"]["value"], "mapping.json");
        assert_eq!(input["column"]["value"], "[redacted]");
        assert_eq!(input["locator_truncated"], true);
        assert_eq!(input["locator_redacted"], true);
        assert!(!text.contains("DO_NOT_LOG_SECRET"));
        let bounded = saddle_core::BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            saddle_core::BoundedDiagnosticCause::new(
                DiagnosticStage::RequestDb,
                DiagnosticCode::new("db.decode.failed").unwrap(),
            ),
        );
        let axes = saddle_core::DiagnosticOutcomeAxes {
            operation: saddle_core::OperationOutcome::Failed,
            physical: saddle_core::PhysicalDispositionFact::Returned,
            business: saddle_core::BusinessOutcome::Failure,
            business_code: saddle_core::RegisteredDiagnosticCode::from_registered(
                "INVALID_ROW",
                &["INVALID_ROW"],
            ),
            delivery: saddle_core::ResponseDelivery::LocalWriteComplete,
            cleanup: saddle_core::CleanupOutcome::Succeeded,
            bytes_written: Some(100),
        };
        assert_eq!(
            emergency.handle().submit_bounded(
                Some(&bounded),
                &axes,
                Some((call.context(), &event))
            ),
            DiagnosticSubmission::Enqueued
        );
        let until = Instant::now() + Duration::from_secs(3);
        while emergency.snapshot().written != 2 && Instant::now() < until {
            thread::sleep(Duration::from_millis(1));
        }
        assert_eq!(emergency.snapshot().written, 2);
        let content = fs::read_to_string(emergency.target()).unwrap();
        let second: serde_json::Value =
            serde_json::from_str(content.lines().nth(1).unwrap()).unwrap();
        assert_eq!(second["axes"]["operation"], "failed");
        assert_eq!(second["axes"]["physical"], "returned");
        assert_eq!(second["axes"]["business_code"], "INVALID_ROW");
        assert_eq!(second["axes"]["delivery"], "local_write_complete");
        assert_eq!(second["axes"]["cleanup"], "succeeded");
        assert_eq!(second["trace_id"], "safe-trace");
        let end = Instant::now() + Duration::from_secs(3);
        while emergency.shutdown() == DiagnosticShutdown::Pending && Instant::now() < end {
            thread::sleep(Duration::from_millis(1));
        }
        assert_eq!(emergency.shutdown(), DiagnosticShutdown::Finished);
        assert_eq!(
            emergency.handle().submit(&diagnostic),
            DiagnosticSubmission::Closed
        );
        call.succeed();
        let _shutdown = observer.shutdown();
        fs::remove_file(emergency.target()).unwrap();
        fs::remove_dir(directory).unwrap();
    }
    #[test]
    fn diagnostic_directory_failure_has_real_os_cause() {
        let _serial = FILE_TEST.lock().unwrap();
        let path = std::env::temp_dir().join(format!(
            "saddle-emergency-file-{}-{}",
            std::process::id(),
            sample().id()
        ));
        fs::write(&path, b"not a directory").unwrap();
        let mut owner =
            EmergencyDiagnostics::start(&FileLoggingConfig::new(&path, crate::Rotation::Daily))
                .unwrap();
        let end = Instant::now() + Duration::from_secs(3);
        while !owner.snapshot().initialized && Instant::now() < end {
            thread::sleep(Duration::from_millis(1));
        }
        let failure = owner.snapshot().first_failure.unwrap();
        assert_eq!(failure.stage, "initialize");
        assert!(failure.os_code.is_some());
        assert_eq!(owner.target(), path.join(EMERGENCY_FILE_NAME));
        while owner.shutdown() == DiagnosticShutdown::Pending && Instant::now() < end {
            thread::sleep(Duration::from_millis(1));
        }
        assert_eq!(owner.shutdown(), DiagnosticShutdown::Finished);
        fs::remove_file(path).unwrap();
    }
    #[test]
    fn diagnostic_stuck_worker_shutdown_never_joins_early() {
        let (sender, _receiver) = mpsc::sync_channel(1);
        let (release, wait) = mpsc::channel();
        let (started, reached) = mpsc::channel();
        let worker = thread::spawn(move || {
            started.send(()).unwrap();
            wait.recv().unwrap();
        });
        reached.recv_timeout(Duration::from_secs(2)).unwrap();
        let mut owner = EmergencyDiagnostics {
            handle: EmergencyDiagnosticHandle {
                sender,
                status: Arc::new(Status::default()),
            },
            worker: Some(worker),
            target: PathBuf::from("unused"),
            source_target_preopened: false,
        };
        assert_eq!(owner.shutdown(), DiagnosticShutdown::Pending);
        release.send(()).unwrap();
        let end = Instant::now() + Duration::from_secs(3);
        while owner.shutdown() == DiagnosticShutdown::Pending && Instant::now() < end {
            thread::yield_now();
        }
        assert_eq!(owner.shutdown(), DiagnosticShutdown::Finished);
    }
    #[test]
    fn diagnostic_double_failure_is_counted_without_recursion() {
        let status = Status::default();
        let mut output: io::Result<Failed> = Err(io::Error::from_raw_os_error(13));
        let mut stderr = Vec::new();
        let mut attempted = false;
        deliver(
            &mut output,
            &mut stderr,
            std::path::Path::new("/safe/logs/saddle.emergency.log"),
            b"{\"original\":true}\n",
            &status,
            &mut attempted,
        );
        let text = String::from_utf8(stderr).unwrap();
        assert!(text.contains("\"os_code\":13") && text.contains("\"original\":true"));
        attempted = false;
        deliver(
            &mut output,
            &mut Failed,
            std::path::Path::new("./logs"),
            b"{}\n",
            &status,
            &mut attempted,
        );
        for _ in 0..10 {
            deliver(
                &mut output,
                &mut Failed,
                std::path::Path::new("./logs"),
                b"{}\n",
                &status,
                &mut attempted,
            );
        }
        assert_eq!(status.snapshot().stderr_failed, 1);
        assert_eq!(status.snapshot().stderr_suppressed, 10);
        assert_eq!(status.snapshot().output_failed, 12);
    }
    #[test]
    fn ingress_deadline_queue_and_writer_failure_are_not_success_evidence() {
        let (sender, receiver) = mpsc::sync_channel(1);
        let status = Arc::new(Status::default());
        let handle = EmergencyDiagnosticHandle { sender, status: status.clone() };
        let observer = crate::Observer::with_writer(Default::default(), io::sink()).unwrap();
        let (call, _) = observer.start_external_call_with_rpc("app", "profusegw", "profusegw", "route",
            Some("trace"), saddle_core::RpcCorrelationId::new("0").unwrap()).unwrap();
        let event = crate::EventContext::new(crate::RequestIdentity::new("request-a").unwrap(),
            crate::RouteIdentity::new("route").unwrap(), 1).unwrap();
        let scope = crate::RequestDiagnosticScope::established(&handle, call.context(), &event);
        assert_eq!(scope.ingress_deadline_bound(12345), DiagnosticSubmission::Enqueued);
        assert_eq!(scope.ingress_deadline_bound(12345), DiagnosticSubmission::Full);
        let packet = receiver.try_recv().unwrap();
        let mut output = Ok(Failed);
        let mut attempted = false;
        deliver(&mut output, &mut Failed, std::path::Path::new("./logs"),
            packet.bytes(), &status, &mut attempted);
        assert_eq!(status.snapshot().written, 0);
        assert_eq!(status.snapshot().output_failed, 1);
        assert!(status.snapshot().first_failure.is_some());
        assert_eq!(handle.snapshot().dropped, 1);
    }

    #[test]
    fn diagnostic_full_queue_submission_does_not_wait() {
        let (sender, _receiver) = mpsc::sync_channel(1);
        let status = Arc::new(Status::default());
        let handle = EmergencyDiagnosticHandle { sender, status };
        let sample = sample();
        assert_eq!(handle.submit(&sample), DiagnosticSubmission::Enqueued);
        assert_eq!(handle.submit(&sample), DiagnosticSubmission::Full);
        assert_eq!(handle.snapshot().dropped, 1);
        assert_eq!(handle.snapshot().written, 0);
    }
}

#[cfg(test)]
mod rpc_stage_frame_tests {
    use super::*;
    #[test]
    fn rpc_stage_native_frame_keeps_actual_call_and_original_slot_identity_with_loss() {
        use saddle_core::request_context::{ContextFact, ContextLabel, RequestLocalFacts,
            RequestRootPublisher, RequestViewPhase};
        let publisher = RequestRootPublisher::create(ContextLabel::checked("rpc-stage").unwrap(),
            ContextFact::NotEstablished).unwrap();
        let view = publisher.reference().view(RequestLocalFacts::new(RequestViewPhase::Reading));
        let (sender, receiver) = mpsc::sync_channel(1);
        let status = Arc::new(Status::default());
        let mut call_id = [0; 256];
        call_id[..12].copy_from_slice(b"actual-child");
        let observer = DetachedRpcStageRelease { context: view.diagnostic_snapshot(),
            output: EmergencyDiagnosticHandle {sender, status: status.clone()},
            process_instance: 77, call_id, call_id_len: 12 };
        let fact = saddle_admission::RpcStageReleaseFact { ledger_instance: 23,
            account: saddle_admission::AccountIdentity {account_slot: 0, account_generation: 7, audit_generation: 11},
            stage_slot: 3, stage_generation: 9 };
        saddle_admission::RpcStageReleaseObserver::released(&observer, fact);
        saddle_admission::RpcStageReleaseObserver::released(&observer, fact);
        assert_eq!(status.dropped.load(Ordering::Relaxed), 1);
        let DiagnosticPacket::Admitted(bytes) = receiver.recv().unwrap() else { panic!("native frame") };
        let row: serde_json::Value = serde_json::from_slice(&bytes.bytes[..bytes.len]).unwrap();
        assert_eq!(row["schema"], "request-rpc-stage/1");
        assert_eq!(row["event"], "rpc_stage_credit_released");
        assert_eq!(row["call_id"], "actual-child");
        assert_eq!(row["stage"]["stage_generation"], 9);
        assert_eq!(row["stage"]["account"]["audit_generation"], 11);
        assert_eq!(row["local_request"], row["context"]["local_request"]);
        assert!(row.get("physical_driver_destroyed").is_none());
        assert_eq!(status.written.load(Ordering::Relaxed), 0);
    }
}

#[cfg(test)]
mod request_stop_decision_tests {
    use super::*;
    #[test]
    fn request_stop_decision_frame_preserves_actual_branch_domain_and_rpc_clock() {
        use saddle_core::request_context::{ContextFact, ContextLabel, RequestLocalFacts,
            RequestRootPublisher, RequestViewPhase};
        let publisher = RequestRootPublisher::create(ContextLabel::checked("stop-decision").unwrap(),
            ContextFact::NotEstablished).unwrap();
        let view = publisher.reference().view(RequestLocalFacts::new(RequestViewPhase::Reading));
        let (sender, receiver) = mpsc::sync_channel(2);
        let status = Arc::new(Status::default());
        let output = EmergencyDiagnosticHandle {sender, status: status.clone()};
        output.record_request_stop_decision(view.diagnostic_snapshot(), "peer_disconnected");
        output.record_request_stop_decision(view.diagnostic_snapshot(), "shutdown-guessed");
        output.record_request_stop_decision(view.diagnostic_snapshot(), "original_deadline");
        let mut rows = Vec::new();
        for _ in 0..2 {
            let DiagnosticPacket::Admitted(bytes) = receiver.recv().unwrap() else { panic!("fixed native frame") };
            rows.push(serde_json::from_slice::<serde_json::Value>(&bytes.bytes[..bytes.len]).unwrap());
        }
        assert_eq!(rows[0]["cause"], "peer_disconnected");
        assert_eq!(rows[1]["cause"], "original_deadline");
        assert_eq!(rows[0]["decision_point"], "handler_select_before_dispatch_drop");
        assert_eq!(rows[0]["schema"], "request-stop/1");
        assert_eq!(rows[0]["local_request"], rows[0]["context"]["local_request"]);
        assert!(rows[1]["sequence"].as_u64().unwrap() > rows[0]["sequence"].as_u64().unwrap());
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        assert!(sequence > rows[1]["sequence"].as_u64().unwrap());
        assert!(monotonic_ns >= rows[1]["monotonic_ns"].as_u64().unwrap() as u128);
        assert!(receiver.try_recv().is_err());
        assert_eq!(status.dropped.load(Ordering::Relaxed), 0);
    }
}

#[cfg(test)]
mod lifecycle_boundary_frame_tests {
    use super::*;
    #[test]
    fn lifecycle_boundary_actual_fixed_frames_share_clock_and_preserve_queue_loss() {
        let (sender, receiver) = mpsc::sync_channel(3);
        let status = Arc::new(Status::default());
        let output = EmergencyDiagnosticHandle { sender, status: status.clone() };
        for (event, phase) in [("shutdown_future_completed", 1), ("claim_cutoff", 2), ("claim_published", 2)] {
            output.record_request_lifecycle("boundary-app", event, phase, 1);
        }
        assert_eq!(status.dropped.load(Ordering::Relaxed), 0);
        output.record_request_lifecycle("boundary-app", "claim_published", 2, 1);
        assert_eq!(status.dropped.load(Ordering::Relaxed), 1);
        let mut previous = 0;
        for event in ["shutdown_future_completed", "claim_cutoff", "claim_published"] {
            let DiagnosticPacket::Admitted(bytes) = receiver.recv().unwrap() else { panic!("fixed native frame") };
            let row: serde_json::Value = serde_json::from_slice(&bytes.bytes[..bytes.len]).unwrap();
            assert_eq!(row["schema"], "request-lifecycle-boundary/1");
            assert_eq!(row["application"], "boundary-app");
            assert_eq!(row["event"], event);
            assert!(row["sequence"].as_u64().unwrap() > previous);
            previous = row["sequence"].as_u64().unwrap();
            assert_eq!(row["process_instance"], serde_json::json!(finalization_process_instance()));
            assert!(row.get("request_id").is_none());
            assert!(row.get("signal_arrival").is_none());
        }
        let (sequence, _) = rpc_lifecycle_stamp();
        assert!(sequence > previous);
        assert!(receiver.try_recv().is_err());
    }
}

#[cfg(test)]
mod read_owner_frame_tests {
    use super::*;
    #[test]
    fn read_owner_frame_binds_unpublished_task_to_actual_scalar_peer() {
        use saddle_core::request_context::{ContextFact,ContextLabel,RequestLocalFacts,RequestRootPublisher,RequestViewPhase};
        let publisher=RequestRootPublisher::create(ContextLabel::checked("read-owner").unwrap(),ContextFact::NotEstablished).unwrap();
        let view=publisher.reference().view(RequestLocalFacts::new(RequestViewPhase::Reading).with_task(ContextFact::Present(9)));
        let (sender,receiver)=mpsc::sync_channel(1);
        let status=Arc::new(Status::default());
        let output=EmergencyDiagnosticHandle{sender,status:status.clone()};
        output.record_request_read_owner(view.diagnostic_snapshot(),3,
            saddle_admission::AccountIdentity{account_slot:0,account_generation:1,audit_generation:2},
            std::net::SocketAddrV4::new(std::net::Ipv4Addr::LOCALHOST,23456));
        let DiagnosticPacket::Admitted(bytes)=receiver.recv().unwrap() else {panic!("native fixed frame")};
        let row:serde_json::Value=serde_json::from_slice(&bytes.bytes[..bytes.len]).unwrap();
        assert_eq!(row["schema"],"request-read-owner/1");
        assert_eq!(row["event"],"recognition_read_head_entered");
        assert_eq!(row["context"]["publication"],0);
        assert_eq!(row["context"]["task"],serde_json::json!({"state":"present","value":9}));
        assert_eq!(row["peer_address"],serde_json::json!([127,0,0,1]));
        assert_eq!(row["peer_port"],23456);
        assert_eq!(row["local_request"],row["context"]["local_request"]);
        assert!(row.get("task_reclaimed").is_none());
        assert_eq!(status.dropped.load(Ordering::Relaxed),0);
    }
}

#[cfg(test)]
mod task_owner_frame_tests {
    use super::*;
    #[test]
    fn owner_binding_is_scalar_creation_not_reclamation() {
        use saddle_core::request_context::{ContextFact,ContextLabel,RequestLocalFacts,RequestRootPublisher,RequestViewPhase};
        let publisher=RequestRootPublisher::create(ContextLabel::checked("task-owner").unwrap(),ContextFact::NotEstablished).unwrap();
        let view=publisher.reference().view(RequestLocalFacts::new(RequestViewPhase::Reading).with_task(ContextFact::Present(9)));
        let (sender,receiver)=mpsc::sync_channel(1);
        let status=Arc::new(Status::default());
        let output=EmergencyDiagnosticHandle{sender,status:status.clone()};
        output.record_request_task_owner(view.diagnostic_snapshot(),3,
            saddle_admission::AccountIdentity{account_slot:0,account_generation:1,audit_generation:2}, "business");
        let DiagnosticPacket::Admitted(bytes)=receiver.recv().unwrap() else {panic!("native frame")};
        let row:serde_json::Value=serde_json::from_slice(&bytes.bytes[..bytes.len]).unwrap();
        assert_eq!(row["schema"],"request-task-roster/2");
        assert_eq!(row["task_role"],"business");
        assert_eq!(row["event"],"task_owner_bound");
        assert_eq!(row["context"]["publication"],0);
        assert_eq!(row["context"]["task"],serde_json::json!({"state":"present","value":9}));
        assert_eq!(row["ledger_instance"],3);
        assert!(row.get("completion").is_none());
        assert_eq!(status.dropped.load(Ordering::Relaxed),0);
        drop(receiver);
        output.record_request_task_owner(view.diagnostic_snapshot(),3,
            saddle_admission::AccountIdentity{account_slot:0,account_generation:1,audit_generation:2}, "recognition");
        assert_eq!(status.dropped.load(Ordering::Relaxed),1);
    }
}

#[cfg(test)]
mod database_admission_frame_tests {
    use super::*;
    #[test]
    fn actual_database_attempt_records_typed_decision_without_claiming_release() {
        use saddle_core::request_context::{ContextFact,ContextLabel,RequestLocalFacts,RequestRootPublisher,RequestViewPhase};
        let publisher=RequestRootPublisher::create(ContextLabel::checked("database-admission").unwrap(),ContextFact::NotEstablished).unwrap();
        let view=publisher.reference().view(RequestLocalFacts::new(RequestViewPhase::Database).with_task(ContextFact::Present(9)));
        let (sender,receiver)=mpsc::sync_channel(2);
        let status=Arc::new(Status::default());
        let output=EmergencyDiagnosticHandle{sender,status:status.clone()};
        let account=saddle_admission::AccountIdentity{account_slot:0,account_generation:1,audit_generation:2};
        output.record_request_database_admission(view.diagnostic_snapshot(),3,account,true,"accepted");
        output.record_request_database_admission(view.diagnostic_snapshot(),3,account,false,"db_credits_exhausted");
        let mut previous=0;
        for accepted in [true,false] {
            let DiagnosticPacket::Admitted(bytes)=receiver.recv().unwrap() else {panic!("native frame")};
            let row:serde_json::Value=serde_json::from_slice(&bytes.bytes[..bytes.len]).unwrap();
            assert_eq!(row["schema"],"request-database-admission/1");
            assert_eq!(row["accepted"],accepted);
            assert_eq!(row["reason"],if accepted {"accepted"} else {"db_credits_exhausted"});
            assert_eq!(row["ledger_instance"],3);
            let sequence=row["sequence"].as_u64().unwrap();assert!(sequence>previous);previous=sequence;
            assert!(row.get("physical_release").is_none());
        }
        assert_eq!(status.dropped.load(Ordering::Relaxed),0);
    }
}

#[cfg(test)]
mod fifo_clock_frame_tests {
    use super::*;
    #[test]
    fn actual_fixed_frame_binds_effective_deadline_and_recording_point_only() {
        let (sender,receiver)=mpsc::sync_channel(1);
        let status=Arc::new(Status::default());
        let output=EmergencyDiagnosticHandle{sender,status:status.clone()};
        let before=std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos();
        output.record_ingress_fifo_clock(Some(saddle_core::request_context::ContextIdentity::checked("queued-R1").unwrap()),7,"removed","deadline",12345);
        let after=std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos();
        let DiagnosticPacket::Admitted(bytes)=receiver.recv().unwrap() else {panic!("native frame")};
        let row:serde_json::Value=serde_json::from_slice(&bytes.bytes[..bytes.len]).unwrap();
        assert_eq!(row["schema"],"ingress-fifo-clock/1");
        assert_eq!(row["request_id"],"queued-R1");
        assert_eq!(row["queue_sequence"],7);
        assert_eq!(row["effective_deadline_unix_ms"],12345);
        let recorded=row["recorded_unix_ns"].as_u64().unwrap() as u128;
        assert!(before<=recorded && recorded<=after);
        assert!(row.get("physical_removal_unix_ns").is_none());
        assert_eq!(status.dropped.load(Ordering::Relaxed),0);
    }
}

#[cfg(test)]
mod signal_receipt_frame_tests {
    use super::*;
    #[test]
    fn signal_receipt_actual_frames_preserve_kind_closed_stream_and_loss() {
        let (sender, receiver) = mpsc::sync_channel(3);
        let status = Arc::new(Status::default());
        let output = EmergencyDiagnosticHandle { sender, status: status.clone() };
        let app = saddle_core::ContextLabel::checked("signal-app").unwrap();
        for (kind, received) in [("SIGTERM", true), ("SIGINT", true), ("SIGTERM", false)] {
            output.record_shutdown_signal_received(app, kind, received);
        }
        output.record_shutdown_signal_received(app, "SIGINT", false);
        assert_eq!(status.dropped.load(Ordering::Relaxed), 1);
        let mut previous = 0;
        for (kind, received) in [("SIGTERM", true), ("SIGINT", true), ("SIGTERM", false)] {
            let DiagnosticPacket::Admitted(bytes) = receiver.recv().unwrap() else { panic!("fixed frame") };
            let row: serde_json::Value = serde_json::from_slice(&bytes.bytes[..bytes.len]).unwrap();
            assert_eq!(row["schema"], "process-signal-receipt/1");
            assert_eq!(row["application"], "signal-app");
            assert_eq!(row["signal"], kind);
            assert_eq!(row["received"], received);
            assert_eq!(row["process_instance"], serde_json::json!(finalization_process_instance()));
            assert!(row["sequence"].as_u64().unwrap() > previous);
            previous = row["sequence"].as_u64().unwrap();
            assert!(row.get("kernel_arrival").is_none());
        }
        assert!(receiver.try_recv().is_err());
    }
}

struct DetachedHandlerPoll {
    context: saddle_core::request_context::RequestContextSnapshot,
    output: EmergencyDiagnosticHandle,
    process_instance: u128,
}
impl saddle_admission::HandlerPollObserver for DetachedHandlerPoll {
    fn before_first_poll(&self) {
        #[derive(Serialize)]
        struct Record<'a> {
            schema: &'static str, event: &'static str, pid: u32, process_instance: u128,
            sequence: u64, monotonic_ns: u128,
            context: &'a saddle_core::request_context::RequestContextSnapshot,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        let _ = self.output.submit_fixed_record(&Record {
            schema: "request-handler-poll/1", event: "before_first_typed_handler_poll",
            pid: std::process::id(), process_instance: self.process_instance,
            sequence, monotonic_ns, context: &self.context,
        });
    }
}
impl EmergencyDiagnosticHandle {
    #[doc(hidden)]
    pub fn prepare_handler_poll(&self, context: saddle_core::request_context::RequestContextSnapshot,
        storage: &saddle_admission::ProcessLogStorage) -> Option<saddle_admission::HandlerPollObservation> {
        saddle_admission::HandlerPollObservation::prepare(storage, DetachedHandlerPoll {
            context, output: self.clone(), process_instance: finalization_process_instance(),
        }).ok()
    }
}

#[cfg(test)]
mod handler_poll_frame_tests {
    use super::*;
    use saddle_admission::HandlerPollObserver;
    #[test]
    fn handler_poll_actual_detached_frame_has_native_context_and_bounded_loss() {
        use saddle_core::request_context::{ContextLabel,ContextFact,RequestRootPublisher,RequestLocalFacts,RequestViewPhase};
        let root=RequestRootPublisher::create(ContextLabel::checked("handler-frame-app").unwrap(),ContextFact::NotEstablished).unwrap();
        let view=root.reference().view(RequestLocalFacts::new(RequestViewPhase::Handler).with_task(ContextFact::Present(9)));
        let (sender,receiver)=mpsc::sync_channel(1);
        let status=Arc::new(Status::default());
        let observer=DetachedHandlerPoll{context:view.diagnostic_snapshot(),
            output:EmergencyDiagnosticHandle{sender,status:status.clone()},process_instance:finalization_process_instance()};
        observer.before_first_poll();
        observer.before_first_poll(); // Queue behavior, not the wrapper's once contract.
        assert_eq!(status.dropped.load(Ordering::Relaxed),1);
        let DiagnosticPacket::Admitted(bytes)=receiver.recv().unwrap() else {panic!("fixed frame")};
        let row:serde_json::Value=serde_json::from_slice(&bytes.bytes[..bytes.len]).unwrap();
        assert_eq!(row["schema"],"request-handler-poll/1");
        assert_eq!(row["event"],"before_first_typed_handler_poll");
        assert_eq!(row["context"]["application"]["value"],"handler-frame-app");
        assert_eq!(row["context"]["task"]["value"],9);
        assert_eq!(row["context"]["publication"],0);
        assert_eq!(row["process_instance"],serde_json::json!(finalization_process_instance()));
        assert!(row["sequence"].as_u64().unwrap()>0);
        assert!(row.get("handler_success").is_none());
    }
}

/// Detached SQL checkout observation; never retains request/account authority.
#[doc(hidden)]
pub struct DbOwnerObservation {
    state: saddle_admission::DbOwnerObservationStorage<DbOwnerObservationState>,
}
struct DbOwnerObservationState {
    context: saddle_core::request_context::RequestContextSnapshot,
    output: EmergencyDiagnosticHandle,
    account: saddle_admission::AccountIdentity,
    ledger_instance: u64,
    owner_id: u64,
    finished: std::sync::atomic::AtomicBool,
    acquired: std::sync::atomic::AtomicBool,
    acknowledged: std::sync::atomic::AtomicBool,
    handed_off: std::sync::atomic::AtomicBool,
    unavailable: std::sync::atomic::AtomicBool,
}
#[doc(hidden)]
#[derive(Clone, Copy)]
pub enum DbOwnerDisposition { Returned, Discarded }
impl DbOwnerObservation {
    fn emit(&self, event: &'static str) {
        let state = self.state.get();
        if state.unavailable.load(Ordering::Acquire) { return; }
        #[derive(Serialize)]
        struct Record<'a> {
            schema: &'static str, pid: u32, process_instance: u128,
            local_request: u64, ledger_instance: u64,
            account: saddle_admission::AccountIdentity,
            context: &'a saddle_core::request_context::RequestContextSnapshot,
            owner_id: u64, event: &'static str, sequence: u64, monotonic_ns: u128,
        }
        let (sequence, monotonic_ns) = rpc_lifecycle_stamp();
        if sequence == 0 { state.unavailable.store(true, Ordering::Release); return; }
        let result = state.output.submit_fixed_record(&Record {
            schema: "request-db-owner/1", pid: std::process::id(),
            process_instance: finalization_process_instance(),
            local_request: state.context.local_request_id(), ledger_instance: state.ledger_instance,
            account: state.account, context: &state.context, owner_id: state.owner_id,
            event, sequence, monotonic_ns,
        });
        if result != DiagnosticSubmission::Enqueued { state.unavailable.store(true, Ordering::Release); }
    }
    pub fn acquired(&self) {
        let s=self.state.get();
        if !s.finished.load(Ordering::Acquire) && !s.acquired.swap(true, Ordering::AcqRel) { self.emit("acquired"); }
    }
    pub fn handoff(&self) {
        let s=self.state.get();
        if s.acquired.load(Ordering::Acquire) && !s.finished.load(Ordering::Acquire) && !s.handed_off.swap(true, Ordering::AcqRel) { self.emit("finalization_handoff"); }
    }
    pub fn commit_ack(&self) {
        let s=self.state.get();
        if s.acquired.load(Ordering::Acquire) && !s.finished.load(Ordering::Acquire) && !s.acknowledged.swap(true, Ordering::AcqRel) { self.emit("commit_ack"); }
    }
    pub fn finish(&mut self, disposition: DbOwnerDisposition) {
        if self.state.get().acquired.load(Ordering::Acquire) && !self.state.get().finished.swap(true, Ordering::AcqRel) {
            self.emit(match disposition { DbOwnerDisposition::Returned => "returned", DbOwnerDisposition::Discarded => "discarded" });
        }
    }
}
impl EmergencyDiagnosticHandle {
    #[doc(hidden)]
    pub fn prepare_db_owner_observation(&self,
        context: saddle_core::request_context::RequestContextSnapshot,
        ledger_instance: u64, account: saddle_admission::AccountIdentity,
        storage: &saddle_admission::ProcessLogStorage) -> Option<DbOwnerObservation> {
        static NEXT_OWNER: AtomicU64 = AtomicU64::new(1);
        let owner_id = NEXT_OWNER.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |id| id.checked_add(1)).ok()?;
        Some(DbOwnerObservation { state: saddle_admission::DbOwnerObservationStorage::prepare(storage, DbOwnerObservationState {
            context, output: self.clone(), account, ledger_instance, owner_id,
            finished: std::sync::atomic::AtomicBool::new(false),
            acquired: std::sync::atomic::AtomicBool::new(false),
            acknowledged: std::sync::atomic::AtomicBool::new(false),
            handed_off: std::sync::atomic::AtomicBool::new(false),
            unavailable: std::sync::atomic::AtomicBool::new(false),
        }).ok()? })
    }
}

#[cfg(test)]
mod db_owner_observation_tests {
    use super::*;
    static TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
    fn ledger() -> saddle_admission::ProfuseGwLightweightProcessOwner {
        use saddle_admission::{freeze_deployment_resource_budget,bind_deployment_resource_budget_bootstrap,prepare_profusegw_lightweight_profile};
        use saddle_core::{BootstrapRendezvousIssuer,GeneratedApplicationFreezeSource,ListenerStartupFreezeSource,pair_bootstrap_rendezvous};
        let pending=freeze_deployment_resource_budget(1,32,5000,1,1,1000000,1000000,1000000,1000000).unwrap().reserve_known_database_memory(0,0,0).ok().unwrap();
        let (application,listener)=BootstrapRendezvousIssuer::issue().freeze_application(GeneratedApplicationFreezeSource::new("db-owner",b"descriptor", &["route"])).unwrap();
        let listener=listener.freeze_listener(ListenerStartupFreezeSource::new("db-owner","127.0.0.1:8000".parse().unwrap(),"127.0.0.1:9000".parse().unwrap(),std::time::Duration::from_millis(5000))).ok().unwrap();
        let (whole,receipt)=pair_bootstrap_rendezvous(application,listener).ok().unwrap();
        prepare_profusegw_lightweight_profile(bind_deployment_resource_budget_bootstrap(pending,whole,receipt).ok().unwrap()).ok().unwrap()
    }
    fn context() -> saddle_core::request_context::RequestContextSnapshot {
        use saddle_core::request_context::*;
        RequestRootPublisher::create(ContextLabel::checked("db-owner").unwrap(),ContextFact::NotEstablished).unwrap()
            .reference().view(RequestLocalFacts::new(RequestViewPhase::Reading).with_task(ContextFact::Present(9))).diagnostic_snapshot()
    }
    fn account() -> saddle_admission::AccountIdentity {
        saddle_admission::AccountIdentity{account_slot:0,account_generation:1,audit_generation:2}
    }
    #[test]
    fn prepaid_detached_move_exact_charge_terminal_once_same_domain() {
        let _guard=TEST_LOCK.lock().unwrap();
        let ledger=ledger();let root=ledger.try_process_storage(saddle_admission::StorageDemand::separate(&[]).unwrap()).unwrap();let storage=root.release_log_storage();let before=ledger.ownership_snapshot();
        let (sender,receiver)=mpsc::sync_channel(8);let status=Arc::new(Status::default());
        let output=EmergencyDiagnosticHandle{sender,status:status.clone()};
        let snapshot=context();let owner=output.prepare_db_owner_observation(snapshot,17,account(),&storage).unwrap();
        let expected=saddle_admission::StorageDemand::embedded(std::alloc::Layout::new::<DbOwnerObservationState>(),std::alloc::Layout::new::<()>(),&[]).unwrap().bytes();
        assert_eq!(ledger.ownership_snapshot().process_storage_bytes-before.process_storage_bytes,expected);
        owner.acquired();owner.acquired();owner.commit_ack();owner.commit_ack();
        let mut moved=owner;moved.handoff();moved.handoff();moved.finish(DbOwnerDisposition::Returned);
        moved.finish(DbOwnerDisposition::Discarded);moved.commit_ack();moved.handoff();
        let mut previous=0;let mut owner_id=None;
        for event in ["acquired","commit_ack","finalization_handoff","returned"] {
            let DiagnosticPacket::Admitted(bytes)=receiver.recv().unwrap() else {panic!("fixed record")};
            let row:serde_json::Value=serde_json::from_slice(&bytes.bytes[..bytes.len]).unwrap();
            assert_eq!(row["schema"],"request-db-owner/1");assert_eq!(row["event"],event);
            assert_eq!(row["ledger_instance"],17);assert_eq!(row["account"],serde_json::to_value(account()).unwrap());
            assert_eq!(row["local_request"],row["context"]["local_request"]);
            let id=row["owner_id"].as_u64().unwrap();assert!(id>0);assert_eq!(*owner_id.get_or_insert(id),id);
            let sequence=row["sequence"].as_u64().unwrap();assert!(sequence>previous);previous=sequence;
        }
        assert!(rpc_lifecycle_stamp().0>previous);assert!(receiver.try_recv().is_err());
        drop(moved);assert_eq!(ledger.ownership_snapshot().process_storage_bytes,before.process_storage_bytes);
        println!("db_owner_state_layout={} handle_layout={} charged={}",std::mem::size_of::<DbOwnerObservationState>(),std::mem::size_of::<DbOwnerObservation>(),expected);
    }
    #[test]
    fn closed_or_full_output_and_failed_prep_are_observation_only() {
        let _guard=TEST_LOCK.lock().unwrap();
        let ledger=ledger();let root=ledger.try_process_storage(saddle_admission::StorageDemand::separate(&[]).unwrap()).unwrap();let storage=root.release_log_storage();let before=ledger.ownership_snapshot();
        for closed in [false,true] {
            let (sender,receiver)=mpsc::sync_channel(0);let status=Arc::new(Status::default());status.closed.store(closed,Ordering::Release);
            let output=EmergencyDiagnosticHandle{sender,status:status.clone()};
            let mut owner=output.prepare_db_owner_observation(context(),17,account(),&storage).unwrap();
            owner.acquired();owner.commit_ack();owner.handoff();owner.finish(DbOwnerDisposition::Discarded);
            assert!(owner.state.get().unavailable.load(Ordering::Acquire));assert_eq!(status.dropped.load(Ordering::Relaxed),1);
            assert!(receiver.try_recv().is_err());drop(owner);
        }
        assert_eq!(ledger.ownership_snapshot().process_storage_bytes,before.process_storage_bytes);
        let (sender,_receiver)=mpsc::sync_channel(1);let output=EmergencyDiagnosticHandle{sender,status:Arc::new(Status::default())};
        let layout=std::alloc::Layout::from_size_align(1024*1024*1024,8).unwrap();
        assert!(storage.try_reserve(layout).is_err());
        drop(root);assert!(output.prepare_db_owner_observation(context(),17,account(),&storage).is_none());
    }
}