fix(agent): consult credential pool in _restore_primary_runtime to avoid stale snapshot key

run_agent.py

@@ -8225,30 +8225,42 @@ def _restore_primary_runtime(self) -> bool:
             self.api_key = rt["api_key"]
             self._client_kwargs = dict(rt["client_kwargs"])
             self._use_prompt_caching = rt["use_prompt_caching"]
-            # Default to native layout when the restored snapshot predates the
-            # native-vs-proxy split (older sessions saved before this PR).
             self._use_native_cache_layout = rt.get(
                 "use_native_cache_layout",
                 self.api_mode == "anthropic_messages" and self.provider == "anthropic",
             )
 
-            # ── Rebuild client for the primary provider ──
-            if self.api_mode == "anthropic_messages":
-                from agent.anthropic_adapter import build_anthropic_client
-                self._anthropic_api_key = rt["anthropic_api_key"]
-                self._anthropic_base_url = rt["anthropic_base_url"]
-                self._anthropic_client = build_anthropic_client(
-                    rt["anthropic_api_key"], rt["anthropic_base_url"],
-                    timeout=get_provider_request_timeout(self.provider, self.model),
-                )
-                self._is_anthropic_oauth = rt["is_anthropic_oauth"]
-                self.client = None
+            # ── Consult credential pool for current best entry ──
+            # The snapshot may hold a stale key if the pool rotated during
+            # the previous turn (e.g. after 401/429/402).  Override with
+            # the pool's current selection when available.
+            pool = self._credential_pool
+            pool_entry = None
+            if pool is not None:
+                try:
+                    pool_entry = pool.current() or pool.select()
+                except Exception:
+                    pool_entry = None
+            if pool_entry is not None:
+                self._swap_credential(pool_entry)
             else:
-                self.client = self._create_openai_client(
-                    dict(rt["client_kwargs"]),
-                    reason="restore_primary",
-                    shared=True,
-                )
+                # ── Rebuild client for the primary provider ──
+                if self.api_mode == "anthropic_messages":
+                    from agent.anthropic_adapter import build_anthropic_client
+                    self._anthropic_api_key = rt["anthropic_api_key"]
+                    self._anthropic_base_url = rt["anthropic_base_url"]
+                    self._anthropic_client = build_anthropic_client(
+                        rt["anthropic_api_key"], rt["anthropic_base_url"],
+                        timeout=get_provider_request_timeout(self.provider, self.model),
+                    )
+                    self._is_anthropic_oauth = rt["is_anthropic_oauth"]
+                    self.client = None
+                else:
+                    self.client = self._create_openai_client(
+                        dict(rt["client_kwargs"]),
+                        reason="restore_primary",
+                        shared=True,
+                    )
 
             # ── Restore context engine state ──
             cc = self.context_compressor
@@ -8332,22 +8344,35 @@ def _try_recover_primary_transport(
                 self._transport_cache.clear()
             self.api_key = rt["api_key"]
 
-            if self.api_mode == "anthropic_messages":
-                from agent.anthropic_adapter import build_anthropic_client
-                self._anthropic_api_key = rt["anthropic_api_key"]
-                self._anthropic_base_url = rt["anthropic_base_url"]
-                self._anthropic_client = build_anthropic_client(
-                    rt["anthropic_api_key"], rt["anthropic_base_url"],
-                    timeout=get_provider_request_timeout(self.provider, self.model),
-                )
-                self._is_anthropic_oauth = rt["is_anthropic_oauth"]
-                self.client = None
+            # Consult credential pool for current best entry, same as
+            # _restore_primary_runtime — the snapshot key may be stale
+            # after pool rotation in a previous turn.
+            pool = self._credential_pool
+            pool_entry = None
+            if pool is not None:
+                try:
+                    pool_entry = pool.current() or pool.select()
+                except Exception:
+                    pool_entry = None
+            if pool_entry is not None:
+                self._swap_credential(pool_entry)
             else:
-                self.client = self._create_openai_client(
-                    dict(rt["client_kwargs"]),
-                    reason="primary_recovery",
-                    shared=True,
-                )
+                if self.api_mode == "anthropic_messages":
+                    from agent.anthropic_adapter import build_anthropic_client
+                    self._anthropic_api_key = rt["anthropic_api_key"]
+                    self._anthropic_base_url = rt["anthropic_base_url"]
+                    self._anthropic_client = build_anthropic_client(
+                        rt["anthropic_api_key"], rt["anthropic_base_url"],
+                        timeout=get_provider_request_timeout(self.provider, self.model),
+                    )
+                    self._is_anthropic_oauth = rt["is_anthropic_oauth"]
+                    self.client = None
+                else:
+                    self.client = self._create_openai_client(
+                        dict(rt["client_kwargs"]),
+                        reason="primary_recovery",
+                        shared=True,
+                    )
 
             wait_time = min(3 + retry_count, 8)
             self._vprint(

tests/run_agent/test_primary_runtime_restore.py

@@ -528,3 +528,112 @@ def test_cooldown_not_set_when_already_on_fallback(self):
 
         # second call should not have extended the cooldown
         assert second_cooldown == first_cooldown
+
+
+# =============================================================================
+# Credential pool bypass fix (issue #25205)
+# =============================================================================
+
+class TestCredentialPoolBypass:
+    """Verify _restore_primary_runtime consults the credential pool instead
+    of blindly restoring a stale snapshot key."""
+
+    def test_restore_uses_pool_current_entry(self):
+        """When the pool has a current entry, restore uses it (not snapshot)."""
+        agent = _make_agent(
+            fallback_model={"provider": "openrouter", "model": "anthropic/claude-sonnet-4"},
+        )
+
+        mock_client = _mock_resolve()
+        with patch("agent.auxiliary_client.resolve_provider_client", return_value=(mock_client, None)):
+            agent._try_activate_fallback()
+
+        assert agent._fallback_activated is True
+
+        pool_entry = SimpleNamespace(
+            runtime_api_key="fresh-pool-key-12345678",
+            access_token="fresh-pool-key-12345678",
+            runtime_base_url="https://api.openai.com/v1",
+            base_url="https://api.openai.com/v1",
+        )
+        mock_pool = MagicMock()
+        mock_pool.current.return_value = pool_entry
+        mock_pool.select.return_value = pool_entry
+        agent._credential_pool = mock_pool
+
+        snapshot_key = agent._primary_runtime["api_key"]
+
+        with patch.object(agent, "_swap_credential") as mock_swap:
+            result = agent._restore_primary_runtime()
+
+        assert result is True
+        assert agent._fallback_activated is False
+        mock_swap.assert_called_once_with(pool_entry)
+        assert agent.api_key != snapshot_key or mock_swap.called
+
+    def test_restore_falls_back_to_snapshot_when_no_pool(self):
+        """Without a credential pool, restore uses the snapshot as before."""
+        agent = _make_agent(
+            fallback_model={"provider": "openrouter", "model": "anthropic/claude-sonnet-4"},
+        )
+        agent._credential_pool = None
+
+        mock_client = _mock_resolve()
+        with patch("agent.auxiliary_client.resolve_provider_client", return_value=(mock_client, None)):
+            agent._try_activate_fallback()
+
+        assert agent._fallback_activated is True
+
+        with patch("run_agent.OpenAI", return_value=MagicMock()):
+            result = agent._restore_primary_runtime()
+
+        assert result is True
+        assert agent.api_key == agent._primary_runtime["api_key"]
+
+    def test_restore_handles_pool_exception_gracefully(self):
+        """If pool.select() raises, restore falls back to snapshot."""
+        agent = _make_agent(
+            fallback_model={"provider": "openrouter", "model": "anthropic/claude-sonnet-4"},
+        )
+
+        mock_client = _mock_resolve()
+        with patch("agent.auxiliary_client.resolve_provider_client", return_value=(mock_client, None)):
+            agent._try_activate_fallback()
+
+        mock_pool = MagicMock()
+        mock_pool.current.side_effect = RuntimeError("pool corrupted")
+        mock_pool.select.side_effect = RuntimeError("pool corrupted")
+        agent._credential_pool = mock_pool
+
+        with patch("run_agent.OpenAI", return_value=MagicMock()):
+            result = agent._restore_primary_runtime()
+
+        assert result is True
+        assert agent.api_key == agent._primary_runtime["api_key"]
+
+    def test_restore_prefers_current_over_select(self):
+        """pool.current() is tried first; select() is the fallback."""
+        agent = _make_agent(
+            fallback_model={"provider": "openrouter", "model": "anthropic/claude-sonnet-4"},
+        )
+
+        mock_client = _mock_resolve()
+        with patch("agent.auxiliary_client.resolve_provider_client", return_value=(mock_client, None)):
+            agent._try_activate_fallback()
+
+        current_entry = SimpleNamespace(
+            runtime_api_key="current-key",
+            access_token="current-key",
+            runtime_base_url="https://api.openai.com/v1",
+            base_url="https://api.openai.com/v1",
+        )
+        mock_pool = MagicMock()
+        mock_pool.current.return_value = current_entry
+        mock_pool.select.return_value = None
+        agent._credential_pool = mock_pool
+
+        with patch.object(agent, "_swap_credential") as mock_swap:
+            agent._restore_primary_runtime()
+
+        mock_swap.assert_called_once_with(current_entry)
+        mock_pool.select.assert_not_called()