v0.13.1

src/engine/automations.rs

@@ -161,6 +161,9 @@ async fn run_automations(state: &AppState) -> Result<()> {
                 item.last_fired_at = Some(fired_at);
                 item.updated_at = Utc::now();
                 state.db.save_automation(&item)?;
+                // last_fired_at/next_ready_at are part of ControlPlan.rules. Runtime execution
+                // does not emit automation.updated, so invalidate the plan directly.
+                state.invalidate_control_plan();
                 state.log("info", "automation.fired", &format!("Automation {} fired", item.name), json!({
                     "automation_id": item.id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id, "group_id": item.action_group_id, "device_id": item.action_device_id
                 }));
@@ -181,6 +184,7 @@ async fn run_automations(state: &AppState) -> Result<()> {
                 item.last_fired_at = Some(Utc::now());
                 item.updated_at = Utc::now();
                 state.db.save_automation(&item)?;
+                state.invalidate_control_plan();
                 state.log("error", "automation.error", &err.to_string(), json!({"automation_id": item.id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id}));
             }
         }

src/engine/polling.rs

@@ -1,3 +1,15 @@
+fn device_runtime_change_affects_control_plan(before: &Device, after: &Device) -> bool {
+    // build_control_plan() consumes only these runtime device fields. Poll heartbeat data such
+    // as last_seen/response_time_ms remains available through device.updated but no longer
+    // forces an expensive control-plan rebuild.
+    before.name != after.name
+        || before.power != after.power
+        || before.mode != after.mode
+        || before.target_temperature != after.target_temperature
+        || before.online != after.online
+        || before.communication_failures != after.communication_failures
+}
+
 async fn lock_poll_zone_operations(state: &AppState, device_id: &str) -> Result<Vec<tokio::sync::OwnedMutexGuard<()>>, AppError> {
     let mut zone_ids: Vec<String> = state.db.list_zones()?.into_iter()
         .filter(|zone| zone.device_id == device_id)
@@ -35,7 +47,11 @@ async fn poll_one_locked(state: &AppState, device_id: &str) -> Result<Device, Ap
     }
     state.db.save_device(&device)?;
     record_reading(state, &device)?;
-    state.broadcast("device.updated", serde_json::to_value(&device).unwrap_or_default());
+    state.broadcast_with_control_plan_invalidation(
+        "device.updated",
+        serde_json::to_value(&device).unwrap_or_default(),
+        device_runtime_change_affects_control_plan(&before, &device),
+    );
     Ok(device)
 }
 
@@ -182,6 +198,9 @@ fn record_poll_failure(device: &mut Device, error: &str) {
 fn register_device_failure(state: &AppState, device: &mut Device, error: &str) -> Result<(), AppError> {
     record_poll_failure(device, error);
     state.db.save_device(device)?;
+    // Command failures can change online/communication health without a device.updated frame.
+    // Keep the materialized plan current explicitly instead of relying on log-event prefixes.
+    state.invalidate_control_plan();
     state.log("warn", "device.communication_error", &format!("{}: {error}", device.name), json!({
         "device_id": device.id,
         "consecutive_failures": device.communication_failures,

src/engine/tests.rs

@@ -233,6 +233,60 @@ mod tests {
         }
     }
 
+    #[test]
+    fn regulator_zone_event_dedup_ignores_only_updated_at_heartbeat() {
+        let zone = test_zone("device");
+        let mut heartbeat = zone.clone();
+        heartbeat.updated_at = zone.updated_at.clone() + chrono::Duration::seconds(5);
+        assert_eq!(
+            zone_event_semantic_value(&zone).unwrap(),
+            zone_event_semantic_value(&heartbeat).unwrap()
+        );
+
+        heartbeat.device_temperature = Some(24.5);
+        assert_ne!(
+            zone_event_semantic_value(&zone).unwrap(),
+            zone_event_semantic_value(&heartbeat).unwrap()
+        );
+    }
+
+    #[test]
+    fn regulator_zone_plan_invalidation_tracks_only_plan_inputs() {
+        let zone = test_zone("combined");
+        let mut diagnostics_only = zone.clone();
+        diagnostics_only.device_temperature = Some(24.0);
+        diagnostics_only.external_temperature = Some(23.0);
+        diagnostics_only.last_action_at = Some(Utc::now());
+        assert!(!zone_runtime_change_affects_control_plan(&zone, &diagnostics_only));
+
+        let mut temperature_changed = zone.clone();
+        temperature_changed.current_temperature = Some(23.5);
+        assert!(zone_runtime_change_affects_control_plan(&zone, &temperature_changed));
+
+        let mut demand_changed = zone.clone();
+        demand_changed.demand = !zone.demand;
+        assert!(zone_runtime_change_affects_control_plan(&zone, &demand_changed));
+    }
+
+    #[test]
+    fn device_poll_heartbeat_does_not_invalidate_control_plan() {
+        let device = Device::simulated_default();
+        let mut heartbeat = device.clone();
+        heartbeat.last_seen = Some(Utc::now() + chrono::Duration::seconds(15));
+        heartbeat.updated_at = Utc::now() + chrono::Duration::seconds(15);
+        heartbeat.response_time_ms = Some(42);
+        heartbeat.current_temperature = Some(25.1);
+        assert!(!device_runtime_change_affects_control_plan(&device, &heartbeat));
+
+        let mut power_changed = device.clone();
+        power_changed.power = !device.power;
+        assert!(device_runtime_change_affects_control_plan(&device, &power_changed));
+
+        let mut health_changed = device.clone();
+        health_changed.communication_failures = 1;
+        assert!(device_runtime_change_affects_control_plan(&device, &health_changed));
+    }
+
     #[test]
     fn zone_can_use_separate_heating_and_cooling_hysteresis() {
         let mut zone = test_zone("device");

src/engine/zone_actions.rs

@@ -1,14 +1,107 @@
+#[derive(Debug)]
+struct PersistedZoneCycle {
+    previous: Option<Zone>,
+    zone: Zone,
+}
+
+fn zone_event_semantic_value(zone: &Zone) -> Result<Value, AppError> {
+    let mut value = serde_json::to_value(zone)?;
+    if let Some(object) = value.as_object_mut() {
+        // updated_at is a persistence/concurrency and observation timestamp. It advances on
+        // every regulator pass even when no user-visible zone state changed, so it must not
+        // by itself create a full zone.updated WebSocket frame.
+        object.remove("updated_at");
+    }
+    Ok(value)
+}
+
+fn zone_runtime_change_affects_control_plan(before: &Zone, after: &Zone) -> bool {
+    // Keep this projection aligned with build_control_plan(). Fields omitted here are runtime
+    // diagnostics/history only and do not alter the materialized plan. Configuration/API
+    // mutations still use the normal broadcast() path, which conservatively invalidates it.
+    let plan_input = |zone: &Zone| {
+        json!({
+            "id": zone.id,
+            "name": zone.name,
+            "device_id": zone.device_id,
+            "enabled": zone.enabled,
+            "mode": zone.mode,
+            "inherit_house_mode": zone.inherit_house_mode,
+            "setpoint": zone.setpoint,
+            "profile_version": zone.profile_version,
+            "cool_comfort_setpoint": zone.cool_comfort_setpoint,
+            "cool_sleep_setpoint": zone.cool_sleep_setpoint,
+            "cool_away_setpoint": zone.cool_away_setpoint,
+            "heat_comfort_setpoint": zone.heat_comfort_setpoint,
+            "heat_sleep_setpoint": zone.heat_sleep_setpoint,
+            "heat_away_setpoint": zone.heat_away_setpoint,
+            "current_temperature": zone.current_temperature,
+            "control_temperature_source": zone.control_temperature_source,
+            "manual_preset": zone.manual_preset,
+            "manual_setpoint": zone.manual_setpoint,
+            "manual_override_until": zone.manual_override_until,
+            "local_thermostat_power": zone.local_thermostat_power,
+            "local_thermostat_resume_at": zone.local_thermostat_resume_at,
+            "temporary_quick_thermostat": zone.temporary_quick_thermostat,
+            "device_manual_override": zone.device_manual_override,
+            "device_manual_override_until": zone.device_manual_override_until,
+            "control_owner": zone.control_owner,
+            "control_source": zone.control_source,
+            "control_since": zone.control_since,
+            "control_resume_at": zone.control_resume_at,
+            "control_reason": zone.control_reason,
+            "lockout_until": zone.lockout_until,
+            "lockout_reason": zone.lockout_reason,
+            "effective_setpoint": zone.effective_setpoint,
+            "demand": zone.demand,
+        })
+    };
+    plan_input(before) != plan_input(after)
+}
+
+fn publish_persisted_zone_cycle(
+    state: &AppState,
+    persisted: PersistedZoneCycle,
+) -> Result<Zone, AppError> {
+    let event_changed = match persisted.previous.as_ref() {
+        Some(previous) => {
+            zone_event_semantic_value(previous)? != zone_event_semantic_value(&persisted.zone)?
+        }
+        None => true,
+    };
+    if event_changed {
+        let invalidates_control_plan = persisted
+            .previous
+            .as_ref()
+            .map(|previous| zone_runtime_change_affects_control_plan(previous, &persisted.zone))
+            .unwrap_or(true);
+        state.broadcast_with_control_plan_invalidation(
+            "zone.updated",
+            serde_json::to_value(&persisted.zone)?,
+            invalidates_control_plan,
+        );
+    }
+    Ok(persisted.zone)
+}
+
 fn persist_zone_cycle(
     state: &AppState,
     computed: &Zone,
     cycle_started_at: DateTime<Utc>,
-) -> Result<Zone, AppError> {
+) -> Result<PersistedZoneCycle, AppError> {
     let Some(mut latest) = state.db.get_zone(&computed.id)? else {
-        return Ok(computed.clone());
+        return Ok(PersistedZoneCycle {
+            previous: None,
+            zone: computed.clone(),
+        });
     };
+    let previous = latest.clone();
     if latest.updated_at <= cycle_started_at {
         state.db.save_zone(computed)?;
-        return Ok(computed.clone());
+        return Ok(PersistedZoneCycle {
+            previous: Some(previous),
+            zone: computed.clone(),
+        });
     }
     // Another actor changed this zone while the regulator was doing network I/O. Never
     // write the old controller snapshot over fresh configuration or manual takeover state.
@@ -21,7 +114,10 @@ fn persist_zone_cycle(
         latest.updated_at = Utc::now();
         state.db.save_zone(&latest)?;
     }
-    Ok(latest)
+    Ok(PersistedZoneCycle {
+        previous: Some(previous),
+        zone: latest,
+    })
 }
 
 async fn thermostat_ownership_is_current(

src/engine/zone_control.rs

@@ -200,8 +200,7 @@ fn persist_zone_cycle_with_history(
 ) -> Result<Zone> {
     record_zone_history(state, zone, outdoor_temperature, poll_interval_seconds);
     let persisted = persist_zone_cycle(state, zone, cycle_started_at)?;
-    state.broadcast("zone.updated", serde_json::to_value(&persisted)?);
-    Ok(persisted)
+    Ok(publish_persisted_zone_cycle(state, persisted)?)
 }
 
 async fn handle_zone_pre_control_state(
@@ -607,8 +606,10 @@ async fn control_zones(state: &AppState) -> Result<()> {
                 outdoor_temperature,
                 settings.poll_interval_seconds,
             );
-            let persisted_zone = persist_zone_cycle(state, &zone, cycle_started_at)?;
-            state.broadcast("zone.updated", serde_json::to_value(&persisted_zone)?);
+            publish_persisted_zone_cycle(
+                state,
+                persist_zone_cycle(state, &zone, cycle_started_at)?,
+            )?;
             continue;
         }
 
@@ -686,8 +687,10 @@ async fn control_zones(state: &AppState) -> Result<()> {
                 outdoor_temperature,
                 settings.poll_interval_seconds,
             );
-            let persisted_zone = persist_zone_cycle(state, &zone, cycle_started_at)?;
-            state.broadcast("zone.updated", serde_json::to_value(&persisted_zone)?);
+            publish_persisted_zone_cycle(
+                state,
+                persist_zone_cycle(state, &zone, cycle_started_at)?,
+            )?;
             continue;
         }
 
@@ -702,8 +705,10 @@ async fn control_zones(state: &AppState) -> Result<()> {
                 outdoor_temperature,
                 settings.poll_interval_seconds,
             );
-            let persisted_zone = persist_zone_cycle(state, &zone, cycle_started_at)?;
-            state.broadcast("zone.updated", serde_json::to_value(&persisted_zone)?);
+            publish_persisted_zone_cycle(
+                state,
+                persist_zone_cycle(state, &zone, cycle_started_at)?,
+            )?;
             continue;
         };
 
@@ -866,8 +871,10 @@ async fn control_zones(state: &AppState) -> Result<()> {
                     outdoor_temperature,
                     settings.poll_interval_seconds,
                 );
-                let persisted_zone = persist_zone_cycle(state, &zone, cycle_started_at)?;
-                state.broadcast("zone.updated", serde_json::to_value(&persisted_zone)?);
+                publish_persisted_zone_cycle(
+                    state,
+                    persist_zone_cycle(state, &zone, cycle_started_at)?,
+                )?;
                 continue;
             }
             if let Some(last_change) = zone.last_power_change_at {
@@ -884,8 +891,10 @@ async fn control_zones(state: &AppState) -> Result<()> {
                         outdoor_temperature,
                         settings.poll_interval_seconds,
                     );
-                    let persisted_zone = persist_zone_cycle(state, &zone, cycle_started_at)?;
-                    state.broadcast("zone.updated", serde_json::to_value(&persisted_zone)?);
+                    publish_persisted_zone_cycle(
+                        state,
+                        persist_zone_cycle(state, &zone, cycle_started_at)?,
+                    )?;
                     continue;
                 }
             }
@@ -924,8 +933,10 @@ async fn control_zones(state: &AppState) -> Result<()> {
                 outdoor_temperature,
                 settings.poll_interval_seconds,
             );
-            let persisted_zone = persist_zone_cycle(state, &zone, cycle_started_at)?;
-            state.broadcast("zone.updated", serde_json::to_value(&persisted_zone)?);
+            publish_persisted_zone_cycle(
+                state,
+                persist_zone_cycle(state, &zone, cycle_started_at)?,
+            )?;
             continue;
         }
         if !device.power {
@@ -944,8 +955,10 @@ async fn control_zones(state: &AppState) -> Result<()> {
                     outdoor_temperature,
                     settings.poll_interval_seconds,
                 );
-                let persisted_zone = persist_zone_cycle(state, &zone, cycle_started_at)?;
-                state.broadcast("zone.updated", serde_json::to_value(&persisted_zone)?);
+                publish_persisted_zone_cycle(
+                    state,
+                    persist_zone_cycle(state, &zone, cycle_started_at)?,
+                )?;
                 continue;
             }
             if settings.compressor_protection_enabled {
@@ -963,8 +976,10 @@ async fn control_zones(state: &AppState) -> Result<()> {
                             outdoor_temperature,
                             settings.poll_interval_seconds,
                         );
-                        let persisted_zone = persist_zone_cycle(state, &zone, cycle_started_at)?;
-                        state.broadcast("zone.updated", serde_json::to_value(&persisted_zone)?);
+                        publish_persisted_zone_cycle(
+                            state,
+                            persist_zone_cycle(state, &zone, cycle_started_at)?,
+                        )?;
                         continue;
                     }
                 }
@@ -1052,8 +1067,10 @@ async fn control_zones(state: &AppState) -> Result<()> {
                         outdoor_temperature,
                         settings.poll_interval_seconds,
                     );
-                    let persisted_zone = persist_zone_cycle(state, &zone, cycle_started_at)?;
-                    state.broadcast("zone.updated", serde_json::to_value(&persisted_zone)?);
+                    publish_persisted_zone_cycle(
+                        state,
+                        persist_zone_cycle(state, &zone, cycle_started_at)?,
+                    )?;
                     continue;
                 }
                 Err(err) => state.log(
@@ -1071,8 +1088,10 @@ async fn control_zones(state: &AppState) -> Result<()> {
             outdoor_temperature,
             settings.poll_interval_seconds,
         );
-        let persisted_zone = persist_zone_cycle(state, &zone, cycle_started_at)?;
-        state.broadcast("zone.updated", serde_json::to_value(&persisted_zone)?);
+        publish_persisted_zone_cycle(
+            state,
+            persist_zone_cycle(state, &zone, cycle_started_at)?,
+        )?;
     }
 
     Ok(())

src/state.rs

@@ -142,11 +142,14 @@ impl AppState {
         self.control_plan_wakeup.notify_one();
     }
 
-    pub fn broadcast(&self, event: impl Into<String>, data: Value) {
-        let event = event.into();
-        let invalidates_control_plan = control_plan_event_affects_plan(&event);
+    pub fn broadcast_with_control_plan_invalidation(
+        &self,
+        event: impl Into<String>,
+        data: Value,
+        invalidates_control_plan: bool,
+    ) {
         let _ = self.events.send(ApiEvent {
-            event,
+            event: event.into(),
             timestamp: Utc::now(),
             data,
         });
@@ -155,10 +158,13 @@ impl AppState {
         }
     }
 
+    pub fn broadcast(&self, event: impl Into<String>, data: Value) {
+        let event = event.into();
+        let invalidates_control_plan = control_plan_event_affects_plan(&event);
+        self.broadcast_with_control_plan_invalidation(event, data, invalidates_control_plan);
+    }
+
     pub fn log(&self, level: &str, kind: &str, message: &str, metadata: Value) {
-        if control_plan_event_affects_plan(kind) {
-            self.invalidate_control_plan();
-        }
         if let Err(err) = self.db.log_event(level, kind, message, &metadata) {
             tracing::warn!(error=?err, "cannot persist event log");
         }