import json import importlib import os import subprocess from types import SimpleNamespace import investibook.config as config_module from investibook.core.eval import semantic_evaluator as semantic_evaluator_module from investibook.core.eval.eval_judge import judge_case_output from investibook.core.eval.eval_registry import load_case from investibook.core.eval.semantic_adapter_models import ( SemanticAnswerContext, SemanticContextCase, SemanticContextV1, SemanticDeterministicContext, SemanticGroundingContext, SemanticJudgeCheckSpec, ) from investibook.core.eval.semantic_evaluator import ( SEMANTIC_CONTEXT_FILENAME, SEMANTIC_PACKET_FILENAME, _split_opencode_model_variant, build_semantic_context, evaluate_semantic_context, maybe_run_semantic_evaluator, ) from .test_eval_judge import PASSING_OUTPUT, _minimal_analysis_patch, _setup_audit_run def _minimal_semantic_context() -> SemanticContextV1: return SemanticContextV1( version="semantic_context_v1", case=SemanticContextCase( case_id="cpu-spike-during-incident-window", ground_truth_root_cause="node count reduction increased per-node load", expected_cluster_id="10518506906906971664", allowed_window_ids=["workspace_observed_window"], semantic_checks=[ SemanticJudgeCheckSpec( check_id=1, title="primary path", description="capacity shift is the primary driver", max_score=3, required=True, ) ], ), deterministic=SemanticDeterministicContext(), grounding=SemanticGroundingContext(), answer=SemanticAnswerContext(output_text="capacity reduction increased per-node load"), ) def test_maybe_run_semantic_evaluator_writes_context_and_packet(tmp_path, monkeypatch) -> None: monkeypatch.setattr("investibook.config.SEMANTIC_EVAL_PROVIDER", "heuristic") 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(), ) judgment = judge_case_output( case, output_text=PASSING_OUTPUT, workspace_id="ws-semantic-evaluator", audit_dir=audit_dir, ) result = maybe_run_semantic_evaluator( case=case, output_text=PASSING_OUTPUT, judgment=judgment, audit_dir=audit_dir, ) assert result is not None context_path = audit_dir / SEMANTIC_CONTEXT_FILENAME packet_path = audit_dir / SEMANTIC_PACKET_FILENAME assert context_path.exists() assert packet_path.exists() context = json.loads(context_path.read_text()) packet = json.loads(packet_path.read_text()) assert context["version"] == "semantic_context_v1" assert context["case"]["case_id"] == "component-restart-during-incident-window" assert [item["check_id"] for item in context["case"]["semantic_checks"]] == [1, 2, 4] assert packet["version"] == "semantic_packet_v1" assert packet["status"] == "ok" assert result["shadow_available"] is True assert [item["check_id"] for item in packet["check_assessments"]] == [1, 2, 4] def test_opencode_semantic_evaluator_splits_model_variant(monkeypatch) -> None: captured = {} packet = { "version": "semantic_packet_v1", "model": { "provider": "placeholder", "model": "placeholder", "prompt_version": "semantic_case_local_v1", "schema_version": "semantic_packet_v1", }, "status": "ok", "case_id": "cpu-spike-during-incident-window", "check_assessments": [ { "check_id": 1, "status": "pass", "confidence": 0.9, "reason": "capacity shift is primary", "evidence_refs": [], "notes": [], } ], "disagreement_flags": [], "notes": [], } def fake_run(cmd, **kwargs): captured["cmd"] = cmd return subprocess.CompletedProcess( cmd, 0, stdout=json.dumps({"type": "text", "part": {"text": json.dumps(packet)}}) + "\n", stderr="", ) monkeypatch.setattr(semantic_evaluator_module.cfg, "SEMANTIC_EVAL_OPENCODE_BIN", "/fake/opencode") monkeypatch.setattr(semantic_evaluator_module.cfg, "SEMANTIC_EVAL_OPENCODE_MODEL", "chat/claude-standard/high") monkeypatch.setattr(semantic_evaluator_module.subprocess, "run", fake_run) result = semantic_evaluator_module._evaluate_with_opencode(_minimal_semantic_context()) assert captured["cmd"][:5] == [ "/fake/opencode", "run", "--format", "json", "--dir", ] assert captured["cmd"][5] assert "--model" in captured["cmd"] assert captured["cmd"][captured["cmd"].index("--model") + 1] == "chat/claude-standard" assert captured["cmd"][captured["cmd"].index("--variant") + 1] == "high" assert result.status == "ok" assert result.model.model == "chat/claude-standard/high" def test_opencode_semantic_evaluator_reports_error_event(monkeypatch) -> None: error_event = { "type": "error", "error": { "name": "UnknownError", "data": {"message": "Model not found: chat/claude-standard/high."}, }, } def fake_run(cmd, **kwargs): return subprocess.CompletedProcess( cmd, 0, stdout=json.dumps(error_event) + "\n", stderr="", ) monkeypatch.setattr(semantic_evaluator_module.cfg, "SEMANTIC_EVAL_OPENCODE_BIN", "/fake/opencode") monkeypatch.setattr(semantic_evaluator_module.cfg, "SEMANTIC_EVAL_OPENCODE_MODEL", "chat/claude-standard/high") monkeypatch.setattr(semantic_evaluator_module.subprocess, "run", fake_run) result = semantic_evaluator_module._evaluate_with_opencode(_minimal_semantic_context()) assert result.status == "error" assert result.notes == [ "opencode_run_failed:UnknownError: Model not found: chat/claude-standard/high." ] def test_maybe_run_semantic_evaluator_marks_error_packet_as_unavailable(tmp_path, monkeypatch) -> None: monkeypatch.setattr("investibook.config.SEMANTIC_EVAL_PROVIDER", "unsupported-provider") 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(), ) judgment = judge_case_output( case, output_text=PASSING_OUTPUT, workspace_id="ws-semantic-evaluator", audit_dir=audit_dir, ) result = maybe_run_semantic_evaluator( case=case, output_text=PASSING_OUTPUT, judgment=judgment, audit_dir=audit_dir, ) assert result is not None assert result["packet_status"] == "error" assert result["shadow_available"] is False def test_semantic_eval_model_falls_back_to_acp_model_standard(monkeypatch) -> None: original_semantic = os.environ.get("INVESTIBOOK_SEMANTIC_EVAL_OPENCODE_MODEL") original_standard = os.environ.get("ACP_MODEL_STANDARD") try: monkeypatch.delenv("INVESTIBOOK_SEMANTIC_EVAL_OPENCODE_MODEL", raising=False) monkeypatch.setenv("ACP_MODEL_STANDARD", "openai/gpt-5.4") reloaded = importlib.reload(config_module) assert reloaded.SEMANTIC_EVAL_OPENCODE_MODEL == "openai/gpt-5.4" monkeypatch.setenv("INVESTIBOOK_SEMANTIC_EVAL_OPENCODE_MODEL", "openai/gpt-5.5") reloaded = importlib.reload(config_module) assert reloaded.SEMANTIC_EVAL_OPENCODE_MODEL == "openai/gpt-5.5" finally: if original_semantic is None: os.environ.pop("INVESTIBOOK_SEMANTIC_EVAL_OPENCODE_MODEL", None) else: os.environ["INVESTIBOOK_SEMANTIC_EVAL_OPENCODE_MODEL"] = original_semantic if original_standard is None: os.environ.pop("ACP_MODEL_STANDARD", None) else: os.environ["ACP_MODEL_STANDARD"] = original_standard importlib.reload(config_module) def test_split_opencode_model_variant_with_reasoning_suffix() -> None: model, variant = _split_opencode_model_variant("chat/claude-standard/high") assert model == "chat/claude-standard" assert variant == "high" def test_split_opencode_model_variant_without_reasoning_suffix() -> None: model, variant = _split_opencode_model_variant("openai/gpt-5.4-mini") assert model == "openai/gpt-5.4-mini" assert variant == "" def test_evaluate_semantic_context_accepts_fenced_json_text_event(monkeypatch) -> None: monkeypatch.setattr("investibook.config.SEMANTIC_EVAL_PROVIDER", "opencode") monkeypatch.setattr("investibook.core.eval.semantic_evaluator.shutil.which", lambda _: "/tmp/opencode") payload = { "version": "semantic_packet_v1", "model": { "provider": "placeholder", "model": "placeholder", "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": [], "notes": [], } ], "disagreement_flags": [], "notes": [], } completed = SimpleNamespace( returncode=0, stdout=json.dumps( { "type": "text", "part": {"text": f"```json\n{json.dumps(payload, ensure_ascii=True)}\n```"}, } ) + "\n", stderr="", ) monkeypatch.setattr( "investibook.core.eval.semantic_evaluator.subprocess.run", lambda *args, **kwargs: completed, ) case = load_case("component-restart-during-incident-window") judgment = judge_case_output( case, output_text=PASSING_OUTPUT, workspace_id="ws-semantic-evaluator", audit_dir=None, ) context = build_semantic_context(case=case, output_text=PASSING_OUTPUT, judgment=judgment) assert context is not None packet = evaluate_semantic_context(context) assert packet.status == "ok" assert packet.check_assessments[0].check_id == 1 assert packet.check_assessments[0].status == "pass" def test_evaluate_semantic_context_passes_model_variant_to_opencode(monkeypatch) -> None: monkeypatch.setattr("investibook.config.SEMANTIC_EVAL_PROVIDER", "opencode") monkeypatch.setattr("investibook.config.SEMANTIC_EVAL_OPENCODE_MODEL", "chat/claude-standard/high") monkeypatch.setattr("investibook.core.eval.semantic_evaluator.shutil.which", lambda _: "/tmp/opencode") payload = { "version": "semantic_packet_v1", "model": { "provider": "placeholder", "model": "placeholder", "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": [], "notes": [], } ], "disagreement_flags": [], "notes": [], } completed = SimpleNamespace( returncode=0, stdout=json.dumps( { "type": "text", "part": {"text": json.dumps(payload, ensure_ascii=True)}, } ) + "\n", stderr="", ) captured: dict[str, list[str]] = {} def _run(cmd, *args, **kwargs): captured["cmd"] = cmd return completed monkeypatch.setattr("investibook.core.eval.semantic_evaluator.subprocess.run", _run) case = load_case("component-restart-during-incident-window") judgment = judge_case_output( case, output_text=PASSING_OUTPUT, workspace_id="ws-semantic-evaluator", audit_dir=None, ) context = build_semantic_context(case=case, output_text=PASSING_OUTPUT, judgment=judgment) assert context is not None packet = evaluate_semantic_context(context) assert packet.status == "ok" assert captured["cmd"][:8] == [ "/tmp/opencode", "run", "--format", "json", "--dir", captured["cmd"][5], "--model", "chat/claude-standard", ] assert "--variant" in captured["cmd"] assert captured["cmd"][captured["cmd"].index("--variant") + 1] == "high" def test_evaluate_semantic_context_accepts_assistant_message_event(monkeypatch) -> None: monkeypatch.setattr("investibook.config.SEMANTIC_EVAL_PROVIDER", "opencode") monkeypatch.setattr("investibook.core.eval.semantic_evaluator.shutil.which", lambda _: "/tmp/opencode") payload = { "version": "semantic_packet_v1", "model": { "provider": "placeholder", "model": "placeholder", "prompt_version": "semantic_case_local_v1", "schema_version": "semantic_packet_v1", }, "status": "ok", "case_id": "component-restart-during-incident-window", "check_assessments": [ { "check_id": 2, "status": "pass", "confidence": 0.85, "reason": "ok", "evidence_refs": [], "notes": [], } ], "disagreement_flags": [], "notes": [], } completed = SimpleNamespace( returncode=0, stdout=json.dumps( { "type": "response_item", "payload": { "type": "message", "role": "assistant", "content": [ { "type": "output_text", "text": json.dumps(payload, ensure_ascii=True), } ], }, } ) + "\n", stderr="", ) monkeypatch.setattr( "investibook.core.eval.semantic_evaluator.subprocess.run", lambda *args, **kwargs: completed, ) case = load_case("component-restart-during-incident-window") judgment = judge_case_output( case, output_text=PASSING_OUTPUT, workspace_id="ws-semantic-evaluator", audit_dir=None, ) context = build_semantic_context(case=case, output_text=PASSING_OUTPUT, judgment=judgment) assert context is not None packet = evaluate_semantic_context(context) assert packet.status == "ok" assert packet.check_assessments[0].check_id == 2 assert packet.check_assessments[0].status == "pass" def test_evaluate_semantic_context_preserves_empty_content_diagnostics(monkeypatch) -> None: monkeypatch.setattr("investibook.config.SEMANTIC_EVAL_PROVIDER", "opencode") monkeypatch.setattr("investibook.core.eval.semantic_evaluator.shutil.which", lambda _: "/tmp/opencode") completed = SimpleNamespace( returncode=0, stdout="", stderr="provider returned no assistant content", ) monkeypatch.setattr( "investibook.core.eval.semantic_evaluator.subprocess.run", lambda *args, **kwargs: completed, ) case = load_case("component-restart-during-incident-window") judgment = judge_case_output( case, output_text=PASSING_OUTPUT, workspace_id="ws-semantic-evaluator", audit_dir=None, ) context = build_semantic_context(case=case, output_text=PASSING_OUTPUT, judgment=judgment) assert context is not None packet = evaluate_semantic_context(context) assert packet.status == "error" assert packet.notes assert packet.notes[0].startswith("opencode_empty_content:returncode=0") assert "provider returned no assistant content" in packet.notes[0]