feat(mcp): auto-reconnect installed servers via a background supervisor (#3312) (#3571)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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());

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
 

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 {

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;
 

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<Instant>,
+}
+
+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<String, BackoffState> = 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<String, BackoffState> = 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<String, BackoffState>, 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)));
+    }
+}

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"
+    );
+}