fix(precompute-engine): wall-clock watermark fallback closes idle windows

asap-query-engine/src/bin/bench_precompute_sketch.rs

@@ -174,6 +174,7 @@ async fn start_engine(
         pass_raw_samples: false,
         raw_mode_aggregation_id: 0,
         late_data_policy: LateDataPolicy::Drop,
+        wall_clock_grace_period_ms: 5_000,
         schema_persist_path: None,
     };
     let engine = PrecomputeEngine::new(

asap-query-engine/src/bin/e2e_quickstart_resource_test.rs

@@ -233,6 +233,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         pass_raw_samples: false,
         raw_mode_aggregation_id: 0,
         late_data_policy: LateDataPolicy::Drop,
+        wall_clock_grace_period_ms: 5_000,
         schema_persist_path: None,
     };
     let output_sink = Arc::new(StoreOutputSink::new(store.clone()));

asap-query-engine/src/bin/precompute_engine.rs

@@ -295,6 +295,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         pass_raw_samples: args.pass_raw_samples,
         raw_mode_aggregation_id: args.raw_mode_aggregation_id,
         late_data_policy: args.late_data_policy,
+        wall_clock_grace_period_ms: 5_000,
         schema_persist_path: None,
     };
 

asap-query-engine/src/bin/test_e2e_precompute.rs

@@ -150,6 +150,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         pass_raw_samples: false,
         raw_mode_aggregation_id: 0,
         late_data_policy: LateDataPolicy::Drop,
+        wall_clock_grace_period_ms: 5_000,
         schema_persist_path: None,
     };
     let output_sink = Arc::new(StoreOutputSink::new(store.clone()));
@@ -295,6 +296,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         pass_raw_samples: true,
         raw_mode_aggregation_id: raw_agg_id,
         late_data_policy: LateDataPolicy::Drop,
+        wall_clock_grace_period_ms: 5_000,
         schema_persist_path: None,
     };
     let raw_sink = Arc::new(RawPassthroughSink::new(store.clone()));
@@ -648,6 +650,7 @@ async fn run_single_bench(
         pass_raw_samples: false,
         raw_mode_aggregation_id: 0,
         late_data_policy: LateDataPolicy::Drop,
+        wall_clock_grace_period_ms: 5_000,
         schema_persist_path: None,
     };
     let engine = PrecomputeEngine::new(

asap-query-engine/src/main.rs

@@ -523,6 +523,7 @@ async fn main() -> Result<()> {
             pass_raw_samples: false,
             raw_mode_aggregation_id: 0,
             late_data_policy: LateDataPolicy::Drop,
+            wall_clock_grace_period_ms: 5_000,
             schema_persist_path: args.schema_persist_path.clone(),
         };
         let output_sink = Arc::new(StoreOutputSink::new(store.clone()));

asap-query-engine/src/precompute_engine/config.rs

@@ -32,6 +32,18 @@ pub struct PrecomputeEngineConfig {
     pub raw_mode_aggregation_id: u64,
     /// Policy for handling late samples that arrive after their window has closed.
     pub late_data_policy: LateDataPolicy,
+    /// Wall-clock grace period (milliseconds) for the watermark fallback in
+    /// `flush_all`. When event-time stagnates (e.g. agents stamp every
+    /// sketch with the same `time_unix_nano`), `flush_all`'s `+1ms`
+    /// watermark advance is a no-op and idle windows never close. The
+    /// wall-clock fallback closes a pane whose creation has been older
+    /// than `window_size_ms + wall_clock_grace_period_ms` of *wall-clock*
+    /// time, regardless of where event-time is. The grace period
+    /// tolerates late-arriving events that would otherwise be evicted as
+    /// "the window already closed". Default: 5000 ms (matches
+    /// `allowed_lateness_ms` default).
+    #[serde(default = "default_wall_clock_grace_period_ms")]
+    pub wall_clock_grace_period_ms: i64,
     /// Optional path where the `SchemaRegistry` persists per-`agg_id`
     /// lifecycle state across restarts (sketch DB Phase 2c). When
     /// set, the registry loads prior `created_at_ms` / `retired_at_ms`
@@ -55,11 +67,16 @@ impl Default for PrecomputeEngineConfig {
             pass_raw_samples: false,
             raw_mode_aggregation_id: 0,
             late_data_policy: LateDataPolicy::Drop,
+            wall_clock_grace_period_ms: default_wall_clock_grace_period_ms(),
             schema_persist_path: None,
         }
     }
 }
 
+fn default_wall_clock_grace_period_ms() -> i64 {
+    5_000
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -76,5 +93,6 @@ mod tests {
         assert!(!config.pass_raw_samples);
         assert_eq!(config.raw_mode_aggregation_id, 0);
         assert_eq!(config.late_data_policy, LateDataPolicy::Drop);
+        assert_eq!(config.wall_clock_grace_period_ms, 5_000);
     }
 }

asap-query-engine/src/precompute_engine/engine.rs

@@ -142,6 +142,7 @@ impl PrecomputeEngine {
                     pass_raw_samples: self.config.pass_raw_samples,
                     raw_mode_aggregation_id: self.config.raw_mode_aggregation_id,
                     late_data_policy: self.config.late_data_policy,
+                    wall_clock_grace_period_ms: self.config.wall_clock_grace_period_ms,
                 },
                 self.diagnostics.worker_group_counts[id].clone(),
                 self.diagnostics.worker_watermarks[id].clone(),