saddle_observability/

request_diagnostic.rs

//! Controlled request failure adapter. Existing callers are migrated explicitly;
//! this module alone does not prove all framework exits are covered.
use crate::{DiagnosticSubmission, EmergencyDiagnosticHandle, EventContext};
use saddle_core::{
    BoundedDiagnostic, BoundedDiagnosticCause, CallContext, CaptureSite, DiagnosticCategory,
    DiagnosticOccurrence, DiagnosticOutcomeAxes,
};
use serde::{Serialize, Serializer};

mod original;

#[derive(Clone, Copy, Eq, PartialEq, Serialize)]
#[serde(tag = "state", content = "value", rename_all = "snake_case")]
enum Field<T> {
    Present(T),
    NotApplicable,
    NotEstablished,
    Unavailable,
}

/// Absence asserted by the component holding the real execution context.
/// Never inferred from a missing optional string or from an endpoint.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DiagnosticContextMissing {
    NotApplicable,
    NotEstablished,
    Unavailable,
}

/// Safe projection of DB's registered static logical operation, not SQL and not
/// registration authority. DB must supply its existing registered declaration.
#[derive(Clone, Copy)]
pub struct DiagnosticDbOperation(&'static str);
impl DiagnosticDbOperation {
    pub fn from_registered(value: &'static str) -> Option<Self> {
        (!value.is_empty()
            && value.len() <= 128
            && value
                .bytes()
                .all(|b| b.is_ascii_alphanumeric() || b"._:-".contains(&b)))
        .then_some(Self(value))
    }
}

/// Fixed safe projection of the already accepted ingress zone. This is NOT
/// ingress validation or routing authority; Service must pass accepted.zone.
#[derive(Clone, Copy)]
pub struct DiagnosticZone(ProtocolId);
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DiagnosticZoneError {
    Empty,
    TooLong,
    Unsafe,
}
impl DiagnosticZone {
    /// Exact copy after bounded safety validation; never hashes or truncates.
    pub fn from_validated_ingress(value: &str) -> Result<Self, DiagnosticZoneError> {
        if value.len() > 256 {
            return Err(DiagnosticZoneError::TooLong);
        }
        if value.trim().is_empty() {
            return Err(DiagnosticZoneError::Empty);
        }
        if value.chars().any(char::is_control)
            || value.contains("://")
            || value.contains(['@', '?', '#', '\\'])
        {
            return Err(DiagnosticZoneError::Unsafe);
        }
        Ok(Self(ProtocolId::copy(value)))
    }
}

#[derive(Clone, Copy, Eq, PartialEq)]
struct SafeText {
    value: ProtocolId,
    truncated: bool,
    redacted: bool,
}
impl SafeText {
    fn metadata(value: &str) -> Self {
        let mut len = value.len().min(256);
        while !value.is_char_boundary(len) {
            len -= 1;
        }
        let prefix = &value[..len];
        let redacted = prefix.contains("://")
            || prefix
                .chars()
                .any(|c| !(c.is_alphanumeric() || "_./:{}*-".contains(c)));
        Self {
            value: ProtocolId::copy(if redacted { "" } else { prefix }),
            truncated: len < value.len(),
            redacted,
        }
    }
}
impl Serialize for SafeText {
    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        use serde::ser::SerializeStruct;
        let mut out = s.serialize_struct("SafeText", 3)?;
        out.serialize_field("value", &self.value)?;
        out.serialize_field("truncated", &self.truncated)?;
        out.serialize_field("redacted", &self.redacted)?;
        out.end()
    }
}
#[derive(Clone, Copy, Eq, PartialEq)]
struct Span(u64);
impl Serialize for Span {
    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        let mut bytes = [b'0'; 16];
        for (i, b) in bytes.iter_mut().enumerate() {
            *b = b"0123456789abcdef"[((self.0 >> ((15 - i) * 4)) & 15) as usize];
        }
        s.serialize_str(std::str::from_utf8(&bytes).unwrap())
    }
}

// Exact protocol identity; never replaces a trace by a hash or generated value.
#[derive(Clone, Copy, Eq, PartialEq)]
struct ProtocolId {
    bytes: [u8; 256],
    len: usize,
}
impl ProtocolId {
    fn copy(value: &str) -> Self {
        // CallContext protocol types have already enforced this bound.
        let mut bytes = [0; 256];
        bytes[..value.len()].copy_from_slice(value.as_bytes());
        Self {
            bytes,
            len: value.len(),
        }
    }
}
impl Serialize for ProtocolId {
    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        s.serialize_str(std::str::from_utf8(&self.bytes[..self.len]).unwrap_or(""))
    }
}

/// Snapshot of existing context, not a second execution context or authority.
#[derive(Clone, Copy, Serialize)]
struct Projection {
    schema_version: u8,
    application: Field<SafeText>,
    module: Field<SafeText>,
    service: Field<SafeText>,
    operation: Field<SafeText>,
    db_operation: Field<&'static str>,
    trace_id: Field<ProtocolId>,
    rpc_id: Field<ProtocolId>,
    span_id: Field<Span>,
    request: Field<SafeText>,
    // Exact comparison only: never serialized, including when the safe display
    // is redacted. Distinct requests must not compare equal after redaction.
    #[serde(skip)]
    request_binding: Option<ProtocolId>,
    route: Field<SafeText>,
    attempt: Field<u32>,
    scope: Field<saddle_core::DbScopeDiagnosticIdentity>,
    task: Field<SafeText>,
    lifecycle: Field<SafeText>,
    zone: Field<ProtocolId>,
    target: Field<SafeText>,
}
impl Projection {
    fn missing(reason: DiagnosticContextMissing) -> Self {
        fn absent<T>(reason: DiagnosticContextMissing) -> Field<T> {
            match reason {
                DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
                DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
                DiagnosticContextMissing::Unavailable => Field::Unavailable,
            }
        }
        Self {
            schema_version: 1,
            application: absent(reason),
            module: absent(reason),
            service: absent(reason),
            operation: absent(reason),
            db_operation: absent(reason),
            trace_id: absent(reason),
            rpc_id: absent(reason),
            span_id: absent(reason),
            request: absent(reason),
            request_binding: None,
            route: absent(reason),
            attempt: absent(reason),
            scope: absent(reason),
            task: absent(reason),
            lifecycle: absent(reason),
            zone: absent(reason),
            target: absent(reason),
        }
    }
    fn existing(call: &CallContext, event: &EventContext) -> Self {
        let text = |s: &str| Field::Present(SafeText::metadata(s));
        Self {
            schema_version: 1,
            application: text(call.application().as_str()),
            module: text(call.module().as_str()),
            service: text(call.service().as_str()),
            operation: text(call.operation().as_str()),
            db_operation: Field::Unavailable,
            trace_id: Field::Present(ProtocolId::copy(call.trace_correlation_id().as_str())),
            rpc_id: call.rpc_correlation_id().map_or(Field::Unavailable, |id| {
                Field::Present(ProtocolId::copy(id.as_str()))
            }),
            span_id: Field::Present(Span(call.span_id().as_u64())),
            request: text(event.diagnostic_request()),
            request_binding: Some(ProtocolId::copy(event.diagnostic_request())),
            route: text(event.diagnostic_route()),
            attempt: Field::Present(event.diagnostic_attempt()),
            // These facts are not present in CallContext/EventContext. Do not
            // infer not_applicable from their absence or mint scope/task IDs.
            scope: Field::Unavailable,
            task: Field::Unavailable,
            lifecycle: Field::Unavailable,
            zone: Field::Unavailable,
            target: Field::Unavailable,
        }
    }
}

/// Closed missing-field vocabulary. This cannot inject arbitrary JSON keys.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DiagnosticContextField {
    Application,
    Module,
    Service,
    Operation,
    DbOperation,
    Trace,
    Rpc,
    Span,
    Request,
    Route,
    Attempt,
    Scope,
    Task,
    Lifecycle,
    Zone,
    Target,
}

/// Atomic binding failure: original projection/output can be recovered unchanged.
#[must_use = "recover the original context; do not silently discard known identity"]
pub struct ContextBindingError<T> {
    field: DiagnosticContextField,
    original: T,
}
impl<T> ContextBindingError<T> {
    pub fn field(&self) -> DiagnosticContextField {
        self.field
    }
    pub fn into_original(self) -> T {
        self.original
    }
}
fn known<T: Copy + Eq>(
    field: &mut Field<T>,
    value: T,
    name: DiagnosticContextField,
) -> Result<(), DiagnosticContextField> {
    if let Field::Present(old) = field
        && *old != value
    {
        return Err(name);
    }
    *field = Field::Present(value);
    Ok(())
}
fn bind_call(context: &mut Projection, call: &CallContext) -> Result<(), DiagnosticContextField> {
    use DiagnosticContextField as F;
    known(
        &mut context.application,
        SafeText::metadata(call.application().as_str()),
        F::Application,
    )?;
    known(
        &mut context.module,
        SafeText::metadata(call.module().as_str()),
        F::Module,
    )?;
    known(
        &mut context.service,
        SafeText::metadata(call.service().as_str()),
        F::Service,
    )?;
    known(
        &mut context.operation,
        SafeText::metadata(call.operation().as_str()),
        F::Operation,
    )?;
    known(
        &mut context.trace_id,
        ProtocolId::copy(call.trace_correlation_id().as_str()),
        F::Trace,
    )?;
    known(&mut context.span_id, Span(call.span_id().as_u64()), F::Span)?;
    if let Some(rpc) = call.rpc_correlation_id() {
        known(&mut context.rpc_id, ProtocolId::copy(rpc.as_str()), F::Rpc)?;
    }
    Ok(())
}
fn bind_event(
    context: &mut Projection,
    event: &EventContext,
) -> Result<(), DiagnosticContextField> {
    use DiagnosticContextField as F;
    bind_request(context, event.diagnostic_request())?;
    known(
        &mut context.route,
        SafeText::metadata(event.diagnostic_route()),
        F::Route,
    )?;
    known(&mut context.attempt, event.diagnostic_attempt(), F::Attempt)
}
fn bind_request(context: &mut Projection, request: &str) -> Result<(), DiagnosticContextField> {
    let exact = ProtocolId::copy(request);
    if context.request_binding.is_some_and(|old| old != exact) {
        return Err(DiagnosticContextField::Request);
    }
    known(
        &mut context.request,
        SafeText::metadata(request),
        DiagnosticContextField::Request,
    )?;
    context.request_binding = Some(exact);
    Ok(())
}

/// Actual request phase supplied by its execution owner; not admission authority.
#[derive(Clone, Copy)]
pub enum DiagnosticRequestPhase {
    SocketAccepted,
    Admission,
    ReadingHead,
    ReadingBody,
    Validation,
    Dispatch,
    Response,
    TaskJoin,
    Finalization,
}
impl DiagnosticRequestPhase {
    fn as_str(self) -> &'static str {
        match self {
            Self::SocketAccepted => "socket_accepted",
            Self::Admission => "admission",
            Self::ReadingHead => "reading_head",
            Self::ReadingBody => "reading_body",
            Self::Validation => "validation",
            Self::Dispatch => "dispatch",
            Self::Response => "response",
            Self::TaskJoin => "task_join",
            Self::Finalization => "finalization",
        }
    }
}

/// Validated read-only rendering of the *existing* runtime task instance ID.
/// This does not allocate a task/sequence or prove task ownership. RP must take
/// the digits from its real runtime ID, never a task-kind label or invented ID.
#[derive(Clone, Copy)]
pub struct DiagnosticTaskId(ProtocolId);
impl DiagnosticTaskId {
    pub fn from_runtime_id(value: &str) -> Option<Self> {
        (!value.is_empty() && value.len() <= 64 && value.bytes().all(|b| b.is_ascii_digit()))
            .then(|| Self(ProtocolId::copy(value)))
    }
}

/// Fixed early-request projection. It is metadata, not a second execution
/// context, admission permission or source receipt. No trace/sequence is minted.
/// ```compile_fail
/// let forged = saddle_observability::EarlyRequestContext { context: todo!() };
/// ```
pub struct EarlyRequestContext {
    context: Projection,
}
// Recovery intentionally returns the fixed inline original, never a heap Box.
#[allow(clippy::result_large_err)]
impl EarlyRequestContext {
    /// No request bytes/identities have been accepted yet. Application is the
    /// existing configured application label, NOT peer/body/header data.
    pub fn socket_accepted(application: &str) -> Self {
        let mut context = Projection::missing(DiagnosticContextMissing::NotEstablished);
        context.application = Field::Present(SafeText::metadata(application));
        context.lifecycle = Field::Present(SafeText::metadata("socket_accepted"));
        Self { context }
    }
    /// Facts cannot be recovered at this boundary. Known facts must be attached
    /// below; this is distinct from an identity that has never been established.
    pub fn unavailable() -> Self {
        Self {
            context: Projection::missing(DiagnosticContextMissing::Unavailable),
        }
    }
    fn update(
        mut self,
        bind: impl FnOnce(&mut Projection) -> Result<(), DiagnosticContextField>,
    ) -> Result<Self, ContextBindingError<Self>> {
        let mut next = self.context;
        if let Err(field) = bind(&mut next) {
            return Err(ContextBindingError {
                field,
                original: self,
            });
        }
        self.context = next;
        Ok(self)
    }
    pub fn with_application(
        self,
        application: &saddle_core::ApplicationId,
    ) -> Result<Self, ContextBindingError<Self>> {
        self.update(|c| {
            known(
                &mut c.application,
                SafeText::metadata(application.as_str()),
                DiagnosticContextField::Application,
            )
        })
    }
    /// An already validated protocol identity. Does not generate a root span.
    pub fn with_trace(
        self,
        trace: &saddle_core::TraceCorrelationId,
    ) -> Result<Self, ContextBindingError<Self>> {
        self.update(|c| {
            known(
                &mut c.trace_id,
                ProtocolId::copy(trace.as_str()),
                DiagnosticContextField::Trace,
            )
        })
    }
    pub fn with_rpc(
        self,
        rpc: &saddle_core::RpcCorrelationId,
    ) -> Result<Self, ContextBindingError<Self>> {
        self.update(|c| {
            known(
                &mut c.rpc_id,
                ProtocolId::copy(rpc.as_str()),
                DiagnosticContextField::Rpc,
            )
        })
    }
    pub fn with_event(self, event: &EventContext) -> Result<Self, ContextBindingError<Self>> {
        self.update(|c| bind_event(c, event))
    }
    pub fn with_call(self, call: &CallContext) -> Result<Self, ContextBindingError<Self>> {
        self.update(|c| bind_call(c, call))
    }
    pub fn with_request(
        self,
        request: &crate::RequestIdentity,
    ) -> Result<Self, ContextBindingError<Self>> {
        self.update(|c| bind_request(c, request.as_str()))
    }
    pub fn with_route(
        self,
        route: &crate::RouteIdentity,
    ) -> Result<Self, ContextBindingError<Self>> {
        self.update(|c| {
            known(
                &mut c.route,
                SafeText::metadata(route.as_str()),
                DiagnosticContextField::Route,
            )
        })
    }
    pub fn with_module(
        self,
        module: &saddle_core::ModuleId,
    ) -> Result<Self, ContextBindingError<Self>> {
        self.update(|c| {
            known(
                &mut c.module,
                SafeText::metadata(module.as_str()),
                DiagnosticContextField::Module,
            )
        })
    }
    pub fn with_service(
        self,
        service: &saddle_core::ServiceId,
    ) -> Result<Self, ContextBindingError<Self>> {
        self.update(|c| {
            known(
                &mut c.service,
                SafeText::metadata(service.as_str()),
                DiagnosticContextField::Service,
            )
        })
    }
    pub fn with_operation(
        self,
        operation: &saddle_core::OperationId,
    ) -> Result<Self, ContextBindingError<Self>> {
        self.update(|c| {
            known(
                &mut c.operation,
                SafeText::metadata(operation.as_str()),
                DiagnosticContextField::Operation,
            )
        })
    }
}

/// Bind once to the existing established request. No empty/default constructor.
///
/// ```compile_fail
/// use saddle_observability::{RequestDiagnosticScope, EmergencyDiagnosticHandle};
/// fn missing_context(output: &EmergencyDiagnosticHandle) {
///     let scope = RequestDiagnosticScope::established(output);
/// }
/// ```
pub struct RequestDiagnosticScope<'a> {
    output: Option<&'a EmergencyDiagnosticHandle>,
    context: Projection,
}
// Same fixed inline recovery model; source capture remains allocation-free.
#[allow(clippy::result_large_err)]
impl<'a> RequestDiagnosticScope<'a> {
    /// A direct transport operation has no established request. This preserves
    /// the original while marking every request fact as not established.
    pub fn unassociated_transport(output: &'a EmergencyDiagnosticHandle) -> Self {
        Self {
            output: Some(output),
            context: Projection::missing(DiagnosticContextMissing::NotEstablished),
        }
    }
    /// Preserve validated wire association when the observer could not create
    /// a CallContext. Missing observer facts stay explicitly unavailable.
    pub fn outbound_wire(
        output: &'a EmergencyDiagnosticHandle,
        event: &EventContext,
        trace_id: &str,
        rpc_id: &str,
        zone: &str,
    ) -> Option<Self> {
        let valid = |s: &str| !s.is_empty() && s.len() <= 256 && !s.chars().any(char::is_control);
        if !valid(trace_id) || !valid(rpc_id) || !valid(zone) { return None; }
        let mut context = Projection::missing(DiagnosticContextMissing::Unavailable);
        context.trace_id = Field::Present(ProtocolId::copy(trace_id));
        context.rpc_id = Field::Present(ProtocolId::copy(rpc_id));
        context.request = Field::Present(SafeText::metadata(event.diagnostic_request()));
        context.request_binding = Some(ProtocolId::copy(event.diagnostic_request()));
        context.route = Field::Present(SafeText::metadata(event.diagnostic_route()));
        context.attempt = Field::Present(event.diagnostic_attempt());
        context.zone = Field::Present(ProtocolId::copy(zone));
        Some(Self { output: Some(output), context })
    }
    /// Capture a borrowed foreign error and its caller-supplied, validated
    /// source target under this exact request projection and occurrence.
    /// Written means every original segment, including the terminator, was
    /// synchronously acknowledged by the existing emergency file writer.
    #[track_caller]
    pub fn source_error_with_target<T: Serialize>(
        &self,
        error: &(dyn std::error::Error + 'static),
        target: &T,
        category: DiagnosticCategory,
        cause: BoundedDiagnosticCause,
    ) -> RequestSourceReceipt {
        let diagnostic = BoundedDiagnostic::capture(category, CaptureSite::FirstObserved, cause);
        let occurrence = diagnostic.occurrence();
        let submission = original::capture(self.output, &self.context, &diagnostic, error, target);
        FrameworkRequestFailure {
            error: (),
            context: self.context.clone(),
            occurrence,
            submission,
            diagnostic,
        }
    }
    /// Capture a borrowed first error before traversing its static source chain.
    #[track_caller]
    pub fn source_borrowed_error_with_target<T: Serialize, E: std::error::Error>(
        &self,
        error: &E,
        target: &T,
        category: DiagnosticCategory,
        cause: BoundedDiagnosticCause,
    ) -> RequestSourceReceipt {
        let diagnostic = BoundedDiagnostic::capture(category, CaptureSite::FirstObserved, cause);
        let occurrence = diagnostic.occurrence();
        let submission = original::capture_borrowed(
            self.output, &self.context, &diagnostic, error, target,
        );
        FrameworkRequestFailure {
            error: (), context: self.context.clone(), occurrence, submission, diagnostic,
        }
    }
    /// Same source issuer for early and established requests. Output availability
    /// is independent of identity absence; None never skips source retention.
    pub fn early(
        output: Option<&'a EmergencyDiagnosticHandle>,
        early: EarlyRequestContext,
    ) -> Self {
        Self {
            output,
            context: early.context,
        }
    }
    /// Move the existing projection to an available/unavailable output without
    /// replacing any context. Useful when an owning task retains the handle.
    pub fn with_output<'b>(
        self,
        output: Option<&'b EmergencyDiagnosticHandle>,
    ) -> RequestDiagnosticScope<'b> {
        RequestDiagnosticScope {
            output,
            context: self.context,
        }
    }
    /// Fixed read-only snapshot for a task's parent/child observation sites.
    /// Copies no receipt or execution owner. Later enrichment is NOT retroactive;
    /// a join fallback can only claim facts actually retained at spawn time.
    pub fn reborrow(&self) -> RequestDiagnosticScope<'a> {
        RequestDiagnosticScope {
            output: self.output,
            context: self.context,
        }
    }
    /// Observe an ordinary completed operation with the exact early/established
    /// projection. Technical failures still require their mandatory carrier.
    pub fn record_nonfailure(
        &self,
        axes: &DiagnosticOutcomeAxes,
    ) -> Result<DiagnosticSubmission, ()> {
        if !matches!(
            axes.operation,
            saddle_core::OperationOutcome::Succeeded | saddle_core::OperationOutcome::Rejected
        ) {
            return Err(());
        }
        #[derive(Serialize)]
        struct NonfailureRecord<'a> {
            event: &'static str,
            timestamp_unix_ms: u128,
            context: &'a Projection,
            diagnostic: Option<&'a BoundedDiagnostic>,
            diagnostic_reference: Option<DiagnosticOccurrence>,
            axes: Option<&'a DiagnosticOutcomeAxes>,
            source_submission: Option<DiagnosticSubmission>,
        }
        Ok(self
            .output
            .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
                output.submit_fixed_record(&NonfailureRecord {
                    event: "framework.boundary.outcome",
                    timestamp_unix_ms: timestamp(),
                    context: &self.context,
                    diagnostic: None::<&BoundedDiagnostic>,
                    diagnostic_reference: None::<DiagnosticOccurrence>,
                    axes: Some(axes),
                    source_submission: None::<DiagnosticSubmission>,
                })
            }))
    }
    pub fn with_task(mut self, task: DiagnosticTaskId) -> Result<Self, ContextBindingError<Self>> {
        if let Err(field) = known(
            &mut self.context.task,
            SafeText {
                value: task.0,
                truncated: false,
                redacted: false,
            },
            DiagnosticContextField::Task,
        ) {
            return Err(ContextBindingError {
                field,
                original: self,
            });
        }
        Ok(self)
    }
    /// Bind only facts actually established by the ingress owner. On conflict no
    /// partial update occurs; task/zone/scope and prior receipts stay unchanged.
    pub fn bind_request_identity(
        mut self,
        request: &crate::RequestIdentity,
    ) -> Result<Self, ContextBindingError<Self>> {
        if let Err(field) = bind_request(&mut self.context, request.as_str()) {
            return Err(ContextBindingError {
                field,
                original: self,
            });
        }
        Ok(self)
    }

    pub fn bind_established(
        mut self,
        call: &CallContext,
        event: &EventContext,
    ) -> Result<Self, ContextBindingError<Self>> {
        let mut next = self.context;
        if let Err(field) = bind_call(&mut next, call).and_then(|()| bind_event(&mut next, event)) {
            return Err(ContextBindingError {
                field,
                original: self,
            });
        }
        self.context = next;
        Ok(self)
    }
    /// Derive the actual outbound call without rewriting the inbound scope.
    /// The caller supplies Boundary's existing child Call/Event (not invented
    /// identities). Trace and request must already be known and match exactly;
    /// RPC must be a direct `parent.<decimal sequence>` child and span must differ.
    /// Missing parent identity cannot establish this relationship. Request
    /// comparison uses a private exact copy, never the redacted display value.
    /// This validates diagnostic metadata, not execution/transport authority.
    pub fn derive_outbound_child(
        &self,
        call: &CallContext,
        event: &EventContext,
    ) -> Result<RequestDiagnosticScope<'a>, DiagnosticContextField> {
        use DiagnosticContextField as F;
        if self.context.trace_id
            != Field::Present(ProtocolId::copy(call.trace_correlation_id().as_str()))
        {
            return Err(F::Trace);
        }
        if self.context.request_binding != Some(ProtocolId::copy(event.diagnostic_request())) {
            return Err(F::Request);
        }
        let Field::Present(parent_rpc) = self.context.rpc_id else {
            return Err(F::Rpc);
        };
        let child_rpc = call.rpc_correlation_id().ok_or(F::Rpc)?.as_str();
        let parent_rpc =
            std::str::from_utf8(&parent_rpc.bytes[..parent_rpc.len]).map_err(|_| F::Rpc)?;
        let sequence = child_rpc
            .strip_prefix(parent_rpc)
            .and_then(|suffix| suffix.strip_prefix('.'))
            .ok_or(F::Rpc)?;
        if sequence.is_empty() || !sequence.bytes().all(|b| b.is_ascii_digit()) {
            return Err(F::Rpc);
        }
        let Field::Present(parent_span) = self.context.span_id else {
            return Err(F::Span);
        };
        if parent_span == Span(call.span_id().as_u64()) {
            return Err(F::Span);
        }
        // Only call-local facts are replaced. All other facts (including their
        // absence states) and the existing output borrow remain the parent's.
        let child = Projection::existing(call, event);
        let mut context = self.context;
        context.application = child.application;
        context.module = child.module;
        context.service = child.service;
        context.operation = child.operation;
        context.rpc_id = child.rpc_id;
        context.span_id = child.span_id;
        context.route = child.route;
        context.attempt = child.attempt;
        Ok(RequestDiagnosticScope {
            output: self.output,
            context,
        })
    }
    /// Advances this projection only; retained source receipts are snapshots.
    pub fn with_phase(mut self, phase: DiagnosticRequestPhase) -> Self {
        self.context.lifecycle = Field::Present(SafeText::metadata(phase.as_str()));
        self
    }
    /// Missing assertions cannot erase known facts. No arbitrary key/value API.
    pub fn with_missing(
        mut self,
        field: DiagnosticContextField,
        reason: DiagnosticContextMissing,
    ) -> Self {
        fn set<T>(field: &mut Field<T>, reason: DiagnosticContextMissing) {
            if !matches!(field, Field::Present(_)) {
                *field = match reason {
                    DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
                    DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
                    DiagnosticContextMissing::Unavailable => Field::Unavailable,
                };
            }
        }
        match field {
            DiagnosticContextField::Application => set(&mut self.context.application, reason),
            DiagnosticContextField::Module => set(&mut self.context.module, reason),
            DiagnosticContextField::Service => set(&mut self.context.service, reason),
            DiagnosticContextField::Operation => set(&mut self.context.operation, reason),
            DiagnosticContextField::DbOperation => set(&mut self.context.db_operation, reason),
            DiagnosticContextField::Trace => set(&mut self.context.trace_id, reason),
            DiagnosticContextField::Rpc => set(&mut self.context.rpc_id, reason),
            DiagnosticContextField::Span => set(&mut self.context.span_id, reason),
            DiagnosticContextField::Request => set(&mut self.context.request, reason),
            DiagnosticContextField::Route => set(&mut self.context.route, reason),
            DiagnosticContextField::Attempt => set(&mut self.context.attempt, reason),
            DiagnosticContextField::Scope => set(&mut self.context.scope, reason),
            DiagnosticContextField::Task => set(&mut self.context.task, reason),
            DiagnosticContextField::Lifecycle => set(&mut self.context.lifecycle, reason),
            DiagnosticContextField::Zone => set(&mut self.context.zone, reason),
            DiagnosticContextField::Target => set(&mut self.context.target, reason),
        }
        self
    }
    /// Projection from a checked live request pair; no SQL-entry side effects.
    pub fn live_db_scope(
        output: Option<&'a EmergencyDiagnosticHandle>,
        projection: &saddle_core::DbScopeDiagnosticContext<(&CallContext, &EventContext)>,
    ) -> Self {
        let ((call, event), scope) = projection.diagnostic_context();
        let mut bound = match output {
            Some(output) => Self::established(output, call, event),
            None => Self::output_unavailable(call, event),
        };
        bound.context.scope = Field::Present(scope);
        bound
    }

    /// Adds only the registered DB operation; original request/scope are intact.
    pub fn with_db_operation(mut self, operation: DiagnosticDbOperation) -> Self {
        self.set_db_operation(operation);
        self
    }
    /// Advances only this scope's next-operation projection. Previously issued
    /// receipts own a value snapshot and are never rewritten.
    pub fn set_db_operation(&mut self, operation: DiagnosticDbOperation) {
        self.context.db_operation = Field::Present(operation.0);
    }
    pub fn with_db_operation_missing(mut self, reason: DiagnosticContextMissing) -> Self {
        if !matches!(self.context.db_operation, Field::Present(_)) {
            self.context.db_operation = match reason {
                DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
                DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
                DiagnosticContextMissing::Unavailable => Field::Unavailable,
            };
        }
        self
    }
    /// Source owner supplies the original accepted zone before source capture.
    /// Every subsequent source/boundary projection includes it automatically.
    pub fn with_zone(mut self, zone: DiagnosticZone) -> Self {
        self.context.zone = Field::Present(zone.0);
        self
    }
    pub fn with_zone_missing(mut self, reason: DiagnosticContextMissing) -> Self {
        // An absence assertion cannot erase an already available zone.
        if !matches!(self.context.zone, Field::Present(_)) {
            self.context.zone = match reason {
                DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
                DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
                DiagnosticContextMissing::Unavailable => Field::Unavailable,
            };
        }
        self
    }
    /// Takes BOTH context and scope from the same checked token, never a raw ID
    /// or a separately supplied CallContext. Borrowing leaves terminal use intact.
    pub fn db_scope(
        output: &'a EmergencyDiagnosticHandle,
        observation: &saddle_core::DbScopeObservation<(crate::Observer, CallContext, EventContext)>,
    ) -> Self {
        let ((_, call, event), scope) = observation.diagnostic_context();
        let mut bound = Self::established(output, call, event);
        bound.context.scope = Field::Present(scope);
        bound
    }
    pub fn db_scope_output_unavailable(
        observation: &saddle_core::DbScopeObservation<(crate::Observer, CallContext, EventContext)>,
    ) -> Self {
        let ((_, call, event), scope) = observation.diagnostic_context();
        let mut bound = Self::output_unavailable(call, event);
        bound.context.scope = Field::Present(scope);
        bound
    }
    pub fn established(
        output: &'a EmergencyDiagnosticHandle,
        call: &CallContext,
        event: &EventContext,
    ) -> Self {
        Self {
            output: Some(output),
            context: Projection::existing(call, event),
        }
    }

    /// Explicit degraded route. Context is still mandatory; no successful
    /// submission or writer readiness is inferred.
    pub fn output_unavailable(call: &CallContext, event: &EventContext) -> Self {
        Self {
            output: None,
            context: Projection::existing(call, event),
        }
    }
    /// Final bound ingress deadline; no clock reads or new identity.
    pub fn ingress_deadline_bound(&self, ingress_deadline_unix_ms: i64) -> DiagnosticSubmission {
        // A redacted/truncated legacy display cannot bind this fact to a request.
        // Preserve the existing privacy boundary instead of exposing request_binding.
        if !matches!(self.context.request, Field::Present(SafeText {
            truncated: false, redacted: false, ..
        })) {
            return DiagnosticSubmission::EncodingFailed;
        }
        #[derive(Serialize)]
        struct DeadlineRecord<'a> {
            schema_version: u8,
            event: &'static str,
            ingress_deadline_unix_ms: i64,
            context: &'a Projection,
        }
        self.output.map_or(DiagnosticSubmission::OutputUnavailable, |output|
            output.submit_fixed_record(&DeadlineRecord {
                schema_version: 1, event: "ingress_deadline_bound",
                ingress_deadline_unix_ms, context: &self.context,
            }))
    }

    /// Consumes existing source facts without capturing a replacement identity.
    pub fn capture_required(&self, diagnostic: BoundedDiagnostic) -> RequestSourceReceipt {
        self.capture_error((), diagnostic)
    }
    /// First submission of an already captured panic/cleanup diagnostic.
    /// Moves the original carrier, never recaptures or clones its stack.
    /// `observer` preserves the existing main-log mirror when one is available.
    /// Uses the legacy allocating 32KiB encoder, NOT the admitted bounded path.
    pub fn capture_existing(
        &self,
        diagnostic: saddle_core::Diagnostic,
        observer: Option<&crate::Observer>,
    ) -> ExistingDiagnosticReceipt {
        let occurrence = diagnostic.occurrence();
        let record = Record {
            event: "framework.diagnostic",
            timestamp_unix_ms: timestamp(),
            context: &self.context,
            diagnostic: Some(&diagnostic),
            diagnostic_reference: occurrence,
            axes: None,
            source_submission: None,
        };
        let submission = self
            .output
            .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
                output.submit_existing_record(&record, diagnostic.deferred_stack())
            });
        if let Some(observer) = observer {
            observer.mirror_existing_record(&record, diagnostic.category());
        }
        FrameworkRequestFailure {
            error: (),
            context: self.context,
            occurrence,
            submission,
            diagnostic,
        }
    }

    /// Capture at the real source, before exposing its error to a caller.
    /// Submission degradation never changes the source error.
    #[track_caller]
    pub fn fail<E>(
        &self,
        error: E,
        category: DiagnosticCategory,
        cause: BoundedDiagnosticCause,
    ) -> FrameworkRequestFailure<E> {
        let diagnostic = BoundedDiagnostic::capture(category, CaptureSite::FirstObserved, cause);
        self.capture_error(error, diagnostic)
    }
    fn capture_error<E>(
        &self,
        error: E,
        diagnostic: BoundedDiagnostic,
    ) -> FrameworkRequestFailure<E> {
        let occurrence = diagnostic.occurrence();
        let submission = self
            .output
            .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
                output.submit_fixed_record(&Record {
                    event: "framework.diagnostic",
                    timestamp_unix_ms: timestamp(),
                    context: &self.context,
                    diagnostic: Some(&diagnostic),
                    diagnostic_reference: occurrence,
                    axes: None,
                    source_submission: None,
                })
            });
        FrameworkRequestFailure {
            error,
            context: self.context,
            occurrence,
            submission,
            diagnostic,
        }
    }
}

/// Only the actual context-bound source submission above constructs this value.
/// No Clone, Default, From<E>, raw identity constructor, or context replacement.
///
/// ```compile_fail
/// use saddle_observability::FrameworkRequestFailure;
/// fn naked() -> Result<(), FrameworkRequestFailure<u32>> { Err(17) }
/// ```
/// ```compile_fail
/// use saddle_observability::FrameworkRequestFailure;
/// fn bypass() -> Result<(), FrameworkRequestFailure<u32>> {
///     let value: Result<(), u32> = Err(17);
///     value?;
///     Ok(())
/// }
/// ```
/// ```compile_fail
/// use saddle_observability::FrameworkRequestFailure;
/// fn forged(source: FrameworkRequestFailure<u32>) -> FrameworkRequestFailure<u32> {
///     FrameworkRequestFailure { error: 17, ..source }
/// }
/// ```
#[must_use = "retain the source receipt with the technical result until its declared boundary"]
pub struct FrameworkRequestFailure<E, D = BoundedDiagnostic> {
    error: E,
    context: Projection,
    occurrence: DiagnosticOccurrence,
    submission: DiagnosticSubmission,
    diagnostic: D,
}
/// Actual source capture/submission receipt; no public construction or Clone.
/// ```compile_fail
/// fn duplicate(receipt: saddle_observability::RequestSourceReceipt) {
///     let _second = receipt.clone();
/// }
/// ```
/// ```compile_fail
/// fn missing() -> saddle_observability::RequestSourceReceipt {
///     Default::default()
/// }
/// ```
pub type RequestSourceReceipt = FrameworkRequestFailure<()>;
pub type ExistingDiagnosticReceipt = FrameworkRequestFailure<(), saddle_core::Diagnostic>;
impl<E, D> FrameworkRequestFailure<E, D> {
    pub fn error(&self) -> &E {
        &self.error
    }
    pub fn submission(&self) -> DiagnosticSubmission {
        self.submission
    }
    pub fn map_error<F>(self, map: impl FnOnce(E) -> F) -> FrameworkRequestFailure<F, D> {
        FrameworkRequestFailure {
            error: map(self.error),
            context: self.context,
            occurrence: self.occurrence,
            submission: self.submission,
            diagnostic: self.diagnostic,
        }
    }
    pub fn record_boundary(
        &self,
        output: &EmergencyDiagnosticHandle,
        axes: &DiagnosticOutcomeAxes,
    ) -> DiagnosticSubmission {
        self.record_boundary_optional(Some(output), axes)
    }
    pub fn record_boundary_optional(
        &self,
        output: Option<&EmergencyDiagnosticHandle>,
        axes: &DiagnosticOutcomeAxes,
    ) -> DiagnosticSubmission {
        output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
            output.submit_fixed_record(&Record {
                event: "framework.boundary.outcome",
                timestamp_unix_ms: timestamp(),
                context: &self.context,
                diagnostic: None::<&BoundedDiagnostic>,
                diagnostic_reference: self.occurrence,
                axes: Some(axes),
                source_submission: Some(self.submission),
            })
        })
    }
    /// Declared terminal observation without discarding the original error or
    /// diagnostic on missing/full/closed output. The returned SAME mandatory
    /// carrier must remain in the enclosing result until its actual terminal.
    /// This does not prove that arbitrary Rust drop/forget can be forbidden.
    pub fn finish_boundary_retained(
        self,
        output: Option<&EmergencyDiagnosticHandle>,
        axes: &DiagnosticOutcomeAxes,
    ) -> (Self, BoundaryDiagnosticDelivery) {
        let delivery = BoundaryDiagnosticDelivery {
            source: self.submission,
            boundary: self.record_boundary_optional(output, axes),
            occurrence: self.occurrence,
        };
        (self, delivery)
    }
    pub fn source_diagnostic(&self) -> &D {
        &self.diagnostic
    }
    /// Sole consuming release of the original error: records its boundary first.
    /// The enclosing component must use this only at its declared public boundary,
    /// not as an adapter back to a legacy internal technical-failure path.
    pub fn finish_boundary(
        self,
        output: &EmergencyDiagnosticHandle,
        axes: &DiagnosticOutcomeAxes,
    ) -> (E, BoundaryDiagnosticDelivery) {
        let boundary = self.record_boundary(output, axes);
        (
            self.error,
            BoundaryDiagnosticDelivery {
                source: self.submission,
                boundary,
                occurrence: self.occurrence,
            },
        )
    }
}
impl<D> FrameworkRequestFailure<(), D> {
    /// Consume a capture receipt into its context-preserving boundary reference.
    pub fn into_reference(self) -> RequestBoundaryReference<D> {
        RequestBoundaryReference {
            context: self.context,
            occurrence: self.occurrence,
            submission: self.submission,
            diagnostic: self.diagnostic,
        }
    }
}
#[must_use = "retain source facts and submission status through terminal consumption"]
pub struct RequestBoundaryReference<D = BoundedDiagnostic> {
    context: Projection,
    occurrence: DiagnosticOccurrence,
    submission: DiagnosticSubmission,
    diagnostic: D,
}
impl<D> RequestBoundaryReference<D> {
    pub(crate) fn record_stage(
        &self,
        output: Option<&EmergencyDiagnosticHandle>,
        axes: &DiagnosticOutcomeAxes,
        stage: &'static str,
        elapsed_ms: u64,
    ) -> BoundaryDiagnosticDelivery {
        #[derive(Serialize)]
        struct StageRecord<'a> {
            #[serde(flatten)]
            record: Record<'a>,
            stage: &'static str,
            elapsed_ms: u64,
        }
        let boundary = output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
            output.submit_fixed_record(&StageRecord {
                record: Record {
                    event: "framework.boundary.outcome",
                    timestamp_unix_ms: timestamp(),
                    context: &self.context,
                    diagnostic: None,
                    diagnostic_reference: self.occurrence,
                    axes: Some(axes),
                    source_submission: Some(self.submission),
                },
                stage,
                elapsed_ms,
            })
        });
        BoundaryDiagnosticDelivery {
            source: self.submission,
            boundary,
            occurrence: self.occurrence,
        }
    }
    pub fn occurrence(&self) -> DiagnosticOccurrence {
        self.occurrence
    }
    pub fn source_submission(&self) -> DiagnosticSubmission {
        self.submission
    }
    /// Retained safe source facts, including when no output could accept them.
    pub fn source_diagnostic(&self) -> &D {
        &self.diagnostic
    }
    pub fn record(
        &self,
        output: &EmergencyDiagnosticHandle,
        axes: &DiagnosticOutcomeAxes,
    ) -> DiagnosticSubmission {
        self.record_optional(Some(output), axes)
    }
    pub fn record_optional(
        &self,
        output: Option<&EmergencyDiagnosticHandle>,
        axes: &DiagnosticOutcomeAxes,
    ) -> DiagnosticSubmission {
        output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
            output.submit_fixed_record(&Record {
                event: "framework.boundary.outcome",
                timestamp_unix_ms: timestamp(),
                context: &self.context,
                diagnostic: None::<&BoundedDiagnostic>,
                diagnostic_reference: self.occurrence,
                axes: Some(axes),
                source_submission: Some(self.submission),
            })
        })
    }
    /// Moves and returns the same retained reference, never a replacement receipt.
    pub fn finish_retained(
        self,
        output: Option<&EmergencyDiagnosticHandle>,
        axes: &DiagnosticOutcomeAxes,
    ) -> (Self, BoundaryDiagnosticDelivery) {
        let delivery = BoundaryDiagnosticDelivery {
            source: self.submission,
            boundary: self.record_optional(output, axes),
            occurrence: self.occurrence,
        };
        (self, delivery)
    }
}
/// Observation only. Neither status claims that a record was written.
pub struct BoundaryDiagnosticDelivery {
    source: DiagnosticSubmission,
    boundary: DiagnosticSubmission,
    occurrence: DiagnosticOccurrence,
}
impl BoundaryDiagnosticDelivery {
    pub fn source_submission(&self) -> DiagnosticSubmission {
        self.source
    }
    pub fn boundary_submission(&self) -> DiagnosticSubmission {
        self.boundary
    }
    pub fn occurrence(&self) -> DiagnosticOccurrence {
        self.occurrence
    }
}
#[derive(Serialize)]
struct Record<'a, D = BoundedDiagnostic> {
    event: &'static str,
    timestamp_unix_ms: u128,
    context: &'a Projection,
    diagnostic: Option<&'a D>,
    diagnostic_reference: DiagnosticOccurrence,
    axes: Option<&'a DiagnosticOutcomeAxes>,
    #[serde(skip_serializing_if = "Option::is_none")]
    source_submission: Option<DiagnosticSubmission>,
}
fn timestamp() -> u128 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis()
}

#[cfg(test)]
mod early_tests {
    use super::*;
    use saddle_core::*;
    fn ok<T>(value: std::result::Result<T, ContextBindingError<T>>) -> T {
        match value {
            Ok(value) => value,
            Err(_) => panic!("unexpected context conflict"),
        }
    }
    fn cause() -> BoundedDiagnosticCause {
        BoundedDiagnosticCause::new(
            DiagnosticStage::RequestDecode,
            DiagnosticCode::new("request.read_failed").unwrap(),
        )
    }
    fn context(scope: &RequestDiagnosticScope<'_>) -> serde_json::Value {
        serde_json::to_value(scope.context).unwrap()
    }
    #[test]
    fn early_absence_and_output_are_independent_and_receipt_retains_error() {
        let scope =
            RequestDiagnosticScope::early(None, EarlyRequestContext::socket_accepted("app"))
                .with_missing(
                    DiagnosticContextField::DbOperation,
                    DiagnosticContextMissing::NotApplicable,
                )
                .with_missing(
                    DiagnosticContextField::Application,
                    DiagnosticContextMissing::Unavailable,
                )
                .with_phase(DiagnosticRequestPhase::ReadingHead);
        let value = context(&scope);
        assert_eq!(value["application"]["value"]["value"], "app");
        for field in ["trace_id", "rpc_id", "span_id", "request", "scope", "task"] {
            assert_eq!(value[field]["state"], "not_established");
            assert!(value[field].get("value").is_none());
        }
        assert_eq!(value["db_operation"]["state"], "not_applicable");
        assert_eq!(value["lifecycle"]["value"]["value"], "reading_head");
        let failure = scope.fail(17u32, DiagnosticCategory::UnexpectedError, cause());
        let id = failure.source_diagnostic().id();
        assert_eq!(
            failure.submission(),
            DiagnosticSubmission::OutputUnavailable
        );
        let mapped = failure.map_error(|code| code + 1);
        let (retained, delivery) =
            mapped.finish_boundary_retained(None, &DiagnosticOutcomeAxes::default());
        assert_eq!(*retained.error(), 18);
        assert_eq!(retained.source_diagnostic().id(), id);
        assert_eq!(
            delivery.source_submission(),
            DiagnosticSubmission::OutputUnavailable
        );
        assert_eq!(
            delivery.boundary_submission(),
            DiagnosticSubmission::OutputUnavailable
        );
        let lost = RequestDiagnosticScope::early(None, EarlyRequestContext::unavailable());
        assert_eq!(context(&lost)["trace_id"]["state"], "unavailable");
    }
    #[test]
    fn known_partial_identity_atomic_conflict_and_promotion_preserve_snapshots() {
        let trace = TraceCorrelationId::new("gateway-opaque-原值").unwrap();
        let foreign = TraceCorrelationId::new("foreign").unwrap();
        let early = ok(EarlyRequestContext::socket_accepted("app").with_trace(&trace));
        let early = match early.with_trace(&foreign) {
            Ok(_) => panic!("foreign trace accepted"),
            Err(error) => {
                assert_eq!(error.field(), DiagnosticContextField::Trace);
                error.into_original()
            }
        };
        let request = crate::RequestIdentity::new("request-1").unwrap();
        let early = ok(early.with_request(&request));
        let scope = ok(RequestDiagnosticScope::early(None, early)
            .with_task(DiagnosticTaskId::from_runtime_id("42").unwrap()));
        let before = context(&scope);
        let receipt = scope
            .fail((), DiagnosticCategory::UnexpectedError, cause())
            .into_reference();
        let call = CallContext::new(
            "app".into(),
            "module".into(),
            "service".into(),
            "operation".into(),
            TraceId::from_u128(1),
            SpanId::from_u64(2),
        )
        .with_trace_correlation_id(foreign);
        let event =
            EventContext::new(request, crate::RouteIdentity::new("/route").unwrap(), 1).unwrap();
        let scope = match scope.bind_established(&call, &event) {
            Ok(_) => panic!("foreign promotion accepted"),
            Err(error) => error.into_original(),
        };
        assert_eq!(context(&scope), before); // module/service partial writes rolled back
        let call = call.with_trace_correlation_id(trace);
        let scope = ok(scope.bind_established(&call, &event));
        let after = context(&scope);
        assert_eq!(after["trace_id"]["value"], "gateway-opaque-原值");
        assert_eq!(after["task"]["value"]["value"], "42");
        assert_eq!(after["span_id"]["value"], "0000000000000002");
        assert_eq!(serde_json::to_value(receipt.context).unwrap(), before);
        assert!(
            scope.with_output(None).context.request
                == Field::Present(SafeText::metadata("request-1"))
        );
    }
    #[test]
    fn task_projection_is_bounded_and_cannot_be_replaced() {
        for unsafe_id in ["", "request_task", "-1", "https://secret", "12\n3"] {
            assert!(DiagnosticTaskId::from_runtime_id(unsafe_id).is_none());
        }
        assert!(DiagnosticTaskId::from_runtime_id(&"1".repeat(65)).is_none());
        let scope = ok(RequestDiagnosticScope::early(
            None,
            EarlyRequestContext::socket_accepted("app"),
        )
        .with_task(DiagnosticTaskId::from_runtime_id("42").unwrap()));
        let original = context(&scope);
        let scope = match scope.with_task(DiagnosticTaskId::from_runtime_id("43").unwrap()) {
            Ok(_) => panic!("changed task identity"),
            Err(error) => error.into_original(),
        };
        assert_eq!(context(&scope), original);
    }
}

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

    fn event(request: &str, route: &str) -> EventContext {
        EventContext::new(
            crate::RequestIdentity::new(request).unwrap(),
            crate::RouteIdentity::new(route).unwrap(),
            1,
        )
        .unwrap()
    }
    fn call(trace: &str, rpc: &str, span: u64) -> CallContext {
        CallContext::new(
            "saddle".into(),
            "zone-a".into(),
            "profusecontract".into(),
            "invoke".into(),
            TraceId::from_u128(1),
            SpanId::from_u64(span),
        )
        .with_trace_correlation_id(TraceCorrelationId::new(trace).unwrap())
        .with_rpc_correlation_id(RpcCorrelationId::new(rpc))
    }
    fn cause() -> BoundedDiagnosticCause {
        BoundedDiagnosticCause::new(
            DiagnosticStage::RequestDecode,
            DiagnosticCode::new("outbound.connect_failed").unwrap(),
        )
    }
    #[test]
    fn outbound_child_preserves_live_scope_and_receipts() {
        let parent = call("opaque-trace", "0.4", 1);
        let child = call("opaque-trace", "0.4.1", 2);
        let parent_event = event("request-1", "/incoming");
        let child_event = event("request-1", "remote.function");
        let (_, issuer) = DbPhysicalDispositionIssuer::issue().into_startup_and_request_issuer();
        let (request, execution) = issuer.issue_request().unwrap();
        let live = request
            .project_diagnostic_context(&execution, (&parent, &parent_event))
            .ok()
            .unwrap();
        let scope = RequestDiagnosticScope::live_db_scope(None, &live)
            .with_task(DiagnosticTaskId::from_runtime_id("42").unwrap())
            .unwrap_or_else(|_| panic!("task"))
            .with_zone(DiagnosticZone::from_validated_ingress("zone-a").unwrap())
            .with_db_operation(DiagnosticDbOperation::from_registered("orders.query").unwrap());
        let original = serde_json::to_value(scope.context).unwrap();
        let old = scope.fail(7, DiagnosticCategory::UnexpectedError, cause());
        let derived = scope.derive_outbound_child(&child, &child_event).unwrap();
        let value = serde_json::to_value(derived.context).unwrap();
        for field in [
            "request",
            "trace_id",
            "task",
            "zone",
            "scope",
            "db_operation",
            "lifecycle",
            "target",
        ] {
            assert_eq!(value[field], original[field], "{field}");
        }
        assert_eq!(value["rpc_id"]["value"], "0.4.1");
        assert_eq!(value["span_id"]["value"], "0000000000000002");
        assert_eq!(value["route"]["value"]["value"], "remote.function");
        assert_eq!(serde_json::to_value(scope.context).unwrap(), original);
        assert_eq!(serde_json::to_value(old.context).unwrap(), original);
        assert!(
            scope
                .reborrow()
                .bind_established(&child, &child_event)
                .is_err()
        );
        let source = derived.fail(9, DiagnosticCategory::UnexpectedError, cause());
        let id = source.source_diagnostic().id();
        let (retained, delivery) =
            source.finish_boundary_retained(None, &DiagnosticOutcomeAxes::default());
        assert_eq!(retained.source_diagnostic().id(), id);
        assert_eq!(*retained.error(), 9);
        assert_eq!(
            delivery.source_submission(),
            DiagnosticSubmission::OutputUnavailable
        );
        assert_eq!(
            delivery.boundary_submission(),
            DiagnosticSubmission::OutputUnavailable
        );
        assert_eq!(serde_json::to_value(retained.context).unwrap(), value);
    }

    #[test]
    fn outbound_child_rejects_foreign_missing_and_redaction_collisions() {
        let parent = call("trace", "0", 1);
        // Both display values redact; equality must use the private exact input.
        let scope = RequestDiagnosticScope::output_unavailable(&parent, &event("a@b", "/in"));
        let before = serde_json::to_value(scope.context).unwrap();
        let child_event = event("a@b", "remote");
        for (child, ev, expected) in [
            (
                call("foreign", "0.1", 2),
                child_event.clone(),
                DiagnosticContextField::Trace,
            ),
            (
                call("trace", "0.1", 2),
                event("c@d", "remote"),
                DiagnosticContextField::Request,
            ),
            (
                call("trace", "01.1", 2),
                child_event.clone(),
                DiagnosticContextField::Rpc,
            ),
            (
                call("trace", "0", 2),
                child_event.clone(),
                DiagnosticContextField::Rpc,
            ),
            (
                call("trace", "0.1.2", 2),
                child_event.clone(),
                DiagnosticContextField::Rpc,
            ),
            (
                call("trace", "0.x", 2),
                child_event.clone(),
                DiagnosticContextField::Rpc,
            ),
            (
                call("trace", "0.1", 1),
                child_event.clone(),
                DiagnosticContextField::Span,
            ),
        ] {
            assert_eq!(
                scope.derive_outbound_child(&child, &ev).err(),
                Some(expected)
            );
            assert_eq!(serde_json::to_value(scope.context).unwrap(), before);
        }
        let child = call("trace", "0.1", 2);
        let derived = scope.derive_outbound_child(&child, &child_event).unwrap();
        let json = serde_json::to_string(&derived.context).unwrap();
        assert!(!json.contains("a@b") && !json.contains("request_binding"));
        assert!(
            RequestDiagnosticScope::early(None, EarlyRequestContext::unavailable())
                .derive_outbound_child(&child, &child_event)
                .is_err()
        );
        let missing_rpc = parent.clone().with_rpc_correlation_id(None);
        assert_eq!(
            RequestDiagnosticScope::output_unavailable(&missing_rpc, &child_event)
                .derive_outbound_child(&child, &child_event)
                .err(),
            Some(DiagnosticContextField::Rpc)
        );
        assert!(
            scope
                .reborrow()
                .bind_request_identity(&crate::RequestIdentity::new("c@d").unwrap())
                .is_err()
        );
        assert!(
            EarlyRequestContext::unavailable()
                .with_request(&crate::RequestIdentity::new("a@b").unwrap())
                .unwrap_or_else(|_| panic!("first identity"))
                .with_request(&crate::RequestIdentity::new("c@d").unwrap())
                .is_err()
        );
    }
}