saddle_observability/root_diagnostic/

original.rs

//! Original error content, streamed into the existing emergency output.
//! These records are privileged originals, not sanitized public error messages.
use super::*;
use std::{error::Error, fmt, io};

const PAYLOAD_BYTES: usize = 6144;

fn written_source_code(error: &(dyn Error + 'static), source_stage: saddle_core::DiagnosticStage,
                       stage: RootRequestEvent) -> DiagnosticCode {
    let generic = || DiagnosticCode::new("framework.error").expect("static code");
    if !matches!(source_stage, saddle_core::DiagnosticStage::RequestDecode)
        || !matches!(stage, RootRequestEvent::Ingress) { return generic(); }
    let Some(mut failure) = error.downcast_ref::<saddle_admission::InputDecodeFailure>() else { return generic(); };
    let mut depth = 0;
    while let saddle_admission::InputDecodeFailure::Context { failure: inner, .. } = failure {
        if depth == 8 { return generic(); }
        failure = inner; depth += 1;
    }
    let code = match failure {
        saddle_admission::InputDecodeFailure::Syntax(_) => "framework.input_decode.syntax",
        saddle_admission::InputDecodeFailure::Utf8(_) => "framework.input_decode.utf8",
        saddle_admission::InputDecodeFailure::Parser => "framework.input_decode.parser",
        _ => return generic(),
    };
    DiagnosticCode::new(code).expect("static code")
}


#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum OriginalCaptureState {
    /// All exposed descriptions and the chain-end marker were accepted, not durable.
    CompleteEnqueued,
    /// The terminating segment and all preceding segments were written by the file writer.
    CompleteWritten,
    OutputUnavailable,
    OutputRejected,
    FormattingFailed,
    CauseCycle,
    LegacyProjectionOnly,
}

#[derive(Serialize)]
pub(crate) struct Segment<'a> {
    schema_version: u8,
    event: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    occurrence: Option<DiagnosticOccurrence>,
    pub(crate) sequence: u64,
    channel: &'static str,
    cause_depth: usize,
    pub(crate) payload: &'a str,
    pub(crate) state: &'static str,
}

pub(crate) trait OriginalOutput {
    fn submit(&self, segment: &Segment<'_>) -> DiagnosticSubmission;
}
impl OriginalOutput for EmergencyDiagnosticHandle {
    fn submit(&self, segment: &Segment<'_>) -> DiagnosticSubmission {
        self.submit_original_record(segment)
    }
}
impl OriginalOutput for crate::BootstrapDiagnostics {
    fn submit(&self, segment: &Segment<'_>) -> DiagnosticSubmission {
        match self.write_original_record(segment) {
            Ok(()) => DiagnosticSubmission::Enqueued,
            Err(_) => DiagnosticSubmission::Closed,
        }
    }
}
struct ConfirmedOutput<'a>(&'a EmergencyDiagnosticHandle);
impl OriginalOutput for ConfirmedOutput<'_> {
    fn submit(&self, segment: &Segment<'_>) -> DiagnosticSubmission {
        match self.0.write_original_record(segment) {
            Ok(()) => DiagnosticSubmission::Enqueued,
            Err(_) => DiagnosticSubmission::Closed,
        }
    }
}

struct Stream<'a, O: OriginalOutput> {
    output: &'a O,
    occurrence: Option<DiagnosticOccurrence>,
    event: &'static str,
    sequence: u64,
    depth: usize,
    channel: &'static str,
    bytes: [u8; PAYLOAD_BYTES],
    len: usize,
    escaped: usize,
    failure: Option<OriginalCaptureState>,
    submission: DiagnosticSubmission,
}
impl<O: OriginalOutput> Stream<'_, O> {
    fn emit(&mut self, state: &'static str) -> fmt::Result {
        let payload = std::str::from_utf8(&self.bytes[..self.len]).map_err(|_| fmt::Error)?;
        let submission = self.output.submit(&Segment {
            schema_version: 1,
            event: self.event,
            occurrence: self.occurrence,
            sequence: self.sequence,
            channel: self.channel,
            cause_depth: self.depth,
            payload,
            state,
        });
        if submission != DiagnosticSubmission::Enqueued {
            self.submission = submission;
            self.failure = Some(OriginalCaptureState::OutputRejected);
            return Err(fmt::Error);
        }
        self.sequence += 1;
        self.len = 0;
        self.escaped = 0;
        Ok(())
    }
    fn push(&mut self, text: &str) -> fmt::Result {
        if self.failure.is_some() {
            return Err(fmt::Error);
        }
        for c in text.chars() {
            let cost = match c {
                '"' | '\\' | '\n' | '\r' | '\t' | '\u{8}' | '\u{c}' => 2,
                '\u{0}'..='\u{1f}' => 6,
                _ => c.len_utf8(),
            };
            if self.escaped + cost > PAYLOAD_BYTES || self.len + c.len_utf8() > PAYLOAD_BYTES {
                self.emit("continuation")?;
            }
            let size = c.len_utf8();
            c.encode_utf8(&mut self.bytes[self.len..self.len + size]);
            self.len += size;
            self.escaped += cost;
        }
        Ok(())
    }
    fn json<T: Serialize>(&mut self, value: &T) -> fmt::Result {
        serde_json::to_writer(&mut *self, value).map_err(|_| fmt::Error)?;
        self.emit("field_end")
    }
    fn report_failure(&mut self, state: OriginalCaptureState) {
        // Preserve the already captured prefix; never mark it complete. If the
        // queue itself rejected, it may also reject this marker. Readback then
        // sees a missing terminal, which is necessarily incomplete.
        if self.failure == Some(OriginalCaptureState::OutputRejected) {
            return;
        }
        self.failure = Some(state);
        let marker = match state {
            OriginalCaptureState::FormattingFailed => "formatting_failed",
            OriginalCaptureState::CauseCycle => "cause_cycle",
            _ => "capture_failed",
        };
        let _ = self.emit(marker);
    }
}
/// Reuse the original UTF-8/escaped-cost framing without making ordinary
/// output mandatory. After rejection consume encoder bytes without allocating
/// a serde I/O error; the first failure and absent terminal remain observable.
pub(crate) fn stream_record<O: OriginalOutput>(output: &O,
    encode: impl Fn(&mut dyn io::Write) -> serde_json::Result<()>) -> DiagnosticSubmission {
    stream_record_identified(output, "context_record_segment", None, encode)
}
fn stream_record_identified<O: OriginalOutput>(output: &O, event: &'static str,
    occurrence: Option<DiagnosticOccurrence>, encode: impl Fn(&mut dyn io::Write) -> serde_json::Result<()>) -> DiagnosticSubmission {
    let mut stream = Stream { output, occurrence, event, sequence: 0,
        depth: 0, channel: "record", bytes: [0; PAYLOAD_BYTES], len: 0, escaped: 0,
        failure: None, submission: DiagnosticSubmission::Enqueued };
    struct Lossy<'a, 'b, O: OriginalOutput>(&'a mut Stream<'b, O>);
    impl<O: OriginalOutput> io::Write for Lossy<'_, '_, O> {
        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
            if self.0.failure.is_none() {
                if let Ok(text) = std::str::from_utf8(bytes) { let _ = self.0.push(text); }
                else { self.0.failure = Some(OriginalCaptureState::FormattingFailed); }
            }
            Ok(bytes.len())
        }
        fn flush(&mut self) -> io::Result<()> { Ok(()) }
    }
    if encode(&mut Lossy(&mut stream)).is_err() { return DiagnosticSubmission::EncodingFailed; }
    if stream.failure == Some(OriginalCaptureState::FormattingFailed) { return DiagnosticSubmission::EncodingFailed; }
    if stream.failure.is_none() { let _ = stream.emit("record_complete"); }
    stream.submission
}

impl<O: OriginalOutput> fmt::Write for Stream<'_, O> {
    fn write_str(&mut self, text: &str) -> fmt::Result {
        self.push(text)
    }
}
impl<O: OriginalOutput> io::Write for Stream<'_, O> {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        // serde_json writes UTF-8 str fragments and ASCII escapes; this adapter
        // rejects other binary writers rather than converting invalid bytes.
        let text = std::str::from_utf8(bytes).map_err(|_| io::ErrorKind::InvalidData)?;
        self.push(text).map_err(|_| io::ErrorKind::Other)?;
        Ok(bytes.len())
    }
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

#[derive(Serialize)]
struct Header<'a> {
    timestamp_unix_ms: u128,
    context: OriginalContext<'a>,
    stage: RootRequestEvent,
    facts: SourceDetail<'a>,
    outcome: RootOutcomeFacts,
    original_contract: &'static str,
    source_interface: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    protocol: Option<RemoteProtocolContext<'a>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    remote_target: Option<RemoteTargetContext<'a>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    grpc: Option<GrpcTargetContext<'a>>,
}
enum OriginalContext<'a> {
    Request(&'a RequestExecutionView),
    Unrooted(crate::AdmissionEventContext<'a>),
}
impl Serialize for OriginalContext<'_> {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        match self {
            Self::Request(view) => view.unified_context().serialize(serializer),
            Self::Unrooted(context) => context.serialize(serializer),
        }
    }
}
#[derive(Serialize)]
pub struct RemoteTargetContext<'a> {
    pub business_unit: &'a str,
    pub function: &'a str,
}
/// Borrowed protocol values from the actual peer response and request. They
/// are serialized while both objects are alive; no bounded summary replaces
/// the original target identity.
#[derive(Serialize)]
pub struct RemoteProtocolContext<'a> {
    pub business_unit: &'a str,
    pub function: &'a str,
    pub technical_code: &'a str,
    pub certainty: &'a str,
}
/// Borrowed facts of the standard RPC attempt; never replaces request context.
#[derive(Serialize)]
pub struct GrpcTargetContext<'a> {
    pub business_unit: &'a str,
    pub dependency_alias: &'a str,
    pub contract_sha256: &'a [u8;32],
    pub service: &'a str,
    pub method: &'a str,
    pub authority_template: &'a str,
    pub authority: Option<&'a str>,
    pub attempt: u64,
    pub deadline_remaining_ns: u128,
    pub phase: &'a str,
    pub send_progress: &'a str,
    pub response_wire_bytes: Option<usize>,
    #[serde(skip_serializing_if="Option::is_none")]
    pub status: Option<GrpcStatusFacts<'a>>,
}
#[derive(Serialize)]
pub struct GrpcStatusFacts<'a> {
    pub code: i32,
    pub message: &'a str,
    pub details: &'a [u8],
}
enum RawSource<'a> {
    Error(&'a (dyn Error + 'static)),
    BorrowedError(&'a (dyn Error + 'a)),
    Description(&'a dyn fmt::Display, &'a dyn fmt::Debug),
}
#[derive(Serialize)]
struct IoFacts {
    kind: Kind,
    os_code: Option<i32>,
}
struct Kind(io::ErrorKind);
impl Serialize for Kind {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        serializer.collect_str(&format_args!("{:?}", self.0))
    }
}

impl RootDiagnosticScope<'_> {
    /// The only source path that can issue a written receipt. A failed write
    /// returns an unconfirmed source receipt for explicit failure handling.
    #[track_caller]
    pub fn source_error_written(
        &self,
        error: &(dyn Error + 'static),
        source_stage: saddle_core::DiagnosticStage,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> Result<super::WrittenRootFailure, RootRequestFailure> {
        let code = written_source_code(error, source_stage, stage);
        let facts = BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            saddle_core::CaptureSite::FirstObserved,
            saddle_core::BoundedDiagnosticCause::new(source_stage, code),
        );
        let receipt = self.original_record_mode(
            RawSource::Error(error), SourceDetail::Bounded(&facts), code, stage, axes, true, None, None,
        );
        if receipt.original_capture == OriginalCaptureState::CompleteWritten {
            Ok(super::WrittenRootFailure(receipt))
        } else {
            // No caller can turn a partial source into a written receipt.
            Err(receipt)
        }
    }
    /// Capture any standard error before lossy mapping. Unknown types need no
    /// registration. Caller location is first observation, not invented origin.
    #[track_caller]
    pub fn source_error(
        &self,
        error: &(dyn Error + 'static),
        source_stage: saddle_core::DiagnosticStage,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> RootRequestFailure {
        let code = DiagnosticCode::new("framework.error").expect("static code");
        let facts = BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            saddle_core::CaptureSite::FirstObserved,
            saddle_core::BoundedDiagnosticCause::new(source_stage, code),
        );
        self.source_error_with_facts(error, facts, code, stage, axes)
    }

    /// Optional typed facts/classification supplement, never replace, original
    /// Display/source. Owns no raw error after returning. Cleanup may supply the
    /// existing primary occurrence in facts; this does not mint a second receipt.
    pub fn source_error_with_facts(
        &self,
        error: &(dyn Error + 'static),
        facts: BoundedDiagnostic,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> RootRequestFailure {
        self.original_record(
            RawSource::Error(error),
            SourceDetail::Bounded(&facts),
            classification,
            stage,
            axes,
        )
    }

    /// Preserve an externally observed source and its existing typed facts;
    /// issue a written receipt only after the complete chain is acknowledged.
    pub fn source_error_with_facts_written(
        &self,
        error: &(dyn Error + 'static),
        facts: BoundedDiagnostic,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> Result<super::WrittenRootFailure, RootRequestFailure> {
        let receipt = self.original_record_mode(
            RawSource::Error(error), SourceDetail::Bounded(&facts),
            classification, stage, axes, true, None, None,
        );
        if receipt.original_capture == OriginalCaptureState::CompleteWritten {
            Ok(super::WrittenRootFailure(receipt))
        } else {
            Err(receipt)
        }
    }

    /// Stream a request-owned error and its borrowed input before either is
    /// projected or released by the supervising scope.
    pub fn source_borrowed_error_with_facts_written<E: Error>(
        &self,
        error: &E,
        facts: BoundedDiagnostic,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> Result<super::WrittenRootFailure, RootRequestFailure> {
        let receipt = self.original_record_mode(
            RawSource::BorrowedError(error), SourceDetail::Bounded(&facts),
            classification, stage, axes, true, None, None,
        );
        if receipt.original_capture == OriginalCaptureState::CompleteWritten {
            Ok(super::WrittenRootFailure(receipt))
        } else {
            Err(receipt)
        }
    }

    /// A panic/string payload may have Display but no Error::source interface.
    /// Preserve it without inventing a cause chain. The original facts/ID are
    /// consumed once; no second capture is required at the logger boundary.
    pub fn source_description<D: fmt::Display + fmt::Debug>(
        &self,
        description: &D,
        facts: BoundedDiagnostic,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> RootRequestFailure {
        self.original_record(
            RawSource::Description(description, description),
            SourceDetail::Bounded(&facts),
            classification,
            stage,
            axes,
        )
    }

    /// Confirm the complete original description before issuing a written
    /// receipt. Panic payloads without an Error interface use this path.
    pub fn source_description_written<D: fmt::Display + fmt::Debug>(
        &self,
        description: &D,
        facts: BoundedDiagnostic,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> Result<super::WrittenRootFailure, RootRequestFailure> {
        self.written_record(
            RawSource::Description(description, description),
            SourceDetail::Bounded(&facts),
            classification,
            stage,
            axes,
        )
    }

    pub fn source_remote_protocol_written<D: fmt::Display + fmt::Debug>(
        &self,
        description: &D,
        facts: BoundedDiagnostic,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
        protocol: RemoteProtocolContext<'_>,
    ) -> Result<super::WrittenRootFailure, RootRequestFailure> {
        let receipt = self.original_record_mode(
            RawSource::Description(description, description),
            SourceDetail::Bounded(&facts), classification, stage, axes, true,
            Some(protocol), None,
        );
        if receipt.original_capture == OriginalCaptureState::CompleteWritten {
            Ok(super::WrittenRootFailure(receipt))
        } else { Err(receipt) }
    }

    pub fn source_remote_error_written(
        &self,
        error: &(dyn Error + 'static),
        facts: BoundedDiagnostic,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
        target: RemoteTargetContext<'_>,
    ) -> Result<super::WrittenRootFailure, RootRequestFailure> {
        let receipt = self.original_record_mode(
            RawSource::Error(error), SourceDetail::Bounded(&facts),
            classification, stage, axes, true, None, Some(target),
        );
        if receipt.original_capture == OriginalCaptureState::CompleteWritten {
            Ok(super::WrittenRootFailure(receipt))
        } else { Err(receipt) }
    }

    /// Compatibility for an already captured panic diagnostic plus its original
    /// payload. Preserves ID/primary/location/stack status, does not recapture.
    pub fn source_existing_description<D: fmt::Display + fmt::Debug>(
        &self,
        description: &D,
        diagnostic: Diagnostic,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> RootRequestFailure {
        self.original_record(
            RawSource::Description(description, description),
            SourceDetail::Existing(&diagnostic),
            classification,
            stage,
            axes,
        )
    }

    pub fn source_existing_description_written<D: fmt::Display + fmt::Debug>(
        &self,
        description: &D,
        diagnostic: Diagnostic,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> Result<super::WrittenRootFailure, RootRequestFailure> {
        self.written_record(
            RawSource::Description(description, description),
            SourceDetail::Existing(&diagnostic),
            classification,
            stage,
            axes,
        )
    }

    pub fn source_existing_error(
        &self,
        error: &(dyn Error + 'static),
        diagnostic: Diagnostic,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> RootRequestFailure {
        self.original_record(
            RawSource::Error(error),
            SourceDetail::Existing(&diagnostic),
            classification,
            stage,
            axes,
        )
    }

    pub fn source_existing_error_written(
        &self,
        error: &(dyn Error + 'static),
        diagnostic: Diagnostic,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> Result<super::WrittenRootFailure, RootRequestFailure> {
        self.written_record(
            RawSource::Error(error),
            SourceDetail::Existing(&diagnostic),
            classification,
            stage,
            axes,
        )
    }

    fn written_record(
        &self,
        raw: RawSource<'_>,
        facts: SourceDetail<'_>,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> Result<super::WrittenRootFailure, RootRequestFailure> {
        let receipt = self.original_record_mode(raw, facts, classification, stage, axes, true, None, None);
        if receipt.original_capture == OriginalCaptureState::CompleteWritten {
            Ok(super::WrittenRootFailure(receipt))
        } else {
            Err(receipt)
        }
    }

    fn original_record(
        &self,
        raw: RawSource<'_>,
        facts: SourceDetail<'_>,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> RootRequestFailure {
        self.original_record_mode(raw, facts, classification, stage, axes, false, None, None)
    }
    fn original_record_mode(
        &self,
        raw: RawSource<'_>,
        facts: SourceDetail<'_>,
        classification: DiagnosticCode,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
        confirmed: bool,
        protocol: Option<RemoteProtocolContext<'_>>,
        remote_target: Option<RemoteTargetContext<'_>>,
    ) -> RootRequestFailure {
        self.original_record_grpc(raw, facts, classification, stage, axes, confirmed, protocol, remote_target, None)
    }
    pub fn source_grpc_error_written<E: Error>(
        &self, error: &E, facts: BoundedDiagnostic, classification: DiagnosticCode,
        stage: RootRequestEvent, axes: RootOutcomeFacts, target: GrpcTargetContext<'_>,
    ) -> Result<super::WrittenRootFailure, RootRequestFailure> {
        let receipt=self.original_record_grpc(RawSource::BorrowedError(error),SourceDetail::Bounded(&facts),classification,stage,axes,true,None,None,Some(target));
        if receipt.original_capture==OriginalCaptureState::CompleteWritten {Ok(super::WrittenRootFailure(receipt))}else{Err(receipt)}
    }
    fn original_record_grpc(
        &self, raw:RawSource<'_>, facts:SourceDetail<'_>,classification:DiagnosticCode,
        stage:RootRequestEvent,axes:RootOutcomeFacts,confirmed:bool,
        protocol:Option<RemoteProtocolContext<'_>>,remote_target:Option<RemoteTargetContext<'_>>,
        grpc:Option<GrpcTargetContext<'_>>,
    ) -> RootRequestFailure {
        let (occurrence, category) = match &facts {
            SourceDetail::Bounded(d) => (d.occurrence(), d.category()),
            SourceDetail::Existing(d) => (d.occurrence(), d.category()),
        };
        let source_interface = match raw {
            RawSource::Error(_) | RawSource::BorrowedError(_) => "error_source",
            RawSource::Description(_, _) => "unavailable_description_debug_only",
        };
        let (submission, original_capture) = match self.output {
            None => (
                DiagnosticSubmission::OutputUnavailable,
                OriginalCaptureState::OutputUnavailable,
            ),
            Some(output) => {
                let header = Header {
                    timestamp_unix_ms: timestamp(),
                    context: OriginalContext::Request(self.view),
                    stage,
                    facts,
                    outcome: axes,
                    original_contract: "display-debug-source/v1",
                    source_interface,
                    protocol,
                    remote_target,
                    grpc,
                };
                if confirmed {
                    let (submission, state) = capture(&ConfirmedOutput(output), occurrence, raw, &header);
                    if submission == DiagnosticSubmission::Enqueued && state == OriginalCaptureState::CompleteEnqueued {
                        (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten)
                    } else {
                        (submission, state)
                    }
                } else {
                    capture(output, occurrence, raw, &header)
                }
            },
        };
        RootRequestFailure {
            source: self.view.clone(),
            occurrence,
            classification,
            source_submission: submission,
            category,
            outcome: axes,
            original_capture,
        }
    }
}

/// Output facts only: no request root, task authority or failure-retention permit.
#[must_use = "capture facts describe submission, not durable output or resource completion"]
pub struct UnrootedCaptureFacts {
    occurrence: DiagnosticOccurrence,
    submission: DiagnosticSubmission,
    original_capture: OriginalCaptureState,
}
/// A root-free original whose terminating segment was acknowledged by the
/// controlled writer. The private field prevents callers from promoting a
/// queued, missing, or partially written record.
///
/// ```compile_fail
/// use saddle_observability::root_diagnostic::WrittenUnrootedFailure;
/// fn forge(facts: saddle_observability::root_diagnostic::UnrootedCaptureFacts)
///     -> WrittenUnrootedFailure { WrittenUnrootedFailure(facts) }
/// ```
#[must_use = "carry the written source to its consumer"]
pub struct WrittenUnrootedFailure(UnrootedCaptureFacts);
impl WrittenUnrootedFailure {
    fn from_facts(facts: UnrootedCaptureFacts) -> Result<Self, UnrootedCaptureFacts> {
        if facts.original_capture == OriginalCaptureState::CompleteWritten {
            Ok(Self(facts))
        } else {
            Err(facts)
        }
    }

    pub fn occurrence(&self) -> DiagnosticOccurrence { self.0.occurrence }
}
impl UnrootedCaptureFacts {
    pub fn occurrence(&self) -> DiagnosticOccurrence {
        self.occurrence
    }
    pub fn submission(&self) -> DiagnosticSubmission {
        self.submission
    }
    pub fn original_capture(&self) -> OriginalCaptureState {
        self.original_capture
    }
}

/// A receipt issued only by the corresponding component source writer.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ComponentSourceKind { Start, Cleanup }

pub struct ComponentSourceReceipt {
    facts: UnrootedCaptureFacts,
    kind: ComponentSourceKind,
}
impl ComponentSourceReceipt {
    pub fn kind(&self) -> ComponentSourceKind { self.kind }
    pub fn occurrence(&self) -> DiagnosticOccurrence { self.facts.occurrence() }
    pub fn original_capture(&self) -> OriginalCaptureState { self.facts.original_capture() }
}
/// A component source confirmed by the same controlled writer. Startup and
/// cleanup remain distinct occurrences even when linked by a primary ID.
///
/// ```compile_fail
/// use saddle_observability::root_diagnostic::WrittenComponentFailure;
/// fn forge(facts: saddle_observability::root_diagnostic::ComponentSourceReceipt)
///     -> WrittenComponentFailure { WrittenComponentFailure(facts) }
/// ```
#[must_use = "carry the written component source to its consumer"]
pub struct WrittenComponentFailure(ComponentSourceReceipt);
impl WrittenComponentFailure {
    fn from_receipt(receipt: ComponentSourceReceipt) -> Result<Self, ComponentSourceReceipt> {
        if receipt.original_capture() == OriginalCaptureState::CompleteWritten {
            Ok(Self(receipt))
        } else {
            Err(receipt)
        }
    }
    pub fn occurrence(&self) -> DiagnosticOccurrence { self.0.occurrence() }
    pub fn kind(&self) -> ComponentSourceKind { self.0.kind() }
}

/// For a failure before a root exists. Does not create identity or acquire storage.
/// Callers with a live view must use RootDiagnosticScope instead.
pub struct UnrootedDiagnosticScope<'a> {
    application: &'a saddle_core::ContextLabel,
    lifecycle: saddle_core::RequestViewPhase,
    output: Option<&'a EmergencyDiagnosticHandle>,
}
impl<'a> UnrootedDiagnosticScope<'a> {
    pub fn new(
        application: &'a saddle_core::ContextLabel,
        lifecycle: saddle_core::RequestViewPhase,
        output: Option<&'a EmergencyDiagnosticHandle>,
    ) -> Self {
        Self {
            application,
            lifecycle,
            output,
        }
    }
    #[track_caller]
    pub fn source_error(
        &self,
        error: &(dyn Error + 'static),
        source_stage: saddle_core::DiagnosticStage,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> UnrootedCaptureFacts {
        let facts = BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            saddle_core::CaptureSite::FirstObserved,
            saddle_core::BoundedDiagnosticCause::new(
                source_stage,
                DiagnosticCode::new("framework.error").unwrap(),
            ),
        );
        self.original_record(RawSource::Error(error), &facts, stage, axes, false)
    }
    /// Confirm every source segment before the root-free error is discarded.
    #[track_caller]
    pub fn source_error_written(
        &self,
        error: &(dyn Error + 'static),
        source_stage: saddle_core::DiagnosticStage,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> UnrootedCaptureFacts {
        let facts = BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            saddle_core::CaptureSite::FirstObserved,
            saddle_core::BoundedDiagnosticCause::new(
                source_stage,
                written_source_code(error, source_stage, stage),
            ),
        );
        self.original_record(RawSource::Error(error), &facts, stage, axes, true)
    }
    /// Preserve the original while borrowed and return a non-forgeable
    /// positive receipt only after the complete file write is acknowledged.
    #[track_caller]
    pub fn source_error_recorded(
        &self,
        error: &(dyn Error + 'static),
        source_stage: saddle_core::DiagnosticStage,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> Result<WrittenUnrootedFailure, UnrootedCaptureFacts> {
        WrittenUnrootedFailure::from_facts(
            self.source_error_written(error, source_stage, stage, axes)
        )
    }
    /// Record an already captured occurrence without minting a second ID.
    /// This is the root-free source path for a paired public SaddleError.
    pub fn source_existing_error_recorded(
        &self,
        error: &(dyn Error + 'static),
        diagnostic: &Diagnostic,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> Result<WrittenUnrootedFailure, UnrootedCaptureFacts> {
        WrittenUnrootedFailure::from_facts(self.original_record_detail(
            RawSource::Error(error), SourceDetail::Existing(diagnostic), stage, axes, true,
        ))
    }
    #[track_caller]
    pub fn source_description<D: fmt::Display + fmt::Debug>(
        &self,
        description: &D,
        source_stage: saddle_core::DiagnosticStage,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> UnrootedCaptureFacts {
        let facts = BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            saddle_core::CaptureSite::FirstObserved,
            saddle_core::BoundedDiagnosticCause::new(
                source_stage,
                DiagnosticCode::new("framework.error").unwrap(),
            ),
        );
        self.original_record(
            RawSource::Description(description, description),
            &facts,
            stage,
            axes,
            false,
        )
    }
    /// Confirm a Display/Debug source with no Error::source implementation.
    #[track_caller]
    pub fn source_description_written<D: fmt::Display + fmt::Debug>(
        &self,
        description: &D,
        source_stage: saddle_core::DiagnosticStage,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
    ) -> UnrootedCaptureFacts {
        let facts = BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            saddle_core::CaptureSite::FirstObserved,
            saddle_core::BoundedDiagnosticCause::new(
                source_stage,
                DiagnosticCode::new("framework.error").unwrap(),
            ),
        );
        self.original_record(RawSource::Description(description, description), &facts, stage, axes, true)
    }
    fn original_record(
        &self,
        raw: RawSource<'_>,
        facts: &BoundedDiagnostic,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
        confirmed: bool,
    ) -> UnrootedCaptureFacts {
        self.original_record_detail(raw, SourceDetail::Bounded(facts), stage, axes, confirmed)
    }
    fn original_record_detail(
        &self,
        raw: RawSource<'_>,
        facts: SourceDetail<'_>,
        stage: RootRequestEvent,
        axes: RootOutcomeFacts,
        confirmed: bool,
    ) -> UnrootedCaptureFacts {
        let occurrence = match facts {
            SourceDetail::Bounded(diagnostic) => diagnostic.occurrence(),
            SourceDetail::Existing(diagnostic) => diagnostic.occurrence(),
        };
        let source_interface = match raw {
            RawSource::Error(_) | RawSource::BorrowedError(_) => "error_source",
            RawSource::Description(_, _) => "unavailable_description_debug_only",
        };
        let (submission, original_capture) = match self.output {
            None => (
                DiagnosticSubmission::OutputUnavailable,
                OriginalCaptureState::OutputUnavailable,
            ),
            Some(output) => {
                let header = Header {
                    timestamp_unix_ms: timestamp(),
                    context: OriginalContext::Unrooted(crate::AdmissionEventContext::Unrooted {
                        application: self.application,
                        lifecycle: self.lifecycle,
                    }),
                    stage,
                    facts,
                    outcome: axes,
                    original_contract: "display-debug-source/v1",
                    source_interface,
                    protocol: None,
                    remote_target: None,
                    grpc: None,
                };
                if confirmed {
                    let (submission, state) = capture(&ConfirmedOutput(output), occurrence, raw, &header);
                    if submission == DiagnosticSubmission::Enqueued && state == OriginalCaptureState::CompleteEnqueued {
                        (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten)
                    } else {
                        (submission, state)
                    }
                } else {
                    capture(output, occurrence, raw, &header)
                }
            },
        };
        UnrootedCaptureFacts {
            occurrence,
            submission,
            original_capture,
        }
    }
}

// Brent's online cycle detector: constant borrowed state, no visited-node list
// and no limit on a finite acyclic chain. Error::source must expose a stable
// chain during this borrow; pathological mutating implementations are not
// converted into an invented finite chain.
struct CycleDetector<'a> {
    anchor: &'a (dyn Error + 'static),
    power: usize,
    distance: usize,
}
impl<'a> CycleDetector<'a> {
    fn new(anchor: &'a (dyn Error + 'static)) -> Self {
        Self {
            anchor,
            power: 1,
            distance: 0,
        }
    }
    fn repeats(&mut self, next: &'a (dyn Error + 'static)) -> bool {
        // Include trait metadata: distinct zero-sized errors may share a data
        // address. Duplicated vtables may delay detection until source traversal
        // revisits the same edge, but must not turn a finite chain into a cycle.
        if std::ptr::eq(self.anchor, next) {
            return true;
        }
        self.distance += 1;
        if self.distance == self.power {
            self.anchor = next;
            self.power = self.power.saturating_mul(2);
            self.distance = 0;
        }
        false
    }
}

/// Reproduce serde_json's Debug bytes without its internal to_string heap.
/// Parser errors have bounded static reasons; a custom oversized reason fails
/// closed, never claiming a complete original after truncation.
pub fn write_json_error_debug<W: fmt::Write>(out: &mut W,
    error: &serde_json::Error) -> fmt::Result {
    struct FixedText<const N: usize> { bytes: [u8; N], len: usize }
    impl<const N: usize> fmt::Write for FixedText<N> {
        fn write_str(&mut self, value: &str) -> fmt::Result {
            if self.len + value.len() > self.bytes.len() { return Err(fmt::Error); }
            self.bytes[self.len..self.len + value.len()].copy_from_slice(value.as_bytes());
            self.len += value.len();
            Ok(())
        }
    }
    let mut shown = FixedText { bytes: [0; 6144], len: 0 };
    fmt::write(&mut shown, format_args!("{error}"))?;
    let shown = std::str::from_utf8(&shown.bytes[..shown.len]).map_err(|_| fmt::Error)?;
    let mut suffix = FixedText { bytes: [0; 64], len: 0 };
    if error.line() != 0 {
        fmt::write(&mut suffix,
            format_args!(" at line {} column {}", error.line(), error.column()))?;
    }
    let suffix = std::str::from_utf8(&suffix.bytes[..suffix.len]).map_err(|_| fmt::Error)?;
    let reason = shown.strip_suffix(suffix).ok_or(fmt::Error)?;
    fmt::write(out, format_args!("Error({reason:?}, line: {}, column: {})",
        error.line(), error.column()))
}

fn write_error_debug<O: OriginalOutput>(stream: &mut Stream<'_, O>,
    error: &(dyn Error + 'static), bounded_input: bool) -> fmt::Result {
    if bounded_input {
        if let Some(json) = error.downcast_ref::<serde_json::Error>() {
            return write_json_error_debug(stream, json);
        }
        if let Some(saddle_admission::InputDecodeFailure::Syntax(json)) =
            error.downcast_ref::<saddle_admission::InputDecodeFailure>() {
            stream.push("Syntax(")?;
            write_json_error_debug(stream, json)?;
            return stream.push(")");
        }
    }
    fmt::write(stream, format_args!("{error:?}"))
}

fn capture<O: OriginalOutput>(
    output: &O,
    occurrence: DiagnosticOccurrence,
    raw: RawSource<'_>,
    header: &impl Serialize,
) -> (DiagnosticSubmission, OriginalCaptureState) {
    // The special case reproduces only known serde JSON Debug bytes. Other
    // error implementations keep their native Debug and source traversal.
    capture_with_mode(output, occurrence, raw, header, true)
}

fn capture_with_mode<O: OriginalOutput>(
    output: &O,
    occurrence: DiagnosticOccurrence,
    raw: RawSource<'_>,
    header: &impl Serialize,
    bounded_input: bool,
) -> (DiagnosticSubmission, OriginalCaptureState) {
    let mut stream = Stream {
        output,
        occurrence: Some(occurrence),
        event: "request_error_original",
        sequence: 0,
        depth: 0,
        channel: "context",
        bytes: [0; PAYLOAD_BYTES],
        len: 0,
        escaped: 0,
        failure: None,
        submission: DiagnosticSubmission::Enqueued,
    };
    let result = (|| -> fmt::Result {
        stream.json(header)?;
        let error = match raw {
            RawSource::Error(error) => error,
            RawSource::BorrowedError(error) => {
                stream.channel = "description";
                if fmt::write(&mut stream, format_args!("{error}")).is_err() {
                    let state = stream.failure.unwrap_or(OriginalCaptureState::FormattingFailed);
                    stream.report_failure(state);
                    return Err(fmt::Error);
                }
                stream.emit("field_end")?;
                stream.channel = "debug";
                if fmt::write(&mut stream, format_args!("{error:?}")).is_err() {
                    let state = stream.failure.unwrap_or(OriginalCaptureState::FormattingFailed);
                    stream.report_failure(state);
                    return Err(fmt::Error);
                }
                stream.emit("field_end")?;
                stream.depth = 1;
                let Some(first) = error.source() else {
                    stream.channel = "terminal";
                    return stream.emit("exposed_chain_complete");
                };
                let mut cycle = CycleDetector::new(first);
                let mut next = Some(first);
                while let Some(current) = next {
                    stream.channel = "description";
                    if fmt::write(&mut stream, format_args!("{current}")).is_err() {
                        let state = stream.failure.unwrap_or(OriginalCaptureState::FormattingFailed);
                        stream.report_failure(state);
                        return Err(fmt::Error);
                    }
                    stream.emit("field_end")?;
                    stream.channel = "debug";
                    if write_error_debug(&mut stream, current, bounded_input).is_err() {
                        let state = stream.failure.unwrap_or(OriginalCaptureState::FormattingFailed);
                        stream.report_failure(state);
                        return Err(fmt::Error);
                    }
                    stream.emit("field_end")?;
                    if let Some(io) = current.downcast_ref::<io::Error>() {
                        stream.channel = "io_facts";
                        stream.json(&IoFacts { kind: Kind(io.kind()), os_code: io.raw_os_error() })?;
                    }
                    next = current.source();
                    stream.depth += 1;
                    if next.is_some_and(|next| cycle.repeats(next)) {
                        stream.channel = "description";
                        stream.report_failure(OriginalCaptureState::CauseCycle);
                        return Err(fmt::Error);
                    }
                }
                stream.channel = "terminal";
                return stream.emit("exposed_chain_complete");
            }
            RawSource::Description(description, debug) => {
                stream.channel = "description";
                if fmt::write(&mut stream, format_args!("{description}")).is_err() {
                    let state = stream
                        .failure
                        .unwrap_or(OriginalCaptureState::FormattingFailed);
                    stream.report_failure(state);
                    return Err(fmt::Error);
                }
                stream.emit("field_end")?;
                stream.channel = "debug";
                if fmt::write(&mut stream, format_args!("{debug:?}")).is_err() {
                    let state = stream
                        .failure
                        .unwrap_or(OriginalCaptureState::FormattingFailed);
                    stream.report_failure(state);
                    return Err(fmt::Error);
                }
                stream.emit("field_end")?;
                stream.depth = 1;
                stream.channel = "terminal";
                return stream.emit("description_complete_source_unavailable");
            }
        };
        let mut cycle = CycleDetector::new(error);
        let mut next = Some(error);
        while let Some(current) = next {
            stream.channel = "description";
            if fmt::write(&mut stream, format_args!("{current}")).is_err() {
                let failure = stream
                    .failure
                    .unwrap_or(OriginalCaptureState::FormattingFailed);
                stream.report_failure(failure);
                return Err(fmt::Error);
            }
            stream.emit("field_end")?;
            stream.channel = "debug";
            if write_error_debug(&mut stream, current, bounded_input).is_err() {
                let state = stream
                    .failure
                    .unwrap_or(OriginalCaptureState::FormattingFailed);
                stream.report_failure(state);
                return Err(fmt::Error);
            }
            stream.emit("field_end")?;
            if let Some(io) = current.downcast_ref::<io::Error>() {
                stream.channel = "io_facts";
                stream.json(&IoFacts {
                    kind: Kind(io.kind()),
                    os_code: io.raw_os_error(),
                })?;
            }
            next = current.source();
            stream.depth += 1;
            if next.is_some_and(|next| cycle.repeats(next)) {
                stream.channel = "description";
                stream.report_failure(OriginalCaptureState::CauseCycle);
                return Err(fmt::Error);
            }
        }
        stream.channel = "terminal";
        stream.emit("exposed_chain_complete")
    })();
    if result.is_ok() {
        (stream.submission, OriginalCaptureState::CompleteEnqueued)
    } else {
        let state = stream
            .failure
            .unwrap_or(OriginalCaptureState::FormattingFailed);
        if stream.submission == DiagnosticSubmission::Enqueued {
            (DiagnosticSubmission::EncodingFailed, state)
        } else {
            (stream.submission, state)
        }
    }
}

/// Source scratch only; the borrowed error owns its own storage. Neither is queued.
pub fn original_capture_layout() -> (std::alloc::Layout, std::alloc::Layout, usize) {
    (
        std::alloc::Layout::new::<Stream<'static, EmergencyDiagnosticHandle>>(),
        std::alloc::Layout::new::<CycleDetector<'static>>(),
        PAYLOAD_BYTES,
    )
}

#[cfg(test)]
mod content_tests {
    use super::*;
    use std::cell::RefCell;

    #[test]
    fn typed_written_input_code_is_borrowed_bounded_and_stage_specific() {
        use saddle_admission::InputDecodeFailure as F;
        use saddle_core::DiagnosticStage as D;
        let code = |e: &(dyn Error + 'static), source, stage| serde_json::to_value(written_source_code(e, source, stage)).unwrap();
        let syntax = F::Syntax(serde_json::from_str::<serde_json::Value>("{").unwrap_err());
        assert_eq!(code(&syntax,D::RequestDecode,RootRequestEvent::Ingress),"framework.input_decode.syntax");
        let bytes = vec![255]; let utf8 = F::Utf8(std::str::from_utf8(&bytes).unwrap_err());
        assert_eq!(code(&utf8,D::RequestDecode,RootRequestEvent::Ingress),"framework.input_decode.utf8");
        assert_eq!(code(&F::Parser,D::RequestDecode,RootRequestEvent::Ingress),"framework.input_decode.parser");
        let mut wrapped = F::Parser;
        for _ in 0..8 { wrapped = F::Context { request_id: "r".into(), failure: Box::new(wrapped) }; }
        assert_eq!(code(&wrapped,D::RequestDecode,RootRequestEvent::Ingress),"framework.input_decode.parser");
        wrapped = F::Context { request_id: "r".into(), failure: Box::new(wrapped) };
        assert_eq!(code(&wrapped,D::RequestDecode,RootRequestEvent::Ingress),"framework.error");
        assert_eq!(code(&F::Resource(saddle_admission::AdmissionError::ForeignManagedSource),D::RequestDecode,RootRequestEvent::Ingress),"framework.error");
        assert_eq!(code(&io::Error::other("Parser"),D::RequestDecode,RootRequestEvent::Ingress),"framework.error");
        assert_eq!(code(&F::Parser,D::RequestDecode,RootRequestEvent::Handler),"framework.error");
    }

    #[test]
    fn typed_written_source_preserves_chain_and_missing_writer_is_unconfirmed() {
        use saddle_core::{ContextLabel, DiagnosticStage, RequestViewPhase};
        let directory = std::env::temp_dir().join(format!("typed-written-{}-{}",std::process::id(),
            std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
        std::fs::create_dir(&directory).unwrap();
        let mut output = crate::EmergencyDiagnostics::start(&crate::FileLoggingConfig::new(directory.clone(),crate::Rotation::Daily)).unwrap();
        let handle = output.handle(); let label = ContextLabel::checked("typed-written").unwrap();
        let error = saddle_admission::InputDecodeFailure::Context { request_id:"original-request".into(),
            failure:Box::new(saddle_admission::InputDecodeFailure::Syntax(serde_json::from_str::<serde_json::Value>("{").unwrap_err())) };
        let description=error.to_string(); let debug=format!("{error:?}");
        let scope=UnrootedDiagnosticScope::new(&label,RequestViewPhase::Reading,Some(&handle));
        assert!(scope.source_error_recorded(&error,DiagnosticStage::RequestDecode,RootRequestEvent::Ingress,RootOutcomeFacts::default()).is_ok());
        let body=std::fs::read_to_string(output.target()).unwrap();
        assert!(body.contains("framework.input_decode.syntax"));
        assert_eq!(error.to_string(),description);assert_eq!(format!("{error:?}"),debug);
        assert!(body.contains("original-request"));assert_eq!(serde_json::from_str::<serde_json::Value>(body.lines().last().unwrap()).unwrap()["channel"],"terminal");
        let missing=UnrootedDiagnosticScope::new(&label,RequestViewPhase::Reading,None);
        assert!(missing.source_error_recorded(&error,DiagnosticStage::RequestDecode,RootRequestEvent::Ingress,RootOutcomeFacts::default()).is_err());
        let deadline=std::time::Instant::now()+std::time::Duration::from_secs(5);
        while output.shutdown()==crate::DiagnosticShutdown::Pending && std::time::Instant::now()<deadline { std::thread::yield_now(); }
        assert_eq!(output.shutdown(),crate::DiagnosticShutdown::Finished);
        drop(output);std::fs::remove_dir_all(directory).unwrap();
    }

    #[test]
    fn written_source_rejects_formatting_truncation_and_actual_file_failure() {
        #[derive(Debug)] struct Partial;
        impl fmt::Display for Partial { fn fmt(&self,f:&mut fmt::Formatter<'_>)->fmt::Result { f.write_str("partial-source")?; Err(fmt::Error) } }
        impl Error for Partial {}
        let directory=std::env::temp_dir().join(format!("typed-failed-{}-{}",std::process::id(),std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
        std::fs::create_dir(&directory).unwrap();
        let mut output=crate::EmergencyDiagnostics::start(&crate::FileLoggingConfig::new(directory.clone(),crate::Rotation::Daily)).unwrap();
        let handle=output.handle();let label=saddle_core::ContextLabel::checked("typed-failed").unwrap();
        let scope=UnrootedDiagnosticScope::new(&label,saddle_core::RequestViewPhase::Reading,Some(&handle));
        assert!(scope.source_error_recorded(&Partial,saddle_core::DiagnosticStage::RequestDecode,RootRequestEvent::Ingress,RootOutcomeFacts::default()).is_err());
        let deadline=std::time::Instant::now()+std::time::Duration::from_secs(5);
        while output.shutdown()==crate::DiagnosticShutdown::Pending && std::time::Instant::now()<deadline {std::thread::yield_now();}
        assert_eq!(output.shutdown(),crate::DiagnosticShutdown::Finished);
        // Finished drains the async queue; it does not revoke this live source handle.
        assert!(scope.source_error_recorded(&saddle_admission::InputDecodeFailure::Parser,saddle_core::DiagnosticStage::RequestDecode,RootRequestEvent::Ingress,RootOutcomeFacts::default()).is_ok());
        drop(output);std::fs::remove_dir_all(directory).unwrap();
        #[cfg(unix)] {
            let child=std::process::Command::new("sh")
                .args(["-c", "trap '' XFSZ; ulimit -f 1; exec \"$@\"", "source-file-limit"])
                .arg(std::env::current_exe().unwrap())
                .args(["root_diagnostic::original::content_tests::written_source_file_limit_child", "--exact", "--nocapture"])
                .env("SADDLE_SOURCE_LIMIT_CONTROL", "1").output().unwrap();
            assert!(child.status.success(), "{}", String::from_utf8_lossy(&child.stdout));
            assert!(String::from_utf8_lossy(&child.stdout).contains("actual file write failed: unconfirmed"));
        }
    }

    #[cfg(unix)]
    #[test]
    fn written_source_file_limit_child() {
        if std::env::var("SADDLE_SOURCE_LIMIT_CONTROL").as_deref()!=Ok("1") { return; }
        let directory=std::env::temp_dir().join(format!("source-limit-{}",std::process::id()));
        std::fs::create_dir(&directory).unwrap();
        let mut output=crate::EmergencyDiagnostics::start(&crate::FileLoggingConfig::new(directory.clone(),crate::Rotation::Daily)).unwrap();
        let handle=output.handle();let label=saddle_core::ContextLabel::checked("source-limit").unwrap();
        let scope=UnrootedDiagnosticScope::new(&label,saddle_core::RequestViewPhase::Reading,Some(&handle));
        let error=io::Error::other("x".repeat(16384));
        assert!(scope.source_error_recorded(&error,saddle_core::DiagnosticStage::RequestDecode,RootRequestEvent::Ingress,RootOutcomeFacts::default()).is_err());
        println!("actual file write failed: unconfirmed; original={}",output.target().display());
        let deadline=std::time::Instant::now()+std::time::Duration::from_secs(5);
        while output.shutdown()==crate::DiagnosticShutdown::Pending && std::time::Instant::now()<deadline {std::thread::yield_now();}
    }

    #[test]
    fn json_parser_debug_stream_matches_native_bytes() {
        for input in ["{broken", "[", "{\"a\":1,}", "\"\\uD800\""] {
            let error = serde_json::from_str::<serde_json::Value>(input).unwrap_err();
            let mut streamed = String::new();
            write_json_error_debug(&mut streamed, &error).unwrap();
            assert_eq!(streamed, format!("{error:?}"));
        }
    }

    #[test]
    fn owned_source_is_written_before_projection_and_mismatch_keeps_original() {
        use saddle_core::{CaptureSite, DiagnosticCause, DiagnosticStage, ErrorKind,
            RequestViewPhase, SaddleError};
        let directory = std::env::temp_dir().join(format!(
            "saddle-owned-source-{}-{}", std::process::id(),
            std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)
                .unwrap().as_nanos()));
        std::fs::create_dir(&directory).unwrap();
        let mut output = crate::EmergencyDiagnostics::start_checked(
            &crate::FileLoggingConfig::new(directory.clone(), crate::Rotation::Daily)).unwrap();
        let selected = output.source_output().expect("checked selected output");
        let label = saddle_core::ContextLabel::checked("owned-test").unwrap();
        let diagnostic = Diagnostic::capture(DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            DiagnosticCause::new(DiagnosticStage::StartupService,
                DiagnosticCode::new("owned.test").unwrap()));
        let target = output.target().to_owned();
        let written = super::RecordedSaddleError::capture_unrooted(
            io::Error::other("original before projection 9271"), diagnostic,
            &label, RequestViewPhase::Admitted, &selected,
            RootRequestEvent::Admission, RootOutcomeFacts::default(),
            |_, diagnostic| {
                assert!(std::fs::read_to_string(&target).unwrap()
                    .contains("original before projection 9271"));
                SaddleError::new(ErrorKind::Infrastructure, "owned.test", "safe")
                    .with_diagnostic(diagnostic)
            }).unwrap_or_else(|_| panic!("written and paired"));
        assert_eq!(written.safe().code(), "owned.test");
        let component = Diagnostic::capture(DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            DiagnosticCause::new(DiagnosticStage::ShutdownComponent,
                DiagnosticCode::new("owned.cleanup").unwrap()));
        let mismatch = super::RecordedSaddleError::capture_component(
            io::Error::other("owned cleanup original 9272"), component,
            saddle_core::ContextFact::Present(label.clone()), &selected,
            ComponentSourceKind::Cleanup,
            |_, _| SaddleError::new(ErrorKind::Infrastructure, "owned.cleanup", "safe")
                .with_diagnostic(Diagnostic::capture(DiagnosticCategory::UnexpectedError,
                    CaptureSite::FirstObserved,
                    DiagnosticCause::new(DiagnosticStage::ShutdownComponent,
                        DiagnosticCode::new("owned.cleanup").unwrap()))),
        ).err().expect("wrong occurrence rejected");
        assert!(mismatch.safe().source_unavailable());
        assert_eq!(mismatch.source().to_string(), "owned cleanup original 9272");
        assert!(std::fs::read_to_string(&target).unwrap()
            .contains("owned cleanup original 9272"));
        drop(written);
        drop(selected);
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        while output.shutdown() == crate::DiagnosticShutdown::Pending
            && std::time::Instant::now() < deadline { std::thread::yield_now(); }
        assert_eq!(output.shutdown(), crate::DiagnosticShutdown::Finished);
        drop(output);
        std::fs::remove_dir_all(directory).unwrap();
    }

    #[test]
    fn sealed_unrooted_failure_requires_written_matching_original() {
        use saddle_core::{CaptureSite, DiagnosticCause, DiagnosticStage, ErrorKind,
            RequestViewPhase, SaddleError};
        let directory = std::env::temp_dir().join(format!(
            "saddle-sealed-source-{}-{}", std::process::id(),
            std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)
                .unwrap().as_nanos()));
        std::fs::create_dir(&directory).unwrap();
        let mut output = crate::EmergencyDiagnostics::start(
            &crate::FileLoggingConfig::new(directory.clone(), crate::Rotation::Daily)).unwrap();
        let handle = output.handle();
        let label = saddle_core::ContextLabel::checked("sealed-test").unwrap();
        let diagnostic = Diagnostic::capture(DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            DiagnosticCause::new(DiagnosticStage::StartupService,
                DiagnosticCode::new("sealed.test").unwrap()));
        let original = io::Error::other("private original 1849");
        let scope = UnrootedDiagnosticScope::new(&label, RequestViewPhase::Admitted,
            Some(&handle));
        let written = scope.source_existing_error_recorded(&original, &diagnostic,
            RootRequestEvent::Admission, RootOutcomeFacts::default()).ok().expect("written");
        let safe = SaddleError::new(ErrorKind::Infrastructure, "sealed.test", "safe")
            .with_diagnostic(diagnostic);
        let recorded = super::RecordedSaddleError::from_written(safe, written).unwrap();
        assert!(!recorded.to_string().contains("private original 1849"));
        assert!(std::fs::read_to_string(output.target()).unwrap()
            .contains("private original 1849"));

        let other = Diagnostic::capture(DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            DiagnosticCause::new(DiagnosticStage::StartupService,
                DiagnosticCode::new("sealed.test").unwrap()));
        let mismatch = scope.source_existing_error_recorded(&original, &other,
            RootRequestEvent::Admission, RootOutcomeFacts::default()).ok().expect("written");
        let wrong = SaddleError::new(ErrorKind::Infrastructure, "sealed.test", "safe")
            .with_diagnostic(Diagnostic::capture(DiagnosticCategory::UnexpectedError,
                CaptureSite::FirstObserved,
                DiagnosticCause::new(DiagnosticStage::StartupService,
                    DiagnosticCode::new("sealed.test").unwrap())));
        assert!(super::RecordedSaddleError::from_written(wrong, mismatch)
            .unwrap_err().source_unavailable());
        let missing = UnrootedDiagnosticScope::new(&label, RequestViewPhase::Admitted, None);
        let no_output = Diagnostic::capture(DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            DiagnosticCause::new(DiagnosticStage::StartupService,
                DiagnosticCode::new("sealed.test").unwrap()));
        assert!(missing.source_existing_error_recorded(&original, &no_output,
            RootRequestEvent::Admission, RootOutcomeFacts::default()).is_err());
        let component = Diagnostic::capture(DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            DiagnosticCause::new(DiagnosticStage::StartupService,
                DiagnosticCode::new("sealed.component").unwrap()));
        let component_written = process_component_start_recorded(Some(&handle),
            saddle_core::ContextFact::Unavailable, &component, &original)
            .ok().expect("component written");
        let component_safe = SaddleError::new(ErrorKind::Infrastructure,
            "sealed.component", "safe").with_diagnostic(component);
        let component_recorded = super::RecordedSaddleError::from_component_written(
            component_safe, component_written).unwrap();
        assert_eq!(component_recorded.component_kind(), Some(ComponentSourceKind::Start));
        drop(scope);
        drop(recorded);
        drop(component_recorded);
        drop(handle);
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        while output.shutdown() == crate::DiagnosticShutdown::Pending
            && std::time::Instant::now() < deadline { std::thread::yield_now(); }
        assert_eq!(output.shutdown(), crate::DiagnosticShutdown::Finished);
        drop(output);
        std::fs::remove_dir_all(directory).unwrap();
    }

    struct Node {
        text: String,
        next: Option<Box<Node>>,
    }
    impl fmt::Display for Node {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.write_str(&self.text)
        }
    }
    impl fmt::Debug for Node {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.write_str(&self.text)
        }
    }
    impl Error for Node {
        fn source(&self) -> Option<&(dyn Error + 'static)> {
            self.next.as_deref().map(|v| v as &dyn Error)
        }
    }
    struct Readback<'a> {
        expected: &'a [String],
        unrooted: bool,
        // sequence, per-field offset, completed Display/Debug fields, terminal
        state: RefCell<(u64, usize, usize, bool)>,
    }
    impl OriginalOutput for Readback<'_> {
        fn submit(&self, segment: &Segment<'_>) -> DiagnosticSubmission {
            // Exercise real JSON escaping and decoding but not output capacity:
            // a full queue is a different contract from content truncation.
            let encoded = serde_json::to_vec(segment).unwrap();
            assert!(encoded.len() < 8192);
            let row: serde_json::Value = serde_json::from_slice(&encoded).unwrap();
            let mut s = self.state.borrow_mut();
            assert_eq!(row["sequence"].as_u64().unwrap(), s.0);
            s.0 += 1;
            if segment.channel == "context" && segment.state == "field_end" {
                let header: serde_json::Value = serde_json::from_str(segment.payload).unwrap();
                if self.unrooted { assert_eq!(header["context"]["schema_version"], 2); }
                else {
                    assert_eq!(header["context"]["schemaVersion"], "1.0");
                    assert_eq!(header["context"]["_meta"]["fieldStates"]["/requestInfo/requestId"], "not_established");
                }
            }
            if matches!(segment.channel, "description" | "debug") {
                let expected = &self.expected[segment.cause_depth];
                let payload = row["payload"].as_str().unwrap();
                assert_eq!(payload, &expected[s.1..s.1 + payload.len()]);
                s.1 += payload.len();
                if segment.state == "field_end" {
                    assert_eq!(s.1, expected.len(), "original tail missing");
                    s.1 = 0;
                    s.2 += 1;
                }
            }
            if segment.channel == "terminal" {
                assert_eq!(segment.state, "exposed_chain_complete");
                assert_eq!(segment.cause_depth, self.expected.len());
                assert_eq!(s.2, self.expected.len() * 2);
                s.3 = true;
            }
            DiagnosticSubmission::Enqueued
        }
    }
    fn assert_content(expected: Vec<String>) -> u64 {
        assert_content_context(expected, false)
    }
    fn assert_content_context(expected: Vec<String>, unrooted: bool) -> u64 {
        let mut head = None;
        for text in expected.iter().rev() {
            head = Some(Box::new(Node {
                text: text.clone(),
                next: head,
            }));
        }
        let publisher = saddle_core::RequestRootPublisher::create(
            saddle_core::ContextLabel::checked("content-test").unwrap(),
            saddle_core::ContextFact::NotEstablished,
        )
        .unwrap();
        let view = publisher
            .reference()
            .view(saddle_core::RequestLocalFacts::new(
                saddle_core::RequestViewPhase::Reading,
            ));
        let diagnostic = BoundedDiagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            saddle_core::CaptureSite::FirstObserved,
            saddle_core::BoundedDiagnosticCause::new(
                saddle_core::DiagnosticStage::RequestDecode,
                DiagnosticCode::new("content.original").unwrap(),
            ),
        );
        let output = Readback {
            expected: &expected,
            unrooted,
            state: RefCell::new((0, 0, 0, false)),
        };
        let application = saddle_core::ContextLabel::checked("content-test").unwrap();
        let result = capture(
            &output,
            diagnostic.occurrence(),
            RawSource::Error(head.as_deref().unwrap()),
            &Header {
                timestamp_unix_ms: 0,
                context: if unrooted {
                    OriginalContext::Unrooted(crate::AdmissionEventContext::Unrooted {
                        application: &application,
                        lifecycle: saddle_core::RequestViewPhase::Reading,
                    })
                } else {
                    OriginalContext::Request(&view)
                },
                stage: RootRequestEvent::Ingress,
                facts: SourceDetail::Bounded(&diagnostic),
                outcome: RootOutcomeFacts::default(),
                original_contract: "display-debug-source/v1",
                source_interface: "error_source",
                protocol: None,
                remote_target: None,
                grpc: None,
            },
        );
        assert_eq!(
            result,
            (
                DiagnosticSubmission::Enqueued,
                OriginalCaptureState::CompleteEnqueued
            )
        );
        assert!(output.state.borrow().3);
        output.state.into_inner().0
    }
    #[test]
    fn rooted_and_unrooted_share_lossless_long_content() {
        let expected = vec![
            format!("HEAD-中文\n\"{}\"-TAIL", "原文".repeat(2200)),
            "nested-cause\n完整".into(),
        ];
        let rooted = assert_content_context(expected.clone(), false);
        let unrooted = assert_content_context(expected, true);
        assert_eq!(rooted, unrooted);
    }
    #[test]
    fn finite_chain_beyond_old_256_keeps_every_cause() {
        let segments = assert_content(
            (0..300)
                .map(|n| format!("unknown-{n}-原文\n\"tail\""))
                .collect(),
        );
        assert!(segments > 600);
        println!("content readback: 300 original causes and Debug fields complete");
    }
    #[test]
    fn cycle_detector_handles_prefix_and_nontrivial_cycle() {
        let nodes: Vec<Node> = (0..4)
            .map(|n| Node {
                text: n.to_string(),
                next: None,
            })
            .collect();
        let mut cycle = CycleDetector::new(&nodes[0]);
        // A -> B -> C -> D -> B; no equal text or depth-limit heuristic.
        for node in [&nodes[1], &nodes[2], &nodes[3]] {
            assert!(!cycle.repeats(node));
        }
        let found = (0..32).any(|step| cycle.repeats(&nodes[1 + step % 3]));
        assert!(found);
    }
    #[test]
    fn finite_text_beyond_old_4096_segments_keeps_tail() {
        let text = format!("{}END-原文", "x".repeat(6144 * 4097));
        let segments = assert_content(vec![text]);
        assert!(segments > 8194);
        println!("content readback: {segments} segments, complete Display/Debug tails");
    }
}

/// Original database source outside a request; never manufactures a trace or
/// request account. `maintenance` distinguishes a tracked sweep from unscoped
/// startup/cleanup where request association is unavailable.
#[doc(hidden)]
pub fn database_background_error(
    output: Option<&EmergencyDiagnosticHandle>,
    datasource: u64,
    connection: Option<u64>,
    sweep: u64,
    phase: &'static str,
    maintenance: bool,
    error: &(dyn Error + 'static),
    facts: BoundedDiagnostic,
) -> UnrootedCaptureFacts {
    #[derive(Serialize)]
    struct Context {
        schema_version: u8,
        datasource: u64,
        connection: ContextFact<u64>,
        maintenance_sweep: u64,
        maintenance_phase: &'static str,
        request: ContextFact<()>,
        trace_id: ContextFact<()>,
        application: ContextFact<()>,
    }
    #[derive(Serialize)]
    struct BackgroundHeader<'a> {
        timestamp_unix_ms: u128,
        context: Context,
        stage: &'static str,
        facts: &'a BoundedDiagnostic,
        original_contract: &'static str,
        source_interface: &'static str,
    }
    let occurrence = facts.occurrence();
    let missing = if maintenance { ContextFact::NotApplicable } else { ContextFact::Unavailable };
    let header = BackgroundHeader {
        timestamp_unix_ms: timestamp(),
        context: Context { schema_version: 2, datasource,
            connection: connection.map(ContextFact::Present).unwrap_or(ContextFact::NotEstablished),
            maintenance_sweep: sweep, maintenance_phase: phase,
            request: missing, trace_id: missing, application: ContextFact::Unavailable },
        stage: if maintenance { "database_maintenance" } else { "database_connection" }, facts: &facts,
        original_contract: "display-debug-source/v1", source_interface: "error_source",
    };
    let (submission, original_capture) = match output {
        Some(output) => {
            let (submission, state) = capture(&ConfirmedOutput(output), occurrence, RawSource::Error(error), &header);
            if submission == DiagnosticSubmission::Enqueued && state == OriginalCaptureState::CompleteEnqueued {
                (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten)
            } else {
                (submission, state)
            }
        },
        None => (DiagnosticSubmission::OutputUnavailable, OriginalCaptureState::OutputUnavailable),
    };
    UnrootedCaptureFacts { occurrence, submission, original_capture }
}

/// Preserve a transaction error after its request task has handed cleanup to
/// the database worker. The retained call supplies real correlation, while a
/// request identity absent from that call remains explicitly unavailable.
#[doc(hidden)]
pub fn database_transaction_error(
    output: Option<&EmergencyDiagnosticHandle>,
    call: &saddle_core::CallContext,
    phase: &'static str,
    error: &(dyn Error + 'static),
    facts: &Diagnostic,
) -> UnrootedCaptureFacts {
    #[derive(Serialize)]
    struct Context<'a> {
        schema_version: u8,
        application: &'a str,
        module: &'a str,
        service: &'a str,
        operation: &'a str,
        trace_id: String,
        span_id: String,
        request: ContextFact<()>,
        transaction_phase: &'static str,
    }
    #[derive(Serialize)]
    struct Header<'a> {
        timestamp_unix_ms: u128,
        context: Context<'a>,
        stage: &'static str,
        facts: &'a Diagnostic,
        original_contract: &'static str,
        source_interface: &'static str,
    }
    let occurrence = facts.occurrence();
    let header = Header {
        timestamp_unix_ms: timestamp(),
        context: Context {
            schema_version: 2,
            application: call.application().as_str(),
            module: call.module().as_str(),
            service: call.service().as_str(),
            operation: call.operation().as_str(),
            trace_id: call.trace_id().to_string(),
            span_id: call.span_id().to_string(),
            request: ContextFact::Unavailable,
            transaction_phase: phase,
        },
        stage: if phase == "begin" || phase == "commit" { "database_transaction" }
            else { "database_transaction_cleanup" },
        facts,
        original_contract: "display-debug-source/v1",
        source_interface: "error_source",
    };
    let (submission, original_capture) = match output {
        Some(output) => {
            let (submission, state) = capture(
                &ConfirmedOutput(output), occurrence, RawSource::Error(error), &header);
            if submission == DiagnosticSubmission::Enqueued
                && state == OriginalCaptureState::CompleteEnqueued {
                (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten)
            } else {
                (submission, state)
            }
        }
        None => (DiagnosticSubmission::OutputUnavailable, OriginalCaptureState::OutputUnavailable),
    };
    UnrootedCaptureFacts { occurrence, submission, original_capture }
}

/// First reception of a SQLx error in the legacy direct query/write surface.
/// The current child call provides actual correlation; a request identity is
/// unavailable in this call and is never inferred from its trace identifier.
#[doc(hidden)]
pub fn database_operation_error(
    output: Option<&EmergencyDiagnosticHandle>,
    call: &saddle_core::CallContext,
    phase: &'static str,
    error: &(dyn Error + 'static),
    diagnostic: &Diagnostic,
) -> UnrootedCaptureFacts {
    database_operation_source(output, call, phase, RawSource::Error(error), diagnostic, diagnostic.occurrence(), false)
}

/// Borrow raw statement construction inputs through the synchronous original
/// writer without cloning unbounded SQL or parameter bytes.
#[doc(hidden)]
pub fn database_operation_borrowed_error<D: Error>(
    output: Option<&EmergencyDiagnosticHandle>,
    call: &saddle_core::CallContext,
    phase: &'static str,
    source: &D,
    diagnostic: &Diagnostic,
) -> UnrootedCaptureFacts {
    database_operation_source(output, call, phase,
        RawSource::BorrowedError(source), diagnostic, diagnostic.occurrence(), false)
}

/// Bounded request-side source capture: the diagnostic owns fixed fields only.
#[doc(hidden)]
pub fn database_operation_borrowed_input_bounded<D: Error>(
    output: Option<&EmergencyDiagnosticHandle>,
    call: &saddle_core::CallContext,
    phase: &'static str,
    source: &D,
    diagnostic: &BoundedDiagnostic,
) -> UnrootedCaptureFacts {
    database_operation_source(output, call, phase,
        RawSource::BorrowedError(source), diagnostic, diagnostic.occurrence(), true)
}

fn database_operation_source<D: Serialize>(
    output: Option<&EmergencyDiagnosticHandle>,
    call: &saddle_core::CallContext,
    phase: &'static str,
    source: RawSource<'_>,
    diagnostic: &D,
    occurrence: DiagnosticOccurrence,
    bounded_input: bool,
) -> UnrootedCaptureFacts {
    struct DisplayId<T>(T);
    impl<T: fmt::Display> Serialize for DisplayId<T> {
        fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
            serializer.collect_str(&self.0)
        }
    }
    #[derive(Serialize)]
    struct Context<'a> {
        schema_version: u8,
        application: &'a str,
        module: &'a str,
        service: &'a str,
        operation: &'a str,
        trace_id: DisplayId<saddle_core::TraceId>,
        trace_correlation_id: &'a str,
        span_id: DisplayId<saddle_core::SpanId>,
        rpc_correlation_id: Option<&'a str>,
        request: ContextFact<()>,
        operation_phase: &'static str,
    }
    #[derive(Serialize)]
    struct Header<'a, D> {
        timestamp_unix_ms: u128,
        context: Context<'a>,
        stage: &'static str,
        facts: &'a D,
        original_contract: &'static str,
        source_interface: &'static str,
    }
    let header = Header {
        timestamp_unix_ms: timestamp(),
        context: Context {
            schema_version: 2,
            application: call.application().as_str(),
            module: call.module().as_str(),
            service: call.service().as_str(),
            operation: call.operation().as_str(),
            trace_id: DisplayId(call.trace_id()),
            trace_correlation_id: call.trace_correlation_id().as_str(),
            span_id: DisplayId(call.span_id()),
            rpc_correlation_id: call.rpc_correlation_id().map(|id| id.as_str()),
            request: ContextFact::Unavailable,
            operation_phase: phase,
        },
        stage: "database_operation",
        facts: diagnostic,
        original_contract: "display-debug-source/v1",
        source_interface: match &source {
            RawSource::Error(_) | RawSource::BorrowedError(_) => "error_source",
            RawSource::Description(_, _) => "description_debug_only",
        },
    };
    let (submission, original_capture) = match output {
        Some(output) => {
            let (submission, state) = capture_with_mode(
                &ConfirmedOutput(output), occurrence, source, &header, bounded_input);
            if submission == DiagnosticSubmission::Enqueued
                && state == OriginalCaptureState::CompleteEnqueued {
                (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten)
            } else { (submission, state) }
        }
        None => (DiagnosticSubmission::OutputUnavailable, OriginalCaptureState::OutputUnavailable),
    };
    UnrootedCaptureFacts { occurrence, submission, original_capture }
}

/// A listener bind failure occurs before any request or task root exists.
/// Preserve its complete I/O source without inventing a request identity.
#[doc(hidden)]
#[track_caller]
pub fn process_startup_error(
    output: Option<&EmergencyDiagnosticHandle>,
    application: &saddle_core::ContextLabel,
    key: &'static str,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> UnrootedCaptureFacts {
    #[derive(Serialize)]
    struct StartupContext<'a> {
        schema_version: u8,
        application: &'a saddle_core::ContextLabel,
        lifecycle: &'static str,
        config_key: &'static str,
        request: ContextFact<()>,
        trace_id: ContextFact<()>,
    }
    #[derive(Serialize)]
    struct StartupHeader<'a> {
        timestamp_unix_ms: u128,
        context: StartupContext<'a>,
        stage: &'static str,
        facts: &'a Diagnostic,
        original_contract: &'static str,
        source_interface: &'static str,
    }
    let occurrence = diagnostic.occurrence();
    let header = StartupHeader {
        timestamp_unix_ms: timestamp(),
        context: StartupContext { schema_version: 2, application, lifecycle: "startup",
            config_key: key, request: ContextFact::NotEstablished,
            trace_id: ContextFact::NotEstablished },
        stage: "startup_listener", facts: diagnostic,
        original_contract: "display-debug-source/v1", source_interface: "error_source",
    };
    let (submission, original_capture) = match output {
        Some(output) => {
            let (submission, state) = capture(&ConfirmedOutput(output), occurrence, RawSource::Error(error), &header);
            if submission == DiagnosticSubmission::Enqueued && state == OriginalCaptureState::CompleteEnqueued {
                (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten)
            } else { (submission, state) }
        }
        None => (DiagnosticSubmission::OutputUnavailable, OriginalCaptureState::OutputUnavailable),
    };
    UnrootedCaptureFacts { occurrence, submission, original_capture }
}

/// Configuration parsing precedes establishment of an application identity.
/// Record the parser's exposed source before converting it to a safe error.
#[doc(hidden)]
#[track_caller]
pub fn process_config_error(
    output: Option<&EmergencyDiagnosticHandle>,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> UnrootedCaptureFacts {
    match output {
        Some(output) => capture_config(&ConfirmedOutput(output), diagnostic, error),
        None => UnrootedCaptureFacts {
            occurrence: diagnostic.occurrence(),
            submission: DiagnosticSubmission::OutputUnavailable,
            original_capture: OriginalCaptureState::OutputUnavailable,
        },
    }
}

/// Configuration-independent source output, before parameters or config exist.
#[doc(hidden)]
#[track_caller]
pub fn process_bootstrap_error(
    output: &crate::BootstrapDiagnostics,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> UnrootedCaptureFacts {
    capture_config(output, diagnostic, error)
}

/// Service registration can precede Observer construction. Both variants use
/// an already selected controlled file; the bootstrap variant adds no worker.
#[derive(Clone, Copy)]
pub enum ServiceRegistrySourceOutput<'a> {
    Bootstrap(&'a crate::BootstrapDiagnostics),
    Emergency(&'a EmergencyDiagnosticHandle),
}

#[track_caller]
pub fn service_registry_source_error(
    output: Option<ServiceRegistrySourceOutput<'_>>,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
    lifecycle: &'static str,
) -> UnrootedCaptureFacts {
    service_registry_source(output, diagnostic, RawSource::Error(error), lifecycle)
}

/// Production Service facts are built before the process writer starts. The
/// already selected bootstrap file confirms the source before this assembly
/// failure can cross into the Facade's safe startup response.
#[track_caller]
pub fn service_production_fact_source_error(
    output: &crate::BootstrapDiagnostics,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
    application: &saddle_core::ApplicationId,
) -> UnrootedCaptureFacts {
    service_compiled_build_source_error(output, diagnostic, error, application, "service_production_fact")
}

/// Synchronously confirms an original Service build failure while generated
/// code still owns the exact error object and the Facade's startup file.
#[track_caller]
pub fn service_compiled_build_source_error(
    output: &crate::BootstrapDiagnostics,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
    application: &saddle_core::ApplicationId,
    source_stage: &'static str,
) -> UnrootedCaptureFacts {
    #[derive(Serialize)]
    struct FactContext<'a> {
        schema_version: u8,
        application: &'a str,
        lifecycle: &'static str,
        request: ContextFact<()>,
        trace_id: ContextFact<()>,
    }
    #[derive(Serialize)]
    struct FactHeader<'a> {
        timestamp_unix_ms: u128,
        context: FactContext<'a>,
        stage: &'static str,
        facts: &'a Diagnostic,
        original_contract: &'static str,
        source_interface: &'static str,
    }
    let occurrence = diagnostic.occurrence();
    let header = FactHeader {
        timestamp_unix_ms: timestamp(),
        context: FactContext {
            schema_version: 2, application: application.as_str(), lifecycle: "startup",
            request: ContextFact::NotEstablished, trace_id: ContextFact::NotEstablished,
        },
        stage: source_stage, facts: diagnostic,
        original_contract: "display-debug-source/v1", source_interface: "error_source",
    };
    let (submission, original_capture) = capture(output, occurrence, RawSource::Error(error), &header);
    let (submission, original_capture) = match (submission, original_capture) {
        (DiagnosticSubmission::Enqueued, OriginalCaptureState::CompleteEnqueued) =>
            (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten),
        facts => facts,
    };
    UnrootedCaptureFacts { occurrence, submission, original_capture }
}

/// Records the actual facts exposed by an opaque registry value. `Any` has no
/// Error interface, so this record explicitly ends without a source chain.
#[track_caller]
pub fn service_registry_source_description(
    output: Option<ServiceRegistrySourceOutput<'_>>,
    diagnostic: &Diagnostic,
    description: &(impl fmt::Display + fmt::Debug),
    lifecycle: &'static str,
) -> UnrootedCaptureFacts {
    service_registry_source(output, diagnostic, RawSource::Description(description, description), lifecycle)
}

/// A Service route is registered before a request exists. The selected
/// emergency file still records its actual application and route facts.
#[track_caller]
pub fn service_route_source_description(
    output: Option<&EmergencyDiagnosticHandle>,
    diagnostic: &Diagnostic,
    description: &(impl fmt::Display + fmt::Debug),
    application: &saddle_core::ApplicationId,
    route: &str,
) -> UnrootedCaptureFacts {
    service_entry_source(output, diagnostic, RawSource::Description(description, description),
        application, route, None, "startup", "service_route_registration")
}

/// Trace validation fails before a request root exists. Preserve the actual
/// parser error and supplied identifier without inventing a trace identity.
#[track_caller]
pub fn service_entry_trace_source_error(
    output: Option<&EmergencyDiagnosticHandle>,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
    application: &saddle_core::ApplicationId,
    route: &str,
    inbound_trace_id: &str,
) -> UnrootedCaptureFacts {
    service_entry_source(output, diagnostic, RawSource::Error(error), application, route,
        Some(inbound_trace_id), "request", "service_trace_validation")
}

#[track_caller]
fn service_entry_source(
    output: Option<&EmergencyDiagnosticHandle>,
    diagnostic: &Diagnostic,
    raw: RawSource<'_>,
    application: &saddle_core::ApplicationId,
    route: &str,
    inbound_trace_id: Option<&str>,
    lifecycle: &'static str,
    stage: &'static str,
) -> UnrootedCaptureFacts {
    #[derive(Serialize)]
    struct RouteContext<'a> {
        schema_version: u8,
        application: &'a str,
        route: &'a str,
        inbound_trace_id: ContextFact<&'a str>,
        lifecycle: &'static str,
        request: ContextFact<()>,
        trace_id: ContextFact<()>,
    }
    #[derive(Serialize)]
    struct RouteHeader<'a> {
        timestamp_unix_ms: u128,
        context: RouteContext<'a>,
        stage: &'static str,
        facts: &'a Diagnostic,
        original_contract: &'static str,
        source_interface: &'static str,
    }
    let occurrence = diagnostic.occurrence();
    let header = RouteHeader {
        timestamp_unix_ms: timestamp(),
        context: RouteContext {
            schema_version: 2, application: application.as_str(), route,
            inbound_trace_id: inbound_trace_id.map(ContextFact::Present)
                .unwrap_or(ContextFact::NotEstablished),
            lifecycle, request: ContextFact::NotEstablished,
            trace_id: ContextFact::NotEstablished,
        },
        stage, facts: diagnostic,
        original_contract: "display-debug-source/v1",
        source_interface: match raw {
            RawSource::Error(_) | RawSource::BorrowedError(_) => "error_source",
            RawSource::Description(_, _) => "unavailable_description_debug_only",
        },
    };
    let (submission, original_capture) = match output {
        Some(output) => {
            let (submission, capture_state) = capture(&ConfirmedOutput(output), occurrence,
                raw, &header);
            match (submission, capture_state) {
                (DiagnosticSubmission::Enqueued, OriginalCaptureState::CompleteEnqueued) =>
                    (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten),
                facts => facts,
            }
        }
        None => (DiagnosticSubmission::OutputUnavailable, OriginalCaptureState::OutputUnavailable),
    };
    UnrootedCaptureFacts { occurrence, submission, original_capture }
}

#[track_caller]
fn service_registry_source(
    output: Option<ServiceRegistrySourceOutput<'_>>,
    diagnostic: &Diagnostic,
    raw: RawSource<'_>,
    lifecycle: &'static str,
) -> UnrootedCaptureFacts {
    #[derive(Serialize)]
    struct RegistryContext {
        schema_version: u8,
        application: ContextFact<()>,
        lifecycle: &'static str,
        request: ContextFact<()>,
        trace_id: ContextFact<()>,
    }
    #[derive(Serialize)]
    struct RegistryHeader<'a> {
        timestamp_unix_ms: u128,
        context: RegistryContext,
        stage: &'static str,
        facts: &'a Diagnostic,
        original_contract: &'static str,
        source_interface: &'static str,
    }
    let occurrence = diagnostic.occurrence();
    let header = RegistryHeader {
        timestamp_unix_ms: timestamp(),
        context: RegistryContext {
            schema_version: 2,
            application: if lifecycle == "startup" { ContextFact::NotEstablished }
                else { ContextFact::Unavailable },
            lifecycle,
            request: if lifecycle == "startup" { ContextFact::NotEstablished }
                else { ContextFact::Unavailable },
            trace_id: if lifecycle == "startup" { ContextFact::NotEstablished }
                else { ContextFact::Unavailable },
        },
        stage: "service_registry", facts: diagnostic,
        original_contract: "display-debug-source/v1", source_interface: match raw {
            RawSource::Error(_) | RawSource::BorrowedError(_) => "error_source",
            RawSource::Description(_, _) => "unavailable_description_debug_only",
        },
    };
    let result = match output {
        Some(ServiceRegistrySourceOutput::Bootstrap(output)) =>
            Some(capture(output, occurrence, raw, &header)),
        Some(ServiceRegistrySourceOutput::Emergency(output)) =>
            Some(capture(&ConfirmedOutput(output), occurrence, raw, &header)),
        None => None,
    };
    let (submission, original_capture) = match result {
        Some((DiagnosticSubmission::Enqueued, OriginalCaptureState::CompleteEnqueued)) =>
            (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten),
        Some(result) => result,
        None => (DiagnosticSubmission::OutputUnavailable, OriginalCaptureState::OutputUnavailable),
    };
    UnrootedCaptureFacts { occurrence, submission, original_capture }
}

/// The legacy Service JSON boundary has a real CallContext even though it has
/// no reserved request view. Borrow every available identity while the first
/// external error and the one selected emergency writer are still alive.
#[track_caller]
pub fn service_request_source_error(
    output: Option<&EmergencyDiagnosticHandle>,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
    call: &saddle_core::CallContext,
    stage: &'static str,
) -> UnrootedCaptureFacts {
    service_request_source(output, diagnostic, RawSource::Error(error), call, stage)
}

#[track_caller]
pub fn service_request_source_description(
    output: Option<&EmergencyDiagnosticHandle>,
    diagnostic: &Diagnostic,
    description: &(impl fmt::Display + fmt::Debug),
    call: &saddle_core::CallContext,
    stage: &'static str,
) -> UnrootedCaptureFacts {
    service_request_source(output, diagnostic, RawSource::Description(description, description), call, stage)
}

/// Runtime records the actual atomic admission decision before it projects
/// a public rejection. The call is already established, but no request has
/// been admitted or assigned a request identity.
#[track_caller]
pub fn runtime_admission_source_description(
    output: Option<&EmergencyDiagnosticHandle>,
    diagnostic: &Diagnostic,
    description: &(impl fmt::Display + fmt::Debug),
    call: &saddle_core::CallContext,
) -> UnrootedCaptureFacts {
    service_request_source(output, diagnostic, RawSource::Description(description, description),
        call, "runtime_request_admission")
}

/// Internal Runtime admission paths can reject before any call or request
/// exists. The application binding is real; later identities are absent.
#[track_caller]
pub fn runtime_unrooted_admission_source_description(
    output: Option<&EmergencyDiagnosticHandle>,
    diagnostic: &Diagnostic,
    description: &(impl fmt::Display + fmt::Debug),
    application: Option<&saddle_core::ApplicationId>,
) -> UnrootedCaptureFacts {
    #[derive(Serialize)]
    struct AdmissionContext<'a> {
        schema_version: u8,
        application: ContextFact<&'a str>,
        lifecycle: &'static str,
        request: ContextFact<()>,
        trace_id: ContextFact<()>,
    }
    #[derive(Serialize)]
    struct AdmissionHeader<'a> {
        timestamp_unix_ms: u128,
        context: AdmissionContext<'a>,
        stage: &'static str,
        facts: &'a Diagnostic,
        original_contract: &'static str,
        source_interface: &'static str,
    }
    let occurrence = diagnostic.occurrence();
    let header = AdmissionHeader {
        timestamp_unix_ms: timestamp(),
        context: AdmissionContext {
            schema_version: 2,
            application: application.map(|id| ContextFact::Present(id.as_str()))
                .unwrap_or(ContextFact::Unavailable),
            lifecycle: "request", request: ContextFact::NotEstablished,
            trace_id: ContextFact::NotEstablished,
        },
        stage: "runtime_unrooted_admission", facts: diagnostic,
        original_contract: "display-debug-source/v1",
        source_interface: "unavailable_description_debug_only",
    };
    let (submission, original_capture) = match output {
        Some(output) => {
            let (submission, state) = capture(&ConfirmedOutput(output), occurrence,
                RawSource::Description(description, description), &header);
            if submission == DiagnosticSubmission::Enqueued
                && state == OriginalCaptureState::CompleteEnqueued {
                (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten)
            } else { (submission, state) }
        }
        None => (DiagnosticSubmission::OutputUnavailable, OriginalCaptureState::OutputUnavailable),
    };
    UnrootedCaptureFacts { occurrence, submission, original_capture }
}

#[track_caller]
fn service_request_source(
    output: Option<&EmergencyDiagnosticHandle>,
    diagnostic: &Diagnostic,
    raw: RawSource<'_>,
    call: &saddle_core::CallContext,
    stage: &'static str,
) -> UnrootedCaptureFacts {
    #[derive(Serialize)]
    struct ServiceContext<'a> {
        schema_version: u8,
        application: &'a str,
        module: &'a str,
        service: &'a str,
        operation: &'a str,
        trace_id: u128,
        trace_correlation_id: &'a str,
        span_id: u64,
        rpc_correlation_id: Option<&'a str>,
        request: ContextFact<()>,
        lifecycle: &'static str,
    }
    #[derive(Serialize)]
    struct ServiceHeader<'a> {
        timestamp_unix_ms: u128,
        context: ServiceContext<'a>,
        stage: &'static str,
        facts: &'a Diagnostic,
        original_contract: &'static str,
        source_interface: &'static str,
    }
    let occurrence = diagnostic.occurrence();
    let header = ServiceHeader {
        timestamp_unix_ms: timestamp(),
        context: ServiceContext {
            schema_version: 2,
            application: call.application().as_str(),
            module: call.module().as_str(),
            service: call.service().as_str(),
            operation: call.operation().as_str(),
            trace_id: call.trace_id().as_u128(),
            trace_correlation_id: call.trace_correlation_id().as_str(),
            span_id: call.span_id().as_u64(),
            rpc_correlation_id: call.rpc_correlation_id().map(|id| id.as_str()),
            request: ContextFact::Unavailable,
            lifecycle: "request",
        },
        stage, facts: diagnostic,
        original_contract: "display-debug-source/v1", source_interface: match raw {
            RawSource::Error(_) | RawSource::BorrowedError(_) => "error_source",
            RawSource::Description(_, _) => "unavailable_description_debug_only",
        },
    };
    let (submission, original_capture) = match output {
        Some(output) => {
            let (submission, state) = capture(
                &ConfirmedOutput(output), occurrence, raw, &header,
            );
            if submission == DiagnosticSubmission::Enqueued
                && state == OriginalCaptureState::CompleteEnqueued {
                (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten)
            } else { (submission, state) }
        }
        None => (DiagnosticSubmission::OutputUnavailable, OriginalCaptureState::OutputUnavailable),
    };
    UnrootedCaptureFacts { occurrence, submission, original_capture }
}

fn capture_config<O: OriginalOutput>(
    output: &O,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> UnrootedCaptureFacts {
    #[derive(Serialize)]
    struct ConfigContext {
        schema_version: u8,
        application: ContextFact<()>,
        lifecycle: &'static str,
        request: ContextFact<()>,
        trace_id: ContextFact<()>,
    }
    #[derive(Serialize)]
    struct ConfigHeader<'a> {
        timestamp_unix_ms: u128,
        context: ConfigContext,
        stage: &'static str,
        facts: &'a Diagnostic,
        original_contract: &'static str,
        source_interface: &'static str,
    }
    let occurrence = diagnostic.occurrence();
    let header = ConfigHeader {
        timestamp_unix_ms: timestamp(),
        context: ConfigContext { schema_version: 2, application: ContextFact::NotEstablished,
            lifecycle: "startup", request: ContextFact::NotEstablished,
            trace_id: ContextFact::NotEstablished },
        stage: "startup_config", facts: diagnostic,
        original_contract: "display-debug-source/v1", source_interface: "error_source",
    };
    let (submission, state) = capture(output, occurrence, RawSource::Error(error), &header);
    let (submission, original_capture) = if submission == DiagnosticSubmission::Enqueued
        && state == OriginalCaptureState::CompleteEnqueued {
        (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten)
    } else { (submission, state) };
    UnrootedCaptureFacts { occurrence, submission, original_capture }
}

/// Record a process resource check while Runtime still owns the output, after
/// request/task cleanup and before diagnostic shutdown. It has no request root.
#[doc(hidden)]
pub fn process_finalization_error(
    output: Option<&EmergencyDiagnosticHandle>,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> UnrootedCaptureFacts {
    process_lifecycle_error(output, ContextFact::Unavailable, "shutdown", "process_finalization", diagnostic, error)
}

/// Record a component startup failure before it becomes a safe projection.
#[doc(hidden)]
pub fn process_component_start_error(
    output: Option<&EmergencyDiagnosticHandle>,
    application: ContextFact<saddle_core::ContextLabel>,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> ComponentSourceReceipt {
    ComponentSourceReceipt {
        facts: process_lifecycle_error(output, application, "startup", "component_start", diagnostic, error),
        kind: ComponentSourceKind::Start,
    }
}

#[doc(hidden)]
pub fn process_component_start_recorded(
    output: Option<&EmergencyDiagnosticHandle>,
    application: ContextFact<saddle_core::ContextLabel>,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> Result<WrittenComponentFailure, ComponentSourceReceipt> {
    WrittenComponentFailure::from_receipt(process_component_start_error(
        output, application, diagnostic, error))
}

/// Record a component shutdown failure while the Application owns the output.
#[doc(hidden)]
pub fn process_component_cleanup_error(
    output: Option<&EmergencyDiagnosticHandle>,
    application: ContextFact<saddle_core::ContextLabel>,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> ComponentSourceReceipt {
    ComponentSourceReceipt {
        facts: process_lifecycle_error(output, application, "shutdown", "component_cleanup", diagnostic, error),
        kind: ComponentSourceKind::Cleanup,
    }
}

#[doc(hidden)]
pub fn process_component_cleanup_recorded(
    output: Option<&EmergencyDiagnosticHandle>,
    application: ContextFact<saddle_core::ContextLabel>,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> Result<WrittenComponentFailure, ComponentSourceReceipt> {
    WrittenComponentFailure::from_receipt(process_component_cleanup_error(
        output, application, diagnostic, error))
}

/// Record a managed shutdown signal failure at its Application receipt.
#[doc(hidden)]
pub fn process_shutdown_signal_error(
    output: Option<&EmergencyDiagnosticHandle>,
    application: ContextFact<saddle_core::ContextLabel>,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> UnrootedCaptureFacts {
    process_lifecycle_error(output, application, "shutdown", "shutdown_signal", diagnostic, error)
}

/// Record an Application request drain failure before it is classified.
#[doc(hidden)]
pub fn process_request_drain_error(
    output: Option<&EmergencyDiagnosticHandle>,
    application: ContextFact<saddle_core::ContextLabel>,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> UnrootedCaptureFacts {
    process_lifecycle_error(output, application, "shutdown", "request_drain", diagnostic, error)
}

/// Record the finalizer's first result receipt before process output closes.
#[doc(hidden)]
pub fn process_post_driver_error(
    output: Option<&EmergencyDiagnosticHandle>,
    application: ContextFact<saddle_core::ContextLabel>,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> UnrootedCaptureFacts {
    process_lifecycle_error(output, application, "shutdown", "post_driver", diagnostic, error)
}

fn process_lifecycle_error(
    output: Option<&EmergencyDiagnosticHandle>,
    application: ContextFact<saddle_core::ContextLabel>,
    lifecycle: &'static str,
    stage: &'static str,
    diagnostic: &Diagnostic,
    error: &(dyn Error + 'static),
) -> UnrootedCaptureFacts {
    #[derive(Serialize)]
    struct FinalizationContext {
        schema_version: u8,
        application: ContextFact<saddle_core::ContextLabel>,
        lifecycle: &'static str,
        request: ContextFact<()>,
        trace_id: ContextFact<()>,
    }
    #[derive(Serialize)]
    struct FinalizationHeader<'a> {
        timestamp_unix_ms: u128,
        context: FinalizationContext,
        stage: &'static str,
        facts: &'a Diagnostic,
        original_contract: &'static str,
        source_interface: &'static str,
    }
    let occurrence = diagnostic.occurrence();
    let header = FinalizationHeader {
        timestamp_unix_ms: timestamp(),
        context: FinalizationContext { schema_version: 2,
            application, lifecycle,
            request: if lifecycle == "startup" { ContextFact::NotEstablished } else { ContextFact::NotApplicable },
            trace_id: if lifecycle == "startup" { ContextFact::NotEstablished } else { ContextFact::NotApplicable } },
        stage, facts: diagnostic,
        original_contract: "display-debug-source/v1", source_interface: "error_source",
    };
    let (submission, original_capture) = match output {
        Some(output) => {
            let (submission, state) = capture(&ConfirmedOutput(output), occurrence, RawSource::Error(error), &header);
            if submission == DiagnosticSubmission::Enqueued && state == OriginalCaptureState::CompleteEnqueued {
                (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten)
            } else { (submission, state) }
        }
        None => (DiagnosticSubmission::OutputUnavailable, OriginalCaptureState::OutputUnavailable),
    };
    UnrootedCaptureFacts { occurrence, submission, original_capture }
}

/// The account may already be closed. The fixed snapshot grants no authority
/// and this capture uses only the startup-selected output's own storage.
pub fn request_terminal_audit_error(
    output: Option<&EmergencyDiagnosticHandle>,
    context: Option<&saddle_core::request_context::RequestContextSnapshot>,
    diagnostic: &BoundedDiagnostic,
    source_occurrence: Option<DiagnosticOccurrence>,
    failure: &saddle_admission::ProfuseGwTerminalAuditFailure,
) -> UnrootedCaptureFacts {
    #[derive(Serialize)]
    struct AuditHeader<'a> {
        timestamp_unix_ms: u128,
        context: Option<&'a saddle_core::request_context::RequestContextSnapshot>,
        stage: &'static str,
        facts: &'a BoundedDiagnostic,
        source_occurrence: Option<DiagnosticOccurrence>,
        audit: Option<saddle_admission::RequestAuditSnapshot>,
        database: saddle_admission::ProfuseGwDatabaseDisposition,
        original_contract: &'static str,
        source_interface: &'static str,
    }
    let occurrence = diagnostic.occurrence();
    let header = AuditHeader {
        timestamp_unix_ms: timestamp(), context, stage: "request_terminal_audit",
        facts: diagnostic, source_occurrence, audit: failure.audit, database: failure.database,
        original_contract: "display-debug-source/v1", source_interface: "error_source",
    };
    let (submission, original_capture) = match output {
        Some(output) => {
            let (submission, state) = capture(&ConfirmedOutput(output), occurrence, RawSource::Error(failure), &header);
            if submission == DiagnosticSubmission::Enqueued && state == OriginalCaptureState::CompleteEnqueued {
                (DiagnosticSubmission::Written, OriginalCaptureState::CompleteWritten)
            } else { (submission, state) }
        }
        None => (DiagnosticSubmission::OutputUnavailable, OriginalCaptureState::OutputUnavailable),
    };
    UnrootedCaptureFacts { occurrence, submission, original_capture }
}

pub fn request_terminal_audit_disposition(
    output: Option<&EmergencyDiagnosticHandle>,
    context: Option<&saddle_core::request_context::RequestContextSnapshot>,
    source_occurrence: Option<DiagnosticOccurrence>,
    captured: &UnrootedCaptureFacts,
    audit: Option<saddle_admission::RequestAuditSnapshot>,
) -> DiagnosticSubmission {
    #[derive(Serialize)]
    struct Record<'a> {
        schema_version: u8,
        timestamp_unix_ms: u128,
        event: &'static str,
        context: Option<&'a saddle_core::request_context::RequestContextSnapshot>,
        occurrence: DiagnosticOccurrence,
        source_occurrence: Option<DiagnosticOccurrence>,
        source_submission: DiagnosticSubmission,
        original_capture: OriginalCaptureState,
        disposition: &'static str,
        audit: Option<saddle_admission::RequestAuditSnapshot>,
    }
    let record = Record {
        schema_version: 2, timestamp_unix_ms: timestamp(), event: "request_audit_disposition",
        context, occurrence: captured.occurrence, source_occurrence,
        source_submission: captured.submission, original_capture: captured.original_capture,
        disposition: "process_untrusted", audit,
    };
    match output {
        Some(output) if output.write_original_record(&record).is_ok() => DiagnosticSubmission::Written,
        Some(_) => DiagnosticSubmission::OutputUnavailable,
        None => DiagnosticSubmission::OutputUnavailable,
    }
}

/// A completed audit of an ordinary source failure is not process failure.
pub fn request_terminal_audit_passed(
    output: Option<&EmergencyDiagnosticHandle>,
    context: Option<&saddle_core::request_context::RequestContextSnapshot>,
    source: DiagnosticOccurrence,
    audit: saddle_admission::RequestAuditSnapshot,
) -> DiagnosticSubmission {
    #[derive(Serialize)]
    struct Record<'a> {
        schema_version: u8,
        event: &'static str,
        occurrence: DiagnosticOccurrence,
        context: Option<&'a saddle_core::request_context::RequestContextSnapshot>,
        audit: saddle_admission::RequestAuditSnapshot,
        disposition: &'static str,
    }
    let record = Record { schema_version: 2, event: "request_audit_passed", occurrence: source,
        context, audit, disposition: "request_failed_process_healthy" };
    match output {
        Some(output) if output.write_original_record(&record).is_ok() => DiagnosticSubmission::Written,
        _ => DiagnosticSubmission::OutputUnavailable,
    }
}

/// Real listener cleanup counters, after matched in-flight owners were joined.
/// Framework backing still owned by lifecycle is reported rather than hidden.
pub fn request_terminal_audit_cleanup(
    output: Option<&EmergencyDiagnosticHandle>,
    context: Option<&saddle_core::request_context::RequestContextSnapshot>,
    occurrence: DiagnosticOccurrence,
    snapshot: &saddle_admission::ProcessSnapshot,
) -> DiagnosticSubmission {
    #[derive(Serialize)]
    struct Record<'a> {
        schema_version: u8,
        event: &'static str,
        occurrence: DiagnosticOccurrence,
        context: Option<&'a saddle_core::request_context::RequestContextSnapshot>,
        active_accounts: usize,
        charged: usize,
        committed: usize,
        framework_charged: usize,
        task_charged: usize,
        healthy: bool,
        cleanup: &'static str,
    }
    let record = Record { schema_version: 2, event: "request_audit_cleanup", occurrence,
        context, active_accounts: snapshot.active_accounts, charged: snapshot.charged,
        committed: snapshot.committed, framework_charged: snapshot.framework_charged,
        task_charged: snapshot.task_charged, healthy: snapshot.healthy,
        cleanup: if snapshot.active_accounts == 0 && snapshot.charged == 0 && snapshot.task_charged == 0 {
            "requests_returned_framework_still_owned"
        } else { "request_resources_remaining" },
    };
    match output {
        Some(output) if output.write_original_record(&record).is_ok() => DiagnosticSubmission::Written,
        _ => DiagnosticSubmission::OutputUnavailable,
    }
}

/// Confirmed terminal records consume only the pre-account-close context bytes.
fn write_owned_terminal(output: Option<&EmergencyDiagnosticHandle>, event: &'static str,
    occurrence: DiagnosticOccurrence, record: &impl Serialize) -> DiagnosticSubmission {
    let Some(output) = output else { return DiagnosticSubmission::OutputUnavailable; };
    match stream_record_identified(&ConfirmedOutput(output), event, Some(occurrence),
        |writer| serde_json::to_writer(writer, record)) {
        DiagnosticSubmission::Enqueued => DiagnosticSubmission::Written,
        submission => submission,
    }
}

pub fn request_terminal_audit_error_owned(output: Option<&EmergencyDiagnosticHandle>,
    context: Result<&crate::OwnedUnifiedContext, DiagnosticSubmission>, diagnostic: &BoundedDiagnostic,
    source_occurrence: Option<DiagnosticOccurrence>, failure: &saddle_admission::ProfuseGwTerminalAuditFailure)
    -> UnrootedCaptureFacts {
    #[derive(Serialize)]
    struct Header<'a> {
        timestamp_unix_ms: u128, context: Option<&'a crate::OwnedUnifiedContext>,
        context_capture: Option<DiagnosticSubmission>, stage: &'static str, facts: &'a BoundedDiagnostic,
        source_occurrence: Option<DiagnosticOccurrence>, audit: Option<saddle_admission::RequestAuditSnapshot>,
        database: saddle_admission::ProfuseGwDatabaseDisposition,
        original_contract: &'static str, source_interface: &'static str,
    }
    let rejected = context.err();
    let header = Header { timestamp_unix_ms: timestamp(), context: context.ok(), context_capture: rejected,
        stage: "request_terminal_audit", facts: diagnostic, source_occurrence, audit: failure.audit,
        database: failure.database, original_contract: "display-debug-source/v1", source_interface: "error_source" };
    let occurrence = diagnostic.occurrence();
    let (submission, state) = match output {
        Some(output) => capture(&ConfirmedOutput(output), occurrence, RawSource::Error(failure), &header),
        None => (DiagnosticSubmission::OutputUnavailable, OriginalCaptureState::OutputUnavailable),
    };
    let complete = submission == DiagnosticSubmission::Enqueued && state == OriginalCaptureState::CompleteEnqueued;
    UnrootedCaptureFacts { occurrence,
        submission: if complete { DiagnosticSubmission::Written } else { submission },
        original_capture: if rejected.is_some() { OriginalCaptureState::OutputRejected }
            else if complete { OriginalCaptureState::CompleteWritten } else { state } }
}

pub fn request_terminal_audit_disposition_owned(output: Option<&EmergencyDiagnosticHandle>,
    context: Result<&crate::OwnedUnifiedContext, DiagnosticSubmission>, source_occurrence: Option<DiagnosticOccurrence>,
    captured: &UnrootedCaptureFacts, audit: Option<saddle_admission::RequestAuditSnapshot>) -> DiagnosticSubmission {
    #[derive(Serialize)]
    struct Record<'a> { #[serde(rename = "recordSchema")] record_schema: &'static str, event: &'static str,
        context: Option<&'a crate::OwnedUnifiedContext>, context_capture: Option<DiagnosticSubmission>,
        occurrence: DiagnosticOccurrence, source_occurrence: Option<DiagnosticOccurrence>,
        source_submission: DiagnosticSubmission, original_capture: OriginalCaptureState,
        disposition: &'static str, audit: Option<saddle_admission::RequestAuditSnapshot> }
    let rejected = context.err();
    let record = Record { record_schema: "1.0", event: "request_audit_disposition", context: context.ok(),
        context_capture: rejected, occurrence: captured.occurrence, source_occurrence,
        source_submission: captured.submission, original_capture: captured.original_capture,
        disposition: "process_untrusted", audit };
    let submission = write_owned_terminal(output, record.event, record.occurrence, &record);
    if rejected.is_some() { DiagnosticSubmission::OutputUnavailable } else { submission }
}

pub fn request_terminal_audit_passed_owned(output: Option<&EmergencyDiagnosticHandle>,
    context: Result<&crate::OwnedUnifiedContext, DiagnosticSubmission>, source: DiagnosticOccurrence,
    audit: saddle_admission::RequestAuditSnapshot) -> DiagnosticSubmission {
    #[derive(Serialize)]
    struct Record<'a> { #[serde(rename = "recordSchema")] record_schema: &'static str, event: &'static str, occurrence: DiagnosticOccurrence,
        context: Option<&'a crate::OwnedUnifiedContext>, context_capture: Option<DiagnosticSubmission>,
        audit: saddle_admission::RequestAuditSnapshot, disposition: &'static str }
    let rejected = context.err();
    let record = Record { record_schema: "1.0", event: "request_audit_passed", occurrence: source,
        context: context.ok(), context_capture: rejected, audit, disposition: "request_failed_process_healthy" };
    let submission = write_owned_terminal(output, record.event, source, &record);
    if rejected.is_some() { DiagnosticSubmission::OutputUnavailable } else { submission }
}

pub fn request_terminal_audit_cleanup_owned(output: Option<&EmergencyDiagnosticHandle>,
    context: Result<&crate::OwnedUnifiedContext, DiagnosticSubmission>, occurrence: DiagnosticOccurrence,
    snapshot: &saddle_admission::ProcessSnapshot) -> DiagnosticSubmission {
    #[derive(Serialize)]
    struct Record<'a> { #[serde(rename = "recordSchema")] record_schema: &'static str, event: &'static str, occurrence: DiagnosticOccurrence,
        context: Option<&'a crate::OwnedUnifiedContext>, context_capture: Option<DiagnosticSubmission>,
        active_accounts: usize, charged: usize, committed: usize, framework_charged: usize,
        task_charged: usize, healthy: bool, cleanup: &'static str }
    let rejected = context.err();
    let record = Record { record_schema: "1.0", event: "request_audit_cleanup", occurrence, context: context.ok(),
        context_capture: rejected, active_accounts: snapshot.active_accounts, charged: snapshot.charged,
        committed: snapshot.committed, framework_charged: snapshot.framework_charged,
        task_charged: snapshot.task_charged, healthy: snapshot.healthy,
        cleanup: if snapshot.active_accounts == 0 && snapshot.charged == 0 && snapshot.task_charged == 0 {
            "requests_returned_framework_still_owned"
        } else { "request_resources_remaining" } };
    let submission = write_owned_terminal(output, record.event, occurrence, &record);
    if rejected.is_some() { DiagnosticSubmission::OutputUnavailable } else { submission }
}

/// Encoding selection borrows the immutable original source; oversized records
/// use the existing confirmed segmented writer without truncation or heap copy.
pub(crate) fn write_encoding_selection(output: &EmergencyDiagnosticHandle, record: &impl Serialize) -> DiagnosticSubmission {
    struct Count(usize);
    impl std::io::Write for Count {
        fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> { self.0 = self.0.saturating_add(bytes.len()); Ok(bytes.len()) }
        fn flush(&mut self) -> std::io::Result<()> { Ok(()) }
    }
    let mut count = Count(0);
    if serde_json::to_writer(&mut count, record).is_err() { return DiagnosticSubmission::EncodingFailed; }
    if count.0 < 8192 {
        return if output.write_original_record(record).is_ok() { DiagnosticSubmission::Written } else { DiagnosticSubmission::OutputUnavailable };
    }
    match stream_record_identified(&ConfirmedOutput(output), "context_record_segment", None, |writer| serde_json::to_writer(writer, record)) {
        DiagnosticSubmission::Enqueued => DiagnosticSubmission::Written, submission => submission,
    }
}