from __future__ import annotations

import importlib.util
import io
import json
import os
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock, patch


SCRIPT = Path(__file__).parents[1] / "scripts" / "serverless_daily_inspection.py"
SPEC = importlib.util.spec_from_file_location("serverless_daily_inspection", SCRIPT)
assert SPEC is not None and SPEC.loader is not None
MODULE = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = MODULE
SPEC.loader.exec_module(MODULE)


class ServerlessDailyInspectionTest(unittest.TestCase):
    CONTROL_PLANE = "us-west-2-f04/aws-us-west-2/devtier-infra"
    PROMQL_CONTROL_PLANE = "us-west-2-f04/aws-us-west-2"

    @staticmethod
    def pool_and_window():
        pool = MODULE.PoolIdentity(
            "us-west-2-f04/aws-us-west-2/devtier-infra", "prod-us-west-2-f04"
        )
        start = datetime(2026, 7, 23, tzinfo=timezone.utc)
        return pool, MODULE.Window(start, start + timedelta(hours=2), "UTC")

    @staticmethod
    def range_response(labels, points):
        return {
            "status": "success",
            "data": {"result": [{"metric": labels, "values": points}]},
        }

    @staticmethod
    def loki_response(streams):
        return {
            "status": "success",
            "data": {"resultType": "streams", "result": streams},
        }

    @staticmethod
    def empty_thread_group_responses():
        return {
            group.name: {"status": "success", "data": {"result": []}}
            for group in MODULE.T1_THREAD_CPU_GROUPS
        }

    def test_identity_parsing_normalizes_and_deduplicates_pairs(self) -> None:
        pools = MODULE.normalize_pools([
            [self.CONTROL_PLANE, "k8s-a"],
            [self.CONTROL_PLANE, "k8s-a"],
            [self.CONTROL_PLANE, "k8s-b"],
        ])
        self.assertEqual(pools, [
            MODULE.PoolIdentity(self.CONTROL_PLANE, "k8s-a"),
            MODULE.PoolIdentity(self.CONTROL_PLANE, "k8s-b"),
        ])
        with self.assertRaises(ValueError):
            MODULE.PoolIdentity.parse("", "k8s-a")
        with self.assertRaises(ValueError):
            MODULE.PoolIdentity.parse("us-west-2-f04/aws-us-west-2", "k8s-a")
        with self.assertRaises(ValueError):
            MODULE.PoolIdentity.parse("us-west-2-f04//devtier-infra", "k8s-a")

    def test_promql_selectors_are_scoped_and_only_unavailable_rules_are_not_queried(self) -> None:
        pool, window = self.pool_and_window()
        queries = {
            **MODULE.build_rule_queries(pool, window),
            **MODULE.build_range_rule_queries(pool),
            **{f"T1:{name}": query for name, query in MODULE.build_thread_group_queries(pool).items()},
        }
        for rule_id, query in queries.items():
            for label in (
                'control_plane_info="us-west-2-f04/aws-us-west-2"',
                'k8s_cluster_info="prod-us-west-2-f04"',
            ):
                self.assertIn(label, query)
            if rule_id not in {"D4", "D5", "D6", "D7"}:
                self.assertIn('namespace="tidb-serverless"', query)
                self.assertIn('component="tikv"', query)
            self.assertNotIn('control_plane_info="us-west-2-f04/aws-us-west-2/devtier-infra"', query)
        self.assertIn("node_disk_io_time_seconds_total", queries["D4"])
        self.assertIn("node_disk_io_time_weighted_seconds_total", queries["D5"])
        self.assertIn("node_disk_read_time_seconds_total", queries["D6"])
        self.assertIn("node_disk_write_time_seconds_total", queries["D7"])
        self.assertNotIn("D9", queries)
        self.assertIn("tikv_scheduler_throttle_duration_seconds_bucket", queries["F1"])
        self.assertIn("raft_engine_wal_write_throttle_duration_seconds_sum", queries["D2"])
        self.assertNotIn("tikv_engine_write_stall", queries["D2"])
        self.assertIn("raft_engine_pending_compaction_wals", queries["D3"])
        self.assertIn('container="tikv"', queries["D3"])
        self.assertNotIn('container=~"tikv|tikv-worker"', queries["D3"])
        self.assertIn("sum by (type) (rate", queries["G1"])
        self.assertEqual(queries["G1"].count("[1m]"), 2)
        self.assertIn("max_over_time", queries["G1"])
        self.assertIn("avg_over_time", queries["G1"])
        self.assertIn('"aggregation", "max", "", ""', queries["G1"])
        self.assertIn('"aggregation", "avg", "", ""', queries["G1"])
        self.assertIn('type=~"coprocessor|kv_get|kv_batch_get|kv_prewrite|kv_commit"', queries["G1"])
        self.assertNotIn("kv_batch_get_command", queries["G1"])
        self.assertIn("kv_batch_get_command", queries["G5"])
        self.assertNotIn('type=~', queries["G2"])
        self.assertIn("sum by (type)", queries["G4"])
        self.assertIn("sum by (instance,type)", queries["C2"])
        self.assertIn("max_over_time", queries["D1"])
        self.assertIn("[7200s:5m]", queries["D1"])
        self.assertIn("tikv_raftstore_append_log_duration_seconds_bucket", queries["D1"])
        self.assertIn("tikv_raftstore_commit_log_duration_seconds_bucket", queries["D1"])
        self.assertIn("tikv_raftstore_apply_log_duration_seconds_bucket", queries["D1"])
        self.assertTrue(queries["D1"].startswith("max(label_replace("))
        self.assertEqual(queries["D1"].count("label_replace("), 3)
        for stage in ("append", "commit", "apply"):
            self.assertIn(f'"raft_stage", "{stage}", "", ""', queries["D1"])
        self.assertIn("sum by (instance)(increase(multilevel_level_elapsed", queries["T3"])
        self.assertIn('name="unified-read-pool"', queries["T3"])
        self.assertIn('level="total"', queries["T3"])
        self.assertTrue(queries["T3"].endswith(" / 1000000"))
        self.assertIn("kv_engine_pending_compaction_shards", queries["K1"])
        self.assertNotIn("raft_engine_pending_compaction_wals", queries["K1"])
        for rule_id in ("K2", "K3"):
            self.assertIn('container=~"tikv|tikv-worker"', queries[rule_id])
        self.assertIn('result=~"failure|local_compaction_disabled"', queries["K2"])
        self.assertIn("kv_engine_compaction_result_counter", queries["K2"])
        self.assertIn("sum by (container,result)", queries["K2"])
        self.assertIn("sum by (container,result)", queries["K3"])
        expected_thread_groups = [
            ("Raft store CPU", r"(raftstore|rs|rfstore)_.*"),
            ("Async apply CPU", r"apply_[0-9]+|apply_follower_[0-9]+"),
            ("Raft async writer CPU", r"raft_io.*"),
            ("Scheduler pool CPU", r"sched_tokio_pool.*|sched_yatp_pool.*|sched_background.*"),
            ("gRPC poll CPU", r"grpc.*"),
            ("KvEngine worker CPU", r"txn-chunk-worker.*|txn_chunk_worker.*"),
            ("RfEngine worker CPU", r"sub-compact-wkr.*|compact-wal-worker.*"),
            ("Unified read pool CPU", r"unified_read_po.*"),
            ("Background worker CPU", r"background.*"),
            ("GC worker CPU", r"gc_worker.*"),
            ("Import CPU", r"sst_.*"),
            ("Backup CPU", r"(backup-worker|bkwkr|backup_endpoint).*"),
            ("CDC worker CPU", r"(cdcwkr.*|tso|cdc_.*)"),
            ("TSO worker CPU", r"tso_worker"),
            ("IME CPU", r"ime.*"),
            ("IA CPU", r"s3fifo.*"),
            ("DFS worker CPU", r"(builtin-dfs.*|memory-fs.*|s3-client.*)"),
            ("Cloud worker CPU", r"worker-server.*"),
            ("Lock Manager Thread CPU", r"(waiter_manager.*|deadlock_detect.*)"),
            ("Resolved TS Worker CPU", r"resolved_ts.*"),
            ("Advance ts Worker CPU", r"advance_ts.*"),
            ("Scan lock Worker CPU", r"inc_scan.*"),
            ("Backup CPU Utilization", r"(b.*k.*w.*k.*|backup_io)"),
        ]
        self.assertEqual(
            [(group.name, group.thread_name_regex) for group in MODULE.T1_THREAD_CPU_GROUPS],
            expected_thread_groups,
        )
        self.assertEqual(list(MODULE.build_thread_group_queries(pool)), [name for name, _ in expected_thread_groups])
        for group, pattern in expected_thread_groups:
            query = MODULE.build_thread_group_queries(pool)[group]
            self.assertTrue(query.startswith("sum by (instance) (rate(tikv_thread_cpu_seconds_total{"))
            self.assertIn(f'name=~"{pattern}"', query)
            self.assertNotIn("by (instance,name)", query)
        all_queries = "\n".join(queries.values())
        for invented in ("tikv_multilevel_level_elapsed", "tikv_storage_mvcc_versions_bucket", "tikv_storage_mvcc_gc_delete_versions_bucket", "tikv_engine_write_stall", "tikv_storage_mvcc_gc_delete_versions_sum"):
            self.assertNotIn(invented, all_queries)
        self.assertEqual(
            queries["F2"],
            'max by (instance) (max_over_time((tikv_scheduler_discard_ratio{control_plane_info="us-west-2-f04/aws-us-west-2",k8s_cluster_info="prod-us-west-2-f04",namespace="tidb-serverless",component="tikv"})[7200s:5m]))',
        )
        for rule_id in ("D8",):
            self.assertNotIn(rule_id, queries)

        b3_queries = MODULE.build_b3_queries(pool, window)
        self.assertEqual(set(b3_queries), {
            "memory_current", "memory_baseline", "region_current", "region_baseline",
            "qps_current", "qps_baseline", "cpu_current", "cpu_baseline",
        })
        for query in b3_queries.values():
            self.assertIn('namespace="tidb-serverless"', query)
            self.assertIn('component="tikv"', query)
        memory_query = MODULE.memory_percentage_query(pool)
        self.assertEqual(
            b3_queries["memory_baseline"],
            f"avg_over_time(({memory_query})[{MODULE.MEMORY_LEAK_BASELINE_WINDOW}:5m] offset {window.prom_window})",
        )

    def test_b3_warns_on_unexplained_memory_growth_and_exempts_workload_growth(self) -> None:
        def response(values):
            return {"status": "success", "data": {"result": [
                {"metric": {"instance": "tikv-0"}, "value": [0, str(values)]},
            ]}}

        unexplained = {
            "memory_current": response(60), "memory_baseline": response(54),
            "region_current": response(100), "region_baseline": response(100),
            "qps_current": response(100), "qps_baseline": response(100),
            "cpu_current": response(20), "cpu_baseline": response(20),
        }
        result = MODULE.b3_result(unexplained)
        self.assertTrue(result.warning)
        self.assertIn("without matching", result.detail)

        explained = dict(unexplained)
        explained["qps_current"] = response(220)
        result = MODULE.b3_result(explained)
        self.assertFalse(result.warning)
        self.assertIn("QPS", result.detail)

        for current_name, current_value, expected_detail in (
            ("region_current", 220, "regions"),
            ("cpu_current", 26, "CPU"),
        ):
            explained = dict(unexplained)
            explained[current_name] = response(current_value)
            result = MODULE.b3_result(explained)
            self.assertFalse(result.warning)
            self.assertIn(expected_detail, result.detail)

        observation = dict(unexplained)
        observation["memory_current"] = response(57)
        result = MODULE.b3_result(observation)
        self.assertFalse(result.warning)
        self.assertIn("observation only below the 5 pp warning line", result.detail)

        missing_support = dict(unexplained)
        missing_support["region_current"] = {"status": "success", "data": {"result": []}}
        missing_support["region_baseline"] = {"status": "success", "data": {"result": []}}
        missing_support["qps_current"] = {"status": "success", "data": {"result": []}}
        missing_support["qps_baseline"] = {"status": "success", "data": {"result": []}}
        missing_support["cpu_current"] = {"status": "success", "data": {"result": []}}
        missing_support["cpu_baseline"] = {"status": "success", "data": {"result": []}}
        result = MODULE.b3_result(missing_support)
        self.assertTrue(result.warning)
        self.assertIn("60.00% current vs 54.00% prior-7d average", result.detail)

    def test_disk_queries_join_unscoped_node_exporter_metrics_to_tikv_storage_pods(self) -> None:
        pool, window = self.pool_and_window()
        queries = MODULE.build_rule_queries(pool, window)
        device_matcher = f'device=~"{MODULE.NON_ROOT_NVME_DEVICE_RE}"'
        pod_selector = (
            'kube_pod_info{control_plane_info="us-west-2-f04/aws-us-west-2",'
            'k8s_cluster_info="prod-us-west-2-f04",namespace="tidb-serverless",'
            f'pod=~"{MODULE.TIKV_STORAGE_POD_RE}"}}'
        )
        for rule_id in ("D4", "D5", "D6", "D7"):
            query = queries[rule_id]
            self.assertIn(device_matcher, query)
            self.assertIn("* on(instance) group_left(pod) label_replace(", query)
            self.assertIn(pod_selector, query)
            self.assertIn('"instance", "$1", "node", "(.*)")', query)
            self.assertIn("max_over_time", query)
            self.assertIn("[7200s:5m]", query)
        for metric_name in (
            "node_disk_io_time_seconds_total",
            "node_disk_io_time_weighted_seconds_total",
            "node_disk_read_time_seconds_total",
            "node_disk_reads_completed_total",
            "node_disk_write_time_seconds_total",
            "node_disk_writes_completed_total",
        ):
            for rule_id in ("D4", "D5", "D6", "D7"):
                query = queries[rule_id]
                if metric_name in query:
                    self.assertNotIn(
                        f'{metric_name}{{control_plane_info=', query,
                    )
                    self.assertNotIn(f'{metric_name}{{namespace=', query)
                    self.assertNotIn(f'{metric_name}{{component=', query)

    def test_disk_latency_warns_strictly_above_twenty_milliseconds_and_empty_is_zero(self) -> None:
        results = {rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose) for rule in MODULE.RULE_SPECS}
        results["D6"] = MODULE.RuleResult("Known", 0.020, "read latency")
        results["D7"] = MODULE.RuleResult("Known", 0.020, "write latency")
        _, findings = MODULE.report_verdict(results)
        self.assertNotIn("WARNING: TiKV data-device read latency exceeded 20 ms.", findings)
        self.assertNotIn("WARNING: TiKV data-device write latency exceeded 20 ms.", findings)

        results["D6"] = MODULE.RuleResult("Known", 0.02001, "read latency")
        results["D7"] = MODULE.RuleResult("Known", 0.030, "write latency")
        verdict, findings = MODULE.report_verdict(results)
        self.assertEqual(verdict, "Warning")
        self.assertIn("WARNING: TiKV data-device read latency exceeded 20 ms.", findings)
        self.assertIn("WARNING: TiKV data-device write latency exceeded 20 ms.", findings)
        for rule_id in ("D5", "D6", "D7"):
            result = MODULE.result_from_prometheus({"status": "success", "data": {"result": []}}, rule_id)
            self.assertEqual(result.status, "Known")
            self.assertEqual(result.value, 0.0)

    def test_disk_latency_queries_zero_idle_disks_without_accepting_nan(self) -> None:
        pool, window = self.pool_and_window()
        queries = MODULE.build_rule_queries(pool, window)
        for rule_id, duration_metric, operations_metric in (
            ("D6", "node_disk_read_time_seconds_total", "node_disk_reads_completed_total"),
            ("D7", "node_disk_write_time_seconds_total", "node_disk_writes_completed_total"),
        ):
            denominator = f"rate({MODULE.node_disk_metric(operations_metric)}[5m])"
            query = queries[rule_id]
            self.assertIn(f"and on(instance,device) ({denominator} > 0)", query)
            self.assertIn(f"or on(instance,device) (0 * {denominator})", query)
            self.assertIn(f"rate({MODULE.node_disk_metric(duration_metric)}[5m])", query)

            idle_disk = MODULE.result_from_prometheus(
                {"status": "success", "data": {"result": [{"value": [0, "0"]}]}}, rule_id
            )
            self.assertEqual(idle_disk.status, "Known")
            self.assertEqual(idle_disk.value, 0.0)

            nan_disk = MODULE.result_from_prometheus(
                {"status": "success", "data": {"result": [{"value": [0, "NaN"]}]}}, rule_id
            )
            self.assertTrue(nan_disk.unknown)

    def test_loki_routes_by_http_control_plane_and_uses_documented_labels(self) -> None:
        pool, window = self.pool_and_window()
        query = MODULE.build_loki_filter(pool)
        for expected in ('cluster="prod-us-west-2-f04"', 'namespace="tidb-serverless"', 'container="tikv-log"', "cdc initialize fail"):
            self.assertIn(expected, query)
        for unsupported_label in ("control_plane_info=", "k8s_cluster_info=", "component="):
            self.assertNotIn(unsupported_label, query)

        captured = {}

        class ParentLogsClient:
            def _get_json(self, path, params):
                captured["path"] = path
                captured["params"] = params
                return {"status": "success", "data": {"result": []}}

        MODULE.ServerlessInspectionClient.log_query(ParentLogsClient(), pool, query, window)
        self.assertEqual(captured["path"], "/data-proxy/serverless-parent-logs/api/v1/query_range")
        self.assertEqual(captured["params"]["control_plane_info"], self.CONTROL_PLANE)
        self.assertEqual(captured["params"]["query"], query)
        self.assertEqual(captured["params"]["limit"], MODULE.LOKI_LOG_QUERY_LIMIT)
        self.assertEqual(captured["params"]["limit"], 1000)

    def test_parent_metric_routes_keep_the_full_grafana_control_plane(self) -> None:
        pool, window = self.pool_and_window()
        captured = []

        class ParentMetricsClient:
            def _get_json(self, path, params):
                captured.append((path, params))
                return {"status": "success", "data": {"result": []}}

        client = ParentMetricsClient()
        MODULE.ServerlessInspectionClient.metric_query(client, pool, "up", window.end)
        MODULE.ServerlessInspectionClient.metric_query_range(client, pool, "up", window)
        self.assertEqual(
            [params["control_plane_info"] for _, params in captured],
            [self.CONTROL_PLANE, self.CONTROL_PLANE],
        )
        self.assertEqual(
            [path for path, _ in captured],
            [
                "/data-proxy/serverless-parent-metrics/api/v1/query",
                "/data-proxy/serverless-parent-metrics/api/v1/query_range",
            ],
        )

    def test_sustained_windows_and_inclusive_thresholds(self) -> None:
        fifteen_minutes = [(index * 300, "1") for index in range(3)]
        self.assertTrue(MODULE.has_sustained_threshold([(float(ts), float(value)) for ts, value in fifteen_minutes], 1.0, 15 * 60))
        self.assertFalse(MODULE.has_sustained_threshold([(0.0, 1.0), (600.0, 1.0), (1200.0, 1.0)], 1.0, 15 * 60))
        thirty_minutes = [[index * 300, "85"] for index in range(6)]
        result = MODULE.sustained_result(self.range_response({"instance": "tikv-0"}, thirty_minutes), "B1", (85.0,), 30 * 60)
        self.assertIn(85.0, result.sustained_at)

        c1 = MODULE.c1_growth_result(self.range_response(
            {"instance": "tikv-0"},
            [[0, "70"], [300, "72"], [600, "74"], [900, "76"], [1200, "78"], [1500, "80"]],
        ))
        self.assertIn(80.0, c1.sustained_at)
        self.assertIn("tikv-0 rose 70.00% -> 80.00%", c1.detail)
        self.assertIn("over 25m", c1.detail)

        results = {rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose) for rule in MODULE.RULE_SPECS}
        results["G2"] = MODULE.RuleResult("Known", 100.0, "rolling maximum")
        results["G3"] = MODULE.RuleResult("Known", 1.0, "sustained", frozenset({1.0}))
        results["B1"] = MODULE.RuleResult("Known", 85.0, "sustained", frozenset({85.0}))
        results["B2"] = MODULE.RuleResult("Known", 90.0, "sustained", frozenset({85.0, 90.0}))
        results["P1"] = MODULE.RuleResult("Known", 100.0, "score")
        verdict, findings = MODULE.report_verdict(results)
        self.assertEqual(verdict, "Error")
        self.assertGreaterEqual(len(findings), 5)

    def test_successful_empty_prometheus_results_are_known_zero_across_paths(self) -> None:
        response = {"status": "success", "data": {"result": []}}
        for rule_id in ("E1", "E5", "D2", "T3", "F2", "K1"):
            empty = MODULE.result_from_prometheus(response, rule_id)
            self.assertEqual(empty.status, "Known")
            self.assertEqual(empty.value, 0.0)
            self.assertIn("successful empty response", empty.detail)

        for parser in (
            lambda: MODULE.g1_result(response),
            lambda: MODULE.g4_result(response),
            lambda: MODULE.c2_result(response),
            lambda: MODULE.compaction_result_by_container_and_result(response),
            lambda: MODULE.range_result(response, "B1")[0],
            lambda: MODULE.sustained_result(response, "B1", (85.0,), 30 * 60),
            lambda: MODULE.c1_growth_result(response),
            lambda: MODULE.thread_group_result(self.empty_thread_group_responses()),
        ):
            result = parser()
            self.assertEqual(result.status, "Known")
            self.assertEqual(result.value, 0.0)

        b3_responses = {
            name: response
            for name in MODULE.build_b3_queries(*self.pool_and_window()).keys()
        }
        b3 = MODULE.b3_result(b3_responses)
        self.assertEqual(b3.status, "Known")
        self.assertEqual(b3.value, 0.0)

    def test_prometheus_errors_and_malformed_nonempty_responses_remain_unknown(self) -> None:
        error_response = {"status": "error", "error": "query failed"}
        malformed_vector = {"status": "success", "data": {"result": [{"metric": {}, "value": [0, "NaN"]}]}}
        malformed_matrix = {"status": "success", "data": {"result": [{"metric": {}, "values": []}]}}

        self.assertTrue(MODULE.result_from_prometheus(error_response, "E1").unknown)
        self.assertTrue(MODULE.result_from_prometheus(malformed_vector, "E1").unknown)
        self.assertTrue(MODULE.g1_result(error_response).unknown)
        self.assertTrue(MODULE.g1_result({"status": "success", "data": {"result": [{"value": [0, "1"]}]}}).unknown)
        self.assertTrue(MODULE.range_result(error_response, "B1")[0].unknown)
        self.assertTrue(MODULE.range_result(malformed_matrix, "B1")[0].unknown)
        self.assertTrue(MODULE.thread_group_result({"Raft store": malformed_matrix}).unknown)

        b3_responses = {
            name: {"status": "success", "data": {"result": []}}
            for name in MODULE.build_b3_queries(*self.pool_and_window()).keys()
        }
        b3_responses["memory_current"] = malformed_vector
        self.assertTrue(MODULE.b3_result(b3_responses).unknown)

    def test_loki_logs_trim_datetimes_group_messages_and_limit_to_ten(self) -> None:
        streams = [
            {"stream": {}, "values": [
                ["1", "2026-07-23T01:02:03.123Z ERROR apply failed for region 7"],
                ["2", "2026/07/23 09:02:04.456 +08:00 ERROR apply failed for region 7"],
                ["3", "2026-07-23 01:02:05 WARN scheduler slow"],
            ]},
            {"stream": {}, "values": [
                [str(index + 4), f"2026-07-23T01:02:{index + 10:02d}Z WARN unique warning {index}"]
                for index in range(11)
            ]},
        ]
        result = MODULE.result_from_loki(self.loki_response(streams))
        self.assertEqual(result.value, 14.0)
        self.assertIn("2 × ERROR apply failed for region 7", result.detail)
        self.assertNotIn("2026-07-23", result.detail)
        self.assertNotIn("2026/07/23", result.detail)
        self.assertEqual(result.detail.count(" × "), MODULE.MAX_LOG_GROUPS)

        pool, window = self.pool_and_window()
        results = {rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose) for rule in MODULE.RULE_SPECS}
        results["L1"] = result
        markdown = MODULE.build_markdown(MODULE.InspectionReport(pool, window, results))
        self.assertIn("top normalized groups", markdown)
        self.assertIn("2 × ERROR apply failed for region 7", markdown)

    def test_loki_log_groups_sort_equal_counts_by_content(self) -> None:
        result = MODULE.result_from_loki(self.loki_response([
            {"stream": {}, "values": [
                ["1", "component-a WARN first"],
                ["2", "component-z ERROR last"],
                ["3", "component-m WARN middle"],
            ]},
        ]))

        self.assertEqual(
            result.log_groups,
            (
                (1, "component-a WARN first"),
                (1, "component-m WARN middle"),
                (1, "component-z ERROR last"),
            ),
        )
        self.assertLess(
            result.detail.index("component-a WARN first"),
            result.detail.index("component-z ERROR last"),
        )

    def test_loki_limit_results_are_lower_bounds_and_incomplete(self) -> None:
        streams = [{"stream": {}, "values": [
            [str(index), "2026-07-23T01:02:03Z WARN log query limit reached"]
            for index in range(MODULE.LOKI_LOG_QUERY_LIMIT)
        ]}]
        result = MODULE.result_from_loki(self.loki_response(streams))

        self.assertTrue(result.truncated)
        self.assertEqual(result.value, float(MODULE.LOKI_LOG_QUERY_LIMIT))
        self.assertIn(f"at least {MODULE.LOKI_LOG_QUERY_LIMIT}", result.detail)
        self.assertIn("incomplete", result.detail)
        self.assertIn(f">= {MODULE.LOKI_LOG_QUERY_LIMIT} ×", result.detail)
        self.assertTrue(json.loads(json.dumps(result, default=MODULE.json_default))["truncated"])
        self.assertEqual(
            MODULE.matrix_result_cell("L1", result),
            f">= {MODULE.LOKI_LOG_QUERY_LIMIT} actionable lines (incomplete)",
        )

        pool, window = self.pool_and_window()
        results = {rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose) for rule in MODULE.RULE_SPECS}
        results["L1"] = result
        report = MODULE.InspectionReport(pool, window, results)
        markdown = MODULE.build_markdown(report)
        log_tables = "\n".join(MODULE.render_log_tables([report]))
        self.assertIn(f">= {MODULE.LOKI_LOG_QUERY_LIMIT:.2f} (incomplete)", markdown)
        self.assertIn("returned counts and normalized groups are incomplete", markdown)
        self.assertIn("incomplete: 1000-entry limit reached", log_tables)
        self.assertIn(f"| WARN log query limit reached | >= {MODULE.LOKI_LOG_QUERY_LIMIT} |", log_tables)

    def test_malformed_loki_responses_remain_unknown(self) -> None:
        malformed_responses = (
            {"status": "success"},
            {"status": "success", "data": {"result": "not-a-list"}},
            {"status": "success", "data": {"resultType": "vector", "result": []}},
            {"status": "success", "data": {"resultType": "streams", "result": ["not-a-stream"]}},
            {"status": "success", "data": {"resultType": "streams", "result": [{"stream": {}, "values": "not-a-list"}]}},
            {"status": "success", "data": {"resultType": "streams", "result": [{"stream": {}, "values": []}]}},
            {"status": "success", "data": {"resultType": "streams", "result": [{"stream": {}, "values": [["1"]]}]}},
            {"status": "success", "data": {"resultType": "streams", "result": [{"stream": {}, "values": [["1", 2]]}]}},
        )
        for response in malformed_responses:
            with self.subTest(response=response):
                result = MODULE.result_from_loki(response)
                self.assertTrue(result.unknown)
                self.assertIn("malformed Loki response", result.detail)

        empty = MODULE.result_from_loki(self.loki_response([]))
        self.assertEqual(empty.status, "Known")
        self.assertEqual(empty.value, 0.0)

    def test_g1_reports_max_and_average_one_minute_rates_by_type(self) -> None:
        response = {
            "status": "success",
            "data": {"result": [
                {"metric": {"aggregation": "max", "type": "kv_get"}, "value": [0, "720"]},
                {"metric": {"aggregation": "avg", "type": "kv_get"}, "value": [0, "360"]},
                {"metric": {"aggregation": "max", "type": "kv_commit"}, "value": [0, "120"]},
                {"metric": {"aggregation": "avg", "type": "kv_commit"}, "value": [0, "60"]},
            ]},
        }
        result = MODULE.g1_result(response)
        self.assertEqual(result.value, 720.0)
        self.assertIn("max kv_get: 720.00 messages/s", result.detail)
        self.assertIn("avg kv_commit: 60.00 messages/s", result.detail)
        serialized = json.loads(json.dumps(result, default=MODULE.json_default))
        self.assertEqual(serialized["g1_rates"][1], {
            "aggregation": "max",
            "request_type": "kv_get",
            "messages_per_second": 720.0,
        })
        self.assertEqual(
            MODULE.format_g1_rates(result, "max", MODULE.G1_READ_GRPC_TYPES),
            "coprocessor: 0.00 messages/s, kv_get: 720.00 messages/s, kv_batch_get: 0.00 messages/s",
        )
        self.assertEqual(
            MODULE.format_g1_rates(result, "avg", MODULE.G1_WRITE_GRPC_TYPES),
            "kv_prewrite: 0.00 messages/s, kv_commit: 60.00 messages/s",
        )

        pool, window = self.pool_and_window()
        results = {
            rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose)
            for rule in MODULE.RULE_SPECS
        }
        results["G1"] = result
        matrix = "\n".join(MODULE.render_rule_matrix(
            "gRPC QPS and Latency",
            [MODULE.InspectionReport(pool, window, results)],
        ))
        self.assertIn(
            "| G1 — gRPC message rate (read, max) | "
            "coprocessor: 0.00 messages/s, kv_get: 720.00 messages/s, kv_batch_get: 0.00 messages/s |",
            matrix,
        )
        self.assertIn(
            "| G1 — gRPC message rate (write, avg) | "
            "kv_prewrite: 0.00 messages/s, kv_commit: 60.00 messages/s |",
            matrix,
        )

        malformed = MODULE.g1_result({
            "status": "success",
            "data": {"result": [
                {"metric": {"aggregation": "peak", "type": "kv_get"}, "value": [0, "1"]},
            ]},
        })
        self.assertTrue(malformed.unknown)
        self.assertIn("malformed G1 series labels", malformed.detail)

    def test_thread_groups_keep_per_instance_evidence_and_render_aggregate_cpu_rates(self) -> None:
        responses = self.empty_thread_group_responses()
        responses["Raft store CPU"] = {
            "status": "success",
            "data": {"result": [
                {"metric": {"instance": "tikv-1"}, "values": [[0, "0.5"], [300, "1.25"], [600, "0.75"]]},
                {"metric": {"instance": "tikv-0"}, "values": [[0, "0.3"], [300, "0.4"]]},
                {"metric": {"instance": "tikv-0"}, "values": [[0, "0.2"], [300, "0.6"]]},
            ]},
        }
        result = MODULE.thread_group_result(responses)

        self.assertEqual(result.status, "Known")
        self.assertFalse(result.warning)
        self.assertEqual(result.value, 1.25)
        rates = {(rate.group, rate.instance): rate for rate in result.t1_cpu_rates}
        self.assertEqual(
            rates[("Raft store CPU", "tikv-0")],
            MODULE.T1CpuRate("Raft store CPU", "tikv-0", 1.0, 0.75),
        )
        self.assertEqual(
            rates[("Raft store CPU", "tikv-1")],
            MODULE.T1CpuRate("Raft store CPU", "tikv-1", 1.25, 2.5 / 3),
        )
        self.assertEqual(
            rates[("Async apply CPU", None)],
            MODULE.T1CpuRate("Async apply CPU", None, 0.0, 0.0),
        )
        self.assertIn("Raft store CPU: max: 1.25 cores, avg: 0.79 cores", result.detail)
        self.assertNotIn("tikv-0", result.detail)
        self.assertNotIn("tikv-1", result.detail)
        self.assertEqual(
            MODULE.format_t1_cpu_rate(result, MODULE.T1_THREAD_CPU_GROUPS[0]),
            "max: 1.25 cores, avg: 0.79 cores",
        )
        self.assertEqual(
            MODULE.format_t1_cpu_rate(result, MODULE.T1_THREAD_CPU_GROUPS[1]),
            "No observed instances; max: 0.00 cores, avg: 0.00 cores",
        )

        serialized = json.loads(json.dumps(result, default=MODULE.json_default))
        self.assertEqual(serialized["t1_cpu_rates"][0], {
            "group": "Raft store CPU",
            "instance": "tikv-0",
            "max_cores": 1.0,
            "avg_cores": 0.75,
            "error": None,
        })

        pool, window = self.pool_and_window()
        results = {rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose) for rule in MODULE.RULE_SPECS}
        results["T1"] = result
        markdown = MODULE.build_markdown(MODULE.InspectionReport(pool, window, results))
        self.assertIn("| Item | Observation |", markdown)
        self.assertIn(
            "| T1 — Raft store CPU | max: 1.25 cores, avg: 0.79 cores |",
            markdown,
        )
        self.assertNotIn("| T1 — Async apply CPU |", markdown)
        t1_section = markdown.split("### Critical Thread Pool CPU", 1)[1].split("### Disk Usage", 1)[0]
        self.assertNotIn("tikv-0", t1_section)
        self.assertNotIn("tikv-1", t1_section)

    def test_t1_report_rows_hide_only_known_low_maxima(self) -> None:
        pool, first_window = self.pool_and_window()
        second_window = MODULE.Window(
            first_window.start + timedelta(days=1),
            first_window.end + timedelta(days=1),
            "UTC",
        )

        def report(window, group_maxima, unknown_group=None, empty_groups=frozenset()):
            results = {
                rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose)
                for rule in MODULE.RULE_SPECS
            }
            t1_rates = tuple(
                MODULE.T1CpuRate(group.name, None, None, None, "query failed")
                if group.name == unknown_group
                else MODULE.T1CpuRate(group.name, "tikv-0", group_maxima[group.name], 0.05)
                if group.name in group_maxima
                else MODULE.T1CpuRate(group.name, None, 0.0, 0.0)
                if group.name in empty_groups
                else MODULE.T1CpuRate(group.name, "tikv-0", 0.09, 0.05)
                for group in MODULE.T1_THREAD_CPU_GROUPS
            )
            results["T1"] = MODULE.RuleResult(
                "Unknown" if unknown_group else "Known",
                max(group_maxima.values(), default=0.09),
                "thread CPU observations",
                t1_cpu_rates=t1_rates,
            )
            return MODULE.InspectionReport(pool, window, results)

        quiet = report(first_window, {})
        quiet_markdown = MODULE.build_markdown(quiet)
        self.assertNotIn("| T1 —", quiet_markdown)

        empty_groups = frozenset(group.name for group in MODULE.T1_THREAD_CPU_GROUPS)
        successful_empty = report(first_window, {}, empty_groups=empty_groups)
        empty_markdown = MODULE.build_markdown(successful_empty)
        self.assertNotIn("| T1 —", empty_markdown)

        threshold = report(first_window, {"Raft store CPU": 0.10})
        threshold_markdown = MODULE.build_markdown(threshold)
        self.assertIn("| T1 — Raft store CPU | max: 0.10 cores, avg: 0.05 cores |", threshold_markdown)
        self.assertNotIn("| T1 — Async apply CPU |", threshold_markdown)

        report_range = MODULE.resolve_report_range([first_window, second_window])
        any_window_visible = MODULE.build_multi_day_report_markdown(
            [quiet, report(second_window, {"Raft store CPU": 0.10})],
            report_range,
        )
        self.assertIn(
            "| T1 — Raft store CPU | max: 0.09 cores, avg: 0.05 cores | "
            "max: 0.10 cores, avg: 0.05 cores |",
            any_window_visible,
        )
        self.assertNotIn("| T1 — Async apply CPU |", any_window_visible)

        empty_with_active_window = MODULE.build_multi_day_report_markdown(
            [successful_empty, report(second_window, {"Raft store CPU": 0.10})],
            report_range,
        )
        self.assertIn(
            "| T1 — Raft store CPU | No observed instances; max: 0.00 cores, avg: 0.00 cores | "
            "max: 0.10 cores, avg: 0.05 cores |",
            empty_with_active_window,
        )

        all_windows_quiet = MODULE.build_multi_day_report_markdown(
            [quiet, report(second_window, {})],
            report_range,
        )
        self.assertNotIn("| T1 —", all_windows_quiet)

        all_windows_empty = MODULE.build_multi_day_report_markdown(
            [successful_empty, report(second_window, {}, empty_groups=empty_groups)],
            report_range,
        )
        self.assertNotIn("| T1 —", all_windows_empty)

        unknown_visible = MODULE.build_multi_day_report_markdown(
            [quiet, report(second_window, {}, unknown_group="Scheduler pool CPU")],
            report_range,
        )
        self.assertIn("| T1 — Scheduler pool CPU |", unknown_visible)
        self.assertIn("Unknown: query failed", unknown_visible)

    def test_g4_and_c2_keep_labeled_evidence(self) -> None:
        g4 = MODULE.g4_result({"status": "success", "data": {"result": [
            {"metric": {"type": "kv_get"}, "value": [0, "0.02"]},
            {"metric": {"type": "kv_commit"}, "value": [0, "0.04"]},
        ]}})
        self.assertEqual(g4.value, 0.04)
        self.assertIn("kv_commit: 0.040000 s", g4.detail)
        self.assertIn("kv_get: 0.020000 s", g4.detail)

        c2 = MODULE.c2_result({"status": "success", "data": {"result": [
            {"metric": {"instance": "tikv-0", "type": "region"}, "value": [0, "120"]},
            {"metric": {"instance": "tikv-0", "type": "leader"}, "value": [0, "60"]},
        ]}})
        self.assertEqual(c2.value, 120.0)
        self.assertIn("tikv-0 leader=60", c2.detail)
        self.assertIn("tikv-0 region=120", c2.detail)

    def test_compaction_rules_keep_labeled_evidence_and_warn_at_thresholds(self) -> None:
        failure_results = [
            {"metric": {"container": "tikv", "result": "failure"}, "value": [0, "2"]},
            {"metric": {"container": "tikv-worker", "result": "local_compaction_disabled"}, "value": [0, "1"]},
        ]
        response = {"status": "success", "data": {"result": [
            *failure_results,
            {"metric": {"container": "tikv", "result": "success"}, "value": [0, "5"]},
        ]}}
        k2 = MODULE.compaction_result_by_container_and_result({
            "status": "success", "data": {"result": failure_results}
        })
        k3 = MODULE.compaction_result_by_container_and_result(response)
        self.assertEqual(k2.value, 3.0)
        self.assertIn("container=tikv, result=failure: 2", k2.detail)
        self.assertIn("container=tikv-worker, result=local_compaction_disabled: 1", k2.detail)
        self.assertEqual(k3.value, 8.0)
        self.assertIn("container=tikv, result=success: 5", k3.detail)

        self.assertEqual(
            MODULE.matrix_result_cell("K3", k3),
            "8.00; container=tikv, result=failure: 2; container=tikv, result=success: 5; "
            "container=tikv-worker, result=local_compaction_disabled: 1",
        )
        empty_k3 = MODULE.compaction_result_by_container_and_result({"status": "success", "data": {"result": []}})
        self.assertEqual(MODULE.matrix_result_cell("K3", empty_k3), "0.00")

        results = {rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose) for rule in MODULE.RULE_SPECS}
        results["D2"] = MODULE.RuleResult("Known", 0.01, "WAL throttle")
        results["F2"] = MODULE.RuleResult("Known", 0.01, "discard ratio")
        results["D3"] = MODULE.RuleResult("Known", 1.0, "pending WALs")
        results["K2"] = k2
        _, findings = MODULE.report_verdict(results)
        self.assertIn("WARNING: Raft Engine WAL write throttling was observed.", findings)
        self.assertIn("WARNING: flow-control discard ratio became positive.", findings)
        self.assertNotIn("WARNING: Raft Engine pending compaction WALs reached 2 or more.", findings)
        self.assertIn("WARNING: compaction failure or local_compaction_disabled results were observed.", findings)

        results["D3"] = MODULE.RuleResult("Known", 2.0, "pending WALs")
        _, findings = MODULE.report_verdict(results)
        self.assertIn("WARNING: Raft Engine pending compaction WALs reached 2 or more.", findings)

        pool, window = self.pool_and_window()
        results["K3"] = k3
        markdown = MODULE.build_markdown(MODULE.InspectionReport(pool, window, results))
        self.assertIn("### Compaction", markdown)
        self.assertIn("container=tikv-worker, result=local_compaction_disabled: 1", markdown)
        self.assertIn("container=tikv, result=success: 5", markdown)

    def test_window_bounds_and_daily_pool_limit(self) -> None:
        window = MODULE.resolve_window(None, None, "2026-07-23T03:00:00Z", "3h", "UTC")
        self.assertEqual(window.seconds, 3 * 60 * 60)
        for start, end, relative in (
            ("2026-07-23T00:00:00Z", None, "3h"),
            (None, "2026-07-23T00:00:00Z", "25h"),
            ("2026-07-23T00:00:00Z", "2026-07-24T00:00:01Z", None),
        ):
            with self.assertRaises(ValueError):
                MODULE.resolve_window(None, start, end, relative, "UTC")
        args = SimpleNamespace(
            pool=[[f"pool-{index}/aws-us-west-2/devtier-infra", f"k8s-{index}"] for index in range(MODULE.MAX_DAILY_REPORT_POOLS + 1)],
            report_date=None, start=None, end=None, relative_window=None, timezone="UTC", env_file="", json=False,
        )
        with self.assertRaises(ValueError):
            MODULE.run_daily_report(args)

    def test_historical_inspect_report_date_ends_at_next_local_midnight(self) -> None:
        window = MODULE.resolve_window("2000-02-29", None, None, None, "UTC")

        self.assertEqual(window.start.isoformat(), "2000-02-29T00:00:00+00:00")
        self.assertEqual(window.end.isoformat(), "2000-03-01T00:00:00+00:00")
        self.assertEqual(window.seconds, 24 * 60 * 60)

    def test_multi_day_daily_report_windows_are_local_day_bounded(self) -> None:
        windows = MODULE.resolve_daily_report_windows(
            None,
            "2026-07-23T06:00:00+00:00",
            "2026-07-25T03:00:00+00:00",
            None,
            "UTC",
        )
        self.assertEqual(
            [(window.start.isoformat(), window.end.isoformat()) for window in windows],
            [
                ("2026-07-23T06:00:00+00:00", "2026-07-24T00:00:00+00:00"),
                ("2026-07-24T00:00:00+00:00", "2026-07-25T00:00:00+00:00"),
                ("2026-07-25T00:00:00+00:00", "2026-07-25T03:00:00+00:00"),
            ],
        )
        self.assertTrue(all(window.seconds <= MODULE.MAX_WINDOW_SECONDS for window in windows))
        maximum_windows = MODULE.resolve_daily_report_windows(
            None,
            "2026-07-01T00:00:00+00:00",
            "2026-08-01T00:00:00+00:00",
            None,
            "UTC",
        )
        self.assertEqual(len(maximum_windows), MODULE.MAX_MULTI_DAY_REPORT_DAYS)
        with self.assertRaisesRegex(ValueError, "at most"):
            MODULE.resolve_daily_report_windows(
                None,
                "2026-01-01T00:00:00+00:00",
                "2026-02-02T00:00:00+00:00",
                None,
                "UTC",
            )

    def test_report_date_uses_last_included_local_date(self) -> None:
        args = SimpleNamespace(
            pool=[[self.CONTROL_PLANE, "k8s-a"]],
            report_date=None,
            start="2025-03-09T00:00:00-05:00",
            end="2025-03-10T00:00:00-04:00",
            relative_window=None,
            timezone="America/New_York",
            env_file="",
            json=True,
        )
        output = io.StringIO()

        def fake_inspect_pool(_client, pool, window):
            results = {
                rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose)
                for rule in MODULE.RULE_SPECS
            }
            report = MODULE.InspectionReport(pool, window, results)
            report.markdown = MODULE.build_markdown(report)
            return report

        with (
            patch.object(MODULE, "ServerlessInspectionClient", return_value=object()),
            patch.object(MODULE, "inspect_pool", side_effect=fake_inspect_pool),
            patch.object(sys, "stdout", output),
        ):
            exit_code = MODULE.run_daily_report(args)

        payload = json.loads(output.getvalue())

        self.assertEqual(exit_code, 0)
        self.assertEqual(payload["report_date"], "2025-03-09")
        self.assertIn("- Inspection Date: 2025-03-09", payload["markdown"])

    def test_multi_day_report_exposes_canonical_range_metadata(self) -> None:
        args = SimpleNamespace(
            pool=[[self.CONTROL_PLANE, "k8s-a"]],
            report_date=None,
            start="2026-07-20T00:00:00+08:00",
            end="2026-07-23T00:00:00+08:00",
            relative_window=None,
            timezone="Asia/Shanghai",
            env_file="",
            json=True,
        )
        output = io.StringIO()

        def fake_inspect_pool(_client, pool, window):
            results = {
                rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose)
                for rule in MODULE.RULE_SPECS
            }
            return MODULE.InspectionReport(pool, window, results)

        with (
            patch.object(MODULE, "ServerlessInspectionClient", return_value=object()),
            patch.object(MODULE, "inspect_pool", side_effect=fake_inspect_pool),
            patch.object(sys, "stdout", output),
        ):
            exit_code = MODULE.run_daily_report(args)

        payload = json.loads(output.getvalue())

        self.assertEqual(exit_code, 0)
        self.assertEqual(payload["report_date"], "2026-07-22")
        self.assertEqual(payload["report_range"], {
            "start": "2026-07-20T00:00:00+08:00",
            "end": "2026-07-23T00:00:00+08:00",
            "timezone_name": "Asia/Shanghai",
            "start_date": "2026-07-20",
            "end_date": "2026-07-22",
            "label": "2026-07-20 to 2026-07-22",
        })
        self.assertIn("- Inspection Date Range: 2026-07-20 to 2026-07-22", payload["markdown"])
        self.assertIn(
            "- Inspection Window: 2026-07-20 00:00:00 CST ~ 2026-07-23 00:00:00 CST",
            payload["markdown"],
        )

    def test_dst_fall_back_windows_use_elapsed_time(self) -> None:
        with self.assertRaisesRegex(ValueError, "must not exceed 24 hours"):
            MODULE.resolve_window(
                None,
                "2025-11-02T00:00:00-04:00",
                "2025-11-03T00:00:00-05:00",
                None,
                "America/New_York",
            )

        windows = MODULE.resolve_daily_report_windows(
            None,
            "2025-11-02T00:00:00-04:00",
            "2025-11-03T00:00:00-05:00",
            None,
            "America/New_York",
        )
        self.assertEqual([window.seconds for window in windows], [24 * 60 * 60, 60 * 60])
        self.assertTrue(all(window.seconds <= MODULE.MAX_WINDOW_SECONDS for window in windows))

        report_date_windows = MODULE.resolve_daily_report_windows(
            "2025-11-02",
            None,
            None,
            None,
            "America/New_York",
        )
        self.assertEqual(
            [window.seconds for window in report_date_windows],
            [24 * 60 * 60, 60 * 60],
        )

        pool, _ = self.pool_and_window()
        results = {
            rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose)
            for rule in MODULE.RULE_SPECS
        }
        reports = [
            MODULE.InspectionReport(pool, window, results)
            for window in report_date_windows
        ]
        labels = MODULE.report_column_labels(reports)
        self.assertEqual(len(labels), len(set(labels)))
        self.assertEqual(
            labels,
            [
                "2025-11-02 00:00 EDT to 2025-11-02 23:00 EST",
                "2025-11-02 23:00 EST to 2025-11-03 00:00 EST",
            ],
        )
        report_range = MODULE.resolve_report_range(report_date_windows)
        markdown = MODULE.build_multi_day_report_markdown(reports, report_range)
        self.assertIn(labels[0], markdown)
        self.assertIn(labels[1], markdown)

    def test_multi_day_report_renders_rule_by_date_and_per_date_log_tables(self) -> None:
        pool, first_window = self.pool_and_window()
        second_window = MODULE.Window(
            first_window.start + timedelta(days=1),
            first_window.end + timedelta(days=1),
            "UTC",
        )
        base_results = {
            rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose)
            for rule in MODULE.RULE_SPECS
        }
        first_results = dict(base_results)
        first_results["T1"] = MODULE.RuleResult(
            "Known",
            1.23,
            "thread CPU observations",
            t1_cpu_rates=(
                MODULE.T1CpuRate("Raft store CPU", "tikv-0", 1.23, 0.45),
                MODULE.T1CpuRate("Raft store CPU", "tikv-1", 0.80, 0.30),
            ),
        )
        first_results["K3"] = MODULE.RuleResult("Known", 5.0, "container=tikv, result=success: 5")
        first_results["L1"] = MODULE.RuleResult(
            "Known",
            3.0,
            "actionable ERROR/FATAL/WARN log lines",
            log_groups=((2, "WARN first|line"), (1, '<image url="https://attacker.example/pixel.png"/>')),
        )
        second_results = dict(base_results)
        second_results["T1"] = MODULE.RuleResult(
            "Known",
            2.34,
            "thread CPU observations",
            t1_cpu_rates=(MODULE.T1CpuRate("Raft store CPU", "tikv-1", 2.34, 0.67),),
        )
        second_results["L1"] = MODULE.RuleResult(
            "Known",
            1.0,
            "actionable ERROR/FATAL/WARN log lines",
            log_groups=((1, "WARN second day"),),
        )
        report_range = MODULE.resolve_report_range([first_window, second_window])
        markdown = MODULE.build_multi_day_report_markdown(
            [
                MODULE.InspectionReport(pool, first_window, first_results),
                MODULE.InspectionReport(pool, second_window, second_results),
            ],
            report_range,
        )
        self.assertIn("- Inspection Date Range: 2026-07-23 to 2026-07-24", markdown)
        self.assertIn(
            "- Inspection Window: 2026-07-23 00:00:00 UTC ~ 2026-07-24 02:00:00 UTC",
            markdown,
        )
        self.assertIn("| Item | 2026-07-23 | 2026-07-24 |", markdown)
        self.assertIn("| E1 — critical errors | 0.00 | 0.00 |", markdown)
        self.assertIn(
            "| T1 — Raft store CPU | max: 1.23 cores, avg: 0.38 cores | "
            "max: 2.34 cores, avg: 0.67 cores |",
            markdown,
        )
        t1_row = next(line for line in markdown.splitlines() if line.startswith("| T1 — Raft store CPU |"))
        self.assertNotIn("tikv-0", t1_row)
        self.assertNotIn("tikv-1", t1_row)
        self.assertIn(
            "| G1 — gRPC message rate (read, max) | "
            "coprocessor: 0.00 messages/s, kv_get: 0.00 messages/s, kv_batch_get: 0.00 messages/s | "
            "coprocessor: 0.00 messages/s, kv_get: 0.00 messages/s, kv_batch_get: 0.00 messages/s |",
            markdown,
        )
        self.assertIn(
            "| G1 — gRPC message rate (write, avg) | "
            "kv_prewrite: 0.00 messages/s, kv_commit: 0.00 messages/s | "
            "kv_prewrite: 0.00 messages/s, kv_commit: 0.00 messages/s |",
            markdown,
        )
        self.assertEqual(markdown.count("| G1 — gRPC message rate ("), 4)
        self.assertIn("#### 2026-07-23 — Top 10 actionable ERROR/WARN log groups", markdown)
        self.assertIn("#### 2026-07-24 — Top 10 actionable ERROR/WARN log groups", markdown)
        self.assertIn("| WARN first&#124;line | 2 |", markdown)
        self.assertNotIn("\\|", markdown)
        self.assertIn(
            "| &lt;image url=\"https://attacker.example/pixel.png\"/&gt; | 1 |",
            markdown,
        )
        self.assertNotIn('<image url="https://attacker.example/pixel.png"/>', markdown)
        self.assertIn("| WARN second day | 1 |", markdown)
        self.assertEqual(markdown.count("| Content | Count |"), 2)
        self.assertIn("### Compaction", markdown)
        self.assertIn("| K1 — pending compaction shards | 0.00 | 0.00 |", markdown)
        self.assertIn(
            "| K3 — compaction result counts by container and result | "
            "5.00; container=tikv, result=success: 5 | 0.00 |",
            markdown,
        )

    def test_json_serializes_nested_reports_dates_and_thresholds(self) -> None:
        pool, window = self.pool_and_window()
        results = {rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose) for rule in MODULE.RULE_SPECS}
        results["B1"] = MODULE.RuleResult("Known", 85.0, "sustained", frozenset({85.0}))
        report = MODULE.InspectionReport(pool, window, results, "markdown")
        inspect_payload = json.loads(json.dumps(report, default=MODULE.json_default))
        self.assertEqual(inspect_payload["window"]["start"], "2026-07-23T00:00:00+00:00")
        self.assertEqual(inspect_payload["results"]["B1"]["sustained_at"], [85.0])
        daily_payload = json.loads(json.dumps({"report_date": "2026-07-23", "reports": [report], "markdown": "all"}, default=MODULE.json_default))
        self.assertEqual(daily_payload["reports"][0]["pool"]["control_plane_info"], self.CONTROL_PLANE)
        self.assertEqual(json.loads(json.dumps(MODULE.RuleResult("Known", 1.0, "logs", log_groups=((1, "WARN log"),)), default=MODULE.json_default))["log_groups"], [[1, "WARN log"]])

    def test_endpoint_env_and_response_security_guards(self) -> None:
        self.assertTrue(callable(MODULE._load_requests().get))
        with patch.object(
            MODULE.importlib,
            "import_module",
            side_effect=ModuleNotFoundError("No module named 'requests'"),
        ):
            with self.assertRaisesRegex(RuntimeError, "pip install -r requirements.txt"):
                MODULE._load_requests()

        for endpoint in (
            "http://clinic.pingcap.com",
            "https://evil.example",
            "https://pingcap.com",
            "https://foo.pingcap.com",
            "https://clinic.pingcap.com.evil.example",
            "https://user@clinic.pingcap.com",
            "https://clinic.pingcap.com/path?x=1",
        ):
            with self.assertRaises(ValueError):
                MODULE.validate_clinic_endpoint(endpoint)
        self.assertEqual(MODULE.validate_clinic_endpoint("https://clinic.pingcap.com/"), "https://clinic.pingcap.com")
        self.assertEqual(MODULE.validate_clinic_endpoint("https://dev-clinic.pingcap.com"), "https://dev-clinic.pingcap.com")
        self.assertEqual(MODULE.validate_clinic_endpoint("https://staging-clinic.pingcap.com"), "https://staging-clinic.pingcap.com")

        with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, {}, clear=True):
            env_file = Path(directory) / "clinic.env"
            env_file.write_text("CLINIC_API_KEY=key\nIGNORED=value\nCLINIC_ENV=dev\n")
            MODULE.load_shell_env(str(env_file))
            self.assertEqual(os.environ["CLINIC_API_KEY"], "key")
            self.assertNotIn("IGNORED", os.environ)

        client = object.__new__(MODULE.ServerlessInspectionClient)
        client.endpoint, client.api_key = "https://clinic.pingcap.com", "key"

        class FakeRequestException(Exception):
            pass

        class FakeResponse:
            def __init__(self, status_code, content):
                self.status_code = status_code
                self.content = content
                self.closed = False

            def __enter__(self):
                return self

            def __exit__(self, exc_type, exc_value, traceback):
                self.closed = True
                return False

            def iter_content(self, chunk_size):
                for offset in range(0, len(self.content), chunk_size):
                    yield self.content[offset:offset + chunk_size]

        def requests_api(*, response=None, error=None):
            request_get = Mock(return_value=response, side_effect=error)
            return MODULE._RequestsApi(request_get, FakeRequestException), request_get

        success = FakeResponse(200, b'{"status":"success"}')
        api, request_get = requests_api(response=success)
        with patch.object(MODULE, "_load_requests", return_value=api):
            self.assertEqual(client._get_json("/x", {"query": "up + 1"}), {"status": "success"})
        self.assertEqual(request_get.call_args.args[0], "https://clinic.pingcap.com/x")
        self.assertEqual(request_get.call_args.kwargs["params"], {"query": "up + 1"})
        self.assertEqual(request_get.call_args.kwargs["headers"]["Authorization"], "Bearer key")
        self.assertEqual(request_get.call_args.kwargs["timeout"], 120)
        self.assertFalse(request_get.call_args.kwargs["allow_redirects"])
        self.assertTrue(request_get.call_args.kwargs["stream"])
        self.assertTrue(success.closed)

        http_error = FakeResponse(502, b"secret body")
        api, _ = requests_api(response=http_error)
        with patch.object(MODULE, "_load_requests", return_value=api):
            self.assertEqual(client._get_json("/x", {})["error"], "Clinic HTTP status 502")
        self.assertTrue(http_error.closed)

        redirect_error = FakeResponse(302, b"")
        api, _ = requests_api(response=redirect_error)
        with patch.object(MODULE, "_load_requests", return_value=api):
            self.assertEqual(client._get_json("/x", {})["error"], "Clinic HTTP status 302")
        self.assertTrue(redirect_error.closed)

        oversized = FakeResponse(200, b"x" * (MODULE.MAX_HTTP_RESPONSE_BYTES + 1))
        api, _ = requests_api(response=oversized)
        with patch.object(MODULE, "_load_requests", return_value=api):
            self.assertEqual(client._get_json("/x", {})["error"], "Clinic response exceeds size limit")
        self.assertTrue(oversized.closed)

        api, _ = requests_api(error=FakeRequestException("offline"))
        with patch.object(MODULE, "_load_requests", return_value=api):
            self.assertIn("offline", client._get_json("/x", {})["error"])

        invalid_json = FakeResponse(200, b"{")
        api, _ = requests_api(response=invalid_json)
        with patch.object(MODULE, "_load_requests", return_value=api):
            self.assertIn("Expecting", client._get_json("/x", {})["error"])
        self.assertTrue(invalid_json.closed)

    def test_identity_and_report_injection_are_rejected_or_sanitized(self) -> None:
        for value in ("bad value", "bad`value", "bad<value>", "bad\nvalue", 'bad"value'):
            with self.assertRaises(ValueError):
                MODULE.PoolIdentity.parse(value, "k8s-a")
        pool, window = self.pool_and_window()
        results = {rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose) for rule in MODULE.RULE_SPECS}
        results["G4"] = MODULE.RuleResult("Known", 1.0, "<script>`payload`\x07")
        markdown = MODULE.build_markdown(MODULE.InspectionReport(pool, window, results))
        self.assertIn("&lt;script&gt;`payload`�", markdown)
        self.assertNotIn("<script>", markdown)

    def test_structurally_unavailable_rules_are_visible_but_never_warn(self) -> None:
        pool, window = self.pool_and_window()
        results = {rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose) for rule in MODULE.RULE_SPECS}
        results.update(MODULE.unavailable_results())
        self.assertEqual(MODULE.unexpected_unknown_rules(results), [])
        verdict, findings = MODULE.report_verdict(results)
        self.assertEqual(verdict, "Healthy")
        self.assertEqual(findings, [])
        report = MODULE.InspectionReport(pool, window, results)
        self.assertEqual(MODULE.report_summary_cell(report), "Healthy")
        self.assertEqual(MODULE.follow_up_action(report), "Continue routine pool-wide monitoring.")
        self.assertEqual(
            MODULE.matrix_result_cell("D8", results["D8"]),
            "Unknown: no aws_ebs_* exceeded-IOPS series exists in Serverless parent metrics",
        )
        markdown = MODULE.build_markdown(report)
        self.assertNotIn("Unknown Rules:", markdown)
        self.assertIn("Unknown (no aws_ebs_* exceeded-IOPS series exists in Serverless parent metrics)", markdown)

        results["E1"] = MODULE.RuleResult("Unknown", None, "Clinic HTTP status 502")
        verdict, _ = MODULE.report_verdict(results)
        self.assertEqual(verdict, "Needs Review")
        self.assertEqual(MODULE.unexpected_unknown_rules(results), ["E1"])
        report = MODULE.InspectionReport(pool, window, results)
        self.assertEqual(MODULE.report_summary_cell(report), "Needs Review; Unknown: E1")
        self.assertEqual(MODULE.follow_up_action(report), "Review Unknown rules before declaring this shared pool healthy.")
        results["L1"] = MODULE.RuleResult("Known", 5.0, "actionable logs")
        report = MODULE.InspectionReport(pool, window, results)
        self.assertEqual(
            MODULE.follow_up_action(report),
            "Investigate Warning findings, Unknown rules, and actionable logs before declaring this shared pool healthy.",
        )
        results["E1"] = MODULE.RuleResult("Known", 0.0, "critical errors")
        report = MODULE.InspectionReport(pool, window, results)
        self.assertEqual(
            MODULE.follow_up_action(report),
            "Investigate Warning findings and actionable logs before declaring this shared pool healthy.",
        )

        results["L1"] = MODULE.RuleResult("Known", 0.0, "actionable logs")
        results["E1"] = MODULE.RuleResult("Known", 1.0, "critical errors")
        report = MODULE.InspectionReport(pool, window, results)
        self.assertEqual(
            MODULE.follow_up_action(report),
            "Investigate Error findings immediately before declaring this shared pool healthy.",
        )
        self.assertIn(
            "Investigate Error findings immediately",
            MODULE.build_markdown(report),
        )

        results["E1"] = MODULE.RuleResult("Known", 0.0, "critical errors")
        results["E2"] = MODULE.RuleResult("Known", 1.0, "channel full")
        report = MODULE.InspectionReport(pool, window, results)
        self.assertEqual(
            MODULE.follow_up_action(report),
            "Investigate Warning findings before declaring this shared pool healthy.",
        )

        results["E1"] = MODULE.RuleResult("Unknown", None, "query failed")
        report = MODULE.InspectionReport(pool, window, results)
        self.assertEqual(
            MODULE.follow_up_action(report),
            "Investigate Warning findings and Unknown rules before declaring this shared pool healthy.",
        )

    def test_report_removes_markdown_escape_backslashes(self) -> None:
        escaped = (
            r"WARNING: local\_compaction\_disabled; aws\_ebs\_\*; "
            r"\[WARN\] \[peer\_fsm.rs:1524\] \{ id: 1 \}"
        )
        rendered = MODULE.markdown_safe(escaped)

        self.assertEqual(
            rendered,
            "WARNING: local_compaction_disabled; aws_ebs_*; "
            + "[WARN] [peer_fsm.rs:1524] { id: 1 }",
        )
        self.assertNotIn("\\", rendered)
        self.assertEqual(MODULE.markdown_safe(r"\<x\&y\>"), "&lt;x&amp;y&gt;")

    def test_identity_validation_requires_positive_dashboard_metric(self) -> None:
        pool, window = self.pool_and_window()

        class Client:
            pool: object | None = None
            query: str = ""
            end: datetime | None = None

            def metric_query(self, received_pool, query, end):
                self.pool, self.query, self.end = received_pool, query, end
                return {"status": "success", "data": {"result": [{"value": [0, "1"]}]}}

        client = Client()
        MODULE.validate_pool_identity(client, pool, window.end)
        self.assertEqual(client.pool, pool)
        self.assertIn("count(tikv_engine_size_bytes", client.query)
        self.assertIn(f'control_plane_info="{self.PROMQL_CONTROL_PLANE}"', client.query)
        self.assertIn('k8s_cluster_info="prod-us-west-2-f04"', client.query)
        self.assertNotIn(f'control_plane_info="{self.CONTROL_PLANE}"', client.query)

        class EmptyClient:
            def metric_query(self, *_):
                return {"status": "success", "data": {"result": []}}

        with self.assertRaisesRegex(ValueError, "identity validation failed"):
            MODULE.validate_pool_identity(EmptyClient(), pool, window.end)

    def test_metric_registries_and_unavailable_reasons_are_complete_when_source_exists(self) -> None:
        unavailable = {rule.rule_id for rule in MODULE.RULE_SPECS if rule.status == "unavailable"}
        self.assertTrue(unavailable.issubset(MODULE.UNAVAILABLE_REASONS))
        expected = {rule.rule_id for rule in MODULE.RULE_SPECS if rule.status in {"exact", "replacement", "partial"}}
        self.assertEqual(
            expected,
            set(MODULE.DASHBOARD_METRIC_REGISTRY) | set(MODULE.PARENT_ONLY_METRIC_REGISTRY),
        )
        self.assertEqual(set(MODULE.PARENT_ONLY_METRIC_REGISTRY), {"T3", "D5", "D6", "D7", "F2"})
        dashboard = Path("/Users/pingyu/workspace/cloud-storage-engine/metrics/grafana/tikv_details_serverless.json")
        if dashboard.exists():
            source = dashboard.read_text()
            for metrics in MODULE.DASHBOARD_METRIC_REGISTRY.values():
                for metric in metrics:
                    self.assertIn(metric, source)

    def test_thread_group_partial_telemetry_keeps_known_stats_without_threshold_warning(self) -> None:
        responses = self.empty_thread_group_responses()
        responses["Raft store CPU"] = self.range_response({"instance": "tikv-0"}, [[0, "1.0"], [300, "0.5"]])
        responses["Async apply CPU"] = {"error": "Clinic HTTP status 502"}
        responses["Raft async writer CPU"] = {
            "status": "success",
            "data": {"result": [{"metric": {}, "values": "not-a-list"}]},
        }
        responses["Scheduler pool CPU"] = self.range_response({}, [[0, "0.5"]])
        result = MODULE.thread_group_result(responses)

        self.assertTrue(result.unknown)
        rates = {(rate.group, rate.instance): rate for rate in result.t1_cpu_rates}
        self.assertEqual(
            rates[("Raft store CPU", "tikv-0")],
            MODULE.T1CpuRate("Raft store CPU", "tikv-0", 1.0, 0.75),
        )
        self.assertEqual(rates[("Async apply CPU", None)].error, "Clinic HTTP status 502")
        self.assertIn("malformed Prometheus matrix series", rates[("Raft async writer CPU", None)].error)
        self.assertEqual(rates[("Scheduler pool CPU", None)].error, "thread CPU series is missing instance label")
        self.assertEqual(
            MODULE.format_t1_cpu_rate(result, MODULE.T1_THREAD_CPU_GROUPS[1]),
            "Unknown: Clinic HTTP status 502",
        )
        self.assertEqual(
            MODULE.format_t1_cpu_rate(result, MODULE.T1_THREAD_CPU_GROUPS[3]),
            "Unknown: thread CPU series is missing instance label",
        )

        results = {rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose) for rule in MODULE.RULE_SPECS}
        results["T1"] = result
        verdict, findings = MODULE.report_verdict(results)
        self.assertEqual(verdict, "Needs Review")
        self.assertNotIn("WARNING: a critical thread pool exceeded its per-instance thread-count CPU threshold.", findings)

    def test_rule_catalog_and_inspection_rules_documentation_match(self) -> None:
        expected = {*(f"E{number}" for number in range(1, 6)), *(f"G{number}" for number in range(1, 6)), *(f"B{number}" for number in range(1, 5)), *(f"T{number}" for number in range(1, 4)), *(f"C{number}" for number in range(1, 3)), *(f"D{number}" for number in range(1, 9)), *(f"F{number}" for number in range(1, 3)), *(f"K{number}" for number in range(1, 4)), "P1", "R1", "L1"}
        self.assertEqual(set(MODULE.RULE_BY_ID), expected)
        self.assertEqual(MODULE.RULE_BY_ID["D4"].status, "exact")
        self.assertEqual(MODULE.RULE_BY_ID["D5"].status, "exact")
        self.assertEqual(MODULE.RULE_BY_ID["D6"].status, "exact")
        self.assertEqual(MODULE.RULE_BY_ID["D7"].status, "exact")
        self.assertEqual(MODULE.RULE_BY_ID["D8"].status, "unavailable")
        self.assertEqual(MODULE.RULE_BY_ID["D3"].status, "exact")
        self.assertEqual(MODULE.RULE_BY_ID["F2"].status, "exact")
        self.assertNotIn("D9", MODULE.RULE_BY_ID)
        self.assertEqual(MODULE.RULE_BY_ID["T3"].status, "exact")
        self.assertEqual(MODULE.RULE_BY_ID["K1"].status, "replacement")
        self.assertEqual(MODULE.RULE_BY_ID["K2"].status, "exact")
        self.assertEqual(MODULE.RULE_BY_ID["K3"].status, "exact")
        rules = (SCRIPT.parents[1] / "references" / "inspection-rules.md").read_text()
        self.assertIn("| Rule ID | Area | Metric or Logic | Threshold | Severity | Status | Source |", rules)
        self.assertNotIn("| D9 |", rules)
        for rule in MODULE.RULE_SPECS:
            self.assertRegex(rules, rf"(?m)^\| {rule.rule_id} \| .* \| {rule.status} \| .+ \|$")

        skill = (SCRIPT.parents[1] / "SKILL.md").read_text()
        feishu = (SCRIPT.parents[1] / "references" / "feishu-template-hints.md").read_text()
        for document in (skill, rules, feishu):
            self.assertIn("Compaction", document)
            self.assertIn("ERROR/WARN Logs", document)
            self.assertLess(document.index("Compaction"), document.index("ERROR/WARN Logs"))
        self.assertIn("kv_engine_compaction_result_counter", skill)
        self.assertIn("components/kvengine/src/metrics.rs", skill)
        self.assertIn("components/kvengine/src/metrics.rs", rules)

    def test_cli_and_markdown_scope_language(self) -> None:
        parser = MODULE.build_parser()
        args = parser.parse_args(["inspect", "--pool", self.CONTROL_PLANE, "k8s-a"])
        self.assertEqual(args.pool, [[self.CONTROL_PLANE, "k8s-a"]])
        with self.assertRaises(SystemExit):
            parser.parse_args(["inspect"])

        pool, window = self.pool_and_window()
        results = {rule.rule_id: MODULE.RuleResult("Known", 0.0, rule.purpose) for rule in MODULE.RULE_SPECS}
        markdown = MODULE.build_markdown(MODULE.InspectionReport(pool, window, results))
        self.assertIn("pool-wide shared Serverless TiKV health", markdown)
        self.assertIn("not a logical-cluster, tenant, or keyspace assessment", markdown)
        self.assertIn("Parent Loki routing", markdown)
        self.assertIn(self.PROMQL_CONTROL_PLANE, markdown)
        self.assertIn("Fixed Metric Scope", markdown)
        self.assertIn("k8s_cluster_info=prod-us-west-2-f04", markdown)

    def test_cli_reports_invalid_timezone_without_traceback(self) -> None:
        stderr = io.StringIO()
        argv = [
            str(SCRIPT),
            "inspect",
            "--pool",
            self.CONTROL_PLANE,
            "k8s-a",
            "--timezone",
            "Not/AZone",
        ]

        with patch.object(sys, "argv", argv), patch.object(sys, "stderr", stderr):
            exit_code = MODULE.main()

        self.assertEqual(exit_code, 1)
        self.assertTrue(stderr.getvalue().startswith("error: "))
        self.assertNotIn("Traceback", stderr.getvalue())

    def test_cli_does_not_hide_unrelated_handler_errors(self) -> None:
        argv = [str(SCRIPT), "inspect", "--pool", self.CONTROL_PLANE, "k8s-a"]

        with (
            patch.object(sys, "argv", argv),
            patch.object(MODULE, "run_inspect", side_effect=TypeError("unexpected handler failure")),
            self.assertRaisesRegex(TypeError, "unexpected handler failure"),
        ):
            MODULE.main()


if __name__ == "__main__":
    unittest.main()