#!/usr/bin/env python3
"""Collect diagnosis evidence and run gate checks.

Usage:
    python3 collect_evidence.py <cluster_id> <start_time> <end_time> [--parent-id PARENT_ID]

Outputs structured evidence summary + gate check verdicts in the same
format consumed by tidb-clinic-diagnosis SKILL.md (sections A1-A6 + Gates).

The script saves categorized evidence JSON files with monitoring/Loki links for
the frontend right-panel artifact under the current workspace.

Zero external dependencies — uses only stdlib + clinic_api.py from clinic-api skill.
"""
from __future__ import annotations

import json
import hashlib
import math
import os
import re
import sys
import threading
import time
from datetime import datetime, timezone
from urllib.parse import quote, urlencode

_EVIDENCES_DIRNAME = "evidences"
COMPONENT_HEALTH_LOOKBACK_SECONDS = 15 * 60
COMPONENT_HEALTH_TAIL_SECONDS = 5 * 60
FOCUSED_LOG_LOOKBACK_SECONDS = 15 * 60
FOCUSED_LOG_TAIL_SECONDS = 5 * 60
TRANSACTION_LOCK_ROUTE_SIGNAL_TERMS = [
    "deadlock",
    "executor:1213",
    "lock wait",
    "lock_wait",
    "pessimistic",
    "write conflict",
    "txnheartbeat",
    "txn heart beat",
    "pessimisticlocknotfound",
]
TRANSACTION_LOG_RECALL_TERMS = TRANSACTION_LOCK_ROUTE_SIGNAL_TERMS + [
    "foreign key",
    "foreign_key",
    "try again later",
]
TRANSACTION_LOG_SIGNAL_REGEX = (
    r"deadlock|executor:1213|lock wait|lock_wait|pessimistic|write conflict|"
    r"txnheartbeat|txn heart beat|pessimisticlocknotfound|foreign key|foreign_key|"
    r"try again later"
)

# Locate clinic_api.py from the current workspace skill layout.
# Search order:
# 1. workspace builtin clinic-cli
# 2. workspace .all-skills nutshell clinic-api
# 3. repo builtin clinic-cli
# 4. repo nutshell clinic-api (legacy fallback)
_CLINIC_API_SEARCH = [
    os.path.join(os.getcwd(), ".opencode", "skills", "builtin", "clinic-cli", "scripts"),
    os.path.join(os.getcwd(), ".opencode", ".all-skills", "nutshell-skills", "platform", "clinic-api", "scripts"),
    os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "builtin", "clinic-cli", "scripts"),
    os.path.join(
        os.path.dirname(os.path.abspath(__file__)),
        "..", "..", "..", "nutshell-skills", "platform", "clinic-api", "scripts",
    ),
]
for _p in _CLINIC_API_SEARCH:
    if os.path.isfile(os.path.join(_p, "clinic_api.py")):
        sys.path.insert(0, _p)
        break

from clinic_api import ClinicAPI  # noqa: E402


def _short_hash(value: str) -> str:
    return hashlib.sha1(value.encode(errors="replace")).hexdigest()[:12]


def _ensure_perf_log() -> str:
    path = os.environ.get("CLINIC_PERF_LOG", "").strip()
    if not path:
        path = os.path.join(_evidence_base_dir(), "perf.jsonl")
        os.environ["CLINIC_PERF_LOG"] = path
    os.makedirs(os.path.dirname(path), exist_ok=True)
    return path


def _perf_event(event: dict) -> None:
    path = os.environ.get("CLINIC_PERF_LOG", "").strip()
    if not path:
        return
    event.setdefault("ts", time.time())
    try:
        with open(path, "a", encoding="utf-8") as f:
            f.write(json.dumps(event, ensure_ascii=False, sort_keys=True) + "\n")
            f.flush()
    except Exception:
        pass


# ---------------------------------------------------------------------------
# Raw evidence capture
# ---------------------------------------------------------------------------
# Successful upstream API responses are archived verbatim under
# <evidences>/raw/<kind>/ with an append-only raw/manifest.jsonl index, so the
# aggregated evidence files above stay reproducible and auditable offline.
# Disable with --no-raw or COLLECT_EVIDENCE_NO_RAW=1.

_RAW_LOCK = threading.Lock()
_RAW_SEQ = {"value": 0}


def _raw_capture_dir() -> str:
    if os.environ.get("COLLECT_EVIDENCE_NO_RAW", "") == "1":
        return ""
    return os.path.join(_evidence_base_dir(), "raw")


def _slug_raw_label(value: str) -> str:
    slug = re.sub(r"[^a-zA-Z0-9._-]+", "_", str(value or "").strip()).strip("_")
    return slug[:80] or "payload"


def _record_raw(kind: str, label: str, payload: object, *, query: str = "",
                scope: str = "", start: str = "", end: str = "") -> None:
    """Archive one raw upstream response. Archiving must never break collection."""
    try:
        base = _raw_capture_dir()
        if not base:
            return
        with _RAW_LOCK:
            _RAW_SEQ["value"] += 1
            seq = _RAW_SEQ["value"]
        kind_dir = os.path.join(base, kind)
        os.makedirs(kind_dir, exist_ok=True)
        filename = f"{seq:04d}_{_slug_raw_label(label)}.json"
        path = os.path.join(kind_dir, filename)
        with open(path, "w", encoding="utf-8") as f:
            # indent=1 keeps raw files line-oriented so agents can read/grep the
            # smaller payloads (logs/queries/events/meta) directly with plain tools.
            json.dump({
                "kind": kind,
                "label": label,
                "query": query,
                "scope": scope,
                "start": str(start),
                "end": str(end),
                "collected_at": datetime.now(tz=timezone.utc).isoformat(),
                "payload": payload,
            }, f, ensure_ascii=False, default=str, indent=1)
        entry = {
            "file": os.path.join("raw", kind, filename),
            "kind": kind,
            "label": label,
            "query_preview": query[:240],
            "scope": scope,
            "start": str(start),
            "end": str(end),
            "bytes": os.path.getsize(path),
        }
        with _RAW_LOCK:
            with open(os.path.join(base, "manifest.jsonl"), "a", encoding="utf-8") as f:
                f.write(json.dumps(entry, ensure_ascii=False, sort_keys=True) + "\n")
    except Exception:
        pass


# ---------------------------------------------------------------------------
# Time parsing helpers
# ---------------------------------------------------------------------------

def _format_utc_iso(ts: int) -> str:
    return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat().replace("+00:00", "Z")


def _parse_time_to_epoch_seconds(value: object) -> int:
    """Parse epoch seconds or timezone-bearing ISO-like datetimes to epoch seconds."""
    text = str(value).strip()
    if not text:
        raise ValueError("empty time value")

    if re.fullmatch(r"\d+", text):
        if len(text) != 10:
            raise ValueError(
                f"invalid epoch seconds {value!r}; use 10-digit epoch seconds "
                "or ISO 8601 like 2026-03-09T04:14:00Z"
            )
        return int(text)

    normalized = text
    if normalized.endswith(("Z", "z")):
        normalized = f"{normalized[:-1]}+00:00"
    if " " in normalized and "T" not in normalized:
        normalized = normalized.replace(" ", "T", 1)

    try:
        dt = datetime.fromisoformat(normalized)
    except ValueError as exc:
        raise ValueError(
            f"invalid time value {value!r}; use 10-digit epoch seconds, "
            "or timezone-bearing ISO 8601 like 2026-03-09T04:14:00Z"
        ) from exc

    if dt.tzinfo is None:
        raise ValueError(
            f"datetime {value!r} has no timezone; use timezone-bearing ISO 8601 "
            "like 2026-03-09T04:14:00Z, or convert with the time-conversion skill"
        )
    return int(dt.timestamp())


def _canonical_time_window(start_value: object, end_value: object) -> dict[str, str]:
    start_epoch = _parse_time_to_epoch_seconds(start_value)
    end_epoch = _parse_time_to_epoch_seconds(end_value)
    if end_epoch <= start_epoch:
        raise ValueError(
            f"end time must be after start time: start={start_value!r} end={end_value!r}"
        )
    return {
        "start": str(start_epoch),
        "end": str(end_epoch),
        "start_iso": _format_utc_iso(start_epoch),
        "end_iso": _format_utc_iso(end_epoch),
        "input_start": str(start_value),
        "input_end": str(end_value),
    }


# ---------------------------------------------------------------------------
# Prometheus helpers
# ---------------------------------------------------------------------------

def _prom_results(resp: dict) -> list[dict]:
    return resp.get("data", {}).get("result", [])


def _query_range_retryable_error(resp: dict) -> bool:
    if not isinstance(resp, dict) or "error" not in resp:
        return False
    text = str(resp.get("error", "")).lower()
    retry_markers = (
        "status 429",
        "status 503",
        "maxconcurrentrequests",
        "couldn't start executing the request",
        "timeout",
        "temporarily unavailable",
    )
    return any(marker in text for marker in retry_markers)


def _prom_results_strict(resp: dict, label: str) -> list[dict]:
    if not isinstance(resp, dict):
        raise RuntimeError(f"{label}: malformed response type={type(resp).__name__}")
    if "error" in resp:
        raise RuntimeError(f"{label}: {resp['error']}")
    data = resp.get("data")
    if not isinstance(data, dict):
        raise RuntimeError(f"{label}: malformed response missing data object")
    result = data.get("result")
    if not isinstance(result, list):
        raise RuntimeError(f"{label}: malformed response missing result list")
    if not result:
        raise RuntimeError(f"{label}: empty result")
    return result


def _prom_peak(result: dict) -> float:
    peak = _prom_observed_peak(result)
    return peak if peak is not None else 0.0


def _prom_observed_peak(result: dict) -> float | None:
    """Keep missing/non-finite observations distinct from an observed zero."""
    vals = result.get("values", [])
    if not vals:
        return None
    nums = []
    for v in vals:
        try:
            f = float(v[1])
            if not (math.isnan(f) or math.isinf(f)):
                nums.append(f)
        except (ValueError, TypeError, IndexError):
            pass
    return max(nums) if nums else None


def _prom_min(result: dict) -> float:
    vals = result.get("values", [])
    if not vals:
        return 0.0
    nums = []
    for v in vals:
        try:
            f = float(v[1])
            if not (math.isnan(f) or math.isinf(f)):
                nums.append(f)
        except (ValueError, IndexError):
            pass
    return min(nums) if nums else 0.0


def _prom_mean(result: dict) -> float:
    vals = result.get("values", [])
    if not vals:
        return 0.0
    nums = []
    for v in vals:
        try:
            f = float(v[1])
            if not (math.isnan(f) or math.isinf(f)):
                nums.append(f)
        except (ValueError, TypeError, IndexError):
            pass
    return sum(nums) / len(nums) if nums else 0.0


def _prom_median(result: dict) -> float:
    vals = result.get("values", [])
    nums = []
    for v in vals:
        try:
            f = float(v[1])
            if not (math.isnan(f) or math.isinf(f)):
                nums.append(f)
        except (ValueError, IndexError):
            pass
    if not nums:
        return 0.0
    nums.sort()
    return nums[len(nums) // 2]


def _instance_name(result: dict) -> str:
    m = result.get("metric", {})
    return m.get("instance", m.get("component", "unknown"))


def _chart_series(results: list[dict], instance_filter: str | None = None, instance_fn=None) -> list[dict]:
    """Convert Prometheus matrix results to workspace report chart_data.series."""
    _inst = instance_fn or _instance_name
    series: list[dict] = []
    for result in results:
        instance = _inst(result)
        if instance_filter and instance != instance_filter:
            continue
        values = result.get("values", [])
        if not values:
            continue
        series.append({
            "instance": instance,
            "values": values,
        })
    return series


def _chart_data(name: str, promql: str, results: list[dict], instance_filter: str | None = None, instance_fn=None) -> dict | None:
    """Build chart_data payload for the frontend diagnosis metric chart."""
    series = _chart_series(results, instance_filter=instance_filter, instance_fn=instance_fn)
    if not series:
        return None
    return {
        "name": name,
        "promql": promql,
        "series": series,
    }


def _chart_data_ref(promql: str, cluster_id: str, start: str, end: str, step: str = "60s") -> dict:
    return {
        "kind": "promql_range",
        "promql": promql,
        "cluster_id": cluster_id,
        "start": start,
        "end": end,
        "step": step,
    }


AI_METRIC_FIELDS = (
    "name",
    "component",
    "metric_group_id",
    "instance",
    "nodegroup",
    "bucket",
    "peak",
    "mean",
    "quota",
    "utilization",
    "unit",
    "severity",
    "availability",
    "observed_series",
    "total_series",
    "missing_instances",
    "promql",
    "link",
)


def compact_metric_for_ai(metric: dict) -> dict:
    """Return a diagnosis metric without raw time-series chart payloads."""
    return {key: metric.get(key) for key in AI_METRIC_FIELDS if key in metric}


def _slug_metric_group_id(value: str) -> str:
    slug = re.sub(r"[^a-z0-9]+", "_", str(value or "").strip().lower())
    return slug.strip("_") or "metric"


def _ensure_metric_group_ids(metrics: list[dict]) -> None:
    for metric in metrics:
        if not isinstance(metric, dict):
            continue
        if metric.get("metric_group_id"):
            continue
        metric["metric_group_id"] = _slug_metric_group_id(
            metric.get("name") or metric.get("promql") or "metric"
        )


def _attach_metric_chart_refs(metrics: list[dict], cluster_id: str, start: str, end: str) -> None:
    for metric in metrics:
        if not isinstance(metric, dict):
            continue
        promql = metric.get("promql")
        if not promql or metric.get("chart_data_ref"):
            continue
        metric["chart_data_ref"] = _chart_data_ref(str(promql), cluster_id, start, end)


def compact_metrics_for_ai(metrics: list[dict]) -> list[dict]:
    """Compact metrics for agent reasoning without raw chart data references."""
    compact_metrics: list[dict] = []
    for metric in metrics:
        if not isinstance(metric, dict):
            continue
        compact_metrics.append(compact_metric_for_ai(metric))
    return compact_metrics


def _log_level(message: str) -> str:
    upper = message.upper()
    if "FATAL" in upper:
        return "FATAL"
    if "ERROR" in upper:
        return "ERROR"
    if "WARN" in upper:
        return "WARN"
    return "INFO"


def compact_logs_for_ai(logs_evidence: dict) -> dict:
    """Summarize raw log entries as counts without carrying full log rows."""
    entries = logs_evidence.get("entries", []) if isinstance(logs_evidence, dict) else []
    grouped: dict[tuple[str, str, str], dict] = {}

    if isinstance(entries, list):
        for entry in entries:
            if not isinstance(entry, dict):
                continue
            message = str(entry.get("message", "")).strip()
            if not message:
                continue
            component = str(entry.get("component", "") or "unknown")
            level = str(entry.get("level", "") or _log_level(message))
            key = (component, level, message)
            item = grouped.setdefault(
                key,
                {
                    "component": component,
                    "level": level,
                    "message": message,
                    "count": 0,
                    "instances": set(),
                    "link": entry.get("link", ""),
                },
            )
            item["count"] += 1
            instance = entry.get("instance")
            if instance:
                item["instances"].add(str(instance))
            if not item.get("link") and entry.get("link"):
                item["link"] = entry.get("link")

    log_counts = []
    for item in grouped.values():
        item["instances"] = sorted(item["instances"])
        log_counts.append(item)

    log_counts.sort(key=lambda item: (-item["count"], item["component"], item["level"], item["message"]))
    return {
        "log_counts": log_counts,
        "warn_count": logs_evidence.get("warn_count", 0) if isinstance(logs_evidence, dict) else 0,
        "loki_url": logs_evidence.get("loki_url", "") if isinstance(logs_evidence, dict) else "",
    }


def _series_label(result: dict, *keys: str, default: str = "unknown") -> str:
    metric = result.get("metric", {}) if isinstance(result, dict) else {}
    if not isinstance(metric, dict):
        return default
    for key in keys:
        value = metric.get(key)
        if value:
            return str(value)
    return default


def compact_workload_for_ai(data: dict) -> list[dict]:
    """Pair statement rate with affected-row rate by SQL type."""
    statements: dict[str, dict] = {}
    for result in data.get("statement_count", []):
        sql_type = _series_label(result, "type", "sql_type")
        statements[sql_type] = {
            "sql_type": sql_type,
            "statement_rate_peak": round(_prom_peak(result), 4),
            "statement_rate_mean": round(_prom_mean(result), 4),
        }

    affected_rows: dict[str, dict] = {}
    for result in data.get("affected_rows", []):
        sql_type = _series_label(result, "sql_type", "type")
        affected_rows[sql_type] = {
            "affected_rows_rate_peak": round(_prom_peak(result), 4),
            "affected_rows_rate_mean": round(_prom_mean(result), 4),
        }

    workload: list[dict] = []
    for sql_type in sorted(set(statements) | set(affected_rows)):
        row = {"sql_type": sql_type}
        row.update(statements.get(sql_type, {}))
        row.update(affected_rows.get(sql_type, {}))
        workload.append(row)
    return sorted(workload, key=lambda row: row.get("statement_rate_peak", 0), reverse=True)


def compact_resource_control_for_ai(data: dict, cluster_id: str, start: str, end: str) -> dict:
    """Summarize resource-control signals without raw metric samples."""
    rcu_usage = [
        {
            "name": _series_label(result, "name", "resource_group"),
            "peak": round(_prom_peak(result), 4),
            "mean": round(_prom_mean(result), 4),
            "unit": "RU/s",
        }
        for result in data.get("rcu_usage", [])
    ]
    request_p90 = [
        {
            "instance": _series_label(result, "instance"),
            "resource_group": _series_label(result, "resource_group", default=""),
            "peak": round(_prom_peak(result), 4),
            "mean": round(_prom_mean(result), 4),
            "unit": "seconds",
            "severity": "warning" if _prom_peak(result) > 0.01 else "normal",
        }
        for result in data.get("resource_manager_request_p90", [])
    ]
    available_ru = [
        {
            "name": _series_label(result, "name", "resource_group"),
            "min": round(_prom_min(result), 4),
            "mean": round(_prom_mean(result), 4),
            "unit": "RU",
            "severity": "warning" if 0 < _prom_min(result) < 1000 else "normal",
        }
        for result in data.get("available_ru", [])
    ]
    return {
        "rcu_usage": sorted(rcu_usage, key=lambda row: row.get("peak", 0), reverse=True)[:10],
        "request_p90": sorted(request_p90, key=lambda row: row.get("peak", 0), reverse=True)[:10],
        "available_ru": sorted(available_ru, key=lambda row: row.get("min", 0))[:10],
        "links": {
            "rcu_usage": _clinic_metrics_url(cluster_id, RESOURCE_RCU_PROMQL, start, end),
            "request_p90": _clinic_metrics_url(cluster_id, RESOURCE_REQUEST_P90_PROMQL, start, end),
            "available_ru": _clinic_metrics_url(cluster_id, AVAILABLE_RU_PROMQL, start, end),
        },
    }


def compact_component_health_for_ai(data: dict, cluster_id: str, start: str, end: str) -> dict:
    """Summarize component counts, restarts, and worker memory for Premium diagnosis."""
    component_count = []
    for result in data.get("component_count", []):
        values = result.get("values", []) if isinstance(result, dict) else []
        if not values:
            continue
        first = _safe_float(values[0][1])
        last = _safe_float(values[-1][1])
        nums = [_safe_float(row[1]) for row in values if len(row) >= 2]
        min_value = min(nums) if nums else 0.0
        max_value = max(nums) if nums else 0.0
        net_delta = round(last - first, 4)
        max_to_last_delta = round(last - max_value, 4)
        capacity_delta = max_to_last_delta if max_to_last_delta < 0 else net_delta
        component_count.append({
            "component": _series_label(result, "component"),
            "first": first,
            "last": last,
            "min": round(min_value, 4),
            "max": round(max_value, 4),
            "changes": capacity_delta,
            "net_delta": net_delta,
            "max_to_last_delta": max_to_last_delta,
            "change_count": sum(1 for i in range(1, len(values)) if values[i][1] != values[i - 1][1]),
        })

    restarts = []
    for result in data.get("restarts", []):
        count = _prom_peak(result)
        if count <= 0:
            continue
        restarts.append({
            "instance": _series_label(result, "instance"),
            "component": _series_label(result, "component", default=""),
            "restart_count": round(count, 4),
        })

    worker_memory = []
    for result in data.get("worker_memory", []):
        worker_memory.append({
            "instance": _series_label(result, "instance"),
            "component": _series_label(result, "component", default=""),
            "peak_gb": round(_prom_peak(result) / (1024 ** 3), 4),
            "mean_gb": round(_prom_mean(result) / (1024 ** 3), 4),
        })

    query_scope = data.get("component_health_scope") or data.get("parent_id") or cluster_id
    return {
        "query_scope": query_scope,
        "parent_id": data.get("parent_id", ""),
        "canonical_window": {"start": start, "end": end},
        "context_window": data.get("component_health_window", {"start": start, "end": end}),
        "component_count": sorted(component_count, key=lambda row: row.get("component", "")),
        "restarts": sorted(restarts, key=lambda row: row.get("restart_count", 0), reverse=True),
        "worker_memory": sorted(worker_memory, key=lambda row: row.get("peak_gb", 0), reverse=True),
        "links": {
            "component_count": _clinic_metrics_url(
                query_scope,
                "count by (component) (process_start_time_seconds)",
                start,
                end,
            ),
            "restarts": _clinic_metrics_url(
                query_scope,
                "changes(process_start_time_seconds[15m])",
                start,
                end,
            ),
            "worker_memory": _clinic_metrics_url(
                query_scope,
                'process_resident_memory_bytes{component=~"coprocessor-worker|tikv-worker"}',
                start,
                end,
            ),
        },
    }


def _query_loki_logs(api: ClinicAPI, cluster_id: str, query: str, start: str, end: str,
                     limit: str = "100") -> dict:
    """Query Loki log streams with a bounded result limit."""
    if hasattr(api, "logs_query_range"):
        resp = api.logs_query_range(
            cluster_id,
            query,
            _loki_time_ns(start),
            _loki_time_ns(end),
            limit=int(limit),
            direction="backward",
        )
    else:
        resp = api._get(
            f"/data-proxy/loki/{cluster_id}/api/v1/query_range",
            {
                "query": query,
                "start": _loki_time_ns(start),
                "end": _loki_time_ns(end),
                "limit": limit,
                "direction": "BACKWARD",
            },
        )
    if isinstance(resp, dict) and "error" not in resp:
        _record_raw("logs", f"loki_{_short_hash(query)}", resp, query=query,
                    scope=cluster_id, start=str(start), end=str(end))
    return resp


def _loki_time_ns(value: str | int | float) -> str:
    """Return a Loki-compatible nanosecond timestamp."""
    try:
        numeric = int(float(value))
    except (TypeError, ValueError):
        return str(value)
    if numeric <= 0:
        return str(numeric)
    return str(numeric if numeric >= 10**18 else numeric * 1_000_000_000)


def _log_relevance_score(message: str) -> int:
    """Rank focused log samples so rare root-cause signals survive noisy queries."""
    text = message.lower()
    if "panic:" in text:
        return 0
    if "memory.(*tracker).consume" in text or "tracker accounting panic" in text:
        return 1
    if "panic in the recoverable goroutine" in text:
        return 2
    if "parallel hash aggregation panicked" in text:
        return 3
    if "query execution was interrupted" in text or "sqlkiller.go" in text or "query is killed" in text:
        return 4
    if "[executor:3024]" in text or "maximum statement execution time exceeded" in text:
        return 5
    if _has_transaction_lock_signal(text):
        return 6
    if "known issue" in text or "fixed release" in text or "upstream issue" in text or "fixed in" in text:
        return 7
    if "oom" in text or "out of memory" in text or "restart" in text or "welcome to tidb" in text:
        return 8
    if "timeout" in text or "backlog" in text or "unavailable" in text:
        return 9
    return 20


_LOGSTATS_FOCUSED_PATTERN = (
    r"panic:|panic in the recoverable goroutine|parallel hash aggregation panicked|"
    r"memory\.\(\*tracker\)\.consume|tracker accounting panic|"
    r"memory.*exceed|exceed.*memory|oom action|tidb_mem_quota|"
    r"executor:3024|maximum statement execution time exceeded|max_execution_time|"
    r"kill query|query execution was interrupted|query is killed|killed remote connection|sqlkiller\.go|"
    + TRANSACTION_LOG_SIGNAL_REGEX + r"|"
    r"oom|out of memory|restarted|restart.*(tidb|tikv|pd|instance|pod)|timeout|backlog|unavailable|"
    r"coprocessor|welcome to tidb|"
    r"known issue|fixed release|upstream issue|fixed in|upgrade"
)


def _has_transaction_lock_signal(text: str) -> bool:
    lowered = text.lower()
    return any(term in lowered for term in TRANSACTION_LOCK_ROUTE_SIGNAL_TERMS)


def _logstats_payload_has_evidence(payload: dict) -> bool:
    if payload.get("log_counts") or payload.get("samples"):
        return True
    summary = payload.get("summary")
    if isinstance(summary, dict):
        for component_summary in summary.values():
            if not isinstance(component_summary, dict):
                continue
            if any(_safe_int(count) > 0 for count in component_summary.values()):
                return True
    timeline = payload.get("timeline")
    if isinstance(timeline, dict):
        for component_timeline in timeline.values():
            if not isinstance(component_timeline, dict):
                continue
            if any(buckets for buckets in component_timeline.values()):
                return True
    return False


def _logstats_backend_enabled(cluster_meta: dict | None) -> bool:
    mode = os.environ.get("INVESTIBOOK_LOGSTATS_BACKEND", "auto").strip().lower()
    if mode in {"", "off", "false", "0", "loki"}:
        return False
    if mode in {"athena", "clinic", "clinic_api", "logstats"}:
        return True
    if mode != "auto":
        return False
    return bool((cluster_meta or {}).get("region"))


def _sql_quote_string(value: object) -> str:
    return "'" + str(value).replace("'", "''") + "'"


def _date_hour_predicates(start: str, end: str) -> str:
    start_ts = int(float(start))
    end_ts = int(float(end))
    if end_ts < start_ts:
        start_ts, end_ts = end_ts, start_ts
    start_hour = datetime.fromtimestamp(start_ts, tz=timezone.utc).replace(minute=0, second=0, microsecond=0)
    end_hour = datetime.fromtimestamp(end_ts, tz=timezone.utc).replace(minute=0, second=0, microsecond=0)

    hours_by_date: dict[str, list[int]] = {}
    current = start_hour
    while current <= end_hour:
        date_key = current.strftime("%Y%m%d")
        hours_by_date.setdefault(date_key, []).append(current.hour)
        current = datetime.fromtimestamp(int(current.timestamp()) + 3600, tz=timezone.utc)

    clauses = []
    for date_key, hours in sorted(hours_by_date.items()):
        clauses.append(
            f"(date = {date_key} AND hour BETWEEN {min(hours)} AND {max(hours)})"
        )
    return "(" + " OR ".join(clauses) + ")"


def _clinic_logstats_where_clause(cluster_meta: dict, cluster_id: str, start: str, end: str) -> str:
    tenant_id = str(cluster_meta.get("tenantID") or cluster_meta.get("tenantId") or "")
    filters = [
        f"cluster = {_sql_quote_string(cluster_id)}",
        _date_hour_predicates(start, end),
    ]
    if tenant_id:
        filters.append(f"org = {_sql_quote_string(tenant_id)}")
    return " AND ".join(filters)


def _clinic_logstats_focused_like_clause() -> str:
    keywords = [
        "panic",
        "tracker",
        "memory",
        "oom",
        "tidb_mem_quota",
        "executor:3024",
        "maximum statement execution time exceeded",
        "max_execution_time",
        "kill",
        "query execution was interrupted",
        "sqlkiller",
        "restart",
        "timeout",
        "backlog",
        "unavailable",
        "coprocessor",
        "known issue",
        "fixed release",
        "upstream issue",
        "fixed in",
        "upgrade",
    ] + TRANSACTION_LOG_RECALL_TERMS
    return " OR ".join(
        f"lower(coalesce(msg_digest, '')) LIKE {_sql_quote_string('%' + keyword + '%')}"
        for keyword in keywords
    )


def _run_clinic_logstats_query(
    api: ClinicAPI,
    cluster_id: str,
    sql: str,
    label: str,
    timeout_s: int | None = None,
) -> tuple[list[dict], dict, str | None]:
    timeout_s = timeout_s or int(os.environ.get(
        "INVESTIBOOK_LOGSTATS_QUERY_TIMEOUT_S",
        os.environ.get("INVESTIBOOK_LOGSTATS_ATHENA_TIMEOUT_S", "45"),
    ))
    op_id = f"A6:clinic-logstats:{label}:{int(time.time() * 1000)}:{os.getpid()}"
    _perf_event({
        "event": "collector_op_start",
        "op_id": op_id,
        "op_kind": "clinic_logstats",
        "label": label,
        "cluster_id": cluster_id,
        "query_hash": _short_hash(sql),
        "query_preview": sql[:240],
        "timeout_s": timeout_s,
    })
    t0 = time.monotonic()

    try:
        payload = api.query_sql(cluster_id, sql, timeout=timeout_s)
    except Exception as exc:  # noqa: BLE001 - collector should fall back to Loki on provider errors.
        elapsed_ms = round((time.monotonic() - t0) * 1000, 3)
        _perf_event({
            "event": "collector_op_end",
            "op_id": op_id,
            "op_kind": "clinic_logstats",
            "label": label,
            "elapsed_ms": elapsed_ms,
            "status": "error",
            "error": str(exc)[:500],
        })
        return [], {}, str(exc)

    if not isinstance(payload, dict):
        return [], {}, f"Clinic LogStats returned unexpected payload type: {type(payload).__name__}"
    if payload.get("error"):
        elapsed_ms = round((time.monotonic() - t0) * 1000, 3)
        _perf_event({
            "event": "collector_op_end",
            "op_id": op_id,
            "op_kind": "clinic_logstats",
            "label": label,
            "elapsed_ms": elapsed_ms,
            "status": "error",
            "error": str(payload.get("error"))[:500],
        })
        return [], payload, str(payload.get("error"))

    _record_raw("logs", f"logstats_{label}", payload, query=sql, scope=cluster_id)

    raw_columns = payload.get("columns") or []
    columns = [
        str(column.get("name") if isinstance(column, dict) else column)
        for column in raw_columns
    ]
    rows: list[dict] = []
    for raw_row in payload.get("rows") or []:
        if isinstance(raw_row, dict):
            rows.append(raw_row)
        elif isinstance(raw_row, list):
            rows.append({
                columns[index]: raw_row[index] if index < len(raw_row) else ""
                for index in range(len(columns))
            })

    elapsed_ms = round((time.monotonic() - t0) * 1000, 3)
    _perf_event({
        "event": "collector_op_end",
        "op_id": op_id,
        "op_kind": "clinic_logstats",
        "label": label,
        "elapsed_ms": elapsed_ms,
        "status": "ok",
        "rows": len(rows),
    })
    return rows, payload, None


def _safe_int(value: object, default: int = 0) -> int:
    try:
        return int(float(str(value)))
    except (TypeError, ValueError):
        return default


def _logstats_hour_epoch(date_value: str, hour_value: str) -> int:
    try:
        dt = datetime.strptime(f"{date_value}{int(hour_value):02d}", "%Y%m%d%H").replace(tzinfo=timezone.utc)
        return int(dt.timestamp())
    except (TypeError, ValueError):
        return 0


def _logstats_sample_timestamp_ns(sample: str) -> str:
    match = re.search(r"(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d+))?Z", sample)
    if not match:
        return ""
    try:
        base = match.group(1)
        fractional = (match.group(2) or "")
        microseconds = (fractional[:6]).ljust(6, "0")
        dt = datetime.fromisoformat(f"{base}.{microseconds}+00:00")
    except ValueError:
        return ""
    nanos = int(dt.timestamp()) * 1_000_000_000
    if fractional:
        nanos += int((fractional[:9]).ljust(9, "0"))
    return str(nanos)


def _logstats_sample_instance(sample: str) -> str:
    match = re.search(r"\bdb-(?:tidb|tikv|pd)[^;\s]*", sample)
    return match.group(0) if match else ""


def _log_interval_scope(begin: int, finish: int, start: str, end: str) -> str:
    canonical_start = _parse_time_to_epoch_seconds(start)
    canonical_end = _parse_time_to_epoch_seconds(end)
    if canonical_start <= begin and finish <= canonical_end:
        return "canonical_window"
    if finish < canonical_start or begin > canonical_end:
        return "outside_window"
    return "overlaps_window"


def _annotate_log_time_scope(payload: dict) -> dict:
    """Preserve context logs, but never equate bucket time with sample time."""
    window = payload["canonical_window"]
    start, end = window["start"], window["end"]
    for row in payload.get("samples", []):
        try:
            timestamp_ns = int(row.get("timestamp_ns") or "")
            # Compare nanoseconds so a fractional sample after the end stays outside.
            begin = _parse_time_to_epoch_seconds(start) * 1_000_000_000
            finish = _parse_time_to_epoch_seconds(end) * 1_000_000_000
            row["time_scope"] = "canonical_window" if begin <= timestamp_ns <= finish else "outside_window"
        except (ValueError, TypeError):
            row["time_scope"] = "unknown"
    for row in payload.get("log_counts", []):
        if payload.get("source") == "clinic_logstats":
            begin = _logstats_hour_epoch(row.get("date", ""), row.get("hour", ""))
            if not begin:
                row["time_scope"] = "unknown"
                continue
            finish = begin + 3600
            row["count_semantics"] = "hourly_aggregate_not_sample_count"
        else:
            context = payload["focused_log_context_window"]
            begin, finish = (_parse_time_to_epoch_seconds(context[key]) for key in ("start", "end"))
            row["count_semantics"] = "returned_samples_in_context_window_not_total_events"
        row["aggregation_window"] = {"start": str(begin), "end": str(finish)}
        row["time_scope"] = _log_interval_scope(begin, finish, start, end)
    if payload.get("source") == "clinic_logstats":
        payload["count_semantics"] = (
            "Hourly bucket totals, not exact canonical-window counts. A representative "
            "sample has its own timestamp; it cannot place the entire bucket count "
            "inside the incident window. Missing/outside samples do not prove absence."
        )
    return payload


def _collect_log_stats_from_clinic_api(
    api: ClinicAPI,
    cluster_meta: dict,
    cluster_id: str,
    start: str,
    end: str,
) -> tuple[dict | None, str | None]:
    where_clause = _clinic_logstats_where_clause(cluster_meta, cluster_id, start, end)
    table_ref = "log_stats"
    components_sql = "'tidb', 'tikv', 'pd'"

    summary_sql = f"""
SELECT date, hour, component, level, count
FROM {table_ref}
WHERE {where_clause}
  AND component IN ({components_sql})
  AND upper(level) IN ('ERROR', 'WARN', 'INFO', 'FATAL', 'PANIC')
LIMIT 5000
""".strip()
    summary_rows, _, summary_error = _run_clinic_logstats_query(api, cluster_id, summary_sql, "A6:logstats_summary")
    if summary_error:
        return None, f"Clinic LogStats summary failed: {summary_error}"

    focused_start, focused_end = _focused_log_context_window(start, end)
    focused_where_clause = _clinic_logstats_where_clause(cluster_meta, cluster_id, focused_start, focused_end)
    focused_sql = f"""
SELECT date, hour, component, level, count, msg_digest, sample
FROM {table_ref}
WHERE {focused_where_clause}
  AND component IN ({components_sql})
  AND upper(level) IN ('ERROR', 'WARN', 'FATAL', 'PANIC')
  AND ({_clinic_logstats_focused_like_clause()})
LIMIT 500
""".strip()
    focused_rows, _, focused_error = _run_clinic_logstats_query(api, cluster_id, focused_sql, "A6:logstats_focused")
    if focused_error:
        return None, f"Clinic LogStats focused failed: {focused_error}"

    components = ["tidb", "tikv", "pd"]
    levels = ["error", "warn", "info"]
    timeline: dict[str, dict[str, list[dict]]] = {component: {level: [] for level in levels} for component in components}
    timeline_counts: dict[tuple[str, str, int], int] = {}
    for row in summary_rows:
        component = str(row.get("component") or "")
        if component not in timeline:
            continue
        raw_level = str(row.get("level") or "").upper()
        level = "error" if raw_level in {"ERROR", "FATAL", "PANIC"} else raw_level.lower()
        if level not in levels:
            continue
        window_start = _logstats_hour_epoch(str(row.get("date") or ""), str(row.get("hour") or ""))
        if not window_start:
            continue
        key = (component, level, window_start)
        timeline_counts[key] = timeline_counts.get(key, 0) + _safe_int(row.get("count"))

    for (component, level, window_start), count in sorted(timeline_counts.items()):
        if count > 0:
            timeline[component][level].append({
                "window_start": window_start,
                "window_end": window_start + 3600,
                "count": count,
                "time_scope": _log_interval_scope(window_start, window_start + 3600, start, end),
            })

    summary = {
        component: {
            level: sum(bucket["count"] for bucket in timeline[component].get(level, []))
            for level in levels
        }
        for component in components
    }

    focused_groups: dict[tuple[str, str, str, str, str], dict] = {}
    for row in focused_rows:
        digest = str(row.get("msg_digest") or "").strip()
        if not digest:
            continue
        key = (
            str(row.get("date") or ""),
            str(row.get("hour") or ""),
            str(row.get("component") or ""),
            str(row.get("level") or "").upper() or "INFO",
            digest,
        )
        group = focused_groups.setdefault(key, {
            "date": key[0],
            "hour": key[1],
            "component": key[2],
            "level": key[3],
            "msg_digest": digest,
            "sample": "",
            "total": 0,
        })
        group["total"] += _safe_int(row.get("count"))
        if not group["sample"]:
            group["sample"] = str(row.get("sample") or "").strip()

    log_counts = []
    samples = []
    for row in sorted(
        focused_groups.values(),
        key=lambda item: (_log_relevance_score(str(item.get("msg_digest") or item.get("sample") or "")), -_safe_int(item.get("total"))),
    )[:20]:
        digest = str(row.get("msg_digest") or "").strip()
        sample = str(row.get("sample") or "").strip()
        component = str(row.get("component") or "")
        level = str(row.get("level") or "").upper() or "INFO"
        total = _safe_int(row.get("total"))
        if digest:
            log_counts.append({
                "component": component,
                "level": level,
                "message": digest[:500],
                "count": total,
                "source": "clinic_logstats",
                "date": str(row.get("date") or ""),
                "hour": _safe_int(row.get("hour")),
            })
        if sample:
            samples.append({
                "timestamp_ns": _logstats_sample_timestamp_ns(sample),
                "component": component,
                "level": level,
                "instance": _logstats_sample_instance(sample),
                "line": sample[:1000],
                "source": "clinic_logstats",
            })

    query = (
        "clinic-logstats:log_stats "
        f"cluster={cluster_id} pattern={_LOGSTATS_FOCUSED_PATTERN}"
    )
    return _annotate_log_time_scope({
        "query": query,
        "source": "clinic_logstats",
        "canonical_window": {"start": start, "end": end},
        "focused_log_context_window": {"start": focused_start, "end": focused_end},
        "timeline": timeline,
        "summary": summary,
        "log_counts": log_counts,
        "samples": sorted(
            samples,
            key=lambda item: (_log_relevance_score(str(item.get("line", ""))), str(item.get("timestamp_ns", ""))),
        )[:20],
        "step_seconds": 3600,
        "errors": [],
        "loki_url": _loki_log_url(cluster_id, '{component=~"tidb|tikv|pd"} |~ "(?i)(ERROR|FATAL|WARN)"', start, end, cluster_meta),
    }), None


def _collect_focused_log_samples(
    api: ClinicAPI,
    cluster_id: str,
    start: str,
    end: str,
) -> tuple[list[dict], list[dict], str | None]:
    """Collect bounded raw log samples for incident signatures that aggregations hide."""
    exhaustive = os.environ.get("INVESTIBOOK_FOCUSED_LOG_EXHAUSTIVE", "").strip().lower() in {
        "1",
        "true",
        "yes",
    }
    primary_selectors = [
        '{container=~"tidb|tikv|pd|coprocessor-worker|statementlog"}',
        '{component=~"tidb|tikv|pd|coprocessor-worker|statementlog"}',
    ]
    fallback_selectors = [
        f'{{cluster_id="{cluster_id}"}}',
        f'{{namespace=~"tidb{cluster_id}"}}',
    ]
    selectors = primary_selectors + fallback_selectors if exhaustive else primary_selectors
    literal_terms = [
        "panic:",
        "memory.(*Tracker).Consume",
        "Query execution was interrupted",
        "query is killed",
        "sqlkiller.go",
        "panic in the recoverable goroutine",
        "[executor:3024]",
        "maximum statement execution time exceeded",
    ] if exhaustive else []
    high_priority_filters = [
        '(?i)(panic in the recoverable goroutine|parallel hash aggregation panicked|'
        'executor:3024|maximum statement execution time exceeded|max_execution_time)',
        '(?i)(panic:|memory\\\\.\\\\(\\\\*Tracker\\\\)\\\\.Consume|tracker accounting panic|'
        'memory.*exceed|exceed.*memory|oom action|tidb_mem_quota)',
        '(?i)(kill query|query execution was interrupted|query is killed|killed remote connection|sqlkiller\\\\.go)',
        f'(?i)({TRANSACTION_LOG_SIGNAL_REGEX})',
        '(?i)(known issue|fixed release|upstream issue|fixed in|upgrade)',
    ]
    focused_filter = (
        '(?i)(panic:|panic in the recoverable goroutine|parallel hash aggregation panicked|'
        'memory\\\\.\\\\(\\\\*Tracker\\\\)\\\\.Consume|tracker accounting panic|'
        'memory.*exceed|exceed.*memory|oom action|tidb_mem_quota|'
        'executor:3024|maximum statement execution time exceeded|max_execution_time|'
        'kill query|query execution was interrupted|query is killed|killed remote connection|sqlkiller\\\\.go|'
        f'{TRANSACTION_LOG_SIGNAL_REGEX}|'
        'OOM|out of memory|restart|timeout|backlog|unavailable|coprocessor|Welcome to TiDB|'
        'known issue|fixed release|upstream issue|fixed in|upgrade)'
    )
    queries = [
        (f'{selector} |= "{term}"', "50")
        for term in literal_terms
        for selector in selectors
    ]
    queries.extend([
        (f'{selector} |~ "{log_filter}"', "200")
        for log_filter in high_priority_filters
        for selector in selectors
    ])
    broad_selectors = selectors if exhaustive else primary_selectors
    queries.extend([(f'{selector} |~ "{focused_filter}"', "100") for selector in broad_selectors])
    fallback_queries = [] if exhaustive else [
        (f'{selector} |~ "{log_filter}"', "200")
        for log_filter in high_priority_filters
        for selector in fallback_selectors
    ]
    if not exhaustive:
        fallback_queries.extend(
            (f'{selector} |~ "{focused_filter}"', "100")
            for selector in fallback_selectors
        )

    counts: dict[tuple[str, str, str], int] = {}
    sample_candidates: list[dict] = []
    seen_samples: set[tuple[str, str, str]] = set()
    errors: list[str] = []

    from concurrent.futures import ThreadPoolExecutor

    def _run_query(query_item: tuple[str, str]) -> tuple[str, dict]:
        query, limit = query_item
        return query, _query_loki_logs(api, cluster_id, query, start, end, limit)

    def _run_query_batch(query_items: list[tuple[str, str]]) -> list[tuple[str, dict]]:
        if not query_items:
            return []
        max_workers = min(8 if exhaustive else 6, len(query_items))
        query_responses: list[tuple[str, dict]] = []
        with ThreadPoolExecutor(max_workers=max_workers) as pool:
            future_by_index = {
                index: pool.submit(_run_query, query_item)
                for index, query_item in enumerate(query_items)
            }
            for index in sorted(future_by_index):
                query_responses.append(future_by_index[index].result())
        return query_responses

    def _ingest_query_responses(query_responses: list[tuple[str, dict]]) -> None:
        for query, resp in query_responses:
            if "error" in resp:
                errors.append(f"{query}: {resp['error']}")
                continue
            for stream in _prom_results(resp):
                labels = stream.get("stream", {}) if isinstance(stream, dict) else {}
                component = str(
                    labels.get("component")
                    or labels.get("container")
                    or labels.get("app")
                    or labels.get("namespace")
                    or ""
                )
                instance = str(labels.get("instance") or labels.get("pod") or labels.get("filename") or "")
                for ts, line in stream.get("values", []):
                    message = str(line)
                    level = "INFO"
                    upper = message.upper()
                    if "PANIC" in upper or "FATAL" in upper:
                        level = "PANIC" if "PANIC" in upper else "FATAL"
                    elif "ERROR" in upper:
                        level = "ERROR"
                    elif "WARN" in upper:
                        level = "WARN"
                    key = (component, level, message[:500])
                    counts[key] = counts.get(key, 0) + 1
                    sample_key = (str(ts), instance, message)
                    if sample_key not in seen_samples and len(sample_candidates) < 300:
                        seen_samples.add(sample_key)
                        sample_candidates.append({
                            "timestamp_ns": str(ts),
                            "component": component,
                            "level": level,
                            "instance": instance,
                            "line": message[:1000],
                        })

    _ingest_query_responses(_run_query_batch(queries))
    if fallback_queries and not sample_candidates:
        _ingest_query_responses(_run_query_batch(fallback_queries))

    log_counts = [
        {"component": component, "level": level, "message": message, "count": count}
        for (component, level, message), count in sorted(
            counts.items(),
            key=lambda item: (_log_relevance_score(item[0][2]), -item[1]),
        )[:20]
    ]
    samples = sorted(
        sample_candidates,
        key=lambda item: (_log_relevance_score(str(item.get("line", ""))), str(item.get("timestamp_ns", ""))),
    )[:20]
    return log_counts, samples, "; ".join(errors) if errors and not log_counts else None


def _tidb_replica_counts(cluster_meta: dict | None) -> list[int]:
    if not isinstance(cluster_meta, dict):
        return []
    components = cluster_meta.get("components")
    if not isinstance(components, dict):
        return []
    tidb_component = components.get("COMPONENT_TYPE_TIDB")
    if not isinstance(tidb_component, dict):
        return []
    counts: list[int] = []
    replicas = tidb_component.get("replicas")
    if isinstance(replicas, (int, float)):
        counts.append(int(replicas))
    groups = tidb_component.get("tidbGroups")
    if isinstance(groups, list):
        for group in groups:
            if isinstance(group, dict) and isinstance(group.get("replicas"), (int, float)):
                counts.append(int(group["replicas"]))
    return counts


def _cluster_parent_id(cluster_meta: dict | None) -> str:
    if not isinstance(cluster_meta, dict):
        return ""
    for key in ("parent_id", "parentID", "parentId", "resourcePoolID", "resource_pool_id"):
        value = cluster_meta.get(key)
        if value:
            return str(value)
    return ""


def _component_health_context_window(start: str, end: str) -> tuple[str, str]:
    """Use a small health-only context window for scrape-edge and pre-symptom events."""
    try:
        start_i = int(float(start))
        end_i = int(float(end))
    except (TypeError, ValueError):
        return start, end
    return (
        str(max(0, start_i - COMPONENT_HEALTH_LOOKBACK_SECONDS)),
        str(end_i + COMPONENT_HEALTH_TAIL_SECONDS),
    )


def _focused_log_context_window(start: str, end: str) -> tuple[str, str]:
    """Use a small log-only context window for restart/panic/kill boundary evidence."""
    try:
        start_i = int(float(start))
        end_i = int(float(end))
    except (TypeError, ValueError):
        return start, end
    return (
        str(max(0, start_i - FOCUSED_LOG_LOOKBACK_SECONDS)),
        str(end_i + FOCUSED_LOG_TAIL_SECONDS),
    )


def compact_events_for_ai(data: dict) -> list[dict]:
    """Keep Activity Hub events compact but searchable for diagnosis handoff."""
    compact: list[dict] = []
    events = data.get("events", {}).get("events") if isinstance(data.get("events"), dict) else []
    if not isinstance(events, list):
        return compact
    for event in events[:20]:
        if not isinstance(event, dict):
            continue
        payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
        compact.append({
            "name": event.get("name") or event.get("display_name") or payload.get("display_event_type") or "",
            "display_name": event.get("display_name") or event.get("displayName") or "",
            "severity": event.get("severity", ""),
            "source": event.get("source", ""),
            "calibration_time": event.get("calibration_time", ""),
            "summary": payload.get("summary") or event.get("summary") or "",
            "status": payload.get("display_status") or (event.get("tags") or {}).get("status", ""),
            "detail": str(payload.get("detail") or "")[:1000],
        })
    return compact


def build_critical_evidence_hints(summary: dict, cluster_meta: dict | None = None) -> list[dict]:
    """Build compact, evidence-derived hints for analyst handoff and evaluation."""
    hints: list[dict] = []

    metrics_text = json.dumps(summary.get("metrics", []), ensure_ascii=False).lower()
    if "tidb cpu" in metrics_text or "process_cpu_seconds_total" in metrics_text:
        hints.append({
            "id": "primary_symptom_evidence_present",
            "family": "cpu_spike",
            "summary": "TiDB CPU metric evidence is present.",
        })

    # Per-instance latency skew: the primary overview comparison for slowness
    # claims (TiDB query latency by instance, TiKV gRPC latency by store).
    for group_name, hint_id, floor_seconds in (
        ("TiKV gRPC P99", "store_latency_skew_present", 0.05),
        ("TiDB P99 Latency", "tidb_latency_skew_present", 0.1),
    ):
        instance_rows = [
            row for row in summary.get("metrics", [])
            if isinstance(row, dict) and row.get("name") == group_name
            and not str(row.get("instance", "")).startswith("all (")
        ]
        avg_rows = [
            row for row in summary.get("metrics", [])
            if isinstance(row, dict) and row.get("name") == f"{group_name} Avg"
        ]
        if not instance_rows or not avg_rows:
            continue
        try:
            top_row = max(instance_rows, key=lambda row: float(row.get("peak") or 0))
            top_peak = float(top_row.get("peak") or 0)
            avg_peak = float(avg_rows[0].get("peak") or 0)
        except (TypeError, ValueError):
            continue
        if top_peak >= floor_seconds and avg_peak > 0 and top_peak >= 2.0 * avg_peak:
            hints.append({
                "id": hint_id,
                "family": "latency_skew",
                "summary": (
                    f"{group_name} skew: {top_row.get('instance')} peak {top_peak:.3f}s vs "
                    f"all-instance average peak {avg_peak:.3f}s; verify this instance against peers."
                ),
            })

    replica_counts = _tidb_replica_counts(cluster_meta)
    nodegroups = sorted({
        str(row.get("nodegroup"))
        for row in summary.get("metrics", [])
        if row.get("nodegroup")
    })
    if len(set(replica_counts)) >= 2 or len(nodegroups) >= 2:
        hints.append({
            "id": "capacity_comparison_context_present",
            "family": "cpu_spike",
            "summary": (
                "Capacity comparison context is present: per-node baseline comparison can use "
                f"node count / replicas {replica_counts or 'unknown'} and nodegroups {nodegroups} "
                "to decide whether fewer nodes handled the same workload."
            ),
        })

    workload_text = json.dumps(summary.get("workload", []), ensure_ascii=False).lower()
    if "delete" in workload_text or "ntdml-delete" in workload_text:
        hints.append({
            "id": "primary_workload_shift_present",
            "family": "sustained_cpu_pressure",
            "summary": "Delete workload evidence is present; check statement_rate_peak and affected_rows_rate_peak.",
        })

    metrics = summary.get("metrics", [])
    queries = summary.get("queries", {})
    gc_warning_metrics = [
        row for row in metrics
        if isinstance(row, dict)
        and "gc" in str(row.get("metric_group_id") or row.get("name") or "").lower()
        and str(row.get("severity") or "").lower() in {"warning", "critical"}
    ]
    query_pressure_text = json.dumps(queries, ensure_ascii=False).lower()
    has_query_memory_pressure = any(
        keyword in query_pressure_text
        for keyword in ("max_mem", "memory", "heap", "large in", "in-list", "indexlookup", "process_keys")
    )
    if gc_warning_metrics or has_query_memory_pressure:
        source = "GC pressure" if gc_warning_metrics else "allocation rate / heap growth"
        hints.append({
            "id": "allocation_pressure_signal_present",
            "family": "sustained_cpu_pressure",
            "summary": (
                f"{source} evidence is present; inspect TiDB GC pause/duration metrics and "
                "large query memory / IndexLookUp / process_keys signals for allocation pressure."
            ),
        })

    metrics_and_queries_text = json.dumps(
        {"metrics": summary.get("metrics", []), "queries": summary.get("queries", {})},
        ensure_ascii=False,
    ).lower()
    if "go_gc" in metrics_and_queries_text or "gc pause" in metrics_and_queries_text:
        hints.append({
            "id": "gc_signal_present",
            "family": "sql_tail_latency",
            "summary": "GC / heap pressure signals are present; check go_gc and heap-related evidence.",
        })
    if any(keyword in metrics_and_queries_text for keyword in ("heap", "inuse_space", "indexlookup", "process_keys", "large scan")):
        hints.append({
            "id": "heap_profile_signal_present",
            "family": "sql_tail_latency",
            "summary": "Heap, profile, or large-scan query evidence is present.",
        })

    events_text = json.dumps(summary.get("events", []), ensure_ascii=False).lower()
    if any(keyword in events_text for keyword in ("overwriteimagetag", "configchange", "upgrade", "fixed release", "fixed in")):
        hints.append({
            "id": "known_issue_correlation_context_present",
            "family": "sql_tail_latency",
            "summary": (
                "Known issue / fixed release / upgrade context is present in Activity Hub events; "
                "correlate abnormal heap or allocation evidence with version-change context before "
                "stopping at workload-only explanations."
            ),
        })

    component_health = summary.get("component_health", {}) if isinstance(summary.get("component_health"), dict) else {}
    if component_health.get("restarts"):
        hints.append({
            "id": "restart_signal_present",
            "family": "component_restart",
            "summary": "Restart metrics are present in component_health.restarts.",
        })
    for row in component_health.get("component_count", []) or []:
        if "coprocessor-worker" in str(row.get("component", "")) and _safe_float(row.get("changes")) < 0:
            hints.append({
                "id": "capacity_loss_signal_present",
                "family": "system_unavailability",
                "summary": "coprocessor-worker capacity loss is present; component_count changes is below 0.",
            })
            break

    logs_payload = summary.get("logs", {}) if isinstance(summary.get("logs"), dict) else {}
    log_hint_terms = {
        "panic_signal_present": ("panic:", "memory.(*tracker).consume", "tracker accounting panic"),
        "query_interruption_signal_present": ("kill query", "query execution was interrupted", "query is killed", "sqlkiller.go"),
        "query_timeout_interrupt_signal_present": ("[executor:3024]", "maximum statement execution time exceeded", "max_execution_time"),
        "restart_or_oom_signal_present": ("oom", "out of memory", "restart", "coprocessor"),
        "coprocessor_cascade_signal_present": ("timeout", "backlog", "unavailable"),
        "transaction_lock_signal_present": TRANSACTION_LOCK_ROUTE_SIGNAL_TERMS,
    }
    logs_text = json.dumps(
        {
            "log_counts": logs_payload.get("log_counts", []),
            "samples": logs_payload.get("samples", []),
        },
        ensure_ascii=False,
    ).lower()
    if any(keyword in logs_text for keyword in log_hint_terms["panic_signal_present"]):
        hints.append({
            "id": "panic_signal_present",
            "family": "component_restart",
            "summary": "Panic log samples are present.",
        })
    if any(keyword in logs_text for keyword in log_hint_terms["query_interruption_signal_present"]):
        hints.append({
            "id": "query_interruption_signal_present",
            "family": "component_restart",
            "summary": "Query interruption / kill log samples are present.",
        })
    if any(keyword in logs_text for keyword in log_hint_terms["query_timeout_interrupt_signal_present"]):
        hints.append({
            "id": "query_timeout_interrupt_signal_present",
            "family": "recoverable_panic",
            "summary": "Query timeout interruption panic evidence is present.",
        })
    if any(keyword in logs_text for keyword in log_hint_terms["restart_or_oom_signal_present"]):
        hints.append({
            "id": "restart_or_oom_signal_present",
            "family": "system_unavailability",
            "summary": "Restart, OOM, or coprocessor log samples are present.",
        })
    if any(keyword in logs_text for keyword in log_hint_terms["coprocessor_cascade_signal_present"]):
        hints.append({
            "id": "coprocessor_cascade_signal_present",
            "family": "system_unavailability",
            "summary": "Potential outage / latency spike cascade log samples are present.",
        })
    transaction_evidence_text = " ".join([logs_text, metrics_and_queries_text])
    transaction_signals = [
        term for term in TRANSACTION_LOCK_ROUTE_SIGNAL_TERMS
        if term in transaction_evidence_text
    ]
    if transaction_signals:
        hints.append({
            "id": "transaction_lock_signal_present",
            "family": "transaction_or_lock_contention",
            "summary": (
                "Transaction / lock evidence is present; route diagnosis through "
                "transaction-or-lock-contention before ruling out write-path causes. "
                f"Matched signals: {', '.join(transaction_signals[:6])}."
            ),
        })

    # Context matches remain routing leads; only time-qualified matches can be
    # promoted to incident evidence. Qualify each signature independently.
    for hint in hints:
        terms = log_hint_terms.get(hint["id"])
        if terms is None:
            continue
        in_window = any(
            row.get("time_scope") == "canonical_window"
            and any(term in str(row.get("line") or row.get("message") or "").lower() for term in terms)
            for row in logs_payload.get("samples", []) + logs_payload.get("log_counts", [])
        )
        if hint["id"] == "transaction_lock_signal_present":
            in_window = in_window or any(term in metrics_and_queries_text for term in terms)
        hint["time_scope"] = "canonical_window" if in_window else "context_only"
        if not in_window:
            hint["summary"] = (
                "Context-only lead; occurrence inside the canonical window is unverified. "
                "Do not use it to establish or exclude a root cause. " + hint["summary"]
            )

    return hints


def _panic_classification_context(logs: dict, gate_checks: dict) -> dict | None:
    """Expose query-timeout panic context in query artifacts for deterministic judging."""
    logs_text = json.dumps(
        {
            "log_counts": [row for row in logs.get("log_counts", []) if row.get("time_scope") == "canonical_window"],
            "samples": [row for row in logs.get("samples", []) if row.get("time_scope") == "canonical_window"],
        },
        ensure_ascii=False,
    )
    matched = [
        keyword for keyword in (
            "max_execution_time",
            "executor:3024",
            "Query execution was interrupted",
            "maximum statement execution time exceeded",
        )
        if keyword.lower() in logs_text.lower()
    ]
    if not matched:
        return None
    gate_text = json.dumps(gate_checks, ensure_ascii=False)
    if "INFRA CLEAR" in gate_text or "HOTSPOT" in gate_text:
        matched.append("gate_checks")
    return {
        "source": "logs_and_gate_checks",
        "classification": "recoverable query-timeout panic context",
        "matched_keywords": sorted(set(matched)),
        "summary": (
            "Query timeout / interruption context is present for alert classification; "
            "logs include executor:3024 or Query execution was interrupted signals."
        ),
    }


# ---------------------------------------------------------------------------
# Link generation (clinic.pingcap.com)
# ---------------------------------------------------------------------------

CLINIC_BASE = "https://clinic.pingcap.com"

RESOURCE_RCU_PROMQL = (
    "sum by (name) (rate(resource_manager_resource_unit_read_request_unit_sum[1m])) + "
    "sum by (name) (rate(resource_manager_resource_unit_write_request_unit_sum[1m])) + "
    "sum by (name) (rate(resource_manager_resource_unit_sql_layer_request_unit_sum[1m]))"
)
RESOURCE_REQUEST_P90_PROMQL = (
    "histogram_quantile(0.90, "
    "sum(rate(resource_manager_client_request_success_bucket[1m])) by (le, instance, resource_group))"
)
AVAILABLE_RU_PROMQL = "resource_manager_resource_unit_available_ru"
TIDB_CPU_PROMQL = 'rate(process_cpu_seconds_total{component="tidb"}[1m])'
TIKV_CPU_PROMQL = 'rate(process_cpu_seconds_total{component="tikv"}[1m])'
TIDB_QPS_PROMQL = 'sum(rate(tidb_server_query_total[1m])) by (result,type,instance)'
COMPONENT_MEMORY_PROMQL = 'process_resident_memory_bytes{component=~"tidb|tikv|pd|tiflash"}'
CONNECTION_IDLE_P99_PROMQL = 'histogram_quantile(0.99, sum(rate(tidb_server_conn_idle_duration_seconds_bucket[1m])) by (le,in_txn,instance))'
PARSE_P99_PROMQL = 'histogram_quantile(0.99, sum(rate(tidb_session_parse_duration_seconds_bucket[1m])) by (le,sql_type,instance))'
COMPILE_P99_PROMQL = 'histogram_quantile(0.99, sum(rate(tidb_session_compile_duration_seconds_bucket[1m])) by (le,sql_type,instance))'
KV_REQUEST_P99_PROMQL = 'histogram_quantile(0.99, sum(rate(tidb_tikvclient_request_seconds_bucket[1m])) by (le,type,store,instance))'
KV_BACKOFF_P99_PROMQL = 'histogram_quantile(0.99, sum(rate(tidb_tikvclient_backoff_seconds_bucket[1m])) by (le,type,instance))'
TSO_WAIT_P99_PROMQL = 'histogram_quantile(0.99, sum(rate(pd_client_cmd_handle_cmds_duration_seconds_bucket{type="wait"}[1m])) by (le,instance))'
TIKV_GRPC_BY_TYPE_P99_PROMQL = 'histogram_quantile(0.99, sum(rate(tikv_grpc_msg_duration_seconds_bucket[1m])) by (le,type,instance))'
TIKV_SCHEDULER_COMMAND_P99_PROMQL = 'histogram_quantile(0.99, sum(rate(tikv_scheduler_command_duration_seconds_bucket[1m])) by (le,instance,type))'
TIKV_SCHEDULER_LATCH_P99_PROMQL = 'histogram_quantile(0.99, sum(rate(tikv_scheduler_latch_wait_duration_seconds_bucket[1m])) by (le,instance,type))'
TIKV_STORAGE_ASYNC_P99_PROMQL = 'histogram_quantile(0.99, sum(rate(tikv_storage_engine_async_request_duration_seconds_bucket[1m])) by (le,instance,type))'
TIKV_RAFT_COMMIT_P99_PROMQL = 'histogram_quantile(0.99, sum(rate(tikv_raftstore_commit_log_duration_seconds_bucket[1m])) by (le,instance))'
TIKV_RAFT_APPLY_P99_PROMQL = 'histogram_quantile(0.99, sum(rate(tikv_raftstore_apply_log_duration_seconds_bucket[1m])) by (le,instance))'
TIKV_PENDING_COMPACTION_PROMQL = 'tikv_engine_pending_compaction_bytes'


def _clinic_metrics_url(cluster_id: str, query: str, start_ts: str, end_ts: str,
                        cluster_meta: dict | None = None) -> str:
    """Generate Clinic Portal monitoring URL.

    Format: /portal/#/orgs/{tenantID}/clusters/{clusterID}
    Falls back to cluster-only path when tenantID is unavailable.
    """
    meta = cluster_meta or {}
    tenant_id = str(meta.get("tenantID") or meta.get("tenantId") or "")
    if tenant_id:
        return f"{CLINIC_BASE}/portal/#/orgs/{tenant_id}/clusters/{cluster_id}"
    return f"{CLINIC_BASE}/portal/#/clusters/{cluster_id}"


def _loki_datasource_name(cluster_meta: dict | None) -> str:
    """Derive the Grafana Loki datasource name from cluster metadata.

    Datasources in Clinic Grafana follow the pattern:
        Loki-{provider}-{region}[-{deployType}]
    e.g. Loki-aws-ap-southeast-1-premium, Loki-alicloud-ap-southeast-1
    """
    meta = cluster_meta or {}
    provider = str(meta.get("provider") or "").lower()
    region = str(meta.get("region") or "")
    deploy_type = str(meta.get("deployType") or "").lower()
    if provider and region:
        parts = ["Loki", provider, region]
        if deploy_type:
            parts.append(deploy_type)
        return "-".join(parts)
    # Fallback to the known default UID when metadata is missing
    return "PF9DF9566127982DF"


def _loki_log_url(cluster_id: str, query: str, start_ts: str, end_ts: str,
                  cluster_meta: dict | None = None) -> str:
    """Generate Grafana Loki explore URL on Clinic.

    Uses the Grafana explore endpoint with the Loki datasource, matching the
    format: /grafana/explore?orgId=1&left={datasource,queries,range}
    """
    datasource = _loki_datasource_name(cluster_meta)
    left = json.dumps({
        "datasource": datasource,
        "queries": [{"refId": "A", "editorMode": "code", "expr": query, "queryType": "range"}],
        "range": {"from": str(int(start_ts) * 1000), "to": str(int(end_ts) * 1000)},
    }, separators=(",", ":"))
    return f"{CLINIC_BASE}/grafana/explore?orgId=1&left={quote(left)}"


def _fetch_cluster_meta(api: ClinicAPI, cluster_id: str) -> dict:
    """Fetch and normalize authoritative + dashboard cluster metadata.

    The canonical exact-ID response is authoritative for tier routing and the
    BYOC logical-to-physical relationship.  The dashboard row is still merged
    because it contains tenant/project fields used by legacy links and
    Activity Hub.  Missing optional endpoints are tolerated for older fakes and
    non-BYOC deployments.
    """
    empty: dict = {
        "provider": "", "region": "", "tenantID": "", "tenantName": "",
        "projectID": "", "clusterName": "", "deployType": "", "version": "",
        "parent_id": "",
    }
    canonical: dict = {}
    if hasattr(api, "get_cluster_by_id"):
        try:
            canonical_resp = api.get_cluster_by_id(cluster_id)
            if isinstance(canonical_resp, dict) and "error" not in canonical_resp:
                candidate = canonical_resp.get("data", canonical_resp)
                if isinstance(candidate, dict):
                    canonical = candidate
                    _record_raw("meta", "cluster_meta_canonical", canonical_resp, scope=cluster_id)
        except Exception:
            canonical = {}

    item: dict = {}
    try:
        resp = api._get(
            "/clinic/api/v1/dashboard/clusters",
            {"cluster_id": cluster_id, "show_deleted": "true", "limit": 10, "page": 1},
        )
        if isinstance(resp, dict) and "error" not in resp:
            _record_raw("meta", "cluster_meta", resp, scope=cluster_id)
            items = resp.get("items") or []
            item = next(
                (i for i in items if str(i.get("clusterID", "")) == str(cluster_id)),
                {},
            )
    except Exception:
        item = {}

    if not canonical and not item:
        return empty

    diagnostic_context = canonical.get("diagnosticContext") or {}
    if not isinstance(diagnostic_context, dict):
        diagnostic_context = {}
    deploy_type = str(
        canonical.get("deployType")
        or canonical.get("clusterDeployTypeV2")
        or canonical.get("clusterDeployType")
        or item.get("clusterDeployType")
        or item.get("clusterDeployTypeV2")
        or ""
    ).lower().replace("_", "-").replace(" ", "-")
    parent_id = str(
        diagnostic_context.get("parentId")
        or canonical.get("parentId")
        or item.get("parentID")
        or item.get("parentId")
        or item.get("parent_id")
        or item.get("resourcePoolID")
        or item.get("resource_pool_id")
        or ""
    )
    normalized = {
        # provider: clusterProviderName (list row), cloudProvider / provider / vendor fallbacks
        "provider": str(canonical.get("provider") or canonical.get("cloudProvider") or item.get("clusterProviderName") or item.get("cloudProvider") or item.get("provider") or item.get("vendor") or "").lower(),
        # region: clusterRegionName (list row) or region / cloudRegion fallbacks
        "region": str(canonical.get("region") or canonical.get("cloudRegion") or item.get("clusterRegionName") or item.get("region") or item.get("cloudRegion") or ""),
        "orgID": str(canonical.get("orgId") or canonical.get("orgID") or item.get("orgID") or item.get("orgId") or ""),
        "orgId": str(canonical.get("orgId") or canonical.get("orgID") or item.get("orgID") or item.get("orgId") or ""),
        # tenantID: user-facing tenant ID (x-org-id in NGM), NOT internal orgID
        "tenantID": str(item.get("tenantID") or item.get("tenantId") or ""),
        "tenantName": str(item.get("tenantName") or item.get("orgName") or ""),
        "projectID": str(item.get("projectID") or item.get("projectId") or ""),
        "clusterName": str(canonical.get("name") or canonical.get("clusterName") or item.get("clusterName") or item.get("name") or ""),
        "deployType": deploy_type,
        # version: clusterVersion (list row) or tidbVersion / version fallbacks
        "version": str(canonical.get("version") or canonical.get("tidbVersion") or item.get("clusterVersion") or item.get("tidbVersion") or item.get("version") or ""),
        "parent_id": parent_id,
        "physical_cluster_id": parent_id if deploy_type == "byoc" else (cluster_id if deploy_type == "byoc-host" else ""),
        "diagnosticContext": diagnostic_context,
    }
    # Preserve canonical fields for future consumers without allowing the
    # dashboard row to override authoritative tier/routing metadata.
    return {**canonical, **normalized}


def _cluster_deploy_type(cluster_meta: dict | None) -> str:
    meta = cluster_meta or {}
    return str(
        meta.get("deployType")
        or meta.get("clusterDeployTypeV2")
        or meta.get("clusterDeployType")
        or ""
    ).lower().replace("_", "-").replace(" ", "-")


def _is_byoc_cluster_meta(cluster_meta: dict | None) -> bool:
    return _cluster_deploy_type(cluster_meta) in {"byoc", "byoc-host", "byoc-resource-pool"}


def _response_records(response: object) -> list[dict]:
    """Extract object rows from canonical list/envelope response shapes."""
    if isinstance(response, list):
        return [dict(row) for row in response if isinstance(row, dict)]
    if not isinstance(response, dict) or response.get("error"):
        return []
    payload: object = response.get("data", response)
    if isinstance(payload, list):
        return [dict(row) for row in payload if isinstance(row, dict)]
    if isinstance(payload, dict):
        for key in ("rows", "items", "records", "result", "statements", "plans"):
            rows = payload.get(key)
            if isinstance(rows, list):
                return [dict(row) for row in rows if isinstance(row, dict)]
    return []


def _response_object(response: object) -> dict:
    if not isinstance(response, dict) or response.get("error"):
        return {}
    payload = response.get("data", response)
    return dict(payload) if isinstance(payload, dict) else {}


def _records_table(records: list[dict]) -> dict:
    columns: list[str] = []
    for record in records:
        for key in record:
            if key not in columns:
                columns.append(key)
    return {
        "columns": columns,
        "rows": [[record.get(column) for column in columns] for record in records],
    }


def _byoc_slow_query_table(response: object, sample_limit: int = 100) -> dict:
    """Summarize a latency-ranked BYOC slow-query sample without implying window totals."""
    grouped: dict[str, dict] = {}
    for index, row in enumerate(_response_records(response)):
        digest = str(row.get("digest") or row.get("sql_digest") or row.get("sqlDigest") or "")
        key = digest or f"row-{index}"
        try:
            query_time = float(row.get("query_time") or row.get("queryTime") or 0)
        except (TypeError, ValueError):
            query_time = 0.0
        try:
            memory_max = float(
                row.get("memory_max") or row.get("memoryMax") or row.get("mem_max") or 0
            )
        except (TypeError, ValueError):
            memory_max = 0.0
        current = grouped.setdefault(key, {
            "digest": digest,
            "sample_sql": str(
                row.get("query") or row.get("sample_sql") or row.get("normalized_sql") or ""
            ),
            "sample_count": 0,
            "total_time": 0.0,
            "max_time": 0.0,
            "max_mem": 0.0,
        })
        current["sample_count"] += 1
        current["total_time"] += query_time
        current["max_time"] = max(current["max_time"], query_time)
        current["max_mem"] = max(current["max_mem"], memory_max)
    records = []
    for current in grouped.values():
        count = int(current.pop("sample_count"))
        total_time = float(current.pop("total_time"))
        current["sample_count"] = count
        current["sample_avg_time"] = total_time / count if count else 0.0
        records.append(current)
    records.sort(key=lambda row: (row["max_time"], row["sample_count"]), reverse=True)
    table = _records_table(records)
    table.update({
        "scope": "latency_ranked_sample",
        "ranking": "query_time_desc",
        "sample_limit": int(sample_limit),
    })
    return table


def _byoc_statement_summary_table(response: object) -> dict:
    """Aggregate full-window BYOC Statement Summary buckets by digest and schema."""
    grouped: dict[tuple[str, str], dict] = {}
    for index, row in enumerate(_response_records(response)):
        digest = str(row.get("digest") or row.get("sql_digest") or row.get("sqlDigest") or "")
        schema_name = str(row.get("schema_name") or row.get("schemaName") or "")
        key = (digest, schema_name) if digest else (f"row-{index}", schema_name)
        exec_count_value = _safe_float(row.get("exec_count", row.get("EXEC_COUNT", 0)))
        exec_count = int(exec_count_value) if math.isfinite(exec_count_value) and exec_count_value > 0 else 0
        sum_latency_ns = _safe_float(row.get("sum_latency", row.get("SUM_LATENCY", 0)))
        max_latency_ns = _safe_float(row.get("max_latency", row.get("MAX_LATENCY", 0)))
        max_mem = _safe_float(row.get("max_mem", row.get("MAX_MEM", 0)))
        current = grouped.setdefault(key, {
            "digest": digest,
            "sample_sql": str(
                row.get("normalized_sql")
                or row.get("normalizedSql")
                or row.get("query_sample_text")
                or row.get("digest_text")
                or row.get("sample_sql")
                or row.get("query")
                or ""
            ),
            "schema_name": schema_name,
            "stmt_type": str(row.get("stmt_type") or row.get("stmtType") or ""),
            "exec_count": 0,
            "sum_latency_ns": 0.0,
            "max_latency_ns": 0.0,
            "max_mem": 0.0,
        })
        current["exec_count"] += exec_count
        current["sum_latency_ns"] += sum_latency_ns
        current["max_latency_ns"] = max(current["max_latency_ns"], max_latency_ns)
        current["max_mem"] = max(current["max_mem"], max_mem)

    records: list[dict] = []
    for current in grouped.values():
        count = int(current.pop("exec_count"))
        total_latency_ns = float(current.pop("sum_latency_ns"))
        max_latency_ns = float(current.pop("max_latency_ns"))
        current["exec_count"] = count
        current["avg_time"] = total_latency_ns / count / 1e9 if count else 0.0
        current["max_time"] = max_latency_ns / 1e9
        records.append(current)
    records.sort(key=lambda row: (row["exec_count"], row["max_time"]), reverse=True)
    table = _records_table(records)
    table.update({"scope": "full_window", "ranking": "exec_count_desc"})
    return table


def _byoc_top_sql_instance_names(response: object, instance_type: str) -> list[str]:
    payload: object = response
    if isinstance(response, dict) and "error" not in response:
        payload = response.get("data", response)
    candidates: object = payload
    if isinstance(payload, dict):
        candidates = (
            payload.get(instance_type)
            or payload.get(f"{instance_type}s")
            or payload.get("instances")
            or payload.get("items")
            or []
        )
    if isinstance(candidates, dict):
        candidates = (
            candidates.get(instance_type)
            or candidates.get(f"{instance_type}s")
            or []
        )
    if not isinstance(candidates, list):
        return []
    names: list[str] = []
    for item in candidates:
        if isinstance(item, str):
            name = item
        elif isinstance(item, dict):
            row_type = str(
                item.get("instance_type")
                or item.get("instanceType")
                or item.get("component")
                or item.get("type")
                or ""
            ).lower()
            if row_type and row_type != instance_type:
                continue
            name = str(
                item.get("instance") or item.get("address") or item.get("name") or item.get("id") or ""
            )
        else:
            name = ""
        if name and name not in names:
            names.append(name)
    return names


def _byoc_top_sql_table(responses: list[tuple[str, object]]) -> dict:
    records: list[dict] = []
    for instance, response in responses:
        for row in _response_records(response):
            normalized = dict(row)
            normalized["instance"] = str(row.get("instance") or instance)
            normalized["sql_digest"] = str(
                row.get("sql_digest") or row.get("sqlDigest") or row.get("digest") or ""
            )
            normalized["sample_sql"] = str(
                row.get("sample_sql")
                or row.get("normalized_sql")
                or row.get("normalizedSql")
                or row.get("sql")
                or ""
            )
            normalized["total_cpu_ms"] = (
                row.get("total_cpu_ms")
                if row.get("total_cpu_ms") is not None
                else row.get("cpu_time_ms", row.get("cpuTimeMs", row.get("cpu_time", 0)))
            )
            normalized["cpu_pct"] = row.get(
                "cpu_pct", row.get("cpuPct", row.get("cpu_percent", row.get("cpuPercent", 0)))
            )
            records.append(normalized)
    records.sort(key=lambda row: _safe_float(row.get("total_cpu_ms", 0)), reverse=True)
    return _records_table(records)


def _byoc_statement_plan_table(detail_response: object) -> dict:
    detail = _response_object(detail_response)
    if not detail:
        records = _response_records(detail_response)
        detail = records[0] if records else {}
    if not detail:
        return {}
    normalized = dict(detail)
    normalized["decoded_plan"] = str(
        detail.get("binary_plan_text")
        or detail.get("decoded_plan")
        or detail.get("plan")
        or detail.get("plan_text")
        or ""
    )
    normalized["full_sql"] = str(
        detail.get("full_sql")
        or detail.get("query")
        or detail.get("sample_sql")
        or detail.get("normalized_sql")
        or ""
    )
    normalized["stats_info"] = detail.get("stats_info", detail.get("stats", ""))
    normalized["indexes_used"] = detail.get("indexes_used", detail.get("index_names", ""))
    return _records_table([normalized])


def collect_activity_events(api: ClinicAPI, cluster_id: str, start: str, end: str,
                            cluster_meta: dict | None = None) -> dict:
    """Collect Activity Hub events for version/config/operation correlation."""
    meta = cluster_meta or {}
    org_id = str(meta.get("orgID") or meta.get("orgId") or meta.get("org_id") or "")
    if not org_id:
        return {
            "events": [],
            "time_window": {"start": int(float(start)), "end": int(float(end))},
            "error": "orgID missing from cluster metadata",
        }
    resp = api._get(
        f"/clinic/api/v1/activityhub/applications/{org_id}/targets/{cluster_id}/activities",
        {
            "begin_ts": int(float(start)),
            "end_ts": int(float(end)),
            "severity": 0,
            "limit": 100,
        },
    )
    if not isinstance(resp, dict) or resp.get("error"):
        return {
            "events": [],
            "time_window": {"start": int(float(start)), "end": int(float(end))},
            "error": resp.get("error") if isinstance(resp, dict) else "malformed activity response",
        }
    _record_raw("events", "activity_hub", resp, scope=cluster_id,
                start=str(start), end=str(end))
    events = resp.get("activities") or resp.get("events") or resp.get("items") or []
    return {
        "events": [item for item in events if isinstance(item, dict)],
        "time_window": {"start": int(float(start)), "end": int(float(end))},
    }


def _ngm_page_params(cluster_id: str, cluster_meta: dict | None = None) -> dict:
    """Build common NGM page query params, omitting optional empty values."""
    meta = cluster_meta or {}
    params = {
        "clusterId": str(meta.get("clusterID") or meta.get("clusterId") or cluster_id),
    }
    for key, value in (
        ("provider", str(meta.get("provider") or meta.get("clusterProviderName") or "").lower()),
        ("region", str(meta.get("region") or meta.get("clusterRegionName") or "")),
        ("orgName", str(meta.get("orgName") or meta.get("tenantName") or "")),
        ("orgId", str(meta.get("tenantID") or meta.get("tenantId") or meta.get("orgId")
                      or meta.get("orgID") or "")),
        ("projectId", str(meta.get("projectID") or meta.get("projectId") or "")),
        ("clusterName", str(meta.get("clusterName") or "")),
        ("deployType", str(meta.get("clusterDeployTypeV2") or meta.get("deployType")
                           or meta.get("clusterDeployType") or "").lower()),
        ("version", str(meta.get("clusterVersion") or meta.get("version") or "")),
    ):
        if value:
            params[key] = value
    return params


def _slow_query_url(cluster_id: str, start_ts: str, end_ts: str,
                    cluster_meta: dict | None = None, digest: str = "") -> str:
    """Generate Clinic NGM slow query list URL."""
    slow_query_params: dict[str, str] = {}
    if digest:
        slow_query_params["digest"] = digest
    if start_ts:
        slow_query_params["from"] = start_ts
    if end_ts:
        slow_query_params["to"] = end_ts
    slow_query_params["show_internal"] = "true"
    slow_query_params["term"] = ""
    return (
        f"{CLINIC_BASE}/portal/dashboard/cloud/ngm.html?"
        f"{urlencode(_ngm_page_params(cluster_id, cluster_meta), quote_via=quote)}"
        f"#/slow_query?{urlencode(slow_query_params, quote_via=quote)}"
    )


def _topsql_url(cluster_id: str, start_ts: str, end_ts: str,
                cluster_meta: dict | None = None, instance: str = "") -> str:
    """Generate Clinic NGM TopSQL page URL.

    orgName, clusterName, and version are optional. If other metadata is missing,
    keep generating an NGM URL with the fields we do have instead of returning
    the deprecated TopSQL route.
    """
    topsql_params = {"from": start_ts, "to": end_ts}
    if instance:
        topsql_params["instance"] = instance

    return (
        f"{CLINIC_BASE}/portal/dashboard/cloud/ngm.html?"
        f"{urlencode(_ngm_page_params(cluster_id, cluster_meta), quote_via=quote)}"
        f"#/topsql?{urlencode(topsql_params, quote_via=quote)}"
    )


def _topsql_url_template(cluster_id: str, start_ts: str, end_ts: str,
                         cluster_meta: dict | None = None) -> str:
    """Generate a compact TopSQL URL template for agent text output.

    The agent should replace {urlencoded_instance} with the selected row's
    URL-encoded instance value.
    """
    return _topsql_url(cluster_id, start_ts, end_ts, cluster_meta, "__INSTANCE__").replace(
        "instance=__INSTANCE__",
        "instance={urlencoded_instance}",
    )


# ---------------------------------------------------------------------------
# Data collection
# ---------------------------------------------------------------------------

def _collection_profile_flags(profile: str) -> tuple[bool, bool]:
    """Return (collect_slow_query, collect_wave2_attribution)."""
    if profile == "full":
        return True, True
    if profile == "fast":
        return False, False
    raise ValueError(f"Unsupported collection profile: {profile}")


_BYOC_PROFILE_WINDOW_SECONDS = 10_799


def _byoc_profile_windows(start: str, end: str) -> list[tuple[str, str]]:
    """Split a BYOC profiling range into API-safe, non-overlapping windows.

    The upstream endpoint rejects minute-rounded ranges of 10,800 seconds or
    more. Keeping each inclusive range to 10,799 seconds leaves room for that
    rounding behaviour.
    """
    start_ts, end_ts = int(start), int(end)
    if end_ts < start_ts:
        return []
    windows: list[tuple[str, str]] = []
    cursor = start_ts
    while cursor <= end_ts:
        window_end = min(cursor + _BYOC_PROFILE_WINDOW_SECONDS, end_ts)
        windows.append((str(cursor), str(window_end)))
        cursor = window_end + 1
    return windows


def _byoc_available_statement_fields(response: object) -> set[str]:
    """Normalize the BYOC Statements field-discovery response."""
    payload = response.get("data", response) if isinstance(response, dict) else response
    if not isinstance(payload, list):
        return set()
    return {str(field) for field in payload if isinstance(field, str)}


def _byoc_metric_scope(cluster_id: str, physical_cluster_id: str, is_byoc: bool,
                       *, data_plane: bool) -> str:
    """Select the safe metrics scope for a logical BYOC cluster."""
    if is_byoc and data_plane and physical_cluster_id:
        return physical_cluster_id
    return cluster_id


def collect(api: ClinicAPI, cluster_id: str, start: str, end: str,
            parent_id: str | None = None, profile: str = "full") -> dict:
    """Collect evidence sections A1-A6 + CPU quota.

    ``parent_id``: For Premium clusters, TiKV/PD/worker metrics live under the
    Resource Pool cluster.  When provided, component health queries
    (count, restarts, worker memory) are sent to ``parent_id`` instead of
    ``cluster_id``.

    All independent API calls run in parallel via ThreadPoolExecutor. The
    ``fast`` profile keeps the broad Wave 1 routing signals but skips the
    SlowQuery scan and Wave 2 attribution queries.
    Dependency graph:
        Wave 1 (parallel): A0, A1×3, A2×16, A3, BYOC A3a, A4×2, A6 (log_stats)
        Wave 2 (after wave 1):
            A3b decoded_plan  — needs A3 + A4 results
            A5 tikv_thread_cpu — needs A2 cpu data
    """
    from concurrent.futures import ThreadPoolExecutor, Future

    prefetched_cluster_meta: dict = _fetch_cluster_meta(api, cluster_id)
    is_byoc = _is_byoc_cluster_meta(prefetched_cluster_meta)
    if not parent_id:
        parent_id = _cluster_parent_id(prefetched_cluster_meta) or None
    physical_cluster_id = str(
        prefetched_cluster_meta.get("physical_cluster_id")
        or (parent_id if is_byoc else "")
        or (cluster_id if _cluster_deploy_type(prefetched_cluster_meta) == "byoc-host" else "")
    )
    byoc_namespace = f"tidb{physical_cluster_id}" if is_byoc and physical_cluster_id else ""

    collect_slow_query, collect_wave2_attribution = _collection_profile_flags(profile)

    data: dict = {
        "errors": [],
        "parent_id": parent_id or "",
        "collection_profile": profile,
        "is_byoc": is_byoc,
        "physical_cluster_id": physical_cluster_id,
        "data_plane_namespace": byoc_namespace,
        "statements": [],
        "profiles": [],
    }
    _timings: list[tuple[str, float, str]] = []
    _timings_lock = __import__("threading").Lock()
    _errors_lock = __import__("threading").Lock()
    _collect_t0 = time.monotonic()
    _collector_id = f"collector-{int(time.time() * 1000)}-{os.getpid()}"
    _debug = os.environ.get("COLLECT_EVIDENCE_DEBUG", "") == "1"
    _ensure_perf_log()
    _perf_event({
        "event": "collector_start",
        "collector_id": _collector_id,
        "cluster_id": cluster_id,
        "start": start,
        "end": end,
        "parent_id": parent_id or "",
        "profile": profile,
    })

    def _record_timing(label: str, elapsed: float, extra: str = "") -> None:
        with _timings_lock:
            _timings.append((label, elapsed, extra))

    def _record_error(msg: str) -> None:
        with _errors_lock:
            data["errors"].append(msg)

    # A BYOC logical cluster exposes tenant-scoped TiDB metrics, but TiKV/PD
    # metrics are served by its physical BYOC host. Query the latter for
    # component-health and data-plane signals; keep TiDB/QPS on ``cluster_id``
    # so tenant-scoped request metrics cannot include neighbouring tenants.
    byoc_data_plane_scope = _byoc_metric_scope(
        cluster_id, physical_cluster_id, is_byoc, data_plane=True,
    )
    health_scope = byoc_data_plane_scope if is_byoc else (parent_id or cluster_id)
    data["component_health_scope"] = health_scope
    component_start, component_end = _component_health_context_window(start, end)
    data["component_health_window"] = {"start": component_start, "end": component_end}

    def _safe_range(query: str, label: str, scope: str | None = None,
                    range_start: str | None = None, range_end: str | None = None,
                    attempts: int = 1) -> list[dict]:
        target = scope or cluster_id
        query_start = range_start or start
        query_end = range_end or end
        resp: dict = {}
        result: list[dict] = []
        for attempt in range(1, max(1, attempts) + 1):
            op_id = f"{_collector_id}:{label}:{int(time.time() * 1000)}:{attempt}"
            _perf_event({
                "event": "collector_op_start",
                "collector_id": _collector_id,
                "op_id": op_id,
                "op_kind": "promql_range",
                "label": label,
                "attempt": attempt,
                "cluster_id": target,
                "query_hash": _short_hash(query),
                "query_preview": query[:240],
                "start": query_start,
                "end": query_end,
                "step": "60",
            })
            t0 = time.monotonic()
            resp = api.query_range(target, query, query_start, query_end)
            elapsed = time.monotonic() - t0
            result = _prom_results(resp) if "error" not in resp else []
            timing_label = label if attempts <= 1 else f"{label}:attempt{attempt}"
            extra = f"series={len(result)}"
            if "error" in resp:
                extra += f" error={str(resp['error'])[:160]}"
            _record_timing(timing_label, elapsed, extra)
            _perf_event({
                "event": "collector_op_end",
                "collector_id": _collector_id,
                "op_id": op_id,
                "op_kind": "promql_range",
                "label": label,
                "attempt": attempt,
                "elapsed_ms": round(elapsed * 1000, 3),
                "status": "error" if "error" in resp else "ok",
                "error": str(resp.get("error", ""))[:500] if isinstance(resp, dict) else "",
                "series": len(result),
            })
            if "error" not in resp or not _query_range_retryable_error(resp) or attempt >= attempts:
                break
            time.sleep(min(8.0, 2.0 * attempt))
        if "error" in resp:
            _record_error(f"{label}: {resp['error']}")
        else:
            _record_raw("metrics", label, resp, query=query, scope=target,
                        start=query_start, end=query_end)
        return result

    def _safe_sql(sql: str, label: str, timeout: int = 60, attempts: int = 1) -> dict:
        op_id = f"{_collector_id}:{label}:{int(time.time() * 1000)}"
        _perf_event({
            "event": "collector_op_start",
            "collector_id": _collector_id,
            "op_id": op_id,
            "op_kind": "sql",
            "label": label,
            "cluster_id": cluster_id,
            "query_hash": _short_hash(sql),
            "query_preview": sql[:240],
            "timeout_s": timeout,
            "attempts": attempts,
        })
        op_t0 = time.monotonic()
        resp, attempt_infos = _query_sql_with_retries(
            api, cluster_id, sql, timeout=timeout, attempts=attempts,
        )
        for info in attempt_infos:
            timing_label = label if attempts <= 1 else f"{label}:attempt{info['attempt']}"
            extra = f"rows={info['rows']}"
            if info.get("error"):
                extra += f" error={info['error']}"
            _record_timing(timing_label, float(info["elapsed"]), extra)
        if "error" in resp:
            _record_error(f"{label}: {resp['error']}")
            _perf_event({
                "event": "collector_op_end",
                "collector_id": _collector_id,
                "op_id": op_id,
                "op_kind": "sql",
                "label": label,
                "elapsed_ms": round((time.monotonic() - op_t0) * 1000, 3),
                "status": "error",
                "error": str(resp.get("error", ""))[:500],
                "attempts": attempt_infos,
            })
            return {}
        _perf_event({
            "event": "collector_op_end",
            "collector_id": _collector_id,
            "op_id": op_id,
            "op_kind": "sql",
            "label": label,
            "elapsed_ms": round((time.monotonic() - op_t0) * 1000, 3),
            "status": "ok",
            "rows": len(resp.get("rows", []) or []),
            "attempts": attempt_infos,
        })
        _record_raw("queries", label, resp, query=sql, scope=cluster_id,
                    start=start, end=end)
        return resp

    def _safe_byoc_slow_query(label: str) -> dict:
        t0 = time.monotonic()
        sample_limit = 100
        resp = api.byoc_slow_query_list(
            cluster_id,
            start,
            end,
            limit=sample_limit,
            order_by="query_time",
            desc=True,
            fields=(
                "query,timestamp,query_time,memory_max,total_keys,process_keys,"
                "digest,connection_id"
            ),
        )
        elapsed = time.monotonic() - t0
        if isinstance(resp, dict) and resp.get("error"):
            _record_timing(label, elapsed, f"error={str(resp['error'])[:160]}")
            _record_error(f"{label}: {resp['error']}")
            return {}
        table = _byoc_slow_query_table(resp, sample_limit=sample_limit)
        _record_timing(label, elapsed, f"rows={len(table.get('rows', []))}")
        _record_raw("queries", label, resp, query="slow-query:list", scope=cluster_id,
                    start=start, end=end)
        return table

    def _safe_byoc_statement_summary(label: str) -> dict:
        t0 = time.monotonic()
        available = api.byoc_statements_available_fields(cluster_id)
        available_fields = _byoc_available_statement_fields(available)
        # `normalized_sql` is not part of this BYOC API contract. Request a
        # SQL sample only when field discovery advertises a compatible field.
        requested = [
            field for field in (
                "digest", "stmt_type", "schema_name", "digest_text",
                "query_sample_text", "exec_count", "sum_latency",
                "max_latency", "max_mem",
            )
            if field in available_fields
        ]
        resp = api.byoc_statements_list(
            cluster_id,
            start,
            end,
            # If discovery is unavailable or empty, omit `fields` so the API
            # applies its own safe default instead of receiving guessed names.
            fields=",".join(requested) if requested else None,
        )
        elapsed = time.monotonic() - t0
        if isinstance(resp, dict) and resp.get("error"):
            _record_timing(label, elapsed, f"error={str(resp['error'])[:160]}")
            _record_error(f"{label}: {resp['error']}")
            return {}
        table = _byoc_statement_summary_table(resp)
        _record_timing(label, elapsed, f"rows={len(table.get('rows', []))}")
        _record_raw("queries", label, resp, query="statements:list", scope=cluster_id,
                    start=start, end=end)
        return table

    def _safe_byoc_continuous_profiles(label: str) -> list[dict]:
        if not physical_cluster_id:
            _record_error(f"{label}: BYOC physical cluster ID is unavailable")
            return []
        profiles: list[dict] = []
        seen: set[str] = set()
        for index, (window_start, window_end) in enumerate(_byoc_profile_windows(start, end), 1):
            t0 = time.monotonic()
            response = api.byoc_continuous_profiling_group_profiles(
                physical_cluster_id, window_start, window_end,
            )
            elapsed = time.monotonic() - t0
            window_label = f"{label}:{index}"
            if isinstance(response, dict) and response.get("error"):
                _record_timing(window_label, elapsed, f"error={str(response['error'])[:160]}")
                _record_error(f"{window_label}: {response['error']}")
                continue
            records = [item for item in _response_records(response) if isinstance(item, dict)]
            _record_timing(window_label, elapsed, f"profiles={len(records)}")
            _record_raw("profiles", window_label, response,
                        query="continuous-profiling:group-profiles",
                        scope=physical_cluster_id, start=window_start, end=window_end)
            for item in records:
                key = json.dumps(item, sort_keys=True, default=str)
                if key not in seen:
                    seen.add(key)
                    profiles.append(item)
        return profiles

    def _safe_byoc_top_sql(instance_type: str, label: str,
                           instance: str | None = None) -> dict:
        t0 = time.monotonic()
        instances = [instance] if instance else []
        if not instances:
            inventory = api.byoc_top_sql_instances(cluster_id, start, end)
            if isinstance(inventory, dict) and inventory.get("error"):
                elapsed = time.monotonic() - t0
                _record_timing(label, elapsed, f"error={str(inventory['error'])[:160]}")
                _record_error(f"{label}: {inventory['error']}")
                return {}
            _record_raw("queries", f"{label}:instances", inventory,
                        query="top-sql:instances", scope=cluster_id, start=start, end=end)
            instances = _byoc_top_sql_instance_names(inventory, instance_type)[:3]

        summaries: list[tuple[str, object]] = []
        for target_instance in instances:
            if not target_instance:
                continue
            response = api.byoc_top_sql_summary(
                cluster_id,
                start,
                end,
                "1m",
                20,
                target_instance,
                instance_type,
            )
            if isinstance(response, dict) and response.get("error"):
                _record_error(f"{label}:{target_instance}: {response['error']}")
                continue
            summaries.append((target_instance, response))
            _record_raw("queries", f"{label}:{target_instance}", response,
                        query=f"top-sql:summary:{instance_type}", scope=cluster_id,
                        start=start, end=end)
        table = _byoc_top_sql_table(summaries)
        _record_timing(label, time.monotonic() - t0, f"rows={len(table.get('rows', []))}")
        return table

    def _safe_disk() -> list[dict]:
        query = 'rate(node_disk_io_time_seconds_total{device=~"nvme.*"}[1m])'
        if not is_byoc:
            return _safe_range(query, "A1:disk")
        if not byoc_namespace:
            _record_error("A1:disk: BYOC physical namespace is unavailable")
            return []
        t0 = time.monotonic()
        resp = api.query_range_join_instance_to_pod(
            cluster_id,
            query,
            byoc_namespace,
            ".*(tikv|tiflash).*",
            start,
            end,
            step="60",
        )
        result = _prom_results(resp) if isinstance(resp, dict) and "error" not in resp else []
        _record_timing("A1:disk", time.monotonic() - t0, f"series={len(result)}")
        if not isinstance(resp, dict) or resp.get("error"):
            _record_error(f"A1:disk: {resp.get('error', 'malformed response') if isinstance(resp, dict) else 'malformed response'}")
            return []
        _record_raw("metrics", "A1:disk", resp, query=query, scope=cluster_id,
                    start=start, end=end)
        return result

    def _safe_exceeded_iops(component: str, label: str) -> list[dict]:
        op_id = f"{_collector_id}:{label}:{int(time.time() * 1000)}"
        _perf_event({
            "event": "collector_op_start",
            "collector_id": _collector_id,
            "op_id": op_id,
            "op_kind": "promql_exceeded_iops",
            "label": label,
            "cluster_id": cluster_id,
            "component": component,
            "start": start,
            "end": end,
            "step": "60",
            "rate_window": "1m",
        })
        t0 = time.monotonic()
        kwargs = {"step": "60", "rate_window": "1m"}
        if is_byoc and byoc_namespace:
            kwargs["namespace"] = byoc_namespace
        resp = api.query_exceeded_iops_by_pod(
            cluster_id, component, start, end, **kwargs)
        elapsed = time.monotonic() - t0
        result = _prom_results(resp) if "error" not in resp else []
        _record_timing(label, elapsed, f"series={len(result)}")
        _perf_event({
            "event": "collector_op_end",
            "collector_id": _collector_id,
            "op_id": op_id,
            "op_kind": "promql_exceeded_iops",
            "label": label,
            "elapsed_ms": round(elapsed * 1000, 3),
            "status": "error" if "error" in resp else "ok",
            "error": str(resp.get("error", ""))[:500] if isinstance(resp, dict) else "",
            "series": len(result),
        })
        if "error" in resp:
            _record_error(f"{label}: {resp['error']}")
            return []
        _record_raw("metrics", label, resp, query=f"exceeded_iops:{component}",
                    scope=cluster_id, start=start, end=end)
        return result

    # Pre-compute date strings used by A3/A4.
    date_str = datetime.fromtimestamp(float(start), tz=timezone.utc).strftime("%Y%m%d")
    topsql_date = datetime.fromtimestamp(float(start), tz=timezone.utc).strftime("%Y-%m-%d")

    # SQL statements for A3/A4.
    slow_sql = (
        f"SELECT digest, any_value(query) AS sample_sql, COUNT(*) AS exec_count, "
        f"MAX(query_time) AS max_time, AVG(query_time) AS avg_time, MAX(mem_max) AS max_mem "
        f"FROM clinic_data_proxy.slow_query_logs "
        f"WHERE date = '{date_str}' AND time >= {start} AND time <= {end} "
        f"GROUP BY digest ORDER BY exec_count DESC LIMIT 20"
    )
    tidb_topsql_sql = (
        f"SELECT instance, sql_digest, any_value(normalized_sql) AS sample_sql, "
        f"SUM(topsql_cpu_time_ms) AS total_cpu_ms, "
        f"ROUND(SUM(topsql_cpu_time_ms) * 100.0 / SUM(SUM(topsql_cpu_time_ms)) OVER (), 2) AS cpu_pct "
        f"FROM clinic_data_proxy.topsql "
        f"WHERE date = '{topsql_date}' AND timestamps BETWEEN {start} AND {end} "
        f"AND component = 'tidb' "
        f"GROUP BY instance, sql_digest ORDER BY total_cpu_ms DESC LIMIT 15"
    )
    tikv_topsql_sql = (
        f"SELECT instance, sql_digest, any_value(normalized_sql) AS sample_sql, "
        f"SUM(topsql_cpu_time_ms) AS total_cpu_ms "
        f"FROM clinic_data_proxy.topsql "
        f"WHERE date = '{topsql_date}' AND component = 'tikv' "
        f"AND timestamps BETWEEN {start} AND {end} "
        f"GROUP BY instance, sql_digest ORDER BY total_cpu_ms DESC LIMIT 20"
    )

    # ------------------------------------------------------------------
    # Wave 1: fire all independent requests in parallel
    # ------------------------------------------------------------------
    print("[collect_evidence] Collecting evidence (parallel) ...", file=sys.stderr, flush=True)
    wave1_t0 = time.monotonic()

    with ThreadPoolExecutor(max_workers=24) as pool:
        # A0: Cluster metadata
        def _fetch_meta():
            t0 = time.monotonic()
            result = prefetched_cluster_meta if prefetched_cluster_meta is not None else _fetch_cluster_meta(api, cluster_id)
            _record_timing("A0:cluster_meta", time.monotonic() - t0)
            return result
        fut_meta: Future = pool.submit(_fetch_meta)

        # A1: Infra
        fut_ebs_tikv = pool.submit(_safe_exceeded_iops, "tikv", "A1:ebs_tikv")
        fut_ebs_tiflash = pool.submit(_safe_exceeded_iops, "tiflash", "A1:ebs_tiflash")
        fut_disk = pool.submit(_safe_disk)

        # A2: App metrics (16 independent queries)
        fut_latency = pool.submit(_safe_range,
            "histogram_quantile(0.99, sum(rate(tidb_server_handle_query_duration_seconds_bucket[1m])) by (le, instance))",
            "A2:latency")
        fut_grpc = pool.submit(_safe_range,
            'histogram_quantile(0.99, sum(rate(tikv_grpc_msg_duration_seconds_bucket{type!="raft"}[1m])) by (le, instance))',
            "A2:grpc", byoc_data_plane_scope)
        fut_connection_idle = pool.submit(_safe_range, CONNECTION_IDLE_P99_PROMQL, "A2:connection_idle_p99")
        fut_parse = pool.submit(_safe_range, PARSE_P99_PROMQL, "A2:parse_p99")
        fut_compile = pool.submit(_safe_range, COMPILE_P99_PROMQL, "A2:compile_p99")
        fut_kv_request = pool.submit(_safe_range, KV_REQUEST_P99_PROMQL, "A2:kv_request_p99")
        fut_kv_backoff = pool.submit(_safe_range, KV_BACKOFF_P99_PROMQL, "A2:kv_backoff_p99")
        fut_tso_wait = pool.submit(_safe_range, TSO_WAIT_P99_PROMQL, "A2:tso_wait_p99")
        fut_grpc_by_type = pool.submit(_safe_range, TIKV_GRPC_BY_TYPE_P99_PROMQL, "A2:grpc_by_type_p99", byoc_data_plane_scope)
        fut_scheduler_command = pool.submit(_safe_range, TIKV_SCHEDULER_COMMAND_P99_PROMQL, "A2:scheduler_command_p99", byoc_data_plane_scope)
        fut_scheduler_latch = pool.submit(_safe_range, TIKV_SCHEDULER_LATCH_P99_PROMQL, "A2:scheduler_latch_p99", byoc_data_plane_scope)
        fut_storage_async = pool.submit(_safe_range, TIKV_STORAGE_ASYNC_P99_PROMQL, "A2:storage_async_p99", byoc_data_plane_scope)
        fut_raft_commit = pool.submit(_safe_range, TIKV_RAFT_COMMIT_P99_PROMQL, "A2:raft_commit_p99", byoc_data_plane_scope)
        fut_raft_apply = pool.submit(_safe_range, TIKV_RAFT_APPLY_P99_PROMQL, "A2:raft_apply_p99", byoc_data_plane_scope)
        fut_pending_compaction = pool.submit(_safe_range, TIKV_PENDING_COMPACTION_PROMQL, "A2:pending_compaction_bytes", byoc_data_plane_scope)
        fut_qps = pool.submit(_safe_range, TIDB_QPS_PROMQL, "A2:tidb_qps")
        fut_tidb_memory = pool.submit(_safe_range,
            'process_resident_memory_bytes{component="tidb"}', "A2:tidb_memory")
        fut_store_memory = pool.submit(_safe_range,
            'process_resident_memory_bytes{component=~"tikv|pd|tiflash"}',
            "A2:store_memory", byoc_data_plane_scope)
        fut_tidb_cpu = pool.submit(_safe_range, TIDB_CPU_PROMQL, "A2:tidb_cpu")
        fut_tikv_cpu = pool.submit(_safe_range, TIKV_CPU_PROMQL, "A2:tikv_cpu", byoc_data_plane_scope)
        fut_tidb_cpu_quota = pool.submit(_safe_range, "tidb_server_maxprocs", "A2:tidb_cpu_quota")
        fut_store_cpu_quota = pool.submit(_safe_range,
            "tikv_server_cpu_cores_quota or pd_service_maxprocs",
            "A2:store_cpu_quota", byoc_data_plane_scope)
        fut_gc_freq = pool.submit(_safe_range,
            'rate(go_gc_duration_seconds_count{component="tidb"}[1m])',
            "A2:tidb_gc_frequency")
        fut_gc_dur = pool.submit(_safe_range,
            'rate(go_gc_duration_seconds_sum{component="tidb"}[1m])',
            "A2:tidb_gc_duration_rate")
        fut_gc_pause = pool.submit(_safe_range,
            'histogram_quantile(0.99, sum(rate(go_gc_pauses_seconds_bucket{component="tidb"}[1m])) by (le, instance))',
            "A2:tidb_gc_pause_p99")
        fut_stmt = pool.submit(_safe_range,
            "sum(rate(tidb_executor_statement_total[1m])) by (type)",
            "A2:statement_count")
        fut_rows = pool.submit(_safe_range,
            "sum(rate(tidb_executor_affected_rows[1m])) by (sql_type)",
            "A2:affected_rows")
        fut_rcu = pool.submit(_safe_range, RESOURCE_RCU_PROMQL, "A2:resource_control_rcu_usage")
        fut_req_p90 = pool.submit(_safe_range, RESOURCE_REQUEST_P90_PROMQL, "A2:resource_control_request_p90")
        fut_avail_ru = pool.submit(_safe_range, AVAILABLE_RU_PROMQL, "A2:resource_control_available_ru")
        fut_comp_count = pool.submit(_safe_range,
            "count by (component) (process_start_time_seconds)", "A2:component_count",
            health_scope, component_start, component_end, 4)
        fut_restarts = pool.submit(_safe_range,
            'changes(process_start_time_seconds{component=~"tidb|tikv|pd|coprocessor-worker|tikv-worker|tiproxy|auditlog"}[5m])',
            "A2:restarts", health_scope, component_start, component_end, 4)
        fut_worker_mem = pool.submit(_safe_range,
            'process_resident_memory_bytes{component=~"coprocessor-worker|tikv-worker"}',
            "A2:worker_memory", health_scope, component_start, component_end, 4)

        # A3 is attribution evidence; defer it until a fast diagnosis selects
        # the query/plan direction.
        fut_slow = (
            pool.submit(
                _safe_byoc_slow_query if is_byoc else _safe_sql,
                "A3:slow_queries" if is_byoc else slow_sql,
                *(() if is_byoc else ("A3:slow_queries",)),
            )
            if collect_slow_query else None
        )
        fut_statement_summary = (
            pool.submit(_safe_byoc_statement_summary, "A3a:statements_summary")
            if is_byoc and collect_slow_query else None
        )
        fut_profiles = (
            pool.submit(_safe_byoc_continuous_profiles, "A8:continuous_profiling")
            if is_byoc and collect_slow_query else None
        )

        # A4: TopSQL
        if is_byoc:
            fut_topsql_tidb = pool.submit(_safe_byoc_top_sql, "tidb", "A4:topsql_tidb")
            fut_topsql_tikv = pool.submit(_safe_byoc_top_sql, "tikv", "A4:topsql_tikv")
        else:
            fut_topsql_tidb = pool.submit(_safe_sql, tidb_topsql_sql, "A4:topsql_tidb", 60, 2)
            fut_topsql_tikv = pool.submit(_safe_sql, tikv_topsql_sql, "A4:topsql_tikv", 60, 2)

        # A6: Logs
        fut_log_stats: Future = pool.submit(collect_log_stats, api, cluster_id, start, end, prefetched_cluster_meta)

    # Harvest wave 1 results.
    data["cluster_meta"] = fut_meta.result()
    data["ebs"] = fut_ebs_tikv.result() + fut_ebs_tiflash.result()
    data["disk"] = fut_disk.result()
    data["latency"] = fut_latency.result()
    data["grpc"] = fut_grpc.result()
    data["connection_idle_p99"] = fut_connection_idle.result()
    data["parse_p99"] = fut_parse.result()
    data["compile_p99"] = fut_compile.result()
    data["kv_request_p99"] = fut_kv_request.result()
    data["kv_backoff_p99"] = fut_kv_backoff.result()
    data["tso_wait_p99"] = fut_tso_wait.result()
    data["grpc_by_type_p99"] = fut_grpc_by_type.result()
    data["scheduler_command_p99"] = fut_scheduler_command.result()
    data["scheduler_latch_p99"] = fut_scheduler_latch.result()
    data["storage_async_p99"] = fut_storage_async.result()
    data["raft_commit_p99"] = fut_raft_commit.result()
    data["raft_apply_p99"] = fut_raft_apply.result()
    data["pending_compaction_bytes"] = fut_pending_compaction.result()
    data["tidb_qps"] = fut_qps.result()
    data["component_memory"] = fut_tidb_memory.result() + fut_store_memory.result()
    data["tidb_cpu"] = fut_tidb_cpu.result()
    data["tikv_cpu"] = fut_tikv_cpu.result()
    data["cpu"] = data["tidb_cpu"] + data["tikv_cpu"]
    data["cpu_quota"] = fut_tidb_cpu_quota.result() + fut_store_cpu_quota.result()
    data["tidb_gc_frequency"] = fut_gc_freq.result()
    data["tidb_gc_duration_rate"] = fut_gc_dur.result()
    data["tidb_gc_pause_p99"] = fut_gc_pause.result()
    data["statement_count"] = fut_stmt.result()
    data["affected_rows"] = fut_rows.result()
    data["rcu_usage"] = fut_rcu.result()
    data["resource_manager_request_p90"] = fut_req_p90.result()
    data["available_ru"] = fut_avail_ru.result()
    data["component_count"] = fut_comp_count.result()
    data["restarts"] = fut_restarts.result()
    data["worker_memory"] = fut_worker_mem.result()
    data["slow"] = fut_slow.result() if fut_slow is not None else {}
    data["statement_summary"] = (
        fut_statement_summary.result() if fut_statement_summary is not None else {}
    )
    data["profiles"] = fut_profiles.result() if fut_profiles is not None else []
    data["topsql_tidb"] = fut_topsql_tidb.result()
    data["topsql_tikv"] = fut_topsql_tikv.result()
    data["log_stats"] = fut_log_stats.result()

    t0 = time.monotonic()
    data["events"] = collect_activity_events(api, cluster_id, start, end, data.get("cluster_meta"))
    _record_timing(
        "A7:activity_events",
        time.monotonic() - t0,
        f"events={len(data.get('events', {}).get('events', []))}",
    )
    if isinstance(data.get("events"), dict) and data["events"].get("error"):
        _record_error(f"A7:activity_events: {data['events']['error']}")

    print(f"[collect_evidence] Wave 1 done ({time.monotonic() - wave1_t0:.1f}s)", file=sys.stderr, flush=True)

    if not collect_wave2_attribution:
        data["topsql_tikv_hot_instances"] = []
        data["tikv_thread_cpu"] = {}
        data["top_plan"] = {}
        data["top_plan_source"] = ""
        data["top_plan_digest"] = ""
        total_elapsed = time.monotonic() - _collect_t0
        print(
            f"[collect_evidence] Fast profile complete after Wave 1: {total_elapsed:.1f}s | "
            f"errors: {len(data['errors'])}",
            file=sys.stderr,
            flush=True,
        )
        _perf_event({
            "event": "collector_end",
            "collector_id": _collector_id,
            "cluster_id": cluster_id,
            "profile": profile,
            "elapsed_ms": round(total_elapsed * 1000, 3),
            "errors": len(data["errors"]),
            "status": "ok",
        })
        return data

    # ------------------------------------------------------------------
    # Wave 2: depends on wave 1 results
    #   A4b TiKV hot instance TopSQL — needs A2 cpu data
    #   A3b decoded_plan — needs A3 + A4/A4b
    #   A5 tikv_thread_cpu — needs A2 cpu data
    # ------------------------------------------------------------------
    wave2_t0 = time.monotonic()
    print("[collect_evidence] Wave 2 (hot TopSQL + plan + thread CPU) ...", file=sys.stderr, flush=True)

    hot_tikv_candidates = _select_tikv_thread_cpu_instances(data)

    with ThreadPoolExecutor(max_workers=max(2, len(hot_tikv_candidates) + 1)) as pool:
        # A4b: hotspot-instance TopSQL. Global TiKV TopSQL is only background;
        # query attribution must use these instance-scoped rows.
        fut_hot_topsql = []
        for candidate in hot_tikv_candidates:
            instance = str(candidate.get("instance", ""))
            if not instance:
                continue
            sql = "" if is_byoc else _build_tikv_instance_topsql_sql(topsql_date, start, end, instance)
            fut_hot_topsql.append((
                instance,
                sql,
                pool.submit(
                    _safe_byoc_top_sql if is_byoc else _safe_sql,
                    *( ("tikv", f"A4b:topsql_tikv_hot:{instance}", instance)
                       if is_byoc
                       else (sql, f"A4b:topsql_tikv_hot:{instance}", 60, 2) ),
                ),
            ))

        # A5: TiKV thread CPU
        def _fetch_tikv_thread_cpu() -> None:
            t0 = time.monotonic()
            data["tikv_thread_cpu"] = collect_tikv_thread_cpu(api, cluster_id, start, end, data)
            _record_timing("A5:tikv_thread_cpu", time.monotonic() - t0)

        fut_thread = pool.submit(_fetch_tikv_thread_cpu)

    hot_topsql_results: list[dict] = []
    for instance, sql, fut in fut_hot_topsql:
        resp = fut.result()
        hot_topsql_results.append({
            "instance": instance,
            "query": sql,
            "columns": resp.get("columns", []),
            "rows": resp.get("rows", []),
        })
    data["topsql_tikv_hot_instances"] = hot_topsql_results
    fut_thread.result()

    with ThreadPoolExecutor(max_workers=1) as pool:
        # A3b: decoded_plan for the most impactful digest.
        def _fetch_decoded_plan() -> None:
            plan_digest = ""
            plan_digest_source = ""
            plan_schema_name = ""

            # Prefer hotspot-instance TopSQL. Global TiKV TopSQL can miss the
            # instance-specific driver and must not override scoped evidence.
            for scoped in data.get("topsql_tikv_hot_instances", []):
                scoped_cols = scoped.get("columns", [])
                scoped_rows = scoped.get("rows", [])
                if scoped_rows and scoped_cols:
                    top_scoped = dict(zip(scoped_cols, scoped_rows[0]))
                    candidate = str(top_scoped.get("sql_digest", ""))
                    if candidate:
                        plan_digest = candidate
                        plan_digest_source = f"topsql_tikv_hot:{scoped.get('instance', '')}"
                        break

            # Fallback to global TopSQL TiKV: pick the digest with highest total_cpu_ms.
            if not plan_digest:
                tikv_ts_cols = data.get("topsql_tikv", {}).get("columns", [])
                tikv_ts_rows = data.get("topsql_tikv", {}).get("rows", [])
                if tikv_ts_rows and tikv_ts_cols:
                    top_tikv = dict(zip(tikv_ts_cols, tikv_ts_rows[0]))
                    candidate = str(top_tikv.get("sql_digest", ""))
                    if candidate:
                        plan_digest = candidate
                        plan_digest_source = "topsql_tikv_global"

            # BYOC Statement Summary covers the full incident window. Slow Query
            # is only a latency-ranked sample and must not drive workload or plan
            # attribution. Non-BYOC keeps its server-side grouped slow-query fallback.
            if not plan_digest and is_byoc:
                statement_rows = data.get("statement_summary", {}).get("rows", [])
                statement_cols = data.get("statement_summary", {}).get("columns", [])
                if statement_rows and statement_cols:
                    statement_records = [dict(zip(statement_cols, row)) for row in statement_rows]
                    top_statement = max(
                        statement_records,
                        key=lambda row: (_safe_int(row.get("exec_count")), _safe_float(row.get("max_time"))),
                    )
                    candidate = str(top_statement.get("digest", ""))
                    plan_schema_name = str(top_statement.get("schema_name", ""))
                    if candidate:
                        plan_digest = candidate
                        plan_digest_source = "statement_summary"
            elif not plan_digest:
                slow_rows = data["slow"].get("rows", [])
                slow_cols = data["slow"].get("columns", [])
                if slow_rows and slow_cols:
                    top_slow = dict(zip(slow_cols, slow_rows[0]))
                    candidate = str(top_slow.get("digest", ""))
                    if candidate:
                        plan_digest = candidate
                        plan_digest_source = "slow_query"

            if plan_digest and is_byoc:
                matching_statements = [
                    dict(zip(data.get("statement_summary", {}).get("columns", []), row))
                    for row in data.get("statement_summary", {}).get("rows", [])
                ]
                matching_statements = [
                    row for row in matching_statements
                    if str(row.get("digest", "")) == plan_digest
                ]
                if matching_statements:
                    plan_schema_name = str(max(
                        matching_statements,
                        key=lambda row: _safe_int(row.get("exec_count")),
                    ).get("schema_name", ""))
                plans_resp = api.byoc_statements_plans(
                    cluster_id,
                    start,
                    end,
                    plan_digest,
                    schema_name=plan_schema_name or None,
                )
                if isinstance(plans_resp, dict) and plans_resp.get("error"):
                    _record_error(f"A3b:statements:plans: {plans_resp['error']}")
                    return
                plan_records = _response_records(plans_resp)
                _record_raw("queries", "A3b:statements:plans", plans_resp,
                            query="statements:plans", scope=cluster_id,
                            start=start, end=end)
                data.setdefault("statements", []).append({
                    "operation": "plans",
                    "digest": plan_digest,
                    "data": plan_records,
                })
                plan_candidates = []
                for record in plan_records:
                    plan_id = str(
                        record.get("plan_digest")
                        or record.get("planDigest")
                        or record.get("digest")
                        or ""
                    )
                    count = _safe_int(record.get("exec_count", record.get("execCount", 0)))
                    if plan_id:
                        plan_candidates.append((plan_id, count))
                selected_plan = ""
                if plan_candidates:
                    counts = {count for _, count in plan_candidates}
                    if len(counts) == 1 and next(iter(counts)) == 0:
                        _record_error("A3b:statements:plans: plan frequency unavailable; plan detail not uniquely attributable")
                    else:
                        selected_plan = max(plan_candidates, key=lambda item: item[1])[0]
                detail_resp = api.byoc_statements_plan_detail(
                    cluster_id,
                    start,
                    end,
                    plan_digest,
                    plans=[selected_plan] if selected_plan else None,
                    schema_name=plan_schema_name or None,
                )
                if isinstance(detail_resp, dict) and detail_resp.get("error"):
                    _record_error(f"A3b:statements:plan-detail: {detail_resp['error']}")
                    return
                detail = _response_object(detail_resp)
                _record_raw("queries", "A3b:statements:plan-detail", detail_resp,
                            query="statements:plan-detail", scope=cluster_id,
                            start=start, end=end)
                data.setdefault("statements", []).append({
                    "operation": "plan-detail",
                    "digest": plan_digest,
                    "plan_digest": selected_plan,
                    "data": detail,
                })
                data["top_plan"] = _byoc_statement_plan_table(detail_resp)
                data["top_plan_source"] = plan_digest_source
                data["top_plan_digest"] = plan_digest
            elif plan_digest:
                # Lowercase the digest in Python so Athena sees a literal match
                # against the lowercase slow_query_logs.digest column.  This
                # enables Parquet predicate pushdown (≈3s / 12 MB) instead of
                # LOWER()+LIKE which defeats row-group statistics (25s / 562 MB).
                plan_digest_lower = plan_digest.lower()
                _where = (
                    f"WHERE date = '{date_str}' AND time >= {start} AND time <= {end} "
                    f"AND digest = '{plan_digest_lower}'"
                )
                # Split into two queries to avoid scanning the large decoded_plan
                # column during aggregation (24s → ~3s).
                # Query A: MAX aggregates only (small columns, fast).
                agg_sql = (
                    f"SELECT "
                    f"MAX(query_time) AS query_time, "
                    f"MAX(process_time) AS process_time, MAX(wait_time) AS wait_time, "
                    f"MAX(total_keys) AS total_keys, MAX(process_keys) AS process_keys "
                    f"FROM clinic_data_proxy.slow_query_logs {_where}"
                )
                # Query B: single row with all columns including decoded_plan
                # (LIMIT 1, no aggregation — Athena stops at first match).
                row_sql = (
                    f"SELECT digest, query AS full_sql, "
                    f"query_time, process_time, wait_time, total_keys, process_keys, "
                    f"decoded_plan, stats AS stats_info, "
                    f"backoff_types AS backoff, index_names AS indexes_used "
                    f"FROM clinic_data_proxy.slow_query_logs {_where} "
                    f"AND decoded_plan IS NOT NULL AND decoded_plan <> '' "
                    f"LIMIT 1"
                )
                with ThreadPoolExecutor(max_workers=2) as _plan_pool:
                    _fut_agg = _plan_pool.submit(_safe_sql, agg_sql, "A3b:agg")
                    _fut_row = _plan_pool.submit(_safe_sql, row_sql, "A3b:row")
                agg_resp = _fut_agg.result()
                row_resp = _fut_row.result()

                # Merge: use row_resp as base, overlay MAX values from agg_resp.
                row_cols = row_resp.get("columns", [])
                row_rows = row_resp.get("rows", [])
                agg_cols = agg_resp.get("columns", [])
                agg_rows = agg_resp.get("rows", [])
                if row_rows and agg_rows:
                    rec = dict(zip(row_cols, row_rows[0]))
                    agg = dict(zip(agg_cols, agg_rows[0]))
                    for k in ("query_time", "process_time", "wait_time",
                              "total_keys", "process_keys"):
                        if k in agg:
                            rec[k] = agg[k]
                    merged_cols = list(rec.keys())
                    merged_rows = [list(rec.values())]
                    data["top_plan"] = {"columns": merged_cols, "rows": merged_rows}
                elif row_rows:
                    data["top_plan"] = row_resp
                elif agg_rows:
                    data["top_plan"] = agg_resp
                else:
                    data["top_plan"] = {}

                data["top_plan_source"] = plan_digest_source
                data["top_plan_digest"] = plan_digest

        fut_plan = pool.submit(_fetch_decoded_plan)

    fut_plan.result()

    print(f"[collect_evidence] Wave 2 done ({time.monotonic() - wave2_t0:.1f}s)", file=sys.stderr, flush=True)

    total_elapsed = time.monotonic() - _collect_t0
    print(f"[collect_evidence] Total: {total_elapsed:.1f}s | errors: {len(data['errors'])}", file=sys.stderr, flush=True)
    _perf_event({
        "event": "collector_end",
        "collector_id": _collector_id,
        "cluster_id": cluster_id,
        "elapsed_ms": round(total_elapsed * 1000, 3),
        "errors": len(data["errors"]),
        "status": "ok",
    })

    # Debug mode: write detailed per-query timing to /tmp
    if _debug:
        import tempfile
        timing_path = os.path.join(tempfile.gettempdir(), f"collect_evidence_timing_{cluster_id}.log")
        with open(timing_path, "w") as f:
            f.write(f"=== collect_evidence timing ({total_elapsed:.1f}s total) ===\n")
            for label, elapsed, extra in sorted(_timings, key=lambda t: t[1], reverse=True):
                bar = "#" * int(elapsed * 2)
                line = f"  {label:40s} {elapsed:6.2f}s  {bar}"
                if extra:
                    line += f"  ({extra})"
                f.write(line + "\n")
            f.write(f"  {'TOTAL':40s} {total_elapsed:6.2f}s\n")
        print(f"[collect_evidence] Debug timing written to {timing_path}", file=sys.stderr, flush=True)

    return data


def _select_tikv_thread_cpu_instances(data: dict, limit: int = 3) -> list[dict]:
    cpu_quota: dict[str, float] = {}
    for r in data.get("cpu_quota", []):
        inst = _instance_name(r)
        median = _prom_median(r)
        if median > 0:
            cpu_quota[inst] = median

    tikv_cpus: list[dict] = []
    for r in data.get("cpu", []):
        inst = _instance_name(r)
        if "tikv" not in inst.lower():
            continue
        peak = _prom_peak(r)
        quota = cpu_quota.get(inst)
        util = peak / quota if quota and quota > 0 else None
        tikv_cpus.append({
            "instance": inst,
            "peak": peak,
            "mean": _prom_mean(r),
            "quota": quota,
            "utilization": util,
        })

    saturated = [c for c in tikv_cpus if c["utilization"] is not None and c["utilization"] >= 0.8]
    if saturated:
        return sorted(saturated, key=lambda c: c["utilization"] or 0, reverse=True)
    return sorted(tikv_cpus, key=lambda c: c["peak"], reverse=True)[:limit]


def _prom_regex_literal(value: str) -> str:
    return re.sub(r'([.^$*+?{}\[\]|()])', r'\\\1', value).replace('"', r'\"')


def _sql_string_literal(value: str) -> str:
    return value.replace("'", "''")


def _sql_like_contains_literal(value: str) -> str:
    escaped = _sql_string_literal(value)
    return f"'%{escaped}%'"


def _build_tikv_instance_topsql_sql(
    topsql_date: str,
    start: str,
    end: str,
    instance: str,
    limit: int = 20,
) -> str:
    instance_like = _sql_like_contains_literal(instance)
    return (
        "SELECT instance, sql_digest, any_value(normalized_sql) AS sample_sql, "
        "SUM(topsql_cpu_time_ms) AS total_cpu_ms, "
        "ROUND(SUM(topsql_cpu_time_ms) * 100.0 / SUM(SUM(topsql_cpu_time_ms)) OVER (), 2) AS cpu_pct "
        "FROM clinic_data_proxy.topsql "
        f"WHERE date = '{_sql_string_literal(topsql_date)}' AND component = 'tikv' "
        f"AND instance LIKE {instance_like} "
        f"AND timestamps BETWEEN {start} AND {end} "
        f"GROUP BY instance, sql_digest ORDER BY total_cpu_ms DESC LIMIT {int(limit)}"
    )


def _query_sql_with_retries(
    api: ClinicAPI,
    cluster_id: str,
    sql: str,
    timeout: int = 60,
    attempts: int = 1,
    retry_delay: float = 1.0,
) -> tuple[dict, list[dict]]:
    attempt_count = max(1, int(attempts))
    attempt_infos: list[dict] = []
    last_resp: dict = {}

    for attempt in range(1, attempt_count + 1):
        t0 = time.monotonic()
        try:
            resp = api.query_sql(cluster_id, sql, timeout=timeout)
            if not isinstance(resp, dict):
                resp = {"error": f"malformed SQL response type={type(resp).__name__}"}
        except Exception as exc:  # pragma: no cover - exact network exceptions vary by backend
            resp = {"error": str(exc)}

        elapsed = time.monotonic() - t0
        rows = len(resp.get("rows", []) or []) if isinstance(resp, dict) else 0
        info = {"attempt": attempt, "elapsed": elapsed, "rows": rows}
        if "error" in resp:
            info["error"] = str(resp.get("error", "unknown error"))
        attempt_infos.append(info)
        last_resp = resp

        if "error" not in resp:
            return resp, attempt_infos
        if attempt < attempt_count and retry_delay > 0:
            time.sleep(retry_delay)

    return last_resp, attempt_infos


def _classify_tikv_thread(name: str) -> str:
    n = name.lower()
    if "grpc" in n:
        return "grpc"
    if "cop" in n:
        return "cop"
    if "raftstore" in n or "raft-store" in n:
        return "raftstore"
    if "apply" in n:
        return "apply"
    if "sched" in n or "scheduler" in n:
        return "sched"
    if "unified-read" in n or "unified_read" in n:
        return "unified_read"
    if "rocksdb" in n:
        return "rocksdb"
    if "background" in n or "bg" in n:
        return "background"
    return "other"


def collect_tikv_thread_cpu(api: ClinicAPI, cluster_id: str, start: str, end: str, data: dict) -> dict:
    """Strictly collect TiKV thread CPU evidence when TiKV CPU data is present."""
    candidates = _select_tikv_thread_cpu_instances(data)
    if not candidates:
        return {
            "status": "skipped",
            "reason": "no TiKV CPU candidates found",
            "selected_instances": [],
            "promql": "",
            "threads": [],
            "buckets": [],
        }

    instance_regex = "|".join(_prom_regex_literal(c["instance"]) for c in candidates)
    promql = (
        "sum by (instance, name) ("
        f'rate(tikv_thread_cpu_seconds_total{{instance=~"{instance_regex}"}}[1m])'
        ")"
    )
    op_id = f"tikv_thread_cpu:{int(time.time() * 1000)}:{os.getpid()}"
    _perf_event({
        "event": "collector_op_start",
        "op_id": op_id,
        "op_kind": "promql_range",
        "label": "A5:tikv_thread_cpu",
        "cluster_id": cluster_id,
        "query_hash": _short_hash(promql),
        "query_preview": promql[:240],
        "start": start,
        "end": end,
        "step": "60",
    })
    op_t0 = time.monotonic()
    try:
        results = _prom_results_strict(
            api.query_range(cluster_id, promql, start, end),
            "TiKV_THREAD_CPU",
        )
    except Exception as exc:
        _perf_event({
            "event": "collector_op_end",
            "op_id": op_id,
            "op_kind": "promql_range",
            "label": "A5:tikv_thread_cpu",
            "elapsed_ms": round((time.monotonic() - op_t0) * 1000, 3),
            "status": "error",
            "error": str(exc)[:500],
        })
        raise
    _perf_event({
        "event": "collector_op_end",
        "op_id": op_id,
        "op_kind": "promql_range",
        "label": "A5:tikv_thread_cpu",
        "elapsed_ms": round((time.monotonic() - op_t0) * 1000, 3),
        "status": "ok",
        "series": len(results),
    })

    selected = {c["instance"] for c in candidates}
    threads: list[dict] = []
    bucket_totals: dict[tuple[str, str], float] = {}
    instances_with_data: set[str] = set()

    for r in results:
        metric = r.get("metric")
        if not isinstance(metric, dict):
            raise RuntimeError("TiKV_THREAD_CPU: malformed result missing metric labels")
        inst = str(metric.get("instance") or "")
        name = str(metric.get("name") or "")
        if not inst or not name:
            raise RuntimeError("TiKV_THREAD_CPU: malformed result missing instance/name label")
        if inst not in selected:
            continue
        values = r.get("values")
        if not isinstance(values, list) or not values:
            raise RuntimeError(f"TiKV_THREAD_CPU: empty values for {inst} {name}")
        peak = _prom_peak(r)
        mean = _prom_mean(r)
        bucket = _classify_tikv_thread(name)
        instances_with_data.add(inst)
        bucket_totals[(inst, bucket)] = bucket_totals.get((inst, bucket), 0.0) + peak
        threads.append({
            "instance": inst,
            "name": name,
            "bucket": bucket,
            "peak": round(peak, 4),
            "mean": round(mean, 4),
        })

    if not threads:
        raise RuntimeError("TiKV_THREAD_CPU: query returned no matching selected TiKV instances")

    missing_instances = sorted(selected - instances_with_data)
    if missing_instances:
        raise RuntimeError(
            "TiKV_THREAD_CPU: missing thread CPU data for selected instance(s): "
            + ", ".join(missing_instances)
        )

    buckets = [
        {"instance": inst, "bucket": bucket, "peak": round(peak, 4)}
        for (inst, bucket), peak in sorted(bucket_totals.items(), key=lambda item: item[1], reverse=True)
    ]

    return {
        "status": "collected",
        "selected_instances": candidates,
        "promql": promql,
        "threads": sorted(threads, key=lambda t: t["peak"], reverse=True),
        "buckets": buckets,
    }


def collect_log_stats(api: ClinicAPI, cluster_id: str, start: str, end: str,
                      cluster_meta: dict | None = None) -> dict:
    """Collect log statistics for section A6.

    Prefer O11y LogStats via Clinic Data Proxy when available.  It avoids broad
    Loki query_range scans and returns hourly aggregates plus representative samples.
    If LogStats is disabled/unavailable, fall back to the original Loki path.
    """
    meta = cluster_meta if isinstance(cluster_meta, dict) else None
    if meta is None and api is not None:
        meta = _fetch_cluster_meta(api, cluster_id)
    is_byoc = _is_byoc_cluster_meta(meta)
    if api is not None and not is_byoc and _logstats_backend_enabled(meta):
        payload, logstats_error = _collect_log_stats_from_clinic_api(api, meta or {}, cluster_id, start, end)
        if payload is not None and _logstats_payload_has_evidence(payload):
            return payload
        if payload is not None:
            logstats_error = "Clinic LogStats returned no log counts, samples, or timeline evidence"
        _perf_event({
            "event": "collector_op_end",
            "op_kind": "clinic_logstats",
            "label": "A6:logstats_fallback",
            "status": "fallback",
            "error": str(logstats_error or "")[:500],
        })

    step = "300"  # 5-minute buckets
    components = ["tidb", "tikv", "pd"]
    levels = {
        "error": '|~ "(?i)(ERROR|FATAL)"',
        "warn":  '|~ "(?i)WARN"',
        # LogQL string literals reject bare `\[` / `\]` escapes. Double-escape
        # the regex so Loki receives literal bracket matching for `[INFO]`.
        "info":  '|~ "(?i)\\\\[INFO\\\\]"',
    }

    errors: list[str] = []
    # timeline: { component -> { level -> [ {window_start, count} ] } }
    timeline: dict[str, dict[str, list[dict]]] = {}
    timeline_label = "container" if is_byoc else "component"

    # Fire all 3×3 = 9 Loki queries in parallel.
    from concurrent.futures import ThreadPoolExecutor

    def _loki_one(component: str, level: str, filter_expr: str) -> tuple[str, str, list[dict], str | None]:
        stream_selector = f'{{{timeline_label}="{component}"}}'
        metric_query = f"sum(count_over_time({stream_selector} {filter_expr} [{step}s]))"
        label = f"A6:logs:{component}:{level}"
        op_id = f"{label}:{int(time.time() * 1000)}:{os.getpid()}"
        _perf_event({
            "event": "collector_op_start",
            "op_id": op_id,
            "op_kind": "loki_range",
            "label": label,
            "cluster_id": cluster_id,
            "query_hash": _short_hash(metric_query),
            "query_preview": metric_query[:240],
            "start": start,
            "end": end,
            "step": step,
        })
        t0 = time.monotonic()
        resp = _query_loki(api, cluster_id, metric_query, start, end, step)
        elapsed_ms = round((time.monotonic() - t0) * 1000, 3)
        if "error" in resp:
            _perf_event({
                "event": "collector_op_end",
                "op_id": op_id,
                "op_kind": "loki_range",
                "label": label,
                "elapsed_ms": elapsed_ms,
                "status": "error",
                "error": str(resp["error"])[:500],
            })
            return component, level, [], f"A6:{component}_{level}: {resp['error']}"
        buckets: list[dict] = []
        for r in _prom_results(resp):
            for ts, val in r.get("values", []):
                try:
                    count = int(float(val))
                except (ValueError, TypeError):
                    count = 0
                if count > 0:
                    buckets.append({"window_start": int(ts), "count": count})
        _perf_event({
            "event": "collector_op_end",
            "op_id": op_id,
            "op_kind": "loki_range",
            "label": label,
            "elapsed_ms": elapsed_ms,
            "status": "ok",
            "buckets": len(buckets),
        })
        return component, level, buckets, None

    with ThreadPoolExecutor(max_workers=9) as pool:
        futures = [
            pool.submit(_loki_one, comp, lvl, filt)
            for comp in components for lvl, filt in levels.items()
        ]

    for comp in components:
        timeline[comp] = {}
    for fut in futures:
        comp, lvl, buckets, err = fut.result()
        timeline[comp][lvl] = buckets
        if err:
            errors.append(err)

    # Build per-component summary totals
    summary: dict[str, dict[str, int]] = {}
    for component in components:
        summary[component] = {
            level: sum(b["count"] for b in timeline[component].get(level, []))
            for level in levels
        }

    focused_start, focused_end = _focused_log_context_window(start, end)
    log_counts, samples, focused_error = _collect_focused_log_samples(
        api,
        cluster_id,
        focused_start,
        focused_end,
    )
    if focused_error:
        errors.append(focused_error)

    result = {
        "query": (
            f'{{{timeline_label}=~"tidb|tikv|pd|coprocessor-worker"}} |~ '
            f'"(?i)(panic|interrupt|kill|oom|restart|timeout|backlog|unavailable|coprocessor|{TRANSACTION_LOG_SIGNAL_REGEX})"'
        ),
        "source": "loki",
        "canonical_window": {"start": start, "end": end},
        "focused_log_context_window": {"start": focused_start, "end": focused_end},
        "timeline": timeline,
        "summary": summary,
        "log_counts": log_counts,
        "samples": samples,
        "step_seconds": int(step),
        "errors": errors,
        "loki_url": _loki_log_url(
            cluster_id,
            f'{{{timeline_label}=~"tidb|tikv|pd"}} |~ "(?i)(ERROR|FATAL|WARN)"',
            start,
            end,
            meta,
        ),
    }
    if is_byoc:
        result["warehouse_logstats"] = {
            "status": "unsupported",
            "reason": "BYOC uses canonical customer-side logs; warehouse LogStats is unavailable",
        }
    return _annotate_log_time_scope(result)


def _query_loki(api: ClinicAPI, cluster_id: str, query: str, start: str, end: str,
                 step: str = "300") -> dict:
    """Query a Loki metric range with a fixed step and no stream-only parameters.

    Log stream queries use ``_query_loki_logs`` instead.
    """
    if hasattr(api, "logs_query_range"):
        return api.logs_query_range(
            cluster_id,
            query,
            _loki_time_ns(start),
            _loki_time_ns(end),
            limit=None,
            direction=None,
            step=step,
        )
    return api._get(
        f"/data-proxy/loki/{cluster_id}/api/v1/query_range",
        {"query": query, "start": _loki_time_ns(start), "end": _loki_time_ns(end), "step": step},
    )


# ---------------------------------------------------------------------------
# Gate checks
# ---------------------------------------------------------------------------

def check_infra_gate(data: dict) -> str:
    """Return infra gate verdict string."""
    issues: list[str] = []
    missing: list[str] = []
    observed: dict[str, list[dict]] = {}
    for name in ("ebs", "disk"):
        rows = data.get(name, [])
        observed[name] = [r for r in rows if _prom_observed_peak(r) is not None]
        if not observed[name]:
            missing.append(f"{name}: no finite samples (unavailable, not zero)")
        elif len(observed[name]) < len(rows):
            missing.append(f"{name}: {len(rows) - len(observed[name])} series without finite samples")

    # EBS throttle
    severe = 0
    moderate = 0
    top_severe: list[str] = []
    for r in observed["ebs"]:
        peak = _prom_peak(r)
        if peak > 0.20:
            severe += 1
            m = r.get("metric", {})
            vol = m.get("volume_id", "?")
            pod = m.get("pod", "")
            label = f"{pod} {vol}" if pod else vol
            top_severe.append(f"{label}={peak:.0%}")
        elif peak > 0.05:
            moderate += 1
    total_vols = len(observed["ebs"])

    if severe > 0:
        issues.append(
            f"EBS IOPS throttling: {severe} volume(s) >20% "
            f"(total={total_vols}, moderate_5-20%={moderate}). "
            f"Top: {', '.join(top_severe[:3])}"
        )

    # Disk IO saturation
    max_util = 0.0
    max_disk_info = ""
    for r in observed["disk"]:
        peak = _prom_peak(r)
        if peak > max_util:
            max_util = peak
            m = r.get("metric", {})
            max_disk_info = f"{m.get('instance', '?')} {m.get('device', '?')}"
    if max_util > 0.8:
        issues.append(f"Disk IO saturation: {max_disk_info} util={max_util:.2f}")

    if issues:
        return "INFRA ABNORMAL: observed signals (causality unverified): " + "; ".join(issues + missing)

    details = []
    if observed["ebs"]:
        details.append(f"EBS total_volumes={total_vols} severe_gt20pct={severe} moderate_5_20pct={moderate}")
    if observed["disk"]:
        details.append(f"disk_io_util max={max_util:.2f}")
    prefix = "INFRA UNKNOWN: incomplete observations; " if missing else "INFRA CLEAR: queried series below thresholds only; "
    return prefix + "; ".join(details + missing)


def detect_hotspot(data: dict) -> str:
    """Return hotspot detection verdict string."""
    hotspots: list[str] = []

    # Parse CPU and quota data
    cpu_instances: list[dict] = []
    for r in data.get("cpu", []):
        inst = _instance_name(r)
        cpu_instances.append({"instance": inst, "peak": _prom_peak(r), "mean": _prom_mean(r)})

    cpu_quota: dict[str, float] = {}
    for r in data.get("cpu_quota", []):
        inst = _instance_name(r)
        median = _prom_median(r)
        if median > 0:
            cpu_quota[inst] = median

    tikv_cpus = [c for c in cpu_instances if "tikv" in c["instance"].lower()]
    tidb_cpus = [c for c in cpu_instances if "tidb" in c["instance"].lower()]

    # --- TiKV gRPC P99 hotspot (>2x peers) ---
    grpc_instances = []
    for r in data.get("grpc", []):
        grpc_instances.append({"instance": _instance_name(r), "peak": _prom_peak(r)})
    if len(grpc_instances) >= 2:
        avg = sum(g["peak"] for g in grpc_instances) / len(grpc_instances)
        if avg > 0:
            for g in grpc_instances:
                ratio = g["peak"] / avg
                if ratio >= 2.0:
                    hotspots.append(
                        f"Instance hotspot (grpc_p99): {g['instance']} "
                        f"value={g['peak']:.3f} peer_avg={avg:.3f} ratio={ratio:.1f}x"
                    )

    # --- TiKV CPU hotspot (>2x peers) ---
    flagged_by_relative: set[str] = set()
    if len(tikv_cpus) >= 2:
        avg = sum(c["peak"] for c in tikv_cpus) / len(tikv_cpus)
        if avg > 0:
            for c in tikv_cpus:
                ratio = c["peak"] / avg
                if ratio >= 2.0:
                    flagged_by_relative.add(c["instance"])
                    hotspots.append(
                        f"Instance hotspot (cpu): {c['instance']} "
                        f"value={c['peak']:.3f} peer_avg={avg:.3f} ratio={ratio:.1f}x"
                    )

    # --- CPU saturation (>80% utilization, absolute threshold) ---
    for c in cpu_instances:
        if c["instance"] in flagged_by_relative:
            continue
        q = cpu_quota.get(c["instance"])
        if not q or q <= 0:
            continue
        util = c["peak"] / q
        if util >= 0.8:
            hotspots.append(
                f"CPU saturation: {c['instance']} "
                f"peak={c['peak']:.1f}/{q:.0f} cores ({util:.0%} utilization)"
            )

    # --- TiDB CPU hotspot per-nodegroup (>2x within same group) ---
    tidb_by_ng: dict[str, list[dict]] = {}
    for c in tidb_cpus:
        inst = c["instance"]
        if "-ac-" in inst:
            import re
            m = re.search(r"db-tidb-\d+-(ac-\S+)", inst)
            ng = m.group(1) if m else "default"
        elif "db-tidb-" in inst:
            ng = "default"
        else:
            continue
        tidb_by_ng.setdefault(ng, []).append(c)

    for ng, members in tidb_by_ng.items():
        if len(members) < 2:
            continue
        avg = sum(c["peak"] for c in members) / len(members)
        if avg <= 0:
            continue
        label = "default node group" if ng == "default" else f"non-default node group ({ng})"
        for c in members:
            ratio = c["peak"] / avg
            if ratio >= 2.0:
                hotspots.append(
                    f"Instance hotspot (tidb_cpu({label})): {c['instance']} "
                    f"value={c['peak']:.3f} peer_avg={avg:.3f} ratio={ratio:.1f}x"
                )

    # --- TopSQL TiKV query hotspot (single query >60% CPU on one hot instance) ---
    # Only use hotspot-instance-scoped TopSQL. Global TopSQL is background
    # workload evidence and cannot safely prove dominance on a hotspot instance.
    for scoped in data.get("topsql_tikv_hot_instances", []):
        scoped_inst = str(scoped.get("instance", "unknown"))
        cols = scoped.get("columns", [])
        entries = [dict(zip(cols, row)) for row in scoped.get("rows", [])]
        if len(entries) < 2:
            continue
        total_cpu = sum(_safe_int(e.get("total_cpu_ms", 0)) for e in entries)
        if total_cpu <= 0:
            continue
        top_entry = max(entries, key=lambda e: _safe_int(e.get("total_cpu_ms", 0)))
        top_cpu = _safe_int(top_entry.get("total_cpu_ms", 0))
        top_pct = _safe_float(top_entry.get("cpu_pct", 0))
        if top_pct <= 0:
            top_pct = top_cpu * 100.0 / total_cpu
        if top_pct > 60:
            full_digest = str(top_entry.get("sql_digest", ""))
            row_inst = str(top_entry.get("instance", "")) or scoped_inst
            hotspots.append(
                f"Query hotspot: digest={full_digest} cpu_pct={top_pct:.1f}% on {row_inst}"
            )

    if not hotspots:
        return "HOTSPOT: not detected (instance metrics within normal variance)"
    return "HOTSPOT DETECTED: " + "; ".join(hotspots)


# ---------------------------------------------------------------------------
# Summary formatting
# ---------------------------------------------------------------------------

def format_summary(data: dict, infra_verdict: str, hotspot_verdict: str,
                   cluster_id: str = "", start: str = "", end: str = "") -> str:
    """Format collected data into the structured evidence text."""
    lines: list[str] = []

    # A1: Infra
    lines.append("## A1. Infrastructure Metrics")
    ebs = [r for r in data.get("ebs", []) if _prom_observed_peak(r) is not None]
    severe = sum(1 for r in ebs if _prom_peak(r) > 0.20)
    moderate = sum(1 for r in ebs if 0.05 < _prom_peak(r) <= 0.20)
    if ebs:
        lines.append(f"EBS_THROTTLE: observed_volumes={len(ebs)} severe_gt20pct={severe} moderate_5_20pct={moderate}")
    else:
        lines.append("EBS_THROTTLE: unavailable (no finite samples, not zero)")

    disk = [r for r in data.get("disk", []) if _prom_observed_peak(r) is not None]
    if disk:
        max_d = max(disk, key=_prom_peak)
        m = max_d.get("metric", {})
        lines.append(f"DISK_IO_UTIL: max={_prom_peak(max_d):.2f} ({m.get('instance', '?')} {m.get('device', '?')})")
    else:
        lines.append("DISK_IO_UTIL: unavailable (no finite samples, not zero)")

    # A2: App metrics
    lines.append("\n## A2. Application Metrics")

    # CPU quota lookup
    cpu_quota: dict[str, float] = {}
    for r in data.get("cpu_quota", []):
        inst = _instance_name(r)
        median = _prom_median(r)
        if median > 0:
            cpu_quota[inst] = median

    # TiDB P99 latency
    if data.get("latency"):
        lines.append("TiDB_P99_latency:")
        for r in sorted(data["latency"], key=_prom_peak, reverse=True)[:5]:
            lines.append(f"  {_instance_name(r)}: peak={_prom_peak(r):.3f}s mean={_prom_mean(r):.3f}s")

    # TiKV gRPC P99
    if data.get("grpc"):
        lines.append("TiKV_gRPC_P99 (top instances):")
        for r in sorted(data["grpc"], key=_prom_peak, reverse=True)[:3]:
            lines.append(f"  {_instance_name(r)}: peak={_prom_peak(r):.3f}s mean={_prom_mean(r):.3f}s")

    # CPU by instance (with utilization)
    if data.get("cpu"):
        lines.append("CPU_by_instance:")
        for r in sorted(data["cpu"], key=_prom_peak, reverse=True):
            inst = _instance_name(r)
            peak = _prom_peak(r)
            mean = _prom_mean(r)
            q = cpu_quota.get(inst)
            if q and q > 0:
                util = peak / q
                lines.append(f"  {inst}: peak={peak:.2f}/{q:.0f} cores ({util:.0%}) mean={mean:.2f}")
            else:
                lines.append(f"  {inst}: peak={peak:.2f} cores mean={mean:.2f}")

    thread_cpu = data.get("tikv_thread_cpu", {})
    lines.append("TiKV_THREAD_CPU:")
    if isinstance(thread_cpu, dict) and thread_cpu.get("status") == "collected":
        selected = thread_cpu.get("selected_instances", [])
        selected_names = [str(c.get("instance", "")) for c in selected if c.get("instance")]
        lines.append(f"  selected_instances={', '.join(selected_names)}")
        for row in thread_cpu.get("buckets", [])[:10]:
            lines.append(
                f"  {row.get('instance', '?')} {row.get('bucket', 'other')}: "
                f"peak={_safe_float(row.get('peak', 0)):.3f} cores"
            )
    elif isinstance(thread_cpu, dict):
        lines.append(f"  skipped: {thread_cpu.get('reason', 'not collected')}")
    else:
        lines.append("  not collected")

    lines.append("TIDB_GC_CHECK:")
    gc_freq = {_instance_name(r): r for r in data.get("tidb_gc_frequency", [])}
    gc_duration = {_instance_name(r): r for r in data.get("tidb_gc_duration_rate", [])}
    gc_pause = {_instance_name(r): r for r in data.get("tidb_gc_pause_p99", [])}
    gc_instances = sorted(set(gc_freq) | set(gc_duration) | set(gc_pause))
    if gc_instances:
        for inst in gc_instances:
            freq_peak = _prom_peak(gc_freq.get(inst, {}))
            duration_peak = _prom_peak(gc_duration.get(inst, {}))
            pause_peak = _prom_peak(gc_pause.get(inst, {}))
            abnormal = freq_peak > 10 or duration_peak > 0.1 or pause_peak > 0.1
            verdict = "abnormal" if abnormal else "normal"
            lines.append(
                f"  instance={inst} gc_frequency_peak={freq_peak:.2f}/s "
                f"gc_duration_rate_peak={duration_peak:.3f}s/s "
                f"gc_pause_p99_peak={pause_peak:.3f}s verdict={verdict}"
            )
    else:
        lines.append("  metrics unavailable or empty: check go_gc_duration_seconds and go_gc_pauses_seconds series before concluding TiDB CPU root cause")

    workload = compact_workload_for_ai(data)
    lines.append("WORKLOAD_STATEMENT_VS_AFFECTED_ROWS:")
    if workload:
        for row in workload[:10]:
            lines.append(
                f"  sql_type={row.get('sql_type', 'unknown')} "
                f"statement_rate_peak={_safe_float(row.get('statement_rate_peak', 0)):.3f}/s "
                f"affected_rows_rate_peak={_safe_float(row.get('affected_rows_rate_peak', 0)):.3f}/s"
            )
    else:
        lines.append("  unavailable or empty")

    resource_control = compact_resource_control_for_ai(data, cluster_id, start, end)
    lines.append("RESOURCE_CONTROL:")
    request_p90 = resource_control.get("request_p90", [])
    available_ru = resource_control.get("available_ru", [])
    rcu_usage = resource_control.get("rcu_usage", [])
    if request_p90:
        top = request_p90[0]
        lines.append(
            f"  request_p90_peak={_safe_float(top.get('peak', 0)):.3f}s "
            f"instance={top.get('instance', 'unknown')} resource_group={top.get('resource_group', '')}"
        )
    if available_ru:
        low = available_ru[0]
        lines.append(
            f"  available_ru_min={_safe_float(low.get('min', 0)):.3f} "
            f"name={low.get('name', 'unknown')}"
        )
    if rcu_usage:
        top = rcu_usage[0]
        lines.append(
            f"  rcu_usage_peak={_safe_float(top.get('peak', 0)):.3f} RU/s "
            f"name={top.get('name', 'unknown')}"
        )
    if not (request_p90 or available_ru or rcu_usage):
        lines.append("  unavailable or empty")

    # A2 continued: Component health
    lines.append("\nComponent_Health:")
    comp_count = data.get("component_count", [])
    if comp_count:
        lines.append("  Component count (active instances):")
        for r in comp_count:
            comp = r.get("metric", {}).get("component", "unknown")
            vals = r.get("values", [])
            if vals:
                first_v = vals[0][1]
                last_v = vals[-1][1]
                changes = sum(1 for i in range(1, len(vals)) if vals[i][1] != vals[i-1][1])
                change_str = f" (CHANGED {changes}x during window)" if changes > 0 else ""
                lines.append(f"    {comp}: {first_v} -> {last_v}{change_str}")
    else:
        if data.get("parent_id"):
            lines.append("  Component count: no data (queried with parent_id, check if cluster has workers)")
        else:
            lines.append("  Component count: no data (for Premium clusters, re-run with --parent-id <resource_pool_id>)")

    restarts = data.get("restarts", [])
    if restarts:
        restarted = [(r, _prom_peak(r)) for r in restarts if _prom_peak(r) > 0]
        if restarted:
            lines.append("  Restarts detected (changes in process_start_time):")
            for r, count in sorted(restarted, key=lambda x: -x[1]):
                inst = _instance_name(r)
                comp = r.get("metric", {}).get("component", "")
                lines.append(f"    {inst} (component={comp}): {count:.0f} restart(s)")
        else:
            lines.append("  Restarts: none detected")
    else:
        lines.append("  Restarts: no data")

    worker_mem = data.get("worker_memory", [])
    if worker_mem:
        lines.append("  Worker memory (coprocessor-worker / tikv-worker):")
        for r in sorted(worker_mem, key=_prom_peak, reverse=True):
            inst = _instance_name(r)
            comp = r.get("metric", {}).get("component", "")
            peak_gb = _prom_peak(r) / (1024 ** 3)
            mean_gb = _prom_mean(r) / (1024 ** 3)
            vals = r.get("values", [])
            gaps = []
            for i in range(1, len(vals)):
                dt = float(vals[i][0]) - float(vals[i - 1][0])
                if dt > 120:
                    gap_start = datetime.fromtimestamp(float(vals[i - 1][0]), tz=timezone.utc).strftime("%H:%M")
                    gap_end = datetime.fromtimestamp(float(vals[i][0]), tz=timezone.utc).strftime("%H:%M")
                    gaps.append(f"{gap_start}-{gap_end}")
            gap_str = f" OFFLINE_GAPS=[{','.join(gaps)}]" if gaps else ""
            lines.append(f"    {inst} ({comp}): peak={peak_gb:.2f}GB mean={mean_gb:.2f}GB{gap_str}")

    # A3: Slow queries
    if data.get("is_byoc"):
        sample_limit = int(data.get("slow", {}).get("sample_limit", 100))
        lines.append(f"\n## A3. Slow Queries (latency-ranked sample, limit={sample_limit})")
    else:
        lines.append("\n## A3. Slow Queries (top by exec_count)")
    slow = data.get("slow", {})
    slow_cols = slow.get("columns", [])
    slow_rows = slow.get("rows", [])
    if slow_rows:
        for row in slow_rows[:10]:
            rec = dict(zip(slow_cols, row))
            sql_preview = str(rec.get("sample_sql", ""))[:120]
            if len(str(rec.get("sample_sql", ""))) > 120:
                sql_preview += "..."
            if data.get("is_byoc"):
                lines.append(
                    f"  digest={str(rec.get('digest', ''))} "
                    f"sample_count={rec.get('sample_count', 0)} "
                    f"sample_avg_time={_safe_float(rec.get('sample_avg_time', 0)):.2f}s "
                    f"max_time={_safe_float(rec.get('max_time', 0)):.2f}s "
                    f"max_mem={rec.get('max_mem', 0)} sql={sql_preview}"
                )
            else:
                lines.append(
                    f"  digest={str(rec.get('digest', ''))} "
                    f"exec_count={rec.get('exec_count', 0)} "
                    f"avg_time={_safe_float(rec.get('avg_time', 0)):.2f}s "
                    f"max_time={_safe_float(rec.get('max_time', 0)):.2f}s "
                    f"max_mem={rec.get('max_mem', 0)} sql={sql_preview}"
                )
    else:
        lines.append("  (no slow queries in window)")

    statement_summary = data.get("statement_summary", {})
    statement_cols = statement_summary.get("columns", [])
    statement_rows = statement_summary.get("rows", [])
    if data.get("is_byoc"):
        lines.append("\n## A3a. Statements Summary (full window, top by exec_count)")
        if statement_rows:
            for row in statement_rows[:10]:
                rec = dict(zip(statement_cols, row))
                lines.append(
                    f"  digest={str(rec.get('digest', ''))} "
                    f"exec_count={rec.get('exec_count', 0)} "
                    f"avg_time={_safe_float(rec.get('avg_time', 0)):.2f}s "
                    f"max_time={_safe_float(rec.get('max_time', 0)):.2f}s "
                    f"schema={rec.get('schema_name', '')}"
                )
        else:
            lines.append("  (no statement summary data in window)")

    # A3b: decoded_plan
    top_plan = data.get("top_plan", {})
    plan_cols = top_plan.get("columns", [])
    plan_rows = top_plan.get("rows", [])
    if plan_rows:
        rec = dict(zip(plan_cols, plan_rows[0]))
        plan_text = str(rec.get("decoded_plan", ""))
        plan_source = data.get("top_plan_source", "slow_query")
        plan_digest = data.get("top_plan_digest", rec.get("digest", "unknown"))
        full_sql = str(rec.get("full_sql", ""))
        lines.append(
            f"\n## A3b. decoded_plan (source={plan_source}, digest={plan_digest})\n"
            f"full_sql: {full_sql}\n"
            f"stats_info: {rec.get('stats_info', '')}\n"
            f"backoff: {rec.get('backoff', '')}\n"
            f"indexes_used: {rec.get('indexes_used', '')}\n"
            f"query_time={rec.get('query_time', '')} process_time={rec.get('process_time', '')} "
            f"wait_time={rec.get('wait_time', '')} total_keys={rec.get('total_keys', '')} "
            f"process_keys={rec.get('process_keys', '')}\n"
            f"{plan_text}"
        )

    # A4: TopSQL
    lines.append("\n## A4. TopSQL")

    lines.append("TopSQL TiDB (by CPU):")
    tidb_ts = data.get("topsql_tidb", {})
    ts_cols = tidb_ts.get("columns", [])
    ts_rows = tidb_ts.get("rows", [])
    if ts_rows:
        for row in ts_rows[:10]:
            rec = dict(zip(ts_cols, row))
            sql_preview = str(rec.get("sample_sql", ""))[:100]
            if len(str(rec.get("sample_sql", ""))) > 100:
                sql_preview += "..."
            lines.append(
                f"  digest={str(rec.get('sql_digest', ''))} "
                f"cpu_ms={rec.get('total_cpu_ms', 0)} "
                f"cpu_pct={_safe_float(rec.get('cpu_pct', 0)):.1f}% sql={sql_preview}"
            )
    else:
        lines.append("  (empty or unavailable)")

    lines.append("TopSQL TiKV (by CPU):")
    tikv_ts = data.get("topsql_tikv", {})
    ts_cols = tikv_ts.get("columns", [])
    ts_rows = tikv_ts.get("rows", [])
    if ts_rows:
        total_tikv_cpu = sum(_safe_int(dict(zip(ts_cols, r)).get("total_cpu_ms", 0)) for r in ts_rows)
        for row in ts_rows[:10]:
            rec = dict(zip(ts_cols, row))
            sql_preview = str(rec.get("sample_sql", ""))[:100]
            if len(str(rec.get("sample_sql", ""))) > 100:
                sql_preview += "..."
            cpu_ms = _safe_int(rec.get("total_cpu_ms", 0))
            cpu_pct = round(cpu_ms * 100.0 / total_tikv_cpu, 2) if total_tikv_cpu > 0 else 0
            inst = rec.get("instance", "")
            inst_str = f" instance={inst}" if inst else ""
            lines.append(
                f"  digest={str(rec.get('sql_digest', ''))} "
                f"cpu_ms={cpu_ms} cpu_pct={cpu_pct:.1f}%{inst_str} sql={sql_preview}"
            )
    else:
        lines.append("  (empty or unavailable)")

    lines.append("TopSQL TiKV hot instances (instance-scoped, for attribution):")
    hot_groups = data.get("topsql_tikv_hot_instances", [])
    rendered_hot = 0
    for group in hot_groups:
        inst_label = str(group.get("instance", ""))
        hot_cols = group.get("columns", [])
        hot_rows = group.get("rows", [])
        if not hot_rows:
            lines.append(f"  instance={inst_label} (empty or unavailable)")
            continue
        lines.append(f"  instance={inst_label}")
        for row in hot_rows[:5]:
            rec = dict(zip(hot_cols, row))
            sql_preview = str(rec.get("sample_sql", ""))[:100]
            if len(str(rec.get("sample_sql", ""))) > 100:
                sql_preview += "..."
            lines.append(
                f"    digest={str(rec.get('sql_digest', ''))} "
                f"cpu_ms={_safe_int(rec.get('total_cpu_ms', 0))} "
                f"cpu_pct={_safe_float(rec.get('cpu_pct', 0)):.1f}% "
                f"row_instance={rec.get('instance', '')} sql={sql_preview}"
            )
            rendered_hot += 1
    if not hot_groups or rendered_hot == 0:
        lines.append("  (no hot-instance TopSQL data)")

    # Collection errors
    if data.get("errors"):
        lines.append("\n## Collection Errors (non-fatal)")
        for e in data["errors"]:
            lines.append(f"  - {e}")

    # A6: Aggregation granularity depends on the source.
    log_stats = data.get("log_stats", {})
    summary = log_stats.get("summary", {}) if isinstance(log_stats, dict) else {}
    lines.append(f"\n## A6. Log Statistics ({log_stats.get('step_seconds', 300)}-second buckets)")
    if log_stats.get("count_semantics"):
        lines.append(f"  {log_stats['count_semantics']}")
    if summary:
        for comp in ("tidb", "tikv", "pd"):
            comp_summary = summary.get(comp, {})
            if not comp_summary:
                continue
            error_total = comp_summary.get("error", 0)
            warn_total = comp_summary.get("warn", 0)
            info_total = comp_summary.get("info", 0)
            lines.append(
                f"  {comp.upper()}: error={error_total} warn={warn_total} info={info_total}"
            )
    else:
        lines.append("  (no log statistics available)")
    loki_url = log_stats.get("loki_url", "") if isinstance(log_stats, dict) else ""
    if loki_url:
        lines.append(f"  Loki Explorer: {loki_url}")

    # Gate check results
    lines.append(f"\n## Gate Check Results")
    lines.append(f"- Infra: {infra_verdict}")
    lines.append(f"- Hotspot: {hotspot_verdict}")

    # Quick links
    logs_query = '{component=~"tidb|tikv|pd"} |~ "(?i)(ERROR|FATAL|WARN)"'
    cluster_meta = data.get("cluster_meta")
    lines.append(f"\n## Quick Links")
    lines.append(f"- Clinic Overview: {_clinic_metrics_url(cluster_id, 'process_cpu_seconds_total', start, end, cluster_meta)}")
    lines.append(f"- Slow Queries: {_slow_query_url(cluster_id, start, end, cluster_meta)}")
    lines.append(f"- Logs: {_loki_log_url(cluster_id, logs_query, start, end, cluster_meta)}")
    lines.append(f"- TopSQL: {_topsql_url(cluster_id, start, end, cluster_meta)}")

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _safe_int(v) -> int:
    try:
        return int(float(str(v)))
    except (ValueError, TypeError):
        return 0


def _safe_float(v) -> float:
    try:
        return float(str(v))
    except (ValueError, TypeError):
        return 0.0


def _workspace_root() -> str:
    script_dir = os.path.abspath(os.path.dirname(__file__))
    cur = script_dir
    while True:
        if os.path.basename(cur).startswith("ws-"):
            return cur
        parent = os.path.dirname(cur)
        if parent == cur:
            break
        cur = parent
    env_root = os.environ.get("INVESTIBOOK_WORKSPACE_DIR", "").strip()
    if env_root:
        return os.path.abspath(os.path.expanduser(env_root))
    cwd = os.path.abspath(os.getcwd())
    if os.path.basename(cwd).startswith("ws-"):
        return cwd
    raise RuntimeError("INVESTIBOOK_WORKSPACE_DIR is required for diagnosis evidence collection")


def _evidence_base_dir() -> str:
    return os.path.join(_workspace_root(), _EVIDENCES_DIRNAME)


def _load_json_file(path: str) -> dict | None:
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except (OSError, ValueError, json.JSONDecodeError):
        return None
    return data if isinstance(data, dict) else None


def _looks_like_run_dir(path: str) -> bool:
    return (
        os.path.isdir(os.path.join(path, "summary"))
        or os.path.isdir(os.path.join(path, "evidence"))
        or os.path.basename(path) == _EVIDENCES_DIRNAME
    )


def _build_run_output_dir(base_output_dir: str, cluster_id: str, start: str, end: str) -> str:
    """Use the stable workspace evidence root."""
    if _looks_like_run_dir(base_output_dir):
        return base_output_dir
    return base_output_dir


def _stable_json_key(value) -> str:
    return json.dumps(value, sort_keys=True, ensure_ascii=False, default=str)


def _merge_list(existing, incoming) -> list:
    items = []
    if isinstance(existing, list):
        items.extend(existing)
    if isinstance(incoming, list):
        items.extend(incoming)
    seen = set()
    merged = []
    for item in items:
        key = _stable_json_key(item)
        if key in seen:
            continue
        seen.add(key)
        merged.append(item)
    return merged


def _merge_dict(existing, incoming) -> dict:
    base = existing if isinstance(existing, dict) else {}
    extra = incoming if isinstance(incoming, dict) else {}
    result = dict(base)
    for key, value in extra.items():
        current = result.get(key)
        if isinstance(current, dict) and isinstance(value, dict):
            result[key] = _merge_dict(current, value)
        elif isinstance(current, list) and isinstance(value, list):
            result[key] = _merge_list(current, value)
        else:
            result[key] = value
    return result


def _write_merged_json(path: str, incoming: dict) -> None:
    existing = _load_json_file(path) or {}
    _write_json(path, _merge_dict(existing, incoming))


# ---------------------------------------------------------------------------
# Evidence file output
# ---------------------------------------------------------------------------

def _build_evidence_data(data: dict, cluster_id: str, start: str, end: str) -> tuple:
    """Build structured evidence from raw collected data.

    Returns (metrics, queries, plan_info, topsql, logs_evidence) where metrics
    still contain chart_data for raw storage. Use compact_metrics_for_ai() to
    strip chart_data for agent-readable files.
    """
    # --- CPU quota lookup ---
    cpu_quota: dict[str, float] = {}
    for r in data.get("cpu_quota", []):
        inst = _instance_name(r)
        median = _prom_median(r)
        if median > 0:
            cpu_quota[inst] = median

    # --- Metrics evidence ---
    metrics: list[dict] = []
    ebs_promql = "rate(aws_ebs_csi_exceeded_iops_seconds_total[1m])  # client-side join by pod"
    disk_promql = 'rate(node_disk_io_time_seconds_total{device=~"nvme.*"}[1m])'
    latency_promql = "histogram_quantile(0.99, sum(rate(tidb_server_handle_query_duration_seconds_bucket[1m])) by (le, instance))"
    grpc_promql = 'histogram_quantile(0.99, sum(rate(tikv_grpc_msg_duration_seconds_bucket{type!="raft"}[1m])) by (le, instance))'
    gc_frequency_promql = 'rate(go_gc_duration_seconds_count{component="tidb"}[1m])'
    gc_duration_promql = 'rate(go_gc_duration_seconds_sum{component="tidb"}[1m])'
    gc_pause_promql = 'histogram_quantile(0.99, sum(rate(go_gc_pauses_seconds_bucket{component="tidb"}[1m])) by (le, instance))'

    def _observed_with_coverage(results, instance_fn=None):
        observed = [r for r in results if _prom_observed_peak(r) is not None]
        coverage = {}
        if len(observed) != len(results):
            coverage = {
                "availability": "partial" if observed else "missing",
                "observed_series": len(observed),
                "total_series": len(results),
                "missing_instances": [
                    (instance_fn or _instance_name)(r) for r in results
                    if _prom_observed_peak(r) is None
                ],
            }
        return observed, coverage

    def _avg_and_top(name: str, results: list[dict], promql: str, top_n: int,
                     unit: str, precision: int = 3,
                     severity_fn=None, extra_fn=None,
                     instance_fn=None, all_results=None,
                     metric_group_id: str | None = None):
        """Emit one avg summary row then top-N rows for a metric group."""
        if not results:
            return
        observed, coverage = _observed_with_coverage(results, instance_fn)
        peaks = [_prom_peak(r) for r in observed]
        means = [_prom_mean(r) for r in observed]
        avg_row = {
            "name": f"{name} Avg",
            "instance": f"all ({len(results)} instances)",
            "peak": round(sum(peaks) / len(peaks), precision) if peaks else None,
            "mean": round(sum(means) / len(means), precision) if means else None,
            "unit": unit,
            "promql": promql,
            "link": _clinic_metrics_url(cluster_id, promql, start, end),
            **coverage,
        }
        if not observed:
            avg_row["severity"] = "unknown"
        if metric_group_id:
            avg_row["metric_group_id"] = metric_group_id
        metrics.append(avg_row)
        for r in sorted(results, key=lambda r: (_prom_observed_peak(r) is not None, _prom_peak(r)), reverse=True)[:top_n]:
            inst = instance_fn(r) if instance_fn else _instance_name(r)
            peak = _prom_observed_peak(r)
            row = {
                "name": name,
                "instance": inst,
                "peak": round(peak, precision) if peak is not None else None,
                "mean": round(_prom_mean(r), precision) if peak is not None else None,
                "unit": unit,
                "promql": promql,
                "chart_data": _chart_data(name, promql, all_results or results, instance_filter=inst, instance_fn=instance_fn),
                "link": _clinic_metrics_url(cluster_id, promql, start, end),
            }
            if metric_group_id:
                row["metric_group_id"] = metric_group_id
            if peak is None:
                row.update(availability="missing", severity="unknown")
            elif severity_fn:
                row["severity"] = severity_fn(r)
            if extra_fn and peak is not None:
                row.update(extra_fn(r))
            metrics.append(row)

    # A1: Infra
    def _ebs_instance(r):
        m = r.get("metric", {})
        vol = m.get("volume_id", "?")
        pod = m.get("pod", "")
        return f"{pod} {vol}" if pod else vol

    def _ebs_severity(r):
        p = _prom_peak(r)
        return "critical" if p > 0.20 else ("warning" if p > 0.05 else "normal")

    _avg_and_top("EBS IOPS Throttle", data.get("ebs", []), ebs_promql, 5,
                 unit="fraction", precision=4, severity_fn=_ebs_severity,
                 instance_fn=_ebs_instance, metric_group_id="ebs_iops_throttle")

    def _disk_severity(r):
        return "critical" if _prom_peak(r) > 0.8 else "normal"

    def _disk_instance(r):
        m = r.get("metric", {})
        return f"{m.get('instance', '?')} {m.get('device', '?')}"

    _avg_and_top("Disk IO Utilization", data.get("disk", []), disk_promql, 5,
                 unit="fraction", precision=4, severity_fn=_disk_severity,
                 instance_fn=_disk_instance, metric_group_id="disk_io_utilization")

    # A2: App metrics
    _avg_and_top("TiDB P99 Latency", data.get("latency", []), latency_promql, 5,
                 unit="seconds", metric_group_id="tidb_p99_latency")
    _avg_and_top("TiKV gRPC P99", data.get("grpc", []), grpc_promql, 3,
                 unit="seconds", metric_group_id="tikv_grpc_p99")

    def _instance_with_labels(*label_names: str):
        def _format(result: dict) -> str:
            metric = result.get("metric", {})
            base = str(metric.get("instance", "?"))
            labels = [f"{name}={metric[name]}" for name in label_names if metric.get(name) not in (None, "")]
            return f"{base} ({', '.join(labels)})" if labels else base
        return _format

    # Database-time routing and read/write path decomposition. These metrics are
    # already documented by the builtin clinic-cli metric schema; keeping them
    # in the broad pass lets the analyst select a path before targeted digging.
    _avg_and_top("TiDB Connection Idle P99", data.get("connection_idle_p99", []),
                 CONNECTION_IDLE_P99_PROMQL, 6, unit="seconds",
                 instance_fn=_instance_with_labels("in_txn"),
                 metric_group_id="tidb_connection_idle_p99")
    _avg_and_top("TiDB Parse P99", data.get("parse_p99", []), PARSE_P99_PROMQL, 6,
                 unit="seconds", instance_fn=_instance_with_labels("sql_type"),
                 metric_group_id="tidb_parse_p99")
    _avg_and_top("TiDB Compile P99", data.get("compile_p99", []), COMPILE_P99_PROMQL, 6,
                 unit="seconds", instance_fn=_instance_with_labels("sql_type"),
                 metric_group_id="tidb_compile_p99")
    _avg_and_top("TiDB KV Request P99", data.get("kv_request_p99", []),
                 KV_REQUEST_P99_PROMQL, 10, unit="seconds",
                 instance_fn=_instance_with_labels("type", "store"),
                 metric_group_id="tidb_kv_request_p99")
    _avg_and_top("TiDB KV Backoff P99", data.get("kv_backoff_p99", []),
                 KV_BACKOFF_P99_PROMQL, 10, unit="seconds",
                 instance_fn=_instance_with_labels("type"),
                 metric_group_id="tidb_kv_backoff_p99")
    _avg_and_top("PD TSO Wait P99", data.get("tso_wait_p99", []), TSO_WAIT_P99_PROMQL, 5,
                 unit="seconds", metric_group_id="pd_tso_wait_p99")
    _avg_and_top("TiKV gRPC P99 By Type", data.get("grpc_by_type_p99", []),
                 TIKV_GRPC_BY_TYPE_P99_PROMQL, 10, unit="seconds",
                 instance_fn=_instance_with_labels("type"),
                 metric_group_id="tikv_grpc_by_type_p99")
    _avg_and_top("TiKV Scheduler Command P99", data.get("scheduler_command_p99", []),
                 TIKV_SCHEDULER_COMMAND_P99_PROMQL, 10, unit="seconds",
                 instance_fn=_instance_with_labels("type"),
                 metric_group_id="tikv_scheduler_command_p99")
    _avg_and_top("TiKV Scheduler Latch Wait P99", data.get("scheduler_latch_p99", []),
                 TIKV_SCHEDULER_LATCH_P99_PROMQL, 10, unit="seconds",
                 instance_fn=_instance_with_labels("type"),
                 metric_group_id="tikv_scheduler_latch_wait_p99")
    _avg_and_top("TiKV Storage Async Request P99", data.get("storage_async_p99", []),
                 TIKV_STORAGE_ASYNC_P99_PROMQL, 10, unit="seconds",
                 instance_fn=_instance_with_labels("type"),
                 metric_group_id="tikv_storage_async_request_p99")
    _avg_and_top("TiKV Raft Commit Log P99", data.get("raft_commit_p99", []),
                 TIKV_RAFT_COMMIT_P99_PROMQL, 5, unit="seconds",
                 metric_group_id="tikv_raft_commit_log_p99")
    _avg_and_top("TiKV Raft Apply Log P99", data.get("raft_apply_p99", []),
                 TIKV_RAFT_APPLY_P99_PROMQL, 5, unit="seconds",
                 metric_group_id="tikv_raft_apply_log_p99")
    _avg_and_top("TiKV Pending Compaction Bytes", data.get("pending_compaction_bytes", []),
                 TIKV_PENDING_COMPACTION_PROMQL, 5, unit="bytes",
                 instance_fn=_instance_with_labels("db"),
                 metric_group_id="tikv_pending_compaction_bytes")
    _avg_and_top("TiDB QPS", data.get("tidb_qps", []), TIDB_QPS_PROMQL, 10,
                 unit="requests_per_second",
                 instance_fn=_instance_with_labels("result", "type"),
                 metric_group_id="tidb_qps")
    _avg_and_top("Component Resident Memory", data.get("component_memory", []),
                 COMPONENT_MEMORY_PROMQL, 10, unit="bytes",
                 instance_fn=_instance_with_labels("component"),
                 metric_group_id="component_resident_memory")

    # CPU: group by component + nodegroup, with component-specific PromQL.
    cpu_groups: dict[tuple[str, str], list[dict]] = {}
    for r in data.get("cpu", []):
        m = r.get("metric", {})
        comp = m.get("component", "unknown")
        ng = m.get("cluster", "db")
        cpu_groups.setdefault((comp, ng), []).append(r)

    cpu_top_n = {"tidb": 5, "tikv": 5}
    cpu_promql_by_component = {"tidb": TIDB_CPU_PROMQL, "tikv": TIKV_CPU_PROMQL}
    cpu_results_by_component = {
        "tidb": data.get("tidb_cpu") or [r for r in data.get("cpu", []) if r.get("metric", {}).get("component") == "tidb"],
        "tikv": data.get("tikv_cpu") or [r for r in data.get("cpu", []) if r.get("metric", {}).get("component") == "tikv"],
    }
    for (comp, ng), group in sorted(cpu_groups.items()):
        label = f"{comp.upper()} CPU"
        promql = cpu_promql_by_component.get(comp, f'rate(process_cpu_seconds_total{{component="{comp}"}}[1m])')
        metric_group_id = f"{comp}_cpu:{ng}"
        observed, coverage = _observed_with_coverage(group)
        peaks = [_prom_peak(r) for r in observed]
        means = [_prom_mean(r) for r in observed]
        metrics.append({
            "name": f"{label} Avg",
            "component": comp,
            "metric_group_id": metric_group_id,
            "instance": f"all ({ng}, {len(group)} instances)",
            "nodegroup": ng,
            "peak": round(sum(peaks) / len(peaks), 2) if peaks else None,
            "mean": round(sum(means) / len(means), 2) if means else None,
            "unit": "cores",
            "promql": promql,
            "link": _clinic_metrics_url(cluster_id, promql, start, end),
            **coverage,
        })
        top_n = cpu_top_n.get(comp, 5)
        for r in sorted(group, key=lambda r: (_prom_observed_peak(r) is not None, _prom_peak(r)), reverse=True)[:top_n]:
            inst = _instance_name(r)
            peak = _prom_observed_peak(r)
            q = cpu_quota.get(inst)
            util = peak / q if peak is not None and q and q > 0 else None
            metrics.append({
                "name": label,
                "component": comp,
                "metric_group_id": metric_group_id,
                "instance": inst,
                "nodegroup": ng,
                "peak": round(peak, 2) if peak is not None else None,
                "quota": round(q, 0) if q else None,
                "utilization": f"{util:.0%}" if util is not None else None,
                "unit": "cores",
                "severity": "unknown" if peak is None else ("critical" if (util or 0) >= 0.8 else ("warning" if (util or 0) >= 0.6 else "normal")),
                **({"availability": "missing"} if peak is None else {}),
                "promql": promql,
                "chart_data": _chart_data(label, promql, cpu_results_by_component.get(comp, group), instance_filter=inst),
                "link": _clinic_metrics_url(cluster_id, promql, start, end),
            })

    # GC metrics
    def _gc_freq_severity(r):
        return "warning" if _prom_peak(r) > 10 else "normal"

    def _gc_dur_severity(r):
        return "warning" if _prom_peak(r) > 0.1 else "normal"

    _avg_and_top("TiDB GC Frequency", data.get("tidb_gc_frequency", []),
                 gc_frequency_promql, 5, unit="events/s", severity_fn=_gc_freq_severity,
                 metric_group_id="tidb_gc_frequency")
    _avg_and_top("TiDB GC Duration Rate", data.get("tidb_gc_duration_rate", []),
                 gc_duration_promql, 5, unit="seconds/second", severity_fn=_gc_dur_severity,
                 metric_group_id="tidb_gc_duration_rate")
    _avg_and_top("TiDB GC Pause P99", data.get("tidb_gc_pause_p99", []),
                 gc_pause_promql, 5, unit="seconds", severity_fn=_gc_dur_severity,
                 metric_group_id="tidb_gc_pause_p99")

    # TiKV thread CPU buckets → metrics (compact summary only; full threads stay in evidence/)
    thread_cpu = data.get("tikv_thread_cpu", {})
    if isinstance(thread_cpu, dict) and thread_cpu.get("status") == "collected":
        thread_cpu_promql = thread_cpu.get("promql", "")
        for b in thread_cpu.get("buckets", []):
            metrics.append({
                "name": "TiKV Thread CPU",
                "instance": b.get("instance", ""),
                "bucket": b.get("bucket", ""),
                "peak": b.get("peak", 0),
                "unit": "cores",
                "promql": thread_cpu_promql,
            })

    _ensure_metric_group_ids(metrics)
    _attach_metric_chart_refs(metrics, cluster_id, start, end)

    # --- Queries evidence ---
    queries: list[dict] = []
    slow = data.get("slow", {})
    slow_cols = slow.get("columns", [])
    for row in slow.get("rows", [])[:10]:
        rec = dict(zip(slow_cols, row))
        query_entry = {
            "source": "slow_query",
            "digest": str(rec.get("digest", "")),
            "sql_preview": str(rec.get("sample_sql", ""))[:200],
            "max_time": f"{_safe_float(rec.get('max_time', 0)):.2f}s",
            "max_mem": rec.get("max_mem", 0),
            "link": _slow_query_url(cluster_id, start, end, data.get("cluster_meta"),
                                    str(rec.get("digest", ""))),
        }
        if slow.get("scope") == "latency_ranked_sample":
            query_entry.update({
                "scope": "latency_ranked_sample",
                "ranking": slow.get("ranking", "query_time_desc"),
                "sample_limit": slow.get("sample_limit", 100),
                "sample_count": rec.get("sample_count", 0),
                "sample_avg_time": f"{_safe_float(rec.get('sample_avg_time', 0)):.2f}s",
            })
        else:
            query_entry.update({
                "exec_count": rec.get("exec_count", 0),
                "avg_time": f"{_safe_float(rec.get('avg_time', 0)):.2f}s",
            })
        queries.append(query_entry)

    # A3b: decoded plan
    top_plan = data.get("top_plan", {})
    plan_cols = top_plan.get("columns", [])
    plan_rows = top_plan.get("rows", [])
    plan_info: dict = {}
    if plan_rows:
        rec = dict(zip(plan_cols, plan_rows[0]))
        plan_info = {
            "digest": data.get("top_plan_digest", ""),
            "source": data.get("top_plan_source", ""),
            "full_sql": str(rec.get("full_sql", "")),
            "decoded_plan": str(rec.get("decoded_plan", "")),
            "stats_info": str(rec.get("stats_info", "")),
            "backoff": str(rec.get("backoff", "")),
            "indexes_used": str(rec.get("indexes_used", "")),
            "query_time": rec.get("query_time", ""),
            "process_time": rec.get("process_time", ""),
            "wait_time": rec.get("wait_time", ""),
            "total_keys": rec.get("total_keys", ""),
            "process_keys": rec.get("process_keys", ""),
        }

    # TopSQL
    topsql: list[dict] = []
    for label, key in [("TiDB", "topsql_tidb"), ("TiKV", "topsql_tikv")]:
        ts = data.get(key, {})
        ts_cols = ts.get("columns", [])
        for row in ts.get("rows", [])[:10]:
            rec = dict(zip(ts_cols, row))
            topsql.append({
                "component": label,
                "digest": str(rec.get("sql_digest", "")),
                "sql_preview": str(rec.get("sample_sql", ""))[:200],
                "cpu_ms": rec.get("total_cpu_ms", 0),
                "cpu_pct": _safe_float(rec.get("cpu_pct", 0)),
                "instance": rec.get("instance", ""),
                "link": _topsql_url(cluster_id, start, end, data.get("cluster_meta"),
                                     str(rec.get("instance", ""))),
            })

    for group in data.get("topsql_tikv_hot_instances", []):
        ts_cols = group.get("columns", [])
        for row in group.get("rows", [])[:10]:
            rec = dict(zip(ts_cols, row))
            row_instance = str(rec.get("instance", "")) or str(group.get("instance", ""))
            topsql.append({
                "component": "TiKV hot instance",
                "digest": str(rec.get("sql_digest", "")),
                "sql_preview": str(rec.get("sample_sql", ""))[:200],
                "cpu_ms": rec.get("total_cpu_ms", 0),
                "cpu_pct": _safe_float(rec.get("cpu_pct", 0)),
                "instance": row_instance,
                "link": _topsql_url(cluster_id, start, end, data.get("cluster_meta"), row_instance),
            })

    # --- Log stats evidence ---
    log_stats_data = data.get("log_stats", {})
    log_stats_evidence = log_stats_data if isinstance(log_stats_data, dict) else {}

    return metrics, queries, plan_info, topsql, log_stats_evidence


def build_agent_summary(data: dict, infra_verdict: str, hotspot_verdict: str,
                        cluster_id: str, start: str, end: str,
                        run_dir: str = "") -> dict:
    """Build the single canonical agent summary object.

    This is the source of truth for evidence_summary_for_agent.json,
    workspace_evidence_report.json, and all categorized evidence files.
    """
    metrics, queries, plan_info, topsql, log_stats_evidence = _build_evidence_data(
        data, cluster_id, start, end,
    )

    cluster_meta = dict(data.get("cluster_meta") or {})
    cluster_meta.setdefault("cluster_id", cluster_id)
    time_range = {
        "start": start,
        "end": end,
        "start_iso": _format_utc_iso(int(start)),
        "end_iso": _format_utc_iso(int(end)),
    }
    input_time_range = data.get("_input_time_range")
    if isinstance(input_time_range, dict):
        if input_time_range.get("input_start"):
            time_range["input_start"] = str(input_time_range["input_start"])
        if input_time_range.get("input_end"):
            time_range["input_end"] = str(input_time_range["input_end"])
    links = {
        "clinic_overview": _clinic_metrics_url(
            cluster_id, "process_cpu_seconds_total", start, end, cluster_meta,
        ),
        "slow_queries": _slow_query_url(cluster_id, start, end, cluster_meta),
        "logs": _loki_log_url(
            cluster_id,
            '{component=~"tidb|tikv|pd"} |~ "(?i)(ERROR|FATAL|WARN)"',
            start,
            end,
            cluster_meta,
        ),
        "topsql": _topsql_url(cluster_id, start, end, cluster_meta),
        "topsql_template": _topsql_url_template(cluster_id, start, end, cluster_meta),
    }

    gate_checks = {
        "infra": infra_verdict,
        "hotspot": hotspot_verdict,
    }
    query_summary = {
        "slow_queries": queries,
        "topsql": topsql,
        "decoded_plan": plan_info,
        "statement_summary": [
            {"source": "statement_summary", **dict(zip(
                data.get("statement_summary", {}).get("columns", []), row,
            ))}
            for row in data.get("statement_summary", {}).get("rows", [])[:20]
        ],
        "statements": [
            {"source": "statement", **entry}
            for entry in data.get("statements", [])
            if isinstance(entry, dict)
        ],
    }
    panic_context = _panic_classification_context(log_stats_evidence, gate_checks)
    if panic_context:
        query_summary["panic_classification_context"] = panic_context

    summary = {
        "metadata": {
            "cluster_id": cluster_id,
            "cluster": cluster_meta or {},
            "collection_profile": data.get("collection_profile", "full"),
            "time_range": time_range,
            "collected_at": datetime.now(tz=timezone.utc).isoformat(),
            "run_dir": run_dir,
        },
        "metrics": compact_metrics_for_ai(metrics),
        "workload": compact_workload_for_ai(data),
        "resource_control": compact_resource_control_for_ai(data, cluster_id, start, end),
        "component_health": compact_component_health_for_ai(data, cluster_id, start, end),
        "events": compact_events_for_ai(data),
        "queries": query_summary,
        "profiles": data.get("profiles", []),
        "logs": log_stats_evidence,
        "gate_checks": gate_checks,
        "links": links,
        "errors": data.get("errors", []),
        # Keep raw metrics for categorized file output (not written to agent summary file).
        "_raw_metrics": metrics,
    }
    summary["critical_evidence_hints"] = build_critical_evidence_hints(summary, cluster_meta)
    return summary


def _format_agent_text(summary: dict) -> str:
    """Render agent summary as compact, section-indexed plain text.

    Designed to minimise token usage while remaining fully readable by LLMs.
    Tables use TSV; metrics are grouped by promql/link; decoded_plan uses an
    indented text tree.
    """
    lines: list[str] = []
    section_starts: dict[str, int] = {}

    def _line(text: str = "") -> None:
        lines.append(text)

    def _section(name: str) -> None:
        _line("")
        section_starts[name] = len(lines) + 1  # 1-based line numbers
        _line(f"--- {name} ---")

    # --- header ---
    meta = summary.get("metadata", {})
    tr = meta.get("time_range", {})
    _line(f"# Evidence — Cluster {meta.get('cluster_id', '?')}")
    start_iso = tr.get("start_iso")
    end_iso = tr.get("end_iso")
    if start_iso and end_iso:
        _line(
            f"# Time: {start_iso} – {end_iso} "
            f"(epoch {tr.get('start', '?')} – {tr.get('end', '?')})"
        )
    else:
        _line(f"# Time: {tr.get('start', '?')} – {tr.get('end', '?')}")
    _line(f"# Collected: {meta.get('collected_at', '?')}")

    # --- gate checks (key findings) ---
    gc = summary.get("gate_checks", {})
    _line("")
    _line("## Key Findings")
    _line(f"- Infra: {gc.get('infra', 'N/A')}")
    _line(f"- Hotspot: {gc.get('hotspot', 'N/A')}")
    hints = summary.get("critical_evidence_hints", [])
    if hints:
        _line("- Critical evidence hints:")
        for hint in hints:
            _line(f"  - {hint.get('id', '')}: {hint.get('summary', '')}")

    # --- placeholder for section index (filled in at the end) ---
    _line("")
    index_line = len(lines)  # we'll insert the index here
    _line("## Section Index")  # placeholder

    # --- metrics (grouped by promql+link) ---
    _section("metrics")
    metrics = summary.get("metrics", [])
    from collections import OrderedDict
    groups: OrderedDict[tuple[str, str, str, str], list[dict]] = OrderedDict()
    for m in metrics:
        key = (
            m.get("metric_group_id", ""),
            m.get("component", ""),
            m.get("promql", ""),
            m.get("link", ""),
        )
        groups.setdefault(key, []).append(m)

    for (_metric_group_id, _component, promql, link), items in groups.items():
        group_name = items[0].get("name", "").replace(" Avg", "")
        unit = items[0].get("unit", "")
        _line(f"### {group_name} (unit: {unit})")
        if _metric_group_id:
            _line(f"metric_group_id: {_metric_group_id}")
        if promql:
            _line(f"promql: {promql}")
        if link:
            _line(f"link: {link}")
        # Determine columns present in this group
        has_severity = any(m.get("severity") for m in items)
        has_bucket = any(m.get("bucket") for m in items)
        has_nodegroup = any(m.get("nodegroup") for m in items)
        has_quota = any(m.get("quota") is not None for m in items)
        has_util = any(m.get("utilization") for m in items)
        cols = ["instance"]
        if has_nodegroup:
            cols.append("nodegroup")
        if has_bucket:
            cols.append("bucket")
        cols.extend(["peak", "mean"])
        if has_quota:
            cols.append("quota")
        if has_util:
            cols.append("utilization")
        if has_severity:
            cols.append("severity")
        for field in ("availability", "observed_series", "total_series", "missing_instances"):
            if any(field in m for m in items):
                cols.append(field)
        _line("\t".join(cols))
        for m in items:
            vals = [str(m.get(c, "")) for c in cols]
            _line("\t".join(vals))
        _line("")

    # --- workload ---
    _section("workload")
    workload = summary.get("workload", [])
    if workload:
        wk_cols = list(workload[0].keys())
        _line("\t".join(wk_cols))
        for row in workload:
            _line("\t".join(str(row.get(c, "")) for c in wk_cols))
    else:
        _line("(no data)")

    # --- resource_control ---
    _section("resource_control")
    rc = summary.get("resource_control", {})
    for sub_key in ("rcu_usage", "request_p90", "available_ru"):
        items = rc.get(sub_key, [])
        if items:
            _line(f"### {sub_key}")
            rc_cols = list(items[0].keys())
            _line("\t".join(rc_cols))
            for row in items:
                _line("\t".join(str(row.get(c, "")) for c in rc_cols))
    rc_links = rc.get("links", {})
    if rc_links:
        for k, v in rc_links.items():
            _line(f"link.{k}: {v}")

    # --- component_health ---
    _section("component_health")
    ch = summary.get("component_health", {})
    if ch.get("parent_id"):
        _line(f"query_scope: {ch.get('query_scope', '')} (parent_id: {ch['parent_id']})")
    cc = ch.get("component_count", [])
    if cc:
        _line("### component_count")
        _line("component\tfirst\tlast\tchanges")
        for row in cc:
            _line(f"{row.get('component','')}\t{row.get('first','')}\t{row.get('last','')}\t{row.get('changes','')}")
    restarts = ch.get("restarts", [])
    if restarts:
        _line("### restarts")
        _line("instance\tcomponent\trestart_count")
        for row in restarts:
            _line(f"{row.get('instance','')}\t{row.get('component','')}\t{row.get('restart_count','')}")
    wm = ch.get("worker_memory", [])
    if wm:
        _line("### worker_memory")
        _line("instance\tcomponent\tpeak_gb\tmean_gb")
        for row in wm:
            _line(f"{row.get('instance','')}\t{row.get('component','')}\t{row.get('peak_gb','')}\t{row.get('mean_gb','')}")
    ch_links = ch.get("links", {})
    if ch_links:
        for k, v in ch_links.items():
            _line(f"link.{k}: {v}")

    # --- slow_queries ---
    _section("slow_queries")
    queries = summary.get("queries", {})
    slow = queries.get("slow_queries", [])
    if slow:
        if any(row.get("scope") == "latency_ranked_sample" for row in slow):
            sq_cols = [
                "source", "scope", "ranking", "sample_limit", "digest", "sql_preview",
                "sample_count", "sample_avg_time", "max_time", "max_mem",
            ]
        else:
            sq_cols = ["source", "digest", "sql_preview", "exec_count", "avg_time", "max_time", "max_mem"]
        _line("\t".join(sq_cols))
        for row in slow:
            _line("\t".join(str(row.get(c, "")) for c in sq_cols))
        sq_link = slow[0].get("link", "")
        if sq_link:
            _line(f"link: {sq_link}")
    else:
        _line("(no slow queries)")

    statement_summary = queries.get("statement_summary", [])
    if statement_summary:
        _section("statement_summary")
        statement_cols = [
            "source", "digest", "sample_sql", "schema_name", "stmt_type",
            "exec_count", "avg_time", "max_time", "max_mem",
        ]
        _line("\t".join(statement_cols))
        for row in statement_summary:
            _line("\t".join(str(row.get(c, "")) for c in statement_cols))

    # --- topsql ---
    _section("topsql")
    topsql = queries.get("topsql", [])
    if topsql:
        link_template = summary.get("links", {}).get("topsql_template", "")
        if link_template:
            _line(f"link_template: {link_template}")
            _line("link_template_note: replace {urlencoded_instance} with the URL-encoded instance from the selected row")
        ts_cols = ["component", "digest", "sql_preview", "cpu_ms", "cpu_pct", "instance"]
        _line("\t".join(ts_cols))
        for row in topsql:
            _line("\t".join(str(row.get(c, "")) for c in ts_cols))
    else:
        _line("(no topsql data)")

    # --- decoded_plan (text tree) ---
    _section("decoded_plan")
    dp = queries.get("decoded_plan", {})
    if dp and dp.get("decoded_plan"):
        _line(f"digest: {dp.get('digest', '')}")
        _line(f"source: {dp.get('source', '')}")
        _line(f"full_sql: {dp.get('full_sql', '')}")
        _line(f"stats_info: {dp.get('stats_info', '')}")
        _line(f"indexes_used: {dp.get('indexes_used', '')}")
        _line(f"query_time={dp.get('query_time','')} process_time={dp.get('process_time','')} "
              f"wait_time={dp.get('wait_time','')} total_keys={dp.get('total_keys','')} "
              f"process_keys={dp.get('process_keys','')}")
        # Render plan as indented text tree
        plan_str = dp["decoded_plan"]
        try:
            plan_obj = json.loads(plan_str)
            main_tree = plan_obj.get("main") if isinstance(plan_obj, dict) else None
            if isinstance(main_tree, dict):
                _line("")
                _render_plan_tree(main_tree, 0, lines)
            else:
                # Not a tree structure; dump compact JSON
                _line(json.dumps(plan_obj, separators=(",", ":")))
        except (json.JSONDecodeError, TypeError):
            # Already plain text
            _line(plan_str)
    else:
        _line("(no decoded plan)")

    # --- logs ---
    _section("logs")
    logs = summary.get("logs", {})
    if logs.get("count_semantics"):
        _line(logs["count_semantics"])
    if logs.get("canonical_window"):
        _line(f"canonical_window: {json.dumps(logs['canonical_window'])}")
    log_summary = logs.get("summary", {}) if isinstance(logs, dict) else {}
    if log_summary:
        _line("component\terror\twarn\tinfo")
        for comp in ("tidb", "tikv", "pd"):
            cs = log_summary.get(comp, {})
            _line(f"{comp}\t{cs.get('error',0)}\t{cs.get('warn',0)}\t{cs.get('info',0)}")
    else:
        _line("(no log statistics)")
    if logs.get("loki_url"):
        _line(f"loki: {logs['loki_url']}")

    # --- links ---
    _section("links")
    for k, v in summary.get("links", {}).items():
        _line(f"{k}: {v}")

    # --- errors ---
    errs = summary.get("errors", [])
    if errs:
        _section("errors")
        for e in errs:
            _line(f"- {e}")

    # --- fill in section index ---
    index_entries = []
    line_offset = len(section_starts)
    for name, start_line in section_starts.items():
        index_entries.append(f"  [{name}] line {start_line + line_offset}")
    lines[index_line] = "## Section Index\n" + "\n".join(index_entries)

    return "\n".join(lines)


def _render_plan_tree(node: dict, depth: int, lines: list[str]) -> None:
    """Recursively render a TiDB execution plan node as an indented text line."""
    if not isinstance(node, dict):
        return
    indent = "  " * depth
    name = node.get("name", "?")
    parts = [f"{indent}{name}"]
    task = node.get("taskType", "")
    if task:
        parts.append(f"task={task}")
    store = node.get("storeType", "")
    if store:
        parts.append(f"store={store}")
    est = node.get("estRows", "")
    if est:
        parts.append(f"est={est}")
    act = node.get("actRows", "")
    if act:
        parts.append(f"act={act}")
    oi = node.get("operatorInfo", "")
    if oi:
        parts.append(f"op={oi}")
    lines.append("  ".join(parts))
    for child in node.get("children", []):
        _render_plan_tree(child, depth + 1, lines)


def save_evidence_files(data: dict, infra_verdict: str, hotspot_verdict: str,
                        cluster_id: str, start: str, end: str, output_dir: str) -> str:
    """Save evidence as JSON files under the workspace evidence root.

    Returns the run_dir path.

    Layout:
        <workspace>/evidences/evidence_summary_for_agent.json
        <workspace>/evidences/workspace_evidence_report.json
        <workspace>/evidences/cluster_meta.json
        <workspace>/evidences/metrics.json
        <workspace>/evidences/workload.json
        <workspace>/evidences/resource_control.json
        <workspace>/evidences/component_health.json
        <workspace>/evidences/queries.json
        <workspace>/evidences/logs.json
        <workspace>/evidences/tikv_thread_cpu.json
        <workspace>/evidences/links.json
        <workspace>/evidences/gate_checks.json
        <workspace>/evidences/collection_errors.json
    """
    run_dir = _build_run_output_dir(output_dir, cluster_id, start, end)
    os.makedirs(run_dir, exist_ok=True)

    agent_summary = build_agent_summary(
        data, infra_verdict, hotspot_verdict, cluster_id, start, end, run_dir,
    )

    # Extract raw data, then strip internal keys from the agent summary file.
    raw_metrics = agent_summary.pop("_raw_metrics", [])

    # --- Agent summary (compact text for agent reasoning) ---
    agent_summary_txt_path = os.path.join(run_dir, "evidence_summary_for_agent.txt")
    with open(agent_summary_txt_path, "w", encoding="utf-8") as f:
        f.write(_format_agent_text(agent_summary))

    # --- Agent summary JSON (consumed by backend for diagnosis report hydration) ---
    agent_summary_path = os.path.join(run_dir, "evidence_summary_for_agent.json")
    _write_json(agent_summary_path, agent_summary)

    # --- Workspace evidence report (written by Python, not by agent) ---
    workspace_evidence_report_path = os.path.join(run_dir, "workspace_evidence_report.json")
    _write_merged_json(workspace_evidence_report_path, {
        "type": "workspace_evidence_report",
        "metadata": agent_summary["metadata"],
        "evidence": {
            "metrics": agent_summary["metrics"],
            "queries": (agent_summary.get("queries", {}).get("slow_queries", [])
                        + agent_summary.get("queries", {}).get("topsql", [])
                        + agent_summary.get("queries", {}).get("statement_summary", [])
                        + agent_summary.get("queries", {}).get("statements", [])),
            "logs": agent_summary["logs"].get("summary", {}) if isinstance(agent_summary["logs"], dict) else {},
            "workload": agent_summary["workload"],
            "component_health": agent_summary["component_health"],
            "events": agent_summary["events"],
            "profiles": agent_summary.get("profiles", []),
        },
        "resource_control": agent_summary["resource_control"],
        "gate_checks": agent_summary["gate_checks"],
        "links": agent_summary["links"],
    })

    # --- Categorized workspace evidence files ---
    _write_merged_json(os.path.join(run_dir, "metrics.json"), {
        "metrics": raw_metrics,
    })
    _write_merged_json(os.path.join(run_dir, "cluster_meta.json"), {
        "cluster": agent_summary.get("metadata", {}).get("cluster", {}),
    })
    _write_merged_json(os.path.join(run_dir, "workload.json"), {
        "workload": agent_summary["workload"],
    })
    _write_merged_json(os.path.join(run_dir, "resource_control.json"), {
        "resource_control": agent_summary["resource_control"],
    })
    _write_merged_json(os.path.join(run_dir, "component_health.json"), {
        "component_health": agent_summary["component_health"],
    })
    _write_merged_json(os.path.join(run_dir, "events.json"), data.get("events", {"events": []}))
    _write_merged_json(os.path.join(run_dir, "profiles.json"), {
        "profiles": agent_summary.get("profiles", []),
        "inventory_only": bool(data.get("is_byoc")),
        "target_cluster_id": data.get("physical_cluster_id") or cluster_id,
    })
    queries_file = {
        "slow_queries": agent_summary["queries"]["slow_queries"],
        "topsql": agent_summary["queries"]["topsql"],
        "decoded_plan": agent_summary["queries"]["decoded_plan"],
        "statement_summary": agent_summary["queries"].get("statement_summary", []),
        "statements": agent_summary["queries"].get("statements", []),
    }
    if agent_summary["queries"].get("panic_classification_context"):
        queries_file["panic_classification_context"] = agent_summary["queries"]["panic_classification_context"]
    _write_merged_json(os.path.join(run_dir, "queries.json"), queries_file)
    _write_merged_json(os.path.join(run_dir, "logs.json"), agent_summary["logs"] if isinstance(agent_summary["logs"], dict) else {})
    _write_merged_json(os.path.join(run_dir, "tikv_thread_cpu.json"),
                data.get("tikv_thread_cpu", {}))
    _write_merged_json(os.path.join(run_dir, "links.json"), agent_summary["links"])
    _write_merged_json(os.path.join(run_dir, "gate_checks.json"), agent_summary["gate_checks"])
    _write_merged_json(os.path.join(run_dir, "collection_errors.json"), {
        "errors": agent_summary["errors"],
    })

    print(f"Evidence files saved to {run_dir}/", file=sys.stderr)
    return run_dir


def _write_json(path: str, obj: dict) -> None:
    with open(path, "w", encoding="utf-8") as f:
        json.dump(obj, f, indent=2, ensure_ascii=False, default=str)


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main():
    import argparse
    parser = argparse.ArgumentParser(
        description="Collect TiDB diagnosis evidence",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""Time inputs:
  - Preferred: timezone-bearing ISO 8601, e.g. 2026-03-09T04:14:00Z
  - Also accepted: 10-digit epoch seconds, e.g. 1773029640
  - Datetime strings must include a timezone; do not pass 2026-03-09 04:14:00

Examples:
  python3 collect_evidence.py 10805087899823024166 2026-03-09T04:14:00Z 2026-03-09T05:14:00Z --parent-id 2028998946604449792
  python3 collect_evidence.py 10805087899823024166 2026-03-09T04:14:00Z 2026-03-09T05:14:00Z --profile fast
  python3 collect_evidence.py 10805087899823024166 1773029640 1773033240 --parent-id 2028998946604449792

Use the time-conversion skill when you need conversion or timezone normalization:
  python3 .opencode/skills/builtin/time-conversion/scripts/time_convert.py dt-to-ts --dt '2026-03-09T04:14:00Z' --field epoch_s
  python3 .opencode/skills/builtin/time-conversion/scripts/time_convert.py dt-to-ts --dt '2026-03-09 12:14:00' --assume-tz Asia/Shanghai --field utc_iso
""",
    )
    parser.add_argument("cluster_id")
    parser.add_argument(
        "start_time",
        help="10-digit epoch seconds or timezone-bearing ISO 8601, e.g. 2026-03-09T04:14:00Z",
    )
    parser.add_argument(
        "end_time",
        help="10-digit epoch seconds or timezone-bearing ISO 8601, e.g. 2026-03-09T05:14:00Z",
    )
    parser.add_argument("--parent-id", default="",
                        help="Premium Resource Pool cluster ID for component health queries "
                             "(TiKV/PD/worker metrics). Required for Premium clusters.")
    parser.add_argument("--no-raw", action="store_true",
                        help="Do not archive raw upstream API responses under evidences/raw/.")
    parser.add_argument(
        "--profile",
        choices=("full", "fast"),
        default="full",
        help="Collection depth. fast stops after broad Wave 1 routing signals; full also runs attribution queries.",
    )
    args = parser.parse_args()

    cluster_id = args.cluster_id
    parent_id = args.parent_id or None
    if args.no_raw:
        os.environ["COLLECT_EVIDENCE_NO_RAW"] = "1"

    try:
        time_window = _canonical_time_window(args.start_time, args.end_time)
        start = time_window["start"]
        end = time_window["end"]

        print(
            "[collect_evidence] Prerequisite: tidb-knowledge-base Step A (tier-architecture.json + "
            "metric routing) and Step B (incorrect-conclusions.json) should be done before this script "
            "(see skills). This script does not validate routing or pitfall rules.",
            file=sys.stderr,
        )
        if parent_id:
            print(f"[collect_evidence] Premium mode: component health queries will use parent_id={parent_id}",
                  file=sys.stderr)
        print(
            "[collect_evidence] Canonical time window: "
            f"{time_window['start_iso']} to {time_window['end_iso']} "
            f"(epoch {start} to {end}; input {time_window['input_start']!r} to {time_window['input_end']!r})",
            file=sys.stderr,
        )

        api = ClinicAPI()

        print(f"Collecting evidence for cluster {cluster_id} ({start} to {end})...",
              file=sys.stderr)

        data = collect(
            api,
            cluster_id,
            start,
            end,
            parent_id=parent_id,
            profile=args.profile,
        )
        data["_input_time_range"] = time_window
        infra_verdict = check_infra_gate(data)
        hotspot_verdict = detect_hotspot(data)

        run_dir = ""
        agent_summary_txt_path = ""
        agent_summary_path = ""
        workspace_evidence_report_path = ""
        errors = data.get("errors", [])

        output_dir = _evidence_base_dir()
        run_dir = save_evidence_files(data, infra_verdict, hotspot_verdict,
                                      cluster_id, start, end, output_dir)
        agent_summary_txt_path = os.path.join(run_dir, "evidence_summary_for_agent.txt")
        agent_summary_path = os.path.join(run_dir, "evidence_summary_for_agent.json")
        workspace_evidence_report_path = os.path.join(run_dir, "workspace_evidence_report.json")

        raw_base = _raw_capture_dir()
        raw_dir = raw_base if raw_base and os.path.isdir(raw_base) else ""

        # stdout: structured JSON command result only
        print(json.dumps({
            "status": "ok",
            "run_dir": run_dir,
            "report_path": workspace_evidence_report_path,
            "agent_summary_txt_path": agent_summary_txt_path,
            "agent_summary_path": agent_summary_path,
            "workspace_evidence_report_path": workspace_evidence_report_path,
            "raw_dir": raw_dir,
            "collection_profile": args.profile,
            "errors": errors,
        }, indent=2))
    except Exception as exc:
        print(json.dumps({
            "status": "error",
            "run_dir": "",
            "report_path": "",
            "agent_summary_txt_path": "",
            "agent_summary_path": "",
            "workspace_evidence_report_path": "",
            "error": str(exc),
            "hint": (
                "Run `python3 collect_evidence.py -h` for supported time formats "
                "and time-conversion skill examples."
            ),
        }, indent=2))
        sys.exit(1)


if __name__ == "__main__":
    main()