From dfe8a8ffc7e20fd4d858b87644be08a639c0f328 Mon Sep 17 00:00:00 2001 From: Jaganmohan Reddy Sanikommu Date: Tue, 8 Sep 2026 10:29:41 -0400 Subject: [PATCH] feat(agent): refresh the device inventory on an interval Discovery ran once during daemon startup and that snapshot was cloned into both the local API and the controller stream, so a tool installed, removed, or reconfigured afterwards stayed invisible until the daemon restarted. The snapshot now lives in a watch channel that a background task refreshes every `inventoryInterval`, defaulting to 15m. The local API serves the latest snapshot, and the controller stream sends a fresh Inventory message whenever one is published. `replace_inventory` already replaces a device's stored rows, so resending is idempotent. A refreshed snapshot is published only when it differs from the previous one, so an idle device neither logs nor wakes the controller stream. The refresh loop takes its discovery function as a parameter so that behaviour is covered by tests, and the stream branch stops watching rather than spinning if the refresher ever goes away. `time::interval_at` panics on a zero period, so parsing rejects a zero `inventoryInterval` and the refresh loop refuses one as well rather than leaving that panic for a future caller. Refs #62. Signed-off-by: Jaganmohan Reddy Sanikommu --- crates/agent/Cargo.toml | 2 + crates/agent/src/api.rs | 8 +- crates/agent/src/daemon.rs | 188 ++++++++++++++++++++++++++++++++----- crates/agent/src/remote.rs | 186 ++++++++++++++++++++++++------------ crates/core/src/config.rs | 61 +++++++++++- crates/core/src/model.rs | 4 +- schema/daemon-config.json | 5 + schema/daemon-config.md | 1 + 8 files changed, 366 insertions(+), 89 deletions(-) diff --git a/crates/agent/Cargo.toml b/crates/agent/Cargo.toml index 0446554c..d0844fea 100644 --- a/crates/agent/Cargo.toml +++ b/crates/agent/Cargo.toml @@ -54,6 +54,8 @@ futures-util = "0.3" tempfile = "3" testcontainers = "0.28" tracing-subscriber.workspace = true +# `start_paused` keeps the interval-driven inventory tests deterministic. +tokio = { workspace = true, features = ["test-util"] } [target.'cfg(unix)'.dependencies] libc.workspace = true diff --git a/crates/agent/src/api.rs b/crates/agent/src/api.rs index a8286a5c..72d204f5 100644 --- a/crates/agent/src/api.rs +++ b/crates/agent/src/api.rs @@ -1,4 +1,4 @@ -use std::{net::SocketAddr, path::PathBuf}; +use std::{net::SocketAddr, path::PathBuf, sync::Arc}; use axum::{ Json, Router, @@ -7,7 +7,7 @@ use axum::{ routing::{get, post}, }; use serde::{Deserialize, Serialize}; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::{mpsc, oneshot, watch}; use agentdesktop_core::{ config::{DaemonConfig, LlmGatewayAuthentication, ProgramAuthentication, valid_client_id}, @@ -21,7 +21,7 @@ use crate::{enrollment::EnrollmentState, gateway_oidc, remote, subscription}; #[derive(Clone)] pub struct AppState { pub config: DaemonConfig, - pub discovery: Discovery, + pub discovery: watch::Receiver>, pub enrollment: EnrollmentState, pub state_dir: PathBuf, pub oidc_callback_listen: Option, @@ -176,7 +176,7 @@ async fn remote_config( } async fn discover(State(state): State) -> Json { - Json(state.discovery) + Json((**state.discovery.borrow()).clone()) } async fn enrollment(State(state): State) -> Json { diff --git a/crates/agent/src/daemon.rs b/crates/agent/src/daemon.rs index a9138535..cb5defe3 100644 --- a/crates/agent/src/daemon.rs +++ b/crates/agent/src/daemon.rs @@ -2,6 +2,8 @@ use std::{ future::Future, net::SocketAddr, path::{Path, PathBuf}, + sync::Arc, + time::Duration, }; #[cfg(unix)] @@ -20,7 +22,10 @@ use hyper_util::{rt::TokioIo, service::TowerToHyperService}; use tokio::net::UnixListener; #[cfg(windows)] use tokio::net::windows::named_pipe::{NamedPipeServer, ServerOptions}; -use tokio::sync::mpsc; +use tokio::{ + sync::{mpsc, watch}, + time, +}; #[cfg(windows)] use windows_sys::Win32::Security::SECURITY_ATTRIBUTES; @@ -328,29 +333,19 @@ where ); } let discovery = reconciler.discover().await; - for agent in &discovery.agents { - tracing::info!( - kind = %agent.kind, - executable = %agent.executable.display(), - version = agent.version.as_deref().unwrap_or("unknown"), - mcps = agent.mcp_servers.len(), - skills = agent.skills.len(), - "discovered program" - ); - } - for runtime in &discovery.model_runtimes { - tracing::info!( - kind = %runtime.kind, - models = runtime.models.len(), - "discovered model runtime" - ); - } + log_discovery(&discovery); + let (inventory_sender, inventory) = watch::channel(Arc::new(discovery)); + tokio::spawn(refresh_inventory( + inventory_sender, + config.inventory_interval, + reconciler.clone(), + )); let (telemetry_sender, telemetry_receiver) = mpsc::channel(256); let telemetry = config.controller.as_ref().map(|_| telemetry_sender.clone()); let (logout_sender, logout_receiver) = mpsc::channel(1); let logout = config.controller.as_ref().map(|_| logout_sender); if let Some(controller) = config.controller.clone() { - let remote_discovery = discovery.clone(); + let remote_discovery = inventory.clone(); let state_dir = args.state_dir.clone(); let oidc_callback_listen = args.oidc_callback_listen; let remote_enrollment = enrollment.clone(); @@ -376,7 +371,7 @@ where } let app = api::router(api::AppState { config, - discovery, + discovery: inventory, enrollment, state_dir: args.state_dir, oidc_callback_listen: args.oidc_callback_listen, @@ -393,6 +388,72 @@ where Ok(()) } +fn log_discovery(discovery: &agentdesktop_core::model::Discovery) { + for agent in &discovery.agents { + tracing::info!( + kind = %agent.kind, + executable = %agent.executable.display(), + version = agent.version.as_deref().unwrap_or("unknown"), + mcps = agent.mcp_servers.len(), + skills = agent.skills.len(), + "discovered program" + ); + } + for runtime in &discovery.model_runtimes { + tracing::info!( + kind = %runtime.kind, + models = runtime.models.len(), + "discovered model runtime" + ); + } +} + +/// Re-runs discovery on an interval so the inventory reflects tools installed, +/// removed, or reconfigured after the daemon started. +/// +/// The snapshot is published only when it differs from the previous one, so +/// idle devices neither log nor wake the controller stream. +async fn refresh_inventory( + sender: watch::Sender>, + interval: Duration, + reconciler: reconcile::Reconciler, +) { + refresh_inventory_with(sender, interval, || reconciler.discover()).await; +} + +async fn refresh_inventory_with( + sender: watch::Sender>, + interval: Duration, + mut discover: F, +) where + F: FnMut() -> Fut, + Fut: Future, +{ + // Configuration rejects a zero interval, but `interval_at` panics on one, + // so refuse it here too rather than leaving a landmine for another caller. + if interval.is_zero() { + tracing::error!("inventory interval must be greater than zero; not refreshing inventory"); + return; + } + let mut ticker = time::interval_at(time::Instant::now() + interval, interval); + ticker.set_missed_tick_behavior(time::MissedTickBehavior::Delay); + loop { + ticker.tick().await; + if sender.is_closed() { + return; + } + let discovery = discover().await; + if **sender.borrow() == discovery { + continue; + } + tracing::info!("inventory changed"); + log_discovery(&discovery); + if sender.send(Arc::new(discovery)).is_err() { + return; + } + } +} + fn start_gateway_authentication( config: &agentdesktop_core::config::DaemonConfig, state_dir: PathBuf, @@ -753,12 +814,97 @@ mod tests { pin::Pin, sync::{ Arc, - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, AtomicUsize, Ordering}, }, task::{Context, Poll}, + time::Duration, }; use agentdesktop_core::config::parse_daemon; + use agentdesktop_core::model::{Agent, Discovery}; + use tokio::sync::watch; + + fn discovery(kinds: &[&str]) -> Discovery { + Discovery { + agents: kinds + .iter() + .map(|kind| Agent { + kind: (*kind).to_owned(), + executable: Path::new("/usr/bin").join(kind), + version: None, + mcp_servers: Vec::new(), + skills: Vec::new(), + }) + .collect(), + model_runtimes: Vec::new(), + } + } + + #[tokio::test(start_paused = true)] + async fn refuses_to_schedule_a_zero_interval() { + let (sender, receiver) = watch::channel(Arc::new(discovery(&["codex"]))); + super::refresh_inventory_with(sender, Duration::ZERO, || async { + panic!("a zero interval must not schedule a scan") + }) + .await; + assert_eq!(receiver.borrow().agents.len(), 1); + } + + #[tokio::test(start_paused = true)] + async fn publishes_inventory_only_when_it_changes() { + let (sender, mut receiver) = watch::channel(Arc::new(discovery(&["codex"]))); + let scans = Arc::new(AtomicUsize::new(0)); + let refresher = tokio::spawn({ + let scans = Arc::clone(&scans); + super::refresh_inventory_with(sender, Duration::from_secs(60), move || { + let scan = scans.fetch_add(1, Ordering::SeqCst); + // The first scan repeats the boot snapshot; every later scan + // reports a newly installed tool. + async move { + if scan == 0 { + discovery(&["codex"]) + } else { + discovery(&["codex", "cursor"]) + } + } + }) + }); + + // Paused time auto-advances between ticks, so this resolves on the + // first scan that actually differs rather than on the first tick. + receiver.changed().await.unwrap(); + assert_eq!(receiver.borrow_and_update().agents.len(), 2); + assert!( + scans.load(Ordering::SeqCst) >= 2, + "an unchanged scan should not have published a snapshot" + ); + + // Repeating that same state does not republish it. + assert!( + tokio::time::timeout(Duration::from_secs(600), receiver.changed()) + .await + .is_err() + ); + refresher.abort(); + } + + #[tokio::test(start_paused = true)] + async fn stops_refreshing_once_every_reader_is_gone() { + let (sender, receiver) = watch::channel(Arc::new(discovery(&["codex"]))); + let refresher = tokio::spawn(super::refresh_inventory_with( + sender, + Duration::from_secs(60), + || async { discovery(&["codex", "cursor"]) }, + )); + drop(receiver); + + assert!( + tokio::time::timeout(Duration::from_secs(600), refresher) + .await + .is_ok() + ); + } + use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf}; use super::{ diff --git a/crates/agent/src/remote.rs b/crates/agent/src/remote.rs index 0cc74622..9c20a678 100644 --- a/crates/agent/src/remote.rs +++ b/crates/agent/src/remote.rs @@ -2,6 +2,7 @@ use std::{ future::Future, net::SocketAddr, path::{Path, PathBuf}, + sync::Arc, time::{Duration, SystemTime, UNIX_EPOCH}, }; @@ -9,7 +10,7 @@ use anyhow::{Context, bail}; use rcgen::{CertificateParams, ExtendedKeyUsagePurpose, KeyPair, KeyUsagePurpose}; use sha2::{Digest, Sha256}; use tokio::{ - sync::{Mutex, mpsc, oneshot}, + sync::{Mutex, mpsc, oneshot, watch}, time, }; use tokio_stream::wrappers::ReceiverStream; @@ -55,7 +56,7 @@ pub struct Requests { pub async fn run( controller: ControllerConnectionConfig, - discovered: AgentDiscovery, + mut discovered: watch::Receiver>, state_dir: PathBuf, oidc_callback_listen: Option, reconciler: Reconciler, @@ -133,7 +134,7 @@ pub async fn run( result = connect( &controller, &identity, - &discovered, + &mut discovered, &state_dir, &reconciler, &mut telemetry, @@ -305,7 +306,7 @@ pub async fn llm_gateway_credential( async fn connect( controller: &ControllerConnectionConfig, identity: &Identity, - discovered: &AgentDiscovery, + discovered: &mut watch::Receiver>, state_dir: &Path, reconciler: &Reconciler, telemetry: &mut mpsc::Receiver, @@ -333,61 +334,8 @@ async fn connect( ) .await?; - send( - &sender, - agent_message::Message::Inventory(Inventory { - discoveries: discovered - .agents - .iter() - .map(|agent| Discovery { - kind: agent.kind.clone(), - version: agent.version.clone().unwrap_or_default(), - path: agent.executable.display().to_string(), - mcp_servers: agent - .mcp_servers - .iter() - .map(|server| agentdesktop_proto::fleet::McpServer { - name: server.name.clone(), - transport: server.transport.clone(), - command: server.command.clone().unwrap_or_default(), - url: server.url.clone().unwrap_or_default(), - enabled: server.enabled, - source: server.source.display().to_string(), - }) - .collect(), - skills: agent - .skills - .iter() - .map(|skill| agentdesktop_proto::fleet::Skill { - path: skill.path.display().to_string(), - front_matter_json: serde_json::to_vec(&skill.front_matter) - .expect("skill front matter is JSON-compatible"), - }) - .collect(), - }) - .collect(), - model_runtimes: discovered - .model_runtimes - .iter() - .map(|runtime| agentdesktop_proto::fleet::ModelRuntime { - kind: runtime.kind.clone(), - models: runtime - .models - .iter() - .map(|model| agentdesktop_proto::fleet::LocalModel { - name: model.name.clone(), - }) - .collect(), - }) - .collect(), - }), - ) - .await?; - info!( - discoveries = discovered.agents.len(), - model_runtimes = discovered.model_runtimes.len(), - "reported inventory to controller" - ); + let snapshot = discovered.borrow_and_update().clone(); + send_inventory(&sender, &snapshot).await?; info!(address = %controller.address, "connected to controller"); let mut heartbeat = time::interval(controller.heartbeat_interval); @@ -413,6 +361,9 @@ async fn connect( Some(event) = telemetry.recv() => { send(&sender, agent_message::Message::Telemetry(telemetry_to_proto(event))).await?; } + snapshot = next_inventory(discovered) => { + send_inventory(&sender, &snapshot).await?; + } message = inbound.message() => { let Some(message) = message.context("read controller stream")? else { return Ok(()); @@ -514,6 +465,84 @@ fn apply_daemon_config( } } +/// Sends the current inventory snapshot to the controller. +/// +/// The controller replaces a device's stored inventory on each message, so +/// resending a refreshed snapshot is safe and is how post-boot changes reach +/// the fleet view. +async fn send_inventory( + sender: &mpsc::Sender, + discovered: &AgentDiscovery, +) -> anyhow::Result<()> { + send( + sender, + agent_message::Message::Inventory(Inventory { + discoveries: discovered + .agents + .iter() + .map(|agent| Discovery { + kind: agent.kind.clone(), + version: agent.version.clone().unwrap_or_default(), + path: agent.executable.display().to_string(), + mcp_servers: agent + .mcp_servers + .iter() + .map(|server| agentdesktop_proto::fleet::McpServer { + name: server.name.clone(), + transport: server.transport.clone(), + command: server.command.clone().unwrap_or_default(), + url: server.url.clone().unwrap_or_default(), + enabled: server.enabled, + source: server.source.display().to_string(), + }) + .collect(), + skills: agent + .skills + .iter() + .map(|skill| agentdesktop_proto::fleet::Skill { + path: skill.path.display().to_string(), + front_matter_json: serde_json::to_vec(&skill.front_matter) + .expect("skill front matter is JSON-compatible"), + }) + .collect(), + }) + .collect(), + model_runtimes: discovered + .model_runtimes + .iter() + .map(|runtime| agentdesktop_proto::fleet::ModelRuntime { + kind: runtime.kind.clone(), + models: runtime + .models + .iter() + .map(|model| agentdesktop_proto::fleet::LocalModel { + name: model.name.clone(), + }) + .collect(), + }) + .collect(), + }), + ) + .await?; + info!( + discoveries = discovered.agents.len(), + model_runtimes = discovered.model_runtimes.len(), + "reported inventory to controller" + ); + Ok(()) +} + +/// Resolves with the next inventory snapshot, and never resolves once the +/// refresher has stopped, so the controller stream keeps running without it. +async fn next_inventory( + discovered: &mut watch::Receiver>, +) -> Arc { + if discovered.changed().await.is_err() { + std::future::pending::<()>().await; + } + discovered.borrow_and_update().clone() +} + async fn send( sender: &mpsc::Sender, message: agent_message::Message, @@ -672,9 +701,46 @@ mod tests { use super::{ MAX_RETRY_DELAY, enroll_with_retry, is_oauth_refresh_rejected, is_unauthenticated, - next_retry_delay, normalize_hostname, + next_inventory, next_retry_delay, normalize_hostname, }; use crate::enrollment::EnrollmentState; + use agentdesktop_core::model::Discovery as AgentDiscovery; + use tokio::{sync::watch, time}; + + fn empty_inventory() -> AgentDiscovery { + AgentDiscovery { + agents: Vec::new(), + model_runtimes: Vec::new(), + } + } + + #[tokio::test] + async fn reports_each_refreshed_inventory_snapshot() { + let (sender, mut receiver) = watch::channel(Arc::new(empty_inventory())); + let mut refreshed = empty_inventory(); + refreshed + .model_runtimes + .push(agentdesktop_core::model::ModelRuntime { + kind: "ollama".to_owned(), + models: Vec::new(), + }); + sender.send(Arc::new(refreshed)).unwrap(); + + let snapshot = next_inventory(&mut receiver).await; + assert_eq!(snapshot.model_runtimes.len(), 1); + } + + /// A stopped refresher must not turn the controller stream into a busy loop. + #[tokio::test(start_paused = true)] + async fn never_reports_again_once_the_refresher_stops() { + let (sender, mut receiver) = watch::channel(Arc::new(empty_inventory())); + drop(sender); + assert!( + time::timeout(Duration::from_secs(3600), next_inventory(&mut receiver)) + .await + .is_err() + ); + } #[test] fn recognizes_contextualized_unauthenticated_status() { diff --git a/crates/core/src/config.rs b/crates/core/src/config.rs index 33febb39..f333043f 100644 --- a/crates/core/src/config.rs +++ b/crates/core/src/config.rs @@ -12,7 +12,7 @@ use serde::{Deserialize, Serialize}; use url::Url; /// Configuration for an Agentdesktop daemon. -#[derive(Clone, Debug, Default, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[cfg_attr(feature = "schema", derive(schemars::JsonSchema))] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct DaemonConfig { @@ -31,6 +31,27 @@ pub struct DaemonConfig { /// Per-program settings reconciled on this device. #[serde(default, skip_serializing_if = "ProgramsConfig::is_empty")] pub programs: ProgramsConfig, + /// Interval between inventory refreshes. Defaults to `15m`, and must be + /// greater than zero. + /// + /// Discovery walks user home directories and developer-tool configuration + /// files, so this trades inventory freshness against local disk activity. + #[serde(default = "default_inventory_interval", with = "humantime_serde")] + #[cfg_attr(feature = "schema", schemars(with = "String"))] + pub inventory_interval: Duration, +} + +impl Default for DaemonConfig { + fn default() -> Self { + Self { + controller: None, + llm_gateway: None, + sandbox: None, + telemetry: TelemetryConfig::default(), + programs: ProgramsConfig::default(), + inventory_interval: default_inventory_interval(), + } + } } /// Local execution restrictions applied to managed developer tools. @@ -358,6 +379,10 @@ fn default_gateway_jwt_lifetime() -> Duration { Duration::from_secs(5 * 60) } +fn default_inventory_interval() -> Duration { + Duration::from_secs(15 * 60) +} + fn default_heartbeat_interval() -> Duration { Duration::from_secs(30) } @@ -614,6 +639,9 @@ pub fn parse_daemon(contents: &str) -> anyhow::Result { { anyhow::bail!("controller address must use HTTPS"); } + if config.inventory_interval.is_zero() { + anyhow::bail!("inventoryInterval must be greater than zero"); + } validate_daemon( config.llm_gateway.as_ref(), config.sandbox.as_ref(), @@ -816,7 +844,7 @@ fn is_true(value: &bool) -> bool { #[cfg(test)] mod tests { - use super::{LlmGatewayAuthentication, parse_controller, parse_daemon}; + use super::{DaemonConfig, LlmGatewayAuthentication, parse_controller, parse_daemon}; #[test] fn daemon_configuration_supports_local_and_managed_options() { @@ -834,6 +862,35 @@ programs: { claudeCode: { useLlmGateway: false } } assert!(!daemon.programs.claude_code.unwrap().use_llm_gateway); } + #[test] + fn daemon_inventory_interval_defaults_and_parses_durations() { + let default = parse_daemon("programs: {}").expect("valid daemon configuration"); + assert_eq!( + default.inventory_interval, + std::time::Duration::from_secs(15 * 60) + ); + assert_eq!( + DaemonConfig::default().inventory_interval, + default.inventory_interval + ); + + let configured = parse_daemon("inventoryInterval: 2m").expect("valid daemon configuration"); + assert_eq!( + configured.inventory_interval, + std::time::Duration::from_secs(120) + ); + } + + #[test] + fn daemon_inventory_interval_rejects_zero() { + let error = parse_daemon("inventoryInterval: 0s") + .expect_err("a zero inventory interval is not schedulable"); + assert!( + error.to_string().contains("inventoryInterval"), + "unexpected error: {error}" + ); + } + #[test] fn daemon_controller_requires_https() { let plaintext = r#" diff --git a/crates/core/src/model.rs b/crates/core/src/model.rs index 44dcced7..bc122e17 100644 --- a/crates/core/src/model.rs +++ b/crates/core/src/model.rs @@ -2,7 +2,7 @@ use std::{collections::BTreeMap, path::PathBuf}; use serde::{Deserialize, Serialize}; -#[derive(Clone, Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] #[serde(rename_all = "camelCase")] pub struct Discovery { pub agents: Vec, @@ -25,7 +25,7 @@ pub struct LocalModel { pub name: String, } -#[derive(Clone, Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] #[serde(rename_all = "camelCase")] pub struct Agent { pub kind: String, diff --git a/schema/daemon-config.json b/schema/daemon-config.json index b3520f50..550dfc5c 100644 --- a/schema/daemon-config.json +++ b/schema/daemon-config.json @@ -15,6 +15,11 @@ } ] }, + "inventoryInterval": { + "description": "Interval between inventory refreshes. Defaults to `15m`, and must be\ngreater than zero.\n\nDiscovery walks user home directories and developer-tool configuration\nfiles, so this trades inventory freshness against local disk activity.", + "type": "string", + "default": "15m" + }, "llmGateway": { "description": "LLM gateway used by managed developer tools.", "anyOf": [ diff --git a/schema/daemon-config.md b/schema/daemon-config.md index 5d90c196..883c4a94 100644 --- a/schema/daemon-config.md +++ b/schema/daemon-config.md @@ -6,6 +6,7 @@ |`controller.address`|string|HTTPS address of the controller's fleet API.| |`controller.caCertificatePath`|string|Path to a PEM-encoded CA certificate used to verify the controller.

Omit this field to use the operating system's trusted certificate roots.| |`controller.heartbeatInterval`|string|Interval between device heartbeats. Defaults to `30s`.| +|`inventoryInterval`|string|Interval between inventory refreshes. Defaults to `15m`, and must be
greater than zero.

Discovery walks user home directories and developer-tool configuration
files, so this trades inventory freshness against local disk activity.| |`llmGateway`|object|LLM gateway used by managed developer tools.| |`llmGateway.authentication`|object|Authentication mechanism used when connecting to this gateway.| |`llmGateway.authentication.allowedClientIds`|[]string|Client identifiers permitted to request credentials for this gateway.|