fix(cluster): honor read-only active task fallbacks

CHANGELOG.md

@@ -65,6 +65,12 @@ under **Unreleased** until a new version is selected and published.
 
 ### Fixed
 
+- Aligned `doris_cluster.list_active_tasks` capability detection with its
+  read-only execution fallbacks, so restricted Doris accounts can use the
+  `information_schema.active_queries` or process-list source when
+  `SHOW PROC \"/current_queries\"` is unavailable. The public input schema now
+  advertises only the query and compaction task types implemented by the
+  runtime.
 - Prevented MetricFlow sidecar processes from inheriting Doris credentials,
   bearer tokens, OAuth/JWT secrets, and unrelated MCP Server environment
   configuration by launching each provider with a fixed minimal environment.

doris_mcp_server/tools/capability_detector.py

@@ -204,10 +204,21 @@ class CapabilityRouteChangedError(CapabilityDetectionError):
         (
             'SHOW PROC "/current_queries"',
             (
-                "legacy_task_views_readable",
+                "current_queries_proc_readable",
                 "unified_task_progress_readable",
             ),
         ),
+        (
+            (
+                "SELECT 1 AS active_query_probe "
+                "FROM information_schema.active_queries LIMIT 1"
+            ),
+            ("active_queries_view_readable",),
+        ),
+        (
+            "SHOW FULL PROCESSLIST",
+            ("processlist_readable",),
+        ),
         (
             (
                 "SELECT BE_ID, METRIC_NAME "
@@ -1849,6 +1860,16 @@ def _combine_query_evidence_probe(
 def _combine_cluster_evidence_probes(
     probes: Mapping[str, CapabilityProbeEvidence],
 ) -> dict[str, CapabilityProbeEvidence]:
+    active_tasks = _combine_any_runtime_probe(
+        "legacy_task_views_readable",
+        probes,
+        (
+            "current_queries_proc_readable",
+            "active_queries_view_readable",
+            "processlist_readable",
+        ),
+        supported_reason="LEGACY_TASK_VIEW_READABLE",
+    )
     audit = probes.get("metrics_history_readable")
     storage = probes.get("resource_storage_history_readable")
     full = (
@@ -1905,6 +1926,7 @@ def partial_source(
         reason_code="PARTITION_CREATION_HISTORY_ONLY",
     )
     return {
+        active_tasks.probe_id: active_tasks,
         full.probe_id: full,
         audit_only.probe_id: audit_only,
         storage_only.probe_id: storage_only,

doris_mcp_server/tools/domain_catalog.py

@@ -1415,10 +1415,13 @@ def _domain(
                 "doris_cluster",
                 "list_active_tasks",
                 "List active tasks",
-                "List visible query, load, schema-change, and compaction tasks.",
+                "List visible active query and compaction tasks.",
                 _input_schema(
                     {
-                        "task_types": _string_array("Task types to include."),
+                        "task_types": _string_array(
+                            "Task types to include.",
+                            enum=("query", "compaction"),
+                        ),
                         "states": _string_array("Task states to include."),
                         "limit": _integer("Maximum results.", minimum=1),
                     }

test/tools/test_capability_detector.py

@@ -251,6 +251,36 @@ async def test_cluster_history_keeps_storage_fallback_without_audit_access() ->
     assert storage.reason_code == "PARTITION_CREATION_HISTORY_ONLY"
 
 
+@pytest.mark.asyncio
+async def test_cluster_active_tasks_accepts_read_only_query_view_fallback() -> None:
+    connection = _ProbeConnection()
+    proc_probe = 'SHOW PROC "/current_queries"'
+    connection.failures[proc_probe] = RuntimeError(
+        "Access denied; user lacks ADMIN privilege"
+    )
+    manager = _ProbeConnectionManager(connection)
+    detector = DorisCapabilityDetector(manager)  # type: ignore[arg-type]
+    base = await detector.detect_base(
+        None,
+        capability_generation=1,
+        provider_generation="provider.cluster",
+    )
+
+    cluster = await detector.detect_domain(base, "doris_cluster", None)
+
+    assert (
+        cluster.probe("current_queries_proc_readable").status
+        is not CapabilityProbeStatus.SUPPORTED
+    )
+    assert (
+        cluster.probe("active_queries_view_readable").status
+        is CapabilityProbeStatus.SUPPORTED
+    )
+    active_tasks = cluster.probe("legacy_task_views_readable")
+    assert active_tasks.status is CapabilityProbeStatus.SUPPORTED
+    assert active_tasks.reason_code == "LEGACY_TASK_VIEW_READABLE"
+
+
 @pytest.mark.asyncio
 async def test_lakehouse_probes_derive_target_sensitive_advanced_facets() -> None:
     connection = _ProbeConnection()

test/tools/test_domain_catalog.py

@@ -131,6 +131,18 @@ def test_adbc_is_inside_query_and_cluster_has_exactly_eleven_children() -> None:
     assert len(cluster.children) == 11
 
 
+def test_cluster_active_task_contract_matches_runtime_sources() -> None:
+    child = DORIS_DOMAIN_CATALOG.resolve_child(
+        "doris_cluster",
+        "list_active_tasks",
+    )
+    task_types = _wire_input(child)["properties"]["task_types"]
+
+    assert task_types["items"]["enum"] == ["query", "compaction"]
+    assert "load" not in child.canonical_description.casefold()
+    assert "schema-change" not in child.canonical_description.casefold()
+
+
 def test_every_child_uses_the_exact_feature_matrix_contract() -> None:
     feature_contracts = {
         feature.feature_id: feature.support_contract

test/utils/test_cluster_runtime.py

@@ -178,6 +178,41 @@ async def test_list_cluster_nodes_normalizes_real_fe_and_be_rows() -> None:
     assert manager.calls == ["SHOW FRONTENDS", "SHOW BACKENDS"]
 
 
+@pytest.mark.asyncio
+async def test_active_tasks_falls_back_to_read_only_active_queries_view() -> None:
+    proc_statement = 'SHOW PROC "/current_queries"'
+    view_statement = "SELECT * FROM information_schema.active_queries"
+    runtime, manager, _ = _runtime(
+        rows={
+            view_statement: [
+                {
+                    "QUERY_ID": "query-1",
+                    "STATE": "RUNNING",
+                    "COMMAND": "Query",
+                }
+            ]
+        },
+        failures={proc_statement: RuntimeError(1105, "access denied")},
+    )
+
+    result = await runtime.list_active_tasks(
+        task_types=["query"],
+        states=None,
+        limit=10,
+    )
+
+    assert result["status"] == "success"
+    assert result["data"]["items"] == [
+        {
+            "query_id": "query-1",
+            "state": "RUNNING",
+            "command": "Query",
+            "task_type": "query",
+        }
+    ]
+    assert manager.calls == [proc_statement, view_statement]
+
+
 @pytest.mark.asyncio
 async def test_memory_stats_only_returns_observed_metrics() -> None:
     runtime, _, _ = _runtime()