v0.15.9

src/engine/automations.rs

@@ -157,6 +157,7 @@ async fn run_automations(state: &AppState) -> Result<()> {
             Ok(true) => {
                 for device_id in target_devices { claimed_devices.insert(device_id); }
                 let fired_at = Utc::now();
+                flow_acknowledge_change_gates(&mut item);
                 flow_record_rate_limited_execution(&mut item, fired_at.clone());
                 item.last_fired_at = Some(fired_at);
                 item.updated_at = Utc::now();
@@ -405,10 +406,27 @@ fn flow_state_duration_update(
     (within_min && within_max, elapsed)
 }
 
-fn flow_change_gate_update(state: &mut crate::models::FlowRuntimeNodeState, current: Value) -> bool {
+fn flow_change_gate_update(
+    state: &mut crate::models::FlowRuntimeNodeState,
+    current: Value,
+    input: bool,
+) -> (bool, bool) {
     let changed = state.last_value.as_ref().is_some_and(|previous| previous != &current);
     state.last_value = Some(current);
-    changed
+    if !input {
+        state.pending = false;
+    } else if changed {
+        state.pending = true;
+    }
+    (changed, state.pending)
+}
+
+fn flow_acknowledge_change_gates(item: &mut Automation) {
+    for condition in item.flow_conditions.iter().filter(|condition| condition.kind == "on_change") {
+        if let Some(state) = item.flow_runtime.get_mut(&condition.id) {
+            state.pending = false;
+        }
+    }
 }
 
 fn flow_rate_limit_status(
@@ -444,9 +462,13 @@ fn flow_rolling_stat_update(
     statistic: &str,
     now: DateTime<Utc>,
 ) -> Option<f64> {
+    let started_at = state.since.get_or_insert_with(|| now.clone()).clone();
     let cutoff = now.clone() - chrono::Duration::seconds(window_seconds as i64);
     state.samples.retain(|item| item.at >= cutoff);
-    state.samples.push(crate::models::FlowRuntimeSample { at: now, value: sample });
+    state.samples.push(crate::models::FlowRuntimeSample { at: now.clone(), value: sample });
+    if now.signed_duration_since(started_at).num_seconds().max(0) < window_seconds as i64 {
+        return None;
+    }
     let mut values = state.samples.iter().map(|item| item.value).collect::<Vec<_>>();
     if values.is_empty() { return None; }
     if statistic == "median" {
@@ -730,13 +752,14 @@ pub async fn evaluate_flow_conditions_trace(
                 let mode = condition.config.get("mode").and_then(Value::as_str).unwrap_or("result");
                 let observed = if mode == "value" { actual_values.get(&input_id).cloned() } else { Some(json!(input)) };
                 let mut changed = false;
+                let mut pending = false;
                 let mut previous = None;
                 if let (Some(current), Some(map)) = (observed.clone(), runtime.as_deref_mut()) {
                     let state = map.entry(condition.id.clone()).or_default();
                     previous = state.last_value.clone();
-                    changed = flow_change_gate_update(state, current);
+                    (changed, pending) = flow_change_gate_update(state, current, input);
                 }
-                (input && changed, json!({"input": input, "mode": mode, "previous": previous, "current": observed, "changed": changed}))
+                (input && pending, json!({"input": input, "mode": mode, "previous": previous, "current": observed, "changed": changed, "pending": pending}))
             }
             "rate_limit" => {
                 let input = condition.inputs.len() == 1 && values.get(&condition.inputs[0]).copied().unwrap_or(false);
@@ -772,12 +795,17 @@ pub async fn evaluate_flow_conditions_trace(
                 let window = condition.config.get("window_seconds").and_then(Value::as_u64).unwrap_or(60);
                 let mut aggregate = None;
                 let mut count = 0usize;
-                if let (Some(sample), Some(map)) = (sample, runtime.as_deref_mut()) {
+                if let Some(map) = runtime.as_deref_mut() {
                     let node_state = map.entry(condition.id.clone()).or_default();
-                    aggregate = flow_rolling_stat_update(
-                        node_state, sample, window, condition.config.get("statistic").and_then(Value::as_str).unwrap_or("mean"), now.with_timezone(&Utc),
-                    );
-                    count = node_state.samples.len();
+                    if let Some(sample) = sample {
+                        aggregate = flow_rolling_stat_update(
+                            node_state, sample, window, condition.config.get("statistic").and_then(Value::as_str).unwrap_or("mean"), now.with_timezone(&Utc),
+                        );
+                        count = node_state.samples.len();
+                    } else {
+                        node_state.since = None;
+                        node_state.samples.clear();
+                    }
                 }
                 let expected = condition.config.get("value").and_then(Value::as_f64);
                 let matched = predecessors_match && aggregate.zip(expected).map(|(a,e)| flow_compare(a, condition.config.get("operator").and_then(Value::as_str).unwrap_or("lt"), e)).unwrap_or(false);