feat: DAG data-flow timeline — per-operator distribution protocol + playhead UI

crates/analyzer/src/error.rs

@@ -28,4 +28,6 @@ pub enum AnalyzerError {
     FsmExitTransitionConversion,
     #[error("invalid argument: {0}")]
     InvalidArgument(String),
+    #[error("this analyzer does not support the requested capability")]
+    Unsupported,
 }

crates/analyzer/src/timeline/binned/categorical.rs

@@ -0,0 +1,171 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+//! Binned timelines of weighted values keyed by categories.
+//!
+//! A categorical timeline describes, per opaque series (e.g. an operator in a
+//! query engine), how some weighted quantity (a "measure", e.g. an entity
+//! count) breaks down over the states of a finite state machine and an
+//! application-defined dimension (e.g. which resource holds the entity's
+//! data), for each time bin of a window. Bin values are absolute,
+//! time-weighted quantities — not normalized shares.
+//!
+//! This module is application-agnostic: series, measures, states, and
+//! dimension keys are opaque to the aggregation. Downstream analyzers decide
+//! what they mean and are expected to keep dimension keys a small enumerable
+//! set.
+
+use std::hash::Hash;
+
+use rustc_hash::FxHashMap as HashMap;
+
+use quent_time::{SpanNanoSec, bin::BinnedSpan};
+
+use crate::{
+    AnalyzerResult,
+    timeline::binned::{BinnedTimelineAggregator, KeyedAggregator},
+};
+
+/// Identity of one aggregation cell of a categorical timeline.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct CategoricalKey<S, M, St, D> {
+    /// Opaque series the sample belongs to (e.g. an operator id downstream).
+    pub series: S,
+    /// The measure this weight contributes to (e.g. an entity count).
+    pub measure: M,
+    /// The FSM state name during the span.
+    pub state: St,
+    /// Application-defined dimension key (opaque to the aggregation).
+    pub dimension: D,
+}
+
+/// A binned timeline of weighted (state, dimension) values for multiple
+/// series and measures.
+#[derive(Clone, Debug)]
+pub struct CategoricalTimeline<S, M, St, D> {
+    pub config: BinnedSpan,
+    pub data: HashMap<CategoricalKey<S, M, St, D>, Vec<f64>>,
+}
+
+/// Builds a [`CategoricalTimeline`] from weighted samples.
+///
+/// Aggregation is span-weighted: each sample contributes
+/// `weight * overlap_fraction` to every bin its span intersects, so bin values
+/// are time-weighted averages over the bin, not instantaneous snapshots.
+pub struct CategoricalTimelineBuilder<S, M, St, D> {
+    aggregator: KeyedAggregator<CategoricalKey<S, M, St, D>>,
+}
+
+impl<S, M, St, D> CategoricalTimelineBuilder<S, M, St, D>
+where
+    S: Eq + Hash,
+    M: Eq + Hash,
+    St: Eq + Hash,
+    D: Eq + Hash,
+{
+    pub fn new(config: BinnedSpan) -> Self {
+        Self {
+            aggregator: KeyedAggregator::new(config),
+        }
+    }
+
+    /// Return the configuration of the binned timeline.
+    pub fn config(&self) -> BinnedSpan {
+        self.aggregator.config()
+    }
+
+    /// Attempt to push one weighted sample spanning `span` into the timeline.
+    pub fn try_push(
+        &mut self,
+        key: CategoricalKey<S, M, St, D>,
+        span: SpanNanoSec,
+        weight: f64,
+    ) -> AnalyzerResult<()> {
+        self.aggregator.try_push(span, (key, weight))
+    }
+
+    pub fn build(self) -> CategoricalTimeline<S, M, St, D> {
+        CategoricalTimeline {
+            config: self.aggregator.config(),
+            data: self.aggregator.finish(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::num::NonZero;
+
+    use super::*;
+
+    fn test_config() -> BinnedSpan {
+        BinnedSpan::try_new(
+            SpanNanoSec::try_new(0, 1000).unwrap(),
+            NonZero::try_from(10).unwrap(),
+        )
+        .unwrap()
+    }
+
+    fn key<'a>(
+        series: u32,
+        measure: &'a str,
+        state: &'a str,
+        dimension: &'a str,
+    ) -> CategoricalKey<u32, &'a str, &'a str, &'a str> {
+        CategoricalKey {
+            series,
+            measure,
+            state,
+            dimension,
+        }
+    }
+
+    /// Samples differing in any key component land in distinct cells; the
+    /// binning math itself is covered by the aggregator's own tests.
+    #[test]
+    fn distinct_series_measures_states_dimensions() -> AnalyzerResult<()> {
+        let mut builder = CategoricalTimelineBuilder::new(test_config());
+        let span = SpanNanoSec::try_new(0, 1000).unwrap();
+
+        builder.try_push(key(1, "count", "a", "x"), span, 1.0)?;
+        builder.try_push(key(2, "count", "a", "x"), span, 1.0)?;
+        builder.try_push(key(1, "bytes", "a", "x"), span, 100.0)?;
+        builder.try_push(key(1, "count", "b", "x"), span, 1.0)?;
+        builder.try_push(key(1, "count", "a", "y"), span, 1.0)?;
+
+        let timeline = builder.build();
+        assert_eq!(timeline.data.len(), 5);
+        assert_eq!(
+            timeline.data.get(&key(1, "bytes", "a", "x")).unwrap()[..],
+            [100.0; 10]
+        );
+        assert_eq!(
+            timeline.data.get(&key(2, "count", "a", "x")).unwrap()[..],
+            [1.0; 10]
+        );
+        Ok(())
+    }
+
+    /// Non-string key components only need `Eq + Hash` — no stringification.
+    #[test]
+    fn non_string_key_components() -> AnalyzerResult<()> {
+        #[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
+        enum Measure {
+            Count,
+        }
+
+        let mut builder: CategoricalTimelineBuilder<u32, Measure, u8, u16> =
+            CategoricalTimelineBuilder::new(test_config());
+        let cell = CategoricalKey {
+            series: 7u32,
+            measure: Measure::Count,
+            state: 3u8,
+            dimension: 9u16,
+        };
+        builder.try_push(cell.clone(), SpanNanoSec::try_new(0, 1000).unwrap(), 2.0)?;
+
+        let timeline = builder.build();
+        assert_eq!(timeline.data.get(&cell).unwrap()[..], [2.0; 10]);
+        Ok(())
+    }
+}

crates/analyzer/src/timeline/binned/mod.rs

@@ -9,6 +9,7 @@ use quent_time::{SpanNanoSec, bin::BinnedSpan};
 
 use crate::AnalyzerResult;
 
+pub mod categorical;
 pub mod resource;
 
 /// A trait for types that can aggregate items into a sequence of time bins.
@@ -146,4 +147,50 @@ mod tests {
 
         Ok(())
     }
+
+    fn ten_bin_config() -> BinnedSpan {
+        BinnedSpan::try_new(
+            SpanNanoSec::try_new(0, 1000).unwrap(),
+            NonZero::new(10).unwrap(),
+        )
+        .unwrap()
+    }
+
+    /// Overlapping spans accumulate span-weighted fractions per bin.
+    #[test]
+    fn keyed_aggregator_span_weighting_across_bin_boundaries() -> AnalyzerResult<()> {
+        let mut aggregator: KeyedAggregator<&str> = KeyedAggregator::new(ten_bin_config());
+
+        aggregator.try_push(SpanNanoSec::try_new(0, 300).unwrap(), ("k", 1.0))?;
+        aggregator.try_push(SpanNanoSec::try_new(250, 450).unwrap(), ("k", 1.0))?;
+
+        let bins = aggregator.finish();
+        assert_eq!(
+            bins.get("k").unwrap()[..],
+            [1.0, 1.0, 1.5, 1.0, 0.5, 0.0, 0.0, 0.0, 0.0, 0.0]
+        );
+        Ok(())
+    }
+
+    /// Zero-duration spans create the key but contribute nothing.
+    #[test]
+    fn keyed_aggregator_zero_duration_span_is_noop() -> AnalyzerResult<()> {
+        let mut aggregator: KeyedAggregator<&str> = KeyedAggregator::new(ten_bin_config());
+        aggregator.try_push(SpanNanoSec::try_new(500, 500).unwrap(), ("k", 1.0))?;
+
+        let bins = aggregator.finish();
+        assert_eq!(bins.get("k").unwrap()[..], [0.0; 10]);
+        Ok(())
+    }
+
+    /// Spans entirely outside the window contribute nothing.
+    #[test]
+    fn keyed_aggregator_out_of_window_span_contributes_nothing() -> AnalyzerResult<()> {
+        let mut aggregator: KeyedAggregator<&str> = KeyedAggregator::new(ten_bin_config());
+        aggregator.try_push(SpanNanoSec::try_new(2000, 3000).unwrap(), ("k", 1.0))?;
+
+        let bins = aggregator.finish();
+        assert_eq!(bins.get("k").unwrap()[..], [0.0; 10]);
+        Ok(())
+    }
 }

crates/ui/src/timeline/categorical.rs

@@ -0,0 +1,84 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+//! Requests and responses for categorical timelines: binned timelines of
+//! weighted values broken down by (FSM state, application-defined dimension),
+//! for one or more application-declared measures. Bin values are absolute
+//! time-weighted quantities, not normalized shares.
+//!
+//! All semantics are declared by the downstream analyzer: which FSM the states
+//! belong to, what the dimension keys mean, and which measures exist. The UI
+//! renders the declared names verbatim.
+
+use std::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+use ts_rs::TS;
+
+use crate::{quantity::CapacityKind, timeline::request::TimelineConfig};
+
+/// Request for a categorical timeline.
+#[derive(TS, Debug, Clone, Serialize, Deserialize)]
+pub struct CategoricalTimelineRequest<GlobalParams> {
+    /// Names of the measures to compute. Empty means all declared measures.
+    pub measures: Vec<String>,
+    /// The configuration of the window and number of bins.
+    pub config: TimelineConfig,
+    /// Global application-specific parameters, e.g. filters.
+    pub app_params: GlobalParams,
+}
+
+/// A measure declared by the downstream analyzer for a categorical timeline,
+/// e.g. an entity count or a number of bytes.
+#[derive(TS, Debug, Clone, Serialize)]
+pub struct MeasureDecl {
+    /// Unique name; key into [`CategoricalSeries::values`].
+    pub name: String,
+    /// Human-friendly display name.
+    pub display_name: String,
+    /// Key into the application's quantity specs map for unit formatting.
+    pub quantity: String,
+    /// Display semantics of a bin value (Occupancy = time-weighted level).
+    pub kind: CapacityKind,
+}
+
+/// One key of the application-defined dimension of a categorical timeline.
+#[derive(TS, Debug, Clone, Serialize)]
+pub struct DimensionKeyDecl {
+    /// The key used in [`CategoricalSeries::values`].
+    pub key: String,
+    /// Human-friendly display name.
+    pub display_name: String,
+}
+
+/// Presentation metadata for a categorical timeline, declared by the
+/// downstream analyzer.
+///
+/// Dimension keys are expected to be a small enumerable set; unbounded key
+/// cardinality is a downstream misuse.
+#[derive(TS, Debug, Clone, Serialize)]
+pub struct CategoricalDecl {
+    /// The FSM type whose states are broken down. References an entry in the
+    /// application's FSM type declarations (e.g. `QueryBundle` fsm_types) for
+    /// the state graph, names, and ordering.
+    pub entity_type_name: String,
+    /// Display name of the dimension, e.g. an application may use a dimension
+    /// describing where an entity's data resides.
+    pub dimension_name: String,
+    /// The dimension keys in stable stacking/legend order.
+    pub dimension_keys: Vec<DimensionKeyDecl>,
+    /// The measures present in this response.
+    pub measures: Vec<MeasureDecl>,
+    /// The measure the UI should select by default; must name an entry in
+    /// `measures`. `None` means the first declared measure.
+    pub default_measure: Option<String>,
+}
+
+/// Binned values of one categorical timeline series:
+/// measure name -> state name -> dimension key -> one value per time bin.
+///
+/// Absent inner entries mean all-zero bins.
+#[derive(TS, Debug, Clone, Default, Serialize)]
+pub struct CategoricalSeries {
+    pub values: HashMap<String, HashMap<String, HashMap<String, Vec<f64>>>>,
+}

crates/ui/src/timeline/mod.rs

@@ -2,5 +2,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
 //! Requests and responses for timelines.
+pub mod categorical;
 pub mod request;
 pub mod response;

domains/query_engine/analyzer/src/ui.rs

@@ -4,13 +4,14 @@
 use std::collections::HashMap;
 use std::path::Path;
 
-use quent_analyzer::AnalyzerResult;
+use quent_analyzer::{AnalyzerError, AnalyzerResult};
 use quent_events::Event;
 use quent_model::io::ImporterResult;
 use quent_query_engine_ui as ui;
 use quent_ui::{
     entities::{request::EntityListRequest, response::EntityListResponse},
     timeline::{
+        categorical::CategoricalTimelineRequest,
         request::{BulkChunkedTimelineRequest, BulkTimelineRequest, SingleTimelineRequest},
         response::{
             BulkChunkedTimelinesResponse, BulkTimelinesResponse, BulkTimelinesResponseEntry,
@@ -116,6 +117,21 @@ pub trait UiAnalyzer {
 
         Ok(BulkChunkedTimelinesResponse { entries })
     }
+
+    /// Return, for every operator of a query, a binned categorical timeline
+    /// over (entity state, analyzer-defined dimension), for one or more
+    /// analyzer-declared measures. Powers the UI's data-flow-over-time view of
+    /// the query plan.
+    ///
+    /// The default implementation returns [`AnalyzerError::Unsupported`]
+    /// (served as HTTP 501), so existing analyzers keep compiling and the UI
+    /// hides the view.
+    fn data_flow_timeline(
+        &self,
+        _request: CategoricalTimelineRequest<ui::QueryFilter>,
+    ) -> AnalyzerResult<ui::DataFlowTimelineBinned> {
+        Err(AnalyzerError::Unsupported)
+    }
 }
 
 /// Boxed owned stream of an analyzer's [`UiAnalyzer::Event`] from

domains/query_engine/server/src/error.rs

@@ -27,6 +27,12 @@ pub type ServerResult<T> = std::result::Result<T, ServerError>;
 impl From<ServerError> for StatusCode {
     fn from(value: ServerError) -> Self {
         match value {
+            // A capability the analyzer opted out of is not a server fault:
+            // clients probe for optional endpoints (e.g. the data-flow view)
+            // and hide the feature on 501.
+            ServerError::Analyzer(quent_analyzer::AnalyzerError::Unsupported) => {
+                StatusCode::NOT_IMPLEMENTED
+            }
             ServerError::Importer(_)
             | ServerError::Analyzer(_)
             | ServerError::Io(_)