fix(credential_pool): run next_available_at under the pool lock

Review fold on the #67642 salvage: next_available_at() called
_available_entries() — which prunes DEAD entries, syncs tokens, and
persists — and iterated self._entries with no lock, racing concurrent
select()/rotation exactly as has_available()'s comment warns. Wrap the
method body in self._lock and pin it with a non-blocking-acquire probe
test.

agent/credential_pool.py

@@ -631,18 +631,21 @@ class CredentialPool:
 
         Like :meth:`has_available`, expired cooldowns are left uncleared
         (``clear_expired=False``); the only writes are the same
-        re-auth/token sync paths ``has_available`` already performs.
+        re-auth/token sync paths ``has_available`` already performs — which
+        is exactly why this must run under ``self._lock`` like every other
+        ``_available_entries`` caller (see the comment on ``has_available``).
         """
-        if self._available_entries():
-            return None
-        candidates: List[float] = []
-        for entry in self._entries:
-            if entry.last_status != STATUS_EXHAUSTED:
-                continue
-            until = _exhausted_until(entry)
-            if until is not None:
-                candidates.append(until)
-        return min(candidates) if candidates else None
+        with self._lock:
+            if self._available_entries():
+                return None
+            candidates: List[float] = []
+            for entry in self._entries:
+                if entry.last_status != STATUS_EXHAUSTED:
+                    continue
+                until = _exhausted_until(entry)
+                if until is not None:
+                    candidates.append(until)
+            return min(candidates) if candidates else None
 
     def entries(self) -> List[PooledCredential]:
         with self._lock:

tests/run_agent/test_reset_aware_primary_restore.py

@@ -193,6 +193,25 @@ class TestNextAvailableAt:
         pool = CredentialPool("openrouter", [])
         assert pool.next_available_at() is None
 
+    def test_runs_under_the_pool_lock(self):
+        """next_available_at must hold self._lock like every other
+        _available_entries caller — a concurrent select()/rotation can
+        otherwise tear self._entries mid-iteration (see has_available)."""
+        pool = CredentialPool("openrouter", [])
+        held = {}
+
+        original = pool._available_entries
+
+        def _probe(**kwargs):
+            held["locked"] = not pool._lock.acquire(blocking=False)
+            if not held["locked"]:
+                pool._lock.release()
+            return original(**kwargs)
+
+        pool._available_entries = _probe
+        pool.next_available_at()
+        assert held["locked"], "next_available_at called _available_entries without self._lock"
+
 
 # =============================================================================
 # restore_primary_runtime() gate