import json
import shutil
from copy import deepcopy
from dataclasses import replace
from pathlib import Path

from investibook.core.eval.eval_judge import (
    CRITICAL_EVIDENCE_BUCKET_MAX,
    DIAGNOSIS_BUCKET_MAX,
    EVIDENCE_COLLECTION_BUCKET_MAX,
    FAILING_HARD_GATE_SCORE_CAP,
    PASSING_HARD_GATE_SCORE_FLOOR,
    WORKFLOW_BUCKET_MAX,
    _apply_score_band,
    _extract_classification_contract_facts,
    _extract_output_json_for_evaluation_mode,
    _extract_output_json,
    _normalize_total_score,
    judge_case_output,
)
from investibook.core.eval.eval_registry import load_case


PASSING_OUTPUT = """
TiDB v8.5.2 known bug GitHub #63956.
KILL QUERY on an idle transaction triggers single-instance panic/restart in memory.(*Tracker).Consume during rollback cleanup.
Upgrade to v8.5.3.
"""


def _classification_artifact(*, agent: str, selected_skills: list[str]) -> dict:
    return {
        "agent": agent,
        "signal_tags": [
            {
                "id": "signal",
                "source": "evidence",
                "component": "tidb",
                "metric_or_signal": "latency",
                "direction": "increased",
                "baseline": "before_after_event",
                "severity": "high",
                "confidence": "high",
                "evidence_ref": "evidences/metrics.json",
            }
        ],
        "fault_classification": {
            "primary_class": "sql_plan_or_query",
            "candidates": [
                {
                    "class": "sql_plan_or_query",
                    "confidence": 0.9,
                    "role": "primary",
                    "evidence_for": ["signal"],
                    "evidence_against": [],
                }
            ],
            "rejected_classes": [],
            "classification_confidence": "high",
            "missing_evidence_for_higher_confidence": [],
        },
        "selected_diagnosis_skills": selected_skills,
    }


def test_classification_contract_requires_real_analyst_and_expected_specialist(tmp_path):
    artifact_path = tmp_path / "diagnosis-analyst.json"
    expected_routes = [
        {"any_of": ["tidb-query-tuning-autox", "tidb-query-tuning"]}
    ]

    artifact_path.write_text(
        json.dumps(
            _classification_artifact(
                agent="diagnosis-analyst",
                selected_skills=["tidb-clinic-diagnosis", "tidb-query-tuning-autox"],
            )
        )
    )
    facts = _extract_classification_contract_facts(
        artifact_path,
        expected_diagnosis_skill_routes=expected_routes,
        diagnosis_analyst_dispatched=True,
    )
    assert facts["diagnosis_analyst_authored"] is True
    assert facts["diagnosis_analyst_dispatched"] is True
    assert facts["expected_diagnosis_skill_routes_selected"] is True
    assert facts["missing_expected_diagnosis_skill_routes"] == []

    artifact_path.write_text(
        json.dumps(
            _classification_artifact(
                agent="diagnosis-analyst",
                selected_skills=[
                    "tidb-query-tuning-autox (SQL specialist - selected after classification)"
                ],
            )
        )
    )
    facts = _extract_classification_contract_facts(
        artifact_path,
        expected_diagnosis_skill_routes=expected_routes,
        diagnosis_analyst_dispatched=True,
    )
    assert facts["expected_diagnosis_skill_routes_selected"] is True
    assert facts["selected_diagnosis_skill_ids"] == ["tidb-query-tuning-autox"]

    artifact_path.write_text(
        json.dumps(
            _classification_artifact(
                agent="investigator",
                selected_skills=["tidb-clinic-diagnosis", "transaction-oncall-handbook"],
            )
        )
    )
    facts = _extract_classification_contract_facts(
        artifact_path,
        expected_diagnosis_skill_routes=expected_routes,
    )
    assert facts["diagnosis_analyst_contract_present"] is True
    assert facts["diagnosis_analyst_authored"] is False
    assert facts["diagnosis_analyst_dispatched"] is False
    assert facts["expected_diagnosis_skill_routes_selected"] is False
    assert facts["missing_expected_diagnosis_skill_routes"] == [
        ["tidb-query-tuning-autox", "tidb-query-tuning"]
    ]


def test_required_specialist_routing_contract_blocks_bypass(tmp_path):
    case = load_case("component-restart-during-incident-window")
    data = deepcopy(case.data)
    data["ground_truth"]["expected_diagnosis_skill_routes"] = [
        {"any_of": ["tidb-query-tuning-autox", "tidb-query-tuning"]}
    ]
    case = replace(case, data=data)
    case = _case_with_evidence_routing(
        case,
        evidence_routing={
            "weight": 5,
            "checks": [
                {
                    "id": "specialist_routing_contract",
                    "weight": 5,
                    "required": True,
                    "description": "real analyst and expected specialist are required",
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "diagnosis_analyst_dispatched",
                        },
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "expected_diagnosis_skill_routes_selected",
                        },
                    ],
                }
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_analyst_payload=_classification_artifact(
            agent="diagnosis-analyst",
            selected_skills=["tidb-clinic-diagnosis", "tidb-query-tuning-autox"],
        ),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-specialist-bypass",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["evidence_routing"]["status"] == "fail"
    assert judgment["facts"]["routing"]["diagnosis_analyst_authored"] is True
    assert judgment["facts"]["routing"]["diagnosis_analyst_dispatched"] is False

    dispatch_path = audit_dir / "runtime" / "agent_dispatches.json"
    dispatch_path.parent.mkdir(parents=True, exist_ok=True)
    dispatch_path.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "source": "messages_v2_acp_metadata",
                "subagent_executions": [
                    {
                        "subagent_name": "diagnosis-analyst",
                        "parent_task_call_id": "task-diagnosis-1",
                        "subagent_session_id": "session-diagnosis-1",
                    }
                ],
            }
        ),
        encoding="utf-8",
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-specialist-dispatched",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["evidence_routing"]["status"] == "pass"
    assert judgment["facts"]["routing"]["diagnosis_analyst_dispatched"] is True


def test_web_audit_db_dump_proves_linked_analyst_dispatch_for_same_workspace(tmp_path):
    case = load_case("component-restart-during-incident-window")
    data = deepcopy(case.data)
    data["ground_truth"]["expected_diagnosis_skill_routes"] = [
        {"any_of": ["tidb-query-tuning-autox"]}
    ]
    case = replace(case, data=data)
    case = _case_with_evidence_routing(
        case,
        evidence_routing={
            "weight": 5,
            "checks": [
                {
                    "id": "specialist_routing_contract",
                    "weight": 5,
                    "required": True,
                    "description": "real analyst and expected specialist are required",
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "diagnosis_analyst_dispatched",
                        },
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "expected_diagnosis_skill_routes_selected",
                        },
                    ],
                }
            ],
        },
    )
    workspace_id = "ws-web-e2e"
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_analyst_payload=_classification_artifact(
            agent="diagnosis-analyst",
            selected_skills=["tidb-query-tuning-autox (SQL specialist)"],
        ),
    )
    (audit_dir / "db_dump.json").write_text(
        json.dumps(
            {
                "workspace_id": workspace_id,
                "messages_v2": [
                    {
                        "workspace_id": workspace_id,
                        "role": "assistant",
                        "parts": json.dumps(
                            [
                                {
                                    "type": "tool",
                                    "state": {
                                        "meta": {
                                            "opencode/subagent": {
                                                "subagentName": "diagnosis-analyst",
                                                "parentTaskCallId": "task-diagnosis-1",
                                                "subagentSessionId": "session-diagnosis-1",
                                            }
                                        }
                                    },
                                }
                            ]
                        ),
                    }
                ],
            }
        ),
        encoding="utf-8",
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id=workspace_id,
        audit_dir=audit_dir,
    )
    assert judgment["packet"]["evidence_routing"]["status"] == "pass"
    assert judgment["facts"]["routing"]["diagnosis_analyst_dispatched"] is True

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-different",
        audit_dir=audit_dir,
    )
    assert judgment["packet"]["evidence_routing"]["status"] == "fail"
    assert judgment["facts"]["routing"]["diagnosis_analyst_dispatched"] is False


def _default_telemetry(case_id: str) -> list[str]:
    return [
        f"issues/{case_id}/issue_meta.json",
        f"issues/{case_id}/cluster_meta.json",
        "tier-architecture.json",
        "incorrect-conclusions.json",
        "Initialize diagnosis run directory",
        "collect_evidence.py",
    ]


def _setup_audit_run(
    tmp_path,
    case_id: str,
    *,
    telemetry_lines: list[str] | None = None,
    include_diagnosis_report: bool = True,
    include_queries: bool = True,
    queries_payload: dict | None = None,
    include_gate_checks: bool = True,
    gate_checks_payload: dict | None = None,
    include_collection_errors: bool = True,
    collection_errors_payload: dict | None = None,
    include_log_stats: bool = True,
    log_stats_payload: dict | None = None,
    include_logs_dir: bool = True,
    logs_files: dict[str, str] | None = None,
    include_summary_artifact: bool = False,
    component_health_payload: dict | None = None,
    metric_files: dict[str, dict] | None = None,
    analysis_patch_text: str | None = None,
    diagnosis_report_payload: dict | None = None,
    diagnosis_analyst_payload: dict | None = None,
):
    audit_dir = tmp_path / "audit" / "evalrun-1" / "evalcase-1"
    evidence_dir = audit_dir / "evidences"
    run_dir = evidence_dir
    logs_dir = evidence_dir / "logs"
    evidence_dir.mkdir(parents=True)
    if include_logs_dir:
        logs_dir.mkdir(parents=True, exist_ok=True)
        for filename, content in (
            logs_files
            or {
                "loki-targeted-panic.log": "\n".join(
                    [
                        "panic: memory.(*Tracker).Consume",
                        "kill query",
                        "query execution was interrupted",
                        "sqlkiller.go",
                    ]
                )
            }
        ).items():
            (logs_dir / filename).write_text(content)

    if include_diagnosis_report:
        payload = diagnosis_report_payload or {
            "metadata": {
                "cluster_id": "10766841556523235152",
                "time_range": {"start": "1775039760", "end": "1775043360"},
                "run_dir": str(run_dir),
                "evidence_dir": str(evidence_dir),
            },
            "evidence": {
                "metrics": [],
            },
        }
        if isinstance(payload.get("metadata"), dict):
            payload["metadata"].setdefault("run_dir", str(run_dir))
            payload["metadata"].setdefault("evidence_dir", str(evidence_dir))
        payload.setdefault("type", "workspace_evidence_report")
        (run_dir / "workspace_evidence_report.json").write_text(json.dumps(payload))

    if include_summary_artifact:
        (run_dir / "evidence_summary_for_agent.txt").write_text("panic: [executor:1317]")

    (run_dir / "evidence_summary_for_agent.json").write_text(
        json.dumps(
            {
                "metadata": {
                    "cluster_id": "10766841556523235152",
                    "time_range": {"start": "1775039760", "end": "1775043360"},
                    "run_dir": str(run_dir),
                }
            }
        )
    )

    if include_queries:
        payload = queries_payload or {
            "slow_queries": [{"digest": "abc", "source": "slow_query"}],
            "topsql": [],
            "decoded_plan": {},
        }
        (evidence_dir / "queries.json").write_text(json.dumps(payload))

    (evidence_dir / "metrics.json").write_text(json.dumps({"signals": []}))

    if include_gate_checks:
        payload = gate_checks_payload or {"infra": "INFRA CLEAR", "hotspot": "HOTSPOT: not detected"}
        (evidence_dir / "gate_checks.json").write_text(json.dumps(payload))

    if include_collection_errors:
        payload = collection_errors_payload or {"errors": []}
        (evidence_dir / "collection_errors.json").write_text(json.dumps(payload))

    if include_log_stats:
        payload = log_stats_payload or {"summary": {"tidb": {"error": 0}}}
        (evidence_dir / "logs.json").write_text(json.dumps(payload))

    component_health = component_health_payload or {
        "component_health": {
            "query_scope": "10766841556523235152",
            "component_count": [],
            "restarts": [
                {
                    "component": "tidb",
                    "instance": "db-tidb-60",
                    "time": 1775043348,
                    "reason": "panic restart",
                }
            ],
            "worker_memory": [],
            "links": {},
        }
    }
    (evidence_dir / "component_health.json").write_text(json.dumps(component_health))

    metrics_dir = run_dir / "metrics"
    if metric_files:
        metrics_dir.mkdir(parents=True, exist_ok=True)
        for filename, payload in metric_files.items():
            (metrics_dir / filename).write_text(json.dumps(payload))

    if analysis_patch_text is not None:
        (run_dir / "analysis_patch.json").write_text(analysis_patch_text)
    if diagnosis_analyst_payload is not None:
        agent_runs_dir = run_dir / "agent-runs"
        agent_runs_dir.mkdir(parents=True, exist_ok=True)
        (agent_runs_dir / "diagnosis-analyst.json").write_text(json.dumps(diagnosis_analyst_payload))

    (audit_dir / ".opencode").mkdir(parents=True, exist_ok=True)
    lines = telemetry_lines if telemetry_lines is not None else _default_telemetry(case_id)
    (audit_dir / ".opencode" / "telemetry.log").write_text("\n".join(lines))
    return audit_dir


def _evidence_run_dir(audit_dir: Path) -> Path:
    return audit_dir / "evidences"


def _evidence_file(audit_dir: Path, *parts: str) -> Path:
    return _evidence_run_dir(audit_dir).joinpath(*parts)


def _minimal_analysis_patch() -> str:
    return json.dumps(
        {
            "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
            "supporting_evidence": [{"type": "log", "name": "warn", "reason": "corroborating logs"}],
        }
    )


def test_apply_score_band_assigns_zero_to_invalid_runs():
    assert _apply_score_band(raw_total_score=57, overall_verdict="invalid", hard_gate_passed=False) == 0


def test_apply_score_band_promotes_hard_gate_passes_into_70_plus_band():
    assert _apply_score_band(raw_total_score=62, overall_verdict="pass", hard_gate_passed=True) == PASSING_HARD_GATE_SCORE_FLOOR


def test_apply_score_band_caps_non_invalid_hard_gate_misses_below_70():
    assert _apply_score_band(raw_total_score=88, overall_verdict="fail", hard_gate_passed=False) == FAILING_HARD_GATE_SCORE_CAP


def test_normalize_total_score_maps_active_case_max_to_100():
    assert _normalize_total_score(raw_total_score=85, total_max_score=85) == 100


def test_normalize_total_score_scales_partial_score_with_active_case_max():
    assert _normalize_total_score(raw_total_score=42, total_max_score=84) == 50


def _semantic_packet_fixture_text() -> str:
    return json.dumps(
        {
            "version": "semantic_packet_v1",
            "model": {
                "provider": "heuristic",
                "model": "semantic-shadow-test",
                "prompt_version": "semantic_case_local_v1",
                "schema_version": "semantic_packet_v1",
            },
            "status": "ok",
            "case_id": "component-restart-during-incident-window",
            "check_assessments": [
                {
                    "check_id": 1,
                    "status": "pass",
                    "confidence": 0.91,
                    "reason": "primary path grounded",
                    "evidence_refs": ["restart_signal_present", "panic_signal_present"],
                    "notes": [],
                },
                {
                    "check_id": 2,
                    "status": "pass",
                    "confidence": 0.9,
                    "reason": "chain complete",
                    "evidence_refs": ["query_interruption_signal_present", "ke-2"],
                    "notes": [],
                },
                {
                    "check_id": 3,
                    "status": "pass",
                    "confidence": 0.88,
                    "reason": "single instance scope aligned",
                    "evidence_refs": ["ke-1", "ke-3"],
                    "notes": [],
                },
                {
                    "check_id": 4,
                    "status": "pass",
                    "confidence": 0.93,
                    "reason": "remediation aligned",
                    "evidence_refs": [],
                    "notes": [],
                },
            ],
            "disagreement_flags": [],
            "notes": [],
        }
    )


def _case_with_execution_contract(
    case,
    *,
    execution_contract: dict | None,
):
    data = deepcopy(case.data)
    judge = data.setdefault("judge", {})
    if execution_contract is None:
        judge.pop("execution_contract", None)
    else:
        judge["execution_contract"] = execution_contract
    return replace(case, data=data)


def _case_with_judge_contract(case, *, judge_contract: dict | None):
    data = deepcopy(case.data)
    judge = data.setdefault("judge", {})
    if judge_contract is None:
        judge.pop("contract", None)
    else:
        judge["contract"] = judge_contract
    return replace(case, data=data)


def _case_with_judge_workflow(case, *, judge_workflow: dict | None):
    data = deepcopy(case.data)
    judge = data.setdefault("judge", {})
    if judge_workflow is None:
        judge.pop("workflow", None)
    else:
        judge["workflow"] = judge_workflow
    return replace(case, data=data)


def _case_with_judge_evidence_collection(case, *, evidence_collection: dict | None):
    data = deepcopy(case.data)
    judge = data.setdefault("judge", {})
    if evidence_collection is None:
        judge.pop("evidence_collection", None)
    else:
        judge["evidence_collection"] = evidence_collection
    return replace(case, data=data)


def _case_with_diagnosis_supporting(case, *, supporting: dict | None):
    data = deepcopy(case.data)
    judge = data.setdefault("judge", {})
    if supporting is None:
        diagnosis = judge.get("diagnosis")
        if isinstance(diagnosis, dict):
            diagnosis.pop("supporting", None)
            if not diagnosis:
                judge.pop("diagnosis", None)
    else:
        judge.setdefault("diagnosis", {})["supporting"] = supporting
    return replace(case, data=data)


def _case_with_diagnosis_hard_checks(case, *, hard_checks: list[dict] | None):
    data = deepcopy(case.data)
    judge = data.setdefault("judge", {})
    diagnosis = judge.setdefault("diagnosis", {})
    if hard_checks is None:
        diagnosis.pop("hard_checks", None)
        if not diagnosis:
            judge.pop("diagnosis", None)
    else:
        diagnosis["hard_checks"] = hard_checks
    return replace(case, data=data)


def _case_with_diagnosis_checks(case, *, checks: list[dict] | None):
    data = deepcopy(case.data)
    judge = data.setdefault("judge", {})
    diagnosis = judge.setdefault("diagnosis", {})
    if checks is None:
        diagnosis.pop("checks", None)
        if not diagnosis:
            judge.pop("diagnosis", None)
    else:
        diagnosis["checks"] = checks
    return replace(case, data=data)


def _case_with_evidence_routing(case, *, evidence_routing: dict | None):
    data = deepcopy(case.data)
    judge = data.setdefault("judge", {})
    if evidence_routing is None:
        judge.pop("evidence_routing", None)
    else:
        judge["evidence_routing"] = evidence_routing
    return replace(case, data=data)


def _case_with_critical_evidence(case, *, critical_evidence: dict | None):
    data = deepcopy(case.data)
    judge = data.setdefault("judge", {})
    if critical_evidence is None:
        judge.pop("critical_evidence", None)
    else:
        judge["critical_evidence"] = critical_evidence
    return replace(case, data=data)


def _realistic_restart_audit_sample() -> dict:
    slow_digest = "7a117029ace0cf816155a39a163838a1b0e32ca78de6da0830168baa1b1f6102"
    return {
        "diagnosis_report_payload": {
            "metadata": {
                "cluster_id": "10766841556523235152",
                "time_range": {"start": "1775039760", "end": "1775043360"},
            },
            "evidence": {
                "metrics": [
                    {
                        "name": "TiDB P99 Latency Avg",
                        "metric_group_id": "tidb_p99_latency",
                        "instance": "all (30 instances)",
                        "peak": 0.011,
                        "mean": 0.009,
                        "unit": "seconds",
                    },
                    {
                        "name": "TiDB P99 Latency",
                        "metric_group_id": "tidb_p99_latency",
                        "instance": "db-tidb-70",
                        "peak": 0.015,
                        "mean": 0.012,
                        "unit": "seconds",
                    },
                    {
                        "name": "TIKV CPU Avg",
                        "component": "tikv",
                        "metric_group_id": "tikv_cpu:db",
                        "instance": "all (db, 15 instances)",
                        "nodegroup": "db",
                        "peak": 4.32,
                        "mean": 2.55,
                        "unit": "cores",
                    },
                    {
                        "name": "TIKV CPU",
                        "component": "tikv",
                        "metric_group_id": "tikv_cpu:db",
                        "instance": "db-tikv-51",
                        "nodegroup": "db",
                        "peak": 7.17,
                        "quota": 16.0,
                        "utilization": "45%",
                        "unit": "cores",
                    },
                ],
            },
        },
        "queries_payload": {
            "slow_queries": [
                {
                    "source": "slow_query",
                    "digest": slow_digest,
                    "sql_preview": "select ... from account_ranking where joined_account_id = ?",
                    "exec_count": 6219,
                    "avg_time": "0.34s",
                    "max_time": "0.69s",
                    "max_mem": 46668,
                }
            ],
            "topsql": [],
            "decoded_plan": {
                "digest": slow_digest,
                "source": "slow_query",
                "indexes_used": "None",
                "process_keys": "986360",
            },
        },
        "log_stats_payload": {
            "timeline": {
                "tidb": {"error": [], "warn": [], "info": []},
                "tikv": {"error": [], "warn": [], "info": []},
                "pd": {"error": [], "warn": [], "info": []},
            },
            "summary": {
                "tidb": {"error": 0, "warn": 0, "info": 0},
                "tikv": {"error": 0, "warn": 0, "info": 0},
                "pd": {"error": 0, "warn": 0, "info": 0},
            },
            "step_seconds": 300,
            "errors": [
                "A6:tidb_info: HTTP 400: parse error at line 1, col 101: invalid char escape",
                "A6:tikv_info: HTTP 400: parse error at line 1, col 101: invalid char escape",
            ],
            "loki_url": "https://clinic.example/grafana/explore",
        },
        "collection_errors_payload": {
            "errors": [
                "A4:topsql_tidb: HTTP 500: Delta Lake external table invalidQuery",
            ]
        },
        "analysis_patch_text": json.dumps(
            {
                "severity": "Low",
                "key_evidence": [
                    {
                        "type": "metric",
                        "name": "TiDB P99 latency stays low in window",
                        "metric_ref": {
                            "metric_group_id": "tidb_p99_latency",
                            "instances": ["all (30 instances)"],
                        },
                        "reason": "The latency aggregate row stays low across the diagnosis window.",
                    },
                    {
                        "type": "metric",
                        "name": "No CPU saturation / hotspot detected",
                        "metric_ref": {
                            "metric_group_id": "tikv_cpu:db",
                            "instances": ["all (db, 15 instances)"],
                        },
                        "reason": "The CPU aggregate row stays below saturation for the busiest TiKV group.",
                    },
                    {
                        "type": "log",
                        "name": "TiDB instance startup banner observed near alert time",
                        "instance": "db-tidb-60",
                        "reason": "Loki startup banner remains the strongest restart signal in the available artifacts.",
                    },
                ],
                "supporting_evidence": [
                    {
                        "type": "metric",
                        "name": "Component health restart metric returned empty",
                        "reason": "Restart detection via metrics did not report any component restarts in this fixed window.",
                    },
                    {
                        "type": "log",
                        "name": "No panic/fatal matches in TiDB/TiKV/PD for the fixed window",
                        "reason": "The available Loki snapshot does not contain panic or fatal matches in the fixed window.",
                    },
                    {
                        "type": "slow_query",
                        "digest": slow_digest,
                        "reason": "The slow query decoded plan gives a concrete optimization candidate but not the primary outage driver.",
                    },
                    {
                        "type": "file",
                        "name": "TopSQL collection failed (A4)",
                        "reason": "TopSQL CPU attribution is unavailable for this run because the collector returned HTTP 500.",
                    },
                    {
                        "type": "file",
                        "name": "Targeted Loki query snapshot miss",
                        "reason": "The narrower Loki query snapshot is missing, so restart reason cannot be confirmed from the remaining data.",
                    },
                ],
            }
        ),
    }


def test_restart_window_case_deterministic_judge_contract_invalid_subtype_contract(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10766841556523235152",
                "time_range": {"start": "1775030000", "end": "1775033600"},
            },
            "evidence": {"metrics": []},
        },
        analysis_patch_text=_minimal_analysis_patch(),
    )
    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-contract-invalid",
        audit_dir=audit_dir,
    )

    assert judgment["judgment_verdict"] == "invalid"
    assert judgment["primary_failure_code"] == "contract_failed"
    assert judgment["invalid_subtype"] == "contract"
    assert judgment["packet"]["contract"]["status"] == "fail"
    assert judgment["score"] == 0
    assert judgment["packet"]["scoring"]["raw_total_score"] == 0
    assert judgment["packet"]["scoring"]["final_score"] == 0
    assert judgment["packet"]["workflow"]["score"] == 0
    assert judgment["packet"]["evidence_collection"]["score"] == 0
    assert judgment["packet"]["critical_evidence"]["score"] == 0
    assert judgment["packet"]["evidence_routing"]["score"] == 0
    assert judgment["packet"]["diagnosis"]["score"] == 0


def test_high_cpu_case_uses_contract_bundle_and_latest_quality_bundle(tmp_path):
    case = load_case("service-degradation-during-incident-window")
    audit_dir = tmp_path / "audit" / "evalrun-1" / "evalcase-1"
    contract_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=[
            "tier-architecture.json",
            "incorrect-conclusions.json",
            "Initialize diagnosis run directory",
            "collect_evidence.py",
            "evidence_summary_for_agent.txt",
        ],
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10322500445237050222",
                "time_range": {"start": "1775551620", "end": "1775558520"},
            },
            "evidence": {"metrics": []},
        },
        queries_payload={
            "slow_queries": [
                {"digest": "fe60d105157bc3b63640b11cf7afd70fea2d37005981e81570c38f7d433a63c1", "source": "slow_query"}
            ],
            "topsql": [],
            "decoded_plan": {},
        },
        gate_checks_payload={
            "infra": "INFRA CLEAR",
            "hotspot": "HOTSPOT DETECTED: CPU saturation on db-tidb-0",
        },
        analysis_patch_text=None,
        include_summary_artifact=True,
    )
    run_dir = _evidence_run_dir(contract_dir)
    (_evidence_file(contract_dir, "workspace_evidence_report.json")).write_text(
        json.dumps(
            {
                "metadata": {
                    "cluster_id": "10322500445237050222",
                    "time_range": {"start": "1775551620", "end": "1775558520"},
                    "run_dir": str(run_dir),
                    "evidence_dir": str(run_dir),
                },
                "evidence": {
                    "metrics": [
                        {"metric_group_id": "tidb_cpu:db", "instance": "db-tidb-0", "peak": 7.42},
                        {"metric_group_id": "tidb_p99_latency", "instance": "db-tidb-0", "peak": 2.644},
                    ],
                },
            }
        )
    )
    (_evidence_file(contract_dir, "queries.json")).write_text(
        json.dumps(
            {
                "slow_queries": [
                    {"digest": "fe60d105157bc3b63640b11cf7afd70fea2d37005981e81570c38f7d433a63c1", "source": "slow_query"}
                ],
                "topsql": [
                    {"digest": "FE60D105157BC3B63640B11CF7AFD70FEA2D37005981E81570C38F7D433A63C1", "component": "TiDB"}
                ],
                "decoded_plan": {
                    "digest": "57429589f0a642037f139d2012c12b51df3b971d1a4177ee864bbe9d2bc1208a",
                    "summary": "delete hotspot with no index usage",
                },
            }
        )
    )
    (_evidence_file(contract_dir, "analysis_patch.json")).write_text(
        json.dumps(
            {
                "key_evidence": [
                    {
                        "type": "metric",
                        "name": "TiDB CPU saturation on db-tidb-0",
                        "metric_ref": {"metric_group_id": "tidb_cpu:db", "instances": ["db-tidb-0"]},
                        "reason": "Focused window hotspot.",
                    },
                    {
                        "type": "topsql",
                        "digest": "FE60D105157BC3B63640B11CF7AFD70FEA2D37005981E81570C38F7D433A63C1",
                        "reason": "Focused window primary query.",
                    },
                ],
            }
        )
    )

    output = """
Root Cause:
- A TiDB CPU hotspot on db-tidb-0 caused the latency spike.
- The dominant query path is FE60...63C1, and uninitialized statistics led to plan misestimation plus intermediate result amplification.
"""
    judgment = judge_case_output(
        case,
        output_text=output,
        workspace_id="ws-layered-bundle",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["contract"]["status"] == "pass"
    assert judgment["packet"]["diagnosis"]["status"] == "pass"
    assert judgment["facts"]["artifacts"]["run_dir"] == "evidences"
    assert judgment["packet"]["artifact_resolution"]["status"] == "resolved"


def test_judge_contract_yaml_policy_passes_on_matching_trace_facts(tmp_path):
    case = _case_with_judge_contract(
        load_case("component-restart-during-incident-window"),
        judge_contract={
            "checks": [
                {
                    "id": "time_window_match",
                    "description": "window matches",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {"source": "trace_fact", "matcher": "bool_true", "path": "time_window_match"}
                    ],
                },
                {
                    "id": "cluster_scope_match",
                    "description": "cluster matches",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {"source": "trace_fact", "matcher": "bool_true", "path": "cluster_scope_match"}
                    ],
                },
            ],
        },
    )
    audit_dir = _setup_audit_run(tmp_path, case.id, analysis_patch_text=_minimal_analysis_patch())

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-contract-yaml-pass", audit_dir=audit_dir)

    assert judgment["packet"]["contract"]["status"] == "pass"
    assert judgment["primary_failure_code"] is None


def test_judge_contract_prefers_nested_cluster_id_from_workspace_report(tmp_path):
    case = _case_with_judge_contract(
        load_case("component-restart-during-incident-window"),
        judge_contract={
            "checks": [
                {
                    "id": "cluster_scope_match",
                    "description": "cluster matches",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {"source": "trace_fact", "matcher": "bool_true", "path": "cluster_scope_match"}
                    ],
                },
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "2013516099403907072",
                "cluster": {"cluster_id": "10766841556523235152"},
                "time_range": {"start": "1775039760", "end": "1775043360"},
            },
            "evidence": {"metrics": []},
        },
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-contract-prefers-nested-cluster",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["contract"]["status"] == "pass"
    assert judgment["facts"]["cluster"]["actual_cluster_id"] == "10766841556523235152"
    assert judgment["facts"]["contract"]["expected_parent_id"] == ""


def test_judge_contract_allows_premium_parent_id_match(tmp_path):
    case = _case_with_judge_contract(
        load_case("tail-latency-surge-during-incident-window"),
        judge_contract={
            "checks": [
                {
                    "id": "cluster_scope_match",
                    "description": "cluster matches",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {"source": "trace_fact", "matcher": "bool_true", "path": "cluster_scope_match"}
                    ],
                },
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "2028998946604449792",
                "time_range": {"start": "1773029640", "end": "1773033240"},
            },
            "evidence": {"metrics": []},
        },
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-contract-premium-parent-match",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["contract"]["status"] == "pass"
    assert judgment["facts"]["contract"]["cluster_scope_match"] is True
    assert judgment["facts"]["contract"]["expected_cluster_id"] == "10805087899823024166"
    assert judgment["facts"]["contract"]["expected_parent_id"] == "2028998946604449792"
    assert judgment["facts"]["cluster"]["allowed_cluster_ids"] == [
        "10805087899823024166",
        "2028998946604449792",
    ]
    assert judgment["facts"]["cluster"]["actual_cluster_id"] == "2028998946604449792"


def test_judge_contract_still_rejects_unlisted_premium_cluster_id(tmp_path):
    case = _case_with_judge_contract(
        load_case("tail-latency-surge-during-incident-window"),
        judge_contract={
            "checks": [
                {
                    "id": "cluster_scope_match",
                    "description": "cluster matches",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {"source": "trace_fact", "matcher": "bool_true", "path": "cluster_scope_match"}
                    ],
                },
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "99999999999999999999",
                "time_range": {"start": "1773029640", "end": "1773033240"},
            },
            "evidence": {"metrics": []},
        },
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-contract-premium-parent-miss",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["contract"]["status"] == "fail"
    assert judgment["primary_failure_code"] == "contract_failed"
    assert judgment["facts"]["contract"]["cluster_scope_match"] is False


def test_judge_contract_yaml_policy_fails_on_required_miss(tmp_path):
    case = _case_with_judge_contract(
        load_case("component-restart-during-incident-window"),
        judge_contract={
            "checks": [
                {
                    "id": "time_window_match",
                    "description": "window matches",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {"source": "trace_fact", "matcher": "bool_true", "path": "time_window_match"}
                    ],
                }
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10766841556523235152",
                "time_range": {"start": "1775030000", "end": "1775033600"},
            },
            "evidence": {"metrics": []},
        },
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-contract-yaml-fail", audit_dir=audit_dir)

    assert judgment["packet"]["contract"]["status"] == "fail"
    assert judgment["primary_failure_code"] == "contract_failed"


def test_judge_contract_prefers_runtime_resolved_window_over_static_allowed_list(tmp_path):
    case = _case_with_judge_contract(
        load_case("component-restart-during-incident-window"),
        judge_contract={
            "checks": [
                {
                    "id": "time_window_match",
                    "description": "window matches",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {"source": "trace_fact", "matcher": "bool_true", "path": "time_window_match"}
                    ],
                }
            ],
        },
    )
    data = deepcopy(case.data)
    data["time_window"]["allowed"] = [
        {
            "id": "wrong-window",
            "start": "2026-04-01T00:00:00Z",
            "end": "2026-04-01T01:00:00Z",
        }
    ]
    data["time_window"]["resolved_by_output"] = "resolved_window"
    data["runtime"] = {
        "hooks": [
            {
                "id": "resolve-window",
                "stage": "before_judge",
                "kind": "python_script",
                "path": "runtime_hook.py",
                "output_id": "resolved_window",
            }
        ]
    }
    case = replace(case, data=data)
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    (audit_dir / "runtime_context.json").write_text(
        json.dumps(
            {
                "outputs": {
                    "resolved_window": {
                        "data": {
                            "id": "runtime-window",
                            "start": "2026-04-01T10:36:00Z",
                            "end": "2026-04-01T11:36:00Z",
                        }
                    }
                }
            }
        ),
        encoding="utf-8",
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-contract-runtime-window",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["contract"]["status"] == "pass"
    assert judgment["facts"]["runtime_context_present"] is True
    assert judgment["facts"]["window"]["expected_allowed_window_id"] == "runtime-window"
    assert judgment["facts"]["window"]["matched_allowed_window"] is True
    assert judgment["facts"]["contract"]["expected_allowed_window_id"] == "runtime-window"


def test_extract_output_json_reads_json_code_block_after_prose():
    output_text = """
结论如下。

```json
{"cluster_id":"c-1","top_nodes_by_avg_read_iops":[{"instance":"n1","avg_read_iops":12.5}]}
```
"""

    parsed = _extract_output_json(output_text)

    assert parsed == {
        "cluster_id": "c-1",
        "top_nodes_by_avg_read_iops": [{"instance": "n1", "avg_read_iops": 12.5}],
    }


def test_extract_output_json_for_adhoc_mode_reads_inline_json_after_prose():
    output_text = """
先说明一下结论。
{"cluster_id":"c-1","evidence_sufficient":true}
"""

    assert _extract_output_json(output_text) is None
    assert _extract_output_json_for_evaluation_mode(
        output_text,
        evaluation_mode="adhoc_evidence",
    ) == {
        "cluster_id": "c-1",
        "evidence_sufficient": True,
    }
    assert _extract_output_json_for_evaluation_mode(
        output_text,
        evaluation_mode="diagnosis",
    ) is None


def test_diagnosis_output_json_can_match_runtime_reference_payload(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_diagnosis_checks(
        case,
        checks=[
            {
                "id": "match_runtime_reference_json",
                "mode": "deterministic_only",
                "weight": 3,
                "required": True,
                "description": "Matches the structured answer against runtime reference JSON.",
                "deterministic": {
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_json",
                            "matcher": "json_path_equals_reference",
                            "path": "cluster_id",
                            "reference_output_id": "reference",
                            "reference_path": "cluster_id",
                        },
                        {
                            "source": "output_json",
                            "matcher": "json_array_matches_reference",
                            "path": "top_nodes_by_avg_read_iops",
                            "reference_output_id": "reference",
                            "reference_path": "top_nodes_by_avg_read_iops",
                            "key_fields": ["instance"],
                            "numeric_fields": ["avg_read_iops"],
                            "threshold": 0.01,
                        },
                    ],
                },
            }
        ],
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    (audit_dir / "runtime_context.json").write_text(
        json.dumps(
            {
                "outputs": {
                    "reference": {
                        "data": {
                            "cluster_id": "10766841556523235152",
                            "top_nodes_by_avg_read_iops": [
                                {"instance": "tikv-1", "avg_read_iops": 18234.5},
                                {"instance": "tikv-2", "avg_read_iops": 17110.2},
                            ],
                        }
                    }
                }
            }
        ),
        encoding="utf-8",
    )
    output_text = """
映射后可直接比较节点级 read IOPS，结果如下。

```json
{
  "cluster_id": "10766841556523235152",
  "top_nodes_by_avg_read_iops": [
    {"instance": "tikv-1", "avg_read_iops": 18234.5},
    {"instance": "tikv-2", "avg_read_iops": 17110.2}
  ]
}
```
"""

    judgment = judge_case_output(
        case,
        output_text=output_text,
        workspace_id="ws-diagnosis-runtime-reference",
        audit_dir=audit_dir,
    )

    diagnosis = judgment["packet"]["diagnosis"]
    check = diagnosis["checks"][0]

    assert judgment["judgment_verdict"] == "pass"
    assert check["passed"] is True
    assert all(rule["matched"] is True for rule in check["deterministic"]["rules"])
    assert diagnosis["score"] == DIAGNOSIS_BUCKET_MAX


def test_diagnosis_output_json_nested_reference_can_match_with_tolerance(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_diagnosis_checks(
        case,
        checks=[
            {
                "id": "match_nested_runtime_reference_json",
                "mode": "deterministic_only",
                "weight": 3,
                "required": True,
                "description": "Matches nested structured output against runtime reference JSON with numeric tolerance.",
                "deterministic": {
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_json",
                            "matcher": "json_path_equals_reference",
                            "path": "nodes_by_cluster",
                            "reference_output_id": "reference",
                            "reference_path": "nodes_by_cluster",
                            "threshold": 0.01,
                        }
                    ],
                },
            }
        ],
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    (audit_dir / "runtime_context.json").write_text(
        json.dumps(
            {
                "outputs": {
                    "reference": {
                        "data": {
                            "nodes_by_cluster": [
                                {
                                    "cluster_id": "c-1",
                                    "nodes": [
                                        ["db-tikv-0", 123.456],
                                        ["db-tikv-1", 100.001],
                                    ],
                                },
                                {
                                    "cluster_id": "c-2",
                                    "nodes": [
                                        ["db-tikv-2", 99.994],
                                    ],
                                },
                            ],
                        }
                    }
                }
            }
        ),
        encoding="utf-8",
    )
    output_text = """
```json
{
  "nodes_by_cluster": [
    {
      "cluster_id": "c-1",
      "nodes": [
        ["db-tikv-0", 123.46],
        ["db-tikv-1", 100.0]
      ]
    },
    {
      "cluster_id": "c-2",
      "nodes": [
        ["db-tikv-2", 99.99]
      ]
    }
  ]
}
```
"""

    judgment = judge_case_output(
        case,
        output_text=output_text,
        workspace_id="ws-diagnosis-nested-runtime-reference",
        audit_dir=audit_dir,
    )

    diagnosis = judgment["packet"]["diagnosis"]
    check = diagnosis["checks"][0]

    assert judgment["judgment_verdict"] == "pass"
    assert check["passed"] is True
    assert check["deterministic"]["rules"][0]["matched"] is True


def test_diagnosis_output_json_missing_runtime_reference_is_invalid_precondition(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_diagnosis_checks(
        case,
        checks=[
            {
                "id": "match_runtime_reference_json",
                "mode": "deterministic_only",
                "weight": 3,
                "required": True,
                "description": "Matches the structured answer against runtime reference JSON.",
                "deterministic": {
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_json",
                            "matcher": "json_path_equals_reference",
                            "path": "cluster_id",
                            "reference_output_id": "reference",
                            "reference_path": "cluster_id",
                        }
                    ],
                },
            }
        ],
    )
    data = deepcopy(case.data)
    data["runtime"] = {
        "hooks": [
            {
                "id": "build-reference",
                "stage": "before_judge",
                "kind": "python_script",
                "path": "runtime_hook.py",
                "output_id": "reference",
            }
        ]
    }
    case = replace(case, data=data)
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    output_text = """
```json
{"cluster_id":"10766841556523235152"}
```
"""

    judgment = judge_case_output(
        case,
        output_text=output_text,
        workspace_id="ws-diagnosis-missing-runtime-reference",
        audit_dir=audit_dir,
    )

    assert judgment["judgment_verdict"] == "invalid"
    assert judgment["invalid_subtype"] == "precondition"
    assert judgment["primary_failure_code"] == "runtime_reference_missing"
    assert judgment["facts"]["runtime_context"]["missing_output_ids"] == ["reference"]


def test_diagnosis_output_json_array_match_rejects_extra_ranked_rows(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_diagnosis_checks(
        case,
        checks=[
            {
                "id": "match_runtime_reference_json",
                "mode": "deterministic_only",
                "weight": 3,
                "required": True,
                "description": "Matches the structured answer against runtime reference JSON.",
                "deterministic": {
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_json",
                            "matcher": "json_array_matches_reference",
                            "path": "top_nodes_by_avg_read_iops",
                            "reference_output_id": "reference",
                            "reference_path": "top_nodes_by_avg_read_iops",
                            "key_fields": ["instance"],
                            "numeric_fields": ["avg_read_iops"],
                            "threshold": 0.01,
                        }
                    ],
                },
            }
        ],
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    (audit_dir / "runtime_context.json").write_text(
        json.dumps(
            {
                "outputs": {
                    "reference": {
                        "data": {
                            "top_nodes_by_avg_read_iops": [
                                {"instance": "tikv-1", "avg_read_iops": 18234.5},
                                {"instance": "tikv-2", "avg_read_iops": 17110.2},
                            ],
                        }
                    }
                }
            }
        ),
        encoding="utf-8",
    )
    output_text = """
```json
{
  "top_nodes_by_avg_read_iops": [
    {"instance": "tikv-1", "avg_read_iops": 18234.5},
    {"instance": "tikv-2", "avg_read_iops": 17110.2},
    {"instance": "tikv-3", "avg_read_iops": 16500.0}
  ]
}
```
"""

    judgment = judge_case_output(
        case,
        output_text=output_text,
        workspace_id="ws-diagnosis-extra-ranked-rows",
        audit_dir=audit_dir,
    )

    diagnosis = judgment["packet"]["diagnosis"]
    assert judgment["judgment_verdict"] == "fail"
    assert diagnosis["checks"][0]["passed"] is False


def test_critical_evidence_output_json_can_match_runtime_reference_payload(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_critical_evidence(
        case,
        critical_evidence={
            "weight": 80,
            "checks": [
                {
                    "id": "match_runtime_reference_json",
                    "weight": 3,
                    "required": True,
                    "description": "Matches the structured answer against runtime reference JSON.",
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_json",
                            "matcher": "json_path_equals_reference",
                            "path": "cluster_id",
                            "reference_output_id": "reference",
                            "reference_path": "cluster_id",
                        },
                        {
                            "source": "output_json",
                            "matcher": "json_array_matches_reference",
                            "path": "top_nodes_by_avg_read_iops",
                            "reference_output_id": "reference",
                            "reference_path": "top_nodes_by_avg_read_iops",
                            "key_fields": ["instance"],
                            "numeric_fields": ["avg_read_iops"],
                            "threshold": 0.01,
                        },
                    ],
                }
            ],
        },
    )
    case = _case_with_diagnosis_checks(case, checks=[])
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    (audit_dir / "runtime_context.json").write_text(
        json.dumps(
            {
                "outputs": {
                    "reference": {
                        "data": {
                            "cluster_id": "10766841556523235152",
                            "top_nodes_by_avg_read_iops": [
                                {"instance": "tikv-1", "avg_read_iops": 18234.5},
                                {"instance": "tikv-2", "avg_read_iops": 17110.2},
                            ],
                        }
                    }
                }
            }
        ),
        encoding="utf-8",
    )
    output_text = """
```json
{
  "cluster_id": "10766841556523235152",
  "top_nodes_by_avg_read_iops": [
    {"instance": "tikv-1", "avg_read_iops": 18234.5},
    {"instance": "tikv-2", "avg_read_iops": 17110.2}
  ]
}
```
"""

    judgment = judge_case_output(
        case,
        output_text=output_text,
        workspace_id="ws-critical-runtime-reference",
        audit_dir=audit_dir,
    )

    critical = judgment["packet"]["critical_evidence"]

    assert judgment["judgment_verdict"] == "pass"
    assert critical["status"] == "pass"
    assert critical["max_score"] == 80
    assert critical["score"] == 80
    assert all(rule["matched"] is True for rule in critical["checks"][0]["rules"])


def test_diagnosis_output_json_can_match_topsql_url_semantics(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_diagnosis_checks(
        case,
        checks=[
            {
                "id": "match_topsql_url",
                "mode": "deterministic_only",
                "weight": 3,
                "required": True,
                "description": "Matches a TopSQL URL against runtime reference semantics.",
                "deterministic": {
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_json",
                            "matcher": "topsql_url_semantically_matches_reference",
                            "path": "link",
                            "reference_output_id": "reference",
                            "reference_path": "link",
                        }
                    ],
                },
            }
        ],
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    (audit_dir / "runtime_context.json").write_text(
        json.dumps(
            {
                "outputs": {
                    "reference": {
                        "data": {
                            "link": {
                                "url": "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html"
                                "?clusterId=10518506906906971664&provider=aws&region=us-east-1"
                                "#/topsql?from=1776558000&to=1776558300"
                                "&instance=db-tikv-23.db-tikv-peer.tidb10518506906906971664.svc%3A20160"
                            }
                        }
                    }
                }
            }
        ),
        encoding="utf-8",
    )
    output_text = """
```json
{
  "link": {
    "url": "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html?provider=aws&clusterId=10518506906906971664&region=us-east-1#/topsql?instance=db-tikv-23.db-tikv-peer.tidb10518506906906971664.svc:20160&to=1776558300&from=1776558000"
  }
}
```
"""

    judgment = judge_case_output(
        case,
        output_text=output_text,
        workspace_id="ws-diagnosis-topsql-url",
        audit_dir=audit_dir,
    )

    diagnosis = judgment["packet"]["diagnosis"]
    assert judgment["judgment_verdict"] == "pass"
    assert diagnosis["checks"][0]["passed"] is True
    assert diagnosis["checks"][0]["deterministic"]["rules"][0]["matched"] is True


def test_diagnosis_output_json_rejects_topsql_url_with_wrong_instance(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_diagnosis_checks(
        case,
        checks=[
            {
                "id": "match_topsql_url",
                "mode": "deterministic_only",
                "weight": 3,
                "required": True,
                "description": "Matches a TopSQL URL against runtime reference semantics.",
                "deterministic": {
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_json",
                            "matcher": "topsql_url_semantically_matches_reference",
                            "path": "link",
                            "reference_output_id": "reference",
                            "reference_path": "link",
                        }
                    ],
                },
            }
        ],
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    (audit_dir / "runtime_context.json").write_text(
        json.dumps(
            {
                "outputs": {
                    "reference": {
                        "data": {
                            "link": {
                                "url": "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html"
                                "?clusterId=10518506906906971664#/topsql?from=1776558000&to=1776558300"
                                "&instance=db-tikv-23.db-tikv-peer.tidb10518506906906971664.svc%3A20160"
                            }
                        }
                    }
                }
            }
        ),
        encoding="utf-8",
    )
    output_text = """
```json
{
  "link": {
    "url": "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html?clusterId=10518506906906971664#/topsql?from=1776558000&to=1776558300&instance=db-tikv-21.db-tikv-peer.tidb10518506906906971664.svc%3A20160"
  }
}
```
"""

    judgment = judge_case_output(
        case,
        output_text=output_text,
        workspace_id="ws-diagnosis-topsql-url-wrong-instance",
        audit_dir=audit_dir,
    )

    diagnosis = judgment["packet"]["diagnosis"]
    assert judgment["judgment_verdict"] == "fail"
    assert diagnosis["checks"][0]["passed"] is False


def test_critical_evidence_output_json_can_match_topsql_url_array_semantics(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_critical_evidence(
        case,
        critical_evidence={
            "weight": 80,
            "checks": [
                {
                    "id": "match_topsql_links",
                    "weight": 3,
                    "required": True,
                    "description": "Matches TopSQL link semantics against runtime reference JSON.",
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_json",
                            "matcher": "json_array_matches_reference",
                            "path": "topsql_links",
                            "reference_output_id": "reference",
                            "reference_path": "topsql_links",
                            "key_fields": ["tikv_node"],
                        },
                        {
                            "source": "output_json",
                            "matcher": "topsql_url_semantically_matches_reference",
                            "path": "topsql_links",
                            "reference_output_id": "reference",
                            "reference_path": "topsql_links",
                        },
                    ],
                }
            ],
        },
    )
    case = _case_with_diagnosis_checks(case, checks=[])
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    (audit_dir / "runtime_context.json").write_text(
        json.dumps(
            {
                "outputs": {
                    "reference": {
                        "data": {
                            "topsql_links": [
                                {
                                    "tikv_node": "db-tikv-23",
                                    "url": "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html"
                                    "?clusterId=10518506906906971664#/topsql?from=1776558000&to=1776558300"
                                    "&instance=db-tikv-23.db-tikv-peer.tidb10518506906906971664.svc%3A20160",
                                },
                                {
                                    "tikv_node": "db-tikv-21",
                                    "url": "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html"
                                    "?clusterId=10518506906906971664#/topsql?from=1776558000&to=1776558300"
                                    "&instance=db-tikv-21.db-tikv-peer.tidb10518506906906971664.svc%3A20160",
                                },
                            ]
                        }
                    }
                }
            }
        ),
        encoding="utf-8",
    )
    output_text = """
```json
{
  "topsql_links": [
    {
      "tikv_node": "db-tikv-23",
      "url": "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html?clusterId=10518506906906971664#/topsql?instance=db-tikv-23.db-tikv-peer.tidb10518506906906971664.svc:20160&to=1776558300&from=1776558000"
    },
    {
      "tikv_node": "db-tikv-21",
      "url": "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html?clusterId=10518506906906971664#/topsql?to=1776558300&from=1776558000&instance=db-tikv-21.db-tikv-peer.tidb10518506906906971664.svc:20160"
    }
  ]
}
```
"""

    judgment = judge_case_output(
        case,
        output_text=output_text,
        workspace_id="ws-critical-topsql-link-array",
        audit_dir=audit_dir,
    )

    critical = judgment["packet"]["critical_evidence"]
    assert judgment["judgment_verdict"] == "pass"
    assert critical["status"] == "pass"
    assert all(rule["matched"] is True for rule in critical["checks"][0]["rules"])


def test_restart_window_case_deterministic_judge_fails_query_misdiagnosis(tmp_path):
    case = load_case("component-restart-during-incident-window")
    output = """
Root Cause:
- A high-frequency point-filter query on fan_prd.account_ranking is executing
  as a TiKV full table scan due to no usable index on joined_account_id.

Recommendations:
- CREATE INDEX idx_account_ranking_joined_account_id ...
- ANALYZE TABLE fan_prd.account_ranking ...
"""
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    judgment = judge_case_output(
        case,
        output_text=output,
        workspace_id="ws-query-misdiagnosis",
        audit_dir=audit_dir,
    )
    assert judgment["packet"]["diagnosis"]["status"] == "fail"
    assert judgment["packet"]["diagnosis"]["score"] == 0


def test_restart_window_case_deterministic_judge_accepts_panic_alias_phrasing(tmp_path):
    case = load_case("component-restart-during-incident-window")
    output = """
Root Cause:
- Query interruption triggered a single-instance TiDB panic/restart event in the known GitHub #63956 bug chain.
- The chain starts from KILL QUERY and is not a missing-index issue.
Recommendation:
- Upgrade from v8.5.2 to a release that contains the fix, such as v8.5.3.
"""
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    judgment = judge_case_output(
        case,
        output_text=output,
        workspace_id="ws-panic-alias",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["diagnosis"]["status"] == "pass"
    diagnosis_checks = {item["id"]: item for item in judgment["packet"]["diagnosis"]["checks"]}
    assert diagnosis_checks["identify_kill_query_cleanup_chain"]["passed"] is True
    assert diagnosis_checks["reject_missing_index_root_cause"]["passed"] is True
    assert diagnosis_checks["identify_known_bug_restart_chain"]["passed"] is True
    assert diagnosis_checks["mention_github_issue_id"]["passed"] is True
    assert diagnosis_checks["recommend_fixed_version_upgrade"]["passed"] is True


def test_high_cpu_case_deterministic_judge_accepts_cpu_hotspot_and_stats_issue(tmp_path):
    case = load_case("service-degradation-during-incident-window")
    output = """
Root Cause:
- A TiDB CPU hotspot on db-tidb-0 caused the latency spike, with P99 latency peaking during the incident.
- The dominant hot JOIN path was the specific SQL path behind the hotspot, and its plan shifted from TiFlash to TiKV.
- The dominant problem is a bad join plan under uninitialized statistics, where the optimizer underestimated join row counts and caused intermediate result explosion.

Supporting Evidence:
- TopSQL and slow query both point to the same hot JOIN path.
- The observed TiDB restarts are secondary effects rather than the primary root cause.
"""
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=[
            "tier-architecture.json",
            "incorrect-conclusions.json",
            "Initialize diagnosis run directory",
            "collect_evidence.py",
            "evidence_summary_for_agent.txt",
        ],
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10322500445237050222",
                "time_range": {"start": "1775551620", "end": "1775558520"},
            },
            "evidence": {
                "metrics": [
                    {"metric_group_id": "tidb_cpu:db", "instance": "db-tidb-0", "peak": 7.76},
                    {"metric_group_id": "tidb_p99_latency", "instance": "db-tidb-0", "peak": 2.644},
                ],
            },
        },
        queries_payload={
            "slow_queries": [
                {"digest": "fe60d105157bc3b63640b11cf7afd70fea2d37005981e81570c38f7d433a63c1", "source": "slow_query"}
            ],
            "topsql": [
                {"digest": "FE60D105157BC3B63640B11CF7AFD70FEA2D37005981E81570C38F7D433A63C1", "component": "TiDB"}
            ],
            "decoded_plan": {
                "digest": "fe60d105157bc3b63640b11cf7afd70fea2d37005981e81570c38f7d433a63c1",
                "summary": "stocks.id and enabled statistics are unInitialized; 177577 rows after join explosion",
            },
        },
        gate_checks_payload={
            "infra": "INFRA CLEAR",
            "hotspot": "HOTSPOT DETECTED: CPU saturation on db-tidb-0",
        },
        component_health_payload={
            "component_health": {
                "query_scope": "10322500445237050222",
                "component_count": [],
                "restarts": [
                    {"component": "tidb", "instance": "db-tidb-0", "restart_count": 1.0},
                    {"component": "tidb", "instance": "db-tidb-2", "restart_count": 1.0},
                ],
                "worker_memory": [],
                "links": {},
            }
        },
        include_logs_dir=False,
        analysis_patch_text=json.dumps(
                {
                    "key_evidence": [
                        {
                            "type": "metric",
                            "name": "TiDB CPU saturation on db-tidb-0",
                            "metric_ref": {"metric_group_id": "tidb_cpu:db", "instances": ["db-tidb-0"]},
                            "reason": "db-tidb-0 is the hotspot instance.",
                        },
                        {
                            "type": "slow_query",
                            "digest": "fe60d105157bc3b63640b11cf7afd70fea2d37005981e81570c38f7d433a63c1",
                            "reason": "The dominant slow query matches the hotspot path.",
                        },
                    ],
                    "supporting_evidence": [
                        {
                            "type": "metric",
                            "name": "P99 latency spike",
                            "metric_ref": {"metric_group_id": "tidb_p99_latency", "instances": ["db-tidb-0"]},
                            "reason": "Latency spike aligns with the hotspot interval.",
                        }
                    ],
                }
        ),
        include_summary_artifact=True,
    )

    judgment = judge_case_output(
        case,
        output_text=output,
        workspace_id="ws-high-cpu-case",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["diagnosis"]["status"] == "pass"
    assert judgment["packet"]["overall_verdict"] == "pass"
    diagnosis_checks = {item["id"]: item for item in judgment["packet"]["diagnosis"]["checks"]}
    assert diagnosis_checks["identify_specific_sql_hotspot_path"]["passed"] is True
    assert diagnosis_checks["identify_specific_sql_hotspot_path"]["deterministic_pass"] is True
    assert diagnosis_checks["identify_root_cause_mechanism"]["passed"] is True
    assert diagnosis_checks["identify_root_cause_mechanism"]["mode"] == "blended"
    assert diagnosis_checks["identify_sql_plan_shift"]["passed"] is True
    assert diagnosis_checks["identify_tiflash_to_tikv_execution_path_change"]["passed"] is True


def test_recoverable_panic_case_accepts_timeout_interrupt_classification(tmp_path):
    case = load_case("panic-in-recoverable-goroutine")
    output = """
Root Cause:
- The recoverable goroutine panic is real, not a fake alarm.
- Loki shows [executor:3024] Query execution was interrupted, maximum statement execution time exceeded.
- This is a query timeout interruption from max_execution_time, expected behavior and not a TiDB bug.

Impact:
- The impact is limited to the query that exceeded the timeout; TiDB recovered the goroutine.

Recommendation:
- No TiDB fix needed for the panic itself. Optionally inspect the timed-out query and tune it or the timeout.
"""
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=[
            "issues/panic-in-recoverable-goroutine/issue_meta.json",
            "issues/panic-in-recoverable-goroutine/cluster_meta.json",
            "tier-architecture.json",
            "incorrect-conclusions.json",
            "Initialize diagnosis run directory",
            "collect_evidence.py",
            "loki-targeted-panic.log",
            "evidence_summary_for_agent.txt",
        ],
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10518506906906971664",
                "time_range": {"start": "1776631800", "end": "1776639000"},
            },
            "evidence": {"metrics": []},
        },
        logs_files={
            "loki-targeted-panic.log": "\n".join(
                [
                    '[2026/04/19 21:50:00.323 +00:00] [ERROR] [misc.go:95] ["panic in the recoverable goroutine"]',
                    'r="[executor:3024]Query execution was interrupted, maximum statement execution time exceeded"',
                ]
            )
        },
        queries_payload={
            "slow_queries": [
                {
                    "digest": "timeout-query",
                    "source": "slow_query",
                    "message": "Query execution was interrupted by max_execution_time",
                }
            ],
            "topsql": [],
            "decoded_plan": {},
        },
        gate_checks_payload={
            "infra": "INFRA CLEAR",
            "hotspot": "HOTSPOT: not detected",
        },
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [
                    {
                        "type": "log",
                        "name": "recoverable panic",
                        "reason": "Direct panic log contains executor:3024 timeout interruption.",
                    }
                ],
                "supporting_evidence": [
                    {
                        "type": "slow_query",
                        "digest": "timeout-query",
                        "reason": "Query context for the timeout interruption.",
                    }
                ],
            }
        ),
        include_summary_artifact=True,
    )

    judgment = judge_case_output(
        case,
        output_text=output,
        workspace_id="ws-panic-recoverable-goroutine",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["overall_verdict"] == "pass"
    hard_checks = {item["id"]: item for item in judgment["packet"]["diagnosis"]["checks"]}
    supporting_checks = hard_checks
    assert hard_checks["identify_real_panic_event"]["passed"] is True
    assert hard_checks["identify_real_panic_event"]["required"] is False
    assert hard_checks["identify_query_timeout_interrupt"]["passed"] is True
    assert hard_checks["identify_query_timeout_interrupt"]["required"] is True
    assert hard_checks["identify_query_timeout_interrupt"]["mode"] == "deterministic_only"
    assert hard_checks["reject_fake_alarm_or_unrelated_root_cause"]["passed"] is True
    assert hard_checks["reject_fake_alarm_or_unrelated_root_cause"]["required"] is False
    assert supporting_checks["identify_expected_behavior"]["passed"] is True
    assert supporting_checks["identify_expected_behavior"]["required"] is False
    assert supporting_checks["recommend_no_tidb_fix_needed"]["passed"] is True


def test_tail_latency_case_accepts_lock_contention_hotspot_diagnosis(tmp_path):
    case = load_case("tail-latency-surge-during-incident-window")
    output = """
Root Cause:
- A repeated point-update write path caused pessimistic lock contention during the incident window.
- The pressure concentrated on a single TiKV instance and formed one TiKV hotspot rather than a cluster-wide infrastructure failure.

Supporting Evidence:
- Lock manager conflicts and pessimistic lock wait signals rose sharply, while the workload became commit-heavy with elevated aborts.
- The dominant path is the repeated point update workload; the large JOIN UPDATE is secondary and adds extra pressure but is not the primary driver.

Recommendation:
- Reduce update frequency with batch update or batch aggregation, and consider async aggregation / periodic flush for this write path.
"""
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=[
            "issues/tail-latency-surge-during-incident-window/issue_meta.json",
            "issues/tail-latency-surge-during-incident-window/cluster_meta.json",
            "tier-architecture.json",
            "incorrect-conclusions.json",
            "Initialize diagnosis run directory",
            "collect_evidence.py",
            "evidence_summary_for_agent.txt",
        ],
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10805087899823024166",
                "time_range": {"start": "1773029640", "end": "1773033240"},
            },
            "evidence": {
                "metrics": [
                    {
                        "metric_group_id": "tikv_lock_manager_waiter_lifetime_duration",
                        "instance": "db-a924bb0f-tikv-29grsn",
                        "peak": 124313,
                    },
                    {
                        "metric_group_id": "tidb_server_handle_query_duration_seconds_sum",
                        "instance": "db-tidb-0",
                        "peak": 216.4,
                    },
                ]
            },
        },
        queries_payload={
            "slow_queries": [
                {
                    "digest": "d8f5b58-lock-heavy-update",
                    "query": "UPDATE t_merchant_online SET amount_by_day = amount_by_day + ? WHERE merchant_id = ?",
                    "source": "slow_query",
                },
                {
                    "digest": "c823661-secondary-join-update",
                    "query": "UPDATE t_merchant_online JOIN (...)",
                    "source": "slow_query",
                },
            ],
            "topsql": [],
            "decoded_plan": {},
        },
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [
                    {
                        "type": "metric",
                        "name": "Lock contention on hotspot TiKV",
                        "metric_ref": {
                            "metric_group_id": "tikv_lock_manager_waiter_lifetime_duration",
                            "instances": ["db-a924bb0f-tikv-29grsn"],
                        },
                        "reason": "The hotspot TiKV instance carries the lock pressure.",
                    },
                    {
                        "type": "slow_query",
                        "digest": "d8f5b58-lock-heavy-update",
                        "reason": "The repeated t_merchant_online update is the dominant write path.",
                    },
                ],
                "supporting_evidence": [
                    {
                        "type": "metric",
                        "name": "Commit and abort pressure",
                        "metric_ref": {
                            "metric_group_id": "tidb_server_handle_query_duration_seconds_sum",
                            "instances": ["db-tidb-0"],
                        },
                        "reason": "Commit-heavy DB time and abort pressure align with lock contention.",
                    }
                ],
            }
        ),
        include_summary_artifact=True,
    )
    _evidence_file(audit_dir, "evidence_summary_for_agent.json").write_text(
        json.dumps(
            {
                "summary": [
                    "lock contention on a single TiKV instance",
                    "high-frequency point update workload plus a secondary join update",
                    "commit-heavy execution with elevated aborts",
                ]
            }
        )
    )
    _evidence_file(audit_dir, "metrics.json").write_text(
        json.dumps(
            {
                "signals": [
                    {
                        "metric": "tikv_lock_manager_waiter_lifetime_duration",
                        "instance": "db-a924bb0f-tikv-29grsn",
                        "peak": 124313,
                    },
                    {
                        "metric": "tikv_grpc_msg_duration_seconds",
                        "instance": "db-a924bb0f-tikv-29grsn",
                        "p99_ms": 98,
                    },
                    {
                        "metric": "acquire_pessimistic_lock",
                        "instance": "db-a924bb0f-tikv-29grsn",
                        "p99_ms": 45,
                    },
                ]
            }
        )
    )

    judgment = judge_case_output(
        case,
        output_text=output,
        workspace_id="ws-tail-latency-case",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["diagnosis"]["status"] == "pass"
    assert judgment["packet"]["overall_verdict"] == "pass"
    diagnosis_checks = {item["id"]: item for item in judgment["packet"]["diagnosis"]["checks"]}
    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}
    assert diagnosis_checks["identify_lock_contention_as_primary_path"]["passed"] is True
    assert diagnosis_checks["identify_single_tikv_hotspot"]["passed"] is True
    assert diagnosis_checks["recommend_write_path_mitigation"]["passed"] is True
    assert critical_checks["lock_contention_signal_present"]["passed"] is True
    assert critical_checks["hotspot_instance_evidence_present"]["passed"] is True
    assert critical_checks["primary_write_path_evidence_present"]["passed"] is True


def test_tail_latency_case_accepts_write_contention_aliases(tmp_path):
    case = load_case("tail-latency-surge-during-incident-window")
    output = """
Root Cause:
- A repeated point-update write path caused concurrent write contention during the incident window.
- The resulting lock wait and hotspot key serialization concentrated pressure on one TiKV instance rather than a cluster-wide resource bottleneck.

Supporting Evidence:
- The workload became commit-heavy with elevated aborts, which is consistent with waiting on the same hot write path.

Recommendation:
- Reduce update frequency and batch the write path.
"""
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10805087899823024166",
                "time_range": {"start": "1773029640", "end": "1773033240"},
            },
            "evidence": {
                "metrics": [
                    {
                        "metric_group_id": "tikv_lock_manager_waiter_lifetime_duration",
                        "instance": "db-a924bb0f-tikv-29grsn",
                        "peak": 124313,
                    }
                ]
            },
        },
        queries_payload={
            "slow_queries": [
                {
                    "digest": "d8f5b58-lock-heavy-update",
                    "query": "UPDATE t_merchant_online SET amount_by_day = amount_by_day + ? WHERE merchant_id = ?",
                    "source": "slow_query",
                }
            ],
            "topsql": [],
            "decoded_plan": {},
        },
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [
                    {
                        "type": "metric",
                        "name": "Lock contention on hotspot TiKV",
                        "metric_ref": {
                            "metric_group_id": "tikv_lock_manager_waiter_lifetime_duration",
                            "instances": ["db-a924bb0f-tikv-29grsn"],
                        },
                        "reason": "The hotspot TiKV instance carries the lock pressure.",
                    }
                ]
            }
        ),
    )
    _evidence_file(audit_dir, "evidence_summary_for_agent.json").write_text(
        json.dumps(
            {
                "summary": [
                    "lock contention on a single TiKV instance",
                    "high-frequency point update workload",
                    "commit-heavy execution with elevated aborts",
                ]
            }
        )
    )
    _evidence_file(audit_dir, "metrics.json").write_text(
        json.dumps(
            {
                "signals": [
                    {
                        "metric": "tikv_lock_manager_waiter_lifetime_duration",
                        "instance": "db-a924bb0f-tikv-29grsn",
                        "peak": 124313,
                    }
                ]
            }
        )
    )

    judgment = judge_case_output(
        case,
        output_text=output,
        workspace_id="ws-tail-latency-aliases",
        audit_dir=audit_dir,
    )

    diagnosis_checks = {item["id"]: item for item in judgment["packet"]["diagnosis"]["checks"]}
    assert diagnosis_checks["identify_lock_contention_as_primary_path"]["passed"] is True


def test_tikv_node_stall_case_accepts_compaction_read_pool_stall_diagnosis(tmp_path):
    case = load_case("tikv-node-stall-during-incident-window")
    output = """
Root Cause:
- A RocksDB default CF compaction burst on db-tikv-19, triggered by SST ingestion after a Raft leader transfer, saturated the unified read-pool.
- The pressure concentrated on a single TiKV instance and caused a short node-level freeze rather than a cluster-wide TiDB CPU bottleneck.

Supporting Evidence:
- read_pool.rs reported estimated_wait_ms spikes and request rejection while grpc and coprocessor wait latency exploded.
- LevelMaxLevelSize compaction, RocksDB CPU, pending peers, and scheduler worker delay all align with the same 03:43 UTC window.

Recommendation:
- Rebalance Raft leaders away from db-tikv-19, reduce batch DELETE tombstone pressure, review RocksDB compaction sizing, and keep TopSQL/profiling follow-up.
"""
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=[
            "issues/tikv-node-stall-during-incident-window/issue_meta.json",
            "issues/tikv-node-stall-during-incident-window/cluster_meta.json",
            "tier-architecture.json",
            "incorrect-conclusions.json",
            "Initialize diagnosis run directory",
            "collect_evidence.py",
            "evidence_summary_for_agent.txt",
        ],
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10143988032832670331",
                "time_range": {"start": "1780630875", "end": "1780631115"},
            },
            "evidence": {
                "metrics": [
                    {
                        "metric_group_id": "tikv_unified_read_pool_wait_duration",
                        "instance": "db-tikv-19",
                        "detail": "read_pool.rs estimated_wait_ms=4690 and request rejected",
                    },
                    {
                        "metric_group_id": "rocksdb_compaction_reason",
                        "instance": "db-tikv-19",
                        "detail": "default-LevelMaxLevelSize compaction with RocksDB CPU spike",
                    },
                    {
                        "metric_group_id": "tikv_grpc_coprocessor_wait",
                        "instance": "db-tikv-19",
                        "detail": "gRPC and coprocessor wait spike while read pool queued requests",
                    },
                ]
            },
        },
        queries_payload={
            "slow_queries": [
                {
                    "digest": "point-get-during-stall",
                    "query": "Point_Get on item_description hit db-tikv-19 during the stall",
                    "source": "slow_query",
                },
                {
                    "digest": "large-batch-delete-precondition",
                    "query": "Large batch DELETEs created tombstone pressure before db-tikv-19 compaction",
                    "source": "slow_query",
                },
            ],
            "topsql": [],
            "decoded_plan": {},
        },
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [
                    {
                        "type": "metric",
                        "name": "Read pool queue overflow on single TiKV",
                        "metric_ref": {
                            "metric_group_id": "tikv_unified_read_pool_wait_duration",
                            "instances": ["db-tikv-19"],
                        },
                        "reason": "The hotspot TiKV instance carried the read-pool queue pressure.",
                    },
                    {
                        "type": "metric",
                        "name": "RocksDB compaction trigger",
                        "metric_ref": {
                            "metric_group_id": "rocksdb_compaction_reason",
                            "instances": ["db-tikv-19"],
                        },
                        "reason": "default-LevelMaxLevelSize compaction followed SST ingestion after Raft leader transfer.",
                    },
                ],
                "supporting_evidence": [
                    {
                        "type": "metric",
                        "name": "gRPC and coprocessor wait spike",
                        "metric_ref": {
                            "metric_group_id": "tikv_grpc_coprocessor_wait",
                            "instances": ["db-tikv-19"],
                        },
                        "reason": "Requests waited in queue rather than spending time in coprocessor handle work.",
                    }
                ],
            }
        ),
        include_summary_artifact=True,
    )
    summary_json_path = _evidence_file(audit_dir, "evidence_summary_for_agent.json")
    summary_json_path.write_text(
        json.dumps(
            {
                "summary": [
                    "node-level freeze on db-tikv-19",
                    "read pool queue overflow and request rejected signals",
                    "LevelMaxLevelSize compaction burst after SST ingestion and Raft leader transfer",
                ]
            }
        )
    )
    metrics_json_path = _evidence_file(audit_dir, "metrics.json")
    metrics_json_path.write_text(
        json.dumps(
            {
                "signals": [
                    {
                        "metric": "tikv_unified_read_pool_wait_duration",
                        "instance": "db-tikv-19",
                        "detail": "read pool pending tasks and estimated_wait_ms",
                    },
                    {
                        "metric": "tikv_grpc_msg_duration_seconds",
                        "instance": "db-tikv-19",
                        "p99_s": 32,
                    },
                    {
                        "metric": "rocksdb_compaction_reason",
                        "instance": "db-tikv-19",
                        "reason": "default-LevelMaxLevelSize compaction burst",
                    },
                ]
            }
        )
    )

    judgment = judge_case_output(
        case,
        output_text=output,
        workspace_id="ws-tikv-node-stall-case",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["diagnosis"]["status"] == "pass"
    assert judgment["packet"]["overall_verdict"] == "pass"
    diagnosis_checks = {item["id"]: item for item in judgment["packet"]["diagnosis"]["checks"]}
    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}
    assert diagnosis_checks["identify_compaction_read_pool_stall_as_primary_path"]["passed"] is True
    assert diagnosis_checks["identify_single_tikv_stall_scope"]["passed"] is True
    assert diagnosis_checks["identify_specific_hotspot_node_scope"]["passed"] is True
    assert diagnosis_checks["recommend_tikv_stall_mitigation"]["passed"] is True
    assert critical_checks["node_local_stall_signal_present"]["passed"] is True
    assert critical_checks["single_tikv_scope_evidence_present"]["passed"] is True
    assert critical_checks["compaction_trigger_evidence_present"]["passed"] is True


def test_tikv_node_stall_case_accepts_stall_aliases(tmp_path):
    case = load_case("tikv-node-stall-during-incident-window")
    output = """
Root Cause:
- SST ingestion after a raft leader transfer triggered compaction on db-tikv-19.
- The TiKV thread pool queue overflow caused request rejection and a short latency stall on one TiKV instance.

Supporting Evidence:
- gRPC and coprocessor waits spiked while requests queued in the read pool.

Recommendation:
- Rebalance leaders and reduce batch delete tombstone pressure.
"""
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10143988032832670331",
                "time_range": {"start": "1780630875", "end": "1780631115"},
            },
            "evidence": {
                "metrics": [
                    {
                        "metric_group_id": "tikv_read_pool_wait_duration",
                        "instance": "db-tikv-19",
                        "detail": "read pool queue overflow and request rejection",
                    }
                ]
            },
        },
        queries_payload={
            "slow_queries": [
                {
                    "digest": "point-get-during-stall",
                    "query": "Point_Get hit db-tikv-19 during request rejection",
                    "source": "slow_query",
                }
            ],
            "topsql": [],
            "decoded_plan": {},
        },
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [
                    {
                        "type": "metric",
                        "name": "Read pool queue overflow on hotspot TiKV",
                        "metric_ref": {
                            "metric_group_id": "tikv_read_pool_wait_duration",
                            "instances": ["db-tikv-19"],
                        },
                        "reason": "The hotspot TiKV instance carries request rejection pressure.",
                    }
                ]
            }
        ),
    )
    _evidence_file(audit_dir, "evidence_summary_for_agent.json").write_text(
        json.dumps(
            {
                "summary": [
                    "read pool queue overflow on db-tikv-19",
                    "RocksDB compaction after SST ingestion",
                    "single TiKV instance request rejection",
                ]
            }
        )
    )
    _evidence_file(audit_dir, "metrics.json").write_text(
        json.dumps(
            {
                "signals": [
                    {
                        "metric": "tikv_read_pool_wait_duration",
                        "instance": "db-tikv-19",
                        "detail": "request rejected after queue overflow",
                    }
                ]
            }
        )
    )

    judgment = judge_case_output(
        case,
        output_text=output,
        workspace_id="ws-tikv-node-stall-aliases",
        audit_dir=audit_dir,
    )

    diagnosis_checks = {item["id"]: item for item in judgment["packet"]["diagnosis"]["checks"]}
    assert diagnosis_checks["identify_compaction_read_pool_stall_as_primary_path"]["passed"] is True
    assert diagnosis_checks["identify_specific_hotspot_node_scope"]["passed"] is True


def test_tikv_node_stall_case_requires_explicit_hotspot_node_name(tmp_path):
    case = load_case("tikv-node-stall-during-incident-window")
    output = """
Root Cause:
- A RocksDB default CF compaction burst after Raft leader transfer saturated the unified read-pool.
- The pressure concentrated on a single TiKV instance and caused request rejection plus a short latency stall.

Supporting Evidence:
- read_pool.rs reported estimated_wait_ms spikes while grpc and coprocessor waits rose sharply.

Recommendation:
- Rebalance leaders and reduce batch delete tombstone pressure.
"""
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10143988032832670331",
                "time_range": {"start": "1780630875", "end": "1780631115"},
            },
            "evidence": {
                "metrics": [
                    {
                        "metric_group_id": "tikv_read_pool_wait_duration",
                        "instance": "db-tikv-19",
                        "detail": "read pool queue overflow and request rejection",
                    },
                    {
                        "metric_group_id": "rocksdb_compaction_reason",
                        "instance": "db-tikv-19",
                        "detail": "default-LevelMaxLevelSize compaction after SST ingestion",
                    }
                ]
            },
        },
        queries_payload={
            "slow_queries": [
                {
                    "digest": "point-get-during-stall",
                    "query": "Point_Get hit db-tikv-19 during request rejection",
                    "source": "slow_query",
                },
                {
                    "digest": "compaction-trigger-window",
                    "query": "MsgTimeoutNow and SST ingestion preceded LevelMaxLevelSize compaction on db-tikv-19",
                    "source": "slow_query",
                }
            ],
            "topsql": [],
            "decoded_plan": {},
        },
        analysis_patch_text=None,
    )
    _evidence_file(audit_dir, "evidence_summary_for_agent.json").write_text(
        json.dumps(
            {
                "summary": [
                    "node-level freeze on db-tikv-19",
                    "read pool queue overflow and request rejected signals",
                    "LevelMaxLevelSize compaction burst after SST ingestion and Raft leader transfer",
                ]
            }
        )
    )
    _evidence_file(audit_dir, "metrics.json").write_text(
        json.dumps(
            {
                "signals": [
                    {
                        "metric": "tikv_read_pool_wait_duration",
                        "instance": "db-tikv-19",
                        "detail": "request rejected after queue overflow",
                    },
                    {
                        "metric": "rocksdb_compaction_reason",
                        "instance": "db-tikv-19",
                        "reason": "default-LevelMaxLevelSize compaction burst",
                    },
                ]
            }
        )
    )

    judgment = judge_case_output(
        case,
        output_text=output,
        workspace_id="ws-tikv-node-stall-missing-explicit-node-name",
        audit_dir=audit_dir,
    )

    diagnosis_checks = {item["id"]: item for item in judgment["packet"]["diagnosis"]["checks"]}
    assert diagnosis_checks["identify_compaction_read_pool_stall_as_primary_path"]["passed"] is True
    assert diagnosis_checks["identify_specific_hotspot_node_scope"]["passed"] is False
    assert judgment["packet"]["diagnosis"]["status"] == "fail"
    assert judgment["packet"]["overall_verdict"] == "fail"
    assert judgment["primary_failure_code"] == "diagnosis_critical_gate_failed"


def test_panic_case_critical_evidence_reads_logs_json_messages(tmp_path):
    case = load_case("panic-in-recoverable-goroutine")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=[
            "issues/panic-in-recoverable-goroutine/issue_meta.json",
            "issues/panic-in-recoverable-goroutine/cluster_meta.json",
            "tier-architecture.json",
            "incorrect-conclusions.json",
            "Initialize diagnosis run directory",
            "collect_evidence.py",
        ],
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10518506906906971664",
                "time_range": {"start": "1776631800", "end": "1776639000"},
            },
            "evidence": {"metrics": []},
        },
        include_queries=False,
        include_gate_checks=False,
        log_stats_payload={
            "query": "{container=\"tidb\"} |~ `(?i)panic|recover|goroutine`",
            "log_counts": [
                {
                    "component": "tidb",
                    "level": "PANIC",
                    "message": "[2026/04/19 21:50:00.323 +00:00] [ERROR] [misc.go:95] [\"panic in the recoverable goroutine\"] [r=\"[executor:3024]Query execution was interrupted, maximum statement execution time exceeded\"]",
                    "count": 1,
                }
            ],
            "samples": [
                {
                    "component": "tidb",
                    "level": "PANIC",
                    "line": "[2026/04/19 21:50:00.323 +00:00] [ERROR] [misc.go:95] [\"panic in the recoverable goroutine\"] [r=\"[executor:3024]Query execution was interrupted, maximum statement execution time exceeded\"]",
                }
            ],
            "summary": {"tidb": {"error": 0, "warn": 0, "info": 0}},
        },
        analysis_patch_text=None,
    )

    judgment = judge_case_output(
        case,
        output_text="stub",
        workspace_id="ws-panic-logs-json",
        audit_dir=audit_dir,
    )

    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}
    assert critical_checks["panic_signal_present"]["passed"] is True
    assert critical_checks["query_timeout_interrupt_signal_present"]["passed"] is True


def test_restart_case_critical_evidence_reads_logs_json_messages(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10766841556523235152",
                "time_range": {"start": "1775039760", "end": "1775043360"},
            },
            "evidence": {"metrics": []},
        },
        log_stats_payload={
            "query": "{container=\"tidb\"} |~ `(?i)panic|interrupt|kill`",
            "log_counts": [
                {
                    "component": "tidb",
                    "level": "PANIC",
                    "message": "panic: [executor:1317]Query execution was interrupted",
                    "count": 1,
                },
                {
                    "component": "tidb",
                    "level": "INFO",
                    "message": "[INFO] [2pc.go:1070] [\"query is killed\"] [signal=1]",
                    "count": 1,
                },
            ],
            "samples": [
                {
                    "component": "tidb",
                    "level": "PANIC",
                    "line": "panic: [executor:1317]Query execution was interrupted",
                }
            ],
            "summary": {"tidb": {"error": 0, "warn": 0, "info": 0}},
        },
        analysis_patch_text=None,
    )

    judgment = judge_case_output(
        case,
        output_text="stub",
        workspace_id="ws-restart-logs-json",
        audit_dir=audit_dir,
    )

    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}
    assert critical_checks["panic_signal_present"]["passed"] is True
    assert critical_checks["query_interruption_signal_present"]["passed"] is True


def test_cpu_spike_capacity_context_accepts_structured_cluster_meta(tmp_path):
    case = load_case("cpu-spike-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10329350811427781110",
                "time_range": {"start": "1778291160", "end": "1778294460"},
            },
            "evidence": {"metrics": []},
        },
        include_summary_artifact=False,
        analysis_patch_text=None,
    )
    _evidence_file(audit_dir, "cluster_meta.json").write_text(
        json.dumps(
                {
                    "cluster": {
                        "components": {
                            "COMPONENT_TYPE_TIDB": {
                                "tidbGroups": [
                                    {"groupName": "baseline", "replicas": 11},
                                    {"groupName": "incident", "replicas": 14},
                                ]
                            }
                        }
                }
            }
        )
    )

    judgment = judge_case_output(
        case,
        output_text="stub",
        workspace_id="ws-cpu-structured-capacity",
        audit_dir=audit_dir,
    )

    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}
    assert critical_checks["capacity_comparison_context_present"]["passed"] is True
    assert any(
        rule["matcher"] == "json_path_unique_count_gte" and rule["matched"]
        for rule in critical_checks["capacity_comparison_context_present"]["rules"]
    )


def test_tikv_node_stall_critical_evidence_accepts_compaction_stall_fragments(tmp_path):
    case = load_case("tikv-node-stall-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10143988032832670331",
                "time_range": {"start": "1780630875", "end": "1780631115"},
            },
            "evidence": {"metrics": []},
        },
        queries_payload={
            "slow_queries": [
                {
                    "source": "slow_query",
                    "digest": "point-get-during-db-tikv-19-stall",
                    "sql_preview": "Point_Get on item_description saw db-tikv-19 latency during the stall",
                }
            ],
            "topsql": [],
            "decoded_plan": {
                "digest": "point-get-during-db-tikv-19-stall",
                "source": "slow_query",
                "full_sql": "point get workload hit db-tikv-19 during the latency stall",
                "decoded_plan": (
                    "{\n"
                    '  "main": {\n'
                    '    "rootGroupExecInfo": [\n'
                    '      "store: db-tikv-19, read_pool.rs estimated_wait_ms=4690, request rejected, '
                    'RocksDB default-LevelMaxLevelSize compaction, IngestExternalFile after MsgTimeoutNow Raft leader transfer"\n'
                    "    ]\n"
                    "  }\n"
                    "}"
                ),
            },
        },
        analysis_patch_text=None,
    )

    judgment = judge_case_output(
        case,
        output_text="stub",
        workspace_id="ws-tikv-node-stall-exact-fragments",
        audit_dir=audit_dir,
    )

    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}
    assert critical_checks["node_local_stall_signal_present"]["passed"] is True
    assert critical_checks["single_tikv_scope_evidence_present"]["passed"] is True
    assert critical_checks["compaction_trigger_evidence_present"]["passed"] is True


def test_sustained_cpu_critical_evidence_reads_workload_and_ignores_optional_heap_profile_miss(tmp_path):
    case = load_case("sustained-cpu-pressure-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10518506906906971664",
                "time_range": {"start": "1773741600", "end": "1773797400"},
            },
            "evidence": {
                "metrics": [],
                "profiles": [],
            },
        },
        analysis_patch_text=None,
    )
    _evidence_file(audit_dir, "workload.json").write_text(
        json.dumps(
            {
                "workload": [
                    {
                        "sql_type": "NTDML-Delete",
                        "statement_rate_peak": 3825.6167,
                        "statement_rate_mean": 582.1657,
                        "affected_rows_rate_peak": 985.1167,
                        "affected_rows_rate_mean": 37.2491,
                    }
                ]
            }
        )
    )
    _evidence_file(audit_dir, "workspace_evidence_report.json").write_text(
        json.dumps(
            {
                "type": "workspace_evidence_report",
                "metadata": {
                    "cluster_id": "10518506906906971664",
                    "time_range": {"start": "1773741600", "end": "1773797400"},
                    "run_dir": str(_evidence_run_dir(audit_dir)),
                    "evidence_dir": str(_evidence_run_dir(audit_dir)),
                },
                "evidence": {
                    "metrics": [],
                    "profiles": [],
                },
            }
        )
    )

    judgment = judge_case_output(
        case,
        output_text="stub",
        workspace_id="ws-sustained-workload",
        audit_dir=audit_dir,
    )

    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}
    assert critical_checks["primary_workload_shift_present"]["passed"] is True
    assert critical_checks["allocation_pressure_signal_present"]["passed"] is False
    assert critical_checks["heap_profile_hotspot_present"]["passed"] is False
    assert critical_checks["heap_profile_hotspot_present"]["required"] is False
    assert judgment["packet"]["critical_evidence"]["status"] == "partial"


def test_json_values_contains_any_matches_leaf_values_not_json_keys(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    case = _case_with_critical_evidence(
        base_case,
        critical_evidence={
            "weight": 30,
            "checks": [
                {
                    "id": "pseudo_stats_present",
                    "weight": 10,
                    "description": "Pseudo stats evidence is present.",
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "artifact_json",
                            "artifacts": ["queries"],
                            "matcher": "json_values_contains_any",
                            "path": "slow_queries",
                            "keywords": ["stats:pseudo"],
                        }
                    ],
                }
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        queries_payload={
                "slow_queries": [
                    {
                        "stats_info": "stats:pseudo[140;4738]",
                        "decoded_plan": "CARTESIAN inner join",
                    }
                ],
            "topsql": [],
            "decoded_plan": {},
        },
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-json-values", audit_dir=audit_dir)
    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}

    assert critical_checks["pseudo_stats_present"]["passed"] is True


def test_json_values_contains_any_does_not_match_json_key_names(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    case = _case_with_critical_evidence(
        base_case,
        critical_evidence={
            "weight": 30,
            "checks": [
                {
                    "id": "field_name_only",
                    "weight": 10,
                    "description": "Field-name-only evidence should not match.",
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "artifact_json",
                            "artifacts": ["component_health"],
                            "matcher": "json_values_contains_any",
                            "path": "component_health",
                            "keywords": ["worker_memory"],
                        }
                    ],
                }
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        component_health_payload={
            "component_health": {
                "component_count": [],
                "restarts": [],
                "worker_memory": [],
                "links": {},
            }
        },
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-json-values-keys", audit_dir=audit_dir)
    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}

    assert critical_checks["field_name_only"]["passed"] is False


def test_json_values_nonempty_matches_nested_metric_series_values(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    case = _case_with_critical_evidence(
        base_case,
        critical_evidence={
            "weight": 30,
            "checks": [
                {
                    "id": "metric_series_nonempty",
                    "weight": 10,
                    "description": "Metrics evidence includes at least one non-empty series payload.",
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "artifact_json",
                            "artifacts": ["metrics"],
                            "matcher": "json_values_nonempty",
                            "path": "metrics.chart_data.series.values",
                        }
                    ],
                }
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    _evidence_file(audit_dir, "metrics.json").write_text(
        json.dumps(
            {
                "metrics": [
                    {
                        "name": "TiDB P99 Latency",
                        "chart_data": {
                            "series": [
                                {
                                    "instance": "db-tidb-0",
                                    "values": [
                                        [1775039760, "0.0123"],
                                        [1775039820, "0.0118"],
                                    ],
                                }
                            ]
                        },
                    }
                ]
            },
            ensure_ascii=False,
        )
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-json-values-nonempty",
        audit_dir=audit_dir,
    )
    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}

    assert critical_checks["metric_series_nonempty"]["passed"] is True


def test_json_values_nonempty_fails_for_empty_metric_series_values(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    case = _case_with_critical_evidence(
        base_case,
        critical_evidence={
            "weight": 30,
            "checks": [
                {
                    "id": "metric_series_nonempty",
                    "weight": 10,
                    "description": "Metrics evidence includes at least one non-empty series payload.",
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "artifact_json",
                            "artifacts": ["metrics"],
                            "matcher": "json_values_nonempty",
                            "path": "metrics.chart_data.series.values",
                        }
                    ],
                }
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    _evidence_file(audit_dir, "metrics.json").write_text(
        json.dumps(
            {
                "metrics": [
                    {
                        "name": "TiDB P99 Latency",
                        "chart_data": {
                            "series": [
                                {
                                    "instance": "db-tidb-0",
                                    "values": [],
                                }
                            ]
                        },
                    }
                ]
            },
            ensure_ascii=False,
        )
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-json-values-empty",
        audit_dir=audit_dir,
    )
    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}

    assert critical_checks["metric_series_nonempty"]["passed"] is False


def test_json_text_contains_any_matches_unescaped_json_string_values(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    case = _case_with_critical_evidence(
        base_case,
        critical_evidence={
            "weight": 30,
            "checks": [
                {
                    "id": "quoted_label_filter_present",
                    "weight": 10,
                    "description": "Metrics evidence includes the quoted label filter in a string value.",
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "artifact_json",
                            "artifacts": ["metrics"],
                            "matcher": "json_text_contains_any",
                            "keywords": ['type="Get"'],
                        }
                    ],
                }
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    _evidence_file(audit_dir, "metrics.json").write_text(
        json.dumps(
            {
                "metrics": [
                    {
                        "metric_group_id": "tidb_tikvclient_request_seconds_bucket",
                        "promql": 'histogram_quantile(0.99, sum by (le, instance) (rate(tidb_tikvclient_request_seconds_bucket{type="Get"}[1m])))',
                    }
                ]
            },
            ensure_ascii=False,
        )
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-json-text-quoted-values",
        audit_dir=audit_dir,
    )
    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}

    assert critical_checks["quoted_label_filter_present"]["passed"] is True


def test_json_values_not_contains_any_can_forbid_broad_instance_regex(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    case = _case_with_critical_evidence(
        base_case,
        critical_evidence={
            "weight": 30,
            "checks": [
                {
                    "id": "broad_instance_regex_absent",
                    "weight": 10,
                    "description": "Metrics evidence must not use a broad db-tikv-9.* instance regex.",
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "artifact_json",
                            "artifacts": ["metrics"],
                            "matcher": "json_values_not_contains_any",
                            "path": "metrics.promql",
                            "keywords": ['instance=~"db-tikv-9.*"'],
                        }
                    ],
                }
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    _evidence_file(audit_dir, "metrics.json").write_text(
        json.dumps(
            {
                "metrics": [
                    {
                        "metric_group_id": "tikv_engine_pending_compaction_bytes",
                        "promql": 'max_over_time(tikv_engine_pending_compaction_bytes{instance=~"db-tikv-9.*",cf="write"}[5m])',
                    }
                ]
            }
        )
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-broad-instance-regex",
        audit_dir=audit_dir,
    )
    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}

    assert critical_checks["broad_instance_regex_absent"]["passed"] is False
    assert critical_checks["broad_instance_regex_absent"]["rules"][0]["matched"] is False
    assert critical_checks["broad_instance_regex_absent"]["rules"][0]["matched_keyword"] == 'instance=~"db-tikv-9.*"'


def test_logs_json_query_text_does_not_satisfy_log_keyword_rules(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=False,
        log_stats_payload={
            "query": "{component=~\"tidb\"} |= \"panic|kill query|sqlkiller.go\"",
            "log_counts": [],
            "samples": [],
        },
        component_health_payload={
            "component_health": {
                "component_count": [],
                "restarts": [],
                "worker_memory": [],
                "links": {},
            }
        },
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-logs-query-only",
        audit_dir=audit_dir,
    )

    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}
    assert critical_checks["panic_signal_present"]["passed"] is False
    assert critical_checks["query_interruption_signal_present"]["passed"] is False


def test_critical_evidence_optional_miss_does_not_downgrade_pass(tmp_path):
    case = load_case("sustained-cpu-pressure-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10518506906906971664",
                "time_range": {"start": "1773741600", "end": "1773797400"},
            },
            "evidence": {
                "metrics": [
                    {
                        "name": "GC duration count",
                        "promql": "rate(go_gc_duration_seconds_count{component=\"tidb\"}[1m])",
                    }
                ],
                "profiles": [],
            },
        },
        analysis_patch_text=None,
    )
    _evidence_file(audit_dir, "workload.json").write_text(
        json.dumps(
            {
                "workload": [
                    {
                        "sql_type": "NTDML-Delete",
                        "statement_rate_peak": 3825.6167,
                        "affected_rows_rate_peak": 985.1167,
                    }
                ]
            }
        )
    )
    _evidence_file(audit_dir, "workspace_evidence_report.json").write_text(
        json.dumps(
            {
                "type": "workspace_evidence_report",
                "metadata": {
                    "cluster_id": "10518506906906971664",
                    "time_range": {"start": "1773741600", "end": "1773797400"},
                    "run_dir": str(_evidence_run_dir(audit_dir)),
                    "evidence_dir": str(_evidence_run_dir(audit_dir)),
                },
                "evidence": {
                    "metrics": [
                        {
                            "name": "GC duration count",
                            "promql": "rate(go_gc_duration_seconds_count{component=\"tidb\"}[1m])",
                        }
                    ],
                    "profiles": [],
                },
            }
        )
    )
    _evidence_file(audit_dir, "metrics.json").write_text(
        json.dumps(
            {
                "signals": [
                    {
                        "name": "GC duration count",
                        "promql": "rate(go_gc_duration_seconds_count{component=\"tidb\"}[1m])",
                    }
                ]
            }
        )
    )

    judgment = judge_case_output(
        case,
        output_text="stub",
        workspace_id="ws-sustained-optional-heap-profile",
        audit_dir=audit_dir,
    )

    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}
    assert critical_checks["primary_workload_shift_present"]["passed"] is True
    assert critical_checks["allocation_pressure_signal_present"]["passed"] is False
    assert critical_checks["heap_profile_hotspot_present"]["passed"] is False
    assert judgment["packet"]["critical_evidence"]["status"] == "partial"


def test_unified_diagnosis_checks_use_yaml_rules(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-diagnosis-supporting-yaml",
        audit_dir=audit_dir,
    )

    diagnosis = judgment["packet"]["diagnosis"]
    checks_by_id = {item["id"]: item for item in diagnosis["checks"]}

    assert diagnosis["max_score"] == DIAGNOSIS_BUCKET_MAX
    assert sorted(checks_by_id) == [
        "identify_kill_query_cleanup_chain",
        "identify_known_bug_restart_chain",
        "mention_github_issue_id",
        "recommend_fixed_version_upgrade",
        "reject_missing_index_root_cause",
    ]
    assert checks_by_id["identify_known_bug_restart_chain"]["deterministic_pass"] is True
    assert checks_by_id["identify_known_bug_restart_chain"]["earned_score"] == 2.0
    assert checks_by_id["identify_kill_query_cleanup_chain"]["deterministic"]["rules"][0]["matcher"] == "bool_true"
    assert checks_by_id["reject_missing_index_root_cause"]["mode"] == "deterministic_only"
    assert all(item["passed"] is True for item in checks_by_id.values())
    assert judgment["judgment_verdict"] == "pass"


def test_unified_diagnosis_blended_check_accepts_without_fix_alias(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_diagnosis_checks(
        case,
        checks=[
            {
                "id": "identify_version_without_fix",
                "mode": "blended",
                "weight": 3,
                "required": True,
                "description": "States that v8.5.2 predates or does not include the fix.",
                "deterministic": {
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_fact",
                            "matcher": "bool_true",
                            "path": "mentions_v8_5_2",
                        },
                        {
                            "source": "output_fact",
                            "matcher": "any_true",
                            "paths": [
                                "mentions_v8_5_3_or_fixed_release",
                                "mentions_without_fix",
                            ],
                        },
                    ],
                },
                "semantic": {
                    "id": 1,
                    "title": "版本关系判断",
                    "description": "判断回答是否说明 v8.5.2 不包含修复。",
                },
            }
        ],
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    output = """
Root Cause:
- TiDB v8.5.2 predates the fix and is running without the fix.
"""

    judgment = judge_case_output(
        case,
        output_text=output,
        workspace_id="ws-diagnosis-supporting-any-true",
        audit_dir=audit_dir,
    )

    diagnosis = judgment["packet"]["diagnosis"]
    assert diagnosis["checks"][0]["deterministic_pass"] is True
    assert diagnosis["checks"][0]["earned_score"] == 2.0


def test_unified_diagnosis_supporting_falls_back_to_legacy_logic_when_schema_missing(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_diagnosis_checks(case, checks=None)
    case = _case_with_diagnosis_hard_checks(
        case,
        hard_checks=[
            {
                "id": "identify_single_instance_panic_restart",
                "description": "Identifies a TiDB panic event and/or single-instance restart.",
                "match": "all_of",
                "rules": [
                    {
                        "source": "output_fact",
                        "matcher": "bool_true",
                        "path": "diagnosis_single_instance_panic_restart_asserted",
                    }
                ],
            }
        ],
    )
    case = _case_with_diagnosis_supporting(
        case,
        supporting={
            "weight": 12,
            "checks": [
                {
                    "id": "identify_version_without_fix",
                    "weight": 12,
                    "description": "States that v8.5.2 predates or does not include the fix.",
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_fact",
                            "matcher": "bool_true",
                            "path": "mentions_v8_5_2",
                        },
                        {
                            "source": "output_fact",
                            "matcher": "any_true",
                            "paths": [
                                "mentions_v8_5_3_or_fixed_release",
                                "mentions_without_fix",
                            ],
                        },
                    ],
                }
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-diagnosis-supporting-legacy",
        audit_dir=audit_dir,
    )

    diagnosis = judgment["packet"]["diagnosis"]
    assert len(diagnosis["checks"]) == 2
    assert all(item["passed"] is True for item in diagnosis["checks"])


def test_unified_diagnosis_bool_false_with_unknown_fact_fails_closed(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_diagnosis_checks(
        case,
        checks=[
            {
                "id": "reject_unknown_flag",
                "mode": "deterministic_only",
                "weight": 3,
                "required": True,
                "description": "Invalid output_fact paths must not accidentally pass negated checks.",
                "deterministic": {
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_fact",
                            "matcher": "bool_false",
                            "path": "unknown_output_fact_typo",
                        }
                    ],
                },
            }
        ],
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-diagnosis-supporting-invalid-bool-false",
        audit_dir=audit_dir,
    )

    diagnosis = judgment["packet"]["diagnosis"]
    assert diagnosis["score"] == 0
    assert diagnosis["max_score"] == DIAGNOSIS_BUCKET_MAX
    assert diagnosis["checks"][0]["passed"] is False
    assert diagnosis["checks"][0]["deterministic"]["rules"][0]["matched"] is False


def test_unified_diagnosis_any_true_with_mixed_valid_invalid_paths_fails_closed(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_diagnosis_checks(
        case,
        checks=[
            {
                "id": "mixed_any_true_paths",
                "mode": "deterministic_only",
                "weight": 3,
                "required": True,
                "description": "Mixed valid and invalid output_fact paths must not partially match.",
                "deterministic": {
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_fact",
                            "matcher": "any_true",
                            "paths": [
                                "mentions_without_fix",
                                "unknown_output_fact_typo",
                            ],
                        }
                    ],
                },
            }
        ],
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-diagnosis-supporting-invalid-any-true",
        audit_dir=audit_dir,
    )

    diagnosis = judgment["packet"]["diagnosis"]
    assert diagnosis["score"] == 0
    assert diagnosis["max_score"] == DIAGNOSIS_BUCKET_MAX
    assert diagnosis["checks"][0]["passed"] is False
    assert diagnosis["checks"][0]["deterministic"]["rules"][0]["matched"] is False


def test_unified_diagnosis_penalty_check_reduces_score_without_failing_required_gate(tmp_path):
    case = load_case("component-restart-during-incident-window")
    case = _case_with_diagnosis_checks(
        case,
        checks=[
            {
                "id": "identify_known_bug_chain",
                "mode": "deterministic_only",
                "weight": 3,
                "required": True,
                "description": "Identifies the known-bug restart chain.",
                "deterministic": {
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_fact",
                            "matcher": "bool_true",
                            "path": "diagnosis_single_instance_panic_restart_asserted",
                        }
                    ],
                },
            },
            {
                "id": "reject_missing_index_root_cause",
                "mode": "deterministic_only",
                "scoring_mode": "penalty",
                "weight": 3,
                "required": False,
                "description": "Penalizes missing-index misdiagnosis.",
                "deterministic": {
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_fact",
                            "matcher": "bool_false",
                            "path": "missing_index_primary_root_cause_asserted",
                        }
                    ],
                },
            },
        ],
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    output = """
Root Cause:
- A single-instance TiDB panic/restart occurred.
- Missing index is the root cause.
"""

    judgment = judge_case_output(
        case,
        output_text=output,
        workspace_id="ws-diagnosis-penalty-mode",
        audit_dir=audit_dir,
    )

    diagnosis = judgment["packet"]["diagnosis"]
    checks_by_id = {item["id"]: item for item in diagnosis["checks"]}
    assert diagnosis["status"] == "pass"
    assert diagnosis["score"] == 0
    assert checks_by_id["identify_known_bug_chain"]["earned_score"] == 3.0
    assert checks_by_id["reject_missing_index_root_cause"]["scoring_mode"] == "penalty"
    assert checks_by_id["reject_missing_index_root_cause"]["earned_score"] == -3.0
    assert checks_by_id["reject_missing_index_root_cause"]["penalty_applied"] == 3.0


def test_artifact_resolution_failure_is_invalid_with_artifact_resolution_subtype(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_diagnosis_report=False,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-artifact-resolution-fail",
        audit_dir=audit_dir,
    )

    assert judgment["judgment_verdict"] == "invalid"
    assert judgment["primary_failure_code"] == "artifact_resolution_failed"
    assert judgment["invalid_subtype"] == "artifact_resolution"
    assert judgment["packet"]["artifact_resolution"]["status"] == "fail"
    assert judgment["audit"]["artifact_resolution"]["selected_scope"] == "audit_dir"


def test_artifact_resolution_success_exposes_resolved_paths_in_facts(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-artifact-resolution-pass",
        audit_dir=audit_dir,
    )

    artifacts = judgment["facts"]["artifacts"]
    assert judgment["judgment_verdict"] == "pass"
    assert artifacts["source_scope"] == "audit_dir"
    assert artifacts["run_dir"] == "evidences"
    assert artifacts["resolution_mode"] == "workspace_evidence_root"
    assert artifacts["diagnosis_report_path"].endswith("workspace_evidence_report.json")


def test_artifact_resolution_uses_workspace_evidence_root_mode(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-artifact-resolution-fallback",
        audit_dir=audit_dir,
    )

    assert judgment["judgment_verdict"] == "pass"
    assert judgment["facts"]["artifacts"]["resolution_mode"] == "workspace_evidence_root"
    assert judgment["audit"]["artifact_resolution"]["resolution_mode"] == "workspace_evidence_root"


def test_artifact_resolution_invalid_when_workspace_report_json_malformed(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    _evidence_file(audit_dir, "workspace_evidence_report.json").write_text("{bad json")

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-artifact-resolution-malformed-report",
        audit_dir=audit_dir,
    )

    assert judgment["judgment_verdict"] == "invalid"
    assert judgment["primary_failure_code"] == "artifact_resolution_failed"
    assert judgment["invalid_subtype"] == "artifact_resolution"


def test_execution_contract_can_require_summary_read_completed(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    execution_contract = deepcopy(base_case.data["judge"]["execution_contract"])
    execution_contract["checks"][0]["rules"][0]["paths"].append("summary_read_completed")
    case = _case_with_execution_contract(
        base_case,
        execution_contract=execution_contract,
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_summary_artifact=False,
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-contract-summary-required",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["workflow"]["status"] == "pass"
    assert judgment["packet"]["execution_contract"]["status"] == "fail"
    assert judgment["judgment_verdict"] == "invalid"
    assert judgment["primary_failure_code"] == "execution_contract_failed"
    failed_check = next(
        item
        for item in judgment["facts"]["execution_contract"]["checks"]
        if item["id"] == "required_workflow_present"
    )
    assert failed_check["passed"] is False
    assert any(
        rule["matcher"] == "subset_observed" and "summary_read_completed" in rule["paths"]
        for rule in failed_check["rules"]
    )


def test_execution_contract_can_require_summary_artifact(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    execution_contract = deepcopy(base_case.data["judge"]["execution_contract"])
    execution_contract["checks"][1]["rules"][0]["paths"].append("summary_artifact_present")
    case = _case_with_execution_contract(
        base_case,
        execution_contract=execution_contract,
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_summary_artifact=False,
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-contract-summary-artifact",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["execution_contract"]["status"] == "fail"
    assert judgment["judgment_verdict"] == "invalid"
    assert judgment["primary_failure_code"] == "execution_contract_failed"
    assert judgment["score"] == 0
    assert judgment["packet"]["scoring"]["raw_total_score"] == 0
    assert judgment["packet"]["workflow"]["score"] == 0
    assert judgment["packet"]["evidence_collection"]["score"] == 0
    assert judgment["packet"]["critical_evidence"]["score"] == 0
    assert judgment["packet"]["evidence_routing"]["score"] == 0
    assert judgment["packet"]["diagnosis"]["score"] == 0
    failed_check = next(
        item
        for item in judgment["facts"]["execution_contract"]["checks"]
        if item["id"] == "required_artifacts_present"
    )
    assert failed_check["passed"] is False
    assert any(
        rule["matcher"] == "subset_observed" and "summary_artifact_present" in rule["paths"]
        for rule in failed_check["rules"]
    )


def test_adhoc_evidence_mode_derives_collection_completed_from_workspace_report(tmp_path):
    base_case = load_case("cpu-spike-during-incident-window")
    data = deepcopy(base_case.data)
    data["evaluation"] = {"mode": "adhoc_evidence"}
    data["judge"]["workflow"] = {
        "checks": [
            {
                "id": "diagnosis_run_initialized",
                "weight": 5,
                "required": True,
                "description": "Diagnosis run is initialized.",
                "match": "all_of",
                "rules": [{"source": "workflow_fact", "matcher": "bool_true", "path": "diagnosis_run_initialized"}],
            },
            {
                "id": "evidence_collection_completed",
                "weight": 10,
                "required": True,
                "description": "Ad-hoc evidence collection completed.",
                "match": "all_of",
                "rules": [{"source": "workflow_fact", "matcher": "bool_true", "path": "evidence_collection_completed"}],
            },
        ]
    }
    data["judge"]["execution_contract"] = {
        "checks": [
            {
                "id": "required_workflow_present",
                "required": True,
                "description": "Required ad-hoc workflow milestones are present.",
                "match": "all_of",
                "rules": [
                    {
                        "source": "workflow_check",
                        "matcher": "subset_observed",
                        "paths": ["diagnosis_run_initialized", "evidence_collection_completed"],
                    }
                ],
            },
            {
                "id": "required_report_present",
                "required": True,
                "description": "Workspace evidence report is present.",
                "match": "all_of",
                "rules": [
                    {
                        "source": "artifact_fact",
                        "matcher": "subset_observed",
                        "paths": ["diagnosis_report_present"],
                    }
                ],
            },
            {
                "id": "preferred_channels_minimum_observed",
                "required": True,
                "description": "At least one ad-hoc evidence channel is observed.",
                "match": "all_of",
                "rules": [
                    {
                        "source": "channel_fact",
                        "matcher": "count_gte",
                        "paths": ["query_or_metric"],
                        "threshold": 1,
                    }
                ],
            },
        ]
    }
    data["judge"]["evidence_collection"] = {
        "weight": 5,
        "checks": [
            {
                "id": "required_report_present",
                "weight": 3,
                "required": True,
                "description": "Workspace evidence report is present.",
                "match": "all_of",
                "rules": [
                    {
                        "source": "artifact_fact",
                        "matcher": "subset_observed",
                        "paths": ["diagnosis_report_present"],
                    }
                ],
            },
            {
                "id": "preferred_channels_minimum_observed",
                "weight": 2,
                "required": True,
                "description": "At least one ad-hoc evidence channel is observed.",
                "match": "all_of",
                "rules": [
                    {
                        "source": "channel_fact",
                        "matcher": "count_gte",
                        "paths": ["query_or_metric"],
                        "threshold": 1,
                    }
                ],
            },
        ],
    }
    data["judge"]["critical_evidence"] = {
        "weight": 5,
        "checks": [
            {
                "id": "metric_signal_present",
                "weight": 5,
                "required": True,
                "description": "Metric evidence is present.",
                "match": "any_of",
                "rules": [
                    {
                        "source": "artifact_json",
                        "artifacts": ["metrics"],
                        "matcher": "json_text_contains_any",
                        "keywords": ["adhoc_total_qps"],
                    }
                ],
            }
        ],
    }
    data["judge"]["diagnosis"] = {
        "checks": [
            {
                "id": "answer_mentions_qps",
                "mode": "deterministic_only",
                "weight": 5,
                "required": True,
                "description": "Answer mentions QPS conclusion.",
                "deterministic": {
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "output_text",
                            "matcher": "keyword_any",
                            "keywords": ["QPS"],
                        }
                    ],
                },
            }
        ]
    }
    case = replace(base_case, data=data)
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=["Initialize diagnosis run directory"],
        include_queries=False,
        include_gate_checks=False,
        include_collection_errors=False,
        include_log_stats=False,
        include_logs_dir=False,
        include_summary_artifact=False,
        diagnosis_report_payload={
            "type": "workspace_evidence_report",
            "metadata": {
                "cluster_id": "10518506906906971664",
                "time_range": {"start": "1775599200", "end": "1775611800"},
            },
            "evidence": {
                "metrics": [
                    {
                        "metric_group_id": "adhoc_total_qps",
                        "name": "adhoc_total_qps",
                        "summary": {"avg": 42.0, "max": 57.0},
                    }
                ]
            },
        },
    )
    _evidence_file(audit_dir, "metrics.json").write_text(
        json.dumps(
            {
                "metrics": [
                    {
                        "metric_group_id": "adhoc_total_qps",
                        "name": "adhoc_total_qps",
                        "summary": {"avg": 42.0, "max": 57.0},
                    }
                ]
            }
        )
    )

    judgment = judge_case_output(
        case,
        output_text="QPS 在该时间窗内明显下降。",
        workspace_id="ws-adhoc-evidence-mode",
        audit_dir=audit_dir,
    )

    assert judgment["facts"]["workflow"]["evidence_collection_completed"] is True


def test_adhoc_critical_evidence_accepts_prose_before_inline_json(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    data = deepcopy(base_case.data)
    data["evaluation"] = {"mode": "adhoc_evidence"}
    data["judge"]["workflow"] = {
        "checks": [
            {
                "id": "final_answer_emitted",
                "weight": 0,
                "required": True,
                "description": "Final answer is emitted.",
                "match": "all_of",
                "rules": [{"source": "workflow_fact", "matcher": "bool_true", "path": "final_answer_emitted"}],
            }
        ]
    }
    data["judge"]["execution_contract"] = {
        "checks": [
            {
                "id": "required_workflow_present",
                "required": True,
                "description": "Minimal ad-hoc workflow is present.",
                "match": "all_of",
                "rules": [
                    {
                        "source": "workflow_check",
                        "matcher": "subset_observed",
                        "paths": ["final_answer_emitted"],
                    }
                ],
            }
        ]
    }
    data["judge"]["evidence_collection"] = {"weight": 0, "checks": []}
    data["judge"]["critical_evidence"] = {
        "weight": 80,
        "checks": [
            {
                "id": "cluster_id_present",
                "weight": 10,
                "required": True,
                "description": "Output JSON cluster_id is present.",
                "match": "all_of",
                "rules": [
                    {
                        "source": "output_json",
                        "matcher": "json_values_nonempty",
                        "path": "cluster_id",
                    }
                ],
            }
        ],
    }
    data["judge"]["diagnosis"] = {
        "supporting": {"weight": 0, "checks": []},
        "semantic": {"enabled": False, "checks": []},
    }
    case = replace(base_case, data=data)

    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    output_text = """
这是补充说明。
{"cluster_id":"10766841556523235152"}
"""

    judgment = judge_case_output(
        case,
        output_text=output_text,
        workspace_id="ws-adhoc-inline-json",
        audit_dir=audit_dir,
    )

    critical_checks = {item["id"]: item for item in judgment["packet"]["critical_evidence"]["checks"]}
    assert critical_checks["cluster_id_present"]["passed"] is True
    assert judgment["packet"]["critical_evidence"]["status"] == "pass"
    assert judgment["packet"]["workflow"]["status"] == "pass"
    assert judgment["packet"]["execution_contract"]["status"] == "pass"
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["judgment_verdict"] == "pass"


def test_execution_contract_can_forbid_collect_evidence_for_adhoc_case(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    workflow = deepcopy(base_case.data["judge"]["workflow"])
    workflow["checks"].append(
        {
            "id": "collect_evidence_not_used",
            "weight": 0,
            "required": True,
            "description": "ad-hoc path must not invoke collect_evidence.py",
            "match": "all_of",
            "rules": [
                {"source": "workflow_fact", "matcher": "bool_false", "path": "collect_evidence_invoked"}
            ],
        }
    )
    execution_contract = deepcopy(base_case.data["judge"]["execution_contract"])
    execution_contract["checks"].append(
        {
            "id": "collect_evidence_not_used",
            "required": True,
            "description": "ad-hoc path must not invoke collect_evidence.py",
            "match": "all_of",
            "rules": [
                {
                    "source": "workflow_check",
                    "matcher": "subset_observed",
                    "paths": ["collect_evidence_not_used"],
                }
            ],
        }
    )
    case = _case_with_execution_contract(
        _case_with_judge_workflow(base_case, judge_workflow=workflow),
        execution_contract=execution_contract,
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=_default_telemetry(case.id),
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-forbid-collect-evidence",
        audit_dir=audit_dir,
    )

    assert judgment["facts"]["workflow"]["collect_evidence_invoked"] is True
    assert judgment["packet"]["workflow"]["status"] == "partial"
    assert judgment["packet"]["execution_contract"]["status"] == "fail"
    assert judgment["judgment_verdict"] == "invalid"
    assert judgment["primary_failure_code"] == "execution_contract_failed"


def test_execution_contract_can_relax_channel_requirement_to_logs_only(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    execution_contract = deepcopy(base_case.data["judge"]["execution_contract"])
    execution_contract["checks"][2]["rules"][0]["paths"] = ["logs"]
    execution_contract["checks"][3]["rules"][0]["paths"] = ["logs"]
    case = _case_with_execution_contract(
        base_case,
        execution_contract=execution_contract,
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        queries_payload={"slow_queries": [], "topsql": [], "decoded_plan": {}},
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-contract-logs-only",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["execution_contract"]["status"] == "pass"
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["judgment_verdict"] == "pass"
    assert judgment["facts"]["execution_contract"]["channel_facts"]["logs"] is True
    assert judgment["facts"]["execution_contract"]["channel_facts"]["query_or_metric"] is False


def test_workflow_optional_checks_excluded_from_denominator(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=_default_telemetry(case.id),
        include_summary_artifact=False,
        include_logs_dir=False,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "direct restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "interrupted", "reason": "query interrupted logs"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-optional-workflow", audit_dir=audit_dir)
    workflow = judgment["packet"]["workflow"]

    assert workflow["status"] == "pass"
    assert workflow["raw_max_score"] == 25
    assert workflow["raw_score"] == 25
    assert workflow["score"] == WORKFLOW_BUCKET_MAX
    assert judgment["facts"]["workflow"]["summary_read_completed"] is False
    assert judgment["facts"]["workflow"]["targeted_panic_log_query_completed"] is False


def test_judge_workflow_yaml_policy_excludes_optional_checks_from_denominator(tmp_path):
    case = _case_with_judge_workflow(
        load_case("component-restart-during-incident-window"),
        judge_workflow={
            "checks": [
                {
                    "id": "architecture_context_loaded",
                    "weight": 5,
                    "required": True,
                    "description": "kb step a",
                    "match": "all_of",
                    "rules": [
                        {"source": "workflow_fact", "matcher": "bool_true", "path": "architecture_context_loaded"}
                    ],
                },
                {
                    "id": "misdiagnosis_guardrails_loaded",
                    "weight": 5,
                    "required": True,
                    "description": "kb step b",
                    "match": "all_of",
                    "rules": [
                        {"source": "workflow_fact", "matcher": "bool_true", "path": "misdiagnosis_guardrails_loaded"}
                    ],
                },
                {
                    "id": "diagnosis_run_initialized",
                    "weight": 5,
                    "required": True,
                    "description": "init",
                    "match": "all_of",
                    "rules": [
                        {"source": "workflow_fact", "matcher": "bool_true", "path": "diagnosis_run_initialized"}
                    ],
                },
                {
                    "id": "evidence_collection_completed",
                    "weight": 10,
                    "required": True,
                    "description": "evidence completed",
                    "match": "all_of",
                    "rules": [
                        {"source": "workflow_fact", "matcher": "bool_true", "path": "evidence_collection_completed"}
                    ],
                },
                {
                    "id": "summary_read_completed",
                    "weight": 5,
                    "required": False,
                    "description": "summary optional",
                    "match": "all_of",
                    "rules": [
                        {"source": "workflow_fact", "matcher": "bool_true", "path": "summary_read_completed"}
                    ],
                },
            ]
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=_default_telemetry(case.id),
        include_summary_artifact=False,
        include_logs_dir=False,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "direct restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "interrupted", "reason": "query interrupted logs"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-workflow-yaml-optional", audit_dir=audit_dir)
    workflow = judgment["packet"]["workflow"]

    assert workflow["status"] == "pass"
    assert workflow["raw_max_score"] == 25
    assert workflow["raw_score"] == 25
    assert workflow["score"] == WORKFLOW_BUCKET_MAX


def test_judge_workflow_yaml_policy_partial_when_required_checks_miss_after_progress(tmp_path):
    case = _case_with_judge_workflow(
        load_case("component-restart-during-incident-window"),
        judge_workflow={
            "checks": [
                {
                    "id": "architecture_context_loaded",
                    "weight": 5,
                    "required": True,
                    "description": "kb step a",
                    "match": "all_of",
                    "rules": [
                        {"source": "workflow_fact", "matcher": "bool_true", "path": "architecture_context_loaded"}
                    ],
                },
                {
                    "id": "diagnosis_run_initialized",
                    "weight": 5,
                    "required": True,
                    "description": "init",
                    "match": "all_of",
                    "rules": [
                        {"source": "workflow_fact", "matcher": "bool_true", "path": "diagnosis_run_initialized"}
                    ],
                },
            ]
        },
    )
    telemetry = [
        f"issues/{case.id}/issue_meta.json",
        "Initialize diagnosis run directory",
    ]
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=telemetry,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "warn", "reason": "corroborating logs"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-workflow-yaml-partial", audit_dir=audit_dir)

    assert judgment["packet"]["workflow"]["status"] == "partial"
    assert judgment["packet"]["execution_contract"]["status"] == "fail"
    assert judgment["primary_failure_code"] == "execution_contract_failed"


def test_evidence_collection_pass_without_summary_read(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=_default_telemetry(case.id),
        include_summary_artifact=False,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "error", "reason": "corroborating logs"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-no-summary", audit_dir=audit_dir)
    assert judgment["packet"]["workflow"]["status"] == "pass"
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"


def test_evidence_collection_does_not_treat_log_stats_shell_as_logs_channel(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=False,
        log_stats_payload={"summary": {}, "timeline": {}, "errors": []},
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "error", "reason": "corroborating logs"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-empty-log-stats", audit_dir=audit_dir)
    assert judgment["facts"]["evidence"]["logs_channel_observed"] is True
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["primary_failure_code"] == "critical_evidence_partial"


def test_evidence_collection_pass_when_metrics_cover_query_or_metric_channel(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        queries_payload={"slow_queries": [], "topsql": [], "decoded_plan": {}},
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10766841556523235152",
                "time_range": {"start": "1775039760", "end": "1775043360"},
            },
            "evidence": {
                "metrics": [
                    {
                        "name": "Process Restart Count",
                        "metric_group_id": "process_restarts",
                        "instance": "db-tidb-60",
                        "peak": 1,
                        "unit": "count",
                    }
                ],
            },
        },
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "warn", "reason": "corroborating logs"}],
            }
        ),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-metric-channel",
        audit_dir=audit_dir,
    )

    assert judgment["facts"]["evidence"]["query_or_metric_channel_observed"] is True
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"


def test_evidence_collection_marks_events_channel_observed_from_events_sidecar(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    _evidence_file(audit_dir, "events.json").write_text(json.dumps({"events": []}))

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-events-sidecar",
        audit_dir=audit_dir,
    )

    assert judgment["facts"]["evidence"]["events_channel_observed"] is True
    assert judgment["facts"]["execution_contract"]["channel_facts"]["events"] is True


def test_evidence_collection_marks_events_channel_observed_from_report(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
        diagnosis_report_payload={
            "metadata": {
                "cluster_id": "10766841556523235152",
                "time_range": {"start": "1775039760", "end": "1775043360"},
            },
            "evidence": {
                "metrics": [],
                "events": [],
            },
        },
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-events-report",
        audit_dir=audit_dir,
    )

    assert judgment["facts"]["evidence"]["events_channel_observed"] is True
    assert judgment["facts"]["execution_contract"]["channel_facts"]["events"] is True


def test_evidence_collection_deducts_score_when_events_query_attempted_without_events_output_file(tmp_path):
    case = load_case("component-restart-during-incident-window")
    telemetry = _default_telemetry(case.id) + ["Query Activity Hub events for incident window"]
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=telemetry,
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-events-attempted-no-output",
        audit_dir=audit_dir,
    )

    assert judgment["facts"]["evidence"]["events_collection_attempted"] is True
    assert judgment["facts"]["evidence"]["events_output_file_present"] is False
    assert judgment["facts"]["evidence"]["events_attempted_implies_artifact_present"] is False
    assert judgment["packet"]["execution_contract"]["status"] == "pass"
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["evidence_collection"]["score"] == 4
    assert judgment["judgment_verdict"] == "pass"
    assert judgment["primary_failure_code"] is None


def test_evidence_collection_accepts_empty_events_output_file_after_events_query_attempt(tmp_path):
    case = load_case("component-restart-during-incident-window")
    telemetry = _default_telemetry(case.id) + ["Query Activity Hub events for incident window"]
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=telemetry,
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    _evidence_file(audit_dir, "events.json").write_text(
        json.dumps({"events": [], "time_window": {"start": 1775039760, "end": 1775043360}})
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-events-attempted-empty-output",
        audit_dir=audit_dir,
    )

    assert judgment["facts"]["evidence"]["events_collection_attempted"] is True
    assert judgment["facts"]["evidence"]["events_output_file_present"] is True
    assert judgment["facts"]["evidence"]["events_attempted_implies_artifact_present"] is True
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["evidence_collection"]["score"] == 5


def test_evidence_collection_detects_live_fetch_activity_events_wording(tmp_path):
    case = load_case("component-restart-during-incident-window")
    telemetry = _default_telemetry(case.id) + ["Fetch activity events during incident window"]
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=telemetry,
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-events-fetch-wording",
        audit_dir=audit_dir,
    )

    assert judgment["facts"]["evidence"]["events_collection_attempted"] is True
    assert judgment["facts"]["evidence"]["events_output_file_present"] is False
    assert judgment["packet"]["evidence_collection"]["score"] == 4


def test_evidence_collection_ignores_activity_events_help_only(tmp_path):
    case = load_case("component-restart-during-incident-window")
    telemetry = _default_telemetry(case.id) + ["Show activity-events command help"]
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=telemetry,
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-events-help-only",
        audit_dir=audit_dir,
    )

    assert judgment["facts"]["evidence"]["events_collection_attempted"] is False
    assert judgment["packet"]["evidence_collection"]["score"] == 5


def test_evidence_collection_partial_when_one_channel_missing(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        queries_payload={"slow_queries": [], "topsql": [], "decoded_plan": {}},
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "warn", "reason": "corroborating logs"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-collection-partial", audit_dir=audit_dir)

    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["judgment_verdict"] == "pass"
    assert judgment["primary_failure_code"] is None


def test_evidence_collection_fail_when_core_bundle_missing(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "warn", "reason": "corroborating logs"}],
            }
        ),
    )
    _evidence_file(audit_dir, "logs.json").unlink()

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-collection-fail", audit_dir=audit_dir)
    assert judgment["packet"]["evidence_collection"]["status"] == "fail"
    assert judgment["packet"]["execution_contract"]["status"] == "fail"
    assert judgment["judgment_verdict"] == "invalid"
    assert judgment["primary_failure_code"] == "execution_contract_failed"
    assert judgment["invalid_subtype"] == "execution_contract"


def test_evidence_collection_fail_when_core_bundle_json_unreadable(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "warn", "reason": "corroborating logs"}],
            }
        ),
    )
    _evidence_file(audit_dir, "queries.json").write_text("{bad json")

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-collection-bad-json", audit_dir=audit_dir)

    assert judgment["packet"]["evidence_collection"]["status"] == "fail"
    assert judgment["packet"]["execution_contract"]["status"] == "fail"
    assert judgment["judgment_verdict"] == "invalid"
    assert judgment["primary_failure_code"] == "execution_contract_failed"


def test_evidence_collection_accepts_workspace_logs_json_without_log_stats_alias(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_log_stats=False,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    _evidence_file(audit_dir, "logs.json").write_text(
        json.dumps(
            {
                "log_counts": [
                    {
                        "component": "tidb",
                        "level": "ERROR",
                        "message": "[ERROR] x",
                        "count": 1,
                        "instances": ["tidb-0"],
                    }
                ]
            }
        )
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-legacy-logs-only",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["evidence_collection"]["status"] in {"pass", "partial"}


def test_evidence_collection_yaml_policy_passes_when_required_and_optional_checks_pass(tmp_path):
    case = _case_with_judge_evidence_collection(
        load_case("component-restart-during-incident-window"),
        evidence_collection={
            "weight": 10,
            "checks": [
                {
                    "id": "required_artifacts_present",
                    "description": "bundle exists",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {"source": "artifact_fact", "matcher": "subset_observed", "paths": ["core_evidence_bundle_present"]}
                    ],
                    "weight": 4,
                },
                {
                    "id": "preferred_channels_minimum_observed",
                    "description": "minimum channels observed",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "channel_fact",
                            "matcher": "count_gte",
                            "paths": ["logs", "query_or_metric"],
                            "threshold": 1,
                        }
                    ],
                    "weight": 3,
                },
                {
                    "id": "preferred_channels_all_observed",
                    "description": "all preferred channels observed",
                    "required": False,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "channel_fact",
                            "matcher": "subset_observed",
                            "paths": ["logs", "query_or_metric"],
                        }
                    ],
                    "weight": 3,
                },
            ]
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-collection-yaml-pass", audit_dir=audit_dir)

    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["evidence_collection"]["raw_score"] == 12
    assert judgment["packet"]["evidence_collection"]["raw_max_score"] == 12
    assert judgment["packet"]["evidence_collection"]["score"] == EVIDENCE_COLLECTION_BUCKET_MAX


def test_evidence_collection_yaml_policy_partial_when_optional_check_misses(tmp_path):
    case = _case_with_judge_evidence_collection(
        load_case("component-restart-during-incident-window"),
        evidence_collection={
            "weight": 10,
            "checks": [
                {
                    "id": "required_artifacts_present",
                    "description": "bundle exists",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {"source": "artifact_fact", "matcher": "subset_observed", "paths": ["core_evidence_bundle_present"]}
                    ],
                    "weight": 4,
                },
                {
                    "id": "preferred_channels_minimum_observed",
                    "description": "minimum channels observed",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "channel_fact",
                            "matcher": "count_gte",
                            "paths": ["logs", "query_or_metric"],
                            "threshold": 1,
                        }
                    ],
                    "weight": 3,
                },
                {
                    "id": "preferred_channels_all_observed",
                    "description": "all preferred channels observed",
                    "required": False,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "channel_fact",
                            "matcher": "subset_observed",
                            "paths": ["logs", "query_or_metric"],
                        }
                    ],
                    "weight": 3,
                },
            ]
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        queries_payload={"slow_queries": [], "topsql": [], "decoded_plan": {}},
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-collection-yaml-partial", audit_dir=audit_dir)

    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["evidence_collection"]["raw_score"] == 9
    assert judgment["primary_failure_code"] is None


def test_evidence_collection_yaml_policy_does_not_inherit_execution_contract_preferred_channel_override(tmp_path):
    base_case = _case_with_judge_evidence_collection(
        load_case("component-restart-during-incident-window"),
        evidence_collection={
            "weight": 10,
            "checks": [
                {
                    "id": "required_artifacts_present",
                    "description": "bundle exists",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {"source": "artifact_fact", "matcher": "subset_observed", "paths": ["core_evidence_bundle_present"]}
                    ],
                    "weight": 4,
                },
                {
                    "id": "preferred_channels_minimum_observed",
                    "description": "minimum channels observed",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "channel_fact",
                            "matcher": "count_gte",
                            "paths": ["logs", "query_or_metric"],
                            "threshold": 1,
                        }
                    ],
                    "weight": 3,
                },
                {
                    "id": "preferred_channels_all_observed",
                    "description": "all preferred channels observed",
                    "required": False,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "channel_fact",
                            "matcher": "subset_observed",
                            "paths": ["logs", "query_or_metric"],
                        }
                    ],
                    "weight": 3,
                },
            ]
        },
    )
    execution_contract = deepcopy(base_case.data["judge"]["execution_contract"])
    execution_contract["checks"][2]["rules"][0]["paths"] = ["logs"]
    execution_contract["checks"][3]["rules"][0]["paths"] = ["logs"]
    case = _case_with_execution_contract(
        base_case,
        execution_contract=execution_contract,
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        queries_payload={"slow_queries": [], "topsql": [], "decoded_plan": {}},
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-collection-yaml-override", audit_dir=audit_dir)

    assert judgment["packet"]["execution_contract"]["status"] == "pass"
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["evidence_collection"]["raw_score"] == 9


def test_evidence_collection_yaml_policy_supports_evidence_fact_rules(tmp_path):
    case = _case_with_judge_evidence_collection(
        load_case("component-restart-during-incident-window"),
        evidence_collection={
            "weight": 10,
            "checks": [
                {
                    "id": "logs_dir_present",
                    "description": "logs directory is present",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {"source": "evidence_fact", "matcher": "bool_true", "path": "logs_dir_present"}
                    ],
                    "weight": 10,
                },
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-collection-evidence-fact",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["evidence_collection"]["score"] == EVIDENCE_COLLECTION_BUCKET_MAX


def test_evidence_routing_pass_when_analysis_patch_missing(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(tmp_path, case.id, include_logs_dir=True, analysis_patch_text=None)

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-partial", audit_dir=audit_dir)
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["evidence_routing"]["status"] == "pass"


def test_evidence_routing_fail_on_invalid_patch_json(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text="{bad json",
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-invalid-json", audit_dir=audit_dir)
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["evidence_routing"]["status"] == "fail"
    assert judgment["primary_failure_code"] == "evidence_routing_failed"


def test_evidence_routing_fail_on_malformed_items(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": ["not-a-dict"],
                "supporting_evidence": [{"type": "log", "name": "panic", "reason": "corroborating signal"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-bad-shape", audit_dir=audit_dir)
    assert judgment["packet"]["evidence_routing"]["status"] == "fail"
    assert judgment["primary_failure_code"] == "evidence_routing_failed"


def test_evidence_routing_partial_without_supporting_evidence(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-no-support", audit_dir=audit_dir)
    assert judgment["packet"]["evidence_routing"]["status"] == "partial"
    assert judgment["primary_failure_code"] == "evidence_routing_partial"


def test_evidence_routing_fail_on_unknown_item_type(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "query", "digest": "abc", "reason": "unsupported generic query type"}],
                "supporting_evidence": [{"type": "log", "name": "panic", "reason": "corroborating signal"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-bad-type", audit_dir=audit_dir)
    assert judgment["packet"]["evidence_routing"]["status"] == "fail"
    assert judgment["primary_failure_code"] == "evidence_routing_failed"


def test_evidence_routing_pass_on_resolvable_query_refs(tmp_path):
    case = load_case("component-restart-during-incident-window")
    queries_payload = {
        "slow_queries": [{"digest": "abc", "source": "slow_query"}],
        "topsql": [{"digest": "def", "source": "topsql", "instance": "db-tidb-0:10080"}],
        "decoded_plan": {},
    }
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        queries_payload=queries_payload,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [
                    {
                        "type": "topsql",
                        "digest": "def",
                        "instance": "db-tidb-0:10080",
                        "reason": "TopSQL digest lines up with the hotspot instance.",
                    }
                ],
                "supporting_evidence": [
                    {"type": "slow_query", "digest": "abc", "reason": "slow query corroborates the SQL fingerprint."}
                ],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-pass", audit_dir=audit_dir)
    assert judgment["judgment_verdict"] == "pass"
    assert judgment["primary_failure_code"] is None
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["evidence_routing"]["status"] == "pass"


def test_evidence_routing_accepts_delta_analysis_patch_fields(tmp_path):
    case = load_case("component-restart-during-incident-window")
    queries_payload = {
        "slow_queries": [{"digest": "abc", "source": "slow_query"}],
        "topsql": [{"digest": "def", "source": "topsql", "instance": "db-tidb-0:10080"}],
        "decoded_plan": {},
    }
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        queries_payload=queries_payload,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "add_key_evidence": [
                    {
                        "type": "topsql",
                        "digest": "def",
                        "instance": "db-tidb-0:10080",
                        "reason": "Delta patch points at the routed TopSQL row.",
                    }
                ],
                "add_supporting_evidence": [
                    {"type": "slow_query", "digest": "abc", "reason": "slow query corroborates the digest."}
                ],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-delta-pass", audit_dir=audit_dir)

    assert judgment["packet"]["evidence_routing"]["status"] == "pass"
    assert judgment["facts"]["routing"]["analysis_has_key_evidence"] is True
    assert judgment["facts"]["routing"]["analysis_has_supporting_evidence"] is True
    assert judgment["primary_failure_code"] is None


def test_evidence_routing_fail_on_topsql_instance_mismatch(tmp_path):
    case = load_case("component-restart-during-incident-window")
    queries_payload = {
        "slow_queries": [],
        "topsql": [{"digest": "def", "source": "topsql", "instance": "db-tidb-0:10080"}],
        "decoded_plan": {},
    }
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        queries_payload=queries_payload,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [
                    {
                        "type": "topsql",
                        "digest": "def",
                        "instance": "db-tidb-1:10080",
                        "reason": "instance should match the routed TopSQL row.",
                    }
                ],
                "supporting_evidence": [{"type": "log", "name": "panic", "reason": "corroborating signal"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-instance-mismatch", audit_dir=audit_dir)
    assert judgment["packet"]["evidence_routing"]["status"] == "fail"
    assert judgment["primary_failure_code"] == "evidence_routing_failed"


def test_evidence_routing_pass_on_multi_instance_metric_ref(tmp_path):
    case = load_case("component-restart-during-incident-window")
    report_payload = {
        "metadata": {
            "cluster_id": "10766841556523235152",
            "time_range": {"start": "1775039760", "end": "1775043360"},
        },
        "evidence": {
            "metrics": [
                {"metric_group_id": "tidb_cpu", "instance": "db-tidb-0:10080"},
                {"metric_group_id": "tidb_cpu", "instance": "db-tidb-1:10080"},
            ],
        },
    }
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload=report_payload,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [
                    {
                        "type": "metric",
                        "name": "TiDB CPU",
                        "metric_ref": {
                            "metric_group_id": "tidb_cpu",
                            "instances": ["db-tidb-0:10080", "db-tidb-1:10080"],
                        },
                        "reason": "both instances are explicitly referenced from the same metric group.",
                    }
                ],
                "supporting_evidence": [{"type": "log", "name": "panic", "reason": "corroborating signal"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-metric-pass", audit_dir=audit_dir)
    assert judgment["packet"]["evidence_routing"]["status"] == "pass"
    assert judgment["primary_failure_code"] is None


def test_realistic_audit_sample_mixed_artifacts_pass_collection_and_routing(tmp_path):
    case = load_case("component-restart-during-incident-window")
    sample = _realistic_restart_audit_sample()
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload=sample["diagnosis_report_payload"],
        queries_payload=sample["queries_payload"],
        log_stats_payload=sample["log_stats_payload"],
        collection_errors_payload=sample["collection_errors_payload"],
        analysis_patch_text=sample["analysis_patch_text"],
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-realistic-audit", audit_dir=audit_dir)

    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["evidence_routing"]["status"] == "pass"
    assert judgment["facts"]["routing"]["analysis_metric_refs_resolvable"] is True
    assert judgment["facts"]["routing"]["analysis_query_refs_resolvable"] is True
    assert judgment["facts"]["evidence"]["topsql_collection_error_recorded"] is True


def test_routing_facts_include_diagnosis_classification_contract(tmp_path):
    case = _case_with_evidence_routing(
        load_case("component-restart-during-incident-window"),
        evidence_routing={
            "weight": 5,
            "checks": [
                {
                    "id": "classification_contract_valid",
                    "description": "diagnosis analyst wrote stable classification contract",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "classification_contract_valid",
                        },
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "primary_fault_class_resource_control_or_capacity",
                        },
                    ],
                    "weight": 5,
                }
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
        diagnosis_analyst_payload={
            "agent": "diagnosis-analyst",
            "status": "ok",
            "root_cause": "capacity reduction increased per-node load",
            "confidence": "medium",
            "key_evidence": [],
            "rejected_causes": [],
            "needs_more_evidence": False,
            "missing_evidence": [],
            "quality_checks": [],
            "signal_tags": [
                {
                    "id": "capacity_comparison_context_present",
                    "source": "critical_evidence_hints",
                    "component": "cluster",
                    "metric_or_signal": "capacity_comparison_context_present",
                    "direction": "changed",
                    "baseline": "before_after_event",
                    "severity": "critical",
                    "confidence": "high",
                    "evidence_ref": "evidences/agent-runs/evidence-agent.json#critical_evidence_hints",
                }
            ],
            "fault_classification": {
                "primary_class": "resource_control_or_capacity",
                "candidates": [
                    {
                        "class": "resource_control_or_capacity",
                        "confidence": 0.9,
                        "role": "primary",
                        "evidence_for": ["capacity_comparison_context_present"],
                        "evidence_against": [],
                    },
                    {
                        "class": "sql_plan_or_query",
                        "confidence": 0.35,
                        "role": "secondary",
                        "evidence_for": ["slow query evidence exists"],
                        "evidence_against": ["capacity comparison explains the differential better"],
                    },
                ],
                "rejected_classes": [
                    {
                        "class": "sql_plan_or_query_as_primary",
                        "confidence": 0.2,
                        "evidence_for": ["slow query evidence exists"],
                        "evidence_against": ["node count reduction is the action-driving cause"],
                    }
                ],
                "classification_confidence": "high",
                "missing_evidence_for_higher_confidence": [],
            },
            "selected_diagnosis_skills": ["tidb-knowledge-base", "tidb-clinic-diagnosis"],
            "critical_evidence_hints": [],
            "final_response": "capacity reduction increased per-node load",
        },
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-classification", audit_dir=audit_dir)

    assert judgment["packet"]["evidence_routing"]["status"] == "pass"
    assert judgment["facts"]["routing"]["classification_contract_valid"] is True
    assert judgment["facts"]["routing"]["primary_fault_class_resource_control_or_capacity"] is True


def test_evidence_routing_yaml_policy_passes_when_all_required_and_optional_checks_pass(tmp_path):
    case = _case_with_evidence_routing(
        load_case("component-restart-during-incident-window"),
        evidence_routing={
            "weight": 10,
            "checks": [
                {
                    "id": "analysis_patch_present",
                    "description": "analysis patch exists",
                    "required": False,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "analysis_patch_present",
                        }
                    ],
                    "weight": 3,
                },
                {
                    "id": "analysis_patch_valid_json",
                    "description": "analysis patch parses",
                    "required": True,
                    "match": "any_of",
                    "rules": [
                        {
                            "source": "routing_fact",
                            "matcher": "bool_false",
                            "path": "analysis_patch_present",
                        },
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "analysis_patch_valid_json",
                        },
                    ],
                    "weight": 2,
                },
                {
                    "id": "analysis_key_evidence_present",
                    "description": "key evidence exists",
                    "required": True,
                    "match": "any_of",
                    "rules": [
                        {
                            "source": "routing_fact",
                            "matcher": "bool_false",
                            "path": "analysis_patch_present",
                        },
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "analysis_has_key_evidence",
                        },
                    ],
                    "weight": 2,
                },
                {
                    "id": "analysis_refs_resolve",
                    "description": "at least one ref family resolves",
                    "required": False,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "routing_fact",
                            "matcher": "any_true",
                            "paths": [
                                "analysis_metric_refs_resolvable",
                                "analysis_query_refs_resolvable",
                            ],
                        }
                    ],
                    "weight": 0,
                },
            ],
        },
    )
    queries_payload = {
        "slow_queries": [],
        "topsql": [{"digest": "def", "source": "topsql", "instance": "db-tidb-0:10080"}],
        "decoded_plan": {},
    }
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        queries_payload=queries_payload,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [
                    {
                        "type": "topsql",
                        "digest": "def",
                        "instance": "db-tidb-0:10080",
                        "reason": "TopSQL digest lines up with the hotspot instance.",
                    }
                ],
                "supporting_evidence": [{"type": "log", "name": "panic", "reason": "corroborating signal"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-yaml-pass", audit_dir=audit_dir)

    assert judgment["packet"]["evidence_routing"]["status"] == "pass"
    assert judgment["primary_failure_code"] is None


def test_evidence_routing_yaml_policy_partial_when_optional_check_misses(tmp_path):
    case = _case_with_evidence_routing(
        load_case("component-restart-during-incident-window"),
        evidence_routing={
            "weight": 10,
            "checks": [
                {
                    "id": "analysis_patch_present",
                    "description": "analysis patch exists",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "analysis_patch_present",
                        }
                    ],
                    "weight": 3,
                },
                {
                    "id": "analysis_patch_valid_json",
                    "description": "analysis patch parses",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "analysis_patch_valid_json",
                        }
                    ],
                    "weight": 2,
                },
                {
                    "id": "analysis_has_supporting_evidence",
                    "description": "supporting evidence exists",
                    "required": False,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "analysis_has_supporting_evidence",
                        }
                    ],
                    "weight": 0,
                },
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-yaml-partial", audit_dir=audit_dir)

    assert judgment["packet"]["evidence_routing"]["status"] == "partial"
    assert judgment["primary_failure_code"] == "evidence_routing_partial"


def test_evidence_routing_yaml_policy_fail_when_required_check_misses(tmp_path):
    case = _case_with_evidence_routing(
        load_case("component-restart-during-incident-window"),
        evidence_routing={
            "weight": 10,
            "checks": [
                {
                    "id": "analysis_patch_present",
                    "description": "analysis patch exists",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "analysis_patch_present",
                        }
                    ],
                    "weight": 3,
                },
                {
                    "id": "analysis_patch_valid_json",
                    "description": "analysis patch parses",
                    "required": True,
                    "match": "all_of",
                    "rules": [
                        {
                            "source": "routing_fact",
                            "matcher": "bool_true",
                            "path": "analysis_patch_valid_json",
                        }
                    ],
                    "weight": 2,
                },
            ],
        },
    )
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text="{bad json",
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-routing-yaml-fail", audit_dir=audit_dir)

    assert judgment["packet"]["evidence_routing"]["status"] == "fail"
    assert judgment["primary_failure_code"] == "evidence_routing_failed"


def test_realistic_audit_sample_collection_errors_do_not_change_collection_score(tmp_path):
    case = load_case("component-restart-during-incident-window")
    sample = _realistic_restart_audit_sample()
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        diagnosis_report_payload=sample["diagnosis_report_payload"],
        queries_payload=sample["queries_payload"],
        log_stats_payload=sample["log_stats_payload"],
        collection_errors_payload=sample["collection_errors_payload"],
        analysis_patch_text=sample["analysis_patch_text"],
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-realistic-score", audit_dir=audit_dir)
    collection = judgment["packet"]["evidence_collection"]

    assert judgment["facts"]["evidence"]["topsql_collection_error_recorded"] is True
    assert collection["status"] == "pass"
    assert collection["raw_score"] == collection["raw_max_score"] == 12
    assert collection["score"] == EVIDENCE_COLLECTION_BUCKET_MAX


def test_critical_evidence_pass_with_structured_and_log_rules(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-critical-pass", audit_dir=audit_dir)
    critical = judgment["packet"]["critical_evidence"]

    assert critical["status"] == "pass"
    assert critical["score"] == CRITICAL_EVIDENCE_BUCKET_MAX
    assert [item["id"] for item in critical["checks"] if item["passed"]] == [
        "restart_signal_present",
        "panic_signal_present",
        "query_interruption_signal_present",
    ]
    assert judgment["audit"]["critical_evidence_verdict"] == "pass"
    assert judgment["packet"]["scoring"]["hard_gate_passed"] is True
    assert judgment["score"] >= PASSING_HARD_GATE_SCORE_FLOOR


def test_critical_evidence_partial_when_only_restart_signal_exists(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=False,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-critical-partial", audit_dir=audit_dir)
    critical = judgment["packet"]["critical_evidence"]

    assert critical["status"] == "partial"
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["primary_failure_code"] == "critical_evidence_partial"
    assert judgment["judgment_verdict"] == "fail"
    assert critical["checks"][0]["passed"] is True
    assert critical["checks"][1]["passed"] is False
    assert critical["checks"][2]["passed"] is False


def test_critical_evidence_partial_is_decisive_when_collection_passes(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        logs_files={
            "loki-targeted-panic.log": "panic: memory.(*Tracker).Consume",
        },
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-critical-partial-decisive",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["critical_evidence"]["status"] == "partial"
    assert judgment["primary_failure_code"] == "critical_evidence_partial"
    assert judgment["judgment_verdict"] == "fail"
    assert judgment["packet"]["scoring"]["hard_gate_passed"] is False
    assert judgment["score"] <= FAILING_HARD_GATE_SCORE_CAP


def test_critical_evidence_fail_when_direct_signals_missing(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=False,
        component_health_payload={
            "component_health": {
                "query_scope": "10766841556523235152",
                "component_count": [],
                "restarts": [],
                "worker_memory": [],
                "links": {},
            }
        },
        metric_files={
            "process_restarts_changes_1h_instant.json": {
                "status": "success",
                "data": {"result": [{"metric": {"instance": "db-tidb-0"}, "value": [1775043360, "0"]}]},
            },
            "kube_restarts_increase_1h_instant.json": {
                "status": "success",
                "data": {"result": []},
            },
        },
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-critical-fail", audit_dir=audit_dir)
    critical = judgment["packet"]["critical_evidence"]

    assert critical["status"] == "fail"
    assert critical["score"] == 0
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["primary_failure_code"] == "critical_evidence_missing"
    assert judgment["judgment_verdict"] == "fail"


def test_critical_evidence_fail_is_decisive_when_collection_passes(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        logs_files={
            "loki-targeted-panic.log": "background info only",
        },
        component_health_payload={
            "component_health": {
                "query_scope": "10766841556523235152",
                "component_count": [],
                "restarts": [],
                "worker_memory": [],
                "links": {},
            }
        },
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-critical-fail-decisive",
        audit_dir=audit_dir,
    )

    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["packet"]["critical_evidence"]["status"] == "fail"
    assert judgment["primary_failure_code"] == "critical_evidence_missing"
    assert judgment["judgment_verdict"] == "fail"
    assert judgment["packet"]["scoring"]["hard_gate_passed"] is False
    assert judgment["score"] <= FAILING_HARD_GATE_SCORE_CAP


def test_critical_evidence_restart_rule_accepts_restart_metric_change(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=False,
        component_health_payload={
            "component_health": {
                "query_scope": "10766841556523235152",
                "component_count": [],
                "restarts": [],
                "worker_memory": [],
                "links": {},
            }
        },
        metric_files={
            "process_restarts_changes_1h_instant.json": {
                "status": "success",
                "data": {
                    "result": [
                        {"metric": {"instance": "db-tidb-0"}, "value": [1775043360, "1"]},
                    ]
                },
            },
        },
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-critical-metric", audit_dir=audit_dir)
    restart_check = next(
        item
        for item in judgment["packet"]["critical_evidence"]["checks"]
        if item["id"] == "restart_signal_present"
    )

    assert restart_check["passed"] is True
    assert any(
        rule["matcher"] == "numeric_series_any_gt" and rule["matched"]
        for rule in restart_check["rules"]
    )


def test_critical_evidence_fail_without_raw_logs_even_if_log_stats_exist(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=False,
        log_stats_payload={"summary": {"tidb": {"error": 3}}},
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(
        case,
        output_text=PASSING_OUTPUT,
        workspace_id="ws-no-raw-logs",
        audit_dir=audit_dir,
    )

    assert judgment["facts"]["evidence"]["logs_channel_observed"] is True
    assert judgment["packet"]["evidence_collection"]["status"] == "pass"
    assert judgment["primary_failure_code"] == "critical_evidence_partial"


def test_workflow_partial_is_invalid_with_workflow_invalid_subtype(tmp_path):
    case = load_case("component-restart-during-incident-window")
    telemetry = [
        f"issues/{case.id}/issue_meta.json",
        "Initialize diagnosis run directory",
    ]
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        telemetry_lines=telemetry,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "warn", "reason": "corroborating logs"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-workflow-partial", audit_dir=audit_dir)
    assert judgment["packet"]["workflow"]["status"] == "partial"
    assert judgment["packet"]["execution_contract"]["status"] == "fail"
    assert judgment["judgment_verdict"] == "invalid"
    assert judgment["primary_failure_code"] == "execution_contract_failed"
    assert judgment["invalid_subtype"] == "execution_contract"


def test_judge_exposes_resolved_artifact_paths_in_facts(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        include_summary_artifact=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "warn", "reason": "corroborating logs"}],
            }
        ),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-artifact-facts", audit_dir=audit_dir)
    artifacts = judgment["facts"]["artifacts"]

    assert judgment["judgment_verdict"] == "pass"
    assert artifacts["source_scope"] == "audit_dir"
    assert artifacts["run_dir"] == "evidences"
    assert artifacts["resolution_mode"] == "workspace_evidence_root"
    assert artifacts["diagnosis_report_path"] == "evidences/workspace_evidence_report.json"
    assert artifacts["evidence_dir"] == "evidences"


def test_judge_exposes_semantic_shadow_when_packet_is_available(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        include_summary_artifact=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )
    (audit_dir / "semantic_packet_v1.json").write_text(_semantic_packet_fixture_text())

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-semantic-shadow", audit_dir=audit_dir)

    semantic_shadow = judgment["packet"]["diagnosis"]["semantic_shadow"]

    assert semantic_shadow["status"] == "ok"
    assert semantic_shadow["available"] is True
    assert semantic_shadow["packet_status"] == "ok"
    assert [item["check_id"] for item in semantic_shadow["checks"]] == [
        1,
        2,
        4,
    ]
    assert all(item["status"] == "pass" for item in semantic_shadow["checks"])
    assert semantic_shadow["summary"]["shadow_score"] == 9
    assert judgment["audit"]["semantic_shadow"]["status"] == "ok"
    assert judgment["audit"]["semantic_shadow"]["available"] is True


def test_judge_reports_semantic_shadow_not_available_without_packet(tmp_path):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=_minimal_analysis_patch(),
    )

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id="ws-semantic-shadow-missing", audit_dir=audit_dir)

    semantic_shadow = judgment["packet"]["diagnosis"]["semantic_shadow"]

    assert semantic_shadow["status"] == "not_available"
    assert semantic_shadow["available"] is False
    assert semantic_shadow["reason"] == "semantic_packet_missing"
    assert semantic_shadow["checks"] == []
    assert semantic_shadow["summary"]["shadow_score"] == 0


def test_judge_prefers_broken_audit_scope_over_valid_workspace_scope(tmp_path, monkeypatch):
    case = load_case("component-restart-during-incident-window")
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "warn", "reason": "corroborating logs"}],
            }
        ),
    )
    _evidence_file(audit_dir, "workspace_evidence_report.json").unlink()

    workspace_root = tmp_path / "workspaces"
    monkeypatch.setattr("investibook.core.eval.eval_judge.WORKSPACES_DIR", workspace_root)
    workspace_id = "ws-audit-priority"
    workspace_dir = workspace_root / workspace_id
    workspace_case_tmp = tmp_path / "workspace_case_tmp"
    workspace_audit_dir = _setup_audit_run(
        workspace_case_tmp,
        case.id,
        include_logs_dir=True,
        analysis_patch_text=json.dumps(
            {
                "key_evidence": [{"type": "log", "name": "panic", "reason": "restart signal"}],
                "supporting_evidence": [{"type": "log", "name": "warn", "reason": "corroborating logs"}],
            }
        ),
    )
    shutil.copytree(workspace_audit_dir / ".opencode", workspace_dir / ".opencode", dirs_exist_ok=True)
    shutil.copytree(workspace_audit_dir / "evidences", workspace_dir / "evidences", dirs_exist_ok=True)

    judgment = judge_case_output(case, output_text=PASSING_OUTPUT, workspace_id=workspace_id, audit_dir=audit_dir)

    assert judgment["judgment_verdict"] == "invalid"
    assert judgment["invalid_subtype"] == "artifact_resolution"
    assert judgment["primary_failure_code"] == "artifact_resolution_failed"
    assert judgment["audit"]["artifact_resolution"]["selected_scope"] == "audit_dir"


def test_evidence_only_judge_passes_on_evidence_artifacts(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    case = replace(base_case, data={**base_case.data, "evaluation": {"mode": "evidence_only"}})
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        include_summary_artifact=True,
    )
    evidence_agent_path = _evidence_file(audit_dir, "agent-runs", "evidence-agent.json")
    evidence_agent_path.parent.mkdir(parents=True, exist_ok=True)
    evidence_agent_path.write_text(json.dumps({"agent": "evidence-agent", "status": "ok"}))

    judgment = judge_case_output(
        case,
        output_text="Evidence collection completed. Artifacts are ready.",
        workspace_id="ws-evidence-only-pass",
        audit_dir=audit_dir,
    )

    assert judgment["judgment_verdict"] == "pass"
    assert judgment["score"] == 100
    assert judgment["primary_failure_code"] is None
    assert judgment["packet"]["strategy"] == "evidence_only_bucketed_v1"
    assert judgment["packet"]["mode_contract"]["score"] == 10
    assert judgment["packet"]["evidence_collection"]["score"] == 30
    assert judgment["packet"]["critical_evidence"]["score"] == 60
    assert judgment["packet"]["scoring"]["raw_total_score"] == 100
    assert judgment["facts"]["evaluation_mode"] == "evidence_only"
    assert judgment["facts"]["evidence_only"]["required_failed"] == []


def test_evidence_only_judge_rejects_diagnosis_agent_artifact(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    case = replace(base_case, data={**base_case.data, "evaluation": {"mode": "evidence_only"}})
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        include_summary_artifact=True,
        diagnosis_analyst_payload={"status": "ok"},
    )
    evidence_agent_path = _evidence_file(audit_dir, "agent-runs", "evidence-agent.json")
    evidence_agent_path.parent.mkdir(parents=True, exist_ok=True)
    evidence_agent_path.write_text(json.dumps({"agent": "evidence-agent", "status": "ok"}))

    judgment = judge_case_output(
        case,
        output_text="Evidence collection completed. Artifacts are ready.",
        workspace_id="ws-evidence-only-boundary",
        audit_dir=audit_dir,
    )

    assert judgment["judgment_verdict"] == "invalid"
    assert judgment["invalid_subtype"] == "execution_contract"
    assert judgment["primary_failure_code"] == "evidence_only_stage_boundary_failed"
    failed = judgment["facts"]["evidence_only"]["required_failed"]
    assert failed == ["diagnosis_agent_not_invoked"]


def test_evidence_only_judge_fails_when_critical_evidence_missing(tmp_path):
    base_case = load_case("component-restart-during-incident-window")
    case = replace(base_case, data={**base_case.data, "evaluation": {"mode": "evidence_only"}})
    audit_dir = _setup_audit_run(
        tmp_path,
        case.id,
        include_logs_dir=True,
        logs_files={"loki-unrelated.log": "no matching restart or panic evidence"},
        include_summary_artifact=True,
        component_health_payload={"component_health": {"restarts": []}},
    )
    evidence_agent_path = _evidence_file(audit_dir, "agent-runs", "evidence-agent.json")
    evidence_agent_path.parent.mkdir(parents=True, exist_ok=True)
    evidence_agent_path.write_text(json.dumps({"agent": "evidence-agent", "status": "ok"}))

    judgment = judge_case_output(
        case,
        output_text="Evidence collection completed. Artifacts are ready.",
        workspace_id="ws-evidence-only-critical-missing",
        audit_dir=audit_dir,
    )

    assert judgment["judgment_verdict"] == "fail"
    assert judgment["primary_failure_code"] == "critical_evidence_missing"
    assert judgment["score"] <= FAILING_HARD_GATE_SCORE_CAP
    assert judgment["packet"]["mode_contract"]["score"] == 10
    assert judgment["packet"]["evidence_collection"]["score"] == 30
    assert judgment["packet"]["critical_evidence"]["score"] == 0