v0.13.0

src/api.rs

@@ -204,6 +204,10 @@ pub fn router(state: AppState) -> Router {
         .route_layer(middleware::from_fn_with_state(state.clone(), auth));
 
     let home_assistant_api = Router::new()
+        .route(
+            "/api/integrations/home-assistant/snapshot",
+            get(home_assistant_snapshot),
+        )
         .route(
             "/api/integrations/home-assistant/devices",
             get(list_devices),

src/api/groups.rs

@@ -278,7 +278,7 @@ async fn list_home_assistant_groups(
     let groups = state.db.list_groups()?;
     let zones = state.db.list_zones()?;
     let devices = state.db.list_devices()?;
-    let plan = engine::build_control_plan(&state).await?;
+    let plan = engine::get_control_plan_snapshot(&state).await?;
     let settings = state.settings.read().await.clone();
     let mut output = Vec::with_capacity(groups.len());
 
@@ -288,6 +288,7 @@ async fn list_home_assistant_groups(
             .filter(|zone| group.zone_ids.iter().any(|zone_id| zone_id == &zone.id))
             .collect::<Vec<_>>();
         let planned_members = plan
+            .plan
             .zones
             .iter()
             .filter(|zone| {

src/api/history.rs

@@ -485,7 +485,6 @@ async fn history(
 }
 
 async fn control_plan(State(state): State<AppState>) -> Result<Json<Value>, AppError> {
-    Ok(Json(serde_json::to_value(
-        engine::build_control_plan(&state).await?,
-    )?))
+    let snapshot = engine::get_control_plan_snapshot(&state).await?;
+    Ok(Json(serde_json::to_value(snapshot.plan.as_ref())?))
 }

src/api/integrations.rs

@@ -1,3 +1,16 @@
+async fn home_assistant_snapshot(
+    State(state): State<AppState>,
+) -> Result<Json<Value>, AppError> {
+    let control_plan = engine::get_control_plan_snapshot(&state).await?;
+    let groups = list_home_assistant_groups(State(state.clone())).await?.0;
+    Ok(Json(json!({
+        "devices": state.db.list_devices()?,
+        "control_plan": control_plan.plan.as_ref(),
+        "control_plan_revision": control_plan.revision,
+        "groups": groups,
+    })))
+}
+
 #[derive(Debug, Deserialize)]
 struct HaTestRequest {
     entity_id: Option<String>,

src/api/system.rs

@@ -20,6 +20,7 @@ async fn build_bootstrap(state: &AppState) -> Result<Value, AppError> {
     let online_count = devices.iter().filter(|value| value.online).count();
     let simulator_count = devices.iter().filter(|value| value.simulated).count();
     let (received_frames_total, received_frames_by_device) = state.gree.received_frame_stats();
+    let control_plan = engine::get_control_plan_snapshot(state).await?;
     Ok(json!({
         "devices": devices,
         "zones": state.db.list_zones()?,
@@ -30,6 +31,8 @@ async fn build_bootstrap(state: &AppState) -> Result<Value, AppError> {
         "access_tokens": state.db.list_api_tokens()?,
         "house": {"mode": settings.house_mode},
         "outdoor_temperature": *state.outdoor_temperature.read().await,
+        "control_plan": control_plan.plan.as_ref(),
+        "control_plan_revision": control_plan.revision,
         "system": {
             "version": env!("CARGO_PKG_VERSION"),
             "uptime_seconds": state.started.elapsed().as_secs(),

src/api/websocket.rs

@@ -14,21 +14,60 @@ async fn websocket(
     Ok(ws.on_upgrade(move |socket| websocket_loop(state, socket)))
 }
 
-async fn websocket_loop(state: AppState, mut socket: WebSocket) {
-    let initial = match build_bootstrap(&state).await {
-        Ok(data) => json!({"event":"bootstrap","timestamp":Utc::now(),"data":data}),
-        Err(err) => {
-            json!({"event":"error","timestamp":Utc::now(),"data":{"message":err.to_string()}})
+fn control_plan_ws_message(snapshot: &crate::state::ControlPlanSnapshot) -> Value {
+    json!({
+        "event": "control_plan.updated",
+        "timestamp": Utc::now(),
+        "data": {
+            "revision": snapshot.revision,
+            "plan": snapshot.plan.as_ref(),
         }
+    })
+}
+
+async fn send_ws_bootstrap(state: &AppState, socket: &mut WebSocket) -> Result<Option<u64>, ()> {
+    let (message, revision) = match build_bootstrap(state).await {
+        Ok(data) => {
+            let revision = data.get("control_plan_revision").and_then(Value::as_u64);
+            (
+                json!({"event":"bootstrap","timestamp":Utc::now(),"data":data}),
+                revision,
+            )
+        }
+        Err(err) => (
+            json!({"event":"error","timestamp":Utc::now(),"data":{"message":err.to_string()}}),
+            None,
+        ),
     };
-    if socket
-        .send(Message::Text(initial.to_string()))
+    socket
+        .send(Message::Text(message.to_string()))
         .await
-        .is_err()
-    {
-        return;
-    }
+        .map_err(|_| ())?;
+    Ok(revision)
+}
+
+async fn websocket_loop(state: AppState, mut socket: WebSocket) {
+    // Subscribe before building the bootstrap so state changes during bootstrap generation are
+    // queued and can be applied immediately after the first frame.
     let mut receiver = state.events.subscribe();
+    let mut control_plan = state.subscribe_control_plan();
+
+    let bootstrap_revision = match send_ws_bootstrap(&state, &mut socket).await {
+        Ok(revision) => revision,
+        Err(()) => return,
+    };
+    let mut last_control_plan_revision = bootstrap_revision;
+
+    // If the watch value is exactly the plan embedded in bootstrap, mark it seen to avoid a
+    // duplicate control_plan.updated frame. A newer revision remains pending and is sent below.
+    let current_revision = control_plan
+        .borrow()
+        .as_ref()
+        .map(|snapshot| snapshot.revision);
+    if current_revision == bootstrap_revision {
+        control_plan.borrow_and_update();
+    }
+
     loop {
         tokio::select! {
             event = receiver.recv() => {
@@ -38,10 +77,31 @@ async fn websocket_loop(state: AppState, mut socket: WebSocket) {
                             if socket.send(Message::Text(text)).await.is_err() { break; }
                         }
                     }
-                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+                    Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
+                        tracing::warn!(skipped, "websocket client lagged; sending full bootstrap resync");
+                        // Drop the retained stale backlog before taking the replacement snapshot.
+                        // Events created while bootstrap is built are queued on this fresh receiver.
+                        receiver = state.events.subscribe();
+                        match send_ws_bootstrap(&state, &mut socket).await {
+                            Ok(revision) => last_control_plan_revision = revision,
+                            Err(()) => break,
+                        }
+                    }
                     Err(_) => break,
                 }
             }
+            changed = control_plan.changed() => {
+                if changed.is_err() { break; }
+                let snapshot = control_plan.borrow_and_update().clone();
+                if let Some(snapshot) = snapshot {
+                    if last_control_plan_revision == Some(snapshot.revision) {
+                        continue;
+                    }
+                    let text = control_plan_ws_message(snapshot.as_ref()).to_string();
+                    if socket.send(Message::Text(text)).await.is_err() { break; }
+                    last_control_plan_revision = Some(snapshot.revision);
+                }
+            }
             message = socket.next() => {
                 match message {
                     Some(Ok(Message::Ping(value))) => { if socket.send(Message::Pong(value)).await.is_err() { break; } }

src/engine.rs

@@ -6,14 +6,14 @@ use crate::{
         DeviceCommand, GroupControlPatch, HaReading, NightModeSettings, Reading, RuntimeSettings,
         Schedule, TemporaryQuickThermostat, Zone, ZoneControlPlan, ZoneReading,
     },
-    state::{AppState, PendingControllerCommand},
+    state::{AppState, ControlPlanSnapshot, PendingControllerCommand},
 };
 use anyhow::Result;
 use chrono::{DateTime, Datelike, Local, NaiveTime, Timelike, Utc, Weekday};
 use serde_json::{json, Value};
 use std::{
-    collections::HashMap,
-    sync::atomic::Ordering,
+    collections::{HashMap, HashSet},
+    sync::{atomic::Ordering, Arc},
     time::{Duration, Instant},
 };
 use tokio::time::sleep;

src/engine/control_plan.rs

@@ -1,3 +1,100 @@
+const CONTROL_PLAN_DEBOUNCE: Duration = Duration::from_millis(120);
+const CONTROL_PLAN_PERIODIC_REFRESH: Duration = Duration::from_secs(10);
+
+pub async fn get_control_plan_snapshot(
+    state: &AppState,
+) -> Result<Arc<ControlPlanSnapshot>, AppError> {
+    refresh_control_plan_cache(state, false).await
+}
+
+async fn refresh_control_plan_cache(
+    state: &AppState,
+    force: bool,
+) -> Result<Arc<ControlPlanSnapshot>, AppError> {
+    let _guard = state.control_plan_build_lock.lock().await;
+    if !force {
+        if let Some(snapshot) = state.control_plan_snapshot() {
+            if !state.control_plan_dirty.load(Ordering::Acquire) {
+                return Ok(snapshot);
+            }
+        }
+    }
+
+    // Clear before calculating. Any relevant mutation that happens during the build sets the
+    // flag again and schedules another pass, so an update cannot be lost.
+    state.control_plan_dirty.store(false, Ordering::Release);
+    let plan = match build_control_plan(state).await {
+        Ok(plan) => plan,
+        Err(err) => {
+            state.control_plan_dirty.store(true, Ordering::Release);
+            state.control_plan_wakeup.notify_one();
+            return Err(err);
+        }
+    };
+
+    if let Some(previous) = state.control_plan_snapshot() {
+        if control_plans_semantically_equal(previous.plan.as_ref(), &plan)? {
+            // Keep generated_at fresh for HTTP/bootstrap consumers without creating a new
+            // semantic revision. WebSocket clients suppress same-revision watch updates.
+            let refreshed = Arc::new(ControlPlanSnapshot {
+                revision: previous.revision,
+                plan: Arc::new(plan),
+            });
+            state.control_plan.send_replace(Some(refreshed.clone()));
+            return Ok(refreshed);
+        }
+    }
+
+    let revision = state
+        .control_plan_revision
+        .fetch_add(1, Ordering::AcqRel)
+        .saturating_add(1);
+    let snapshot = Arc::new(ControlPlanSnapshot {
+        revision,
+        plan: Arc::new(plan),
+    });
+    state.control_plan.send_replace(Some(snapshot.clone()));
+    Ok(snapshot)
+}
+
+fn control_plans_semantically_equal(
+    left: &ControlPlan,
+    right: &ControlPlan,
+) -> Result<bool, AppError> {
+    fn semantic_value(plan: &ControlPlan) -> Result<Value, AppError> {
+        let mut value = serde_json::to_value(plan)?;
+        if let Some(object) = value.as_object_mut() {
+            object.remove("generated_at");
+        }
+        Ok(value)
+    }
+    Ok(semantic_value(left)? == semantic_value(right)?)
+}
+
+async fn control_plan_cache_loop(state: AppState) {
+    let mut periodic = tokio::time::interval(CONTROL_PLAN_PERIODIC_REFRESH);
+    periodic.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    // Tokio intervals tick immediately once; consume that tick because startup is refreshed below.
+    periodic.tick().await;
+
+    if let Err(err) = refresh_control_plan_cache(&state, false).await {
+        tracing::warn!(error=?err, "cannot build initial control plan snapshot");
+    }
+
+    loop {
+        let force = tokio::select! {
+            _ = state.control_plan_wakeup.notified() => {
+                sleep(CONTROL_PLAN_DEBOUNCE).await;
+                false
+            }
+            _ = periodic.tick() => true,
+        };
+        if let Err(err) = refresh_control_plan_cache(&state, force).await {
+            tracing::warn!(error=?err, "cannot refresh control plan snapshot");
+        }
+    }
+}
+
 pub async fn build_control_plan(state: &AppState) -> Result<ControlPlan, AppError> {
     let settings = state.settings.read().await.clone();
     let schedules = state.db.list_schedules()?;
@@ -333,11 +430,22 @@ fn next_schedule_events(
     if mode == "off" {
         return Vec::new();
     }
+    let boundary_minutes = schedule_boundary_minutes(&zone.id, schedules);
+    if boundary_minutes.is_empty() {
+        return Vec::new();
+    }
     let mut events = Vec::new();
     let mut current = active_schedule_for_zone(zone, schedules, now).map(|item| item.id.as_str());
     let base = minute_floor(now);
     for minute in 1..=(8 * 24 * 60) {
         let candidate = base + chrono::Duration::minutes(minute);
+        let minute_of_day = candidate.hour() * 60 + candidate.minute();
+        let previous_local = base + chrono::Duration::minutes(minute - 1);
+        let clock_discontinuity = candidate.naive_local() - previous_local.naive_local()
+            != chrono::Duration::minutes(1);
+        if !boundary_minutes.contains(&minute_of_day) && !clock_discontinuity {
+            continue;
+        }
         let next = active_schedule_for_zone(zone, schedules, candidate);
         let next_id = next.map(|item| item.id.as_str());
         if next_id == current {

src/engine/runtime.rs

@@ -24,6 +24,11 @@ pub fn start(state: AppState) {
     if let Err(err) = reset_temporary_condition_observations_after_restart(&state) {
         tracing::warn!(error=?err, "cannot reset temporary thermostat observation continuity after restart");
     }
+
+    let control_plan_state = state.clone();
+    tokio::spawn(async move {
+        control_plan_cache_loop(control_plan_state).await;
+    });
     let poll_state = state.clone();
     tokio::spawn(async move {
         sleep(Duration::from_millis(500)).await;

src/engine/schedules.rs

@@ -17,6 +17,21 @@ fn minute_floor(now: DateTime<Local>) -> DateTime<Local> {
         .unwrap_or(now)
 }
 
+fn schedule_boundary_minutes(zone_id: &str, schedules: &[Schedule]) -> HashSet<u32> {
+    let mut result = HashSet::new();
+    for item in schedules
+        .iter()
+        .filter(|item| item.enabled && item.zone_id == zone_id)
+    {
+        for raw in [&item.start_time, &item.end_time] {
+            if let Ok(time) = NaiveTime::parse_from_str(raw, "%H:%M") {
+                result.insert(time.hour() * 60 + time.minute());
+            }
+        }
+    }
+    result
+}
+
 pub fn next_schedule_boundary_utc(
     zone_id: &str,
     schedules: &[Schedule],
@@ -27,10 +42,22 @@ pub fn next_schedule_boundary_utc(
         .filter(|item| item.enabled && item.zone_id == zone_id && schedule_active(item, now))
         .max_by_key(|item| item.updated_at)
         .map(|item| item.id.as_str());
+    let boundary_minutes = schedule_boundary_minutes(zone_id, schedules);
+    if boundary_minutes.is_empty() {
+        return None;
+    }
     let base = minute_floor(now);
-    // Eight days cover a complete weekly schedule plus the next transition.
+    // State normally changes on a configured start/end minute. Also evaluate local-clock
+    // discontinuities so DST jumps/repeats preserve the original absolute-minute behavior.
     for minute in 1..=(8 * 24 * 60) {
         let candidate = base + chrono::Duration::minutes(minute);
+        let minute_of_day = candidate.hour() * 60 + candidate.minute();
+        let previous_local = base + chrono::Duration::minutes(minute - 1);
+        let clock_discontinuity = candidate.naive_local() - previous_local.naive_local()
+            != chrono::Duration::minutes(1);
+        if !boundary_minutes.contains(&minute_of_day) && !clock_discontinuity {
+            continue;
+        }
         let next = schedules
             .iter()
             .filter(|item| {

src/engine/tests.rs

@@ -11,6 +11,31 @@ mod tests {
         assert_eq!(effective_sensor_stale_after_seconds(120_000, 600), 86_400);
     }
 
+    #[test]
+    fn control_plan_semantic_equality_ignores_only_generation_timestamp() {
+        let generated_at = Utc.with_ymd_and_hms(2026, 9, 4, 8, 0, 0).unwrap();
+        let left = ControlPlan {
+            generated_at: generated_at.clone(),
+            house_mode: "cool".into(),
+            house_preset: Some("comfort".into()),
+            house_power: true,
+            outdoor_temperature: Some(24.0),
+            control_strategy: "thermostat".into(),
+            night_mode_active: false,
+            night_mode_start: "22:00".into(),
+            night_mode_end: "06:00".into(),
+            night_mode_max_fan_speed: 2,
+            next_events: Vec::new(),
+            zones: Vec::new(),
+            rules: Vec::new(),
+        };
+        let mut right = left.clone();
+        right.generated_at = generated_at + chrono::Duration::seconds(30);
+        assert!(control_plans_semantically_equal(&left, &right).unwrap());
+        right.house_mode = "heat".into();
+        assert!(!control_plans_semantically_equal(&left, &right).unwrap());
+    }
+
     #[test]
     fn overnight_schedule_works() {
         let now = Utc.with_ymd_and_hms(2025, 1, 7, 1, 0, 0).unwrap().with_timezone(&Local); // Tuesday

src/main.rs

@@ -18,13 +18,16 @@ use models::Device;
 use protocol::GreeClient;
 use state::AppState;
 use std::{
-    sync::{atomic::AtomicBool, Arc},
+    sync::{
+        atomic::{AtomicBool, AtomicU64},
+        Arc,
+    },
     time::{Duration, Instant},
 };
 use tokio::{
     net::TcpListener,
     signal,
-    sync::{broadcast, Notify, RwLock},
+    sync::{broadcast, watch, Notify, RwLock},
 };
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
@@ -56,6 +59,7 @@ async fn main() -> Result<()> {
     }
 
     let (events, _) = broadcast::channel(512);
+    let (control_plan, _) = watch::channel(None);
     let debug_gree_frames = Arc::new(AtomicBool::new(runtime_settings.debug.gree_frames));
     let http = reqwest::Client::builder()
         .timeout(Duration::from_secs(10))
@@ -78,6 +82,11 @@ async fn main() -> Result<()> {
         debug_gree_frames,
         initial_device_sync_complete: Arc::new(AtomicBool::new(false)),
         zone_control_wakeup: Arc::new(Notify::new()),
+        control_plan,
+        control_plan_wakeup: Arc::new(Notify::new()),
+        control_plan_dirty: Arc::new(AtomicBool::new(true)),
+        control_plan_revision: Arc::new(AtomicU64::new(0)),
+        control_plan_build_lock: Arc::new(tokio::sync::Mutex::new(())),
         device_operation_locks: Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())),
         zone_operation_locks: Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())),
         group_operation_locks: Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())),

src/state.rs

@@ -1,17 +1,20 @@
 use crate::{
     config::Config,
     db::Db,
-    models::{ApiEvent, DeviceCommand, RuntimeSettings},
+    models::{ApiEvent, ControlPlan, DeviceCommand, RuntimeSettings},
     protocol::GreeClient,
 };
 use chrono::Utc;
 use serde_json::Value;
 use std::{
     collections::HashMap,
-    sync::{atomic::AtomicBool, Arc},
+    sync::{
+        atomic::{AtomicBool, AtomicU64, Ordering},
+        Arc,
+    },
     time::Instant,
 };
-use tokio::sync::{broadcast, Mutex, Notify, OwnedMutexGuard, RwLock};
+use tokio::sync::{broadcast, watch, Mutex, Notify, OwnedMutexGuard, RwLock};
 
 #[derive(Debug, Clone)]
 pub(crate) struct PendingControllerCommand {
@@ -24,6 +27,12 @@ pub(crate) struct PendingControllerCommand {
     pub expires_at: Instant,
 }
 
+#[derive(Debug, Clone)]
+pub struct ControlPlanSnapshot {
+    pub revision: u64,
+    pub plan: Arc<ControlPlan>,
+}
+
 #[derive(Clone)]
 pub struct AppState {
     pub db: Db,
@@ -39,6 +48,14 @@ pub struct AppState {
     pub initial_device_sync_complete: Arc<AtomicBool>,
     /// Explicit thermostat changes wake the regulator instead of waiting for the next fixed interval.
     pub zone_control_wakeup: Arc<Notify>,
+    /// Latest materialized control plan. WebSocket clients subscribe to this state stream,
+    /// while HTTP uses the same snapshot as a compatibility/fallback surface.
+    pub control_plan: watch::Sender<Option<Arc<ControlPlanSnapshot>>>,
+    /// Coalesces changes that can affect the derived control plan.
+    pub control_plan_wakeup: Arc<Notify>,
+    pub control_plan_dirty: Arc<AtomicBool>,
+    pub(crate) control_plan_revision: Arc<AtomicU64>,
+    pub(crate) control_plan_build_lock: Arc<Mutex<()>>,
     pub(crate) device_operation_locks: Arc<Mutex<HashMap<String, Arc<Mutex<()>>>>>,
     pub(crate) zone_operation_locks: Arc<Mutex<HashMap<String, Arc<Mutex<()>>>>>,
     pub(crate) group_operation_locks: Arc<Mutex<HashMap<String, Arc<Mutex<()>>>>>,
@@ -112,15 +129,36 @@ impl AppState {
         self.zone_control_wakeup.notify_one();
     }
 
+    pub fn control_plan_snapshot(&self) -> Option<Arc<ControlPlanSnapshot>> {
+        self.control_plan.borrow().clone()
+    }
+
+    pub fn subscribe_control_plan(&self) -> watch::Receiver<Option<Arc<ControlPlanSnapshot>>> {
+        self.control_plan.subscribe()
+    }
+
+    pub fn invalidate_control_plan(&self) {
+        self.control_plan_dirty.store(true, Ordering::Release);
+        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);
         let _ = self.events.send(ApiEvent {
-            event: event.into(),
+            event,
             timestamp: Utc::now(),
             data,
         });
+        if invalidates_control_plan {
+            self.invalidate_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");
         }
@@ -144,3 +182,44 @@ impl AppState {
         }
     }
 }
+
+fn control_plan_event_affects_plan(event: &str) -> bool {
+    event.starts_with("device.")
+        || event == "devices.discovered"
+        || event.starts_with("zone.")
+        || event.starts_with("group.")
+        || event.starts_with("schedule.")
+        || event.starts_with("automation.")
+        || event.starts_with("flow.")
+        || event.starts_with("settings.")
+        || matches!(
+            event,
+            "house.mode_changed" | "outdoor.updated" | "configuration.imported"
+        )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::control_plan_event_affects_plan;
+
+    #[test]
+    fn control_plan_invalidation_tracks_state_events_but_not_diagnostics() {
+        for event in [
+            "device.updated",
+            "zone.updated",
+            "group.updated",
+            "schedule.updated",
+            "automation.updated",
+            "flow.updated",
+            "settings.night.updated",
+            "house.mode_changed",
+            "outdoor.updated",
+            "configuration.imported",
+        ] {
+            assert!(control_plan_event_affects_plan(event), "{event}");
+        }
+        for event in ["log.created", "api.request", "gree.frame"] {
+            assert!(!control_plan_event_affects_plan(event), "{event}");
+        }
+    }
+}