fix(emit): B3-pop + B4-window bundle (B2-Sum deferred)

control_plane/src/emit/agent.rs

@@ -137,7 +137,15 @@ fn build_processor_block(cfg: &AgentCollectorConfig) -> Value {
 
     if cfg.mode == ProcessorMode::Window {
         if let Some(wd) = cfg.window_duration {
-            m.insert("window_duration".into(), Value::String(format_duration(wd)));
+            // MVP blocker B4: clamp `window_duration` to [5, 60] so
+            // the legacy agent emitter matches the typed L5 emitter's
+            // bounds — without this, a `[5m]` workload landing here
+            // mints a 300s sketch window whose closed answer never
+            // falls inside the user's replay range.
+            let clamped = super::stage_config::clamp_window_secs(Some(wd.as_secs()))
+                .map(std::time::Duration::from_secs)
+                .unwrap_or(wd);
+            m.insert("window_duration".into(), Value::String(format_duration(clamped)));
         }
     }
 
@@ -303,9 +311,50 @@ mod tests {
     #[test]
     fn contains_window_duration() {
         let yaml = generate_agent_collector_config(&ddsketch_cfg(), "ws://ctrl:4320/v1/opamp").unwrap();
+        // MVP blocker B4: the fixture's 5m window clamps to 60s
+        // (`MAX_WINDOW_SECS`). Assert on the clamped form — a window
+        // larger than 60s would put the sketch close outside any
+        // sensible replay range. Pre-B4 this test asserted "5m".
         assert!(
-            yaml.contains("5m"),
-            "YAML should contain window_duration\n{yaml}"
+            yaml.contains("window_duration: 1m") || yaml.contains("window_duration: 60s"),
+            "YAML should contain clamped window_duration (1m / 60s)\n{yaml}"
+        );
+    }
+
+    #[test]
+    fn clamps_oversize_window_to_max() {
+        let mut cfg = ddsketch_cfg();
+        cfg.window_duration = Some(Duration::from_secs(3600)); // 1h
+        let yaml = generate_agent_collector_config(&cfg, "ws://ctrl:4320/v1/opamp").unwrap();
+        assert!(
+            !yaml.contains("window_duration: 1h"),
+            "1h window must clamp to MAX_WINDOW_SECS, not pass through\n{yaml}"
+        );
+        assert!(
+            yaml.contains("window_duration: 1m") || yaml.contains("window_duration: 60s"),
+            "clamped window must be 60s\n{yaml}"
+        );
+    }
+
+    #[test]
+    fn clamps_undersize_window_to_min() {
+        let mut cfg = ddsketch_cfg();
+        cfg.window_duration = Some(Duration::from_secs(1)); // 1s
+        let yaml = generate_agent_collector_config(&cfg, "ws://ctrl:4320/v1/opamp").unwrap();
+        assert!(
+            yaml.contains("window_duration: 5s"),
+            "1s window must clamp UP to MIN_WINDOW_SECS=5s\n{yaml}"
+        );
+    }
+
+    #[test]
+    fn preserves_window_inside_clamp_range() {
+        let mut cfg = ddsketch_cfg();
+        cfg.window_duration = Some(Duration::from_secs(30));
+        let yaml = generate_agent_collector_config(&cfg, "ws://ctrl:4320/v1/opamp").unwrap();
+        assert!(
+            yaml.contains("window_duration: 30s"),
+            "30s window is inside [5, 60] and must pass through verbatim\n{yaml}"
         );
     }
 

control_plane/src/emit/asapquery_backend.rs

@@ -41,9 +41,18 @@ use crate::types::{AgentCollectorConfig, CollectionPlan, SketchType};
 /// serialization fails.
 pub fn generate_streaming_config_yaml(metric: &str, plan: &CollectionPlan) -> Result<String> {
     let agg = &plan.agent_config;
+    // MVP blocker B4: clamp the workload's `window_duration` to
+    // `[MIN_WINDOW_SECS, MAX_WINDOW_SECS]` so the legacy YAML emit
+    // matches the typed L5 JSON emit's `windowSize` clamp. Without
+    // this, the legacy and typed paths can disagree (e.g. typed
+    // clamps `[5m]` → 60, legacy passes 300 → backend reducer keys
+    // a 300s window the agent never closes).
     let window_secs = agg
         .window_duration
-        .map(|d: Duration| d.as_secs())
+        .map(|d: Duration| {
+            super::stage_config::clamp_window_secs(Some(d.as_secs()))
+                .expect("clamp preserves Some")
+        })
         .unwrap_or(0);
     if window_secs == 0 {
         anyhow::bail!(

control_plane/src/emit/mod.rs

@@ -497,13 +497,20 @@ mod runtime_tests {
         use crate::types_v2;
         let analyzer = Analyzer::new();
         for entry in registry.entries() {
+            // Mirrors main.rs's QuerySpec construction post-B3/B4:
+            // thread grouping_labels into group_by_labels; let the
+            // parser drive time_window when query_string is present.
             let spec = QuerySpec {
                 query_string: entry.query_string.clone(),
                 metric_name: entry.metric_name.clone(),
                 label_filters: Default::default(),
-                group_by_labels: vec![],
+                group_by_labels: entry.grouping_labels.clone(),
                 aggregations: vec!["quantile".into()],
-                time_window: "5m".into(),
+                time_window: if entry.query_string.is_some() {
+                    String::new()
+                } else {
+                    "5m".into()
+                },
                 repeat_every: None,
                 accuracy_sla: entry.accuracy_sla,
                 latency_sla: None,
@@ -602,4 +609,114 @@ mod runtime_tests {
             "routing table should have 5 entries (5 sketches; raw declines), got: {map:?}"
         );
     }
+
+    // ── B3 regression: WorkloadEntry.grouping_labels populates emit ───────
+    //
+    // Pre-B3 the WorkloadEntry YAML had no way to declare grouping
+    // labels — the analyzer pulled them only from PromQL `by (...)`
+    // clauses. Bare `quantile_over_time(0.99, metric[30s])` carries no
+    // `by`, so `QueryWorkload.group_by_labels` ended up empty, so
+    // `collect_metric_to_grouping_labels` returned `{metric: vec![]}`,
+    // so the 5-sketch routing emitter wrote
+    // `keep_keys(datapoint.attributes, [])` — stripping ALL attrs
+    // instead of keeping `["zone"]`. Sid catalog ended up with one sid
+    // per metric instead of one per (metric × zone).
+    //
+    // Post-B3 a declarative `grouping_labels: [zone]` on WorkloadEntry
+    // is threaded through the pre-pop QuerySpec → analyzer →
+    // QueryWorkload.group_by_labels → collect_metric_to_grouping_labels
+    // → the emitter's keep_keys list. Without this round-trip the
+    // smoke test's sid catalog stays empty-per-zone.
+    #[test]
+    fn workload_entry_grouping_labels_round_trip_through_emit_to_keep_keys() {
+        let yaml = r#"
+- metric_name: http_requests_total_latency_ms
+  query_string: "quantile_over_time(0.99, http_requests_total_latency_ms[30s])"
+  accuracy_sla: 0.01
+  assign_to_role: agent
+  grouping_labels: ["zone"]
+"#;
+        let entries: Vec<crate::workload::WorkloadEntry> =
+            serde_yaml::from_str(yaml).expect("parse workload yaml");
+        assert_eq!(entries.len(), 1);
+        assert_eq!(
+            entries[0].grouping_labels,
+            vec!["zone".to_string()],
+            "WorkloadEntry must surface grouping_labels from YAML"
+        );
+
+        let registry = crate::workload::WorkloadRegistry::from_entries(entries);
+        let store = WorkloadStore::new();
+        populate_store_from_registry(&registry, &store);
+
+        // The analyzer must have threaded grouping_labels into
+        // QueryWorkload.group_by_labels.
+        let map = collect_metric_to_grouping_labels(&registry, &store);
+        assert_eq!(
+            map.get("http_requests_total_latency_ms"),
+            Some(&vec!["zone".to_string()]),
+            "collect_metric_to_grouping_labels must surface entry.grouping_labels — \
+             without this the agent strips ALL attrs and the sid catalog ends up \
+             with one sid per metric instead of one per (metric, zone)\nmap: {map:?}"
+        );
+    }
+
+    /// Belt-and-braces companion: the emit-side keep_keys statement
+    /// must contain the per-entry grouping_labels VERBATIM. Catches a
+    /// regression where the pre-pop loop populates the workload store
+    /// but the round-trip through the emitter drops the labels.
+    #[test]
+    fn workload_entry_grouping_labels_surface_in_emit_keep_keys_list() {
+        use crate::physical::colored_dag::emitter::{EdgeStageConfig, ExportTarget};
+        use crate::physical::colored_dag::stage_id::StageId;
+        use crate::sketch_algebra::params::SketchKind;
+
+        let yaml = r#"
+- metric_name: http_requests_total_latency_ms
+  query_string: "quantile_over_time(0.99, http_requests_total_latency_ms[30s])"
+  accuracy_sla: 0.01
+  assign_to_role: agent
+  grouping_labels: ["zone"]
+"#;
+        let entries: Vec<crate::workload::WorkloadEntry> =
+            serde_yaml::from_str(yaml).expect("parse workload yaml");
+        let registry = crate::workload::WorkloadRegistry::from_entries(entries);
+        let store = WorkloadStore::new();
+        populate_store_from_registry(&registry, &store);
+
+        let mut edge_cfg = EdgeStageConfig {
+            source_metric: Some("http_requests_total_latency_ms".to_string()),
+            label_filters: Vec::new(),
+            window_secs: Some(30),
+            sketch_processors: Vec::new(),
+            exporter_target: ExportTarget::Stage(StageId::Gateway),
+            prometheus_archive_metrics: Vec::new(),
+            archive_tier_metrics: Vec::new(),
+            warm_passthrough_metrics: Vec::new(),
+            metric_to_family: std::collections::HashMap::from([(
+                "http_requests_total_latency_ms".to_string(),
+                SketchKind::DDSketch,
+            )]),
+            metric_to_grouping_labels: std::collections::HashMap::new(),
+        };
+        edge_cfg.metric_to_grouping_labels = collect_metric_to_grouping_labels(&registry, &store);
+
+        let yaml_out = crate::emit::emit_edge_yaml(&edge_cfg, "ws://c/", "test-agent")
+            .expect("emit ok");
+        assert!(
+            yaml_out.contains(
+                "keep_keys(datapoint.attributes, [\"zone\"]) where metric.name == \"http_requests_total_latency_ms\""
+            ),
+            "keep_keys must list `zone` (NOT empty) for the YAML-declared grouping_labels\n{yaml_out}",
+        );
+        // Belt-and-braces: the bug surface is specifically
+        // `keep_keys(..., [])`. Make sure we don't accidentally emit
+        // the empty-list form for this metric.
+        assert!(
+            !yaml_out.contains(
+                "keep_keys(datapoint.attributes, []) where metric.name == \"http_requests_total_latency_ms\""
+            ),
+            "empty keep_keys would strip all attrs and break per-zone sid splitting\n{yaml_out}",
+        );
+    }
 }