Skip to main content

saddle_observability/
request_diagnostic.rs

1//! Controlled request failure adapter. Existing callers are migrated explicitly;
2//! this module alone does not prove all framework exits are covered.
3use crate::{DiagnosticSubmission, EmergencyDiagnosticHandle, EventContext};
4use saddle_core::{
5    BoundedDiagnostic, BoundedDiagnosticCause, CallContext, CaptureSite, DiagnosticCategory,
6    DiagnosticOccurrence, DiagnosticOutcomeAxes,
7};
8use serde::{Serialize, Serializer};
9
10mod original;
11
12#[derive(Clone, Copy, Eq, PartialEq, Serialize)]
13#[serde(tag = "state", content = "value", rename_all = "snake_case")]
14enum Field<T> {
15    Present(T),
16    NotApplicable,
17    NotEstablished,
18    Unavailable,
19}
20
21/// Absence asserted by the component holding the real execution context.
22/// Never inferred from a missing optional string or from an endpoint.
23#[derive(Clone, Copy, Debug, Eq, PartialEq)]
24pub enum DiagnosticContextMissing {
25    NotApplicable,
26    NotEstablished,
27    Unavailable,
28}
29
30/// Safe projection of DB's registered static logical operation, not SQL and not
31/// registration authority. DB must supply its existing registered declaration.
32#[derive(Clone, Copy)]
33pub struct DiagnosticDbOperation(&'static str);
34impl DiagnosticDbOperation {
35    pub fn from_registered(value: &'static str) -> Option<Self> {
36        (!value.is_empty()
37            && value.len() <= 128
38            && value
39                .bytes()
40                .all(|b| b.is_ascii_alphanumeric() || b"._:-".contains(&b)))
41        .then_some(Self(value))
42    }
43}
44
45/// Fixed safe projection of the already accepted ingress zone. This is NOT
46/// ingress validation or routing authority; Service must pass accepted.zone.
47#[derive(Clone, Copy)]
48pub struct DiagnosticZone(ProtocolId);
49#[derive(Clone, Copy, Debug, Eq, PartialEq)]
50pub enum DiagnosticZoneError {
51    Empty,
52    TooLong,
53    Unsafe,
54}
55impl DiagnosticZone {
56    /// Exact copy after bounded safety validation; never hashes or truncates.
57    pub fn from_validated_ingress(value: &str) -> Result<Self, DiagnosticZoneError> {
58        if value.len() > 256 {
59            return Err(DiagnosticZoneError::TooLong);
60        }
61        if value.trim().is_empty() {
62            return Err(DiagnosticZoneError::Empty);
63        }
64        if value.chars().any(char::is_control)
65            || value.contains("://")
66            || value.contains(['@', '?', '#', '\\'])
67        {
68            return Err(DiagnosticZoneError::Unsafe);
69        }
70        Ok(Self(ProtocolId::copy(value)))
71    }
72}
73
74#[derive(Clone, Copy, Eq, PartialEq)]
75struct SafeText {
76    value: ProtocolId,
77    truncated: bool,
78    redacted: bool,
79}
80impl SafeText {
81    fn metadata(value: &str) -> Self {
82        let mut len = value.len().min(256);
83        while !value.is_char_boundary(len) {
84            len -= 1;
85        }
86        let prefix = &value[..len];
87        let redacted = prefix.contains("://")
88            || prefix
89                .chars()
90                .any(|c| !(c.is_alphanumeric() || "_./:{}*-".contains(c)));
91        Self {
92            value: ProtocolId::copy(if redacted { "" } else { prefix }),
93            truncated: len < value.len(),
94            redacted,
95        }
96    }
97}
98impl Serialize for SafeText {
99    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
100        use serde::ser::SerializeStruct;
101        let mut out = s.serialize_struct("SafeText", 3)?;
102        out.serialize_field("value", &self.value)?;
103        out.serialize_field("truncated", &self.truncated)?;
104        out.serialize_field("redacted", &self.redacted)?;
105        out.end()
106    }
107}
108#[derive(Clone, Copy, Eq, PartialEq)]
109struct Span(u64);
110impl Serialize for Span {
111    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
112        let mut bytes = [b'0'; 16];
113        for (i, b) in bytes.iter_mut().enumerate() {
114            *b = b"0123456789abcdef"[((self.0 >> ((15 - i) * 4)) & 15) as usize];
115        }
116        s.serialize_str(std::str::from_utf8(&bytes).unwrap())
117    }
118}
119
120// Exact protocol identity; never replaces a trace by a hash or generated value.
121#[derive(Clone, Copy, Eq, PartialEq)]
122struct ProtocolId {
123    bytes: [u8; 256],
124    len: usize,
125}
126impl ProtocolId {
127    fn copy(value: &str) -> Self {
128        // CallContext protocol types have already enforced this bound.
129        let mut bytes = [0; 256];
130        bytes[..value.len()].copy_from_slice(value.as_bytes());
131        Self {
132            bytes,
133            len: value.len(),
134        }
135    }
136}
137impl Serialize for ProtocolId {
138    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
139        s.serialize_str(std::str::from_utf8(&self.bytes[..self.len]).unwrap_or(""))
140    }
141}
142
143/// Snapshot of existing context, not a second execution context or authority.
144#[derive(Clone, Copy, Serialize)]
145struct Projection {
146    schema_version: u8,
147    application: Field<SafeText>,
148    module: Field<SafeText>,
149    service: Field<SafeText>,
150    operation: Field<SafeText>,
151    db_operation: Field<&'static str>,
152    trace_id: Field<ProtocolId>,
153    rpc_id: Field<ProtocolId>,
154    span_id: Field<Span>,
155    request: Field<SafeText>,
156    // Exact comparison only: never serialized, including when the safe display
157    // is redacted. Distinct requests must not compare equal after redaction.
158    #[serde(skip)]
159    request_binding: Option<ProtocolId>,
160    route: Field<SafeText>,
161    attempt: Field<u32>,
162    scope: Field<saddle_core::DbScopeDiagnosticIdentity>,
163    task: Field<SafeText>,
164    lifecycle: Field<SafeText>,
165    zone: Field<ProtocolId>,
166    target: Field<SafeText>,
167}
168impl Projection {
169    fn missing(reason: DiagnosticContextMissing) -> Self {
170        fn absent<T>(reason: DiagnosticContextMissing) -> Field<T> {
171            match reason {
172                DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
173                DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
174                DiagnosticContextMissing::Unavailable => Field::Unavailable,
175            }
176        }
177        Self {
178            schema_version: 1,
179            application: absent(reason),
180            module: absent(reason),
181            service: absent(reason),
182            operation: absent(reason),
183            db_operation: absent(reason),
184            trace_id: absent(reason),
185            rpc_id: absent(reason),
186            span_id: absent(reason),
187            request: absent(reason),
188            request_binding: None,
189            route: absent(reason),
190            attempt: absent(reason),
191            scope: absent(reason),
192            task: absent(reason),
193            lifecycle: absent(reason),
194            zone: absent(reason),
195            target: absent(reason),
196        }
197    }
198    fn existing(call: &CallContext, event: &EventContext) -> Self {
199        let text = |s: &str| Field::Present(SafeText::metadata(s));
200        Self {
201            schema_version: 1,
202            application: text(call.application().as_str()),
203            module: text(call.module().as_str()),
204            service: text(call.service().as_str()),
205            operation: text(call.operation().as_str()),
206            db_operation: Field::Unavailable,
207            trace_id: Field::Present(ProtocolId::copy(call.trace_correlation_id().as_str())),
208            rpc_id: call.rpc_correlation_id().map_or(Field::Unavailable, |id| {
209                Field::Present(ProtocolId::copy(id.as_str()))
210            }),
211            span_id: Field::Present(Span(call.span_id().as_u64())),
212            request: text(event.diagnostic_request()),
213            request_binding: Some(ProtocolId::copy(event.diagnostic_request())),
214            route: text(event.diagnostic_route()),
215            attempt: Field::Present(event.diagnostic_attempt()),
216            // These facts are not present in CallContext/EventContext. Do not
217            // infer not_applicable from their absence or mint scope/task IDs.
218            scope: Field::Unavailable,
219            task: Field::Unavailable,
220            lifecycle: Field::Unavailable,
221            zone: Field::Unavailable,
222            target: Field::Unavailable,
223        }
224    }
225}
226
227/// Closed missing-field vocabulary. This cannot inject arbitrary JSON keys.
228#[derive(Clone, Copy, Debug, Eq, PartialEq)]
229pub enum DiagnosticContextField {
230    Application,
231    Module,
232    Service,
233    Operation,
234    DbOperation,
235    Trace,
236    Rpc,
237    Span,
238    Request,
239    Route,
240    Attempt,
241    Scope,
242    Task,
243    Lifecycle,
244    Zone,
245    Target,
246}
247
248/// Atomic binding failure: original projection/output can be recovered unchanged.
249#[must_use = "recover the original context; do not silently discard known identity"]
250pub struct ContextBindingError<T> {
251    field: DiagnosticContextField,
252    original: T,
253}
254impl<T> ContextBindingError<T> {
255    pub fn field(&self) -> DiagnosticContextField {
256        self.field
257    }
258    pub fn into_original(self) -> T {
259        self.original
260    }
261}
262fn known<T: Copy + Eq>(
263    field: &mut Field<T>,
264    value: T,
265    name: DiagnosticContextField,
266) -> Result<(), DiagnosticContextField> {
267    if let Field::Present(old) = field
268        && *old != value
269    {
270        return Err(name);
271    }
272    *field = Field::Present(value);
273    Ok(())
274}
275fn bind_call(context: &mut Projection, call: &CallContext) -> Result<(), DiagnosticContextField> {
276    use DiagnosticContextField as F;
277    known(
278        &mut context.application,
279        SafeText::metadata(call.application().as_str()),
280        F::Application,
281    )?;
282    known(
283        &mut context.module,
284        SafeText::metadata(call.module().as_str()),
285        F::Module,
286    )?;
287    known(
288        &mut context.service,
289        SafeText::metadata(call.service().as_str()),
290        F::Service,
291    )?;
292    known(
293        &mut context.operation,
294        SafeText::metadata(call.operation().as_str()),
295        F::Operation,
296    )?;
297    known(
298        &mut context.trace_id,
299        ProtocolId::copy(call.trace_correlation_id().as_str()),
300        F::Trace,
301    )?;
302    known(&mut context.span_id, Span(call.span_id().as_u64()), F::Span)?;
303    if let Some(rpc) = call.rpc_correlation_id() {
304        known(&mut context.rpc_id, ProtocolId::copy(rpc.as_str()), F::Rpc)?;
305    }
306    Ok(())
307}
308fn bind_event(
309    context: &mut Projection,
310    event: &EventContext,
311) -> Result<(), DiagnosticContextField> {
312    use DiagnosticContextField as F;
313    bind_request(context, event.diagnostic_request())?;
314    known(
315        &mut context.route,
316        SafeText::metadata(event.diagnostic_route()),
317        F::Route,
318    )?;
319    known(&mut context.attempt, event.diagnostic_attempt(), F::Attempt)
320}
321fn bind_request(context: &mut Projection, request: &str) -> Result<(), DiagnosticContextField> {
322    let exact = ProtocolId::copy(request);
323    if context.request_binding.is_some_and(|old| old != exact) {
324        return Err(DiagnosticContextField::Request);
325    }
326    known(
327        &mut context.request,
328        SafeText::metadata(request),
329        DiagnosticContextField::Request,
330    )?;
331    context.request_binding = Some(exact);
332    Ok(())
333}
334
335/// Actual request phase supplied by its execution owner; not admission authority.
336#[derive(Clone, Copy)]
337pub enum DiagnosticRequestPhase {
338    SocketAccepted,
339    Admission,
340    ReadingHead,
341    ReadingBody,
342    Validation,
343    Dispatch,
344    Response,
345    TaskJoin,
346    Finalization,
347}
348impl DiagnosticRequestPhase {
349    fn as_str(self) -> &'static str {
350        match self {
351            Self::SocketAccepted => "socket_accepted",
352            Self::Admission => "admission",
353            Self::ReadingHead => "reading_head",
354            Self::ReadingBody => "reading_body",
355            Self::Validation => "validation",
356            Self::Dispatch => "dispatch",
357            Self::Response => "response",
358            Self::TaskJoin => "task_join",
359            Self::Finalization => "finalization",
360        }
361    }
362}
363
364/// Validated read-only rendering of the *existing* runtime task instance ID.
365/// This does not allocate a task/sequence or prove task ownership. RP must take
366/// the digits from its real runtime ID, never a task-kind label or invented ID.
367#[derive(Clone, Copy)]
368pub struct DiagnosticTaskId(ProtocolId);
369impl DiagnosticTaskId {
370    pub fn from_runtime_id(value: &str) -> Option<Self> {
371        (!value.is_empty() && value.len() <= 64 && value.bytes().all(|b| b.is_ascii_digit()))
372            .then(|| Self(ProtocolId::copy(value)))
373    }
374}
375
376/// Fixed early-request projection. It is metadata, not a second execution
377/// context, admission permission or source receipt. No trace/sequence is minted.
378/// ```compile_fail
379/// let forged = saddle_observability::EarlyRequestContext { context: todo!() };
380/// ```
381pub struct EarlyRequestContext {
382    context: Projection,
383}
384// Recovery intentionally returns the fixed inline original, never a heap Box.
385#[allow(clippy::result_large_err)]
386impl EarlyRequestContext {
387    /// No request bytes/identities have been accepted yet. Application is the
388    /// existing configured application label, NOT peer/body/header data.
389    pub fn socket_accepted(application: &str) -> Self {
390        let mut context = Projection::missing(DiagnosticContextMissing::NotEstablished);
391        context.application = Field::Present(SafeText::metadata(application));
392        context.lifecycle = Field::Present(SafeText::metadata("socket_accepted"));
393        Self { context }
394    }
395    /// Facts cannot be recovered at this boundary. Known facts must be attached
396    /// below; this is distinct from an identity that has never been established.
397    pub fn unavailable() -> Self {
398        Self {
399            context: Projection::missing(DiagnosticContextMissing::Unavailable),
400        }
401    }
402    fn update(
403        mut self,
404        bind: impl FnOnce(&mut Projection) -> Result<(), DiagnosticContextField>,
405    ) -> Result<Self, ContextBindingError<Self>> {
406        let mut next = self.context;
407        if let Err(field) = bind(&mut next) {
408            return Err(ContextBindingError {
409                field,
410                original: self,
411            });
412        }
413        self.context = next;
414        Ok(self)
415    }
416    pub fn with_application(
417        self,
418        application: &saddle_core::ApplicationId,
419    ) -> Result<Self, ContextBindingError<Self>> {
420        self.update(|c| {
421            known(
422                &mut c.application,
423                SafeText::metadata(application.as_str()),
424                DiagnosticContextField::Application,
425            )
426        })
427    }
428    /// An already validated protocol identity. Does not generate a root span.
429    pub fn with_trace(
430        self,
431        trace: &saddle_core::TraceCorrelationId,
432    ) -> Result<Self, ContextBindingError<Self>> {
433        self.update(|c| {
434            known(
435                &mut c.trace_id,
436                ProtocolId::copy(trace.as_str()),
437                DiagnosticContextField::Trace,
438            )
439        })
440    }
441    pub fn with_rpc(
442        self,
443        rpc: &saddle_core::RpcCorrelationId,
444    ) -> Result<Self, ContextBindingError<Self>> {
445        self.update(|c| {
446            known(
447                &mut c.rpc_id,
448                ProtocolId::copy(rpc.as_str()),
449                DiagnosticContextField::Rpc,
450            )
451        })
452    }
453    pub fn with_event(self, event: &EventContext) -> Result<Self, ContextBindingError<Self>> {
454        self.update(|c| bind_event(c, event))
455    }
456    pub fn with_call(self, call: &CallContext) -> Result<Self, ContextBindingError<Self>> {
457        self.update(|c| bind_call(c, call))
458    }
459    pub fn with_request(
460        self,
461        request: &crate::RequestIdentity,
462    ) -> Result<Self, ContextBindingError<Self>> {
463        self.update(|c| bind_request(c, request.as_str()))
464    }
465    pub fn with_route(
466        self,
467        route: &crate::RouteIdentity,
468    ) -> Result<Self, ContextBindingError<Self>> {
469        self.update(|c| {
470            known(
471                &mut c.route,
472                SafeText::metadata(route.as_str()),
473                DiagnosticContextField::Route,
474            )
475        })
476    }
477    pub fn with_module(
478        self,
479        module: &saddle_core::ModuleId,
480    ) -> Result<Self, ContextBindingError<Self>> {
481        self.update(|c| {
482            known(
483                &mut c.module,
484                SafeText::metadata(module.as_str()),
485                DiagnosticContextField::Module,
486            )
487        })
488    }
489    pub fn with_service(
490        self,
491        service: &saddle_core::ServiceId,
492    ) -> Result<Self, ContextBindingError<Self>> {
493        self.update(|c| {
494            known(
495                &mut c.service,
496                SafeText::metadata(service.as_str()),
497                DiagnosticContextField::Service,
498            )
499        })
500    }
501    pub fn with_operation(
502        self,
503        operation: &saddle_core::OperationId,
504    ) -> Result<Self, ContextBindingError<Self>> {
505        self.update(|c| {
506            known(
507                &mut c.operation,
508                SafeText::metadata(operation.as_str()),
509                DiagnosticContextField::Operation,
510            )
511        })
512    }
513}
514
515/// Bind once to the existing established request. No empty/default constructor.
516///
517/// ```compile_fail
518/// use saddle_observability::{RequestDiagnosticScope, EmergencyDiagnosticHandle};
519/// fn missing_context(output: &EmergencyDiagnosticHandle) {
520///     let scope = RequestDiagnosticScope::established(output);
521/// }
522/// ```
523pub struct RequestDiagnosticScope<'a> {
524    output: Option<&'a EmergencyDiagnosticHandle>,
525    context: Projection,
526}
527// Same fixed inline recovery model; source capture remains allocation-free.
528#[allow(clippy::result_large_err)]
529impl<'a> RequestDiagnosticScope<'a> {
530    /// A direct transport operation has no established request. This preserves
531    /// the original while marking every request fact as not established.
532    pub fn unassociated_transport(output: &'a EmergencyDiagnosticHandle) -> Self {
533        Self {
534            output: Some(output),
535            context: Projection::missing(DiagnosticContextMissing::NotEstablished),
536        }
537    }
538    /// Preserve validated wire association when the observer could not create
539    /// a CallContext. Missing observer facts stay explicitly unavailable.
540    pub fn outbound_wire(
541        output: &'a EmergencyDiagnosticHandle,
542        event: &EventContext,
543        trace_id: &str,
544        rpc_id: &str,
545        zone: &str,
546    ) -> Option<Self> {
547        let valid = |s: &str| !s.is_empty() && s.len() <= 256 && !s.chars().any(char::is_control);
548        if !valid(trace_id) || !valid(rpc_id) || !valid(zone) { return None; }
549        let mut context = Projection::missing(DiagnosticContextMissing::Unavailable);
550        context.trace_id = Field::Present(ProtocolId::copy(trace_id));
551        context.rpc_id = Field::Present(ProtocolId::copy(rpc_id));
552        context.request = Field::Present(SafeText::metadata(event.diagnostic_request()));
553        context.request_binding = Some(ProtocolId::copy(event.diagnostic_request()));
554        context.route = Field::Present(SafeText::metadata(event.diagnostic_route()));
555        context.attempt = Field::Present(event.diagnostic_attempt());
556        context.zone = Field::Present(ProtocolId::copy(zone));
557        Some(Self { output: Some(output), context })
558    }
559    /// Capture a borrowed foreign error and its caller-supplied, validated
560    /// source target under this exact request projection and occurrence.
561    /// Written means every original segment, including the terminator, was
562    /// synchronously acknowledged by the existing emergency file writer.
563    #[track_caller]
564    pub fn source_error_with_target<T: Serialize>(
565        &self,
566        error: &(dyn std::error::Error + 'static),
567        target: &T,
568        category: DiagnosticCategory,
569        cause: BoundedDiagnosticCause,
570    ) -> RequestSourceReceipt {
571        let diagnostic = BoundedDiagnostic::capture(category, CaptureSite::FirstObserved, cause);
572        let occurrence = diagnostic.occurrence();
573        let submission = original::capture(self.output, &self.context, &diagnostic, error, target);
574        FrameworkRequestFailure {
575            error: (),
576            context: self.context.clone(),
577            occurrence,
578            submission,
579            diagnostic,
580        }
581    }
582    /// Capture a borrowed first error before traversing its static source chain.
583    #[track_caller]
584    pub fn source_borrowed_error_with_target<T: Serialize, E: std::error::Error>(
585        &self,
586        error: &E,
587        target: &T,
588        category: DiagnosticCategory,
589        cause: BoundedDiagnosticCause,
590    ) -> RequestSourceReceipt {
591        let diagnostic = BoundedDiagnostic::capture(category, CaptureSite::FirstObserved, cause);
592        let occurrence = diagnostic.occurrence();
593        let submission = original::capture_borrowed(
594            self.output, &self.context, &diagnostic, error, target,
595        );
596        FrameworkRequestFailure {
597            error: (), context: self.context.clone(), occurrence, submission, diagnostic,
598        }
599    }
600    /// Same source issuer for early and established requests. Output availability
601    /// is independent of identity absence; None never skips source retention.
602    pub fn early(
603        output: Option<&'a EmergencyDiagnosticHandle>,
604        early: EarlyRequestContext,
605    ) -> Self {
606        Self {
607            output,
608            context: early.context,
609        }
610    }
611    /// Move the existing projection to an available/unavailable output without
612    /// replacing any context. Useful when an owning task retains the handle.
613    pub fn with_output<'b>(
614        self,
615        output: Option<&'b EmergencyDiagnosticHandle>,
616    ) -> RequestDiagnosticScope<'b> {
617        RequestDiagnosticScope {
618            output,
619            context: self.context,
620        }
621    }
622    /// Fixed read-only snapshot for a task's parent/child observation sites.
623    /// Copies no receipt or execution owner. Later enrichment is NOT retroactive;
624    /// a join fallback can only claim facts actually retained at spawn time.
625    pub fn reborrow(&self) -> RequestDiagnosticScope<'a> {
626        RequestDiagnosticScope {
627            output: self.output,
628            context: self.context,
629        }
630    }
631    /// Observe an ordinary completed operation with the exact early/established
632    /// projection. Technical failures still require their mandatory carrier.
633    pub fn record_nonfailure(
634        &self,
635        axes: &DiagnosticOutcomeAxes,
636    ) -> Result<DiagnosticSubmission, ()> {
637        if !matches!(
638            axes.operation,
639            saddle_core::OperationOutcome::Succeeded | saddle_core::OperationOutcome::Rejected
640        ) {
641            return Err(());
642        }
643        #[derive(Serialize)]
644        struct NonfailureRecord<'a> {
645            event: &'static str,
646            timestamp_unix_ms: u128,
647            context: &'a Projection,
648            diagnostic: Option<&'a BoundedDiagnostic>,
649            diagnostic_reference: Option<DiagnosticOccurrence>,
650            axes: Option<&'a DiagnosticOutcomeAxes>,
651            source_submission: Option<DiagnosticSubmission>,
652        }
653        Ok(self
654            .output
655            .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
656                output.submit_fixed_record(&NonfailureRecord {
657                    event: "framework.boundary.outcome",
658                    timestamp_unix_ms: timestamp(),
659                    context: &self.context,
660                    diagnostic: None::<&BoundedDiagnostic>,
661                    diagnostic_reference: None::<DiagnosticOccurrence>,
662                    axes: Some(axes),
663                    source_submission: None::<DiagnosticSubmission>,
664                })
665            }))
666    }
667    pub fn with_task(mut self, task: DiagnosticTaskId) -> Result<Self, ContextBindingError<Self>> {
668        if let Err(field) = known(
669            &mut self.context.task,
670            SafeText {
671                value: task.0,
672                truncated: false,
673                redacted: false,
674            },
675            DiagnosticContextField::Task,
676        ) {
677            return Err(ContextBindingError {
678                field,
679                original: self,
680            });
681        }
682        Ok(self)
683    }
684    /// Bind only facts actually established by the ingress owner. On conflict no
685    /// partial update occurs; task/zone/scope and prior receipts stay unchanged.
686    pub fn bind_request_identity(
687        mut self,
688        request: &crate::RequestIdentity,
689    ) -> Result<Self, ContextBindingError<Self>> {
690        if let Err(field) = bind_request(&mut self.context, request.as_str()) {
691            return Err(ContextBindingError {
692                field,
693                original: self,
694            });
695        }
696        Ok(self)
697    }
698
699    pub fn bind_established(
700        mut self,
701        call: &CallContext,
702        event: &EventContext,
703    ) -> Result<Self, ContextBindingError<Self>> {
704        let mut next = self.context;
705        if let Err(field) = bind_call(&mut next, call).and_then(|()| bind_event(&mut next, event)) {
706            return Err(ContextBindingError {
707                field,
708                original: self,
709            });
710        }
711        self.context = next;
712        Ok(self)
713    }
714    /// Derive the actual outbound call without rewriting the inbound scope.
715    /// The caller supplies Boundary's existing child Call/Event (not invented
716    /// identities). Trace and request must already be known and match exactly;
717    /// RPC must be a direct `parent.<decimal sequence>` child and span must differ.
718    /// Missing parent identity cannot establish this relationship. Request
719    /// comparison uses a private exact copy, never the redacted display value.
720    /// This validates diagnostic metadata, not execution/transport authority.
721    pub fn derive_outbound_child(
722        &self,
723        call: &CallContext,
724        event: &EventContext,
725    ) -> Result<RequestDiagnosticScope<'a>, DiagnosticContextField> {
726        use DiagnosticContextField as F;
727        if self.context.trace_id
728            != Field::Present(ProtocolId::copy(call.trace_correlation_id().as_str()))
729        {
730            return Err(F::Trace);
731        }
732        if self.context.request_binding != Some(ProtocolId::copy(event.diagnostic_request())) {
733            return Err(F::Request);
734        }
735        let Field::Present(parent_rpc) = self.context.rpc_id else {
736            return Err(F::Rpc);
737        };
738        let child_rpc = call.rpc_correlation_id().ok_or(F::Rpc)?.as_str();
739        let parent_rpc =
740            std::str::from_utf8(&parent_rpc.bytes[..parent_rpc.len]).map_err(|_| F::Rpc)?;
741        let sequence = child_rpc
742            .strip_prefix(parent_rpc)
743            .and_then(|suffix| suffix.strip_prefix('.'))
744            .ok_or(F::Rpc)?;
745        if sequence.is_empty() || !sequence.bytes().all(|b| b.is_ascii_digit()) {
746            return Err(F::Rpc);
747        }
748        let Field::Present(parent_span) = self.context.span_id else {
749            return Err(F::Span);
750        };
751        if parent_span == Span(call.span_id().as_u64()) {
752            return Err(F::Span);
753        }
754        // Only call-local facts are replaced. All other facts (including their
755        // absence states) and the existing output borrow remain the parent's.
756        let child = Projection::existing(call, event);
757        let mut context = self.context;
758        context.application = child.application;
759        context.module = child.module;
760        context.service = child.service;
761        context.operation = child.operation;
762        context.rpc_id = child.rpc_id;
763        context.span_id = child.span_id;
764        context.route = child.route;
765        context.attempt = child.attempt;
766        Ok(RequestDiagnosticScope {
767            output: self.output,
768            context,
769        })
770    }
771    /// Advances this projection only; retained source receipts are snapshots.
772    pub fn with_phase(mut self, phase: DiagnosticRequestPhase) -> Self {
773        self.context.lifecycle = Field::Present(SafeText::metadata(phase.as_str()));
774        self
775    }
776    /// Missing assertions cannot erase known facts. No arbitrary key/value API.
777    pub fn with_missing(
778        mut self,
779        field: DiagnosticContextField,
780        reason: DiagnosticContextMissing,
781    ) -> Self {
782        fn set<T>(field: &mut Field<T>, reason: DiagnosticContextMissing) {
783            if !matches!(field, Field::Present(_)) {
784                *field = match reason {
785                    DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
786                    DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
787                    DiagnosticContextMissing::Unavailable => Field::Unavailable,
788                };
789            }
790        }
791        match field {
792            DiagnosticContextField::Application => set(&mut self.context.application, reason),
793            DiagnosticContextField::Module => set(&mut self.context.module, reason),
794            DiagnosticContextField::Service => set(&mut self.context.service, reason),
795            DiagnosticContextField::Operation => set(&mut self.context.operation, reason),
796            DiagnosticContextField::DbOperation => set(&mut self.context.db_operation, reason),
797            DiagnosticContextField::Trace => set(&mut self.context.trace_id, reason),
798            DiagnosticContextField::Rpc => set(&mut self.context.rpc_id, reason),
799            DiagnosticContextField::Span => set(&mut self.context.span_id, reason),
800            DiagnosticContextField::Request => set(&mut self.context.request, reason),
801            DiagnosticContextField::Route => set(&mut self.context.route, reason),
802            DiagnosticContextField::Attempt => set(&mut self.context.attempt, reason),
803            DiagnosticContextField::Scope => set(&mut self.context.scope, reason),
804            DiagnosticContextField::Task => set(&mut self.context.task, reason),
805            DiagnosticContextField::Lifecycle => set(&mut self.context.lifecycle, reason),
806            DiagnosticContextField::Zone => set(&mut self.context.zone, reason),
807            DiagnosticContextField::Target => set(&mut self.context.target, reason),
808        }
809        self
810    }
811    /// Projection from a checked live request pair; no SQL-entry side effects.
812    pub fn live_db_scope(
813        output: Option<&'a EmergencyDiagnosticHandle>,
814        projection: &saddle_core::DbScopeDiagnosticContext<(&CallContext, &EventContext)>,
815    ) -> Self {
816        let ((call, event), scope) = projection.diagnostic_context();
817        let mut bound = match output {
818            Some(output) => Self::established(output, call, event),
819            None => Self::output_unavailable(call, event),
820        };
821        bound.context.scope = Field::Present(scope);
822        bound
823    }
824
825    /// Adds only the registered DB operation; original request/scope are intact.
826    pub fn with_db_operation(mut self, operation: DiagnosticDbOperation) -> Self {
827        self.set_db_operation(operation);
828        self
829    }
830    /// Advances only this scope's next-operation projection. Previously issued
831    /// receipts own a value snapshot and are never rewritten.
832    pub fn set_db_operation(&mut self, operation: DiagnosticDbOperation) {
833        self.context.db_operation = Field::Present(operation.0);
834    }
835    pub fn with_db_operation_missing(mut self, reason: DiagnosticContextMissing) -> Self {
836        if !matches!(self.context.db_operation, Field::Present(_)) {
837            self.context.db_operation = match reason {
838                DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
839                DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
840                DiagnosticContextMissing::Unavailable => Field::Unavailable,
841            };
842        }
843        self
844    }
845    /// Source owner supplies the original accepted zone before source capture.
846    /// Every subsequent source/boundary projection includes it automatically.
847    pub fn with_zone(mut self, zone: DiagnosticZone) -> Self {
848        self.context.zone = Field::Present(zone.0);
849        self
850    }
851    pub fn with_zone_missing(mut self, reason: DiagnosticContextMissing) -> Self {
852        // An absence assertion cannot erase an already available zone.
853        if !matches!(self.context.zone, Field::Present(_)) {
854            self.context.zone = match reason {
855                DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
856                DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
857                DiagnosticContextMissing::Unavailable => Field::Unavailable,
858            };
859        }
860        self
861    }
862    /// Takes BOTH context and scope from the same checked token, never a raw ID
863    /// or a separately supplied CallContext. Borrowing leaves terminal use intact.
864    pub fn db_scope(
865        output: &'a EmergencyDiagnosticHandle,
866        observation: &saddle_core::DbScopeObservation<(crate::Observer, CallContext, EventContext)>,
867    ) -> Self {
868        let ((_, call, event), scope) = observation.diagnostic_context();
869        let mut bound = Self::established(output, call, event);
870        bound.context.scope = Field::Present(scope);
871        bound
872    }
873    pub fn db_scope_output_unavailable(
874        observation: &saddle_core::DbScopeObservation<(crate::Observer, CallContext, EventContext)>,
875    ) -> Self {
876        let ((_, call, event), scope) = observation.diagnostic_context();
877        let mut bound = Self::output_unavailable(call, event);
878        bound.context.scope = Field::Present(scope);
879        bound
880    }
881    pub fn established(
882        output: &'a EmergencyDiagnosticHandle,
883        call: &CallContext,
884        event: &EventContext,
885    ) -> Self {
886        Self {
887            output: Some(output),
888            context: Projection::existing(call, event),
889        }
890    }
891
892    /// Explicit degraded route. Context is still mandatory; no successful
893    /// submission or writer readiness is inferred.
894    pub fn output_unavailable(call: &CallContext, event: &EventContext) -> Self {
895        Self {
896            output: None,
897            context: Projection::existing(call, event),
898        }
899    }
900    /// Final bound ingress deadline; no clock reads or new identity.
901    pub fn ingress_deadline_bound(&self, ingress_deadline_unix_ms: i64) -> DiagnosticSubmission {
902        // A redacted/truncated legacy display cannot bind this fact to a request.
903        // Preserve the existing privacy boundary instead of exposing request_binding.
904        if !matches!(self.context.request, Field::Present(SafeText {
905            truncated: false, redacted: false, ..
906        })) {
907            return DiagnosticSubmission::EncodingFailed;
908        }
909        #[derive(Serialize)]
910        struct DeadlineRecord<'a> {
911            schema_version: u8,
912            event: &'static str,
913            ingress_deadline_unix_ms: i64,
914            context: &'a Projection,
915        }
916        self.output.map_or(DiagnosticSubmission::OutputUnavailable, |output|
917            output.submit_fixed_record(&DeadlineRecord {
918                schema_version: 1, event: "ingress_deadline_bound",
919                ingress_deadline_unix_ms, context: &self.context,
920            }))
921    }
922
923    /// Consumes existing source facts without capturing a replacement identity.
924    pub fn capture_required(&self, diagnostic: BoundedDiagnostic) -> RequestSourceReceipt {
925        self.capture_error((), diagnostic)
926    }
927    /// First submission of an already captured panic/cleanup diagnostic.
928    /// Moves the original carrier, never recaptures or clones its stack.
929    /// `observer` preserves the existing main-log mirror when one is available.
930    /// Uses the legacy allocating 32KiB encoder, NOT the admitted bounded path.
931    pub fn capture_existing(
932        &self,
933        diagnostic: saddle_core::Diagnostic,
934        observer: Option<&crate::Observer>,
935    ) -> ExistingDiagnosticReceipt {
936        let occurrence = diagnostic.occurrence();
937        let record = Record {
938            event: "framework.diagnostic",
939            timestamp_unix_ms: timestamp(),
940            context: &self.context,
941            diagnostic: Some(&diagnostic),
942            diagnostic_reference: occurrence,
943            axes: None,
944            source_submission: None,
945        };
946        let submission = self
947            .output
948            .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
949                output.submit_existing_record(&record, diagnostic.deferred_stack())
950            });
951        if let Some(observer) = observer {
952            observer.mirror_existing_record(&record, diagnostic.category());
953        }
954        FrameworkRequestFailure {
955            error: (),
956            context: self.context,
957            occurrence,
958            submission,
959            diagnostic,
960        }
961    }
962
963    /// Capture at the real source, before exposing its error to a caller.
964    /// Submission degradation never changes the source error.
965    #[track_caller]
966    pub fn fail<E>(
967        &self,
968        error: E,
969        category: DiagnosticCategory,
970        cause: BoundedDiagnosticCause,
971    ) -> FrameworkRequestFailure<E> {
972        let diagnostic = BoundedDiagnostic::capture(category, CaptureSite::FirstObserved, cause);
973        self.capture_error(error, diagnostic)
974    }
975    fn capture_error<E>(
976        &self,
977        error: E,
978        diagnostic: BoundedDiagnostic,
979    ) -> FrameworkRequestFailure<E> {
980        let occurrence = diagnostic.occurrence();
981        let submission = self
982            .output
983            .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
984                output.submit_fixed_record(&Record {
985                    event: "framework.diagnostic",
986                    timestamp_unix_ms: timestamp(),
987                    context: &self.context,
988                    diagnostic: Some(&diagnostic),
989                    diagnostic_reference: occurrence,
990                    axes: None,
991                    source_submission: None,
992                })
993            });
994        FrameworkRequestFailure {
995            error,
996            context: self.context,
997            occurrence,
998            submission,
999            diagnostic,
1000        }
1001    }
1002}
1003
1004/// Only the actual context-bound source submission above constructs this value.
1005/// No Clone, Default, From<E>, raw identity constructor, or context replacement.
1006///
1007/// ```compile_fail
1008/// use saddle_observability::FrameworkRequestFailure;
1009/// fn naked() -> Result<(), FrameworkRequestFailure<u32>> { Err(17) }
1010/// ```
1011/// ```compile_fail
1012/// use saddle_observability::FrameworkRequestFailure;
1013/// fn bypass() -> Result<(), FrameworkRequestFailure<u32>> {
1014///     let value: Result<(), u32> = Err(17);
1015///     value?;
1016///     Ok(())
1017/// }
1018/// ```
1019/// ```compile_fail
1020/// use saddle_observability::FrameworkRequestFailure;
1021/// fn forged(source: FrameworkRequestFailure<u32>) -> FrameworkRequestFailure<u32> {
1022///     FrameworkRequestFailure { error: 17, ..source }
1023/// }
1024/// ```
1025#[must_use = "retain the source receipt with the technical result until its declared boundary"]
1026pub struct FrameworkRequestFailure<E, D = BoundedDiagnostic> {
1027    error: E,
1028    context: Projection,
1029    occurrence: DiagnosticOccurrence,
1030    submission: DiagnosticSubmission,
1031    diagnostic: D,
1032}
1033/// Actual source capture/submission receipt; no public construction or Clone.
1034/// ```compile_fail
1035/// fn duplicate(receipt: saddle_observability::RequestSourceReceipt) {
1036///     let _second = receipt.clone();
1037/// }
1038/// ```
1039/// ```compile_fail
1040/// fn missing() -> saddle_observability::RequestSourceReceipt {
1041///     Default::default()
1042/// }
1043/// ```
1044pub type RequestSourceReceipt = FrameworkRequestFailure<()>;
1045pub type ExistingDiagnosticReceipt = FrameworkRequestFailure<(), saddle_core::Diagnostic>;
1046impl<E, D> FrameworkRequestFailure<E, D> {
1047    pub fn error(&self) -> &E {
1048        &self.error
1049    }
1050    pub fn submission(&self) -> DiagnosticSubmission {
1051        self.submission
1052    }
1053    pub fn map_error<F>(self, map: impl FnOnce(E) -> F) -> FrameworkRequestFailure<F, D> {
1054        FrameworkRequestFailure {
1055            error: map(self.error),
1056            context: self.context,
1057            occurrence: self.occurrence,
1058            submission: self.submission,
1059            diagnostic: self.diagnostic,
1060        }
1061    }
1062    pub fn record_boundary(
1063        &self,
1064        output: &EmergencyDiagnosticHandle,
1065        axes: &DiagnosticOutcomeAxes,
1066    ) -> DiagnosticSubmission {
1067        self.record_boundary_optional(Some(output), axes)
1068    }
1069    pub fn record_boundary_optional(
1070        &self,
1071        output: Option<&EmergencyDiagnosticHandle>,
1072        axes: &DiagnosticOutcomeAxes,
1073    ) -> DiagnosticSubmission {
1074        output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
1075            output.submit_fixed_record(&Record {
1076                event: "framework.boundary.outcome",
1077                timestamp_unix_ms: timestamp(),
1078                context: &self.context,
1079                diagnostic: None::<&BoundedDiagnostic>,
1080                diagnostic_reference: self.occurrence,
1081                axes: Some(axes),
1082                source_submission: Some(self.submission),
1083            })
1084        })
1085    }
1086    /// Declared terminal observation without discarding the original error or
1087    /// diagnostic on missing/full/closed output. The returned SAME mandatory
1088    /// carrier must remain in the enclosing result until its actual terminal.
1089    /// This does not prove that arbitrary Rust drop/forget can be forbidden.
1090    pub fn finish_boundary_retained(
1091        self,
1092        output: Option<&EmergencyDiagnosticHandle>,
1093        axes: &DiagnosticOutcomeAxes,
1094    ) -> (Self, BoundaryDiagnosticDelivery) {
1095        let delivery = BoundaryDiagnosticDelivery {
1096            source: self.submission,
1097            boundary: self.record_boundary_optional(output, axes),
1098            occurrence: self.occurrence,
1099        };
1100        (self, delivery)
1101    }
1102    pub fn source_diagnostic(&self) -> &D {
1103        &self.diagnostic
1104    }
1105    /// Sole consuming release of the original error: records its boundary first.
1106    /// The enclosing component must use this only at its declared public boundary,
1107    /// not as an adapter back to a legacy internal technical-failure path.
1108    pub fn finish_boundary(
1109        self,
1110        output: &EmergencyDiagnosticHandle,
1111        axes: &DiagnosticOutcomeAxes,
1112    ) -> (E, BoundaryDiagnosticDelivery) {
1113        let boundary = self.record_boundary(output, axes);
1114        (
1115            self.error,
1116            BoundaryDiagnosticDelivery {
1117                source: self.submission,
1118                boundary,
1119                occurrence: self.occurrence,
1120            },
1121        )
1122    }
1123}
1124impl<D> FrameworkRequestFailure<(), D> {
1125    /// Consume a capture receipt into its context-preserving boundary reference.
1126    pub fn into_reference(self) -> RequestBoundaryReference<D> {
1127        RequestBoundaryReference {
1128            context: self.context,
1129            occurrence: self.occurrence,
1130            submission: self.submission,
1131            diagnostic: self.diagnostic,
1132        }
1133    }
1134}
1135#[must_use = "retain source facts and submission status through terminal consumption"]
1136pub struct RequestBoundaryReference<D = BoundedDiagnostic> {
1137    context: Projection,
1138    occurrence: DiagnosticOccurrence,
1139    submission: DiagnosticSubmission,
1140    diagnostic: D,
1141}
1142impl<D> RequestBoundaryReference<D> {
1143    pub(crate) fn record_stage(
1144        &self,
1145        output: Option<&EmergencyDiagnosticHandle>,
1146        axes: &DiagnosticOutcomeAxes,
1147        stage: &'static str,
1148        elapsed_ms: u64,
1149    ) -> BoundaryDiagnosticDelivery {
1150        #[derive(Serialize)]
1151        struct StageRecord<'a> {
1152            #[serde(flatten)]
1153            record: Record<'a>,
1154            stage: &'static str,
1155            elapsed_ms: u64,
1156        }
1157        let boundary = output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
1158            output.submit_fixed_record(&StageRecord {
1159                record: Record {
1160                    event: "framework.boundary.outcome",
1161                    timestamp_unix_ms: timestamp(),
1162                    context: &self.context,
1163                    diagnostic: None,
1164                    diagnostic_reference: self.occurrence,
1165                    axes: Some(axes),
1166                    source_submission: Some(self.submission),
1167                },
1168                stage,
1169                elapsed_ms,
1170            })
1171        });
1172        BoundaryDiagnosticDelivery {
1173            source: self.submission,
1174            boundary,
1175            occurrence: self.occurrence,
1176        }
1177    }
1178    pub fn occurrence(&self) -> DiagnosticOccurrence {
1179        self.occurrence
1180    }
1181    pub fn source_submission(&self) -> DiagnosticSubmission {
1182        self.submission
1183    }
1184    /// Retained safe source facts, including when no output could accept them.
1185    pub fn source_diagnostic(&self) -> &D {
1186        &self.diagnostic
1187    }
1188    pub fn record(
1189        &self,
1190        output: &EmergencyDiagnosticHandle,
1191        axes: &DiagnosticOutcomeAxes,
1192    ) -> DiagnosticSubmission {
1193        self.record_optional(Some(output), axes)
1194    }
1195    pub fn record_optional(
1196        &self,
1197        output: Option<&EmergencyDiagnosticHandle>,
1198        axes: &DiagnosticOutcomeAxes,
1199    ) -> DiagnosticSubmission {
1200        output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
1201            output.submit_fixed_record(&Record {
1202                event: "framework.boundary.outcome",
1203                timestamp_unix_ms: timestamp(),
1204                context: &self.context,
1205                diagnostic: None::<&BoundedDiagnostic>,
1206                diagnostic_reference: self.occurrence,
1207                axes: Some(axes),
1208                source_submission: Some(self.submission),
1209            })
1210        })
1211    }
1212    /// Moves and returns the same retained reference, never a replacement receipt.
1213    pub fn finish_retained(
1214        self,
1215        output: Option<&EmergencyDiagnosticHandle>,
1216        axes: &DiagnosticOutcomeAxes,
1217    ) -> (Self, BoundaryDiagnosticDelivery) {
1218        let delivery = BoundaryDiagnosticDelivery {
1219            source: self.submission,
1220            boundary: self.record_optional(output, axes),
1221            occurrence: self.occurrence,
1222        };
1223        (self, delivery)
1224    }
1225}
1226/// Observation only. Neither status claims that a record was written.
1227pub struct BoundaryDiagnosticDelivery {
1228    source: DiagnosticSubmission,
1229    boundary: DiagnosticSubmission,
1230    occurrence: DiagnosticOccurrence,
1231}
1232impl BoundaryDiagnosticDelivery {
1233    pub fn source_submission(&self) -> DiagnosticSubmission {
1234        self.source
1235    }
1236    pub fn boundary_submission(&self) -> DiagnosticSubmission {
1237        self.boundary
1238    }
1239    pub fn occurrence(&self) -> DiagnosticOccurrence {
1240        self.occurrence
1241    }
1242}
1243#[derive(Serialize)]
1244struct Record<'a, D = BoundedDiagnostic> {
1245    event: &'static str,
1246    timestamp_unix_ms: u128,
1247    context: &'a Projection,
1248    diagnostic: Option<&'a D>,
1249    diagnostic_reference: DiagnosticOccurrence,
1250    axes: Option<&'a DiagnosticOutcomeAxes>,
1251    #[serde(skip_serializing_if = "Option::is_none")]
1252    source_submission: Option<DiagnosticSubmission>,
1253}
1254fn timestamp() -> u128 {
1255    std::time::SystemTime::now()
1256        .duration_since(std::time::UNIX_EPOCH)
1257        .unwrap_or_default()
1258        .as_millis()
1259}
1260
1261#[cfg(test)]
1262mod early_tests {
1263    use super::*;
1264    use saddle_core::*;
1265    fn ok<T>(value: std::result::Result<T, ContextBindingError<T>>) -> T {
1266        match value {
1267            Ok(value) => value,
1268            Err(_) => panic!("unexpected context conflict"),
1269        }
1270    }
1271    fn cause() -> BoundedDiagnosticCause {
1272        BoundedDiagnosticCause::new(
1273            DiagnosticStage::RequestDecode,
1274            DiagnosticCode::new("request.read_failed").unwrap(),
1275        )
1276    }
1277    fn context(scope: &RequestDiagnosticScope<'_>) -> serde_json::Value {
1278        serde_json::to_value(scope.context).unwrap()
1279    }
1280    #[test]
1281    fn early_absence_and_output_are_independent_and_receipt_retains_error() {
1282        let scope =
1283            RequestDiagnosticScope::early(None, EarlyRequestContext::socket_accepted("app"))
1284                .with_missing(
1285                    DiagnosticContextField::DbOperation,
1286                    DiagnosticContextMissing::NotApplicable,
1287                )
1288                .with_missing(
1289                    DiagnosticContextField::Application,
1290                    DiagnosticContextMissing::Unavailable,
1291                )
1292                .with_phase(DiagnosticRequestPhase::ReadingHead);
1293        let value = context(&scope);
1294        assert_eq!(value["application"]["value"]["value"], "app");
1295        for field in ["trace_id", "rpc_id", "span_id", "request", "scope", "task"] {
1296            assert_eq!(value[field]["state"], "not_established");
1297            assert!(value[field].get("value").is_none());
1298        }
1299        assert_eq!(value["db_operation"]["state"], "not_applicable");
1300        assert_eq!(value["lifecycle"]["value"]["value"], "reading_head");
1301        let failure = scope.fail(17u32, DiagnosticCategory::UnexpectedError, cause());
1302        let id = failure.source_diagnostic().id();
1303        assert_eq!(
1304            failure.submission(),
1305            DiagnosticSubmission::OutputUnavailable
1306        );
1307        let mapped = failure.map_error(|code| code + 1);
1308        let (retained, delivery) =
1309            mapped.finish_boundary_retained(None, &DiagnosticOutcomeAxes::default());
1310        assert_eq!(*retained.error(), 18);
1311        assert_eq!(retained.source_diagnostic().id(), id);
1312        assert_eq!(
1313            delivery.source_submission(),
1314            DiagnosticSubmission::OutputUnavailable
1315        );
1316        assert_eq!(
1317            delivery.boundary_submission(),
1318            DiagnosticSubmission::OutputUnavailable
1319        );
1320        let lost = RequestDiagnosticScope::early(None, EarlyRequestContext::unavailable());
1321        assert_eq!(context(&lost)["trace_id"]["state"], "unavailable");
1322    }
1323    #[test]
1324    fn known_partial_identity_atomic_conflict_and_promotion_preserve_snapshots() {
1325        let trace = TraceCorrelationId::new("gateway-opaque-原值").unwrap();
1326        let foreign = TraceCorrelationId::new("foreign").unwrap();
1327        let early = ok(EarlyRequestContext::socket_accepted("app").with_trace(&trace));
1328        let early = match early.with_trace(&foreign) {
1329            Ok(_) => panic!("foreign trace accepted"),
1330            Err(error) => {
1331                assert_eq!(error.field(), DiagnosticContextField::Trace);
1332                error.into_original()
1333            }
1334        };
1335        let request = crate::RequestIdentity::new("request-1").unwrap();
1336        let early = ok(early.with_request(&request));
1337        let scope = ok(RequestDiagnosticScope::early(None, early)
1338            .with_task(DiagnosticTaskId::from_runtime_id("42").unwrap()));
1339        let before = context(&scope);
1340        let receipt = scope
1341            .fail((), DiagnosticCategory::UnexpectedError, cause())
1342            .into_reference();
1343        let call = CallContext::new(
1344            "app".into(),
1345            "module".into(),
1346            "service".into(),
1347            "operation".into(),
1348            TraceId::from_u128(1),
1349            SpanId::from_u64(2),
1350        )
1351        .with_trace_correlation_id(foreign);
1352        let event =
1353            EventContext::new(request, crate::RouteIdentity::new("/route").unwrap(), 1).unwrap();
1354        let scope = match scope.bind_established(&call, &event) {
1355            Ok(_) => panic!("foreign promotion accepted"),
1356            Err(error) => error.into_original(),
1357        };
1358        assert_eq!(context(&scope), before); // module/service partial writes rolled back
1359        let call = call.with_trace_correlation_id(trace);
1360        let scope = ok(scope.bind_established(&call, &event));
1361        let after = context(&scope);
1362        assert_eq!(after["trace_id"]["value"], "gateway-opaque-原值");
1363        assert_eq!(after["task"]["value"]["value"], "42");
1364        assert_eq!(after["span_id"]["value"], "0000000000000002");
1365        assert_eq!(serde_json::to_value(receipt.context).unwrap(), before);
1366        assert!(
1367            scope.with_output(None).context.request
1368                == Field::Present(SafeText::metadata("request-1"))
1369        );
1370    }
1371    #[test]
1372    fn task_projection_is_bounded_and_cannot_be_replaced() {
1373        for unsafe_id in ["", "request_task", "-1", "https://secret", "12\n3"] {
1374            assert!(DiagnosticTaskId::from_runtime_id(unsafe_id).is_none());
1375        }
1376        assert!(DiagnosticTaskId::from_runtime_id(&"1".repeat(65)).is_none());
1377        let scope = ok(RequestDiagnosticScope::early(
1378            None,
1379            EarlyRequestContext::socket_accepted("app"),
1380        )
1381        .with_task(DiagnosticTaskId::from_runtime_id("42").unwrap()));
1382        let original = context(&scope);
1383        let scope = match scope.with_task(DiagnosticTaskId::from_runtime_id("43").unwrap()) {
1384            Ok(_) => panic!("changed task identity"),
1385            Err(error) => error.into_original(),
1386        };
1387        assert_eq!(context(&scope), original);
1388    }
1389}
1390
1391#[cfg(test)]
1392mod outbound_child_tests {
1393    use super::*;
1394    use saddle_core::*;
1395
1396    fn event(request: &str, route: &str) -> EventContext {
1397        EventContext::new(
1398            crate::RequestIdentity::new(request).unwrap(),
1399            crate::RouteIdentity::new(route).unwrap(),
1400            1,
1401        )
1402        .unwrap()
1403    }
1404    fn call(trace: &str, rpc: &str, span: u64) -> CallContext {
1405        CallContext::new(
1406            "saddle".into(),
1407            "zone-a".into(),
1408            "profusecontract".into(),
1409            "invoke".into(),
1410            TraceId::from_u128(1),
1411            SpanId::from_u64(span),
1412        )
1413        .with_trace_correlation_id(TraceCorrelationId::new(trace).unwrap())
1414        .with_rpc_correlation_id(RpcCorrelationId::new(rpc))
1415    }
1416    fn cause() -> BoundedDiagnosticCause {
1417        BoundedDiagnosticCause::new(
1418            DiagnosticStage::RequestDecode,
1419            DiagnosticCode::new("outbound.connect_failed").unwrap(),
1420        )
1421    }
1422    #[test]
1423    fn outbound_child_preserves_live_scope_and_receipts() {
1424        let parent = call("opaque-trace", "0.4", 1);
1425        let child = call("opaque-trace", "0.4.1", 2);
1426        let parent_event = event("request-1", "/incoming");
1427        let child_event = event("request-1", "remote.function");
1428        let (_, issuer) = DbPhysicalDispositionIssuer::issue().into_startup_and_request_issuer();
1429        let (request, execution) = issuer.issue_request().unwrap();
1430        let live = request
1431            .project_diagnostic_context(&execution, (&parent, &parent_event))
1432            .ok()
1433            .unwrap();
1434        let scope = RequestDiagnosticScope::live_db_scope(None, &live)
1435            .with_task(DiagnosticTaskId::from_runtime_id("42").unwrap())
1436            .unwrap_or_else(|_| panic!("task"))
1437            .with_zone(DiagnosticZone::from_validated_ingress("zone-a").unwrap())
1438            .with_db_operation(DiagnosticDbOperation::from_registered("orders.query").unwrap());
1439        let original = serde_json::to_value(scope.context).unwrap();
1440        let old = scope.fail(7, DiagnosticCategory::UnexpectedError, cause());
1441        let derived = scope.derive_outbound_child(&child, &child_event).unwrap();
1442        let value = serde_json::to_value(derived.context).unwrap();
1443        for field in [
1444            "request",
1445            "trace_id",
1446            "task",
1447            "zone",
1448            "scope",
1449            "db_operation",
1450            "lifecycle",
1451            "target",
1452        ] {
1453            assert_eq!(value[field], original[field], "{field}");
1454        }
1455        assert_eq!(value["rpc_id"]["value"], "0.4.1");
1456        assert_eq!(value["span_id"]["value"], "0000000000000002");
1457        assert_eq!(value["route"]["value"]["value"], "remote.function");
1458        assert_eq!(serde_json::to_value(scope.context).unwrap(), original);
1459        assert_eq!(serde_json::to_value(old.context).unwrap(), original);
1460        assert!(
1461            scope
1462                .reborrow()
1463                .bind_established(&child, &child_event)
1464                .is_err()
1465        );
1466        let source = derived.fail(9, DiagnosticCategory::UnexpectedError, cause());
1467        let id = source.source_diagnostic().id();
1468        let (retained, delivery) =
1469            source.finish_boundary_retained(None, &DiagnosticOutcomeAxes::default());
1470        assert_eq!(retained.source_diagnostic().id(), id);
1471        assert_eq!(*retained.error(), 9);
1472        assert_eq!(
1473            delivery.source_submission(),
1474            DiagnosticSubmission::OutputUnavailable
1475        );
1476        assert_eq!(
1477            delivery.boundary_submission(),
1478            DiagnosticSubmission::OutputUnavailable
1479        );
1480        assert_eq!(serde_json::to_value(retained.context).unwrap(), value);
1481    }
1482
1483    #[test]
1484    fn outbound_child_rejects_foreign_missing_and_redaction_collisions() {
1485        let parent = call("trace", "0", 1);
1486        // Both display values redact; equality must use the private exact input.
1487        let scope = RequestDiagnosticScope::output_unavailable(&parent, &event("a@b", "/in"));
1488        let before = serde_json::to_value(scope.context).unwrap();
1489        let child_event = event("a@b", "remote");
1490        for (child, ev, expected) in [
1491            (
1492                call("foreign", "0.1", 2),
1493                child_event.clone(),
1494                DiagnosticContextField::Trace,
1495            ),
1496            (
1497                call("trace", "0.1", 2),
1498                event("c@d", "remote"),
1499                DiagnosticContextField::Request,
1500            ),
1501            (
1502                call("trace", "01.1", 2),
1503                child_event.clone(),
1504                DiagnosticContextField::Rpc,
1505            ),
1506            (
1507                call("trace", "0", 2),
1508                child_event.clone(),
1509                DiagnosticContextField::Rpc,
1510            ),
1511            (
1512                call("trace", "0.1.2", 2),
1513                child_event.clone(),
1514                DiagnosticContextField::Rpc,
1515            ),
1516            (
1517                call("trace", "0.x", 2),
1518                child_event.clone(),
1519                DiagnosticContextField::Rpc,
1520            ),
1521            (
1522                call("trace", "0.1", 1),
1523                child_event.clone(),
1524                DiagnosticContextField::Span,
1525            ),
1526        ] {
1527            assert_eq!(
1528                scope.derive_outbound_child(&child, &ev).err(),
1529                Some(expected)
1530            );
1531            assert_eq!(serde_json::to_value(scope.context).unwrap(), before);
1532        }
1533        let child = call("trace", "0.1", 2);
1534        let derived = scope.derive_outbound_child(&child, &child_event).unwrap();
1535        let json = serde_json::to_string(&derived.context).unwrap();
1536        assert!(!json.contains("a@b") && !json.contains("request_binding"));
1537        assert!(
1538            RequestDiagnosticScope::early(None, EarlyRequestContext::unavailable())
1539                .derive_outbound_child(&child, &child_event)
1540                .is_err()
1541        );
1542        let missing_rpc = parent.clone().with_rpc_correlation_id(None);
1543        assert_eq!(
1544            RequestDiagnosticScope::output_unavailable(&missing_rpc, &child_event)
1545                .derive_outbound_child(&child, &child_event)
1546                .err(),
1547            Some(DiagnosticContextField::Rpc)
1548        );
1549        assert!(
1550            scope
1551                .reborrow()
1552                .bind_request_identity(&crate::RequestIdentity::new("c@d").unwrap())
1553                .is_err()
1554        );
1555        assert!(
1556            EarlyRequestContext::unavailable()
1557                .with_request(&crate::RequestIdentity::new("a@b").unwrap())
1558                .unwrap_or_else(|_| panic!("first identity"))
1559                .with_request(&crate::RequestIdentity::new("c@d").unwrap())
1560                .is_err()
1561        );
1562    }
1563}