import importlib.util
import json
from pathlib import Path

import pytest


SCRIPT_PATH = (
    Path(__file__).resolve().parents[2]
    / "agentkit"
    / "skills"
    / "builtin"
    / "tidb-clinic-diagnosis"
    / "scripts"
    / "collect_evidence.py"
)


def _load_collect_evidence():
    spec = importlib.util.spec_from_file_location("collect_evidence_for_test", SCRIPT_PATH)
    module = importlib.util.module_from_spec(spec)
    assert spec.loader is not None
    spec.loader.exec_module(module)
    return module


ce = _load_collect_evidence()


@pytest.mark.parametrize("rows", [[], [{"values": []}], [{"values": [[1, "NaN"], [2, "+Inf"], [3, None]]}]])
def test_infra_gate_does_not_treat_missing_samples_as_zero(rows):
    verdict = ce.check_infra_gate({"ebs": rows, "disk": rows})
    assert verdict.startswith("INFRA UNKNOWN")
    assert "unavailable, not zero" in verdict
    assert "max=0.00" not in verdict
    assert "total_volumes=0" not in verdict


def test_infra_gate_distinguishes_observed_zero_and_partial_coverage():
    zero = {"values": [[1, "0"], [2, "0"]]}
    assert "INFRA CLEAR" in ce.check_infra_gate({"ebs": [zero], "disk": [zero]})
    partial = ce.check_infra_gate({"ebs": [], "disk": [zero]})
    assert "INFRA UNKNOWN" in partial
    assert "disk_io_util max=0.00" in partial
    invalid_series = ce.check_infra_gate({"ebs": [zero], "disk": [zero, {"values": [[1, "NaN"]]}]})
    assert "INFRA UNKNOWN" in invalid_series
    assert "1 series without finite samples" in invalid_series
    abnormal = ce.check_infra_gate({"disk": [{"values": [[1, "0.95"]]}]})
    assert "INFRA ABNORMAL" in abnormal
    assert "Disk IO saturation" in abnormal
    assert "ebs: no finite samples" in abnormal
    assert "causality unverified" in abnormal


@pytest.mark.parametrize("finite_values", [None, [[1, "0"], [2, "0"]], [[1, "0.2"], [2, "0.4"], [3, "NaN"]]])
def test_metric_missing_state_survives_artifact_pipeline(tmp_path, finite_values):
    missing = {"metric": {"instance": "missing", "device": "data", "volume_id": "missing"},
               "values": [[1, "NaN"], [2, "+Inf"]]}
    rows = [missing]
    if finite_values is not None:
        rows.append({"metric": {"instance": "observed", "device": "data", "volume_id": "observed"},
                     "values": finite_values})
    data = {"disk": rows, "ebs": rows, "errors": []}
    gate = ce.check_infra_gate(data)
    assert not gate.startswith("INFRA CLEAR")
    run_dir = Path(ce.save_evidence_files(data, gate, "unknown", "c1", "1787106300", "1787107200", str(tmp_path)))
    summary = json.loads((tmp_path / "evidence_summary_for_agent.json").read_text())
    report = json.loads((tmp_path / "workspace_evidence_report.json").read_text())
    raw = json.loads((run_dir / "metrics.json").read_text())
    for metrics in (summary["metrics"], report["evidence"]["metrics"], raw["metrics"]):
        for group in ("disk_io_utilization", "ebs_iops_throttle"):
            items = [m for m in metrics if m.get("metric_group_id") == group]
            avg = next(m for m in items if m["name"].endswith(" Avg"))
            assert avg["observed_series"] == int(finite_values is not None)
            assert avg["total_series"] == len(rows)
            assert "missing" in str(avg["missing_instances"])
            assert avg["availability"] == ("partial" if finite_values is not None else "missing")
            absent = next(m for m in items if m["instance"].startswith("missing"))
            assert absent["peak"] is None and absent["mean"] is None
            assert absent["severity"] == "unknown" and absent["availability"] == "missing"
            if finite_values is None:
                assert avg["peak"] is None and avg["mean"] is None
            else:
                assert avg["peak"] == ce._prom_peak(rows[1])
                assert avg["mean"] == round(ce._prom_mean(rows[1]), 4)
    text = (tmp_path / "evidence_summary_for_agent.txt").read_text()
    assert "availability" in text and "observed_series" in text and "missing_instances" in text
    assert "missing" in text
    if finite_values is None:
        diagnostic_text = ce.format_summary(data, gate, "unknown", "c1", "1787106300", "1787107200")
        assert "max=0.00" not in diagnostic_text
        assert "EBS_THROTTLE: unavailable" in diagnostic_text


def test_cpu_summary_does_not_average_missing_series_as_zero():
    data = {"cpu": [
        {"metric": {"component": "tidb", "instance": "missing"}, "values": [[1, "NaN"]]},
        {"metric": {"component": "tidb", "instance": "valid"}, "values": [[1, "4"]]},
    ]}
    summary = ce.build_agent_summary(data, "unknown", "unknown", "c1", "1787106300", "1787107200")
    avg = next(m for m in summary["metrics"] if m["name"] == "TIDB CPU Avg")
    assert avg["peak"] == 4 and avg["availability"] == "partial"
    missing = next(m for m in summary["metrics"] if m["instance"] == "missing")
    assert missing["peak"] is None and missing["severity"] == "unknown"


def test_logstats_keeps_context_counts_and_samples_without_promoting_them(monkeypatch):
    start = ce._parse_time_to_epoch_seconds("2026-08-01T02:25:00Z")
    end = ce._parse_time_to_epoch_seconds("2026-08-01T02:40:00Z")
    rows = [
        {"date": "20260801", "hour": "2", "component": "tidb", "level": "WARN",
         "count": "500", "msg_digest": "Query execution was interrupted",
         "sample": "2026-08-01T02:41:00Z;Query execution was interrupted"},
        {"date": "20260801", "hour": "2", "component": "tidb", "level": "WARN",
         "count": "10", "msg_digest": "deadlock", "sample": "2026-08-01T02:30:00Z;deadlock"},
        {"date": "20260801", "hour": "2", "component": "tidb", "level": "WARN",
         "count": "5", "msg_digest": "panic:", "sample": "panic: timestamp unavailable"},
    ]
    monkeypatch.setattr(ce, "_run_clinic_logstats_query", lambda *args: (rows, None, None))
    logs, error = ce._collect_log_stats_from_clinic_api(None, {}, "c1", str(start), str(end))
    assert error is None
    assert logs["summary"]["tidb"]["warn"] == 515
    assert sum(r["count"] for r in logs["log_counts"]) == 515
    assert len(logs["samples"]) == 3
    assert all(r["time_scope"] == "overlaps_window" for r in logs["log_counts"])
    assert all(r["count_semantics"] == "hourly_aggregate_not_sample_count" for r in logs["log_counts"])
    scopes = {r["line"].split(";")[-1]: r["time_scope"] for r in logs["samples"]}
    assert scopes["Query execution was interrupted"] == "outside_window"
    assert scopes["deadlock"] == "canonical_window"
    assert scopes["panic: timestamp unavailable"] == "unknown"
    hints = {h["id"]: h for h in ce.build_critical_evidence_hints({"logs": logs})}
    assert hints["panic_signal_present"]["time_scope"] == "context_only"
    assert hints["query_interruption_signal_present"]["time_scope"] == "context_only"
    assert hints["transaction_lock_signal_present"]["time_scope"] == "canonical_window"
    assert ce._panic_classification_context(logs, {}) is None
    text = ce._format_agent_text({"logs": logs, "critical_evidence_hints": list(hints.values())})
    assert "Hourly bucket totals, not exact canonical-window counts" in text
    assert "Context-only lead" in text
    assert logs["timeline"]["tidb"]["warn"][0]["window_end"] - logs["timeline"]["tidb"]["warn"][0]["window_start"] == 3600


def test_log_time_scope_keeps_in_window_evidence_and_fractional_boundaries():
    logs = ce._annotate_log_time_scope({
        "source": "loki", "canonical_window": {"start": "1700000100", "end": "1700000200"},
        "focused_log_context_window": {"start": "1700000090", "end": "1700000210"},
        "log_counts": [{"message": "Query execution was interrupted", "count": 3}],
        "samples": [
            {"timestamp_ns": "1700000200000000000", "line": "Query execution was interrupted"},
            {"timestamp_ns": "1700000200000000001", "line": "panic: boundary"},
            {"timestamp_ns": "", "line": "panic: undated"},
        ],
    })
    assert [r["time_scope"] for r in logs["samples"]] == ["canonical_window", "outside_window", "unknown"]
    assert logs["log_counts"][0]["time_scope"] == "overlaps_window"
    assert ce._panic_classification_context(logs, {}) is not None
    hints = {h["id"]: h for h in ce.build_critical_evidence_hints({"logs": logs})}
    assert hints["query_interruption_signal_present"]["time_scope"] == "canonical_window"
    assert hints["panic_signal_present"]["time_scope"] == "context_only"


def test_full_hour_aggregate_is_qualified_but_undated_sample_is_not():
    begin = ce._logstats_hour_epoch("20260801", "2")
    logs = ce._annotate_log_time_scope({
        "source": "clinic_logstats", "canonical_window": {"start": str(begin), "end": str(begin + 3600)},
        "log_counts": [{"date": "20260801", "hour": 2, "message": "deadlock", "count": 4}],
        "samples": [{"line": "deadlock"}],
    })
    assert logs["log_counts"][0]["time_scope"] == "canonical_window"
    assert logs["samples"][0]["time_scope"] == "unknown"
    assert ce.build_critical_evidence_hints({"logs": logs})[0]["time_scope"] == "canonical_window"


def test_collection_profiles_keep_full_as_default_and_bound_fast_work():
    assert ce.collect.__defaults__ == (None, "full")
    assert ce._collection_profile_flags("full") == (True, True)
    assert ce._collection_profile_flags("fast") == (False, False)

    with pytest.raises(ValueError, match="Unsupported collection profile"):
        ce._collection_profile_flags("unknown")


def test_collect_evidence_time_parser_accepts_epoch_and_iso_variants():
    expected_start = 1773029640
    assert ce._parse_time_to_epoch_seconds("1773029640") == expected_start
    assert ce._parse_time_to_epoch_seconds("2026-03-09T04:14:00Z") == expected_start
    assert ce._parse_time_to_epoch_seconds("2026-03-09 04:14:00+00:00") == expected_start


def test_collect_evidence_time_parser_rejects_datetime_without_timezone():
    with pytest.raises(ValueError, match="has no timezone"):
        ce._parse_time_to_epoch_seconds("2026-03-09 04:14:00")


def test_collect_evidence_time_parser_rejects_epoch_ms_and_ns():
    with pytest.raises(ValueError, match="10-digit epoch seconds"):
        ce._parse_time_to_epoch_seconds("1773029640000")
    with pytest.raises(ValueError, match="10-digit epoch seconds"):
        ce._parse_time_to_epoch_seconds("1773029640000000000")


def test_collect_evidence_canonical_time_window_records_inputs_and_utc_iso():
    window = ce._canonical_time_window(
        "2026-03-09T04:14:00Z",
        "2026-03-09 05:14:00+00:00",
    )

    assert window == {
        "start": "1773029640",
        "end": "1773033240",
        "start_iso": "2026-03-09T04:14:00Z",
        "end_iso": "2026-03-09T05:14:00Z",
        "input_start": "2026-03-09T04:14:00Z",
        "input_end": "2026-03-09 05:14:00+00:00",
    }


def test_collect_evidence_summary_metadata_includes_canonical_time_window():
    summary = ce.build_agent_summary(
        {"_input_time_range": ce._canonical_time_window("2026-03-09T04:14:00Z", "2026-03-09T05:14:00Z")},
        "infra",
        "hotspot",
        "cluster-1",
        "1773029640",
        "1773033240",
    )

    assert summary["metadata"]["time_range"] == {
        "start": "1773029640",
        "end": "1773033240",
        "start_iso": "2026-03-09T04:14:00Z",
        "end_iso": "2026-03-09T05:14:00Z",
        "input_start": "2026-03-09T04:14:00Z",
        "input_end": "2026-03-09T05:14:00Z",
    }
    assert summary["metadata"]["collection_profile"] == "full"
    text = ce._format_agent_text(summary)
    assert "# Time: 2026-03-09T04:14:00Z – 2026-03-09T05:14:00Z (epoch 1773029640 – 1773033240)" in text


def _series(instance, values, metric=None):
    labels = {"instance": instance}
    if metric:
        labels.update(metric)
    return {
        "metric": labels,
        "values": [[str(i), str(v)] for i, v in enumerate(values, start=1)],
    }


class FakeAPI:
    def __init__(self, resp):
        self.resp = resp
        self.queries = []

    def query_range(self, cluster_id, query, start, end):
        self.queries.append(query)
        return self.resp


class FakeSQLAPI:
    def __init__(self, responses):
        self.responses = list(responses)
        self.calls = []

    def query_sql(self, cluster_id, sql, timeout=60):
        self.calls.append({"cluster_id": cluster_id, "sql": sql, "timeout": timeout})
        if self.responses:
            return self.responses.pop(0)
        return {"columns": [], "rows": []}


class FakeBYOCCollectorAPI:
    def __init__(self):
        self.calls = []

    def get_cluster_by_id(self, cluster_id):
        self.calls.append(("metadata", cluster_id))
        return {
            "id": cluster_id,
            "orgId": "org-1",
            "deployType": "byoc",
            "diagnosticContext": {"parentId": "physical-1"},
        }

    def _get(self, path, params=None):
        self.calls.append(("get", path, params))
        if path == "/clinic/api/v1/dashboard/clusters":
            return {"items": []}
        if "/activities" in path:
            return {"activities": []}
        return {"status": "success", "data": {"result": []}}

    def query_range(self, cluster_id, query, start, end, step="60"):
        self.calls.append(("metrics", cluster_id, query))
        return {"status": "success", "data": {"result": []}}

    def query_range_join_instance_to_pod(
        self, cluster_id, query, namespace, pod_pattern, start, end, step="60",
    ):
        self.calls.append(("disk", cluster_id, namespace, pod_pattern))
        return {"status": "success", "data": {"result": []}}

    def query_exceeded_iops_by_pod(self, cluster_id, component, start, end, **kwargs):
        self.calls.append(("iops", cluster_id, component, kwargs.get("namespace")))
        return {"status": "success", "data": {"result": []}}

    def query_sql(self, *args, **kwargs):
        raise AssertionError("BYOC collection must not use warehouse SQL")

    def byoc_slow_query_list(self, cluster_id, start, end, **kwargs):
        self.calls.append(("slow-query", cluster_id, kwargs))
        return [
            {
                "digest": "a" * 64,
                "query": "select * from t",
                "query_time": 2.5,
                "memory_max": 1024,
                "timestamp": int(start),
                "connection_id": 1,
            }
        ]

    def byoc_statements_list(self, cluster_id, start, end, **kwargs):
        self.calls.append(("statements-list", cluster_id, kwargs))
        return [
            {
                "digest": "a" * 64,
                "stmt_type": "Select",
                "schema_name": "test",
                "normalized_sql": "select * from t",
                "exec_count": 25,
                "sum_latency": 5_000_000_000,
                "max_latency": 500_000_000,
                "max_mem": 2048,
            }
        ]

    def byoc_statements_available_fields(self, cluster_id):
        self.calls.append(("statements-fields", cluster_id))
        return [
            "digest", "stmt_type", "schema_name", "normalized_sql", "exec_count",
            "sum_latency", "max_latency", "max_mem",
        ]

    def byoc_continuous_profiling_group_profiles(self, cluster_id, start, end):
        self.calls.append(("profiles", cluster_id, start, end))
        return []

    def byoc_top_sql_instances(self, cluster_id, start, end):
        self.calls.append(("top-sql-instances", cluster_id))
        return {
            "status": "success",
            "data": {"tidb": ["tidb-0:4000"], "tikv": ["tikv-0:20160"]},
        }

    def byoc_top_sql_summary(
        self, cluster_id, start, end, window, top, instance, instance_type,
    ):
        self.calls.append(("top-sql-summary", cluster_id, instance, instance_type))
        return {
            "status": "success",
            "data": [
                {
                    "sql_digest": "a" * 64,
                    "normalized_sql": "select * from t",
                    "cpu_time_ms": 250,
                    "cpu_pct": 75,
                }
            ],
        }

    def byoc_statements_plans(self, cluster_id, start, end, digest, schema_name=None):
        self.calls.append(("statements-plans", cluster_id, digest))
        return [{"plan_digest": "b" * 64, "exec_count": 3}]

    def byoc_statements_plan_detail(
        self, cluster_id, start, end, digest, plans=None, schema_name=None,
    ):
        self.calls.append(("statements-plan-detail", cluster_id, digest, plans))
        return {
            "digest": digest,
            "plan_digest": "b" * 64,
            "binary_plan_text": "TableReader_1 root",
            "query": "select * from t",
        }

    def logs_query_range(
        self, cluster_id, query, start, end, limit=100, direction="backward", step=None,
    ):
        self.calls.append(("logs", cluster_id, query, limit, direction, step))
        return {"status": "success", "data": {"result": []}}


def _cpu_data():
    return {
        "cpu": [
            _series("db-tikv-0", [1.2, 3.4]),
            _series("db-tikv-1", [0.8, 1.0]),
        ],
        "cpu_quota": [
            _series("db-tikv-0", [4.0, 4.0]),
            _series("db-tikv-1", [4.0, 4.0]),
        ],
    }


def test_fetch_cluster_meta_prefers_canonical_byoc_routing_fields():
    api = FakeBYOCCollectorAPI()

    meta = ce._fetch_cluster_meta(api, "logical-1")

    assert meta["deployType"] == "byoc"
    assert meta["orgID"] == "org-1"
    assert meta["parent_id"] == "physical-1"
    assert meta["physical_cluster_id"] == "physical-1"
    assert ce._is_byoc_cluster_meta(meta)


def test_byoc_response_normalizers_preserve_collector_contract():
    slow = ce._byoc_slow_query_table([
        {"digest": "d1", "query": "select 1", "query_time": 1, "memory_max": 10},
        {"digest": "d1", "query": "select 1", "query_time": 3, "memory_max": 20},
    ])
    slow_row = dict(zip(slow["columns"], slow["rows"][0]))
    assert slow["ranking"] == "query_time_desc"
    assert slow["scope"] == "latency_ranked_sample"
    assert slow_row["sample_count"] == 2
    assert slow_row["sample_avg_time"] == 2
    assert "exec_count" not in slow_row
    assert "avg_time" not in slow_row
    assert slow_row["max_time"] == 3
    assert slow_row["max_mem"] == 20

    statements = ce._byoc_statement_summary_table([
        {
            "digest": "d1",
            "stmt_type": "Select",
            "schema_name": "test",
            "normalized_sql": "select 1",
            "exec_count": 2,
            "sum_latency": 2_000_000_000,
            "max_latency": 1_500_000_000,
            "max_mem": 10,
        },
        {
            "digest": "d1",
            "stmt_type": "Select",
            "schema_name": "test",
            "normalized_sql": "select 1",
            "exec_count": 3,
            "sum_latency": 6_000_000_000,
            "max_latency": 2_500_000_000,
            "max_mem": 20,
        },
    ])
    statement_row = dict(zip(statements["columns"], statements["rows"][0]))
    assert statement_row["exec_count"] == 5
    assert statement_row["avg_time"] == 1.6
    assert statement_row["max_time"] == 2.5
    assert statement_row["max_mem"] == 20

    top = ce._byoc_top_sql_table([
        ("tikv-0", {"data": [{"digest": "d1", "normalized_sql": "select 1", "cpuTimeMs": 8}]})
    ])
    top_row = dict(zip(top["columns"], top["rows"][0]))
    assert top_row["instance"] == "tikv-0"
    assert top_row["sql_digest"] == "d1"
    assert top_row["sample_sql"] == "select 1"
    assert top_row["total_cpu_ms"] == 8


def test_byoc_statement_summary_keeps_same_digest_per_schema():
    table = ce._byoc_statement_summary_table([
        {"digest": "d1", "schema_name": "sales", "exec_count": 10},
        {"digest": "d1", "schema_name": "analytics", "exec_count": 20},
    ])
    rows = [dict(zip(table["columns"], row)) for row in table["rows"]]
    assert [(row["schema_name"], row["exec_count"]) for row in rows] == [
        ("analytics", 20), ("sales", 10),
    ]


def test_collect_byoc_routes_to_o11y_and_targeted_statements(monkeypatch, tmp_path):
    monkeypatch.setenv("COLLECT_EVIDENCE_NO_RAW", "1")
    monkeypatch.setenv("CLINIC_PERF_LOG", str(tmp_path / "perf.jsonl"))
    api = FakeBYOCCollectorAPI()

    data = ce.collect(api, "logical-1", "100", "200", profile="full")

    assert data["is_byoc"] is True
    assert data["parent_id"] == "physical-1"
    assert data["data_plane_namespace"] == "tidbphysical-1"
    assert data["component_health_scope"] == "physical-1"
    assert any(call[:3] == ("iops", "logical-1", "tikv") and call[3] == "tidbphysical-1" for call in api.calls)
    assert ("disk", "logical-1", "tidbphysical-1", ".*(tikv|tiflash).*") in api.calls
    assert any(call[0] == "slow-query" for call in api.calls)
    assert any(call[0] == "statements-list" for call in api.calls)
    assert any(call[0] == "top-sql-summary" for call in api.calls)
    assert any(call[0] == "statements-plans" for call in api.calls)
    assert any(call[0] == "statements-plan-detail" for call in api.calls)
    assert data["top_plan_source"] == "topsql_tikv_global"
    assert data["statements"][0]["operation"] == "plans"
    assert data["statements"][1]["operation"] == "plan-detail"
    top_plan = dict(zip(data["top_plan"]["columns"], data["top_plan"]["rows"][0]))
    assert top_plan["decoded_plan"] == "TableReader_1 root"
    assert data["log_stats"]["warehouse_logstats"]["status"] == "unsupported"

    summary = ce.build_agent_summary(
        data, "INFRA CLEAR", "HOTSPOT CLEAR", "logical-1", "100", "200",
    )
    assert summary["queries"]["statements"][0]["source"] == "statement"
    run_dir = ce.save_evidence_files(
        data, "INFRA CLEAR", "HOTSPOT CLEAR", "logical-1", "100", "200", str(tmp_path),
    )
    queries = json.loads((Path(run_dir) / "queries.json").read_text())
    assert [entry["operation"] for entry in queries["statements"]] == [
        "plans", "plan-detail",
    ]
    assert all(entry["source"] == "statement" for entry in queries["statements"])
    assert queries["statement_summary"][0]["exec_count"] == 25
    assert queries["statement_summary"][0]["source"] == "statement_summary"
    assert queries["slow_queries"][0]["scope"] == "latency_ranked_sample"
    assert queries["slow_queries"][0]["sample_count"] == 1
    assert "exec_count" not in queries["slow_queries"][0]


def test_byoc_plan_fallback_uses_full_window_statement_summary(monkeypatch, tmp_path):
    monkeypatch.setenv("COLLECT_EVIDENCE_NO_RAW", "1")
    monkeypatch.setenv("CLINIC_PERF_LOG", str(tmp_path / "perf.jsonl"))

    class StatementRankedAPI(FakeBYOCCollectorAPI):
        def byoc_slow_query_list(self, cluster_id, start, end, **kwargs):
            self.calls.append(("slow-query", cluster_id, kwargs))
            return [
                {"digest": "a" * 64, "query": "select slow", "query_time": 20},
                {"digest": "a" * 64, "query": "select slow", "query_time": 19},
            ]

        def byoc_statements_list(self, cluster_id, start, end, **kwargs):
            self.calls.append(("statements-list", cluster_id, kwargs))
            return [
                {
                    "digest": "b" * 64,
                    "normalized_sql": "select frequent",
                    "exec_count": 10_000,
                    "sum_latency": 50_000_000_000,
                    "max_latency": 100_000_000,
                }
            ]

        def byoc_top_sql_instances(self, cluster_id, start, end):
            self.calls.append(("top-sql-instances", cluster_id))
            return {"data": {"tidb": [], "tikv": []}}

    api = StatementRankedAPI()
    data = ce.collect(api, "logical-1", "100", "200", profile="full")

    assert data["top_plan_digest"] == "b" * 64
    assert data["top_plan_source"] == "statement_summary"
    assert any(
        call[0] == "statements-plans" and call[2] == "b" * 64
        for call in api.calls
    )
    assert not any(
        call[0] == "statements-plans" and call[2] == "a" * 64
        for call in api.calls
    )


def test_byoc_plan_selection_uses_highest_plan_exec_count(monkeypatch, tmp_path):
    monkeypatch.setenv("COLLECT_EVIDENCE_NO_RAW", "1")
    monkeypatch.setenv("CLINIC_PERF_LOG", str(tmp_path / "perf.jsonl"))

    class MultiplePlansAPI(FakeBYOCCollectorAPI):
        def byoc_statements_plans(self, cluster_id, start, end, digest, schema_name=None):
            self.calls.append(("statements-plans", cluster_id, digest, schema_name))
            return [
                {"plan_digest": "low", "exec_count": 1},
                {"plan_digest": "high", "exec_count": 9},
            ]

        def byoc_statements_plan_detail(
            self, cluster_id, start, end, digest, plans=None, schema_name=None,
        ):
            self.calls.append(("statements-plan-detail", cluster_id, digest, plans, schema_name))
            return {"digest": digest, "plan_digest": plans[0] if plans else ""}

    api = MultiplePlansAPI()
    data = ce.collect(api, "logical-1", "100", "200", profile="full")
    detail_call = next(call for call in api.calls if call[0] == "statements-plan-detail")
    assert detail_call[3] == ["high"]
    assert detail_call[4] == "test"
    assert data["top_plan_source"] == "topsql_tikv_global"


def test_byoc_slow_query_report_is_labeled_as_latency_ranked_sample():
    slow = ce._byoc_slow_query_table([
        {"digest": "d1", "query": "select 1", "query_time": 3},
        {"digest": "d1", "query": "select 1", "query_time": 1},
    ])

    report = ce.format_summary(
        {"is_byoc": True, "slow": slow},
        "INFRA CLEAR",
        "HOTSPOT CLEAR",
        "logical-1",
        "100",
        "200",
    )

    assert "Slow Queries (latency-ranked sample, limit=100)" in report
    assert "sample_count=2" in report
    assert "sample_avg_time=2.00s" in report
    assert "Slow Queries (top by exec_count)" not in report


def test_byoc_does_not_use_latency_sample_count_for_plan_fallback(monkeypatch, tmp_path):
    monkeypatch.setenv("COLLECT_EVIDENCE_NO_RAW", "1")
    monkeypatch.setenv("CLINIC_PERF_LOG", str(tmp_path / "perf.jsonl"))

    class SampleOnlyAPI(FakeBYOCCollectorAPI):
        def byoc_top_sql_instances(self, cluster_id, start, end):
            return {"data": {"tidb": [], "tikv": []}}

        def byoc_statements_list(self, cluster_id, start, end, **kwargs):
            self.calls.append(("statements-list", cluster_id, kwargs))
            return []

    api = SampleOnlyAPI()
    data = ce.collect(api, "logical-1", "100", "200", profile="full")

    assert data.get("top_plan_digest", "") == ""
    assert data.get("top_plan_source", "") == ""
    assert not any(call[0] == "statements-plans" for call in api.calls)


def test_query_sql_with_retries_returns_second_success_without_final_error():
    api = FakeSQLAPI([
        {"error": "context canceled"},
        {"columns": ["digest"], "rows": [["abc"]]},
    ])

    resp, attempts = ce._query_sql_with_retries(
        api, "c1", "SELECT * FROM clinic_data_proxy.topsql", attempts=2, retry_delay=0,
    )

    assert resp == {"columns": ["digest"], "rows": [["abc"]]}
    assert len(api.calls) == 2
    assert len(attempts) == 2
    assert attempts[0]["error"] == "context canceled"
    assert attempts[1]["rows"] == 1


def test_build_tikv_instance_topsql_sql_scopes_to_hot_instance():
    sql = ce._build_tikv_instance_topsql_sql(
        "2026-05-11", "1778481420", "1778488620",
        "db-tikv-79.db-tikv-peer.tidb1.svc:20160",
    )

    assert "component = 'tikv'" in sql
    assert "instance LIKE '%db-tikv-79.db-tikv-peer.tidb1.svc:20160%'" in sql
    assert "SUM(topsql_cpu_time_ms)" in sql
    assert "AS cpu_pct" in sql


def test_detect_hotspot_ignores_global_tikv_topsql_for_query_attribution():
    data = {
        "topsql_tikv": {
            "columns": ["instance", "sql_digest", "total_cpu_ms"],
            "rows": [
                ["db-tikv-41", "digest-global", "900"],
                ["db-tikv-41", "digest-small", "100"],
            ],
        },
        "topsql_tikv_hot_instances": [],
    }

    verdict = ce.detect_hotspot(data)

    assert "Query hotspot" not in verdict


def test_detect_hotspot_uses_hot_instance_topsql_for_query_attribution():
    data = {
        "topsql_tikv": {
            "columns": ["instance", "sql_digest", "total_cpu_ms"],
            "rows": [
                ["db-tikv-41", "digest-global", "900"],
                ["db-tikv-41", "digest-small", "100"],
            ],
        },
        "topsql_tikv_hot_instances": [
            {
                "instance": "db-tikv-79",
                "columns": ["instance", "sql_digest", "total_cpu_ms", "cpu_pct"],
                "rows": [
                    ["db-tikv-79.db-tikv-peer.tidb1.svc:20160", "digest-hot", "700", "70.0"],
                    ["db-tikv-79.db-tikv-peer.tidb1.svc:20160", "digest-other", "300", "30.0"],
                ],
            }
        ],
    }

    verdict = ce.detect_hotspot(data)

    assert "Query hotspot: digest=digest-hot cpu_pct=70.0%" in verdict
    assert "db-tikv-79.db-tikv-peer.tidb1.svc:20160" in verdict


def test_collect_tikv_thread_cpu_triggers_for_saturated_tikv():
    api = FakeAPI({
        "data": {
            "result": [
                _series("db-tikv-0", [0.3, 0.8], {"name": "grpc-server"}),
                _series("db-tikv-0", [0.2, 0.4], {"name": "raftstore"}),
            ]
        }
    })

    result = ce.collect_tikv_thread_cpu(api, "c1", "100", "200", _cpu_data())

    assert result["status"] == "collected"
    assert result["selected_instances"][0]["instance"] == "db-tikv-0"
    assert result["threads"][0]["bucket"] == "grpc"
    assert "tikv_thread_cpu_seconds_total" in api.queries[0]
    assert 'instance=~"db-tikv-0"' in api.queries[0]


def test_collect_tikv_thread_cpu_api_error_fails_strictly():
    api = FakeAPI({"error": "upstream unavailable"})

    with pytest.raises(RuntimeError, match="TiKV_THREAD_CPU: upstream unavailable"):
        ce.collect_tikv_thread_cpu(api, "c1", "100", "200", _cpu_data())


def test_collect_tikv_thread_cpu_empty_result_fails_strictly():
    api = FakeAPI({"data": {"result": []}})

    with pytest.raises(RuntimeError, match="TiKV_THREAD_CPU: empty result"):
        ce.collect_tikv_thread_cpu(api, "c1", "100", "200", _cpu_data())


def test_build_evidence_data_includes_database_time_and_read_write_path_metrics():
    data = {
        "connection_idle_p99": [_series("db-tidb-0", [0.01, 0.02], {"in_txn": "1"})],
        "compile_p99": [_series("db-tidb-0", [0.03], {"sql_type": "Select"})],
        "kv_request_p99": [_series("db-tidb-0", [0.04], {"type": "Cop", "store": "1"})],
        "scheduler_command_p99": [_series("db-tikv-0", [0.05], {"type": "prewrite"})],
        "raft_commit_p99": [_series("db-tikv-0", [0.06])],
        "pending_compaction_bytes": [_series("db-tikv-0", [1024], {"db": "kv"})],
        "tidb_qps": [_series("db-tidb-0", [100], {"result": "OK", "type": "Query"})],
        "component_memory": [_series("db-tikv-0", [2048], {"component": "tikv"})],
    }

    metrics, _, _, _, _ = ce._build_evidence_data(data, "c1", "100", "200")
    group_ids = {row.get("metric_group_id") for row in metrics}

    assert "tidb_connection_idle_p99" in group_ids
    assert "tidb_compile_p99" in group_ids
    assert "tidb_kv_request_p99" in group_ids
    assert "tikv_scheduler_command_p99" in group_ids
    assert "tikv_raft_commit_log_p99" in group_ids
    assert "tikv_pending_compaction_bytes" in group_ids
    assert "tidb_qps" in group_ids
    assert "component_resident_memory" in group_ids
    kv_row = next(row for row in metrics if row.get("metric_group_id") == "tidb_kv_request_p99" and "store=1" in row["instance"])
    assert "type=Cop" in kv_row["instance"]


def test_save_evidence_files_writes_tikv_thread_cpu_sidecar(tmp_path):
    data = {
        "errors": [],
        "tikv_thread_cpu": {
            "status": "collected",
            "selected_instances": [{"instance": "db-tikv-0"}],
            "promql": "sum by (instance, name) (rate(tikv_thread_cpu_seconds_total[1m]))",
            "threads": [{"instance": "db-tikv-0", "name": "grpc-server", "bucket": "grpc", "peak": 0.8}],
            "buckets": [{"instance": "db-tikv-0", "bucket": "grpc", "peak": 0.8}],
        },
    }

    ce.save_evidence_files(data, "INFRA CLEAR", "HOTSPOT DETECTED", "c1", "1713139200", "1713142800", str(tmp_path))

    sidecar = tmp_path / "tikv_thread_cpu.json"
    assert sidecar.exists()
    assert "tikv_thread_cpu_seconds_total" in sidecar.read_text()
    report = json.loads((tmp_path / "workspace_evidence_report.json").read_text())
    thread_metrics = [
        item
        for item in report["evidence"]["metrics"]
        if item.get("name") == "TiKV Thread CPU"
    ]
    assert thread_metrics
    assert all(item.get("metric_group_id") == "tikv_thread_cpu" for item in thread_metrics)


def test_compact_metric_for_ai_strips_raw_chart_series():
    metric = {
        "name": "CPU Usage",
        "component": "tidb",
        "metric_group_id": "tidb_cpu:db",
        "instance": "db-tidb-0",
        "peak": 3.5,
        "mean": 2.1,
        "quota": 4,
        "utilization": "88%",
        "unit": "cores",
        "severity": "critical",
        "promql": "rate(process_cpu_seconds_total[1m])",
        "link": "https://clinic.pingcap.com/clusters/c1/monitor",
        "chart_data": {
            "name": "CPU Usage",
            "promql": "rate(process_cpu_seconds_total[1m])",
            "series": [{"instance": "db-tidb-0", "values": [["1", "1.0"]]}],
        },
    }

    compact = ce.compact_metric_for_ai(metric)

    assert compact == {
        "name": "CPU Usage",
        "component": "tidb",
        "metric_group_id": "tidb_cpu:db",
        "instance": "db-tidb-0",
        "peak": 3.5,
        "mean": 2.1,
        "quota": 4,
        "utilization": "88%",
        "unit": "cores",
        "severity": "critical",
        "promql": "rate(process_cpu_seconds_total[1m])",
        "link": "https://clinic.pingcap.com/clusters/c1/monitor",
    }
    assert "chart_data" not in compact
    assert "chart_data_ref" not in compact


def test_save_evidence_files_splits_tidb_and_tikv_cpu_metrics(tmp_path):
    data = {
        "errors": [],
        "tidb_cpu": [_series("db-tidb-0", [1.0, 3.5], {"component": "tidb", "cluster": "db"})],
        "tikv_cpu": [_series("db-tikv-0", [2.0, 4.5], {"component": "tikv", "cluster": "db"})],
        "cpu": [
            _series("db-tidb-0", [1.0, 3.5], {"component": "tidb", "cluster": "db"}),
            _series("db-tikv-0", [2.0, 4.5], {"component": "tikv", "cluster": "db"}),
        ],
        "cpu_quota": [
            _series("db-tidb-0", [4.0, 4.0]),
            _series("db-tikv-0", [8.0, 8.0]),
        ],
    }

    ce.save_evidence_files(data, "INFRA CLEAR", "HOTSPOT CLEAR", "c1", "1713139200", "1713142800", str(tmp_path))

    run_dir = tmp_path
    raw_metrics = json.loads((run_dir / "metrics.json").read_text())["metrics"]
    agent_metrics = json.loads((run_dir / "evidence_summary_for_agent.json").read_text())["metrics"]

    tidb_cpu = next(m for m in raw_metrics if m.get("name") == "TIDB CPU" and m.get("instance") == "db-tidb-0")
    tikv_cpu = next(m for m in raw_metrics if m.get("name") == "TIKV CPU" and m.get("instance") == "db-tikv-0")

    assert tidb_cpu["promql"] == 'rate(process_cpu_seconds_total{component="tidb"}[1m])'
    assert tikv_cpu["promql"] == 'rate(process_cpu_seconds_total{component="tikv"}[1m])'
    assert tidb_cpu["metric_group_id"] == "tidb_cpu:db"
    assert tikv_cpu["metric_group_id"] == "tikv_cpu:db"
    assert tidb_cpu["component"] == "tidb"
    assert tikv_cpu["component"] == "tikv"
    assert tidb_cpu["chart_data"]["promql"] != tikv_cpu["chart_data"]["promql"]
    assert tidb_cpu["chart_data_ref"]["cluster_id"] == "c1"
    assert tidb_cpu["chart_data_ref"]["start"] == "1713139200"
    assert tidb_cpu["chart_data_ref"]["end"] == "1713142800"

    compact_tidb = next(m for m in agent_metrics if m.get("name") == "TIDB CPU" and m.get("instance") == "db-tidb-0")
    compact_tikv = next(m for m in agent_metrics if m.get("name") == "TIKV CPU" and m.get("instance") == "db-tikv-0")
    assert compact_tidb["metric_group_id"] == "tidb_cpu:db"
    assert compact_tikv["metric_group_id"] == "tikv_cpu:db"


def test_save_evidence_files_merges_existing_workspace_metrics(tmp_path):
    (tmp_path / "metrics.json").write_text(
        json.dumps({
            "metrics": [
                {
                    "name": "Existing CLI Metric",
                    "metric_group_id": "existing_cli_metric",
                    "instance": "db-tidb-0",
                    "peak": 1.0,
                }
            ]
        }),
        encoding="utf-8",
    )
    data = {
        "errors": [],
        "cpu": [_series("db-tidb-0", [1.0, 3.5])],
        "cpu_quota": [_series("db-tidb-0", [4.0, 4.0])],
    }

    ce.save_evidence_files(data, "INFRA CLEAR", "HOTSPOT CLEAR", "c1", "1713139200", "1713142800", str(tmp_path))

    metrics = json.loads((tmp_path / "metrics.json").read_text())["metrics"]
    assert any(m.get("metric_group_id") == "existing_cli_metric" for m in metrics)
    assert any(m.get("name") == "UNKNOWN CPU" for m in metrics)


def test_topsql_url_uses_ngm_dashboard_format_when_metadata_available():
    link = ce._topsql_url(
        "10989049060142230334",
        "1776301644",
        "1776308844",
        {
            "clusterProviderName": "AWS",
            "clusterRegionName": "us-east-1",
            "tenantID": "1372813089209061633",
            "projectID": "1372813089454544954",
            "clusterID": "10989049060142230334",
            "clusterDeployType": "DEDICATED",
            "orgName": "Buffet's Org",
            "clusterName": "prod-us-saas-gosaas-region-tidb-v85-01",
            "clusterVersion": "v8.5.4",
        },
        "db-2006050899364941824-4af81c75-tidb-10.db-2006050899364941824-4af81c75-tidb-peer.tidb10989049060142230334.svc:10080",
    )

    assert link.startswith("https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html?")
    assert "provider=aws" in link
    assert "region=us-east-1" in link
    assert "orgName=Buffet%27s%20Org" in link
    assert "orgId=1372813089209061633" in link
    assert "projectId=1372813089454544954" in link
    assert "clusterId=10989049060142230334" in link
    assert "deployType=dedicated" in link
    assert "version=v8.5.4" in link
    assert "#/topsql?" in link
    assert "instance=db-2006050899364941824-4af81c75-tidb-10" in link
    assert "%3A10080" in link
    assert "from=1776301644" in link
    assert "to=1776308844" in link


def test_topsql_url_uses_ngm_dashboard_format_when_metadata_missing():
    assert ce._topsql_url("c1", "100", "200") == (
        "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html?clusterId=c1#/topsql?from=100&to=200"
    )


def test_build_evidence_data_topsql_link_includes_tidb_instance():
    data = {
        "cluster_meta": {
            "provider": "gcp",
            "region": "asia-northeast1",
            "tenantID": "1372813089201471287",
            "projectID": "1372813089199431289",
            "clusterName": "main",
            "deployType": "dedicated",
            "clusterVersion": "v8.5.3",
        },
        "topsql_tidb": {
            "columns": ["instance", "sql_digest", "sample_sql", "total_cpu_ms", "cpu_pct"],
            "rows": [[
                "db-tidb-1120.db-tidb-peer.tidb10123293495242577103.svc:10080",
                "digest-1",
                "SELECT 1",
                1234,
                91.2,
            ]],
        },
        "topsql_tikv": {"columns": [], "rows": []},
    }

    _, _, _, topsql, _ = ce._build_evidence_data(data, "10123293495242577103", "1800", "now")

    assert topsql[0]["instance"] == "db-tidb-1120.db-tidb-peer.tidb10123293495242577103.svc:10080"
    assert "instance=db-tidb-1120.db-tidb-peer.tidb10123293495242577103.svc%3A10080" in topsql[0]["link"]
    assert "from=1800" in topsql[0]["link"]
    assert "to=now" in topsql[0]["link"]


def test_slow_query_url_uses_ngm_list_format_with_digest():
    link = ce._slow_query_url(
        "10989049060142230334",
        "1776301644",
        "1776308844",
        {
            "provider": "aws",
            "region": "us-east-1",
            "tenantID": "1372813089209061633",
            "projectID": "1372813089454544954",
            "clusterName": "prod-us-saas-gosaas-region-tidb-v85-01",
            "deployType": "dedicated",
        },
        "adf0384ee3b7b8ce35f0e5c8124b07d2d9864bcee03b0f6c04f1e0ac8e89e711",
    )

    assert link.startswith("https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html?")
    assert "provider=aws" in link
    assert "region=us-east-1" in link
    assert "orgId=1372813089209061633" in link
    assert "projectId=1372813089454544954" in link
    assert "clusterId=10989049060142230334" in link
    assert "clusterName=prod-us-saas-gosaas-region-tidb-v85-01" in link
    assert "deployType=dedicated" in link
    assert "#/slow_query?" in link
    assert "digest=adf0384ee3b7b8ce35f0e5c8124b07d2d9864bcee03b0f6c04f1e0ac8e89e711" in link
    assert "from=1776301644" in link
    assert "to=1776308844" in link
    assert "show_internal=true" in link
    assert "term=" in link
    assert "connection_id" not in link
    assert "timestamp" not in link


def test_slow_query_url_uses_ngm_list_format_when_metadata_missing():
    assert ce._slow_query_url("c1", "100", "200") == (
        "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html?clusterId=c1#/slow_query?from=100&to=200&show_internal=true&term="
    )


def test_loki_log_url_uses_millisecond_timestamps():
    url = ce._loki_log_url("c1", '{component="tidb"}', "1776301644", "1776308844")
    assert "grafana/explore" in url
    # range timestamps must be milliseconds (13 digits), not seconds (10 digits)
    assert "1776301644000" in url
    assert "1776308844000" in url
    assert "1776301644%22" not in url  # seconds-only value must not appear quoted


def test_loki_log_url_derives_datasource_from_cluster_meta():
    meta = {"provider": "aws", "region": "ap-southeast-1", "deployType": "premium"}
    url = ce._loki_log_url("c1", '{component="tidb"}', "1776301644", "1776308844", meta)
    assert "Loki-aws-ap-southeast-1-premium" in url


def test_loki_log_url_datasource_without_deploy_type():
    meta = {"provider": "alicloud", "region": "ap-southeast-1", "deployType": ""}
    url = ce._loki_log_url("c1", '{component="tidb"}', "1776301644", "1776308844", meta)
    assert "Loki-alicloud-ap-southeast-1" in url
    assert "Loki-alicloud-ap-southeast-1-" not in url


def test_loki_log_url_falls_back_to_uid_when_no_meta():
    url = ce._loki_log_url("c1", '{component="tidb"}', "1776301644", "1776308844", None)
    assert "PF9DF9566127982DF" in url


def test_collect_log_stats_double_escapes_info_regex_for_logql(monkeypatch):
    captured_queries = []

    def fake_query_loki(api, cluster_id, query, start, end, step="300"):
        captured_queries.append(query)
        return {"data": {"result": []}}

    monkeypatch.setattr(ce, "_query_loki", fake_query_loki)
    monkeypatch.setattr(ce, "_query_loki_logs", lambda *args, **kwargs: {"data": {"result": []}})

    payload = ce.collect_log_stats(api=None, cluster_id="c1", start="100", end="200")

    info_queries = [query for query in captured_queries if "INFO" in query]
    assert len(captured_queries) == 9
    assert len(info_queries) == 3
    assert all("\\\\[INFO\\\\]" in query for query in info_queries)
    assert all("\\[INFO\\]" not in query.replace("\\\\[INFO\\\\]", "") for query in info_queries)
    assert payload["errors"] == []
    assert payload["log_counts"] == []
    assert payload["samples"] == []


def test_query_loki_metric_uses_step_without_stream_parameters():
    calls = []

    class RecordingAPI:
        def logs_query_range(
            self,
            cluster_id,
            query,
            start,
            end,
            limit=100,
            direction="backward",
            step=None,
        ):
            calls.append({
                "cluster_id": cluster_id,
                "query": query,
                "start": start,
                "end": end,
                "limit": limit,
                "direction": direction,
                "step": step,
            })
            return {"status": "success", "data": {"result": []}}

    ce._query_loki(
        RecordingAPI(),
        "byoc-1",
        'sum(count_over_time({container="tidb"}[300s]))',
        "100",
        "200",
        "300",
    )

    assert calls == [{
        "cluster_id": "byoc-1",
        "query": 'sum(count_over_time({container="tidb"}[300s]))',
        "start": "100000000000",
        "end": "200000000000",
        "limit": None,
        "direction": None,
        "step": "300",
    }]


def test_collect_log_stats_byoc_uses_container_timeline_selectors(monkeypatch):
    captured_queries = []

    def fake_query_loki(api, cluster_id, query, start, end, step="300"):
        captured_queries.append(query)
        return {"data": {"result": []}}

    monkeypatch.setattr(ce, "_query_loki", fake_query_loki)
    monkeypatch.setattr(
        ce,
        "_collect_focused_log_samples",
        lambda *args, **kwargs: ([], [], None),
    )

    payload = ce.collect_log_stats(
        api=object(),
        cluster_id="byoc-1",
        start="100",
        end="200",
        cluster_meta={"deployType": "byoc"},
    )

    assert len(captured_queries) == 9
    assert all('{container="' in query for query in captured_queries)
    assert all('{component="' not in query for query in captured_queries)
    assert payload["step_seconds"] == 300
    assert payload["warehouse_logstats"]["status"] == "unsupported"


def test_collect_log_stats_uses_clinic_logstats_without_loki(monkeypatch):
    monkeypatch.setenv("INVESTIBOOK_LOGSTATS_BACKEND", "athena")

    def fail_loki(*args, **kwargs):
        raise AssertionError("Loki should not be queried when Clinic LogStats succeeds")

    class FakeLogStatsAPI:
        def __init__(self):
            self.calls = []

        def query_sql(self, cluster_id, sql, timeout=60):
            self.calls.append((cluster_id, sql, timeout))
            assert cluster_id == "105"
            assert "FROM log_stats" in sql
            assert "cluster = '105'" in sql
            assert "org = '137'" in sql
            assert "date = 20260419" in sql
            if "msg_digest" in sql:
                return {
                    "columns": ["date", "hour", "component", "level", "count", "msg_digest", "sample"],
                    "rows": [[
                        "20260419",
                        "1",
                        "tidb",
                        "WARN",
                        "1",
                        '["kill initiated"] [reason="[executor:?]Query execution was interrupted"]',
                        (
                            '2026-04-19T01:09:21.123Z;tidb;warn;db-tidb-57;'
                            '[WARN] [sqlkiller.go:61] ["kill initiated"]'
                        ),
                    ]],
                }
            if "msg_digest" not in sql:
                return {
                    "columns": ["date", "hour", "component", "level", "count"],
                    "rows": [
                        ["20260419", "1", "tidb", "WARN", "2"],
                        ["20260419", "1", "tidb", "ERROR", "1"],
                    ],
                }
            raise AssertionError(f"unexpected sql {sql}")

    monkeypatch.setattr(ce, "_query_loki", fail_loki)
    monkeypatch.setattr(ce, "_query_loki_logs", fail_loki)
    api = FakeLogStatsAPI()

    payload = ce.collect_log_stats(
        api=api,
        cluster_id="105",
        start="1776559200",
        end="1776562800",
        cluster_meta={"provider": "aws", "region": "us-east-1", "tenantID": "137"},
    )

    assert len(api.calls) == 2
    assert payload["source"] == "clinic_logstats"
    assert payload["step_seconds"] == 3600
    assert payload["summary"]["tidb"]["warn"] == 2
    assert payload["summary"]["tidb"]["error"] == 1
    assert payload["log_counts"][0]["source"] == "clinic_logstats"
    assert "Query execution was interrupted" in payload["log_counts"][0]["message"]
    assert payload["samples"][0]["instance"] == "db-tidb-57"
    assert payload["samples"][0]["timestamp_ns"].startswith("1776560961")


def test_collect_log_stats_forced_backend_allows_non_aws_region(monkeypatch):
    monkeypatch.setenv("INVESTIBOOK_LOGSTATS_BACKEND", "athena")

    def fail_loki(*args, **kwargs):
        raise AssertionError("Loki should not be queried when forced Clinic LogStats succeeds")

    class FakeLogStatsAPI:
        def query_sql(self, cluster_id, sql, timeout=60):
            assert "FROM log_stats" in sql
            assert "cluster = '107'" in sql
            if "msg_digest" not in sql:
                return {
                    "columns": ["date", "hour", "component", "level", "count"],
                    "rows": [["20260401", "11", "tidb", "ERROR", "1"]],
                }
            if "msg_digest" in sql:
                return {"columns": ["date", "hour", "component", "level", "count", "msg_digest", "sample"], "rows": []}
            raise AssertionError(f"unexpected sql {sql}")

    monkeypatch.setattr(ce, "_query_loki", fail_loki)
    monkeypatch.setattr(ce, "_query_loki_logs", fail_loki)

    payload = ce.collect_log_stats(
        api=FakeLogStatsAPI(),
        cluster_id="107",
        start="1775039760",
        end="1775043360",
        cluster_meta={"provider": "gcp", "region": "asia-northeast1", "tenantID": "137"},
    )

    assert payload["source"] == "clinic_logstats"
    assert payload["summary"]["tidb"]["error"] == 1


def test_collect_log_stats_falls_back_when_clinic_logstats_is_empty(monkeypatch):
    monkeypatch.setenv("INVESTIBOOK_LOGSTATS_BACKEND", "athena")
    loki_log_queries = []

    class FakeLogStatsAPI:
        def query_sql(self, cluster_id, sql, timeout=60):
            assert cluster_id == "105"
            if "msg_digest" in sql:
                return {"columns": ["date", "hour", "component", "level", "count", "msg_digest", "sample"], "rows": []}
            return {"columns": ["date", "hour", "component", "level", "count"], "rows": []}

    def fake_query_loki(*args, **kwargs):
        return {"data": {"result": []}}

    def fake_query_loki_logs(api, cluster_id, query, start, end, limit):
        loki_log_queries.append(query)
        if "deadlock" not in query:
            return {"data": {"result": []}}
        return {
            "data": {
                "result": [
                    {
                        "stream": {"component": "tidb", "pod": "tidb-0"},
                        "values": [
                            [
                                "1776635400323971241",
                                '[ERROR] [session.go:2392] ["run statement failed"] '
                                '[error="[executor:1213]Deadlock found when trying to get lock; try restarting transaction"]',
                            ]
                        ],
                    }
                ]
            }
        }

    monkeypatch.setattr(ce, "_query_loki", fake_query_loki)
    monkeypatch.setattr(ce, "_query_loki_logs", fake_query_loki_logs)

    payload = ce.collect_log_stats(
        api=FakeLogStatsAPI(),
        cluster_id="105",
        start="1776631800",
        end="1776639000",
        cluster_meta={"provider": "aws", "region": "us-east-1", "tenantID": "137"},
    )

    assert payload["source"] == "loki"
    assert any("deadlock" in query for query in loki_log_queries)
    assert payload["log_counts"]
    assert "Deadlock found" in payload["samples"][0]["line"]


def test_save_evidence_files_keeps_raw_metrics_but_compacts_summary(tmp_path):
    data = {
        "errors": [],
        "cpu": [_series("db-tidb-0", [1.0, 3.5])],
        "cpu_quota": [_series("db-tidb-0", [4.0, 4.0])],
    }

    ce.save_evidence_files(data, "INFRA CLEAR", "HOTSPOT CLEAR", "c1", "1713139200", "1713142800", str(tmp_path))

    run_dir = tmp_path
    raw_metrics = json.loads((run_dir / "metrics.json").read_text())
    agent_summary = json.loads((run_dir / "evidence_summary_for_agent.json").read_text())

    # First metric is the group avg (no chart_data), second is top instance (has chart_data)
    top_instance = next(m for m in raw_metrics["metrics"] if m.get("chart_data"))
    assert top_instance["chart_data"]["series"]
    assert "chart_data" not in agent_summary["metrics"][0]
    assert "chart_data_ref" not in agent_summary["metrics"][0]


def test_save_evidence_files_summarizes_logs_in_agent_summary(tmp_path):
    data = {
        "errors": [],
        "log_stats": {
            "timeline": {
                "tidb": {
                    "error": [
                        {"window_start": 1713139200, "count": 5},
                        {"window_start": 1713139500, "count": 3},
                    ],
                    "warn": [],
                    "info": [{"window_start": 1713139200, "count": 100}],
                },
                "tikv": {
                    "error": [{"window_start": 1713139200, "count": 2}],
                    "warn": [{"window_start": 1713139200, "count": 7}],
                    "info": [],
                },
                "pd": {"error": [], "warn": [], "info": []},
            },
            "summary": {
                "tidb": {"error": 8, "warn": 0, "info": 100},
                "tikv": {"error": 2, "warn": 7, "info": 0},
                "pd": {"error": 0, "warn": 0, "info": 0},
            },
            "step_seconds": 300,
            "errors": [],
            "loki_url": "https://clinic.pingcap.com/logs",
        },
    }

    ce.save_evidence_files(data, "INFRA CLEAR", "HOTSPOT CLEAR", "c1", "1713139200", "1713142800", str(tmp_path))

    run_dir = tmp_path
    evidence_logs = json.loads((run_dir / "logs.json").read_text())
    agent_summary = json.loads((run_dir / "evidence_summary_for_agent.json").read_text())

    # logs.json has timeline and summary (aggregated format)
    assert "timeline" in evidence_logs
    assert "summary" in evidence_logs
    assert "entries" not in evidence_logs
    assert evidence_logs["summary"]["tidb"]["error"] == 8
    assert evidence_logs["summary"]["tikv"]["warn"] == 7
    # agent summary has log_stats directly
    assert "entries" not in agent_summary["logs"]
    assert agent_summary["logs"]["summary"]["tikv"]["error"] == 2
    assert agent_summary["logs"]["loki_url"] == "https://clinic.pingcap.com/logs"


def test_compact_workload_pairs_statement_count_and_affected_rows():
    data = {
        "statement_count": [
            _series("tidb-0", [10.0, 20.0], {"type": "Update"}),
            _series("tidb-0", [2.0, 3.0], {"type": "Insert"}),
        ],
        "affected_rows": [
            _series("tidb-0", [1.0, 2.0], {"sql_type": "Update"}),
            _series("tidb-0", [20.0, 30.0], {"sql_type": "Insert"}),
        ],
    }

    workload = ce.compact_workload_for_ai(data)

    assert workload[0] == {
        "sql_type": "Update",
        "statement_rate_peak": 20.0,
        "statement_rate_mean": 15.0,
        "affected_rows_rate_peak": 2.0,
        "affected_rows_rate_mean": 1.5,
    }
    assert workload[1]["sql_type"] == "Insert"


def test_save_evidence_files_writes_workload_and_resource_control_summary(tmp_path):
    data = {
        "errors": [],
        "statement_count": [_series("tidb-0", [10.0, 20.0], {"type": "Update"})],
        "affected_rows": [_series("tidb-0", [1.0, 2.0], {"sql_type": "Update"})],
        "rcu_usage": [_series("ks1", [100.0, 200.0], {"name": "ks1"})],
        "resource_manager_request_p90": [
            _series("tidb-0", [0.005, 0.02], {"resource_group": "default"})
        ],
        "available_ru": [_series("ks1", [500.0, 1500.0], {"name": "ks1"})],
    }

    ce.save_evidence_files(data, "INFRA CLEAR", "HOTSPOT CLEAR", "c1", "1713139200", "1713142800", str(tmp_path))

    agent_summary = json.loads(
        (tmp_path / "evidence_summary_for_agent.json").read_text()
    )

    assert agent_summary["workload"][0]["sql_type"] == "Update"
    assert agent_summary["workload"][0]["statement_rate_peak"] == 20.0
    assert agent_summary["workload"][0]["affected_rows_rate_peak"] == 2.0
    assert agent_summary["resource_control"]["rcu_usage"][0]["peak"] == 200.0
    assert agent_summary["resource_control"]["request_p90"][0]["severity"] == "warning"
    assert agent_summary["resource_control"]["available_ru"][0]["severity"] == "warning"


def test_save_evidence_files_new_layout(tmp_path):
    """Verify the complete new file layout."""
    data = {
        "errors": ["test error"],
        "cpu": [_series("db-tidb-0", [1.0, 3.5])],
        "cpu_quota": [_series("db-tidb-0", [4.0, 4.0])],
        "tikv_thread_cpu": {"status": "skipped", "reason": "no saturated instance"},
        "log_stats": {"timeline": {}, "summary": {}, "step_seconds": 300, "errors": [], "loki_url": ""},
    }

    run_dir_str = ce.save_evidence_files(
        data, "INFRA CLEAR", "HOTSPOT CLEAR", "c1", "1713139200", "1713142800", str(tmp_path),
    )
    run_dir = Path(run_dir_str)

    # Root files
    assert (run_dir / "evidence_summary_for_agent.txt").exists()
    assert (run_dir / "evidence_summary_for_agent.json").exists()
    assert (run_dir / "workspace_evidence_report.json").exists()
    assert not (run_dir / "diagnosis_report.json").exists()
    assert not (run_dir / "run_manifest.json").exists()

    # Evidence category files
    assert (run_dir / "metrics.json").exists()
    assert (run_dir / "workload.json").exists()
    assert (run_dir / "resource_control.json").exists()
    assert (run_dir / "component_health.json").exists()
    assert (run_dir / "queries.json").exists()
    assert (run_dir / "logs.json").exists()
    assert (run_dir / "tikv_thread_cpu.json").exists()
    assert (run_dir / "links.json").exists()
    assert (run_dir / "gate_checks.json").exists()
    assert (run_dir / "collection_errors.json").exists()

    # Old layout dirs should NOT exist
    assert not (run_dir / "metrics").exists()
    assert not (run_dir / "queries").exists()
    assert not (run_dir / "logs").exists()
    assert not (run_dir / "summary").exists()

    # Verify workspace_evidence_report.json written by Python
    report = json.loads((run_dir / "workspace_evidence_report.json").read_text())
    assert report["type"] == "workspace_evidence_report"
    assert "metadata" in report
    assert "evidence" in report
    assert report["links"]["clinic_overview"].startswith("https://clinic.pingcap.com")

    agent_summary = json.loads((run_dir / "evidence_summary_for_agent.json").read_text())
    assert agent_summary["links"]["clinic_overview"].startswith("https://clinic.pingcap.com")

    # Verify collection errors
    errors = json.loads((run_dir / "collection_errors.json").read_text())
    assert errors["errors"] == ["test error"]


def test_save_evidence_files_returns_run_dir(tmp_path):
    data = {"errors": []}
    run_dir = ce.save_evidence_files(
        data, "INFRA CLEAR", "HOTSPOT CLEAR", "c1", "1713139200", "1713142800", str(tmp_path),
    )
    assert run_dir
    assert Path(run_dir).is_dir()


def test_build_agent_summary_schema():
    """Verify the agent summary has the expected top-level keys."""
    data = {
        "errors": [],
        "cpu": [_series("db-tidb-0", [1.0, 3.5])],
        "cpu_quota": [_series("db-tidb-0", [4.0, 4.0])],
        "tikv_thread_cpu": {"status": "skipped", "reason": "no saturated instance"},
        "log_stats": {"timeline": {}, "summary": {}, "step_seconds": 300, "errors": [], "loki_url": ""},
    }

    summary = ce.build_agent_summary(data, "INFRA CLEAR", "HOTSPOT CLEAR", "c1", "100", "200")
    # Remove internal keys
    summary.pop("_raw_metrics", None)
    summary.pop("_raw_logs_evidence", None)

    expected_keys = {
        "metadata", "metrics", "workload", "resource_control", "component_health", "profiles",
        "events", "queries", "logs", "gate_checks", "links", "errors", "critical_evidence_hints",
    }
    assert set(summary.keys()) == expected_keys
    assert summary["queries"].keys() == {
        "slow_queries", "topsql", "decoded_plan", "statement_summary", "statements",
    }
    assert summary["gate_checks"] == {"infra": "INFRA CLEAR", "hotspot": "HOTSPOT CLEAR"}
    assert summary["links"]["topsql_template"] == (
        "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html?clusterId=c1"
        "#/topsql?from=100&to=200&instance={urlencoded_instance}"
    )


def test_format_agent_text_has_section_index():
    """Verify the compact text format contains expected sections and index."""
    data = {
        "errors": [],
        "cpu": [_series("db-tidb-0", [1.0, 3.5])],
        "cpu_quota": [_series("db-tidb-0", [4.0, 4.0])],
        "tikv_thread_cpu": {"status": "skipped", "reason": "no saturated instance"},
        "log_stats": {"timeline": {}, "summary": {}, "step_seconds": 300, "errors": [], "loki_url": ""},
    }

    summary = ce.build_agent_summary(data, "INFRA CLEAR", "HOTSPOT CLEAR", "c1", "100", "200")
    summary.pop("_raw_metrics", None)
    summary.pop("_raw_logs_evidence", None)

    text = ce._format_agent_text(summary)

    # Has header
    assert "# Evidence — Cluster c1" in text
    # Has key findings
    assert "INFRA CLEAR" in text
    assert "HOTSPOT CLEAR" in text
    # Has section index with line numbers
    assert "[metrics] line" in text
    assert "[slow_queries] line" in text
    assert "[decoded_plan] line" in text
    assert "[logs] line" in text
    # Has section delimiters
    assert "--- metrics ---" in text
    assert "--- slow_queries ---" in text
    assert "--- decoded_plan ---" in text
    assert "--- logs ---" in text
    # Significantly smaller than JSON
    import json
    json_size = len(json.dumps(summary))
    text_size = len(text)
    assert text_size < json_size


def test_format_agent_text_section_index_line_numbers_match_sections():
    data = {
        "errors": ["oops"],
        "cpu": [_series("db-tidb-0", [1.0, 3.5])],
        "cpu_quota": [_series("db-tidb-0", [4.0, 4.0])],
        "tikv_thread_cpu": {"status": "skipped", "reason": "no saturated instance"},
        "log_stats": {"timeline": {}, "summary": {}, "step_seconds": 300, "errors": [], "loki_url": ""},
    }

    summary = ce.build_agent_summary(data, "INFRA CLEAR", "HOTSPOT CLEAR", "c1", "100", "200")
    summary.pop("_raw_metrics", None)
    summary.pop("_raw_logs_evidence", None)

    text = ce._format_agent_text(summary)
    lines = text.splitlines()
    actual = {
        line.removeprefix("--- ").removesuffix(" ---"): idx
        for idx, line in enumerate(lines, start=1)
        if line.startswith("--- ") and line.endswith(" ---")
    }

    for line in lines:
        if not line.startswith("  ["):
            continue
        name, start = line.strip()[1:].split("] line ")
        assert actual[name] == int(start)


def test_format_agent_text_includes_topsql_link_template_not_row_links():
    summary = {
        "metadata": {"cluster_id": "c1", "time_range": {"start": "100", "end": "200"}},
        "gate_checks": {"infra": "INFRA CLEAR", "hotspot": "HOTSPOT CLEAR"},
        "metrics": [],
        "workload": [],
        "resource_control": {},
        "component_health": {},
        "links": {
            "topsql_template": "https://clinic.example/ngm.html#/topsql?instance={urlencoded_instance}&from=100&to=200",
        },
        "queries": {
            "slow_queries": [],
            "topsql": [
                {
                    "component": "TiKV",
                    "digest": "digest-1",
                    "sql_preview": "SELECT 1",
                    "cpu_ms": 1234,
                    "cpu_pct": 66.4,
                    "instance": "db-tikv-24.db-tikv-peer.tidb1.svc:20180",
                }
            ],
            "decoded_plan": {},
        },
        "logs": {},
        "errors": [],
    }

    text = ce._format_agent_text(summary)

    assert "link_template: https://clinic.example/ngm.html#/topsql?instance={urlencoded_instance}&from=100&to=200" in text
    assert "component\tdigest\tsql_preview\tcpu_ms\tcpu_pct\tinstance" in text
    assert "db-tikv-24.db-tikv-peer.tidb1.svc:20180" in text
    assert "db-tikv-24.db-tikv-peer.tidb1.svc%3A20180" not in text


def test_format_agent_text_writes_txt_file(tmp_path):
    """Verify save_evidence_files writes the .txt file."""
    data = {
        "errors": [],
        "cpu": [_series("db-tidb-0", [1.0, 3.5])],
        "cpu_quota": [_series("db-tidb-0", [4.0, 4.0])],
        "tikv_thread_cpu": {"status": "skipped", "reason": "no saturated instance"},
        "log_stats": {"timeline": {}, "summary": {}, "step_seconds": 300, "errors": [], "loki_url": ""},
    }

    ce.save_evidence_files(data, "INFRA CLEAR", "HOTSPOT CLEAR", "c1", "1713139200", "1713142800", str(tmp_path))

    run_dir = tmp_path
    txt_path = run_dir / "evidence_summary_for_agent.txt"
    assert txt_path.exists()
    content = txt_path.read_text()
    assert "# Evidence — Cluster c1" in content
    assert "--- metrics ---" in content


def test_compact_component_health_for_ai_uses_parent_scope():
    data = {
        "parent_id": "rp1",
        "component_count": [
            _series("ignored", [2.0, 3.0], {"component": "coprocessor-worker"}),
        ],
        "restarts": [
            _series("worker-0", [0.0, 1.0], {"component": "coprocessor-worker"}),
        ],
        "worker_memory": [
            _series("worker-0", [1024 ** 3, 2 * 1024 ** 3], {"component": "coprocessor-worker"}),
        ],
    }

    summary = ce.compact_component_health_for_ai(data, "c1", "100", "200")

    assert summary["query_scope"] == "rp1"
    assert summary["parent_id"] == "rp1"
    assert summary["component_count"][0] == {
        "component": "coprocessor-worker",
        "first": 2.0,
        "last": 3.0,
        "min": 2.0,
        "max": 3.0,
        "changes": 1,
        "net_delta": 1,
        "max_to_last_delta": 0,
        "change_count": 1,
    }
    assert summary["restarts"][0]["restart_count"] == 1.0
    assert summary["worker_memory"][0]["peak_gb"] == 2.0


def test_component_count_changes_records_capacity_delta():
    data = {
        "component_count": [
            _series("ignored", [3.0, 1.0], {"component": "coprocessor-worker"}),
        ],
    }

    summary = ce.compact_component_health_for_ai(data, "c1", "100", "200")

    assert summary["component_count"][0]["changes"] == -2.0
    assert summary["component_count"][0]["net_delta"] == -2.0
    assert summary["component_count"][0]["max_to_last_delta"] == -2.0
    assert summary["component_count"][0]["change_count"] == 1


def test_component_count_changes_records_transient_capacity_loss():
    data = {
        "component_count": [
            _series("ignored", [2.0, 3.0, 2.0], {"component": "coprocessor-worker"}),
        ],
    }

    summary = ce.compact_component_health_for_ai(data, "c1", "100", "200")

    row = summary["component_count"][0]
    assert row["first"] == 2.0
    assert row["last"] == 2.0
    assert row["min"] == 2.0
    assert row["max"] == 3.0
    assert row["changes"] == -1.0
    assert row["net_delta"] == 0.0
    assert row["max_to_last_delta"] == -1.0
    assert row["change_count"] == 2


def test_fetch_cluster_meta_includes_parent_id_for_premium():
    class FakeAPI:
        def _get(self, path, params):
            assert path == "/clinic/api/v1/dashboard/clusters"
            return {
                "items": [
                    {
                        "clusterID": "c1",
                        "clusterDeployType": "premium",
                        "parentID": "rp1",
                        "orgID": "org1",
                    }
                ]
            }

    meta = ce._fetch_cluster_meta(FakeAPI(), "c1")

    assert meta["deployType"] == "premium"
    assert meta["parent_id"] == "rp1"
    assert meta["orgID"] == "org1"


def test_component_health_context_window_adds_small_guard_band():
    assert ce._component_health_context_window("1000", "2000") == ("100", "2300")


def test_focused_log_context_window_adds_small_guard_band():
    assert ce._focused_log_context_window("1000", "2000") == ("100", "2300")


def test_loki_time_ns_converts_epoch_seconds_only():
    assert ce._loki_time_ns("1776635400") == "1776635400000000000"
    assert ce._loki_time_ns("1776635400000000000") == "1776635400000000000"


def test_collect_focused_log_samples_uses_namespace_fallback_and_ns_window():
    calls = []

    class FakeAPI:
        def _get(self, path, params):
            calls.append((path, params))
            if "namespace=~" not in params["query"]:
                return {"data": {"result": []}}
            return {
                "data": {
                    "result": [
                        {
                            "stream": {"namespace": "tidb105", "pod": "tidb-0"},
                            "values": [
                                [
                                    "1776635400323971241",
                                    '[ERROR] ["panic in the recoverable goroutine"] '
                                    '[r="[executor:3024]Query execution was interrupted, '
                                    'maximum statement execution time exceeded"]',
                                ]
                            ],
                        }
                    ]
                }
            }

    counts, samples, error = ce._collect_focused_log_samples(FakeAPI(), "105", "1776631800", "1776639000")

    assert error is None
    assert counts[0]["component"] == "tidb105"
    assert "panic in the recoverable goroutine" in counts[0]["message"]
    assert samples[0]["instance"] == "tidb-0"
    assert calls[0][1]["start"] == "1776631800000000000"
    assert calls[0][1]["end"] == "1776639000000000000"
    assert any('namespace=~"tidb105"' in call[1]["query"] for call in calls)


def test_collect_activity_events_uses_org_id_from_cluster_meta():
    class FakeAPI:
        def _get(self, path, params):
            assert path == "/clinic/api/v1/activityhub/applications/org1/targets/c1/activities"
            assert params["begin_ts"] == 100
            assert params["end_ts"] == 200
            return {"activities": [{"name": "configchangeapply", "payload": {"summary": "Configuration change applying"}}]}

    events = ce.collect_activity_events(FakeAPI(), "c1", "100", "200", {"orgID": "org1"})

    assert events["events"][0]["name"] == "configchangeapply"


def test_build_critical_evidence_hints_includes_capacity_and_log_signals():
    summary = {
        "metadata": {},
        "metrics": [
            {"name": "TIDB CPU", "nodegroup": "db", "promql": "process_cpu_seconds_total"},
            {"name": "TIDB CPU", "nodegroup": "db-small", "promql": "process_cpu_seconds_total"},
            {"name": "TiDB GC Pause P99", "metric_group_id": "tidb_gc_pause_p99", "severity": "warning"},
        ],
        "workload": [{"sql_type": "Delete", "statement_rate_peak": 10.0, "affected_rows_rate_peak": 1.0}],
        "queries": {"decoded_plan": "IndexLookUp process_keys=100000"},
        "component_health": {
            "component_count": [
                {"component": "coprocessor-worker", "first": 3.0, "last": 1.0, "changes": -2.0}
            ],
            "restarts": [{"instance": "worker-0", "restart_count": 1.0}],
        },
        "logs": {
            "log_counts": [
                {"message": "panic: [executor:3024]Query execution was interrupted, maximum statement execution time exceeded", "count": 1},
                {"message": "request backlog timeout unavailable coprocessor", "count": 1},
                {"message": "[executor:1213]Deadlock found when trying to get lock; pessimistic lock retry", "count": 2},
            ]
        },
        "events": [
            {
                "name": "configchangeapply",
                "detail": '{"componentName":"tidb","overwriteImageTag":"v8.5.4-nextgen.202510.14"}',
            }
        ],
    }
    cluster_meta = {
        "components": {
            "COMPONENT_TYPE_TIDB": {
                "tidbGroups": [{"replicas": 4}, {"replicas": 1}],
            }
        }
    }

    hints = ce.build_critical_evidence_hints(summary, cluster_meta)

    ids = {hint["id"] for hint in hints}
    assert "capacity_comparison_context_present" in ids
    assert "primary_workload_shift_present" in ids
    assert "allocation_pressure_signal_present" in ids
    assert "capacity_loss_signal_present" in ids
    assert "restart_signal_present" in ids
    assert "panic_signal_present" in ids
    assert "query_interruption_signal_present" in ids
    assert "query_timeout_interrupt_signal_present" in ids
    assert "coprocessor_cascade_signal_present" in ids
    assert "known_issue_correlation_context_present" in ids
    assert "transaction_lock_signal_present" in ids


def test_build_critical_evidence_hints_does_not_route_on_broad_transaction_recall_terms_only():
    summary = {
        "metadata": {},
        "metrics": [],
        "workload": [],
        "queries": {},
        "component_health": {},
        "logs": {
            "log_counts": [
                {"message": "foreign key metadata loaded; try again later", "count": 1},
            ],
            "samples": [],
        },
        "events": [],
    }

    hints = ce.build_critical_evidence_hints(summary, {})

    ids = {hint["id"] for hint in hints}
    assert "transaction_lock_signal_present" not in ids