feat(precompute): separate idle and deadline closure

data_plane/src/main.rs

@@ -513,7 +513,10 @@ 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,
+            wall_clock_idle_grace_period_ms: 5_000,
+            // Enabled only after deadline-triggered late corrections are
+            // guaranteed to append rather than drop.
+            wall_clock_max_open_grace_period_ms: 0,
             schema_persist_path: args.schema_persist_path.clone(),
         };
         // M2.3.6 — sketch-only sink. Precompute writes now go to

data_plane/src/precompute_engine/config.rs

@@ -30,18 +30,20 @@ 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,
+    /// Additional idle grace after one window duration. A pane closes when it
+    /// has received no input for `window_size + idle_grace`. The legacy YAML
+    /// name is accepted as a serde alias. Non-positive disables idle closure.
+    #[serde(
+        default = "default_wall_clock_idle_grace_period_ms",
+        alias = "wall_clock_grace_period_ms"
+    )]
+    pub wall_clock_idle_grace_period_ms: i64,
+    /// Additional grace for the absolute wall-clock deadline. A pane closes
+    /// after `window_size + max_open_grace` from its first input even if it is
+    /// still active. Non-positive disables the deadline. It stays disabled by
+    /// default until late corrections are guaranteed not to be dropped.
+    #[serde(default)]
+    pub wall_clock_max_open_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`
@@ -64,13 +66,14 @@ 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(),
+            wall_clock_idle_grace_period_ms: default_wall_clock_idle_grace_period_ms(),
+            wall_clock_max_open_grace_period_ms: 0,
             schema_persist_path: None,
         }
     }
 }
 
-fn default_wall_clock_grace_period_ms() -> i64 {
+fn default_wall_clock_idle_grace_period_ms() -> i64 {
     5_000
 }
 
@@ -89,6 +92,27 @@ 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);
+        assert_eq!(config.wall_clock_idle_grace_period_ms, 5_000);
+        assert_eq!(config.wall_clock_max_open_grace_period_ms, 0);
+    }
+
+    #[test]
+    fn legacy_wall_clock_grace_deserializes_as_idle_grace() {
+        let config: PrecomputeEngineConfig = serde_yaml::from_str(
+            r#"
+num_workers: 1
+allowed_lateness_ms: 100
+max_buffer_per_series: 10
+flush_interval_ms: 1000
+channel_buffer_size: 10
+pass_raw_samples: false
+raw_mode_aggregation_id: 0
+late_data_policy: Drop
+wall_clock_grace_period_ms: 7000
+"#,
+        )
+        .expect("legacy config should deserialize");
+        assert_eq!(config.wall_clock_idle_grace_period_ms, 7_000);
+        assert_eq!(config.wall_clock_max_open_grace_period_ms, 0);
     }
 }

data_plane/src/precompute_engine/engine.rs

@@ -134,7 +134,10 @@ 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,
+                    wall_clock_idle_grace_period_ms: self.config.wall_clock_idle_grace_period_ms,
+                    wall_clock_max_open_grace_period_ms: self
+                        .config
+                        .wall_clock_max_open_grace_period_ms,
                 },
                 self.diagnostics.worker_group_counts[id].clone(),
                 self.diagnostics.worker_watermarks[id].clone(),

data_plane/src/precompute_engine/precompute_engine_design_doc.md

@@ -147,6 +147,8 @@ pub struct PrecomputeEngineConfig {
     pub pass_raw_samples: bool,          // default: false
     pub raw_mode_aggregation_id: u64,    // default: 0
     pub late_data_policy: LateDataPolicy, // default: Drop
+    pub wall_clock_idle_grace_period_ms: i64, // default: 5,000
+    pub wall_clock_max_open_grace_period_ms: i64, // default: 0 (disabled)
 }
 
 pub enum LateDataPolicy {
@@ -155,6 +157,12 @@ pub enum LateDataPolicy {
 }
 ```
 
+For a window of duration `W`, idle closure fires after `W + idle_grace`
+without a touch. When enabled, the absolute deadline fires after
+`W + max_open_grace` from the first touch even if input remains active. These
+wall-clock decisions advance only the closure watermark; they never modify the
+maximum observed event timestamp.
+
 ### 3.3 SeriesRouter (`series_router.rs`)
 
 Deterministic hash-based routing using XXHash64: