Skip to main content

query_runtime/observability/
attempt_report.rs

1//! Typed sealed execution-attempt reports and terminal facts.
2
3use std::fmt;
4
5use query_planning::error::{
6    AllocationSource, AttemptFailureCode, QueryError, QueryErrorCode, QueryErrorStatus, Result,
7};
8use serde::{
9    Deserialize, Serialize,
10    de::{self, SeqAccess, Visitor},
11};
12
13use crate::observability::{AttemptComposedScheduleProjection, AttemptId};
14
15/// Schema label of [`QueryAttemptReport`]. Decoding rejects every other label.
16pub const QUERY_ATTEMPT_REPORT_SCHEMA: &str = "query_engine.query_attempt_report";
17/// Schema label for the bounded public projection of a sealed attempt report.
18///
19/// The projection deliberately omits correlations, diagnostics, allocator
20/// events, and other payloads that are not suitable for fixed-cardinality
21/// consumers such as DataFusion metrics and retained Flight control metadata.
22pub const SEALED_ATTEMPT_PROJECTION_SCHEMA: &str = "query_engine.sealed_attempt_projection";
23/// Maximum UTF-8 byte length for caller-supplied attempt query correlation.
24pub const MAX_ATTEMPT_QUERY_ID_BYTES: usize = 256;
25/// Maximum UTF-8 byte length for caller-supplied opaque trace context.
26pub const MAX_ATTEMPT_TRACE_CONTEXT_BYTES: usize = 1024;
27
28macro_rules! attempt_correlation {
29    ($name:ident, $limit:ident, $field:literal, $doc:literal) => {
30        #[doc = $doc]
31        #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize)]
32        #[serde(transparent)]
33        pub struct $name(String);
34        impl $name {
35            /// Preserves the supplied UTF-8 correlation after validation.
36            ///
37            /// Returns `InvalidAttemptCorrelation` for blank text or text
38            /// exceeding this correlation type's byte limit. Whitespace is
39            /// used to detect blank values and is otherwise preserved.
40            pub fn new(value: impl Into<String>) -> Result<Self> {
41                let value = value.into();
42                if value.trim().is_empty() || value.len() > $limit {
43                    return Err(QueryError::new(
44                        QueryErrorCode::InvalidAttemptCorrelation,
45                        "query_engine.attempt.correlation",
46                        "attempt correlation must be non-blank and within its byte limit",
47                    )
48                    .with_fact("field", $field)
49                    .with_fact("max_bytes", $limit)
50                    .with_fact("reason", if value.trim().is_empty() { "blank" } else { "too_long" }));
51                }
52                Ok(Self(value))
53            }
54            /// Returns the original validated correlation text.
55            pub fn as_str(&self) -> &str {
56                &self.0
57            }
58        }
59        impl fmt::Display for $name {
60            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
61                f.write_str(&self.0)
62            }
63        }
64        impl<'de> Deserialize<'de> for $name {
65            fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
66            where D: serde::Deserializer<'de> {
67                // Validation is part of the value contract on every
68                // construction path, including report decoding.
69                let value = String::deserialize(deserializer)?;
70                Self::new(value).map_err(de::Error::custom)
71            }
72        }
73    };
74}
75attempt_correlation!(
76    AttemptQueryId,
77    MAX_ATTEMPT_QUERY_ID_BYTES,
78    "query_id",
79    "Opaque caller query correlation, limited to 256 UTF-8 bytes and required to be non-blank."
80);
81attempt_correlation!(
82    AttemptTraceContext,
83    MAX_ATTEMPT_TRACE_CONTEXT_BYTES,
84    "trace_context",
85    "Opaque caller trace correlation, limited to 1024 UTF-8 bytes and required to be non-blank."
86);
87
88/// Execution coverage assigned after planning, independently of terminal outcome.
89#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
90#[serde(rename_all = "snake_case")]
91pub enum AttemptDisposition {
92    /// The attempt did not obtain a classified physical plan.
93    Unplanned,
94    /// The physical plan executes through DataFusion without native operators.
95    CpuFallback,
96    /// The physical plan contains native and DataFusion execution.
97    PartialNative,
98    /// The physical plan has native execution coverage.
99    Native,
100}
101
102/// Cleanup result fixed before the terminal report is published.
103#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
104#[serde(rename_all = "snake_case")]
105pub enum AttemptCleanupDisposition {
106    /// No admitted GPU cleanup was required.
107    NotRequired,
108    /// Required cleanup finished without recorded failures.
109    Completed,
110    /// Cleanup finished with at least one recorded failure.
111    CompletedWithFailures,
112}
113
114/// Ownership of admitted capacity after the terminal release commit.
115#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
116#[serde(rename_all = "snake_case")]
117pub enum AttemptCapacityDisposition {
118    /// The attempt acquired no GPU admission grant.
119    NotAdmitted,
120    /// Capacity returned to the device ledger after cleanup.
121    Released,
122    /// Capacity remains owned by a resident result after attempt cleanup.
123    FrozenResident,
124    /// Capacity remains reserved because cleanup or device health prevents reuse.
125    RetainedQuarantine,
126}
127
128/// The device-health result decided by an admitted attempt's terminal release
129/// commit.
130///
131/// This is deliberately distinct from [`AttemptCapacityDisposition`]: both a
132/// device-local fault and an unknown-isolation backend fault retain capacity,
133/// but they have materially different health consequences.
134#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
135#[serde(rename_all = "snake_case")]
136pub enum AttemptHealthTransition {
137    /// Cleanup and health evidence permit capacity reuse on the selected device.
138    HealthyReuse,
139    /// A fault isolated to the selected device closed its admission ledger.
140    DeviceQuarantined,
141    /// Fault isolation was unavailable and admission closed across the backend.
142    BackendUnhealthy,
143}
144
145/// Maximum retained UTF-8 bytes in one stable failure-text projection.
146///
147/// The source error remains untouched on its returned path. This is only the
148/// bounded, redacted summary that may enter an immutable attempt report.
149pub(crate) const MAX_ATTEMPT_FAILURE_TEXT_BYTES: usize = 256;
150#[cfg(test)]
151const MAX_ATTEMPT_FAILURE_TEXT_INSPECTION_BYTES: usize = 1024;
152const MAX_ATTEMPT_FAILURE_FACTS: usize = 4;
153
154/// Owning layer that classified a stable attempt failure.
155#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
156#[serde(rename_all = "snake_case")]
157pub enum AttemptFailureOrigin {
158    /// Classified by the native planning, runtime or execution layer.
159    Native,
160    /// Classified by a DataFusion boundary.
161    Datafusion,
162    /// Classified by attempt lifecycle control, such as backend closure.
163    AttemptControl,
164}
165
166/// Closed stability/recovery status for a report-visible failure.
167#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
168#[serde(rename_all = "snake_case")]
169pub enum AttemptFailureStatus {
170    /// The owning failure taxonomy provides no native recovery status.
171    NotApplicable,
172    /// The condition requires changed input, configuration or code.
173    Permanent,
174    /// The condition may clear; retry still requires sealed evidence.
175    Temporary,
176    /// A dependency condition is known or observed to persist.
177    Persistent,
178}
179
180impl AttemptFailureStatus {
181    pub(crate) const fn from_query_error_status(status: QueryErrorStatus) -> Self {
182        match status {
183            QueryErrorStatus::Permanent => Self::Permanent,
184            QueryErrorStatus::Temporary => Self::Temporary,
185            QueryErrorStatus::Persistent => Self::Persistent,
186        }
187    }
188}
189
190/// Coarse owning operation for a stable failure projection.
191#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
192#[serde(rename_all = "snake_case")]
193pub enum AttemptFailureOperation {
194    /// Logical or physical plan construction.
195    Planning,
196    /// Capacity placement, queuing or grant acquisition.
197    Admission,
198    /// Execution after planning or admission.
199    Execution,
200    /// Release of resources after a terminal outcome.
201    Cleanup,
202    /// A DataFusion-owned failure without a more specific native operation.
203    Datafusion,
204    /// Attempt lifecycle control outside a native operation.
205    AttemptControl,
206}
207
208/// Stable allowlisted fact keys for bounded failure summaries.
209#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
210#[serde(rename_all = "snake_case")]
211pub enum AttemptFailureFactKey {
212    /// Closed classification of the condition that ended the attempt.
213    Reason,
214    /// Native allocator family that reported allocation failure.
215    AllocationSource,
216}
217
218/// Stable allowlisted fact values for bounded failure summaries.
219#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
220#[serde(rename_all = "snake_case")]
221pub enum AttemptFailureFactValue {
222    /// Admission or attempt creation encountered a closing backend.
223    BackendClosing,
224    /// The invocation's absolute monotonic deadline elapsed.
225    DeadlineExpired,
226    /// The caller requested cooperative cancellation.
227    CallerCancelled,
228    /// The shared admission queue had reached its waiter bound.
229    AdmissionQueueFull,
230    /// A queued request exceeded the configured maximum queue wait.
231    QueueWaitElapsed,
232    /// The admission wait ended before a memory grant was available.
233    MemoryGrantWaitTimeout,
234    /// The required plan memory exceeds the eligible device capacity.
235    RequiredFloorExceedsDeviceCapacity,
236    /// The requested query cap does not fit any eligible device once resident
237    /// ownership is deducted. Retrying the same request cannot succeed.
238    QueryCapExceedsResidentCapacity,
239    /// Device or backend health prevents new admission.
240    BackendUnhealthy,
241    /// cuGraph reported an RMM out-of-memory failure.
242    CugraphRmmOutOfMemory,
243    /// cuGraph reported an RMM allocation exception.
244    CugraphRmmBadAlloc,
245    /// cuGraph reported a standard allocation exception.
246    CugraphStdBadAlloc,
247}
248
249/// One closed, bounded fact retained beside an attempt failure summary.
250#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
251pub struct AttemptFailureFact {
252    key: AttemptFailureFactKey,
253    value: AttemptFailureFactValue,
254}
255
256impl AttemptFailureFact {
257    /// Returns the allowlisted classification represented by this fact.
258    pub const fn key(&self) -> AttemptFailureFactKey {
259        self.key
260    }
261
262    /// Returns the closed value retained by the failure projection.
263    pub const fn value(&self) -> AttemptFailureFactValue {
264        self.value
265    }
266}
267
268/// Bounded, UTF-8-safe and redacted diagnostic text.
269#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
270pub(crate) struct AttemptFailureText {
271    value: String,
272    truncated: bool,
273    redacted: bool,
274    invalid_utf8: bool,
275}
276
277#[derive(Deserialize)]
278struct AttemptFailureTextWire {
279    #[serde(deserialize_with = "deserialize_bounded_attempt_failure_text_value")]
280    value: String,
281    truncated: bool,
282    redacted: bool,
283    invalid_utf8: bool,
284}
285
286fn deserialize_bounded_attempt_failure_text_value<'de, D>(deserializer: D) -> std::result::Result<String, D::Error>
287where D: serde::Deserializer<'de> {
288    struct BoundedTextVisitor;
289
290    impl Visitor<'_> for BoundedTextVisitor {
291        type Value = String;
292
293        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
294            write!(formatter, "a UTF-8 string no longer than {MAX_ATTEMPT_FAILURE_TEXT_BYTES} bytes")
295        }
296
297        fn visit_str<E>(self, value: &str) -> std::result::Result<Self::Value, E>
298        where E: de::Error {
299            if value.len() > MAX_ATTEMPT_FAILURE_TEXT_BYTES {
300                return Err(E::custom("attempt failure text exceeds its byte bound"));
301            }
302            Ok(value.to_owned())
303        }
304
305        fn visit_string<E>(self, value: String) -> std::result::Result<Self::Value, E>
306        where E: de::Error {
307            if value.len() > MAX_ATTEMPT_FAILURE_TEXT_BYTES {
308                return Err(E::custom("attempt failure text exceeds its byte bound"));
309            }
310            Ok(value)
311        }
312    }
313
314    deserializer.deserialize_string(BoundedTextVisitor)
315}
316
317impl<'de> Deserialize<'de> for AttemptFailureText {
318    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
319    where D: serde::Deserializer<'de> {
320        let wire = AttemptFailureTextWire::deserialize(deserializer)?;
321        if wire.value.len() > MAX_ATTEMPT_FAILURE_TEXT_BYTES {
322            return Err(de::Error::custom("attempt failure text exceeds its byte bound"));
323        }
324        if wire.redacted {
325            if wire.value != "<redacted>" {
326                return Err(de::Error::custom("redacted attempt failure text must not retain source text"));
327            }
328        } else if attempt_failure_text_requires_redaction(&wire.value.to_ascii_lowercase()) {
329            return Err(de::Error::custom("attempt failure text requires redaction"));
330        }
331        Ok(Self {
332            value: wire.value,
333            truncated: wire.truncated,
334            redacted: wire.redacted,
335            invalid_utf8: wire.invalid_utf8,
336        })
337    }
338}
339
340impl AttemptFailureText {
341    /// Retains only a deterministic safe projection of arbitrary error bytes.
342    #[cfg(test)]
343    pub(crate) fn from_bytes(bytes: &[u8]) -> Self {
344        let inspected_len = bytes.len().min(MAX_ATTEMPT_FAILURE_TEXT_INSPECTION_BYTES);
345        let inspected = &bytes[..inspected_len];
346        let invalid_utf8 = std::str::from_utf8(inspected).is_err();
347        let mut value = String::from_utf8_lossy(inspected).split_whitespace().collect::<Vec<_>>().join(" ");
348        let mut redacted = false;
349        let lowercase = value.to_ascii_lowercase();
350        if attempt_failure_text_requires_redaction(&lowercase) {
351            value = "<redacted>".to_string();
352            redacted = true;
353        }
354        let (value, truncated_by_bound) = truncate_utf8(&value, MAX_ATTEMPT_FAILURE_TEXT_BYTES);
355        Self { value, truncated: bytes.len() > inspected_len || truncated_by_bound, redacted, invalid_utf8 }
356    }
357
358    pub fn value(&self) -> &str {
359        &self.value
360    }
361
362    #[cfg(test)]
363    pub(crate) const fn truncated(&self) -> bool {
364        self.truncated
365    }
366
367    #[cfg(test)]
368    pub(crate) const fn redacted(&self) -> bool {
369        self.redacted
370    }
371
372    #[cfg(test)]
373    pub(crate) const fn invalid_utf8(&self) -> bool {
374        self.invalid_utf8
375    }
376}
377
378fn attempt_failure_text_requires_redaction(lowercase: &str) -> bool {
379    lowercase.contains("://")
380        || lowercase.contains("password")
381        || lowercase.contains("passwd")
382        || lowercase.contains("credential")
383        || lowercase.contains("authorization")
384        || lowercase.contains("token")
385        || lowercase.contains("secret")
386        || lowercase.contains("api_key")
387        || lowercase.contains("access_key")
388        || lowercase.contains("bearer ")
389        || looks_like_sql_statement(lowercase)
390}
391
392fn looks_like_sql_statement(value: &str) -> bool {
393    let value = value.trim_start();
394    ["select", "insert", "update", "delete", "merge", "with", "create", "alter", "drop"].into_iter().any(|keyword| {
395        value.strip_prefix(keyword).is_some_and(|suffix| {
396            suffix.is_empty()
397                || suffix
398                    .chars()
399                    .next()
400                    .is_some_and(|character| character.is_whitespace() || matches!(character, '*' | '('))
401        })
402    })
403}
404
405#[cfg(test)]
406fn truncate_utf8(value: &str, max_bytes: usize) -> (String, bool) {
407    if value.len() <= max_bytes {
408        return (value.to_string(), false);
409    }
410    let mut end = max_bytes;
411    while !value.is_char_boundary(end) {
412        end -= 1;
413    }
414    (value[..end].to_string(), true)
415}
416
417/// Bounded retained classification for a primary terminal failure.
418#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
419pub struct AttemptFailureSummary {
420    origin: AttemptFailureOrigin,
421    code: AttemptFailureCode,
422    status: AttemptFailureStatus,
423    operation: AttemptFailureOperation,
424    text: Option<AttemptFailureText>,
425    facts: Vec<AttemptFailureFact>,
426}
427
428#[derive(Deserialize)]
429struct AttemptFailureSummaryWire {
430    origin: AttemptFailureOrigin,
431    code: AttemptFailureCode,
432    status: AttemptFailureStatus,
433    operation: AttemptFailureOperation,
434    #[serde(default)]
435    text: Option<AttemptFailureText>,
436    #[serde(default, deserialize_with = "deserialize_bounded_attempt_failure_facts")]
437    facts: Vec<AttemptFailureFact>,
438}
439
440fn deserialize_bounded_attempt_failure_facts<'de, D>(
441    deserializer: D,
442) -> std::result::Result<Vec<AttemptFailureFact>, D::Error>
443where D: serde::Deserializer<'de> {
444    struct BoundedFactsVisitor;
445
446    impl<'de> Visitor<'de> for BoundedFactsVisitor {
447        type Value = Vec<AttemptFailureFact>;
448
449        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
450            write!(formatter, "at most {MAX_ATTEMPT_FAILURE_FACTS} attempt failure facts")
451        }
452
453        fn visit_seq<A>(self, mut sequence: A) -> std::result::Result<Self::Value, A::Error>
454        where A: SeqAccess<'de> {
455            let mut facts = Vec::with_capacity(MAX_ATTEMPT_FAILURE_FACTS);
456            while let Some(fact) = sequence.next_element()? {
457                if facts.len() == MAX_ATTEMPT_FAILURE_FACTS {
458                    return Err(de::Error::custom("attempt failure facts exceed their fixed bound"));
459                }
460                facts.push(fact);
461            }
462            Ok(facts)
463        }
464    }
465
466    deserializer.deserialize_seq(BoundedFactsVisitor)
467}
468
469impl<'de> Deserialize<'de> for AttemptFailureSummary {
470    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
471    where D: serde::Deserializer<'de> {
472        let wire = AttemptFailureSummaryWire::deserialize(deserializer)?;
473        Self::from_parts(wire.origin, wire.code, wire.status, wire.operation, wire.text, wire.facts)
474            .map_err(de::Error::custom)
475    }
476}
477
478impl AttemptFailureSummary {
479    fn from_parts(
480        origin: AttemptFailureOrigin,
481        code: AttemptFailureCode,
482        status: AttemptFailureStatus,
483        operation: AttemptFailureOperation,
484        text: Option<AttemptFailureText>,
485        facts: Vec<AttemptFailureFact>,
486    ) -> std::result::Result<Self, &'static str> {
487        if facts.len() > MAX_ATTEMPT_FAILURE_FACTS {
488            return Err("attempt failure facts exceed their fixed bound");
489        }
490        if facts.windows(2).any(|pair| pair[0].key >= pair[1].key) {
491            return Err("attempt failure facts must use unique deterministic key order");
492        }
493        Ok(Self { origin, code, status, operation, text, facts })
494    }
495
496    /// Captures a native error by reference without touching its source chain.
497    ///
498    /// The native message is deliberately not copied into the stable report:
499    /// it can contain arbitrary application metadata even when it does not
500    /// match a known URI, credential, or SQL pattern. The returned
501    /// [`query_planning::error::QueryError`] keeps that complete source text on its
502    /// ordinary error path.
503    pub fn from_native_error(error: &QueryError, operation: AttemptFailureOperation) -> Self {
504        let mut facts = Vec::with_capacity(2);
505        let reason = match error.fact_value("reason") {
506            Some("backend_closing") => Some(AttemptFailureFactValue::BackendClosing),
507            Some("deadline_expired") => Some(AttemptFailureFactValue::DeadlineExpired),
508            Some("caller_cancelled") => Some(AttemptFailureFactValue::CallerCancelled),
509            Some("admission_queue_full") => Some(AttemptFailureFactValue::AdmissionQueueFull),
510            Some("queue_wait_elapsed") => Some(AttemptFailureFactValue::QueueWaitElapsed),
511            Some("memory_grant_wait_timeout") => Some(AttemptFailureFactValue::MemoryGrantWaitTimeout),
512            Some("required_floor_exceeds_device_capacity") | Some("impossible_requirement") => {
513                Some(AttemptFailureFactValue::RequiredFloorExceedsDeviceCapacity)
514            },
515            Some("query_cap_exceeds_resident_capacity") => {
516                Some(AttemptFailureFactValue::QueryCapExceedsResidentCapacity)
517            },
518            Some("backend_unhealthy") => Some(AttemptFailureFactValue::BackendUnhealthy),
519            _ => None,
520        };
521        if let Some(value) = reason {
522            facts.push(AttemptFailureFact { key: AttemptFailureFactKey::Reason, value });
523        }
524        let allocation_source = match error.context().allocation_source {
525            Some(AllocationSource::RmmOutOfMemory) => Some(AttemptFailureFactValue::CugraphRmmOutOfMemory),
526            Some(AllocationSource::RmmBadAlloc) => Some(AttemptFailureFactValue::CugraphRmmBadAlloc),
527            Some(AllocationSource::StdBadAlloc) => Some(AttemptFailureFactValue::CugraphStdBadAlloc),
528            Some(AllocationSource::Unknown) | None => None,
529        };
530        if let Some(value) = allocation_source {
531            facts.push(AttemptFailureFact { key: AttemptFailureFactKey::AllocationSource, value });
532        }
533        Self {
534            origin: AttemptFailureOrigin::Native,
535            code: AttemptFailureCode::from_query_error_code(error.code()),
536            status: AttemptFailureStatus::from_query_error_status(error.status()),
537            operation,
538            text: None,
539            facts,
540        }
541    }
542
543    /// Closed root-adapter fallback. It intentionally retains no adapter-side
544    /// display text, source-chain text, SQL, URI, or credential material.
545    pub const fn datafusion_fallback() -> Self {
546        Self {
547            origin: AttemptFailureOrigin::Datafusion,
548            code: AttemptFailureCode::DatafusionError,
549            status: AttemptFailureStatus::Permanent,
550            operation: AttemptFailureOperation::Datafusion,
551            text: None,
552            facts: Vec::new(),
553        }
554    }
555
556    /// Returns a text-free admission failure for a closing backend.
557    pub const fn backend_closing() -> Self {
558        Self {
559            origin: AttemptFailureOrigin::AttemptControl,
560            code: AttemptFailureCode::BackendClosing,
561            status: AttemptFailureStatus::Permanent,
562            operation: AttemptFailureOperation::Admission,
563            text: None,
564            facts: Vec::new(),
565        }
566    }
567
568    const fn control(code: AttemptFailureCode) -> Self {
569        Self {
570            origin: AttemptFailureOrigin::AttemptControl,
571            code,
572            status: AttemptFailureStatus::NotApplicable,
573            operation: AttemptFailureOperation::AttemptControl,
574            text: None,
575            facts: Vec::new(),
576        }
577    }
578
579    /// Returns the layer that classified this failure.
580    pub const fn origin(&self) -> AttemptFailureOrigin {
581        self.origin
582    }
583
584    /// Returns the closed machine identity retained from the primary failure.
585    pub const fn code(&self) -> AttemptFailureCode {
586        self.code
587    }
588
589    /// Returns the source condition classification, without granting retry permission.
590    pub const fn status(&self) -> AttemptFailureStatus {
591        self.status
592    }
593
594    /// Returns the lifecycle operation in which the failure was recorded.
595    pub const fn operation(&self) -> AttemptFailureOperation {
596        self.operation
597    }
598
599    pub(crate) fn text(&self) -> Option<&AttemptFailureText> {
600        self.text.as_ref()
601    }
602
603    /// Returns allowlisted facts in unique deterministic key order.
604    /// Arbitrary error facts and source text are excluded.
605    pub fn facts(&self) -> &[AttemptFailureFact] {
606        &self.facts
607    }
608}
609
610/// Closed lifecycle phase at which a secondary cleanup failure occurred.
611#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
612#[serde(rename_all = "snake_case")]
613pub enum CleanupFailurePhase {
614    /// Scheduler cycle cleanup.
615    SchedulingCycle,
616    /// Cleanup after runnable-task selection.
617    RunnableTaskSelection,
618    /// Cleanup after task input and resource preparation.
619    TaskPreparation,
620    /// Cleanup after native task execution returned an error.
621    TaskExecution,
622    /// Cleanup after a task panic was caught.
623    TaskPanic,
624    /// Cleanup during task result completion.
625    TaskCompletion,
626    /// Release of resources retained by a completed task.
627    CompletedResourceRelease,
628    /// Final scheduler validation before release.
629    FinalValidation,
630    /// Synchronization of the admitted query stream failed.
631    QueryServiceStreamSynchronization,
632    /// An admitted query cleanup callback was abandoned.
633    QueryServiceCallbackAbandoned,
634    /// Cleanup owned by the admission service.
635    QueryServiceCleanup,
636}
637
638/// First bounded cleanup failure retained by an attempt lifecycle.
639#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
640pub struct CleanupFailureSummary {
641    phase: CleanupFailurePhase,
642    failure: AttemptFailureSummary,
643    /// `false` means another cleanup failure followed this first retained
644    /// summary. The exact count remains in the enclosing attempt report; the
645    /// field is required so decoding never fabricates exactness.
646    complete: bool,
647}
648
649impl CleanupFailureSummary {
650    /// Projects one native cleanup failure without copying its message or source.
651    /// The summary is initially complete; subsequent cleanup failures can make
652    /// the retained first summary incomplete.
653    pub fn from_native_error(phase: CleanupFailurePhase, error: &QueryError) -> Self {
654        Self {
655            phase,
656            failure: AttemptFailureSummary::from_native_error(error, AttemptFailureOperation::Cleanup),
657            complete: true,
658        }
659    }
660
661    pub(crate) const fn stream_synchronization_failed() -> Self {
662        Self {
663            phase: CleanupFailurePhase::QueryServiceStreamSynchronization,
664            failure: AttemptFailureSummary::control(AttemptFailureCode::StreamSynchronizationFailed),
665            complete: true,
666        }
667    }
668
669    pub(crate) const fn callback_abandoned() -> Self {
670        Self {
671            phase: CleanupFailurePhase::QueryServiceCallbackAbandoned,
672            failure: AttemptFailureSummary::control(AttemptFailureCode::CleanupCallbackAbandoned),
673            complete: true,
674        }
675    }
676
677    /// Returns the cleanup boundary that recorded the failure.
678    pub const fn phase(&self) -> CleanupFailurePhase {
679        self.phase
680    }
681
682    /// Returns the bounded failure identity and allowlisted facts.
683    pub const fn failure(&self) -> &AttemptFailureSummary {
684        &self.failure
685    }
686
687    /// Returns whether this first summary accounts for all recorded cleanup failures.
688    /// The enclosing report carries the count and its separate completeness flag.
689    pub const fn complete(&self) -> bool {
690        self.complete
691    }
692
693    pub(crate) fn mark_incomplete(&mut self) {
694        self.complete = false;
695    }
696}
697
698/// Exact final counters observed for one attempt allocation domain.
699///
700/// A report carries this only when its final sample completed successfully.
701/// `None` in [`QueryAttemptReport::final_domain_usage`] means that no
702/// trustworthy final sample is available; it never represents a synthetic
703/// all-zero domain. This includes releases without an allocation domain,
704/// failed final samples, and unknown-isolation cleanup paths.
705#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
706pub struct AttemptFinalDomainUsage {
707    /// Identity of the admitted allocation domain.
708    pub domain_id: u64,
709    pub(crate) active: bool,
710    pub(crate) active_members: u64,
711    pub(crate) ceiling_bytes: u64,
712    pub(crate) allocation_limit_bytes: u64,
713    /// Logical live allocation bytes at the final sample.
714    pub current_bytes: u64,
715    /// Highest logical live allocation bytes observed over the domain lifetime.
716    pub peak_bytes: u64,
717    pub(crate) total_bytes: u64,
718    pub(crate) current_aligned_bytes: u64,
719    /// Highest allocator-aligned live allocation bytes over the domain lifetime.
720    pub peak_aligned_bytes: u64,
721    pub(crate) total_allocations: u64,
722    pub(crate) current_allocations: u64,
723    pub(crate) peak_allocations: u64,
724    pub(crate) failed_allocation_count: u64,
725    pub(crate) largest_failed_allocation_bytes: u64,
726}
727
728impl From<cudf::QueryMemoryDomainUsage> for AttemptFinalDomainUsage {
729    fn from(usage: cudf::QueryMemoryDomainUsage) -> Self {
730        Self {
731            domain_id: usage.domain_id,
732            active: usage.active,
733            active_members: usage.active_members,
734            ceiling_bytes: usage.ceiling_bytes,
735            allocation_limit_bytes: usage.allocation_limit_bytes,
736            current_bytes: usage.current_bytes,
737            peak_bytes: usage.peak_bytes,
738            total_bytes: usage.total_bytes,
739            current_aligned_bytes: usage.current_aligned_bytes,
740            peak_aligned_bytes: usage.peak_aligned_bytes,
741            total_allocations: usage.total_allocations,
742            current_allocations: usage.current_allocations,
743            peak_allocations: usage.peak_allocations,
744            failed_allocation_count: usage.failed_allocation_count,
745            largest_failed_allocation_bytes: usage.largest_failed_allocation_bytes,
746        }
747    }
748}
749
750/// Isolation decision committed with one device-backed attempt's terminal
751/// native-health observation.
752///
753/// `DeviceLocal` means the release path proved that the relevant cleanup and
754/// health facts belong to the attempt's immutable device. `Unknown` means the
755/// cleanup path could not establish that boundary, so the engine closes the
756/// entire backend without guessing a device owner.
757#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
758#[serde(rename_all = "snake_case")]
759pub(crate) enum AttemptFaultIsolation {
760    DeviceLocal,
761    Unknown,
762}
763
764/// Cleanup state retained with terminal native-health evidence.
765///
766/// This is an engine-owned projection of the release worker's result. It does
767/// not infer a CUDA status from a later runtime query.
768#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
769#[serde(rename_all = "snake_case")]
770pub(crate) enum AttemptTeardownHealth {
771    Completed,
772    StreamSynchronizationFailed,
773    CallbackAbandoned,
774}
775
776/// Health state read from cuDF's process-lifetime device-health authority at
777/// the terminal release commit.
778#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
779#[serde(rename_all = "snake_case")]
780pub(crate) enum AttemptDeviceHealthStatus {
781    Healthy,
782    Unhealthy,
783}
784
785/// Closed projection of the device-fault kind supplied by cuDF and cuGraph.
786///
787/// The source enum is non-exhaustive. A newly introduced sibling kind remains
788/// observable as `Unknown` without making a report falsely claim an existing
789/// classification.
790#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
791#[serde(rename_all = "snake_case")]
792pub(crate) enum AttemptNativeFaultKind {
793    CurrentDeviceQueryFailed,
794    OwnerDeviceEntryFailed,
795    AmbientDeviceRestoreFailed,
796    FatalDeviceOperation,
797    TeardownQuarantined,
798    Unknown,
799}
800
801/// Exact status family and value retained by the native runtime.
802///
803/// cuDF and CUDA status values intentionally remain distinct. The engine
804/// receives this value from the sibling-owned device-health record; it never
805/// parses display text or samples CUDA after the original native failure.
806#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
807#[serde(rename_all = "snake_case")]
808pub(crate) enum AttemptNativeStatus {
809    Cudf(u32),
810    Cuda(i32),
811}
812
813/// First structured native fault retained by cuDF's device-health authority.
814#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
815pub(crate) struct AttemptNativeFault {
816    pub(crate) kind: AttemptNativeFaultKind,
817    pub(crate) status: Option<AttemptNativeStatus>,
818}
819
820/// Terminal health counters from the authoritative cuDF device record.
821#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
822pub(crate) struct AttemptDeviceHealth {
823    pub(crate) status: AttemptDeviceHealthStatus,
824    pub(crate) fault_count: u64,
825    pub(crate) rejected_operation_count: u64,
826    pub(crate) quarantined_teardown_count: u64,
827}
828
829/// Bounded cuDF/cuGraph native-fault evidence assembled by the release commit.
830///
831/// This snapshot is deliberately terminal-only. Same-sequence allocator and
832/// ledger evidence is carried separately; this type does not imply that a
833/// native fault was caused by a particular allocation or reconstruct a failure
834/// instant.
835#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
836pub struct AttemptNativeFaultEvidence {
837    pub(crate) isolation: AttemptFaultIsolation,
838    pub(crate) teardown_health: AttemptTeardownHealth,
839    /// No device health sample is retained when cleanup lost isolation before
840    /// the engine could establish a truthful device-local observation.
841    pub(crate) device_health: Option<AttemptDeviceHealth>,
842    pub(crate) first_fault: Option<AttemptNativeFault>,
843}
844
845/// How one bounded resource-evidence field was obtained.
846///
847/// The release commit does not infer unavailable values, and it never upgrades
848/// a device-level observation into an attempt fact merely because it was read
849/// near an attempt's terminal boundary.
850#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
851#[serde(rename_all = "snake_case")]
852pub enum AttemptEvidenceProvenance {
853    /// The value is immutable for, or was joined exactly to, this attempt.
854    AttemptExact,
855    /// The value is a bounded snapshot of the selected CUDA device only.
856    DeviceSnapshot,
857    /// The value was sampled after the attempt reached its terminal boundary,
858    /// whatever its outcome; it is not reconstructed as the instant of an
859    /// earlier failure. Outcome lives in the report's terminal fields.
860    SampledAtTerminalBoundary,
861    /// No truthful value was available at the terminal release commit.
862    Unavailable,
863}
864
865/// One bounded resource-evidence value together with its provenance.
866///
867/// In engine-produced reports, `value` is absent if and only if the provenance
868/// is [`AttemptEvidenceProvenance::Unavailable`]. Consumers must retain the
869/// provenance. An absent value is neither an ordinal zero nor an exact sample.
870#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
871pub struct AttemptEvidenceField<T> {
872    /// Scope and sampling classification that consumers must retain with the value.
873    pub provenance: AttemptEvidenceProvenance,
874    /// Observed value, or `None` when provenance is `Unavailable`.
875    pub value: Option<T>,
876}
877
878impl<T> AttemptEvidenceField<T> {
879    pub(crate) const fn attempt_exact(value: T) -> Self {
880        Self { provenance: AttemptEvidenceProvenance::AttemptExact, value: Some(value) }
881    }
882
883    pub(crate) const fn device_snapshot(value: T) -> Self {
884        Self { provenance: AttemptEvidenceProvenance::DeviceSnapshot, value: Some(value) }
885    }
886
887    pub(crate) const fn sampled_at_terminal_boundary(value: T) -> Self {
888        Self { provenance: AttemptEvidenceProvenance::SampledAtTerminalBoundary, value: Some(value) }
889    }
890
891    pub(crate) const fn unavailable() -> Self {
892        Self { provenance: AttemptEvidenceProvenance::Unavailable, value: None }
893    }
894}
895
896impl<T> Default for AttemptEvidenceField<T> {
897    fn default() -> Self {
898        Self::unavailable()
899    }
900}
901
902/// Exact, separately timed usage sample from an admitted allocation domain.
903#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
904pub struct AttemptTimedDomainUsage {
905    /// Query-relative monotonic time recorded from the attempt clock while
906    /// sampling the usage. This is intentionally separate from the outcome
907    /// terminal timestamp and from the post-cleanup sample time.
908    pub(crate) sampled_ns: u64,
909    pub(crate) usage: AttemptFinalDomainUsage,
910    /// Kept separate from logical usage so physical pool pages never widen the grant.
911    pub pool_usage: Option<AttemptAllocatorPoolUsage>,
912}
913
914/// Closed source category copied from one cuDF allocator-ring record.
915#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
916#[serde(rename_all = "snake_case")]
917pub enum AttemptAllocatorEventSource {
918    /// The underlying allocator could not satisfy the request.
919    Physical,
920    /// The request exceeded the allocation domain reservation ceiling.
921    ReservationLimit,
922    /// The request exceeded a logical allocation limit.
923    AllocationLimit,
924    /// The native source category is unavailable or unrecognized.
925    Unknown,
926}
927
928/// Closed allocator callback action copied from one allocator-ring record.
929#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
930#[serde(rename_all = "snake_case")]
931pub(crate) enum AttemptAllocatorEventAction {
932    Decline,
933    Retry,
934    TrimAndRetry,
935    CallbackPanicked,
936    Unknown,
937}
938
939/// Bounded async-pool counters copied from one allocator-ring record.
940#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
941pub struct AttemptAllocatorPoolUsage {
942    pub(crate) reserved_current_bytes: u64,
943    /// Highest bytes reserved by the CUDA async allocation pool.
944    pub reserved_high_bytes: u64,
945    pub(crate) used_current_bytes: u64,
946    /// Highest bytes used within the CUDA async allocation pool.
947    pub used_high_bytes: u64,
948}
949
950/// One allocator-ring record retained in an attempt resource-evidence block.
951///
952/// The enclosing evidence field is `attempt_exact` only after the engine's
953/// bounded domain-generation join proves ownership. Optional callback fields
954/// retain their own provenance so an unavailable callback timestamp, action,
955/// or pool sample is never replaced with a zero value.
956#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
957pub struct AttemptAllocatorEvent {
958    /// Monotonic identity in the device allocator event ring.
959    pub sequence: u64,
960    pub(crate) monotonic_time_ns: AttemptEvidenceField<u64>,
961    /// Size in bytes of the allocation request associated with this event.
962    pub requested_bytes: u64,
963    /// Closed allocator failure source retained from the native record.
964    pub source: AttemptAllocatorEventSource,
965    pub(crate) action: AttemptEvidenceField<AttemptAllocatorEventAction>,
966    pub(crate) callback_attempt: u32,
967    pub(crate) trim_retry_count: u32,
968    pub(crate) pool_usage: AttemptEvidenceField<AttemptAllocatorPoolUsage>,
969}
970
971/// A device-level allocator sample which was deliberately not attributed to
972/// the attempt. It is diagnostic context only and does not establish cause.
973#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
974pub struct AttemptUnattributedAllocatorEvent {
975    pub(crate) device_ordinal: usize,
976    /// Query-relative monotonic time at which the release commit sampled this
977    /// retained ring record.
978    pub(crate) sampled_ns: u64,
979    pub(crate) event: AttemptAllocatorEvent,
980}
981
982/// One bounded, truthful resource-evidence block committed with an admitted
983/// attempt's post-cleanup terminal disposition.
984///
985/// Every field carries an explicit provenance. The ledger values are obtained
986/// from one lock-held sequence; the profile and grant/domain caps are
987/// immutable attempt facts; and both usage samples retain their own times.
988#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
989pub struct AttemptResourceEvidence {
990    /// Device ledger transition sequence from the terminal release snapshot.
991    pub ledger_sequence: AttemptEvidenceField<u64>,
992    pub(crate) ledger_sample_ns: AttemptEvidenceField<u64>,
993    pub(crate) configured_limit_bytes: AttemptEvidenceField<u64>,
994    pub(crate) headroom_bytes: AttemptEvidenceField<u64>,
995    pub(crate) backend_reserve_bytes: AttemptEvidenceField<u64>,
996    pub(crate) managed_capacity_bytes: AttemptEvidenceField<u64>,
997    pub(crate) live_grants_bytes: AttemptEvidenceField<u64>,
998    pub(crate) resident_owner_bytes: AttemptEvidenceField<u64>,
999    /// Cached resident bytes eligible for eviction in the device snapshot.
1000    pub cache_evictable_resident_bytes: AttemptEvidenceField<u64>,
1001    /// Cached resident bytes pinned by active use in the device snapshot.
1002    pub cache_pinned_in_use_bytes: AttemptEvidenceField<u64>,
1003    pub(crate) active_attempt_count: AttemptEvidenceField<u64>,
1004    /// Immutable identity of this attempt's admission grant.
1005    pub grant_id: AttemptEvidenceField<u64>,
1006    pub(crate) grant_cap_bytes: AttemptEvidenceField<u64>,
1007    /// Immutable identity of this attempt's allocation domain.
1008    pub domain_id: AttemptEvidenceField<u64>,
1009    pub(crate) domain_cap_bytes: AttemptEvidenceField<u64>,
1010    /// Domain sample taken at the outcome terminal boundary, before cleanup.
1011    pub outcome_terminal_usage: AttemptEvidenceField<AttemptTimedDomainUsage>,
1012    pub(crate) post_cleanup_usage: AttemptEvidenceField<AttemptTimedDomainUsage>,
1013    /// Allocator event attributed to this attempt by a bounded domain-generation join.
1014    pub allocator_event: AttemptEvidenceField<AttemptAllocatorEvent>,
1015    /// Device allocator context whose ownership could not be attributed to the attempt.
1016    pub unattributed_device_allocator_event: AttemptEvidenceField<AttemptUnattributedAllocatorEvent>,
1017}
1018
1019impl AttemptResourceEvidence {
1020    /// Starts a block with every non-derivable value explicitly unavailable.
1021    pub(crate) const fn unavailable() -> Self {
1022        Self {
1023            ledger_sequence: AttemptEvidenceField::unavailable(),
1024            ledger_sample_ns: AttemptEvidenceField::unavailable(),
1025            configured_limit_bytes: AttemptEvidenceField::unavailable(),
1026            headroom_bytes: AttemptEvidenceField::unavailable(),
1027            backend_reserve_bytes: AttemptEvidenceField::unavailable(),
1028            managed_capacity_bytes: AttemptEvidenceField::unavailable(),
1029            live_grants_bytes: AttemptEvidenceField::unavailable(),
1030            resident_owner_bytes: AttemptEvidenceField::unavailable(),
1031            cache_evictable_resident_bytes: AttemptEvidenceField::unavailable(),
1032            cache_pinned_in_use_bytes: AttemptEvidenceField::unavailable(),
1033            active_attempt_count: AttemptEvidenceField::unavailable(),
1034            grant_id: AttemptEvidenceField::unavailable(),
1035            grant_cap_bytes: AttemptEvidenceField::unavailable(),
1036            domain_id: AttemptEvidenceField::unavailable(),
1037            domain_cap_bytes: AttemptEvidenceField::unavailable(),
1038            outcome_terminal_usage: AttemptEvidenceField::unavailable(),
1039            post_cleanup_usage: AttemptEvidenceField::unavailable(),
1040            allocator_event: AttemptEvidenceField::unavailable(),
1041            unattributed_device_allocator_event: AttemptEvidenceField::unavailable(),
1042        }
1043    }
1044}
1045
1046/// The immutable cleanup and capacity facts committed after an admitted
1047/// attempt's device cleanup and before its observation report is sealed.
1048///
1049/// This stays crate-internal because the query service is the sole authority
1050/// that observes the cleanup result and returns (or retains) capacity. The
1051/// attempt owner projects this value into [`QueryAttemptReport`].
1052#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1053pub(crate) struct AttemptTerminalCommit {
1054    capacity_disposition: AttemptCapacityDisposition,
1055    final_domain_usage: Option<AttemptFinalDomainUsage>,
1056    health_transition: AttemptHealthTransition,
1057    native_fault_evidence: Option<AttemptNativeFaultEvidence>,
1058    resource_evidence: Option<AttemptResourceEvidence>,
1059    alternate_device_eligibility: AttemptAlternateDeviceEligibility,
1060}
1061
1062/// Result of the post-quarantine check over the immutable compatible-device
1063/// set captured at admission. This is internal finalizer input that only gates
1064/// retry advice in the sealed report; it adds no second scheduler decision.
1065#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1066pub(crate) enum AttemptAlternateDeviceEligibility {
1067    NotApplicable,
1068    Eligible,
1069    NoEligibleAlternate,
1070    Unavailable,
1071}
1072
1073impl AttemptTerminalCommit {
1074    pub(crate) const fn new(
1075        capacity_disposition: AttemptCapacityDisposition,
1076        final_domain_usage: Option<AttemptFinalDomainUsage>,
1077        health_transition: AttemptHealthTransition,
1078    ) -> Self {
1079        Self {
1080            capacity_disposition,
1081            final_domain_usage,
1082            health_transition,
1083            native_fault_evidence: None,
1084            resource_evidence: None,
1085            alternate_device_eligibility: AttemptAlternateDeviceEligibility::NotApplicable,
1086        }
1087    }
1088
1089    /// Adds the sole terminal native-health observation from the owning
1090    /// release commit. The finalizer remains the only report producer.
1091    pub(crate) const fn with_native_fault_evidence(mut self, evidence: AttemptNativeFaultEvidence) -> Self {
1092        self.native_fault_evidence = Some(evidence);
1093        self
1094    }
1095
1096    /// Adds the sole bounded resource-evidence block from the owning release
1097    /// commit. The finalizer remains the only report producer.
1098    pub(crate) const fn with_resource_evidence(mut self, evidence: AttemptResourceEvidence) -> Self {
1099        self.resource_evidence = Some(evidence);
1100        self
1101    }
1102
1103    /// Adds the one post-quarantine eligibility result taken from the frozen
1104    /// compatible-device set. A new admission still owns its own placement;
1105    /// this value never reserves or selects an alternate device.
1106    pub(crate) const fn with_alternate_device_eligibility(
1107        mut self,
1108        eligibility: AttemptAlternateDeviceEligibility,
1109    ) -> Self {
1110        self.alternate_device_eligibility = eligibility;
1111        self
1112    }
1113
1114    pub(crate) const fn capacity_disposition(self) -> AttemptCapacityDisposition {
1115        self.capacity_disposition
1116    }
1117
1118    pub(crate) const fn final_domain_usage(self) -> Option<AttemptFinalDomainUsage> {
1119        self.final_domain_usage
1120    }
1121
1122    pub(crate) const fn health_transition(self) -> AttemptHealthTransition {
1123        self.health_transition
1124    }
1125
1126    pub(crate) const fn native_fault_evidence(self) -> Option<AttemptNativeFaultEvidence> {
1127        self.native_fault_evidence
1128    }
1129
1130    pub(crate) const fn resource_evidence(self) -> Option<AttemptResourceEvidence> {
1131        self.resource_evidence
1132    }
1133
1134    pub(crate) const fn alternate_device_eligibility(self) -> AttemptAlternateDeviceEligibility {
1135        self.alternate_device_eligibility
1136    }
1137
1138    pub(crate) const fn has_consistent_capacity_health_transition(self) -> bool {
1139        matches!(
1140            (self.capacity_disposition, self.health_transition),
1141            (AttemptCapacityDisposition::Released, AttemptHealthTransition::HealthyReuse)
1142                | (AttemptCapacityDisposition::FrozenResident, AttemptHealthTransition::HealthyReuse)
1143                | (AttemptCapacityDisposition::RetainedQuarantine, AttemptHealthTransition::DeviceQuarantined)
1144                | (AttemptCapacityDisposition::RetainedQuarantine, AttemptHealthTransition::BackendUnhealthy)
1145        )
1146    }
1147
1148    pub(crate) const fn is_admitted_cleanup_commit(self) -> bool {
1149        !matches!(self.capacity_disposition, AttemptCapacityDisposition::NotAdmitted)
1150            && self.has_consistent_capacity_health_transition()
1151    }
1152}
1153
1154/// Closed recovery action advised by one sealed attempt report.
1155///
1156/// This is advice only. The engine never retries an attempt from this value;
1157/// callers must satisfy the paired [`AttemptRetryPrerequisite`] before they
1158/// choose to submit a new attempt.
1159#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1160#[serde(rename_all = "snake_case")]
1161pub enum AttemptRetryRecovery {
1162    /// The attempt succeeded and requires no further recovery action.
1163    NoRetryNeeded,
1164    /// The attempt was cancelled, dropped, or panicked; do not retry it from
1165    /// this report.
1166    DoNotRetry,
1167    /// Retry may be considered only after bounded capacity evidence supports
1168    /// waiting for the service to drain.
1169    Backoff,
1170    /// The attempt cannot succeed until its query, capability selection, or
1171    /// configured resource limits change.
1172    ChangeConfiguration,
1173    /// Retry may be considered only on another eligible device after the
1174    /// failed device has been retained in quarantine.
1175    RetryAnotherEligibleDevice,
1176    /// The backend must recover before a new attempt can be considered.
1177    RecoverBackend,
1178    /// The bounded terminal report cannot truthfully recommend recovery.
1179    Indeterminate,
1180}
1181
1182impl AttemptRetryRecovery {
1183    /// Stable closed label for a public projection of sealed retry advice.
1184    pub const fn as_str(self) -> &'static str {
1185        match self {
1186            Self::NoRetryNeeded => "no_retry_needed",
1187            Self::DoNotRetry => "do_not_retry",
1188            Self::Backoff => "backoff",
1189            Self::ChangeConfiguration => "change_configuration",
1190            Self::RetryAnotherEligibleDevice => "retry_another_eligible_device",
1191            Self::RecoverBackend => "recover_backend",
1192            Self::Indeterminate => "indeterminate",
1193        }
1194    }
1195}
1196
1197/// Closed condition that must hold before acting on an
1198/// [`AttemptRetryRecovery`] recommendation.
1199#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1200#[serde(rename_all = "snake_case")]
1201pub enum AttemptRetryPrerequisite {
1202    /// No prerequisite accompanies success or a do-not-retry recommendation.
1203    None,
1204    /// Capacity evidence permits a new admission after waiting.
1205    CapacityAvailable,
1206    /// The query, capability selection or configured limit has changed.
1207    ConfigurationChanged,
1208    /// The failed device remains quarantined and another device is eligible.
1209    QuarantinedDeviceAndEligibleAlternate,
1210    /// Backend recovery completed before a new attempt.
1211    BackendRecovered,
1212    /// The sealed evidence cannot establish a recovery prerequisite.
1213    EvidenceUnavailable,
1214}
1215
1216impl AttemptRetryPrerequisite {
1217    /// Stable closed label for a public projection of sealed retry advice.
1218    pub const fn as_str(self) -> &'static str {
1219        match self {
1220            Self::None => "none",
1221            Self::CapacityAvailable => "capacity_available",
1222            Self::ConfigurationChanged => "configuration_changed",
1223            Self::QuarantinedDeviceAndEligibleAlternate => "quarantined_device_and_eligible_alternate",
1224            Self::BackendRecovered => "backend_recovered",
1225            Self::EvidenceUnavailable => "evidence_unavailable",
1226        }
1227    }
1228}
1229
1230/// One bounded retry/recovery recommendation sealed with an attempt report.
1231#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1232pub struct AttemptRetryAdvice {
1233    /// Recovery action supported by the terminal evidence.
1234    pub recovery: AttemptRetryRecovery,
1235    /// Condition required before submitting a new attempt for that action.
1236    pub prerequisite: AttemptRetryPrerequisite,
1237}
1238
1239#[derive(Deserialize)]
1240struct AttemptRetryAdviceWire {
1241    recovery: AttemptRetryRecovery,
1242    prerequisite: AttemptRetryPrerequisite,
1243}
1244
1245impl<'de> Deserialize<'de> for AttemptRetryAdvice {
1246    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
1247    where D: serde::Deserializer<'de> {
1248        let wire = AttemptRetryAdviceWire::deserialize(deserializer)?;
1249        Self::from_parts(wire.recovery, wire.prerequisite).map_err(de::Error::custom)
1250    }
1251}
1252
1253impl AttemptRetryAdvice {
1254    fn from_parts(
1255        recovery: AttemptRetryRecovery,
1256        prerequisite: AttemptRetryPrerequisite,
1257    ) -> std::result::Result<Self, &'static str> {
1258        let valid = matches!(
1259            (recovery, prerequisite),
1260            (AttemptRetryRecovery::NoRetryNeeded, AttemptRetryPrerequisite::None)
1261                | (AttemptRetryRecovery::DoNotRetry, AttemptRetryPrerequisite::None)
1262                | (AttemptRetryRecovery::Backoff, AttemptRetryPrerequisite::CapacityAvailable)
1263                | (AttemptRetryRecovery::ChangeConfiguration, AttemptRetryPrerequisite::ConfigurationChanged)
1264                | (
1265                    AttemptRetryRecovery::RetryAnotherEligibleDevice,
1266                    AttemptRetryPrerequisite::QuarantinedDeviceAndEligibleAlternate
1267                )
1268                | (AttemptRetryRecovery::RecoverBackend, AttemptRetryPrerequisite::BackendRecovered)
1269                | (AttemptRetryRecovery::Indeterminate, AttemptRetryPrerequisite::EvidenceUnavailable)
1270        );
1271        valid.then_some(Self { recovery, prerequisite }).ok_or("retry recovery and prerequisite are inconsistent")
1272    }
1273
1274    /// Returns the success recommendation with no recovery prerequisite.
1275    pub const fn no_retry_needed() -> Self {
1276        Self { recovery: AttemptRetryRecovery::NoRetryNeeded, prerequisite: AttemptRetryPrerequisite::None }
1277    }
1278
1279    /// Returns the terminal control recommendation with no recovery prerequisite.
1280    pub const fn do_not_retry() -> Self {
1281        Self { recovery: AttemptRetryRecovery::DoNotRetry, prerequisite: AttemptRetryPrerequisite::None }
1282    }
1283
1284    /// Returns an indeterminate recommendation because recovery evidence is unavailable.
1285    pub const fn indeterminate() -> Self {
1286        Self {
1287            recovery: AttemptRetryRecovery::Indeterminate,
1288            prerequisite: AttemptRetryPrerequisite::EvidenceUnavailable,
1289        }
1290    }
1291
1292    fn backoff() -> Self {
1293        Self { recovery: AttemptRetryRecovery::Backoff, prerequisite: AttemptRetryPrerequisite::CapacityAvailable }
1294    }
1295
1296    fn change_configuration() -> Self {
1297        Self {
1298            recovery: AttemptRetryRecovery::ChangeConfiguration,
1299            prerequisite: AttemptRetryPrerequisite::ConfigurationChanged,
1300        }
1301    }
1302
1303    fn retry_another_eligible_device() -> Self {
1304        Self {
1305            recovery: AttemptRetryRecovery::RetryAnotherEligibleDevice,
1306            prerequisite: AttemptRetryPrerequisite::QuarantinedDeviceAndEligibleAlternate,
1307        }
1308    }
1309
1310    fn recover_backend() -> Self {
1311        Self {
1312            recovery: AttemptRetryRecovery::RecoverBackend,
1313            prerequisite: AttemptRetryPrerequisite::BackendRecovered,
1314        }
1315    }
1316
1317    /// Derives the only retry advice retained for an immutable attempt
1318    /// report. This consumes only closed report facts and terminal evidence;
1319    /// it deliberately never turns a coarse temporary error status or an
1320    /// arbitrary error message into a retry recommendation.
1321    pub(crate) fn from_sealed_attempt(
1322        outcome: &RuntimeTerminalOutcome,
1323        primary_failure: Option<&AttemptFailureSummary>,
1324        capacity_disposition: AttemptCapacityDisposition,
1325        health_transition: Option<AttemptHealthTransition>,
1326        native_fault_evidence: Option<AttemptNativeFaultEvidence>,
1327        resource_evidence: Option<AttemptResourceEvidence>,
1328        alternate_device_eligibility: AttemptAlternateDeviceEligibility,
1329    ) -> Self {
1330        if requires_backend_recovery(primary_failure, health_transition, native_fault_evidence) {
1331            return Self::recover_backend();
1332        }
1333
1334        match outcome {
1335            RuntimeTerminalOutcome::Succeeded => Self::no_retry_needed(),
1336            RuntimeTerminalOutcome::Cancelled | RuntimeTerminalOutcome::Dropped | RuntimeTerminalOutcome::Panicked => {
1337                Self::do_not_retry()
1338            },
1339            RuntimeTerminalOutcome::Failed { .. } => {
1340                if has_proven_device_local_fault(capacity_disposition, health_transition, native_fault_evidence) {
1341                    return match alternate_device_eligibility {
1342                        AttemptAlternateDeviceEligibility::Eligible => Self::retry_another_eligible_device(),
1343                        AttemptAlternateDeviceEligibility::NoEligibleAlternate => Self::do_not_retry(),
1344                        AttemptAlternateDeviceEligibility::NotApplicable
1345                        | AttemptAlternateDeviceEligibility::Unavailable => Self::indeterminate(),
1346                    };
1347                }
1348
1349                let Some(failure) = primary_failure else {
1350                    return Self::indeterminate();
1351                };
1352                if is_configuration_failure(failure) {
1353                    return Self::change_configuration();
1354                }
1355                if has_reason(failure, AttemptFailureFactValue::RequiredFloorExceedsDeviceCapacity)
1356                    || has_reason(failure, AttemptFailureFactValue::QueryCapExceedsResidentCapacity)
1357                {
1358                    // An impossible cap is not queue pressure: backing off
1359                    // cannot change the outcome, so the caller must change the
1360                    // cap, the device, or release resident owners.
1361                    return Self::change_configuration();
1362                }
1363                if is_queue_overload(failure) {
1364                    return Self::backoff();
1365                }
1366                if matches!(failure.code(), AttemptFailureCode::Query(QueryErrorCode::MemoryReservationFailed)) {
1367                    return match exact_allocator_event_source(resource_evidence) {
1368                        Some(
1369                            AttemptAllocatorEventSource::ReservationLimit
1370                            | AttemptAllocatorEventSource::AllocationLimit,
1371                        ) => Self::change_configuration(),
1372                        // An exact physical allocator failure establishes
1373                        // neither that capacity will drain nor that the same
1374                        // demand can fit on a subsequent attempt. Backoff is
1375                        // defined only for queue overload.
1376                        Some(AttemptAllocatorEventSource::Physical) => Self::indeterminate(),
1377                        Some(AttemptAllocatorEventSource::Unknown) | None => Self::indeterminate(),
1378                    };
1379                }
1380                Self::indeterminate()
1381            },
1382        }
1383    }
1384}
1385
1386fn has_reason(failure: &AttemptFailureSummary, expected: AttemptFailureFactValue) -> bool {
1387    failure.facts().iter().any(|fact| fact.key() == AttemptFailureFactKey::Reason && fact.value() == expected)
1388}
1389
1390fn is_configuration_failure(failure: &AttemptFailureSummary) -> bool {
1391    matches!(
1392        failure.code(),
1393        AttemptFailureCode::Query(QueryErrorCode::UnsupportedPipelineContract)
1394            | AttemptFailureCode::Query(QueryErrorCode::UnsupportedOperator)
1395            | AttemptFailureCode::Query(QueryErrorCode::UnsupportedExpression)
1396            | AttemptFailureCode::Query(QueryErrorCode::UnsupportedType)
1397            | AttemptFailureCode::Query(QueryErrorCode::CudfUnsupported)
1398            | AttemptFailureCode::Query(QueryErrorCode::QueryMemoryLimitExceeded)
1399    )
1400}
1401
1402fn is_queue_overload(failure: &AttemptFailureSummary) -> bool {
1403    failure.code() == AttemptFailureCode::Query(QueryErrorCode::ServiceOverloaded)
1404        && failure.facts().iter().any(|fact| {
1405            fact.key() == AttemptFailureFactKey::Reason
1406                && matches!(
1407                    fact.value(),
1408                    AttemptFailureFactValue::AdmissionQueueFull
1409                        | AttemptFailureFactValue::QueueWaitElapsed
1410                        | AttemptFailureFactValue::MemoryGrantWaitTimeout
1411                )
1412        })
1413}
1414
1415fn requires_backend_recovery(
1416    primary_failure: Option<&AttemptFailureSummary>,
1417    health_transition: Option<AttemptHealthTransition>,
1418    native_fault_evidence: Option<AttemptNativeFaultEvidence>,
1419) -> bool {
1420    matches!(health_transition, Some(AttemptHealthTransition::BackendUnhealthy))
1421        || native_fault_evidence.is_some_and(|evidence| evidence.isolation == AttemptFaultIsolation::Unknown)
1422        || primary_failure.is_some_and(|failure| has_reason(failure, AttemptFailureFactValue::BackendUnhealthy))
1423}
1424
1425fn has_proven_device_local_fault(
1426    capacity_disposition: AttemptCapacityDisposition,
1427    health_transition: Option<AttemptHealthTransition>,
1428    native_fault_evidence: Option<AttemptNativeFaultEvidence>,
1429) -> bool {
1430    matches!(capacity_disposition, AttemptCapacityDisposition::RetainedQuarantine)
1431        && matches!(health_transition, Some(AttemptHealthTransition::DeviceQuarantined))
1432        && native_fault_evidence.as_ref().is_some_and(has_proven_device_local_fault_evidence)
1433}
1434
1435/// The native evidence forms that establish a device-local fault, as distinct
1436/// from a conservative quarantine. The release finalizer and sealed report share
1437/// this predicate so alternate eligibility cannot be calculated for weaker
1438/// evidence than the report will accept.
1439pub(crate) fn has_proven_device_local_fault_evidence(evidence: &AttemptNativeFaultEvidence) -> bool {
1440    if evidence.isolation != AttemptFaultIsolation::DeviceLocal {
1441        return false;
1442    }
1443    matches!(evidence.teardown_health, AttemptTeardownHealth::StreamSynchronizationFailed)
1444        || (evidence.device_health.is_some_and(|health| health.status == AttemptDeviceHealthStatus::Unhealthy)
1445            && evidence.first_fault.is_some())
1446}
1447
1448fn exact_allocator_event_source(
1449    resource_evidence: Option<AttemptResourceEvidence>,
1450) -> Option<AttemptAllocatorEventSource> {
1451    let allocator_event = resource_evidence?.allocator_event;
1452    (allocator_event.provenance == AttemptEvidenceProvenance::AttemptExact)
1453        .then_some(allocator_event.value?)
1454        .map(|event| event.source)
1455}
1456
1457#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1458/// Immutable device grant and wait duration committed by admission.
1459pub struct AttemptAdmissionFact {
1460    /// Grant publication sequence, independent of FIFO submission identity.
1461    pub admission_sequence: u64,
1462    /// Elapsed nanoseconds from admission submission to the grant.
1463    pub admission_wait_ns: u64,
1464    /// CUDA device selected for the immutable grant.
1465    pub device_ordinal: usize,
1466    /// Immutable admitted query capacity in bytes.
1467    pub grant_bytes: u64,
1468}
1469
1470/// Final queue state captured when a queued admission receives its immutable
1471/// grant.
1472///
1473/// This is a sealed controller fact and is not a runner-side estimate.
1474/// `submission_sequence` is FIFO identity, independent of the later
1475/// admission sequence.
1476#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1477pub struct AttemptAdmissionWaitFact {
1478    /// FIFO admission queue identity fixed at submission.
1479    pub submission_sequence: u64,
1480}
1481#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1482/// Attempt lifecycle times in nanoseconds from one monotonic attempt clock.
1483/// Optional boundaries are absent when they were not recorded. These values
1484/// are elapsed times, without a wall-clock epoch.
1485pub struct AttemptTimestamps {
1486    /// Time at attempt creation.
1487    pub submitted_ns: u64,
1488    /// Time planning began, or `None` when planning never started.
1489    pub planning_started_ns: Option<u64>,
1490    /// Time planning finished, or `None` when no finish was recorded.
1491    pub planning_finished_ns: Option<u64>,
1492    /// Sampled when the attempt claims its one admission, the first step of
1493    /// the query-service admission request.
1494    pub admission_requested_ns: Option<u64>,
1495    /// Time the immutable grant was published, or `None` when unadmitted.
1496    pub admitted_ns: Option<u64>,
1497    /// Time physical execution began, or `None` when it never began.
1498    pub physical_execution_started_ns: Option<u64>,
1499    /// Time first GPU dispatch began, or `None` when there was no dispatch.
1500    pub gpu_execution_started_ns: Option<u64>,
1501    /// Time the first terminal execution outcome was committed.
1502    pub outcome_terminal_ns: u64,
1503    /// Time cleanup completed, or `None` when that boundary was not recorded.
1504    pub cleanup_complete_ns: Option<u64>,
1505    /// Time the terminal report was fixed for publication, or `None` when absent.
1506    pub report_ready_ns: Option<u64>,
1507}
1508fn deserialize_query_attempt_report_schema<'de, D>(deserializer: D) -> std::result::Result<String, D::Error>
1509where D: serde::Deserializer<'de> {
1510    let schema = String::deserialize(deserializer)?;
1511    if schema != QUERY_ATTEMPT_REPORT_SCHEMA {
1512        return Err(de::Error::custom(format!(
1513            "unsupported attempt report schema {schema:?}; expected {QUERY_ATTEMPT_REPORT_SCHEMA:?}"
1514        )));
1515    }
1516    Ok(schema)
1517}
1518
1519#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1520/// Immutable terminal attempt record published after cleanup and sealing.
1521///
1522/// The report preserves the execution outcome separately from cleanup, health
1523/// and capacity disposition. Retry advice is derived from those committed
1524/// facts; delivery failures do not change the report. Duration helpers
1525/// return `None` for missing boundaries and saturate reversed intervals at zero.
1526pub struct QueryAttemptReport {
1527    #[serde(deserialize_with = "deserialize_query_attempt_report_schema")]
1528    /// Schema identity accepted by report decoding.
1529    pub schema: String,
1530    /// Runtime identity shared by the attempt and its observations.
1531    pub attempt_id: AttemptId,
1532    /// Validated caller query correlation, or `None` when none was supplied.
1533    pub query_id: Option<AttemptQueryId>,
1534    /// Validated opaque caller trace correlation, or `None` when none was supplied.
1535    pub trace_context: Option<AttemptTraceContext>,
1536    /// Planning coverage of this attempt, independent of execution success.
1537    pub disposition: AttemptDisposition,
1538    /// First terminal execution outcome committed by the attempt owner.
1539    pub outcome: RuntimeTerminalOutcome,
1540    /// Bounded primary terminal-failure facts retained by the attempt
1541    /// lifecycle. Successful attempts have no primary summary.
1542    pub primary_failure: Option<AttemptFailureSummary>,
1543    /// The sole evidence-gated recovery recommendation derived as this
1544    /// report is sealed. It is required and never inferred from an error
1545    /// status at read time.
1546    pub retry_advice: AttemptRetryAdvice,
1547    /// Selected CUDA device, or `None` when no device-backed admission occurred.
1548    pub device_ordinal: Option<usize>,
1549    /// Immutable grant facts, or `None` when the attempt was unadmitted.
1550    pub admission: Option<AttemptAdmissionFact>,
1551    /// Final queue facts for an admitted request. `None` means the
1552    /// execution path did not retain a queue-backed admission handle; it is
1553    /// never reconstructed from snapshots after the report is sealed.
1554    pub admission_wait: Option<AttemptAdmissionWaitFact>,
1555    /// Immutable engine-owned composed schedule, when this admitted attempt
1556    /// used the composed graph execution contract. Ordinary attempts leave it
1557    /// absent; it is never reconstructed from runtime observations.
1558    #[serde(default)]
1559    pub composed_schedule: Option<AttemptComposedScheduleProjection>,
1560    /// Monotonic lifecycle boundary times for duration calculations.
1561    pub timestamps: AttemptTimestamps,
1562    /// Cleanup completion classification committed before publication.
1563    pub cleanup_disposition: AttemptCleanupDisposition,
1564    /// Exact when `cleanup_failure_count_complete` is true. The lifecycle
1565    /// never publishes a saturated arithmetic result as exact.
1566    pub cleanup_failure_count: u64,
1567    /// Required so decoding never fabricates an exact cleanup-failure count.
1568    pub cleanup_failure_count_complete: bool,
1569    /// First bounded cleanup failure. Further failures affect only the exact
1570    /// count; timeline retention is intentionally not authoritative.
1571    pub first_cleanup_failure: Option<CleanupFailureSummary>,
1572    /// Disposition of the immutable grant after terminal cleanup.
1573    pub capacity_disposition: AttemptCapacityDisposition,
1574    /// Exact final domain usage after admitted cleanup, when sampling
1575    /// succeeded. `None` is an unavailable sample, never a zero sample.
1576    pub final_domain_usage: Option<AttemptFinalDomainUsage>,
1577    /// Health result of the admitted release commit. `None` means no
1578    /// trustworthy admitted terminal release-commit fact was recorded (for
1579    /// example, an unadmitted attempt).
1580    pub health_transition: Option<AttemptHealthTransition>,
1581    /// Bounded cuDF/cuGraph native status, isolation, and teardown facts from
1582    /// the same terminal release commit. `None` means the attempt did not own
1583    /// a device-backed domain. No native device status applies to such an attempt.
1584    pub native_fault_evidence: Option<AttemptNativeFaultEvidence>,
1585    /// Bounded ledger, immutable-cap, domain-usage, and allocator evidence
1586    /// from the sole terminal release commit. `None` means the attempt did
1587    /// not own a device-backed admission domain.
1588    pub resource_evidence: Option<AttemptResourceEvidence>,
1589    /// Observation records lost because retention or recording failed.
1590    pub observation_lost_records: u64,
1591    /// Records retained in the sealed bounded observation.
1592    pub observation_retained_records: u64,
1593    /// Deterministic logical bytes retained in the sealed observation.
1594    pub observation_retained_bytes: u64,
1595    /// Whether retention limits prevented complete observation capture.
1596    pub observation_truncated: bool,
1597    /// Whether recording degraded for reasons other than configured retention limits.
1598    pub observation_degraded: bool,
1599    /// Optional wall-clock nanosecond anchor associated with the observation clock.
1600    pub unix_epoch_ns: Option<u64>,
1601}
1602
1603impl QueryAttemptReport {
1604    /// Returns the compact, fixed-shape projection used by post-seal public
1605    /// consumers. The complete report remains the canonical retained record;
1606    /// this helper is only a read-only projection and never reconstructs facts
1607    /// from an error or delivery result.
1608    pub fn sealed_projection(&self) -> SealedAttemptProjection {
1609        SealedAttemptProjection::from(self)
1610    }
1611}
1612/// Bounded projection of facts from one already-sealed attempt report.
1613///
1614/// Unlike [`QueryAttemptReport`], this type is intentionally suitable for
1615/// fixed-cardinality metrics and small transport metadata. It contains no
1616/// caller correlations, diagnostic text, failure facts, allocator event data,
1617/// or query payload. The attempt id is retained solely for delivery surfaces
1618/// that already identify a single attempt; metric consumers must not use it as
1619/// a label.
1620#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1621pub struct SealedAttemptProjection {
1622    /// Schema label for this bounded projection.
1623    pub schema: &'static str,
1624    /// Attempt identity for individual delivery; excluded from metric labels.
1625    pub attempt_id: AttemptId,
1626    /// Planning coverage copied from the report.
1627    pub disposition: AttemptDisposition,
1628    /// Terminal execution outcome copied from the report.
1629    pub outcome: RuntimeTerminalOutcome,
1630    /// Cleanup result copied from the report.
1631    pub cleanup_disposition: AttemptCleanupDisposition,
1632    /// Recorded cleanup failure count; consult its completeness flag.
1633    pub cleanup_failure_count: u64,
1634    /// Whether the cleanup failure count is exact.
1635    pub cleanup_failure_count_complete: bool,
1636    /// Grant ownership after cleanup.
1637    pub capacity_disposition: AttemptCapacityDisposition,
1638    /// Committed admitted-device health result, or `None` when unavailable.
1639    pub health_transition: Option<AttemptHealthTransition>,
1640    /// Recovery recommendation derived at report seal.
1641    pub retry_advice: AttemptRetryAdvice,
1642    /// Whether configured retention bounds truncated observation capture.
1643    pub observation_truncated: bool,
1644    /// Whether recording encountered degradation beyond retention limits.
1645    pub observation_degraded: bool,
1646    /// Whether the full report contains native-fault evidence.
1647    pub native_fault_evidence_present: bool,
1648    /// Whether the full report contains terminal resource evidence.
1649    pub resource_evidence_present: bool,
1650}
1651
1652/// Stable fixed-cardinality labels derived from one sealed attempt projection.
1653///
1654/// Every consumer that renders attempt facts with labels (DataFusion metrics,
1655/// Flight query stats, and future bounded exporters) must use this projection
1656/// and must not duplicate enum-to-string mappings at its own boundary.
1657#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1658pub struct SealedAttemptLabels {
1659    /// Stable planning coverage label.
1660    pub disposition: &'static str,
1661    /// Stable terminal outcome label.
1662    pub outcome: &'static str,
1663    /// Stable cleanup result label.
1664    pub cleanup_disposition: &'static str,
1665    /// `true` or `false` for cleanup count exactness.
1666    pub cleanup_failure_count_complete: &'static str,
1667    /// Stable post-cleanup grant ownership label.
1668    pub capacity_disposition: &'static str,
1669    /// Stable health label, including `none` for unavailable evidence.
1670    pub health_transition: &'static str,
1671    /// Stable advised recovery label.
1672    pub retry_recovery: &'static str,
1673    /// Stable recovery prerequisite label.
1674    pub retry_prerequisite: &'static str,
1675    /// `true` or `false` for retention truncation.
1676    pub observation_truncated: &'static str,
1677    /// `true` or `false` for recording degradation.
1678    pub observation_degraded: &'static str,
1679    /// `present` or `absent` for native-fault evidence.
1680    pub native_fault_evidence: &'static str,
1681    /// `present` or `absent` for resource evidence.
1682    pub resource_evidence: &'static str,
1683}
1684
1685impl SealedAttemptProjection {
1686    /// Returns the sole stable label mapping for bounded post-seal consumers.
1687    pub const fn labels(&self) -> SealedAttemptLabels {
1688        SealedAttemptLabels {
1689            disposition: match self.disposition {
1690                AttemptDisposition::Unplanned => "unplanned",
1691                AttemptDisposition::CpuFallback => "cpu_fallback",
1692                AttemptDisposition::PartialNative => "partial_native",
1693                AttemptDisposition::Native => "native",
1694            },
1695            outcome: self.outcome.stable_label(),
1696            cleanup_disposition: match self.cleanup_disposition {
1697                AttemptCleanupDisposition::NotRequired => "not_required",
1698                AttemptCleanupDisposition::Completed => "completed",
1699                AttemptCleanupDisposition::CompletedWithFailures => "completed_with_failures",
1700            },
1701            cleanup_failure_count_complete: if self.cleanup_failure_count_complete { "true" } else { "false" },
1702            capacity_disposition: match self.capacity_disposition {
1703                AttemptCapacityDisposition::NotAdmitted => "not_admitted",
1704                AttemptCapacityDisposition::Released => "released",
1705                AttemptCapacityDisposition::FrozenResident => "frozen_resident",
1706                AttemptCapacityDisposition::RetainedQuarantine => "retained_quarantine",
1707            },
1708            health_transition: match self.health_transition {
1709                None => "none",
1710                Some(AttemptHealthTransition::HealthyReuse) => "healthy_reuse",
1711                Some(AttemptHealthTransition::DeviceQuarantined) => "device_quarantined",
1712                Some(AttemptHealthTransition::BackendUnhealthy) => "backend_unhealthy",
1713            },
1714            retry_recovery: self.retry_advice.recovery.as_str(),
1715            retry_prerequisite: self.retry_advice.prerequisite.as_str(),
1716            observation_truncated: if self.observation_truncated { "true" } else { "false" },
1717            observation_degraded: if self.observation_degraded { "true" } else { "false" },
1718            native_fault_evidence: if self.native_fault_evidence_present { "present" } else { "absent" },
1719            resource_evidence: if self.resource_evidence_present { "present" } else { "absent" },
1720        }
1721    }
1722}
1723
1724impl From<&QueryAttemptReport> for SealedAttemptProjection {
1725    fn from(report: &QueryAttemptReport) -> Self {
1726        Self {
1727            schema: SEALED_ATTEMPT_PROJECTION_SCHEMA,
1728            attempt_id: report.attempt_id,
1729            disposition: report.disposition,
1730            outcome: report.outcome.clone(),
1731            cleanup_disposition: report.cleanup_disposition,
1732            cleanup_failure_count: report.cleanup_failure_count,
1733            cleanup_failure_count_complete: report.cleanup_failure_count_complete,
1734            capacity_disposition: report.capacity_disposition,
1735            health_transition: report.health_transition,
1736            retry_advice: report.retry_advice,
1737            observation_truncated: report.observation_truncated,
1738            observation_degraded: report.observation_degraded,
1739            native_fault_evidence_present: report.native_fault_evidence.is_some(),
1740            resource_evidence_present: report.resource_evidence.is_some(),
1741        }
1742    }
1743}
1744impl QueryAttemptReport {
1745    /// Returns submission-to-report-ready duration in nanoseconds.
1746    /// Returns `None` when report readiness was not recorded.
1747    pub fn total_ns(&self) -> Option<u64> {
1748        self.timestamps.report_ready_ns.map(|ready| ready.saturating_sub(self.timestamps.submitted_ns))
1749    }
1750    /// Returns planning-start-to-planning-finish duration in nanoseconds.
1751    /// Returns `None` when either boundary is absent.
1752    pub fn planning_ns(&self) -> Option<u64> {
1753        self.timestamps
1754            .planning_started_ns
1755            .zip(self.timestamps.planning_finished_ns)
1756            .map(|(start, end)| end.saturating_sub(start))
1757    }
1758    /// Returns admission-request-to-grant duration in nanoseconds.
1759    /// Returns `None` when either boundary is absent.
1760    pub fn queue_wait_ns(&self) -> Option<u64> {
1761        self.timestamps
1762            .admission_requested_ns
1763            .zip(self.timestamps.admitted_ns)
1764            .map(|(start, end)| end.saturating_sub(start))
1765    }
1766    /// Returns planning-finish-to-admission-request duration in nanoseconds.
1767    /// Returns `None` when either boundary is absent.
1768    pub fn post_planning_pre_admission_ns(&self) -> Option<u64> {
1769        self.timestamps
1770            .planning_finished_ns
1771            .zip(self.timestamps.admission_requested_ns)
1772            .map(|(start, end)| end.saturating_sub(start))
1773    }
1774    /// Returns grant-to-physical-execution-start duration in nanoseconds.
1775    /// Returns `None` when either boundary is absent.
1776    pub fn admitted_to_physical_execution_ns(&self) -> Option<u64> {
1777        self.timestamps
1778            .admitted_ns
1779            .zip(self.timestamps.physical_execution_started_ns)
1780            .map(|(start, end)| end.saturating_sub(start))
1781    }
1782    /// Returns physical-execution-start-to-GPU-dispatch duration in nanoseconds.
1783    /// Returns `None` when either boundary is absent.
1784    pub fn physical_to_gpu_dispatch_ns(&self) -> Option<u64> {
1785        self.timestamps
1786            .physical_execution_started_ns
1787            .zip(self.timestamps.gpu_execution_started_ns)
1788            .map(|(start, end)| end.saturating_sub(start))
1789    }
1790    /// Returns grant-to-first-GPU-dispatch duration in nanoseconds.
1791    /// Returns `None` when either boundary is absent.
1792    pub fn admitted_to_gpu_dispatch_ns(&self) -> Option<u64> {
1793        self.timestamps
1794            .admitted_ns
1795            .zip(self.timestamps.gpu_execution_started_ns)
1796            .map(|(start, end)| end.saturating_sub(start))
1797    }
1798    /// Returns submission-to-first-GPU-dispatch duration in nanoseconds.
1799    /// Returns `None` when GPU dispatch was not recorded.
1800    pub fn submitted_to_gpu_dispatch_ns(&self) -> Option<u64> {
1801        self.timestamps.gpu_execution_started_ns.map(|end| end.saturating_sub(self.timestamps.submitted_ns))
1802    }
1803    /// Returns first-GPU-dispatch-to-terminal-outcome duration in nanoseconds.
1804    /// Returns `None` when GPU dispatch was not recorded.
1805    pub fn gpu_dispatch_to_terminal_ns(&self) -> Option<u64> {
1806        self.timestamps.gpu_execution_started_ns.map(|start| self.timestamps.outcome_terminal_ns.saturating_sub(start))
1807    }
1808    /// Returns physical-execution-start-to-terminal-outcome duration in nanoseconds.
1809    /// Returns `None` when physical execution did not start.
1810    pub fn physical_execution_ns(&self) -> Option<u64> {
1811        self.timestamps
1812            .physical_execution_started_ns
1813            .map(|start| self.timestamps.outcome_terminal_ns.saturating_sub(start))
1814    }
1815    /// Returns terminal-outcome-to-cleanup-complete duration in nanoseconds.
1816    /// Returns `None` when cleanup completion was not recorded.
1817    pub fn cleanup_ns(&self) -> Option<u64> {
1818        self.timestamps.cleanup_complete_ns.map(|end| end.saturating_sub(self.timestamps.outcome_terminal_ns))
1819    }
1820}
1821/// Query terminal outcome stored outside the bounded record collection.
1822#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1823#[serde(tag = "kind", rename_all = "snake_case")]
1824pub enum RuntimeTerminalOutcome {
1825    /// Execution reached successful completion.
1826    Succeeded,
1827    /// A terminal error classified from the terminal structured error itself.
1828    Failed {
1829        /// Stable machine identity of the terminal error.
1830        error_code: AttemptFailureCode,
1831        /// Source condition classification; recovery requires the sealed retry advice.
1832        error_status: AttemptFailureStatus,
1833    },
1834    /// Caller cancellation or deadline expiry ended execution.
1835    Cancelled,
1836    /// The attempt or result stream was dropped before another terminal outcome.
1837    Dropped,
1838    /// An unwind ended the attempt.
1839    Panicked,
1840}
1841
1842impl RuntimeTerminalOutcome {
1843    /// Stable closed label for bounded post-seal projections.
1844    pub const fn stable_label(&self) -> &'static str {
1845        match self {
1846            Self::Succeeded => "succeeded",
1847            Self::Failed { .. } => "failed",
1848            Self::Cancelled => "cancelled",
1849            Self::Dropped => "dropped",
1850            Self::Panicked => "panicked",
1851        }
1852    }
1853
1854    /// Copies the stable code and status from a primary failure summary.
1855    pub const fn failed(summary: &AttemptFailureSummary) -> Self {
1856        Self::Failed { error_code: summary.code(), error_status: summary.status() }
1857    }
1858
1859    /// Reduces a native error to its closed terminal projection. Attempt
1860    /// owners must still call the borrowed summary recorder first when they
1861    /// publish an attempt report.
1862    pub fn from_query_error(error: &QueryError) -> Self {
1863        if error.code() == QueryErrorCode::Cancelled {
1864            Self::Cancelled
1865        } else {
1866            Self::Failed {
1867                error_code: AttemptFailureCode::from_query_error_code(error.code()),
1868                error_status: AttemptFailureStatus::from_query_error_status(error.status()),
1869            }
1870        }
1871    }
1872
1873    /// Returns the bounded primary-failure projection that must accompany a
1874    /// terminal native error. Ordinary caller cancellation intentionally
1875    /// remains a text-free control outcome, while an expired deadline retains
1876    /// its stable reason beside the same `Cancelled` terminal outcome.
1877    pub(crate) fn primary_failure_from_query_error(
1878        error: &QueryError,
1879        operation: AttemptFailureOperation,
1880    ) -> Option<AttemptFailureSummary> {
1881        match error.code() {
1882            QueryErrorCode::Cancelled if error.fact_value("reason") != Some("deadline_expired") => None,
1883            _ => Some(AttemptFailureSummary::from_native_error(error, operation)),
1884        }
1885    }
1886
1887    pub(crate) const fn from_primary_failure(summary: &AttemptFailureSummary) -> Self {
1888        match summary.code() {
1889            AttemptFailureCode::Query(QueryErrorCode::Cancelled) => Self::Cancelled,
1890            _ => Self::failed(summary),
1891        }
1892    }
1893
1894    pub(crate) fn fallback_failure_summary(&self) -> Option<AttemptFailureSummary> {
1895        match self {
1896            Self::Succeeded => None,
1897            Self::Failed { error_code, error_status } => Some(AttemptFailureSummary {
1898                origin: AttemptFailureOrigin::Native,
1899                code: *error_code,
1900                status: *error_status,
1901                operation: AttemptFailureOperation::Execution,
1902                text: None,
1903                facts: Vec::new(),
1904            }),
1905            Self::Cancelled => {
1906                Some(AttemptFailureSummary::control(AttemptFailureCode::Query(QueryErrorCode::Cancelled)))
1907            },
1908            Self::Dropped => Some(AttemptFailureSummary::control(AttemptFailureCode::Dropped)),
1909            Self::Panicked => Some(AttemptFailureSummary::control(AttemptFailureCode::Panicked)),
1910        }
1911    }
1912}
1913#[cfg(test)]
1914mod tests;