knowledge_runtime/obs/
trace.rs

1//! Query and projection observability traces.
2//!
3//! Every query execution produces a `QueryTrace` that records classification,
4//! route plan, per-leg timing, merge statistics, and any degradation warnings.
5//! Projection lifecycle actions produce a `ProjectionTrace`.
6
7use crate::ids::ScopeKey;
8use crate::projection::lifecycle::{ProjectionHealth, StaleCause};
9use crate::query::classify::ClassifyResult;
10use crate::query::merge::MergedResults;
11use crate::query::route::{RetrievalStrategy, RoutePlan};
12use schemars::JsonSchema;
13use serde::{Deserialize, Serialize};
14use stack_ids::TraceCtx;
15use std::time::Duration;
16
17/// A warning or degradation notice emitted during query execution.
18///
19/// These make runtime behavior transparent to callers — especially
20/// when the runtime falls back to simpler behavior than requested.
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
22#[serde(rename_all = "snake_case")]
23pub enum QueryWarning {
24    /// A temporal search leg was downgraded to hybrid search because
25    /// temporal execution is not implemented.
26    TemporalDowngradedToHybrid {
27        /// The original temporal expression from the query.
28        temporal_expr: String,
29    },
30    /// Scope enforcement is limited to namespace-only at the adapter level.
31    /// Full scope (domain, workspace_id, repo_id) was used for runtime-owned
32    /// logic but NOT enforced during upstream search.
33    ScopePartiallyEnforced {
34        /// The full scope that was requested.
35        full_scope: ScopeKey,
36    },
37    /// Entity resolution fell back to a broader scope.
38    EntityScopeFallback {
39        mention: String,
40        queried_scope: ScopeKey,
41    },
42    /// Projection data for this scope is stale due to import lag or failure.
43    ProjectionImportStale {
44        /// Scope affected.
45        scope: ScopeKey,
46        /// When the last successful import occurred, if ever.
47        last_import_at: Option<String>,
48    },
49}
50
51/// Logical view through which a query leg is executed.
52#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
53#[serde(rename_all = "snake_case")]
54pub enum RuntimeView {
55    Semantic,
56    Temporal,
57    Entity,
58    Causal,
59    Control,
60}
61
62/// Disclosure record emitted when a query leg is widened or degraded from its requested view.
63#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
64pub struct WideningDisclosure {
65    pub leg_index: usize,
66    pub requested_view: RuntimeView,
67    pub executed_view: RuntimeView,
68    pub reason: String,
69}
70
71/// Runtime-level summary of a semantic-memory derived candidate receipt.
72///
73/// This mirrors semantic-memory receipt fields without depending on proveKV or
74/// compression crate APIs.
75#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
76pub struct RuntimeDerivedCandidateTraceV1 {
77    pub candidate_backend: String,
78    #[serde(default, skip_serializing_if = "Option::is_none")]
79    pub codec_family: Option<String>,
80    #[serde(default, skip_serializing_if = "Option::is_none")]
81    pub generation_id: Option<String>,
82    #[serde(default, skip_serializing_if = "Option::is_none")]
83    pub embedding_snapshot_digest: Option<String>,
84    #[serde(default, skip_serializing_if = "Option::is_none")]
85    pub pool_manifest_digest: Option<String>,
86    pub exact_rerank: bool,
87    pub approximate: bool,
88    #[serde(default, skip_serializing_if = "Option::is_none")]
89    pub fallback: Option<String>,
90    pub raw_candidate_count: usize,
91    pub post_filter_count: usize,
92    pub final_result_count: usize,
93}
94
95/// Observability record for a single query execution.
96#[derive(Debug, Clone, Serialize, Deserialize)]
97pub struct QueryTrace {
98    /// Canonical trace context for cross-crate correlation.
99    pub trace_ctx: TraceCtx,
100    /// Phase status: compatibility / migration-only.
101    /// Legacy trace_id extracted from trace_ctx for backward compat.
102    #[serde(skip_serializing_if = "Option::is_none")]
103    pub trace_id: Option<String>,
104    /// Scope used for this query.
105    pub scope: ScopeKey,
106    /// Classification result.
107    pub classification: ClassifyResult,
108    /// Route plan that was executed.
109    pub plan: RoutePlan,
110    /// Per-leg timing in milliseconds.
111    pub leg_timings_ms: Vec<u64>,
112    /// Total query duration.
113    #[serde(with = "duration_ms")]
114    pub total_duration: Duration,
115    /// Number of raw results before merge.
116    pub raw_result_count: usize,
117    /// Number of results after merge.
118    pub merged_result_count: usize,
119    /// Number of duplicates fused during merge.
120    pub duplicates_fused: usize,
121    /// Distinct views requested by the route plan.
122    #[serde(default, skip_serializing_if = "Vec::is_empty")]
123    pub requested_views: Vec<RuntimeView>,
124    /// Distinct views actually executed after degradation/widening.
125    #[serde(default, skip_serializing_if = "Vec::is_empty")]
126    pub executed_views: Vec<RuntimeView>,
127    /// Explicit widening/degradation disclosures.
128    #[serde(default, skip_serializing_if = "Vec::is_empty")]
129    pub widenings: Vec<WideningDisclosure>,
130    /// Warnings and degradation notices.
131    pub warnings: Vec<QueryWarning>,
132    /// Semantic-memory derived candidate backend receipts observed while executing legs.
133    #[serde(default, skip_serializing_if = "Vec::is_empty")]
134    pub derived_candidate_receipts: Vec<RuntimeDerivedCandidateTraceV1>,
135    /// Kernel scheduler degradation surfaced by attached runtime inference, if any.
136    #[serde(default, skip_serializing_if = "Option::is_none")]
137    pub kernel_degraded_reason: Option<String>,
138    /// Bitemporal valid-time coordinate used for this query, if temporal.
139    #[serde(default, skip_serializing_if = "Option::is_none")]
140    pub valid_as_of: Option<String>,
141    /// Bitemporal recorded-time ceiling used for this query, if temporal.
142    #[serde(default, skip_serializing_if = "Option::is_none")]
143    pub recorded_as_of: Option<String>,
144    /// How the temporal coordinates were resolved: "exact", "downgraded", "none".
145    #[serde(default, skip_serializing_if = "Option::is_none")]
146    pub temporal_mode: Option<String>,
147}
148
149/// Observability record for a projection lifecycle action.
150#[derive(Debug, Clone, Serialize, Deserialize)]
151pub struct ProjectionTrace {
152    /// Canonical trace context.
153    pub trace_ctx: TraceCtx,
154    /// Phase status: compatibility / migration-only.
155    #[serde(skip_serializing_if = "Option::is_none")]
156    pub trace_id: Option<String>,
157    /// Which projection was affected.
158    pub projection_id: String,
159    /// Scope of the projection.
160    pub scope: ScopeKey,
161    /// What action was taken.
162    pub action: ProjectionAction,
163    /// Duration in milliseconds (for builds).
164    #[serde(skip_serializing_if = "Option::is_none")]
165    pub duration_ms: Option<u64>,
166    /// Number of source items processed (for builds).
167    #[serde(skip_serializing_if = "Option::is_none")]
168    pub source_count: Option<usize>,
169    /// Whether the action succeeded.
170    pub success: bool,
171    /// Error message if action failed.
172    #[serde(skip_serializing_if = "Option::is_none")]
173    pub error: Option<String>,
174    /// Resulting health after the action.
175    pub resulting_health: ProjectionHealth,
176    /// Stale cause if the projection is now stale.
177    #[serde(skip_serializing_if = "Option::is_none")]
178    pub stale_cause: Option<StaleCause>,
179}
180
181/// What lifecycle action was taken on a projection.
182#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
183#[serde(rename_all = "snake_case")]
184pub enum ProjectionAction {
185    Build,
186    Invalidate,
187    Remove,
188    Clear,
189    Failure,
190}
191
192impl QueryTrace {
193    /// Create a trace from query pipeline components.
194    #[allow(clippy::too_many_arguments)]
195    pub fn from_pipeline(
196        trace_ctx: TraceCtx,
197        scope: ScopeKey,
198        classification: ClassifyResult,
199        plan: RoutePlan,
200        leg_timings_ms: Vec<u64>,
201        total_duration: Duration,
202        merged: &MergedResults,
203        warnings: Vec<QueryWarning>,
204    ) -> Self {
205        let legacy_id = Some(trace_ctx.to_legacy_trace_id().to_string());
206        let (requested_views, executed_views) = merged_trace_plan_views(&warnings, &plan);
207        let widenings = build_widenings(&plan, &warnings);
208        Self {
209            trace_ctx,
210            trace_id: legacy_id,
211            scope,
212            classification,
213            plan,
214            leg_timings_ms,
215            total_duration,
216            raw_result_count: merged.total_raw,
217            merged_result_count: merged.results.len(),
218            duplicates_fused: merged.duplicates_fused,
219            requested_views: collect_requested_views(&requested_views),
220            executed_views: collect_requested_views(&executed_views),
221            widenings,
222            warnings,
223            derived_candidate_receipts: Vec::new(),
224            kernel_degraded_reason: None,
225            valid_as_of: None,
226            recorded_as_of: None,
227            temporal_mode: None,
228        }
229    }
230
231    /// Whether any temporal requests were downgraded.
232    pub fn has_temporal_downgrade(&self) -> bool {
233        self.warnings
234            .iter()
235            .any(|w| matches!(w, QueryWarning::TemporalDowngradedToHybrid { .. }))
236    }
237
238    /// Whether scope enforcement was only partial.
239    pub fn has_scope_enforcement_warning(&self) -> bool {
240        self.warnings
241            .iter()
242            .any(|w| matches!(w, QueryWarning::ScopePartiallyEnforced { .. }))
243    }
244
245    /// Whether import staleness was detected.
246    pub fn has_import_staleness_warning(&self) -> bool {
247        self.warnings
248            .iter()
249            .any(|w| matches!(w, QueryWarning::ProjectionImportStale { .. }))
250    }
251
252    /// Whether any degradation warnings were emitted.
253    pub fn is_degraded(&self) -> bool {
254        !self.warnings.is_empty()
255    }
256
257    /// Build a versioned provenance snapshot from this trace for cross-crate audit.
258    pub fn runtime_query_provenance(&self) -> RuntimeQueryProvenanceV1 {
259        RuntimeQueryProvenanceV1 {
260            schema_version: "runtime_query_provenance_v1".into(),
261            trace_ctx: self.trace_ctx.clone(),
262            scope: self.scope.clone(),
263            classification_mode: self.classification.mode.kind().into(),
264            classification_reason: self.classification.reason.clone(),
265            query: self.plan.query.clone(),
266            requested_views: self.requested_views.clone(),
267            executed_views: self.executed_views.clone(),
268            widenings: self.widenings.clone(),
269            warnings: self.warnings.clone(),
270            kernel_degraded_reason: self.kernel_degraded_reason.clone(),
271            leg_strategies: self
272                .plan
273                .legs
274                .iter()
275                .map(|leg| leg.strategy.kind().into())
276                .collect(),
277            leg_timings_ms: self.leg_timings_ms.clone(),
278            total_duration_ms: self.total_duration.as_millis() as u64,
279            raw_result_count: self.raw_result_count,
280            merged_result_count: self.merged_result_count,
281            duplicates_fused: self.duplicates_fused,
282            valid_as_of: self.valid_as_of.clone(),
283            recorded_as_of: self.recorded_as_of.clone(),
284            temporal_mode: self.temporal_mode.clone(),
285        }
286    }
287}
288
289/// Versioned provenance record summarizing a full query execution for downstream audit.
290#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
291#[schemars(title = "RuntimeQueryProvenanceV1")]
292pub struct RuntimeQueryProvenanceV1 {
293    pub schema_version: String,
294    pub trace_ctx: TraceCtx,
295    pub scope: ScopeKey,
296    pub classification_mode: String,
297    #[serde(default, skip_serializing_if = "Option::is_none")]
298    pub classification_reason: Option<String>,
299    pub query: String,
300    #[serde(default, skip_serializing_if = "Vec::is_empty")]
301    pub requested_views: Vec<RuntimeView>,
302    #[serde(default, skip_serializing_if = "Vec::is_empty")]
303    pub executed_views: Vec<RuntimeView>,
304    #[serde(default, skip_serializing_if = "Vec::is_empty")]
305    pub widenings: Vec<WideningDisclosure>,
306    #[serde(default, skip_serializing_if = "Vec::is_empty")]
307    pub warnings: Vec<QueryWarning>,
308    #[serde(default, skip_serializing_if = "Option::is_none")]
309    pub kernel_degraded_reason: Option<String>,
310    #[serde(default, skip_serializing_if = "Vec::is_empty")]
311    pub leg_strategies: Vec<String>,
312    #[serde(default, skip_serializing_if = "Vec::is_empty")]
313    pub leg_timings_ms: Vec<u64>,
314    pub total_duration_ms: u64,
315    pub raw_result_count: usize,
316    pub merged_result_count: usize,
317    pub duplicates_fused: usize,
318    /// Bitemporal valid-time coordinate used for this query, if temporal.
319    #[serde(default, skip_serializing_if = "Option::is_none")]
320    pub valid_as_of: Option<String>,
321    /// Bitemporal recorded-time ceiling used for this query, if temporal.
322    #[serde(default, skip_serializing_if = "Option::is_none")]
323    pub recorded_as_of: Option<String>,
324    /// How the temporal coordinates were resolved: "exact", "downgraded", "none".
325    #[serde(default, skip_serializing_if = "Option::is_none")]
326    pub temporal_mode: Option<String>,
327}
328
329fn strategy_view(strategy: &RetrievalStrategy) -> RuntimeView {
330    match strategy {
331        RetrievalStrategy::HybridSearch => RuntimeView::Semantic,
332        RetrievalStrategy::EntitySearch { .. } => RuntimeView::Entity,
333        RetrievalStrategy::TemporalSearch { .. } => RuntimeView::Temporal,
334    }
335}
336
337fn collect_requested_views(views: &[RuntimeView]) -> Vec<RuntimeView> {
338    let mut deduped = Vec::new();
339    for view in views {
340        if !deduped.contains(view) {
341            deduped.push(view.clone());
342        }
343    }
344    deduped
345}
346
347fn merged_trace_plan_views(
348    warnings: &[QueryWarning],
349    plan: &RoutePlan,
350) -> (Vec<RuntimeView>, Vec<RuntimeView>) {
351    let requested = plan
352        .legs
353        .iter()
354        .map(|leg| strategy_view(&leg.strategy))
355        .collect::<Vec<_>>();
356    let mut executed = requested.clone();
357    if warnings
358        .iter()
359        .any(|warning| matches!(warning, QueryWarning::TemporalDowngradedToHybrid { .. }))
360    {
361        for (index, leg) in plan.legs.iter().enumerate() {
362            if matches!(leg.strategy, RetrievalStrategy::TemporalSearch { .. }) {
363                executed[index] = RuntimeView::Semantic;
364            }
365        }
366    }
367    (requested, executed)
368}
369
370fn build_widenings(plan: &RoutePlan, warnings: &[QueryWarning]) -> Vec<WideningDisclosure> {
371    let mut widenings = Vec::new();
372    for (index, leg) in plan.legs.iter().enumerate() {
373        match &leg.strategy {
374            RetrievalStrategy::TemporalSearch { temporal_expr } => {
375                if warnings.iter().any(|warning| {
376                    matches!(
377                        warning,
378                        QueryWarning::TemporalDowngradedToHybrid { temporal_expr: seen }
379                            if seen == temporal_expr
380                    )
381                }) {
382                    widenings.push(WideningDisclosure {
383                        leg_index: index,
384                        requested_view: RuntimeView::Temporal,
385                        executed_view: RuntimeView::Semantic,
386                        reason: "temporal route degraded to semantic hybrid execution".into(),
387                    });
388                }
389            }
390            RetrievalStrategy::EntitySearch { mention } => {
391                if warnings.iter().any(|warning| {
392                    matches!(
393                        warning,
394                        QueryWarning::EntityScopeFallback { mention: seen, .. } if seen == mention
395                    )
396                }) {
397                    widenings.push(WideningDisclosure {
398                        leg_index: index,
399                        requested_view: RuntimeView::Entity,
400                        executed_view: RuntimeView::Entity,
401                        reason: "entity resolution widened to a broader candidate scope".into(),
402                    });
403                }
404            }
405            RetrievalStrategy::HybridSearch => {}
406        }
407    }
408    if let Some(scope_warning) = warnings
409        .iter()
410        .find(|warning| matches!(warning, QueryWarning::ScopePartiallyEnforced { .. }))
411    {
412        let reason = match scope_warning {
413            QueryWarning::ScopePartiallyEnforced { .. } => {
414                "adapter enforced namespace-only scope; wider scope was disclosed".to_string()
415            }
416            // LIB-HIGH-001: replaced unreachable!() with safe fallback
417            _ => "unknown".into(),
418        };
419        widenings.push(WideningDisclosure {
420            leg_index: 0,
421            requested_view: RuntimeView::Semantic,
422            executed_view: RuntimeView::Semantic,
423            reason,
424        });
425    }
426    widenings
427}
428
429/// Serde helper for Duration as milliseconds.
430mod duration_ms {
431    use serde::{Deserialize, Deserializer, Serialize, Serializer};
432    use std::time::Duration;
433
434    pub fn serialize<S: Serializer>(d: &Duration, s: S) -> Result<S::Ok, S::Error> {
435        d.as_millis().serialize(s)
436    }
437
438    pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Duration, D::Error> {
439        let ms = u64::deserialize(d)?;
440        Ok(Duration::from_millis(ms))
441    }
442}
knowledge_runtime/obs/trace.rs

knowledge_runtime/obs/
trace.rs
