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

import argparse
import concurrent.futures
import contextlib
import fcntl
import getpass
import http.cookiejar
import io
import json
import math
import os
import re
import shlex
import signal
import subprocess
import sys
import tempfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

SKILL_ROOT = Path(__file__).resolve().parents[1]
CLUSTERS_FILE = SKILL_ROOT / "data" / "clusters.json"
SOURCE = "wetech-op-grafana"
DEFAULT_REGION = "asia-prod"
DEFAULT_2FA_SCRIPT = SKILL_ROOT / "scripts" / "get_verify_code.py"
SESSION_CACHE_ROOT = Path.home() / ".cache" / "wetech-op-grafana" / "browser"
DIAGNOSTIC_ROOT = Path.home() / ".cache" / "wetech-op-grafana" / "diagnostics"
CONCURRENCY_ROOT = Path.home() / ".cache" / "wetech-op-grafana" / "concurrency"
METRICS_CONCURRENCY = 5
DASHBOARD_CONCURRENCY = 2
DIRECT_SESSION_FILENAME = "direct-session-cookies.json"
_OPERATION_THREAD_STATE = threading.local()
_ACTIVE_SUBCOMMANDS: set[subprocess.Popen[str]] = set()
_ACTIVE_SUBCOMMANDS_LOCK = threading.Lock()
LOGIN_CREDENTIALS = {
    "p81_vpn": {
        "username": os.getenv("WETECH_P81_USERNAME", ""),
        "password": os.getenv("WETECH_P81_PASSWORD", ""),
    },
    "grafana": {
        "username": os.getenv("WETECH_GRAFANA_USERNAME", ""),
        "password": os.getenv("WETECH_GRAFANA_PASSWORD", ""),
    },
    "dashboard": {
        "username": os.getenv("WETECH_DASHBOARD_USERNAME", ""),
        "password": os.getenv("WETECH_DASHBOARD_PASSWORD", ""),
    },
}
SENSITIVE_HEADER_NAMES = {
    "authorization",
    "cookie",
    "x-csrf-token",
    "x-grafana-token",
    "x-auth-token",
    "x-id-token",
    "x-access-token",
}
HOP_BY_HOP_HEADERS = {
    "connection",
    "content-length",
    "host",
    "origin",
    "sec-fetch-dest",
    "sec-fetch-mode",
    "sec-fetch-site",
}
IGNORED_OPTIONS_WITH_VALUE = {
    "--connect-timeout",
    "--max-time",
    "--retry",
    "--retry-delay",
    "--retry-max-time",
    "--http1.1",
    "--http2",
    "--output",
    "-o",
}

STATEMENT_DEFAULT_FIELDS = (
    "digest_text",
    "sum_latency",
    "avg_latency",
    "max_latency",
    "min_latency",
    "exec_count",
    "plan_count",
)
# Curated from the Clinic slow_query_logs contract plus the evidence required by
# tidb-query-tuning, tidb-latency-diagnosis, and op-diagnosis. Large plan
# payloads stay opt-in; use --fields to request binary_plan for a narrow query.
SLOW_QUERY_REQUIRED_FIELDS = ("digest", "connection_id", "timestamp")
SLOW_QUERY_EXECUTION_INTERVAL_FIELDS = (
    "execution_interval_status",
    "time_semantics",
    "start_time",
    "finish_time",
    "overlaps_incident_window",
    "incident_overlap_seconds",
)
SLOW_QUERY_COMPACT_IDENTITY_FIELDS = (
    "digest",
    "query",
    "instance",
    "db",
    "query_time",
)
SLOW_QUERY_DIAGNOSTIC_FIELDS = (
    "query",
    "instance",
    "db",
    "success",
    "query_time",
    "parse_time",
    "compile_time",
    "rewrite_time",
    "preproc_subqueries_time",
    "optimize_time",
    "wait_ts",
    "cop_time",
    "lock_keys_time",
    "write_sql_response_total",
    "exec_retry_time",
    "memory_max",
    "disk_max",
    "index_names",
    "process_time",
    "wait_time",
    "backoff_time",
    "backoff_types",
    "local_latch_wait_time",
    "resolve_lock_time",
    "cop_proc_avg",
    "cop_proc_p90",
    "cop_proc_max",
    "cop_wait_avg",
    "cop_wait_p90",
    "cop_wait_max",
    "request_count",
    "process_keys",
    "total_keys",
    "rocksdb_delete_skipped_count",
    "rocksdb_key_skipped_count",
    "rocksdb_block_cache_hit_count",
    "rocksdb_block_read_count",
    "rocksdb_block_read_byte",
    "write_keys",
    "write_size",
    "txn_retry",
    "ru",
    "time_queued_by_rc",
    "resource_group",
)
SLOW_QUERY_CORE_FIELDS = (
    "query",
    "instance",
    "db",
    "success",
    "query_time",
    "memory_max",
    "disk_max",
    "process_time",
    "wait_time",
    "backoff_time",
    "request_count",
    "process_keys",
    "total_keys",
    "ru",
    "time_queued_by_rc",
    "resource_group",
)
SLOW_QUERY_REFINEMENT_FIELDS = (
    "stats",
    "warnings",
    "cop_proc_addr",
    "cop_wait_addr",
)
SLOW_QUERY_HANDOFF_FIELDS_CONTRACT = (
    SKILL_ROOT / "references" / "slow-query-handoff-fields.json"
)


def _load_slow_query_handoff_fields() -> frozenset[str]:
    payload = json.loads(SLOW_QUERY_HANDOFF_FIELDS_CONTRACT.read_text(encoding="utf-8"))
    fields = payload.get("fields") if isinstance(payload, dict) else None
    if (
        not isinstance(fields, list)
        or not fields
        or any(not isinstance(field, str) or not field for field in fields)
        or len(fields) != len(set(fields))
    ):
        raise RuntimeError("Slow Query handoff field contract is invalid.")
    return frozenset(fields)


SLOW_QUERY_HANDOFF_FIELDS = _load_slow_query_handoff_fields()
SLOW_QUERY_MAX_HANDOFF_FIELDS = 12
SLOW_QUERY_OVERLAP_LOOKAHEAD_SECONDS = 15 * 60
SLOW_QUERY_MINIMUM_CANDIDATE_LIMIT = 100
SLOW_QUERY_MAX_PLAN_DECODES = 3
OP_BATCH_MAX_QUERIES = 12
OP_BATCH_QUERY_ID_PATTERN = re.compile(r"^[a-z][a-z0-9]*(?:-[a-z0-9]+)*$")

TOPSQL_RECOMMENDED_SUMMARY_FIELDS = (
    "sql_digest",
    "sql_text",
    "is_other",
    "cpu_time_ms",
    "exec_count_per_sec",
    "duration_per_exec_ms",
    "scan_records_per_sec",
    "scan_indexes_per_sec",
    "network_bytes",
    "logical_io_bytes",
    "plans",
)
TOPSQL_RECOMMENDED_PLAN_FIELDS = (
    "plan_digest",
    "plan_text",
    "timestamp_sec",
    "cpu_time_ms",
    "exec_count_per_sec",
    "duration_per_exec_ms",
    "scan_records_per_sec",
    "scan_indexes_per_sec",
    "network_bytes",
    "logical_io_bytes",
)

PROVIDER_RESULT_VERSION = "provider-result-v1"
PROVIDER_STDOUT_MAX_BYTES = 32 * 1024
PROVIDER_SUMMARY_MAX_ROWS = 10
PROVIDER_SUMMARY_MAX_FIELDS = 24
PROVIDER_SUMMARY_MAX_DEPTH = 4
PROVIDER_SUMMARY_MAX_STRING = 300
DEFAULT_TOPSQL_TOP = 10
DEFAULT_TOPSQL_COMMAND_TIMEOUT_SECONDS = 120
MAX_BATCH_TOPSQL_RANGE_SECONDS = 15 * 60
PROVIDER_REPORT_MAX_ENTRIES = 10
METRIC_MAX_DECISION_LABELS = 12
METRIC_DECISION_LABEL_FIELDS = (
    "instance",
    "type",
    "job",
    "component",
    "store",
    "store_id",
    "address",
    "pod",
    "namespace",
    "node",
    "host",
    "cluster",
)
METRIC_NUMERIC_SUMMARY_FIELDS = (
    "series_count",
    "point_count",
    "min",
    "max",
    "max_at",
    "mean",
    "last",
)
_DEFAULT_RUN_OUTPUT_DIR: Path | None = None
_STDOUT_FORBIDDEN_KEYS = {
    "auth_token",
    "authorization",
    "binary_plan",
    "binary_plan_json",
    "binary_plan_text",
    "chart_data",
    "cookie",
    "decoded_plan",
    "execution_plan",
    "password",
    "payload",
    "physical_plan",
    "plan",
    "plan_detail",
    "plan_text",
    "plan_tree",
    "plans",
    "secret",
    "session",
    "token",
}


class ProviderError(RuntimeError):
    def __init__(self, code: str, message: str, *, detail: Any = None):
        super().__init__(message)
        self.code = code
        self.message = message
        self.detail = detail


MAX_PROMETHEUS_RESPONSE_BYTES = 8 * 1024 * 1024


def _print_json(payload: dict[str, Any]) -> None:
    print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))


def _compact_provider_value(
    value: Any,
    *,
    depth: int = 0,
    max_string: int = PROVIDER_SUMMARY_MAX_STRING,
) -> Any:
    if isinstance(value, str):
        if len(value) <= max_string:
            return value
        return value[:max_string] + "...<truncated>"
    if isinstance(value, (bool, int, float)) or value is None:
        return value
    if depth >= PROVIDER_SUMMARY_MAX_DEPTH:
        return "<truncated:depth>"
    if isinstance(value, dict):
        result: dict[str, Any] = {}
        omitted = 0
        safe_plan_summary = (
            isinstance(value.get("status"), str)
            and isinstance(value.get("operator_count"), int)
            and isinstance(value.get("operators"), list)
        )
        for key, child in value.items():
            normalized_key = str(key)
            if normalized_key.lower() in _STDOUT_FORBIDDEN_KEYS:
                omitted += 1
                continue
            if len(result) >= PROVIDER_SUMMARY_MAX_FIELDS:
                omitted += 1
                continue
            if safe_plan_summary and normalized_key == "operators":
                result[normalized_key] = [
                    operator
                    for operator in child[:12]
                    if isinstance(operator, str)
                    and re.fullmatch(
                        r"[A-Za-z][A-Za-z0-9]{0,79}\([A-Za-z0-9_.;=-]{1,240}\)",
                        operator,
                    )
                ]
                continue
            result[normalized_key] = _compact_provider_value(
                child,
                depth=depth + 1,
                max_string=max_string,
            )
        if omitted:
            result["_omitted_fields"] = omitted
        return result
    if isinstance(value, (list, tuple)):
        return [
            _compact_provider_value(child, depth=depth + 1, max_string=max_string)
            for child in list(value)[:PROVIDER_SUMMARY_MAX_ROWS]
        ]
    return _compact_provider_value(str(value), depth=depth, max_string=max_string)


def _print_provider_result(
    *,
    operation: str,
    status: str,
    kind: str | None = None,
    summary: dict[str, Any] | None = None,
    artifacts: dict[str, Any] | None = None,
    error: dict[str, Any] | None = None,
) -> None:
    result: dict[str, Any] = {
        "schema_version": PROVIDER_RESULT_VERSION,
        "status": status,
        "operation": operation,
    }
    if kind:
        result["kind"] = kind
    if summary:
        result["summary"] = _compact_provider_value(summary)
    if artifacts:
        result["artifacts"] = _compact_provider_value(artifacts)
    if error:
        result["error"] = _compact_provider_value(error)

    encoded = json.dumps(result, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
    if len(encoded.encode("utf-8")) > PROVIDER_STDOUT_MAX_BYTES:
        result["summary"] = {
            "truncated": True,
            "reason": "provider stdout byte budget exceeded",
        }
        encoded = json.dumps(result, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
    if len(encoded.encode("utf-8")) > PROVIDER_STDOUT_MAX_BYTES:
        result.pop("summary", None)
        encoded = json.dumps(result, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
    print(encoded)


def _load_clusters() -> list[dict[str, Any]]:
    return json.loads(CLUSTERS_FILE.read_text(encoding="utf-8"))


def _normalize_alias(value: str) -> str:
    return re.sub(r"[^a-z0-9]+", "", value.lower())


def _cluster_aliases(cluster: dict[str, Any]) -> set[str]:
    values = [cluster.get("alias")]
    aliases = cluster.get("aliases")
    if isinstance(aliases, list):
        values.extend(aliases)
    return {
        normalized
        for value in values
        if value is not None and (normalized := _normalize_alias(str(value)))
    }


def _find_cluster(alias: str, region: str | None = None) -> dict[str, Any]:
    wanted = _normalize_alias(alias)
    exact_dashboard_matches = [
        row
        for row in _load_clusters()
        if _normalize_alias(str(row.get("dashboard_cluster_name") or "")) == wanted
    ]
    if len(exact_dashboard_matches) == 1:
        return exact_dashboard_matches[0]
    if len(exact_dashboard_matches) > 1:
        regions = [str(row.get("region") or "legacy") for row in exact_dashboard_matches]
        raise SystemExit(
            f"dashboard cluster name {alias!r} is ambiguous; matching regions: {regions}"
        )
    wanted_region = _normalize_alias(region or DEFAULT_REGION)
    matches = [
        row
        for row in _load_clusters()
        if wanted in _cluster_aliases(row)
        and (wanted_region is None or _normalize_alias(_cluster_region(row)) == wanted_region)
    ]
    if len(matches) == 1:
        return matches[0]
    if region is None:
        all_alias_matches = [
            row for row in _load_clusters() if wanted in _cluster_aliases(row)
        ]
        if len(all_alias_matches) == 1:
            return all_alias_matches[0]
    if len(matches) > 1:
        regions = [str(row.get("region") or "legacy") for row in matches]
        raise SystemExit(f"cluster alias {alias!r} is ambiguous; pass --region. matching regions: {regions}")
    suffix = f" in region {region or DEFAULT_REGION!r}"
    raise SystemExit(f"unknown cluster alias: {alias}{suffix}")


def _cluster_key(cluster: dict[str, Any]) -> str:
    return _normalize_alias(str(cluster.get("alias") or ""))


def _cluster_region(cluster: dict[str, Any]) -> str:
    return str(cluster.get("region") or "legacy")


def _shared_grafana_clusters(clusters: list[dict[str, Any]], cluster: dict[str, Any]) -> list[dict[str, Any]]:
    urls = set(cluster.get("grafana") or [])
    if not urls:
        return [cluster]
    return [
        row
        for row in clusters
        if row is not cluster and urls.intersection(set(row.get("grafana") or []))
    ]


def _redact(value: Any) -> str:
    text = str(value)
    if not text:
        return ""
    return "<redacted>"


def _redact_headers(headers: dict[str, str]) -> dict[str, str]:
    out: dict[str, str] = {}
    for name, value in sorted(headers.items(), key=lambda item: item[0].lower()):
        if name.lower() in SENSITIVE_HEADER_NAMES:
            out[name] = _redact(value)
        else:
            out[name] = value
    return out


def _redact_sensitive_text(value: Any) -> str:
    text = str(value)
    for header_name in SENSITIVE_HEADER_NAMES:
        pattern = re.compile(rf"(?im)^(\s*[-\w]*{re.escape(header_name)}[-\w]*\s*:\s*).*$")
        text = pattern.sub(r"\1<redacted>", text)
    text = re.sub(r"(?i)(cookie=)[^;\s]+", r"\1<redacted>", text)
    text = re.sub(r"(?i)((?:authorization|token|session|csrf)[-_a-z0-9]*=)[^;&\s]+", r"\1<redacted>", text)
    return text


def _sanitize_error_text(value: Any) -> str:
    return _redact_sensitive_text(value)[:2000]


def _redact_sensitive_value(value: Any) -> Any:
    if isinstance(value, dict):
        result: dict[str, Any] = {}
        for key, child in value.items():
            normalized_key = str(key)
            if re.search(r"authorization|cookie|password|secret|session|csrf|token", normalized_key, re.I):
                result[normalized_key] = "<redacted>"
            else:
                result[normalized_key] = _redact_sensitive_value(child)
        return result
    if isinstance(value, list):
        return [_redact_sensitive_value(child) for child in value]
    if isinstance(value, tuple):
        return [_redact_sensitive_value(child) for child in value]
    if isinstance(value, str):
        return _redact_sensitive_text(value)
    return value


def _datasource_text(datasource: dict[str, Any]) -> str:
    pieces = [
        datasource.get("name"),
        datasource.get("uid"),
        datasource.get("url"),
        datasource.get("typeName"),
        datasource.get("database"),
    ]
    json_data = datasource.get("jsonData")
    if isinstance(json_data, dict):
        pieces.extend(str(value) for value in json_data.values() if isinstance(value, (str, int, float)))
    return " ".join(str(piece) for piece in pieces if piece)


def _datasource_match_score(cluster: dict[str, Any], datasource: dict[str, Any]) -> int:
    alias = str(cluster.get("alias") or "")
    normalized_alias = _normalize_alias(alias)
    normalized_text = _normalize_alias(_datasource_text(datasource))
    if not normalized_alias or not normalized_text:
        return 0
    score = 0
    if normalized_alias == normalized_text:
        score += 200
    if normalized_alias in normalized_text:
        score += 100 + len(normalized_alias)
    for token in re.split(r"[^A-Za-z0-9]+", alias):
        token = token.strip()
        if len(token) < 3:
            continue
        normalized_token = _normalize_alias(token)
        if normalized_token and normalized_token in normalized_text:
            score += len(normalized_token)
    return score


def _redact_url(url: str) -> str:
    parsed = urllib.parse.urlparse(url)
    pairs = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True)
    redacted_pairs = []
    for key, value in pairs:
        if re.search(r"token|session|cookie|password|secret|auth", key, re.I):
            redacted_pairs.append((key, "<redacted>"))
        else:
            redacted_pairs.append((key, value))
    query = urllib.parse.urlencode(redacted_pairs)
    return urllib.parse.urlunparse(parsed._replace(query=query))


def _parse_header(raw: str) -> tuple[str, str]:
    if ":" not in raw:
        raise ValueError(f"invalid header: {raw!r}")
    name, value = raw.split(":", 1)
    return name.strip(), value.strip()


def _parse_curl(curl_file: Path) -> dict[str, Any]:
    text = curl_file.read_text(encoding="utf-8")
    text = text.replace("\\\r\n", " ").replace("\\\n", " ")
    tokens = shlex.split(text)
    if not tokens or tokens[0] != "curl":
        raise ValueError("curl file must start with a curl command")

    url = ""
    headers: dict[str, str] = {}
    method = "GET"
    i = 1
    while i < len(tokens):
        token = tokens[i]
        if token in {"-H", "--header"}:
            i += 1
            if i >= len(tokens):
                raise ValueError(f"{token} requires a value")
            name, value = _parse_header(tokens[i])
            headers[name] = value
        elif token.startswith("--header="):
            name, value = _parse_header(token.split("=", 1)[1])
            headers[name] = value
        elif token.startswith("-H") and token != "-H":
            name, value = _parse_header(token[2:])
            headers[name] = value
        elif token in {"-b", "--cookie", "--cookie-jar"}:
            i += 1
            if i >= len(tokens):
                raise ValueError(f"{token} requires a value")
            if token != "--cookie-jar":
                headers["Cookie"] = tokens[i]
        elif token.startswith("--cookie="):
            headers["Cookie"] = token.split("=", 1)[1]
        elif token in {"-X", "--request"}:
            i += 1
            if i >= len(tokens):
                raise ValueError(f"{token} requires a value")
            method = tokens[i].upper()
        elif token.startswith("--request="):
            method = token.split("=", 1)[1].upper()
        elif token.startswith("http://") or token.startswith("https://"):
            url = token
        elif token in {"--compressed", "-s", "-S", "-L", "-k", "--insecure"}:
            pass
        elif token in IGNORED_OPTIONS_WITH_VALUE:
            i += 1
            if i >= len(tokens):
                raise ValueError(f"{token} requires a value")
        elif any(token.startswith(f"{option}=") for option in IGNORED_OPTIONS_WITH_VALUE if option.startswith("--")):
            pass
        elif token.startswith("-"):
            # Ignore browser curl noise that does not affect GET auth context.
            pass
        else:
            if not url and (token.startswith("http://") or token.startswith("https://")):
                url = token
        i += 1

    if method != "GET":
        raise ValueError(f"only GET curl commands are supported, got {method}")
    if not url:
        raise ValueError("curl command does not contain a URL")

    parsed = urllib.parse.urlparse(url)
    if parsed.scheme != "https":
        raise ValueError("only HTTPS Grafana URLs are supported")
    if "/api/datasources/proxy/" not in parsed.path:
        raise ValueError("only Grafana datasource proxy URLs are supported")
    endpoint = _grafana_endpoint(parsed.path)
    if endpoint is None:
        raise ValueError("only Prometheus query_range/query proxy URLs are supported")

    return {"url": url, "headers": headers, "method": method, "endpoint": endpoint}


def _grafana_endpoint(path: str) -> str | None:
    if path.endswith("/api/v1/query_range"):
        return "query_range"
    if path.endswith("/api/v1/query"):
        return "query"
    return None


def _with_query_overrides(url: str, args: argparse.Namespace) -> str:
    parsed = urllib.parse.urlparse(url)
    params = dict(urllib.parse.parse_qsl(parsed.query, keep_blank_values=True))
    for attr in ("query", "start", "end", "step"):
        value = getattr(args, attr, None)
        if value is not None and value != "":
            params[attr] = str(value)
    query = urllib.parse.urlencode(params)
    return urllib.parse.urlunparse(parsed._replace(query=query))


def _safe_request_headers(headers: dict[str, str]) -> dict[str, str]:
    out: dict[str, str] = {}
    for name, value in headers.items():
        if name.lower() in HOP_BY_HOP_HEADERS:
            continue
        out[name] = value
    if "User-Agent" not in out and "user-agent" not in {k.lower() for k in out}:
        out["User-Agent"] = "Mozilla/5.0 wetech-op-grafana"
    out.setdefault("Accept", "application/json, text/plain, */*")
    return out


def _http_get(url: str, headers: dict[str, str], timeout: int = 30) -> dict[str, Any]:
    request = urllib.request.Request(url, headers=_safe_request_headers(headers), method="GET")
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            body = response.read()
            status = response.getcode()
            content_type = response.headers.get("content-type", "")
    except urllib.error.HTTPError as exc:
        body = exc.read()
        return {
            "http_status": exc.code,
            "error": f"HTTP {exc.code}",
            "body_prefix": body.decode(errors="replace")[:500],
        }
    except Exception as exc:
        return {"http_status": 0, "error": _sanitize_error_text(exc)}

    raw = body.decode(errors="replace")
    try:
        payload = json.loads(raw)
    except json.JSONDecodeError:
        return {
            "http_status": status,
            "content_type": content_type,
            "error": "response was not JSON",
            "body_prefix": raw[:500],
        }
    return {"http_status": status, "content_type": content_type, "payload": payload}


def _series_points(payload: dict[str, Any]) -> list[float]:
    points: list[float] = []
    for series in payload.get("data", {}).get("result") or []:
        values = series.get("values")
        if values is None and "value" in series:
            values = [series.get("value")]
        if not isinstance(values, list):
            continue
        for pair in values:
            if not isinstance(pair, list) or len(pair) < 2:
                continue
            try:
                value = float(pair[1])
            except (TypeError, ValueError):
                continue
            if math.isfinite(value):
                points.append(value)
    return points


def _summarize_numeric(payload: dict[str, Any]) -> dict[str, Any]:
    points = _series_points(payload)
    summary: dict[str, Any] = {
        "point_count": len(points),
        "series_count": _series_count(payload),
    }
    if points:
        summary.update(
            {
                "min": min(points),
                "max": max(points),
                "mean": sum(points) / len(points),
                "last": points[-1],
            }
        )
    return summary


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


def _series_summaries(payload: dict[str, Any]) -> list[dict[str, Any]]:
    summaries: list[dict[str, Any]] = []
    for index, series in enumerate(payload.get("data", {}).get("result") or []):
        values = series.get("values")
        if values is None and "value" in series:
            values = [series.get("value")]
        child = {"data": {"result": [{"values": values or []}]}}
        summary = _summarize_numeric(child)
        maximum_value: float | None = None
        maximum_at: int | None = None
        for pair in values or []:
            if not isinstance(pair, list) or len(pair) < 2:
                continue
            try:
                timestamp = float(pair[0])
                value = float(pair[1])
            except (TypeError, ValueError):
                continue
            if not math.isfinite(timestamp) or not math.isfinite(value):
                continue
            if maximum_value is None or value > maximum_value:
                maximum_value = value
                maximum_at = int(timestamp)
        if maximum_at is not None:
            summary["max_at"] = maximum_at
        summaries.append(
            {
                "index": index,
                "metric": series.get("metric") or {},
                "summary": summary,
            }
        )
    return summaries


def _time_window(start: Any, end: Any, step: Any) -> dict[str, Any]:
    return {"start": start, "end": end, "step": str(step) if step is not None else ""}


def _parse_duration_seconds(value: str) -> int:
    match = re.fullmatch(r"\s*(\d+)\s*([smhd])\s*", value, re.I)
    if not match:
        raise ValueError("duration must look like 30m, 1h, 2d")
    amount = int(match.group(1))
    unit = match.group(2).lower()
    multipliers = {"s": 1, "m": 60, "h": 3600, "d": 86400}
    return amount * multipliers[unit]


def _positive_int(value: str) -> int:
    parsed = int(value)
    if parsed <= 0:
        raise argparse.ArgumentTypeError("value must be a positive integer")
    return parsed


def _topsql_command_timeout(value: str) -> int:
    parsed = _positive_int(value)
    if parsed > DEFAULT_TOPSQL_COMMAND_TIMEOUT_SECONDS:
        raise argparse.ArgumentTypeError(
            f"value must not exceed {DEFAULT_TOPSQL_COMMAND_TIMEOUT_SECONDS} seconds"
        )
    return parsed


def _resolve_range_args(args: argparse.Namespace) -> tuple[str, str, str]:
    if args.end:
        end = int(float(args.end))
    else:
        end = int(time.time())
    if args.start:
        start = int(float(args.start))
    else:
        start = end - _parse_duration_seconds(args.last)
    if start >= end:
        raise ValueError("start must be earlier than end")
    return str(start), str(end), str(args.step)


def _resolve_dashboard_range_args(args: argparse.Namespace) -> tuple[str, str]:
    if args.end:
        end = int(float(args.end))
    else:
        end = int(time.time())
    if args.start:
        start = int(float(args.start))
    else:
        start = end - _parse_duration_seconds(args.last)
    if start >= end:
        raise ValueError("start must be earlier than end")
    return str(start), str(end)


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


def _metric_row_from_payload(
    *,
    name: str,
    metric_group_id: str,
    unit: str,
    promql: str,
    start: str,
    end: str,
    step: str,
    payload: dict[str, Any],
    source_kind: str,
    origin: str,
    path: str,
) -> tuple[dict[str, Any], dict[str, Any]]:
    time_range = _time_window(start, end, step)
    row = {
        "id": metric_group_id,
        "group": metric_group_id,
        "ref": {
            "kind": "metric",
            "source": SOURCE,
            "metric_group_id": metric_group_id,
        },
        "source": SOURCE,
        "metric_group_id": metric_group_id,
        "name": name,
        "promql": promql,
        "unit": unit or "",
        "summary": _summarize_numeric(payload),
        "series_summaries": _series_summaries(payload),
        "time_window": time_range,
        "chart_data_ref": {
            "kind": source_kind,
            "source": SOURCE,
            "origin": origin,
            "path": path,
            "promql": promql,
            "start": start,
            "end": end,
            "step": step,
        },
    }
    return row, time_range


@contextlib.contextmanager
def _evidence_lock(evidence_dir: Path):
    evidence_dir.mkdir(parents=True, exist_ok=True)
    lock_path = evidence_dir / ".workspace_evidence.lock"
    with lock_path.open("a+") as lock:
        fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
        try:
            yield
        finally:
            fcntl.flock(lock.fileno(), fcntl.LOCK_UN)


@contextlib.contextmanager
def _operation_slot(operation: str):
    if operation == "browser-query-range":
        pool, limit = "metrics", METRICS_CONCURRENCY
    elif operation == "dashboard-query":
        pool, limit = "dashboard", DASHBOARD_CONCURRENCY
    else:
        raise ValueError(f"unsupported concurrent Wetech operation: {operation}")
    pool_dir = CONCURRENCY_ROOT / pool
    pool_dir.mkdir(parents=True, exist_ok=True)
    while True:
        for slot in range(limit):
            lock = (pool_dir / f"slot-{slot}.lock").open("a+")
            try:
                fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                lock.close()
                continue
            try:
                _OPERATION_THREAD_STATE.slot = slot
                yield slot
            finally:
                _OPERATION_THREAD_STATE.slot = None
                fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
                lock.close()
            return
        time.sleep(0.05)


@contextlib.contextmanager
def _profile_lock(session_dir: Path):
    session_dir.mkdir(parents=True, exist_ok=True)
    lock_path = session_dir.parent / f".{session_dir.name}.profile.lock"
    with lock_path.open("a+") as lock:
        fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
        try:
            yield
        finally:
            fcntl.flock(lock.fileno(), fcntl.LOCK_UN)


def _json_load(path: Path, default: Any) -> Any:
    if not path.exists():
        return default
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return default


def _atomic_write_json(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile("w", delete=False, dir=str(path.parent), encoding="utf-8") as tmp:
        json.dump(payload, tmp, ensure_ascii=False, indent=2, sort_keys=True)
        tmp.write("\n")
        tmp_path = Path(tmp.name)
    os.replace(tmp_path, path)


def _compact_metric_labels(labels: Any, *, max_string: int) -> dict[str, Any]:
    if not isinstance(labels, dict):
        return {}
    selected_keys = [
        key for key in METRIC_DECISION_LABEL_FIELDS
        if key in labels
    ]
    for key in labels:
        normalized_key = str(key)
        if normalized_key in selected_keys:
            continue
        selected_keys.append(normalized_key)
        if len(selected_keys) >= METRIC_MAX_DECISION_LABELS:
            break
    return {
        key: _compact_provider_value(labels.get(key), max_string=max_string)
        for key in selected_keys[:METRIC_MAX_DECISION_LABELS]
    }


def _compact_metric_numeric_summary(summary: Any) -> dict[str, Any]:
    if not isinstance(summary, dict):
        return {}
    return {
        field: summary.get(field)
        for field in METRIC_NUMERIC_SUMMARY_FIELDS
        if isinstance(summary.get(field), (int, float)) and not isinstance(summary.get(field), bool)
    }


def _compact_metric_series_summaries(
    series_summaries: Any,
    *,
    max_string: int = PROVIDER_SUMMARY_MAX_STRING,
) -> list[dict[str, Any]]:
    if not isinstance(series_summaries, list):
        return []
    ranked_series = sorted(
        (
            series
            for series in series_summaries
            if isinstance(series, dict)
        ),
        key=lambda series: (
            series.get("summary", {}).get("max")
            if isinstance(series.get("summary"), dict)
            and isinstance(series.get("summary", {}).get("max"), (int, float))
            else float("-inf")
        ),
        reverse=True,
    )
    result: list[dict[str, Any]] = []
    for index, series in enumerate(ranked_series[:PROVIDER_SUMMARY_MAX_ROWS]):
        result.append(
            {
                "index": series.get("index", index),
                "metric": _compact_metric_labels(
                    series.get("metric"),
                    max_string=max_string,
                ),
                "summary": _compact_metric_numeric_summary(series.get("summary")),
            }
        )
    return result


def _compact_metric(
    row: dict[str, Any],
    *,
    max_string: int = PROVIDER_SUMMARY_MAX_STRING,
) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key in (
        "id",
        "group",
        "ref",
        "source",
        "metric_group_id",
        "name",
        "promql",
        "unit",
        "time_window",
        "chart_data_ref",
    ):
        if key in row:
            result[key] = _compact_provider_value(row[key], max_string=max_string)
    result["summary"] = _compact_metric_numeric_summary(row.get("summary"))
    result["series_summaries"] = _compact_metric_series_summaries(
        row.get("series_summaries"),
        max_string=max_string,
    )
    return result


def _report_size_bytes(report: dict[str, Any]) -> int:
    return len(
        json.dumps(report, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode("utf-8")
    )


def _bounded_report_rows(rows: Any, compact_row: Any) -> list[dict[str, Any]]:
    if not isinstance(rows, list):
        return []
    compacted = [
        compact_row(row)
        for row in rows
        if isinstance(row, dict)
    ]
    return compacted[-PROVIDER_REPORT_MAX_ENTRIES:]


def _bound_workspace_report(report: dict[str, Any]) -> dict[str, Any]:
    evidence = report.get("evidence")
    if not isinstance(evidence, dict):
        evidence = {}
    evidence["metrics"] = _bounded_report_rows(evidence.get("metrics"), _compact_metric)
    evidence["dashboard"] = _bounded_report_rows(
        evidence.get("dashboard"),
        _compact_dashboard_evidence,
    )
    report["evidence"] = evidence

    while _report_size_bytes(report) > PROVIDER_STDOUT_MAX_BYTES:
        candidates = [
            (key, rows)
            for key, rows in evidence.items()
            if isinstance(rows, list) and len(rows) > 1
        ]
        if not candidates:
            break
        key, rows = max(
            candidates,
            key=lambda item: len(
                json.dumps(item[1], ensure_ascii=False, separators=(",", ":")).encode("utf-8")
            ),
        )
        evidence[key] = rows[1:]

    if _report_size_bytes(report) > PROVIDER_STDOUT_MAX_BYTES:
        for max_string in (120, 60, 30):
            for key, rows in list(evidence.items()):
                if not isinstance(rows, list):
                    continue
                if key == "metrics":
                    evidence[key] = [
                        _compact_metric(row, max_string=max_string)
                        for row in rows
                        if isinstance(row, dict)
                    ]
                elif key == "dashboard":
                    evidence[key] = [
                        _compact_dashboard_evidence(row, max_string=max_string)
                        for row in rows
                        if isinstance(row, dict)
                    ]
                else:
                    evidence[key] = [
                        _compact_provider_value(row, max_string=max_string)
                        for row in rows
                    ]
            if _report_size_bytes(report) <= PROVIDER_STDOUT_MAX_BYTES:
                break

    if _report_size_bytes(report) > PROVIDER_STDOUT_MAX_BYTES:
        evidence["metrics"] = [
            {
                "id": row.get("id"),
                "kind": "metric",
                "summary": row.get("summary"),
                "report_truncated": True,
            }
            for row in evidence.get("metrics", [])
            if isinstance(row, dict)
        ][-1:]
        evidence["dashboard"] = [
            {
                "id": row.get("id"),
                "kind": row.get("kind"),
                "cluster": row.get("cluster"),
                "time_window": row.get("time_window"),
                "record_count": row.get("record_count"),
                "report_truncated": True,
            }
            for row in evidence.get("dashboard", [])
            if isinstance(row, dict)
        ][-1:]
    return report


def _resolve_evidence_dir(output_dir: str | None = None) -> Path:
    global _DEFAULT_RUN_OUTPUT_DIR

    if output_dir:
        return Path(output_dir).expanduser().resolve()

    env_output = os.environ.get("WETECH_OP_GRAFANA_OUTPUT_DIR", "").strip()
    if env_output:
        return Path(env_output).expanduser().resolve()

    workspace_dir = os.environ.get("INVESTIBOOK_WORKSPACE_DIR", "").strip()
    if workspace_dir:
        return Path(workspace_dir).resolve() / "evidences"

    if _DEFAULT_RUN_OUTPUT_DIR is None:
        run_id = f"run-{int(time.time() * 1000)}-{os.getpid()}"
        _DEFAULT_RUN_OUTPUT_DIR = (Path.cwd() / "evidences" / run_id).resolve()
    return _DEFAULT_RUN_OUTPUT_DIR


def _persist_error_artifact(
    *,
    operation: str,
    code: str,
    message: str,
    detail: Any,
    output_dir: str | None = None,
) -> dict[str, str]:
    evidence_dir = _resolve_evidence_dir(output_dir)
    error_dir = evidence_dir / "errors"
    error_path = error_dir / f"{_metric_group_id(operation)}-{time.time_ns()}.json"
    artifact = {
        "schema_version": PROVIDER_RESULT_VERSION,
        "status": "error",
        "operation": operation,
        "code": code,
        "message": message,
        "detail": _redact_sensitive_value(detail),
        "recorded_at": int(time.time()),
    }
    with _evidence_lock(evidence_dir):
        _atomic_write_json(error_path, artifact)
    return {"error_path": str(error_path), "output_dir": str(evidence_dir)}


def _provider_error_parts(
    exc: Exception,
    *,
    default_code: str,
) -> tuple[str, str, Any]:
    if isinstance(exc, ProviderError):
        return exc.code, exc.message[:PROVIDER_SUMMARY_MAX_STRING], exc.detail
    full_message = _sanitize_error_text(exc)
    first_line = full_message.splitlines()[0]
    if any(marker in first_line.lower() for marker in ("payload", "response body")) or any(
        marker in first_line for marker in ("{", "[")
    ):
        message = "Provider operation failed; inspect the error artifact."
    else:
        message = first_line[:PROVIDER_SUMMARY_MAX_STRING]
    return (
        default_code,
        message or "provider operation failed",
        {
            "exception_type": type(exc).__name__,
            "message": full_message,
        },
    )


def _print_provider_error(
    *,
    operation: str,
    exc: Exception,
    output_dir: str | None = None,
    kind: str | None = None,
    default_code: str = "provider_operation_failed",
) -> None:
    code, message, detail = _provider_error_parts(exc, default_code=default_code)
    artifacts: dict[str, Any] = {}
    try:
        artifacts = _persist_error_artifact(
            operation=operation,
            code=code,
            message=message,
            detail=detail,
            output_dir=output_dir,
        )
    except Exception:
        artifacts = {}
    _print_provider_result(
        operation=operation,
        status="error",
        kind=kind,
        artifacts=artifacts,
        error={"code": code, "message": message},
    )


def _persist_metric(
    row: dict[str, Any],
    time_range: dict[str, Any],
    *,
    output_dir: str | None = None,
) -> dict[str, str]:

    evidence_dir = _resolve_evidence_dir(output_dir)
    metrics_path = evidence_dir / "metrics.json"
    report_path = evidence_dir / "workspace_evidence_report.json"

    with _evidence_lock(evidence_dir):
        metrics_payload = _json_load(metrics_path, {"metrics": [], "time_range": time_range})
        metrics = metrics_payload.get("metrics")
        if not isinstance(metrics, list):
            metrics = []
        metrics.append(row)
        metrics_payload["metrics"] = metrics
        metrics_payload["time_range"] = time_range
        _atomic_write_json(metrics_path, metrics_payload)

        report = _json_load(report_path, {"evidence": {}})
        evidence = report.get("evidence")
        if not isinstance(evidence, dict):
            evidence = {}
        evidence["metrics"] = [
            _compact_metric(raw_row)
            for raw_row in metrics[-PROVIDER_REPORT_MAX_ENTRIES:]
            if isinstance(raw_row, dict)
        ]
        report["evidence"] = evidence
        report["updated_at"] = int(time.time())
        report = _bound_workspace_report(report)
        _atomic_write_json(report_path, report)

    return {"evidence_file_path": str(metrics_path), "report_path": str(report_path)}


def _truncate_value(
    value: Any,
    *,
    max_string: int = PROVIDER_SUMMARY_MAX_STRING,
    depth: int = 0,
) -> Any:
    if depth >= PROVIDER_SUMMARY_MAX_DEPTH:
        return "<truncated:depth>"
    if isinstance(value, str):
        if len(value) <= max_string:
            return value
        return value[:max_string] + "...<truncated>"
    if isinstance(value, dict):
        result: dict[str, Any] = {}
        omitted = 0
        for key, child in value.items():
            normalized_key = str(key)
            if normalized_key.lower() in _STDOUT_FORBIDDEN_KEYS:
                omitted += 1
                continue
            if len(result) >= PROVIDER_SUMMARY_MAX_FIELDS:
                omitted += 1
                continue
            result[normalized_key] = _truncate_value(
                child,
                max_string=max_string,
                depth=depth + 1,
            )
        if omitted:
            result["_omitted_fields"] = omitted
        return result
    if isinstance(value, list):
        return [
            _truncate_value(child, max_string=max_string, depth=depth + 1)
            for child in value[:PROVIDER_SUMMARY_MAX_ROWS]
        ]
    return value


def _dashboard_record_count(payload: Any) -> int:
    return len(_dashboard_rows_for_preview(payload))


def _dashboard_rows_for_preview(payload: Any) -> list[Any]:
    if isinstance(payload, list):
        return payload
    if isinstance(payload, dict):
        summary = payload.get("summary")
        if isinstance(summary, dict):
            rows = _dashboard_rows_from_mapping(summary)
            if rows:
                return rows
        return _dashboard_rows_from_mapping(payload)
    return []


def _dashboard_rows_from_mapping(payload: dict[str, Any]) -> list[Any]:
    for key in ("data", "items", "result", "results"):
        value = payload.get(key)
        if isinstance(value, list):
            return value
    return []


def _dashboard_preview_rows(
    payload: Any,
    limit: int,
    selected_fields: list[str] | None = None,
) -> list[Any]:
    rows = _dashboard_rows_for_preview(payload)
    bounded_limit = max(0, min(limit, PROVIDER_SUMMARY_MAX_ROWS))
    if selected_fields is None or "*" in selected_fields:
        return [_truncate_value(row) for row in rows[:bounded_limit]]
    selected = list(dict.fromkeys(selected_fields))
    return [
        _truncate_value({key: row[key] for key in selected if key in row})
        if isinstance(row, dict)
        else _truncate_value(row)
        for row in rows[:bounded_limit]
    ]


def _slow_query_execution_rows(
    payload: Any,
    incident_start: int,
    incident_end: int,
    result_limit: int,
) -> tuple[list[dict[str, Any]], dict[str, int]]:
    classified: list[dict[str, Any]] = []
    overlapping = 0
    unclassified = 0
    excluded = 0
    for candidate in _dashboard_rows_for_preview(payload):
        if not isinstance(candidate, dict):
            unclassified += 1
            continue
        row = dict(candidate)
        row["query_plan_summary"] = _safe_slow_query_plan_summary(row)
        finish = _slow_query_epoch_seconds(row.get("timestamp", row.get("time")))
        query_time = _finite_number(row.get("query_time"))
        if finish is None or query_time is None or query_time < 0:
            row.update({
                "execution_interval_status": "unavailable",
                "time_semantics": "slow_log_time_is_finish_print_time",
                "start_time": None,
                "finish_time": finish,
                "overlaps_incident_window": None,
                "incident_overlap_seconds": None,
            })
            classified.append(row)
            unclassified += 1
            continue
        start = finish - query_time
        overlaps = finish >= incident_start and start <= incident_end
        overlap_seconds = max(
            0.0,
            min(finish, incident_end) - max(start, incident_start),
        )
        row.update({
            "execution_interval_status": "derived",
            "time_semantics": "slow_log_time_is_finish_print_time",
            "start_time": round(start, 6),
            "finish_time": round(finish, 6),
            "overlaps_incident_window": overlaps,
            "incident_overlap_seconds": round(overlap_seconds, 6),
        })
        if overlaps:
            classified.append(row)
            overlapping += 1
        else:
            excluded += 1
    bounded_limit = max(1, min(200, result_limit))
    return classified[:bounded_limit], {
        "raw_record_count": len(_dashboard_rows_for_preview(payload)),
        "overlapping_record_count": overlapping,
        "unclassified_record_count": unclassified,
        "excluded_non_overlapping_record_count": excluded,
    }


def _slow_query_preview_fields(request: dict[str, Any]) -> list[str]:
    fields = [
        *SLOW_QUERY_COMPACT_IDENTITY_FIELDS,
        *SLOW_QUERY_EXECUTION_INTERVAL_FIELDS,
        "query_plan_summary",
    ]
    requested = request.get("field_selection", {}).get("requested_fields", [])
    if isinstance(requested, list):
        fields.extend(str(field) for field in requested)
    return list(dict.fromkeys(fields))


def _safe_slow_query_plan_summary(row: dict[str, Any]) -> dict[str, Any]:
    decoded = _decoded_plan_object(row.get("binary_plan_json"))
    if decoded is not None:
        operators: list[str] = []
        _collect_safe_plan_operators(decoded, operators, 0)
        return {
            "status": "complete" if operators else "decoded_plan_empty",
            "available": bool(operators),
            "operator_count": len(operators),
            "operators": operators,
        }
    text = row.get("binary_plan_text")
    if isinstance(text, str) and text.strip():
        operators = _safe_text_plan_operators(text)
        return {
            "status": "complete" if operators else "decoded_plan_unparseable",
            "available": bool(operators),
            "operator_count": len(operators),
            "operators": operators,
        }
    if isinstance(row.get("binary_plan"), str) and row["binary_plan"].strip():
        return {
            "status": "encoded_plan_only",
            "available": False,
            "operator_count": 0,
            "operators": [],
        }
    return {
        "status": "query_plan_unavailable",
        "available": False,
        "operator_count": 0,
        "operators": [],
    }


def _decoded_plan_object(value: Any) -> dict[str, Any] | None:
    if isinstance(value, dict):
        parsed = value
    elif isinstance(value, str) and 0 < len(value) <= 1024 * 1024:
        try:
            parsed = json.loads(value)
        except json.JSONDecodeError:
            return None
        if not isinstance(parsed, dict):
            return None
    else:
        return None
    main = parsed.get("main")
    return main if isinstance(main, dict) else parsed


def _collect_safe_plan_operators(
    node: dict[str, Any],
    output: list[str],
    depth: int,
) -> None:
    if len(output) >= 32 or depth > 16:
        return
    name = _safe_plan_operator_name(node.get("name", node.get("id")))
    if name:
        output.append(_safe_plan_operator_fact(
            name=name,
            depth=depth,
            est_rows=_safe_plan_number(node.get("estRows", node.get("est_rows"))),
            act_rows=_safe_plan_number(node.get("actRows", node.get("act_rows"))),
            task_type=_safe_plan_category(node.get("taskType", node.get("task_type"))),
            execution_time_ms=_safe_execution_time_ms(
                node.get("executionInfo", node.get("execution_info"))
            ),
        ))
    children = node.get("children")
    if isinstance(children, list):
        for child in children:
            if isinstance(child, dict):
                _collect_safe_plan_operators(child, output, depth + 1)


def _safe_text_plan_operators(value: str) -> list[str]:
    operators: list[str] = []
    for line in value.splitlines()[:64]:
        columns = [column.strip() for column in line.split("\t")]
        if len(columns) < 3:
            columns = re.split(r"\s{2,}", line.strip())
        if not columns:
            continue
        name = _safe_plan_operator_name(columns[0])
        if not name or name.lower() in {"id", "operator"}:
            continue
        operators.append(_safe_plan_operator_fact(
            name=name,
            depth=max(0, min(16, len(columns[0]) - len(columns[0].lstrip(" │├└─")))),
            est_rows=_safe_plan_number(columns[1] if len(columns) > 1 else None),
            act_rows=_safe_plan_number(columns[2] if len(columns) > 2 else None),
            task_type=_safe_plan_category(columns[3] if len(columns) > 3 else None),
            execution_time_ms=_safe_execution_time_ms(columns[5] if len(columns) > 5 else None),
        ))
        if len(operators) >= 32:
            break
    return operators


def _safe_plan_operator_fact(
    *,
    name: str,
    depth: int,
    est_rows: float | int | None,
    act_rows: float | int | None,
    task_type: str | None,
    execution_time_ms: float | None,
) -> str:
    fields = [f"depth={depth}"]
    for key, value in (
        ("estRows", est_rows),
        ("actRows", act_rows),
        ("timeMs", execution_time_ms),
    ):
        if value is not None:
            fields.append(f"{key}={value}")
    if task_type:
        fields.append(f"task={task_type}")
    return f"{name}({';'.join(fields)})"


def _safe_plan_operator_name(value: Any) -> str | None:
    match = re.search(r"(?:^|[─\s])([A-Za-z][A-Za-z0-9]*)(?:_[0-9]+)?(?:$|\s)", str(value or "").strip())
    return match.group(1)[:80] if match else None


def _safe_plan_category(value: Any) -> str | None:
    text = str(value or "").strip().lower()
    return text if re.fullmatch(r"[a-z][a-z0-9_-]{0,63}", text) else None


def _safe_plan_number(value: Any) -> float | int | None:
    numeric = _finite_number(value)
    if numeric is None or abs(numeric) > 9007199254740991:
        return None
    return int(numeric) if numeric.is_integer() else numeric


def _safe_execution_time_ms(value: Any) -> float | None:
    match = re.search(
        r"(?:^|[,;\s])time\s*:\s*([0-9]+(?:\.[0-9]+)?)\s*(ns|us|µs|ms|s|m|h)\b",
        str(value or ""),
        re.IGNORECASE,
    )
    if not match:
        return None
    scale = {"ns": 0.000001, "us": 0.001, "µs": 0.001, "ms": 1, "s": 1000, "m": 60000, "h": 3600000}
    return round(float(match.group(1)) * scale[match.group(2).lower()], 3)


def _slow_query_epoch_seconds(value: Any) -> float | None:
    numeric = _finite_number(value)
    if numeric is not None:
        if numeric >= 1e15:
            return numeric / 1_000_000
        if numeric >= 1e12:
            return numeric / 1_000
        return numeric
    if not isinstance(value, str) or not value.strip():
        return None
    text = value.strip().replace(" ", "T")
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.timestamp()


def _finite_number(value: Any) -> float | None:
    if isinstance(value, bool) or value is None:
        return None
    try:
        numeric = float(value)
    except (TypeError, ValueError):
        return None
    return numeric if math.isfinite(numeric) else None


def _field_list(value: str | None, defaults: tuple[str, ...]) -> list[str]:
    if value is None or not value.strip():
        return list(defaults)
    fields = [field.strip() for field in value.split(",") if field.strip()]
    if fields == ["*"]:
        return fields
    return list(dict.fromkeys(fields))


def _slow_query_field_selection(
    requested_fields: list[str],
    available_payload: Any,
    available_status: int,
) -> tuple[list[str], dict[str, Any]]:
    available_fields = [
        str(field)
        for field in available_payload
        if isinstance(field, str) and field
    ] if available_status == 200 and isinstance(available_payload, list) else []
    diagnostic_profile = requested_fields == list(SLOW_QUERY_DIAGNOSTIC_FIELDS)
    effective_requested_fields = (
        list(dict.fromkeys((*SLOW_QUERY_REQUIRED_FIELDS, *requested_fields)))
        if diagnostic_profile else requested_fields
    )
    if effective_requested_fields == ["*"]:
        selected_fields = ["*"]
        unsupported_fields: list[str] = []
    elif available_fields:
        available = set(available_fields)
        selected_fields = [field for field in effective_requested_fields if field in available]
        unsupported_fields = [field for field in effective_requested_fields if field not in available]
    else:
        selected_fields = effective_requested_fields
        unsupported_fields = []
    return selected_fields, {
        "mode": "requested_projection",
        "profile": "diagnostic" if diagnostic_profile else "explicit",
        "availability_status": "available" if available_fields else "unavailable",
        "requested_fields": selected_fields,
        "unsupported_fields": unsupported_fields,
    }


def _topsql_field_selection(payload: Any) -> dict[str, Any]:
    rows = _dashboard_rows_for_preview(payload)
    summary_fields: set[str] = set()
    plan_fields: set[str] = set()
    for row in rows:
        if not isinstance(row, dict):
            continue
        summary_fields.update(str(key) for key in row)
        plans = row.get("plans")
        if isinstance(plans, list):
            for plan in plans:
                if isinstance(plan, dict):
                    plan_fields.update(str(key) for key in plan)
    return {
        "mode": "fixed_response",
        "returned_summary_fields": sorted(summary_fields),
        "returned_plan_fields": sorted(plan_fields),
        "missing_recommended_summary_fields": [
            field for field in TOPSQL_RECOMMENDED_SUMMARY_FIELDS if field not in summary_fields
        ],
        "missing_recommended_plan_fields": [
            field for field in TOPSQL_RECOMMENDED_PLAN_FIELDS if field not in plan_fields
        ],
    }


def _compact_dashboard_evidence(
    row: dict[str, Any],
    *,
    max_string: int = PROVIDER_SUMMARY_MAX_STRING,
) -> dict[str, Any]:
    compact = _compact_provider_value(
        {
            key: value
            for key, value in row.items()
            if key not in {"payload", "auth_token"}
        },
        max_string=max_string,
    )
    return compact if isinstance(compact, dict) else {}


def _persist_dashboard_evidence(row: dict[str, Any], *, output_dir: str | None = None) -> dict[str, str]:
    evidence_dir = _resolve_evidence_dir(output_dir)
    dashboard_path = evidence_dir / "dashboard.json"
    report_path = evidence_dir / "workspace_evidence_report.json"

    with _evidence_lock(evidence_dir):
        dashboard_payload = _json_load(dashboard_path, {"dashboard": []})
        dashboard_rows = dashboard_payload.get("dashboard")
        if not isinstance(dashboard_rows, list):
            dashboard_rows = []
        dashboard_rows.append(row)
        dashboard_payload["dashboard"] = dashboard_rows
        _atomic_write_json(dashboard_path, dashboard_payload)

        report = _json_load(report_path, {"evidence": {}})
        evidence = report.get("evidence")
        if not isinstance(evidence, dict):
            evidence = {}
        report_rows = _bounded_report_rows(
            evidence.get("dashboard"),
            _compact_dashboard_evidence,
        )
        report_rows.append(_compact_dashboard_evidence(row))
        evidence["dashboard"] = report_rows[-PROVIDER_REPORT_MAX_ENTRIES:]
        report["evidence"] = evidence
        report["updated_at"] = int(time.time())
        report = _bound_workspace_report(report)
        _atomic_write_json(report_path, report)

    return {"evidence_file_path": str(dashboard_path), "report_path": str(report_path)}


def _import_playwright():
    try:
        from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
        from playwright.sync_api import sync_playwright
    except Exception as exc:
        raise RuntimeError(
            "Playwright is required for browser-query-range and dashboard-query. Run "
            "`python3 -m pip install -r wetech-op-grafana/requirements.txt` and "
            "`python3 -m playwright install chromium`."
        ) from exc
    return sync_playwright, PlaywrightTimeoutError


def _cluster_grafana_url(args: argparse.Namespace) -> tuple[dict[str, Any], str]:
    cluster = _find_cluster(args.cluster, getattr(args, "region", None))
    if args.grafana_url:
        return cluster, args.grafana_url
    urls = cluster.get("grafana") or []
    if not urls:
        raise ValueError(f"cluster {args.cluster!r} has no Grafana URL")
    return cluster, str(urls[0])


def _cluster_dashboard_urls(args: argparse.Namespace) -> tuple[dict[str, Any], list[str]]:
    cluster = _find_cluster(args.cluster, getattr(args, "region", None))
    if args.dashboard_url:
        return cluster, [args.dashboard_url]
    urls = cluster.get("dashboard") or []
    if not urls:
        raise ValueError(f"cluster {args.cluster!r} has no TiDB Dashboard URL")
    return cluster, [str(url) for url in urls]


def _cluster_dashboard_name(cluster: dict[str, Any], override: str | None = None) -> str | None:
    value = str(override or "").strip()
    if value:
        return value
    if cluster.get("dashboard_direct") is True:
        return None
    value = str(cluster.get("dashboard_cluster_name") or "").strip()
    if value:
        return value
    raise ValueError(
        f"cluster {cluster.get('alias')!r} has no dashboard_cluster_name; pass --dashboard-cluster-name"
    )


def _cluster_datasource_selector(cluster: dict[str, Any], override_regex: str | None) -> dict[str, Any]:
    datasource_id = cluster.get("datasource_id")
    if datasource_id is not None and not str(datasource_id).strip():
        datasource_id = None
    return {
        "id": datasource_id,
        "uid": str(cluster.get("datasource_uid") or "") or None,
        "name_regex": override_regex or str(cluster.get("datasource_name_regex") or "") or None,
    }


def _session_dir(alias: str, grafana_url: str) -> Path:
    parsed = urllib.parse.urlparse(grafana_url)
    slug = _metric_group_id(f"{alias}-{parsed.netloc}")
    return SESSION_CACHE_ROOT / slug


def _session_dir_for_cluster(cluster: dict[str, Any], grafana_url: str) -> Path:
    # Reuse one browser profile per region+Grafana host so clusters sharing a
    # Grafana origin do not repeatedly go through P81/Grafana login.
    key = str(cluster.get("region") or cluster.get("alias") or "grafana")
    return _session_dir(key, grafana_url)


def _direct_session_path(session_dir: Path) -> Path:
    return session_dir / DIRECT_SESSION_FILENAME


def _save_direct_session_cookies(
    session_dir: Path, cookies: list[dict[str, Any]], *, user_agent: str | None = None
) -> None:
    session_dir.mkdir(parents=True, exist_ok=True)
    os.chmod(session_dir, 0o700)
    allowed_fields = ("name", "value", "domain", "path", "expires", "httpOnly", "secure", "sameSite")
    payload = [
        {key: cookie[key] for key in allowed_fields if key in cookie}
        for cookie in cookies
        if isinstance(cookie, dict) and cookie.get("name") and cookie.get("value") is not None
    ]
    path = _direct_session_path(session_dir)
    fd, temporary = tempfile.mkstemp(prefix=".direct-session-", dir=session_dir)
    try:
        os.fchmod(fd, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(
                {"version": 1, "cookies": payload, "userAgent": user_agent},
                handle,
                separators=(",", ":"),
            )
        os.replace(temporary, path)
        os.chmod(path, 0o600)
    except Exception:
        with contextlib.suppress(OSError):
            os.close(fd)
        with contextlib.suppress(OSError):
            os.unlink(temporary)
        raise


def _load_direct_session_cookies(session_dir: Path) -> list[dict[str, Any]]:
    path = _direct_session_path(session_dir)
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (FileNotFoundError, OSError, json.JSONDecodeError):
        return []
    cookies = payload.get("cookies") if isinstance(payload, dict) else None
    return [cookie for cookie in cookies or [] if isinstance(cookie, dict)]


def _load_direct_session_user_agent(session_dir: Path) -> str | None:
    try:
        payload = json.loads(_direct_session_path(session_dir).read_text(encoding="utf-8"))
    except (FileNotFoundError, OSError, json.JSONDecodeError):
        return None
    value = payload.get("userAgent") if isinstance(payload, dict) else None
    return str(value).strip() if value else None


def _cookie_header_for_url(cookies: list[dict[str, Any]], url: str, *, now: float | None = None) -> str:
    parsed = urllib.parse.urlparse(url)
    host = (parsed.hostname or "").lower()
    request_path = parsed.path or "/"
    current_time = time.time() if now is None else now
    matched: list[tuple[int, str]] = []
    for cookie in cookies:
        name = str(cookie.get("name") or "")
        value = str(cookie.get("value") or "")
        domain = str(cookie.get("domain") or host).lstrip(".").lower()
        path = str(cookie.get("path") or "/")
        expires = cookie.get("expires")
        if not name or not domain or (host != domain and not host.endswith("." + domain)):
            continue
        if not request_path.startswith(path):
            continue
        if cookie.get("secure") and parsed.scheme != "https":
            continue
        if isinstance(expires, (int, float)) and expires > 0 and expires <= current_time:
            continue
        matched.append((len(path), f"{name}={value}"))
    matched.sort(key=lambda item: item[0], reverse=True)
    return "; ".join(value for _length, value in matched)


def _cookie_jar(cookies: list[dict[str, Any]]) -> http.cookiejar.CookieJar:
    jar = http.cookiejar.CookieJar()
    for item in cookies:
        name = str(item.get("name") or "")
        value = str(item.get("value") or "")
        raw_domain = str(item.get("domain") or "")
        if not name or not raw_domain:
            continue
        expires_value = item.get("expires")
        expires = int(expires_value) if isinstance(expires_value, (int, float)) and expires_value > 0 else None
        jar.set_cookie(
            http.cookiejar.Cookie(
                version=0,
                name=name,
                value=value,
                port=None,
                port_specified=False,
                domain=raw_domain,
                domain_specified=raw_domain.startswith("."),
                domain_initial_dot=raw_domain.startswith("."),
                path=str(item.get("path") or "/"),
                path_specified=True,
                secure=bool(item.get("secure")),
                expires=expires,
                discard=expires is None,
                comment=None,
                comment_url=None,
                rest={"HttpOnly": bool(item.get("httpOnly"))},
                rfc2109=False,
            )
        )
    return jar


def _dashboard_session_dir_for_cluster(cluster: dict[str, Any], dashboard_url: str) -> Path:
    parsed = urllib.parse.urlparse(dashboard_url)
    key = f"{cluster.get('region') or cluster.get('alias') or 'dashboard'}-{parsed.netloc}-dashboard"
    slot = os.getenv("WETECH_DASHBOARD_CONCURRENCY_SLOT", "").strip()
    if slot:
        key = f"{key}-slot-{slot}"
    return SESSION_CACHE_ROOT / _metric_group_id(str(key))


def _origin_from_url(url: str) -> str:
    parsed = urllib.parse.urlparse(url)
    return f"{parsed.scheme}://{parsed.netloc}"


def _try_fill(locator: Any, value: str, timeout_ms: int = 1200) -> bool:
    try:
        target = locator.first
        target.wait_for(state="visible", timeout=timeout_ms)
        target.fill(value)
        return True
    except Exception:
        return False


def _try_click(locator: Any, timeout_ms: int = 1200) -> bool:
    try:
        target = locator.first
        target.wait_for(state="visible", timeout=timeout_ms)
        target.click()
        return True
    except Exception:
        return False


def _page_text(page: Any) -> str:
    try:
        return page.locator("body").inner_text(timeout=1000)
    except Exception:
        return ""


def _write_browser_diagnostic(page: Any, prefix: str, *, include_screenshot: bool = True) -> dict[str, str]:
    DIAGNOSTIC_ROOT.mkdir(parents=True, exist_ok=True)
    stamp = int(time.time())
    slug = _metric_group_id(prefix)
    screenshot = DIAGNOSTIC_ROOT / f"{stamp}-{slug}.png"
    text_path = DIAGNOSTIC_ROOT / f"{stamp}-{slug}.txt"
    info: dict[str, str] = {}
    if include_screenshot:
        try:
            page.screenshot(path=str(screenshot), full_page=True)
            info["screenshot"] = str(screenshot)
        except Exception as exc:
            info["screenshot_error"] = _sanitize_error_text(exc)
    try:
        text_path.write_text(f"url: {_redact_url(page.url)}\n\n{_page_text(page)[:4000]}\n", encoding="utf-8")
        info["page_text"] = str(text_path)
    except Exception as exc:
        info["page_text_error"] = _sanitize_error_text(exc)
    return info


def _redact_codes(text: str) -> str:
    return re.sub(r"\b\d{6,8}\b", "<redacted-code>", text)


def _is_p81_page(page: Any) -> bool:
    text = _page_text(page).lower()
    url = page.url.lower()
    return ("perimeter81.com" in url or "pzero.perimeter81.com" in url) and (
        "check point" in text
        or "sase" in text
        or "sign in with your email" in text
        or "verification" in text
        or "your session has ended" in text
        or "connect" in url
    )


def _is_p81_auth_choice_text(text: str, url: str) -> bool:
    lowered = text.lower()
    lowered_url = url.lower()
    return (
        "/sign-in" in lowered_url
        or "sign in with microsoft" in lowered
        or "sign in with your email" in lowered
        or "email and password" in lowered
    )


def _is_p81_auth_choice_page(page: Any) -> bool:
    return _is_p81_page(page) and _is_p81_auth_choice_text(_page_text(page), page.url)


def _is_grafana_login_page(page: Any) -> bool:
    text = _page_text(page).lower()
    url = page.url.lower()
    return "login" in url and ("grafana" in text or "email or username" in text or "username" in text)


def _is_dashboard_login_frame(frame: Any) -> bool:
    text = _page_text(frame).lower()
    url = frame.url.lower()
    return (
        "/dashboard/#/signin" in url
        or "sql user sign in" in text
        or ("username" in text and "password" in text and "sign in" in text)
    )


def _is_dashboard_authentication_rejected(frame: Any) -> bool:
    text = " ".join(_page_text(frame).lower().split())
    failed = "sign in failed" in text or "登录失败" in text
    rejected = (
        "tidb authentication failed" in text
        or ("tidb" in text and "身份验证失败" in text)
    )
    return failed and rejected


def _looks_like_2fa(page: Any) -> bool:
    text = _page_text(page).lower()
    if any(marker in text for marker in ("verification code", "authenticator", "two-factor", "2fa", "one-time")):
        return True
    try:
        visible_inputs = page.locator("input:visible").count()
    except Exception:
        visible_inputs = 0
    return _is_p81_page(page) and visible_inputs <= 3 and ("code" in text or "verify" in text)


def _server_error_marker(page: Any) -> str | None:
    text = _page_text(page).lower()
    markers = (
        "502 bad gateway",
        "bad gateway",
        "500 internal server error",
        "internal server error",
        "503 service unavailable",
        "service unavailable",
    )
    for marker in markers:
        if marker in text:
            return marker
    title = ""
    with contextlib.suppress(Exception):
        title = str(page.title()).lower()
    for marker in markers:
        if marker in title:
            return marker
    return None


def _is_p81_auth_failed_page(page: Any) -> bool:
    text = _page_text(page).lower()
    url = page.url.lower()
    is_perimeter_url = "perimeter81.com" in url or "pzero.perimeter81.com" in url
    has_auth_failed_marker = (
        "authorization failed" in text
        or "connection error: please login again" in text
        or "access is not allowed" in text
    )
    return is_perimeter_url and has_auth_failed_marker


def _recover_p81_auth_failed(page: Any) -> bool:
    if not _is_p81_auth_failed_page(page):
        return False
    parsed = urllib.parse.urlparse(page.url)
    from_payload = {
        "pathname": parsed.path or "/connect",
        "search": f"?{parsed.query}" if parsed.query else "",
        "hash": f"#{parsed.fragment}" if parsed.fragment else "",
    }
    sign_in_url = urllib.parse.urlunparse(
        parsed._replace(
            path="/sign-in",
            query=urllib.parse.urlencode({"from": json.dumps(from_payload, separators=(",", ":"))}),
            fragment="",
        )
    )
    with contextlib.suppress(Exception):
        page.goto(sign_in_url, wait_until="domcontentloaded", timeout=10000)
        page.wait_for_timeout(1000)
        return True
    if _try_click(page.get_by_role("button", name=re.compile(r"close|ok|关闭", re.I)), timeout_ms=1500):
        page.wait_for_timeout(2000)
        return True
    if _try_click(page.get_by_text(re.compile(r"close|ok|关闭", re.I)), timeout_ms=1500):
        page.wait_for_timeout(2000)
        return True
    return True


def _api_server_error(status: int, payload: Any) -> str | None:
    if status < 500:
        return None
    if isinstance(payload, str):
        return payload[:200]
    return f"HTTP {status}"


def _required_login_credentials(scope: str) -> dict[str, str]:
    credentials = LOGIN_CREDENTIALS.get(scope, {})
    username = credentials.get("username", "").strip()
    password = credentials.get("password", "").strip()
    if not username or not password:
        raise ProviderError(
            "provider_credentials_missing",
            f"Required {scope} credentials are not configured in the runtime environment.",
        )
    return {"username": username, "password": password}


def _fill_p81_login(page: Any) -> bool:
    credentials = _required_login_credentials("p81_vpn")
    filled_user = _try_fill(
        page.locator(
            "input[type='email'], input[autocomplete='username'], input[name*='email'], input[id*='email'], input[type='text']"
        ),
        credentials["username"],
    )
    password = page.locator("input[type='password']").first
    filled_pass = _try_fill(page.locator("input[type='password']"), credentials["password"])
    if filled_user or filled_pass:
        if filled_pass:
            password.press("Enter")
        else:
            _try_click(page.locator("form:has(input[type='password']) button[type='submit'], form:has(input[type='password']) button"))
        return True
    return False


def _open_p81_password_login(page: Any) -> bool:
    if not _is_p81_auth_choice_page(page):
        return False
    patterns = (
        r"sign in with your email",
        r"email and password",
        r"email.*password",
        r"password",
    )
    for pattern in patterns:
        regex = re.compile(pattern, re.I)
        if _try_click(page.get_by_role("button", name=regex), timeout_ms=1000):
            page.wait_for_timeout(1000)
            return True
        if _try_click(page.get_by_role("link", name=regex), timeout_ms=1000):
            page.wait_for_timeout(1000)
            return True
        try:
            locator = page.get_by_text(regex).filter(has_not_text=re.compile(r"microsoft", re.I))
            if _try_click(locator, timeout_ms=1000):
                page.wait_for_timeout(1000)
                return True
        except Exception:
            pass
    return False


def _fill_grafana_login(page: Any) -> bool:
    credentials = _required_login_credentials("grafana")
    filled_user = _try_fill(
        page.locator(
            "input[name='user'], input[name='username'], input[autocomplete='username'], input[type='email'], input[type='text']"
        ),
        credentials["username"],
    )
    password = page.locator("input[name='password'], input[type='password']").first
    filled_pass = _try_fill(page.locator("input[name='password'], input[type='password']"), credentials["password"])
    if filled_user or filled_pass:
        if filled_pass:
            password.press("Enter")
        else:
            _try_click(page.locator("form:has(input[type='password']) button[type='submit'], form:has(input[type='password']) button"))
        return True
    return False


def _fill_dashboard_login(frame: Any) -> bool:
    credentials = _required_login_credentials("dashboard")
    filled_user = _try_fill(
        frame.locator("input[name='username'], input[name='user'], input[placeholder*='Username'], input[type='text']"),
        credentials["username"],
        timeout_ms=2500,
    )
    password = frame.locator("input[name='password'], input[placeholder*='Password'], input[type='password']").first
    filled_pass = _try_fill(
        frame.locator("input[name='password'], input[placeholder*='Password'], input[type='password']"),
        credentials["password"],
        timeout_ms=2500,
    )
    if filled_user or filled_pass:
        if filled_pass:
            password.press("Enter")
        else:
            _try_click(frame.get_by_role("button", name=re.compile(r"sign in|log in|登录", re.I)), timeout_ms=1500)
        return True
    return False


def _prompt_and_fill_2fa(page: Any, *, headless: bool) -> bool:
    if not sys.stdin.isatty():
        raise RuntimeError("P81 2FA is required, but stdin is not interactive")
    if headless:
        print("P81 2FA required. Enter the current 6-digit code.", file=sys.stderr)
    else:
        print("P81 2FA required. You can enter it here or directly in the browser.", file=sys.stderr)
    code = getpass.getpass("P81 2FA code: ").strip()
    if not re.fullmatch(r"\d{6,8}", code):
        raise RuntimeError("2FA code must be 6-8 digits")
    _fill_and_submit_2fa_code(page, code)
    return True


def _fill_and_submit_2fa_code(page: Any, code: str) -> None:
    candidates = page.locator(
        "input[autocomplete='one-time-code'], input[name*='code'], input[id*='code'], input[type='tel'], input[type='text']"
    )
    try:
        count = candidates.count()
    except Exception:
        count = 0

    filled = False
    if count >= len(code):
        filled_digits = 0
        for index, digit in enumerate(code):
            try:
                field = candidates.nth(index)
                field.wait_for(state="visible", timeout=500)
                field.fill(digit)
                filled_digits += 1
            except Exception:
                break
        filled = filled_digits == len(code)

    if not filled:
        if not _try_fill(candidates, code, timeout_ms=3000):
            raise RuntimeError("could not find a visible 2FA code input")

    page.wait_for_timeout(300)
    with contextlib.suppress(Exception):
        page.keyboard.press("Enter")
    page.wait_for_timeout(800)

    button_patterns = (
        r"verify|continue|submit|sign in|log in|next",
        r"确认|继续|下一步|验证|登录",
    )
    for pattern in button_patterns:
        if _try_click(page.get_by_role("button", name=re.compile(pattern, re.I)), timeout_ms=1000):
            page.wait_for_timeout(1000)
            return
    if _try_click(page.locator("button[type='submit'], input[type='submit']"), timeout_ms=1000):
        page.wait_for_timeout(1000)
        return
    with contextlib.suppress(Exception):
        page.keyboard.press("Enter")


def _run_2fa_script(script_path: Path, timeout_seconds: int = 20) -> str:
    resolved = script_path.expanduser().resolve()
    if not resolved.exists():
        raise RuntimeError(f"2FA code script does not exist: {resolved}")
    result = subprocess.run(
        [sys.executable, str(resolved)],
        check=False,
        capture_output=True,
        text=True,
        timeout=timeout_seconds,
    )
    if result.returncode != 0:
        stderr = _redact_codes((result.stderr or "").strip())
        detail = f": {stderr}" if stderr else ""
        raise RuntimeError(f"2FA code script failed with exit code {result.returncode}{detail}")
    lines = [line.strip() for line in (result.stdout or "").splitlines() if line.strip()]
    if not lines:
        raise RuntimeError("2FA code script returned no code")
    code = lines[-1]
    if not re.fullmatch(r"\d{6,8}", code):
        raise RuntimeError("2FA code script output was not a 6-8 digit code")
    return code


def _script_and_fill_2fa(page: Any, script_path: Path) -> bool:
    code = _run_2fa_script(script_path)
    _fill_and_submit_2fa_code(page, code)
    return True


def _grafana_api_get(
    page: Any,
    origin: str,
    path: str,
    params: dict[str, str] | None = None,
    *,
    timeout_ms: int = 5000,
) -> tuple[int, Any]:
    url = origin.rstrip("/") + path
    if params:
        url += "?" + urllib.parse.urlencode(params)
    request_headers = {
        "Accept": "application/json, text/plain, */*",
        "Origin": origin.rstrip("/"),
        "Referer": origin.rstrip("/") + "/",
        "X-Grafana-Org-Id": "1",
    }
    try:
        response = page.context.request.get(url, headers=request_headers, timeout=timeout_ms)
    except Exception as exc:
        return 0, _sanitize_error_text(exc)
    status = response.status
    try:
        payload = response.json()
    except Exception:
        payload = response.text()[:500]
    return status, payload


def _grafana_page_fetch_get(
    page: Any,
    origin: str,
    path: str,
    params: dict[str, str] | None = None,
    *,
    timeout_ms: int = 5000,
) -> tuple[int, Any]:
    url = origin.rstrip("/") + path
    if params:
        url += "?" + urllib.parse.urlencode(params)
    try:
        result = page.evaluate(
            """async ({url, timeoutMs}) => {
                const controller = new AbortController();
                const timeout = setTimeout(() => controller.abort(), timeoutMs);
                try {
                    const response = await fetch(url, {
                        credentials: 'include',
                        headers: {
                            'Accept': 'application/json, text/plain, */*',
                            'X-Grafana-Org-Id': '1'
                        },
                        signal: controller.signal
                    });
                    const text = await response.text();
                    return {
                        status: response.status,
                        contentType: response.headers.get('content-type') || '',
                        text
                    };
                } finally {
                    clearTimeout(timeout);
                }
            }""",
            {"url": url, "timeoutMs": timeout_ms},
        )
    except Exception as exc:
        return 0, _sanitize_error_text(exc)[:500]
    raw = str(result.get("text") or "")
    try:
        payload = json.loads(raw)
    except json.JSONDecodeError:
        payload = raw[:500]
    return int(result.get("status") or 0), payload


def _grafana_api_get_with_page_fallback(
    page: Any,
    origin: str,
    path: str,
    params: dict[str, str] | None = None,
    *,
    timeout_ms: int = 5000,
) -> tuple[int, Any]:
    status, payload = _grafana_api_get(page, origin, path, params, timeout_ms=timeout_ms)
    if status != 403:
        return status, payload
    page_origin = _origin_from_url(page.url) if page.url.startswith(("http://", "https://")) else ""
    if page_origin.rstrip("/") != origin.rstrip("/"):
        return status, payload
    return _grafana_page_fetch_get(page, origin, path, params, timeout_ms=timeout_ms)


def _direct_grafana_api_get(
    cookies: list[dict[str, Any]],
    origin: str,
    path: str,
    params: dict[str, str] | None = None,
    *,
    timeout_ms: int = 5000,
    user_agent: str | None = None,
    max_response_bytes: int | None = None,
) -> tuple[int, Any]:
    url = origin.rstrip("/") + path
    if params:
        url += "?" + urllib.parse.urlencode(params)
    headers = {
        "Accept": "application/json, text/plain, */*",
        "Origin": origin.rstrip("/"),
        "Referer": origin.rstrip("/") + "/",
        # P81 binds the application authorization to the browser UA used when
        # the session was established. Reuse that public client metadata with
        # the exported cookies; no live browser context is required.
        "User-Agent": user_agent or "Mozilla/5.0 wetech-op-grafana-direct",
        "X-Grafana-Org-Id": "1",
    }
    request = urllib.request.Request(url, headers=headers, method="GET")
    # Let the cookie jar follow the P81 /connect authorization hop while
    # applying only the cookies that match each redirect host. A static Cookie
    # header would either fail the cross-host flow or leak cookies to it.
    opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(_cookie_jar(cookies)))
    try:
        with opener.open(request, timeout=max(timeout_ms / 1000, 0.1)) as response:
            status = int(response.status)
            final_url = response.geturl()
            content_length = response.headers.get("content-length")
            if max_response_bytes is not None and content_length:
                try:
                    parsed_content_length = int(content_length)
                except ValueError:
                    parsed_content_length = None
                if parsed_content_length is not None and parsed_content_length > max_response_bytes:
                    raise ProviderError(
                        "grafana_query_result_too_large",
                        f"Grafana response exceeds the {max_response_bytes}-byte result limit.",
                        detail={
                            "content_length": parsed_content_length,
                            "limit_bytes": max_response_bytes,
                        },
                    )
            raw = response.read(
                max_response_bytes + 1 if max_response_bytes is not None else -1
            )
            if max_response_bytes is not None and len(raw) > max_response_bytes:
                raise ProviderError(
                    "grafana_query_result_too_large",
                    f"Grafana response exceeds the {max_response_bytes}-byte result limit.",
                    detail={"observed_bytes": len(raw), "limit_bytes": max_response_bytes},
                )
            content_type = str(response.headers.get("content-type") or "")
    except urllib.error.HTTPError as exc:
        status = int(exc.code)
        final_url = exc.geturl()
        raw = exc.read()
        content_type = str(exc.headers.get("content-type") or "")
    except ProviderError:
        raise
    except Exception as exc:
        return 0, _sanitize_error_text(exc)
    if 300 <= status < 400:
        return 401, "authentication required"
    final_origin = _origin_from_url(final_url) if final_url.startswith(("http://", "https://")) else ""
    if final_origin.rstrip("/") != origin.rstrip("/") or "sign-in" in final_url.lower():
        return 401, "authentication required"
    text = raw.decode("utf-8", errors="replace")
    if "json" in content_type.lower():
        with contextlib.suppress(json.JSONDecodeError):
            return status, json.loads(text)
    with contextlib.suppress(json.JSONDecodeError):
        return status, json.loads(text)
    return status, text[:500]


def _dashboard_frame(page: Any) -> Any | None:
    if "/dashboard/" in (page.url or ""):
        return page.main_frame
    frames = [frame for frame in page.frames if "/dashboard/" in (frame.url or "")]
    return frames[-1] if frames else None


def _dashboard_app_url(origin: str, dashboard_cluster_name: str | None, route: str = "overview") -> str:
    if dashboard_cluster_name is None:
        return origin.rstrip("/") + f"/#/{route.lstrip('#/')}"
    return (
        origin.rstrip("/")
        + f"/proxy/{urllib.parse.quote(dashboard_cluster_name, safe='')}/dashboard/#/{route.lstrip('#/')}"
    )


def _dashboard_api_url(
    origin: str,
    dashboard_cluster_name: str | None,
    path: str,
    params: dict[str, Any] | None = None,
) -> str:
    query = ""
    if params:
        query = "?" + urllib.parse.urlencode({key: str(value) for key, value in params.items()})
    if dashboard_cluster_name is None:
        return origin.rstrip("/") + path + query
    return origin.rstrip("/") + f"/proxy/{urllib.parse.quote(dashboard_cluster_name, safe='')}" + path + query


def _dashboard_request_base(dashboard_url: str, dashboard_cluster_name: str | None) -> str:
    if dashboard_cluster_name is not None:
        return _origin_from_url(dashboard_url)
    parsed = urllib.parse.urlparse(dashboard_url)
    path = parsed.path.rstrip("/")
    if not path:
        raise ValueError("direct TiDB Dashboard URL must include its application base path")
    return urllib.parse.urlunparse((parsed.scheme, parsed.netloc, path, "", "", ""))


def _open_dashboard_iframe(
    page: Any,
    origin: str,
    dashboard_cluster_name: str | None,
    route: str = "overview",
) -> None:
    if dashboard_cluster_name is None:
        page.goto(
            _dashboard_app_url(origin, None, route),
            wait_until="domcontentloaded",
            timeout=30000,
        )
        return
    relative_url = f"/proxy/{urllib.parse.quote(dashboard_cluster_name, safe='')}/dashboard/"
    url = _dashboard_app_url(origin, dashboard_cluster_name, route)
    page.evaluate(
        """({clusterName, relativeUrl, url}) => {
            if (typeof window.openTab === 'function') {
                window.openTab(clusterName, relativeUrl);
                const iframe = document.getElementById('iframe-' + clusterName);
                if (iframe && url.includes('#/')) iframe.src = url;
                return;
            }
            const iframeId = 'iframe-' + clusterName;
            let iframe = document.getElementById(iframeId);
            if (!iframe) {
                iframe = document.createElement('iframe');
                iframe.id = iframeId;
                iframe.style.width = '100%';
                iframe.style.height = '100%';
                iframe.style.border = '0';
                iframe.className = 'active';
                const container = document.querySelector('#content-body') || document.body;
                container.appendChild(iframe);
            }
            iframe.src = url;
        }""",
        {"clusterName": dashboard_cluster_name, "relativeUrl": relative_url, "url": url},
    )


def _navigate_dashboard_app(page: Any, origin: str, dashboard_cluster_name: str | None, route: str) -> None:
    frame = _dashboard_frame(page)
    if frame is not None and frame is not page.main_frame:
        frame.goto(_dashboard_app_url(origin, dashboard_cluster_name, route), wait_until="domcontentloaded", timeout=30000)
    elif "/dashboard/" in (page.url or ""):
        page.goto(_dashboard_app_url(origin, dashboard_cluster_name, route), wait_until="domcontentloaded", timeout=30000)
    else:
        _open_dashboard_iframe(page, origin, dashboard_cluster_name, route)


def _dashboard_auth_token(page: Any, dashboard_cluster_name: str | None) -> str | None:
    token_key = dashboard_cluster_name or "direct"
    with contextlib.suppress(Exception):
        token = page.evaluate(
            """(clusterName) =>
                localStorage.getItem(clusterName + '_dashboard_auth_token') ||
                localStorage.getItem('dashboard_auth_token')""",
            token_key,
        )
        if isinstance(token, str) and token.strip():
            return token.strip()
    return None


def _clear_dashboard_auth_token(page: Any, dashboard_cluster_name: str | None) -> None:
    token_key = dashboard_cluster_name or "direct"
    with contextlib.suppress(Exception):
        page.evaluate(
            """(clusterName) => {
                localStorage.removeItem(clusterName + '_dashboard_auth_token');
                localStorage.removeItem(clusterName + '_dashboard_last_login_username');
                localStorage.removeItem('dashboard_auth_token');
                localStorage.removeItem('dashboard_last_login_username');
            }""",
            token_key,
        )


def _dashboard_api_get(
    page: Any,
    origin: str,
    dashboard_cluster_name: str | None,
    path: str,
    params: dict[str, Any] | None = None,
    *,
    timeout_ms: int = 30000,
) -> tuple[int, Any]:
    token = _dashboard_auth_token(page, dashboard_cluster_name)
    if not token:
        return 401, {"error": True, "message": "missing dashboard auth token"}
    url = _dashboard_api_url(origin, dashboard_cluster_name, path, params)
    evaluator = _dashboard_frame(page) or page
    for attempt in range(2):
        try:
            result = evaluator.evaluate(
                """async ({url, token, timeoutMs}) => {
                    const controller = new AbortController();
                    const timeout = setTimeout(() => controller.abort(), timeoutMs);
                    try {
                        const response = await fetch(url, {
                            credentials: 'include',
                            headers: {
                                'Accept': 'application/json, text/plain, */*',
                                'Authorization': 'Bearer ' + token
                            },
                            signal: controller.signal
                        });
                        const text = await response.text();
                        return {
                            status: response.status,
                            contentType: response.headers.get('content-type') || '',
                            text
                        };
                    } finally {
                        clearTimeout(timeout);
                    }
                }""",
                {"url": url, "token": token, "timeoutMs": timeout_ms},
            )
        except Exception as exc:
            return 0, _sanitize_error_text(exc)[:500]
        raw = str(result.get("text") or "")
        try:
            payload = json.loads(raw)
        except json.JSONDecodeError:
            payload = raw[:500]
        status = int(result.get("status") or 0)
        if attempt == 0 and status in {500, 502, 503, 504}:
            with contextlib.suppress(Exception):
                page.wait_for_timeout(250)
            continue
        return status, payload
    raise AssertionError("unreachable dashboard retry state")


def _dashboard_api_post_json(
    page: Any,
    origin: str,
    dashboard_cluster_name: str | None,
    path: str,
    body: dict[str, Any],
    *,
    timeout_ms: int = 30000,
) -> tuple[int, Any]:
    token = _dashboard_auth_token(page, dashboard_cluster_name)
    if not token:
        return 401, {"error": True, "message": "missing dashboard auth token"}
    url = _dashboard_api_url(origin, dashboard_cluster_name, path)
    evaluator = _dashboard_frame(page) or page
    for attempt in range(2):
        try:
            result = evaluator.evaluate(
                """async ({url, token, timeoutMs, body}) => {
                    const controller = new AbortController();
                    const timeout = setTimeout(() => controller.abort(), timeoutMs);
                    try {
                        const response = await fetch(url, {
                            method: 'POST',
                            credentials: 'include',
                            headers: {
                                'Accept': 'application/json, text/plain, */*',
                                'Authorization': 'Bearer ' + token,
                                'Content-Type': 'application/json'
                            },
                            body: JSON.stringify(body),
                            signal: controller.signal
                        });
                        const text = await response.text();
                        return {
                            status: response.status,
                            contentType: response.headers.get('content-type') || '',
                            text
                        };
                    } finally {
                        clearTimeout(timeout);
                    }
                }""",
                {"url": url, "token": token, "timeoutMs": timeout_ms, "body": body},
            )
        except Exception as exc:
            return 0, _sanitize_error_text(exc)[:500]
        raw = str(result.get("text") or "")
        try:
            payload = json.loads(raw)
        except json.JSONDecodeError:
            payload = raw[:500]
        status = int(result.get("status") or 0)
        if attempt == 0 and status in {500, 502, 503, 504}:
            with contextlib.suppress(Exception):
                page.wait_for_timeout(250)
            continue
        return status, payload
    raise AssertionError("unreachable dashboard retry state")


def _ensure_dashboard_session(
    page: Any,
    dashboard_url: str,
    dashboard_cluster_name: str | None,
    *,
    headless: bool,
    terminal_2fa: bool = False,
    twofa_script: Path = DEFAULT_2FA_SCRIPT,
    login_timeout_seconds: int = 300,
    debug_login: bool = False,
) -> str:
    origin = _dashboard_request_base(dashboard_url, dashboard_cluster_name)
    page.goto(_dashboard_api_url(origin, dashboard_cluster_name, "/api/info/info"), wait_until="domcontentloaded", timeout=60000)
    reported_2fa = False
    refreshed_dashboard_login = False
    submitted_dashboard_login = False
    last_state = "unknown"
    deadline = time.time() + login_timeout_seconds
    last_debug = 0.0

    while time.time() < deadline:
        page.wait_for_timeout(1000)
        if _is_p81_page(page) and _recover_p81_auth_failed(page):
            last_state = "p81_auth_failed_recover"
            continue
        if "/api/" in (page.url or "") and not _is_p81_page(page):
            page.goto(_dashboard_app_url(origin, dashboard_cluster_name, "overview"), wait_until="domcontentloaded", timeout=30000)
            last_state = "dashboard_direct_open"
            page.wait_for_timeout(2000)
            continue
        status, whoami = _dashboard_api_get(page, origin, dashboard_cluster_name, "/api/info/whoami", timeout_ms=2500)
        if status == 200 and isinstance(whoami, dict) and not whoami.get("error"):
            return origin
        if "cluster not found" in _page_text(page).lower():
            diagnostic = _write_browser_diagnostic(page, "dashboard-cluster-not-found")
            raise RuntimeError(
                "TiDB Dashboard proxy cluster route was not found; "
                f"diagnostic={diagnostic}"
            )
        auth_frame = _dashboard_frame(page)
        if status == 0 and auth_frame is not None and "network connection error" in _page_text(page).lower():
            page.goto(_dashboard_api_url(origin, dashboard_cluster_name, "/api/info/info"), wait_until="domcontentloaded", timeout=30000)
            last_state = "dashboard_api_auth_probe"
            page.wait_for_timeout(2000)
            continue
        if (
            status == 401
            and auth_frame is not None
            and not _is_dashboard_login_frame(auth_frame)
            and not refreshed_dashboard_login
        ):
            _clear_dashboard_auth_token(page, dashboard_cluster_name)
            _navigate_dashboard_app(page, origin, dashboard_cluster_name, "signin")
            refreshed_dashboard_login = True
            last_state = "dashboard_token_refresh"
            page.wait_for_timeout(2000)
            continue
        if auth_frame is None and not _is_p81_page(page):
            with contextlib.suppress(Exception):
                page.goto(
                    _dashboard_app_url(origin, dashboard_cluster_name, "overview"),
                    wait_until="domcontentloaded",
                    timeout=30000,
                )
                last_state = "dashboard_direct_open"
                page.wait_for_timeout(2000)
                continue
        if debug_login and time.time() - last_debug >= 5:
            text = _page_text(page).replace("\n", " ")[:180]
            frame = _dashboard_frame(page)
            frame_url = frame.url if frame else ""
            print(
                f"dashboard-login-state url={page.url!r} frame={frame_url!r} "
                f"whoami_status={status!r} state={last_state!r} text={text!r}",
                file=sys.stderr,
            )
            last_debug = time.time()
        api_error = _api_server_error(status, whoami)
        if api_error:
            diagnostic = _write_browser_diagnostic(page, "dashboard-api-server-error")
            raise RuntimeError(
                f"TiDB Dashboard API returned server error HTTP {status}: {api_error!r}; diagnostic={diagnostic}"
            )
        if not headless:
            page_server_error = _server_error_marker(page)
            if page_server_error:
                diagnostic = _write_browser_diagnostic(page, "dashboard-browser-server-error")
                raise RuntimeError(f"Browser page shows server error {page_server_error!r}; diagnostic={diagnostic}")

        frame = _dashboard_frame(page)
        if (
            submitted_dashboard_login
            and frame is not None
            and _is_dashboard_authentication_rejected(frame)
        ):
            diagnostic = _write_browser_diagnostic(
                page, "dashboard-authentication-rejected"
            )
            raise ProviderError(
                "wetech_dashboard_authentication_failed",
                "TiDB Dashboard rejected the configured SQL credentials.",
                detail={"diagnostic": diagnostic},
            )
        if frame is not None and _is_dashboard_login_frame(frame) and _fill_dashboard_login(frame):
            submitted_dashboard_login = True
            last_state = "dashboard_login"
            page.wait_for_timeout(3000)
            continue
        if _looks_like_2fa(page):
            last_state = "p81_2fa"
            if terminal_2fa:
                _prompt_and_fill_2fa(page, headless=headless)
            else:
                if not reported_2fa:
                    print(f"P81 2FA required. Fetching code through {twofa_script}.", file=sys.stderr)
                    reported_2fa = True
                _script_and_fill_2fa(page, twofa_script)
            with contextlib.suppress(Exception):
                page.wait_for_load_state("domcontentloaded", timeout=30000)
            continue
        if _is_p81_page(page) and _fill_p81_login(page):
            last_state = "p81_login"
            page.wait_for_load_state("domcontentloaded", timeout=30000)
            continue
        if _is_p81_page(page) and _open_p81_password_login(page):
            last_state = "p81_password_login_choice"
            continue
        if _is_p81_page(page) and _try_continue_from_p81(page):
            last_state = "p81_continue"
            continue

        if not headless:
            print("Waiting for TiDB Dashboard login to complete...", file=sys.stderr)
        last_state = "waiting"
        page.wait_for_timeout(2000)

    diagnostic = _write_browser_diagnostic(page, "dashboard-session-not-ready")
    raise RuntimeError(
        f"TiDB Dashboard session is not ready after {login_timeout_seconds}s; "
        f"last_state={last_state}; diagnostic={diagnostic}"
    )


def _is_datasource_list(status: int, payload: Any) -> bool:
    return status == 200 and isinstance(payload, list)


def _is_permission_denied(status: int, payload: Any) -> bool:
    if status != 403:
        return False
    if isinstance(payload, dict):
        text = str(payload.get("message") or "").lower()
    else:
        text = str(payload).lower()
    return "permission denied" in text or "origin not allowed" in text


def _looks_like_grafana_home(page: Any) -> bool:
    text = _page_text(page).lower()
    url = page.url.lower()
    return "pzero.perimeter81.com" in url and "login" not in url and (
        "general / home" in text or "dashboards" in text or "welcome to grafana" in text
    )


def _collect_panel_datasources(obj: Any, out: list[Any]) -> None:
    if isinstance(obj, dict):
        datasource = obj.get("datasource")
        if datasource:
            out.append(datasource)
        for key in ("panels", "rows"):
            for item in obj.get(key) or []:
                _collect_panel_datasources(item, out)
    elif isinstance(obj, list):
        for item in obj:
            _collect_panel_datasources(item, out)


def _dashboard_panel_datasources(page: Any, origin: str, limit: int = 12) -> list[dict[str, Any]]:
    status, search = _grafana_api_get_with_page_fallback(
        page,
        origin,
        "/api/search",
        {"type": "dash-db"},
        timeout_ms=10000,
    )
    if status != 200 or not isinstance(search, list):
        raise RuntimeError(f"could not search Grafana dashboards: HTTP {status}: {search!r}")
    raw_datasources: list[Any] = []
    for row in search[:limit]:
        uid = row.get("uid") if isinstance(row, dict) else None
        if not uid:
            continue
        dash_status, dashboard_payload = _grafana_api_get_with_page_fallback(
            page,
            origin,
            f"/api/dashboards/uid/{uid}",
            timeout_ms=10000,
        )
        if dash_status == 200 and isinstance(dashboard_payload, dict):
            _collect_panel_datasources(dashboard_payload.get("dashboard"), raw_datasources)

    rows: list[dict[str, Any]] = []
    seen: set[str] = set()
    for datasource in raw_datasources:
        if isinstance(datasource, dict):
            name = str(datasource.get("name") or datasource.get("uid") or "").strip()
            uid = str(datasource.get("uid") or "").strip() or None
            dtype = str(datasource.get("type") or "").strip() or None
        else:
            name = str(datasource).strip()
            uid = None
            dtype = None
        if not name:
            continue
        key = json.dumps({"name": name, "uid": uid, "type": dtype}, sort_keys=True)
        if key in seen:
            continue
        seen.add(key)
        rows.append({"name": name, "uid": uid, "type": dtype or "prometheus", "source": "dashboard_panels"})
    return rows


def _direct_dashboard_panel_datasources(
    cookies: list[dict[str, Any]], origin: str, limit: int = 12, *, user_agent: str | None = None
) -> list[dict[str, Any]]:
    status, search = _direct_grafana_api_get(
        cookies, origin, "/api/search", {"type": "dash-db"}, timeout_ms=10000, user_agent=user_agent
    )
    if status != 200 or not isinstance(search, list):
        raise RuntimeError(f"could not search Grafana dashboards: HTTP {status}: {search!r}")
    raw_datasources: list[Any] = []
    for row in search[:limit]:
        uid = row.get("uid") if isinstance(row, dict) else None
        if not uid:
            continue
        dash_status, dashboard_payload = _direct_grafana_api_get(
            cookies, origin, f"/api/dashboards/uid/{uid}", timeout_ms=10000, user_agent=user_agent
        )
        if dash_status == 200 and isinstance(dashboard_payload, dict):
            _collect_panel_datasources(dashboard_payload.get("dashboard"), raw_datasources)
    rows: list[dict[str, Any]] = []
    seen: set[str] = set()
    for datasource in raw_datasources:
        if isinstance(datasource, dict):
            name = str(datasource.get("name") or datasource.get("uid") or "").strip()
            uid = str(datasource.get("uid") or "").strip() or None
            dtype = str(datasource.get("type") or "").strip() or None
        else:
            name = str(datasource).strip()
            uid = None
            dtype = None
        if not name:
            continue
        key = json.dumps({"name": name, "uid": uid, "type": dtype}, sort_keys=True)
        if key in seen:
            continue
        seen.add(key)
        rows.append({"name": name, "uid": uid, "type": dtype or "prometheus", "source": "dashboard_panels"})
    return rows


def _try_continue_from_p81(page: Any) -> bool:
    if _is_p81_auth_choice_page(page):
        return False
    clicked = False
    for pattern in (
        r"open|launch|continue|connect|go to|proceed",
        r"tidb|grafana",
    ):
        clicked = _try_click(page.get_by_role("button", name=re.compile(pattern, re.I)), timeout_ms=1000)
        if clicked:
            break
        clicked = _try_click(page.get_by_role("link", name=re.compile(pattern, re.I)), timeout_ms=1000)
        if clicked:
            break
    if clicked:
        page.wait_for_timeout(3000)
    return clicked


def _ensure_grafana_session(
    page: Any,
    grafana_url: str,
    *,
    headless: bool,
    terminal_2fa: bool = False,
    twofa_script: Path = DEFAULT_2FA_SCRIPT,
    login_timeout_seconds: int = 300,
    debug_login: bool = False,
) -> str:
    origin = _origin_from_url(grafana_url)
    page.goto(grafana_url, wait_until="domcontentloaded", timeout=60000)
    reported_2fa = False
    last_state = "unknown"
    deadline = time.time() + login_timeout_seconds
    last_debug = 0.0

    while time.time() < deadline:
        page.wait_for_timeout(1000)
        if _is_p81_page(page) and _recover_p81_auth_failed(page):
            last_state = "p81_auth_failed_recover"
            continue
        current = urllib.parse.urlparse(page.url)
        if current.scheme in {"http", "https"} and current.netloc and "perimeter81.com" in current.netloc:
            origin = f"{current.scheme}://{current.netloc}"
        status, payload = _grafana_api_get_with_page_fallback(page, origin, "/api/datasources", timeout_ms=10000)
        if _is_datasource_list(status, payload):
            return origin
        if _is_permission_denied(status, payload) and _looks_like_grafana_home(page):
            return origin
        if debug_login and time.time() - last_debug >= 5:
            text = _page_text(page).replace("\n", " ")[:160]
            print(
                f"login-state url={page.url!r} probe_origin={origin!r} "
                f"datasources_status={status!r} state={last_state!r} text={text!r}",
                file=sys.stderr,
            )
            last_debug = time.time()
        api_error = _api_server_error(status, payload)
        if api_error:
            diagnostic = _write_browser_diagnostic(page, "grafana-api-server-error")
            raise RuntimeError(
                f"Grafana datasource API returned server error HTTP {status}: {api_error!r}; diagnostic={diagnostic}"
            )
        page_server_error = _server_error_marker(page)
        if page_server_error:
            diagnostic = _write_browser_diagnostic(page, "browser-server-error")
            raise RuntimeError(f"Browser page shows server error {page_server_error!r}; diagnostic={diagnostic}")

        if _looks_like_2fa(page):
            last_state = "p81_2fa"
            if terminal_2fa:
                _prompt_and_fill_2fa(page, headless=headless)
            else:
                if not reported_2fa:
                    print(f"P81 2FA required. Fetching code through {twofa_script}.", file=sys.stderr)
                    reported_2fa = True
                _script_and_fill_2fa(page, twofa_script)
            with contextlib.suppress(Exception):
                page.wait_for_load_state("domcontentloaded", timeout=30000)
            continue
        if _is_p81_page(page) and _fill_p81_login(page):
            last_state = "p81_login"
            page.wait_for_load_state("domcontentloaded", timeout=30000)
            continue
        if _is_p81_page(page) and _open_p81_password_login(page):
            last_state = "p81_password_login_choice"
            continue
        if _is_p81_page(page) and _try_continue_from_p81(page):
            last_state = "p81_continue"
            continue
        if _is_grafana_login_page(page) and _fill_grafana_login(page):
            last_state = "grafana_login"
            page.wait_for_load_state("domcontentloaded", timeout=30000)
            continue

        if not headless:
            print("Waiting for browser login to complete...", file=sys.stderr)
        last_state = "waiting"
        page.wait_for_timeout(2000)

    status, payload = _grafana_api_get_with_page_fallback(page, origin, "/api/datasources")
    diagnostic = _write_browser_diagnostic(
        page,
        "grafana-session-not-ready",
        include_screenshot=last_state != "p81_2fa" and not _looks_like_2fa(page),
    )
    if last_state == "p81_2fa" or _looks_like_2fa(page):
        raise RuntimeError(
            f"P81 2FA was not completed within {login_timeout_seconds}s; diagnostic={diagnostic}"
        )
    raise RuntimeError(
        f"Grafana session is not ready after {login_timeout_seconds}s; "
        f"last_state={last_state}; /api/datasources returned {status}: {payload!r}; diagnostic={diagnostic}"
    )


def _cached_datasources(page: Any, grafana_url: str) -> tuple[str, list[dict[str, Any]]] | None:
    origin = _origin_from_url(grafana_url)
    status, payload = _grafana_api_get_with_page_fallback(page, origin, "/api/datasources", timeout_ms=5000)
    if _is_datasource_list(status, payload):
        return origin, payload
    return None


def _ensure_datasources(
    page: Any,
    grafana_url: str,
    *,
    headless: bool,
    terminal_2fa: bool,
    twofa_script: Path,
    login_timeout_seconds: int,
    debug_login: bool,
) -> tuple[str, list[dict[str, Any]]]:
    cached = _cached_datasources(page, grafana_url)
    if cached:
        return cached
    origin = _ensure_grafana_session(
        page,
        grafana_url,
        headless=headless,
        terminal_2fa=terminal_2fa,
        twofa_script=twofa_script,
        login_timeout_seconds=login_timeout_seconds,
        debug_login=debug_login,
    )
    ds_status, datasources = _grafana_api_get_with_page_fallback(page, origin, "/api/datasources")
    if ds_status != 200 or not isinstance(datasources, list):
        if _is_permission_denied(ds_status, datasources) and _looks_like_grafana_home(page):
            panel_datasources = _dashboard_panel_datasources(page, origin)
            if panel_datasources:
                return origin, panel_datasources
        raise RuntimeError(f"could not list Grafana datasources: HTTP {ds_status}: {datasources!r}")
    return origin, datasources


def _select_prometheus_datasource(
    datasources: list[dict[str, Any]],
    *,
    selector: dict[str, str | None],
    cluster: dict[str, Any],
    shared_grafana: bool,
) -> dict[str, Any]:
    candidates = [row for row in datasources if str(row.get("type") or "").lower() == "prometheus"]
    uid = selector.get("uid")
    if uid:
        uid_candidates = [row for row in candidates if str(row.get("uid") or "") == uid]
        if uid_candidates:
            return uid_candidates[0]
        names = [str(row.get("name") or row.get("uid") or row.get("id")) for row in candidates]
        raise RuntimeError(f"no Prometheus datasource matched configured uid {uid!r}; available datasources: {names}")

    name_regex = selector.get("name_regex")
    if name_regex:
        pattern = re.compile(name_regex, re.I)
        candidates = [row for row in candidates if pattern.search(str(row.get("name") or ""))]
        if len(candidates) == 1:
            return candidates[0]
        if len(candidates) > 1:
            names = [str(row.get("name") or row.get("uid") or row.get("id")) for row in candidates]
            raise RuntimeError(f"datasource regex {name_regex!r} matched multiple Prometheus datasources: {names}")
    if not candidates:
        names = [str(row.get("name") or row.get("uid") or row.get("id")) for row in datasources]
        raise RuntimeError(f"no matching Prometheus datasource found; available datasources: {names}")
    if shared_grafana:
        names = [str(row.get("name") or row.get("uid") or row.get("id")) for row in candidates]
        alias = cluster.get("alias") or "<unknown>"
        raise RuntimeError(
            f"cluster {alias!r} shares a Grafana origin and has no unique datasource selector; "
            f"run rebuild-datasource-map and configure datasource_uid or datasource_name_regex. "
            f"available Prometheus datasources: {names}"
        )
    default_candidates = [row for row in candidates if row.get("isDefault")]
    return (default_candidates or candidates)[0]


def _datasource_proxy_path(
    page: Any,
    origin: str,
    datasource: dict[str, Any],
    api_path: str = "/api/v1/query_range",
) -> str:
    normalized_api_path = "/" + api_path.lstrip("/")
    datasource_uid = str(datasource.get("uid") or "").strip()
    if datasource_uid:
        return (
            "/api/datasources/proxy/uid/"
            + urllib.parse.quote(datasource_uid, safe="")
            + normalized_api_path
        )

    datasource_id = datasource.get("id")
    if datasource_id is not None:
        return (
            f"/api/datasources/proxy/{urllib.parse.quote(str(datasource_id), safe='')}"
            + normalized_api_path
        )

    datasource_name = str(datasource.get("name") or "").strip()
    if not datasource_name:
        raise RuntimeError(f"selected datasource has no uid, id, or name: {datasource!r}")

    encoded_name = urllib.parse.quote(datasource_name, safe="")
    status, resolved = _grafana_api_get_with_page_fallback(
        page,
        origin,
        f"/api/datasources/name/{encoded_name}",
        timeout_ms=10000,
    )
    if status == 200 and isinstance(resolved, dict):
        resolved_uid = str(resolved.get("uid") or "").strip()
        if resolved_uid:
            return (
                "/api/datasources/proxy/uid/"
                + urllib.parse.quote(resolved_uid, safe="")
                + normalized_api_path
            )
        resolved_id = resolved.get("id")
        if resolved_id is not None:
            return (
                f"/api/datasources/proxy/{urllib.parse.quote(str(resolved_id), safe='')}"
                + normalized_api_path
            )

    id_status, id_payload = _grafana_api_get_with_page_fallback(
        page,
        origin,
        f"/api/datasources/id/{encoded_name}",
        timeout_ms=10000,
    )
    resolved_id = id_payload.get("id") if isinstance(id_payload, dict) else id_payload
    if id_status == 200 and resolved_id is not None and str(resolved_id).strip():
        return (
            f"/api/datasources/proxy/{urllib.parse.quote(str(resolved_id), safe='')}"
            + normalized_api_path
        )

    raise RuntimeError(
        "selected datasource name could not be resolved to a Grafana uid or id: "
        f"name={datasource_name!r}, name_status={status}, id_status={id_status}"
    )


def _direct_datasource_proxy_path(
    cookies: list[dict[str, Any]],
    origin: str,
    datasource: dict[str, Any],
    api_path: str = "/api/v1/query_range",
    *,
    user_agent: str | None = None,
) -> str:
    normalized_api_path = "/" + api_path.lstrip("/")
    datasource_uid = str(datasource.get("uid") or "").strip()
    if datasource_uid:
        return "/api/datasources/proxy/uid/" + urllib.parse.quote(datasource_uid, safe="") + normalized_api_path
    datasource_id = datasource.get("id")
    if datasource_id is not None:
        return f"/api/datasources/proxy/{urllib.parse.quote(str(datasource_id), safe='')}" + normalized_api_path
    datasource_name = str(datasource.get("name") or "").strip()
    if not datasource_name:
        raise RuntimeError(f"selected datasource has no uid, id, or name: {datasource!r}")
    encoded_name = urllib.parse.quote(datasource_name, safe="")
    status, resolved = _direct_grafana_api_get(
        cookies, origin, f"/api/datasources/name/{encoded_name}", timeout_ms=10000, user_agent=user_agent
    )
    if status == 200 and isinstance(resolved, dict):
        resolved_uid = str(resolved.get("uid") or "").strip()
        if resolved_uid:
            return "/api/datasources/proxy/uid/" + urllib.parse.quote(resolved_uid, safe="") + normalized_api_path
        if resolved.get("id") is not None:
            return f"/api/datasources/proxy/{urllib.parse.quote(str(resolved['id']), safe='')}" + normalized_api_path
    id_status, id_payload = _direct_grafana_api_get(
        cookies, origin, f"/api/datasources/id/{encoded_name}", timeout_ms=10000, user_agent=user_agent
    )
    resolved_id = id_payload.get("id") if isinstance(id_payload, dict) else id_payload
    if id_status == 200 and resolved_id is not None and str(resolved_id).strip():
        return f"/api/datasources/proxy/{urllib.parse.quote(str(resolved_id), safe='')}" + normalized_api_path
    raise RuntimeError(
        "selected datasource name could not be resolved to a Grafana uid or id: "
        f"name={datasource_name!r}, name_status={status}, id_status={id_status}"
    )


def _refresh_direct_session(
    args: argparse.Namespace, session_dir: Path, grafana_url: str
) -> tuple[str, list[dict[str, Any]], str]:
    sync_playwright, _PlaywrightTimeoutError = _import_playwright()
    with _profile_lock(session_dir):
        with sync_playwright() as playwright:
            context = None
            try:
                context = playwright.chromium.launch_persistent_context(
                    str(session_dir),
                    headless=args.headless,
                    viewport={"width": 1440, "height": 1000},
                )
                page = context.pages[0] if context.pages else context.new_page()
                origin = _ensure_grafana_session(
                    page,
                    grafana_url,
                    headless=args.headless,
                    terminal_2fa=args.terminal_2fa,
                    twofa_script=args.twofa_script,
                    login_timeout_seconds=args.login_timeout,
                    debug_login=args.debug_login,
                )
                cookies = context.cookies()
                user_agent = str(page.evaluate("navigator.userAgent"))
                _save_direct_session_cookies(session_dir, cookies, user_agent=user_agent)
                return origin, cookies, user_agent
            finally:
                if context is not None:
                    context.close()


def _direct_query_range(
    cookies: list[dict[str, Any]],
    origin: str,
    cluster: dict[str, Any],
    datasource_selector: dict[str, Any],
    shared_grafana: bool,
    args: argparse.Namespace,
    start: str,
    end: str,
    step: str,
    user_agent: str | None,
) -> tuple[dict[str, Any], str, int, Any]:
    configured_uid = str(datasource_selector.get("uid") or "").strip()
    if configured_uid:
        datasource = {
            "uid": configured_uid,
            "name": str(cluster.get("datasource_name") or "").strip(),
            "type": "prometheus",
            "source": "configured_exact_uid",
        }
    elif datasource_selector.get("id") is not None:
        datasource = {
            "id": datasource_selector["id"],
            "name": str(cluster.get("datasource_name") or "").strip(),
            "type": "prometheus",
            "source": "configured_exact_id",
        }
    else:
        ds_status, datasources = _direct_grafana_api_get(
            cookies, origin, "/api/datasources", timeout_ms=10000, user_agent=user_agent
        )
        if ds_status == 401 or ds_status == 0:
            raise PermissionError("Grafana direct session is not authenticated")
        if ds_status != 200 or not isinstance(datasources, list):
            if _is_permission_denied(ds_status, datasources):
                datasource_name = str(cluster.get("datasource_name") or "").strip()
                exact_name_patterns = {f"^{re.escape(datasource_name)}$"} if datasource_name else set()
                if datasource_name and re.fullmatch(r"[A-Za-z0-9_-]+", datasource_name):
                    exact_name_patterns.add(f"^{datasource_name}$")
                if datasource_name and datasource_selector.get("name_regex") in exact_name_patterns:
                    datasources = [
                        {
                            "name": datasource_name,
                            "type": "prometheus",
                            "source": "configured_exact_name",
                        }
                    ]
                else:
                    datasources = _direct_dashboard_panel_datasources(cookies, origin, user_agent=user_agent)
            else:
                raise RuntimeError(f"could not list Grafana datasources: HTTP {ds_status}: {datasources!r}")
        datasource = _select_prometheus_datasource(
            datasources,
            selector=datasource_selector,
            cluster=cluster,
            shared_grafana=shared_grafana,
        )
    proxy_path = _direct_datasource_proxy_path(cookies, origin, datasource, user_agent=user_agent)
    query_status, payload = _direct_grafana_api_get(
        cookies,
        origin,
        proxy_path,
        {"query": args.query, "start": start, "end": end, "step": step},
        timeout_ms=30000,
        user_agent=user_agent,
        max_response_bytes=MAX_PROMETHEUS_RESPONSE_BYTES,
    )
    if query_status in {0, 401, 403}:
        raise PermissionError("Grafana direct session expired during query")
    return datasource, proxy_path, query_status, payload


def cmd_browser_query_range(args: argparse.Namespace) -> int:
    try:
        cluster, grafana_url = _cluster_grafana_url(args)
        start, end, step = _resolve_range_args(args)
        session_dir = _session_dir_for_cluster(cluster, grafana_url)
        all_clusters = _load_clusters()
        shared_grafana = bool(_shared_grafana_clusters(all_clusters, cluster))
        datasource_selector = _cluster_datasource_selector(cluster, args.datasource_name_regex)
        session_dir.mkdir(parents=True, exist_ok=True)
        os.chmod(session_dir, 0o700)
        origin = _origin_from_url(grafana_url)
        cookies = _load_direct_session_cookies(session_dir)
        user_agent = _load_direct_session_user_agent(session_dir)
        refreshed = False
        while True:
            if not cookies:
                origin, cookies, user_agent = _refresh_direct_session(args, session_dir, grafana_url)
                refreshed = True
            try:
                datasource, proxy_path, query_status, payload = _direct_query_range(
                    cookies,
                    origin,
                    cluster,
                    datasource_selector,
                    shared_grafana,
                    args,
                    start,
                    end,
                    step,
                    user_agent,
                )
                break
            except PermissionError:
                if refreshed:
                    raise
                origin, cookies, user_agent = _refresh_direct_session(args, session_dir, grafana_url)
                refreshed = True
        if query_status != 200 or not isinstance(payload, dict) or payload.get("status") != "success":
            raise ProviderError(
                "grafana_query_range_failed",
                f"Grafana query_range failed with HTTP {query_status}.",
                detail={"http_status": query_status, "response": payload},
            )
        metric_group_id = args.metric_group_id or _metric_group_id(args.name)
        row, time_range = _metric_row_from_payload(
            name=args.name,
            metric_group_id=metric_group_id,
            unit=args.unit,
            promql=args.query,
            start=start,
            end=end,
            step=step,
            payload=payload,
            source_kind="grafana_cookie_session_range",
            origin=origin,
            path=proxy_path,
        )
        row["chart_data"] = {
            "name": args.name,
            "promql": args.query,
            "result_type": payload.get("data", {}).get("resultType") or "matrix",
            "series": payload.get("data", {}).get("result") or [],
        }
        persisted = _persist_metric(row, time_range, output_dir=args.output_dir)
        _print_provider_result(
            operation="browser-query-range",
            status="ok",
            kind="metric",
            summary={
                "http_status": query_status,
                "cluster": cluster.get("alias") or args.cluster,
                "region": cluster.get("region"),
                "datasource": {
                    "id": datasource.get("id"),
                    "uid": datasource.get("uid"),
                    "name": datasource.get("name"),
                    "type": datasource.get("type"),
                },
                "datasource_selector": datasource_selector,
                "metric_group_id": metric_group_id,
                "numeric": row["summary"],
                "series_summaries": _compact_metric_series_summaries(
                    row.get("series_summaries")
                ),
            },
            artifacts=persisted,
        )
        return 0
    except Exception as exc:
        _print_provider_error(
            operation="browser-query-range",
            exc=exc,
            output_dir=args.output_dir,
            kind="metric",
        )
        return 2


def _dashboard_rows_from_payload(payload: Any) -> list[Any]:
    if isinstance(payload, list):
        return payload
    if isinstance(payload, dict):
        for key in ("data", "items", "result", "results"):
            value = payload.get(key)
            if isinstance(value, list):
                return value
    return []


def _is_timeout_payload(payload: Any) -> bool:
    text = str(payload).lower()
    return "timeout" in text or "aborterror" in text or "signal is aborted" in text


def _is_common_bad_request(payload: Any) -> bool:
    if not isinstance(payload, dict):
        return False
    values = (payload.get("code"), payload.get("error_code"), payload.get("message"))
    return any("common.bad_request" in str(value).lower() for value in values if value is not None)


def _decode_incident_slow_query_plans(
    page: Any,
    origin: str,
    dashboard_cluster_name: str | None,
    payload: Any,
    incident_start: int,
    incident_end: int,
) -> dict[str, int]:
    requested = 0
    decoded = 0
    failed = 0
    for candidate in _dashboard_rows_for_preview(payload):
        if requested >= SLOW_QUERY_MAX_PLAN_DECODES or not isinstance(candidate, dict):
            continue
        if _decoded_plan_object(candidate.get("binary_plan_json")) is not None:
            continue
        binary_plan = candidate.get("binary_plan")
        if not isinstance(binary_plan, str) or not binary_plan.strip():
            continue
        finish = _slow_query_epoch_seconds(candidate.get("timestamp", candidate.get("time")))
        query_time = _finite_number(candidate.get("query_time"))
        if finish is None or query_time is None or query_time < 0:
            continue
        start = finish - query_time
        if finish < incident_start or start > incident_end:
            continue
        requested += 1
        status, decoded_payload = _dashboard_api_post_json(
            page,
            origin,
            dashboard_cluster_name,
            "/api/visualplan/generate",
            {"binary_plan": binary_plan},
            timeout_ms=30000,
        )
        if status == 200 and _decoded_plan_object(decoded_payload) is not None:
            candidate["binary_plan_json"] = decoded_payload
            decoded += 1
        else:
            failed += 1
    return {
        "requested": requested,
        "decoded": decoded,
        "failed": failed,
    }


def _dashboard_query_payload(
    page: Any,
    origin: str,
    dashboard_cluster_name: str | None,
    args: argparse.Namespace,
    start: str,
    end: str,
    *,
    deadline: float | None = None,
) -> tuple[Any, dict[str, Any]]:
    if args.kind == "topsql":
        instance = str(args.instance or "").strip()
        instance_type = str(args.instance_type or "tidb").strip() or "tidb"
        if not instance:
            raise ProviderError(
                "topsql_instance_required",
                "TopSQL requires one explicit --instance selected from time-aligned metrics.",
                detail={
                    "required_action": (
                        "Query time-aligned component metrics first, identify the top or affected "
                        "TiDB instance, then rerun TopSQL with exactly one --instance."
                    ),
                    "start": start,
                    "end": end,
                    "instance_type": instance_type,
                },
            )
        top = args.limit if args.limit is not None else DEFAULT_TOPSQL_TOP
        summary_params = {
            "start": start,
            "end": end,
            "instance": instance,
            "instance_type": instance_type or "tidb",
            "group_by": args.group_by,
            "top": top,
            "window": args.window,
        }
        timeout_ms = DEFAULT_TOPSQL_COMMAND_TIMEOUT_SECONDS * 1000
        if deadline is not None:
            remaining_ms = int((deadline - time.monotonic()) * 1000)
            if remaining_ms <= 0:
                raise ProviderError(
                    "topsql_command_timeout",
                    "TopSQL command exhausted its time budget before the summary query.",
                    detail={"request": summary_params},
                )
            timeout_ms = min(timeout_ms, max(1, remaining_ms))
        status, summary_payload = _dashboard_api_get(
            page,
            origin,
            dashboard_cluster_name,
            "/api/topsql/summary",
            summary_params,
            timeout_ms=timeout_ms,
        )
        if status != 200:
            if _is_timeout_payload(summary_payload):
                raise ProviderError(
                    "topsql_query_timeout",
                    "TopSQL summary query timed out; narrow the time range before retrying.",
                    detail={
                        "http_status": status,
                        "response": summary_payload,
                        "request": summary_params,
                        "timeout_ms": timeout_ms,
                    },
                )
            raise ProviderError(
                "topsql_query_failed",
                f"TopSQL summary query failed with HTTP {status}.",
                detail={"http_status": status, "response": summary_payload, "request": summary_params},
            )
        payload = {"summary": summary_payload}
        request = {
            "path": "/api/topsql/summary",
            "start": start,
            "end": end,
            "instance": instance,
            "instance_type": instance_type or "tidb",
            "group_by": args.group_by,
            "top": top,
            "window": args.window,
            "field_selection": _topsql_field_selection(payload),
        }
        return payload, request

    if args.kind == "statements":
        fields = ",".join(_field_list(args.fields, STATEMENT_DEFAULT_FIELDS))
        status, payload = _dashboard_api_get(
            page,
            origin,
            dashboard_cluster_name,
            "/api/statements/list",
            {
                "begin_time": start,
                "end_time": end,
                "fields": fields,
                "text": args.text or "",
            },
            timeout_ms=60000,
        )
        if status != 200:
            raise ProviderError(
                "statements_query_failed",
                f"SQL statements query failed with HTTP {status}.",
                detail={"http_status": status, "response": payload},
            )
        return payload, {
            "path": "/api/statements/list",
            "begin_time": start,
            "end_time": end,
            "fields": fields,
            "text": args.text or "",
        }

    if args.kind == "slow-query":
        incident_start = int(start)
        incident_end = int(end)
        query_end = incident_end + SLOW_QUERY_OVERLAP_LOOKAHEAD_SECONDS
        requested_limit = args.limit if args.limit is not None else 100
        candidate_limit = min(200, max(SLOW_QUERY_MINIMUM_CANDIDATE_LIMIT, requested_limit))
        requested_fields = _field_list(args.fields, SLOW_QUERY_DIAGNOSTIC_FIELDS)
        available_status, available_payload = _dashboard_api_get(
            page,
            origin,
            dashboard_cluster_name,
            "/api/slow_query/available_fields",
            {},
            timeout_ms=30000,
        )
        selected_fields, field_selection = _slow_query_field_selection(
            requested_fields,
            available_payload,
            available_status,
        )
        fields = ",".join(selected_fields)
        params = {
            "begin_time": str(incident_start),
            "end_time": str(query_end),
            "fields": fields,
            "limit": candidate_limit,
            "text": args.text or "",
            "digest": args.digest or "",
            "orderBy": args.order_by or "",
            "desc": bool(args.desc),
        }
        status, payload = _dashboard_api_get(
            page,
            origin,
            dashboard_cluster_name,
            "/api/slow_query/list",
            params,
            timeout_ms=60000,
        )
        if status == 400 and _is_common_bad_request(payload):
            minimal_order = (
                *SLOW_QUERY_REQUIRED_FIELDS,
                *SLOW_QUERY_CORE_FIELDS,
                args.order_by or "query_time",
            )
            available = set(selected_fields)
            fallback_fields = [field for field in minimal_order if field in available]
            if not fallback_fields:
                fallback_fields = list(SLOW_QUERY_REQUIRED_FIELDS)
            fallback_params = {
                "begin_time": str(incident_start),
                "end_time": str(query_end),
                "fields": ",".join(fallback_fields),
                "limit": candidate_limit,
                "orderBy": args.order_by or "query_time",
                "desc": bool(args.desc),
            }
            if args.text:
                fallback_params["text"] = args.text
            if args.digest:
                fallback_params["digest"] = args.digest
            status, payload = _dashboard_api_get(
                page,
                origin,
                dashboard_cluster_name,
                "/api/slow_query/list",
                fallback_params,
                timeout_ms=60000,
            )
            params = fallback_params
            field_selection = {
                **field_selection,
                "requested_fields": fallback_fields,
                "repair": "common_bad_request_minimal_projection",
            }
        if status != 200:
            raise ProviderError(
                "slow_query_failed",
                f"Slow query list failed with HTTP {status}.",
                detail={"http_status": status, "response": payload, "request": params},
            )
        plan_decode = _decode_incident_slow_query_plans(
            page,
            origin,
            dashboard_cluster_name,
            payload,
            incident_start,
            incident_end,
        )
        return payload, {
            "path": "/api/slow_query/list",
            **params,
            "incident_time_window": {"start": incident_start, "end": incident_end},
            "query_time_window": {"start": incident_start, "end": query_end},
            "result_limit": requested_limit,
            "field_selection": field_selection,
            "plan_decode": plan_decode,
        }

    raise ValueError(f"unsupported dashboard query kind: {args.kind}")


def cmd_dashboard_query(args: argparse.Namespace) -> int:
    try:
        if args.kind == "topsql" and not str(args.instance or "").strip():
            raise ProviderError(
                "topsql_instance_required",
                "TopSQL requires one explicit --instance selected from time-aligned metrics.",
                detail={
                    "required_action": (
                        "Query time-aligned component metrics first, identify the top or affected "
                        "TiDB instance, then rerun TopSQL with exactly one --instance."
                    ),
                    "requested_cluster": args.cluster,
                    "requested_range": {
                        "start": args.start,
                        "end": args.end,
                        "last": args.last,
                    },
                },
            )
        command_deadline = (
            time.monotonic() + args.command_timeout
            if args.kind == "topsql"
            else None
        )
        sync_playwright, _PlaywrightTimeoutError = _import_playwright()
        cluster, dashboard_urls = _cluster_dashboard_urls(args)
        dashboard_cluster_name = _cluster_dashboard_name(cluster, args.dashboard_cluster_name)
        start, end = _resolve_dashboard_range_args(args)
        origin = ""
        payload: Any = None
        request: dict[str, Any] = {}
        with sync_playwright() as p:
            for index, dashboard_url in enumerate(dashboard_urls):
                session_dir = _dashboard_session_dir_for_cluster(cluster, dashboard_url)
                session_dir.mkdir(parents=True, exist_ok=True)
                os.chmod(session_dir, 0o700)
                context = None
                try:
                    context = p.chromium.launch_persistent_context(
                        str(session_dir),
                        headless=args.headless,
                        viewport={"width": 1440, "height": 1000},
                    )
                    page = context.pages[0] if context.pages else context.new_page()
                    try:
                        origin = _ensure_dashboard_session(
                            page,
                            dashboard_url,
                            dashboard_cluster_name,
                            headless=args.headless,
                            terminal_2fa=args.terminal_2fa,
                            twofa_script=args.twofa_script,
                            login_timeout_seconds=(
                                min(
                                    args.login_timeout,
                                    max(1, math.ceil(command_deadline - time.monotonic())),
                                )
                                if command_deadline is not None
                                else args.login_timeout
                            ),
                            debug_login=args.debug_login,
                        )
                    except ProviderError:
                        raise
                    except RuntimeError:
                        if index + 1 >= len(dashboard_urls):
                            raise
                        continue
                    payload, request = _dashboard_query_payload(
                        page,
                        origin,
                        dashboard_cluster_name,
                        args,
                        start,
                        end,
                        deadline=command_deadline,
                    )
                    break
                finally:
                    if context is not None:
                        context.close()

        selected_preview_fields = None
        preview_payload = payload
        record_count = _dashboard_record_count(payload)
        slow_query_counts: dict[str, int] = {}
        if args.kind == "slow-query":
            selected_preview_fields = _slow_query_preview_fields(request)
            preview_payload, slow_query_counts = _slow_query_execution_rows(
                payload,
                int(start),
                int(end),
                int(request.get("result_limit", args.limit if args.limit is not None else 100)),
            )
            record_count = len(preview_payload)
        row = {
            "id": _metric_group_id(f"dashboard_{args.kind}_{cluster.get('alias') or args.cluster}"),
            "kind": args.kind,
            "source": SOURCE,
            "cluster": cluster.get("alias") or args.cluster,
            "region": cluster.get("region"),
            "dashboard_origin": origin,
            "dashboard_route_mode": "direct" if dashboard_cluster_name is None else "proxy",
            "dashboard_cluster_name": dashboard_cluster_name,
            "request": request,
            "time_window": {"start": start, "end": end},
            "record_count": record_count,
            **slow_query_counts,
            "preview_rows": _dashboard_preview_rows(preview_payload, args.preview_limit, selected_preview_fields),
            "payload": payload,
        }
        persisted = _persist_dashboard_evidence(row, output_dir=args.output_dir)
        _print_provider_result(
            operation="dashboard-query",
            status="ok",
            kind=args.kind,
            summary={
                "cluster": row["cluster"],
                "region": row["region"],
                "dashboard_origin": row["dashboard_origin"],
                "dashboard_route_mode": row["dashboard_route_mode"],
                "dashboard_cluster_name": dashboard_cluster_name,
                "request": request,
                "time_window": row["time_window"],
                "record_count": row["record_count"],
                **slow_query_counts,
                "top_rows": row["preview_rows"],
            },
            artifacts=persisted,
        )
        return 0
    except Exception as exc:
        _print_provider_error(
            operation="dashboard-query",
            exc=exc,
            output_dir=args.output_dir,
            kind=args.kind,
        )
        return 2


def cmd_list_clusters(_args: argparse.Namespace) -> int:
    _print_json({"status": "ok", "clusters": _load_clusters()})
    return 0


def _target_clusters(args: argparse.Namespace) -> list[dict[str, Any]]:
    clusters = _load_clusters()
    if args.cluster:
        return [_find_cluster(args.cluster, args.region)]
    if args.region:
        region = _normalize_alias(args.region)
        matched = [row for row in clusters if _normalize_alias(_cluster_region(row)) == region]
        if not matched:
            raise ValueError(f"unknown region: {args.region}")
        return matched
    return clusters


def _datasource_summary(datasource: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": datasource.get("id"),
        "uid": datasource.get("uid"),
        "name": datasource.get("name"),
        "type": datasource.get("type"),
        "isDefault": datasource.get("isDefault"),
    }


def _map_datasources_for_clusters(
    clusters: list[dict[str, Any]],
    datasources: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    prometheus = [row for row in datasources if str(row.get("type") or "").lower() == "prometheus"]
    mappings: list[dict[str, Any]] = []
    for cluster in clusters:
        scored = sorted(
            (
                (_datasource_match_score(cluster, datasource), datasource)
                for datasource in prometheus
            ),
            key=lambda item: item[0],
            reverse=True,
        )
        best_score = scored[0][0] if scored else 0
        best = [datasource for score, datasource in scored if score == best_score and score > 0]
        status = "unmatched"
        datasource: dict[str, Any] | None = None
        if len(best) == 1:
            status = "matched"
            datasource = best[0]
        elif len(best) > 1:
            status = "ambiguous"
        mappings.append(
            {
                "cluster": cluster.get("alias"),
                "region": cluster.get("region"),
                "status": status,
                "score": best_score,
                "datasource": _datasource_summary(datasource) if datasource else None,
                "candidates": [_datasource_summary(row) for row in best[:10]],
            }
        )
    return mappings


def _write_cluster_datasource_mapping(mappings: list[dict[str, Any]]) -> int:
    clusters = _load_clusters()
    by_key = {
        (
            _normalize_alias(str(row.get("region") or "")),
            _normalize_alias(str(row.get("cluster") or "")),
        ): row
        for row in mappings
    }
    changed = 0
    for cluster in clusters:
        mapping = by_key.get((_normalize_alias(_cluster_region(cluster)), _cluster_key(cluster)))
        datasource = mapping.get("datasource") if mapping else None
        if mapping and mapping.get("status") == "matched" and isinstance(datasource, dict):
            uid = str(datasource.get("uid") or "")
            name = str(datasource.get("name") or "")
            if uid and cluster.get("datasource_uid") != uid:
                cluster["datasource_uid"] = uid
                changed += 1
            if name and cluster.get("datasource_name") != name:
                cluster["datasource_name"] = name
                changed += 1
    if changed:
        _atomic_write_json(CLUSTERS_FILE, clusters)
    return changed


def cmd_rebuild_datasource_map(args: argparse.Namespace) -> int:
    try:
        sync_playwright, _PlaywrightTimeoutError = _import_playwright()
        clusters = _target_clusters(args)
        if not clusters:
            raise ValueError("no clusters selected")

        groups: dict[str, dict[str, Any]] = {}
        for cluster in clusters:
            urls = cluster.get("grafana") or []
            if not urls:
                continue
            grafana_url = str(args.grafana_url or urls[0])
            groups.setdefault(grafana_url, {"grafana_url": grafana_url, "clusters": []})["clusters"].append(cluster)

        results: list[dict[str, Any]] = []
        with sync_playwright() as p:
            for group in groups.values():
                grafana_url = group["grafana_url"]
                group_clusters = group["clusters"]
                session_dir = _session_dir_for_cluster(group_clusters[0], grafana_url)
                session_dir.mkdir(parents=True, exist_ok=True)
                os.chmod(session_dir, 0o700)
                context = None
                try:
                    context = p.chromium.launch_persistent_context(
                        str(session_dir),
                        headless=args.headless,
                        viewport={"width": 1440, "height": 1000},
                    )
                    page = context.pages[0] if context.pages else context.new_page()
                    origin, datasources = _ensure_datasources(
                        page,
                        grafana_url,
                        headless=args.headless,
                        terminal_2fa=args.terminal_2fa,
                        twofa_script=args.twofa_script,
                        login_timeout_seconds=args.login_timeout,
                        debug_login=args.debug_login,
                    )
                    mappings = _map_datasources_for_clusters(group_clusters, datasources)
                    results.append(
                        {
                            "grafana_url": grafana_url,
                            "origin": origin,
                            "clusters": [cluster.get("alias") for cluster in group_clusters],
                            "datasource_count": len(datasources),
                            "prometheus_datasource_count": len(
                                [row for row in datasources if str(row.get("type") or "").lower() == "prometheus"]
                            ),
                            "mappings": mappings,
                        }
                    )
                finally:
                    if context is not None:
                        context.close()

        all_mappings = [mapping for result in results for mapping in result["mappings"]]
        changed = _write_cluster_datasource_mapping(all_mappings) if args.write else 0
        _print_json({"status": "ok", "write": bool(args.write), "changed": changed, "results": results})
        return 0
    except Exception as exc:
        _print_json({"status": "error", "error": _sanitize_error_text(exc)})
        return 2


def cmd_datasources(args: argparse.Namespace) -> int:
    try:
        sync_playwright, _PlaywrightTimeoutError = _import_playwright()
        cluster, grafana_url = _cluster_grafana_url(args)
        session_dir = _session_dir_for_cluster(cluster, grafana_url)
        session_dir.mkdir(parents=True, exist_ok=True)
        os.chmod(session_dir, 0o700)
        with sync_playwright() as p:
            context = p.chromium.launch_persistent_context(
                str(session_dir),
                headless=args.headless,
                viewport={"width": 1440, "height": 1000},
            )
            try:
                page = context.pages[0] if context.pages else context.new_page()
                origin, datasources = _ensure_datasources(
                    page,
                    grafana_url,
                    headless=args.headless,
                    terminal_2fa=args.terminal_2fa,
                    twofa_script=args.twofa_script,
                    login_timeout_seconds=args.login_timeout,
                    debug_login=args.debug_login,
                )
            finally:
                context.close()
        rows = [
            _datasource_summary(row)
            for row in datasources
            if not args.prometheus_only or str(row.get("type") or "").lower() == "prometheus"
        ]
        _print_json(
            {
                "status": "ok",
                "cluster": cluster.get("alias"),
                "region": cluster.get("region"),
                "origin": origin,
                "datasources": rows,
            }
        )
        return 0
    except Exception as exc:
        _print_json({"status": "error", "error": _sanitize_error_text(exc)})
        return 2


def cmd_login_guide(args: argparse.Namespace) -> int:
    cluster = _find_cluster(args.cluster, args.region)
    _print_json(
        {
            "status": "ok",
            "cluster": cluster,
            "steps": [
                "Prefer browser-query-range for metric collection.",
                "The command opens a skill-managed Playwright browser profile and reads credentials only from runtime environment variables.",
                "If P81 asks for 2FA, scripts/get_verify_code.py reads WETECH_P81_TOTP_SECRET from the runtime environment and fills the current code in memory.",
                "After the first visible run succeeds, reuse browser-query-range --headless for the same Grafana origin until the session expires.",
                "Fallback only when Playwright is unavailable: copy one /api/datasources/proxy/.../api/v1/query_range request as cURL and run probe/query-range.",
            ],
            "notes": [
                "No username, password, TOTP seed, cookie, or token is stored in this skill package or returned by this command.",
                "Browser automation uses a skill-managed Playwright profile under ~/.cache/wetech-op-grafana/browser/ and does not read the user's Chrome profile.",
                "browser-query-range resolves the Prometheus datasource from datasource_uid, datasource_name_regex, /api/datasources, or dashboard panel JSON.",
                "Session cache is keyed by region and Grafana host so clusters sharing a Grafana origin reuse login state.",
                "A copied cURL remains a last-resort fallback for the same Grafana origin until the browser session expires.",
                "HA hosts are separate origins; probe them separately.",
            ],
        }
    )
    return 0


def cmd_twofa_guide(_args: argparse.Namespace) -> int:
    _print_json(
        {
            "status": "ok",
            "authenticator": "Any standard TOTP authenticator app",
            "not_supported": [
                "writing 6-digit one-time codes to stdout, evidence, or diagnostics from the browser-query flow",
            ],
            "steps": [
                "Provision WETECH_P81_TOTP_SECRET through the Kubernetes Secret used by the runtime.",
                "Use scripts/get_verify_code.py only inside the runtime when a current code is required.",
                "After P81 login succeeds, use browser-query-range or datasources to query through the skill-managed browser session.",
            ],
            "credentialSource": "runtime_environment_only",
        }
    )
    return 0


def _read_otpauth_uri(args: argparse.Namespace) -> str:
    if bool(args.otpauth_uri) == bool(args.otpauth_file):
        raise ValueError("provide exactly one of --otpauth-uri or --otpauth-file")
    if args.otpauth_uri:
        uri = str(args.otpauth_uri).strip()
    else:
        uri = Path(args.otpauth_file).expanduser().read_text(encoding="utf-8").strip()
    if not uri.startswith(("otpauth://", "otpauth-migration://")):
        raise ValueError("2FA URI must start with otpauth:// or otpauth-migration://")
    return uri


def cmd_twofa_qr(args: argparse.Namespace) -> int:
    try:
        uri = _read_otpauth_uri(args)
        import qrcode
        import qrcode.image.svg
    except Exception as exc:
        _print_json({"status": "error", "error": _sanitize_error_text(exc)})
        return 2

    result: dict[str, Any] = {
        "status": "ok",
        "format": "svg",
        "note": "Scan the generated QR with Google Authenticator. This command renders the URI as QR only; it does not generate one-time codes.",
    }

    if args.print_terminal:
        qr = qrcode.QRCode(border=2)
        qr.add_data(uri)
        qr.make(fit=True)
        with contextlib.redirect_stdout(sys.stderr):
            qr.print_ascii(invert=True)
        result["printed_terminal_qr"] = True

    if args.output:
        output = Path(args.output).expanduser().resolve()
        output.parent.mkdir(parents=True, exist_ok=True)
        image = qrcode.make(uri, image_factory=qrcode.image.svg.SvgPathImage)
        image.save(str(output))
        result["output"] = str(output)
    else:
        result["output"] = None

    _print_json(result)
    return 0


def _probe_payload(curl_file: Path, overrides: argparse.Namespace | None = None) -> tuple[dict[str, Any], dict[str, Any]]:
    parsed = _parse_curl(curl_file)
    url = _with_query_overrides(parsed["url"], overrides) if overrides else parsed["url"]
    response = _http_get(url, parsed["headers"])
    request_summary = {
        "url": _redact_url(url),
        "headers": _redact_headers(parsed["headers"]),
    }
    return request_summary, response


def cmd_probe(args: argparse.Namespace) -> int:
    try:
        request_summary, response = _probe_payload(args.curl_file)
    except Exception as exc:
        _print_provider_error(
            operation="probe",
            exc=exc,
            output_dir=args.output_dir,
            kind="metric",
        )
        return 2

    payload = response.get("payload") if isinstance(response.get("payload"), dict) else {}
    ok = response.get("http_status") == 200 and payload.get("status") == "success"
    if not ok:
        exc = ProviderError(
            "grafana_probe_failed",
            f"Grafana probe failed with HTTP {response.get('http_status')}.",
            detail={"response": response, "request": request_summary},
        )
        _print_provider_error(
            operation="probe",
            exc=exc,
            output_dir=args.output_dir,
            kind="metric",
        )
        return 1
    _print_provider_result(
        operation="probe",
        status="ok",
        kind="metric",
        summary={
            "http_status": response.get("http_status"),
            "grafana_status": payload.get("status"),
            "result_type": payload.get("data", {}).get("resultType") if isinstance(payload, dict) else None,
            "series_count": _series_count(payload),
            "point_count": _summarize_numeric(payload).get("point_count", 0),
            "request": request_summary,
        },
    )
    return 0


def cmd_query_range(args: argparse.Namespace) -> int:
    try:
        parsed = _parse_curl(args.curl_file)
        url = _with_query_overrides(parsed["url"], args)
        parsed_endpoint = _grafana_endpoint(urllib.parse.urlparse(url).path)
        if parsed_endpoint != "query_range":
            raise ValueError("query-range requires a /api/v1/query_range URL; use a copied range-query request")
        response = _http_get(url, parsed["headers"])
    except Exception as exc:
        _print_provider_error(
            operation="query-range",
            exc=exc,
            output_dir=args.output_dir,
            kind="metric",
        )
        return 2

    payload = response.get("payload") if isinstance(response.get("payload"), dict) else {}
    if response.get("http_status") != 200 or payload.get("status") != "success":
        exc = ProviderError(
            "grafana_query_range_failed",
            f"Grafana query_range failed with HTTP {response.get('http_status')}.",
            detail={
                "response": response,
                "request": {"url": _redact_url(url), "headers": _redact_headers(parsed["headers"])},
            },
        )
        _print_provider_error(
            operation="query-range",
            exc=exc,
            output_dir=args.output_dir,
            kind="metric",
        )
        return 1

    parsed_url = urllib.parse.urlparse(url)
    params = dict(urllib.parse.parse_qsl(parsed_url.query, keep_blank_values=True))
    metric_group_id = args.metric_group_id or _metric_group_id(args.name)
    time_range = _time_window(params.get("start"), params.get("end"), params.get("step"))
    row = {
        **_metric_row_from_payload(
            name=args.name,
            metric_group_id=metric_group_id,
            unit=args.unit,
            promql=params.get("query", ""),
            start=str(params.get("start") or ""),
            end=str(params.get("end") or ""),
            step=str(params.get("step") or ""),
            payload=payload,
            source_kind="grafana_browser_proxy_range",
            origin=f"{parsed_url.scheme}://{parsed_url.netloc}",
            path=parsed_url.path,
        )[0]
    }
    row["chart_data"] = {
        "name": args.name,
        "promql": params.get("query", ""),
        "result_type": payload.get("data", {}).get("resultType") or "matrix",
        "series": payload.get("data", {}).get("result") or [],
    }

    persisted = _persist_metric(row, time_range, output_dir=args.output_dir)
    _print_provider_result(
        operation="query-range",
        status="ok",
        kind="metric",
        summary={
            "http_status": response.get("http_status"),
            "metric_group_id": metric_group_id,
            "numeric": row["summary"],
            "series_summaries": _compact_metric_series_summaries(
                row.get("series_summaries")
            ),
        },
        artifacts=persisted,
    )
    return 0


def _load_bounded_json_object(path: Path, maximum_bytes: int = 262_144) -> dict[str, Any]:
    if not path.is_file() or path.stat().st_size > maximum_bytes:
        raise ProviderError("op_batch_input_invalid", f"Invalid or oversized JSON input: {path.name}")
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ProviderError("op_batch_input_invalid", f"Invalid JSON input: {path.name}") from exc
    if not isinstance(value, dict):
        raise ProviderError("op_batch_input_invalid", f"JSON input must be an object: {path.name}")
    return value


def _atomic_write_json(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(value, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
    temporary.replace(path)


def _historical_bounds(time_window: Any) -> tuple[int, int] | None:
    if not isinstance(time_window, str):
        return None
    match = re.fullmatch(r"(?:historical|live):absolute;start=(\d+);end=(\d+);.*", time_window)
    if not match:
        return None
    start, end = int(match.group(1)), int(match.group(2))
    if start >= end:
        raise ProviderError("op_batch_window_invalid", "Historical OP batch window is invalid.")
    return start, end


def _comparison_bounds(context: dict[str, Any]) -> tuple[tuple[int, int], ...]:
    value = context.get("comparisonWindows")
    if value is None:
        return ()
    incident = _historical_bounds(context.get("timeWindow"))
    if incident is None or not isinstance(value, list) or len(value) != 7:
        raise ProviderError("op_batch_comparison_windows_invalid", "Historical comparison windows are invalid.")
    windows: list[tuple[int, int]] = []
    for index, item in enumerate(value, start=1):
        if not isinstance(item, dict):
            raise ProviderError("op_batch_comparison_windows_invalid", "Historical comparison windows are invalid.")
        start, end = item.get("start"), item.get("end")
        if (item.get("windowId") != f"day-{index}" or not isinstance(start, int)
                or isinstance(start, bool) or not isinstance(end, int) or isinstance(end, bool)
                or start != incident[0] - index * 86400 or end != incident[1] - index * 86400):
            raise ProviderError("op_batch_comparison_windows_invalid", "Historical comparison windows are invalid.")
        windows.append((start, end))
    return tuple(windows)


def _bounded_batch_range(
    params: dict[str, Any],
    bounds: tuple[int, int] | None,
    comparison_bounds: tuple[tuple[int, int], ...] = (),
) -> tuple[int, int]:
    start = params.get("start", bounds[0] if bounds else None)
    end = params.get("end", bounds[1] if bounds else None)
    if not isinstance(start, int) or isinstance(start, bool) or not isinstance(end, int) or isinstance(end, bool):
        raise ProviderError("op_batch_window_missing", "Every OP batch query requires integer start and end.")
    if start >= end:
        raise ProviderError("op_batch_window_invalid", "OP batch query start must be before end.")
    if comparison_bounds:
        if (start, end) not in comparison_bounds:
            raise ProviderError(
                "op_batch_comparison_window_invalid",
                "Historical comparison queries must use one exact runner-owned daily window.",
            )
        return start, end
    if bounds and (start < bounds[0] or end > bounds[1]):
        raise ProviderError(
            "op_batch_window_out_of_bounds",
            "OP batch query must stay inside the runner-owned historical window.",
        )
    return start, end


def _assert_comparison_plan(
    context: dict[str, Any],
    queries: list[dict[str, Any]],
    bounds: tuple[int, int] | None,
    comparison_bounds: tuple[tuple[int, int], ...],
) -> None:
    if not comparison_bounds:
        return
    if context.get("scenarioId") != "workload-hotspot" or len(queries) != 7:
        raise ProviderError(
            "op_batch_comparison_plan_invalid",
            "Historical comparison requires seven workload-hotspot metric queries.",
        )
    used: set[tuple[int, int]] = set()
    for query in queries:
        if query.get("operation") != "browser-query-range" or not isinstance(query.get("params"), dict):
            raise ProviderError(
                "op_batch_comparison_plan_invalid",
                "Historical comparison supports only separate metric range queries.",
            )
        used.add(_bounded_batch_range(query["params"], bounds, comparison_bounds))
    if len(used) != len(comparison_bounds):
        raise ProviderError(
            "op_batch_comparison_plan_invalid",
            "Historical comparison must cover each runner-owned daily window exactly once.",
        )


def _batch_string(params: dict[str, Any], key: str, *, required: bool = False) -> str | None:
    value = params.get(key)
    if value is None and not required:
        return None
    if not isinstance(value, str) or not value.strip():
        raise ProviderError("op_batch_parameter_invalid", f"OP batch parameter {key} must be a non-empty string.")
    return value.strip()


def _batch_positive_int(params: dict[str, Any], key: str) -> int | None:
    value = params.get(key)
    if value is None:
        return None
    if not isinstance(value, int) or isinstance(value, bool) or value < 1:
        raise ProviderError("op_batch_parameter_invalid", f"OP batch parameter {key} must be a positive integer.")
    return value


def _normalize_wetech_batch_target(
    plan: dict[str, Any], queries: list[dict[str, Any]], context: dict[str, Any] | None = None
) -> dict[str, str | None]:
    target = plan.get("target")
    target = target if isinstance(target, dict) else {}
    cluster = target.get("cluster")
    region = target.get("region")
    context_auth = context.get("auth") if isinstance(context, dict) else None
    context_auth = context_auth if isinstance(context_auth, dict) else {}
    target_references = (
        context_auth.get("targetReference"),
        plan.get("targetReference"),
        target.get("targetReference"),
    )
    for target_reference in target_references:
        if not isinstance(target_reference, str):
            continue
        parts = target_reference.split(":", 3)
        if len(parts) >= 3 and parts[0] == "wetech":
            region = region or parts[1]
            cluster = cluster or parts[2]
    for query in queries:
        params = query.get("params") if isinstance(query, dict) else None
        if not isinstance(params, dict):
            continue
        cluster = cluster or params.get("cluster")
        region = region or params.get("region")
        if cluster:
            break
    if not isinstance(cluster, str) or not cluster.strip():
        raise ProviderError("op_batch_target_invalid", "Wetech OP batch requires an exact cluster target.")
    return {
        "cluster": cluster.strip(),
        "region": region.strip() if isinstance(region, str) and region.strip() else None,
    }


def _normalize_wetech_batch_query(
    query: dict[str, Any], scenario_id: str | None = None
) -> dict[str, Any]:
    params = query.get("params")
    normalized_params = dict(params) if isinstance(params, dict) else {}
    aliases = {
        "metric_group_id": "metricGroupId",
        "datasource_name_regex": "datasourceNameRegex",
        "preview_limit": "previewLimit",
        "instance_from_query_id": "instanceFromQueryId",
        "window_from_query_id": "windowFromQueryId",
        "hotspot_window_seconds": "hotspotWindowSeconds",
        "instance_type": "instanceType",
        "group_by": "groupBy",
        "command_timeout": "commandTimeout",
        "order_by": "orderBy",
    }
    metadata = {
        "queryId", "description", "statement", "operation", "params", "subdeadlineMs",
        "subDeadlineMs", "deadlineMs", "dependsOn", "handoffFields",
        "emptyResultExpectation",
    }
    for key, value in query.items():
        if key in metadata:
            continue
        normalized_params.setdefault(aliases.get(key, key), value)
    operation = query.get("operation")
    empty_expectation = query.get("emptyResultExpectation")
    if operation == "metric":
        if empty_expectation is None:
            empty_expectation = "unknown"
            query["emptyResultExpectation"] = empty_expectation
        if empty_expectation not in {"expected_non_empty", "sparse_allowed", "unknown"}:
            raise ProviderError(
                "op_batch_parameter_invalid",
                "Metric emptyResultExpectation must be expected_non_empty, sparse_allowed, or unknown.",
            )
    elif empty_expectation is not None:
        raise ProviderError(
            "op_batch_parameter_invalid",
            "emptyResultExpectation is valid only for metric operations.",
        )
    operation_aliases = {
        "metric": "browser-query-range",
        "slow-query": "dashboard-query",
        "statements": "dashboard-query",
        "topsql": "dashboard-query",
    }
    normalized_operation = operation_aliases.get(operation, operation)
    if operation in {"slow-query", "statements", "topsql"}:
        normalized_params.setdefault("kind", operation)
    fields = normalized_params.get("fields")
    if isinstance(fields, list) and fields and all(
        isinstance(field, str) and field.strip() for field in fields
    ):
        normalized_params["fields"] = ",".join(
            dict.fromkeys(field.strip() for field in fields)
        )
    if operation == "slow-query":
        default_order = "memory_max" if scenario_id == "memory-oom" else "query_time"
        normalized_params.setdefault("orderBy", default_order)
        normalized_params.setdefault("desc", True)
        selected = _field_list(normalized_params.get("fields"), ())
        handoff_fields, unsupported_handoff_fields = _slow_query_handoff_fields(
            query.get("handoffFields")
        )
        query["handoffFields"] = handoff_fields
        query["unsupportedHandoffFields"] = unsupported_handoff_fields
        selected_handoff_fields = [
            field for field in handoff_fields
            if field not in SLOW_QUERY_EXECUTION_INTERVAL_FIELDS
        ]
        order_field = normalized_params["orderBy"]
        if not isinstance(order_field, str) or not order_field.strip():
            raise ProviderError(
                "op_batch_parameter_invalid", "Slow Query orderBy must be a non-empty field."
            )
        normalized_params["fields"] = ",".join(dict.fromkeys((
            *selected_handoff_fields,
            *SLOW_QUERY_REQUIRED_FIELDS,
            *SLOW_QUERY_CORE_FIELDS,
            *selected,
            order_field.strip(),
        )))
    return {**query, "operation": normalized_operation, "params": normalized_params}


def _slow_query_handoff_fields(value: Any) -> tuple[list[str], list[str]]:
    if value is None:
        return [], []
    if (
        not isinstance(value, list)
        or len(value) > SLOW_QUERY_MAX_HANDOFF_FIELDS
        or any(not isinstance(field, str) or not field.strip() for field in value)
    ):
        raise ProviderError(
            "op_batch_parameter_invalid",
            f"Slow Query handoffFields must contain at most {SLOW_QUERY_MAX_HANDOFF_FIELDS} field names.",
        )
    fields = list(dict.fromkeys(field.strip() for field in value))
    return (
        [field for field in fields if field in SLOW_QUERY_HANDOFF_FIELDS],
        [field for field in fields if field not in SLOW_QUERY_HANDOFF_FIELDS],
    )


def _valid_batch_query_id(value: Any) -> bool:
    return (
        isinstance(value, str)
        and len(value) <= 64
        and OP_BATCH_QUERY_ID_PATTERN.fullmatch(value) is not None
    )


def _batch_instance_from_result(query_id: str, results: dict[str, dict[str, Any]]) -> str:
    payload = results.get(query_id)
    series = payload.get("summary", {}).get("series_summaries", []) if isinstance(payload, dict) else []
    if not isinstance(series, list):
        series = []
    for row in series:
        metric = row.get("metric", {}) if isinstance(row, dict) else {}
        instance = metric.get("instance") if isinstance(metric, dict) else None
        if isinstance(instance, str) and instance.strip():
            return instance.strip()
    raise ProviderError(
        "op_batch_instance_dependency_unresolved",
        f"The prior {query_id} query did not return an exact instance for the dependent TopSQL query.",
    )


def _batch_hotspot_window_from_result(
    query_id: str,
    results: dict[str, dict[str, Any]],
    bounds: tuple[int, int] | None,
    window_seconds: int,
) -> tuple[int, int]:
    if bounds is None:
        raise ProviderError(
            "op_batch_hotspot_window_unresolved",
            "A metric-derived TopSQL window requires runner-owned absolute time bounds.",
        )
    payload = results.get(query_id)
    series = payload.get("summary", {}).get("series_summaries", []) if isinstance(payload, dict) else []
    if not isinstance(series, list):
        series = []
    ranked = sorted(
        (row for row in series if isinstance(row, dict)),
        key=lambda row: (
            row.get("summary", {}).get("max")
            if isinstance(row.get("summary"), dict)
            and isinstance(row.get("summary", {}).get("max"), (int, float))
            else float("-inf")
        ),
        reverse=True,
    )
    hotspot_at: int | None = None
    for row in ranked:
        summary = row.get("summary")
        candidate = summary.get("max_at") if isinstance(summary, dict) else None
        if isinstance(candidate, (int, float)) and not isinstance(candidate, bool):
            hotspot_at = int(candidate)
            break
    if hotspot_at is None or hotspot_at < bounds[0] or hotspot_at > bounds[1]:
        raise ProviderError(
            "op_batch_hotspot_window_unresolved",
            f"The prior {query_id} query did not return a bounded hotspot timestamp for TopSQL.",
        )
    available = bounds[1] - bounds[0]
    width = min(window_seconds, available)
    start = hotspot_at - width // 2
    start = max(bounds[0], min(start, bounds[1] - width))
    return start, start + width


def _wetech_batch_argv(
    target: dict[str, Any],
    query: dict[str, Any],
    bounds: tuple[int, int] | None,
    output_dir: Path,
    prior_results: dict[str, dict[str, Any]],
    comparison_bounds: tuple[tuple[int, int], ...] = (),
) -> list[str]:
    operation = query.get("operation")
    params = query.get("params")
    if operation not in {"browser-query-range", "dashboard-query"} or not isinstance(params, dict):
        raise ProviderError("op_batch_operation_invalid", "Unsupported Wetech OP batch operation.")
    kind = _batch_string(params, "kind") if operation == "dashboard-query" else None
    window_dependency = _batch_string(params, "windowFromQueryId")
    hotspot_window_seconds = _batch_positive_int(params, "hotspotWindowSeconds") or MAX_BATCH_TOPSQL_RANGE_SECONDS
    if window_dependency:
        if kind != "topsql":
            raise ProviderError("op_batch_parameter_invalid", "windowFromQueryId is supported only for TopSQL.")
        if "start" in params or "end" in params:
            raise ProviderError(
                "op_batch_parameter_invalid",
                "Use explicit start/end or windowFromQueryId for TopSQL, not both.",
            )
        if window_dependency not in prior_results or hotspot_window_seconds > MAX_BATCH_TOPSQL_RANGE_SECONDS:
            raise ProviderError("op_batch_parameter_invalid", "TopSQL hotspot window dependency is invalid.")
        start, end = _batch_hotspot_window_from_result(
            window_dependency, prior_results, bounds, hotspot_window_seconds
        )
    else:
        start, end = _bounded_batch_range(params, bounds, comparison_bounds)
    argv = [
        operation, "--cluster", target["cluster"], "--start", str(start), "--end", str(end),
        "--headless", "--output-dir", str(output_dir),
    ]
    region = target.get("region")
    if region:
        argv.extend(["--region", region])
    if operation == "browser-query-range":
        allowed = {
            "cluster", "region", "headless", "name", "query", "step",
            "metricGroupId", "unit", "datasourceNameRegex", "start", "end",
        }
        argv.extend(["--name", _batch_string(params, "name", required=True)])
        argv.extend(["--query", _batch_string(params, "query", required=True)])
        step = params.get("step")
        if step is not None:
            if not isinstance(step, (str, int, float)) or isinstance(step, bool):
                raise ProviderError("op_batch_parameter_invalid", "OP batch parameter step must be a duration or number.")
            argv.extend(["--step", str(step)])
        for key, flag in (
            ("metricGroupId", "--metric-group-id"),
            ("unit", "--unit"), ("datasourceNameRegex", "--datasource-name-regex"),
        ):
            value = _batch_string(params, key)
            if value is not None:
                argv.extend([flag, value])
    else:
        allowed = {
            "kind", "limit", "previewLimit", "fields", "text", "digest", "orderBy",
            "desc", "instance", "instanceFromQueryId", "instanceType", "groupBy", "window",
            "windowFromQueryId", "hotspotWindowSeconds", "commandTimeout", "cluster", "region",
            "headless", "start", "end",
        }
        kind = _batch_string(params, "kind", required=True)
        argv.extend(["--kind", kind])
        if kind == "topsql" and end - start > MAX_BATCH_TOPSQL_RANGE_SECONDS:
            raise ProviderError(
                "op_batch_topsql_window_too_wide",
                (
                    "Wetech batch TopSQL requires an incident-aligned absolute window no longer "
                    f"than {MAX_BATCH_TOPSQL_RANGE_SECONDS} seconds; omit TopSQL when no exact "
                    "hotspot subwindow is known."
                ),
                detail={
                    "start": start,
                    "end": end,
                    "range_seconds": end - start,
                    "maximum_range_seconds": MAX_BATCH_TOPSQL_RANGE_SECONDS,
                },
            )
        instance = _batch_string(params, "instance")
        dependency = _batch_string(params, "instanceFromQueryId")
        if instance and dependency:
            raise ProviderError("op_batch_parameter_invalid", "Use instance or instanceFromQueryId, not both.")
        if dependency:
            if dependency not in prior_results:
                raise ProviderError("op_batch_parameter_invalid", "instanceFromQueryId is invalid.")
            instance = _batch_instance_from_result(dependency, prior_results)
        if instance:
            argv.extend(["--instance", instance])
        for key, flag in (
            ("fields", "--fields"), ("text", "--text"), ("digest", "--digest"),
            ("orderBy", "--order-by"), ("instanceType", "--instance-type"),
            ("groupBy", "--group-by"), ("window", "--window"),
        ):
            value = _batch_string(params, key)
            if value is not None:
                argv.extend([flag, value])
        for key, flag in (
            ("limit", "--limit"), ("previewLimit", "--preview-limit"),
            ("commandTimeout", "--command-timeout"),
        ):
            value = _batch_positive_int(params, key)
            if value is not None:
                argv.extend([flag, str(value)])
        if params.get("desc") is True:
            argv.append("--desc")
        elif params.get("desc") not in {None, False}:
            raise ProviderError("op_batch_parameter_invalid", "desc must be a boolean.")
    # The plan is agent-authored. Only map this operation's explicit allowlist
    # into CLI arguments; unrelated descriptive or future metadata is inert.
    return [str(value) for value in argv]


def _run_wetech_subcommand(argv: list[str]) -> tuple[int, dict[str, Any]]:
    env = os.environ.copy()
    slot = getattr(_OPERATION_THREAD_STATE, "slot", None)
    if argv and argv[0] == "dashboard-query" and slot is not None:
        env["WETECH_DASHBOARD_CONCURRENCY_SLOT"] = str(slot)
    process = subprocess.Popen(
        [sys.executable, str(Path(__file__).resolve()), *argv],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        env=env,
        start_new_session=True,
    )
    with _ACTIVE_SUBCOMMANDS_LOCK:
        _ACTIVE_SUBCOMMANDS.add(process)
    try:
        stdout, stderr = process.communicate()
    finally:
        with _ACTIVE_SUBCOMMANDS_LOCK:
            _ACTIVE_SUBCOMMANDS.discard(process)
    candidate = stdout.strip()
    if not candidate:
        candidate = stderr.strip().splitlines()[-1]
    try:
        payload = json.loads(candidate)
    except (json.JSONDecodeError, IndexError) as exc:
        raise ProviderError("op_batch_result_invalid", "Wetech provider returned invalid compact JSON.") from exc
    if not isinstance(payload, dict):
        raise ProviderError("op_batch_result_invalid", "Wetech provider returned a non-object compact result.")
    return process.returncode, payload


def _install_batch_signal_forwarding() -> None:
    def terminate(signum: int, _frame: Any) -> None:
        with _ACTIVE_SUBCOMMANDS_LOCK:
            active = list(_ACTIVE_SUBCOMMANDS)
        for process in active:
            try:
                os.killpg(process.pid, signal.SIGTERM)
            except ProcessLookupError:
                pass
        raise SystemExit(128 + signum)

    signal.signal(signal.SIGTERM, terminate)
    signal.signal(signal.SIGINT, terminate)


def _wetech_result_empty(operation: str, payload: dict[str, Any]) -> bool:
    summary = payload.get("summary", {})
    if not isinstance(summary, dict):
        return True
    if operation == "browser-query-range":
        numeric = summary.get("numeric", {})
        series = summary.get("series_summaries", [])
        return not series or not isinstance(numeric, dict) or int(numeric.get("point_count", 0)) == 0
    return int(summary.get("record_count", 0)) == 0


def _wetech_dashboard_shared_failure_code(
    payload: dict[str, Any], output_dir: Path | None = None
) -> str | None:
    error = payload.get("error")
    if not isinstance(error, dict):
        return None
    code = str(error.get("code") or "")
    if code == "wetech_dashboard_authentication_failed":
        return code
    detail = error.get("detail")
    detail_message = detail.get("message") if isinstance(detail, dict) else None
    if detail_message is None and output_dir is not None:
        artifacts = payload.get("artifacts")
        error_path = artifacts.get("error_path") if isinstance(artifacts, dict) else None
        if isinstance(error_path, str):
            try:
                resolved_output = output_dir.resolve()
                resolved_error = Path(error_path).resolve()
                resolved_error.relative_to(resolved_output)
                artifact = _load_bounded_json_object(resolved_error)
                artifact_detail = artifact.get("detail")
                detail_message = (
                    artifact_detail.get("message") if isinstance(artifact_detail, dict) else None
                )
            except (OSError, ValueError, ProviderError):
                detail_message = None
    text = " ".join(
        value for value in (error.get("message"), detail_message) if isinstance(value, str)
    ).lower()
    unavailable = any(
        marker in text
        for marker in (
            "dashboard session is not ready",
            "dashboard proxy cluster route was not found",
        )
    )
    return "wetech_dashboard_session_unavailable" if unavailable else None


def cmd_batch(args: argparse.Namespace) -> int:
    try:
        context = _load_bounded_json_object(args.context)
        plan = _load_bounded_json_object(args.plan)
        if context.get("schemaVersion") != "auto-oncall-op-query-context-v1" or context.get("provider") != "wetech":
            raise ProviderError("op_batch_context_invalid", "Wetech OP batch context is invalid.")
        if plan.get("schemaVersion") != "auto-oncall-op-query-batch-plan-v1" or plan.get("provider") != "wetech":
            raise ProviderError("op_batch_plan_invalid", "Wetech OP batch plan is invalid.")
        if plan.get("scenarioId") != context.get("scenarioId") or plan.get("timeWindow") != context.get("timeWindow"):
            raise ProviderError("op_batch_plan_mismatch", "Wetech OP batch plan does not match runner-owned context.")
        queries = plan.get("queries")
        if not isinstance(queries, list) or not 1 <= len(queries) <= OP_BATCH_MAX_QUERIES:
            raise ProviderError(
                "op_batch_query_count_invalid",
                f"Wetech OP batch requires one to {OP_BATCH_MAX_QUERIES} queries.",
            )
        if any(not isinstance(query, dict) for query in queries):
            raise ProviderError("op_batch_query_invalid", "Wetech OP batch queries must be objects.")
        queries = [
            _normalize_wetech_batch_query(query, str(context.get("scenarioId") or ""))
            for query in queries
        ]
        bounds = _historical_bounds(context.get("timeWindow"))
        comparison_bounds = _comparison_bounds(context)
        _assert_comparison_plan(context, queries, bounds, comparison_bounds)
        target = _normalize_wetech_batch_target(plan, queries, context)
        seen: set[str] = set()
        dependencies: dict[str, tuple[str, ...]] = {}
        for query in queries:
            query_id = query.get("queryId")
            if not _valid_batch_query_id(query_id) or query_id in seen:
                raise ProviderError("op_batch_query_invalid", "Wetech OP batch query id is invalid or duplicated.")
            params = query.get("params")
            query_dependencies = tuple(dict.fromkeys(
                dependency for dependency in (
                    _batch_string(params, "instanceFromQueryId"),
                    _batch_string(params, "windowFromQueryId"),
                ) if dependency is not None
            )) if query.get("operation") == "dashboard-query" and isinstance(params, dict) else ()
            if any(dependency not in seen for dependency in query_dependencies):
                raise ProviderError(
                    "op_batch_parameter_invalid",
                    "Query dependencies must reference earlier queries in the same batch.",
                )
            seen.add(query_id)
            dependencies[query_id] = query_dependencies
        prior_results: dict[str, dict[str, Any]] = {}
        dashboard_session_failure: tuple[str, str] | None = None
        completed_by_id: dict[str, dict[str, Any]] = {}
        state_lock = threading.Lock()
        started_wall, started_monotonic = time.time(), time.monotonic()
        result = {
            "schemaVersion": "auto-oncall-op-query-batch-v1",
            "provider": "wetech",
            "scenarioId": context.get("scenarioId"),
            "timeWindow": context.get("timeWindow"),
            "completion": "partial",
            "startedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(started_wall)),
            "durationMs": 0,
            "queries": [],
        }

        def execute_query(query: dict[str, Any]) -> tuple[str, dict[str, Any], dict[str, Any]]:
            nonlocal dashboard_session_failure
            query_id = query["queryId"]
            operation = str(query.get("operation") or "")
            query_started = time.monotonic()
            payload: dict[str, Any] = {}
            try:
                query_output_dir = args.result.parent / query_id / "provider-raw"
                with state_lock:
                    available_results = dict(prior_results)
                argv = _wetech_batch_argv(
                    target, query, bounds, query_output_dir, available_results,
                    comparison_bounds,
                )
                with _operation_slot(operation):
                    with state_lock:
                        prior_dashboard_failure = dashboard_session_failure
                    if operation == "dashboard-query" and prior_dashboard_failure is not None:
                        prior_code, prior_message = prior_dashboard_failure
                        code, payload = 2, {
                            "schema_version": "provider-result-v1",
                            "status": "error",
                            "operation": operation,
                            "error": {
                                "code": prior_code,
                                "message": (
                                    "TiDB Dashboard access already failed earlier in this "
                                    f"batch ({prior_code}: {prior_message})."
                                )[:500],
                            },
                            "summary": {},
                        }
                    else:
                        code, payload = _run_wetech_subcommand(argv)
                    if (
                        operation == "dashboard-query"
                        and code
                        and (
                            shared_failure_code := _wetech_dashboard_shared_failure_code(
                                payload, query_output_dir
                            )
                        )
                    ):
                        slot_error = payload.get("error", {}) if isinstance(payload.get("error"), dict) else {}
                        with state_lock:
                            dashboard_session_failure = (
                                shared_failure_code,
                                str(slot_error.get("message") or "Provider query failed.")[:500],
                            )
                status = "failed" if code else ("empty" if _wetech_result_empty(operation, payload) else "complete")
                error = payload.get("error", {}) if isinstance(payload.get("error"), dict) else {}
                error_code = str(error.get("code") or "wetech_provider_query_failed") if code else None
                error_message = str(error.get("message") or "Provider query failed.")[:500] if code else None
            except Exception as exc:
                code, payload, status = 2, {}, "failed"
                if isinstance(exc, ProviderError):
                    error_code, error_message = exc.code, exc.message[:500]
                else:
                    error_code, error_message = "wetech_provider_query_failed", str(exc)[:500]
            item = {
                "queryId": query_id,
                "operation": operation,
                **({"evidenceRole": "supplemental"}
                   if isinstance(query.get("params"), dict)
                   and query["params"].get("kind") == "statements"
                   else {}),
                "status": status,
                "statement": str(query.get("statement") or (
                    f"Wetech {operation} returned usable provider evidence for {query_id}."
                    if status == "complete" else
                    f"Wetech {operation} returned no usable provider evidence for {query_id}."
                ))[:2000],
                "durationMs": round((time.monotonic() - query_started) * 1000),
                "summary": payload,
                "handoffFields": query.get("handoffFields", []),
                "unsupportedHandoffFields": query.get("unsupportedHandoffFields", []),
                **({"emptyResultExpectation": query.get("emptyResultExpectation")}
                   if query.get("emptyResultExpectation") is not None else {}),
                "errorCode": error_code,
                "errorMessage": error_message,
            }
            return query_id, payload, item

        pending = {query["queryId"]: query for query in queries}
        running: dict[concurrent.futures.Future[tuple[str, dict[str, Any], dict[str, Any]]], dict[str, Any]] = {}
        running_by_operation = {"browser-query-range": 0, "dashboard-query": 0}
        local_limits = {
            "browser-query-range": METRICS_CONCURRENCY,
            "dashboard-query": DASHBOARD_CONCURRENCY,
        }
        with concurrent.futures.ThreadPoolExecutor(
            max_workers=METRICS_CONCURRENCY + DASHBOARD_CONCURRENCY
        ) as executor:
            while pending or running:
                while len(running) < METRICS_CONCURRENCY + DASHBOARD_CONCURRENCY:
                    ready = [
                        query for query_id, query in pending.items()
                        if all(dependency in completed_by_id for dependency in dependencies[query_id])
                        and running_by_operation[str(query.get("operation") or "")]
                            < local_limits[str(query.get("operation") or "")]
                    ]
                    if not ready:
                        break
                    ready.sort(key=lambda query: not dependencies[query["queryId"]])
                    query = ready[0]
                    operation = str(query.get("operation") or "")
                    pending.pop(query["queryId"])
                    future = executor.submit(execute_query, query)
                    running[future] = query
                    running_by_operation[operation] += 1
                if not running:
                    raise ProviderError(
                        "op_batch_parameter_invalid",
                        "Wetech OP batch query dependencies cannot be resolved.",
                    )
                finished, _ = concurrent.futures.wait(
                    running,
                    return_when=concurrent.futures.FIRST_COMPLETED,
                )
                for future in finished:
                    query = running.pop(future)
                    operation = str(query.get("operation") or "")
                    running_by_operation[operation] -= 1
                    query_id, payload, item = future.result()
                    prior_results[query_id] = payload
                    completed_by_id[query_id] = item
                    result["queries"] = [
                        completed_by_id[query["queryId"]]
                        for query in queries
                        if query["queryId"] in completed_by_id
                    ]
                    result["durationMs"] = round((time.monotonic() - started_monotonic) * 1000)
                    _atomic_write_json(args.result, result)
        completed = list(result["queries"])
        result["completion"] = "complete" if all(
            item.get("evidenceRole") == "supplemental" or item["status"] != "failed"
            for item in completed
        ) else "partial"
        result["durationMs"] = round((time.monotonic() - started_monotonic) * 1000)
        _atomic_write_json(args.result, result)
        _print_json(result)
        return 0
    except Exception as exc:
        _print_provider_error(operation="batch", exc=exc, output_dir=str(args.result.parent))
        return 2


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Wetech OP Grafana metrics and TiDB Dashboard browser-proxy helper")
    sub = parser.add_subparsers(dest="command", required=True)

    sub.add_parser("list-clusters", help="list known Wetech OP Grafana/Dashboard URLs")

    rebuild = sub.add_parser(
        "rebuild-datasource-map",
        help="login to regional Grafana and rebuild cluster-to-Prometheus-datasource mapping",
    )
    rebuild.add_argument("--cluster", help="single cluster alias to map")
    rebuild.add_argument("--region", help="region key to map, for example asia-prod")
    rebuild.add_argument("--grafana-url", help="override Grafana URL for all selected clusters")
    rebuild.add_argument("--headless", action="store_true", help="run without showing a browser; reuse cached session and auto-refresh login when possible")
    rebuild.add_argument("--terminal-2fa", action="store_true", help="prompt for the current 2FA code in the terminal instead of using --twofa-script")
    rebuild.add_argument(
        "--twofa-script",
        type=Path,
        default=DEFAULT_2FA_SCRIPT,
        help="script that prints the current P81 2FA code; defaults to scripts/get_verify_code.py",
    )
    rebuild.add_argument("--login-timeout", type=int, default=300, help="seconds to wait for visible P81/Grafana login, default 300")
    rebuild.add_argument("--debug-login", action="store_true", help="print login readiness probe state while waiting")
    rebuild.add_argument("--write", action="store_true", help="write matched datasource_uid/datasource_name values back to data/clusters.json")

    datasources = sub.add_parser("datasources", help="login to a cluster Grafana and list datasources")
    datasources.add_argument("--cluster", required=True)
    datasources.add_argument("--region", help="disambiguate duplicate cluster aliases")
    datasources.add_argument("--grafana-url", help="override the cluster Grafana URL")
    datasources.add_argument("--headless", action="store_true", help="run without showing a browser; reuse cached session and auto-refresh login when possible")
    datasources.add_argument("--terminal-2fa", action="store_true", help="prompt for the current 2FA code in the terminal instead of using --twofa-script")
    datasources.add_argument(
        "--twofa-script",
        type=Path,
        default=DEFAULT_2FA_SCRIPT,
        help="script that prints the current P81 2FA code; defaults to scripts/get_verify_code.py",
    )
    datasources.add_argument("--login-timeout", type=int, default=300, help="seconds to wait for visible P81/Grafana login, default 300")
    datasources.add_argument("--debug-login", action="store_true", help="print login readiness probe state while waiting")
    datasources.add_argument("--prometheus-only", action="store_true", help="only list Prometheus datasources")

    guide = sub.add_parser("login-guide", help="show browser login and curl-copy guide for a cluster")
    guide.add_argument("--cluster", required=True)
    guide.add_argument("--region", help="disambiguate duplicate cluster aliases")

    sub.add_parser("twofa-guide", help="show manual Google Authenticator binding steps")

    twofa_qr = sub.add_parser("twofa-qr", help="render an otpauth URI as a scannable QR code")
    twofa_qr.add_argument("--otpauth-uri", help="otpauth:// or otpauth-migration:// URI to render")
    twofa_qr.add_argument("--otpauth-file", help="file containing an otpauth:// or otpauth-migration:// URI")
    twofa_qr.add_argument("--output", help="SVG output path, for example /tmp/p81-2fa.svg")
    twofa_qr.add_argument("--print-terminal", action="store_true", help="also print a terminal QR code")

    probe = sub.add_parser("probe", help="verify a copied Grafana query curl")
    probe.add_argument("--curl-file", required=True, type=Path)
    probe.add_argument("--output-dir", help="run-scoped directory for raw error artifacts; defaults to a generated ./evidences/run-* directory")

    query = sub.add_parser("query-range", help="fetch Grafana Prometheus query_range data")
    query.add_argument("--curl-file", required=True, type=Path)
    query.add_argument("--name", required=True)
    query.add_argument("--query")
    query.add_argument("--start")
    query.add_argument("--end")
    query.add_argument("--step")
    query.add_argument("--metric-group-id")
    query.add_argument("--unit", default="")
    query.add_argument("--output-dir", help="run-scoped directory for raw metrics and compact report; defaults to a generated ./evidences/run-* directory")
    query.add_argument("--inline-chart-data", action="store_true", help="deprecated compatibility flag; raw chart data is always written to metrics.json and never stdout")

    browser_query = sub.add_parser(
        "browser-query-range",
        help="login with a skill-managed Playwright profile and fetch Grafana Prometheus query_range data",
    )
    browser_query.add_argument("--cluster", required=True)
    browser_query.add_argument("--region", help="disambiguate duplicate cluster aliases")
    browser_query.add_argument("--grafana-url", help="override the cluster Grafana URL")
    browser_query.add_argument("--name", required=True)
    browser_query.add_argument("--query", required=True)
    browser_query.add_argument("--start", help="epoch seconds; defaults to now minus --last")
    browser_query.add_argument("--end", help="epoch seconds; defaults to now")
    browser_query.add_argument("--last", default="1h", help="relative range when --start is omitted, for example 30m or 1h")
    browser_query.add_argument("--step", default="30")
    browser_query.add_argument("--metric-group-id")
    browser_query.add_argument("--unit", default="")
    browser_query.add_argument("--datasource-name-regex", help="optional regex to select a Prometheus datasource by name")
    browser_query.add_argument("--headless", action="store_true", help="run without showing a browser; reuse cached session and auto-refresh login when possible")
    browser_query.add_argument("--terminal-2fa", action="store_true", help="prompt for the current 2FA code in the terminal instead of using --twofa-script")
    browser_query.add_argument(
        "--twofa-script",
        type=Path,
        default=DEFAULT_2FA_SCRIPT,
        help="script that prints the current P81 2FA code; defaults to scripts/get_verify_code.py",
    )
    browser_query.add_argument("--login-timeout", type=int, default=300, help="seconds to wait for visible P81/Grafana login, default 300")
    browser_query.add_argument("--debug-login", action="store_true", help="print login readiness probe state while waiting")
    browser_query.add_argument("--output-dir", help="run-scoped directory for raw metrics and compact report; defaults to a generated ./evidences/run-* directory")
    browser_query.add_argument("--inline-chart-data", action="store_true", help="deprecated compatibility flag; raw chart data is always written to metrics.json and never stdout")

    dashboard_query = sub.add_parser(
        "dashboard-query",
        help="login with the shared P81/Dashboard browser session and fetch TopSQL, statements, or slow-query data",
    )
    dashboard_query.add_argument("--cluster", required=True)
    dashboard_query.add_argument("--region", help="disambiguate duplicate cluster aliases")
    dashboard_query.add_argument("--dashboard-url", help="override the cluster TiDB Dashboard URL")
    dashboard_query.add_argument("--dashboard-cluster-name", help="override the TiDB Dashboard proxy cluster name")
    dashboard_query.add_argument(
        "--kind",
        required=True,
        choices=["topsql", "statements", "slow-query"],
    )
    dashboard_query.add_argument("--start", help="epoch seconds; defaults to now minus --last")
    dashboard_query.add_argument("--end", help="epoch seconds; defaults to now")
    dashboard_query.add_argument("--last", default="30m", help="relative range when --start is omitted, for example 30m or 1h")
    dashboard_query.add_argument("--limit", type=int, help="row limit for slow-query or top count for TopSQL")
    dashboard_query.add_argument("--preview-limit", type=int, default=10, help="compact summary rows, hard-capped at 10")
    dashboard_query.add_argument(
        "--fields",
        help="comma-separated SQL Statements or Slow Query fields; defaults to a kind-specific diagnostic projection",
    )
    dashboard_query.add_argument("--text", default="", help="optional statement/slow-query text filter")
    dashboard_query.add_argument("--digest", default="", help="optional exact Slow Query digest filter")
    dashboard_query.add_argument("--order-by", default="", help="optional Slow Query order field, for example query_time")
    dashboard_query.add_argument("--desc", action="store_true", help="sort Slow Query rows descending by --order-by")
    dashboard_query.add_argument("--instance", help="required for TopSQL; select one affected instance from time-aligned metrics")
    dashboard_query.add_argument("--instance-type", default="tidb", help="TopSQL instance type, default tidb")
    dashboard_query.add_argument("--group-by", default="query", choices=["query", "plan"], help="TopSQL summary grouping")
    dashboard_query.add_argument("--window", default="10s", help="TopSQL summary window, default 10s")
    dashboard_query.add_argument(
        "--command-timeout",
        type=_topsql_command_timeout,
        default=DEFAULT_TOPSQL_COMMAND_TIMEOUT_SECONDS,
        help="TopSQL total command budget in seconds, default and maximum 120; no automatic retry",
    )
    dashboard_query.add_argument("--headless", action="store_true", help="run without showing a browser; reuse cached session and auto-refresh login when possible")
    dashboard_query.add_argument("--terminal-2fa", action="store_true", help="prompt for the current 2FA code in the terminal instead of using --twofa-script")
    dashboard_query.add_argument(
        "--twofa-script",
        type=Path,
        default=DEFAULT_2FA_SCRIPT,
        help="script that prints the current P81 2FA code; defaults to scripts/get_verify_code.py",
    )
    dashboard_query.add_argument("--login-timeout", type=int, default=300, help="seconds to wait for visible P81/Dashboard login, default 300")
    dashboard_query.add_argument("--debug-login", action="store_true", help="print login readiness probe state while waiting")
    dashboard_query.add_argument(
        "--output-dir",
        help="run-scoped directory for raw dashboard evidence and compact report; defaults to a generated ./evidences/run-* directory",
    )

    batch = sub.add_parser("batch", help="execute one bounded OP evidence plan with operation-class concurrency")
    batch.add_argument("context", type=Path)
    batch.add_argument("plan", type=Path)
    batch.add_argument("result", type=Path)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    if args.command == "list-clusters":
        return cmd_list_clusters(args)
    if args.command == "rebuild-datasource-map":
        return cmd_rebuild_datasource_map(args)
    if args.command == "datasources":
        return cmd_datasources(args)
    if args.command == "login-guide":
        return cmd_login_guide(args)
    if args.command == "twofa-guide":
        return cmd_twofa_guide(args)
    if args.command == "twofa-qr":
        return cmd_twofa_qr(args)
    if args.command == "probe":
        return cmd_probe(args)
    if args.command == "query-range":
        return cmd_query_range(args)
    if args.command == "browser-query-range":
        return cmd_browser_query_range(args)
    if args.command == "dashboard-query":
        return cmd_dashboard_query(args)
    if args.command == "batch":
        _install_batch_signal_forwarding()
        return cmd_batch(args)
    parser.error(f"unknown command: {args.command}")
    return 2


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