fix(ai): recover canonical Ollama primary safely

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(),

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:
         """

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"""

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):