From d7d46d347c91a1074c3e9d5d04c51d7c864e6e6f Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 20 Jul 2026 18:03:58 +0200 Subject: [PATCH] feat(#2038): surface degraded mcp servers at turn start (observability) --- hive-agent/src/events.rs | 94 ++++++++++++++++++++++++++++++++++++ hive-agent/src/mcp_config.rs | 23 ++++++++- hive-agent/src/turn.rs | 1 + 3 files changed, 116 insertions(+), 2 deletions(-) diff --git a/hive-agent/src/events.rs b/hive-agent/src/events.rs index e24b5719..0b50c828 100644 --- a/hive-agent/src/events.rs +++ b/hive-agent/src/events.rs @@ -671,6 +671,52 @@ impl Bus { } } + /// Inspect the per-turn `system`/`init` stream event's `mcp_servers` + /// array and surface a Note + `warn` when a configured MCP server + /// failed to connect or is missing entirely. Pure observability: it + /// never touches the session. It exists because the harness is + /// otherwise blind to the init event — the only place claude reports + /// MCP-server status — so a dropped/failed stdio bridge (e.g. matrix + /// after a core bounce) went silent. Emitted every degraded turn on + /// purpose: whether it persists across fresh turns is the signal that + /// picks the real fix. + pub fn observe_mcp_health(&self, v: &serde_json::Value) { + if v.get("type").and_then(|t| t.as_str()) != Some("system") + || v.get("subtype").and_then(|s| s.as_str()) != Some("init") + { + return; + } + // claude reports one entry per configured server: `{name, status}`. + // Absent-from-array = claude dropped the server entirely; a present + // entry with `status != "connected"` = it failed to register. + let reported: std::collections::HashMap<&str, &str> = v + .get("mcp_servers") + .and_then(|m| m.as_array()) + .map(|arr| { + arr.iter() + .filter_map(|s| { + let name = s.get("name").and_then(|n| n.as_str())?; + let status = s + .get("status") + .and_then(|st| st.as_str()) + .unwrap_or("unknown"); + Some((name, status)) + }) + .collect() + }) + .unwrap_or_default(); + let degraded = + degraded_mcp_servers(&crate::mcp_config::configured_server_names(), &reported); + if degraded.is_empty() { + return; + } + let list = degraded.join(", "); + tracing::warn!(degraded = %list, "mcp servers not connected at turn start"); + self.emit(LiveEvent::Note { + text: format!("⚠ MCP degraded at turn start: {list}"), + }); + } + /// Snapshot + clear the per-turn tool-call counter. The harness /// calls this between turns to fold the breakdown into a /// `turn_stats` row, then start the next turn with an empty map. @@ -837,6 +883,25 @@ impl Default for Bus { } } +/// Compare the configured MCP server names against the `{name -> status}` +/// map parsed from the init event, returning display strings for the +/// degraded ones: a configured server absent from the report was dropped +/// entirely; one present with a non-`connected` status failed to register. +/// Pure so it's unit-testable without a live `Bus`. +fn degraded_mcp_servers( + configured: &[String], + reported: &std::collections::HashMap<&str, &str>, +) -> Vec { + configured + .iter() + .filter_map(|name| match reported.get(name.as_str()).copied() { + Some("connected") => None, + Some(status) => Some(format!("{name} ({status})")), + None => Some(format!("{name} (absent)")), + }) + .collect() +} + #[cfg(test)] mod tests { use super::{BusEvent, LiveEvent, StoredEvent}; @@ -869,4 +934,33 @@ mod tests { assert_eq!(v["ts"], 1_700_000_000_i64); assert_eq!(v["kind"], "note"); } + + #[test] + fn degraded_mcp_servers_flags_absent_and_failed_only() { + use std::collections::HashMap; + let configured = [ + "hyperhive".to_owned(), + "matrix".to_owned(), + "bash".to_owned(), + ]; + // hyperhive connected; matrix failed; bash absent (claude dropped it). + let reported: HashMap<&str, &str> = [("hyperhive", "connected"), ("matrix", "failed")] + .into_iter() + .collect(); + let degraded = super::degraded_mcp_servers(&configured, &reported); + assert_eq!( + degraded, + vec!["matrix (failed)".to_owned(), "bash (absent)".to_owned()] + ); + } + + #[test] + fn degraded_mcp_servers_empty_when_all_connected() { + use std::collections::HashMap; + let configured = ["hyperhive".to_owned(), "matrix".to_owned()]; + let reported: HashMap<&str, &str> = [("hyperhive", "connected"), ("matrix", "connected")] + .into_iter() + .collect(); + assert!(super::degraded_mcp_servers(&configured, &reported).is_empty()); + } } diff --git a/hive-agent/src/mcp_config.rs b/hive-agent/src/mcp_config.rs index 8b3223f1..dd8cc823 100644 --- a/hive-agent/src/mcp_config.rs +++ b/hive-agent/src/mcp_config.rs @@ -290,6 +290,26 @@ fn load_extra_mcp() -> std::collections::BTreeMap { /// declared via `hyperhive.extraMcpServers` (those stay stdio bridges). #[must_use] pub fn render_claude_config() -> String { + let config = serde_json::json!({ "mcpServers": build_mcp_servers() }); + serde_json::to_string_pretty(&config).unwrap_or_else(|_| "{}".into()) +} + +/// The set of MCP server names claude is configured with this turn — the +/// keys of the rendered `--mcp-config` (built-in hyperhive HTTP surface + +/// any tool-group-permitted extra stdio servers). The harness compares +/// this against the per-turn `system`/`init` event's `mcp_servers` to +/// detect a configured server that failed to connect or was dropped +/// (the MCP-health instrumentation). +#[must_use] +pub fn configured_server_names() -> Vec { + build_mcp_servers().into_iter().map(|(k, _)| k).collect() +} + +/// Build the `mcpServers` map claude gets in its `--mcp-config`: the +/// built-in hyperhive HTTP surface plus any tool-group-permitted extra +/// stdio servers. Shared by [`render_claude_config`] (serialises it) and +/// [`configured_server_names`] (lists its keys) so the two never drift. +fn build_mcp_servers() -> serde_json::Map { let mut servers = serde_json::Map::new(); // The built-in hyperhive surface is served exclusively over streamable // HTTP by the persistent `hive-mcp-http` daemon (loopback, inside the @@ -344,8 +364,7 @@ pub fn render_claude_config() -> String { }), ); } - let config = serde_json::json!({ "mcpServers": servers }); - serde_json::to_string_pretty(&config).unwrap_or_else(|_| "{}".into()) + servers } #[cfg(test)] diff --git a/hive-agent/src/turn.rs b/hive-agent/src/turn.rs index 644fa2a9..f099d7ab 100644 --- a/hive-agent/src/turn.rs +++ b/hive-agent/src/turn.rs @@ -628,6 +628,7 @@ impl Sink for BusSink<'_> { // and is applied from the run's returned `Telemetry` (see `drive_turn` // → `apply_telemetry`). self.bus.observe_stream(event); + self.bus.observe_mcp_health(event); self.bus.emit(LiveEvent::Stream(event.clone())); }