fmt + add and wire observation retention

what ever new thing could add stuff in CONTEXT.md and docs/adr/ hmmmmmmmmmmmmmmmmmmmmmm

wakey-agent/src/cli.rs

@@ -98,6 +98,10 @@ pub struct InitConfigArgs {
     #[arg(long)]
     pub agent_token: Option<String>,
 
+    /// Days to keep local hook observation rows since last seen. Zero disables pruning.
+    #[arg(long)]
+    pub observation_retention_days: Option<u64>,
+
     /// Replace an existing config file.
     #[arg(long)]
     pub force: bool,

wakey-agent/src/config.rs

@@ -11,6 +11,7 @@ const WAKEY_OBSERVATION_STORE_ENV: &str = "WAKEY_OBSERVATION_STORE";
 const DEFAULT_DHCP_LEASES_PATH: &str = "/tmp/dhcp.leases";
 const DEFAULT_MAC_NAME_CACHE_PATH: &str = "/tmp/wakey_mac_names.json";
 const DEFAULT_OBSERVATION_STORE_PATH: &str = "/tmp/wakey_observations.json";
+pub const DEFAULT_OBSERVATION_RETENTION_DAYS: u64 = 30;
 
 #[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct AgentConfig {
@@ -23,6 +24,8 @@ pub struct AgentConfig {
     pub reconnect_max_ms: u64,
     #[serde(default = "default_observation_sync_interval_seconds")]
     pub observation_sync_interval_seconds: u64,
+    #[serde(default = "default_observation_retention_days")]
+    pub observation_retention_days: u64,
     #[serde(default = "default_pid_file")]
     pub pid_file: PathBuf,
     #[serde(default = "default_dhcp_leases_path")]
@@ -45,6 +48,10 @@ impl fmt::Debug for AgentConfig {
                 "observation_sync_interval_seconds",
                 &self.observation_sync_interval_seconds,
             )
+            .field(
+                "observation_retention_days",
+                &self.observation_retention_days,
+            )
             .field("pid_file", &self.pid_file)
             .field("dhcp_leases_path", &self.dhcp_leases_path)
             .field("mac_name_cache_path", &self.mac_name_cache_path)
@@ -65,6 +72,10 @@ const fn default_observation_sync_interval_seconds() -> u64 {
     60
 }
 
+const fn default_observation_retention_days() -> u64 {
+    DEFAULT_OBSERVATION_RETENTION_DAYS
+}
+
 fn default_pid_file() -> PathBuf {
     DEFAULT_PID_FILE.into()
 }
@@ -183,6 +194,7 @@ mod tests {
             reconnect_base_ms: 123,
             reconnect_max_ms: 456,
             observation_sync_interval_seconds: 7,
+            observation_retention_days: 3,
             pid_file: "/tmp/test-wakey-agent.pid".into(),
             dhcp_leases_path: "/tmp/test-dhcp.leases".into(),
             mac_name_cache_path: "/tmp/test-names.json".into(),

wakey-agent/src/dispatch.rs

@@ -64,6 +64,15 @@ async fn dispatch_devs(req: DevsRequest) -> Result<CommandResult> {
 }
 
 async fn dispatch_inventory(req: InventoryRequest, config: &AgentConfig) -> Result<CommandResult> {
+    Ok(CommandResult::Inventory(
+        inventory_for_config(req, config).await?,
+    ))
+}
+
+pub async fn inventory_for_config(
+    req: InventoryRequest,
+    config: &AgentConfig,
+) -> Result<wakey_core::DeviceInventory> {
     let query = req.into_inventory_query();
     let neighbors = wakey::wakey_linux::devices::query_neighbors(&query).await?;
     let leases = wakey::wakey_linux::dhcp::read_dhcp_leases_with_names_from_paths(
@@ -72,6 +81,20 @@ async fn dispatch_inventory(req: InventoryRequest, config: &AgentConfig) -> Resu
         &config.mac_name_cache_path,
     )
     .await?;
+    match wakey::wakey_linux::observations::prune_stale_observations_from_path(
+        &config.observation_store_path,
+        config.observation_retention_days,
+    )
+    .await
+    {
+        Ok(removed) if removed > 0 => {
+            debug!(removed, "pruned stale local hook observations");
+        }
+        Ok(_) => {}
+        Err(err) => {
+            warn!(error = %err, "failed pruning stale local hook observations");
+        }
+    }
     let observations = match wakey::wakey_linux::observations::list_local_observations_from_path(
         &config.observation_store_path,
     )
@@ -104,7 +127,7 @@ async fn dispatch_inventory(req: InventoryRequest, config: &AgentConfig) -> Resu
         rows = inventory.devices.len(),
         "dispatched inventory command"
     );
-    Ok(CommandResult::Inventory(inventory))
+    Ok(inventory)
 }
 
 async fn dispatch_wake(req: WakeRequest) -> Result<CommandResult> {

wakey-agent/src/enroll.rs

@@ -68,6 +68,9 @@ pub async fn enroll(
         observation_sync_interval_seconds: base_config
             .map(|config| config.observation_sync_interval_seconds)
             .unwrap_or(60),
+        observation_retention_days: base_config
+            .map(|config| config.observation_retention_days)
+            .unwrap_or(crate::config::DEFAULT_OBSERVATION_RETENTION_DAYS),
         pid_file: base_config
             .map(|config| config.pid_file.clone())
             .unwrap_or_else(|| crate::config::DEFAULT_PID_FILE.into()),
@@ -166,6 +169,7 @@ mod tests {
             reconnect_base_ms: 2_000,
             reconnect_max_ms: 60_000,
             observation_sync_interval_seconds: 30,
+            observation_retention_days: 11,
             pid_file: "/tmp/custom-wakey-agent.pid".into(),
             dhcp_leases_path: "/tmp/custom-dhcp.leases".into(),
             mac_name_cache_path: "/tmp/custom-names.json".into(),
@@ -180,6 +184,7 @@ mod tests {
         assert_eq!(config.agent_id, "agent-123");
         assert_eq!(config.agent_token, "token-xyz");
         assert_eq!(config.server_url, "https://control.example.com");
+        assert_eq!(config.observation_retention_days, 11);
         assert_eq!(config.pid_file, base_config.pid_file);
         assert_eq!(config.dhcp_leases_path, base_config.dhcp_leases_path);
         assert!(outcome.backup_path.is_none());

wakey-agent/src/main.rs

@@ -255,6 +255,7 @@ fn init_config(args: InitConfigArgs) -> Result<()> {
             reconnect_base_ms: 1_000,
             reconnect_max_ms: 30_000,
             observation_sync_interval_seconds: 60,
+            observation_retention_days: config::DEFAULT_OBSERVATION_RETENTION_DAYS,
             pid_file: config::DEFAULT_PID_FILE.into(),
             dhcp_leases_path: "/tmp/dhcp.leases".into(),
             mac_name_cache_path: "/tmp/wakey_mac_names.json".into(),
@@ -271,6 +272,9 @@ fn init_config(args: InitConfigArgs) -> Result<()> {
     if let Some(agent_token) = args.agent_token {
         cfg.agent_token = agent_token;
     }
+    if let Some(days) = args.observation_retention_days {
+        cfg.observation_retention_days = days;
+    }
 
     if let Some(path) = &args.config {
         config::save_config(path, &cfg)?;

wakey-agent/src/session.rs

@@ -7,7 +7,7 @@ use tokio_tungstenite::{connect_async, tungstenite::Message};
 use tracing::{debug, error, info, info_span, warn};
 
 use crate::config::AgentConfig;
-use crate::dispatch::dispatch_command;
+use crate::dispatch::{dispatch_command, inventory_for_config};
 use crate::protocol::{AgentCommand, ClientMessage, ErrorPayload, ServerMessage};
 
 pub async fn run(config: AgentConfig) -> Result<()> {
@@ -150,8 +150,15 @@ where
     S: SinkExt<Message> + Unpin,
     <S as futures_util::Sink<Message>>::Error: std::error::Error + Send + Sync + 'static,
 {
-    let query = wakey_core::InventoryQueryBuilder::new().build();
-    let inventory = wakey::inventory(query)
+    let req = crate::protocol::InventoryRequest {
+        query: None,
+        name: None,
+        ips: Vec::new(),
+        devs: Vec::new(),
+        nuds: Vec::new(),
+        macs: Vec::new(),
+    };
+    let inventory = inventory_for_config(req, config)
         .await
         .context("failed to run inventory for device snapshot")?;
     let count = inventory.devices.len();