fix: hotplug-driven inventory and phantom device-card suppression

crates/openlogi-agent-core/Cargo.toml

@@ -15,6 +15,7 @@ openlogi-core = { path = "../openlogi-core" }
 openlogi-inject = { path = "../openlogi-inject" }
 openlogi-hid = { path = "../openlogi-hid" }
 openlogi-hook = { path = "../openlogi-hook" }
+futures-lite = { workspace = true }
 serde = { workspace = true }
 tarpc = { version = "0.37", features = ["serde1", "serde-transport", "serde-transport-bincode", "tokio1"] }
 interprocess = { workspace = true }

crates/openlogi-agent-core/src/orchestrator.rs

@@ -10,7 +10,7 @@
 //! [`DpiCycleState::capabilities`] stays `None` and presets cycle at their raw
 //! (still valid) values — exactly the GUI's "window never opened" behaviour.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashSet};
 use std::sync::atomic::{AtomicI32, Ordering};
 use std::sync::{Arc, RwLock};
 
@@ -80,6 +80,9 @@ pub struct Orchestrator {
     /// set/route/online state looks identical across the sleep gap, so the
     /// next refresh re-applies volatile settings to every online device.
     reapply_all_next_refresh: bool,
+    /// Config keys of devices first sighted last refresh, due one confirming
+    /// re-apply: the first write can race the device's own boot and be lost.
+    reapply_followup: HashSet<String>,
     shared: SharedRuntime,
 }
 
@@ -117,6 +120,7 @@ impl Orchestrator {
             current_app: None,
             inventory: InventoryState::Pending,
             reapply_all_next_refresh: false,
+            reapply_followup: HashSet::new(),
             shared,
         };
         orch.rebuild();
@@ -200,7 +204,11 @@ impl Orchestrator {
         // (new route), a wake from device sleep (offline→online), or — via the
         // flag — a system wake where none of those are observable.
         let reapply_all = std::mem::take(&mut self.reapply_all_next_refresh);
-        for idx in reapply_targets(&self.devices, &devices, reapply_all) {
+        let followup = std::mem::take(&mut self.reapply_followup);
+        let (targets, next_followup) =
+            plan_reapply(&self.devices, &devices, &followup, reapply_all);
+        self.reapply_followup = next_followup;
+        for idx in targets {
             self.reapply_volatile_settings(&devices[idx]);
         }
         let changed = devices.len() != self.devices.len()
@@ -433,6 +441,36 @@ fn reapply_targets(prev: &[AgentDevice], next: &[AgentDevice], reapply_all: bool
         .collect()
 }
 
+/// Plan this refresh's volatile-settings writes: the [`reapply_targets`] set
+/// plus one confirming re-apply for devices first sighted last refresh, and
+/// the follow-up keys to confirm next refresh.
+fn plan_reapply(
+    prev: &[AgentDevice],
+    next: &[AgentDevice],
+    followup: &HashSet<String>,
+    reapply_all: bool,
+) -> (Vec<usize>, HashSet<String>) {
+    let mut targets = reapply_targets(prev, next, reapply_all);
+    let next_followup = targets
+        .iter()
+        .filter(|&&idx| {
+            let id = stable_id(&next[idx]);
+            !prev.iter().any(|p| stable_id(p) == id)
+        })
+        .map(|&idx| next[idx].config_key.clone())
+        .collect();
+    for (idx, dev) in next.iter().enumerate() {
+        if dev.online
+            && dev.route.is_some()
+            && followup.contains(&dev.config_key)
+            && !targets.contains(&idx)
+        {
+            targets.push(idx);
+        }
+    }
+    (targets, next_followup)
+}
+
 /// Index of the selected device: the one whose `config_key` matches the saved
 /// selection, else the first. `build_devices` sorts by the same canonical key
 /// the GUI carousel uses, so "the first" is the same physical device in both
@@ -455,7 +493,7 @@ fn write_value<T>(lock: &RwLock<T>, value: T, name: &str) {
 
 #[cfg(test)]
 mod tests {
-    use super::{AgentDevice, InventoryHealth, Orchestrator, reapply_targets};
+    use super::{AgentDevice, InventoryHealth, Orchestrator, plan_reapply, reapply_targets};
     use openlogi_core::config::Config;
     use openlogi_hid::DeviceRoute;
 
@@ -526,6 +564,52 @@ mod tests {
         assert_eq!(reapply_targets(&prev, &next, true), vec![0]);
     }
 
+    #[test]
+    fn plan_reapply_confirms_a_first_sighting_once() {
+        use std::collections::HashSet;
+        // First sighting: applied now, queued for one confirming re-apply.
+        let (targets, followup) = plan_reapply(&[], &[dev("a", 1, true)], &HashSet::new(), false);
+        assert_eq!(targets, vec![0]);
+        assert_eq!(followup, HashSet::from(["a".to_string()]));
+        // Next refresh: the confirming apply fires, then the queue drains.
+        let prev = [dev("a", 1, true)];
+        let (targets, followup) = plan_reapply(&prev, &prev, &followup, false);
+        assert_eq!(targets, vec![0]);
+        assert!(followup.is_empty());
+        // Steady state after that: nothing.
+        let (targets, _) = plan_reapply(&prev, &prev, &followup, false);
+        assert!(targets.is_empty());
+    }
+
+    #[test]
+    fn plan_reapply_transitions_are_not_queued_for_confirmation() {
+        use std::collections::HashSet;
+        // A wake from device sleep re-applies once — the device was already
+        // booted, so no confirming write is queued.
+        let (targets, followup) = plan_reapply(
+            &[dev("a", 1, false)],
+            &[dev("a", 1, true)],
+            &HashSet::new(),
+            false,
+        );
+        assert_eq!(targets, vec![0]);
+        assert!(followup.is_empty());
+    }
+
+    #[test]
+    fn plan_reapply_skips_a_followup_that_went_offline() {
+        use std::collections::HashSet;
+        let prev = [dev("a", 1, true)];
+        let (targets, followup) = plan_reapply(
+            &prev,
+            &[dev("a", 1, false)],
+            &HashSet::from(["a".to_string()]),
+            false,
+        );
+        assert!(targets.is_empty());
+        assert!(followup.is_empty());
+    }
+
     /// An *empty* snapshot still flips the health to `Ready`: the watcher only
     /// forwards completed enumerations, so "checked and found nothing" must not
     /// be reported as "still scanning" — that's the whole distinction the

crates/openlogi-agent-core/src/watchers/inventory.rs

@@ -1,19 +1,20 @@
-//! Polling HID inventory watcher.
+//! HID inventory watcher: periodic polling, woken early by hotplug events.
 //!
 //! Spawns a dedicated OS thread with a one-shot tokio runtime that calls
 //! `openlogi_hid::enumerate` every `period` and forwards each completed
 //! snapshot over an unbounded mpsc to the agent's select loop, which applies
 //! it via `Orchestrator::refresh_inventory`.
 //!
-//! Polling beats hot-plug event registration on simplicity: HID transport
-//! crates ship different listener APIs across platforms, and `async-hid 0.4`
-//! does not expose any. A 2 s tick is cheap (one HID enumerate per cycle ≤
-//! a few hundred milliseconds) and matches the human-perceptible reconnect
-//! latency budget in PLAN.md.
+//! An OS hotplug event (`openlogi_hid::watch_hotplug`) cuts the wait short so
+//! a just-plugged device is probed — and gets its persisted settings applied —
+//! within a settle delay instead of a full period. The periodic tick stays as
+//! the reconciliation pass (battery refresh, missed events, platforms where
+//! the hotplug stream is unavailable).
 
 use std::thread;
 use std::time::{Duration, SystemTime};
 
+use futures_lite::StreamExt as _;
 use openlogi_core::device::DeviceInventory;
 use tokio::sync::mpsc;
 use tracing::{debug, info, warn};
@@ -25,6 +26,10 @@ use tracing::{debug, info, warn};
 /// live inventory.
 const INITIAL_FAILURE_LIMIT: u8 = 3;
 
+/// Pause between a hotplug event and the early enumerate, so a just-connected
+/// node finishes registering with the OS before the probe opens it.
+const HOTPLUG_SETTLE: Duration = Duration::from_millis(400);
+
 /// Wall-clock slack past `period` before a late tick is read as a sleep/wake
 /// gap. Generously above the worst honest iteration (period + a fully
 /// timed-out probe pass), so only a genuine suspend trips it; a rare false
@@ -126,6 +131,16 @@ pub fn spawn(period: Duration) -> mpsc::UnboundedReceiver<InventoryEvent> {
             let mut enumerator = openlogi_hid::Enumerator::default();
             let mut state = WatchState::default();
             let mut last_tick = SystemTime::now();
+            // `block_on` installs runtime context so a backend that registers an
+            // `AsyncFd` (Linux udev) fails as a catchable `Err`, not a panic that
+            // would take down the whole watcher thread.
+            let mut hotplug = match rt.block_on(async { openlogi_hid::watch_hotplug() }) {
+                Ok(stream) => Some(stream),
+                Err(e) => {
+                    warn!(error = ?e, "hotplug watch unavailable — polling only");
+                    None
+                }
+            };
             loop {
                 // A tick arriving far past its period means the system slept;
                 // `duration_since` errs when the clock stepped backwards, in
@@ -147,7 +162,38 @@ pub fn spawn(period: Duration) -> mpsc::UnboundedReceiver<InventoryEvent> {
                     debug!("inventory watcher receiver dropped — exiting");
                     return;
                 }
-                thread::sleep(period);
+                let stream_alive = rt.block_on(async {
+                    let Some(stream) = hotplug.as_mut() else {
+                        tokio::time::sleep(period).await;
+                        return true;
+                    };
+                    tokio::select! {
+                        () = tokio::time::sleep(period) => true,
+                        event = stream.next() => match event {
+                            Some(event) => {
+                                debug!(?event, "hotplug event — enumerating early");
+                                tokio::time::sleep(HOTPLUG_SETTLE).await;
+                                // Drain the burst so one enumerate covers every node that just
+                                // arrived; a `None` here is the stream closing, so report it now
+                                // rather than after a spurious extra enumerate next tick.
+                                let mut alive = true;
+                                while let Some(drained) =
+                                    futures_lite::future::poll_once(stream.next()).await
+                                {
+                                    if drained.is_none() {
+                                        alive = false;
+                                        break;
+                                    }
+                                }
+                                alive
+                            }
+                            None => false,
+                        },
+                    }
+                });
+                if !stream_alive && hotplug.take().is_some() {
+                    warn!("hotplug stream ended — falling back to pure polling");
+                }
             }
         });
     if let Err(e) = spawn_result {