fix(ai): enforce ollama first for drift governance

apps/api/tests/test_drift_interpreter_ollama_first.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from src.models.drift import DriftIntent
+from src.services import openclaw as openclaw_module
+from src.services.drift_interpreter import NemotronDriftInterpreter
+
+
+class _FakeOpenClaw:
+    def __init__(self) -> None:
+        self.alert_context: dict[str, Any] | None = None
+
+    async def call(
+        self,
+        prompt: str,
+        alert_context: dict[str, Any] | None = None,
+    ) -> tuple[str, str, bool]:
+        self.alert_context = alert_context
+        return (
+            '{"intent":"automated_change","explanation":"HPA 自動調整","risk":"LOW","confidence":0.8}',
+            "ollama_gcp_a",
+            True,
+        )
+
+
+@pytest.mark.asyncio
+async def test_drift_interpreter_declares_ollama_first_governance_lane(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fake_openclaw = _FakeOpenClaw()
+    monkeypatch.setattr(openclaw_module, "get_openclaw", lambda: fake_openclaw)
+
+    result = await NemotronDriftInterpreter()._call_nemotron("請分析漂移")
+
+    assert result.intent == DriftIntent.AUTOMATED_CHANGE
+    assert fake_openclaw.alert_context is not None
+    assert fake_openclaw.alert_context["enforce_ollama_first"] is True
+    assert fake_openclaw.alert_context["task_type"] == "diagnose"
+    assert fake_openclaw.alert_context["allow_gcp_heavy_model"] is True

apps/api/tests/test_openclaw_alert_cloud_fallback_gate.py

@@ -177,6 +177,40 @@ async def test_non_alert_context_keeps_router_cloud_order(
     assert fake_executor.provider_order == ["gemini", "claude", "ollama"]
 
 
+@pytest.mark.asyncio
+async def test_explicit_ai_governance_context_uses_ollama_first(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fake_executor = _FakeExecutor()
+    fake_failover = _FakeFailoverManager()
+
+    monkeypatch.setattr(openclaw_module.settings, "USE_AI_ROUTER", True)
+    monkeypatch.setattr(openclaw_module.settings, "MOCK_MODE", False)
+    monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ALLOW_CLOUD_FALLBACK", True)
+    monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ENFORCE_OLLAMA_FIRST", True)
+    monkeypatch.setattr(ai_control_module, "get_ai_router_enabled", AsyncMock(return_value=None))
+    monkeypatch.setattr(ai_control_module, "get_primary_provider", AsyncMock(return_value=None))
+    monkeypatch.setattr(ai_control_module, "is_provider_disabled", AsyncMock(return_value=False))
+    monkeypatch.setattr(ai_router_module, "get_ai_router", lambda: _FakeRouter())
+    monkeypatch.setattr(ai_router_module, "get_ai_executor", lambda: fake_executor)
+    monkeypatch.setattr(openclaw_module, "get_ollama_failover_manager", lambda: fake_failover)
+
+    service = object.__new__(OpenClawService)
+    await service._call_with_fallback(
+        "analyze config drift",
+        alert_context={
+            "intent_hint": "config",
+            "task_type": "diagnose",
+            "enforce_ollama_first": True,
+            "allow_gcp_heavy_model": True,
+            "target_resource": "config-drift",
+        },
+    )
+
+    assert fake_executor.provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local", "gemini"]
+    assert fake_failover.task_types == ["diagnose"]
+
+
 @pytest.mark.asyncio
 async def test_alert_context_uses_gcp_a_gcp_b_then_111_order(
     monkeypatch: pytest.MonkeyPatch,