From f64174106b4d7752797efade3223620ede46550d Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 10:31:33 +0800 Subject: [PATCH] fix(ai): restore bounded alert automation fallback --- apps/api/src/api/v1/webhooks.py | 6 +- apps/api/src/services/ai_router.py | 30 +++++++- .../src/services/heartbeat_report_service.py | 41 ++++++++--- .../src/services/ollama_endpoint_resolver.py | 22 ++++-- .../src/services/ollama_failover_manager.py | 35 +++++++-- apps/api/src/services/openclaw.py | 73 ++++++++++++++----- apps/api/src/services/signoz_client.py | 33 +++++++-- apps/api/src/services/telegram_gateway.py | 5 +- .../test_ai_router_cache_provider_policy.py | 47 ++++++++++++ .../tests/test_alertmanager_rule_bypass.py | 23 ++++++ apps/api/tests/test_heartbeat_dedup_p0_4.py | 41 ++++++++++- .../tests/test_heartbeat_ollama_endpoints.py | 25 ++++++- .../tests/test_ollama_endpoint_resolver.py | 19 ++++- .../api/tests/test_ollama_failover_manager.py | 22 +++++- ...test_openclaw_alert_cloud_fallback_gate.py | 36 ++++++--- apps/api/tests/test_signoz_fail_closed.py | 60 +++++++++++++++ 16 files changed, 449 insertions(+), 69 deletions(-) create mode 100644 apps/api/tests/test_signoz_fail_closed.py diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py index 9dab1239f..d7e12cbf3 100644 --- a/apps/api/src/api/v1/webhooks.py +++ b/apps/api/src/api/v1/webhooks.py @@ -2414,7 +2414,7 @@ async def _process_new_alert_background( _op_log_fallback = get_alert_operation_log_repository() if repair_candidate_result.candidate_found: await _op_log_fallback.append( - "REPAIR_CANDIDATE_READY", + "PLAYBOOK_DRAFT_CREATED", incident_id=fallback_incident_id, approval_id=str(approval.id), actor="openclaw-repair-candidate", @@ -2429,7 +2429,7 @@ async def _process_new_alert_background( ) elif repair_candidate_result.draft_ready_for_owner_review: await _op_log_fallback.append( - "REPAIR_CANDIDATE_DRAFT_READY", + "PLAYBOOK_DRAFT_CREATED", incident_id=fallback_incident_id, approval_id=str(approval.id), actor="openclaw-repair-candidate", @@ -2447,7 +2447,7 @@ async def _process_new_alert_background( ) else: await _op_log_fallback.append( - "REPAIR_CANDIDATE_BLOCKED", + "STATE_GUARD_BLOCKED", incident_id=fallback_incident_id, approval_id=str(approval.id), actor="openclaw-repair-candidate", diff --git a/apps/api/src/services/ai_router.py b/apps/api/src/services/ai_router.py index 67b9a15c4..2e88fc307 100644 --- a/apps/api/src/services/ai_router.py +++ b/apps/api/src/services/ai_router.py @@ -418,7 +418,11 @@ class AIRouter: if provider == AIProviderEnum.OLLAMA: try: failover_result = await self._failover_manager.select_provider( - task_type=intent.value if intent else "general" + task_type=( + "alert_fast" + if intent == IntentType.ALERT_TRIAGE + else intent.value if intent else "general" + ) ) primary_str = failover_result.primary.provider_name try: @@ -1141,6 +1145,18 @@ class AIRouterExecutor: errors: list[str] = [] attempted_providers: set[str] = set() + provider_timeout_seconds: float | None = None + try: + configured_provider_timeout = float( + (context or {}).get("provider_timeout_seconds") or 0 + ) + if configured_provider_timeout > 0: + provider_timeout_seconds = configured_provider_timeout + except (TypeError, ValueError): + logger.warning( + "ai_router_invalid_provider_timeout", + value=(context or {}).get("provider_timeout_seconds"), + ) alert_requires_ollama_before_cloud = bool( (context or {}).get("alert_requires_ollama_before_cloud") ) @@ -1238,7 +1254,15 @@ class AIRouterExecutor: async with sem: try: attempted_providers.add(provider_name) - result = await provider.analyze(prompt, context) + provider_call = provider.analyze(prompt, context) + result = ( + await asyncio.wait_for( + provider_call, + timeout=provider_timeout_seconds, + ) + if provider_timeout_seconds is not None + else await provider_call + ) if result.success: # 記錄成功 (per-provider CB) @@ -1300,7 +1324,7 @@ class AIRouterExecutor: # NIM 偶爾 GPU 忙碌導致 27s，timeout 不代表 NIM 故障 # 只有明確連線錯誤（非 timeout）才累積 CB 失敗次數 import httpx as _httpx - if not isinstance(e, _httpx.TimeoutException): + if not isinstance(e, (_httpx.TimeoutException, TimeoutError)): cb.record_failure() # 2026-04-27 A2: 記錄失敗的 provider，供下輪迭代計算 fallback metric _last_attempted_provider = provider_name diff --git a/apps/api/src/services/heartbeat_report_service.py b/apps/api/src/services/heartbeat_report_service.py index 4060ac7a0..2e74a56b9 100644 --- a/apps/api/src/services/heartbeat_report_service.py +++ b/apps/api/src/services/heartbeat_report_service.py @@ -112,6 +112,7 @@ class HeartbeatReport: ai_services: dict[str, ProbeResult] = field(default_factory=dict) ollama_models: dict[str, bool] = field(default_factory=dict) ollama_endpoints: dict[str, ProbeResult] = field(default_factory=dict) + ollama_active_endpoint: str | None = None mcp_providers: dict[str, ProbeResult] = field(default_factory=dict) flywheel: FlywheelStats = field(default_factory=FlywheelStats) infra: dict[str, ProbeResult] = field(default_factory=dict) @@ -186,6 +187,7 @@ class HeartbeatReportService: report.ai_services["ollama"] = ollama_data.get("probe", ProbeResult(False, "❌ 無回應")) report.ollama_models = ollama_data.get("models", {}) report.ollama_endpoints = ollama_data.get("endpoints", {}) + report.ollama_active_endpoint = ollama_data.get("active_endpoint") report.ai_services["nemotron"] = collected["_nemotron"] or ProbeResult(False, "❌ 無回應") report.ai_services["gemini"] = collected["_gemini"] or ProbeResult(False, "❌ 無回應") report.ai_services["claude"] = collected["_claude"] or ProbeResult(False, "❌ 無回應") @@ -259,31 +261,44 @@ class HeartbeatReportService: ) endpoint_status = {label: probe for label, probe, _available in results} - primary_probe = endpoint_status.get("GCP-A", ProbeResult(False, "❌ 無回應")) - primary_available = next( - (available for label, _probe, available in results if label == "GCP-A"), - set(), + active = next( + ( + (label, probe, available) + for label, probe, available in results + if probe.ok + ), + None, ) - if primary_probe.ok: + if active: + active_label, active_probe, active_available = active # 也把 short name（無 :tag）加進去方便匹配 - available_short = {n.split(":")[0] for n in primary_available} + available_short = {n.split(":")[0] for n in active_available} model_status: dict[str, bool] = {} for required in settings.OLLAMA_REQUIRED_MODELS: req_short = required.split(":")[0] - ok = required in primary_available or req_short in available_short + ok = required in active_available or req_short in available_short model_status[required] = ok + aggregate_probe = active_probe + if active_label != "GCP-A": + aggregate_probe = ProbeResult( + True, + f"✅ 備援中 ({active_label})", + active_probe.latency_ms, + ) return { - "probe": primary_probe, + "probe": aggregate_probe, "models": model_status, "endpoints": endpoint_status, + "active_endpoint": active_label, } return { - "probe": primary_probe, + "probe": ProbeResult(False, "❌ 所有 Ollama 端點無回應"), "models": {}, "endpoints": endpoint_status, + "active_endpoint": None, } async def _probe_nemotron(self) -> ProbeResult: @@ -762,7 +777,13 @@ class HeartbeatReportService: for name, probe in report.ollama_endpoints.items(): if not probe.ok and not probe.status.startswith("⚠️ 未設定"): - warnings.append(f"Ollama {name} 異常: {probe.status}") + active = report.ollama_active_endpoint + if active: + warnings.append( + f"Ollama {name} 降級: {probe.status}；{active} 備援執行中" + ) + else: + warnings.append(f"Ollama {name} 異常: {probe.status}") # AI 服務異常 for name, probe in report.ai_services.items(): diff --git a/apps/api/src/services/ollama_endpoint_resolver.py b/apps/api/src/services/ollama_endpoint_resolver.py index a8e1e13de..b0a9288f9 100644 --- a/apps/api/src/services/ollama_endpoint_resolver.py +++ b/apps/api/src/services/ollama_endpoint_resolver.py @@ -1,9 +1,10 @@ """ Ollama endpoint resolver for AWOOOI workload placement. -ADR-110 gives AWOOOI three Ollama endpoints. The global order is always -GCP-A -> GCP-B -> 111 local; Gemini is owned by the caller/AI Router as the -final non-Ollama fallback. +ADR-110 gives AWOOOI three Ollama endpoints. Deep and batch workloads use +GCP-A -> GCP-B -> 111 local. Latency-bound alert workloads keep GCP-A first, +then prefer 111 before the slower GCP-B CPU lane. Gemini is owned by the +caller/AI Router as the final non-Ollama fallback. """ from __future__ import annotations @@ -59,13 +60,20 @@ def resolve_ollama_order( *, config: _OllamaSettings | None = None, ) -> tuple[OllamaEndpointSelection, ...]: - """Return the global Ollama fallback order: GCP-A -> GCP-B -> 111.""" + """Return the Ollama fallback order for the requested workload.""" cfg = config or settings + primary = (cfg.OLLAMA_URL, "ollama_gcp_a", "global_primary_gcp_a") + secondary = ( + cfg.OLLAMA_SECONDARY_URL, + "ollama_gcp_b", + "global_secondary_gcp_b", + ) + local = (cfg.OLLAMA_FALLBACK_URL, "ollama_local", "global_local_111") candidates = ( - (cfg.OLLAMA_URL, "ollama_gcp_a", "global_primary_gcp_a"), - (cfg.OLLAMA_SECONDARY_URL, "ollama_gcp_b", "global_secondary_gcp_b"), - (cfg.OLLAMA_FALLBACK_URL, "ollama_local", "global_local_111"), + (primary, local, secondary) + if workload_type == "alert_fast" + else (primary, secondary, local) ) selections: list[OllamaEndpointSelection] = [] seen: set[str] = set() diff --git a/apps/api/src/services/ollama_failover_manager.py b/apps/api/src/services/ollama_failover_manager.py index 148e6e927..ae201d1b6 100644 --- a/apps/api/src/services/ollama_failover_manager.py +++ b/apps/api/src/services/ollama_failover_manager.py @@ -182,15 +182,15 @@ class OllamaFailoverManager: context: dict | None = None, ) -> OllamaRoutingResult: """ - 三層 Ollama 容災路由（ADR-110 修正版 2026-05-04）： - Primary(OLLAMA_URL) → Secondary(OLLAMA_SECONDARY_URL) → Tertiary(OLLAMA_FALLBACK_URL) - → Gemini → Nemotron → Claude + 三層 Ollama 容災路由（ADR-110 修正版 2026-05-04）。一般工作使用 + GCP-A → GCP-B → 111；alert_fast 在 GCP-A 不可用時優先 111， + 避免慢速 GCP-B CPU 推理耗盡告警回應期限。 - 2026-05-04 ogt: URL 優先序已更新（ConfigMap），primary = 111（K8s 內網可達）。 - GCP-A/B 為 secondary/tertiary，待 nginx proxy 架設後再升回 primary。 + Production URL slots are GCP-A primary, GCP-B secondary, and 111 local + fallback; the task policy determines the order after health probing. Args: - task_type: 任務類型（預留，目前未影響路由邏輯） + task_type: 任務類型；alert_fast 使用低延遲 fallback 順序 context: 額外上下文（預留） Returns: @@ -254,6 +254,7 @@ class OllamaFailoverManager: url_gcp_a=url_primary, url_gcp_b=url_secondary, url_local=url_tertiary, + task_type=task_type, ) # Gemini 帳單熔斷（quota gate） @@ -333,6 +334,7 @@ class OllamaFailoverManager: url_gcp_a: str, url_gcp_b: str, url_local: str, + task_type: str = "", ) -> OllamaRoutingResult: """ 三層 Ollama 決策矩陣（2026-05-03 ogt，ADR-110）： @@ -359,6 +361,7 @@ class OllamaFailoverManager: lbl_p = _short(url_gcp_a) # primary label lbl_s = _short(url_gcp_b) # secondary label lbl_t = _short(url_local) # tertiary label + alert_fast = str(task_type or "").strip().lower() == "alert_fast" # Primary HEALTHY → 使用 primary if health_gcp_a.status == HealthStatus.HEALTHY: @@ -371,6 +374,26 @@ class OllamaFailoverManager: health_local=health_local, ) + # Production inference canary (2026-07-11): the same minimal + # gemma3:4b response took ~4s on 111 and ~37s on GCP-B. For the + # bounded alert-response lane, prefer the verified low-latency host. + if alert_fast and health_local.status == HealthStatus.HEALTHY: + fallback_chain = [] + if health_gcp_b.status in {HealthStatus.HEALTHY, HealthStatus.SLOW}: + fallback_chain.append(ep_gcp_b) + fallback_chain.append(_GEMINI_ENDPOINT) + return OllamaRoutingResult( + primary=ep_local, + fallback_chain=fallback_chain, + routing_reason=( + f"primary({lbl_p}) {health_gcp_a.status.value}" + f" → alert_fast tertiary({lbl_t}) at {now_ts}" + ), + health_gcp_a=health_gcp_a, + health_gcp_b=health_gcp_b, + health_local=health_local, + ) + # Primary 不健康，Secondary HEALTHY → 切 secondary if health_gcp_b.status == HealthStatus.HEALTHY: return OllamaRoutingResult( diff --git a/apps/api/src/services/openclaw.py b/apps/api/src/services/openclaw.py index 4cd5b908c..92d42f014 100644 --- a/apps/api/src/services/openclaw.py +++ b/apps/api/src/services/openclaw.py @@ -145,7 +145,8 @@ class OpenClawService: OpenClaw AI 決策服務 - True LLM + SignOz Integration 實作 AI_FALLBACK_ORDER 備援機制。 - 告警/incident 上下文預設套用成本防線，只允許 Ollama GCP-A → GCP-B → 111。 + 告警/incident 上下文預設套用成本與期限防線，依序嘗試 + Ollama GCP-A → 111 → GCP-B，再進入雲端備援。 新增 SignOz 整合: - 自動擷取 Gold Metrics @@ -186,9 +187,7 @@ class OpenClawService: "alert_name", "fingerprint", "incident_id", - "severity", "signals", - "target_resource", } return any(key in alert_context for key in alert_keys) @@ -201,7 +200,7 @@ class OpenClawService: return bool(getattr(settings, "ALERT_AI_ALLOW_CLOUD_FALLBACK", True)) def _alert_enforces_ollama_first(self, alert_context: dict | None) -> bool: - """Alert and AI-governance lanes must try GCP-A/GCP-B/111 before Gemini backup.""" + """Require the workload-specific Ollama lane before cloud backup.""" return ( bool(alert_context) and ( @@ -217,7 +216,7 @@ class OpenClawService: alert_context: dict | None = None, cloud_provider_order: list[str] | None = None, ) -> list[str]: - """Resolve GCP-A/GCP-B/111, then OpenClaw/Nemo before Gemini backup.""" + """Resolve the Ollama lane, then OpenClaw/Nemo and Gemini backup.""" provider_order: list[str] = [] try: route = await get_ollama_failover_manager().select_provider(task_type=task_type) @@ -244,7 +243,11 @@ class OpenClawService: if not self._alert_enforces_ollama_first(alert_context): return deduped - ollama_order = {"ollama_gcp_a": 0, "ollama_gcp_b": 1, "ollama_local": 2} + ollama_order = ( + {"ollama_gcp_a": 0, "ollama_local": 1, "ollama_gcp_b": 2} + if self._is_incident_alert_context(alert_context) + else {"ollama_gcp_a": 0, "ollama_gcp_b": 1, "ollama_local": 2} + ) ordered_ollama = [ provider_name for provider_name in deduped @@ -1108,8 +1111,17 @@ class OpenClawService: router = get_ai_router() executor = get_ai_executor() + routing_context = dict(alert_context or {}) + is_incident_alert = self._is_incident_alert_context( + routing_context + ) + if is_incident_alert: + # Alert payloads already provide a stable intent. Avoid the + # three sequential 5s LLM-classifier probes seen in prod. + routing_context.setdefault("intent_hint", "alert_triage") + # Step 1: 取得路由決策 (含意圖分類 + 複雜度評分) - decision = await router.route(prompt, alert_context) + decision = await router.route(prompt, routing_context) # Step 2: 從 RoutingDecision 建立 provider_order (主 + fallback) # Phase 24 C: Redis primary_provider 覆蓋路由決策 @@ -1137,18 +1149,24 @@ class OpenClawService: except Exception as _e: logger.warning("ai_control_override_failed", error=str(_e)) - if self._alert_enforces_ollama_first(alert_context): + if self._alert_enforces_ollama_first(routing_context): original_provider_order = list(provider_order) provider_order = await self._resolve_alert_provider_order( - task_type=decision.intent.value if decision.intent else "diagnose", - alert_context=alert_context, + task_type=( + "alert_fast" + if is_incident_alert + else decision.intent.value if decision.intent else "diagnose" + ), + alert_context=routing_context, cloud_provider_order=original_provider_order, ) logger.info( "alert_ollama_first_provider_order", original_provider_order=original_provider_order, provider_order=provider_order, - cloud_fallback_allowed=self._cloud_fallback_allowed_for_alert(alert_context), + cloud_fallback_allowed=self._cloud_fallback_allowed_for_alert( + routing_context + ), ) # Step 3: D7 隱私 — CODE_REVIEW 強制 local @@ -1163,15 +1181,36 @@ class OpenClawService: # - 其他/未注入 → OPENCLAW_TIMEOUT=30s（不夠 qwen2.5:7b 推理） # webhooks alert_context 從未注入 task_type → Ollama fallback 永遠 30s timeout # 對齊 decision.intent 後 Ollama fallback 真正能跑完 - exec_context = dict(alert_context) if alert_context else {} + exec_context = dict(routing_context) if decision.intent == IntentType.DIAGNOSE: exec_context["task_type"] = "diagnose" - if self._alert_enforces_ollama_first(alert_context): - exec_context.setdefault("task_type", "diagnose") - exec_context.setdefault("ollama_model", getattr(settings, "ALERT_OLLAMA_MODEL", "qwen3:14b")) - exec_context["allow_gcp_heavy_model"] = bool( - exec_context.get("allow_gcp_heavy_model", True) + if self._alert_enforces_ollama_first(routing_context): + exec_context["task_type"] = ( + "alert_fast" if is_incident_alert else "diagnose" ) + exec_context.setdefault( + "ollama_model", + ( + getattr( + settings, + "OLLAMA_HEALTH_CHECK_MODEL", + "gemma3:4b", + ) + if is_incident_alert + else getattr(settings, "ALERT_OLLAMA_MODEL", "qwen3:14b") + ), + ) + exec_context["allow_gcp_heavy_model"] = bool( + exec_context.get( + "allow_gcp_heavy_model", + not is_incident_alert, + ) + ) + if is_incident_alert: + # Four possible providers must fit inside the 90s + # Alertmanager background budget. Production canaries + # show 111 completes this payload in ~4s. + exec_context.setdefault("provider_timeout_seconds", 18.0) exec_context["alert_requires_ollama_before_cloud"] = True result = await executor.execute( diff --git a/apps/api/src/services/signoz_client.py b/apps/api/src/services/signoz_client.py index 3b1bbd26e..bec831b36 100644 --- a/apps/api/src/services/signoz_client.py +++ b/apps/api/src/services/signoz_client.py @@ -139,7 +139,12 @@ class SignOzClient: # ClickHouse Direct Queries (永久架構) # ========================================================================= - async def _query_clickhouse(self, query: str) -> list[dict]: + async def _query_clickhouse( + self, + query: str, + *, + fail_closed: bool = False, + ) -> list[dict]: """ 執行 ClickHouse 查詢 (原生 httpx，非 curl) @@ -181,16 +186,26 @@ class SignOzClient: response_text=response.text[:200], elapsed_ms=round(elapsed_ms, 2), ) + if fail_closed: + raise RuntimeError( + f"ClickHouse query returned HTTP {response.status_code}" + ) return [] # 解析 JSONEachRow 格式 (每行一個 JSON 物件) results = [] + parse_error_count = 0 for line in response.text.strip().split("\n"): if line: try: results.append(json.loads(line)) except json.JSONDecodeError: - continue + parse_error_count += 1 + + if fail_closed and parse_error_count: + raise RuntimeError( + "ClickHouse returned malformed JSONEachRow output" + ) logger.info( "clickhouse_query_success", @@ -210,6 +225,8 @@ class SignOzClient: query=query[:100], elapsed_ms=round(elapsed_ms, 2), ) + if fail_closed: + raise return [] # ========================================================================= @@ -266,7 +283,7 @@ class SignOzClient: AND serviceName LIKE '%{service_name}%' """ - rps_results = await self._query_clickhouse(rps_query) + rps_results = await self._query_clickhouse(rps_query, fail_closed=True) if rps_results: row = rps_results[0] @@ -292,7 +309,10 @@ class SignOzClient: AND serviceName LIKE '%{service_name}%' """ - latency_results = await self._query_clickhouse(latency_query) + latency_results = await self._query_clickhouse( + latency_query, + fail_closed=True, + ) if latency_results: row = latency_results[0] @@ -311,7 +331,10 @@ class SignOzClient: AND serviceName LIKE '%{service_name}%' """ - trend_results = await self._query_clickhouse(trend_query) + trend_results = await self._query_clickhouse( + trend_query, + fail_closed=True, + ) if trend_results: prev_total = int(trend_results[0].get("prev_requests", 0)) diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index f282ec925..774de7eb1 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -82,7 +82,10 @@ _HEARTBEAT_WARNING_FINGERPRINT_RULES: tuple[tuple[re.Pattern[str], str], ...] = # actionable condition, not volatile probe details. HTTP status, timeout, # latency and count text can vary every 30 minutes while the operator # action stays exactly the same, which caused repeated "heartbeat" noise. - (re.compile(r"^(Ollama\s+[^:]+)\s+異常:.*$"), r"\1 異常"), + ( + re.compile(r"^(Ollama\s+[^:]+)\s+(異常|降級):.*$"), + r"\1 \2", + ), (re.compile(r"^(.+?)\s+服務異常:.*$"), r"\1 服務異常"), (re.compile(r"^(MCP\s+[^:]+):.*$"), r"\1"), (re.compile(r"^(ArgoCD):.*$"), r"\1"), diff --git a/apps/api/tests/test_ai_router_cache_provider_policy.py b/apps/api/tests/test_ai_router_cache_provider_policy.py index 5309850f1..4dfd74de3 100644 --- a/apps/api/tests/test_ai_router_cache_provider_policy.py +++ b/apps/api/tests/test_ai_router_cache_provider_policy.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import json from typing import Any @@ -71,6 +72,21 @@ class _CloudProvider: return AIResult(raw_response='{"provider":"gemini"}', success=True, provider=self.name) +class _SlowLocalProvider(_FailingLocalProvider): + async def analyze( + self, + prompt: str, + context: dict[str, Any] | None = None, + ) -> AIResult: + self.calls += 1 + await asyncio.sleep(1) + return AIResult( + raw_response='{"provider":"late"}', + success=True, + provider=self.name, + ) + + @pytest.mark.asyncio async def test_executor_skips_cached_cloud_provider_when_ollama_lane_is_required( monkeypatch: pytest.MonkeyPatch, @@ -178,3 +194,34 @@ async def test_alert_cloud_backup_allowed_after_ollama_local_attempt( assert gcp_a.calls == 1 assert local.calls == 1 assert gemini.calls == 1 + + +@pytest.mark.asyncio +async def test_alert_provider_timeout_allows_next_fallback( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_redis = _FakeRedis(cached_provider="none") + local = _SlowLocalProvider("ollama_local") + gemini = _CloudProvider() + registry = AIProviderRegistry() + registry.register(local) + registry.register(gemini) + + monkeypatch.setattr(ai_router_module._settings, "MOCK_MODE", False) + monkeypatch.setattr("src.core.redis_client.get_redis", lambda: fake_redis) + + result = await AIRouterExecutor(registry).execute( + prompt="diagnose alert", + provider_order=["ollama_local", "gemini"], + context={ + "intent_hint": "alert_triage", + "alert_type": "HostHighCpuLoad", + "alert_requires_ollama_before_cloud": True, + "provider_timeout_seconds": 0.01, + }, + ) + + assert result.success is True + assert result.provider == "gemini" + assert local.calls == 1 + assert gemini.calls == 1 diff --git a/apps/api/tests/test_alertmanager_rule_bypass.py b/apps/api/tests/test_alertmanager_rule_bypass.py index 742c71d29..cc053e082 100644 --- a/apps/api/tests/test_alertmanager_rule_bypass.py +++ b/apps/api/tests/test_alertmanager_rule_bypass.py @@ -1,13 +1,17 @@ +import ast import asyncio +import inspect from datetime import datetime import pytest +from src.api.v1 import webhooks as webhooks_module from src.api.v1.webhooks import ( _analyze_alertmanager_with_timeout, _should_bypass_alertmanager_llm, _should_use_alertmanager_rule_first, ) +from src.repositories.alert_operation_log_repository import ALERT_EVENT_TYPES from src.services.alertmanager_llm_guard import ( ALERTMANAGER_LLM_INFLIGHT_LOCK_TTL_SECONDS, alertmanager_llm_inflight_key, @@ -20,6 +24,25 @@ from src.services.decision_manager import ( from src.services.telegram_gateway import _format_resolved_guard_stamp +def test_repair_candidate_operation_log_events_use_database_enum() -> None: + tree = ast.parse(inspect.getsource(webhooks_module)) + emitted = { + call.args[0].value + for call in ast.walk(tree) + if isinstance(call, ast.Call) + and isinstance(call.func, ast.Attribute) + and call.func.attr == "append" + and isinstance(call.func.value, ast.Name) + and call.func.value.id == "_op_log_fallback" + and call.args + and isinstance(call.args[0], ast.Constant) + and isinstance(call.args[0].value, str) + } + + assert emitted == {"PLAYBOOK_DRAFT_CREATED", "STATE_GUARD_BLOCKED"} + assert emitted <= ALERT_EVENT_TYPES + + def test_host_resource_yaml_no_action_bypasses_llm(): rule_response = { "rule_id": "host_resource_alert", diff --git a/apps/api/tests/test_heartbeat_dedup_p0_4.py b/apps/api/tests/test_heartbeat_dedup_p0_4.py index aa0ea4609..691c3cf88 100644 --- a/apps/api/tests/test_heartbeat_dedup_p0_4.py +++ b/apps/api/tests/test_heartbeat_dedup_p0_4.py @@ -10,7 +10,7 @@ P0 #4 heartbeat 噪音降頻測試 直接測 telegram_gateway.send_heartbeat()，mock 掉 redis + report + send_to_group。 """ -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, patch import pytest @@ -53,10 +53,12 @@ class FakeRedis: def _make_report(warnings: list[str] | None = None): """構造 fake HeartbeatReport""" - from datetime import datetime, timezone + from datetime import UTC, datetime + from src.services.heartbeat_report_service import HeartbeatReport + return HeartbeatReport( - timestamp=datetime.now(timezone.utc), + timestamp=datetime.now(UTC), warnings=warnings or [], ) @@ -186,6 +188,39 @@ class TestHeartbeatDedup: gw.send_to_group.assert_not_called() gw.send_notification.assert_not_called() + @pytest.mark.asyncio + async def test_ollama_fallback_warning_details_do_not_repeat( + self, + gateway_with_fake_redis, + sre_group_configured, + ): + """備援正常時，單一端點的 volatile 降級細節不重複推送。""" + gw, fake_redis = gateway_with_fake_redis + from src.services.telegram_gateway import _heartbeat_warnings_hash + + fake_redis.preset( + "heartbeat:warnings_hash", + _heartbeat_warnings_hash( + ["Ollama GCP-A 降級: ❌ timeout；111 備援執行中"] + ), + ) + + with patch("src.core.redis_client.get_redis", return_value=fake_redis), \ + patch("src.services.heartbeat_report_service.HeartbeatReportService") as MockSvc, \ + patch("src.services.heartbeat_report_service.report_to_telegram_html", + return_value="warnings"): + MockSvc.return_value.collect = AsyncMock( + return_value=_make_report( + ["Ollama GCP-A 降級: ❌ connect timeout；111 備援執行中"] + ) + ) + + result = await gw.send_heartbeat() + + assert result is True + gw.send_to_group.assert_not_called() + gw.send_notification.assert_not_called() + @pytest.mark.asyncio async def test_warnings_changed_pushes( self, diff --git a/apps/api/tests/test_heartbeat_ollama_endpoints.py b/apps/api/tests/test_heartbeat_ollama_endpoints.py index c015db852..f9559421c 100644 --- a/apps/api/tests/test_heartbeat_ollama_endpoints.py +++ b/apps/api/tests/test_heartbeat_ollama_endpoints.py @@ -21,7 +21,7 @@ class _FakeAsyncClient: def __init__(self, *_args: Any, **_kwargs: Any) -> None: pass - async def __aenter__(self) -> "_FakeAsyncClient": + async def __aenter__(self) -> _FakeAsyncClient: return self async def __aexit__(self, *_args: Any) -> None: @@ -53,6 +53,24 @@ async def test_probe_ollama_reports_each_endpoint(monkeypatch) -> None: assert result["endpoints"]["GCP-A"].ok is True assert result["endpoints"]["GCP-B"].ok is True assert result["endpoints"]["111"].ok is False + assert result["active_endpoint"] == "GCP-A" + + +@pytest.mark.asyncio +async def test_probe_ollama_uses_healthy_fallback_when_gcp_a_is_down(monkeypatch) -> None: + monkeypatch.setattr(heartbeat.httpx, "AsyncClient", _FakeAsyncClient) + monkeypatch.setattr(heartbeat.settings, "OLLAMA_URL", "http://unreachable:11434") + monkeypatch.setattr(heartbeat.settings, "OLLAMA_SECONDARY_URL", "http://gcp-b:11434") + monkeypatch.setattr(heartbeat.settings, "OLLAMA_FALLBACK_URL", "http://local-111:11434") + monkeypatch.setattr(heartbeat.settings, "OLLAMA_REQUIRED_MODELS", ["gemma3:4b"]) + + result = await heartbeat.HeartbeatReportService()._probe_ollama() + + assert result["probe"].ok is True + assert result["probe"].status == "✅ 備援中 (GCP-B)" + assert result["active_endpoint"] == "GCP-B" + assert result["models"] == {"gemma3:4b": True} + assert result["endpoints"]["GCP-A"].ok is False def test_report_to_telegram_html_renders_ollama_endpoint_statuses() -> None: @@ -67,6 +85,7 @@ def test_report_to_telegram_html_renders_ollama_endpoint_statuses() -> None: "GCP-B": heartbeat.ProbeResult(True, "✅ 正常", 1100), "111": heartbeat.ProbeResult(False, "❌ connect failed"), } + report.ollama_active_endpoint = "GCP-A" report.warnings = heartbeat.HeartbeatReportService()._build_warnings(report) text = heartbeat.report_to_telegram_html(report) @@ -74,4 +93,6 @@ def test_report_to_telegram_html_renders_ollama_endpoint_statuses() -> None: assert "GCP-A: ✅ 正常" in text assert "GCP-B: ✅ 正常" in text assert "111: ❌ connect failed" in text - assert "Ollama 111 異常" in "\n".join(report.warnings) + assert "Ollama 111 降級" in "\n".join(report.warnings) + assert "GCP-A 備援執行中" in "\n".join(report.warnings) + assert "ollama 服務異常" not in "\n".join(report.warnings) diff --git a/apps/api/tests/test_ollama_endpoint_resolver.py b/apps/api/tests/test_ollama_endpoint_resolver.py index aba43a979..91176bc17 100644 --- a/apps/api/tests/test_ollama_endpoint_resolver.py +++ b/apps/api/tests/test_ollama_endpoint_resolver.py @@ -47,7 +47,7 @@ def test_all_workloads_prefer_gcp_a_lane() -> None: assert selection.reason == "global_primary_gcp_a" -def test_all_workloads_share_global_ollama_order() -> None: +def test_general_workloads_share_global_ollama_order() -> None: cfg = _settings() for workload in ("interactive", "deep_rca", "local_required", "privacy_sensitive", "dr"): @@ -64,6 +64,23 @@ def test_all_workloads_share_global_ollama_order() -> None: ] +def test_alert_fast_prefers_local_111_before_gcp_b() -> None: + order = resolve_ollama_order("alert_fast", config=_settings()) + + assert [selection.provider_name for selection in order] == [ + "ollama_gcp_a", + "ollama_local", + "ollama_gcp_b", + ] + + +def test_alert_fast_uses_local_when_gcp_a_is_not_configured() -> None: + selection = resolve_ollama_selection("alert_fast", config=_settings(primary="")) + + assert selection.provider_name == "ollama_local" + assert selection.url == "http://192.168.0.110:11437" + + def test_non_sensitive_workloads_fall_back_to_gcp_b_when_primary_missing() -> None: cfg = _settings(primary="") diff --git a/apps/api/tests/test_ollama_failover_manager.py b/apps/api/tests/test_ollama_failover_manager.py index 5c52c248e..240eb5290 100644 --- a/apps/api/tests/test_ollama_failover_manager.py +++ b/apps/api/tests/test_ollama_failover_manager.py @@ -84,7 +84,7 @@ def _offline_health(url: str = URL_GCP_A) -> HealthReport: class TestDecideRoute: - """_decide_route 路由邏輯純函數測試（ADR-110 三層容災：GCP-A → GCP-B → Local → Gemini）""" + """_decide_route 路由邏輯純函數測試。""" def _setup(self) -> OllamaFailoverManager: return _make_manager() @@ -237,6 +237,26 @@ class TestDecideRoute: assert "nemotron" in provider_names assert "claude" in provider_names + def test_alert_fast_prefers_healthy_local_before_gcp_b(self): + manager = self._setup() + + result = manager._decide_route( + health_gcp_a=_make_health(HealthStatus.OFFLINE, URL_GCP_A), + health_gcp_b=_make_health(HealthStatus.HEALTHY, URL_GCP_B), + health_local=_make_health(HealthStatus.HEALTHY, URL_LOCAL), + url_gcp_a=URL_GCP_A, + url_gcp_b=URL_GCP_B, + url_local=URL_LOCAL, + task_type="alert_fast", + ) + + assert result.primary.provider_name == "ollama_local" + assert result.primary.url == URL_LOCAL + assert [endpoint.provider_name for endpoint in result.fallback_chain] == [ + "ollama_gcp_b", + "gemini", + ] + # ------------------------------------------------------------------ # routing_reason 記錄 # ------------------------------------------------------------------ diff --git a/apps/api/tests/test_openclaw_alert_cloud_fallback_gate.py b/apps/api/tests/test_openclaw_alert_cloud_fallback_gate.py index 5fddb2af6..d73bb8384 100644 --- a/apps/api/tests/test_openclaw_alert_cloud_fallback_gate.py +++ b/apps/api/tests/test_openclaw_alert_cloud_fallback_gate.py @@ -53,9 +53,14 @@ class _UnorderedFailoverManager: class _FakeRouter: + def __init__(self) -> None: + self.contexts: list[dict[str, Any]] = [] + async def route(self, prompt: str, context: dict[str, Any]) -> SimpleNamespace: + self.contexts.append(context) return SimpleNamespace( selected_provider=AIProviderEnum.GEMINI, + selected_model="gemini-cloud-only-model", fallback_chain=[ (AIProviderEnum.OPENCLAW_NEMO, "nvidia"), (AIProviderEnum.CLAUDE, "claude"), @@ -69,6 +74,7 @@ class _FakeRouter: class _FakeExecutor: def __init__(self) -> None: self.provider_order: list[str] | None = None + self.context: dict[str, Any] | None = None async def execute( self, @@ -80,6 +86,7 @@ class _FakeExecutor: require_local: bool, ) -> SimpleNamespace: self.provider_order = provider_order + self.context = context return SimpleNamespace( raw_response='{"root_cause":"ok","suggested_action":"NO_ACTION"}', provider=provider_order[0], @@ -96,6 +103,7 @@ async def test_alert_context_uses_ollama_lane_then_openclaw_nemo_before_gemini( ) -> None: fake_executor = _FakeExecutor() fake_failover = _FakeFailoverManager() + fake_router = _FakeRouter() monkeypatch.setattr(openclaw_module.settings, "USE_AI_ROUTER", True) monkeypatch.setattr(openclaw_module.settings, "MOCK_MODE", False) @@ -104,7 +112,8 @@ async def test_alert_context_uses_ollama_lane_then_openclaw_nemo_before_gemini( 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(openclaw_module.settings, "OLLAMA_HEALTH_CHECK_MODEL", "gemma3:4b") + monkeypatch.setattr(ai_router_module, "get_ai_router", lambda: fake_router) monkeypatch.setattr(ai_router_module, "get_ai_executor", lambda: fake_executor) monkeypatch.setattr(openclaw_module, "get_ollama_failover_manager", lambda: fake_failover) @@ -127,12 +136,18 @@ async def test_alert_context_uses_ollama_lane_then_openclaw_nemo_before_gemini( ) assert fake_executor.provider_order == [ "ollama_gcp_a", - "ollama_gcp_b", "ollama_local", + "ollama_gcp_b", "openclaw_nemo", "gemini", ] - assert fake_failover.task_types == ["diagnose"] + assert fake_failover.task_types == ["alert_fast"] + assert fake_router.contexts[0]["intent_hint"] == "alert_triage" + assert fake_executor.context is not None + assert fake_executor.context["task_type"] == "alert_fast" + assert fake_executor.context["ollama_model"] == "gemma3:4b" + assert fake_executor.context["allow_gcp_heavy_model"] is False + assert fake_executor.context["provider_timeout_seconds"] == 18.0 @pytest.mark.asyncio @@ -159,7 +174,7 @@ async def test_alert_context_can_disable_cloud_backup_for_cost_stop( alert_context={"incident_id": "INC-1", "alertname": "HostHighCpuLoad"}, ) - assert fake_executor.provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local"] + assert fake_executor.provider_order == ["ollama_gcp_a", "ollama_local", "ollama_gcp_b"] @pytest.mark.asyncio @@ -225,7 +240,7 @@ async def test_explicit_ai_governance_context_uses_ollama_first( @pytest.mark.asyncio -async def test_alert_context_uses_gcp_a_gcp_b_then_111_order( +async def test_alert_context_uses_gcp_a_111_then_gcp_b_order( monkeypatch: pytest.MonkeyPatch, ) -> None: fake_failover = _FakeFailoverManager() @@ -236,11 +251,12 @@ async def test_alert_context_uses_gcp_a_gcp_b_then_111_order( service = object.__new__(OpenClawService) provider_order = await service._resolve_alert_provider_order( - task_type="diagnose", + task_type="alert_fast", alert_context={"incident_id": "INC-1", "alertname": "HostHighCpuLoad"}, ) - assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local"] + assert provider_order == ["ollama_gcp_a", "ollama_local", "ollama_gcp_b"] + assert fake_failover.task_types == ["alert_fast"] @pytest.mark.asyncio @@ -257,7 +273,7 @@ async def test_alert_context_sorts_ollama_lane_and_drops_cloud_providers( alert_context={"incident_id": "INC-1", "alertname": "HostHighCpuLoad"}, ) - assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local"] + assert provider_order == ["ollama_gcp_a", "ollama_local", "ollama_gcp_b"] @pytest.mark.asyncio @@ -276,7 +292,7 @@ async def test_alert_context_sorts_ollama_lane_before_openclaw_nemo_backup( cloud_provider_order=["claude", "openclaw_nemo", "gemini", "ollama"], ) - assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local", "openclaw_nemo", "gemini"] + assert provider_order == ["ollama_gcp_a", "ollama_local", "ollama_gcp_b", "openclaw_nemo", "gemini"] @pytest.mark.asyncio @@ -295,4 +311,4 @@ async def test_alert_context_respects_disabled_openclaw_nemo_backup( cloud_provider_order=["openclaw_nemo", "gemini"], ) - assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local", "gemini"] + assert provider_order == ["ollama_gcp_a", "ollama_local", "ollama_gcp_b", "gemini"] diff --git a/apps/api/tests/test_signoz_fail_closed.py b/apps/api/tests/test_signoz_fail_closed.py new file mode 100644 index 000000000..96fd2dea6 --- /dev/null +++ b/apps/api/tests/test_signoz_fail_closed.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from src.services import signoz_client as signoz_module +from src.services.openclaw import OpenClawService +from src.services.signoz_client import SignOzClient + + +@pytest.mark.asyncio +async def test_clickhouse_query_can_fail_closed_on_transport_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = SignOzClient() + http_client = SimpleNamespace( + post=AsyncMock(side_effect=RuntimeError("connection refused")) + ) + monkeypatch.setattr( + signoz_module, + "get_clickhouse_client", + AsyncMock(return_value=http_client), + ) + + assert await client._query_clickhouse("SELECT 1") == [] + with pytest.raises(RuntimeError, match="connection refused"): + await client._query_clickhouse("SELECT 1", fail_closed=True) + + +@pytest.mark.asyncio +async def test_gold_metrics_requires_successful_clickhouse_queries( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = SignOzClient() + query = AsyncMock(side_effect=RuntimeError("clickhouse unavailable")) + monkeypatch.setattr(client, "_query_clickhouse", query) + + with pytest.raises(RuntimeError, match="clickhouse unavailable"): + await client.get_gold_metrics("node-exporter-110") + + assert query.await_args.kwargs["fail_closed"] is True + + +@pytest.mark.asyncio +async def test_openclaw_marks_signoz_unavailable_when_metrics_query_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = object.__new__(OpenClawService) + service._signoz = SimpleNamespace( + get_gold_metrics=AsyncMock(side_effect=RuntimeError("clickhouse unavailable")), + generate_trace_url=lambda **_kwargs: "https://unused.example/traces", + ) + monkeypatch.setattr(signoz_module.settings, "SIGNOZ_URL", "https://signoz.example") + + metrics, trace_url = await service.get_signoz_context("node-exporter-110") + + assert metrics is None + assert trace_url == "https://signoz.example/traces?service=node-exporter-110"