Merge pull request #1123 from LibreQoE/fix/stormguard_reset

Fix StormGuard reset when disabled

docs-es/v2.0/stormguard-es.md

@@ -41,6 +41,8 @@ minimum_upload_percentage = 0.5
 
 Si está probando, comience con `dry_run = true`.
 
+Al deshabilitar StormGuard, o al volver a `dry_run = true` después de usarlo en modo activo, las colas administradas recuperan sus tasas garantizadas y límites máximos configurados, y se eliminan los ajustes adaptativos persistidos por StormGuard. Los ajustes administrados por el operador no se modifican.
+
 ## UI y depuración
 
 - WebUI (Node Manager) incluye una pestaña dedicada de StormGuard además de las vistas de estado y depuración.

docs/v2.0/stormguard.md

@@ -41,6 +41,8 @@ minimum_upload_percentage = 0.5
 
 If you are testing, start with `dry_run = true` so you can observe decisions before allowing live limit changes.
 
+Disabling StormGuard, or changing an active deployment back to `dry_run = true`, restores its managed queues to their configured rates and ceilings and removes StormGuard's persisted adaptive overrides. Operator-managed overrides are not changed.
+
 ## UI and Debugging
 
 - WebUI provides a dedicated StormGuard dashboard tab plus status and debug views.

src/rust/lqos_bakery/src/commands.rs

@@ -115,6 +115,19 @@ pub enum ExecutionMode {
     LiveUpdate,
 }
 
+/// One HTB class that StormGuard must restore before leaving live mode.
+#[derive(Debug, Clone, Allocative)]
+pub struct StormGuardRestoreAdjustment {
+    /// Network interface containing the class.
+    pub interface_name: String,
+    /// Fully qualified HTB class identifier.
+    pub class_id: TcHandle,
+    /// Planned guaranteed class rate in Mbps.
+    pub planned_rate: u64,
+    /// Planned class ceiling in Mbps.
+    pub planned_ceil: u64,
+}
+
 /// List of commands that the Bakery system can handle.
 #[derive(Debug, Clone, Allocative)]
 pub enum BakeryCommands {
@@ -255,6 +268,8 @@ pub enum BakeryCommands {
     },
     /// Change a specific HTB class rate on-the-fly; optionally dry-run.
     StormGuardAdjustment {
+        /// Bakery shaping-tree generation used to resolve this class handle.
+        tree_generation: u64,
         /// If true, log the tc command instead of executing it.
         dry_run: bool,
         /// Network interface name (e.g., `eth0`) containing the class.
@@ -263,6 +278,45 @@ pub enum BakeryCommands {
         class_id: String,
         /// New class ceiling rate in Mbps (the handler sets ceil and rate-1).
         new_rate: u64,
+        /// Original guaranteed class rate in Mbps.
+        planned_rate: u64,
+        /// Original class ceiling in Mbps.
+        planned_ceil: u64,
+        /// Optional synchronous completion reply.
+        #[allocative(skip)]
+        reply: Option<ReplySender<Result<(), String>>>,
+    },
+    /// Restore planned HTB rates and clear retained StormGuard replay state after success.
+    ResetStormGuardAdjustments {
+        /// Bakery shaping-tree generation used to resolve the supplied class handles.
+        tree_generation: u64,
+        /// Complete set of StormGuard-managed HTB classes to restore.
+        adjustments: Vec<StormGuardRestoreAdjustment>,
+        /// Restore the supplied adjustments when Bakery has no in-process ownership cache.
+        restore_untracked: bool,
+        /// Optional synchronous completion reply.
+        #[allocative(skip)]
+        reply: Option<ReplySender<Result<(), String>>>,
+    },
+    /// Discard retained StormGuard class ownership after a shaping-tree rebuild.
+    DiscardStormGuardAdjustments {
+        /// Bakery shaping-tree generation whose cached ownership may be discarded.
+        tree_generation: u64,
+        /// Optional synchronous completion reply.
+        #[allocative(skip)]
+        reply: Option<ReplySender<Result<(), String>>>,
+    },
+    /// Apply a StormGuard circuit SQM update immediately and acknowledge the live result.
+    StormGuardCircuitAdjustment {
+        /// Bakery shaping-tree generation used to resolve this circuit.
+        tree_generation: u64,
+        /// Stable circuit hash in Bakery state.
+        circuit_hash: i64,
+        /// Requested per-circuit SQM override, or `None` to restore the configured default.
+        sqm_override: Option<String>,
+        /// Optional synchronous completion reply.
+        #[allocative(skip)]
+        reply: Option<ReplySender<Result<bool, String>>>,
     },
     /// Runtime TreeGuard request to virtualize or restore a non-top-level site without a full reload.
     TreeGuardSetNodeVirtual {

src/rust/lqos_bakery/src/lib.rs

@@ -58,7 +58,7 @@ pub use commands::{
     BakeryCommands, RuntimeNodeOperationAction as BakeryRuntimeNodeOperationAction,
     RuntimeNodeOperationFailureReason as BakeryRuntimeNodeOperationFailureReason,
     RuntimeNodeOperationSnapshot as BakeryRuntimeNodeOperationSnapshot,
-    RuntimeNodeOperationStatus as BakeryRuntimeNodeOperationStatus,
+    RuntimeNodeOperationStatus as BakeryRuntimeNodeOperationStatus, StormGuardRestoreAdjustment,
 };
 use lqos_bus::{
     BusRequest, BusResponse, InsightLicenseSummary, LibreqosBusClient, TcHandle, UrgentSeverity,
@@ -248,6 +248,22 @@ struct StormguardOverrideKey {
     class: TcHandle,
 }
 
+#[derive(Clone, Copy, Debug)]
+struct StormguardOverrideValue {
+    planned_rate: u64,
+    planned_ceil: u64,
+}
+
+struct StormguardAdjustmentRequest {
+    tree_generation: u64,
+    dry_run: bool,
+    interface_name: String,
+    class_id: String,
+    new_rate: u64,
+    planned_rate: u64,
+    planned_ceil: u64,
+}
+
 #[derive(Clone, Debug)]
 struct VirtualizedSiteQdiscHandles {
     down: Option<u16>,
@@ -1483,6 +1499,10 @@ fn circuits_with_pending_migration_targets(
 pub static ACTIVE_CIRCUITS: AtomicUsize = AtomicUsize::new(0);
 /// True while Bakery is applying a full reload batch to `tc`.
 static FULL_RELOAD_IN_PROGRESS: AtomicBool = AtomicBool::new(false);
+/// Monotonic count of committed Bakery shaping-tree changes.
+static STORMGUARD_TREE_GENERATION: AtomicU64 = AtomicU64::new(0);
+/// Latest generation produced by a successful full tree rebuild.
+static STORMGUARD_TREE_REBUILD_GENERATION: AtomicU64 = AtomicU64::new(0);
 #[cfg(test)]
 pub(crate) fn test_state_lock() -> &'static std::sync::Mutex<()> {
     static LOCK: OnceLock<std::sync::Mutex<()>> = OnceLock::new();
@@ -3705,6 +3725,16 @@ pub fn full_reload_in_progress() -> bool {
     FULL_RELOAD_IN_PROGRESS.load(Ordering::Relaxed)
 }
 
+/// Returns the Bakery shaping-tree generation observed by StormGuard.
+pub fn stormguard_tree_generation() -> u64 {
+    STORMGUARD_TREE_GENERATION.load(Ordering::Acquire)
+}
+
+/// Returns the latest generation that replaced class handles with a full tree rebuild.
+pub fn stormguard_tree_rebuild_generation() -> u64 {
+    STORMGUARD_TREE_REBUILD_GENERATION.load(Ordering::Acquire)
+}
+
 #[derive(Debug, Default, Clone, Copy)]
 struct MappedLimitStats {
     enforced_limit: Option<usize>,
@@ -4040,8 +4070,9 @@ fn bakery_main(rx: Receiver<BakeryCommands>, tx: Sender<BakeryCommands>) {
         .ok()
         .map(|config| QdiscHandleState::load(&config))
         .unwrap_or_default();
-    // Persist latest StormGuard ceilings keyed by interface + class so we can replay after rebuilds.
-    let mut stormguard_overrides: HashMap<StormguardOverrideKey, u64> = HashMap::new();
+    // Retain each StormGuard-owned class's original plan until disable/reset reconciliation.
+    let mut stormguard_overrides: HashMap<StormguardOverrideKey, StormguardOverrideValue> =
+        HashMap::new();
     let mut virtualized_sites: HashMap<i64, VirtualizedSiteState> = HashMap::new();
     let mut runtime_node_operations: HashMap<i64, RuntimeNodeOperation> = HashMap::new();
     let mut next_runtime_operation_id: u64 = 1;
@@ -5088,78 +5119,92 @@ fn bakery_main(rx: Receiver<BakeryCommands>, tx: Sender<BakeryCommands>) {
                 );
             }
             BakeryCommands::StormGuardAdjustment {
+                tree_generation,
                 dry_run,
                 interface_name,
                 class_id,
                 new_rate,
+                planned_rate,
+                planned_ceil,
+                reply,
             } => {
-                let has_mq_run = MQ_CREATED.load(Relaxed);
-                if !has_mq_run {
-                    debug!("StormGuardAdjustment received before MQ setup, skipping.");
-                    continue;
+                let result = apply_stormguard_adjustment(
+                    StormguardAdjustmentRequest {
+                        tree_generation,
+                        dry_run,
+                        interface_name,
+                        class_id,
+                        new_rate,
+                        planned_rate,
+                        planned_ceil,
+                    },
+                    &mut stormguard_overrides,
+                );
+                if let Some(reply) = reply {
+                    let _ = reply.send(result);
                 }
-                let Ok(config) = lqos_config::load_config() else {
-                    error!("Failed to load configuration, skipping StormGuardAdjustment.");
-                    continue;
-                };
-                let Ok(tc_handle) = TcHandle::from_string(&class_id) else {
-                    warn!(
-                        "StormGuardAdjustment has invalid class_id [{}], skipping.",
-                        class_id
-                    );
-                    continue;
-                };
-                if !dry_run {
-                    let key = StormguardOverrideKey {
-                        interface: interface_name.to_string(),
-                        class: tc_handle,
-                    };
-                    stormguard_overrides.insert(key, new_rate);
-                }
-                if let Some(reason) = live_tree_mutation_blocker_for_config(&config) {
-                    info!(
-                        "Skipping StormGuard live class change for {} {} because {}.",
-                        interface_name, class_id, reason
-                    );
-                    continue;
-                }
-                let normalized_class = tc_handle.as_tc_string();
-                // Build the HTB command
-                let args = vec![
-                    "class".to_string(),
-                    "replace".to_string(),
-                    "dev".to_string(),
-                    interface_name.to_string(),
-                    "classid".to_string(),
-                    normalized_class.clone(),
-                    "htb".to_string(),
-                    "rate".to_string(),
-                    format!("{}mbit", new_rate.saturating_sub(1)),
-                    "ceil".to_string(),
-                    format!("{}mbit", new_rate),
-                ];
-                if dry_run {
-                    info!("DRY RUN: /sbin/tc {}", args.join(" "));
+            }
+            BakeryCommands::ResetStormGuardAdjustments {
+                tree_generation,
+                adjustments,
+                restore_untracked,
+                reply,
+            } => {
+                let current_generation = stormguard_tree_generation();
+                let result = if tree_generation != current_generation {
+                    Err(format!(
+                        "StormGuard reset used shaping-tree generation {tree_generation}, but Bakery is at generation {current_generation}"
+                    ))
+                } else if batch.is_some() {
+                    Err("Bakery queue rebuild batch is still open".to_string())
                 } else {
-                    let output = std::process::Command::new("/sbin/tc").args(&args).output();
-                    match output {
-                        Err(e) => {
-                            warn!("Failed to run tc command: {}", e);
-                        }
-                        Ok(out) => {
-                            if !out.status.success() {
-                                warn!(
-                                    "tc command failed: {}",
-                                    String::from_utf8_lossy(&out.stderr)
-                                );
-                            } else {
-                                debug!(
-                                    "tc command succeeded: {}",
-                                    String::from_utf8_lossy(&out.stdout)
-                                );
-                            }
-                        }
-                    }
+                    reset_stormguard_adjustments(
+                        &adjustments,
+                        restore_untracked,
+                        &mut stormguard_overrides,
+                    )
+                };
+                if let Some(reply) = reply {
+                    let _ = reply.send(result);
+                }
+            }
+            BakeryCommands::DiscardStormGuardAdjustments {
+                tree_generation,
+                reply,
+            } => {
+                let current_generation = stormguard_tree_generation();
+                let result = if tree_generation != current_generation {
+                    Err(format!(
+                        "StormGuard discard used shaping-tree generation {tree_generation}, but Bakery is at generation {current_generation}"
+                    ))
+                } else if batch.is_some() {
+                    Err("Bakery queue rebuild batch is still open".to_string())
+                } else {
+                    stormguard_overrides.clear();
+                    Ok(())
+                };
+                if let Some(reply) = reply {
+                    let _ = reply.send(result);
+                }
+            }
+            BakeryCommands::StormGuardCircuitAdjustment {
+                tree_generation,
+                circuit_hash,
+                sqm_override,
+                reply,
+            } => {
+                let current_generation = stormguard_tree_generation();
+                let result = if tree_generation != current_generation {
+                    Err(format!(
+                        "StormGuard circuit adjustment used shaping-tree generation {tree_generation}, but Bakery is at generation {current_generation}"
+                    ))
+                } else if batch.is_some() {
+                    Err("Bakery queue rebuild batch is still open".to_string())
+                } else {
+                    apply_stormguard_circuit_adjustment(circuit_hash, sqm_override, &mut circuits)
+                };
+                if let Some(reply) = reply {
+                    let _ = reply.send(result);
                 }
             }
             BakeryCommands::TreeGuardSetNodeVirtual {
@@ -5201,7 +5246,7 @@ fn handle_commit_batch(
     qdisc_handles: &mut QdiscHandleState,
     tx: &Sender<BakeryCommands>,
     migrations: &mut HashMap<i64, Migration>,
-    stormguard_overrides: &HashMap<StormguardOverrideKey, u64>,
+    stormguard_overrides: &HashMap<StormguardOverrideKey, StormguardOverrideValue>,
     virtualized_sites: &mut HashMap<i64, VirtualizedSiteState>,
     runtime_node_operations: &mut HashMap<i64, RuntimeNodeOperation>,
 ) {
@@ -5271,7 +5316,6 @@ fn handle_commit_batch(
             &config,
             new_batch,
             resolved_mq_layout,
-            stormguard_overrides,
             virtualized_sites,
             runtime_node_operations,
             summary,
@@ -5318,7 +5362,6 @@ fn handle_commit_batch(
             &config,
             new_batch,
             resolved_mq_layout,
-            stormguard_overrides,
             virtualized_sites,
             runtime_node_operations,
             summary,
@@ -5364,7 +5407,6 @@ fn handle_commit_batch(
             &config,
             new_batch,
             resolved_mq_layout,
-            stormguard_overrides,
             virtualized_sites,
             runtime_node_operations,
             summary,
@@ -5414,7 +5456,6 @@ fn handle_commit_batch(
             &config,
             new_batch,
             resolved_mq_layout,
-            stormguard_overrides,
             virtualized_sites,
             runtime_node_operations,
             summary.clone(),
@@ -5483,7 +5524,40 @@ fn handle_commit_batch(
             &config,
             new_batch,
             resolved_mq_layout,
-            stormguard_overrides,
+            virtualized_sites,
+            runtime_node_operations,
+            summary,
+        );
+        return;
+    }
+
+    if stormguard_site_changes_intersect_ownership(&site_change_mode, &config, stormguard_overrides)
+    {
+        let summary = "Bakery full reload triggered to keep StormGuard class ownership synchronized with a changed shaping plan.".to_string();
+        let (new_batch, mapped_limit_stats) = filter_batch_by_mapped_circuit_limit(
+            raw_batch.clone(),
+            &baseline_circuits,
+            effective_limit,
+        );
+        log_mapped_limit_decision(
+            "StormGuard-consistent rebuild",
+            mapped_limit,
+            mapped_limit_stats,
+        );
+        if mapped_limit_stats.dropped_mapped > 0 {
+            maybe_emit_mapped_circuit_limit_urgent(&mapped_limit_stats);
+        }
+        announce_full_reload(&summary);
+        full_reload(
+            batch,
+            sites,
+            circuits,
+            live_circuits,
+            mq_layout,
+            qdisc_handles,
+            &config,
+            new_batch,
+            resolved_mq_layout,
             virtualized_sites,
             runtime_node_operations,
             summary,
@@ -5536,7 +5610,6 @@ fn handle_commit_batch(
                     &config,
                     raw_batch.clone(),
                     resolved_mq_layout.clone(),
-                    stormguard_overrides,
                     virtualized_sites,
                     runtime_node_operations,
                     summary,
@@ -5544,6 +5617,12 @@ fn handle_commit_batch(
                 return; // Skip the rest of this CommitBatch processing
             }
         }
+        if stormguard_live_enabled(&config) {
+            // Publish the generation only after every live speed command is in this same channel.
+            // Previously queued StormGuard work retains the old generation and will be rejected;
+            // work created for the new generation is queued behind the new site plan.
+            STORMGUARD_TREE_GENERATION.fetch_add(1, Ordering::Release);
+        }
     }
 
     // Now we can process circuit changes incrementally
@@ -5830,7 +5909,6 @@ fn handle_commit_batch(
                         &config,
                         raw_batch.clone(),
                         resolved_mq_layout.clone(),
-                        stormguard_overrides,
                         virtualized_sites,
                         runtime_node_operations,
                         summary,
@@ -6179,6 +6257,29 @@ fn handle_tick(
     execute_and_record_live_change(&commands, "pruning lazy queues");
 }
 
+fn stormguard_site_changes_intersect_ownership(
+    site_change_mode: &SiteDiffResult,
+    config: &Arc<Config>,
+    overrides: &HashMap<StormguardOverrideKey, StormguardOverrideValue>,
+) -> bool {
+    let SiteDiffResult::SpeedChanges { changes } = site_change_mode else {
+        return false;
+    };
+
+    changes.iter().any(|change| {
+        let Some((down_class, up_class)) = site_class_handles(change) else {
+            return false;
+        };
+        overrides.contains_key(&StormguardOverrideKey {
+            interface: config.isp_interface(),
+            class: down_class,
+        }) || overrides.contains_key(&StormguardOverrideKey {
+            interface: config.internet_interface(),
+            class: up_class,
+        })
+    })
+}
+
 fn handle_change_site_speed_live(
     site_hash: i64,
     download_bandwidth_min: f32,
@@ -6295,7 +6396,14 @@ fn handle_change_site_speed_live(
                 ),
             ],
         ];
-        execute_and_record_live_change(&commands, "changing site speed live");
+        let result = execute_and_record_live_change(&commands, "changing site speed live");
+        if !result.ok {
+            mark_reload_required(format!(
+                "Bakery live site speed update failed: {}",
+                summarize_apply_result("changing site speed live", &result)
+            ));
+            return;
+        }
         // Update the site speeds in the site map - create a new Arc with updated values
         let new_site = Arc::new(BakeryCommands::AddSite {
             site_hash,
@@ -10913,7 +11021,6 @@ fn full_reload(
     config: &Arc<Config>,
     new_batch: Vec<Arc<BakeryCommands>>,
     resolved_mq_layout: Option<MqDeviceLayout>,
-    stormguard_overrides: &HashMap<StormguardOverrideKey, u64>,
     virtualized_sites: &mut HashMap<i64, VirtualizedSiteState>,
     runtime_node_operations: &mut HashMap<i64, RuntimeNodeOperation>,
     trigger_summary: String,
@@ -11045,7 +11152,13 @@ fn full_reload(
             "A successful Bakery full reload re-established baseline state; incremental topology mutations can resume.",
         );
         FIRST_COMMIT_APPLIED.store(true, Ordering::Relaxed);
-        apply_stormguard_overrides(stormguard_overrides, config);
+        let rebuild_generation = STORMGUARD_TREE_GENERATION
+            .load(Ordering::Relaxed)
+            .wrapping_add(1);
+        STORMGUARD_TREE_REBUILD_GENERATION.store(rebuild_generation, Ordering::Release);
+        STORMGUARD_TREE_GENERATION.store(rebuild_generation, Ordering::Release);
+        // StormGuard replays persisted adjustments after the queue tracker publishes the rebuilt
+        // structure. Replaying cached class handles here could target reassigned handles.
     } else {
         *sites = previous_sites;
         *circuits = previous_circuits;
@@ -11185,56 +11298,483 @@ fn process_batch(
     }
 }
 
-fn apply_stormguard_overrides(
-    overrides: &HashMap<StormguardOverrideKey, u64>,
-    config: &Arc<Config>,
+fn stormguard_live_enabled(config: &Config) -> bool {
+    config
+        .stormguard
+        .as_ref()
+        .is_some_and(|stormguard| stormguard.enabled && !stormguard.dry_run)
+}
+
+fn record_stormguard_override(
+    overrides: &mut HashMap<StormguardOverrideKey, StormguardOverrideValue>,
+    key: StormguardOverrideKey,
+    planned_rate: u64,
+    planned_ceil: u64,
 ) {
-    if config.queues.queue_mode.is_observe() {
-        push_bakery_event(
-            "stormguard_override_replay_skipped",
-            "info",
-            "Skipping StormGuard HTB override replay because queue_mode is observe.".to_string(),
+    overrides.entry(key).or_insert(StormguardOverrideValue {
+        planned_rate,
+        planned_ceil,
+    });
+}
+
+fn apply_stormguard_adjustment(
+    request: StormguardAdjustmentRequest,
+    overrides: &mut HashMap<StormguardOverrideKey, StormguardOverrideValue>,
+) -> Result<(), String> {
+    let current_generation = stormguard_tree_generation();
+    if request.tree_generation != current_generation {
+        return Err(format!(
+            "StormGuard class change used shaping-tree generation {}, but Bakery is at generation {current_generation}",
+            request.tree_generation
+        ));
+    }
+    if !MQ_CREATED.load(Relaxed) {
+        return Err("StormGuard class change received before MQ setup".to_string());
+    }
+    let config = lqos_config::load_config().map_err(|error| {
+        format!("failed to load configuration for StormGuard class change: {error}")
+    })?;
+    let tc_handle = TcHandle::from_string(&request.class_id)
+        .map_err(|error| format!("invalid StormGuard class ID {}: {error}", request.class_id))?;
+    if !request.dry_run && !stormguard_live_enabled(&config) {
+        return Err("StormGuard is disabled or in dry-run mode".to_string());
+    }
+    if let Some(reason) = live_tree_mutation_blocker_for_config(&config) {
+        return Err(format!(
+            "StormGuard live class change is blocked because {reason}"
+        ));
+    }
+
+    let args = stormguard_htb_command(
+        &request.interface_name,
+        tc_handle,
+        request.new_rate.saturating_sub(1),
+        request.new_rate,
+    );
+    if request.dry_run {
+        info!("DRY RUN: /sbin/tc {}", args.join(" "));
+        return Ok(());
+    }
+
+    let output = std::process::Command::new("/sbin/tc")
+        .args(&args)
+        .output()
+        .map_err(|error| format!("failed to run StormGuard tc command: {error}"))?;
+    if !output.status.success() {
+        return Err(format!(
+            "StormGuard tc command failed: {}",
+            String::from_utf8_lossy(&output.stderr)
+        ));
+    }
+    record_stormguard_override(
+        overrides,
+        StormguardOverrideKey {
+            interface: request.interface_name,
+            class: tc_handle,
+        },
+        request.planned_rate,
+        request.planned_ceil,
+    );
+    debug!(
+        "StormGuard tc command succeeded: {}",
+        String::from_utf8_lossy(&output.stdout)
+    );
+    Ok(())
+}
+
+fn stormguard_htb_command(
+    interface_name: &str,
+    class_id: TcHandle,
+    rate: u64,
+    ceil: u64,
+) -> Vec<String> {
+    vec![
+        "class".to_string(),
+        "replace".to_string(),
+        "dev".to_string(),
+        interface_name.to_string(),
+        "classid".to_string(),
+        class_id.as_tc_string(),
+        "htb".to_string(),
+        "rate".to_string(),
+        format!("{}mbit", rate),
+        "ceil".to_string(),
+        format!("{}mbit", ceil),
+    ]
+}
+
+fn stormguard_restorations(
+    adjustments: &[StormGuardRestoreAdjustment],
+    restore_untracked: bool,
+    overrides: &HashMap<StormguardOverrideKey, StormguardOverrideValue>,
+) -> Vec<StormGuardRestoreAdjustment> {
+    let mut restorations: HashMap<StormguardOverrideKey, StormGuardRestoreAdjustment> =
+        HashMap::new();
+    if restore_untracked {
+        for adjustment in adjustments {
+            restorations.insert(
+                StormguardOverrideKey {
+                    interface: adjustment.interface_name.clone(),
+                    class: adjustment.class_id,
+                },
+                adjustment.clone(),
+            );
+        }
+    }
+    for (key, value) in overrides {
+        restorations.insert(
+            key.clone(),
+            StormGuardRestoreAdjustment {
+                interface_name: key.interface.clone(),
+                class_id: key.class,
+                planned_rate: value.planned_rate,
+                planned_ceil: value.planned_ceil,
+            },
         );
-        return;
     }
-    if overrides.is_empty() {
-        return;
+    restorations.into_values().collect()
+}
+
+fn reset_stormguard_adjustments(
+    adjustments: &[StormGuardRestoreAdjustment],
+    restore_untracked: bool,
+    overrides: &mut HashMap<StormguardOverrideKey, StormguardOverrideValue>,
+) -> Result<(), String> {
+    if overrides.is_empty() && (!restore_untracked || adjustments.is_empty()) {
+        overrides.clear();
+        debug!("Cleared inactive Bakery StormGuard replay state.");
+        return Ok(());
     }
+    if !MQ_CREATED.load(Relaxed) {
+        return Err("Bakery MQ setup has not completed".to_string());
+    }
+    let config = lqos_config::load_config()
+        .map_err(|error| format!("failed to load configuration: {error}"))?;
+    if let Some(reason) = live_tree_mutation_blocker_for_config(&config) {
+        return Err(format!("live queue mutation is blocked because {reason}"));
+    }
+
+    let restorations = stormguard_restorations(adjustments, restore_untracked, overrides);
+    invalidate_live_tc_snapshots();
+    let mut live_classes: HashMap<String, HashSet<TcHandle>> = HashMap::new();
     let mut commands = Vec::new();
-    for (key, rate) in overrides.iter() {
-        commands.push(vec![
-            "class".to_string(),
-            "replace".to_string(),
-            "dev".to_string(),
-            key.interface.clone(),
-            "classid".to_string(),
-            key.class.as_tc_string(),
-            "htb".to_string(),
-            "rate".to_string(),
-            format!("{}mbit", rate.saturating_sub(1)),
-            "ceil".to_string(),
-            format!("{}mbit", rate),
-        ]);
+    for restoration in restorations {
+        if !live_classes.contains_key(&restoration.interface_name) {
+            let snapshot = read_live_class_snapshot(&restoration.interface_name)?;
+            live_classes.insert(
+                restoration.interface_name.clone(),
+                snapshot.into_keys().collect(),
+            );
+        }
+        if live_classes
+            .get(&restoration.interface_name)
+            .is_some_and(|classes| classes.contains(&restoration.class_id))
+        {
+            commands.push(stormguard_htb_command(
+                &restoration.interface_name,
+                restoration.class_id,
+                restoration.planned_rate,
+                restoration.planned_ceil,
+            ));
+        } else {
+            debug!(
+                "StormGuard restore skipped absent class {} on {}.",
+                restoration.class_id.as_tc_string(),
+                restoration.interface_name
+            );
+        }
     }
-    let result = execute_in_memory(&commands, "replaying StormGuard overrides");
+    if !commands.is_empty() {
+        let result = execute_in_memory(&commands, "restoring StormGuard planned rates");
+        if !result.ok {
+            return Err(summarize_apply_result(
+                "restoring StormGuard planned rates",
+                &result,
+            ));
+        }
+    }
+
+    overrides.clear();
+    debug!("Restored planned StormGuard rates and cleared retained adjustments.");
+    Ok(())
+}
+
+fn apply_stormguard_circuit_adjustment(
+    circuit_hash: i64,
+    sqm_override: Option<String>,
+    circuits: &mut HashMap<i64, Arc<BakeryCommands>>,
+) -> Result<bool, String> {
+    let Some(existing) = circuits.get(&circuit_hash) else {
+        return Ok(false);
+    };
+    let command = circuit_with_sqm_override(existing.as_ref(), sqm_override)?;
+    if !MQ_CREATED.load(Relaxed) {
+        return Err("Bakery MQ setup has not completed".to_string());
+    }
+    let config = lqos_config::load_config()
+        .map_err(|error| format!("failed to load configuration: {error}"))?;
+    if let Some(reason) = live_tree_mutation_blocker_for_config(&config) {
+        return Err(format!("live queue mutation is blocked because {reason}"));
+    }
+    let commands = add_commands_for_circuit(&command, &config, ExecutionMode::Builder)
+        .filter(|commands| !commands.is_empty())
+        .ok_or_else(|| "StormGuard circuit adjustment produced no live tc commands".to_string())?;
+    let result = execute_in_memory(&commands, "applying StormGuard circuit SQM adjustment");
     if !result.ok {
-        push_bakery_event(
-            "stormguard_override_replay_failed",
-            "error",
-            summarize_apply_result("replaying StormGuard overrides", &result),
-        );
+        return Err(summarize_apply_result(
+            "applying StormGuard circuit SQM adjustment",
+            &result,
+        ));
     }
+    circuits.insert(circuit_hash, Arc::new(command));
+    Ok(true)
+}
+
+fn circuit_with_sqm_override(
+    existing: &BakeryCommands,
+    sqm_override: Option<String>,
+) -> Result<BakeryCommands, String> {
+    let mut command = existing.clone();
+    let BakeryCommands::AddCircuit {
+        sqm_override: current,
+        ..
+    } = &mut command
+    else {
+        return Err("existing Bakery circuit state is not AddCircuit".to_string());
+    };
+    *current = sqm_override;
+    Ok(command)
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
-    use lqos_config::Config;
+    use lqos_config::{Config, StormguardConfig};
     use std::sync::Mutex;
     use std::time::{SystemTime, UNIX_EPOCH};
 
     static TC_BYPASS_SETTER_CALLS: Mutex<Vec<bool>> = Mutex::new(Vec::new());
 
+    #[test]
+    fn stormguard_live_requires_enabled_non_dry_run_config() {
+        let mut config = Config::default();
+        assert!(!stormguard_live_enabled(&config));
+
+        config.stormguard = Some(StormguardConfig {
+            enabled: true,
+            dry_run: true,
+            ..StormguardConfig::default()
+        });
+        assert!(!stormguard_live_enabled(&config));
+
+        if let Some(stormguard) = &mut config.stormguard {
+            stormguard.dry_run = false;
+        }
+        assert!(stormguard_live_enabled(&config));
+    }
+
+    #[test]
+    fn stormguard_restore_command_uses_planned_rate() -> anyhow::Result<()> {
+        let command = stormguard_htb_command("eth0", TcHandle::from_string("1:2")?, 4, 10);
+        assert_eq!(
+            command,
+            [
+                "class", "replace", "dev", "eth0", "classid", "0x1:0x2", "htb", "rate", "4mbit",
+                "ceil", "10mbit"
+            ]
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn stormguard_override_retains_original_rate_pair() -> anyhow::Result<()> {
+        let key = StormguardOverrideKey {
+            interface: "eth0".to_string(),
+            class: TcHandle::from_string("1:2")?,
+        };
+        let mut overrides = HashMap::new();
+        record_stormguard_override(&mut overrides, key.clone(), 4, 10);
+        record_stormguard_override(&mut overrides, key.clone(), 99, 100);
+        let Some(value) = overrides.get(&key) else {
+            anyhow::bail!("missing StormGuard override");
+        };
+        assert_eq!(value.planned_rate, 4);
+        assert_eq!(value.planned_ceil, 10);
+        Ok(())
+    }
+
+    #[test]
+    fn stormguard_rejects_stale_tree_generation_before_live_mutation() {
+        let current_generation = stormguard_tree_generation();
+        let result = apply_stormguard_adjustment(
+            StormguardAdjustmentRequest {
+                tree_generation: current_generation.wrapping_add(1),
+                dry_run: false,
+                interface_name: "eth0".to_string(),
+                class_id: "1:2".to_string(),
+                new_rate: 5,
+                planned_rate: 10,
+                planned_ceil: 10,
+            },
+            &mut HashMap::new(),
+        );
+
+        let error = result.expect_err("stale generation must be rejected");
+        assert!(error.contains("generation"));
+    }
+
+    #[test]
+    fn stormguard_circuit_adjustment_reports_missing_bakery_state() {
+        let result =
+            apply_stormguard_circuit_adjustment(42, Some("cake".to_string()), &mut HashMap::new());
+        assert_eq!(result, Ok(false));
+    }
+
+    #[test]
+    fn stormguard_rebuild_gate_only_matches_changed_owned_site() {
+        let config = Arc::new(Config {
+            bridge: Some(lqos_config::BridgeConfig {
+                use_xdp_bridge: false,
+                to_internet: "internet0".to_string(),
+                to_network: "isp0".to_string(),
+                mtu: None,
+            }),
+            ..Config::default()
+        });
+        let changed_site = mk_add_site(42, 0x1_0000, 0x2_0000, 0x10);
+        let change_mode = SiteDiffResult::SpeedChanges {
+            changes: vec![changed_site.as_ref().clone()],
+        };
+        let (down_class, up_class) =
+            site_class_handles(changed_site.as_ref()).expect("site class handles");
+        let mut overrides = HashMap::new();
+        overrides.insert(
+            StormguardOverrideKey {
+                interface: config.isp_interface(),
+                class: down_class,
+            },
+            StormguardOverrideValue {
+                planned_rate: 10,
+                planned_ceil: 100,
+            },
+        );
+
+        assert!(stormguard_site_changes_intersect_ownership(
+            &change_mode,
+            &config,
+            &overrides
+        ));
+
+        overrides.clear();
+        overrides.insert(
+            StormguardOverrideKey {
+                interface: config.internet_interface(),
+                class: up_class,
+            },
+            StormguardOverrideValue {
+                planned_rate: 10,
+                planned_ceil: 100,
+            },
+        );
+        assert!(stormguard_site_changes_intersect_ownership(
+            &change_mode,
+            &config,
+            &overrides
+        ));
+
+        overrides.clear();
+        overrides.insert(
+            StormguardOverrideKey {
+                interface: config.isp_interface(),
+                class: TcHandle::from_u32(0x9_0009),
+            },
+            StormguardOverrideValue {
+                planned_rate: 10,
+                planned_ceil: 100,
+            },
+        );
+        assert!(!stormguard_site_changes_intersect_ownership(
+            &change_mode,
+            &config,
+            &overrides
+        ));
+    }
+
+    #[test]
+    fn stormguard_reset_prefers_successful_bakery_ownership() -> anyhow::Result<()> {
+        let class = TcHandle::from_string("1:2")?;
+        let supplied = [
+            StormGuardRestoreAdjustment {
+                interface_name: "eth0".to_string(),
+                class_id: class,
+                planned_rate: 99,
+                planned_ceil: 100,
+            },
+            StormGuardRestoreAdjustment {
+                interface_name: "eth0".to_string(),
+                class_id: TcHandle::from_string("1:3")?,
+                planned_rate: 7,
+                planned_ceil: 20,
+            },
+        ];
+        let key = StormguardOverrideKey {
+            interface: "eth0".to_string(),
+            class,
+        };
+        let mut overrides = HashMap::new();
+        record_stormguard_override(&mut overrides, key, 4, 10);
+
+        let restorations = stormguard_restorations(&supplied, true, &overrides);
+        assert_eq!(restorations.len(), 2);
+        let cached = restorations
+            .iter()
+            .find(|adjustment| adjustment.class_id == class)
+            .ok_or_else(|| anyhow::anyhow!("missing cached restoration"))?;
+        assert_eq!(cached.planned_rate, 4);
+        assert_eq!(cached.planned_ceil, 10);
+
+        let startup = stormguard_restorations(&supplied, true, &HashMap::new());
+        assert_eq!(startup.len(), 2);
+        assert!(stormguard_restorations(&supplied, false, &HashMap::new()).is_empty());
+        Ok(())
+    }
+
+    #[test]
+    fn stormguard_circuit_sqm_change_preserves_qdisc_handles() -> anyhow::Result<()> {
+        let existing = BakeryCommands::AddCircuit {
+            circuit_hash: 42,
+            circuit_name: Some("Circuit A".to_string()),
+            site_name: Some("Site A".to_string()),
+            parent_class_id: TcHandle::from_string("1:1")?,
+            up_parent_class_id: TcHandle::from_string("2:1")?,
+            class_minor: 3,
+            download_bandwidth_min: 5.0,
+            upload_bandwidth_min: 2.0,
+            download_bandwidth_max: 10.0,
+            upload_bandwidth_max: 4.0,
+            class_major: 1,
+            up_class_major: 2,
+            down_qdisc_handle: Some(0x9000),
+            up_qdisc_handle: Some(0x9001),
+            ip_addresses: "192.0.2.1".to_string(),
+            sqm_override: Some("cake".to_string()),
+        };
+        let updated = circuit_with_sqm_override(&existing, None).map_err(anyhow::Error::msg)?;
+        let BakeryCommands::AddCircuit {
+            down_qdisc_handle,
+            up_qdisc_handle,
+            sqm_override,
+            ..
+        } = updated
+        else {
+            anyhow::bail!("expected AddCircuit");
+        };
+        assert_eq!(down_qdisc_handle, Some(0x9000));
+        assert_eq!(up_qdisc_handle, Some(0x9001));
+        assert_eq!(sqm_override, None);
+        Ok(())
+    }
+
     fn recording_tc_bypass_setter(enabled: bool) -> anyhow::Result<()> {
         TC_BYPASS_SETTER_CALLS
             .lock()

src/rust/lqos_network_devices/src/lib.rs

@@ -23,6 +23,7 @@ use anyhow::{Context, Result};
 use lqos_config::{
     Config, ConfigShapedDevices, NetworkJson, TopologyShapingInputsFile,
 };
+use lqos_topology_compile::TopologyImportFile;
 use std::sync::Arc;
 use tracing::debug;
 
@@ -152,6 +153,16 @@ fn load_shaped_devices_for_config(config: &lqos_config::Config) -> Result<Loaded
     })
 }
 
+fn load_source_shaped_devices_for_config(config: &Config) -> Result<ConfigShapedDevices> {
+    if lqos_config::integration_ingress_enabled(config) {
+        return Ok(TopologyImportFile::load(config)?
+            .map(|topology_import| topology_import.into_imported_bundle().shaped_devices)
+            .unwrap_or_default());
+    }
+
+    ConfigShapedDevices::load_for_config(config).context("Unable to load ShapedDevices.csv")
+}
+
 /// Loads shaped-device configuration from disk.
 ///
 /// When integration-backed topology ingress is enabled, this prefers ready runtime shaping inputs,
@@ -161,6 +172,18 @@ pub fn load_shaped_devices() -> Result<ConfigShapedDevices> {
     Ok(load_shaped_devices_with_source()?.shaped)
 }
 
+/// Loads the operator or integration source shaped-device rows without runtime overrides.
+///
+/// Integration-ingress mode reads `topology_import.json`; manual mode reads
+/// `ShapedDevices.csv`. This deliberately bypasses ready runtime shaping inputs so callers can
+/// detect source settings hidden by an effective adaptive override.
+///
+/// Side effects: reads the configured source file from disk.
+pub fn load_source_shaped_devices() -> Result<ConfigShapedDevices> {
+    let config = lqos_config::load_config().context("Unable to load /etc/lqos.conf")?;
+    load_source_shaped_devices_for_config(config.as_ref())
+}
+
 pub(crate) fn load_shaped_devices_with_source() -> Result<LoadedShapedDevices> {
     let config = lqos_config::load_config().context("Unable to load /etc/lqos.conf")?;
     load_shaped_devices_for_config(config.as_ref())

src/rust/lqos_network_devices/src/tests.rs

@@ -1,6 +1,7 @@
 use crate::{
     DynamicCircuit, ShapedDevicesCatalog, load_shaped_devices_for_config,
-    resolve_parent_node_reference, runtime_inputs, with_network_json_write,
+    load_source_shaped_devices_for_config, resolve_parent_node_reference, runtime_inputs,
+    with_network_json_write,
 };
 use lqos_config::{
     CircuitAnchorsFile, Config, ConfigShapedDevices, NetworkJsonNode, ShapedDevice,
@@ -367,7 +368,7 @@ fn runtime_inputs_leave_runtime_fallback_circuits_unparented() {
 }
 
 #[test]
-fn load_shaped_devices_uses_topology_import_when_runtime_inputs_are_empty() {
+fn source_loader_bypasses_ready_runtime_inputs() {
     let _guard = TEST_LOCK.lock();
 
     let lqos_directory = unique_temp_dir("lqos-network-devices-runtime-empty");
@@ -384,10 +385,24 @@ fn load_shaped_devices_uses_topology_import_when_runtime_inputs_are_empty() {
         "csv-circuit",
         "192.0.2.10/32",
     );
+    let runtime_inputs = TopologyShapingInputsFile {
+        circuits: vec![TopologyShapingCircuitInput {
+            circuit_id: "runtime-circuit".to_string(),
+            circuit_name: "Runtime Circuit".to_string(),
+            effective_parent_node_name: "Runtime Parent".to_string(),
+            effective_parent_node_id: "runtime-parent".to_string(),
+            devices: vec![TopologyShapingDeviceInput {
+                device_id: "runtime-device".to_string(),
+                ..TopologyShapingDeviceInput::default()
+            }],
+            ..TopologyShapingCircuitInput::default()
+        }],
+        shaping_generation: "shape-1".to_string(),
+        ..TopologyShapingInputsFile::default()
+    };
     std::fs::write(
         &runtime_path,
-        serde_json::to_string_pretty(&TopologyShapingInputsFile::default())
-            .expect("empty shaping inputs should encode"),
+        serde_json::to_string_pretty(&runtime_inputs).expect("runtime shaping inputs should encode"),
     )
     .expect("runtime shaping inputs should write");
     let mut import_devices = ConfigShapedDevices::default();
@@ -432,7 +447,12 @@ fn load_shaped_devices_uses_topology_import_when_runtime_inputs_are_empty() {
     let loaded = load_shaped_devices_for_config(&config)
         .expect("preferred shaped-device source should load");
     assert_eq!(loaded.shaped.devices.len(), 1);
-    assert_eq!(loaded.shaped.devices[0].circuit_id, "import-circuit");
+    assert_eq!(loaded.shaped.devices[0].circuit_id, "runtime-circuit");
+
+    let source =
+        load_source_shaped_devices_for_config(&config).expect("raw integration source should load");
+    assert_eq!(source.devices.len(), 1);
+    assert_eq!(source.devices[0].circuit_id, "import-circuit");
 }
 
 #[test]

src/rust/lqos_queue_tracker/src/lib.rs

@@ -23,7 +23,8 @@ pub use interval::set_queue_refresh_interval;
 #[doc(hidden)]
 pub use queue_structure::{
     EFFECTIVE_CIRCUIT_RATES, EFFECTIVE_NODE_RATES, QUEUE_STRUCTURE,
-    QUEUE_STRUCTURE_CHANGED_STORMGUARD, QueueNode, QueueStructure, spawn_queue_structure_monitor,
+    QUEUE_STRUCTURE_CHANGED_STORMGUARD, QueueNode, QueueStructure, reload_queue_structure,
+    spawn_queue_structure_monitor,
 };
 pub use queue_types::deserialize_tc_tree; // Exported for the benchmarker
 pub use tracking::spawn_queue_monitor;

src/rust/lqos_queue_tracker/src/queue_structure/mod.rs

@@ -1,12 +1,12 @@
 mod queing_structure_json_monitor;
 mod queue_network;
 mod queue_node;
-pub use queing_structure_json_monitor::spawn_queue_structure_monitor;
 #[doc(hidden)]
 pub use queing_structure_json_monitor::{
     EFFECTIVE_CIRCUIT_RATES, EFFECTIVE_NODE_RATES, QUEUE_STRUCTURE,
     QUEUE_STRUCTURE_CHANGED_STORMGUARD, QueueStructure,
 };
+pub use queing_structure_json_monitor::{reload_queue_structure, spawn_queue_structure_monitor};
 use queue_network::QueueNetwork;
 #[doc(hidden)]
 pub use queue_node::QueueNode;

src/rust/lqos_queue_tracker/src/queue_structure/queing_structure_json_monitor.rs

@@ -1,5 +1,6 @@
 use crate::queue_structure::{
-    queue_network::QueueNetwork, queue_node::QueueNode, read_queueing_structure,
+    QueueStructureError, queue_network::QueueNetwork, queue_node::QueueNode,
+    read_queueing_structure,
 };
 use arc_swap::ArcSwap;
 use lqos_utils::file_watcher::FileWatcher;
@@ -26,17 +27,14 @@ static INITIAL_EFFECTIVE_RATES: Lazy<EffectiveRates> = Lazy::new(|| {
 ///
 /// This contains only named queue nodes that map cleanly back to authored network-tree
 /// entries. Circuit/device rows and generated placeholder nodes are intentionally excluded.
-pub static EFFECTIVE_NODE_RATES: Lazy<ArcSwap<HashMap<String, (f64, f64)>>> = Lazy::new(|| {
-    ArcSwap::new(Arc::new(INITIAL_EFFECTIVE_RATES.nodes.clone()))
-});
+pub static EFFECTIVE_NODE_RATES: Lazy<ArcSwap<HashMap<String, (f64, f64)>>> =
+    Lazy::new(|| ArcSwap::new(Arc::new(INITIAL_EFFECTIVE_RATES.nodes.clone())));
 /// Global effective circuit-rate overlay derived from `queuingStructure.json`.
 ///
 /// This contains the currently programmed circuit queue rates keyed by normalized circuit ID.
-pub static EFFECTIVE_CIRCUIT_RATES: Lazy<ArcSwap<HashMap<String, (f64, f64)>>> = Lazy::new(|| {
-    ArcSwap::new(Arc::new(INITIAL_EFFECTIVE_RATES.circuits.clone()))
-});
-/// Set to true when the queue structure changes. This is here rather than in StormGuard
-/// to avoid circular dependencies.
+pub static EFFECTIVE_CIRCUIT_RATES: Lazy<ArcSwap<HashMap<String, (f64, f64)>>> =
+    Lazy::new(|| ArcSwap::new(Arc::new(INITIAL_EFFECTIVE_RATES.circuits.clone())));
+/// Set when StormGuard should reconsider a non-live queue-plan snapshot.
 pub static QUEUE_STRUCTURE_CHANGED_STORMGUARD: AtomicBool = AtomicBool::new(false);
 
 #[allow(missing_docs)]
@@ -122,33 +120,39 @@ pub fn spawn_queue_structure_monitor() -> anyhow::Result<()> {
     Ok(())
 }
 
-fn update_queue_structure() {
+/// Reloads the queue structure from its generated JSON file.
+///
+/// This is exposed for consumers that must synchronize an in-memory snapshot with a
+/// successfully applied shaping-tree generation instead of waiting for the file watcher.
+pub fn reload_queue_structure() -> Result<(), QueueStructureError> {
     debug!("queueingStructure.json reload requested");
-    match read_queueing_structure() {
-        Ok(queues) => {
-            let effective_rates = build_effective_rates(&queues);
-            let new_queue_structure = QueueStructure {
-                maybe_queues: Some(queues),
-            };
-            QUEUE_STRUCTURE.store(Arc::new(new_queue_structure));
-            EFFECTIVE_NODE_RATES.store(Arc::new(effective_rates.nodes));
-            EFFECTIVE_CIRCUIT_RATES.store(Arc::new(effective_rates.circuits));
-            QUEUE_STRUCTURE_CHANGED_STORMGUARD.store(true, std::sync::atomic::Ordering::Relaxed);
-        }
-        Err(err) => {
-            if QUEUE_STRUCTURE.load().maybe_queues.is_some() {
-                warn!(
-                    "Failed to reload queuingStructure.json ({err:?}); preserving last-known-good snapshot"
-                );
-            } else {
-                warn!(
-                    "Failed to load queuingStructure.json ({err:?}); leaving queue structure unavailable"
-                );
-                QUEUE_STRUCTURE.store(Arc::new(QueueStructure { maybe_queues: None }));
-                EFFECTIVE_NODE_RATES.store(Arc::new(HashMap::new()));
-                EFFECTIVE_CIRCUIT_RATES.store(Arc::new(HashMap::new()));
-            }
+    let queues = read_queueing_structure()?;
+    let effective_rates = build_effective_rates(&queues);
+    let new_queue_structure = QueueStructure {
+        maybe_queues: Some(queues),
+    };
+    QUEUE_STRUCTURE.store(Arc::new(new_queue_structure));
+    EFFECTIVE_NODE_RATES.store(Arc::new(effective_rates.nodes));
+    EFFECTIVE_CIRCUIT_RATES.store(Arc::new(effective_rates.circuits));
+    Ok(())
+}
+
+fn update_queue_structure() {
+    if let Err(err) = reload_queue_structure() {
+        if QUEUE_STRUCTURE.load().maybe_queues.is_some() {
+            warn!(
+                "Failed to reload queuingStructure.json ({err:?}); preserving last-known-good snapshot"
+            );
+        } else {
+            warn!(
+                "Failed to load queuingStructure.json ({err:?}); leaving queue structure unavailable"
+            );
+            QUEUE_STRUCTURE.store(Arc::new(QueueStructure { maybe_queues: None }));
+            EFFECTIVE_NODE_RATES.store(Arc::new(HashMap::new()));
+            EFFECTIVE_CIRCUIT_RATES.store(Arc::new(HashMap::new()));
         }
+    } else {
+        QUEUE_STRUCTURE_CHANGED_STORMGUARD.store(true, std::sync::atomic::Ordering::Relaxed);
     }
 }
 

src/rust/lqos_stormguard/README.md

@@ -68,6 +68,7 @@ active_ping_timeout_seconds = 1.0
 
 You can list as many sites as you want in `targets`, or turn on `all_sites` and carve out exceptions with `exclude_sites`.
 `dry_run` is the recommended starting point while tuning the thresholds for a network.
+Disabling StormGuard, or switching a live deployment back to `dry_run`, restores its managed queues to their planned rates and clears StormGuard-owned adaptive overrides.
 
 ## How it works
 

src/rust/lqos_stormguard/src/adaptive_actions.rs

@@ -1,11 +1,13 @@
-use anyhow::{Result, anyhow};
+use anyhow::{Context, Result, anyhow};
 use lqos_bakery::BakeryCommands;
 use lqos_config::ShapedDevice;
 use lqos_overrides::{OverrideFile, OverrideLayer, OverrideStore};
-use lqos_queue_tracker::QUEUE_STRUCTURE;
 use lqos_utils::hash_to_i64;
 use std::collections::{BTreeMap, HashMap, HashSet};
+use std::time::{Duration, Instant};
+use tracing::warn;
 
+#[derive(Debug)]
 pub enum CircuitFallbackOutcome {
     Applied { persisted: bool },
     Cleared { persisted: bool },
@@ -25,6 +27,31 @@ pub struct PersistedCircuitFallback {
     pub devices: Vec<ShapedDevice>,
 }
 
+/// Waits asynchronously for an acknowledged Bakery operation.
+///
+/// Side effects: yields to the Tokio runtime while polling the synchronous reply channel.
+pub(crate) async fn wait_for_bakery_reply<T>(
+    reply_receiver: std::sync::mpsc::Receiver<Result<T, String>>,
+    operation: &str,
+) -> Result<T> {
+    let deadline = Instant::now() + Duration::from_secs(10);
+    loop {
+        match reply_receiver.try_recv() {
+            Ok(Ok(value)) => return Ok(value),
+            Ok(Err(error)) => return Err(anyhow!(error)),
+            Err(std::sync::mpsc::TryRecvError::Disconnected) => {
+                return Err(anyhow!("Bakery closed the {operation} reply channel"));
+            }
+            Err(std::sync::mpsc::TryRecvError::Empty) if Instant::now() >= deadline => {
+                return Err(anyhow!("timed out waiting for Bakery to {operation}"));
+            }
+            Err(std::sync::mpsc::TryRecvError::Empty) => {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        }
+    }
+}
+
 pub fn apply_site_override_updates(updates: &[SiteOverrideUpdate]) -> Result<bool> {
     if updates.is_empty() {
         return Ok(false);
@@ -64,11 +91,19 @@ pub fn apply_site_override_updates(updates: &[SiteOverrideUpdate]) -> Result<boo
     Ok(true)
 }
 
-pub fn apply_circuit_fallback(
+/// Removes every adaptive override owned by StormGuard.
+///
+/// Side effects: replaces the persisted StormGuard override layer with an empty layer.
+pub fn clear_stormguard_override_layer() -> Result<()> {
+    OverrideStore::save_layer(OverrideLayer::Stormguard, &OverrideFile::default())
+}
+
+pub async fn apply_circuit_fallback(
     circuit_id: &str,
     sqm_override: &str,
     persist: bool,
     dry_run: bool,
+    tree_generation: u64,
     bakery_sender: crossbeam_channel::Sender<BakeryCommands>,
 ) -> Result<CircuitFallbackOutcome> {
     let devices = load_devices_for_circuit(circuit_id)?;
@@ -88,18 +123,57 @@ pub fn apply_circuit_fallback(
         });
     }
 
+    let reply_receiver = apply_circuit_sqm_override_live(
+        circuit_id,
+        Some(sqm_override),
+        tree_generation,
+        bakery_sender.clone(),
+    )?;
+    if !wait_for_bakery_reply(reply_receiver, "apply StormGuard circuit fallback").await? {
+        return Ok(CircuitFallbackOutcome::Skipped {
+            reason: "Circuit is absent from Bakery state.".to_string(),
+        });
+    }
     let persisted = if persist {
-        set_devices_sqm_override(&devices, sqm_override)?
+        match set_devices_sqm_override(&devices, sqm_override) {
+            Ok(persisted) => persisted,
+            Err(persistence_error) => {
+                let rollback = apply_circuit_sqm_override_live(
+                    circuit_id,
+                    None,
+                    tree_generation,
+                    bakery_sender,
+                );
+                let rollback = match rollback {
+                    Ok(rollback_receiver) => wait_for_bakery_reply(
+                        rollback_receiver,
+                        "roll back unpersisted StormGuard circuit fallback",
+                    )
+                    .await
+                    .map(|_| ()),
+                    Err(error) => Err(error),
+                };
+                if let Err(rollback_error) = rollback {
+                    warn!(
+                        "StormGuard could not persist or roll back circuit fallback for {circuit_id}; retaining in-memory ownership: persistence={persistence_error}; rollback={rollback_error}"
+                    );
+                    return Ok(CircuitFallbackOutcome::Applied { persisted: false });
+                }
+                return Err(persistence_error).context(
+                    "failed to persist StormGuard circuit fallback; live change was rolled back",
+                );
+            }
+        }
     } else {
         false
     };
-    apply_circuit_sqm_override_live(circuit_id, &devices, Some(sqm_override), bakery_sender)?;
     Ok(CircuitFallbackOutcome::Applied { persisted })
 }
 
-pub fn clear_circuit_fallback(
+pub async fn clear_circuit_fallback(
     circuit_id: &str,
     dry_run: bool,
+    tree_generation: u64,
     bakery_sender: crossbeam_channel::Sender<BakeryCommands>,
 ) -> Result<CircuitFallbackOutcome> {
     let fallback = load_persisted_circuit_fallbacks()?
@@ -129,8 +203,15 @@ pub fn clear_circuit_fallback(
         .iter()
         .map(|d| d.device_id.clone())
         .collect();
+    let reply_receiver =
+        apply_circuit_sqm_override_live(circuit_id, None, tree_generation, bakery_sender)?;
+    if !wait_for_bakery_reply(reply_receiver, "clear StormGuard circuit fallback").await? {
+        let persisted = clear_device_overrides(&device_ids)?;
+        return Ok(CircuitFallbackOutcome::Skipped {
+            reason: format!("Circuit is absent from Bakery state; persisted={persisted}"),
+        });
+    }
     let persisted = clear_device_overrides(&device_ids)?;
-    apply_circuit_sqm_override_live(circuit_id, &fallback.devices, None, bakery_sender)?;
     Ok(CircuitFallbackOutcome::Cleared { persisted })
 }
 
@@ -169,6 +250,12 @@ fn load_devices_for_circuit(circuit_id: &str) -> Result<Vec<ShapedDevice>> {
 
 fn conflicting_sqm_owner(devices: &[ShapedDevice]) -> Result<Option<String>> {
     let device_ids: HashSet<&str> = devices.iter().map(|d| d.device_id.as_str()).collect();
+    let source_devices = lqos_network_devices::load_source_shaped_devices()?.devices;
+    if has_source_sqm_override(&device_ids, &source_devices) {
+        return Ok(Some(
+            "The circuit already has a source-configured SQM override.".to_string(),
+        ));
+    }
 
     let operator = OverrideStore::load_layer(OverrideLayer::Operator)?;
     if has_sqm_override_for_device_ids(&operator, &device_ids) {
@@ -187,6 +274,16 @@ fn conflicting_sqm_owner(devices: &[ShapedDevice]) -> Result<Option<String>> {
     Ok(None)
 }
 
+fn has_source_sqm_override(device_ids: &HashSet<&str>, source_devices: &[ShapedDevice]) -> bool {
+    source_devices.iter().any(|device| {
+        device_ids.contains(device.device_id.as_str())
+            && device
+                .sqm_override
+                .as_deref()
+                .is_some_and(|token| !token.trim().is_empty())
+    })
+}
+
 fn has_sqm_override_for_device_ids(overrides: &OverrideFile, device_ids: &HashSet<&str>) -> bool {
     overrides.circuit_adjustments().iter().any(|adj| {
         matches!(
@@ -253,89 +350,19 @@ fn clear_device_overrides(device_ids: &[String]) -> Result<bool> {
 
 fn apply_circuit_sqm_override_live(
     circuit_id: &str,
-    devices: &[ShapedDevice],
     sqm_override: Option<&str>,
+    tree_generation: u64,
     bakery_sender: crossbeam_channel::Sender<BakeryCommands>,
-) -> Result<()> {
-    let snapshot = QUEUE_STRUCTURE.load();
-    let Some(queues) = snapshot.maybe_queues.as_ref() else {
-        return Err(anyhow!("queueingStructure.json not loaded"));
-    };
-
-    let mut stack = Vec::new();
-    for queue in queues.iter() {
-        stack.push(queue);
-    }
-
-    let mut found = None;
-    while let Some(node) = stack.pop() {
-        if node.circuit_id.as_deref() == Some(circuit_id) && node.device_id.is_none() {
-            found = Some(node);
-            break;
-        }
-
-        for child in node.children.iter() {
-            stack.push(child);
-        }
-        for circuit in node.circuits.iter() {
-            stack.push(circuit);
-        }
-        for device in node.devices.iter() {
-            stack.push(device);
-        }
-    }
-
-    let Some(node) = found else {
-        return Err(anyhow!(
-            "circuit not found in queue structure: {circuit_id}"
-        ));
-    };
-
-    let class_minor = u16::try_from(node.class_minor)
-        .map_err(|_| anyhow!("class_minor too large: {}", node.class_minor))?;
-    let class_major = u16::try_from(node.class_major)
-        .map_err(|_| anyhow!("class_major too large: {}", node.class_major))?;
-    let up_class_major = u16::try_from(node.up_class_major)
-        .map_err(|_| anyhow!("up_class_major too large: {}", node.up_class_major))?;
-
-    bakery_sender.send(BakeryCommands::AddCircuit {
+) -> Result<std::sync::mpsc::Receiver<Result<bool, String>>> {
+    let (reply, reply_receiver) = std::sync::mpsc::channel();
+    bakery_sender.send(BakeryCommands::StormGuardCircuitAdjustment {
+        tree_generation,
         circuit_hash: hash_to_i64(circuit_id),
-        circuit_name: node
-            .circuit_name
-            .clone()
-            .or_else(|| Some(circuit_id.to_string())),
-        site_name: node.parent_node.clone(),
-        parent_class_id: node.parent_class_id,
-        up_parent_class_id: node.up_parent_class_id,
-        class_minor,
-        download_bandwidth_min: node.download_bandwidth_mbps_min as f32,
-        upload_bandwidth_min: node.upload_bandwidth_mbps_min as f32,
-        download_bandwidth_max: node.download_bandwidth_mbps as f32,
-        upload_bandwidth_max: node.upload_bandwidth_mbps as f32,
-        class_major,
-        up_class_major,
-        down_qdisc_handle: None,
-        up_qdisc_handle: None,
-        ip_addresses: ip_list(devices),
-        sqm_override: sqm_override.map(|value| value.to_string()),
+        sqm_override: sqm_override.map(str::to_string),
+        reply: Some(reply),
     })?;
 
-    Ok(())
-}
-
-fn ip_list(devices: &[ShapedDevice]) -> String {
-    let mut ips = Vec::new();
-    for dev in devices {
-        for (ip, prefix) in dev.ipv4.iter() {
-            ips.push(format!("{ip}/{prefix}"));
-        }
-        for (ip, prefix) in dev.ipv6.iter() {
-            ips.push(format!("{ip}/{prefix}"));
-        }
-    }
-    ips.sort();
-    ips.dedup();
-    ips.join(",")
+    Ok(reply_receiver)
 }
 
 fn group_circuit_fallbacks(
@@ -513,4 +540,15 @@ mod tests {
         );
         assert!(!grouped.contains_key("c2"));
     }
+
+    #[test]
+    fn source_sqm_change_is_visible_while_effective_device_has_stormguard_token() {
+        let mut effective = sample_device("dev1", "c1");
+        effective.sqm_override = Some("cake".to_string());
+        let mut source = sample_device("dev1", "c1");
+        source.sqm_override = Some("fq_codel".to_string());
+        let device_ids = HashSet::from([effective.device_id.as_str()]);
+
+        assert!(has_source_sqm_override(&device_ids, &[source]));
+    }
 }

src/rust/lqos_stormguard/src/config.rs

@@ -26,6 +26,8 @@ pub struct WatchingSite {
 pub struct WatchingSiteDependency {
     pub name: String,
     pub class_id: TcHandle,
+    pub original_min_download_mbps: u64,
+    pub original_min_upload_mbps: u64,
     pub original_max_download_mbps: u64,
     pub original_max_upload_mbps: u64,
 }

src/rust/lqos_stormguard/src/lib.rs

@@ -15,8 +15,9 @@ use lqos_config::NetworkJsonTransport;
 use lqos_probe::ProbeClient;
 use lqos_queue_tracker::QUEUE_STRUCTURE_CHANGED_STORMGUARD;
 use parking_lot::Mutex;
+use std::collections::HashSet;
 use std::time::Duration;
-use tracing::{debug, info};
+use tracing::{debug, info, warn};
 
 mod active_ping;
 mod adaptive_actions;
@@ -28,6 +29,55 @@ mod site_state;
 const READING_ACCUMULATOR_SIZE: usize = 15;
 const MOVING_AVERAGE_BUFFER_SIZE: usize = 15;
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum StormguardRuntimeMode {
+    Disabled,
+    DryRun,
+    Live,
+}
+
+fn runtime_mode(enabled: bool, dry_run: bool) -> StormguardRuntimeMode {
+    if !enabled {
+        StormguardRuntimeMode::Disabled
+    } else if dry_run {
+        StormguardRuntimeMode::DryRun
+    } else {
+        StormguardRuntimeMode::Live
+    }
+}
+
+fn requested_runtime_mode() -> Option<StormguardRuntimeMode> {
+    lqos_config::load_config().ok().map(|config| {
+        config
+            .stormguard
+            .as_ref()
+            .map(|stormguard| runtime_mode(stormguard.enabled, stormguard.dry_run))
+            .unwrap_or(StormguardRuntimeMode::Disabled)
+    })
+}
+
+fn active_runtime_mode(config: Option<&config::StormguardConfig>) -> StormguardRuntimeMode {
+    config
+        .map(|config| runtime_mode(true, config.dry_run))
+        .unwrap_or(StormguardRuntimeMode::Disabled)
+}
+
+fn requires_live_reset(
+    active_mode: StormguardRuntimeMode,
+    requested_mode: StormguardRuntimeMode,
+) -> bool {
+    active_mode == StormguardRuntimeMode::Live && requested_mode != StormguardRuntimeMode::Live
+}
+
+fn rebuild_occurred_since(observed: u64, current: u64, latest_rebuild: u64) -> bool {
+    observed < latest_rebuild && latest_rebuild <= current
+}
+
+fn clear_published_state() {
+    STORMGUARD_STATS.lock().clear();
+    STORMGUARD_DEBUG.lock().clear();
+}
+
 /// Globally accessible stormguard statistics
 pub static STORMGUARD_STATS: Mutex<Vec<(String, u64, u64)>> = Mutex::new(Vec::new());
 
@@ -50,6 +100,10 @@ pub async fn start_stormguard(
     let mut log_sender: Option<std::sync::mpsc::Sender<datalog::LogCommand>> = None;
     let mut site_state_tracker: Option<site_state::SiteStateTracker> = None;
     let mut active_ping = active_ping::ActivePingManager::new(probe_client);
+    let mut inactive_reconciled = false;
+    let mut live_reset_pending = false;
+    let mut observed_tree_generation = lqos_bakery::stormguard_tree_generation();
+    let mut retained_circuit_fallbacks = HashSet::new();
 
     // Main Cycle - use tokio interval instead of blocking TimerFd
     let mut interval = tokio::time::interval(Duration::from_secs(1));
@@ -58,25 +112,168 @@ pub async fn start_stormguard(
     loop {
         interval.tick().await;
 
-        // Check if queue structure has changed or if we need initial configuration
-        let queue_structure_changed =
-            QUEUE_STRUCTURE_CHANGED_STORMGUARD.swap(false, std::sync::atomic::Ordering::Relaxed);
+        let Some(requested_mode) = requested_runtime_mode() else {
+            warn!("StormGuard could not read the current runtime mode; retaining existing state.");
+            continue;
+        };
+        let active_mode = active_runtime_mode(config.as_ref());
+        let dry_run_plan_changed = QUEUE_STRUCTURE_CHANGED_STORMGUARD
+            .swap(false, std::sync::atomic::Ordering::Relaxed)
+            && active_mode == StormguardRuntimeMode::DryRun
+            && requested_mode == StormguardRuntimeMode::DryRun;
+        let current_tree_generation = lqos_bakery::stormguard_tree_generation();
+        let bakery_tree_changed = current_tree_generation != observed_tree_generation;
+        let bakery_tree_rebuilt = rebuild_occurred_since(
+            observed_tree_generation,
+            current_tree_generation,
+            lqos_bakery::stormguard_tree_rebuild_generation(),
+        );
+        if bakery_tree_changed {
+            if let Err(error) = lqos_queue_tracker::reload_queue_structure() {
+                warn!(
+                    "StormGuard could not load the queue structure for Bakery generation {current_tree_generation}; retrying: {error}"
+                );
+                continue;
+            }
+            if lqos_bakery::stormguard_tree_generation() != current_tree_generation {
+                debug!(
+                    "Bakery tree changed again while StormGuard refreshed its queue snapshot; retrying."
+                );
+                continue;
+            }
+            if bakery_tree_rebuilt
+                && let Err(error) =
+                    site_state::discard_bakery_adjustments(current_tree_generation, bakery.clone())
+                        .await
+            {
+                warn!("StormGuard rebuild reconciliation will retry: {error}");
+                continue;
+            }
+            inactive_reconciled = false;
+        }
+        if active_mode != StormguardRuntimeMode::Live {
+            observed_tree_generation = current_tree_generation;
+        }
+        if active_mode == StormguardRuntimeMode::Live && bakery_tree_changed {
+            if bakery_tree_rebuilt {
+                retained_circuit_fallbacks.clear();
+            } else if let Some(tracker) = &site_state_tracker {
+                retained_circuit_fallbacks.extend(tracker.active_circuit_fallbacks());
+            }
+            observed_tree_generation = current_tree_generation;
+            config = None;
+            site_state_tracker = None;
+            log_sender = None;
+            live_reset_pending = false;
+            clear_published_state();
+        }
+        if !bakery_tree_changed && requires_live_reset(active_mode, requested_mode) {
+            live_reset_pending = true;
+        }
+        if live_reset_pending {
+            let Some(tracker) = &mut site_state_tracker else {
+                warn!("StormGuard reset is pending without an active tracker; retrying.");
+                continue;
+            };
+            info!("StormGuard is restoring planned queue rates before reconfiguration.");
+            match tracker.reset_adjustments(bakery.clone()).await {
+                Ok(()) => {
+                    config = None;
+                    site_state_tracker = None;
+                    log_sender = None;
+                    inactive_reconciled = true;
+                    live_reset_pending = false;
+                    clear_published_state();
+                }
+                Err(error) => {
+                    warn!("StormGuard could not restore planned queue rates; retrying: {error}");
+                    continue;
+                }
+            }
+        } else if active_mode != requested_mode && active_mode != StormguardRuntimeMode::Disabled {
+            config = None;
+            site_state_tracker = None;
+            log_sender = None;
+            clear_published_state();
+        }
 
-        if config.is_none() || queue_structure_changed {
+        if requested_mode != StormguardRuntimeMode::Live
+            && !retained_circuit_fallbacks.is_empty()
+        {
+            if let Err(error) = site_state::clear_retained_circuit_fallbacks(
+                &retained_circuit_fallbacks,
+                current_tree_generation,
+                bakery.clone(),
+            )
+            .await
+            {
+                warn!("StormGuard retained circuit cleanup will retry: {error}");
+                continue;
+            }
+            retained_circuit_fallbacks.clear();
+        }
+
+        if requested_mode != StormguardRuntimeMode::Live && !inactive_reconciled {
+            match site_state::reconcile_inactive_state(bakery.clone()).await {
+                Ok(()) => inactive_reconciled = true,
+                Err(error) => {
+                    warn!("StormGuard inactive-state cleanup will retry: {error}");
+                    continue;
+                }
+            }
+        } else if requested_mode == StormguardRuntimeMode::Live {
+            inactive_reconciled = false;
+        }
+
+        if requested_mode == StormguardRuntimeMode::Disabled {
+            config = None;
+            site_state_tracker = None;
+            log_sender = None;
+            clear_published_state();
+            active_ping.reconfigure(None);
+            continue;
+        }
+
+        if config.is_none() || bakery_tree_changed || dry_run_plan_changed {
+            let configuration_generation = lqos_bakery::stormguard_tree_generation();
             // Try to (re)configure StormGuard
             match config::configure() {
                 Ok(new_config) => {
+                    if lqos_bakery::stormguard_tree_generation() != configuration_generation {
+                        debug!(
+                            "Bakery tree changed while StormGuard was configuring; retrying with the new generation."
+                        );
+                        config = None;
+                        site_state_tracker = None;
+                        continue;
+                    }
                     if new_config.is_empty() {
                         debug!("No StormGuard sites found in queue structure yet");
                         config = None;
+                        site_state_tracker = None;
+                        clear_published_state();
                     } else {
                         info!("StormGuard configuration loaded successfully");
                         // Initialize or reinitialize everything
                         if log_sender.is_none() {
                             log_sender = datalog::start_datalog(&new_config).ok();
                         }
-                        let mut tracker = site_state::SiteStateTracker::from_config(&new_config);
-                        tracker.replay_persisted_adjustments(&new_config, bakery.clone());
+                        let mut tracker = site_state::SiteStateTracker::from_config(
+                            &new_config,
+                            configuration_generation,
+                        );
+                        if let Err(error) = tracker
+                            .replay_persisted_adjustments(&new_config, bakery.clone())
+                            .await
+                        {
+                            warn!("StormGuard persisted adjustment replay will retry: {error}");
+                            config = None;
+                            site_state_tracker = None;
+                            continue;
+                        }
+                        tracker.retain_active_circuit_fallbacks(std::mem::take(
+                            &mut retained_circuit_fallbacks,
+                        ));
                         site_state_tracker = Some(tracker);
                         config = Some(new_config);
                     }
@@ -84,6 +281,8 @@ pub async fn start_stormguard(
                 Err(e) => {
                     debug!("StormGuard configuration not ready: {}", e);
                     config = None;
+                    site_state_tracker = None;
+                    clear_published_state();
                 }
             }
         }
@@ -113,8 +312,51 @@ pub async fn start_stormguard(
             if !recommendations.is_empty()
                 && let Some(sender) = &log_sender
             {
-                tracker.apply_recommendations(recommendations, cfg, sender.clone(), bakery.clone());
+                tracker
+                    .apply_recommendations(recommendations, cfg, sender.clone(), bakery.clone())
+                    .await;
             }
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        StormguardRuntimeMode, rebuild_occurred_since, requires_live_reset, runtime_mode,
+    };
+
+    #[test]
+    fn runtime_mode_tracks_disabled_dry_run_and_live_states() {
+        assert_eq!(runtime_mode(false, false), StormguardRuntimeMode::Disabled);
+        assert_eq!(runtime_mode(false, true), StormguardRuntimeMode::Disabled);
+        assert_eq!(runtime_mode(true, true), StormguardRuntimeMode::DryRun);
+        assert_eq!(runtime_mode(true, false), StormguardRuntimeMode::Live);
+    }
+
+    #[test]
+    fn live_mode_resets_only_when_leaving_live_operation() {
+        assert!(requires_live_reset(
+            StormguardRuntimeMode::Live,
+            StormguardRuntimeMode::Disabled,
+        ));
+        assert!(requires_live_reset(
+            StormguardRuntimeMode::Live,
+            StormguardRuntimeMode::DryRun,
+        ));
+        assert!(!requires_live_reset(
+            StormguardRuntimeMode::Live,
+            StormguardRuntimeMode::Live,
+        ));
+        assert!(!requires_live_reset(
+            StormguardRuntimeMode::DryRun,
+            StormguardRuntimeMode::Disabled,
+        ));
+    }
+
+    #[test]
+    fn incremental_generation_preserves_bakery_ownership() {
+        assert!(!rebuild_occurred_since(4, 5, 4));
+        assert!(rebuild_occurred_since(4, 6, 5));
+    }
+}

src/rust/lqos_stormguard/src/queue_structure.rs

@@ -46,6 +46,8 @@ pub fn find_queue_dependents(parent_name: &str) -> Result<Vec<WatchingSiteDepend
                 dependents.push(WatchingSiteDependency {
                     name: candidate.clone().name.unwrap_or_default(),
                     class_id: candidate.class_id,
+                    original_min_download_mbps: candidate.download_bandwidth_mbps_min,
+                    original_min_upload_mbps: candidate.upload_bandwidth_mbps_min,
                     original_max_download_mbps: candidate.download_bandwidth_mbps,
                     original_max_upload_mbps: candidate.upload_bandwidth_mbps,
                 });

src/rust/lqos_stormguard/src/site_state.rs

@@ -7,7 +7,8 @@ mod stormguard_state;
 use crate::active_ping::TimedRtt;
 use crate::adaptive_actions::{
     CircuitFallbackOutcome, SiteOverrideUpdate, apply_circuit_fallback,
-    apply_site_override_updates, clear_circuit_fallback, load_persisted_circuit_fallbacks,
+    apply_site_override_updates, clear_circuit_fallback, clear_stormguard_override_layer,
+    load_persisted_circuit_fallbacks, wait_for_bakery_reply,
 };
 use crate::config::StormguardConfig;
 use crate::datalog::LogCommand;
@@ -19,21 +20,25 @@ use crate::site_state::ring_buffer::RingBuffer;
 use crate::site_state::site::SiteState;
 use crate::site_state::stormguard_state::StormguardState;
 use crate::{MOVING_AVERAGE_BUFFER_SIZE, READING_ACCUMULATOR_SIZE};
+use anyhow::{Context, anyhow};
 use crossbeam_channel::Sender;
-use lqos_bakery::BakeryCommands;
+use lqos_bakery::{BakeryCommands, StormGuardRestoreAdjustment};
 use lqos_bus::{StormguardDebugDirection, StormguardDebugEntry, TcHandle};
 use lqos_config::NetworkJsonTransport;
+use lqos_overrides::{NetworkAdjustment, OverrideFile, OverrideLayer, OverrideStore};
 use lqos_queue_tracker::QUEUE_STRUCTURE;
 use std::collections::{HashMap, HashSet};
 use std::time::{Duration, Instant};
 use tracing::{debug, info, warn};
 
 pub struct SiteStateTracker {
+    tree_generation: u64,
     sites: HashMap<String, SiteState>,
     active_circuit_fallbacks: HashSet<String>,
 }
 
 struct CircuitQueueRecommendationContext<'a> {
+    tree_generation: u64,
     active_circuit_fallbacks: &'a mut HashSet<String>,
     site: &'a mut SiteState,
     config: &'a StormguardConfig,
@@ -45,8 +50,266 @@ struct CircuitQueueRecommendationContext<'a> {
     bakery_sender: Sender<BakeryCommands>,
 }
 
+struct HtbChangeRequest {
+    tree_generation: u64,
+    dry_run: bool,
+    interface_name: String,
+    class_handle: TcHandle,
+    new_rate: u64,
+    planned_rate: u64,
+    planned_ceil: u64,
+}
+
+async fn reset_bakery_adjustments(
+    tree_generation: u64,
+    adjustments: Vec<StormGuardRestoreAdjustment>,
+    restore_untracked: bool,
+    bakery_sender: Sender<BakeryCommands>,
+) -> anyhow::Result<()> {
+    let (reply_sender, reply_receiver) = std::sync::mpsc::channel();
+    bakery_sender
+        .send(BakeryCommands::ResetStormGuardAdjustments {
+            tree_generation,
+            adjustments,
+            restore_untracked,
+            reply: Some(reply_sender),
+        })
+        .context("failed to send StormGuard reset command to Bakery")?;
+
+    wait_for_bakery_reply(reply_receiver, "reset StormGuard adjustments").await
+}
+
+/// Discards class ownership cached against the previous shaping tree.
+///
+/// Side effects: clears Bakery's in-process StormGuard adjustment cache after acknowledgement.
+pub(crate) async fn discard_bakery_adjustments(
+    tree_generation: u64,
+    bakery_sender: Sender<BakeryCommands>,
+) -> anyhow::Result<()> {
+    let (reply_sender, reply_receiver) = std::sync::mpsc::channel();
+    bakery_sender
+        .send(BakeryCommands::DiscardStormGuardAdjustments {
+            tree_generation,
+            reply: Some(reply_sender),
+        })
+        .context("failed to send StormGuard discard command to Bakery")?;
+    wait_for_bakery_reply(reply_receiver, "discard stale StormGuard ownership").await
+}
+
+/// Clears circuit fallbacks retained across an incremental tracker reload.
+///
+/// Side effects: applies acknowledged Bakery circuit updates and removes any matching persisted
+/// StormGuard device overrides for each retained circuit.
+pub(crate) async fn clear_retained_circuit_fallbacks(
+    circuit_ids: &HashSet<String>,
+    tree_generation: u64,
+    bakery_sender: Sender<BakeryCommands>,
+) -> anyhow::Result<()> {
+    for circuit_id in circuit_ids {
+        match clear_circuit_fallback(
+            circuit_id,
+            false,
+            tree_generation,
+            bakery_sender.clone(),
+        )
+        .await?
+        {
+            CircuitFallbackOutcome::Cleared { .. } | CircuitFallbackOutcome::Skipped { .. } => {}
+            outcome => {
+                return Err(anyhow!(
+                    "unexpected retained circuit cleanup outcome for {circuit_id}: {outcome:?}"
+                ));
+            }
+        }
+    }
+    Ok(())
+}
+
+/// Clears stale StormGuard replay and persistence state while StormGuard is inactive.
+///
+/// Side effects: asks Bakery to discard retained HTB replay entries, then empties the persisted
+/// StormGuard override layer. The persisted layer is unchanged if Bakery cannot acknowledge.
+pub async fn reconcile_inactive_state(bakery_sender: Sender<BakeryCommands>) -> anyhow::Result<()> {
+    let tree_generation = lqos_bakery::stormguard_tree_generation();
+    let override_snapshot = OverrideStore::load_layer(OverrideLayer::Stormguard)?;
+    let adjustments = persisted_restore_adjustments(&override_snapshot)?;
+    if lqos_bakery::stormguard_tree_generation() != tree_generation {
+        return Err(anyhow!(
+            "Bakery shaping tree changed while preparing inactive StormGuard reconciliation"
+        ));
+    }
+    reset_bakery_adjustments(tree_generation, adjustments, true, bakery_sender.clone()).await?;
+    let circuit_ids: Vec<String> = load_persisted_circuit_fallbacks()?.into_keys().collect();
+    for circuit_id in circuit_ids {
+        let outcome =
+            clear_circuit_fallback(&circuit_id, false, tree_generation, bakery_sender.clone())
+                .await;
+        let outcome = match outcome {
+            Ok(outcome) => outcome,
+            Err(error) => {
+                restore_override_snapshot(
+                    &override_snapshot,
+                    "after inactive circuit cleanup failed",
+                )?;
+                return Err(error);
+            }
+        };
+        match outcome {
+            CircuitFallbackOutcome::Cleared { .. } | CircuitFallbackOutcome::Skipped { .. } => {}
+            outcome => {
+                restore_override_snapshot(
+                    &override_snapshot,
+                    "after unexpected inactive circuit cleanup outcome",
+                )?;
+                return Err(anyhow!(
+                    "unexpected inactive reset outcome for circuit {circuit_id}: {outcome:?}"
+                ));
+            }
+        }
+    }
+    ensure_generation_or_restore_overrides(
+        tree_generation,
+        &override_snapshot,
+        "before inactive StormGuard persistence cleanup",
+    )?;
+    clear_stormguard_override_layer().context("failed to clear inactive StormGuard overrides")?;
+    ensure_generation_or_restore_overrides(
+        tree_generation,
+        &override_snapshot,
+        "during inactive StormGuard persistence cleanup",
+    )?;
+    Ok(())
+}
+
+fn ensure_generation_or_restore_overrides(
+    expected_generation: u64,
+    snapshot: &OverrideFile,
+    context: &str,
+) -> anyhow::Result<()> {
+    if lqos_bakery::stormguard_tree_generation() == expected_generation {
+        return Ok(());
+    }
+    restore_override_snapshot(snapshot, &format!("after tree changed {context}"))?;
+    Err(anyhow!("Bakery shaping tree changed {context}"))
+}
+
+fn restore_override_snapshot(snapshot: &OverrideFile, context: &str) -> anyhow::Result<()> {
+    OverrideStore::save_layer(OverrideLayer::Stormguard, snapshot)
+        .with_context(|| format!("failed to restore StormGuard ownership {context}"))
+}
+
+fn persisted_restore_adjustments(
+    overrides: &OverrideFile,
+) -> anyhow::Result<Vec<StormGuardRestoreAdjustment>> {
+    let site_adjustments: Vec<_> = overrides
+        .network_adjustments()
+        .iter()
+        .filter_map(|adjustment| match adjustment {
+            NetworkAdjustment::AdjustSiteSpeed {
+                site_name,
+                download_bandwidth_mbps,
+                upload_bandwidth_mbps,
+                ..
+            } => Some((
+                site_name.clone(),
+                *download_bandwidth_mbps,
+                *upload_bandwidth_mbps,
+            )),
+            _ => None,
+        })
+        .collect();
+    if site_adjustments.is_empty() {
+        return Ok(Vec::new());
+    }
+
+    let config = lqos_config::load_config()?;
+    let queue_snapshot = QUEUE_STRUCTURE.load();
+    let queues = queue_snapshot
+        .maybe_queues
+        .as_ref()
+        .ok_or_else(|| anyhow!("queue structure is unavailable for inactive StormGuard reset"))?;
+    let mut adjustments: HashMap<(String, TcHandle), (u64, u64)> = HashMap::new();
+
+    for (site_name, download_target, upload_target) in site_adjustments {
+        let Some(queue) = queues
+            .iter()
+            .find(|queue| queue.name.as_deref() == Some(site_name.as_str()))
+        else {
+            warn!(
+                "Skipping stale persisted StormGuard site {site_name}: its queue no longer exists"
+            );
+            continue;
+        };
+        let dependencies = crate::queue_structure::find_queue_dependents(&site_name)?;
+        for (direction, target) in [
+            (RecommendationDirection::Download, download_target),
+            (RecommendationDirection::Upload, upload_target),
+        ] {
+            let Some(target) = target else {
+                continue;
+            };
+            let interface = match direction {
+                RecommendationDirection::Download => config.isp_interface(),
+                RecommendationDirection::Upload => config.internet_interface(),
+            };
+            let planned_rate = match direction {
+                RecommendationDirection::Download => queue.download_bandwidth_mbps_min,
+                RecommendationDirection::Upload => queue.upload_bandwidth_mbps_min,
+            };
+            let planned_ceil = match direction {
+                RecommendationDirection::Download => queue.download_bandwidth_mbps,
+                RecommendationDirection::Upload => queue.upload_bandwidth_mbps,
+            };
+            adjustments.insert(
+                (interface.clone(), queue.class_id),
+                (planned_rate, planned_ceil),
+            );
+            for dependency in &dependencies {
+                let (dependency_rate, dependency_ceil) = match direction {
+                    RecommendationDirection::Download => (
+                        dependency.original_min_download_mbps,
+                        dependency.original_max_download_mbps,
+                    ),
+                    RecommendationDirection::Upload => (
+                        dependency.original_min_upload_mbps,
+                        dependency.original_max_upload_mbps,
+                    ),
+                };
+                if dependency_ceil >= target.max(0.0) as u64 {
+                    adjustments.insert(
+                        (interface.clone(), dependency.class_id),
+                        (dependency_rate, dependency_ceil),
+                    );
+                }
+            }
+        }
+    }
+
+    Ok(adjustments
+        .into_iter()
+        .map(
+            |((interface_name, class_id), (planned_rate, planned_ceil))| {
+                StormGuardRestoreAdjustment {
+                    interface_name,
+                    class_id,
+                    planned_rate,
+                    planned_ceil,
+                }
+            },
+        )
+        .collect())
+}
+
 impl SiteStateTracker {
-    pub fn from_config(config: &StormguardConfig) -> Self {
+    pub(crate) fn active_circuit_fallbacks(&self) -> HashSet<String> {
+        self.active_circuit_fallbacks.clone()
+    }
+
+    pub(crate) fn retain_active_circuit_fallbacks(&mut self, circuit_ids: HashSet<String>) {
+        self.active_circuit_fallbacks.extend(circuit_ids);
+    }
+
+    pub fn from_config(config: &StormguardConfig, tree_generation: u64) -> Self {
         let mut sites = HashMap::new();
         for (name, site) in &config.sites {
             let delay_probe = matches!(
@@ -97,47 +360,50 @@ impl SiteStateTracker {
             );
         }
         Self {
+            tree_generation,
             sites,
             active_circuit_fallbacks: HashSet::new(),
         }
     }
 
-    pub fn replay_persisted_adjustments(
+    pub async fn replay_persisted_adjustments(
         &mut self,
         config: &StormguardConfig,
         bakery_sender: Sender<BakeryCommands>,
-    ) {
+    ) -> anyhow::Result<()> {
         if config.dry_run {
-            return;
+            return Ok(());
         }
 
         let Some(queues) = &QUEUE_STRUCTURE.load().maybe_queues else {
             info!("No queue structure - cannot replay StormGuard adjustments");
-            return;
+            return Ok(());
         };
 
-        match load_persisted_circuit_fallbacks() {
-            Ok(fallbacks) => {
-                for (circuit_id, fallback) in fallbacks {
+        let fallbacks = load_persisted_circuit_fallbacks()?;
+        for (circuit_id, fallback) in fallbacks {
+            match apply_circuit_fallback(
+                &circuit_id,
+                &fallback.sqm_override,
+                false,
+                false,
+                self.tree_generation,
+                bakery_sender.clone(),
+            )
+            .await?
+            {
+                CircuitFallbackOutcome::Applied { .. } => {
                     self.active_circuit_fallbacks.insert(circuit_id.clone());
-                    if let Err(e) = apply_circuit_fallback(
-                        &circuit_id,
-                        &fallback.sqm_override,
-                        false,
-                        false,
-                        bakery_sender.clone(),
-                    ) {
-                        warn!(
-                            "Failed to replay persisted StormGuard circuit fallback for {}: {}",
-                            circuit_id, e
-                        );
-                    }
+                }
+                CircuitFallbackOutcome::Skipped { reason } => warn!(
+                    "Skipped persisted StormGuard circuit fallback for {circuit_id}: {reason}"
+                ),
+                outcome => {
+                    return Err(anyhow!(
+                        "unexpected persisted circuit replay outcome for {circuit_id}: {outcome:?}"
+                    ));
                 }
             }
-            Err(e) => warn!(
-                "Failed to load persisted StormGuard circuit fallbacks: {}",
-                e
-            ),
         }
 
         for (name, site) in &self.sites {
@@ -167,23 +433,101 @@ impl SiteStateTracker {
                     "Replaying persisted StormGuard {} override for {}: {} -> {}",
                     direction, name, planned_rate, current_rate
                 );
-                Self::apply_dependents(
-                    &site.config,
-                    direction,
-                    current_rate,
-                    config,
-                    &interface_name,
-                    bakery_sender.clone(),
-                );
-                Self::apply_htb_change(
+                for (class_id, planned_rate, planned_ceil, rate) in
+                    Self::dependent_adjustments(&site.config, direction, current_rate)
+                {
+                    Self::apply_htb_change_confirmed(
+                        config,
+                        &interface_name,
+                        class_id,
+                        rate,
+                        (planned_rate, planned_ceil),
+                        self.tree_generation,
+                        bakery_sender.clone(),
+                    )
+                    .await?;
+                }
+                Self::apply_htb_change_confirmed(
                     config,
                     &interface_name,
                     queue.class_id,
                     current_rate,
+                    (Self::queue_minimum_rate(queue, direction), planned_rate),
+                    self.tree_generation,
                     bakery_sender.clone(),
-                );
+                )
+                .await?;
             }
         }
+        Ok(())
+    }
+
+    /// Restores StormGuard-managed classes and circuit fallbacks to their planned state.
+    ///
+    /// Side effects: applies live Bakery commands and empties the persisted StormGuard override
+    /// layer. Persisted and replay state is cleared only after the live HTB reset succeeds.
+    pub async fn reset_adjustments(
+        &mut self,
+        bakery_sender: Sender<BakeryCommands>,
+    ) -> anyhow::Result<()> {
+        let override_snapshot = OverrideStore::load_layer(OverrideLayer::Stormguard)?;
+        reset_bakery_adjustments(
+            self.tree_generation,
+            Vec::new(),
+            false,
+            bakery_sender.clone(),
+        )
+        .await?;
+
+        let active_circuit_fallbacks: Vec<String> =
+            self.active_circuit_fallbacks.iter().cloned().collect();
+        for circuit_id in active_circuit_fallbacks {
+            match clear_circuit_fallback(
+                &circuit_id,
+                false,
+                self.tree_generation,
+                bakery_sender.clone(),
+            )
+            .await
+            {
+                Ok(CircuitFallbackOutcome::Cleared { .. }) => {}
+                Ok(CircuitFallbackOutcome::Skipped { reason }) => {
+                    warn!(
+                        "StormGuard circuit fallback for {circuit_id} no longer owns the live SQM setting: {reason}"
+                    );
+                }
+                Ok(outcome) => {
+                    restore_override_snapshot(
+                        &override_snapshot,
+                        "after unexpected circuit cleanup outcome",
+                    )?;
+                    return Err(anyhow!(
+                        "unexpected reset outcome for StormGuard circuit fallback {circuit_id}: {outcome:?}"
+                    ));
+                }
+                Err(error) => {
+                    restore_override_snapshot(&override_snapshot, "after circuit cleanup failed")?;
+                    return Err(error).with_context(|| {
+                        format!("failed to clear StormGuard circuit fallback {circuit_id}")
+                    });
+                }
+            }
+        }
+
+        ensure_generation_or_restore_overrides(
+            self.tree_generation,
+            &override_snapshot,
+            "before StormGuard persistence cleanup",
+        )?;
+        clear_stormguard_override_layer()
+            .context("failed to clear persisted StormGuard adjustments")?;
+        ensure_generation_or_restore_overrides(
+            self.tree_generation,
+            &override_snapshot,
+            "during StormGuard persistence cleanup",
+        )?;
+        self.active_circuit_fallbacks.clear();
+        Ok(())
     }
 
     pub fn read_new_tick_data(
@@ -443,7 +787,7 @@ impl SiteStateTracker {
             .collect()
     }
 
-    pub fn apply_recommendations(
+    pub async fn apply_recommendations(
         &mut self,
         recommendations: Vec<(Recommendation, String)>,
         config: &StormguardConfig,
@@ -487,6 +831,7 @@ impl SiteStateTracker {
             // Circuit queues host qdiscs; prefer the TreeGuard-style fallback path.
             if let Some(circuit_id) = queue.circuit_id.as_deref() {
                 Self::handle_circuit_queue_recommendation(CircuitQueueRecommendationContext {
+                    tree_generation: self.tree_generation,
                     active_circuit_fallbacks: &mut self.active_circuit_fallbacks,
                     site,
                     config,
@@ -496,7 +841,8 @@ impl SiteStateTracker {
                     cooldown_secs,
                     log_sender: &log_sender,
                     bakery_sender: bakery_sender.clone(),
-                });
+                })
+                .await;
                 continue;
             }
 
@@ -536,6 +882,7 @@ impl SiteStateTracker {
                     new_rate,
                     config,
                     &interface_name,
+                    self.tree_generation,
                     bakery_sender.clone(),
                 );
                 Self::apply_htb_change(
@@ -543,6 +890,11 @@ impl SiteStateTracker {
                     &interface_name,
                     class_handle,
                     new_rate,
+                    (
+                        Self::queue_minimum_rate(queue, recommendation.direction),
+                        Self::planned_rate(&site.config, recommendation.direction),
+                    ),
+                    self.tree_generation,
                     bakery_sender.clone(),
                 );
                 Self::enter_cooldown(
@@ -570,6 +922,7 @@ impl SiteStateTracker {
                 new_rate,
                 config,
                 &interface_name,
+                self.tree_generation,
                 bakery_sender.clone(),
             );
             Self::apply_htb_change(
@@ -577,6 +930,11 @@ impl SiteStateTracker {
                 &interface_name,
                 class_handle,
                 new_rate,
+                (
+                    Self::queue_minimum_rate(queue, recommendation.direction),
+                    Self::planned_rate(&site.config, recommendation.direction),
+                ),
+                self.tree_generation,
                 bakery_sender.clone(),
             );
             pending_site_updates.insert(site.config.name.clone());
@@ -614,8 +972,9 @@ impl SiteStateTracker {
         }
     }
 
-    fn handle_circuit_queue_recommendation(ctx: CircuitQueueRecommendationContext<'_>) {
+    async fn handle_circuit_queue_recommendation(ctx: CircuitQueueRecommendationContext<'_>) {
         let CircuitQueueRecommendationContext {
+            tree_generation,
             active_circuit_fallbacks,
             site,
             config,
@@ -646,8 +1005,11 @@ impl SiteStateTracker {
                         &config.circuit_fallback_sqm,
                         config.circuit_fallback_persist,
                         config.dry_run,
+                        tree_generation,
                         bakery_sender,
-                    ) {
+                    )
+                    .await
+                    {
                         Ok(outcome) => outcome,
                         Err(e) => {
                             warn!(
@@ -661,7 +1023,14 @@ impl SiteStateTracker {
                     }
                 }
                 RecommendationAction::Increase | RecommendationAction::IncreaseFast => {
-                    match clear_circuit_fallback(circuit_id, config.dry_run, bakery_sender) {
+                    match clear_circuit_fallback(
+                        circuit_id,
+                        config.dry_run,
+                        tree_generation,
+                        bakery_sender,
+                    )
+                    .await
+                    {
                         Ok(outcome) => outcome,
                         Err(e) => {
                             warn!(
@@ -756,6 +1125,16 @@ impl SiteStateTracker {
         }
     }
 
+    fn queue_minimum_rate(
+        queue: &lqos_queue_tracker::QueueNode,
+        direction: RecommendationDirection,
+    ) -> u64 {
+        match direction {
+            RecommendationDirection::Download => queue.download_bandwidth_mbps_min,
+            RecommendationDirection::Upload => queue.upload_bandwidth_mbps_min,
+        }
+    }
+
     fn interface_name(config: &StormguardConfig, direction: RecommendationDirection) -> String {
         match direction {
             RecommendationDirection::Download => config.download_interface.clone(),
@@ -808,30 +1187,55 @@ impl SiteStateTracker {
         new_rate: u64,
         config: &StormguardConfig,
         interface_name: &str,
+        tree_generation: u64,
         bakery_sender: Sender<BakeryCommands>,
     ) {
-        for dependent in &site.dependent_nodes {
-            let max_rate = match direction {
-                RecommendationDirection::Download => dependent.original_max_download_mbps,
-                RecommendationDirection::Upload => dependent.original_max_upload_mbps,
-            };
-            if max_rate < new_rate {
-                continue;
-            }
+        for (class_id, planned_rate, planned_ceil, rate) in
+            Self::dependent_adjustments(site, direction, new_rate)
+        {
             info!(
-                "Applying rate change to dependent {}: {} -> {}",
-                dependent.name, max_rate, new_rate
+                "Applying rate change to dependent class {}: {} Mbps",
+                class_id.as_tc_string(),
+                rate
             );
             Self::apply_htb_change(
                 config,
                 interface_name,
-                dependent.class_id,
-                new_rate,
+                class_id,
+                rate,
+                (planned_rate, planned_ceil),
+                tree_generation,
                 bakery_sender.clone(),
             );
         }
     }
 
+    fn dependent_adjustments(
+        site: &crate::config::WatchingSite,
+        direction: RecommendationDirection,
+        target_rate: u64,
+    ) -> Vec<(TcHandle, u64, u64, u64)> {
+        site.dependent_nodes
+            .iter()
+            .filter_map(|dependent| {
+                let original_rate = match direction {
+                    RecommendationDirection::Download => dependent.original_max_download_mbps,
+                    RecommendationDirection::Upload => dependent.original_max_upload_mbps,
+                };
+                let original_min_rate = match direction {
+                    RecommendationDirection::Download => dependent.original_min_download_mbps,
+                    RecommendationDirection::Upload => dependent.original_min_upload_mbps,
+                };
+                (original_rate >= target_rate).then_some((
+                    dependent.class_id,
+                    original_min_rate,
+                    original_rate,
+                    target_rate,
+                ))
+            })
+            .collect()
+    }
+
     fn site_override_update_from_state(site: &SiteState) -> SiteOverrideUpdate {
         SiteOverrideUpdate {
             site_name: site.config.name.clone(),
@@ -878,14 +1282,19 @@ impl SiteStateTracker {
         interface_name: &str,
         class_handle: TcHandle,
         new_rate: u64,
+        planned_rates: (u64, u64),
+        tree_generation: u64,
         bakery_sender: Sender<BakeryCommands>,
     ) {
-        if let Err(e) = bakery_sender.send(BakeryCommands::StormGuardAdjustment {
-            dry_run: config.dry_run,
-            interface_name: interface_name.to_owned(),
-            class_id: class_handle.as_tc_string(),
+        let request = Self::htb_change_request(
+            config,
+            interface_name,
+            class_handle,
             new_rate,
-        }) {
+            planned_rates,
+            tree_generation,
+        );
+        if let Err(e) = Self::send_htb_change(request, bakery_sender, None) {
             warn!("Failed to send StormGuard adjustment command: {}", e);
         }
 
@@ -923,13 +1332,73 @@ impl SiteStateTracker {
         //     }
         // }
     }
+
+    async fn apply_htb_change_confirmed(
+        config: &StormguardConfig,
+        interface_name: &str,
+        class_handle: TcHandle,
+        new_rate: u64,
+        planned_rates: (u64, u64),
+        tree_generation: u64,
+        bakery_sender: Sender<BakeryCommands>,
+    ) -> anyhow::Result<()> {
+        let request = Self::htb_change_request(
+            config,
+            interface_name,
+            class_handle,
+            new_rate,
+            planned_rates,
+            tree_generation,
+        );
+        let (reply_sender, reply_receiver) = std::sync::mpsc::channel();
+        Self::send_htb_change(request, bakery_sender, Some(reply_sender))?;
+        wait_for_bakery_reply(reply_receiver, "replay persisted StormGuard class change").await
+    }
+
+    fn htb_change_request(
+        config: &StormguardConfig,
+        interface_name: &str,
+        class_handle: TcHandle,
+        new_rate: u64,
+        planned_rates: (u64, u64),
+        tree_generation: u64,
+    ) -> HtbChangeRequest {
+        HtbChangeRequest {
+            tree_generation,
+            dry_run: config.dry_run,
+            interface_name: interface_name.to_owned(),
+            class_handle,
+            new_rate,
+            planned_rate: planned_rates.0,
+            planned_ceil: planned_rates.1,
+        }
+    }
+
+    fn send_htb_change(
+        request: HtbChangeRequest,
+        bakery_sender: Sender<BakeryCommands>,
+        reply: Option<std::sync::mpsc::Sender<Result<(), String>>>,
+    ) -> anyhow::Result<()> {
+        bakery_sender
+            .send(BakeryCommands::StormGuardAdjustment {
+                tree_generation: request.tree_generation,
+                dry_run: request.dry_run,
+                interface_name: request.interface_name,
+                class_id: request.class_handle.as_tc_string(),
+                new_rate: request.new_rate,
+                planned_rate: request.planned_rate,
+                planned_ceil: request.planned_ceil,
+                reply,
+            })
+            .context("failed to send StormGuard class change to Bakery")
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::config::StormguardConfig as RuntimeStormguardConfig;
-    use crate::config::WatchingSite;
+    use crate::config::{WatchingSite, WatchingSiteDependency};
     use lqos_config::StormguardStrategy;
     use std::collections::HashMap;
 
@@ -1023,6 +1492,44 @@ mod tests {
         assert_eq!(update.upload_bandwidth_mbps, None);
     }
 
+    #[test]
+    fn dependent_adjustment_carries_original_directional_rate_pair() -> anyhow::Result<()> {
+        let mut site = site_state(50, 50, 100, 100).config;
+        site.dependent_nodes.push(WatchingSiteDependency {
+            name: "Dependent A".to_string(),
+            class_id: TcHandle::from_string("1:2")?,
+            original_min_download_mbps: 7,
+            original_min_upload_mbps: 3,
+            original_max_download_mbps: 37,
+            original_max_upload_mbps: 19,
+        });
+        let download =
+            SiteStateTracker::dependent_adjustments(&site, RecommendationDirection::Download, 15);
+        let upload =
+            SiteStateTracker::dependent_adjustments(&site, RecommendationDirection::Upload, 15);
+        assert_eq!(download, [(TcHandle::from_string("1:2")?, 7, 37, 15)]);
+        assert_eq!(upload, [(TcHandle::from_string("1:2")?, 3, 19, 15)]);
+        Ok(())
+    }
+
+    #[test]
+    fn dependent_adjustment_skips_classes_below_target_rate() -> anyhow::Result<()> {
+        let mut site = site_state(50, 50, 100, 100).config;
+        site.dependent_nodes.push(WatchingSiteDependency {
+            name: "Dependent A".to_string(),
+            class_id: TcHandle::from_string("1:2")?,
+            original_min_download_mbps: 7,
+            original_min_upload_mbps: 3,
+            original_max_download_mbps: 37,
+            original_max_upload_mbps: 19,
+        });
+        assert!(
+            SiteStateTracker::dependent_adjustments(&site, RecommendationDirection::Download, 40,)
+                .is_empty()
+        );
+        Ok(())
+    }
+
     #[test]
     fn skipped_circuit_outcomes_do_not_enter_cooldown() {
         assert!(!SiteStateTracker::circuit_outcome_enters_cooldown(&false));