refactor(capability_matching): move serving-time routing logic to data_plane

control_plane/src/emit/monitor.rs

@@ -215,7 +215,7 @@ mod tests {
     fn streaming_entry_deserializes_as_monitor_spec() {
         // The emitted JSON must round-trip into the backend's MonitorSpec.
         let entry = streaming_config_monitor_entry(&sum_intent());
-        let spec: asap_types::streaming_config::MonitorSpec =
+        let spec: asap_types::MonitorSpec =
             serde_json::from_value(entry).expect("MonitorSpec deserialize");
         assert_eq!(spec.agg_id, agg_id_for_metric("bytes_sent"));
         assert_eq!(spec.tau, 100.0);

crates/asap_types/src/lib.rs

@@ -1,24 +1,19 @@
 pub mod aggregation_config;
 pub mod aggregation_type;
-pub mod capability_matching;
 pub mod enums;
 pub mod key_by_label_names;
+pub mod monitor_spec;
 pub mod policy_fingerprint;
 pub mod policy_registry;
 pub mod query_requirements;
-pub mod streaming_config;
 pub mod traits;
 pub mod utils;
 
 pub use aggregation_config::*;
 pub use aggregation_type::AggregationType;
-pub use capability_matching::{
-    compatible_storage_backends, parse_storage_backend_engine_id, AccuracyTarget, StorageBackend,
-    CANONICAL_QUERY_ENGINE_IDS, ENGINE_ID_ASAP_QUERY, ENGINE_ID_THANOS_QUERY,
-};
 pub use enums::*;
 pub use key_by_label_names::KeyByLabelNames;
+pub use monitor_spec::MonitorSpec;
 pub use policy_fingerprint::PolicyFingerprint;
 pub use policy_registry::PolicyRegistry;
 pub use query_requirements::*;
-pub use streaming_config::*;

crates/asap_types/src/monitor_spec.rs

@@ -0,0 +1,53 @@
+use serde::{Deserialize, Serialize};
+
+/// One continuous-monitoring (CDM) threshold spec. The data-plane monitor
+/// coordinator owns the AUTHORITATIVE `tau`/`epsilon`/`window_ms` (the edge
+/// copy is advisory), keyed by the same content-addressed `agg_id` the edge and
+/// coordinator share. `key` is the CMS point-frequency key for point monitors
+/// (empty for Sum / whole-stream). See
+/// `ASAPCollector/docs/continuous-monitoring-tumbling-cost-analysis.md`.
+///
+/// Stays here (unlike `data_plane::storage_engines::types::StreamingConfig`,
+/// which holds a `Vec<MonitorSpec>` field) because `control_plane` genuinely
+/// needs it: `emit/monitor.rs` builds the `StreamingConfig.monitors[]` JSON
+/// entry by hand and has a regression test asserting that JSON deserializes
+/// into this exact type. `control_plane` cannot depend on `data_plane` (the
+/// dependency runs the other way), so this type has to live somewhere both
+/// sides can reach — same reasoning as `AggregationConfig`/`PolicyFingerprint`.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MonitorSpec {
+    pub agg_id: u64,
+    /// Additive readout the edge reports: "sum" (default), "cms_point", "f2".
+    /// Pass-through metadata so the edge can auto-learn its reporting mode from
+    /// the pushed config; the coordinator allocation is value-driven and does not
+    /// branch on it (p_i ∝ √(value/rate) is the F2 allocation when value=‖f‖²).
+    #[serde(default)]
+    pub functional: String,
+    /// CMS point-frequency key x; empty (default) for Sum / whole-stream / F2.
+    #[serde(default)]
+    pub key: String,
+    /// Threshold τ (authoritative here, not at the edge).
+    pub tau: f64,
+    /// Relative tolerance ε; the alert fires when the estimate reaches (1−ε)τ.
+    #[serde(default = "default_monitor_epsilon")]
+    pub epsilon: f64,
+    /// Tumbling epoch length in ms; MUST match the edge window for this agg_id.
+    pub window_ms: u64,
+    /// Count-Sketch depth (rows) for whole-sketch `functional="f2"` monitors.
+    /// 0 (default) for scalar monitors; MUST match the edge's Count-Sketch for
+    /// this agg when F2 (both sides square/merge the same cell matrix).
+    #[serde(default)]
+    pub d: usize,
+    /// Count-Sketch width (buckets/row) for F2 monitors; 0 for scalar.
+    #[serde(default)]
+    pub w: usize,
+    /// F2 monitoring variant: "distributed" (default, ship every window) or
+    /// "geometric" (Sharfman–Schuster–Keren safe-zone, ship on local violation).
+    /// Ignored by scalar monitors.
+    #[serde(default)]
+    pub mode: String,
+}
+
+fn default_monitor_epsilon() -> f64 {
+    0.05
+}

crates/asap_types/src/policy_registry.rs

@@ -1,10 +1,11 @@
 //! Content-addressed policy registry.
 //!
-//! Derived view over a `StreamingConfig` that maps
+//! Derived view over a collection of `AggregationConfig`s that maps
 //! [`PolicyFingerprint`] → [`AggregationConfig`]. This is the
 //! merged-sid-identity-chain replacement for the controller-allocated
-//! `aggregation_id`-keyed `HashMap` that today's `StreamingConfig`
-//! carries.
+//! `aggregation_id`-keyed `HashMap` that `data_plane`'s `StreamingConfig`
+//! carries (see `data_plane::storage_engines::types::streaming_config`'s
+//! module doc for why that type lives there, not here).
 //!
 //! ## Dual-keyed transition
 //!
@@ -25,13 +26,12 @@
 //! map produce the same fingerprint, the later one wins (last-write
 //! semantics). In practice the source should never contain duplicates;
 //! if it does, that's a control-plane bug worth surfacing in telemetry
-//! (see `PolicyRegistry::from_streaming_config_with_collisions`).
+//! (see [`PolicyRegistry::from_configs_with_collisions`]).
 
 use std::collections::HashMap;
 
 use crate::aggregation_config::AggregationConfig;
 use crate::policy_fingerprint::PolicyFingerprint;
-use crate::streaming_config::StreamingConfig;
 
 /// Content-addressed lookup table for active aggregation policies.
 #[derive(Debug, Clone, Default)]
@@ -75,17 +75,6 @@ impl PolicyRegistry {
         (Self { policies }, collisions)
     }
 
-    /// Build from a `StreamingConfig`. Sugar over `from_configs` —
-    /// keeps callers from needing to walk the legacy map themselves.
-    pub fn from_streaming_config(cfg: &StreamingConfig) -> Self {
-        Self::from_configs(cfg.aggregation_configs.values().cloned())
-    }
-
-    /// `from_streaming_config` + collision count.
-    pub fn from_streaming_config_with_collisions(cfg: &StreamingConfig) -> (Self, usize) {
-        Self::from_configs_with_collisions(cfg.aggregation_configs.values().cloned())
-    }
-
     /// Look up the config for a fingerprint.
     pub fn get(&self, fp: PolicyFingerprint) -> Option<&AggregationConfig> {
         self.policies.get(&fp)
@@ -176,22 +165,4 @@ mod tests {
         assert_eq!(reg.len(), 2);
         assert_eq!(collisions, 0);
     }
-
-    #[test]
-    fn from_streaming_config_walks_the_map() {
-        let mut map = StdHashMap::new();
-        map.insert(1, cfg(1, "http_lat"));
-        map.insert(2, cfg(2, "cpu_pct"));
-        let sc = StreamingConfig::new(map);
-        let reg = PolicyRegistry::from_streaming_config(&sc);
-        assert_eq!(reg.len(), 2);
-    }
-
-    #[test]
-    fn empty_streaming_config_yields_empty_registry() {
-        let sc = StreamingConfig::new(StdHashMap::new());
-        let reg = PolicyRegistry::from_streaming_config(&sc);
-        assert!(reg.is_empty());
-        assert_eq!(reg.len(), 0);
-    }
 }

data_plane/src/drivers/query/servers/http.rs

@@ -17,11 +17,11 @@ use tracing::{debug, info, warn};
 use crate::drivers::query::adapters::{create_http_adapter, AdapterConfig, HttpProtocolAdapter};
 use crate::drivers::query::servers::metrics as srv_metrics;
 use crate::query_engines::routing::{
-    EngineRouter, EngineRouterError, FreshnessProbeCache, QueryEngine,
+    AccuracyTarget, EngineRouter, EngineRouterError, FreshnessProbeCache, QueryEngine,
 };
 use crate::query_engines::ASAPQueryEngine;
+use crate::storage_engines::types::StorageBackend;
 use asap_types::Statistic;
-use asap_types::{AccuracyTarget, StorageBackend};
 
 // ─── Control-plane-pushed precompute job registry ────────────────────────────
 //
@@ -5330,7 +5330,7 @@ async fn handle_post_streaming_config(
         }
     };
     let new_config =
-        match asap_types::streaming_config::StreamingConfig::from_yaml_data(&yaml_value) {
+        match crate::storage_engines::types::StreamingConfig::from_yaml_data(&yaml_value) {
             Ok(c) => c,
             Err(e) => {
                 let body = serde_json::json!({