from investibook.core.eval.semantic_adapter_models import ( SemanticJudgeCheckSpec, SemanticContextV1, SemanticPacketV1, adapt_diagnosis_semantic_shadow_checks, ) def _semantic_context_payload() -> dict: return { "version": "semantic_context_v1", "case": { "case_id": "component-restart-during-incident-window", "ground_truth_root_cause": "TiDB v8.5.2 known bug GitHub", "expected_cluster_id": "10766841556523235152", "allowed_window_ids": ["started_at_minus_1h"], "semantic_checks": [ { "check_id": 1, "title": "主路径判断", "description": "判断回答是否将本次事件的主路径识别为 TiDB v8.5.2 已知 bug 导致的 single-instance restart chain，而不是 generic restart、missing index 或其它泛化解释。", "max_score": 3, "required": False, }, { "check_id": 2, "title": "因果链完整性", "description": "判断回答是否说明 KILL QUERY / query interruption 是触发因素而不是主根因，并进一步解释 rollback 或 idle transaction cleanup 到 panic/restart 的链路。", "max_score": 3, "required": False, }, { "check_id": 3, "title": "范围对齐", "description": "判断回答是否识别影响范围主要集中在单个 TiDB 实例，而不是 cluster-wide degradation。", "max_score": 3, "required": False, }, { "check_id": 4, "title": "修复建议对齐", "description": "判断回答是否给出了与 ground truth 一致的 remediation，例如升级到包含修复的版本，而不是把建索引或泛化调参当成主修复方案。", "max_score": 3, "required": False, }, ], }, "deterministic": { "verdicts": { "contract": "pass", "workflow": "pass", "execution_contract": "pass", "evidence_collection": "pass", "critical_evidence": "pass", "evidence_routing": "partial", }, "output_facts": { "diagnosis_single_instance_panic_restart_asserted": True, "diagnosis_memory_tracker_panic_asserted": True, "diagnosis_kill_query_chain_asserted": True, "missing_index_primary_root_cause_asserted": False, "claims_panic_not_confirmable": False, "mentions_idle_transaction": True, "mentions_rollback_cleanup": True, "mentions_v8_5_2": True, "mentions_v8_5_3_or_fixed_release": True, "mentions_without_fix": True, }, "workflow_facts": {}, "artifact_facts": {}, "channel_facts": {}, }, "grounding": { "critical_evidence_checks": [], "routing_checks": [], "matched_critical_evidence_ids": [ "restart_signal_present", "panic_signal_present", "query_interruption_signal_present", ], "routed_items": [ { "item_id": "ke-1", "kind": "log", "summary": "panic log shows memory tracker panic on a single TiDB instance during the incident window", "scope": { "component": "tidb", "instance": "tidb-2", "window": "incident", }, }, { "item_id": "ke-2", "kind": "log", "summary": "logs contain kill query and query interruption signals before the panic", "scope": { "component": "tidb", "instance": "tidb-2", "window": "incident", }, }, { "item_id": "ke-3", "kind": "metric", "summary": "restart metrics confirm a single-instance restart during the incident window", "scope": { "component": "tidb", "instance": "tidb-2", "window": "incident", }, }, ], }, "answer": { "output_text": "TiDB v8.5.2 on a single TiDB instance likely hit a known bug path rather than a missing-index issue. Logs show KILL QUERY or query interruption activity, then rollback or idle transaction cleanup, followed by a memory tracker panic and instance restart. This matches the known bug fixed in v8.5.3 or later, so the recommended remediation is to upgrade to a release containing the fix." }, } def test_semantic_context_fixture_parses() -> None: model = SemanticContextV1.model_validate(_semantic_context_payload()) assert model.case.case_id == "component-restart-during-incident-window" assert model.deterministic.verdicts["critical_evidence"] == "pass" assert model.grounding.matched_critical_evidence_ids == [ "restart_signal_present", "panic_signal_present", "query_interruption_signal_present", ] def test_semantic_packet_fixture_parses() -> None: payload = { "version": "semantic_packet_v1", "model": { "provider": "heuristic", "model": "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.9, "reason": "ok", "evidence_refs": ["restart_signal_present"]}, ], "disagreement_flags": [], "notes": [], } model = SemanticPacketV1.model_validate(payload) assert model.case_id == "component-restart-during-incident-window" assert model.check_assessments[0].check_id == 1 assert model.check_assessments[0].status == "pass" def test_adapter_maps_yaml_driven_semantic_shadow_checks() -> None: packet = SemanticPacketV1.model_validate( { "version": "semantic_packet_v1", "model": { "provider": "heuristic", "model": "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.9, "reason": "ok", "evidence_refs": ["restart_signal_present"]}, {"check_id": 2, "status": "pass", "confidence": 0.9, "reason": "ok", "evidence_refs": ["query_interruption_signal_present"]}, {"check_id": 3, "status": "pass", "confidence": 0.9, "reason": "ok", "evidence_refs": ["ke-1"]}, {"check_id": 4, "status": "pass", "confidence": 0.9, "reason": "ok", "evidence_refs": []}, ], } ) specs = [ SemanticJudgeCheckSpec(check_id=1, title="主路径判断", description="...", max_score=3), SemanticJudgeCheckSpec(check_id=2, title="因果链完整性", description="...", max_score=3), SemanticJudgeCheckSpec(check_id=3, title="范围对齐", description="...", max_score=3), SemanticJudgeCheckSpec(check_id=4, title="修复建议对齐", description="...", max_score=3), ] result = adapt_diagnosis_semantic_shadow_checks(packet, specs) assert result.bucket == "diagnosis.semantic" assert [item.check_id for item in result.checks] == [ 1, 2, 3, 4, ] assert all(item.status == "pass" for item in result.checks) assert result.summary.shadow_score == 12 assert result.summary.shadow_max_score == 12