#!/usr/bin/env python3
"""Pool-wide Serverless TiKV daily inspection using Clinic parent endpoints."""
from __future__ import annotations

import argparse
import importlib
import json
import math
import os
import re
import stat
import sys
from collections import Counter
from collections.abc import Callable, Iterable, Mapping, Sequence
from dataclasses import dataclass, fields, is_dataclass
from datetime import date, datetime, time, timedelta, timezone
from pathlib import Path
from typing import Protocol, cast
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError


class _HttpResponse(Protocol):
    status_code: int

    def __enter__(self) -> _HttpResponse: ...

    def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None: ...

    def iter_content(self, chunk_size: int) -> Iterable[bytes]: ...


@dataclass(frozen=True)
class _RequestsApi:
    get: Callable[..., _HttpResponse]
    request_exception: type[Exception]


def _load_requests() -> _RequestsApi:
    try:
        module = importlib.import_module("requests")
    except ModuleNotFoundError as error:
        raise RuntimeError(
            "requests is required; install it with "
            "python3 -m pip install -r requirements.txt from the skill directory"
        ) from error
    request_get = getattr(module, "get", None)
    request_exception = getattr(getattr(module, "exceptions", None), "RequestException", None)
    if (
        not callable(request_get)
        or not isinstance(request_exception, type)
        or not issubclass(request_exception, Exception)
    ):
        raise RuntimeError("installed requests package exposes an unexpected API")
    return _RequestsApi(
        cast(Callable[..., _HttpResponse], request_get),
        cast(type[Exception], request_exception),
    )


DEFAULT_TZ = "Asia/Shanghai"
DEFAULT_ENV_FILE = str(Path.home() / ".env")
SERVERLESS_NAMESPACE = "tidb-serverless"
SERVERLESS_COMPONENT = "tikv"
SERVERLESS_LOG_CONTAINER = "tikv-log"
# TiKV storage pods are StatefulSet-style pods with a numeric ordinal. Keep
# this deliberately narrow so a node-exporter series cannot be attributed to
# an unrelated Serverless workload that merely contains "tikv" in its name.
TIKV_STORAGE_POD_RE = r"^tikv(?:-[a-z0-9]+)*-[0-9]+$"
# nvme0n1 is the node root disk in the validated Serverless pool shape. Do
# not match partitions or the root device when observing TiKV data disks.
NON_ROOT_NVME_DEVICE_RE = r"^nvme[1-9][0-9]*n[1-9][0-9]*$"
RANGE_STEP_SECONDS = 300
SUSTAINED_30M_POINTS = 6
SUSTAINED_15M_POINTS = 3
MAX_WINDOW_SECONDS = 24 * 60 * 60
MAX_DAILY_REPORT_POOLS = 20
MAX_MULTI_DAY_REPORT_DAYS = 31
MAX_ENV_FILE_BYTES = 64 * 1024
MAX_HTTP_RESPONSE_BYTES = 5 * 1024 * 1024
MAX_LOG_GROUPS = 10
LOKI_LOG_QUERY_LIMIT = 1000
ALLOWED_CLINIC_HOSTS = frozenset({
    "clinic.pingcap.com",
    "dev-clinic.pingcap.com",
    "staging-clinic.pingcap.com",
})
ALLOWED_CLINIC_ENDPOINTS = frozenset(
    endpoint
    for hostname in ALLOWED_CLINIC_HOSTS
    for endpoint in (f"https://{hostname}", f"https://{hostname}:443")
)
MEMORY_LEAK_BASELINE_WINDOW = "7d"
MEMORY_LEAK_OBSERVE_DELTA_PCT = 3.0
MEMORY_LEAK_WARN_DELTA_PCT = 5.0
MEMORY_LEAK_EXPLAIN_REGION_DELTA_ABS = 100.0
MEMORY_LEAK_EXPLAIN_REGION_DELTA_RATIO_PCT = 10.0
MEMORY_LEAK_EXPLAIN_QPS_DELTA_ABS = 100.0
MEMORY_LEAK_EXPLAIN_QPS_DELTA_RATIO_PCT = 20.0
MEMORY_LEAK_EXPLAIN_CPU_DELTA_PCT = 5.0
T1_REPORT_MIN_MAX_CORES = 0.1
IDENTITY_VALUE_RE = re.compile(r"[A-Za-z0-9._/+:\-]+\Z")
LEADING_LOG_DATETIME_RE = re.compile(
    r"^\s*(?:\[?\d{4}[-/]\d{2}[-/]\d{2}[ T]\d{2}:\d{2}:\d{2}(?:[.,]\d+)?\s*(?:Z|[+-]\d{2}:?\d{2})?\]?\s*)"
)
LOG_SEVERITY_RE = re.compile(r"\b(ERROR|FATAL|WARN)\b", re.IGNORECASE)
MARKDOWN_ESCAPE_RE = re.compile(r"""\\([!"#$%&'()*+,\-./:;<=>?@\[\]^_`{|}~])""")
GRPC_TYPES = ("kv_prewrite", "kv_commit", "coprocessor", "kv_get", "kv_batch_get", "kv_batch_get_command")
G1_READ_GRPC_TYPES = ("coprocessor", "kv_get", "kv_batch_get")
G1_WRITE_GRPC_TYPES = ("kv_prewrite", "kv_commit")
G1_GRPC_TYPES = (*G1_READ_GRPC_TYPES, *G1_WRITE_GRPC_TYPES)
READ_GRPC_TYPES = "coprocessor|kv_get|kv_batch_get|kv_batch_get_command"
@dataclass(frozen=True)
class ThreadCpuGroupSpec:
    name: str
    thread_name_regex: str


# Ordered stable thread CPU groups from the Serverless TiKV Details dashboard.
# Exact duplicate panels are omitted, while panels with distinct meanings remain.
T1_THREAD_CPU_GROUPS = (
    ThreadCpuGroupSpec("Raft store CPU", r"(raftstore|rs|rfstore)_.*"),
    ThreadCpuGroupSpec("Async apply CPU", r"apply_[0-9]+|apply_follower_[0-9]+"),
    ThreadCpuGroupSpec("Raft async writer CPU", r"raft_io.*"),
    ThreadCpuGroupSpec(
        "Scheduler pool CPU", r"sched_tokio_pool.*|sched_yatp_pool.*|sched_background.*"
    ),
    ThreadCpuGroupSpec("gRPC poll CPU", r"grpc.*"),
    ThreadCpuGroupSpec("KvEngine worker CPU", r"txn-chunk-worker.*|txn_chunk_worker.*"),
    ThreadCpuGroupSpec("RfEngine worker CPU", r"sub-compact-wkr.*|compact-wal-worker.*"),
    ThreadCpuGroupSpec("Unified read pool CPU", r"unified_read_po.*"),
    ThreadCpuGroupSpec("Background worker CPU", r"background.*"),
    ThreadCpuGroupSpec("GC worker CPU", r"gc_worker.*"),
    ThreadCpuGroupSpec("Import CPU", r"sst_.*"),
    ThreadCpuGroupSpec("Backup CPU", r"(backup-worker|bkwkr|backup_endpoint).*"),
    ThreadCpuGroupSpec("CDC worker CPU", r"(cdcwkr.*|tso|cdc_.*)"),
    ThreadCpuGroupSpec("TSO worker CPU", r"tso_worker"),
    ThreadCpuGroupSpec("IME CPU", r"ime.*"),
    ThreadCpuGroupSpec("IA CPU", r"s3fifo.*"),
    ThreadCpuGroupSpec("DFS worker CPU", r"(builtin-dfs.*|memory-fs.*|s3-client.*)"),
    ThreadCpuGroupSpec("Cloud worker CPU", r"worker-server.*"),
    ThreadCpuGroupSpec("Lock Manager Thread CPU", r"(waiter_manager.*|deadlock_detect.*)"),
    ThreadCpuGroupSpec("Resolved TS Worker CPU", r"resolved_ts.*"),
    ThreadCpuGroupSpec("Advance ts Worker CPU", r"advance_ts.*"),
    ThreadCpuGroupSpec("Scan lock Worker CPU", r"inc_scan.*"),
    ThreadCpuGroupSpec("Backup CPU Utilization", r"(b.*k.*w.*k.*|backup_io)"),
)


@dataclass(frozen=True, order=True)
class PoolIdentity:
    control_plane_info: str
    k8s_cluster_info: str

    def __post_init__(self) -> None:
        for label, value in (("CONTROL_PLANE", self.control_plane_info), ("K8S_CLUSTER", self.k8s_cluster_info)):
            if not value or not IDENTITY_VALUE_RE.fullmatch(value):
                raise ValueError(
                    f"--pool {label} must contain only letters, digits, '.', '_', '-', '/', ':', or '+'"
                )
        control_plane_segments = self.control_plane_info.split("/")
        if len(control_plane_segments) != 3 or not all(control_plane_segments):
            raise ValueError(
                "--pool CONTROL_PLANE must be the full Grafana routing value: "
                "<pool>/<vendor-region>/<application>"
            )

    @classmethod
    def parse(cls, control_plane_info: str, k8s_cluster_info: str) -> "PoolIdentity":
        return cls(control_plane_info, k8s_cluster_info)

    @property
    def promql_control_plane_info(self) -> str:
        """Return the two-segment metric label derived from Grafana's route value."""
        return "/".join(self.control_plane_info.split("/")[:2])

    def promql_selector(
        self, extra: Mapping[str, str | tuple[str, str]] | None = None
    ) -> str:
        labels: list[tuple[str, str, str]] = [
            ("control_plane_info", "=", self.promql_control_plane_info),
            ("k8s_cluster_info", "=", self.k8s_cluster_info),
            ("namespace", "=", SERVERLESS_NAMESPACE),
            ("component", "=", SERVERLESS_COMPONENT),
        ]
        fixed_keys = {label for label, _, _ in labels}
        for key, raw_matcher in (extra or {}).items():
            if key in fixed_keys:
                raise ValueError(f"cannot override required Serverless selector label: {key}")
            operator, value = ("=", raw_matcher) if isinstance(raw_matcher, str) else raw_matcher
            if operator not in {"=", "!=", "=~", "!~"}:
                raise ValueError(f"unsupported PromQL matcher operator for {key}: {operator}")
            labels.append((key, operator, value))
        return "{" + ",".join(
            f'{key}{operator}"{escape_label_value(value)}"' for key, operator, value in labels
        ) + "}"

    def promql_tikv_storage_pod_selector(self) -> str:
        """Scope a Kubernetes pod map without putting workload labels on node metrics."""
        labels = (
            ("control_plane_info", "=", self.promql_control_plane_info),
            ("k8s_cluster_info", "=", self.k8s_cluster_info),
            ("namespace", "=", SERVERLESS_NAMESPACE),
            ("pod", "=~", TIKV_STORAGE_POD_RE),
        )
        return "{" + ",".join(
            f'{key}{operator}"{escape_label_value(value)}"' for key, operator, value in labels
        ) + "}"

    def loki_selector(self) -> str:
        return "{" + ",".join(
            (
                f'cluster="{escape_label_value(self.k8s_cluster_info)}"',
                f'namespace="{SERVERLESS_NAMESPACE}"',
                f'container="{SERVERLESS_LOG_CONTAINER}"',
            )
        ) + "}"


@dataclass(frozen=True)
class Window:
    start: datetime
    end: datetime
    timezone_name: str

    @property
    def seconds(self) -> int:
        return max(1, int(self.end.timestamp() - self.start.timestamp()))

    @property
    def prom_window(self) -> str:
        return f"{self.seconds}s"


@dataclass(frozen=True)
class ReportRange:
    start: datetime
    end: datetime
    timezone_name: str
    start_date: str
    end_date: str
    label: str


@dataclass(frozen=True)
class RuleSpec:
    rule_id: str
    status: str
    section: str
    purpose: str


RULE_SPECS = (
    RuleSpec("E1", "exact", "Error Signals", "critical errors"),
    RuleSpec("E2", "exact", "Error Signals", "channel full"),
    RuleSpec("E3", "exact", "Error Signals", "scheduler busy"),
    RuleSpec("E4", "exact", "Error Signals", "coprocessor full"),
    RuleSpec("E5", "exact", "Error Signals", "server report failures"),
    RuleSpec("G1", "exact", "gRPC QPS and Latency", "gRPC message rate"),
    RuleSpec("G2", "exact", "gRPC QPS and Latency", "gRPC failed messages"),
    RuleSpec("G3", "exact", "gRPC QPS and Latency", "write-path gRPC P99"),
    RuleSpec("G4", "exact", "gRPC QPS and Latency", "average gRPC duration"),
    RuleSpec("G5", "exact", "gRPC QPS and Latency", "read-path gRPC P99"),
    RuleSpec("B1", "exact", "CPU and Memory", "TiKV CPU percentage"),
    RuleSpec("B2", "replacement", "CPU and Memory", "TiKV memory percentage"),
    RuleSpec("B3", "replacement", "CPU and Memory", "memory-growth versus baseline judgment"),
    RuleSpec("B4", "exact", "CPU and Memory", "TiKV restarts"),
    RuleSpec("T1", "replacement", "Critical Thread Pool CPU", "thread CPU groups"),
    RuleSpec("T2", "exact", "Critical Thread Pool CPU", "unified read running tasks"),
    RuleSpec("T3", "exact", "Critical Thread Pool CPU", "unified read time used"),
    RuleSpec("C1", "replacement", "Disk Usage and Region Count", "store capacity usage"),
    RuleSpec("C2", "exact", "Disk Usage and Region Count", "region and leader balance"),
    RuleSpec("D1", "exact", "Disk Load", "raft engine P99"),
    RuleSpec("D2", "exact", "Disk Load", "WAL write throttle duration"),
    RuleSpec("D3", "exact", "Disk Load", "Raft Engine pending compaction WALs"),
    RuleSpec("D4", "exact", "Disk Load", "pool-scoped disk utilization"),
    RuleSpec("D5", "exact", "Disk Load", "disk queue"),
    RuleSpec("D6", "exact", "Disk Load", "disk read latency"),
    RuleSpec("D7", "exact", "Disk Load", "disk write latency"),
    RuleSpec("D8", "unavailable", "Disk Load", "exceeded IOPS"),
    RuleSpec("F1", "exact", "Disk Load", "flow-control throttle duration"),
    RuleSpec("F2", "exact", "Disk Load", "flow-control discard ratio"),
    RuleSpec("P1", "replacement", "Disk Load", "store slow score"),
    RuleSpec("R1", "partial", "Disk Load", "raft log lag rate"),
    RuleSpec("K1", "replacement", "Compaction", "pending compaction shards"),
    RuleSpec("K2", "exact", "Compaction", "compaction failures and local compaction disabled"),
    RuleSpec("K3", "exact", "Compaction", "compaction result counts by container and result"),
    RuleSpec("L1", "log", "ERROR/WARN Logs", "actionable TiKV logs"),
)
RULE_BY_ID = {rule.rule_id: rule for rule in RULE_SPECS}
UNAVAILABLE_REASONS = {
    "D8": "no aws_ebs_* exceeded-IOPS series exists in Serverless parent metrics",
}
DASHBOARD_METRIC_REGISTRY: dict[str, tuple[str, ...]] = {
    "E1": ("tikv_critical_error_total",), "E2": ("tikv_channel_full_total",),
    "E3": ("tikv_scheduler_too_busy_total",), "E4": ("tikv_coprocessor_request_error",),
    "E5": ("tikv_server_report_failure_msg_total",),
    "G1": ("tikv_grpc_msg_duration_seconds_count",), "G2": ("tikv_grpc_msg_fail_total",),
    "G3": ("tikv_grpc_msg_duration_seconds_bucket",),
    "G4": ("tikv_grpc_msg_duration_seconds_sum", "tikv_grpc_msg_duration_seconds_count"),
    "G5": ("tikv_grpc_msg_duration_seconds_bucket",),
    "B1": ("process_cpu_seconds_total", "tikv_server_cpu_cores_quota"),
    "B2": ("process_resident_memory_bytes", "tikv_server_memory_quota_bytes"),
    "B3": (
        "process_resident_memory_bytes",
        "tikv_server_memory_quota_bytes",
        "tikv_raftstore_region_count",
        "tikv_grpc_msg_duration_seconds_count",
        "process_cpu_seconds_total",
        "tikv_server_cpu_cores_quota",
    ),
    "B4": ("process_start_time_seconds",), "T1": ("tikv_thread_cpu_seconds_total",),
    "T2": ("tikv_unified_read_pool_running_tasks",),
    "C1": ("tikv_store_size_bytes",), "C2": ("tikv_raftstore_region_count",),
    "D1": ("tikv_raftstore_append_log_duration_seconds_bucket", "tikv_raftstore_commit_log_duration_seconds_bucket", "tikv_raftstore_apply_log_duration_seconds_bucket"),
    "D2": ("raft_engine_wal_write_throttle_duration_seconds",),
    "D3": ("raft_engine_pending_compaction_wals",),
    "D4": ("node_disk_io_time_seconds_total",),
    "F1": ("tikv_scheduler_throttle_duration_seconds_bucket",), "P1": ("store_slow_score",),
    "R1": ("tikv_raftstore_log_lag_sum",),
    "K1": ("kv_engine_pending_compaction_shards",),
    "K2": ("kv_engine_compaction_result_counter",),
    "K3": ("kv_engine_compaction_result_counter",),
}
# These validated parent-only metrics are intentionally absent from the
# external TiKV dashboard source used by the dashboard registry test.
PARENT_ONLY_METRIC_REGISTRY: dict[str, tuple[str, ...]] = {
    "T3": ("multilevel_level_elapsed",),
    "F2": ("tikv_scheduler_discard_ratio",),
    "D5": ("node_disk_io_time_weighted_seconds_total", "kube_pod_info"),
    "D6": (
        "node_disk_read_time_seconds_total",
        "node_disk_reads_completed_total",
        "kube_pod_info",
    ),
    "D7": (
        "node_disk_write_time_seconds_total",
        "node_disk_writes_completed_total",
        "kube_pod_info",
    ),
}


@dataclass(frozen=True)
class G1Rate:
    aggregation: str
    request_type: str
    messages_per_second: float


@dataclass(frozen=True)
class T1CpuRate:
    group: str
    instance: str | None
    max_cores: float | None
    avg_cores: float | None
    error: str | None = None


@dataclass(frozen=True)
class RuleResult:
    status: str
    value: float | None
    detail: str
    sustained_at: frozenset[float] = frozenset()
    warning: bool = False
    log_groups: tuple[tuple[int, str], ...] = ()
    truncated: bool = False
    g1_rates: tuple[G1Rate, ...] = ()
    t1_cpu_rates: tuple[T1CpuRate, ...] = ()

    @property
    def unknown(self) -> bool:
        return self.status == "Unknown"


def escape_label_value(value: str) -> str:
    return value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n").replace("\r", "\\r")


def markdown_safe(value: object) -> str:
    """Render untrusted text literally in Feishu Markdown."""
    text = str(value)
    sanitized = "".join("�" if ord(character) < 32 or ord(character) == 127 else character for character in text)
    unescaped = MARKDOWN_ESCAPE_RE.sub(r"\1", sanitized)
    return unescaped.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")


def normalize_pools(values: Sequence[Sequence[str]]) -> list[PoolIdentity]:
    pools: list[PoolIdentity] = []
    seen: set[PoolIdentity] = set()
    for pair in values:
        if len(pair) != 2:
            raise ValueError("each --pool must contain CONTROL_PLANE and K8S_CLUSTER")
        identity = PoolIdentity.parse(pair[0], pair[1])
        if identity not in seen:
            pools.append(identity)
            seen.add(identity)
    return pools


def load_shell_env(env_file: str) -> None:
    path = Path(env_file).expanduser()
    if not path.exists():
        return
    if path.is_symlink() or not stat.S_ISREG(path.stat().st_mode):
        raise ValueError("env file must be a regular file")
    if path.stat().st_size > MAX_ENV_FILE_BYTES:
        raise ValueError(f"env file exceeds {MAX_ENV_FILE_BYTES} byte limit")
    allowed_keys = {"CLINIC_API_KEY", "CLINIC_ENV", "CLINIC_ENDPOINT"}
    for raw_line in path.read_text().splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith("export "):
            line = line[7:]
        key, separator, value = line.partition("=")
        key = key.strip()
        if separator and key in allowed_keys and key not in os.environ:
            os.environ[key] = value.strip().strip("'\"")


def parse_relative_window(value: str) -> timedelta:
    match = re.fullmatch(r"\s*(\d+)\s*([hd])\s*", value, re.IGNORECASE)
    if not match:
        raise ValueError("relative window must look like 2h, 6h, or 1d")
    amount = int(match.group(1))
    if amount <= 0:
        raise ValueError("relative window must be greater than zero")
    duration = timedelta(hours=amount) if match.group(2).lower() == "h" else timedelta(days=amount)
    if duration.total_seconds() > MAX_WINDOW_SECONDS:
        raise ValueError("inspection window must not exceed 24 hours")
    return duration


def parse_datetime(value: str, timezone_name: str) -> datetime:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    return parsed if parsed.tzinfo else parsed.replace(tzinfo=ZoneInfo(timezone_name))


def resolve_window(
    report_date: str | None,
    start: str | None,
    end: str | None,
    relative_window: str | None,
    timezone_name: str,
) -> Window:
    tz = ZoneInfo(timezone_name)
    if start and relative_window:
        raise ValueError("--start cannot be combined with --relative-window; use --end as the anchor")
    if start and end and relative_window:
        raise ValueError("--relative-window cannot be combined with --start and --end")
    if start or end:
        if start and end:
            start_time = parse_datetime(start, timezone_name)
            end_time = parse_datetime(end, timezone_name)
        elif relative_window:
            end_time = parse_datetime(end, timezone_name) if end else datetime.now(tz)
            duration = parse_relative_window(relative_window)
            start_time = datetime.fromtimestamp(
                end_time.timestamp() - duration.total_seconds(), tz=end_time.tzinfo
            )
        else:
            raise ValueError("--start and --end must be provided together unless --relative-window is used")
    elif relative_window:
        end_time = datetime.now(tz)
        duration = parse_relative_window(relative_window)
        start_time = datetime.fromtimestamp(
            end_time.timestamp() - duration.total_seconds(), tz=end_time.tzinfo
        )
    else:
        now = datetime.now(tz)
        inspection_date = date.fromisoformat(report_date) if report_date else now.date()
        start_time = datetime.combine(inspection_date, time.min, tzinfo=tz)
        end_time = (
            datetime.combine(inspection_date + timedelta(days=1), time.min, tzinfo=tz)
            if inspection_date < now.date()
            else now
        )
    if end_time.timestamp() <= start_time.timestamp():
        raise ValueError("inspection end must be after inspection start")
    if end_time.timestamp() - start_time.timestamp() > MAX_WINDOW_SECONDS:
        raise ValueError("inspection window must not exceed 24 hours")
    return Window(start_time, end_time, timezone_name)


def resolve_daily_report_windows(
    report_date: str | None,
    start: str | None,
    end: str | None,
    relative_window: str | None,
    timezone_name: str,
) -> list[Window]:
    """Resolve daily-report input into one or more <=24-hour local-day slices."""
    if relative_window or bool(start) != bool(end):
        return [resolve_window(report_date, start, end, relative_window, timezone_name)]

    local_timezone = ZoneInfo(timezone_name)
    if start and end:
        start_time = parse_datetime(start, timezone_name)
        end_time = parse_datetime(end, timezone_name)
    else:
        now = datetime.now(local_timezone)
        inspection_date = date.fromisoformat(report_date) if report_date else now.date()
        start_time = datetime.combine(inspection_date, time.min, tzinfo=local_timezone)
        end_time = (
            datetime.combine(
                inspection_date + timedelta(days=1), time.min, tzinfo=local_timezone
            )
            if inspection_date < now.date()
            else now
        )
    if end_time.timestamp() <= start_time.timestamp():
        raise ValueError("inspection end must be after inspection start")
    first_date = start_time.astimezone(local_timezone).date()
    exclusive_end = (
        end_time.astimezone(timezone.utc) - timedelta(microseconds=1)
    ).astimezone(local_timezone)
    last_date = exclusive_end.date()
    if (last_date - first_date).days + 1 > MAX_MULTI_DAY_REPORT_DAYS:
        raise ValueError(f"multi-day daily-report supports at most {MAX_MULTI_DAY_REPORT_DAYS} local dates")

    windows: list[Window] = []
    current = start_time
    while current < end_time:
        local_current = current.astimezone(local_timezone)
        next_midnight = datetime.combine(local_current.date() + timedelta(days=1), time.min, tzinfo=local_timezone)
        elapsed_limit = datetime.fromtimestamp(
            current.timestamp() + MAX_WINDOW_SECONDS, tz=local_timezone
        )
        slice_end = min(next_midnight, elapsed_limit, end_time, key=lambda value: value.timestamp())
        windows.append(Window(current, slice_end, timezone_name))
        current = slice_end
    return windows


def iso_utc(value: datetime) -> str:
    return value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def to_local_string(value: datetime, timezone_name: str) -> str:
    return value.astimezone(ZoneInfo(timezone_name)).strftime("%Y-%m-%d %H:%M:%S %Z")


def last_included_local_date(window: Window) -> str:
    exclusive_end = window.end.astimezone(timezone.utc) - timedelta(microseconds=1)
    return exclusive_end.astimezone(ZoneInfo(window.timezone_name)).date().isoformat()


def resolve_report_range(windows: Sequence[Window]) -> ReportRange:
    if not windows:
        raise ValueError("report range requires at least one inspection window")
    timezone_name = windows[0].timezone_name
    if any(window.timezone_name != timezone_name for window in windows):
        raise ValueError("report windows must use the same timezone")
    start = min((window.start for window in windows), key=lambda value: value.timestamp())
    end = max((window.end for window in windows), key=lambda value: value.timestamp())
    local_timezone = ZoneInfo(timezone_name)
    start_date = start.astimezone(local_timezone).date().isoformat()
    end_date = last_included_local_date(Window(start, end, timezone_name))
    label = start_date if start_date == end_date else f"{start_date} to {end_date}"
    return ReportRange(start, end, timezone_name, start_date, end_date, label)


def metric(pool: PoolIdentity, metric_name: str, **labels: str | tuple[str, str]) -> str:
    return f"{metric_name}{pool.promql_selector(labels)}"


def histogram_p99(pool: PoolIdentity, metric_name: str, **labels: str | tuple[str, str]) -> str:
    buckets = metric(pool, f"{metric_name}_bucket", **labels)
    return f"histogram_quantile(0.99, sum by (le,instance,type) (rate({buckets}[5m])))"


def cpu_percentage_query(pool: PoolIdentity) -> str:
    numerator = f"sum by (instance) (rate({metric(pool, 'process_cpu_seconds_total', job=('=~', '.*(tikv|copr|schema-manager).*'))}[5m]))"
    denominator = f"sum by (instance) ({metric(pool, 'tikv_server_cpu_cores_quota', job=('=~', '.*(tikv|copr|schema-manager).*'))})"
    return f"100 * ({numerator} / {denominator})"


def memory_percentage_query(pool: PoolIdentity) -> str:
    numerator = f"sum by (instance) ({metric(pool, 'process_resident_memory_bytes', job=('=~', '.*(tikv|copr|schema-manager).*'))})"
    denominator = f"sum by (instance) ({metric(pool, 'tikv_server_memory_quota_bytes', job=('=~', '.*(tikv|copr|schema-manager).*'))})"
    return f"100 * ({numerator} / {denominator})"


def store_usage_percentage_query(pool: PoolIdentity) -> str:
    used = f"sum by (instance) ({metric(pool, 'tikv_store_size_bytes', type='all_used')})"
    capacity = f"sum by (instance) ({metric(pool, 'tikv_store_size_bytes', type='capacity')})"
    return f"100 * ({used} / {capacity})"


def node_disk_metric(metric_name: str) -> str:
    """Return an unscoped node-exporter selector for validated TiKV data disks."""
    return f'{metric_name}{{device=~"{NON_ROOT_NVME_DEVICE_RE}"}}'


def node_disk_latency_expression(duration_metric: str, operations_metric: str) -> str:
    numerator = f"rate({node_disk_metric(duration_metric)}[5m])"
    denominator = f"rate({node_disk_metric(operations_metric)}[5m])"
    return (
        f"(({numerator} / {denominator}) and on(instance,device) ({denominator} > 0)) "
        f"or on(instance,device) (0 * {denominator})"
    )


def node_disk_tikv_storage_join(pool: PoolIdentity, node_expression: str) -> str:
    """Map node-exporter disk observations to scoped TiKV storage pods by node."""
    pod_mapping = (
        f'label_replace(kube_pod_info{pool.promql_tikv_storage_pod_selector()}, '
        '"instance", "$1", "node", "(.*)")'
    )
    return f"({node_expression}) * on(instance) group_left(pod) {pod_mapping}"


def peak_tikv_storage_disk_query(pool: PoolIdentity, node_expression: str, window: Window) -> str:
    """Reduce node disk observations per TiKV pod/device using the D1/D4 window maximum."""
    joined = node_disk_tikv_storage_join(pool, node_expression)
    return f"max_over_time((max by (pod,device) ({joined}))[{window.prom_window}:5m])"


def build_rule_queries(pool: PoolIdentity, window: Window) -> dict[str, str]:
    duration = window.prom_window
    grpc_filter = ("=~", "|".join(GRPC_TYPES))
    g1_rate = (
        f"sum by (type) (rate({metric(pool, 'tikv_grpc_msg_duration_seconds_count', type=('=~', '|'.join(G1_GRPC_TYPES)))}[1m]))"
    )
    g1_max = (
        f'label_replace(max_over_time(({g1_rate})[{duration}:1m]), '
        '"aggregation", "max", "", "")'
    )
    g1_avg = (
        f'label_replace(avg_over_time(({g1_rate})[{duration}:1m]), '
        '"aggregation", "avg", "", "")'
    )
    raft_p99 = {
        "append": histogram_p99(pool, "tikv_raftstore_append_log_duration_seconds"),
        "commit": histogram_p99(pool, "tikv_raftstore_commit_log_duration_seconds"),
        "apply": histogram_p99(pool, "tikv_raftstore_apply_log_duration_seconds"),
    }
    raft_window_p99 = [
        f'label_replace(max_over_time(({query})[{duration}:5m]), "raft_stage", "{stage}", "", "")'
        for stage, query in raft_p99.items()
    ]
    return {
        "E1": f"sum(increase({metric(pool, 'tikv_critical_error_total')}[{duration}]))",
        "E2": f"sum(increase({metric(pool, 'tikv_channel_full_total')}[{duration}]))",
        "E3": f"sum(increase({metric(pool, 'tikv_scheduler_too_busy_total')}[{duration}]))",
        "E4": f"sum(increase({metric(pool, 'tikv_coprocessor_request_error', type='full')}[{duration}]))",
        "E5": f"sum(increase({metric(pool, 'tikv_server_report_failure_msg_total')}[{duration}]))",
        "G1": f"{g1_max} or {g1_avg}",
        "G4": f"sum by (type) (increase({metric(pool, 'tikv_grpc_msg_duration_seconds_sum', type=grpc_filter)}[{duration}])) / sum by (type) (increase({metric(pool, 'tikv_grpc_msg_duration_seconds_count', type=grpc_filter)}[{duration}]))",
        "G5": f"max_over_time(({histogram_p99(pool, 'tikv_grpc_msg_duration_seconds', type=('=~', READ_GRPC_TYPES))})[{duration}:5m])",
        "B4": f"sum(changes({metric(pool, 'process_start_time_seconds')}[{duration}]))",
        "T2": f"max_over_time((sum by (instance) ({metric(pool, 'tikv_unified_read_pool_running_tasks')}))[{duration}:5m])",
        "T3": f"sum by (instance)(increase({metric(pool, 'multilevel_level_elapsed', name='unified-read-pool', level='total')}[{duration}])) / 1000000",
        "C2": f"sum by (instance,type) ({metric(pool, 'tikv_raftstore_region_count', type=('=~', 'region|leader'))})",
        "D1": "max(" + " or ".join(raft_window_p99) + ")",
        "D2": f"sum(increase({metric(pool, 'raft_engine_wal_write_throttle_duration_seconds_sum')}[{duration}]))",
        "D3": f"max_over_time(({metric(pool, 'raft_engine_pending_compaction_wals', container='tikv')}[{duration}:5m]))",
        "D4": peak_tikv_storage_disk_query(
            pool, f"rate({node_disk_metric('node_disk_io_time_seconds_total')}[5m])", window
        ),
        "D5": peak_tikv_storage_disk_query(
            pool, f"rate({node_disk_metric('node_disk_io_time_weighted_seconds_total')}[5m])", window
        ),
        "D6": peak_tikv_storage_disk_query(
            pool,
            node_disk_latency_expression(
                "node_disk_read_time_seconds_total", "node_disk_reads_completed_total"
            ),
            window,
        ),
        "D7": peak_tikv_storage_disk_query(
            pool,
            node_disk_latency_expression(
                "node_disk_write_time_seconds_total", "node_disk_writes_completed_total"
            ),
            window,
        ),
        "F1": f"max_over_time(({histogram_p99(pool, 'tikv_scheduler_throttle_duration_seconds')})[{duration}:5m])",
        "F2": f"max by (instance) (max_over_time(({metric(pool, 'tikv_scheduler_discard_ratio')})[{duration}:5m]))",
        "P1": f"max_over_time(({metric(pool, 'store_slow_score')}[{duration}:5m]))",
        "R1": f"max_over_time((sum(rate({metric(pool, 'tikv_raftstore_log_lag_sum')}[5m])))[{duration}:5m])",
        "K1": f"max_over_time(({metric(pool, 'kv_engine_pending_compaction_shards')}[{duration}:5m]))",
        "K2": f"sum by (container,result) (increase({metric(pool, 'kv_engine_compaction_result_counter', container=('=~', 'tikv|tikv-worker'), result=('=~', 'failure|local_compaction_disabled'))}[{duration}]))",
        "K3": f"sum by (container,result) (increase({metric(pool, 'kv_engine_compaction_result_counter', container=('=~', 'tikv|tikv-worker'))}[{duration}]))",
    }


def build_range_rule_queries(pool: PoolIdentity) -> dict[str, str]:
    return {
        "G2": f"sum by (type) (increase({metric(pool, 'tikv_grpc_msg_fail_total')}[30m]))",
        "G3": histogram_p99(pool, "tikv_grpc_msg_duration_seconds", type=("=~", "kv_prewrite|kv_commit")),
        "B1": cpu_percentage_query(pool),
        "B2": memory_percentage_query(pool),
        "C1": store_usage_percentage_query(pool),
    }


def build_b3_queries(pool: PoolIdentity, window: Window) -> dict[str, str]:
    memory_query = memory_percentage_query(pool)
    cpu_query = cpu_percentage_query(pool)
    region_query = (
        f"sum by (instance) ({metric(pool, 'tikv_raftstore_region_count', type='region')})"
    )
    grpc_qps_query = (
        f"sum by (instance) (rate({metric(pool, 'tikv_grpc_msg_duration_seconds_count', type=('=~', '|'.join(GRPC_TYPES)))}[5m]))"
    )

    def average(query: str, duration: str) -> str:
        return f"avg_over_time(({query})[{duration}:5m])"

    def baseline(query: str) -> str:
        return f"avg_over_time(({query})[{MEMORY_LEAK_BASELINE_WINDOW}:5m] offset {window.prom_window})"

    return {
        "memory_current": average(memory_query, window.prom_window),
        "memory_baseline": baseline(memory_query),
        "region_current": average(region_query, window.prom_window),
        "region_baseline": baseline(region_query),
        "qps_current": average(grpc_qps_query, window.prom_window),
        "qps_baseline": baseline(grpc_qps_query),
        "cpu_current": average(cpu_query, window.prom_window),
        "cpu_baseline": baseline(cpu_query),
    }


def build_thread_group_queries(pool: PoolIdentity) -> dict[str, str]:
    return {
        group.name: (
            f"sum by (instance) (rate({metric(pool, 'tikv_thread_cpu_seconds_total', name=('=~', group.thread_name_regex))}[5m]))"
        )
        for group in T1_THREAD_CPU_GROUPS
    }


def build_loki_filter(pool: PoolIdentity) -> str:
    return (
        f"{pool.loki_selector()} |~ `(?i)ERROR|FATAL|WARN` "
        "!~ `TLS handshake|local file gc kv files|monotonic time jumped|wait connect timeout|batch_raft send response fail` "
        '!= "KV:Raft:StepPeerNotFound" '
        '!= "no valid key found for split." '
        '!= "cdc register for a not found region" '
        '!= "cdc initialize fail" '
        '!= "not_leader {" '
        '!= "stale command" '
        '!= "region_not_found {" '
        '!= "epoch_not_match {" '
        '!= "oneshot canceled" '
        '!= "scan canceled"'
    )


def as_mapping(value: object) -> Mapping[str, object]:
    return value if isinstance(value, dict) else {}


def as_sequence(value: object) -> Sequence[object]:
    return value if isinstance(value, list) else ()


def response_error(response: Mapping[str, object]) -> str | None:
    error = response.get("error")
    if error:
        return str(error)
    status = response.get("status")
    if status and status != "success":
        return f"status={status}"
    return None


def finite_float(value: object) -> float | None:
    if not isinstance(value, (str, int, float)):
        return None
    try:
        number = float(value)
    except (TypeError, ValueError):
        return None
    return number if math.isfinite(number) else None


def result_from_prometheus(response: Mapping[str, object], rule_id: str) -> RuleResult:
    error = response_error(response)
    if error:
        return RuleResult("Unknown", None, error)
    payload = as_mapping(response.get("data"))
    raw_results = payload.get("result")
    if not isinstance(raw_results, list):
        return RuleResult("Unknown", None, "malformed Prometheus response: data.result must be a list")
    if not raw_results:
        return RuleResult("Known", 0.0, "successful empty response treated as zero")
    values: list[float] = []
    for raw_item in raw_results:
        if not isinstance(raw_item, dict):
            return RuleResult("Unknown", None, "malformed Prometheus series")
        item = raw_item
        raw_pair = as_sequence(item.get("value"))
        raw_value = raw_pair[1] if len(raw_pair) > 1 else None
        value = finite_float(raw_value)
        if value is None:
            return RuleResult("Unknown", None, "malformed Prometheus series value")
        values.append(value)
    if not values:
        return RuleResult("Unknown", None, "malformed Prometheus response: no values")
    return RuleResult("Known", max(values), RULE_BY_ID[rule_id].purpose)


def labeled_vector(response: Mapping[str, object]) -> tuple[str | None, list[tuple[dict[str, str], float]]]:
    error = response_error(response)
    if error:
        return error, []
    raw_results = as_mapping(response.get("data")).get("result")
    if not isinstance(raw_results, list):
        return "malformed Prometheus response: data.result must be a list", []
    if not raw_results:
        return None, []
    entries: list[tuple[dict[str, str], float]] = []
    for raw_item in raw_results:
        if not isinstance(raw_item, dict):
            return "malformed Prometheus series", []
        item = raw_item
        raw_labels = item.get("metric")
        if not isinstance(raw_labels, dict):
            return "malformed Prometheus series labels", []
        raw_pair = as_sequence(item.get("value"))
        raw_value = raw_pair[1] if len(raw_pair) > 1 else None
        value = finite_float(raw_value)
        if value is None:
            return "malformed Prometheus series value", []
        entries.append(({key: str(label) for key, label in raw_labels.items()}, value))
    return None, entries


def g1_result(response: Mapping[str, object]) -> RuleResult:
    error, entries = labeled_vector(response)
    if error:
        return RuleResult("Unknown", None, error)
    rates: dict[tuple[str, str], float] = {
        (aggregation, request_type): 0.0
        for aggregation in ("max", "avg")
        for request_type in G1_GRPC_TYPES
    }
    for labels, value in entries:
        aggregation = labels.get("aggregation")
        request_type = labels.get("type")
        if aggregation not in {"max", "avg"} or request_type not in G1_GRPC_TYPES:
            return RuleResult("Unknown", None, "malformed G1 series labels")
        rates[(aggregation, request_type)] = value
    evidence = tuple(
        G1Rate(aggregation, request_type, rates[(aggregation, request_type)])
        for aggregation in ("max", "avg")
        for request_type in G1_GRPC_TYPES
    )
    detail = "; ".join(
        f"{rate.aggregation} {rate.request_type}: {rate.messages_per_second:.2f} messages/s"
        for rate in evidence
    )
    if not entries:
        detail = f"successful empty response treated as zero; {detail}"
    return RuleResult(
        "Known",
        max((rate.messages_per_second for rate in evidence), default=0.0),
        detail,
        g1_rates=evidence,
    )


def g4_result(response: Mapping[str, object]) -> RuleResult:
    error, entries = labeled_vector(response)
    if error:
        return RuleResult("Unknown", None, error)
    averages: dict[str, float] = {}
    for labels, value in entries:
        request_type = labels.get("type", "unknown")
        averages[request_type] = value
    if not averages:
        return RuleResult("Known", 0.0, "successful empty response treated as zero")
    detail = "; ".join(f"{request_type}: {value:.6f} s" for request_type, value in sorted(averages.items()))
    return RuleResult("Known", max(averages.values()), detail)


def c2_result(response: Mapping[str, object]) -> RuleResult:
    error, entries = labeled_vector(response)
    if error:
        return RuleResult("Unknown", None, error)
    evidence = sorted(
        (labels.get("instance", "unknown"), labels.get("type", "unknown"), value)
        for labels, value in entries
    )
    if not evidence:
        return RuleResult("Known", 0.0, "successful empty response treated as zero")
    detail = "; ".join(f"{instance} {kind}={value:.0f}" for instance, kind, value in evidence)
    return RuleResult("Known", max(value for _, _, value in evidence), detail)


def compaction_result_by_container_and_result(response: Mapping[str, object]) -> RuleResult:
    error, entries = labeled_vector(response)
    if error:
        return RuleResult("Unknown", None, error)
    if not entries:
        return RuleResult("Known", 0.0, "successful empty response treated as zero")
    totals: dict[tuple[str, str], float] = {}
    for labels, value in entries:
        container = labels.get("container")
        result = labels.get("result")
        if not container or not result:
            return RuleResult("Unknown", None, "compaction result series is missing container or result label")
        key = (container, result)
        totals[key] = totals.get(key, 0.0) + value
    detail = "; ".join(
        f"container={container}, result={result}: {value:.0f}"
        for (container, result), value in sorted(totals.items())
    )
    return RuleResult("Known", sum(totals.values()), detail)


def normalize_loki_log_line(line: str) -> str:
    """Remove a leading datetime, unescape quotes, and normalize whitespace for log grouping."""
    without_datetime = LEADING_LOG_DATETIME_RE.sub("", line, count=1)
    return " ".join(without_datetime.split()).replace('\\"', '"')


def top_loki_log_groups(
    response: Mapping[str, object],
) -> tuple[str | None, int, tuple[tuple[int, str], ...], bool]:
    payload = response.get("data")
    if not isinstance(payload, dict):
        return "malformed Loki response: data must be an object", 0, (), False
    if payload.get("resultType") != "streams":
        return "malformed Loki response: data.resultType must be 'streams'", 0, (), False
    raw_results = payload.get("result")
    if not isinstance(raw_results, list):
        return "malformed Loki response: data.result must be a list", 0, (), False
    groups: Counter[tuple[str, str]] = Counter()
    count = 0
    raw_entry_count = 0
    for raw_stream in raw_results:
        if not isinstance(raw_stream, dict):
            return "malformed Loki response: result stream must be an object", 0, (), False
        if not isinstance(raw_stream.get("stream"), dict):
            return "malformed Loki response: stream labels must be an object", 0, (), False
        if not all(
            isinstance(key, str) and isinstance(value, str)
            for key, value in raw_stream["stream"].items()
        ):
            return "malformed Loki response: stream labels must be strings", 0, (), False
        raw_values = raw_stream.get("values")
        if not isinstance(raw_values, list):
            return "malformed Loki response: stream values must be a list", 0, (), False
        if not raw_values:
            return "malformed Loki response: non-empty stream has no values", 0, (), False
        for raw_value in raw_values:
            if not isinstance(raw_value, list) or len(raw_value) != 2:
                return "malformed Loki response: log value must contain timestamp and line", 0, (), False
            timestamp, line = raw_value[0], raw_value[1]
            if not isinstance(timestamp, str) or not isinstance(line, str):
                return "malformed Loki response: log timestamp and line have invalid types", 0, (), False
            raw_entry_count += 1
            normalized = normalize_loki_log_line(line)
            if not normalized:
                continue
            severity_match = LOG_SEVERITY_RE.search(normalized)
            severity = severity_match.group(1).upper() if severity_match else "ACTIONABLE"
            groups[(severity, normalized)] += 1
            count += 1
    top_groups = tuple(
        (count, message)
        for (_severity, message), count in sorted(
            groups.items(), key=lambda item: (-item[1], item[0][1])
        )[:MAX_LOG_GROUPS]
    )
    return None, count, top_groups, raw_entry_count >= LOKI_LOG_QUERY_LIMIT


def result_from_loki(response: Mapping[str, object]) -> RuleResult:
    error = response_error(response)
    if error:
        return RuleResult("Unknown", None, error)
    parse_error, count, groups, truncated = top_loki_log_groups(response)
    if parse_error:
        return RuleResult("Unknown", None, parse_error)
    detail = (
        f"at least {count} actionable ERROR/FATAL/WARN log lines; "
        + f"log query reached the {LOKI_LOG_QUERY_LIMIT}-entry limit, so counts and top normalized groups are incomplete"
        if truncated
        else "actionable ERROR/FATAL/WARN log lines"
    )
    if groups:
        detail += "; top normalized groups: " + "; ".join(
            f"{'>= ' if truncated else ''}{group_count} × {message}" for group_count, message in groups
        )
    return RuleResult("Known", float(count), detail, log_groups=groups, truncated=truncated)


@dataclass(frozen=True)
class MatrixSeries:
    labels: Mapping[str, str]
    points: tuple[tuple[float, float], ...]


def matrix_from_prometheus(response: Mapping[str, object]) -> tuple[str | None, list[MatrixSeries]]:
    error = response_error(response)
    if error:
        return error, []
    payload = as_mapping(response.get("data"))
    raw_results = payload.get("result")
    if not isinstance(raw_results, list):
        return "malformed Prometheus response: data.result must be a list", []
    if not raw_results:
        return None, []
    series: list[MatrixSeries] = []
    for raw_item in raw_results:
        if not isinstance(raw_item, dict):
            return "malformed Prometheus matrix series", []
        item = raw_item
        raw_labels = item.get("metric")
        raw_points = item.get("values")
        if not isinstance(raw_labels, dict) or not isinstance(raw_points, list):
            return "malformed Prometheus matrix series", []
        labels = {key: str(value) for key, value in raw_labels.items()}
        points: list[tuple[float, float]] = []
        for raw_point in raw_points:
            point = as_sequence(raw_point)
            if len(point) < 2 or not isinstance(point[0], (str, int, float)) or not isinstance(point[1], (str, int, float)):
                return "malformed Prometheus matrix point", []
            timestamp = finite_float(point[0])
            value = finite_float(point[1])
            if timestamp is None or value is None:
                return "malformed Prometheus matrix point", []
            points.append((timestamp, value))
        if not points:
            return "malformed Prometheus matrix series: no values", []
        series.append(MatrixSeries(labels, tuple(sorted(points))))
    if not series:
        return "malformed Prometheus matrix response: no series", []
    return None, series


def range_result(response: Mapping[str, object], rule_id: str) -> tuple[RuleResult, list[MatrixSeries]]:
    error, series = matrix_from_prometheus(response)
    if error:
        return RuleResult("Unknown", None, error), []
    if not series:
        return RuleResult("Known", 0.0, "successful empty response treated as zero"), []
    peak = max(value for item in series for _, value in item.points)
    return RuleResult("Known", peak, RULE_BY_ID[rule_id].purpose), series


def has_sustained_threshold(
    points: Sequence[tuple[float, float]], threshold: float, duration_seconds: int
) -> bool:
    required_points = (
        SUSTAINED_15M_POINTS if duration_seconds == 15 * 60
        else SUSTAINED_30M_POINTS if duration_seconds == 30 * 60
        else max(1, math.ceil(duration_seconds / RANGE_STEP_SECONDS))
    )
    run_points = 0
    previous_timestamp: float | None = None
    for timestamp, value in points:
        contiguous = previous_timestamp is None or timestamp - previous_timestamp <= RANGE_STEP_SECONDS * 1.5
        if value >= threshold and contiguous:
            run_points += 1
        elif value >= threshold:
            run_points = 1
        else:
            run_points = 0
        if run_points >= required_points:
            return True
        previous_timestamp = timestamp
    return False


def sustained_result(
    response: Mapping[str, object], rule_id: str, thresholds: Sequence[float], duration_seconds: int
) -> RuleResult:
    result, series = range_result(response, rule_id)
    if result.unknown:
        return result
    sustained = frozenset(
        threshold
        for threshold in thresholds
        if any(has_sustained_threshold(item.points, threshold, duration_seconds) for item in series)
    )
    if sustained:
        detail = f"sustained for at least {duration_seconds // 60} minutes at >= {', '.join(str(value) for value in sorted(sustained))}"
    else:
        detail = f"no threshold sustained for {duration_seconds // 60} minutes"
    return RuleResult("Known", result.value, detail, sustained)


def c1_growth_result(response: Mapping[str, object]) -> RuleResult:
    result, series = range_result(response, "C1")
    if result.unknown:
        return result
    events: list[str] = []
    for item in series:
        for index, (end_timestamp, end_value) in enumerate(item.points):
            candidate_points = [
                (start_timestamp, start_value)
                for start_timestamp, start_value in item.points[:index]
                if 0 < end_timestamp - start_timestamp <= 30 * 60
            ]
            if not candidate_points:
                continue
            start_timestamp, start_value = min(candidate_points, key=lambda point: (point[1], point[0]))
            if end_value >= 80 and end_value - start_value >= 10:
                elapsed_minutes = (end_timestamp - start_timestamp) / 60
                events.append(
                    f"{item.labels.get('instance', 'unknown')} rose {start_value:.2f}% -> {end_value:.2f}% "
                    f"({end_value - start_value:.2f} pp) over {elapsed_minutes:.0f}m"
                )
    if events:
        return RuleResult("Known", result.value, "; ".join(events[:3]), frozenset({80.0}))
    return RuleResult("Known", result.value, "no >=80% usage with >=10 percentage-point 30-minute rise")


def vector_stats_by_instance(response: Mapping[str, object]) -> tuple[str | None, dict[str, float]]:
    error, entries = labeled_vector(response)
    if error:
        return error, {}
    if not entries:
        return None, {}
    if any(not labels.get("instance") for labels, _value in entries):
        return "Prometheus series is missing instance label", {}
    values = {
        labels.get("instance", "unknown"): value
        for labels, value in entries
    }
    return None, values


def ratio_delta_pct(current: float | None, baseline: float | None) -> float | None:
    if current is None or baseline is None or baseline <= 0:
        return None
    return ((current - baseline) / baseline) * 100.0


def b3_result(responses: Mapping[str, Mapping[str, object]]) -> RuleResult:
    stats: dict[str, dict[str, float]] = {}
    errors: list[str] = []
    for name, response in responses.items():
        error, values = vector_stats_by_instance(response)
        if error:
            errors.append(f"{name}: {error}")
        else:
            stats[name] = values
    if errors:
        return RuleResult("Unknown", None, "B3 comparison unavailable: " + "; ".join(errors))

    if not stats["memory_current"] or not stats["memory_baseline"]:
        return RuleResult("Known", 0.0, "successful empty response treated as zero; no memory-growth observation")

    comparable_instances = sorted(
        set(stats["memory_current"]) & set(stats["memory_baseline"])
    )
    if not comparable_instances:
        return RuleResult("Unknown", None, "B3 comparison unavailable: no matching TiKV instances")

    risers: list[tuple[str, float, list[str]]] = []
    warning_nodes: list[tuple[str, float]] = []
    for instance in comparable_instances:
        current = stats["memory_current"][instance]
        baseline = stats["memory_baseline"][instance]
        delta = current - baseline
        if delta < MEMORY_LEAK_OBSERVE_DELTA_PCT:
            continue
        support_signals: list[str] = []

        region_current = stats.get("region_current", {}).get(instance)
        region_baseline = stats.get("region_baseline", {}).get(instance)
        region_delta = None if region_current is None or region_baseline is None else region_current - region_baseline
        region_ratio = ratio_delta_pct(region_current, region_baseline)
        if (
            region_delta is not None
            and region_ratio is not None
            and region_delta >= MEMORY_LEAK_EXPLAIN_REGION_DELTA_ABS
            and region_ratio >= MEMORY_LEAK_EXPLAIN_REGION_DELTA_RATIO_PCT
        ):
            support_signals.append(f"regions +{region_delta:.0f} ({region_ratio:.1f}%)")

        qps_current = stats.get("qps_current", {}).get(instance)
        qps_baseline = stats.get("qps_baseline", {}).get(instance)
        qps_delta = None if qps_current is None or qps_baseline is None else qps_current - qps_baseline
        qps_ratio = ratio_delta_pct(qps_current, qps_baseline)
        if (
            qps_delta is not None
            and qps_ratio is not None
            and qps_delta >= MEMORY_LEAK_EXPLAIN_QPS_DELTA_ABS
            and qps_ratio >= MEMORY_LEAK_EXPLAIN_QPS_DELTA_RATIO_PCT
        ):
            support_signals.append(f"QPS +{qps_delta:.1f} ({qps_ratio:.1f}%)")

        cpu_current = stats.get("cpu_current", {}).get(instance)
        cpu_baseline = stats.get("cpu_baseline", {}).get(instance)
        cpu_delta = None if cpu_current is None or cpu_baseline is None else cpu_current - cpu_baseline
        if cpu_delta is not None and cpu_delta >= MEMORY_LEAK_EXPLAIN_CPU_DELTA_PCT:
            support_signals.append(f"CPU +{cpu_delta:.1f} pp")

        risers.append((instance, delta, support_signals))
        if delta >= MEMORY_LEAK_WARN_DELTA_PCT and not support_signals:
            warning_nodes.append((instance, delta))

    risers.sort(key=lambda item: item[1], reverse=True)
    warning_nodes.sort(key=lambda item: item[1], reverse=True)
    peak_delta = max((delta for _instance, delta, _support in risers), default=0.0)
    if warning_nodes:
        detail = "; ".join(
            f"{instance} {stats['memory_current'][instance]:.2f}% current vs "
            f"{stats['memory_baseline'][instance]:.2f}% prior-7d average "
            f"(delta {delta:.2f} pp) without matching Region/QPS/CPU growth"
            for instance, delta in warning_nodes[:3]
        )
        return RuleResult("Known", peak_delta, detail, warning=True)
    if risers:
        detail = "; ".join(
            f"{instance} {stats['memory_current'][instance]:.2f}% current vs "
            f"{stats['memory_baseline'][instance]:.2f}% prior-7d average "
            f"(delta +{delta:.2f} pp) "
            + (f"explained by {', '.join(support)}" if support else "observation only below the 5 pp warning line")
            for instance, delta, support in risers[:3]
        )
        return RuleResult("Known", peak_delta, detail)
    return RuleResult("Known", 0.0, "no instance exceeded the 3 percentage-point rising-memory observation line")


def thread_group_result(responses: Mapping[str, Mapping[str, object]]) -> RuleResult:
    evidence: list[str] = []
    rates: list[T1CpuRate] = []
    unknown_groups: list[str] = []
    peak: float | None = None
    for group in T1_THREAD_CPU_GROUPS:
        response = responses.get(group.name)
        if response is None:
            error = "missing T1 group response"
            rates.append(T1CpuRate(group.name, None, None, None, error))
            unknown_groups.append(f"{group.name}: {error}")
            continue
        error, series = matrix_from_prometheus(response)
        if error:
            rates.append(T1CpuRate(group.name, None, None, None, error))
            unknown_groups.append(f"{group.name}: {error}")
            continue
        per_instance_points: dict[str, dict[float, float]] = {}
        missing_instance = False
        for item in series:
            instance = item.labels.get("instance")
            if not instance:
                missing_instance = True
                break
            timestamps = per_instance_points.setdefault(instance, {})
            for timestamp, value in item.points:
                timestamps[timestamp] = timestamps.get(timestamp, 0.0) + value
        if missing_instance:
            error = "thread CPU series is missing instance label"
            rates.append(T1CpuRate(group.name, None, None, None, error))
            unknown_groups.append(f"{group.name}: {error}")
            continue
        if not per_instance_points:
            rates.append(T1CpuRate(group.name, None, 0.0, 0.0))
            evidence.append(f"{group.name}: no observed instances; max: 0.00 cores, avg: 0.00 cores")
            peak = 0.0 if peak is None else max(peak, 0.0)
            continue
        group_rates: list[T1CpuRate] = []
        for instance, timestamps in sorted(per_instance_points.items()):
            values = list(timestamps.values())
            maximum = max(values)
            average = sum(values) / len(values)
            group_rates.append(T1CpuRate(group.name, instance, maximum, average))
            peak = maximum if peak is None else max(peak, maximum)
        rates.extend(group_rates)
        group_maximum = max(cast(float, rate.max_cores) for rate in group_rates)
        group_average = sum(cast(float, rate.avg_cores) for rate in group_rates) / len(group_rates)
        evidence.append(f"{group.name}: max: {group_maximum:.2f} cores, avg: {group_average:.2f} cores")
    if unknown_groups:
        return RuleResult(
            "Unknown",
            peak,
            "; ".join(evidence + unknown_groups),
            t1_cpu_rates=tuple(rates),
        )
    if peak is None:
        return RuleResult(
            "Known",
            0.0,
            "successful empty response treated as zero",
            t1_cpu_rates=tuple(rates),
        )
    return RuleResult("Known", peak, "; ".join(evidence), t1_cpu_rates=tuple(rates))


def validate_clinic_endpoint(value: str) -> str:
    candidate = value[:-1] if value.endswith("/") else value
    normalized = candidate.lower()
    if normalized not in ALLOWED_CLINIC_ENDPOINTS:
        raise ValueError("CLINIC_ENDPOINT must be an approved HTTPS Clinic endpoint without credentials, query, fragment, or path")
    return normalized


class ServerlessInspectionClient:
    def __init__(self, env_file: str) -> None:
        load_shell_env(env_file)
        environment = os.environ.get("CLINIC_ENV", "prod")
        endpoints = {
            "dev": "https://dev-clinic.pingcap.com",
            "staging": "https://staging-clinic.pingcap.com",
            "prod": "https://clinic.pingcap.com",
        }
        self.endpoint = validate_clinic_endpoint(os.environ.get("CLINIC_ENDPOINT", endpoints.get(environment, endpoints["prod"])))
        self.api_key = os.environ.get("CLINIC_API_KEY", "")
        if not self.api_key:
            raise RuntimeError("CLINIC_API_KEY not set")

    def _get_json(self, path: str, params: Mapping[str, str | int]) -> dict[str, object]:
        requests_api = _load_requests()
        try:
            with requests_api.get(
                f"{self.endpoint.rstrip('/')}{path}",
                params=params,
                headers={"Authorization": f"Bearer {self.api_key}"},
                timeout=120,
                allow_redirects=False,
                stream=True,
            ) as response:
                if response.status_code >= 300:
                    return {"error": f"Clinic HTTP status {response.status_code}"}
                body = bytearray()
                for chunk in response.iter_content(chunk_size=64 * 1024):
                    if not chunk:
                        continue
                    body.extend(chunk)
                    if len(body) > MAX_HTTP_RESPONSE_BYTES:
                        return {"error": "Clinic response exceeds size limit"}
            payload = json.loads(bytes(body))
        except requests_api.request_exception as error:
            return {"error": str(error)}
        except (OSError, json.JSONDecodeError) as error:
            return {"error": str(error)}
        return payload if isinstance(payload, dict) else {"error": "unexpected non-object response"}

    def metric_query(self, pool: PoolIdentity, query: str, end: datetime) -> dict[str, object]:
        return self._get_json(
            "/data-proxy/serverless-parent-metrics/api/v1/query",
            {"control_plane_info": pool.control_plane_info, "query": query, "time": iso_utc(end)},
        )

    def metric_query_range(self, pool: PoolIdentity, query: str, window: Window) -> dict[str, object]:
        return self._get_json(
            "/data-proxy/serverless-parent-metrics/api/v1/query_range",
            {
                "control_plane_info": pool.control_plane_info,
                "query": query,
                "start": iso_utc(window.start),
                "end": iso_utc(window.end),
                "step": RANGE_STEP_SECONDS,
            },
        )

    def log_query(self, pool: PoolIdentity, query: str, window: Window) -> dict[str, object]:
        return self._get_json(
            "/data-proxy/serverless-parent-logs/api/v1/query_range",
            {
                "control_plane_info": pool.control_plane_info,
                "query": query,
                "start": int(window.start.timestamp() * 1_000_000_000),
                "end": int(window.end.timestamp() * 1_000_000_000),
                "limit": LOKI_LOG_QUERY_LIMIT,
                "direction": "backward",
            },
        )


def unavailable_results() -> dict[str, RuleResult]:
    return {
        rule.rule_id: RuleResult(
            "Unknown",
            None,
            UNAVAILABLE_REASONS.get(rule.rule_id, "unavailable for shared Serverless pool scope"),
        )
        for rule in RULE_SPECS
        if rule.status == "unavailable"
    }


@dataclass
class InspectionReport:
    pool: PoolIdentity
    window: Window
    results: dict[str, RuleResult]
    markdown: str = ""


def identity_validation_query(pool: PoolIdentity) -> str:
    return f"count({metric(pool, 'tikv_engine_size_bytes')})"


def validate_pool_identity(client: ServerlessInspectionClient, pool: PoolIdentity, end: datetime) -> None:
    result = result_from_prometheus(client.metric_query(pool, identity_validation_query(pool), end), "C1")
    if result.unknown or result.value is None or result.value <= 0:
        raise ValueError(
            "pool identity validation failed: no positive tikv_engine_size_bytes series for the supplied "
            "control_plane_info/k8s_cluster_info pair"
        )


def inspect_pool(client: ServerlessInspectionClient, pool: PoolIdentity, window: Window) -> InspectionReport:
    results = unavailable_results()
    validate_pool_identity(client, pool, window.end)
    for rule_id, query in build_rule_queries(pool, window).items():
        response = client.metric_query(pool, query, window.end)
        results[rule_id] = (
            g1_result(response)
            if rule_id == "G1"
            else g4_result(response)
            if rule_id == "G4"
            else c2_result(response)
            if rule_id == "C2"
            else compaction_result_by_container_and_result(response)
            if rule_id in {"K2", "K3"}
            else result_from_prometheus(response, rule_id)
        )
    range_responses = {
        rule_id: client.metric_query_range(pool, query, window)
        for rule_id, query in build_range_rule_queries(pool).items()
    }
    results["G2"], _ = range_result(range_responses["G2"], "G2")
    results["G3"] = sustained_result(range_responses["G3"], "G3", (1.0,), 15 * 60)
    results["B1"] = sustained_result(range_responses["B1"], "B1", (85.0,), 30 * 60)
    results["B2"] = sustained_result(range_responses["B2"], "B2", (85.0, 90.0), 30 * 60)
    results["C1"] = c1_growth_result(range_responses["C1"])
    b3_queries = build_b3_queries(pool, window)
    results["B3"] = b3_result(
        {
            name: client.metric_query(pool, query, window.end)
            for name, query in b3_queries.items()
        }
    )
    thread_responses = {
        group: client.metric_query_range(pool, query, window)
        for group, query in build_thread_group_queries(pool).items()
    }
    results["T1"] = thread_group_result(thread_responses)
    results["L1"] = result_from_loki(client.log_query(pool, build_loki_filter(pool), window),)
    report = InspectionReport(pool, window, results)
    report.markdown = build_markdown(report)
    return report


def format_result(result: RuleResult) -> str:
    if result.unknown:
        return f"Unknown ({markdown_safe(result.detail)})"
    if result.value is None:
        return "Unknown"
    if result.truncated:
        return f">= {result.value:.2f} (incomplete)"
    return f"{result.value:.2f}"


def format_milliseconds(result: RuleResult) -> str:
    if result.unknown or result.value is None:
        return format_result(result)
    return f"{result.value * 1000:.2f} ms"


def format_g1_rates(result: RuleResult, aggregation: str, request_types: Sequence[str]) -> str:
    if result.unknown:
        return format_result(result)
    rates = {
        (rate.aggregation, rate.request_type): rate.messages_per_second
        for rate in result.g1_rates
    }
    return ", ".join(
        f"{request_type}: {rates.get((aggregation, request_type), 0.0):.2f} messages/s"
        for request_type in request_types
    )


def format_t1_cpu_rate(result: RuleResult, group: ThreadCpuGroupSpec) -> str:
    rates = [rate for rate in result.t1_cpu_rates if rate.group == group.name]
    if not rates:
        return "Unknown: missing T1 group result"
    errors = [rate.error for rate in rates if rate.error]
    if errors:
        return f"Unknown: {errors[0]}"
    observed = [rate for rate in rates if rate.instance is not None]
    if not observed:
        return "No observed instances; max: 0.00 cores, avg: 0.00 cores"
    if any(rate.max_cores is None or rate.avg_cores is None for rate in observed):
        return "Unknown: missing CPU rate statistics"
    maximum = max(cast(float, rate.max_cores) for rate in observed)
    average = sum(cast(float, rate.avg_cores) for rate in observed) / len(observed)
    return f"max: {maximum:.2f} cores, avg: {average:.2f} cores"


def t1_group_visible(group: ThreadCpuGroupSpec, reports: Sequence[InspectionReport]) -> bool:
    """Keep visible evidence for T1 errors and groups reaching the report line."""
    for report in reports:
        rates = [rate for rate in report.results["T1"].t1_cpu_rates if rate.group == group.name]
        if not rates or any(rate.error for rate in rates):
            return True
        observed = [rate for rate in rates if rate.instance is not None]
        if any(rate.max_cores is None or rate.avg_cores is None for rate in rates):
            return True
        maximum = max((cast(float, rate.max_cores) for rate in observed), default=0.0)
        if maximum >= T1_REPORT_MIN_MAX_CORES:
            return True
    return False


def visible_t1_groups(reports: Sequence[InspectionReport]) -> tuple[ThreadCpuGroupSpec, ...]:
    return tuple(group for group in T1_THREAD_CPU_GROUPS if t1_group_visible(group, reports))


def known_above(results: Mapping[str, RuleResult], rule_id: str, threshold: float) -> bool:
    result = results[rule_id]
    return result.status == "Known" and result.value is not None and result.value > threshold


def known_at_or_above(results: Mapping[str, RuleResult], rule_id: str, threshold: float) -> bool:
    result = results[rule_id]
    return result.status == "Known" and result.value is not None and result.value >= threshold


def sustained_at_or_above(results: Mapping[str, RuleResult], rule_id: str, threshold: float) -> bool:
    return threshold in results[rule_id].sustained_at


def unexpected_unknown_rules(results: Mapping[str, RuleResult]) -> list[str]:
    return [
        rule_id
        for rule_id, result in results.items()
        if result.unknown and RULE_BY_ID[rule_id].status != "unavailable"
    ]


def report_verdict(results: Mapping[str, RuleResult]) -> tuple[str, list[str]]:
    findings: list[str] = []
    if known_above(results, "E1", 0):
        findings.append("ERROR: critical TiKV errors were observed pool-wide.")
    for rule_id, label in (("E2", "channel full"), ("E3", "scheduler busy"), ("E4", "coprocessor full"), ("E5", "server report failures")):
        if known_above(results, rule_id, 0):
            findings.append(f"WARNING: {label} events were observed pool-wide.")
    if known_at_or_above(results, "G2", 100):
        findings.append("WARNING: gRPC failed messages reached the 30-minute watch line.")
    if sustained_at_or_above(results, "G3", 1):
        findings.append("WARNING: write-path gRPC P99 stayed at or above 1000 ms for at least 15 minutes.")
    if sustained_at_or_above(results, "B1", 85):
        findings.append("WARNING: TiKV CPU stayed at or above 85% for at least 30 minutes.")
    if sustained_at_or_above(results, "B2", 90):
        findings.append("ERROR: TiKV memory stayed at or above 90% for at least 30 minutes.")
    elif sustained_at_or_above(results, "B2", 85):
        findings.append("WARNING: TiKV memory stayed at or above 85% for at least 30 minutes.")
    if results["B3"].warning:
        findings.append("WARNING: memory growth exceeded the seven-day baseline without matching Region/QPS/CPU growth.")
    if sustained_at_or_above(results, "C1", 80):
        findings.append("WARNING: store usage reached 80% and rose by at least 10 percentage points within 30 minutes.")
    if known_above(results, "D2", 0):
        findings.append("WARNING: Raft Engine WAL write throttling was observed.")
    if known_above(results, "D6", 0.020):
        findings.append("WARNING: TiKV data-device read latency exceeded 20 ms.")
    if known_above(results, "D7", 0.020):
        findings.append("WARNING: TiKV data-device write latency exceeded 20 ms.")
    if known_at_or_above(results, "P1", 100):
        findings.append("WARNING: store slow score reached the high-risk line.")
    if known_above(results, "F2", 0):
        findings.append("WARNING: flow-control discard ratio became positive.")
    if known_at_or_above(results, "D3", 2):
        findings.append("WARNING: Raft Engine pending compaction WALs reached 2 or more.")
    if known_above(results, "K2", 0):
        findings.append("WARNING: compaction failure or local_compaction_disabled results were observed.")
    if known_above(results, "L1", 0):
        findings.append("WARNING: actionable ERROR/FATAL/WARN TiKV log lines were observed.")
    if results["L1"].truncated:
        findings.append(
            f"WARNING: actionable TiKV log query reached the {LOKI_LOG_QUERY_LIMIT}-entry limit; "
            + "returned counts and normalized groups are incomplete."
        )
    if any(item.startswith("ERROR") for item in findings):
        return "Error", findings
    if findings:
        return "Warning", findings
    if unexpected_unknown_rules(results):
        return "Needs Review", findings
    return "Healthy", findings


def build_markdown(report: InspectionReport) -> str:
    pool = report.pool
    window = report.window
    results = report.results
    verdict, findings = report_verdict(results)
    unknown_rules = unexpected_unknown_rules(results)
    finding_lines = [f"- {finding}" for finding in findings] or ["- No evaluated error or warning finding."]
    lines = [
        f"## Shared Pool: `{markdown_safe(pool.control_plane_info)}` / `{markdown_safe(pool.k8s_cluster_info)}`",
        "",
        f"- Inspection Window: {to_local_string(window.start, window.timezone_name)} ~ {to_local_string(window.end, window.timezone_name)}",
        "- Scope: **pool-wide shared Serverless TiKV health**; this section is not a logical-cluster, tenant, or keyspace assessment.",
        f"- Parent API Routing Identity: `control_plane_info={markdown_safe(pool.control_plane_info)}`",
        f"- Fixed Metric Scope: `control_plane_info={markdown_safe(pool.promql_control_plane_info)}`, `k8s_cluster_info={markdown_safe(pool.k8s_cluster_info)}`, `namespace={SERVERLESS_NAMESPACE}`, `component={SERVERLESS_COMPONENT}`",
        f"- Fixed Log Scope: `cluster={pool.k8s_cluster_info}`, `namespace={SERVERLESS_NAMESPACE}`, `container={SERVERLESS_LOG_CONTAINER}`",
        "- Parent Loki routing: `control_plane_info` is sent as an HTTP parameter; the proxy injects/replaces the upstream `cluster` matcher.",
        "- TiKV Dashboard: `Starter-Essential-TiKV-Details-for-Clinic` (UID `_eCx0ZDvz`)",
        "",
        "### Summary",
        f"- Verdict: {verdict}",
        "- Health Summary: " + (
            "Pool-wide error signals require immediate investigation." if verdict == "Error" else
            "Pool-wide pressure should remain under watch." if verdict == "Warning" else
            "Required signals are incomplete; this shared pool needs review." if verdict == "Needs Review" else
            "All evaluated pool-wide signals are healthy in this window."
        ),
        f"- Unknown Rules: {', '.join(sorted(unknown_rules))}" if unknown_rules else None,
        "",
        "### Key Findings",
        *finding_lines,
        "",
        "### Error Signals",
        f"- E1 critical errors: {format_result(results['E1'])}",
        f"- E2-E5 busy/report events: {', '.join(format_result(results[item]) for item in ('E2', 'E3', 'E4', 'E5'))}",
        "",
        "### gRPC QPS and Latency",
        f"- G1 gRPC message rate (read, max): {format_g1_rates(results['G1'], 'max', G1_READ_GRPC_TYPES)}",
        f"- G1 gRPC message rate (read, avg): {format_g1_rates(results['G1'], 'avg', G1_READ_GRPC_TYPES)}",
        f"- G1 gRPC message rate (write, max): {format_g1_rates(results['G1'], 'max', G1_WRITE_GRPC_TYPES)}",
        f"- G1 gRPC message rate (write, avg): {format_g1_rates(results['G1'], 'avg', G1_WRITE_GRPC_TYPES)}",
        f"- G2 maximum rolling 30-minute failures: {format_result(results['G2'])}",
        f"- G3 write P99: {format_result(results['G3'])} s ({markdown_safe(results['G3'].detail)}); G4 average by type: {format_result(results['G4'])} s; {markdown_safe(results['G4'].detail)}; G5 read P99: {format_result(results['G5'])} s",
        "",
        "### CPU and Memory",
        f"- B1 CPU peak: {format_result(results['B1'])}% ({markdown_safe(results['B1'].detail)}); B2 memory peak: {format_result(results['B2'])}% ({markdown_safe(results['B2'].detail)})",
        f"- B3 memory growth versus prior seven-day baseline with Region/QPS/CPU exclusions: {format_result(results['B3'])}",
        f"- B4 start-time changes: {format_result(results['B4'])}",
        "",
        "### Critical Thread Pool CPU",
        *render_markdown_table(
            ["Item", "Observation"],
            [
                [f"T1 — {group.name}", format_t1_cpu_rate(results["T1"], group)]
                for group in visible_t1_groups((report,))
            ],
        ),
        f"- T2 unified-read running tasks: {format_result(results['T2'])}; T3 unified-read elapsed time: {format_result(results['T3'])} s",
        "",
        "### Disk Usage and Region Count",
        f"- C1 all-used/capacity peak: {format_result(results['C1'])}% ({markdown_safe(results['C1'].detail)}); C2 region/leader observation: {format_result(results['C2'])}; {markdown_safe(results['C2'].detail)}",
        "",
        "### Disk Load",
        f"- D1 raft P99: {format_result(results['D1'])} s; D2 Raft Engine WAL write throttle duration: {format_result(results['D2'])} s",
        f"- D3 Raft Engine pending compaction WAL peak: {format_result(results['D3'])} (warn at 2 or more)",
        f"- D4 TiKV-pod-mapped non-root NVMe I/O utilization observation: {format_result(results['D4'])}; D5 weighted disk queue observation: {format_result(results['D5'])}",
        f"- D6 TiKV-pod-mapped non-root NVMe read latency: {format_milliseconds(results['D6'])} (warn only above 20 ms); D7 write latency: {format_milliseconds(results['D7'])} (warn only above 20 ms); D8 exceeded IOPS: {format_result(results['D8'])}",
        f"- F1 throttle-duration P99: {format_result(results['F1'])}; F2 discard ratio: {format_result(results['F2'])}",
        f"- P1 store slow score: {format_result(results['P1'])}; R1 raft-log-lag rate (partial): {format_result(results['R1'])}",
        "",
        "### Compaction",
        f"- K1 pending compaction shards: {format_result(results['K1'])} (observation only; no GiB threshold)",
        f"- K2 failures/local_compaction_disabled by container and result: {format_result(results['K2'])}; {markdown_safe(results['K2'].detail)}",
        f"- K3 all compaction results by container and result: {format_result(results['K3'])}; {markdown_safe(results['K3'].detail)}",
        "",
        "### ERROR/WARN Logs",
        f"- L1 actionable ERROR/FATAL/WARN lines after noise exclusions: {format_result(results['L1'])}",
        f"- {markdown_safe(results['L1'].detail)}" if results["L1"].status == "Known" else None,
        "",
        "### Follow-up Investigation",
        f"- {follow_up_action(report)}",
    ]
    return "\n".join(line for line in lines if line is not None) + "\n"


def build_daily_report_markdown(reports: Iterable[InspectionReport], report_date: str, window: Window) -> str:
    report_list = list(reports)
    lines = [
        f"- Inspection Date: {report_date}",
        f"- Inspection Window: {to_local_string(window.start, window.timezone_name)} ~ {to_local_string(window.end, window.timezone_name)}",
        "- Scope: Every section is a pool-wide shared Serverless TiKV assessment.",
        "- Data Sources: Clinic Serverless Parent Metrics / Clinic Serverless Parent Logs",
        "",
    ]
    for report in report_list:
        lines.append(report.markdown.rstrip())
        lines.append("")
    return "\n".join(lines).rstrip() + "\n"


def report_date_label(report: InspectionReport) -> str:
    return report.window.start.astimezone(ZoneInfo(report.window.timezone_name)).date().isoformat()


def report_column_labels(reports: Sequence[InspectionReport]) -> list[str]:
    date_labels = [report_date_label(report) for report in reports]
    label_counts = Counter(date_labels)
    labels: list[str] = []
    for report, date_label in zip(reports, date_labels):
        if label_counts[date_label] == 1:
            labels.append(date_label)
            continue
        local_timezone = ZoneInfo(report.window.timezone_name)
        local_start = report.window.start.astimezone(local_timezone)
        local_end = report.window.end.astimezone(local_timezone)
        labels.append(
            f"{local_start.strftime('%Y-%m-%d %H:%M %Z')} to "
            f"{local_end.strftime('%Y-%m-%d %H:%M %Z')}"
        )
    return labels


def markdown_table_cell(value: object) -> str:
    return markdown_safe(value).replace("\n", " ").replace("|", "&#124;")


def render_markdown_table(headers: Sequence[str], rows: Iterable[Sequence[object]]) -> list[str]:
    rendered = [
        "| " + " | ".join(markdown_table_cell(header) for header in headers) + " |",
        "| " + " | ".join("---" for _ in headers) + " |",
    ]
    rendered.extend(
        "| " + " | ".join(markdown_table_cell(cell) for cell in row) + " |"
        for row in rows
    )
    return rendered


def matrix_result_cell(rule_id: str, result: RuleResult) -> str:
    if result.unknown:
        return f"Unknown: {result.detail}"
    if result.value is None:
        return "Unknown"
    if rule_id in {"D6", "D7"}:
        value = format_milliseconds(result)
    elif rule_id in {"G3", "G4", "G5", "T3", "D1", "D2", "F1"}:
        value = f"{result.value:.2f} s"
    elif rule_id in {"B1", "B2", "C1"}:
        value = f"{result.value:.2f}%"
    elif rule_id == "B3":
        value = f"{result.value:.2f} pp"
    elif rule_id == "K3":
        if result.detail.startswith("container="):
            value = f"{result.value:.2f}; {result.detail}"
        else:
            value = f"{result.value:.2f}"
    elif rule_id == "L1":
        value = (
            f">= {result.value:.0f} actionable lines (incomplete)"
            if result.truncated
            else f"{result.value:.0f} actionable lines"
        )
    else:
        value = f"{result.value:.2f}"
    return value


def report_summary_cell(report: InspectionReport) -> str:
    verdict, _ = report_verdict(report.results)
    unknown_rules = sorted(unexpected_unknown_rules(report.results))
    return f"{verdict}; Unknown: {', '.join(unknown_rules)}" if unknown_rules else verdict


def render_rule_matrix(
    section: str,
    reports: Sequence[InspectionReport],
    column_labels: Sequence[str] | None = None,
) -> list[str]:
    rules = [rule for rule in RULE_SPECS if rule.section == section]
    headers = ["Item", *(column_labels or report_column_labels(reports))]
    rows: list[list[object]] = []
    for rule in rules:
        if rule.rule_id == "T1":
            for group in visible_t1_groups(reports):
                rows.append([
                    f"T1 — {group.name}",
                    *(format_t1_cpu_rate(report.results["T1"], group) for report in reports),
                ])
            continue
        if rule.rule_id == "G1":
            for path, aggregation, request_types in (
                ("read", "max", G1_READ_GRPC_TYPES),
                ("read", "avg", G1_READ_GRPC_TYPES),
                ("write", "max", G1_WRITE_GRPC_TYPES),
                ("write", "avg", G1_WRITE_GRPC_TYPES),
            ):
                rows.append([
                    f"G1 — gRPC message rate ({path}, {aggregation})",
                    *(format_g1_rates(report.results["G1"], aggregation, request_types) for report in reports),
                ])
            continue
        rows.append([
            f"{rule.rule_id} — {rule.purpose}",
            *(matrix_result_cell(rule.rule_id, report.results[rule.rule_id]) for report in reports),
        ])
    return render_markdown_table(headers, rows)


def render_log_tables(
    reports: Sequence[InspectionReport],
    column_labels: Sequence[str] | None = None,
) -> list[str]:
    lines: list[str] = []
    labels = column_labels or report_column_labels(reports)
    for report, column_label in zip(reports, labels):
        result = report.results["L1"]
        heading = f"#### {column_label} — Top 10 actionable ERROR/WARN log groups"
        if result.truncated:
            heading += f" (incomplete: {LOKI_LOG_QUERY_LIMIT}-entry limit reached)"
        lines.extend(("", heading))
        if result.unknown:
            rows: list[Sequence[object]] = [[f"Unknown: {result.detail}", "—"]]
        elif not result.log_groups:
            rows = [["No actionable ERROR/FATAL/WARN log groups", 0]]
        else:
            rows = [
                [message, f">= {count}" if result.truncated else count]
                for count, message in result.log_groups
            ]
        lines.extend(render_markdown_table(["Content", "Count"], rows))
    return lines


def follow_up_action(report: InspectionReport) -> str:
    verdict, _ = report_verdict(report.results)
    unknown_rules = unexpected_unknown_rules(report.results)
    has_actionable_logs = bool(report.results["L1"].value)
    if verdict == "Error":
        suffix = ""
        if unknown_rules and has_actionable_logs:
            suffix = ", Unknown rules, and actionable logs"
        elif unknown_rules:
            suffix = " and Unknown rules"
        elif has_actionable_logs:
            suffix = " and actionable logs"
        return f"Investigate Error findings immediately{suffix} before declaring this shared pool healthy."
    if verdict == "Warning" and has_actionable_logs:
        unknown_text = ", Unknown rules," if unknown_rules else ""
        return f"Investigate Warning findings{unknown_text} and actionable logs before declaring this shared pool healthy."
    if verdict == "Warning":
        if unknown_rules:
            return "Investigate Warning findings and Unknown rules before declaring this shared pool healthy."
        return "Investigate Warning findings before declaring this shared pool healthy."
    if unknown_rules and has_actionable_logs:
        return "Review Unknown rules and actionable logs before declaring this shared pool healthy."
    if unknown_rules:
        return "Review Unknown rules before declaring this shared pool healthy."
    if has_actionable_logs:
        return "Review actionable ERROR/WARN logs before declaring this shared pool healthy."
    return "Continue routine pool-wide monitoring."


def build_multi_day_report_markdown(
    reports: Iterable[InspectionReport], report_range: ReportRange
) -> str:
    reports_by_pool: dict[PoolIdentity, list[InspectionReport]] = {}
    for report in reports:
        reports_by_pool.setdefault(report.pool, []).append(report)

    lines = [
        f"- Inspection Date Range: {report_range.label}",
        f"- Inspection Window: {to_local_string(report_range.start, report_range.timezone_name)} ~ {to_local_string(report_range.end, report_range.timezone_name)}",
        "- Scope: Every section is a pool-wide shared Serverless TiKV assessment.",
        "- Data Sources: Clinic Serverless Parent Metrics / Clinic Serverless Parent Logs",
        "- Every date/window column reflects an independent inspection window; results are not averaged or merged.",
        "",
    ]
    for pool, pool_reports in reports_by_pool.items():
        ordered_reports = sorted(pool_reports, key=lambda report: report.window.start)
        date_headers = report_column_labels(ordered_reports)
        lines.extend((
            f"## Shared Pool: `{markdown_safe(pool.control_plane_info)}` / `{markdown_safe(pool.k8s_cluster_info)}`",
            "",
            f"- Parent API Routing Identity: `control_plane_info={markdown_safe(pool.control_plane_info)}`",
            f"- Fixed Metric Scope: `control_plane_info={markdown_safe(pool.promql_control_plane_info)}`, `k8s_cluster_info={markdown_safe(pool.k8s_cluster_info)}`, `namespace={SERVERLESS_NAMESPACE}`, `component={SERVERLESS_COMPONENT}`",
            f"- Fixed Log Scope: `cluster={markdown_safe(pool.k8s_cluster_info)}`, `namespace={SERVERLESS_NAMESPACE}`, `container={SERVERLESS_LOG_CONTAINER}`",
            "",
            "### Summary",
            *render_markdown_table(["Item", *date_headers], [["Verdict and unknown rules", *(report_summary_cell(report) for report in ordered_reports)]]),
            "",
            "### Key Findings",
            *render_markdown_table(
                ["Item", *date_headers],
                [["Evaluated findings", *("; ".join(report_verdict(report.results)[1]) or "No evaluated error or warning finding." for report in ordered_reports)]],
            ),
        ))
        for section in (
            "Error Signals",
            "gRPC QPS and Latency",
            "CPU and Memory",
            "Critical Thread Pool CPU",
            "Disk Usage and Region Count",
            "Disk Load",
            "Compaction",
            "ERROR/WARN Logs",
        ):
            lines.extend(("", f"### {section}", *render_rule_matrix(section, ordered_reports, date_headers)))
            if section == "ERROR/WARN Logs":
                lines.extend(render_log_tables(ordered_reports, date_headers))
        lines.extend((
            "",
            "### Follow-up Investigation",
            *render_markdown_table(
                ["Item", *date_headers],
                [["Required action", *(follow_up_action(report) for report in ordered_reports)]],
            ),
            "",
        ))
    return "\n".join(lines).rstrip() + "\n"


def json_default(value: object) -> object:
    if isinstance(value, datetime):
        return value.isoformat()
    if isinstance(value, frozenset):
        return sorted(value, key=repr)
    if is_dataclass(value):
        return {field.name: getattr(value, field.name) for field in fields(value)}
    raise TypeError(f"cannot encode {type(value).__name__} as JSON")


def run_inspect(args: argparse.Namespace) -> int:
    pools = normalize_pools(args.pool)
    if len(pools) != 1:
        raise ValueError("inspect requires exactly one --pool CONTROL_PLANE K8S_CLUSTER")
    window = resolve_window(args.report_date, args.start, args.end, args.relative_window, args.timezone)
    report = inspect_pool(ServerlessInspectionClient(args.env_file), pools[0], window)
    output = json.dumps(report, default=json_default, indent=2) if args.json else report.markdown
    print(output, end="" if not args.json else "\n")
    return 0


def run_daily_report(args: argparse.Namespace) -> int:
    pools = normalize_pools(args.pool)
    if not pools:
        raise ValueError("daily-report requires at least one --pool CONTROL_PLANE K8S_CLUSTER")
    if len(pools) > MAX_DAILY_REPORT_POOLS:
        raise ValueError(f"daily-report supports at most {MAX_DAILY_REPORT_POOLS} pools")
    windows = resolve_daily_report_windows(args.report_date, args.start, args.end, args.relative_window, args.timezone)
    report_range = resolve_report_range(windows)
    client = ServerlessInspectionClient(args.env_file)
    reports = [inspect_pool(client, pool, window) for pool in pools for window in windows]
    report_date = args.report_date or last_included_local_date(windows[-1])
    markdown = (
        build_multi_day_report_markdown(reports, report_range)
        if len(windows) > 1
        else build_daily_report_markdown(reports, report_date, windows[0])
    )
    payload = {
        "report_date": report_date,
        "report_range": report_range,
        "windows": windows,
        "reports": reports,
        "markdown": markdown,
    }
    output = json.dumps(payload, default=json_default, indent=2) if args.json else markdown
    print(output, end="" if not args.json else "\n")
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Serverless shared TiKV pool daily inspection")
    subparsers = parser.add_subparsers(dest="command", required=True)
    for name, handler, many in (("inspect", run_inspect, False), ("daily-report", run_daily_report, True)):
        command = subparsers.add_parser(name)
        command.add_argument("--pool", action="append", nargs=2, metavar=("CONTROL_PLANE", "K8S_CLUSTER"), required=True)
        command.add_argument("--env-file", default=DEFAULT_ENV_FILE)
        command.add_argument("--report-date")
        command.add_argument("--start")
        command.add_argument("--end")
        command.add_argument("--relative-window")
        command.add_argument("--timezone", default=DEFAULT_TZ)
        command.add_argument("--json", action="store_true")
        command.set_defaults(func=handler, accepts_many_pools=many)
    return parser


def main() -> int:
    parser = build_parser()
    try:
        args = parser.parse_args()
        return int(args.func(args))
    except (RuntimeError, ValueError, ZoneInfoNotFoundError) as error:
        print(f"error: {error}", file=sys.stderr)
        return 1


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