feat(#2038): surface degraded mcp servers at turn start (observability)

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<String> {
+    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());
+    }
 }

hive-agent/src/mcp_config.rs

@@ -290,6 +290,26 @@ fn load_extra_mcp() -> std::collections::BTreeMap<String, ExtraMcpServer> {
 /// 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<String> {
+    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<String, serde_json::Value> {
     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)]

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