feat(evolution): diagnose.py aggregation + case packs + run_diagnosis (#8)

tests/unit/test_diagnose.py

@@ -12,11 +12,17 @@ from __future__ import annotations
 
 import json
 from typing import Any
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
 from core.evolution.diagnose import (
+    _percentile,
     _trigrams,
+    aggregate_d2,
+    aggregate_d3,
+    aggregate_d4,
+    aggregate_d5,
     aggregate_soft,
     attribute_error,
     calc_budget_usage,
@@ -28,11 +34,20 @@ from core.evolution.diagnose import (
     calc_tool_usage,
     extract_json_from_response,
     extract_rule_metrics,
+    merge_system_packs,
+    merge_tool_packs,
     question_soft_score,
+    run_diagnosis,
 )
 from core.evolution.types import (
+    CaseSample,
+    DiagnosePrompts,
+    DiagnosisResult,
     QuestionMetrics,
+    SkillStepAdherence,
     SpanMetrics,
+    SystemCasePack,
+    ToolCasePack,
 )
 
 # =========================================================================
@@ -506,3 +521,254 @@ class TestAttributeError:
         qm = _make_qm(evidence_sufficient=True)
         result = attribute_error(qm)
         assert result.reasoning_failure_type is None
+
+
+# =========================================================================
+# E. D2-D5 聚合测试
+# =========================================================================
+
+
+class TestPercentile:
+    """_percentile 辅助函数测试。"""
+
+    def test_empty_returns_zero(self) -> None:
+        """空列表返回 0.0。"""
+        assert _percentile([], 0.5) == 0.0
+
+    def test_single_element(self) -> None:
+        """单元素返回该元素。"""
+        assert _percentile([42.0], 0.5) == 42.0
+
+    def test_median_two_elements(self) -> None:
+        """两元素中位数。"""
+        assert _percentile([1.0, 3.0], 0.5) == 2.0
+
+    def test_quartiles(self) -> None:
+        """四分位线性插值。"""
+        values = [1.0, 2.0, 3.0, 4.0, 5.0]
+        assert _percentile(values, 0.0) == 1.0
+        assert _percentile(values, 1.0) == 5.0
+        p25 = _percentile(values, 0.25)
+        assert abs(p25 - 2.0) < 1e-9
+
+
+class TestAggregation:
+    """D2-D5 聚合函数测试。"""
+
+    def test_d2_empty(self) -> None:
+        """空输入返回空字典。"""
+        assert aggregate_d2([]) == {}
+
+    def test_d2_groups_by_tool(self) -> None:
+        """按工具名分组聚合。"""
+        qm = _make_qm(
+            span_metrics=[
+                _make_span(
+                    step=0,
+                    tool_name="view_node",
+                    extraction_completeness=0.8,
+                    hallucination_rate=0.2,
+                ),
+                _make_span(
+                    step=1,
+                    tool_name="view_node",
+                    extraction_completeness=0.6,
+                    hallucination_rate=0.1,
+                ),
+                _make_span(
+                    step=2,
+                    tool_name="search_similar",
+                    extraction_completeness=0.9,
+                    hallucination_rate=0.0,
+                ),
+            ]
+        )
+        result = aggregate_d2([qm])
+        assert "view_node" in result
+        assert "search_similar" in result
+        assert result["view_node"]["n_calls"] == 2
+        assert result["search_similar"]["n_calls"] == 1
+        assert abs(result["view_node"]["avg_completeness"] - 0.7) < 1e-9
+
+    def test_d3_empty(self) -> None:
+        """空输入返回空字典。"""
+        assert aggregate_d3([]) == {}
+
+    def test_d3_correct_vs_incorrect(self) -> None:
+        """按正误拆分。"""
+        qm_correct = _make_qm(correct=True, task_type="T1", budget_usage=0.5)
+        qm_wrong = _make_qm(correct=False, task_type="T1", budget_usage=0.8)
+        result = aggregate_d3([qm_correct, qm_wrong])
+        assert "T1" in result
+        assert result["T1"]["correct"]["n_questions"] == 1
+        assert result["T1"]["incorrect"]["n_questions"] == 1
+        # avg_steps 存储 budget_usage 均值
+        assert result["T1"]["correct"]["avg_steps"] == 0.5
+        assert result["T1"]["incorrect"]["avg_steps"] == 0.8
+
+    def test_d4_empty(self) -> None:
+        """空输入返回空字典。"""
+        assert aggregate_d4([]) == {}
+
+    def test_d4_adherence_rate(self) -> None:
+        """技能遵循率计算。"""
+        qm = _make_qm(
+            task_type="T1",
+            correct=True,
+            skill_adherence=[
+                SkillStepAdherence(step_label="S1", adhered=True, description=""),
+                SkillStepAdherence(step_label="S1", adhered=False, description=""),
+            ],
+        )
+        result = aggregate_d4([qm])
+        assert "T1" in result
+        assert result["T1"]["overall_adherence"] == 0.5
+
+    def test_d5_empty_returns_zero_structure(self) -> None:
+        """空输入返回完整零结构。"""
+        result = aggregate_d5([])
+        assert "early_submit_rate" in result
+        assert result["early_submit_rate"] == 0.0
+        assert "format_compliance_rate" in result
+        assert "budget_usage_median" in result
+        assert "confirmation_bias_rate" in result
+        assert "per_type_bias" in result
+        assert result["per_type_bias"] == {}
+
+    def test_d5_with_data(self) -> None:
+        """有数据时正确计算。"""
+        qm1 = _make_qm(
+            correct=True,
+            budget_usage=0.5,
+            format_compliance=1.0,
+            confidence_calibration="calibrated",
+            confirmation_bias=False,
+        )
+        qm2 = _make_qm(
+            correct=False,
+            budget_usage=0.2,
+            format_compliance=0.8,
+            confidence_calibration="high_conf_wrong",
+            confirmation_bias=True,
+        )
+        result = aggregate_d5([qm1, qm2])
+        assert result["format_compliance_rate"] == 0.9
+        assert result["high_conf_wrong_rate"] == 0.5
+        assert result["early_submit_rate"] == 1.0  # 1 wrong with budget<0.3
+
+
+# =========================================================================
+# F. Merge 函数测试
+# =========================================================================
+
+
+class TestMerge:
+    """merge_system_packs / merge_tool_packs 测试。"""
+
+    def test_merge_system_packs_none_on_empty(self) -> None:
+        """空列表返回 None。"""
+        assert merge_system_packs([]) is None
+
+    def test_merge_system_packs_wraps_stats(self) -> None:
+        """stats 包裹为 per_step 列表。"""
+        pack = SystemCasePack(stats={"a": 1}, failure_cases=[], success_cases=[])
+        merged = merge_system_packs([pack, pack])
+        assert merged is not None
+        assert "per_step" in merged.stats
+        assert len(merged.stats["per_step"]) == 2
+
+    def test_merge_system_packs_concats_cases(self) -> None:
+        """failure/success cases 拼接。"""
+        case = CaseSample(
+            question_id="q1",
+            video_id="v1",
+            task_type="T1",
+            question="q",
+            options=[],
+            answer="a",
+            prediction="b",
+            correct=False,
+            error_type="mixed",
+            selection_reason="test",
+            metrics={},
+            trace=[],
+        )
+        p1 = SystemCasePack(stats={}, failure_cases=[case], success_cases=[])
+        p2 = SystemCasePack(stats={}, failure_cases=[case], success_cases=[case])
+        merged = merge_system_packs([p1, p2])
+        assert merged is not None
+        assert len(merged.failure_cases) == 2
+        assert len(merged.success_cases) == 1
+
+    def test_merge_tool_packs_empty(self) -> None:
+        """空列表返回空字典。"""
+        assert merge_tool_packs([]) == {}
+
+    def test_merge_tool_packs_groups_by_name(self) -> None:
+        """同名工具合并。"""
+        p1 = ToolCasePack(
+            tool_name="view_node",
+            target_files=["f1.md"],
+            stats={"x": 1},
+            failure_spans=[{"a": 1}],
+            success_spans=[],
+        )
+        p2 = ToolCasePack(
+            tool_name="view_node",
+            target_files=["f1.md"],
+            stats={"x": 2},
+            failure_spans=[{"b": 2}],
+            success_spans=[{"c": 3}],
+        )
+        merged = merge_tool_packs([p1, p2])
+        assert "view_node" in merged
+        vn = merged["view_node"]
+        assert len(vn.failure_spans) == 2
+        assert len(vn.success_spans) == 1
+        assert "per_step" in vn.stats
+
+
+# =========================================================================
+# G. run_diagnosis 入口测试
+# =========================================================================
+
+
+class TestRunDiagnosis:
+    """run_diagnosis 入口测试。"""
+
+    def test_empty_predictions_returns_empty_result(self) -> None:
+        """无预测时返回空 DiagnosisResult。"""
+        import asyncio
+
+        mock_log = AsyncMock()
+        mock_log.get_predictions.return_value = []
+        mock_log.get_traces.return_value = []
+        mock_llm = AsyncMock()
+        mock_store = MagicMock()
+        mock_store.list_skill_files.return_value = []
+        prompts = DiagnosePrompts(
+            defect_vs_lapse="",
+            reasoning_sub="",
+            span_eval_system="",
+            span_eval_user="",
+            missed_nodes="",
+            skill_adherence="",
+            confirmation_bias="",
+            evidence_sufficiency="",
+        )
+        result = asyncio.run(
+            run_diagnosis(
+                "run1",
+                [],
+                {},
+                mock_llm,
+                mock_log,
+                mock_store,
+                prompts,
+                concurrency=1,
+            )
+        )
+        assert isinstance(result, DiagnosisResult)
+        assert result.run_id == "run1"
+        assert result.error_attributions == []
+        assert result.degraded_count == 0