fix(cluster): require recorded history evidence

CHANGELOG.md

@@ -65,6 +65,9 @@ under **Unreleased** until a new version is selected and published.
 
 ### Fixed
 
+- Required at least two recorded time buckets before exposing or executing
+  resource-growth analysis, so readable-but-empty audit and partition metadata
+  no longer masquerade as historical evidence.
 - 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

doris_mcp_server/tools/capability_detector.py

@@ -244,17 +244,6 @@ class CapabilityRouteChangedError(CapabilityDetectionError):
             "SHOW COMPUTE GROUPS",
             ("compute_group_metadata_readable",),
         ),
-        (
-            ("SELECT `time` FROM internal.__internal_schema.audit_log LIMIT 1"),
-            ("metrics_history_readable",),
-        ),
-        (
-            (
-                "SELECT CREATE_TIME, DATA_LENGTH, INDEX_LENGTH "
-                "FROM information_schema.partitions LIMIT 1"
-            ),
-            ("resource_storage_history_readable",),
-        ),
     ),
     "doris_pipeline": (
         (
@@ -516,6 +505,9 @@ async def detect_domain(
                         probes
                     )
                 elif domain_name == "doris_cluster":
+                    probes.update(
+                        await self._probe_cluster_history_sources(auth_context)
+                    )
                     probes.update(await self._safe_probe_cluster_services(auth_context))
                     probes.update(_combine_cluster_evidence_probes(probes))
                 elif domain_name == "doris_pipeline":
@@ -753,6 +745,76 @@ async def _probe_query_services(
         )
         return probes
 
+    async def _probe_cluster_history_sources(
+        self,
+        auth_context: Any | None,
+    ) -> dict[str, CapabilityProbeEvidence]:
+        """Require recorded history, not merely readable metadata objects."""
+        contracts = (
+            (
+                "metrics_history_readable",
+                "SELECT COUNT(DISTINCT DATE(`time`)) AS evidence_bucket_count "
+                "FROM internal.__internal_schema.audit_log "
+                "WHERE `time` >= DATE_SUB(NOW(), INTERVAL 3650 DAY)",
+                "AUDIT_HISTORY_RECORDED",
+                "AUDIT_HISTORY_INSUFFICIENT",
+            ),
+            (
+                "resource_storage_history_readable",
+                "SELECT COUNT(DISTINCT DATE(CREATE_TIME)) "
+                "AS evidence_bucket_count "
+                "FROM information_schema.partitions "
+                "WHERE CREATE_TIME >= DATE_SUB(NOW(), INTERVAL 3650 DAY)",
+                "PARTITION_CREATION_HISTORY_RECORDED",
+                "PARTITION_CREATION_HISTORY_INSUFFICIENT",
+            ),
+        )
+        evidence: dict[str, CapabilityProbeEvidence] = {}
+        route = self.route_identity(auth_context)
+        for index, (
+            probe_id,
+            statement,
+            supported_reason,
+            insufficient_reason,
+        ) in enumerate(contracts):
+            session_id = (
+                f"capability-cluster:history:{index}:"
+                f"{route.fingerprint[:12]}"
+            )
+            async with (
+                self._connection_manager.get_connection_context_for_auth_context(
+                    session_id,
+                    auth_context,
+                ) as connection
+            ):
+                rows, probe = await self._probe_rows(
+                    connection,
+                    statement,
+                    probe_id,
+                )
+            if probe.status is CapabilityProbeStatus.SUPPORTED:
+                bucket_count = _nonnegative_int(
+                    _row_value(rows[0], "evidence_bucket_count")
+                    if rows
+                    else None
+                )
+                probe = CapabilityProbeEvidence(
+                    probe_id=probe_id,
+                    status=(
+                        CapabilityProbeStatus.SUPPORTED
+                        if bucket_count >= 2
+                        else CapabilityProbeStatus.UNKNOWN
+                    ),
+                    reason_code=(
+                        supported_reason
+                        if bucket_count >= 2
+                        else insufficient_reason
+                    ),
+                    evidence_sources=("recorded_history_probe",),
+                )
+            evidence[probe_id] = probe
+        return evidence
+
     async def _probe_search_match_syntax(
         self,
         auth_context: Any | None,
@@ -1886,6 +1948,16 @@ def _combine_cluster_evidence_probes(
             evidence_sources=("runtime_probe",),
         )
     )
+    if audit is not None and storage is not None and not any(
+        source.status is CapabilityProbeStatus.SUPPORTED
+        for source in (audit, storage)
+    ):
+        full = CapabilityProbeEvidence(
+            probe_id=full.probe_id,
+            status=full.status,
+            reason_code="RESOURCE_HISTORY_UNAVAILABLE",
+            evidence_sources=full.evidence_sources,
+        )
 
     def partial_source(
         probe_id: str,
@@ -2349,6 +2421,21 @@ def _row_value(
     return None
 
 
+def _nonnegative_int(value: Any | None) -> int:
+    if isinstance(value, bool):
+        return 0
+    if isinstance(value, int):
+        parsed = value
+    elif isinstance(value, str):
+        try:
+            parsed = int(value.strip())
+        except ValueError:
+            return 0
+    else:
+        return 0
+    return max(0, parsed)
+
+
 def _component_version(value: Any) -> DorisVersion:
     raw = "" if value is None else str(value).strip()
     if not raw:

doris_mcp_server/tools/domain_catalog.py

@@ -1513,7 +1513,8 @@ def _domain(
                 "analyze_resource_growth",
                 "Analyze resource growth",
                 "Analyze recorded resource-growth evidence without inventing "
-                "missing history. For an unqualified cluster-history request, "
+                "missing history. A series requires at least two recorded time "
+                "buckets. For an unqualified cluster-history request, "
                 "omit resource so every currently usable recorded series is "
                 "attempted and partial evidence is preserved.",
                 _input_schema(

doris_mcp_server/utils/cluster_runtime.py

@@ -749,7 +749,22 @@ async def analyze_resource_growth(
                     "value": _number(_row_lookup(row, "value")),
                 }
                 for row in rows
+                if _row_lookup(row, "bucket") is not None
             ]
+            if len(points) < 2:
+                warnings.append(
+                    f"{resource_name} history has fewer than two recorded "
+                    "time buckets."
+                )
+                evidence.append(
+                    {
+                        "resource": resource_name,
+                        "success": False,
+                        "reason_code": "RESOURCE_HISTORY_INSUFFICIENT",
+                        "points": len(points),
+                    }
+                )
+                continue
             series[resource_name] = points
             evidence.append(
                 {

test/tools/test_capability_detector.py

@@ -44,6 +44,16 @@
     "AND TABLE_SCHEMA NOT IN ('information_schema', 'mysql') "
     "ORDER BY TABLE_SCHEMA, TABLE_NAME, ORDINAL_POSITION LIMIT 8"
 )
+_AUDIT_HISTORY_PROBE_SQL = (
+    "SELECT COUNT(DISTINCT DATE(`time`)) AS evidence_bucket_count "
+    "FROM internal.__internal_schema.audit_log "
+    "WHERE `time` >= DATE_SUB(NOW(), INTERVAL 3650 DAY)"
+)
+_STORAGE_HISTORY_PROBE_SQL = (
+    "SELECT COUNT(DISTINCT DATE(CREATE_TIME)) AS evidence_bucket_count "
+    "FROM information_schema.partitions "
+    "WHERE CREATE_TIME >= DATE_SUB(NOW(), INTERVAL 3650 DAY)"
+)
 
 
 class _ProbeConnection:
@@ -113,6 +123,8 @@ async def execute(
                     "DATA_LENGTH": 8,
                 }
             ],
+            _AUDIT_HISTORY_PROBE_SQL: [{"evidence_bucket_count": 3}],
+            _STORAGE_HISTORY_PROBE_SQL: [{"evidence_bucket_count": 2}],
         }
         return SimpleNamespace(data=self.row_overrides.get(sql, rows.get(sql, [])))
 
@@ -222,10 +234,7 @@ async def test_detector_builds_version_vector_and_extends_domains_lazily() -> No
 @pytest.mark.asyncio
 async def test_cluster_history_keeps_storage_fallback_without_audit_access() -> None:
     connection = _ProbeConnection()
-    audit_probe = (
-        "SELECT `time` FROM internal.__internal_schema.audit_log LIMIT 1"
-    )
-    connection.failures[audit_probe] = RuntimeError(
+    connection.failures[_AUDIT_HISTORY_PROBE_SQL] = RuntimeError(
         "Access denied; user lacks SELECT privilege"
     )
     manager = _ProbeConnectionManager(connection)
@@ -251,6 +260,43 @@ 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_history_requires_two_recorded_time_buckets() -> None:
+    connection = _ProbeConnection()
+    connection.row_overrides[_AUDIT_HISTORY_PROBE_SQL] = [
+        {"evidence_bucket_count": 1}
+    ]
+    connection.row_overrides[_STORAGE_HISTORY_PROBE_SQL] = [
+        {"evidence_bucket_count": 0}
+    ]
+    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)
+
+    audit = cluster.probe("metrics_history_readable")
+    storage = cluster.probe("resource_storage_history_readable")
+    assert audit is not None
+    assert audit.status is CapabilityProbeStatus.UNKNOWN
+    assert audit.reason_code == "AUDIT_HISTORY_INSUFFICIENT"
+    assert storage is not None
+    assert storage.status is CapabilityProbeStatus.UNKNOWN
+    assert storage.reason_code == "PARTITION_CREATION_HISTORY_INSUFFICIENT"
+    assert (
+        cluster.probe("resource_history_all_sources_readable").status
+        is CapabilityProbeStatus.UNKNOWN
+    )
+    assert (
+        cluster.probe("resource_history_all_sources_readable").reason_code
+        == "RESOURCE_HISTORY_UNAVAILABLE"
+    )
+
+
 @pytest.mark.asyncio
 async def test_cluster_active_tasks_accepts_read_only_query_view_fallback() -> None:
     connection = _ProbeConnection()

test/utils/test_cluster_runtime.py

@@ -346,3 +346,33 @@ async def test_resource_growth_rejects_unknown_resource_before_sql() -> None:
 
     assert error.value.reason_code == "CLUSTER_ARGUMENT_INVALID"
     assert manager.calls == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "rows",
+    [
+        [],
+        [{"bucket": "2026-07-31", "value": 15}],
+    ],
+)
+async def test_resource_growth_rejects_insufficient_recorded_history(
+    rows: list[dict[str, Any]],
+) -> None:
+    query_sql = (
+        "SELECT DATE(`time`) AS bucket, COUNT(*) AS value "
+        "FROM internal.__internal_schema.audit_log "
+        "WHERE `time` >= DATE_SUB(NOW(), INTERVAL %s DAY) "
+        "GROUP BY bucket ORDER BY bucket"
+    )
+    runtime, manager, _ = _runtime(rows={query_sql: rows})
+
+    with pytest.raises(ClusterRuntimeFailure) as error:
+        await runtime.analyze_resource_growth(
+            resource="query_volume",
+            window_days=30,
+            granularity="day",
+        )
+
+    assert error.value.reason_code == "RESOURCE_HISTORY_UNAVAILABLE"
+    assert manager.calls == [query_sql]