import copy import importlib.util import os from pathlib import Path import pytest import yaml from investibook.core.eval.eval_registry import EvalRegistryError, inspect_suite, list_suites, load_case, load_suite from investibook.core.eval.eval_registry import default_registry_root def _write_case_files(case_dir: Path, payload: dict) -> None: case_payload = copy.deepcopy(payload) judge_payload = { "schema_version": case_payload["schema_version"], "case_id": case_payload["id"], "ground_truth": case_payload.pop("ground_truth"), "judge": case_payload.pop("judge"), } (case_dir / "case.yaml").write_text(yaml.safe_dump(case_payload, sort_keys=False)) (case_dir / "judge.yaml").write_text(yaml.safe_dump(judge_payload, sort_keys=False)) (case_dir / "expected.md").write_text("expected\n") def test_load_core_daily_demo_suite_from_repo(): detail = inspect_suite("core-daily-demo") assert detail["suite_id"] == "core-daily-demo" assert detail["case_count"] == 13 case = detail["cases"][0] assert case["case_id"] == "component-restart-during-incident-window" assert "identify_known_bug_restart_chain" in case["critical_rubric_ids"] assert case["case_version"] assert detail["suite_version"] def test_case_loads_without_source_snapshot_metadata(): suite = load_suite("core-daily-demo") assert [case.id for case in suite.cases] == [ "component-restart-during-incident-window", "cpu-spike-during-incident-window", "system-unavailability-during-incident-window", "region-unavailable-escalation-during-incident-window", "expensive-query-alert-correlation-during-incident-window", "expensive-query-spike-during-incident-window", "periodic-tidb-cpu-spike-during-incident-window", "sql-tail-latency-spike-during-incident-window", "service-degradation-during-incident-window", "panic-in-recoverable-goroutine", "tikv-node-stall-during-incident-window", "ticdc-dataflow-stall-during-incident-window", "tikv-node-slowdown-during-incident-window", ] def test_load_adhoc_evidence_demo_suite_from_repo(): detail = inspect_suite("adhoc-evidence-demo") assert detail["suite_id"] == "adhoc-evidence-demo" assert detail["case_count"] == 12 assert [item["case_id"] for item in detail["cases"]] == [ "tikv-read-iops-ranking-during-incident-window", "tikv-read-iops-full-ranking-across-two-clusters-during-incident-window", "topsql-links-for-hottest-two-tikv-nodes-during-incident-window", "representative-business-slow-queries-in-saas01-during-latest-window", "new-slow-query-digests-between-adjacent-windows-in-saas01", "top-cpu-overload-nodes-and-log-patterns-in-saas01-during-latest-window", "top-tidb-p99-latency-nodes-in-saas01-during-latest-window", "classify-selected-stores-as-tikv-or-tiflash", "compaction-signals-on-db-tikv-9-in-saas01-during-latest-window", "map-selected-tikv-pods-to-worker-nodes-in-saas01", "single-node-cpu-core-scope-compare-during-incident-window", "ttl-row-processing-during-sql-tail-latency-incident-window", ] assert detail["suite_version"] def test_list_suites_sorts_active_first_and_exposes_metadata(tmp_path: Path): root = tmp_path / "regression" suites_dir = root / "suites" suites_dir.mkdir(parents=True) (suites_dir / "draft-suite.yaml").write_text( yaml.safe_dump( { "schema_version": 1, "id": "draft-suite", "status": "draft", "title": "Draft Suite", "description": "Draft only.", "mode": "manual", "baseline_eligible": False, "cases": ["case-b"], }, sort_keys=False, ) ) (suites_dir / "active-suite.yaml").write_text( yaml.safe_dump( { "schema_version": 1, "id": "active-suite", "status": "active", "title": "Active Suite", "description": "Runnable now.", "mode": "scheduled", "baseline_eligible": True, "cases": ["case-a", "case-c"], }, sort_keys=False, ) ) suites = list_suites(registry_root=root) assert [suite.id for suite in suites] == ["active-suite", "draft-suite"] assert suites[0].title == "Active Suite" assert suites[0].status == "active" assert suites[0].mode == "scheduled" assert suites[0].baseline_eligible is True assert suites[0].case_count == 2 assert suites[0].case_ids == ["case-a", "case-c"] def test_case_can_declare_allowed_time_windows(): suite = load_suite("core-daily-demo") allowed = suite.cases[0].data["time_window"]["allowed"] assert len(allowed) == 1 assert allowed[0]["id"] == "started_at_minus_1h" assert allowed[0]["start"] == "2026-04-01T10:36:00Z" def test_case_can_declare_execution_contract(): suite = load_suite("core-daily-demo") contract = suite.cases[0].execution_contract assert [item["id"] for item in contract["checks"]] == [ "required_workflow_present", "required_artifacts_present", "preferred_channels_minimum_observed", "preferred_channels_all_observed", ] assert contract["checks"][0]["rules"][0]["source"] == "workflow_check" assert contract["checks"][1]["rules"][0]["source"] == "artifact_fact" assert contract["checks"][2]["rules"][0]["matcher"] == "count_gte" assert contract["checks"][3]["required"] is False def test_case_can_declare_judge_contract_rules(): suite = load_suite("core-daily-demo") judge_contract = suite.cases[0].judge_contract assert [item["id"] for item in judge_contract["checks"]] == [ "time_window_match", "cluster_scope_match", ] assert judge_contract["checks"][0]["rules"][0]["source"] == "trace_fact" assert judge_contract["checks"][0]["rules"][0]["matcher"] == "bool_true" def test_case_can_declare_judge_workflow_rules(): suite = load_suite("core-daily-demo") judge_workflow = suite.cases[0].judge_workflow assert [item["id"] for item in judge_workflow["checks"][:3]] == [ "architecture_context_loaded", "misdiagnosis_guardrails_loaded", "diagnosis_run_initialized", ] assert judge_workflow["checks"][0]["rules"][0]["source"] == "workflow_fact" assert judge_workflow["checks"][0]["rules"][0]["matcher"] == "bool_true" assert judge_workflow["checks"][4]["required"] is False def test_case_can_declare_judge_evidence_collection_rules(): suite = load_suite("core-daily-demo") evidence_collection = suite.cases[0].judge_evidence_collection assert evidence_collection["weight"] == 5 assert [item["id"] for item in evidence_collection["checks"]] == [ "required_artifacts_present", "preferred_channels_minimum_observed", "preferred_channels_all_observed", ] assert evidence_collection["checks"][0]["rules"][0]["source"] == "artifact_fact" assert evidence_collection["checks"][1]["rules"][0]["matcher"] == "count_gte" assert evidence_collection["checks"][2]["required"] is False def test_case_can_declare_critical_evidence_rules(): suite = load_suite("core-daily-demo") critical_evidence = suite.cases[0].critical_evidence assert critical_evidence["weight"] == 30 assert [item["id"] for item in critical_evidence["checks"]] == [ "restart_signal_present", "panic_signal_present", "query_interruption_signal_present", ] assert critical_evidence["checks"][1]["rules"][0]["source"] == "logs" assert critical_evidence["checks"][1]["rules"][0]["matcher"] == "keyword_any" def test_case_can_declare_evidence_routing_rules(): suite = load_suite("core-daily-demo") evidence_routing = suite.cases[0].evidence_routing assert evidence_routing["weight"] == 5 assert [item["id"] for item in evidence_routing["checks"]] == [ "analysis_patch_valid_json", "analysis_items_shape_valid", "analysis_key_evidence_present", "metric_refs_resolvable", "query_refs_resolvable", "supporting_evidence_present", "analysis_log_or_event_channel_resolvable", ] assert evidence_routing["checks"][0]["required"] is True assert evidence_routing["checks"][0]["rules"][0]["source"] == "routing_fact" assert evidence_routing["checks"][0]["rules"][1]["matcher"] == "bool_true" def test_case_can_declare_unified_diagnosis_checks(): suite = load_suite("core-daily-demo") checks = suite.cases[0].diagnosis_checks assert [item["id"] for item in checks] == [ "identify_known_bug_restart_chain", "identify_kill_query_cleanup_chain", "reject_missing_index_root_cause", "mention_github_issue_id", "recommend_fixed_version_upgrade", ] assert checks[0]["mode"] == "blended" assert checks[0]["deterministic"]["rules"][0]["source"] == "output_fact" assert checks[0]["semantic"]["id"] == 1 assert checks[2]["mode"] == "deterministic_only" assert checks[2]["deterministic"]["rules"][1]["matcher"] == "bool_false" def test_workspace_export_case_loads_from_repo(): case = load_case("service-degradation-during-incident-window") assert "source" not in case.data assert "golden_run" not in case.data assert case.data["environment"]["cluster_id"] == "10322500445237050222" assert case.data["time_window"]["allowed"][0]["start"] == "2026-04-07T08:47:00Z" assert [item["id"] for item in case.diagnosis_checks] == [ "identify_specific_sql_hotspot_path", "identify_root_cause_mechanism", "identify_sql_plan_shift", "identify_tiflash_to_tikv_execution_path_change", ] required_ids = [item["id"] for item in case.diagnosis_checks if item.get("required")] assert required_ids == ["identify_specific_sql_hotspot_path"] def test_cpu_spike_case_loads_from_repo(): case = load_case("cpu-spike-during-incident-window") assert "source" not in case.data assert "golden_run" not in case.data assert case.data["environment"]["cluster_id"] == "10518506906906971664" assert case.data["time_window"]["allowed"][0]["end"] == "2026-04-08T01:30:00Z" assert case.acp_bash_timeout_s == 300 assert [item["id"] for item in case.diagnosis_checks] == [ "identify_capacity_shift_as_primary_driver", "reject_plan_cache_or_secondary_signals_as_primary_root_cause", "recommend_capacity_adjustment_over_cache_tuning", ] required_ids = [item["id"] for item in case.diagnosis_checks if item.get("required")] assert required_ids == ["identify_capacity_shift_as_primary_driver"] def test_sustained_cpu_pressure_case_loads_from_repo(): case = load_case("sustained-cpu-pressure-during-incident-window") assert "source" not in case.data assert "golden_run" not in case.data assert case.data["environment"]["cluster_id"] == "10518506906906971664" assert case.data["time_window"]["allowed"][0]["end"] == "2026-03-17T15:10:00Z" assert case.acp_timeout_s == 1200 assert case.acp_bash_timeout_s == 900 assert case.acp_idle_timeout_s == 900 assert [item["id"] for item in case.diagnosis_checks] == [ "identify_batch_delete_path_as_primary_driver", "correlate_log_context_with_primary_workload", "identify_allocation_or_gc_pressure", "identify_large_in_list_gc_pressure_chain", ] required_ids = [item["id"] for item in case.diagnosis_checks if item.get("required")] assert required_ids == ["identify_batch_delete_path_as_primary_driver"] assert case.diagnosis_checks[2]["mode"] == "blended" assert case.critical_evidence["checks"][2]["id"] == "heap_profile_hotspot_present" assert case.critical_evidence["checks"][2]["required"] is False def test_recoverable_panic_case_loads_with_single_required_gate(): case = load_case("panic-in-recoverable-goroutine") assert [item["id"] for item in case.diagnosis_checks] == [ "identify_real_panic_event", "identify_query_timeout_interrupt", "reject_fake_alarm_or_unrelated_root_cause", "identify_expected_behavior", "recommend_no_tidb_fix_needed", ] required_ids = [item["id"] for item in case.diagnosis_checks if item.get("required")] assert required_ids == ["identify_query_timeout_interrupt"] assert case.diagnosis_checks[1]["mode"] == "deterministic_only" def test_expensive_query_spike_case_loads_with_single_required_gate(): case = load_case("expensive-query-spike-during-incident-window") assert [item["id"] for item in case.diagnosis_checks] == [ "mention_expensive_query_log_evidence", "identify_large_scan_or_missing_access_path", "reject_analyze_as_primary_culprit", "recommend_index_or_access_path_fix", ] required_ids = [item["id"] for item in case.diagnosis_checks if item.get("required")] assert required_ids == ["identify_large_scan_or_missing_access_path"] assert case.diagnosis_checks[2]["mode"] == "deterministic_only" def test_expensive_query_alert_correlation_case_loads_with_single_required_gate(): case = load_case("expensive-query-alert-correlation-during-incident-window") assert [item["id"] for item in case.diagnosis_checks] == [ "identify_expensive_query_workload_as_primary_driver", "identify_correlated_tikv_or_network_impact", "reject_background_noise_as_primary_root_cause", "recommend_query_level_optimization", "mention_related_alert_correlation", "call_out_need_for_query_identification", ] required_ids = [item["id"] for item in case.diagnosis_checks if item.get("required")] assert required_ids == ["identify_expensive_query_workload_as_primary_driver"] assert case.diagnosis_checks[2]["mode"] == "deterministic_only" def test_system_unavailability_case_loads_with_single_required_gate(): case = load_case("system-unavailability-during-incident-window") assert [item["id"] for item in case.diagnosis_checks] == [ "identify_worker_capacity_collapse_as_primary_path", "identify_restart_loop_as_key_amplifier", "reject_query_only_primary_root_cause", "reconstruct_capacity_to_outage_chain", "recommend_capacity_protection", "recommend_worker_health_followup", ] required_ids = [item["id"] for item in case.diagnosis_checks if item.get("required")] assert required_ids == ["identify_worker_capacity_collapse_as_primary_path"] assert case.diagnosis_checks[2]["mode"] == "blended" def test_tail_latency_surge_case_loads_from_repo(): case = load_case("tail-latency-surge-during-incident-window") assert "source" not in case.data assert "golden_run" not in case.data assert case.data["environment"]["cluster_id"] == "10805087899823024166" assert case.data["time_window"]["allowed"][0]["end"] == "2026-03-09T05:14:00Z" assert case.acp_bash_timeout_s == 300 assert case.acp_idle_timeout_s == 300 assert [item["id"] for item in case.diagnosis_checks] == [ "identify_lock_contention_as_primary_path", "identify_single_tikv_hotspot", "recommend_write_path_mitigation", ] required_ids = [item["id"] for item in case.diagnosis_checks if item.get("required")] assert required_ids == ["identify_lock_contention_as_primary_path"] def test_tikv_node_stall_case_loads_from_repo(): case = load_case("tikv-node-stall-during-incident-window") assert "source" not in case.data assert "golden_run" not in case.data assert case.data["environment"]["cluster_id"] == "10143988032832670331" assert case.data["environment"]["cluster_type"] == "dedicated" assert case.data["time_window"]["allowed"][0]["start"] == "2026-06-05T03:41:15Z" assert case.data["time_window"]["allowed"][0]["end"] == "2026-06-05T03:45:15Z" assert case.acp_bash_timeout_s == 300 assert case.acp_idle_timeout_s == 300 assert [item["id"] for item in case.diagnosis_checks] == [ "identify_compaction_read_pool_stall_as_primary_path", "identify_single_tikv_stall_scope", "identify_specific_hotspot_node_scope", "recommend_tikv_stall_mitigation", ] required_ids = [item["id"] for item in case.diagnosis_checks if item.get("required")] assert required_ids == [ "identify_compaction_read_pool_stall_as_primary_path", "identify_specific_hotspot_node_scope", ] def test_tikv_read_iops_ranking_case_loads_as_adhoc(): case = load_case("tikv-read-iops-ranking-during-incident-window") assert case.evaluation_mode == "adhoc_evidence" assert case.data["environment"]["cluster_id"] == "10518506906906971664" assert case.resolved_time_window_output_id == "resolved_window" assert [item["output_id"] for item in case.runtime_hooks] == [ "resolved_window", "reference_answer", ] assert [item["id"] for item in case.diagnosis_checks] == [] assert [item["id"] for item in case.judge_workflow["checks"]] == [ "final_answer_emitted", "collect_evidence_not_used", ] assert [item["id"] for item in case.execution_contract["checks"]] == [ "required_workflow_present", "collect_evidence_not_used", "required_report_present", "query_or_metric_channel_observed", ] assert case.critical_evidence["weight"] == 80 assert [item["id"] for item in case.critical_evidence["checks"]] == [ "raw_disk_read_metric_present", "cluster_scoped_mapping_evidence_present", "output_cluster_id_matches_reference", "output_window_start_matches_reference", "output_window_end_matches_reference", "evidence_sufficient_matches_reference", "top_nodes_present", "top_nodes_have_three_unique_tikv_nodes", "top_nodes_match_runtime_reference", ] assert "--duration-minutes" in case.runtime_hooks[0]["args"] assert "5" in case.runtime_hooks[0]["args"] assert "tikv_node" in case.data["prompt"]["text"] def test_tikv_read_iops_full_ranking_case_loads_as_adhoc(): case = load_case("tikv-read-iops-full-ranking-across-two-clusters-during-incident-window") assert case.evaluation_mode == "adhoc_evidence" assert case.data["environment"]["cluster_id_a"] == "10518506906906971664" assert case.data["environment"]["cluster_id_b"] == "10989049060142230334" assert case.resolved_time_window_output_id == "resolved_window" assert [item["output_id"] for item in case.runtime_hooks] == [ "resolved_window", "reference_answer", ] assert [item["id"] for item in case.diagnosis_checks] == [] assert [item["id"] for item in case.judge_contract["checks"]] == [ "time_window_match", ] assert [item["id"] for item in case.judge_workflow["checks"]] == [ "final_answer_emitted", "collect_evidence_not_used", ] assert [item["id"] for item in case.execution_contract["checks"]] == [ "required_workflow_present", "collect_evidence_not_used", "required_report_present", "query_or_metric_channel_observed", ] assert case.critical_evidence["weight"] == 80 assert [item["id"] for item in case.critical_evidence["checks"]] == [ "raw_disk_read_metric_present", "cluster_scoped_mapping_evidence_present", "output_window_start_matches_reference", "output_window_end_matches_reference", "mapping_strategy_matches_reference", "evidence_sufficient_matches_reference", "nodes_by_cluster_present", "nodes_cover_two_clusters", "full_node_ranking_matches_runtime_reference", ] assert "--cluster-id-a" in case.runtime_hooks[0]["args"] assert "--cluster-id-b" in case.runtime_hooks[0]["args"] assert '"nodes"' in case.data["prompt"]["text"] assert '["db-tikv-0", 12345.67]' in case.data["prompt"]["text"] def test_topsql_link_case_loads_as_adhoc(): case = load_case("topsql-links-for-hottest-two-tikv-nodes-during-incident-window") assert case.evaluation_mode == "adhoc_evidence" assert case.data["environment"]["cluster_id"] == "10518506906906971664" assert case.resolved_time_window_output_id == "resolved_window" assert [item["output_id"] for item in case.runtime_hooks] == [ "resolved_window", "reference_answer", ] assert [item["id"] for item in case.judge_contract["checks"]] == [ "time_window_match", "cluster_scope_match", ] assert [item["id"] for item in case.judge_workflow["checks"]] == [ "final_answer_emitted", "collect_evidence_not_used", ] assert [item["id"] for item in case.execution_contract["checks"]] == [ "required_workflow_present", "collect_evidence_not_used", "required_report_present", "query_or_metric_channel_observed", ] assert case.critical_evidence["weight"] == 80 assert [item["id"] for item in case.critical_evidence["checks"]] == [ "raw_disk_read_metric_present", "cluster_scoped_mapping_evidence_present", "output_cluster_id_matches_reference", "output_window_start_matches_reference", "output_window_end_matches_reference", "mapping_strategy_matches_reference", "evidence_sufficient_matches_reference", "topsql_links_present", "topsql_links_have_two_unique_tikv_nodes", "topsql_link_nodes_match_runtime_reference", "topsql_urls_match_runtime_reference", ] assert "--top-n" in case.runtime_hooks[0]["args"] assert "2" in case.runtime_hooks[0]["args"] assert "topsql_links" in case.data["prompt"]["text"] assert '"url"' in case.data["prompt"]["text"] def test_store_role_classification_case_loads_as_adhoc(): case = load_case("classify-selected-stores-as-tikv-or-tiflash") assert case.evaluation_mode == "adhoc_evidence" assert case.data["environment"]["cluster_id"] == "10518506906906971664" assert case.resolved_time_window_output_id == "case_input" assert [item["output_id"] for item in case.runtime_hooks] == [ "case_input", "reference_answer", ] assert [item["id"] for item in case.diagnosis_checks] == [] assert [item["id"] for item in case.judge_contract["checks"]] == [ "cluster_scope_match", ] assert [item["id"] for item in case.judge_workflow["checks"]] == [ "final_answer_emitted", "collect_evidence_not_used", ] assert [item["id"] for item in case.execution_contract["checks"]] == [ "required_workflow_present", "collect_evidence_not_used", "required_report_present", "query_or_metric_channel_observed", ] assert case.critical_evidence["weight"] == 80 assert [item["id"] for item in case.critical_evidence["checks"]] == [ "output_cluster_id_matches_reference", "evidence_sufficient_matches_reference", "stores_present", "stores_have_expected_unique_store_count", "stores_cover_two_component_roles", "store_roles_match_runtime_reference", ] assert "--stores-per-role" in case.runtime_hooks[0]["args"] assert "3" in case.runtime_hooks[0]["args"] assert "store_ids_json" in case.data["prompt"]["text"] assert '"component": "tikv"' in case.data["prompt"]["text"] def test_representative_business_slow_queries_case_loads_as_adhoc(): case = load_case("representative-business-slow-queries-in-saas01-during-latest-window") assert case.evaluation_mode == "adhoc_evidence" assert case.data["environment"]["cluster_label"] == "saas01" assert case.data["environment"]["cluster_id"] == "10989049060142230334" assert case.resolved_time_window_output_id == "resolved_window" assert [item["output_id"] for item in case.runtime_hooks] == [ "resolved_window", "reference_answer", ] assert [item["id"] for item in case.diagnosis_checks] == [] assert [item["id"] for item in case.judge_contract["checks"]] == [ "cluster_scope_match", ] assert [item["id"] for item in case.judge_workflow["checks"]] == [ "final_answer_emitted", "collect_evidence_not_used", ] assert [item["id"] for item in case.execution_contract["checks"]] == [ "required_workflow_present", "collect_evidence_not_used", "required_report_present", "query_or_metric_channel_observed", ] assert case.critical_evidence["weight"] == 80 assert [item["id"] for item in case.critical_evidence["checks"]] == [ "output_cluster_id_matches_reference", "output_window_start_matches_reference", "output_window_end_matches_reference", "evidence_sufficient_matches_reference", "slow_queries_present", "slow_queries_have_three_unique_digests", "slow_query_digests_match_reference", "slow_query_exec_count_match_reference", "slow_query_avg_time_match_reference", ] assert "--top-n" in case.runtime_hooks[0]["args"] assert "3" in case.runtime_hooks[0]["args"] assert "{{environment.cluster_label}}" in case.data["prompt"]["text"] assert '"slow_queries"' in case.data["prompt"]["text"] def test_representative_business_slow_queries_runtime_filters_control_sql(): hook_path = ( default_registry_root() / "cases" / "representative-business-slow-queries-in-saas01-during-latest-window" / "runtime_hook.py" ) spec = importlib.util.spec_from_file_location("slow_query_runtime_hook", hook_path) assert spec is not None assert spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) assert module._is_trivial_control_sql("COMMIT") is True assert module._is_trivial_control_sql("START TRANSACTION") is True assert module._is_trivial_control_sql("set autocommit=0") is True assert module._is_trivial_control_sql(" use mysql; ") is True assert module._is_trivial_control_sql("SHOW TABLES") is True assert module._is_trivial_control_sql("SELECT * FROM mysql.user") is False def test_new_slow_query_digests_case_loads_as_adhoc(): case = load_case("new-slow-query-digests-between-adjacent-windows-in-saas01") assert case.evaluation_mode == "adhoc_evidence" assert case.data["environment"]["cluster_label"] == "saas01" assert case.data["environment"]["cluster_id"] == "10989049060142230334" assert case.resolved_time_window_output_id == "resolved_window" assert [item["output_id"] for item in case.runtime_hooks] == [ "resolved_window", "reference_answer", ] assert [item["id"] for item in case.diagnosis_checks] == [] assert [item["id"] for item in case.judge_contract["checks"]] == [ "cluster_scope_match", ] assert [item["id"] for item in case.judge_workflow["checks"]] == [ "final_answer_emitted", "collect_evidence_not_used", ] assert [item["id"] for item in case.execution_contract["checks"]] == [ "required_workflow_present", "collect_evidence_not_used", "required_report_present", "query_or_metric_channel_observed", ] assert case.critical_evidence["weight"] == 80 assert [item["id"] for item in case.critical_evidence["checks"]] == [ "output_cluster_id_matches_reference", "previous_window_start_matches_reference", "previous_window_end_matches_reference", "current_window_start_matches_reference", "current_window_end_matches_reference", "evidence_sufficient_matches_reference", "new_sql_digests_match_reference", "new_sql_digest_occurrence_counts_match_reference", ] assert "--top-n" in case.runtime_hooks[0]["args"] assert "10" in case.runtime_hooks[0]["args"] assert "new_sql_digests" in case.data["prompt"]["text"] assert "slow query logs" in case.data["prompt"]["text"] def test_top_cpu_overload_and_error_warn_log_counts_case_loads_as_adhoc(): case = load_case("top-cpu-overload-nodes-and-log-patterns-in-saas01-during-latest-window") assert case.evaluation_mode == "adhoc_evidence" assert case.data["environment"]["cluster_label"] == "saas01" assert case.data["environment"]["cluster_id"] == "10989049060142230334" assert case.resolved_time_window_output_id == "resolved_window" assert [item["output_id"] for item in case.runtime_hooks] == [ "resolved_window", "reference_answer", ] assert [item["id"] for item in case.diagnosis_checks] == [] assert [item["id"] for item in case.judge_contract["checks"]] == [ "cluster_scope_match", ] assert [item["id"] for item in case.judge_workflow["checks"]] == [ "final_answer_emitted", "collect_evidence_not_used", ] assert [item["id"] for item in case.execution_contract["checks"]] == [ "required_workflow_present", "collect_evidence_not_used", "required_report_present", "query_or_metric_channel_observed", ] assert case.critical_evidence["weight"] == 80 assert [item["id"] for item in case.critical_evidence["checks"]] == [ "output_cluster_id_matches_reference", "output_window_start_matches_reference", "output_window_end_matches_reference", "evidence_sufficient_matches_reference", "ranked_nodes_identity_matches_reference", "ranked_nodes_numeric_fields_match_reference", ] assert "--top-n-per-component" in case.runtime_hooks[0]["args"] assert "3" in case.runtime_hooks[0]["args"] assert "ranked_nodes" in case.data["prompt"]["text"] assert "error_warn_log_count" in case.data["prompt"]["text"] assert "top_log_digest" not in case.data["prompt"]["text"] def test_top_tidb_p99_latency_nodes_case_loads_as_adhoc(): case = load_case("top-tidb-p99-latency-nodes-in-saas01-during-latest-window") assert case.evaluation_mode == "adhoc_evidence" assert case.data["environment"]["cluster_label"] == "saas01" assert case.data["environment"]["cluster_id"] == "10989049060142230334" assert case.resolved_time_window_output_id == "resolved_window" assert [item["output_id"] for item in case.runtime_hooks] == [ "resolved_window", "reference_answer", ] assert [item["id"] for item in case.diagnosis_checks] == [] assert [item["id"] for item in case.judge_contract["checks"]] == [ "cluster_scope_match", ] assert [item["id"] for item in case.judge_workflow["checks"]] == [ "final_answer_emitted", "collect_evidence_not_used", ] assert [item["id"] for item in case.execution_contract["checks"]] == [ "required_workflow_present", "collect_evidence_not_used", "required_report_present", "query_or_metric_channel_observed", ] assert case.critical_evidence["weight"] == 80 assert [item["id"] for item in case.critical_evidence["checks"]] == [ "latency_metric_present", "breakdown_metrics_present", "latency_metric_series_nonempty", "output_cluster_id_matches_reference", "output_window_start_matches_reference", "output_window_end_matches_reference", "evidence_sufficient_matches_reference", "cluster_avg_tikv_grpc_matches_reference", "cluster_avg_pd_tso_matches_reference", "top_nodes_match_runtime_reference", ] assert "--step-seconds" in case.runtime_hooks[0]["args"] assert "60" in case.runtime_hooks[0]["args"] assert "top_nodes_by_avg_p99_latency" in case.data["prompt"]["text"] assert "avg_gc_stw_pause_p99_s" in case.data["prompt"]["text"] assert "cluster_avg_tikv_grpc_request_handle_p99_s" in case.data["prompt"]["text"] assert "cluster_avg_pd_tso_request_handle_p99_s" in case.data["prompt"]["text"] assert "TiDB client 侧的 kv_get request duration" in case.data["prompt"]["text"] assert "PD client 侧的 TSO request duration" in case.data["prompt"]["text"] assert "Go GC pause 耗时" in case.data["prompt"]["text"] assert "所有 TiKV 节点的 kv_get gRPC request handle 时间均值" in case.data["prompt"]["text"] assert "所有 PD 节点的 TSO request handle 时间均值" in case.data["prompt"]["text"] assert "保留 6 位小数" in case.data["prompt"]["text"] assert "整包都是空序列" in case.data["prompt"]["text"] def test_top_cpu_overload_runtime_detects_error_warn_log_levels(): hook_path = ( default_registry_root() / "cases" / "top-cpu-overload-nodes-and-log-patterns-in-saas01-during-latest-window" / "runtime_hook.py" ) spec = importlib.util.spec_from_file_location("cpu_overload_runtime_hook", hook_path) assert spec is not None assert spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) assert r"\[warn(?:ing)?\]" in module._node_log_queries("db-tidb-1")[0][1] assert module._error_warn_level("[WARN] ddl worker busy") == "WARN" assert module._error_warn_level("[WARNING] ddl worker busy") == "WARN" assert module._error_warn_level("[ERROR] schema sync failed") == "ERROR" assert module._error_warn_level("[INFO] heartbeat ok") == "" def test_top_tidb_p99_latency_runtime_sorts_by_full_precision_before_rounding(): hook_path = ( default_registry_root() / "cases" / "top-tidb-p99-latency-nodes-in-saas01-during-latest-window" / "runtime_hook.py" ) spec = importlib.util.spec_from_file_location("top_tidb_p99_runtime_hook", hook_path) assert spec is not None assert spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) payload = { "data": { "result": [ { "metric": {"instance": "db-tidb-95"}, "values": [[1, "0.04478741"], [2, "0.04478741"]], }, { "metric": {"instance": "db-tidb-87"}, "values": [[1, "0.04478731"], [2, "0.04478731"]], }, { "metric": {"instance": "db-tidb-5"}, "values": [[1, "0.04467371"], [2, "0.04467371"]], }, ] } } assert module._rank_tidb_nodes_by_avg_p99(payload) == [ {"node": "db-tidb-95", "avg_p99_latency_s": 0.044787}, {"node": "db-tidb-87", "avg_p99_latency_s": 0.044787}, {"node": "db-tidb-5", "avg_p99_latency_s": 0.044674}, ] def test_top_tidb_p99_latency_runtime_limits_breakdown_queries_to_top_nodes(): hook_path = ( default_registry_root() / "cases" / "top-tidb-p99-latency-nodes-in-saas01-during-latest-window" / "runtime_hook.py" ) spec = importlib.util.spec_from_file_location("top_tidb_p99_runtime_hook", hook_path) assert spec is not None assert spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) promql = module._p99_histogram_promql( "tidb_tikvclient_request_seconds_bucket", nodes=["db-tidb-95", "db-tidb-87", "db-tidb-5"], extra_matchers=('type="Get"',), ) assert 'instance=~"^(db-tidb-95|db-tidb-87|db-tidb-5)($|[-.:].*)"' in promql assert 'type="Get"' in promql assert "tidb_tikvclient_request_seconds_bucket" in promql def test_top_tidb_p99_latency_runtime_uses_tidb_client_side_tso_metric(): hook_path = ( default_registry_root() / "cases" / "top-tidb-p99-latency-nodes-in-saas01-during-latest-window" / "runtime_hook.py" ) spec = importlib.util.spec_from_file_location("top_tidb_p99_runtime_hook", hook_path) assert spec is not None assert spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) promql = module._p99_histogram_promql( "pd_client_request_handle_requests_duration_seconds_bucket", nodes=["db-tidb-95", "db-tidb-87", "db-tidb-5"], extra_matchers=('type="tso"',), ) assert 'instance=~"^(db-tidb-95|db-tidb-87|db-tidb-5)($|[-.:].*)"' in promql assert 'type="tso"' in promql assert "pd_client_request_handle_requests_duration_seconds_bucket" in promql def test_top_tidb_p99_latency_runtime_uses_cluster_side_server_metrics(): hook_path = ( default_registry_root() / "cases" / "top-tidb-p99-latency-nodes-in-saas01-during-latest-window" / "runtime_hook.py" ) spec = importlib.util.spec_from_file_location("top_tidb_p99_runtime_hook", hook_path) assert spec is not None assert spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) tikv_promql = module._p99_histogram_promql( "tikv_grpc_msg_duration_seconds_bucket", extra_matchers=('type="kv_get"',), ) pd_promql = module._p99_histogram_promql( "pd_server_handle_tso_duration_seconds_bucket", ) assert 'type="kv_get"' in tikv_promql assert "tikv_grpc_msg_duration_seconds_bucket" in tikv_promql assert "pd_server_handle_tso_duration_seconds_bucket" in pd_promql def test_case_can_declare_runtime_hooks_and_resolved_time_window_output(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "runtime-case" case_dir.mkdir(parents=True) real_case = load_case("cpu-spike-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "runtime-case" payload["runtime"] = { "hooks": [ { "id": "resolve-window", "stage": "before_acp", "kind": "python_script", "path": "runtime_hook.py", "output_id": "resolved_window", "args": ["--cluster", "{{environment.cluster_id}}"], } ] } payload["time_window"]["resolved_by_output"] = "resolved_window" (case_dir / "runtime_hook.py").write_text("print('{}')\n", encoding="utf-8") _write_case_files(case_dir, payload) case = load_case("runtime-case", registry_root=root) assert case.runtime["hooks"][0]["id"] == "resolve-window" assert case.runtime_hooks[0]["stage"] == "before_acp" assert case.runtime_hooks[0]["kind"] == "python_script" assert case.runtime_hooks[0]["output_id"] == "resolved_window" assert case.resolved_time_window_output_id == "resolved_window" def test_case_can_use_pure_runtime_resolved_time_window(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "runtime-only-window-case" case_dir.mkdir(parents=True) real_case = load_case("cpu-spike-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "runtime-only-window-case" payload["time_window"] = { "resolved_by_output": "resolved_window", } payload["runtime"] = { "hooks": [ { "id": "resolve-window", "stage": "before_acp", "kind": "python_script", "path": "runtime_hook.py", "output_id": "resolved_window", } ] } (case_dir / "runtime_hook.py").write_text("print('{}')\n", encoding="utf-8") _write_case_files(case_dir, payload) case = load_case("runtime-only-window-case", registry_root=root) assert case.resolved_time_window_output_id == "resolved_window" assert "incident_at" not in case.data["time_window"] assert "start" not in case.data["time_window"] assert "end" not in case.data["time_window"] def test_rejects_runtime_hook_invalid_stage(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("cpu-spike-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["runtime"] = { "hooks": [ { "id": "resolve-window", "stage": "after_answer", "kind": "python_script", "path": "runtime_hook.py", "output_id": "resolved_window", } ] } (case_dir / "runtime_hook.py").write_text("print('{}')\n", encoding="utf-8") _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="runtime.hooks\\[0\\]\\.stage must be one of"): load_case("bad-case", registry_root=root) def test_rejects_time_window_resolved_by_output_without_declared_hook(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("cpu-spike-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["runtime"] = { "hooks": [ { "id": "resolve-window", "stage": "before_acp", "kind": "python_script", "path": "runtime_hook.py", "output_id": "another_window", } ] } payload["time_window"]["resolved_by_output"] = "resolved_window" (case_dir / "runtime_hook.py").write_text("print('{}')\n", encoding="utf-8") _write_case_files(case_dir, payload) with pytest.raises( EvalRegistryError, match="time_window.resolved_by_output must reference a declared runtime hook output_id", ): load_case("bad-case", registry_root=root) def test_accepts_output_json_reference_rule_without_reference_path(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["runtime"] = { "hooks": [ { "id": "build-reference", "stage": "before_judge", "kind": "python_script", "path": "runtime_hook.py", "output_id": "reference", } ] } payload["judge"]["diagnosis"]["checks"] = [ { "id": "match-reference-json", "mode": "deterministic_only", "weight": 3, "required": True, "description": "Compares output JSON to runtime reference JSON.", "deterministic": { "match": "all_of", "rules": [ { "source": "output_json", "matcher": "json_path_equals_reference", "path": "cluster_id", "reference_output_id": "reference", } ], }, } ] (case_dir / "runtime_hook.py").write_text("print('{}')\n", encoding="utf-8") _write_case_files(case_dir, payload) case = load_case("bad-case", registry_root=root) rules = case.diagnosis_checks[0]["deterministic"]["rules"] assert rules[0]["reference_output_id"] == "reference" assert "reference_path" not in rules[0] def test_rejects_output_json_reference_rule_with_unknown_runtime_output(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["runtime"] = { "hooks": [ { "id": "resolve-window", "stage": "before_judge", "kind": "python_script", "path": "runtime_hook.py", "output_id": "known_reference", } ] } payload["judge"]["diagnosis"]["checks"] = [ { "id": "match-reference-json", "mode": "deterministic_only", "weight": 3, "required": True, "description": "Compares output JSON to runtime reference JSON.", "deterministic": { "match": "all_of", "rules": [ { "source": "output_json", "matcher": "json_path_equals_reference", "path": "cluster_id", "reference_output_id": "unknown_reference", } ], }, } ] (case_dir / "runtime_hook.py").write_text("print('{}')\n", encoding="utf-8") _write_case_files(case_dir, payload) with pytest.raises( EvalRegistryError, match="reference_output_id must reference a declared runtime hook output_id", ): load_case("bad-case", registry_root=root) def test_rejects_non_integer_execution_timeout_hint(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("cpu-spike-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["execution"]["acp_bash_timeout_s"] = "300" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="execution.acp_bash_timeout_s must be an integer"): load_case("bad-case", registry_root=root) def test_rejects_non_integer_execution_idle_timeout_hint(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("cpu-spike-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["execution"]["acp_idle_timeout_s"] = "600" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="execution.acp_idle_timeout_s must be an integer"): load_case("bad-case", registry_root=root) def test_rejects_non_integer_execution_total_timeout_hint(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("cpu-spike-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["execution"]["acp_timeout_s"] = "1200" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="execution.acp_timeout_s must be an integer"): load_case("bad-case", registry_root=root) @pytest.mark.parametrize("mode", ["adhoc_evidence", "evidence_only", "diagnosis_only"]) def test_case_allows_supported_evaluation_modes(tmp_path: Path, mode: str): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("cpu-spike-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["evaluation"] = {"mode": mode} _write_case_files(case_dir, payload) case = load_case("bad-case", registry_root=root) assert case.evaluation_mode == mode def test_rejects_unknown_evaluation_mode(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("cpu-spike-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["evaluation"] = {"mode": "unexpected_mode"} _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="evaluation.mode must be one of"): load_case("bad-case", registry_root=root) def test_rejects_judge_embedded_in_case_yaml(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" (case_dir / "case.yaml").write_text(yaml.safe_dump(payload, sort_keys=False)) (case_dir / "judge.yaml").write_text( yaml.safe_dump( { "schema_version": payload["schema_version"], "case_id": "bad-case", "ground_truth": payload["ground_truth"], "judge": payload["judge"], }, sort_keys=False, ) ) (case_dir / "expected.md").write_text("expected\n") with pytest.raises(EvalRegistryError, match="judge is not allowed; move it to judge.yaml"): load_case("bad-case", registry_root=root) def test_rejects_unknown_execution_contract_workflow_check(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["execution_contract"]["checks"][0]["rules"][0]["paths"][0] = "unknown_workflow_check" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="supported workflow_check id"): load_case("bad-case", registry_root=root) def test_rejects_unknown_judge_contract_rule_source(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["contract"]["checks"][0]["rules"][0]["source"] = "routing_fact" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="judge.contract.checks\\[0\\].rules\\[0\\].source must be one of"): load_case("bad-case", registry_root=root) def test_rejects_unknown_workflow_fact_path(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["workflow"]["checks"][0]["rules"][0]["source"] = "workflow_fact" payload["judge"]["workflow"]["checks"][0]["rules"][0]["path"] = "bad_path" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="supported workflow_fact id"): load_case("bad-case", registry_root=root) def test_rejects_unknown_trace_fact_path(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["contract"]["checks"][0]["rules"][0]["path"] = "window_match" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="supported trace_fact id"): load_case("bad-case", registry_root=root) def test_rejects_unknown_judge_evidence_collection_rule_source(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["evidence_collection"]["checks"][0]["rules"][0]["source"] = "routing_fact" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="judge.evidence_collection.checks\\[0\\].rules\\[0\\].source must be one of"): load_case("bad-case", registry_root=root) def test_rejects_unknown_channel_fact_path_in_evidence_collection(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["evidence_collection"]["checks"][1]["rules"][0]["paths"][1] = "metrics" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="supported channel_fact id"): load_case("bad-case", registry_root=root) def test_rejects_unknown_evidence_fact_path_in_evidence_collection(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["evidence_collection"]["checks"][0]["rules"][0]["source"] = "evidence_fact" payload["judge"]["evidence_collection"]["checks"][0]["rules"][0]["path"] = "unknown_fact" payload["judge"]["evidence_collection"]["checks"][0]["rules"][0].pop("paths", None) payload["judge"]["evidence_collection"]["checks"][0]["rules"][0]["matcher"] = "bool_true" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="supported evidence_fact id"): load_case("bad-case", registry_root=root) def test_rejects_count_gte_without_threshold_in_evidence_collection(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["evidence_collection"]["checks"][1]["rules"][0].pop("threshold", None) _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="threshold must be a number"): load_case("bad-case", registry_root=root) def test_rejects_unknown_critical_evidence_rule_source(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["critical_evidence"]["checks"][0]["rules"][0]["source"] = "metrics" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="source must be one of"): load_case("bad-case", registry_root=root) def test_rejects_unknown_critical_evidence_artifact_id(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["critical_evidence"]["checks"][0]["rules"][0]["artifacts"] = ["restart_metrics"] _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="artifacts\\[0\\] must be one of"): load_case("bad-case", registry_root=root) def test_rejects_artifact_json_keyword_any_without_artifacts(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["critical_evidence"]["checks"][0]["rules"][0]["matcher"] = "keyword_any" payload["judge"]["critical_evidence"]["checks"][0]["rules"][0]["keywords"] = ["panic"] payload["judge"]["critical_evidence"]["checks"][0]["rules"][0].pop("artifacts", None) _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="artifacts must be a non-empty list"): load_case("bad-case", registry_root=root) def test_accepts_summary_json_as_critical_evidence_artifact(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["critical_evidence"]["checks"][0]["rules"][0]["artifacts"] = ["summary_json"] _write_case_files(case_dir, payload) case = load_case("bad-case", registry_root=root) assert case.id == "bad-case" def test_accepts_new_critical_evidence_json_value_matchers(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["critical_evidence"]["checks"][0]["rules"][0] = { "source": "artifact_json", "artifacts": ["queries"], "matcher": "json_values_contains_any", "path": "slow_queries", "keywords": ["stats:pseudo"], } payload["judge"]["critical_evidence"]["checks"][1]["rules"][0] = { "source": "artifact_json", "artifacts": ["component_health"], "matcher": "json_numeric_any_lt", "path": "component_health.component_count.changes", "threshold": 0, } payload["judge"]["critical_evidence"]["checks"][2]["rules"][0] = { "source": "artifact_json", "artifacts": ["metrics"], "matcher": "json_values_not_contains_any", "path": "metrics.promql", "keywords": ['instance=~"db-tikv-9.*"'], } _write_case_files(case_dir, payload) case = load_case("bad-case", registry_root=root) assert case.id == "bad-case" def test_rejects_unknown_evidence_routing_rule_source(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["evidence_routing"]["checks"][0]["rules"][0]["source"] = "output_fact" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="judge.evidence_routing.checks\\[0\\].rules\\[0\\].source must be one of"): load_case("bad-case", registry_root=root) def test_rejects_unknown_routing_fact_path(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["evidence_routing"]["checks"][2]["rules"][1]["path"] = "analysis_patch_shape_valid" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="supported routing_fact id"): load_case("bad-case", registry_root=root) def test_rejects_any_true_without_paths_in_evidence_routing(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["evidence_routing"]["checks"][1]["rules"][1] = { "source": "routing_fact", "matcher": "any_true", } _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="paths must be a non-empty list"): load_case("bad-case", registry_root=root) def test_rejects_unknown_unified_diagnosis_rule_source(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["diagnosis"]["checks"][1]["deterministic"]["rules"][0]["source"] = "unknown_text" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="judge.diagnosis.checks\\[1\\].deterministic.rules\\[0\\].source must be one of"): load_case("bad-case", registry_root=root) def test_rejects_any_true_without_paths_in_unified_diagnosis_check(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["diagnosis"]["checks"][0]["deterministic"]["rules"][0].pop("paths", None) _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="paths must be a non-empty list"): load_case("bad-case", registry_root=root) def test_rejects_unknown_output_fact_path_in_unified_diagnosis_check(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["diagnosis"]["checks"][1]["deterministic"]["rules"][0]["path"] = "mentions_6395_typo" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="supported output_fact id"): load_case("bad-case", registry_root=root) def test_rejects_unknown_output_fact_path_in_any_true_rule(tmp_path: Path): root = tmp_path / "eval-registry" case_dir = root / "cases" / "bad-case" case_dir.mkdir(parents=True) real_case = load_case("component-restart-during-incident-window") payload = copy.deepcopy(real_case.data) payload["id"] = "bad-case" payload["judge"]["diagnosis"]["checks"][0]["deterministic"]["rules"][0]["paths"][1] = "mentions_without_fix_typo" _write_case_files(case_dir, payload) with pytest.raises(EvalRegistryError, match="supported output_fact id"): load_case("bad-case", registry_root=root) def test_rejects_duplicate_case_ids(tmp_path: Path): root = tmp_path / "eval-registry" (root / "suites").mkdir(parents=True) (root / "suites" / "bad.yaml").write_text( "\n".join( [ "schema_version: 1", "id: bad", "cases:", " - A", " - A", ] ) ) with pytest.raises(EvalRegistryError, match="duplicate case ids"): load_suite("bad", registry_root=root) def test_default_registry_root_prefers_env_override(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): root = tmp_path / "custom-regression" (root / "suites").mkdir(parents=True) monkeypatch.setenv("EVAL_REGISTRY_ROOT", os.fspath(root)) assert default_registry_root() == root.resolve()