observe: key migration

ui/src/api.ts

@@ -270,6 +270,7 @@ export function fetchFleetDevices(opts?: {
   presence?: string;
   known?: string;
   agentId?: string;
+  visibility?: "operator" | "all";
   limit?: number;
 }): Promise<FleetDevice[]> {
   const params = new URLSearchParams();
@@ -279,6 +280,7 @@ export function fetchFleetDevices(opts?: {
   }
   if (opts?.known && opts.known !== "all") params.set("known", opts.known);
   if (opts?.agentId) params.set("agent_id", opts.agentId);
+  if (opts?.visibility) params.set("visibility", opts.visibility);
   params.set("limit", String(opts?.limit ?? 500));
   return request<FleetDevice[]>(
     `/api/v1/control/fleet/devices?${params.toString()}`,

wakey-control-plane/src/api/control/fleet.rs

@@ -17,12 +17,13 @@ use crate::state::{
     KnownDeviceSummary,
 };
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Default, Deserialize)]
 pub struct ListFleetDevicesQuery {
     pub query: Option<String>,
     pub presence: Option<String>,
     pub known: Option<String>,
     pub agent_id: Option<String>,
+    pub visibility: Option<String>,
     pub limit: Option<usize>,
 }
 
@@ -307,6 +308,7 @@ pub async fn wake_fleet_device(
         presence: None,
         known: None,
         agent_id: None,
+        visibility: Some("all".into()),
         limit: Some(1000),
     };
     let devices = load_fleet_devices(&state, &query).await.map_err(|err| {
@@ -506,8 +508,14 @@ fn filter_fleet_devices(devices: &mut Vec<FleetDevice>, query: &ListFleetDevices
     let presence = normalize_filter(query.presence.as_deref());
     let known = normalize_filter(query.known.as_deref());
     let agent_id = normalize_filter(query.agent_id.as_deref());
+    let visibility = normalize_filter(query.visibility.as_deref());
 
     devices.retain(|device| {
+        if visibility.as_deref().unwrap_or("operator") != "all"
+            && fleet_device_is_operator_noise(device)
+        {
+            return false;
+        }
         if let Some(presence) = presence.as_deref()
             && presence != "all"
             && device.presence != presence
@@ -551,6 +559,15 @@ fn filter_fleet_devices(devices: &mut Vec<FleetDevice>, query: &ListFleetDevices
     });
 }
 
+fn fleet_device_is_operator_noise(device: &FleetDevice) -> bool {
+    device.known_device.is_none()
+        && device.macs.is_empty()
+        && device.hostnames.is_empty()
+        && device.recommended_route.is_none()
+        && device.ips.is_empty()
+        && device.presence == "offline"
+}
+
 fn observation_group_key(
     observation: &AgentDeviceObservation,
     context: &FleetBuildContext,
@@ -1091,6 +1108,74 @@ mod tests {
         assert!(!devices[0].route_candidates[0].wakeable);
     }
 
+    #[test]
+    fn unknown_ip_only_remove_observation_is_hidden() {
+        let mut offline = observation("agent-a", None, Some("192.168.1.2"), 20);
+        offline.kind = "neigh".into();
+        offline.hostname = None;
+        offline.last_action = "remove".into();
+
+        let mut devices = build_fleet_devices(Vec::new(), vec![offline], &context(&["agent-a"]));
+
+        assert_eq!(devices.len(), 1);
+        assert_eq!(devices[0].display_name, "(unknown device)");
+        filter_fleet_devices(&mut devices, &ListFleetDevicesQuery::default());
+        assert!(devices.is_empty());
+    }
+
+    #[test]
+    fn visibility_all_keeps_unknown_ip_only_remove_observation() {
+        let mut offline = observation("agent-a", None, Some("192.168.1.2"), 20);
+        offline.kind = "neigh".into();
+        offline.hostname = None;
+        offline.last_action = "remove".into();
+
+        let mut devices = build_fleet_devices(Vec::new(), vec![offline], &context(&["agent-a"]));
+
+        filter_fleet_devices(
+            &mut devices,
+            &ListFleetDevicesQuery {
+                visibility: Some("all".into()),
+                ..Default::default()
+            },
+        );
+        assert_eq!(devices.len(), 1);
+        assert_eq!(devices[0].display_name, "(unknown device)");
+    }
+
+    #[test]
+    fn known_ip_only_remove_observation_is_kept() {
+        let known = KnownDevice {
+            device_id: "dev-1".into(),
+            display_name: "lda".into(),
+            pinned: true,
+            created_at_unix: 1,
+            updated_at_unix: 1,
+            notes: None,
+            identifiers: vec![DeviceIdentifier {
+                identifier_key: "ip:192.168.1.2".into(),
+                device_id: "dev-1".into(),
+                kind: "ip".into(),
+                value: "192.168.1.2".into(),
+                created_at_unix: 1,
+            }],
+        };
+        let mut ctx = context(&["agent-a"]);
+        ctx.identifier_map
+            .insert("ip:192.168.1.2".into(), known_device_summary(&known));
+        let mut offline = observation("agent-a", None, Some("192.168.1.2"), 20);
+        offline.kind = "neigh".into();
+        offline.hostname = None;
+        offline.last_action = "remove".into();
+
+        let devices = build_fleet_devices(vec![known], vec![offline], &ctx);
+
+        assert_eq!(devices.len(), 1);
+        assert_eq!(devices[0].device_key, "known:dev-1");
+        assert_eq!(devices[0].display_name, "lda");
+        assert!(devices[0].ips.contains(&"192.168.1.2".to_string()));
+    }
+
     #[test]
     fn inventory_result_maps_to_stored_observations() {
         let observations = inventory_result_to_observations(serde_json::json!({

wakey-linux/src/dhcp.rs

@@ -179,4 +179,61 @@ mod tests {
 
         let _ = tokio::fs::remove_file(observation_path).await;
     }
+
+    #[tokio::test]
+    #[serial]
+    async fn neighbor_observation_migrates_coarse_mac_key_to_mac_ip_pair() {
+        let observation_path = temp_file("neighbor-observations-migrate");
+        let _observation_guard = EnvGuard::set(OBSERVATION_STORE_ENV, &observation_path);
+        let mut neighbors = std::collections::BTreeMap::new();
+        neighbors.insert(
+            "mac:aa:bb:cc:dd:ee:ff".to_string(),
+            ObservedNeighbor {
+                key: "mac:aa:bb:cc:dd:ee:ff".to_string(),
+                mac: Some("aa:bb:cc:dd:ee:ff".to_string()),
+                ip: None,
+                first_seen_unix: 1,
+                last_seen_unix: 1,
+                last_action: "add".to_string(),
+            },
+        );
+        let fixture = LocalObservationStore {
+            dhcp_clients: Default::default(),
+            neighbors,
+        };
+        tokio::fs::write(
+            &observation_path,
+            serde_json::to_string(&fixture).expect("fixture should serialize"),
+        )
+        .await
+        .expect("fixture should write");
+
+        observe_neighbor_event(
+            "update",
+            Some("aa:bb:cc:dd:ee:ff".parse().expect("mac should parse")),
+            Some("192.168.1.2".parse().expect("ip should parse")),
+        )
+        .await
+        .expect("observation should write");
+
+        let store = load_observation_store()
+            .await
+            .expect("observation store should read");
+        assert!(!store.neighbors.contains_key("mac:aa:bb:cc:dd:ee:ff"));
+        let row = store
+            .neighbors
+            .get("mac:aa:bb:cc:dd:ee:ff:ip:192.168.1.2")
+            .expect("coarse key should migrate to pair key");
+        assert_eq!(row.first_seen_unix, 1);
+        assert_eq!(row.mac.as_deref(), Some("aa:bb:cc:dd:ee:ff"));
+        assert_eq!(
+            row.ip
+                .expect("ip should be carried into migrated row")
+                .to_string(),
+            "192.168.1.2"
+        );
+        assert_eq!(row.last_action, "update");
+
+        let _ = tokio::fs::remove_file(observation_path).await;
+    }
 }

wakey-linux/src/dhcp/observations.rs

@@ -230,26 +230,19 @@ pub async fn observe_neighbor_event(
     let mac = mac.map(|value| value.to_string().to_ascii_lowercase());
     let mut store = load_observation_store().await.unwrap_or_default();
     let mut changed = false;
-    store
+    let alias_keys = neighbor_observation_alias_keys(mac.as_deref(), ip, &key);
+    let mut row = store
         .neighbors
-        .entry(key.clone())
+        .remove(&key)
-        .and_modify(|row| {
+        .or_else(|| {
-            if row.mac != mac
+            alias_keys
-                || row.ip != ip
+                .iter()
-                || row.last_action != action
+                .find_map(|alias| store.neighbors.remove(alias))
-                || row.last_seen_unix != now
-            {
-                row.mac = mac.clone();
-                row.ip = ip;
-                row.last_action = action.clone();
-                row.last_seen_unix = now;
-                changed = true;
-            }
         })
-        .or_insert_with(|| {
+        .unwrap_or_else(|| {
             changed = true;
             ObservedNeighbor {
-                key,
+                key: key.clone(),
                 mac: mac.clone(),
                 ip,
                 first_seen_unix: now,
@@ -257,6 +250,25 @@ pub async fn observe_neighbor_event(
                 last_action: action.clone(),
             }
         });
+    for alias in alias_keys {
+        if store.neighbors.remove(&alias).is_some() {
+            changed = true;
+        }
+    }
+    if row.key != key
+        || row.mac != mac
+        || row.ip != ip
+        || row.last_action != action
+        || row.last_seen_unix != now
+    {
+        row.key = key.clone();
+        row.mac = mac.clone();
+        row.ip = ip;
+        row.last_action = action.clone();
+        row.last_seen_unix = now;
+        changed = true;
+    }
+    store.neighbors.insert(key, row);
     if let (Some(mac), Some(ip)) = (mac.as_deref(), ip)
         && !is_offline_neighbor_action(&action)
     {
@@ -296,6 +308,22 @@ fn neighbor_observation_key(mac: Option<MacAddr>, ip: Option<IpAddr>) -> Option<
     }
 }
 
+fn neighbor_observation_alias_keys(
+    mac: Option<&str>,
+    ip: Option<IpAddr>,
+    primary_key: &str,
+) -> Vec<String> {
+    let mut keys = Vec::with_capacity(2);
+    if let Some(mac) = mac {
+        keys.push(format!("mac:{mac}"));
+    }
+    if let Some(ip) = ip {
+        keys.push(format!("ip:{ip}"));
+    }
+    keys.retain(|key| key != primary_key);
+    keys
+}
+
 fn mark_replaced_neighbor_ips_removed(
     store: &mut LocalObservationStore,
     mac: &str,