#!/usr/bin/env python3
from __future__ import annotations

import argparse
import fcntl
import hashlib
import importlib.util
import json
import math
import os
import re
import sys
import tempfile
import time
from contextlib import contextmanager
from copy import deepcopy
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
import urllib.parse
import urllib.request
import zipfile


CONTRACT_VERSION = "2026-04-27"

_EVIDENCE_FILE_BY_KIND = {
    "metric": "metrics.json",
    "query": "queries.json",
    "log": "logs.json",
    "event": "events.json",
    "topology_snapshot": "topology_snapshots.json",
    "profile": "profiles.json",
    "file": "files.json",
    "cluster_meta": "cluster_meta.json",
}

_REPORT_FILENAME = "workspace_evidence_report.json"
_WORKSPACE_EVIDENCES_DIRNAME = "evidences"
_WORKSPACE_EVIDENCE_LOCK_FILENAME = ".workspace_evidence.lock"


def _repo_root() -> Path:
    return Path(__file__).resolve().parents[5]


def _clinic_helper_root() -> Path:
    env = os.environ.get("CLINIC_API_SKILL_ROOT")
    candidates = []
    if env:
        candidates.append(Path(env))
    candidates.extend(
        [
            Path(__file__).resolve().parents[1],
            Path.cwd() / ".opencode" / ".all-skills" / "nutshell-skills" / "platform" / "clinic-api",
            Path.cwd() / ".opencode" / "skills" / "nutshell-skills" / "platform" / "clinic-api",
            _repo_root() / "agentkit" / "skills" / "nutshell-skills" / "platform" / "clinic-api",
            _repo_root().parent.parent / "nutshell-skills/skills/platform/clinic-api",
            Path.home() / "nutshell-skills/skills/platform/clinic-api",
        ]
    )
    for candidate in candidates:
        script = candidate / "scripts/clinic_api.py"
        if script.exists():
            return candidate
    raise RuntimeError(
        "clinic_api.py helper not found; set CLINIC_API_SKILL_ROOT to the directory "
        "containing scripts/clinic_api.py"
    )


def _load_clinic_api_helper():
    root = _clinic_helper_root()
    module_path = root / "scripts/clinic_api.py"
    spec = importlib.util.spec_from_file_location("clinic_api_helper", module_path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"failed to load clinic_api.py from {module_path}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _workspace_root_from_env() -> Path | None:
    for parent in Path(__file__).resolve().parents:
        if parent.name.startswith("ws-"):
            return parent
    workspace_root = os.environ.get("INVESTIBOOK_WORKSPACE_DIR", "").strip()
    if workspace_root:
        return Path(workspace_root).expanduser().resolve()
    cwd = Path.cwd().resolve()
    if cwd.name.startswith("ws-"):
        return cwd
    return None


def _workspace_evidence_dir() -> Path:
    workspace_root = _workspace_root_from_env()
    if workspace_root is not None:
        return workspace_root / _WORKSPACE_EVIDENCES_DIRNAME
    raise RuntimeError("INVESTIBOOK_WORKSPACE_DIR is required for Clinic data commands")


def _ensure_perf_log() -> str:
    path = os.environ.get("CLINIC_PERF_LOG", "").strip()
    if not path:
        path = str(_workspace_evidence_dir() / "perf.jsonl")
        os.environ["CLINIC_PERF_LOG"] = path
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    return path


def _perf_event(event: dict[str, Any]) -> 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


def _command_perf_payload(args: argparse.Namespace) -> dict[str, Any]:
    payload: dict[str, Any] = {"command": getattr(args, "command", "")}
    for key in ("cluster_id", "name", "query_type", "start", "end", "step", "timeout", "limit", "file_type"):
        if hasattr(args, key):
            payload[key] = getattr(args, key)
    query = getattr(args, "query", "")
    sql = getattr(args, "sql", "")
    text = str(sql or query or "")
    if text:
        payload["query_hash"] = hashlib.sha1(text.encode(errors="replace")).hexdigest()[:12]
        payload["query_preview"] = text[:240]
    return payload


def build_api():
    module = _load_clinic_api_helper()
    base_cls = module.ClinicAPI

    class ClinicAPIExtended(base_cls):
        def search_cluster_exact(self, cluster_id: str, show_deleted: bool = True) -> dict[str, Any]:
            raw = self._get(
                "/clinic/api/v1/dashboard/clusters",
                {"cluster_id": cluster_id, "show_deleted": str(show_deleted).lower()},
            )
            items = raw.get("items") or raw.get("data", {}).get("items") or []
            exact = next((item for item in items if str(item.get("clusterID")) == str(cluster_id)), None)
            if exact is None:
                return {"error": f"cluster {cluster_id} not found"}
            return {"status": "success", "item": exact, "items": items}

        def search_clusters(
            self,
            *,
            query: str = "",
            tenant_id: str = "",
            cluster_id: str = "",
            show_deleted: bool = False,
            cluster_status: str = "",
            limit: int = 20,
            page: int = 1,
        ) -> dict[str, Any]:
            params: dict[str, Any] = {
                "show_deleted": str(show_deleted).lower(),
                "limit": limit,
                "page": page,
            }
            if cluster_id:
                params["cluster_id"] = cluster_id
            else:
                effective_query = query or tenant_id
                if effective_query:
                    params["query"] = effective_query
            if cluster_status:
                params["cluster_status"] = cluster_status

            raw = self._get("/clinic/api/v1/dashboard/clusters", params)
            if raw.get("error"):
                return raw

            items = raw.get("items") or raw.get("data", {}).get("items") or []
            if tenant_id:
                items = [
                    item
                    for item in items
                    if str(item.get("tenantID") or item.get("tenantId") or "") == str(tenant_id)
                ]
            if cluster_id:
                items = [
                    item
                    for item in items
                    if str(item.get("clusterID") or item.get("clusterId") or "") == str(cluster_id)
                ]

            return {
                "status": "success",
                "items": items,
                "raw_count": len(raw.get("items") or raw.get("data", {}).get("items") or []),
                "total": raw.get("total") or raw.get("data", {}).get("total"),
                "page": page,
                "limit": limit,
            }

        def get_cluster_detail(self, org_id: str, cluster_id: str) -> dict[str, Any]:
            return self._get(f"/clinic/api/v1/orgs/{org_id}/clusters/{cluster_id}", {})

        def get_resource_pool_components(self, org_id: str, cluster_id: str) -> dict[str, Any]:
            return self._get(
                f"/clinic/api/v1/orgs/{org_id}/clusters/{cluster_id}/resource-pool/components",
                {},
            )

        def _get_with_headers(self, path: str, params: dict[str, Any], headers: dict[str, str]) -> dict[str, Any]:
            qs = urllib.parse.urlencode(params)
            url = f"{self.endpoint}{path}" + (f"?{qs}" if qs else "")
            req = urllib.request.Request(
                url,
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    **{k: v for k, v in headers.items() if v},
                },
            )
            return self._do(req)

        def _get_text_with_headers(self, path: str, params: dict[str, Any], headers: dict[str, str]) -> dict[str, Any]:
            qs = urllib.parse.urlencode(params)
            url = f"{self.endpoint}{path}" + (f"?{qs}" if qs else "")
            req = urllib.request.Request(
                url,
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    **{k: v for k, v in headers.items() if v},
                },
            )
            try:
                with urllib.request.urlopen(req, timeout=120) as resp:
                    return {"status": "success", "text": resp.read().decode(errors="replace").strip()}
            except Exception as exc:
                return {"error": str(exc)}

        def _download_with_headers(
            self,
            path: str,
            params: dict[str, Any],
            headers: dict[str, str],
            output_path: Path,
        ) -> dict[str, Any]:
            qs = urllib.parse.urlencode(params)
            url = f"{self.endpoint}{path}" + (f"?{qs}" if qs else "")
            req = urllib.request.Request(
                url,
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    **{k: v for k, v in headers.items() if v},
                },
            )
            tmp_path: Path | None = None
            try:
                output_path.parent.mkdir(parents=True, exist_ok=True)
                with urllib.request.urlopen(req, timeout=300) as resp:
                    with tempfile.NamedTemporaryFile(
                        "wb",
                        dir=output_path.parent,
                        prefix=f".{output_path.name}.",
                        suffix=".tmp",
                        delete=False,
                    ) as tmp:
                        tmp.write(resp.read())
                        tmp.flush()
                        os.fsync(tmp.fileno())
                        tmp_path = Path(tmp.name)
                os.replace(tmp_path, output_path)
                return {
                    "status": "success",
                    "path": str(output_path),
                    "size_bytes": output_path.stat().st_size,
                }
            except Exception as exc:
                if tmp_path is not None:
                    try:
                        tmp_path.unlink(missing_ok=True)
                    except OSError:
                        pass
                return {"error": str(exc)}

        def activity_events(
            self,
            org_id: str,
            cluster_id: str,
            start: int,
            end: int,
            *,
            name: str = "",
            severity: int = 0,
            limit: int = 100,
            cursor: str = "",
        ) -> dict[str, Any]:
            headers = {"cursor": cursor} if cursor else {}
            params: dict[str, Any] = {
                "begin_ts": start,
                "end_ts": end,
                "severity": severity,
                "limit": limit,
            }
            if name:
                params["name"] = name
            return self._get_with_headers(
                f"/clinic/api/v1/activityhub/applications/{org_id}/targets/{cluster_id}/activities",
                params,
                headers,
            )

        def ngm_context(self, cluster_id: str) -> dict[str, str] | dict[str, Any]:
            search = self.search_cluster_exact(cluster_id)
            if search.get("error"):
                return search
            item = search.get("item") or {}
            org_id = str(item.get("orgID") or item.get("orgId") or "")
            detail = self.get_cluster_detail(org_id, cluster_id) if org_id else {}
            merged = deepcopy(item)
            if isinstance(detail, dict) and not detail.get("error"):
                merged.update(detail)
            deploy_type = str(
                merged.get("clusterDeployType") or merged.get("clusterDeployTypeV2") or merged.get("clusterType") or ""
            ).lower().replace(" ", "_")
            deploy_type_v2 = str(merged.get("clusterDeployTypeV2") or merged.get("clusterTypeV2") or "").lower()
            parent_id = str(merged.get("parentID") or merged.get("parentId") or "")
            ngm_cluster_id = parent_id if deploy_type_v2 in {"premium", "byoc"} and parent_id else cluster_id
            if deploy_type_v2 in {"premium", "byoc"}:
                deploy_type = deploy_type_v2
            return {
                "cluster_id": ngm_cluster_id,
                "tenant_id": str(merged.get("tenantID") or merged.get("tenantId") or ""),
                "project_id": str(merged.get("projectID") or merged.get("projectId") or ""),
                "deploy_type": deploy_type,
                "provider": str(
                    merged.get("provider")
                    or merged.get("clusterProviderName")
                    or merged.get("cloudProvider")
                    or ""
                ).lower(),
                "region": str(merged.get("region") or merged.get("clusterRegionName") or ""),
                "raw_cluster_id": cluster_id,
            }

        def _ngm_headers(self, cluster_id: str) -> dict[str, str] | dict[str, Any]:
            ctx = self.ngm_context(cluster_id)
            if ctx.get("error"):
                return ctx
            headers = {
                "x-cluster-id": str(ctx.get("cluster_id") or ""),
                "x-org-id": str(ctx.get("tenant_id") or ""),
                "x-deploy-type": str(ctx.get("deploy_type") or ""),
                "x-provider": str(ctx.get("provider") or ""),
                "x-region": str(ctx.get("region") or ""),
            }
            project_id = str(ctx.get("project_id") or "")
            if project_id and headers["x-deploy-type"] != "premium":
                headers["x-project-id"] = project_id
            return headers

        def continuous_profiling_group_profiles(self, cluster_id: str, start: int, end: int) -> dict[str, Any]:
            headers = self._ngm_headers(cluster_id)
            if headers.get("error"):
                return headers
            return self._get_with_headers(
                "/ngm/api/v1/continuous_profiling/group_profiles",
                {"begin_time": start, "end_time": end},
                headers,
            )

        def continuous_profiling_detail(self, cluster_id: str, ts: int) -> dict[str, Any]:
            headers = self._ngm_headers(cluster_id)
            if headers.get("error"):
                return headers
            return self._get_with_headers(
                "/ngm/api/v1/continuous_profiling/group_profile/detail",
                {"ts": ts},
                headers,
            )

        def continuous_profiling_action_token(
            self,
            cluster_id: str,
            *,
            ts: int,
            profile_type: str,
            component: str,
            address: str,
        ) -> dict[str, Any]:
            headers = self._ngm_headers(cluster_id)
            if headers.get("error"):
                return headers
            query = (
                f"ts={ts}&profile_type={profile_type}&component={component}"
                f"&address={address}&data_format=protobuf"
            )
            return self._get_text_with_headers(
                "/ngm/api/v1/continuous_profiling/action_token",
                {"q": query},
                headers,
            )

        def continuous_profiling_download(
            self,
            cluster_id: str,
            *,
            token: str,
            output_path: Path,
        ) -> dict[str, Any]:
            headers = self._ngm_headers(cluster_id)
            if headers.get("error"):
                return headers
            return self._download_with_headers(
                "/ngm/api/v1/continuous_profiling/download",
                {"token": token},
                headers,
                output_path,
            )

        def diagnostic_files(self, cluster_id: str, file_type: str, start: int, end: int) -> dict[str, Any]:
            headers = self._ngm_headers(cluster_id)
            if headers.get("error"):
                return headers
            endpoint = {
                "plan_replayer": "/ngm/api/v1/plan_replayer/list",
                "oom_record": "/ngm/api/v1/oom_record/list",
            }[file_type]
            return self._get_with_headers(endpoint, {"start": start, "end": end}, headers)

        def diagnostic_file_download(
            self,
            cluster_id: str,
            *,
            key: str,
            output_path: Path,
            provider: str = "",
            region: str = "",
        ) -> dict[str, Any]:
            headers = self._ngm_headers(cluster_id)
            if headers.get("error"):
                return headers
            return self._download_with_headers(
                "/ngm/api/v1/oom_record/files",
                {
                    "provider": provider or str(headers.get("x-provider") or ""),
                    "region": region or str(headers.get("x-region") or ""),
                    "key": key,
                },
                headers,
                output_path,
            )

        def loki_query_range(
            self,
            cluster_id: str,
            query: str,
            start_ns: int,
            end_ns: int,
            *,
            limit: int = 200,
            direction: str = "backward",
        ) -> dict[str, Any]:
            return self.logs_query_range(
                cluster_id,
                query,
                start_ns,
                end_ns,
                limit=limit,
                direction=direction,
            )

    return ClinicAPIExtended()


def _json_dump(data: dict[str, Any]) -> None:
    json.dump(data, sys.stdout, indent=2, ensure_ascii=False)
    sys.stdout.write("\n")


def _json_load(path: Path) -> dict[str, Any]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError, json.JSONDecodeError):
        return {}
    return data if isinstance(data, dict) else {}


def _json_write(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    content = json.dumps(payload, indent=2, ensure_ascii=False, default=str) + "\n"
    with tempfile.NamedTemporaryFile(
        "w",
        encoding="utf-8",
        dir=path.parent,
        prefix=f".{path.name}.",
        suffix=".tmp",
        delete=False,
    ) as tmp:
        tmp.write(content)
        tmp.flush()
        os.fsync(tmp.fileno())
        tmp_path = Path(tmp.name)
    os.replace(tmp_path, path)


@contextmanager
def _workspace_evidence_lock(evidence_dir: Path):
    evidence_dir.mkdir(parents=True, exist_ok=True)
    lock_path = evidence_dir / _WORKSPACE_EVIDENCE_LOCK_FILENAME
    with lock_path.open("a+", encoding="utf-8") as lock_file:
        fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
        try:
            yield
        finally:
            fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)


def _stable_key(payload: Any) -> str:
    return json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str)


def _merge_lists(existing: Any, incoming: Any) -> list[Any]:
    base = existing if isinstance(existing, list) else []
    extra = incoming if isinstance(incoming, list) else []
    seen: set[str] = set()
    result: list[Any] = []
    for item in [*base, *extra]:
        key = _stable_key(item)
        if key in seen:
            continue
        seen.add(key)
        result.append(item)
    return result


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


def _merge_evidence_payload(kind: str, existing: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]:
    if kind == "metric":
        return {"metrics": _merge_lists(existing.get("metrics"), payload.get("metrics"))}
    if kind == "query":
        merged = _merge_dicts(existing, payload)
        for key in ("slow_queries", "topsql", "statement_summary", "statements", "changefeeds"):
            if isinstance(existing.get(key), list) or isinstance(payload.get(key), list):
                merged[key] = _merge_lists(existing.get(key), payload.get(key))
        if "decoded_plan" in payload:
            merged["decoded_plan"] = deepcopy(payload["decoded_plan"])
        return merged
    if kind == "log":
        return _merge_dicts(existing, payload)
    if kind in {"event", "profile", "file"}:
        key = {"event": "events", "profile": "profiles", "file": "files"}[kind]
        merged = _merge_dicts(existing, payload)
        merged[key] = _merge_lists(existing.get(key), payload.get(key))
        return merged
    if kind == "topology_snapshot":
        merged = _merge_dicts(existing, payload)
        merged["topology_snapshots"] = _merge_lists(
            existing.get("topology_snapshots"),
            payload.get("topology_snapshots"),
        )
        return merged
    if kind == "cluster_meta":
        return _merge_dicts(existing, payload)
    return _merge_dicts(existing, payload)


def _compact_report_evidence(kind: str, merged: dict[str, Any]) -> dict[str, Any]:
    if kind == "metric":
        metrics = []
        for raw in merged.get("metrics") or []:
            if not isinstance(raw, dict):
                continue
            metrics.append({key: deepcopy(value) for key, value in raw.items() if key != "chart_data"})
        return {"metrics": metrics}
    if kind == "query":
        items: list[dict[str, Any]] = []
        for source_key, source_type in (
            ("slow_queries", "slow_query"),
            ("topsql", "topsql"),
            ("statement_summary", "statement_summary"),
            ("statements", "statement"),
            ("changefeeds", "ticdc_changefeed"),
        ):
            for raw in merged.get(source_key) or []:
                if isinstance(raw, dict):
                    row = deepcopy(raw)
                    row.setdefault("source", source_type)
                    items.append(row)
        return {"queries": items}
    if kind == "log":
        return {"logs": merged.get("log_counts") if isinstance(merged.get("log_counts"), list) else []}
    if kind == "event":
        return {"events": merged.get("events") if isinstance(merged.get("events"), list) else []}
    if kind == "topology_snapshot":
        return {
            "topology_snapshots": (
                merged.get("topology_snapshots")
                if isinstance(merged.get("topology_snapshots"), list)
                else []
            )
        }
    if kind == "profile":
        return {"profiles": merged.get("profiles") if isinstance(merged.get("profiles"), list) else []}
    if kind == "file":
        return {"files": merged.get("files") if isinstance(merged.get("files"), list) else []}
    return {}


def _normalized_report_time_range(payload: dict[str, Any]) -> dict[str, Any] | None:
    if not isinstance(payload, dict):
        return None
    raw = payload.get("time_range")
    if not isinstance(raw, dict):
        raw = payload.get("time_window")
    if not isinstance(raw, dict):
        return None

    start = raw.get("start")
    end = raw.get("end")
    if start is None and raw.get("start_ns") is not None:
        start = _parse_time_to_epoch_seconds(raw.get("start_ns"))
    if end is None and raw.get("end_ns") is not None:
        end = _parse_time_to_epoch_seconds(raw.get("end_ns"))
    if start is None or end is None:
        return None

    normalized: dict[str, Any] = {
        "start": int(start),
        "end": int(end),
    }
    step = raw.get("step")
    if step not in (None, ""):
        normalized["step"] = str(step)
    return normalized


def _window_candidate_from_value(value: Any) -> dict[str, Any] | None:
    if not isinstance(value, dict):
        return None
    candidate = _normalized_report_time_range({"time_range": value})
    if candidate is not None:
        return candidate
    if value.get("kind") == "promql_range":
        start = value.get("start")
        end = value.get("end")
        if start is None or end is None:
            return None
        normalized = {"start": int(start), "end": int(end)}
        step = value.get("step")
        if step not in (None, ""):
            normalized["step"] = str(step)
        return normalized
    return None


def _collect_window_candidates(value: Any, *, candidates: list[dict[str, Any]]) -> None:
    if isinstance(value, dict):
        direct = _window_candidate_from_value(value)
        if direct is not None:
            candidates.append(direct)
        for key, nested in value.items():
            if key in {"time_range", "time_window", "chart_data_ref"}:
                candidate = _window_candidate_from_value(nested)
                if candidate is not None:
                    candidates.append(candidate)
            _collect_window_candidates(nested, candidates=candidates)
    elif isinstance(value, list):
        for item in value:
            _collect_window_candidates(item, candidates=candidates)


def _recover_report_time_range(report: dict[str, Any], *, evidence_dir: Path) -> dict[str, Any] | None:
    candidates: list[dict[str, Any]] = []
    evidence = report.get("evidence")
    if isinstance(evidence, dict):
        _collect_window_candidates(evidence, candidates=candidates)

    for filename in sorted(set(_EVIDENCE_FILE_BY_KIND.values())):
        payload = _json_load(evidence_dir / filename)
        if payload:
            _collect_window_candidates(payload, candidates=candidates)

    if not candidates:
        return None

    start = min(int(item["start"]) for item in candidates)
    end = max(int(item["end"]) for item in candidates)
    recovered = {"start": start, "end": end}
    steps = {
        str(item.get("step"))
        for item in candidates
        if item.get("step") not in (None, "")
    }
    if len(steps) == 1:
        recovered["step"] = steps.pop()
    return recovered


def _persist_raw_evidence(command: str, kind: str, cluster_id: str, payload: dict[str, Any] | None) -> dict[str, Any] | None:
    if kind not in _EVIDENCE_FILE_BY_KIND:
        return None
    if not isinstance(payload, dict):
        raise RuntimeError(f"Clinic data command {command} did not produce a raw payload to persist")

    evidence_dir = _workspace_evidence_dir()
    evidence_dir.mkdir(parents=True, exist_ok=True)
    with _workspace_evidence_lock(evidence_dir):
        evidence_path = evidence_dir / _EVIDENCE_FILE_BY_KIND[kind]
        merged = _merge_evidence_payload(kind, _json_load(evidence_path), payload)
        _json_write(evidence_path, merged)

        report_path = evidence_dir / _REPORT_FILENAME
        report = _json_load(report_path)
        if report.get("type") != "workspace_evidence_report":
            report = {
                "type": "workspace_evidence_report",
                "metadata": {},
                "analysis": {},
                "evidence": {
                    "metrics": [],
                    "queries": [],
                    "logs": [],
                    "events": [],
                    "profiles": [],
                    "files": [],
                },
                "links": {},
                "gate_checks": {},
            }
        metadata = report.setdefault("metadata", {})
        metadata["cluster_id"] = cluster_id or str(metadata.get("cluster_id") or "")
        metadata["evidence_dir"] = str(evidence_dir)
        if kind == "cluster_meta":
            metadata["cluster"] = merged.get("cluster") or merged
        else:
            report.setdefault("evidence", {}).update(_compact_report_evidence(kind, merged))

        # metadata.time_range is the investigation scope, not the query window:
        # a targeted slice (e.g. sampling one suspicious hour) must never
        # shrink an established scope. First command seeds it; if the top-level
        # scope disappears later, recover it from already-persisted evidence
        # before falling back to the current payload window.
        time_range = _normalized_report_time_range(payload)
        existing_range = metadata.get("time_range")
        has_existing_range = (
            isinstance(existing_range, dict)
            and existing_range.get("start")
            and existing_range.get("end")
        )
        if not has_existing_range:
            recovered_range = _recover_report_time_range(report, evidence_dir=evidence_dir)
            if recovered_range is not None:
                metadata["time_range"] = recovered_range
            elif time_range is not None:
                metadata["time_range"] = time_range
        metadata["last_raw_evidence"] = {
            "source": f"clinic-cli:{command}",
            "kind": kind,
            "file_path": str(evidence_path),
            **({"time_range": time_range} if time_range is not None else {}),
        }
        _json_write(report_path, report)

    return {
        "evidence_dir": str(evidence_dir),
        "evidence_file_path": str(evidence_path),
        "report_path": str(report_path),
    }


def _ok(command: str, kind: str, cluster_id: str, **extra: Any) -> dict[str, Any]:
    payload = extra.get("payload")
    persisted = _persist_raw_evidence(command, kind, cluster_id, payload if isinstance(payload, dict) else None)
    visible_extra = dict(extra)
    if persisted:
        visible_extra.pop("payload", None)
    result = {
        "status": "ok",
        "contract_version": CONTRACT_VERSION,
        "command": command,
        "kind": kind,
        "cluster_id": cluster_id,
        **visible_extra,
    }
    if persisted:
        result["persisted_evidence"] = persisted
    return result


def _error(command: str, kind: str, cluster_id: str, message: str, **extra: Any) -> dict[str, Any]:
    return {
        "status": "error",
        "contract_version": CONTRACT_VERSION,
        "command": command,
        "kind": kind,
        "cluster_id": cluster_id,
        "error": message,
        **extra,
    }


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


def _metric_ref_fields(metric_group_id: str) -> dict[str, Any]:
    return {
        "id": metric_group_id,
        "group": metric_group_id,
        "ref": {
            "type": "metric",
            "metric_group_id": metric_group_id,
        },
    }


def _coerce_int(value: Any, *, name: str) -> int:
    try:
        return int(str(value))
    except (TypeError, ValueError) as exc:
        raise ValueError(f"invalid integer for {name}: {value!r}") from exc


def _parse_time_to_epoch_seconds(value: Any) -> int:
    text = str(value).strip()
    if not text:
        raise argparse.ArgumentTypeError("empty timestamp")

    if re.fullmatch(r"\d+", text):
        raw = int(text)
        if raw >= 10**18:
            return raw // 1_000_000_000
        if raw >= 10**12:
            return raw // 1_000
        return raw

    normalized = text
    if normalized.endswith("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 argparse.ArgumentTypeError(
            f"invalid timestamp {value!r}; use epoch seconds/ms/ns or ISO8601 like 2026-04-22T06:24:00Z"
        ) from exc
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return int(dt.timestamp())


def _parse_time_to_epoch_or_ns(value: Any) -> int:
    text = str(value).strip()
    if re.fullmatch(r"\d+", text):
        raw = int(text)
        if raw >= 10**18:
            return raw
    return _parse_time_to_epoch_seconds(value)


def _to_ns(ts: int) -> int:
    return ts if ts >= 10**18 else ts * 1_000_000_000


def _time_window(start: int, end: int, step: str | int) -> dict[str, Any]:
    return {"start": int(start), "end": int(end), "step": str(step)}


def _series_points(result: dict[str, Any]) -> list[float]:
    rows = result.get("data", {}).get("result") or []
    values: list[float] = []
    for row in rows:
        series_values = row.get("values")
        if isinstance(series_values, list):
            for pair in series_values:
                if isinstance(pair, list) and len(pair) >= 2:
                    try:
                        values.append(float(pair[1]))
                    except (TypeError, ValueError):
                        continue
        elif "value" in row and isinstance(row.get("value"), list) and len(row["value"]) >= 2:
            try:
                values.append(float(row["value"][1]))
            except (TypeError, ValueError):
                continue
    return values


def _summarize_numeric(values: list[float]) -> dict[str, Any]:
    if not values:
        return {"point_count": 0}
    finite = [item for item in values if math.isfinite(item)]
    if not finite:
        return {"point_count": 0}
    return {
        "point_count": len(finite),
        "min": min(finite),
        "max": max(finite),
        "avg": sum(finite) / len(finite),
    }


def _metric_series_count(result: dict[str, Any]) -> int:
    rows = result.get("data", {}).get("result") or []
    return len(rows) if isinstance(rows, list) else 0


def _metric_series_identities(result: dict[str, Any], *, limit: int = 300) -> dict[str, Any]:
    rows = result.get("data", {}).get("result") or []
    if not isinstance(rows, list):
        return {
            "count": 0,
            "items": [],
            "truncated": False,
            "identity_labels_complete": False,
        }

    seen: set[str] = set()
    identities: list[dict[str, Any]] = []
    rows_with_identity = 0
    identity_keys = ("pod", "instance", "store", "address", "node", "ip")
    for row in rows:
        if not isinstance(row, dict):
            continue
        metric = row.get("metric") if isinstance(row.get("metric"), dict) else {}
        identity = ""
        for key in identity_keys:
            value = metric.get(key)
            if value not in (None, ""):
                identity = f"{key}:{value}"
                break
        if not identity:
            continue
        rows_with_identity += 1
        if identity in seen:
            continue
        seen.add(identity)
        if len(identities) < limit:
            identities.append({"identity": identity, "labels": metric})
    return {
        "count": len(seen),
        "items": identities,
        "truncated": len(seen) > len(identities),
        "identity_labels_complete": bool(rows) and rows_with_identity == len(rows),
    }


def _metric_series_summaries(result: dict[str, Any], *, limit: int = 50) -> list[dict[str, Any]]:
    rows = result.get("data", {}).get("result") or []
    if not isinstance(rows, list):
        return []

    summaries: list[dict[str, Any]] = []
    for row in rows[:limit]:
        if not isinstance(row, dict):
            continue
        metric = row.get("metric") if isinstance(row.get("metric"), dict) else {}
        values = _series_points({"data": {"result": [row]}})
        summary = _summarize_numeric(values)
        summaries.append(
            {
                "labels": metric,
                **summary,
            }
        )
    return summaries


def _metric_row_from_result(
    *,
    name: str,
    promql: str,
    result: dict[str, Any],
    cluster_id: str,
    start: int,
    end: int,
    step: str,
    metric_group_id: str,
    unit: str = "",
    summary_extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
    summary = _summarize_numeric(_series_points(result))
    summary["series_count"] = _metric_series_count(result)
    if summary_extra:
        summary.update(summary_extra)
    return {
        **_metric_ref_fields(metric_group_id),
        "metric_group_id": metric_group_id,
        "name": name,
        "promql": promql,
        "unit": unit,
        "summary": summary,
        "series_summaries": _metric_series_summaries(result),
        "time_window": _time_window(start, end, step),
        "chart_data_ref": {
            "kind": "promql_range",
            "cluster_id": cluster_id,
            "promql": promql,
            "start": int(start),
            "end": int(end),
            "step": str(step),
        },
        "chart_data": {
            "name": name,
            "promql": promql,
            "result_type": result.get("data", {}).get("resultType") or "matrix",
            "series": result.get("data", {}).get("result") or [],
        },
    }


def _series_stat(row: dict[str, Any]) -> dict[str, Any]:
    return _summarize_numeric(_series_points({"data": {"result": [row]}}))


def _rank_metric_series(
    result: dict[str, Any],
    *,
    label_keys: tuple[str, ...],
    limit: int = 10,
    rank_by: str = "max",
) -> list[dict[str, Any]]:
    rows = result.get("data", {}).get("result") or []
    ranked: list[dict[str, Any]] = []
    for row in rows:
        if not isinstance(row, dict):
            continue
        metric = row.get("metric") if isinstance(row.get("metric"), dict) else {}
        stat = _series_stat(row)
        value = stat.get(rank_by)
        if value is None:
            continue
        labels = {key: metric.get(key, "") for key in label_keys if metric.get(key) not in (None, "")}
        if not labels:
            labels = {key: value for key, value in metric.items() if key in {"instance", "pod", "node", "job", "container"}}
        ranked.append({"labels": labels, **stat})
    ranked.sort(key=lambda item: (-float(item.get(rank_by) or 0), json.dumps(item.get("labels", {}), sort_keys=True)))
    return ranked[:limit]


def _metric_result_empty(result: dict[str, Any]) -> bool:
    return _metric_series_count(result) == 0 or not _series_points(result)


def _instant_scalar_value(result: dict[str, Any]) -> float | None:
    rows = result.get("data", {}).get("result") or []
    if not isinstance(rows, list):
        return None
    for row in rows:
        if not isinstance(row, dict):
            continue
        value = row.get("value")
        if not isinstance(value, list) or len(value) < 2:
            continue
        try:
            numeric = float(value[1])
        except (TypeError, ValueError):
            continue
        if math.isfinite(numeric):
            return numeric
    return None


def _metric_node_name(metric: dict[str, Any], pattern: re.Pattern[str]) -> str:
    for label_name in ("pod", "pod_name", "kubernetes_pod_name", "instance", "job", "exported_instance"):
        raw = str(metric.get(label_name) or "").strip()
        match = pattern.search(raw)
        if match:
            return match.group(0)
    return ""


def _avg_metric_by_node(result: dict[str, Any], pattern: re.Pattern[str]) -> dict[str, float]:
    values_by_node: dict[str, list[float]] = {}
    for row in result.get("data", {}).get("result") or []:
        if not isinstance(row, dict):
            continue
        metric = row.get("metric") if isinstance(row.get("metric"), dict) else {}
        node = _metric_node_name(metric, pattern)
        if not node:
            continue
        values = _series_points({"data": {"result": [row]}})
        if values:
            values_by_node.setdefault(node, []).extend(values)
    return {
        node: sum(values) / len(values)
        for node, values in values_by_node.items()
        if values
    }


def _ranked_node_pairs(values_by_node: dict[str, float]) -> list[tuple[str, float]]:
    pairs = list(values_by_node.items())
    pairs.sort(key=lambda item: (-float(item[1]), item[0]))
    return pairs


def _median(values: list[float]) -> float | None:
    if not values:
        return None
    ordered = sorted(values)
    return ordered[len(ordered) // 2]


def _instance_regex(nodes: list[str]) -> str:
    escaped = [re.escape(node).replace(r"\-", "-") for node in nodes if node]
    return f"^({'|'.join(escaped)})($|[-.:].*)" if escaped else ""


def _histogram_p99_promql(
    bucket_metric: str,
    *,
    nodes: list[str] | None = None,
    extra_matchers: tuple[str, ...] = (),
) -> str:
    matchers = [matcher for matcher in extra_matchers if matcher]
    if nodes:
        regex = _instance_regex(nodes)
        if regex:
            matchers.append(f'instance=~"{regex}"')
    selector = f"{{{','.join(matchers)}}}" if matchers else ""
    return (
        "histogram_quantile(0.99, "
        f"sum by (le, instance) (rate({bucket_metric}{selector}[1m])))"
    )


def _ngm_page_params(cluster_id: str, cluster_meta: dict[str, Any] | None = None) -> dict[str, str]:
    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 meta.get("name") 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 meta.get("tidbVersion") or "")),
    ):
        if value:
            params[key] = value
    return params


def _topsql_tikv_instance(cluster_id: str, tikv_node: str) -> str:
    return f"{tikv_node}.db-tikv-peer.tidb{cluster_id}.svc:20160"


def _topsql_url(cluster_id: str, start: int, end: int, *, cluster_meta: dict[str, Any] | None, instance: str) -> str:
    page_query = urllib.parse.urlencode(
        _ngm_page_params(cluster_id, cluster_meta),
        quote_via=urllib.parse.quote,
    )
    topsql_query = urllib.parse.urlencode(
        {"from": str(start), "to": str(end), "instance": instance},
        quote_via=urllib.parse.quote,
    )
    return f"https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html?{page_query}#/topsql?{topsql_query}"


def _cluster_meta_for_links(api: Any, cluster_id: str) -> dict[str, Any]:
    if not hasattr(api, "search_cluster_exact"):
        return {"clusterID": cluster_id}
    search = api.search_cluster_exact(cluster_id)
    if not isinstance(search, dict) or search.get("error"):
        return {"clusterID": cluster_id}
    item = search.get("item") if isinstance(search.get("item"), dict) else {}
    merged = {"clusterID": cluster_id, **item}
    org_id = str(item.get("orgID") or item.get("orgId") or "")
    if org_id and hasattr(api, "get_cluster_detail"):
        detail = api.get_cluster_detail(org_id, cluster_id)
        if isinstance(detail, dict) and not detail.get("error"):
            merged.update(detail)
    return merged


_CPU_COMPONENT_ORDER = ("tidb", "tikv", "pd")
_CPU_PROMQL_BY_COMPONENT = {
    "tidb": 'sum by (pod, instance) (rate(process_cpu_seconds_total{component="tidb"}[1m]))',
    "tikv": 'sum by (pod, instance) (rate(process_cpu_seconds_total{component="tikv"}[1m]))',
    "pd": 'sum by (pod, instance) (rate(process_cpu_seconds_total{component="pd"}[1m]))',
}
_QUOTA_PROMQL_BY_COMPONENT = {
    "tidb": "max by (pod, instance) (tidb_server_maxprocs)",
    "tikv": "max by (pod, instance) (tikv_server_cpu_cores_quota)",
    "pd": "max by (pod, instance) (pd_service_maxprocs)",
}
_NODE_PATTERN_BY_COMPONENT = {
    "tidb": re.compile(r"\bdb-tidb-[A-Za-z0-9-]+\b"),
    "tikv": re.compile(r"\bdb-tikv-[A-Za-z0-9-]+\b"),
    "pd": re.compile(r"\bdb-pd-[A-Za-z0-9-]+\b"),
}


def _component_node_name(metric: dict[str, Any], *, component: str) -> str:
    pattern = _NODE_PATTERN_BY_COMPONENT[component]
    for label_name in ("pod", "pod_name", "kubernetes_pod_name"):
        raw = str(metric.get(label_name) or "").strip()
        if raw.startswith(f"db-{component}-"):
            return raw
    return _metric_node_name(metric, pattern)


def _cpu_overload_rows(
    api: Any,
    *,
    cluster_id: str,
    component: str,
    start: int,
    end: int,
    step: str,
    top_n: int,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], dict[str, Any]]:
    cpu_promql = _CPU_PROMQL_BY_COMPONENT[component]
    quota_promql = _QUOTA_PROMQL_BY_COMPONENT[component]
    cpu_result = api.query_range(cluster_id, cpu_promql, start, end, step=step)
    quota_result = api.query_range(cluster_id, quota_promql, start, end, step=step)
    rows = [
        _metric_row_from_result(
            name=f"{component}_cpu_rate",
            promql=cpu_promql,
            result=cpu_result,
            cluster_id=cluster_id,
            start=start,
            end=end,
            step=step,
            metric_group_id=f"{component}_cpu_rate",
            unit="cores",
        ),
        _metric_row_from_result(
            name=f"{component}_cpu_quota",
            promql=quota_promql,
            result=quota_result,
            cluster_id=cluster_id,
            start=start,
            end=end,
            step=step,
            metric_group_id=f"{component}_cpu_quota",
            unit="cores",
        ),
    ]
    if cpu_result.get("error"):
        return rows, [], {"error": cpu_result["error"]}
    if quota_result.get("error"):
        return rows, [], {"error": quota_result["error"]}

    cpu_by_node: dict[str, float] = {}
    for item in cpu_result.get("data", {}).get("result") or []:
        if not isinstance(item, dict):
            continue
        metric = item.get("metric") if isinstance(item.get("metric"), dict) else {}
        node = _component_node_name(metric, component=component)
        values = _series_points({"data": {"result": [item]}})
        if not node or not values:
            continue
        cpu_by_node[node] = max(cpu_by_node.get(node, 0.0), sum(values) / len(values))

    quota_by_node: dict[str, float] = {}
    for item in quota_result.get("data", {}).get("result") or []:
        if not isinstance(item, dict):
            continue
        metric = item.get("metric") if isinstance(item.get("metric"), dict) else {}
        node = _component_node_name(metric, component=component)
        quota = _median(_series_points({"data": {"result": [item]}}))
        if not node or quota is None or quota <= 0:
            continue
        quota_by_node[node] = max(quota_by_node.get(node, 0.0), quota)

    ranked: list[dict[str, Any]] = []
    for node, mean_cpu in cpu_by_node.items():
        quota = quota_by_node.get(node)
        if quota is None or quota <= 0:
            continue
        ranked.append(
            {
                "component": component,
                "node": node,
                "avg_cpu_cores": round(mean_cpu, 6),
                "cpu_quota": round(quota, 6),
                "avg_cpu_overload": round(mean_cpu / quota, 4),
            }
        )
    ranked.sort(key=lambda item: (-float(item["avg_cpu_overload"]), str(item["node"])))
    selected = ranked[:top_n]
    for idx, item in enumerate(selected, start=1):
        item["rank"] = idx
    return rows, selected, {
        "cpu_query": cpu_promql,
        "quota_query": quota_promql,
        "observed_cpu_nodes": len(cpu_by_node),
        "observed_quota_nodes": len(quota_by_node),
    }


def _loki_streams(result: dict[str, Any]) -> list[dict[str, Any]]:
    data = result.get("data")
    if not isinstance(data, dict):
        return []
    rows = data.get("result")
    return [row for row in rows if isinstance(row, dict)] if isinstance(rows, list) else []


def _error_warn_level(line: str) -> str:
    upper = line.upper()
    if "[ERROR]" in upper or "[FATAL]" in upper:
        return "ERROR"
    if "[WARN]" in upper or "[WARNING]" in upper:
        return "WARN"
    return ""


def _time_slices(start: int, end: int, *, step_seconds: int) -> list[tuple[int, int]]:
    slices: list[tuple[int, int]] = []
    cursor = start
    while cursor < end:
        next_cursor = min(end, cursor + step_seconds)
        slices.append((cursor, next_cursor))
        cursor = next_cursor
    return slices


def _node_log_queries(node: str, log_regex: str) -> list[str]:
    filter_expr = f"|~ `{log_regex}`"
    return [
        f'{{instance="{node}"}} {filter_expr}',
        f'{{pod="{node}"}} {filter_expr}',
        f'{{app="{node}"}} {filter_expr}',
        f'{{service_name="{node}"}} {filter_expr}',
    ]


def _count_error_warn_logs_for_nodes(
    api: Any,
    *,
    cluster_id: str,
    nodes_by_component: dict[str, set[str]],
    start: int,
    end: int,
    log_regex: str,
    limit: int,
    slice_seconds: int,
) -> tuple[dict[str, int], list[dict[str, Any]], list[dict[str, Any]]]:
    counts = {node: 0 for nodes in nodes_by_component.values() for node in nodes}
    log_counts: list[dict[str, Any]] = []
    debug_queries: list[dict[str, Any]] = []
    seen: set[tuple[str, str, str]] = set()
    node_to_component = {
        node: component
        for component, nodes in nodes_by_component.items()
        for node in nodes
    }
    for node in sorted(node_to_component):
        for logql in _node_log_queries(node, log_regex):
            matched = 0
            for slice_start, slice_end in _time_slices(start, end, step_seconds=slice_seconds):
                result = api.loki_query_range(
                    cluster_id,
                    logql,
                    _to_ns(slice_start),
                    _to_ns(slice_end),
                    limit=limit,
                    direction="backward",
                )
                for stream in _loki_streams(result):
                    labels = stream.get("stream") if isinstance(stream.get("stream"), dict) else {}
                    for entry in stream.get("values") or []:
                        if not isinstance(entry, list) or len(entry) < 2:
                            continue
                        ts_ns = str(entry[0])
                        line = str(entry[1])
                        level = _error_warn_level(line)
                        if level not in {"ERROR", "WARN"}:
                            continue
                        raw_labels = " ".join(str(labels.get(key) or "") for key in ("instance", "pod", "app", "service_name"))
                        if node not in raw_labels and node not in line:
                            continue
                        key = (ts_ns, node, line)
                        if key in seen:
                            continue
                        seen.add(key)
                        matched += 1
                        counts[node] = counts.get(node, 0) + 1
                        log_counts.append(
                            {
                                "component": node_to_component[node],
                                "node": node,
                                "level": level,
                                "message": _normalize_log_message(line),
                                "count": 1,
                                "instances": [node],
                                "first_seen_ns": ts_ns,
                                "last_seen_ns": ts_ns,
                            }
                        )
            debug_queries.append({"node": node, "logql": logql, "matched_lines": matched})
    return counts, log_counts, debug_queries


def _date_partitions(start: int, end: int, *, fmt: str) -> list[str]:
    start_dt = datetime.fromtimestamp(int(start), tz=timezone.utc).date()
    end_dt = datetime.fromtimestamp(int(end), tz=timezone.utc).date()
    dates: list[str] = []
    current = start_dt
    while current <= end_dt:
        dates.append(current.strftime(fmt))
        current += timedelta(days=1)
    return dates


def _sql_string_list(values: list[str]) -> str:
    return ", ".join("'" + value.replace("'", "''") + "'" for value in values)


def _component_from_store_address(address: str) -> str:
    lowered = address.lower()
    if "tiflash" in lowered:
        return "tiflash"
    if "tikv" in lowered:
        return "tikv"
    return "unknown"


def _trim_line(line: str, *, limit: int = 240) -> str:
    compact = re.sub(r"\s+", " ", line).strip()
    return compact if len(compact) <= limit else compact[: limit - 3] + "..."


def _guess_log_level(line: str) -> str:
    lowered = line.lower()
    if "panic" in lowered:
        return "PANIC"
    if "fatal" in lowered:
        return "FATAL"
    if "[error]" in lowered or " error " in lowered:
        return "ERROR"
    if "[warn" in lowered or " warning " in lowered:
        return "WARN"
    return "INFO"


def _normalize_log_message(line: str) -> str:
    compact = _trim_line(line, limit=320)
    compact = re.sub(r"^\d{4}-\d{2}-\d{2}[tT ][^ ]+\s*", "", compact)
    compact = re.sub(r"^\[[A-Z]+\]\s*", "", compact)
    return compact


def _rows_to_objects(columns: list[str], rows: list[list[Any]]) -> list[dict[str, Any]]:
    result: list[dict[str, Any]] = []
    for row in rows:
        mapped = {}
        for idx, column in enumerate(columns):
            mapped[column] = row[idx] if idx < len(row) else None
        result.append(mapped)
    return result


def _first_list(raw: Any, keys: tuple[str, ...]) -> list[Any]:
    if isinstance(raw, list):
        return raw
    if not isinstance(raw, dict):
        return []
    for key in keys:
        value = raw.get(key)
        if isinstance(value, list):
            return value
    data = raw.get("data")
    if isinstance(data, dict):
        for key in keys:
            value = data.get(key)
            if isinstance(value, list):
                return value
    if isinstance(data, list):
        return data
    return []


def _api_error(raw: Any) -> str:
    if isinstance(raw, dict) and raw.get("error"):
        return str(raw["error"])
    return ""


def _response_object(raw: Any) -> dict[str, Any]:
    if not isinstance(raw, dict):
        return {}
    data = raw.get("data")
    if isinstance(data, dict):
        return data
    return raw


def _canonical_cluster_metadata(api: Any, cluster_id: str) -> tuple[dict[str, Any], str]:
    getter = getattr(api, "get_cluster_by_id", None)
    if not callable(getter):
        return {}, ""
    try:
        result = getter(cluster_id)
    except Exception as exc:  # noqa: BLE001 - surfaced as a stable CLI error.
        return {}, str(exc)
    if not isinstance(result, dict):
        return {}, "cluster metadata returned an unexpected payload"
    if result.get("error"):
        return {}, str(result["error"])
    metadata = _response_object(result)
    return metadata, ""


def _metadata_is_byoc(api: Any, metadata: dict[str, Any]) -> bool:
    checker = getattr(api, "is_byoc_metadata", None)
    if callable(checker):
        return bool(checker(metadata))
    return _normalize_deploy_type(metadata) in {"byoc", "byoc_host", "byoc_resource_pool"}


def _require_byoc(api: Any, cluster_id: str, *, command: str) -> tuple[dict[str, Any], dict[str, Any] | None]:
    metadata, error = _canonical_cluster_metadata(api, cluster_id)
    if error:
        return {}, _error(command, "query", cluster_id, f"could not resolve cluster tier: {error}")
    if not metadata:
        return {}, _error(command, "query", cluster_id, "cluster tier metadata is unavailable")
    if not _metadata_is_byoc(api, metadata):
        deploy_type = _normalize_deploy_type(metadata) or "unknown"
        return {}, _error(
            command,
            "query",
            cluster_id,
            f"{command} live operations are supported only for BYOC clusters; cluster tier is {deploy_type}",
        )
    return metadata, None


def _query_payload(
    *,
    slow_queries: list[Any] | None = None,
    topsql: list[Any] | None = None,
    statements: list[Any] | None = None,
    changefeeds: list[Any] | None = None,
    time_range: dict[str, Any] | None = None,
) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "slow_queries": slow_queries or [],
        "topsql": topsql or [],
        "statements": statements or [],
    }
    if changefeeds is not None:
        payload["changefeeds"] = changefeeds
    if time_range is not None:
        payload["time_range"] = time_range
    return payload


def _operation_rows(raw: Any, *, operation: str, source: str) -> list[dict[str, Any]]:
    values = _first_list(raw, ("items", "rows", "instances", "summaries", "plans", "statements"))
    if not values:
        value = _response_object(raw)
        envelope_keys = {"status", "data", "metadata"}
        if value and not set(value).issubset(envelope_keys):
            values = [value]
    rows: list[dict[str, Any]] = []
    for value in values:
        if not isinstance(value, dict):
            continue
        row = deepcopy(value)
        row.setdefault("source", source)
        row.setdefault("operation", operation)
        rows.append(row)
    return rows


_TICDC_DIAGNOSTIC_FIELD_ALIASES = {
    "id": ("id",),
    "display_id": ("display_id", "displayId"),
    "display_name": ("display_name", "displayName"),
    "name": ("name",),
    "status": ("status",),
    "state": ("state",),
    "cluster_type": ("cluster_type", "clusterType"),
    "architecture": ("architecture",),
    "downstream_type": ("downstream_type", "downstreamType"),
    "sink_type": ("sink_type", "sinkType"),
    "checkpoint_tso": ("checkpoint_tso", "checkpointTso"),
    "checkpoint_ts": ("checkpoint_ts", "checkpointTs"),
    "resolved_tso": ("resolved_tso", "resolvedTso"),
    "resolved_ts": ("resolved_ts", "resolvedTs"),
    "rcu": ("rcu",),
    "created_at": ("created_at", "createdAt"),
    "updated_at": ("updated_at", "updatedAt"),
}


def _compact_ticdc_changefeed(raw: Any, *, operation: str) -> dict[str, Any]:
    if not isinstance(raw, dict):
        return {}
    row: dict[str, Any] = {}
    for output_key, aliases in _TICDC_DIAGNOSTIC_FIELD_ALIASES.items():
        value = next((raw[key] for key in aliases if raw.get(key) not in (None, "")), None)
        if isinstance(value, (str, int, float, bool)):
            row[output_key] = deepcopy(value)
    row["source"] = "ticdc_changefeed"
    row["operation"] = operation
    return row


def _top_sql_instance_rows(raw: Any) -> list[dict[str, Any]]:
    payload = _response_object(raw)
    rows: list[dict[str, Any]] = []
    if isinstance(payload, dict):
        for instance_type in ("tidb", "tikv", "pd"):
            values = payload.get(instance_type) or payload.get(f"{instance_type}s")
            if not isinstance(values, list):
                continue
            for value in values:
                if isinstance(value, str):
                    row = {"instance": value}
                elif isinstance(value, dict):
                    row = deepcopy(value)
                else:
                    continue
                row.setdefault("instance_type", instance_type)
                row.setdefault("source", "topsql")
                row.setdefault("operation", "instances")
                rows.append(row)
    return rows or _operation_rows(raw, operation="instances", source="topsql")


def _flatten_diagnostic_files(raw: dict[str, Any], *, file_type: str) -> list[dict[str, Any]]:
    files: list[dict[str, Any]] = []
    records = _first_list(raw, ("records", "items", "files"))
    for record in records:
        if not isinstance(record, dict):
            continue
        nested_files = record.get("files")
        if isinstance(nested_files, list):
            for item in nested_files:
                if isinstance(item, dict):
                    files.append(
                        {
                            "kind": file_type,
                            "instance": record.get("instance") or record.get("target") or "",
                            **item,
                        }
                    )
        else:
            files.append({"kind": file_type, **record})
    return files


def _sanitize_filename(value: str, *, fallback: str) -> str:
    name = Path(value).name
    name = re.sub(r"[^A-Za-z0-9._-]+", "_", name).strip("._")
    return name or fallback


def _short_hash(*parts: Any) -> str:
    raw = "\0".join(str(part) for part in parts)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:12]


def _filename_with_identity_hash(filename: str, *identity: Any) -> str:
    digest = _short_hash(*identity)
    path = Path(filename)
    suffix = path.suffix
    stem = path.name[: -len(suffix)] if suffix else path.name
    return f"{stem}-{digest}{suffix}"


def _workspace_download_dir(kind: str) -> Path:
    try:
        root = _workspace_evidence_dir()
    except RuntimeError as exc:
        raise RuntimeError("INVESTIBOOK_WORKSPACE_DIR is required for downloads") from exc
    path = root / "downloads" / kind
    path.mkdir(parents=True, exist_ok=True)
    return path


def _safe_unzip(archive: Path, output_dir: Path | None = None) -> list[str]:
    if not zipfile.is_zipfile(archive):
        return []
    target_dir = (output_dir or archive.with_suffix("")).resolve()
    target_dir.mkdir(parents=True, exist_ok=True)
    extracted: list[str] = []
    with zipfile.ZipFile(archive) as zf:
        for member in zf.infolist():
            dest = (target_dir / member.filename).resolve()
            if target_dir not in (dest, *dest.parents):
                raise RuntimeError(f"unsafe zip member path: {member.filename}")
            if member.is_dir():
                dest.mkdir(parents=True, exist_ok=True)
                continue
            dest.parent.mkdir(parents=True, exist_ok=True)
            with zf.open(member) as src, dest.open("wb") as out:
                out.write(src.read())
            extracted.append(str(dest))
    return extracted


def _download_extract_dir(archive: Path) -> Path:
    return archive.parent / f"{archive.name}.extracted-{os.getpid()}"


_TOPOLOGY_SUPPORTED_DEPLOY_TYPES = {
    "dedicated",
    "resource_pool",
    "byoc_resource_pool",
    "premium_resource_pool",
    "premium",
    "byoc",
}
_TOPOLOGY_UNSUPPORTED_DEPLOY_TYPES = {"shared", "starter", "essential"}
_TOPOLOGY_CPU_METRICS = {
    "tidb": ("tidb_server_maxprocs", 'tidb_server_maxprocs'),
    "tikv": ("tikv_server_cpu_cores_quota", 'tikv_server_cpu_cores_quota'),
    "tiflash": ("tiflash_proxy_tikv_server_cpu_cores_quota", 'tiflash_proxy_tikv_server_cpu_cores_quota'),
    "pd": ("pd_service_maxprocs", 'pd_service_maxprocs'),
}
_TOPOLOGY_ACTIVITY_ACTION_RE = re.compile(
    r"(?i)(scale|resize|modify|change|update|adjust|capacity|topology|"
    r"扩容|缩容|扩缩|变配|规格|容量|拓扑|节点)"
)
_TOPOLOGY_ACTIVITY_COMPONENT_RE = re.compile(
    r"(?i)(tidb|tikv|tiflash|tiproxy|pd|component|node|instance|"
    r"组件|节点|实例)"
)


def _normalize_deploy_type(*sources: dict[str, Any]) -> str:
    for source in sources:
        for key in ("clusterDeployType", "clusterDeployTypeV2", "clusterType", "clusterTypeV2", "deployType"):
            value = source.get(key) if isinstance(source, dict) else None
            if value:
                return re.sub(r"[ -]+", "_", str(value).strip().lower())
    return ""


def _event_time_seconds(event: dict[str, Any]) -> int:
    for value in (
        event.get("calibration_time"),
        (event.get("payload") or {}).get("happened_at_ms") if isinstance(event.get("payload"), dict) else None,
        (event.get("payload") or {}).get("HappenedAtMs") if isinstance(event.get("payload"), dict) else None,
    ):
        if value in (None, ""):
            continue
        try:
            raw = int(str(value))
        except (TypeError, ValueError):
            continue
        return raw // 1000 if raw >= 10**12 else raw
    return 0


def _event_summary_text(event: dict[str, Any]) -> str:
    payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
    parts = [
        event.get("name"),
        event.get("display_name") or event.get("displayName"),
        payload.get("summary"),
        payload.get("display_event_type"),
        payload.get("display_status"),
    ]
    detail = payload.get("detail")
    if isinstance(detail, dict):
        parts.append(json.dumps(detail, ensure_ascii=False, default=str))
    return " ".join(str(part) for part in parts if part not in (None, ""))


def _topology_activity_event(event: dict[str, Any]) -> dict[str, Any] | None:
    text = _event_summary_text(event)
    if not (
        _TOPOLOGY_ACTIVITY_ACTION_RE.search(text)
        and _TOPOLOGY_ACTIVITY_COMPONENT_RE.search(text)
    ):
        return None
    payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
    return {
        "time": _event_time_seconds(event),
        "name": event.get("name") or "",
        "display_name": event.get("display_name") or event.get("displayName") or payload.get("display_event_type") or "",
        "status": (event.get("tags") or {}).get("status") or payload.get("display_status") or "",
        "summary": str(payload.get("summary") or event.get("summary") or "")[:1000],
    }


def _component_metric_snapshot(
    *,
    api: Any,
    cluster_id: str,
    component: str,
    promql: str,
    at: int,
) -> dict[str, Any]:
    result = api.query_instant(cluster_id, promql, time=at)
    if result.get("error"):
        return {
            "component": component,
            "promql": promql,
            "status": "error",
            "evidence_status": "error",
            "historical_metric_available": False,
            "error": result["error"],
        }
    values = _series_points(result)
    finite = [value for value in values if math.isfinite(value)]
    distinct_values = sorted({float(value) for value in finite})
    summary = _summarize_numeric(finite)
    series_count = _metric_series_count(result)
    instance_identities = _metric_series_identities(result)
    identity_labels_complete = bool(instance_identities["identity_labels_complete"])
    instances = instance_identities["items"] if identity_labels_complete else []
    instance_count = instance_identities["count"] if identity_labels_complete else series_count
    instances_truncated = bool(instance_identities["truncated"]) if identity_labels_complete else False
    instance_count_source = "identity_labels" if identity_labels_complete else "series_count_fallback"
    historical_metric_available = series_count > 0
    single_node_cores = distinct_values[0] if len(distinct_values) == 1 else None
    return {
        "component": component,
        "promql": promql,
        "status": "ok",
        "evidence_status": "ok" if historical_metric_available else "no_data",
        "at": at,
        "series_count": series_count,
        "instance_count": instance_count,
        "instance_count_source": instance_count_source,
        "identity_labels_complete": identity_labels_complete,
        "instances": instances,
        "instances_truncated": instances_truncated,
        "historical_metric_available": historical_metric_available,
        "single_node_cores": single_node_cores,
        "distinct_core_values": distinct_values[:20],
        "summary": {
            **summary,
            "series_count": series_count,
            "instance_count": instance_count,
            "instance_count_source": instance_count_source,
            "identity_labels_complete": identity_labels_complete,
            "instances_truncated": instances_truncated,
            "historical_metric_available": historical_metric_available,
        },
    }


def _current_components_for_snapshot(api: Any, org_id: str, cluster_id: str, item: dict[str, Any], detail: dict[str, Any]) -> dict[str, Any]:
    deploy_type = _normalize_deploy_type(item, detail)
    if deploy_type in _TOPOLOGY_UNSUPPORTED_DEPLOY_TYPES:
        return {
            "status": "unsupported",
            "reason": "topology is not supported for shared/starter/essential clusters",
        }
    if deploy_type not in _TOPOLOGY_SUPPORTED_DEPLOY_TYPES:
        return {
            "status": "unsupported",
            "reason": f"cluster type {deploy_type!r} has no known topology path",
        }
    raw_components = api.get_resource_pool_components(org_id, cluster_id)
    if isinstance(raw_components, dict) and raw_components.get("error"):
        return {"status": "error", "error": raw_components["error"]}
    return {
        "status": "ok",
        "source": "resource_pool_components_current",
        "components": _first_list(raw_components, ("items", "components", "component_infos")),
    }


def _activity_events_with_premium_fallback(
    *,
    api: Any,
    org_id: str,
    cluster_id: str,
    cluster_item: dict[str, Any],
    start: int,
    end: int,
    name: str = "",
    severity: int = 0,
    limit: int = 100,
    cursor: str = "",
) -> tuple[dict[str, Any], str, bool, str]:
    deploy_type_v2 = str(cluster_item.get("clusterDeployTypeV2") or cluster_item.get("clusterTypeV2") or "").lower()
    parent_id = str(cluster_item.get("parentID") or cluster_item.get("parentId") or "")
    fallback_target_id = parent_id if deploy_type_v2 == "premium" and parent_id and parent_id != cluster_id else ""

    result = api.activity_events(
        org_id,
        cluster_id,
        start,
        end,
        name=name,
        severity=severity,
        limit=limit,
        cursor=cursor,
    )
    target_cluster_id = cluster_id
    fallback_error = ""
    fallback_used = False
    if result.get("error") and fallback_target_id:
        fallback_error = str(result["error"])
        fallback_result = api.activity_events(
            org_id,
            fallback_target_id,
            start,
            end,
            name=name,
            severity=severity,
            limit=limit,
            cursor=cursor,
        )
        if not fallback_result.get("error"):
            result = fallback_result
            target_cluster_id = fallback_target_id
            fallback_used = True

    return result, target_cluster_id, fallback_used, fallback_error


def _cluster_compact(
    detail: dict[str, Any],
    search_item: dict[str, Any],
    components: Any = None,
    *,
    topology_cluster_id: str = "",
) -> dict[str, Any]:
    merged = deepcopy(detail) if isinstance(detail, dict) else {}
    merged.update(search_item or {})
    cluster = {
        "cluster_id": str(merged.get("clusterID") or merged.get("cluster_id") or ""),
        "cluster_name": merged.get("name") or merged.get("clusterName") or "",
        "org_id": str(merged.get("orgID") or merged.get("orgId") or ""),
        "tenant_id": str(merged.get("tenantID") or merged.get("tenantId") or ""),
        "cluster_type": merged.get("clusterDeployType") or merged.get("clusterType") or "",
        "cluster_type_v2": merged.get("clusterDeployTypeV2") or merged.get("clusterTypeV2") or "",
        "cluster_status": merged.get("status") or merged.get("clusterStatus") or "",
        "parent_id": str(merged.get("parentID") or merged.get("parentId") or ""),
        "cloud_provider": merged.get("cloudProvider") or "",
        "region": merged.get("region") or "",
    }
    if components is not None:
        cluster["components"] = components
        cluster["topology_source"] = "resource_pool_components"
        cluster["topology_cluster_id"] = topology_cluster_id or cluster["cluster_id"]
    return cluster


def _cluster_search_row(item: dict[str, Any]) -> dict[str, Any]:
    return _cluster_compact({}, item)


def _filter_cluster_search_items(
    items: list[Any],
    *,
    tenant_id: str = "",
    cluster_id: str = "",
) -> list[dict[str, Any]]:
    filtered = []
    for item in items:
        if not isinstance(item, dict):
            continue
        if tenant_id and str(item.get("tenantID") or item.get("tenantId") or "") != str(tenant_id):
            continue
        if cluster_id and str(item.get("clusterID") or item.get("clusterId") or "") != str(cluster_id):
            continue
        filtered.append(item)
    return filtered


def run_search_clusters(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    effective_query = args.query or args.tenant_id
    result = api.search_clusters(
        query=effective_query,
        tenant_id=args.tenant_id,
        cluster_id=args.cluster_id,
        show_deleted=args.show_deleted,
        cluster_status=args.cluster_status,
        limit=args.limit,
        page=args.page,
    )
    if result.get("error"):
        return _error("search clusters", "cluster_search", args.cluster_id, result["error"])

    items = _filter_cluster_search_items(
        _first_list(result, ("items", "clusters")),
        tenant_id=args.tenant_id,
        cluster_id=args.cluster_id,
    )
    clusters = [_cluster_search_row(item) for item in items]
    summary = {
        "count": len(clusters),
        "raw_count": result.get("raw_count"),
        "total": result.get("total"),
        "query": args.query,
        "tenant_id": args.tenant_id,
        "cluster_id": args.cluster_id,
        "cluster_status": args.cluster_status,
        "show_deleted": args.show_deleted,
        "page": args.page,
        "limit": args.limit,
    }
    return _ok(
        "search clusters",
        "cluster_search",
        args.cluster_id,
        summary=summary,
        clusters=clusters,
    )


def run_cluster_metadata(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    search = api.search_cluster_exact(args.cluster_id)
    if search.get("error"):
        return _error("cluster-metadata", "cluster_meta", args.cluster_id, search["error"])

    item = search.get("item") or {}
    org_id = str(item.get("orgID") or item.get("orgId") or "")
    if not org_id:
        return _error("cluster-metadata", "cluster_meta", args.cluster_id, "orgID missing from cluster lookup")

    detail = api.get_cluster_detail(org_id, args.cluster_id)
    if detail.get("error"):
        return _error("cluster-metadata", "cluster_meta", args.cluster_id, detail["error"], search_item=item)

    components = None
    topology_cluster_id = ""
    if args.include_components:
        deploy_type = _normalize_deploy_type(item, detail)
        if deploy_type in _TOPOLOGY_UNSUPPORTED_DEPLOY_TYPES:
            return _error(
                "cluster-metadata",
                "cluster_meta",
                args.cluster_id,
                "topology is not supported for shared/starter/essential clusters; use cluster metadata plus metrics/logs",
                search_item=item,
                detail=detail,
            )
        if deploy_type not in _TOPOLOGY_SUPPORTED_DEPLOY_TYPES:
            return _error(
                "cluster-metadata",
                "cluster_meta",
                args.cluster_id,
                f"cluster type {deploy_type!r} has no known topology path",
                search_item=item,
                detail=detail,
            )

        parent_id = str(detail.get("parentID") or item.get("parentID") or "")
        if args.resource_pool_scope == "parent":
            if not parent_id or parent_id == "null":
                return _error(
                    "cluster-metadata",
                    "cluster_meta",
                    args.cluster_id,
                    "parent resource pool requested but parentID is missing",
                    search_item=item,
                    detail=detail,
                )
            topology_cluster_id = parent_id
        else:
            topology_cluster_id = args.cluster_id

        raw_components = api.get_resource_pool_components(org_id, topology_cluster_id)
        if isinstance(raw_components, dict) and raw_components.get("error"):
            return _error(
                "cluster-metadata",
                "cluster_meta",
                args.cluster_id,
                raw_components["error"],
                search_item=item,
                detail=detail,
            )
        components = _first_list(raw_components, ("items", "components", "component_infos"))

    cluster = _cluster_compact(detail, item, components, topology_cluster_id=topology_cluster_id)
    return _ok(
        "cluster-metadata",
        "cluster_meta",
        args.cluster_id,
        cluster=cluster,
        payload={"cluster": cluster},
    )


def run_topology_snapshot(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    search = api.search_cluster_exact(args.cluster_id)
    if search.get("error"):
        return _error("topology-snapshot", "topology_snapshot", args.cluster_id, search["error"])

    item = search.get("item") or {}
    org_id = str(args.org_id or item.get("orgID") or item.get("orgId") or "")
    if not org_id:
        return _error("topology-snapshot", "topology_snapshot", args.cluster_id, "orgID missing from cluster lookup")

    detail = api.get_cluster_detail(org_id, args.cluster_id)
    if detail.get("error"):
        return _error("topology-snapshot", "topology_snapshot", args.cluster_id, detail["error"], search_item=item)

    at = int(args.at)
    before = max(1, int(args.metric_window_minutes)) * 60
    metric_start = at - before
    metric_end = at + before
    components = []
    for component, (_, promql) in _TOPOLOGY_CPU_METRICS.items():
        components.append(
            _component_metric_snapshot(
                api=api,
                cluster_id=args.cluster_id,
                component=component,
                promql=promql,
                at=at,
            )
        )

    activity_start = at - max(0, int(args.activity_before_days)) * 86400
    activity_end = at + max(0, int(args.activity_after_days)) * 86400
    activity_result, activity_target_cluster_id, activity_fallback_used, activity_fallback_error = _activity_events_with_premium_fallback(
        api=api,
        org_id=org_id,
        cluster_id=args.cluster_id,
        cluster_item=item,
        start=activity_start,
        end=activity_end,
        severity=args.severity,
        limit=args.limit,
    )
    activity_status: dict[str, Any]
    topology_events: list[dict[str, Any]] = []
    if activity_result.get("error"):
        activity_status = {"status": "error", "error": activity_result["error"]}
    else:
        events = [event for event in _first_list(activity_result, ("activities", "events", "items")) if isinstance(event, dict)]
        for event in events:
            item_event = _topology_activity_event(event)
            if item_event:
                topology_events.append(item_event)
        topology_events.sort(key=lambda event: int(event.get("time") or 0))
        activity_status = {
            "status": "ok",
            "event_count": len(events),
            "topology_event_count": len(topology_events),
            "partial": activity_result.get("partial"),
            "cursor": activity_result.get("cursor"),
            "requested_cluster_id": args.cluster_id,
            "target_cluster_id": activity_target_cluster_id,
            "fallback_used": activity_fallback_used,
        }
        if activity_fallback_used:
            activity_status["fallback_reason"] = "premium_activityhub_parent_target"
            activity_status["fallback_original_error"] = activity_fallback_error

    current_topology = _current_components_for_snapshot(api, org_id, args.cluster_id, item, detail)
    cluster = _cluster_compact(detail, item)
    missing_metric_components = [
        str(item.get("component") or "")
        for item in components
        if not item.get("historical_metric_available")
    ]
    metric_components_available = [
        str(item.get("component") or "")
        for item in components
        if item.get("status") == "ok" and item.get("historical_metric_available")
    ]
    error_metric_components = [
        str(item.get("component") or "")
        for item in components
        if item.get("status") == "error"
    ]
    historical_metrics_complete = not missing_metric_components and len(metric_components_available) == len(components)
    snapshot = {
        "cluster": cluster,
        "at": at,
        "metric_query": {
            "type": "instant",
            "at": at,
            "node_count_source": (
                "per_component_identity_labels_or_series_count_fallback_at_requested_timestamp"
            ),
            "metric_window_usage": "report_metadata_only_not_used_for_node_count",
            "historical_metrics_complete": historical_metrics_complete,
            "metric_components_available": metric_components_available,
            "missing_metric_components": missing_metric_components,
            "error_metric_components": error_metric_components,
        },
        "metric_window": {"start": metric_start, "end": metric_end, "step": str(args.step)},
        "components": components,
        "current_topology": current_topology,
        "activity_window": {"start": activity_start, "end": activity_end},
        "activity": {
            **activity_status,
            "topology_events": topology_events[:50],
        },
        "interpretation": {
            "metric_series_count_is_point_in_time_evidence": True,
            "metric_window_is_not_range_query_evidence": True,
            "current_topology_is_not_historical_evidence": True,
            "current_topology_is_not_historical_fallback_when_metrics_missing": True,
            "activity_events_are_change_evidence": True,
            "do_not_infer_scale_change_from_current_topology_difference_alone": True,
            "do_not_infer_historical_topology_from_no_activity_events_alone": True,
            "confidence": (
                "higher_when_cpu metrics exist and activity window has no topology changes; "
                "lower when activity is partial or errors"
            ),
        },
    }
    summary = {
        "at": at,
        "metric_query_type": "instant",
        "metric_window_usage": "report_metadata_only_not_used_for_node_count",
        "historical_metrics_complete": historical_metrics_complete,
        "metric_components_available": metric_components_available,
        "missing_metric_components": missing_metric_components,
        "error_metric_components": error_metric_components,
        "current_topology_usage": "comparison_only_not_historical_fallback",
        "cluster_name": cluster.get("cluster_name"),
        "component_count": len(components),
        "activity_status": activity_status.get("status"),
        "topology_event_count": len(topology_events),
        "activity_partial": activity_status.get("partial"),
        "current_topology_status": current_topology.get("status"),
        "components": [
            {
                "component": item.get("component"),
                "series_count": item.get("series_count"),
                "instance_count": item.get("instance_count"),
                "instance_count_source": item.get("instance_count_source"),
                "identity_labels_complete": item.get("identity_labels_complete"),
                "instances_truncated": item.get("instances_truncated"),
                "historical_metric_available": item.get("historical_metric_available"),
                "evidence_status": item.get("evidence_status"),
                "single_node_cores": item.get("single_node_cores"),
                "distinct_core_values": item.get("distinct_core_values"),
                "status": item.get("status"),
            }
            for item in components
        ],
    }
    payload = {
        "topology_snapshots": [snapshot],
        "time_window": {"start": metric_start, "end": metric_end, "step": str(args.step)},
    }
    return _ok(
        "topology-snapshot",
        "topology_snapshot",
        args.cluster_id,
        summary=summary,
        payload=payload,
    )


def run_metrics(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    result = api.query_range(args.cluster_id, args.query, args.start, args.end, step=str(args.step))
    if result.get("error"):
        return _error("metrics", "metric", args.cluster_id, result["error"])
    if result.get("status") != "success":
        return _error("metrics", "metric", args.cluster_id, "metrics query did not return success", raw_result=result)

    summary = _summarize_numeric(_series_points(result))
    summary["series_count"] = _metric_series_count(result)
    series_summaries = _metric_series_summaries(result)
    metric_group_id = args.metric_group_id or _slug(args.name)
    row = {
        **_metric_ref_fields(metric_group_id),
        "metric_group_id": metric_group_id,
        "name": args.name,
        "promql": args.query,
        "unit": args.unit,
        "summary": summary,
        "series_summaries": series_summaries,
        "time_window": _time_window(args.start, args.end, args.step),
        "chart_data_ref": {
            "kind": "promql_range",
            "cluster_id": args.cluster_id,
            "promql": args.query,
            "start": int(args.start),
            "end": int(args.end),
            "step": str(args.step),
        },
    }
    if args.inline_chart_data:
        row["chart_data"] = {
            "name": args.name,
            "promql": args.query,
            "result_type": result.get("data", {}).get("resultType") or "matrix",
            "series": result.get("data", {}).get("result") or [],
        }

    payload = {
        "metrics": [row],
        "time_range": _time_window(args.start, args.end, args.step),
    }
    return _ok(
        "metrics",
        "metric",
        args.cluster_id,
        summary=summary,
        metric_group_id=metric_group_id,
        payload=payload,
    )


def run_sql(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    metadata, metadata_error = _canonical_cluster_metadata(api, args.cluster_id)
    if metadata_error:
        return _error("sql", "query", args.cluster_id, f"could not resolve cluster tier: {metadata_error}")
    if metadata and _metadata_is_byoc(api, metadata):
        return _error(
            "sql",
            "query",
            args.cluster_id,
            "warehouse SQL is not supported for BYOC Slow Query, Top SQL, or Statements; use the live slow-query, top-sql, or statements command",
        )
    result = api.query_sql(args.cluster_id, args.sql, timeout=int(args.timeout))
    if result.get("error"):
        return _error("sql", "query", args.cluster_id, result["error"])

    columns = result.get("columns") or []
    rows = result.get("rows") or []
    objects = _rows_to_objects(columns, rows)
    for item in objects:
        item.setdefault("source", args.query_type)
        item.setdefault("title", args.name)
        item.setdefault("query_sql", args.sql)
    payload = {"slow_queries": [], "topsql": [], "statements": []}
    payload_key = "topsql" if args.query_type == "topsql" else "slow_queries"
    payload[payload_key] = objects
    summary = {
        "row_count": len(objects),
        "columns": columns,
        "query_type": args.query_type,
    }
    if objects:
        summary["sample"] = objects[: min(len(objects), args.sample_limit)]
    return _ok(
        "sql",
        "query",
        args.cluster_id,
        query_type=args.query_type,
        summary=summary,
        payload=payload,
    )


def run_slow_query(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    command = f"slow-query {args.slow_query_operation}"
    _, failure = _require_byoc(api, args.cluster_id, command=command)
    if failure is not None:
        return failure

    operation = args.slow_query_operation
    if operation == "available-fields":
        result = api.byoc_slow_query_available_fields(args.cluster_id)
    elif operation == "list":
        result = api.byoc_slow_query_list(
            args.cluster_id,
            args.start,
            args.end,
            limit=args.limit,
            order_by=args.order_by,
            desc=args.desc,
            fields=args.fields,
            text=args.text,
            digest=args.digest,
            show_internal=args.show_internal,
            db=args.db,
            plans=args.plan,
        )
    else:
        result = api.byoc_slow_query_detail(
            args.cluster_id,
            args.digest,
            args.timestamp,
            args.connect_id,
            show_internal=args.show_internal,
        )
    error = _api_error(result)
    if error:
        return _error(command, "query", args.cluster_id, error)
    if operation == "available-fields":
        fields = _first_list(result, ("fields", "items"))
        return _ok(
            command,
            "query_metadata",
            args.cluster_id,
            summary={"field_count": len(fields)},
            data=result,
        )

    rows = _operation_rows(result, operation=operation, source="slow_query")
    time_range = (
        {"start": args.start, "end": args.end}
        if operation == "list"
        else None
    )
    return _ok(
        command,
        "query",
        args.cluster_id,
        summary={"row_count": len(rows), "operation": operation},
        payload=_query_payload(slow_queries=rows, time_range=time_range),
    )


def run_top_sql(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    operation = args.top_sql_operation
    command = f"top-sql {operation}"
    _, failure = _require_byoc(api, args.cluster_id, command=command)
    if failure is not None:
        return failure
    if operation == "instances":
        result = api.byoc_top_sql_instances(args.cluster_id, args.start, args.end)
    else:
        result = api.byoc_top_sql_summary(
            args.cluster_id,
            args.start,
            args.end,
            args.window,
            args.top,
            args.instance,
            args.instance_type,
        )
    error = _api_error(result)
    if error:
        return _error(command, "query", args.cluster_id, error)
    rows = (
        _top_sql_instance_rows(result)
        if operation == "instances"
        else _operation_rows(result, operation=operation, source="topsql")
    )
    return _ok(
        command,
        "query",
        args.cluster_id,
        summary={"row_count": len(rows), "operation": operation},
        payload=_query_payload(
            topsql=rows,
            time_range={"start": args.start, "end": args.end},
        ),
    )


def run_statements(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    operation = args.statements_operation
    command = f"statements {operation}"
    _, failure = _require_byoc(api, args.cluster_id, command=command)
    if failure is not None:
        return failure

    if operation == "available-fields":
        result = api.byoc_statements_available_fields(args.cluster_id)
    elif operation == "statement-types":
        result = api.byoc_statements_statement_types(args.cluster_id)
    elif operation == "list":
        result = api.byoc_statements_list(
            args.cluster_id,
            args.start,
            args.end,
            schemas=args.schema,
            stmt_types=args.statement_type,
            text=args.text,
            fields=args.fields,
        )
    elif operation == "plans":
        result = api.byoc_statements_plans(
            args.cluster_id,
            args.start,
            args.end,
            args.digest,
            schema_name=args.schema_name,
        )
    else:
        result = api.byoc_statements_plan_detail(
            args.cluster_id,
            args.start,
            args.end,
            args.digest,
            plans=args.plan,
            schema_name=args.schema_name,
        )
    error = _api_error(result)
    if error:
        return _error(command, "query", args.cluster_id, error)
    if operation in {"available-fields", "statement-types"}:
        values = _first_list(result, ("fields", "statement_types", "items"))
        return _ok(
            command,
            "query_metadata",
            args.cluster_id,
            summary={"value_count": len(values), "operation": operation},
            data=result,
        )

    rows = _operation_rows(result, operation=operation, source="statement")
    return _ok(
        command,
        "query",
        args.cluster_id,
        summary={"row_count": len(rows), "operation": operation},
        payload=_query_payload(
            statements=rows,
            time_range={"start": args.start, "end": args.end},
        ),
    )


def run_ticdc(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    operation = args.ticdc_operation
    command = f"ticdc {operation}"
    if operation == "list":
        result = api.ticdc_changefeeds_list(args.cluster_id, org_id=args.org_id)
        error = _api_error(result)
        if error:
            return _error(command, "query", args.cluster_id, error)
        raw_rows = _first_list(result, ("changefeeds", "items"))
        rows = [
            compact
            for raw in raw_rows
            if (compact := _compact_ticdc_changefeed(raw, operation=operation))
        ]
    else:
        result = api.ticdc_changefeed_get(
            args.cluster_id,
            args.changefeed_id,
            org_id=args.org_id,
        )
        error = _api_error(result)
        if error:
            return _error(command, "query", args.cluster_id, error)
        raw = _response_object(result).get("changefeed")
        if not isinstance(raw, dict):
            raw = _response_object(result)
        compact = _compact_ticdc_changefeed(raw, operation=operation)
        if not compact.get("id"):
            return _error(
                command,
                "query",
                args.cluster_id,
                "TiCDC detail response did not contain a kernel changefeed id",
            )
        rows = [compact]

    return _ok(
        command,
        "query",
        args.cluster_id,
        summary={"changefeed_count": len(rows), "operation": operation},
        changefeeds=rows,
        payload=_query_payload(changefeeds=rows),
    )


def run_logs(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    start_ns = _to_ns(int(args.start))
    end_ns = _to_ns(int(args.end))
    query_logs = getattr(api, "logs_query_range", None) or api.loki_query_range
    result = query_logs(
        args.cluster_id,
        args.query,
        start_ns,
        end_ns,
        limit=int(args.limit),
        direction=args.direction,
    )
    if result.get("error"):
        return _error("logs", "log", args.cluster_id, result["error"])
    if result.get("status") != "success":
        return _error("logs", "log", args.cluster_id, "logs query did not return success", raw_result=result)

    grouped: dict[tuple[str, str, str], dict[str, Any]] = {}
    sample_lines: list[dict[str, Any]] = []
    total_lines = 0
    for stream in result.get("data", {}).get("result") or []:
        stream_labels = stream.get("stream") or {}
        component = str(stream_labels.get("container") or stream_labels.get("component") or "unknown")
        instance = str(stream_labels.get("instance") or stream_labels.get("pod") or "")
        for entry in stream.get("values") or []:
            if not isinstance(entry, list) or len(entry) < 2:
                continue
            total_lines += 1
            ts_ns = str(entry[0])
            line = str(entry[1])
            level = _guess_log_level(line)
            message = _normalize_log_message(line)
            key = (component, level, message)
            row = grouped.setdefault(
                key,
                {
                    "component": component,
                    "level": level,
                    "message": message,
                    "count": 0,
                    "instances": set(),
                    "first_seen_ns": ts_ns,
                    "last_seen_ns": ts_ns,
                },
            )
            row["count"] += 1
            if instance:
                row["instances"].add(instance)
            row["last_seen_ns"] = ts_ns
            if len(sample_lines) < args.sample_limit:
                sample_lines.append(
                    {
                        "timestamp_ns": ts_ns,
                        "component": component,
                        "level": level,
                        "instance": instance,
                        "line": _trim_line(line, limit=400),
                    }
                )

    log_counts = []
    for item in grouped.values():
        log_counts.append(
            {
                **item,
                "instances": sorted(item["instances"]),
            }
        )
    log_counts.sort(key=lambda item: (-int(item["count"]), item["component"], item["level"], item["message"]))

    payload = {
        "query": args.query,
        "time_window": {"start_ns": start_ns, "end_ns": end_ns},
        "log_counts": log_counts,
        "samples": sample_lines,
    }
    summary = {
        "stream_count": len(result.get("data", {}).get("result") or []),
        "line_count": total_lines,
        "pattern_count": len(log_counts),
        "top_patterns": log_counts[: min(len(log_counts), args.sample_limit)],
    }
    return _ok("logs", "log", args.cluster_id, summary=summary, payload=payload)


def run_activity_events(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    search = api.search_cluster_exact(args.cluster_id)
    if search.get("error"):
        return _error("activity-events", "event", args.cluster_id, search["error"])
    item = search.get("item") or {}
    org_id = str(args.org_id or item.get("orgID") or item.get("orgId") or "")
    if not org_id:
        return _error("activity-events", "event", args.cluster_id, "orgID missing from cluster lookup")

    result, target_cluster_id, fallback_used, fallback_error = _activity_events_with_premium_fallback(
        api=api,
        org_id=org_id,
        cluster_id=args.cluster_id,
        cluster_item=item,
        start=args.start,
        end=args.end,
        name=args.name or "",
        severity=args.severity,
        limit=args.limit,
        cursor=args.cursor or "",
    )

    if result.get("error"):
        if fallback_error:
            return _error(
                "activity-events",
                "event",
                args.cluster_id,
                f"{result['error']} (premium parent fallback also failed; original error: {fallback_error})",
            )
        return _error("activity-events", "event", args.cluster_id, result["error"])
    events = [item for item in _first_list(result, ("activities", "events", "items")) if isinstance(item, dict)]
    payload = {
        "events": events,
        "time_window": {"start": args.start, "end": args.end},
        "requested_cluster_id": args.cluster_id,
        "target_cluster_id": target_cluster_id,
    }
    summary = {
        "event_count": len(events),
        "partial": result.get("partial"),
        "cursor": result.get("cursor"),
        "requested_cluster_id": args.cluster_id,
        "target_cluster_id": target_cluster_id,
        "fallback_used": fallback_used,
    }
    if fallback_used:
        summary["fallback_reason"] = "premium_activityhub_parent_target"
        summary["fallback_original_error"] = fallback_error
        payload["fallback"] = {
            "reason": "premium_activityhub_parent_target",
            "original_error": fallback_error,
        }
    return _ok(
        "activity-events",
        "event",
        args.cluster_id,
        summary=summary,
        payload=payload,
    )


_BYOC_PROFILE_WINDOW_SECONDS = 10_799


def _byoc_continuous_profiling_inventory(
    api: Any, cluster_id: str, start: int, end: int,
) -> dict[str, Any]:
    """Fetch a BYOC profile inventory without exceeding the API window limit."""
    profiles: list[dict[str, Any]] = []
    seen: set[str] = set()
    cursor = start
    while cursor <= end:
        window_end = min(cursor + _BYOC_PROFILE_WINDOW_SECONDS, end)
        result = api.byoc_continuous_profiling_group_profiles(
            cluster_id, cursor, window_end,
        )
        error = _api_error(result)
        if error:
            return {"error": error}
        for profile in _first_list(
            result, ("target_profiles", "group_profiles", "profiles", "items"),
        ):
            if not isinstance(profile, dict):
                continue
            key = json.dumps(profile, sort_keys=True, default=str)
            if key not in seen:
                seen.add(key)
                profiles.append(profile)
        cursor = window_end + 1
    return {"group_profiles": profiles}


def run_continuous_profiling(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    metadata, metadata_error = _canonical_cluster_metadata(api, args.cluster_id)
    if metadata_error:
        return _error(
            "continuous-profiling",
            "profile",
            args.cluster_id,
            f"could not resolve cluster tier: {metadata_error}",
        )
    is_byoc = bool(metadata) and _metadata_is_byoc(api, metadata)
    target_cluster_id = args.cluster_id
    if is_byoc:
        try:
            target_cluster_id = api.byoc_physical_cluster_id(args.cluster_id, metadata)
        except Exception as exc:  # noqa: BLE001 - stable CLI failure.
            return _error("continuous-profiling", "profile", args.cluster_id, str(exc))
        if args.ts is not None:
            result = api.byoc_continuous_profiling_group_detail(target_cluster_id, args.ts)
            command = "continuous-profiling-detail"
            time_window = None
        else:
            result = _byoc_continuous_profiling_inventory(
                api, target_cluster_id, args.start, args.end,
            )
            command = "continuous-profiling"
            time_window = {"start": args.start, "end": args.end}
    elif args.ts is not None:
        result = api.continuous_profiling_detail(args.cluster_id, args.ts)
        command = "continuous-profiling-detail"
        time_window = None
    else:
        result = api.continuous_profiling_group_profiles(args.cluster_id, args.start, args.end)
        command = "continuous-profiling"
        time_window = {"start": args.start, "end": args.end}
    error = _api_error(result)
    if error:
        return _error(command, "profile", args.cluster_id, error)
    profiles = [
        item for item in _first_list(result, ("target_profiles", "group_profiles", "profiles", "items"))
        if isinstance(item, dict)
    ]
    payload: dict[str, Any] = {
        "profiles": profiles,
        "requested_cluster_id": args.cluster_id,
        "target_cluster_id": target_cluster_id,
        "inventory_only": is_byoc,
    }
    if time_window is not None:
        payload["time_window"] = time_window
    if args.ts is not None:
        payload["ts"] = args.ts
    return _ok(
        command,
        "profile",
        args.cluster_id,
        summary={
            "profile_count": len(profiles),
            "target_cluster_id": target_cluster_id,
            "inventory_only": is_byoc,
        },
        payload=payload,
    )


def run_continuous_profiling_download(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    metadata, metadata_error = _canonical_cluster_metadata(api, args.cluster_id)
    if metadata_error:
        return _error(
            "continuous-profiling-download",
            "profile",
            args.cluster_id,
            f"could not resolve cluster tier: {metadata_error}",
        )
    if metadata and _metadata_is_byoc(api, metadata):
        return _error(
            "continuous-profiling-download",
            "profile",
            args.cluster_id,
            "BYOC continuous profiling is inventory-only; token and profile downloads are not supported",
        )
    output_dir = _workspace_download_dir("profiles")
    base_filename = _sanitize_filename(
        f"pprof-{args.ts}-{args.component}-{args.address}-{args.profile_type}.pb",
        fallback=f"pprof-{args.ts}.pb",
    )
    filename = _filename_with_identity_hash(
        base_filename,
        args.cluster_id,
        args.ts,
        args.component,
        args.address,
        args.profile_type,
    )
    output_path = output_dir / filename
    token_result = api.continuous_profiling_action_token(
        args.cluster_id,
        ts=args.ts,
        profile_type=args.profile_type,
        component=args.component,
        address=args.address,
    )
    if token_result.get("error"):
        return _error("continuous-profiling-download", "profile", args.cluster_id, token_result["error"])
    token = str(token_result.get("text") or "")
    if not token:
        return _error("continuous-profiling-download", "profile", args.cluster_id, "empty continuous profiling download token")
    download = api.continuous_profiling_download(args.cluster_id, token=token, output_path=output_path)
    if download.get("error"):
        return _error("continuous-profiling-download", "profile", args.cluster_id, download["error"])
    extracted_files = _safe_unzip(output_path, _download_extract_dir(output_path)) if args.unzip else []
    profile = {
        "ts": args.ts,
        "profile_type": args.profile_type,
        "component": args.component,
        "address": args.address,
        "file_path": str(output_path),
        "file_size_bytes": download.get("size_bytes"),
        "extracted_files": extracted_files,
    }
    return _ok(
        "continuous-profiling-download",
        "profile",
        args.cluster_id,
        summary={"downloaded": True, "file_path": str(output_path), "extracted_file_count": len(extracted_files)},
        payload={"profiles": [profile]},
    )


def run_diagnostic_files(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    metadata, metadata_error = _canonical_cluster_metadata(api, args.cluster_id)
    if metadata_error:
        return _error("diagnostic-files", "file", args.cluster_id, f"could not resolve cluster tier: {metadata_error}")
    if metadata and _metadata_is_byoc(api, metadata):
        return _error(
            "diagnostic-files",
            "file",
            args.cluster_id,
            "Plan Replayer and OOM diagnostic files are not supported for BYOC clusters",
        )
    result = api.diagnostic_files(args.cluster_id, args.file_type, args.start, args.end)
    if result.get("error"):
        return _error("diagnostic-files", "file", args.cluster_id, result["error"])
    files = _flatten_diagnostic_files(result, file_type=args.file_type)
    payload = {
        "files": files,
        "time_window": {"start": args.start, "end": args.end},
    }
    return _ok(
        "diagnostic-files",
        "file",
        args.cluster_id,
        file_type=args.file_type,
        summary={"file_count": len(files)},
        payload=payload,
    )


def run_diagnostic_file_download(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    metadata, metadata_error = _canonical_cluster_metadata(api, args.cluster_id)
    if metadata_error:
        return _error("diagnostic-file-download", "file", args.cluster_id, f"could not resolve cluster tier: {metadata_error}")
    if metadata and _metadata_is_byoc(api, metadata):
        return _error(
            "diagnostic-file-download",
            "file",
            args.cluster_id,
            "Plan Replayer and OOM diagnostic file downloads are not supported for BYOC clusters",
        )
    output_dir = _workspace_download_dir("files")
    base_filename = _sanitize_filename(args.filename or args.key, fallback="diagnostic-file.bin")
    filename = _filename_with_identity_hash(
        base_filename,
        args.cluster_id,
        args.file_type,
        args.key,
        args.provider or "",
        args.region or "",
    )
    output_path = output_dir / filename
    download = api.diagnostic_file_download(
        args.cluster_id,
        key=args.key,
        output_path=output_path,
        provider=args.provider or "",
        region=args.region or "",
    )
    if download.get("error"):
        return _error("diagnostic-file-download", "file", args.cluster_id, download["error"])
    extracted_files = _safe_unzip(output_path, _download_extract_dir(output_path)) if args.unzip else []
    file_row = {
        "kind": args.file_type,
        "key": args.key,
        "file_path": str(output_path),
        "file_size_bytes": download.get("size_bytes"),
        "extracted_files": extracted_files,
    }
    return _ok(
        "diagnostic-file-download",
        "file",
        args.cluster_id,
        file_type=args.file_type,
        summary={"downloaded": True, "file_path": str(output_path), "extracted_file_count": len(extracted_files)},
        payload={"files": [file_row]},
    )


def run_ad_hoc_store_roles(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    promql = 'pd_scheduler_store_status{type="store_capacity"}'
    result = api.query_range(args.cluster_id, promql, args.start, args.end, step=str(args.step))
    if result.get("error"):
        return _error("ad-hoc-store-roles", "metric", args.cluster_id, result["error"])
    if result.get("status") != "success":
        return _error("ad-hoc-store-roles", "metric", args.cluster_id, "store role query did not return success", raw_result=result)

    stores: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    for row_item in result.get("data", {}).get("result") or []:
        metric = row_item.get("metric") if isinstance(row_item, dict) and isinstance(row_item.get("metric"), dict) else {}
        address = str(metric.get("address") or metric.get("instance") or "")
        store = str(metric.get("store") or metric.get("store_id") or "")
        key = (store, address)
        if key in seen:
            continue
        seen.add(key)
        stores.append(
            {
                "store": store,
                "address": address,
                "component": _component_from_store_address(address),
                "labels": metric,
                "summary": _series_stat(row_item),
            }
        )
    stores.sort(key=lambda item: (item["component"], item["address"], item["store"]))
    row = _metric_row_from_result(
        name="pd_scheduler_store_status_store_roles",
        promql=promql,
        result=result,
        cluster_id=args.cluster_id,
        start=args.start,
        end=args.end,
        step=str(args.step),
        metric_group_id=args.metric_group_id or "store_roles",
        summary_extra={"stores": stores},
    )
    payload = {"metrics": [row], "time_range": _time_window(args.start, args.end, args.step)}
    return _ok(
        "ad-hoc-store-roles",
        "metric",
        args.cluster_id,
        summary={"store_count": len(stores), "stores": stores},
        metric_group_id=row["metric_group_id"],
        payload=payload,
    )


def run_ad_hoc_slow_query_top(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    dates = _date_partitions(args.start, args.end, fmt="%Y%m%d")
    order_map = {
        "max_query_time": "max_query_time",
        "exec_count": "exec_count",
        "total_process_time": "total_process_time",
        "total_wait_time": "total_wait_time",
    }
    order_by = order_map[args.order_by]
    sql = f"""
SELECT
  digest,
  any_value(query) AS sample_sql,
  COUNT(*) AS exec_count,
  MAX(query_time) AS max_query_time,
  AVG(query_time) AS avg_query_time,
  SUM(process_time) AS total_process_time,
  SUM(wait_time) AS total_wait_time
FROM clinic_data_proxy.slow_query_logs
WHERE date IN ({_sql_string_list(dates)})
  AND time >= {int(args.start)}
  AND time <= {int(args.end)}
GROUP BY digest
ORDER BY {order_by} DESC
LIMIT {int(args.limit)}
""".strip()
    namespace = argparse.Namespace(
        cluster_id=args.cluster_id,
        query_type="slow_queries",
        name=args.name or f"slow_query_top_by_{args.order_by}",
        sql=sql,
        timeout=args.timeout,
        sample_limit=args.limit,
    )
    result = run_sql(namespace, api=api)
    if result.get("status") == "ok":
        result["ad_hoc_sql"] = sql
        result["summary"]["date_partitions"] = dates
        result["summary"]["order_by"] = args.order_by
    return result


def run_ad_hoc_ttl_rows(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    window_seconds = max(1, int(args.end) - int(args.start))
    metrics = [
        ("ttl_insert_rows_delta", "tidb_server_ttl_insert_rows"),
        ("ttl_processed_expired_rows_delta", "tidb_server_ttl_processed_expired_rows"),
    ]
    rows: list[dict[str, Any]] = []
    signals: dict[str, int | None] = {}
    errors: list[dict[str, Any]] = []
    for field, metric_name in metrics:
        promql = f"sum(increase({metric_name}[{window_seconds}s]))"
        result = api.query_instant(args.cluster_id, promql, time=args.end)
        if result.get("error"):
            errors.append({"field": field, "metric": metric_name, "error": result["error"]})
            signals[field] = None
            continue
        value = _instant_scalar_value(result)
        signals[field] = max(0, int(round(value))) if value is not None else None
        rows.append(
            {
                **_metric_ref_fields(field),
                "metric_group_id": field,
                "name": field,
                "promql": promql,
                "unit": "rows",
                "summary": {
                    "value": signals[field],
                    "series_count": _metric_series_count(result),
                },
                "time_window": {"start": args.start, "end": args.end, "query_time": args.end},
                "chart_data_ref": {
                    "kind": "promql_instant",
                    "cluster_id": args.cluster_id,
                    "promql": promql,
                    "time": int(args.end),
                },
                "chart_data": {
                    "name": field,
                    "promql": promql,
                    "result_type": result.get("data", {}).get("resultType") or "vector",
                    "series": result.get("data", {}).get("result") or [],
                },
            }
        )

    evidence_sufficient = not errors and all(value is not None for value in signals.values())
    ttl_activity_present = bool(
        evidence_sufficient
        and any(int(value or 0) > 0 for value in signals.values())
    )
    payload = {"metrics": rows, "time_range": _time_window(args.start, args.end, "instant")}
    return _ok(
        "ad-hoc-ttl-rows",
        "metric",
        args.cluster_id,
        summary={
            "window": {"start": args.start, "end": args.end},
            "evidence_sufficient": evidence_sufficient,
            "ttl_activity_present": ttl_activity_present,
            "signals": signals,
            "errors": errors,
        },
        payload=payload,
    )


def run_ad_hoc_topsql_links(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    namespace = args.namespace or f"tidb{args.cluster_id}"
    promql = (
        f'sum by (instance) (rate(node_disk_reads_completed_total'
        f'{{instance=~"{args.instance_filter}",device!~"{args.device_exclude_regex}"}}'
        f'[{args.rate_window}]))'
    )
    result = api.query_range_join_instance_to_pod(
        args.cluster_id,
        promql,
        namespace,
        ".*tikv.*",
        args.start,
        args.end,
        step=str(args.step),
    )
    if result.get("error"):
        return _error("ad-hoc-topsql-links", "metric", args.cluster_id, result["error"])
    if result.get("status") != "success":
        return _error("ad-hoc-topsql-links", "metric", args.cluster_id, "read IOPS query did not return success", raw_result=result)

    ranked = _rank_metric_series(result, label_keys=("pod", "instance", "device"), limit=args.top_n, rank_by="avg")
    cluster_meta = _cluster_meta_for_links(api, args.cluster_id)
    topsql_links = []
    for item in ranked:
        labels = item.get("labels") if isinstance(item.get("labels"), dict) else {}
        tikv_node = str(labels.get("pod") or "")
        if not tikv_node:
            continue
        instance = _topsql_tikv_instance(args.cluster_id, tikv_node)
        topsql_links.append(
            {
                "tikv_node": tikv_node,
                "url": _topsql_url(args.cluster_id, args.start, args.end, cluster_meta=cluster_meta, instance=instance),
                "tikv_instance": instance,
                "avg_read_iops": item.get("avg"),
                "metric_labels": labels,
            }
        )

    row = _metric_row_from_result(
        name="tikv_read_iops_for_topsql_links",
        promql=f"client-side join: {promql} -> kube_pod_info namespace={namespace}",
        result=result,
        cluster_id=args.cluster_id,
        start=args.start,
        end=args.end,
        step=str(args.step),
        metric_group_id=args.metric_group_id or "tikv_read_iops_for_topsql_links",
        unit="ops/s",
        summary_extra={"top_series": ranked, "topsql_links": topsql_links},
    )
    payload = {"metrics": [row], "time_range": _time_window(args.start, args.end, args.step)}
    return _ok(
        "ad-hoc-topsql-links",
        "metric",
        args.cluster_id,
        summary={
            "mapping_strategy": "node_filter",
            "evidence_sufficient": len(topsql_links) >= args.top_n,
            "topsql_links": topsql_links,
            "top_series": ranked,
        },
        payload=payload,
    )


def run_ad_hoc_tidb_p99_breakdown(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    tidb_pattern = re.compile(r"\bdb-tidb-[A-Za-z0-9-]+\b")
    tikv_pattern = re.compile(r"\bdb-tikv-[A-Za-z0-9-]+\b")
    pd_pattern = re.compile(r"\bdb-pd-[A-Za-z0-9-]+\b")
    primary_promql = _histogram_p99_promql("tidb_server_handle_query_duration_seconds_bucket")
    primary = api.query_range(args.cluster_id, primary_promql, args.start, args.end, step=str(args.step))
    if primary.get("error"):
        return _error("ad-hoc-tidb-p99-breakdown", "metric", args.cluster_id, primary["error"])
    primary_values = _avg_metric_by_node(primary, tidb_pattern)
    top_pairs = _ranked_node_pairs(primary_values)[: args.top_n]
    top_nodes = [node for node, _ in top_pairs]

    metric_rows = [
        _metric_row_from_result(
            name="tidb_query_p99_latency",
            promql=primary_promql,
            result=primary,
            cluster_id=args.cluster_id,
            start=args.start,
            end=args.end,
            step=str(args.step),
            metric_group_id="tidb_query_p99_latency",
            unit="seconds",
        )
    ]
    breakdown_specs = [
        ("avg_plan_build_p99_s", "tidb_session_compile_duration_seconds_bucket", ()),
        ("avg_distributed_execution_p99_s", "tidb_distsql_handle_query_duration_seconds_bucket", ()),
        ("avg_kv_request_p99_s", "tidb_tikvclient_request_seconds_bucket", ('type="Get"',)),
        ("avg_get_tso_wait_p99_s", "pd_client_request_handle_requests_duration_seconds_bucket", ('type="tso"',)),
        ("avg_gc_stw_pause_p99_s", "go_gc_pauses_seconds_bucket", ('component="tidb"',)),
    ]
    breakdown_values: dict[str, dict[str, float]] = {}
    missing_signals: list[str] = []
    for field, metric_name, extra_matchers in breakdown_specs:
        promql = _histogram_p99_promql(metric_name, nodes=top_nodes, extra_matchers=extra_matchers)
        result = api.query_range(args.cluster_id, promql, args.start, args.end, step=str(args.step))
        values = _avg_metric_by_node(result, tidb_pattern)
        breakdown_values[field] = values
        if not values:
            missing_signals.append(field)
        metric_rows.append(
            _metric_row_from_result(
                name=field,
                promql=promql,
                result=result,
                cluster_id=args.cluster_id,
                start=args.start,
                end=args.end,
                step=str(args.step),
                metric_group_id=field,
                unit="seconds",
            )
        )

    cluster_specs = [
        ("cluster_avg_tikv_grpc_request_handle_p99_s", "tikv_grpc_msg_duration_seconds_bucket", ('type="kv_get"',), tikv_pattern),
        ("cluster_avg_pd_tso_request_handle_p99_s", "pd_server_handle_tso_duration_seconds_bucket", (), pd_pattern),
    ]
    cluster_values: dict[str, float | None] = {}
    for field, metric_name, extra_matchers, pattern in cluster_specs:
        promql = _histogram_p99_promql(metric_name, extra_matchers=extra_matchers)
        result = api.query_range(args.cluster_id, promql, args.start, args.end, step=str(args.step))
        values = _avg_metric_by_node(result, pattern)
        cluster_values[field] = (sum(values.values()) / len(values)) if values else None
        if cluster_values[field] is None:
            missing_signals.append(field)
        metric_rows.append(
            _metric_row_from_result(
                name=field,
                promql=promql,
                result=result,
                cluster_id=args.cluster_id,
                start=args.start,
                end=args.end,
                step=str(args.step),
                metric_group_id=field,
                unit="seconds",
            )
        )

    evidence_sufficient = len(top_pairs) >= args.top_n and not missing_signals and all(
        node in breakdown_values[field]
        for node, _ in top_pairs
        for field, _, _ in breakdown_specs
    )
    top_nodes_by_avg_p99_latency = []
    if evidence_sufficient:
        for node, value in top_pairs:
            row = {"node": node, "avg_p99_latency_s": round(value, args.precision)}
            for field, _, _ in breakdown_specs:
                row[field] = round(breakdown_values[field][node], args.precision)
            top_nodes_by_avg_p99_latency.append(row)

    payload = {"metrics": metric_rows, "time_range": _time_window(args.start, args.end, args.step)}
    return _ok(
        "ad-hoc-tidb-p99-breakdown",
        "metric",
        args.cluster_id,
        summary={
            "window": {"start": args.start, "end": args.end},
            "evidence_sufficient": evidence_sufficient,
            "missing_signals": sorted(set(missing_signals)),
            "cluster_avg_tikv_grpc_request_handle_p99_s": (
                round(cluster_values["cluster_avg_tikv_grpc_request_handle_p99_s"], args.precision)
                if cluster_values["cluster_avg_tikv_grpc_request_handle_p99_s"] is not None
                else None
            ),
            "cluster_avg_pd_tso_request_handle_p99_s": (
                round(cluster_values["cluster_avg_pd_tso_request_handle_p99_s"], args.precision)
                if cluster_values["cluster_avg_pd_tso_request_handle_p99_s"] is not None
                else None
            ),
            "top_nodes_by_avg_p99_latency": top_nodes_by_avg_p99_latency,
        },
        payload=payload,
    )


def run_ad_hoc_logs(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    query = f'{{container=~"{args.component_regex}"}} |~ `{args.regex}`'
    namespace = argparse.Namespace(
        cluster_id=args.cluster_id,
        query=query,
        start=args.start,
        end=args.end,
        limit=args.limit,
        direction=args.direction,
        sample_limit=args.sample_limit,
    )
    result = run_logs(namespace, api=api)
    if result.get("status") == "ok":
        result["ad_hoc_logql"] = query
    return result


def run_ad_hoc_cpu_logs(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    promql = (
        f'sum(rate(process_cpu_seconds_total{{job=~"{args.job_regex}"}}[{args.rate_window}])) '
        "by (job,instance)"
    )
    metric_result = api.query_range(args.cluster_id, promql, args.start, args.end, step=str(args.step))
    if metric_result.get("error"):
        return _error("ad-hoc-cpu-logs", "metric", args.cluster_id, metric_result["error"])
    if metric_result.get("status") != "success":
        return _error("ad-hoc-cpu-logs", "metric", args.cluster_id, "cpu query did not return success", raw_result=metric_result)

    top_cpu = _rank_metric_series(metric_result, label_keys=("job", "instance"), limit=args.limit)
    metric_row = _metric_row_from_result(
        name="top_component_cpu",
        promql=promql,
        result=metric_result,
        cluster_id=args.cluster_id,
        start=args.start,
        end=args.end,
        step=str(args.step),
        metric_group_id=args.metric_group_id or "top_component_cpu",
        unit="cores",
        summary_extra={"top_series": top_cpu},
    )

    log_query = f'{{container=~"{args.log_component_regex}"}} |~ `{args.log_regex}`'
    log_args = argparse.Namespace(
        cluster_id=args.cluster_id,
        query=log_query,
        start=args.start,
        end=args.end,
        limit=args.log_limit,
        direction="backward",
        sample_limit=args.sample_limit,
    )
    log_result = run_logs(log_args, api=api)
    if log_result.get("status") != "ok":
        return log_result

    metric_payload = {"metrics": [metric_row], "time_range": _time_window(args.start, args.end, args.step)}
    metric_persisted = _persist_raw_evidence("ad-hoc-cpu-logs", "metric", args.cluster_id, metric_payload)
    return {
        "status": "ok",
        "contract_version": CONTRACT_VERSION,
        "command": "ad-hoc-cpu-logs",
        "kind": "ad_hoc",
        "cluster_id": args.cluster_id,
        "summary": {
            "top_cpu": top_cpu,
            "log_summary": log_result.get("summary", {}),
            "log_query": log_query,
        },
        "persisted_evidence": {
            "metrics": metric_persisted,
            "logs": log_result.get("persisted_evidence"),
        },
    }


def run_ad_hoc_cpu_overload_logs(args: argparse.Namespace, api: Any | None = None) -> dict[str, Any]:
    api = api or build_api()
    metric_rows: list[dict[str, Any]] = []
    ranked_nodes: list[dict[str, Any]] = []
    nodes_by_component: dict[str, set[str]] = {}
    debug: dict[str, Any] = {"components": {}}

    for component in _CPU_COMPONENT_ORDER:
        rows, selected, component_debug = _cpu_overload_rows(
            api,
            cluster_id=args.cluster_id,
            component=component,
            start=args.start,
            end=args.end,
            step=str(args.step),
            top_n=args.top_n_per_component,
        )
        metric_rows.extend(rows)
        debug["components"][component] = component_debug
        if component_debug.get("error"):
            return _error("ad-hoc-cpu-overload-logs", "metric", args.cluster_id, component_debug["error"], component=component)
        nodes_by_component[component] = {str(item["node"]) for item in selected}
        ranked_nodes.extend(selected)

    log_counts_by_node, log_counts, log_debug = _count_error_warn_logs_for_nodes(
        api,
        cluster_id=args.cluster_id,
        nodes_by_component=nodes_by_component,
        start=args.start,
        end=args.end,
        log_regex=args.log_regex,
        limit=args.log_limit,
        slice_seconds=args.log_slice_seconds,
    )
    debug["log_queries"] = log_debug

    final_ranked = []
    for component in _CPU_COMPONENT_ORDER:
        rows = [item for item in ranked_nodes if item["component"] == component]
        rows.sort(key=lambda item: int(item["rank"]))
        for item in rows:
            final_ranked.append(
                {
                    "component": component,
                    "rank": int(item["rank"]),
                    "node": str(item["node"]),
                    "avg_cpu_overload": float(item["avg_cpu_overload"]),
                    "error_warn_log_count": int(log_counts_by_node.get(str(item["node"]), 0)),
                    "avg_cpu_cores": item.get("avg_cpu_cores"),
                    "cpu_quota": item.get("cpu_quota"),
                }
            )

    metric_payload = {"metrics": metric_rows, "time_range": _time_window(args.start, args.end, args.step)}
    log_payload = {
        "time_window": {"start_ns": _to_ns(args.start), "end_ns": _to_ns(args.end)},
        "query": "per-node ERROR/WARN selectors",
        "log_counts": log_counts,
        "samples": log_counts[: args.sample_limit],
    }
    metric_persisted = _persist_raw_evidence("ad-hoc-cpu-overload-logs", "metric", args.cluster_id, metric_payload)
    log_persisted = _persist_raw_evidence("ad-hoc-cpu-overload-logs", "log", args.cluster_id, log_payload)
    return {
        "status": "ok",
        "contract_version": CONTRACT_VERSION,
        "command": "ad-hoc-cpu-overload-logs",
        "kind": "ad_hoc",
        "cluster_id": args.cluster_id,
        "summary": {
            "window": {"start": args.start, "end": args.end},
            "evidence_sufficient": all(
                len(nodes_by_component.get(component, set())) >= args.top_n_per_component
                for component in _CPU_COMPONENT_ORDER
            ),
            "ranked_nodes": final_ranked,
            "debug": debug,
        },
        "persisted_evidence": {
            "metrics": metric_persisted,
            "logs": log_persisted,
        },
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Stable Clinic CLI wrappers")
    sub = parser.add_subparsers(dest="command", required=True)

    search = sub.add_parser("search", help="Clinic search helpers")
    search_sub = search.add_subparsers(dest="search_target", required=True)
    search_clusters = search_sub.add_parser("clusters", help="search cluster list")
    search_clusters.add_argument("--query", default="", help="fuzzy search text: tenant name/id, cluster id, or cluster name")
    search_clusters.add_argument("--tenant-id", default="", help="exact user-facing tenantID filter")
    search_clusters.add_argument("--cluster-id", default="", help="exact clusterID filter")
    search_clusters.add_argument("--cluster-status", default="", help="optional Clinic cluster_status filter, for example active or deleted")
    search_clusters.add_argument("--show-deleted", action="store_true")
    search_clusters.add_argument("--limit", default=20, type=int)
    search_clusters.add_argument("--page", default=1, type=int)

    cluster = sub.add_parser("cluster-metadata", help="exact cluster metadata lookup")
    cluster.add_argument("--cluster-id", required=True)
    cluster.add_argument("--include-components", action="store_true")
    cluster.add_argument(
        "--resource-pool-scope",
        choices=["auto", "current", "parent"],
        default="current",
        help="topology cluster id to query; auto is kept as an alias of current",
    )

    topology_snapshot = sub.add_parser("topology-snapshot", help="point-in-time topology evidence from metrics plus Activity Hub changes")
    topology_snapshot.add_argument("--cluster-id", required=True)
    topology_snapshot.add_argument("--at", required=True, type=_parse_time_to_epoch_seconds, help="target timestamp")
    topology_snapshot.add_argument("--org-id", default="", help="Clinic internal orgID override")
    topology_snapshot.add_argument("--metric-window-minutes", default=5, type=int)
    topology_snapshot.add_argument("--activity-before-days", default=30, type=int)
    topology_snapshot.add_argument("--activity-after-days", default=1, type=int)
    topology_snapshot.add_argument("--severity", default=0, type=int)
    topology_snapshot.add_argument("--limit", default=100, type=int)
    topology_snapshot.add_argument("--step", default="60")

    metrics = sub.add_parser("metrics", help="PromQL range query")
    metrics.add_argument("--cluster-id", required=True)
    metrics.add_argument("--name", required=True)
    metrics.add_argument("--query", required=True)
    metrics.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    metrics.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    metrics.add_argument("--step", default="60")
    metrics.add_argument("--metric-group-id")
    metrics.add_argument("--unit", default="")
    metrics.add_argument("--inline-chart-data", action="store_true")

    sql = sub.add_parser("sql", help="Data Proxy SQL query")
    sql.add_argument("--cluster-id", required=True)
    sql.add_argument("--query-type", required=True, choices=["slow_queries", "topsql"])
    sql.add_argument("--name", required=True)
    sql.add_argument("--sql", required=True)
    sql.add_argument("--timeout", default=60, type=int)
    sql.add_argument("--sample-limit", default=3, type=int)

    slow_query = sub.add_parser("slow-query", help="Tier-aware Slow Query operations")
    slow_query_sub = slow_query.add_subparsers(dest="slow_query_operation", required=True)
    slow_fields = slow_query_sub.add_parser("available-fields", help="List available BYOC Slow Query fields")
    slow_fields.add_argument("--cluster-id", required=True)
    slow_list = slow_query_sub.add_parser("list", help="List BYOC Slow Query rows")
    slow_list.add_argument("--cluster-id", required=True)
    slow_list.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    slow_list.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    slow_list.add_argument("--limit", default=100, type=int)
    slow_list.add_argument(
        "--order-by",
        default="timestamp",
        choices=["timestamp", "query_time", "memory_max", "total_keys", "process_keys"],
    )
    slow_list.add_argument("--desc", action="store_true")
    slow_list.add_argument("--fields", default="query,timestamp,query_time,memory_max")
    slow_list.add_argument("--text")
    slow_list.add_argument("--digest")
    slow_list.add_argument("--show-internal", action="store_true")
    slow_list.add_argument("--db", action="append", default=[])
    slow_list.add_argument("--plan", action="append", default=[])
    slow_detail = slow_query_sub.add_parser("detail", help="Fetch one BYOC Slow Query detail")
    slow_detail.add_argument("--cluster-id", required=True)
    slow_detail.add_argument("--digest", required=True)
    slow_detail.add_argument("--timestamp", required=True)
    slow_detail.add_argument("--connect-id", required=True)
    slow_detail.add_argument("--show-internal", action="store_true")

    top_sql = sub.add_parser("top-sql", help="Tier-aware Top SQL operations")
    top_sql_sub = top_sql.add_subparsers(dest="top_sql_operation", required=True)
    top_instances = top_sql_sub.add_parser("instances", help="List BYOC Top SQL instances")
    top_instances.add_argument("--cluster-id", required=True)
    top_instances.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    top_instances.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    top_summary = top_sql_sub.add_parser("summary", help="Fetch BYOC Top SQL summary")
    top_summary.add_argument("--cluster-id", required=True)
    top_summary.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    top_summary.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    top_summary.add_argument("--window", default="1m")
    top_summary.add_argument("--top", default=20, type=int)
    top_summary.add_argument("--instance", required=True)
    top_summary.add_argument("--instance-type", required=True, choices=["tidb", "tikv", "pd"])

    statements = sub.add_parser("statements", help="Tier-aware live Statements operations")
    statements_sub = statements.add_subparsers(dest="statements_operation", required=True)
    statement_fields = statements_sub.add_parser("available-fields", help="List BYOC Statements fields")
    statement_fields.add_argument("--cluster-id", required=True)
    statement_types = statements_sub.add_parser("statement-types", help="List BYOC statement types")
    statement_types.add_argument("--cluster-id", required=True)
    statement_list = statements_sub.add_parser("list", help="List BYOC live Statements")
    statement_list.add_argument("--cluster-id", required=True)
    statement_list.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    statement_list.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    statement_list.add_argument("--schema", action="append", default=[])
    statement_list.add_argument("--statement-type", action="append", default=[])
    statement_list.add_argument("--text")
    statement_list.add_argument("--fields")
    statement_plans = statements_sub.add_parser("plans", help="List plans for a BYOC SQL digest")
    statement_plans.add_argument("--cluster-id", required=True)
    statement_plans.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    statement_plans.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    statement_plans.add_argument("--digest", required=True)
    statement_plans.add_argument("--schema-name")
    statement_plan_detail = statements_sub.add_parser("plan-detail", help="Fetch BYOC Statements plan detail")
    statement_plan_detail.add_argument("--cluster-id", required=True)
    statement_plan_detail.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    statement_plan_detail.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    statement_plan_detail.add_argument("--digest", required=True)
    statement_plan_detail.add_argument("--plan", action="append", default=[])
    statement_plan_detail.add_argument("--schema-name")

    ticdc = sub.add_parser("ticdc", help="TiCDC changefeed metadata operations")
    ticdc_sub = ticdc.add_subparsers(dest="ticdc_operation", required=True)
    ticdc_list = ticdc_sub.add_parser("list", help="List changefeeds for a cluster")
    ticdc_list.add_argument("--cluster-id", required=True)
    ticdc_list.add_argument("--org-id", default="", help="canonical DBAAS org ID override")
    ticdc_detail = ticdc_sub.add_parser("detail", help="Get one changefeed by kernel/internal ID")
    ticdc_detail.add_argument("--cluster-id", required=True)
    ticdc_detail.add_argument("--changefeed-id", required=True, help="kernel/internal ID returned by ticdc list")
    ticdc_detail.add_argument("--org-id", default="", help="canonical DBAAS org ID override")

    logs = sub.add_parser("logs", help="Loki range query")
    logs.add_argument("--cluster-id", required=True)
    logs.add_argument("--query", required=True)
    logs.add_argument("--start", required=True, type=_parse_time_to_epoch_or_ns, help="epoch seconds/ms/ns or ISO8601")
    logs.add_argument("--end", required=True, type=_parse_time_to_epoch_or_ns, help="epoch seconds/ms/ns or ISO8601")
    logs.add_argument("--limit", default=200, type=int)
    logs.add_argument("--direction", default="backward", choices=["backward", "forward"])
    logs.add_argument("--sample-limit", default=5, type=int)

    events = sub.add_parser("activity-events", help="Activity Hub events")
    events.add_argument("--cluster-id", required=True)
    events.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    events.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    events.add_argument("--org-id", default="", help="Clinic internal orgID override")
    events.add_argument("--name", default="")
    events.add_argument("--severity", default=0, type=int)
    events.add_argument("--limit", default=100, type=int)
    events.add_argument("--cursor", default="")

    profiles = sub.add_parser("continuous-profiling", help="Continuous profiling metadata")
    profiles.add_argument("--cluster-id", required=True)
    profiles.add_argument("--start", type=_parse_time_to_epoch_seconds)
    profiles.add_argument("--end", type=_parse_time_to_epoch_seconds)
    profiles.add_argument("--ts", type=_parse_time_to_epoch_seconds, help="when set, query profile detail for this snapshot")

    profile_download = sub.add_parser(
        "continuous-profiling-download",
        help="Download continuous profiling blob",
    )
    profile_download.add_argument("--cluster-id", required=True)
    profile_download.add_argument("--ts", required=True, type=_parse_time_to_epoch_seconds)
    profile_download.add_argument("--profile-type", required=True, choices=["cpu", "heap", "mutex", "goroutine"])
    profile_download.add_argument("--component", required=True)
    profile_download.add_argument("--address", required=True)
    profile_download.add_argument("--unzip", action="store_true")

    files = sub.add_parser("diagnostic-files", help="Diagnostic file list")
    files.add_argument("--cluster-id", required=True)
    files.add_argument("--file-type", required=True, choices=["plan_replayer", "oom_record"])
    files.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    files.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)

    file_download = sub.add_parser(
        "diagnostic-file-download",
        help="Download a diagnostic file",
    )
    file_download.add_argument("--cluster-id", required=True)
    file_download.add_argument("--key", required=True)
    file_download.add_argument("--file-type", required=True, choices=["plan_replayer", "oom_record", "diagnostic"])
    file_download.add_argument("--filename")
    file_download.add_argument("--provider")
    file_download.add_argument("--region")
    file_download.add_argument("--unzip", action="store_true")

    ad_hoc = sub.add_parser("ad-hoc", help="Stable recipes for common ad-hoc Clinic queries")
    ad_hoc_sub = ad_hoc.add_subparsers(dest="ad_hoc_target", required=True)

    store_roles = ad_hoc_sub.add_parser("store-roles", help="Map PD stores to TiKV/TiFlash roles")
    store_roles.add_argument("--cluster-id", required=True)
    store_roles.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    store_roles.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    store_roles.add_argument("--step", default="60")
    store_roles.add_argument("--metric-group-id", default="")

    slow_top = ad_hoc_sub.add_parser("slow-query-top", help="Top slow query digests for a bounded window")
    slow_top.add_argument("--cluster-id", required=True)
    slow_top.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    slow_top.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    slow_top.add_argument("--order-by", default="max_query_time", choices=["max_query_time", "exec_count", "total_process_time", "total_wait_time"])
    slow_top.add_argument("--limit", default=20, type=int)
    slow_top.add_argument("--timeout", default=60, type=int)
    slow_top.add_argument("--name", default="")

    logs_recipe = ad_hoc_sub.add_parser("logs", help="Focused Loki log pattern aggregation")
    logs_recipe.add_argument("--cluster-id", required=True)
    logs_recipe.add_argument("--start", required=True, type=_parse_time_to_epoch_or_ns)
    logs_recipe.add_argument("--end", required=True, type=_parse_time_to_epoch_or_ns)
    logs_recipe.add_argument("--component-regex", default="tidb|tikv|pd|tiflash")
    logs_recipe.add_argument("--regex", default=r"(?i)panic|fatal|\[error\]|error|timeout|unavailable|server is busy|oom")
    logs_recipe.add_argument("--limit", default=1000, type=int)
    logs_recipe.add_argument("--direction", default="backward", choices=["backward", "forward"])
    logs_recipe.add_argument("--sample-limit", default=10, type=int)

    ttl_rows = ad_hoc_sub.add_parser("ttl-rows", help="TTL row processing counters for a bounded window")
    ttl_rows.add_argument("--cluster-id", required=True)
    ttl_rows.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    ttl_rows.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)

    topsql_links = ad_hoc_sub.add_parser("topsql-links", help="Top TiKV read IOPS nodes and Clinic TopSQL links")
    topsql_links.add_argument("--cluster-id", required=True)
    topsql_links.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    topsql_links.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    topsql_links.add_argument("--step", default="60")
    topsql_links.add_argument("--rate-window", default="60s")
    topsql_links.add_argument("--instance-filter", default=".*")
    topsql_links.add_argument("--device-exclude-regex", default="dm.*")
    topsql_links.add_argument("--namespace", default="")
    topsql_links.add_argument("--top-n", default=2, type=int)
    topsql_links.add_argument("--metric-group-id", default="")

    tidb_p99 = ad_hoc_sub.add_parser("tidb-p99-breakdown", help="Top TiDB P99 latency nodes with required breakdowns")
    tidb_p99.add_argument("--cluster-id", required=True)
    tidb_p99.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    tidb_p99.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    tidb_p99.add_argument("--step", default="60")
    tidb_p99.add_argument("--top-n", default=3, type=int)
    tidb_p99.add_argument("--precision", default=6, type=int)

    cpu_logs = ad_hoc_sub.add_parser("cpu-logs", help="Top component CPU plus focused log patterns")
    cpu_logs.add_argument("--cluster-id", required=True)
    cpu_logs.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    cpu_logs.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    cpu_logs.add_argument("--step", default="60")
    cpu_logs.add_argument("--rate-window", default="1m")
    cpu_logs.add_argument("--job-regex", default="tidb|tikv|pd|tiflash")
    cpu_logs.add_argument("--log-component-regex", default="tidb|tikv|pd|tiflash")
    cpu_logs.add_argument("--log-regex", default=r"(?i)panic|fatal|\[error\]|error|timeout|unavailable|server is busy|oom")
    cpu_logs.add_argument("--limit", default=10, type=int)
    cpu_logs.add_argument("--log-limit", default=1000, type=int)
    cpu_logs.add_argument("--sample-limit", default=10, type=int)
    cpu_logs.add_argument("--metric-group-id", default="")

    cpu_overload_logs = ad_hoc_sub.add_parser(
        "cpu-overload-logs",
        help="Top tidb/tikv/pd CPU-overload nodes plus exact ERROR/WARN log counts",
    )
    cpu_overload_logs.add_argument("--cluster-id", required=True)
    cpu_overload_logs.add_argument("--start", required=True, type=_parse_time_to_epoch_seconds)
    cpu_overload_logs.add_argument("--end", required=True, type=_parse_time_to_epoch_seconds)
    cpu_overload_logs.add_argument("--step", default="60")
    cpu_overload_logs.add_argument("--top-n-per-component", default=3, type=int)
    cpu_overload_logs.add_argument("--log-regex", default=r"(?i)\[error\]|\[warn(?:ing)?\]")
    cpu_overload_logs.add_argument("--log-limit", default=1000, type=int)
    cpu_overload_logs.add_argument("--log-slice-seconds", default=60, type=int)
    cpu_overload_logs.add_argument("--sample-limit", default=10, type=int)
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    command_id = f"cli-{int(time.time() * 1000)}-{os.getpid()}"
    command_t0 = time.monotonic()
    try:
        _ensure_perf_log()
        _perf_event({
            "event": "cli_start",
            "command_id": command_id,
            **_command_perf_payload(args),
        })
    except Exception:
        pass
    try:
        if args.command == "search" and args.search_target == "clusters":
            result = run_search_clusters(args)
        elif args.command == "cluster-metadata":
            result = run_cluster_metadata(args)
        elif args.command == "topology-snapshot":
            result = run_topology_snapshot(args)
        elif args.command == "metrics":
            result = run_metrics(args)
        elif args.command == "sql":
            result = run_sql(args)
        elif args.command == "slow-query":
            result = run_slow_query(args)
        elif args.command == "top-sql":
            result = run_top_sql(args)
        elif args.command == "statements":
            result = run_statements(args)
        elif args.command == "ticdc":
            result = run_ticdc(args)
        elif args.command == "logs":
            result = run_logs(args)
        elif args.command == "activity-events":
            result = run_activity_events(args)
        elif args.command == "continuous-profiling":
            if args.ts is None and (args.start is None or args.end is None):
                parser.error("continuous-profiling requires --start/--end unless --ts is set")
                return 2
            result = run_continuous_profiling(args)
        elif args.command == "continuous-profiling-download":
            result = run_continuous_profiling_download(args)
        elif args.command == "diagnostic-files":
            result = run_diagnostic_files(args)
        elif args.command == "diagnostic-file-download":
            result = run_diagnostic_file_download(args)
        elif args.command == "ad-hoc":
            if args.ad_hoc_target == "store-roles":
                result = run_ad_hoc_store_roles(args)
            elif args.ad_hoc_target == "slow-query-top":
                result = run_ad_hoc_slow_query_top(args)
            elif args.ad_hoc_target == "logs":
                result = run_ad_hoc_logs(args)
            elif args.ad_hoc_target == "ttl-rows":
                result = run_ad_hoc_ttl_rows(args)
            elif args.ad_hoc_target == "topsql-links":
                result = run_ad_hoc_topsql_links(args)
            elif args.ad_hoc_target == "tidb-p99-breakdown":
                result = run_ad_hoc_tidb_p99_breakdown(args)
            elif args.ad_hoc_target == "cpu-logs":
                result = run_ad_hoc_cpu_logs(args)
            elif args.ad_hoc_target == "cpu-overload-logs":
                result = run_ad_hoc_cpu_overload_logs(args)
            else:
                parser.error(f"unsupported ad-hoc target: {args.ad_hoc_target}")
                return 2
        else:
            parser.error(f"unsupported command: {args.command}")
            return 2
    except Exception as exc:  # pragma: no cover - CLI safety net
        kind_map = {
            "search": "cluster_search",
            "cluster-metadata": "cluster_meta",
            "topology-snapshot": "topology_snapshot",
            "metrics": "metric",
            "sql": "query",
            "slow-query": "query",
            "top-sql": "query",
            "statements": "query",
            "ticdc": "query",
            "logs": "log",
            "activity-events": "event",
            "continuous-profiling": "profile",
            "continuous-profiling-download": "profile",
            "diagnostic-files": "file",
            "diagnostic-file-download": "file",
            "ad-hoc": "ad_hoc",
        }
        cluster_id = getattr(args, "cluster_id", "")
        result = _error(args.command, kind_map.get(args.command, "unknown"), cluster_id, str(exc))
    _perf_event({
        "event": "cli_end",
        "command_id": command_id,
        "elapsed_ms": round((time.monotonic() - command_t0) * 1000, 3),
        "status": result.get("status"),
        "error": str(result.get("error", ""))[:500],
        **_command_perf_payload(args),
    })
    _json_dump(result)
    return 0 if result.get("status") == "ok" else 1


if __name__ == "__main__":
    raise SystemExit(main())