feat: add infra diagnostics tool to the Moose Dev MCP (#2915)

<!-- CURSOR_SUMMARY -->
> [!NOTE]
> Adds a new MCP tool `diagnose_infrastructure` with multiple ClickHouse
diagnostics and wires it into the server alongside tests and docs
updates.
> 
> - **MCP Server**:
> - Registers new tool `diagnose_infrastructure` in `list_tools` and
routes in `call_tool`.
> - Updates server instructions to mention diagnostics and source
browsing.
>   - Updates tests to account for 6 tools and validates tool names.
> - **New Tool: `infra_issues` (ClickHouse diagnostics)**:
> - Core module `infra_issues/mod.rs` with params parsing, severity
filtering, Redis-backed infra map usage, and aggregated JSON output with
summaries.
>   - Diagnostic providers:
> - `MutationDiagnostic` (`system.mutations`): stuck/failed mutations.
> - `PartsDiagnostic` (`system.parts`): excessive parts per partition.
>     - `MergeDiagnostic` (`system.merges`): long-running merges.
> - `ErrorStatsDiagnostic` (`system.errors`): top recurring errors
(system-wide).
> - `S3QueueDiagnostic` (`system.s3queue_log`): ingestion failures
(S3Queue tables).
> - `ReplicationDiagnostic`
(`system.replication_queue`/`system.replicas`): lag and health issues
(replicated tables).
> - `MergeFailureDiagnostic` (`system.metrics`): background merge
failures (system-wide).
> - `StoppedOperationsDiagnostic`: merges/replication possibly stopped.
> - **Tool Registry**:
>   - Exposes `infra_issues` in `tools/mod.rs`.
> - **Misc**:
> - `get_source.rs`: remove unused `SourceLocationNotAvailable` error
variant.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
7c40c36. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: callicles

apps/framework-cli/src/mcp/server.rs

@@ -14,7 +14,9 @@ use rmcp::{
 use std::sync::Arc;
 
 use super::embedded_docs;
-use super::tools::{create_error_result, get_source, infra_map, logs, query_olap, sample_stream};
+use super::tools::{
+    create_error_result, get_source, infra_issues, infra_map, logs, query_olap, sample_stream,
+};
 use crate::cli::processing_coordinator::ProcessingCoordinator;
 use crate::infrastructure::olap::clickhouse::config::ClickHouseConfig;
 use crate::infrastructure::redis::redis_client::RedisClient;
@@ -69,7 +71,7 @@ impl ServerHandler for MooseMcpHandler {
                 website_url: None,
             },
             instructions: Some(
-                "Moose MCP Server - Access dev server logs, infrastructure map, query the OLAP database, sample streaming topics, and browse embedded Moose documentation"
+                "Moose MCP Server - Access dev server logs, infrastructure map, diagnose infrastructure issues, query the OLAP database, sample streaming topics, browse code sources, and read embedded Moose documentation"
                     .to_string(),
             ),
         }
@@ -84,6 +86,7 @@ impl ServerHandler for MooseMcpHandler {
             tools: vec![
                 logs::tool_definition(),
                 infra_map::tool_definition(),
+                infra_issues::tool_definition(),
                 query_olap::tool_definition(),
                 sample_stream::tool_definition(),
                 get_source::tool_definition(),
@@ -108,6 +111,12 @@ impl ServerHandler for MooseMcpHandler {
                 self.redis_client.clone(),
             )
             .await),
+            "diagnose_infrastructure" => Ok(infra_issues::handle_call(
+                param.arguments.as_ref(),
+                self.redis_client.clone(),
+                &self.clickhouse_config,
+            )
+            .await),
             "query_olap" => Ok(query_olap::handle_call(
                 &self.clickhouse_config,
                 param.arguments.as_ref(),
@@ -219,19 +228,21 @@ mod tests {
         // Test that all expected tools are returned
         let logs_tool = logs::tool_definition();
         let infra_tool = infra_map::tool_definition();
+        let infra_issues_tool = infra_issues::tool_definition();
         let olap_tool = query_olap::tool_definition();
         let stream_tool = sample_stream::tool_definition();
         let get_source_tool = get_source::tool_definition();
 
-        // Ensure we have 5 tools
+        // Ensure we have 6 tools
         let all_tools = vec![
             &logs_tool,
             &infra_tool,
+            &infra_issues_tool,
             &olap_tool,
             &stream_tool,
             &get_source_tool,
         ];
-        assert_eq!(all_tools.len(), 5);
+        assert_eq!(all_tools.len(), 6);
 
         // Verify each tool has required fields
         for tool in all_tools {
@@ -250,18 +261,21 @@ mod tests {
             "query_olap",
             "get_stream_sample",
             "get_source",
+            "diagnose_infrastructure",
         ];
 
         let logs_tool = logs::tool_definition();
         let infra_tool = infra_map::tool_definition();
         let olap_tool = query_olap::tool_definition();
         let stream_tool = sample_stream::tool_definition();
         let get_source_tool = get_source::tool_definition();
+        let infra_issues_tool = infra_issues::tool_definition();
 
         assert_eq!(logs_tool.name, expected_tools[0]);
         assert_eq!(infra_tool.name, expected_tools[1]);
         assert_eq!(olap_tool.name, expected_tools[2]);
         assert_eq!(stream_tool.name, expected_tools[3]);
         assert_eq!(get_source_tool.name, expected_tools[4]);
+        assert_eq!(infra_issues_tool.name, expected_tools[5]);
     }
 }

apps/framework-cli/src/mcp/tools/get_source.rs

@@ -29,13 +29,6 @@ pub enum GetSourceError {
         component_type: String,
         name: String,
     },
-
-    #[allow(dead_code)]
-    #[error("Source location not available for {component_type}/{name}")]
-    SourceLocationNotAvailable {
-        component_type: String,
-        name: String,
-    },
 }
 
 /// Parameters for the get_source tool

apps/framework-cli/src/mcp/tools/infra_issues/errors.rs

@@ -0,0 +1,129 @@
+//! Diagnostic provider for checking system-wide errors
+
+use log::debug;
+use serde_json::{json, Map, Value};
+
+use super::{Component, DiagnoseError, DiagnosticProvider, Issue, Severity};
+use crate::infrastructure::olap::clickhouse::client::ClickHouseClient;
+use crate::infrastructure::olap::clickhouse::config::ClickHouseConfig;
+use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+
+/// Query timeout for diagnostic checks (30 seconds)
+const DIAGNOSTIC_QUERY_TIMEOUT_SECS: u64 = 30;
+
+/// Diagnostic provider for checking system-wide errors
+pub struct ErrorStatsDiagnostic;
+
+#[async_trait::async_trait]
+impl DiagnosticProvider for ErrorStatsDiagnostic {
+    fn name(&self) -> &str {
+        "ErrorStatsDiagnostic"
+    }
+
+    fn applicable_to(&self, _component: &Component, _engine: Option<&ClickhouseEngine>) -> bool {
+        // Error stats are system-wide, not component-specific
+        // This should be run separately outside the component loop
+        false
+    }
+
+    fn is_system_wide(&self) -> bool {
+        true
+    }
+
+    async fn diagnose(
+        &self,
+        component: &Component,
+        _engine: Option<&ClickhouseEngine>,
+        config: &ClickHouseConfig,
+        _since: Option<&str>,
+    ) -> Result<Vec<Issue>, DiagnoseError> {
+        let client = ClickHouseClient::new(config)
+            .map_err(|e| DiagnoseError::ClickHouseConnection(format!("{}", e)))?;
+
+        // Get recent errors with significant counts
+        let query = "SELECT
+                name,
+                value,
+                last_error_time,
+                last_error_message
+             FROM system.errors
+             WHERE value > 0
+             ORDER BY value DESC
+             LIMIT 10
+             FORMAT JSON";
+
+        debug!("Executing errors query: {}", query);
+
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(DIAGNOSTIC_QUERY_TIMEOUT_SECS),
+            client.execute_sql(query),
+        )
+        .await
+        .map_err(|_| DiagnoseError::QueryTimeout(DIAGNOSTIC_QUERY_TIMEOUT_SECS))?
+        .map_err(|e| DiagnoseError::QueryFailed(format!("{}", e)))?;
+
+        let json_response: Value = serde_json::from_str(&result)
+            .map_err(|e| DiagnoseError::ParseError(format!("{}", e)))?;
+
+        let data = json_response
+            .get("data")
+            .and_then(|v| v.as_array())
+            .ok_or_else(|| {
+                DiagnoseError::ParseError("Missing 'data' field in response".to_string())
+            })?;
+
+        let mut issues = Vec::new();
+
+        for row in data {
+            let name = row
+                .get("name")
+                .and_then(|v| v.as_str())
+                .unwrap_or("UNKNOWN");
+            let value = row.get("value").and_then(|v| v.as_u64()).unwrap_or(0);
+            let last_error_message = row
+                .get("last_error_message")
+                .and_then(|v| v.as_str())
+                .unwrap_or("");
+
+            // Skip if no occurrences
+            if value == 0 {
+                continue;
+            }
+
+            let severity = if value > 100 {
+                Severity::Error
+            } else if value > 10 {
+                Severity::Warning
+            } else {
+                Severity::Info
+            };
+
+            let mut details = Map::new();
+            details.insert("error_name".to_string(), json!(name));
+            details.insert("occurrence_count".to_string(), json!(value));
+            details.insert(
+                "last_error_time".to_string(),
+                row.get("last_error_time").cloned().unwrap_or(json!("")),
+            );
+            if !last_error_message.is_empty() {
+                details.insert("last_error_message".to_string(), json!(last_error_message));
+            }
+
+            issues.push(Issue {
+                severity,
+                source: "system.errors".to_string(),
+                component: component.clone(),
+                error_type: "system_error".to_string(),
+                message: format!("Error '{}' occurred {} times. Last: {}", name, value, last_error_message),
+                details,
+                suggested_action: "Review error pattern and recent query logs. Check ClickHouse server logs for more details.".to_string(),
+                related_queries: vec![
+                    "SELECT * FROM system.errors WHERE value > 0 ORDER BY value DESC".to_string(),
+                    format!("SELECT * FROM system.query_log WHERE exception LIKE '%{}%' ORDER BY event_time DESC LIMIT 10", name),
+                ],
+            });
+        }
+
+        Ok(issues)
+    }
+}

apps/framework-cli/src/mcp/tools/infra_issues/merge_failures.rs

@@ -0,0 +1,105 @@
+//! Diagnostic provider for checking merge failures from system.metrics
+
+use log::debug;
+use serde_json::{json, Map, Value};
+
+use super::{Component, DiagnoseError, DiagnosticProvider, Issue, Severity};
+use crate::infrastructure::olap::clickhouse::client::ClickHouseClient;
+use crate::infrastructure::olap::clickhouse::config::ClickHouseConfig;
+use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+
+/// Query timeout for diagnostic checks (30 seconds)
+const DIAGNOSTIC_QUERY_TIMEOUT_SECS: u64 = 30;
+
+/// Diagnostic provider for checking merge failures from system.metrics
+pub struct MergeFailureDiagnostic;
+
+#[async_trait::async_trait]
+impl DiagnosticProvider for MergeFailureDiagnostic {
+    fn name(&self) -> &str {
+        "merge_failures"
+    }
+
+    fn applicable_to(&self, _component: &Component, _engine: Option<&ClickhouseEngine>) -> bool {
+        // Merge failures from system.metrics are system-wide, not component-specific
+        // This should be run separately outside the component loop
+        false
+    }
+
+    fn is_system_wide(&self) -> bool {
+        true
+    }
+
+    async fn diagnose(
+        &self,
+        component: &Component,
+        _engine: Option<&ClickhouseEngine>,
+        config: &ClickHouseConfig,
+        _since: Option<&str>,
+    ) -> Result<Vec<Issue>, DiagnoseError> {
+        let client = ClickHouseClient::new(config)
+            .map_err(|e| DiagnoseError::ClickHouseConnection(format!("{}", e)))?;
+
+        let mut issues = Vec::new();
+
+        // Check system.metrics for background merge failures
+        // Note: This is a system-wide metric, not per-table, but we report it per-table for context
+        let metrics_query =
+            "SELECT value FROM system.metrics WHERE metric = 'FailedBackgroundMerges' FORMAT JSON";
+
+        debug!("Executing merge failure metrics query: {}", metrics_query);
+
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(DIAGNOSTIC_QUERY_TIMEOUT_SECS),
+            client.execute_sql(metrics_query),
+        )
+        .await
+        .map_err(|_| DiagnoseError::QueryTimeout(DIAGNOSTIC_QUERY_TIMEOUT_SECS))?
+        .map_err(|e| DiagnoseError::QueryFailed(format!("{}", e)))?;
+
+        let json_response: Value = serde_json::from_str(&result)
+            .map_err(|e| DiagnoseError::ParseError(format!("{}", e)))?;
+
+        let failed_merges = json_response
+            .get("data")
+            .and_then(|v| v.as_array())
+            .and_then(|arr| arr.first())
+            .and_then(|row| row.get("value"))
+            .and_then(|v| v.as_u64())
+            .unwrap_or(0);
+
+        if failed_merges > 0 {
+            let severity = if failed_merges > 10 {
+                Severity::Error
+            } else {
+                Severity::Warning
+            };
+
+            let mut details = Map::new();
+            details.insert("failed_merges".to_string(), json!(failed_merges));
+
+            issues.push(Issue {
+                severity,
+                source: "system.metrics".to_string(),
+                component: component.clone(),
+                error_type: "merge_failures".to_string(),
+                message: format!(
+                    "Background merge failures detected: {} failed merges currently in system. This may affect table maintenance.",
+                    failed_merges
+                ),
+                details,
+                suggested_action: "Check system.errors for merge failure details. Review disk space and memory availability. Consider increasing merge-related settings if failures persist.".to_string(),
+                related_queries: vec![
+                    "SELECT * FROM system.errors WHERE name LIKE '%Merge%' ORDER BY last_error_time DESC LIMIT 10".to_string(),
+                    "SELECT * FROM system.metrics WHERE metric LIKE '%Merge%'".to_string(),
+                    format!(
+                        "SELECT * FROM system.merges WHERE database = '{}' AND table = '{}'",
+                        config.db_name, component.name
+                    ),
+                ],
+            });
+        }
+
+        Ok(issues)
+    }
+}