diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 5b47190bd..77c253a70 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -2259,6 +2259,29 @@ pub async fn bootstrap_core_runtime(host_kind: crate::core::types::HostKind) { }); } + // --- MCP registry reconnect supervisor (#3312) ----------------------- + // Keep installed MCP servers connected for the life of the process: + // transports drop silently over long uptimes (subprocess exits, HTTP + // session expires) and the boot spawn above only runs once. The + // supervisor periodically probes each connection and reconnects dropped + // ones with per-server backoff. First tick is delayed so it doesn't race + // the boot spawn. + // + // Guarded by `Once`: `bootstrap_core_runtime` is documented idempotent, and + // unlike the one-shot boot spawn above the supervisor is an infinite tick + // loop — a second instance would race the first on the shared connections + // registry (duplicate probes, reconnect thrashing, nondeterministic backoff). + { + use std::sync::Once; + static SUPERVISOR_SPAWNED: Once = Once::new(); + SUPERVISOR_SPAWNED.call_once(|| { + let cfg = cfg.clone(); + tokio::spawn(async move { + crate::openhuman::mcp_registry::supervisor::run(cfg).await; + }); + }); + } + // --- Socket manager bootstrap --- let socket_mgr = Arc::new(SocketManager::new()); set_global_socket_manager(socket_mgr.clone()); diff --git a/src/openhuman/mcp_registry/README.md b/src/openhuman/mcp_registry/README.md index 24f9d57a4..28ac25cd6 100644 --- a/src/openhuman/mcp_registry/README.md +++ b/src/openhuman/mcp_registry/README.md @@ -28,6 +28,7 @@ The transport primitives themselves (stdio + HTTP MCP clients) live in the sibli | `src/openhuman/mcp_registry/registries/mcp_official.rs` | Official modelcontextprotocol/registry adapter; cursor→page mapping with a bounded sequential cursor walk; optional `MCP_OFFICIAL_REGISTRY_*` env overrides. | | `src/openhuman/mcp_registry/connections.rs` | Global in-process connection registry; `connect`/`disconnect`/`call_tool`/`all_status`/`all_connected_tools`; dispatches stdio vs HTTP via `ActiveClient`. | | `src/openhuman/mcp_registry/boot.rs` | `spawn_installed_servers` — boot-time connect of all installs; per-server failures logged, never fatal. | +| `src/openhuman/mcp_registry/supervisor.rs` | Background reconnect loop (#3312): every 60s probes each connected transport (`connections::probe_alive`) and reconnects dropped/never-connected enabled servers with per-server exponential backoff. Spawned once at startup after the boot pass; connectivity-only (publishes no health events). | | `src/openhuman/mcp_registry/ops.rs` | `mcp_clients_*` RPC handler implementations + `resolve_command` + `config_assist` inference call. | | `src/openhuman/mcp_registry/setup.rs` | Opaque secret-ref machinery (`SecretRef`, mint/fulfill/await/resolve/consume/gc) for the setup agent. | | `src/openhuman/mcp_registry/setup_ops.rs` | `mcp_setup_*` RPC handlers (search/get/request_secret/submit_secret/test_connection/install_and_connect) + connection `pick_connection`. | @@ -41,7 +42,7 @@ From `mod.rs`: - `mcp_registry_schemas` (`schemas::schemas`). - Types `ConnStatus`, `InstalledServer`, `McpTool`. -`pub mod boot`, `bus`, `connections`, `setup`, `setup_ops`, `store`, `types` are public; `ops`, `registries`, `registry`, `schemas` are private (reached via the schema handlers). Notably `connections::all_connected_tools()` is consumed by `tool_registry`, and `boot::spawn_installed_servers` is called from core startup. +`pub mod boot`, `bus`, `connections`, `setup`, `setup_ops`, `store`, `supervisor`, `types` are public; `ops`, `registries`, `registry`, `schemas` are private (reached via the schema handlers). Notably `connections::all_connected_tools()` is consumed by `tool_registry`, and both `boot::spawn_installed_servers` and `supervisor::run` are spawned from core startup. ## RPC / controllers diff --git a/src/openhuman/mcp_registry/connections.rs b/src/openhuman/mcp_registry/connections.rs index 11a209024..858747be1 100644 --- a/src/openhuman/mcp_registry/connections.rs +++ b/src/openhuman/mcp_registry/connections.rs @@ -225,6 +225,55 @@ async fn connect_inner(config: &Config, server: &InstalledServer) -> anyhow::Res Ok(tools) } +/// Whether `server_id` currently has a live entry in the connection +/// registry. Note this only reflects map membership — a silently-dropped +/// transport stays in the map until something probes it. Use +/// [`probe_alive`] for an actual liveness check. +pub async fn is_connected(server_id: &str) -> bool { + connections().read().await.contains_key(server_id) +} + +/// Actively probe a connected server's transport by issuing a `tools/list` +/// round-trip under `timeout`. Returns `true` only when the call succeeds. +/// +/// This is the detection mechanism the reconnect supervisor (#3312) relies +/// on: MCP transports can drop silently (subprocess exits, HTTP session +/// expires) while their `Connection` stays in the registry, so "is it in the +/// map" (`is_connected`) is not enough — a dead transport only surfaces on the +/// next actual call. A periodic lightweight probe converts that latent failure +/// into an observable one so the supervisor can disconnect + reconnect. +/// +/// Returns `false` (rather than erroring) for a missing connection, a transport +/// error, or a timeout — all of which mean "not usable, reconnect". +pub async fn probe_alive(server_id: &str, timeout: std::time::Duration) -> bool { + tracing::trace!("[mcp-registry] probe_alive server_id={server_id} timeout={timeout:?}"); + let conn = { + let map = connections().read().await; + map.get(server_id).cloned() + }; + let Some(conn) = conn else { + return false; + }; + match tokio::time::timeout(timeout, conn.client.list_tools()).await { + Ok(Ok(_)) => { + tracing::trace!("[mcp-registry] probe_alive server_id={server_id} alive"); + true + } + Ok(Err(err)) => { + tracing::debug!( + "[mcp-registry] probe_alive server_id={server_id} transport error: {err}" + ); + false + } + Err(_) => { + tracing::debug!( + "[mcp-registry] probe_alive server_id={server_id} timed out after {timeout:?}" + ); + false + } + } +} + /// Disconnect and remove from the registry. Also clears any recorded /// connect error so the next status poll starts from a clean slate. pub async fn disconnect(server_id: &str) -> bool { diff --git a/src/openhuman/mcp_registry/mod.rs b/src/openhuman/mcp_registry/mod.rs index 8d21b7a75..6fcc9308d 100644 --- a/src/openhuman/mcp_registry/mod.rs +++ b/src/openhuman/mcp_registry/mod.rs @@ -60,6 +60,7 @@ mod schemas; pub mod setup; pub mod setup_ops; pub mod store; +pub mod supervisor; pub mod tools; pub mod types; diff --git a/src/openhuman/mcp_registry/supervisor.rs b/src/openhuman/mcp_registry/supervisor.rs new file mode 100644 index 000000000..e5e161189 --- /dev/null +++ b/src/openhuman/mcp_registry/supervisor.rs @@ -0,0 +1,223 @@ +//! Background supervisor that keeps installed MCP servers connected (#3312). +//! +//! Problem: MCP transports drop silently over a long-running deployment — a +//! stdio subprocess exits, or an HTTP-remote session expires — and nothing +//! re-establishes them. Connections are brought up only once, at boot +//! ([`super::boot::spawn_installed_servers`]). After a few hours a headless +//! deployment ends up with 0 MCP tools and no way back short of a restart. +//! +//! This supervisor mirrors the cron scheduler's tick loop: every +//! [`TICK_INTERVAL`] it walks every enabled installed server and, for each: +//! - if it is in the registry, actively probes the transport +//! ([`super::connections::probe_alive`]) — a silent drop only surfaces under +//! an actual round-trip, so map membership alone is not trusted; +//! - if the probe fails, disconnects the dead transport and reconnects; +//! - if it is not connected, reconnects — subject to per-server exponential +//! backoff so a genuinely-down or misconfigured server isn't hammered. +//! +//! Scope: this PR is connectivity only. The supervisor deliberately does **not** +//! publish health events to the global health bus — doing so would flip the +//! whole container to `unhealthy`/503 whenever a single MCP server is down, +//! which is exactly the all-or-nothing coupling the granular-health work +//! addresses separately. Reconnect outcomes are logged. + +use std::collections::HashMap; +use std::time::{Duration, Instant}; + +use crate::openhuman::config::Config; + +use super::{connections, store}; + +/// How often the supervisor walks the installed-server list. +const TICK_INTERVAL: Duration = Duration::from_secs(60); + +/// Per-server liveness-probe timeout. A `tools/list` round-trip should be fast; +/// a server that can't answer within this window is treated as dropped. +const PROBE_TIMEOUT: Duration = Duration::from_secs(8); + +/// Base unit for exponential reconnect backoff. +const BACKOFF_BASE: Duration = Duration::from_secs(5); + +/// Cap on reconnect backoff so a long-down server is still retried every few +/// minutes (its operator may fix it without touching OpenHuman). +const BACKOFF_MAX: Duration = Duration::from_secs(300); + +/// Exponential backoff delay after `failures` consecutive failed reconnects: +/// `BASE * 2^(failures-1)`, capped at [`BACKOFF_MAX`]. `failures == 0` yields +/// [`BACKOFF_BASE`] (treated as "no failures yet → try immediately-ish"). +fn backoff_delay(failures: u32) -> Duration { + if failures == 0 { + return BACKOFF_BASE; + } + // Saturating shift so a large failure count can't overflow; cap below. + let shifted = BACKOFF_BASE + .as_secs() + .saturating_mul(1u64.checked_shl(failures - 1).unwrap_or(u64::MAX)); + Duration::from_secs(shifted.min(BACKOFF_MAX.as_secs())) +} + +/// Per-server reconnect backoff state. +#[derive(Default)] +struct BackoffState { + failures: u32, + next_attempt_at: Option, +} + +impl BackoffState { + /// Whether a reconnect may be attempted at `now`. A fresh state (no prior + /// failure) is always ready. + fn ready(&self, now: Instant) -> bool { + self.next_attempt_at.is_none_or(|t| now >= t) + } + + /// Record a failed reconnect and schedule the next eligible attempt. + fn record_failure(&mut self, now: Instant) { + self.failures = self.failures.saturating_add(1); + self.next_attempt_at = Some(now + backoff_delay(self.failures)); + } +} + +/// Run the supervisor loop forever. Spawned once at core startup, after the +/// boot connect pass. The first tick is delayed by [`TICK_INTERVAL`] so it does +/// not race the boot spawn. +pub async fn run(config: Config) { + let start = Instant::now() + TICK_INTERVAL; + let mut interval = tokio::time::interval_at(start.into(), TICK_INTERVAL); + let mut backoff: HashMap = HashMap::new(); + + tracing::info!( + "[mcp-supervisor] started: tick={}s probe_timeout={}s", + TICK_INTERVAL.as_secs(), + PROBE_TIMEOUT.as_secs() + ); + + loop { + interval.tick().await; + tick_once(&config, &mut backoff, Instant::now()).await; + } +} + +/// Run exactly one supervision cycle with a fresh backoff map. Exposed for +/// integration tests (the internal [`tick_once`] can't be `pub` because its +/// `BackoffState` parameter is a private type). Not used by production code — +/// the live loop calls [`tick_once`] directly with persistent backoff state. +#[doc(hidden)] +pub async fn run_single_tick_for_test(config: &Config) { + let mut backoff: HashMap = HashMap::new(); + tick_once(config, &mut backoff, Instant::now()).await; +} + +/// One supervision cycle, extracted so the loop body is driven without owning a +/// `tokio::time::interval`. `now` is injected for deterministic backoff timing. +async fn tick_once(config: &Config, backoff: &mut HashMap, now: Instant) { + let servers = match store::list_servers(config) { + Ok(s) => s, + Err(err) => { + tracing::warn!("[mcp-supervisor] tick: list_servers failed: {err}"); + return; + } + }; + + for server in servers { + let id = server.server_id.clone(); + + if !server.enabled { + // Drop any stale backoff for a server that's been disabled; the + // disable flow owns tearing down its live connection. + backoff.remove(&id); + continue; + } + + if connections::is_connected(&id).await { + if connections::probe_alive(&id, PROBE_TIMEOUT).await { + // Healthy — clear any lingering backoff and move on. + backoff.remove(&id); + continue; + } + tracing::warn!( + "[mcp-supervisor] server_id={id} qualified={} transport dropped; reconnecting", + server.qualified_name + ); + connections::disconnect(&id).await; + } + + // Not connected (never connected, or just-disconnected dead transport). + // Gate the reconnect attempt on the per-server backoff schedule. + let ready = { + let st = backoff.entry(id.clone()).or_default(); + st.ready(now) + }; + if !ready { + continue; + } + + match connections::connect(config, &server).await { + Ok(tools) => { + backoff.remove(&id); + tracing::info!( + "[mcp-supervisor] reconnected server_id={id} qualified={} tools={}", + server.qualified_name, + tools.len() + ); + } + Err(err) => { + let st = backoff.entry(id.clone()).or_default(); + st.record_failure(now); + tracing::warn!( + "[mcp-supervisor] reconnect failed server_id={id} qualified={} \ + failures={} next_retry_in={}s err={err}", + server.qualified_name, + st.failures, + backoff_delay(st.failures).as_secs() + ); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn backoff_delay_grows_exponentially_then_caps() { + assert_eq!(backoff_delay(0), BACKOFF_BASE); + assert_eq!(backoff_delay(1), Duration::from_secs(5)); + assert_eq!(backoff_delay(2), Duration::from_secs(10)); + assert_eq!(backoff_delay(3), Duration::from_secs(20)); + assert_eq!(backoff_delay(4), Duration::from_secs(40)); + // Caps at BACKOFF_MAX and never overflows for absurd failure counts. + assert_eq!(backoff_delay(20), BACKOFF_MAX); + assert_eq!(backoff_delay(u32::MAX), BACKOFF_MAX); + } + + #[test] + fn fresh_state_is_ready_immediately() { + let st = BackoffState::default(); + assert!(st.ready(Instant::now())); + } + + #[test] + fn record_failure_defers_next_attempt_until_backoff_elapses() { + let base = Instant::now(); + let mut st = BackoffState::default(); + st.record_failure(base); + assert_eq!(st.failures, 1); + // Not ready immediately after a failure... + assert!(!st.ready(base)); + assert!(!st.ready(base + Duration::from_secs(4))); + // ...ready once the 5s base backoff has elapsed. + assert!(st.ready(base + Duration::from_secs(5))); + } + + #[test] + fn consecutive_failures_lengthen_the_backoff_window() { + let base = Instant::now(); + let mut st = BackoffState::default(); + st.record_failure(base); // failures=1 → 5s + st.record_failure(base); // failures=2 → 10s from `base` + assert_eq!(st.failures, 2); + assert!(!st.ready(base + Duration::from_secs(9))); + assert!(st.ready(base + Duration::from_secs(10))); + } +} diff --git a/tests/mcp_registry_e2e.rs b/tests/mcp_registry_e2e.rs index 02b4fba39..9de89bdd9 100644 --- a/tests/mcp_registry_e2e.rs +++ b/tests/mcp_registry_e2e.rs @@ -347,3 +347,87 @@ async fn update_env_on_disabled_server_persists_but_does_not_reconnect() { .unwrap(); assert_eq!(mine.status.as_str(), "disabled"); } + +// ── Reconnect supervisor (#3312) ─────────────────────────────────────────────── + +#[tokio::test] +async fn probe_alive_reflects_transport_liveness() { + let (_tmp, cfg) = fresh_workspace_config(); + let server = make_installed_server(); + store::insert_server(&cfg, &server).expect("insert installed server"); + + connections::connect(&cfg, &server).await.expect("connect"); + assert!(connections::is_connected(&server.server_id).await); + assert!( + connections::probe_alive(&server.server_id, std::time::Duration::from_secs(8)).await, + "a live stub answers the tools/list probe" + ); + + connections::disconnect(&server.server_id).await; + assert!(!connections::is_connected(&server.server_id).await); + assert!( + !connections::probe_alive(&server.server_id, std::time::Duration::from_secs(8)).await, + "a disconnected server is not alive" + ); +} + +#[tokio::test] +async fn supervisor_reconnects_a_dropped_server() { + use openhuman_core::openhuman::mcp_registry::supervisor; + + let (_tmp, cfg) = fresh_workspace_config(); + let server = make_installed_server(); + store::insert_server(&cfg, &server).expect("insert installed server"); + + // Bring it up, then simulate a silent transport drop by disconnecting while + // it stays installed + enabled in the store. + connections::connect(&cfg, &server).await.expect("connect"); + connections::disconnect(&server.server_id).await; + assert!(!connections::is_connected(&server.server_id).await); + + // One supervisor tick should notice the enabled-but-disconnected server and + // reconnect it. + supervisor::run_single_tick_for_test(&cfg).await; + + assert!( + connections::is_connected(&server.server_id).await, + "supervisor reconnects a dropped-but-installed server" + ); + assert!(connections::probe_alive(&server.server_id, std::time::Duration::from_secs(8)).await); + + connections::disconnect(&server.server_id).await; +} + +#[tokio::test] +async fn supervisor_leaves_a_healthy_connection_intact() { + use openhuman_core::openhuman::mcp_registry::supervisor; + + let (_tmp, cfg) = fresh_workspace_config(); + let server = make_installed_server(); + store::insert_server(&cfg, &server).expect("insert installed server"); + connections::connect(&cfg, &server).await.expect("connect"); + + // A tick over a healthy server must keep it connected (probe succeeds → no + // disconnect/reconnect churn). + supervisor::run_single_tick_for_test(&cfg).await; + assert!(connections::is_connected(&server.server_id).await); + + connections::disconnect(&server.server_id).await; +} + +#[tokio::test] +async fn supervisor_skips_a_disabled_server() { + use openhuman_core::openhuman::mcp_registry::supervisor; + + let (_tmp, cfg) = fresh_workspace_config(); + let mut server = make_installed_server(); + server.enabled = false; + store::insert_server(&cfg, &server).expect("insert installed server"); + + // A disabled server must never be connected by the supervisor. + supervisor::run_single_tick_for_test(&cfg).await; + assert!( + !connections::is_connected(&server.server_id).await, + "supervisor does not connect disabled servers" + ); +}