diff --git a/apps/api/src/services/ollama_auto_recovery.py b/apps/api/src/services/ollama_auto_recovery.py index 82deb2bc0..63a3ea5d5 100644 --- a/apps/api/src/services/ollama_auto_recovery.py +++ b/apps/api/src/services/ollama_auto_recovery.py @@ -31,6 +31,7 @@ from __future__ import annotations import asyncio import datetime +import time from datetime import timedelta, timezone from typing import Any, Protocol, runtime_checkable @@ -117,6 +118,11 @@ class OllamaAutoRecoveryService: # 修法:判斷改用 _is_ollama_primary(),比對 startswith("ollama_") self._current_primary: str = "ollama_gcp_a" # fallback_manager 傳入的 provider_name self._consecutive_healthy: int = 0 + self._state_lock = asyncio.Lock() + self._provider_transition_epoch: int = 0 + self._provider_transition_completed_epoch: int = 0 + self._last_observation_generation: int = 0 + self._last_preferred_recovery_completed_at_monotonic: float = 0.0 self._task: asyncio.Task | None = None # ------------------------------------------------------------------------- @@ -151,8 +157,9 @@ class OllamaAutoRecoveryService: current_primary=self._current_primary, ) - # 若 primary 不是 Ollama,立刻評估一次(不等 30s interval) - if not self._is_ollama_primary(self._current_primary): + # 若目前不在 preferred GCP-A lane,立刻評估一次(不等 30s interval)。 + # GCP-B / Local 雖是 Ollama,仍是 degraded fallback,必須持續追蹤 GCP-A。 + if not self._is_preferred_primary(self._current_primary): try: await self._check_and_recover() except Exception: @@ -199,35 +206,81 @@ class OllamaAutoRecoveryService: # 狀態控制(供整合層設定當前 primary,例如 failover 觸發時更新) # ------------------------------------------------------------------------- - async def set_current_primary(self, provider: str) -> None: + async def set_current_primary( + self, + provider: str, + *, + observation_generation: int | None = None, + observation_started_at_monotonic: float | None = None, + ) -> bool: """ 通知 recovery service 當前路由 primary 是哪個 provider。 OllamaFailoverManager.select_provider() 觸發 failover 後呼叫此方法更新狀態。 + 回傳 True 代表 observation 已接受(含 provider 未變);generation 或 + recovery barrier 判定為 stale 時回傳 False,讓 manager 抑制舊告警。 # 2026-04-25 critic-fix Part2 H5+H6 by Claude Engineer-C2 # 改為 async,切換時同步寫 Redis 持久化(跨重啟恢復) """ - if self._current_primary != provider: - logger.info( - "ollama_auto_recovery_primary_changed", - service="ollama_auto_recovery", - from_primary=self._current_primary, - to_primary=provider, - ) - self._current_primary = provider - # Redis 持久化(跨重啟恢復) - await self._persist_primary(provider) + async with self._state_lock: + if observation_generation is not None: + if observation_generation <= self._last_observation_generation: + logger.info( + "ollama_auto_recovery_stale_generation_ignored", + service="ollama_auto_recovery", + provider=provider, + observation_generation=observation_generation, + last_observation_generation=( + self._last_observation_generation + ), + ) + return False + self._last_observation_generation = observation_generation - # 2026-05-04 ogt: B5/B6 修復 — 判斷改用 _is_ollama_primary() - # 原設計:provider != "ollama",但 callback 傳 "ollama_gcp_a" → 永遠觸發 tracking - if not self._is_ollama_primary(provider): - # 切換到非 Ollama(gemini/nemotron/claude)→ 重置 counter,開始監控恢復 + if ( + observation_started_at_monotonic is not None + and observation_started_at_monotonic + <= self._last_preferred_recovery_completed_at_monotonic + ): + logger.info( + "ollama_auto_recovery_stale_observation_ignored", + service="ollama_auto_recovery", + provider=provider, + observation_generation=observation_generation, + ) + return False + + await self._set_current_primary_locked(provider) + return True + + async def _set_current_primary_locked(self, provider: str) -> bool: + """Apply one serialized provider transition and persist its exact lane.""" + if self._current_primary == provider: + return False + logger.info( + "ollama_auto_recovery_primary_changed", + service="ollama_auto_recovery", + from_primary=self._current_primary, + to_primary=provider, + ) + self._current_primary = provider + self._provider_transition_epoch += 1 + transition_epoch = self._provider_transition_epoch + if not self._is_preferred_primary(provider): + # 新切到 fallback lane 時重置 counter;相同 lane 的重複 + # observation 不得抹掉已累積的 GCP-A stable window。 self._consecutive_healthy = 0 + try: + await self._persist_primary(provider) + finally: + self._provider_transition_completed_epoch = transition_epoch + if not self._is_preferred_primary(provider): logger.info( "ollama_auto_recovery_tracking_started", service="ollama_auto_recovery", current_primary=provider, ) + return True # ------------------------------------------------------------------------- # Redis 持久化(跨重啟恢復) @@ -249,8 +302,10 @@ class OllamaAutoRecoveryService: try: from src.core.redis_client import get_redis redis = get_redis() - await redis.set(self._REDIS_PRIMARY_KEY, primary) - await redis.set(self._REDIS_PRIMARY_KEY_LEGACY, primary) # Phase A 雙寫 + await redis.mset({ + self._REDIS_PRIMARY_KEY: primary, + self._REDIS_PRIMARY_KEY_LEGACY: primary, + }) except Exception as e: logger.warning( "ollama_auto_recovery_persist_failed", @@ -308,6 +363,11 @@ class OllamaAutoRecoveryService: """Return true for both legacy "ollama" and ADR-110 provider names.""" return provider == "ollama" or provider.startswith("ollama_") + @staticmethod + def _is_preferred_primary(provider: str) -> bool: + """Return whether routing already points at the canonical GCP-A lane.""" + return provider in {"ollama", "ollama_gcp_a"} + # ------------------------------------------------------------------------- # 背景監控迴圈 # ------------------------------------------------------------------------- @@ -336,6 +396,27 @@ class OllamaAutoRecoveryService: 抽取為獨立方法,方便單元測試。 """ host = self._settings.OLLAMA_URL + probe_invoked_epoch = self._provider_transition_epoch + probe_invoked_completed_epoch = ( + self._provider_transition_completed_epoch + ) + + async with self._state_lock: + if ( + probe_invoked_epoch != probe_invoked_completed_epoch + or self._provider_transition_epoch != probe_invoked_epoch + ): + logger.info( + "ollama_auto_recovery_transition_probe_ignored", + service="ollama_auto_recovery", + probe_invoked_epoch=probe_invoked_epoch, + probe_invoked_completed_epoch=( + probe_invoked_completed_epoch + ), + current_epoch=self._provider_transition_epoch, + ) + return + probe_epoch = self._provider_transition_epoch try: health = await self._monitor.check(host) @@ -345,45 +426,72 @@ class OllamaAutoRecoveryService: service="ollama_auto_recovery", host=host, ) - self._consecutive_healthy = 0 + async with self._state_lock: + if self._provider_transition_epoch == probe_epoch: + self._consecutive_healthy = 0 return # 2026-05-04 ogt: SLOW 視為可用(GCP 高負載 10-30s 仍優於 Gemini quota 限制) # 原設計只接受 HEALTHY,GCP 若因負載偏高落入 SLOW 區會永久卡在 Gemini is_usable = health.status in (HealthStatus.HEALTHY, HealthStatus.SLOW) - if is_usable: - self._consecutive_healthy += 1 - logger.debug( - "ollama_auto_recovery_healthy_tick", - service="ollama_auto_recovery", - consecutive=self._consecutive_healthy, - required=self._stable_count_required, - current_primary=self._current_primary, - actual_status=health.status.value, - ) - - if ( - self._consecutive_healthy >= self._stable_count_required - and not self._is_ollama_primary(self._current_primary) - ): - await self._switch_back_to_ollama() - else: - # DEGRADED / OFFLINE → counter 歸零,繼續等 - if self._consecutive_healthy > 0: - logger.debug( - "ollama_auto_recovery_counter_reset", + should_recover = False + async with self._state_lock: + if self._provider_transition_epoch != probe_epoch: + logger.info( + "ollama_auto_recovery_stale_health_probe_ignored", service="ollama_auto_recovery", - previous_count=self._consecutive_healthy, - status=health.status.value, + probe_epoch=probe_epoch, + current_epoch=self._provider_transition_epoch, + current_primary=self._current_primary, ) - self._consecutive_healthy = 0 + return + + if self._is_preferred_primary(self._current_primary): + self._consecutive_healthy = 0 + return + + if is_usable: + self._consecutive_healthy += 1 + logger.debug( + "ollama_auto_recovery_healthy_tick", + service="ollama_auto_recovery", + consecutive=self._consecutive_healthy, + required=self._stable_count_required, + current_primary=self._current_primary, + actual_status=health.status.value, + ) + + should_recover = ( + self._consecutive_healthy + >= self._stable_count_required + and not self._is_preferred_primary(self._current_primary) + ) + else: + # DEGRADED / OFFLINE → counter 歸零,繼續等 + if self._consecutive_healthy > 0: + logger.debug( + "ollama_auto_recovery_counter_reset", + service="ollama_auto_recovery", + previous_count=self._consecutive_healthy, + status=health.status.value, + ) + self._consecutive_healthy = 0 + + if should_recover: + await self._switch_back_to_ollama( + expected_provider_epoch=probe_epoch, + ) # ------------------------------------------------------------------------- # 切回 Ollama 111 # ------------------------------------------------------------------------- - async def _switch_back_to_ollama(self) -> None: + async def _switch_back_to_ollama( + self, + *, + expected_provider_epoch: int | None = None, + ) -> None: """ 切回 Ollama 111: 1. 更新 _current_primary @@ -392,34 +500,79 @@ class OllamaAutoRecoveryService: 4. Telegram 通知(若 alerter 已設定) 5. structlog audit """ - from_provider = self._current_primary - self._current_primary = "ollama" - stable_count = self._consecutive_healthy - # counter 不立即清零:保留供 audit,下次 loop 開始後自然計數 - self._consecutive_healthy = 0 + async with self._state_lock: + if ( + expected_provider_epoch is not None + and self._provider_transition_epoch + != expected_provider_epoch + ): + logger.info( + "ollama_auto_recovery_stale_switch_ignored", + service="ollama_auto_recovery", + expected_provider_epoch=expected_provider_epoch, + current_epoch=self._provider_transition_epoch, + current_primary=self._current_primary, + ) + return + if self._is_preferred_primary(self._current_primary): + self._consecutive_healthy = 0 + return + if self._consecutive_healthy < self._stable_count_required: + logger.debug( + "ollama_auto_recovery_debounce_pending", + service="ollama_auto_recovery", + consecutive=self._consecutive_healthy, + required=self._stable_count_required, + current_primary=self._current_primary, + ) + return + from_provider = self._current_primary + stable_count = self._consecutive_healthy + self._consecutive_healthy = 0 + await self._set_current_primary_locked("ollama_gcp_a") - # 取台北時間(標準庫保證 +8,禁 UTC) - # 2026-04-25 critic-fix Part2 B4 by Claude Engineer-C2 - recovery_time = datetime.datetime.now(TAIPEI_TZ) + # 取台北時間(標準庫保證 +8,禁 UTC) + # 2026-04-25 critic-fix Part2 B4 by Claude Engineer-C2 + recovery_time = datetime.datetime.now(TAIPEI_TZ) - # 清空 failover manager 快取(讓下次 select_provider 重新評估) - try: - await self._failover_manager.clear_cache() - except Exception: - logger.exception( - "ollama_auto_recovery_clear_cache_error", - service="ollama_auto_recovery", + # 清空 failover manager 快取(讓下次 select_provider 重新評估) + try: + await self._failover_manager.clear_cache() + except Exception: + logger.exception( + "ollama_auto_recovery_clear_cache_error", + service="ollama_auto_recovery", + ) + + # 通知 failover_manager + try: + self._failover_manager.notify_recovery("ollama_gcp_a") + except Exception: + logger.exception( + "ollama_auto_recovery_notify_error", + service="ollama_auto_recovery", + ) + self._last_preferred_recovery_completed_at_monotonic = ( + time.monotonic() + ) + await self._emit_recovery_side_effects_locked( + from_provider=from_provider, + stable_count=stable_count, + recovery_time=recovery_time, ) - # 通知 failover_manager - try: - # 2026-05-03 ogt: ADR-110 — provider name 改為 "ollama_gcp_a"(GCP-A Primary) - self._failover_manager.notify_recovery("ollama_gcp_a") - except Exception: - logger.exception( - "ollama_auto_recovery_notify_error", - service="ollama_auto_recovery", - ) + async def _emit_recovery_side_effects_locked( + self, + *, + from_provider: str, + stable_count: int, + recovery_time: datetime.datetime, + ) -> None: + """Emit one serialized recovery notification, metric, and audit event. + + The caller holds ``_state_lock`` so a newer fallback cannot become + canonical while this recovery lifecycle notification is in flight. + """ # Telegram 通知(P1.5 Engineer-D 接入) # 2026-04-25 P1.5 by Claude Engineer-D — 接 FailoverAlerter,self._alerter 優先, @@ -442,7 +595,7 @@ class OllamaAutoRecoveryService: _provider_label = "Ollama" await alerter.alert_recovery({ "from": from_provider, - "to": "ollama", + "to": "ollama_gcp_a", "recovered_host": f"{_provider_label} {_recovered_url}", "stable_count": stable_count, "recovery_time": recovery_time.isoformat(), diff --git a/apps/api/src/services/ollama_failover_manager.py b/apps/api/src/services/ollama_failover_manager.py index ada2d6d2e..df3baa139 100644 --- a/apps/api/src/services/ollama_failover_manager.py +++ b/apps/api/src/services/ollama_failover_manager.py @@ -31,6 +31,8 @@ from __future__ import annotations import asyncio import datetime +import inspect +import time from dataclasses import dataclass from datetime import timedelta, timezone @@ -188,7 +190,11 @@ class OllamaFailoverManager: # recovery_callback: async callable(provider_name: str) → None # OllamaAutoRecoveryService.set_current_primary 在 failover 時被通知, # 避免重啟後 _current_primary 停留在 "ollama" 而永不啟動恢復監控 - self._recovery_callback = recovery_callback + self._recovery_callback = None + self._recovery_callback_accepts_observation_metadata = False + self.set_recovery_callback(recovery_callback) + self._recovery_observation_generation = 0 + self._recovery_side_effect_lock = asyncio.Lock() # ------------------------------------------------------------------------- # Public API @@ -218,6 +224,15 @@ class OllamaFailoverManager: Returns: OllamaRoutingResult """ + recovery_observation_generation: int | None = None + recovery_observation_started_at_monotonic: float | None = None + if emit_side_effects: + self._recovery_observation_generation += 1 + recovery_observation_generation = ( + self._recovery_observation_generation + ) + recovery_observation_started_at_monotonic = time.monotonic() + # 2026-05-04 ogt: 改用語意中性名稱 primary/secondary/tertiary, # 避免 gcp_a/gcp_b/local 與實際 URL 脫鉤造成 log 誤導 url_primary = self._settings.OLLAMA_URL # canonical GCP-A endpoint @@ -284,12 +299,6 @@ class OllamaFailoverManager: # Provider selection never consumes paid quota. Claude/Gemini providers # each reserve requests/tokens/cost atomically immediately before calls. - # Status/readback probes must not emit failover metrics, Telegram, or - # recovery-state writes. Production action callers retain the existing - # behavior and must own the explicit side-effect boundary. - if emit_side_effects: - await self._write_failover_audit(result) - def _status(report: HealthReport | None) -> str: return report.status.value if report else "not_checked" @@ -310,22 +319,14 @@ class OllamaFailoverManager: side_effects_emitted=emit_side_effects, ) - # Recovery tracks the first health-observed fallback candidate, while - # the selected execution primary remains fixed at GCP-A. - # 2026-04-25 critic-fix Part2 H5+H6 by Claude Engineer-C2 - if emit_side_effects and self._recovery_callback is not None: - try: - recovery_observation = ( - result.observed_fallback.provider_name - if result.observed_fallback - else result.primary.provider_name - ) - await self._recovery_callback(recovery_observation) - except Exception as e: - logger.warning( - "ollama_failover_recovery_callback_failed", - error=str(e), - ) + if emit_side_effects: + await self._emit_recovery_observation_side_effects( + result, + observation_generation=recovery_observation_generation, + observation_started_at_monotonic=( + recovery_observation_started_at_monotonic + ), + ) return result @@ -533,14 +534,108 @@ class OllamaFailoverManager: # Recovery API(供 OllamaAutoRecoveryService 呼叫) # ------------------------------------------------------------------------- + async def _emit_recovery_observation_side_effects( + self, + result: OllamaRoutingResult, + *, + observation_generation: int | None, + observation_started_at_monotonic: float | None, + ) -> None: + """Serialize one accepted provider observation and its alert lifecycle.""" + if ( + observation_generation is None + or observation_started_at_monotonic is None + ): + logger.warning( + "ollama_failover_observation_metadata_missing", + service="ollama_failover", + ) + return + + async with self._recovery_side_effect_lock: + if observation_generation != self._recovery_observation_generation: + logger.info( + "ollama_failover_stale_observation_side_effects_ignored", + service="ollama_failover", + observation_generation=observation_generation, + latest_generation=self._recovery_observation_generation, + ) + return + + # Recovery tracks the first health-observed fallback candidate, + # while the selected execution primary remains fixed at GCP-A. + recovery_callback = self._recovery_callback + callback_accepted = True + if recovery_callback is not None: + recovery_observation = ( + result.observed_fallback.provider_name + if result.observed_fallback + else result.primary.provider_name + ) + try: + if self._recovery_callback_accepts_observation_metadata: + callback_result = await recovery_callback( + recovery_observation, + observation_generation=observation_generation, + observation_started_at_monotonic=( + observation_started_at_monotonic + ), + ) + else: + callback_result = await recovery_callback( + recovery_observation + ) + callback_accepted = callback_result is not False + except Exception as e: + callback_accepted = False + logger.warning( + "ollama_failover_recovery_callback_failed", + error=str(e), + ) + + if not callback_accepted: + logger.info( + "ollama_failover_rejected_observation_alert_suppressed", + service="ollama_failover", + observation_generation=observation_generation, + ) + return + + # State acceptance is ordered before recipient-visible degraded + # audit/Telegram, and concurrent selections cannot overtake it. + await self._write_failover_audit(result) + def set_recovery_callback(self, callback) -> None: """ 設定 recovery callback(供 lifespan wiring 使用)。 - callback signature: async (provider_name: str) -> None + Legacy callback signature remains async (provider_name: str) -> None. + Metadata-capable callbacks may additionally declare both + observation_generation and observation_started_at_monotonic keyword + parameters, or accept arbitrary keyword arguments. Returning literal + False rejects the observation and suppresses its degraded alert; + legacy None remains accepted. # 2026-04-25 P1.2 by Claude Engineer-A2 — failover 整合到 ai_router + lifespan """ self._recovery_callback = callback + self._recovery_callback_accepts_observation_metadata = False + if callback is None: + return + try: + parameters = inspect.signature(callback).parameters.values() + except (TypeError, ValueError): + return + parameter_names = {parameter.name for parameter in parameters} + self._recovery_callback_accepts_observation_metadata = ( + { + "observation_generation", + "observation_started_at_monotonic", + }.issubset(parameter_names) + or any( + parameter.kind is inspect.Parameter.VAR_KEYWORD + for parameter in parameters + ) + ) async def clear_cache(self) -> None: """ diff --git a/apps/api/tests/test_ollama_auto_recovery.py b/apps/api/tests/test_ollama_auto_recovery.py index 49ad649a4..2ba5101da 100644 --- a/apps/api/tests/test_ollama_auto_recovery.py +++ b/apps/api/tests/test_ollama_auto_recovery.py @@ -22,7 +22,8 @@ OllamaAutoRecoveryService 單元測試 - P1.1d from __future__ import annotations import asyncio -from unittest.mock import AsyncMock, MagicMock, call, patch +import time +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -147,6 +148,25 @@ class TestCheckAndRecover: mock_switch.assert_awaited_once() + @pytest.mark.asyncio + async def test_gcp_b_fallback_tracks_gcp_a_recovery(self): + """GCP-B fallback must not disable the preferred GCP-A recovery probe.""" + svc, mock_monitor, _ = _make_service( + current_primary="ollama_gcp_b", + stable_count=3, + ) + mock_monitor.check.return_value = _make_health(HealthStatus.HEALTHY) + svc._consecutive_healthy = 2 + + with patch.object( + svc, + "_switch_back_to_ollama", + new_callable=AsyncMock, + ) as mock_switch: + await svc._check_and_recover() + + mock_switch.assert_awaited_once() + @pytest.mark.asyncio async def test_two_healthy_not_yet_switch(self): """連續 2 次 HEALTHY(未達 3 次門檻)→ 不觸發切回""" @@ -235,8 +255,8 @@ class TestDebounce: for _ in range(3): await svc._check_and_recover() - # 觸發後 current_primary 應切回 ollama - assert svc.current_primary == "ollama" + # 觸發後 current_primary 應切回 canonical GCP-A lane + assert svc.current_primary == "ollama_gcp_a" mock_failover.clear_cache.assert_awaited_once() # 2026-05-03 ogt: ADR-110 — notify_recovery 改為 "ollama_gcp_a"(GCP-A Primary) mock_failover.notify_recovery.assert_called_once_with("ollama_gcp_a") @@ -251,20 +271,38 @@ class TestSwitchBackToOllama: """_switch_back_to_ollama 切回邏輯""" @pytest.mark.asyncio - async def test_sets_current_primary_to_ollama(self): + async def test_sets_current_primary_to_gcp_a(self): svc, _, _ = _make_service(current_primary="gemini") + svc._consecutive_healthy = svc._stable_count_required await svc._switch_back_to_ollama() - assert svc.current_primary == "ollama" + assert svc.current_primary == "ollama_gcp_a" + + @pytest.mark.asyncio + async def test_persists_exact_gcp_a_identity_after_recovery(self): + svc, _, _ = _make_service(current_primary="ollama_gcp_b") + svc._consecutive_healthy = svc._stable_count_required + + with patch.object( + svc, + "_persist_primary", + new_callable=AsyncMock, + ) as persist_primary: + await svc._switch_back_to_ollama() + + assert svc.current_primary == "ollama_gcp_a" + persist_primary.assert_awaited_once_with("ollama_gcp_a") @pytest.mark.asyncio async def test_calls_clear_cache(self): svc, _, mock_failover = _make_service(current_primary="gemini") + svc._consecutive_healthy = svc._stable_count_required await svc._switch_back_to_ollama() mock_failover.clear_cache.assert_awaited_once() @pytest.mark.asyncio async def test_calls_notify_recovery(self): svc, _, mock_failover = _make_service(current_primary="gemini") + svc._consecutive_healthy = svc._stable_count_required await svc._switch_back_to_ollama() # 2026-05-03 ogt: ADR-110 — notify_recovery 改為 "ollama_gcp_a"(GCP-A Primary) mock_failover.notify_recovery.assert_called_once_with("ollama_gcp_a") @@ -282,28 +320,30 @@ class TestSwitchBackToOllama: mock_alerter.alert_recovery.assert_awaited_once() payload = mock_alerter.alert_recovery.call_args[0][0] assert payload["from"] == "gemini" - assert payload["to"] == "ollama" # 2026-05-03 ogt: ADR-110 — "to" 改為 "ollama"(不再指定 111) + assert payload["to"] == "ollama_gcp_a" assert payload["stable_count"] == 3 @pytest.mark.asyncio async def test_no_alerter_does_not_crash(self): """alerter = None → 不 crash""" svc, _, _ = _make_service(current_primary="gemini", alerter=None) + svc._consecutive_healthy = svc._stable_count_required # 不應 raise await svc._switch_back_to_ollama() - assert svc.current_primary == "ollama" + assert svc.current_primary == "ollama_gcp_a" @pytest.mark.asyncio async def test_clear_cache_failure_does_not_crash(self): """clear_cache 失敗 → 靜默繼續,不 crash""" svc, _, mock_failover = _make_service(current_primary="gemini") mock_failover.clear_cache.side_effect = RuntimeError("Redis down") + svc._consecutive_healthy = svc._stable_count_required # 不應 raise await svc._switch_back_to_ollama() # 儘管 clear_cache 失敗,current_primary 應已更新 - assert svc.current_primary == "ollama" + assert svc.current_primary == "ollama_gcp_a" @pytest.mark.asyncio async def test_alerter_failure_does_not_crash(self): @@ -311,11 +351,83 @@ class TestSwitchBackToOllama: mock_alerter = AsyncMock() mock_alerter.alert_recovery = AsyncMock(side_effect=RuntimeError("TG timeout")) svc, _, _ = _make_service(current_primary="gemini", alerter=mock_alerter) + svc._consecutive_healthy = svc._stable_count_required # 不應 raise await svc._switch_back_to_ollama() - assert svc.current_primary == "ollama" + assert svc.current_primary == "ollama_gcp_a" + + @pytest.mark.asyncio + async def test_does_not_switch_before_stability_threshold(self): + mock_alerter = AsyncMock() + svc, _, mock_failover = _make_service( + current_primary="ollama_gcp_b", + stable_count=3, + alerter=mock_alerter, + ) + svc._consecutive_healthy = 2 + + await svc._switch_back_to_ollama() + + assert svc.current_primary == "ollama_gcp_b" + assert svc.consecutive_healthy == 2 + mock_failover.clear_cache.assert_not_awaited() + mock_failover.notify_recovery.assert_not_called() + mock_alerter.alert_recovery.assert_not_awaited() + + @pytest.mark.asyncio + async def test_orders_newer_fallback_after_recovery_notification(self): + recovery_alert_started = asyncio.Event() + release_recovery_alert = asyncio.Event() + observed_primary: list[tuple[str, str]] = [] + + async def blocking_recovery_alert(payload): + observed_primary.append(("started", svc.current_primary)) + recovery_alert_started.set() + await release_recovery_alert.wait() + observed_primary.append(("completed", svc.current_primary)) + + mock_alerter = AsyncMock() + mock_alerter.alert_recovery = AsyncMock( + side_effect=blocking_recovery_alert, + ) + svc, _, _ = _make_service( + current_primary="ollama_gcp_b", + stable_count=3, + alerter=mock_alerter, + ) + svc._consecutive_healthy = svc._stable_count_required + + with patch.object( + svc, + "_persist_primary", + new_callable=AsyncMock, + ): + recovery_task = asyncio.create_task( + svc._switch_back_to_ollama() + ) + await recovery_alert_started.wait() + fallback_task = asyncio.create_task( + svc.set_current_primary( + "ollama_local", + observation_generation=1, + observation_started_at_monotonic=time.monotonic(), + ) + ) + await asyncio.sleep(0) + fallback_waited_for_recovery = fallback_task.done() is False + primary_during_recovery_alert = svc.current_primary + release_recovery_alert.set() + await asyncio.gather(recovery_task, fallback_task) + + assert fallback_waited_for_recovery is True + assert primary_during_recovery_alert == "ollama_gcp_a" + assert observed_primary == [ + ("started", "ollama_gcp_a"), + ("completed", "ollama_gcp_a"), + ] + assert svc.current_primary == "ollama_local" # ============================================================================= @@ -423,6 +535,162 @@ class TestStateManagement: assert svc.consecutive_healthy == 0 + @pytest.mark.asyncio + async def test_set_current_primary_to_gcp_b_resets_recovery_counter(self): + svc, _, _ = _make_service(current_primary="ollama_gcp_a") + svc._consecutive_healthy = 5 + + with patch("src.core.redis_client.get_redis", side_effect=RuntimeError("no redis")): + await svc.set_current_primary("ollama_gcp_b") + + assert svc.current_primary == "ollama_gcp_b" + assert svc.consecutive_healthy == 0 + + @pytest.mark.asyncio + async def test_same_gcp_b_observation_preserves_recovery_counter(self): + svc, _, _ = _make_service(current_primary="ollama_gcp_b") + svc._consecutive_healthy = 2 + mock_redis = AsyncMock() + + with patch("src.core.redis_client.get_redis", return_value=mock_redis): + await svc.set_current_primary("ollama_gcp_b") + + assert svc.consecutive_healthy == 2 + mock_redis.set.assert_not_awaited() + + @pytest.mark.asyncio + async def test_stale_fallback_observation_cannot_reopen_recovered_primary(self): + mock_alerter = AsyncMock() + svc, _, _ = _make_service( + current_primary="ollama_gcp_b", + alerter=mock_alerter, + ) + svc._consecutive_healthy = svc._stable_count_required + stale_observation_started_at = time.monotonic() + + with patch.object( + svc, + "_persist_primary", + new_callable=AsyncMock, + ) as persist_primary: + await svc._switch_back_to_ollama() + await svc.set_current_primary( + "ollama_gcp_b", + observation_generation=1, + observation_started_at_monotonic=stale_observation_started_at, + ) + + assert svc.current_primary == "ollama_gcp_a" + persist_primary.assert_awaited_once_with("ollama_gcp_a") + mock_alerter.alert_recovery.assert_awaited_once() + + @pytest.mark.asyncio + async def test_inflight_stale_fallback_waits_for_atomic_recovery_transition(self): + mock_alerter = AsyncMock() + svc, _, _ = _make_service( + current_primary="ollama_gcp_b", + alerter=mock_alerter, + ) + svc._consecutive_healthy = svc._stable_count_required + stale_observation_started_at = time.monotonic() + recovery_persist_started = asyncio.Event() + release_recovery_persist = asyncio.Event() + + async def blocking_persist(provider: str) -> None: + assert provider == "ollama_gcp_a" + recovery_persist_started.set() + await release_recovery_persist.wait() + + with patch.object( + svc, + "_persist_primary", + side_effect=blocking_persist, + ) as persist_primary: + recovery_task = asyncio.create_task(svc._switch_back_to_ollama()) + await recovery_persist_started.wait() + stale_callback_task = asyncio.create_task( + svc.set_current_primary( + "ollama_gcp_b", + observation_generation=1, + observation_started_at_monotonic=( + stale_observation_started_at + ), + ) + ) + await asyncio.sleep(0) + release_recovery_persist.set() + await asyncio.gather(recovery_task, stale_callback_task) + + assert svc.current_primary == "ollama_gcp_a" + assert svc.consecutive_healthy == 0 + persist_primary.assert_awaited_once_with("ollama_gcp_a") + mock_alerter.alert_recovery.assert_awaited_once() + + @pytest.mark.asyncio + async def test_newer_local_transition_invalidates_inflight_health_tick(self): + mock_alerter = AsyncMock() + svc, mock_monitor, _ = _make_service( + current_primary="ollama_gcp_b", + stable_count=3, + alerter=mock_alerter, + ) + svc._consecutive_healthy = 2 + mock_monitor.check.return_value = _make_health(HealthStatus.HEALTHY) + local_persist_started = asyncio.Event() + release_local_persist = asyncio.Event() + + async def blocking_persist(provider: str) -> None: + if provider == "ollama_local": + local_persist_started.set() + await release_local_persist.wait() + + with patch.object( + svc, + "_persist_primary", + side_effect=blocking_persist, + ) as persist_primary: + local_transition = asyncio.create_task( + svc.set_current_primary( + "ollama_local", + observation_generation=1, + observation_started_at_monotonic=time.monotonic(), + ) + ) + await local_persist_started.wait() + inflight_health_tick = asyncio.create_task( + svc._check_and_recover() + ) + await asyncio.sleep(0) + release_local_persist.set() + await asyncio.gather(local_transition, inflight_health_tick) + + assert svc.current_primary == "ollama_local" + assert svc.consecutive_healthy == 0 + persist_primary.assert_awaited_once_with("ollama_local") + mock_alerter.alert_recovery.assert_not_awaited() + + @pytest.mark.asyncio + async def test_older_observation_generation_cannot_overwrite_newer_lane(self): + svc, _, _ = _make_service(current_primary="ollama_gcp_a") + with patch.object( + svc, + "_persist_primary", + new_callable=AsyncMock, + ) as persist_primary: + await svc.set_current_primary( + "ollama_gcp_b", + observation_generation=2, + observation_started_at_monotonic=time.monotonic(), + ) + await svc.set_current_primary( + "ollama_local", + observation_generation=1, + observation_started_at_monotonic=time.monotonic(), + ) + + assert svc.current_primary == "ollama_gcp_b" + persist_primary.assert_awaited_once_with("ollama_gcp_b") + @pytest.mark.asyncio async def test_set_current_primary_to_ollama_no_counter_reset(self): """切換到 ollama(正常路由)→ counter 不重置(直接標記)""" @@ -451,7 +719,7 @@ class TestRedisPersistence: @pytest.mark.asyncio async def test_set_current_primary_persists_to_redis(self): - """set_current_primary("gemini") → Phase A 雙寫新舊 Redis key""" + """set_current_primary("gemini") atomically persists both Redis keys.""" svc, _, _ = _make_service(current_primary="ollama") mock_redis = AsyncMock() mock_redis.set = AsyncMock() @@ -459,12 +727,13 @@ class TestRedisPersistence: with patch("src.core.redis_client.get_redis", return_value=mock_redis): await svc.set_current_primary("gemini") - mock_redis.set.assert_has_awaits( - [ - call("platform:ollama:current_primary", "gemini"), - call("ollama:current_primary", "gemini"), - ] + mock_redis.mset.assert_awaited_once_with( + { + "platform:ollama:current_primary": "gemini", + "ollama:current_primary": "gemini", + } ) + mock_redis.set.assert_not_awaited() @pytest.mark.asyncio async def test_set_current_primary_same_value_no_redis_write(self): @@ -552,6 +821,26 @@ class TestRedisPersistence: finally: await svc.stop() + @pytest.mark.asyncio + async def test_start_immediate_check_when_primary_is_gcp_b_fallback(self): + svc, _, _ = _make_service(current_primary="ollama_gcp_a") + mock_redis = AsyncMock() + mock_redis.get = AsyncMock(return_value=b"ollama_gcp_b") + + with patch.object( + svc, + "_check_and_recover", + new_callable=AsyncMock, + ) as immediate_check, patch( + "src.core.redis_client.get_redis", + return_value=mock_redis, + ): + try: + await svc.start() + immediate_check.assert_awaited_once() + finally: + await svc.stop() + @pytest.mark.asyncio async def test_start_no_immediate_check_when_primary_is_ollama(self): """start() 時 primary=ollama(正常狀態)→ 不立刻執行 check""" diff --git a/apps/api/tests/test_ollama_failover_manager.py b/apps/api/tests/test_ollama_failover_manager.py index f82f38024..262a80c21 100644 --- a/apps/api/tests/test_ollama_failover_manager.py +++ b/apps/api/tests/test_ollama_failover_manager.py @@ -22,7 +22,8 @@ OllamaFailoverManager 單元測試 - P1.1c v4.0 from __future__ import annotations -from unittest.mock import AsyncMock, MagicMock, patch +import asyncio +from unittest.mock import ANY, AsyncMock, MagicMock, patch import pytest @@ -383,7 +384,7 @@ class TestSelectProvider: local_status=HealthStatus.OFFLINE, ) recovery_callback = AsyncMock() - manager._recovery_callback = recovery_callback + manager.set_recovery_callback(recovery_callback) audit = AsyncMock() with patch.object(manager, "_write_failover_audit", audit): @@ -402,14 +403,156 @@ class TestSelectProvider: local_status=HealthStatus.OFFLINE, ) recovery_callback = AsyncMock() - manager._recovery_callback = recovery_callback + manager.set_recovery_callback(recovery_callback) audit = AsyncMock() with patch.object(manager, "_write_failover_audit", audit): result = await manager.select_provider(emit_side_effects=True) audit.assert_awaited_once_with(result) - recovery_callback.assert_awaited_once_with("ollama_gcp_b") + recovery_callback.assert_awaited_once_with( + "ollama_gcp_b", + observation_generation=1, + observation_started_at_monotonic=ANY, + ) + + @pytest.mark.asyncio + async def test_legacy_one_argument_recovery_callback_remains_supported(self): + manager, _ = self._make_three_layer_mock( + gcp_a_status=HealthStatus.OFFLINE, + gcp_b_status=HealthStatus.HEALTHY, + local_status=HealthStatus.OFFLINE, + ) + observed_providers: list[str] = [] + + async def legacy_callback(provider_name: str) -> None: + observed_providers.append(provider_name) + + manager.set_recovery_callback(legacy_callback) + + with patch.object( + manager, + "_write_failover_audit", + new_callable=AsyncMock, + ): + await manager.select_provider(emit_side_effects=True) + + assert observed_providers == ["ollama_gcp_b"] + + @pytest.mark.asyncio + async def test_recovery_state_callback_precedes_degraded_alert_audit(self): + manager, _ = self._make_three_layer_mock( + gcp_a_status=HealthStatus.OFFLINE, + gcp_b_status=HealthStatus.HEALTHY, + local_status=HealthStatus.OFFLINE, + ) + event_order: list[str] = [] + + async def recovery_callback(provider_name: str, **metadata) -> None: + event_order.append(f"state:{provider_name}") + + async def degraded_audit(result) -> None: + event_order.append( + f"alert:{result.observed_fallback.provider_name}" + ) + + manager.set_recovery_callback(recovery_callback) + + with patch.object( + manager, + "_write_failover_audit", + side_effect=degraded_audit, + ): + await manager.select_provider(emit_side_effects=True) + + assert event_order == [ + "state:ollama_gcp_b", + "alert:ollama_gcp_b", + ] + + @pytest.mark.asyncio + async def test_rejected_recovery_observation_suppresses_degraded_alert(self): + manager, _ = self._make_three_layer_mock( + gcp_a_status=HealthStatus.OFFLINE, + gcp_b_status=HealthStatus.HEALTHY, + local_status=HealthStatus.OFFLINE, + ) + + async def stale_callback(provider_name: str, **metadata) -> bool: + return False + + manager.set_recovery_callback(stale_callback) + audit = AsyncMock() + + with patch.object(manager, "_write_failover_audit", audit): + await manager.select_provider(emit_side_effects=True) + + audit.assert_not_awaited() + + @pytest.mark.asyncio + async def test_overtaken_selection_cannot_emit_stale_degraded_alert(self): + manager = _make_manager() + older_secondary_started = asyncio.Event() + release_older_secondary = asyncio.Event() + newer_selection_started = False + + async def ordered_health_check(url: str) -> HealthReport: + if url == URL_GCP_A: + return _make_health(HealthStatus.OFFLINE, url) + if url == URL_GCP_B and not newer_selection_started: + older_secondary_started.set() + await release_older_secondary.wait() + return _make_health(HealthStatus.HEALTHY, url) + if url == URL_GCP_B: + return _make_health(HealthStatus.OFFLINE, url) + return _make_health( + HealthStatus.HEALTHY + if newer_selection_started + else HealthStatus.OFFLINE, + url, + ) + + manager._monitor.check = AsyncMock(side_effect=ordered_health_check) + last_accepted_generation = 0 + + async def generation_callback( + provider_name: str, + *, + observation_generation: int, + observation_started_at_monotonic: float, + ) -> bool: + nonlocal last_accepted_generation + if observation_generation <= last_accepted_generation: + return False + last_accepted_generation = observation_generation + return True + + manager.set_recovery_callback(generation_callback) + alerted_providers: list[str] = [] + + async def capture_audit(result: OllamaRoutingResult) -> None: + alerted_providers.append(result.observed_fallback.provider_name) + + with patch.object( + manager, + "_write_failover_audit", + side_effect=capture_audit, + ): + older_selection = asyncio.create_task( + manager.select_provider(emit_side_effects=True), + name="older-provider-selection", + ) + await older_secondary_started.wait() + newer_selection_started = True + newer_selection = asyncio.create_task( + manager.select_provider(emit_side_effects=True), + name="newer-provider-selection", + ) + await newer_selection + release_older_secondary.set() + await older_selection + + assert alerted_providers == ["ollama_local"] @pytest.mark.asyncio async def test_select_provider_gcp_a_healthy_primary_ollama(self):