Reuse credential tokens for Decision and chat model requests

sdk/evaluation/azure-ai-evaluation/CHANGELOG.md

@@ -1,5 +1,17 @@
 # Release History
 
+## 1.18.12 (Unreleased)
+
+### Features Added
+
+### Breaking Changes
+
+### Bugs Fixed
+
+- Decision requests, and Azure OpenAI chat model requests that authenticate with Microsoft Entra ID, reuse a credential's token until five minutes before it expires, as azure-core's bearer token policy does, instead of requesting a token for every request. Concurrent evaluations that share a synchronous credential wait for one token request, without blocking their event loops. A credential whose `get_token` is a coroutine function, or that implements `AsyncTokenCredential` but not the synchronous context manager protocol, is asked for its token on the caller's event loop, and any other credential on a worker thread, so a plain `get_token` that needs the running event loop must belong to an `AsyncTokenCredential`. With developer credentials such as `AzureCliCredential`, which start a process per request, concurrent evaluations were slow and could fail. Requests that waited on a token request that failed authentication share its failure instead of each requesting a token again; after any other failure, such as a token service outage, each waiting request makes its own attempt, so a transient failure fails only the request that hit it. Decision and chat requests behave the same way. A token rejected with HTTP 401 is dropped and requested again; a token whose request began before the rejection is not kept for reuse, and requests that waited on it use it rather than each requesting another one. The default credential that a chat evaluator creates when no `credential` is passed is shared the same way. Tokens are kept per scope, and the cache never keeps a credential alive. A token without an expiry, or that expires within five minutes, is not kept, but the requests that waited for it use it.
+
+### Other Changes
+
 ## 1.18.11 (2026-11-03)
 
 ### Features Added
@@ -17,6 +29,7 @@
 - `evaluate()` computes `<evaluator>.binary_aggregate` over the rows whose result is `pass` or `fail`. Other rows, such as not applicable or errored rows, used to count as failures, although the mean score leaves them out. An evaluator without `pass` or `fail` rows, including a custom evaluator whose results have other values, has no `binary_aggregate`, as one without scores has no mean.
 - An evaluator's inputs keep the order in which its `__call__` overloads declare them, instead of the order of a Python `set`, which varies with the process's hash seed. An evaluator that sends these inputs to a Decision model unchanged, as several Microsoft Foundry catalog evaluators do, could list the same fields in a different order in each process, and the Decision model could answer the same input differently. The built-in evaluators in this package already send their Decision inputs in a fixed order.
 
+
 ## 1.18.10 (2026-10-22)
 
 ### Features Added

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluators/_common/_model_runners.py

@@ -16,14 +16,17 @@
 import math
 import os
 import re
+import threading
 import time
 from typing import Any, Callable, Dict, List, Optional
 from urllib.parse import urlsplit
 
+from azure.core.exceptions import ClientAuthenticationError
 from azure.core.pipeline.policies import AsyncRedirectPolicy, AsyncRetryPolicy, CustomHookPolicy
 from azure.core.pipeline.transport import AioHttpTransport
 
 from azure.ai.evaluation._exceptions import ErrorBlame, ErrorCategory, ErrorTarget, EvaluationException
+from azure.ai.evaluation._legacy._common._token_cache import TOKENS as _TOKENS, is_async_credential
 from ..._common.utils import construct_prompty_model_config, validate_model_config
 from ..._http_utils import get_async_http_client
 
@@ -44,6 +47,9 @@ def value(self) -> str:
 
 DECISION_MODEL_TYPE = "decision"
 _DECISION_PATH = "/providers/microsoft/v1/systemone"
+# Decision endpoints also accept the chat scope, https://ai.azure.com/.default, but the hosted evaluation service
+# relies on this audience to call Decision with its exchanged or managed identity token rather than the caller's own
+# token, so Decision requests keep this scope and their own cached token.
 _DECISION_TOKEN_SCOPE = "https://cognitiveservices.azure.com/.default"
 LOGGER = logging.getLogger(__name__)
 _PREFIX = "azure.ai.evaluation.decision."
@@ -362,6 +368,25 @@ def _is_finite_number(value: Any) -> bool:
         return False
 
 
+def _token_failure(error: BaseException) -> Optional[Dict[str, Any]]:
+    """Describe an authentication failure for the requests that waited on it, or return None for any other failure.
+
+    As on the chat path, only an authentication failure is shared: any other failure, such as a token service outage,
+    is not, so each waiting request makes its own attempt and one transient failure fails only the request that hit
+    it. The description never keeps the error, whose traceback references the credential.
+    """
+    if not isinstance(error, ClientAuthenticationError):
+        return None
+    message = "Could not acquire a token for the decision model."
+    return {
+        "message": message,
+        "internal_message": message,
+        "target": ErrorTarget.EVALUATE,
+        "category": ErrorCategory.FAILED_EXECUTION,
+        "blame": ErrorBlame.USER_ERROR,
+    }
+
+
 def _provider_error_code(payload: Any) -> Optional[str]:
     """Return the provider's nested ``error.code`` when it is a bounded identifier, otherwise None."""
     provider_error = payload.get("error") if isinstance(payload, dict) else None
@@ -695,25 +720,72 @@ async def _await_request(self, content: bytes, deadline: float, execution: Decis
                 task.cancel()
                 await asyncio.gather(task, return_exceptions=True)
 
-    async def _get_token(self) -> str:
-        if inspect.iscoroutinefunction(self._credential.get_token):
-            return (await self._credential.get_token(_DECISION_TOKEN_SCOPE)).token
-        token = await asyncio.to_thread(self._credential.get_token, _DECISION_TOKEN_SCOPE)
-        if inspect.isawaitable(token):
-            token = await token
-        return token.token
+    async def _get_token(self, deadline: float) -> Optional[str]:
+        """Return a bearer token, or None when the deadline passed while waiting for another request's token.
+
+        A credential whose ``get_token`` is a coroutine function, or that implements ``AsyncTokenCredential`` but not
+        the synchronous context manager protocol, is asked on this event loop; any other credential is asked on a
+        worker thread, once for concurrent requests.
+
+        :param float deadline: Absolute ``time.monotonic()`` instant by which the request must complete, not a
+            duration or a wall-clock time; the caller derives it from the request timeout.
+        :return: The bearer token, or None when the deadline passed while this request waited for a token that
+            another request was acquiring.
+        :rtype: Optional[str]
+        """
+        credential = self._credential
+        token = _TOKENS.cached(credential, _DECISION_TOKEN_SCOPE)
+        if token is None:
+            # An acquisition that begins before a service rejects a token may return that token; the generation it
+            # began under lets the cache refuse it.
+            generation = _TOKENS.generation(credential, _DECISION_TOKEN_SCOPE)
+            if is_async_credential(credential):
+                # An asynchronous credential's token request may need this event loop.
+                token = credential.get_token(_DECISION_TOKEN_SCOPE)
+                if inspect.isawaitable(token):
+                    token = await token
+                _TOKENS.store(credential, _DECISION_TOKEN_SCOPE, token, generation)
+            else:
+                abandoned = threading.Event()
+                try:
+                    # A request waiting for another request's token holds a default-executor thread while it waits;
+                    # past the executor's size, further requests queue for a thread before they reach the cache.
+                    token = await asyncio.to_thread(
+                        _TOKENS.fetch,
+                        credential,
+                        _DECISION_TOKEN_SCOPE,
+                        abandoned=abandoned,
+                        deadline=deadline,
+                        describe_failure=_token_failure,
+                    )
+                finally:
+                    abandoned.set()
+                if token is None and time.monotonic() >= deadline:
+                    return None
+                if inspect.isawaitable(token):
+                    token = await token
+                    _TOKENS.store(credential, _DECISION_TOKEN_SCOPE, token, generation)
+        value = getattr(token, "token", None)
+        if not isinstance(value, str):
+            raise _failed("Could not acquire a token for the decision model.", ErrorBlame.USER_ERROR)
+        return value
 
     async def _post(self, content: bytes, *, deadline: float, execution: DecisionExecution) -> Any:
         headers = {"x-ms-useragent": self._user_agent, "Content-Type": "application/json"}
+        token = None
         if self._api_key:
             headers["api-key"] = self._api_key
         else:
             try:
-                headers["Authorization"] = f"Bearer {await self._get_token()}"
+                token = await self._get_token(deadline)
             except EvaluationException:
                 raise
             except Exception:  # pylint: disable=broad-except
                 raise _failed("Could not acquire a token for the decision model.", ErrorBlame.USER_ERROR) from None
+            if token is None:
+                execution.outcome = "deadline"
+                raise _failed("The decision model request timed out.")
+            headers["Authorization"] = "Bearer " + token
 
         policies: Dict[str, Any] = {
             "custom_hook_policy": execution,
@@ -766,6 +838,9 @@ async def _post(self, content: bytes, *, deadline: float, execution: DecisionExe
                     raise _failed("The decision model returned invalid JSON.") from None
                 payload = None
         if status != 200:
+            if status == 401 and token is not None:
+                # A rejected token is not reused.
+                _TOKENS.discard(self._credential, _DECISION_TOKEN_SCOPE, token)
             blame = ErrorBlame.USER_ERROR if status in (400, 401, 403, 404, 422) else ErrorBlame.SYSTEM_ERROR
             code = _provider_error_code(payload)
             kind = _provider_error_kind(status, payload)

sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_common/_async_token_provider.py

@@ -4,6 +4,7 @@
 
 import logging
 import os
+import threading
 from typing import Any, AsyncContextManager, Optional
 
 from azure.core.credentials import AccessToken, TokenCredential
@@ -27,11 +28,13 @@ def __init__(self, *, base_url: Optional[str] = None, **kwargs: Any) -> None:
         """Initialize the AsyncAzureTokenProvider."""
         self._credential: Optional[TokenCredential] = None
         self._env_client: Optional[AzureEnvironmentClient] = AzureEnvironmentClient(base_url=base_url, **kwargs)
+        self._credential_lock = threading.Lock()
 
     async def close(self) -> None:
-        credential, client = self._credential, self._env_client
-        self._credential = None
-        self._env_client = None
+        with self._credential_lock:
+            credential, client = self._credential, self._env_client
+            self._credential = None
+            self._env_client = None
         try:
             close_credential = getattr(credential, "close", None)
             if close_credential is not None:
@@ -40,31 +43,58 @@ async def close(self) -> None:
             if client is not None:
                 await client.close()
 
-    async def get_token(
-        self,
-        *scopes: str,
-        claims: Optional[str] = None,
-        tenant_id: Optional[str] = None,
-        enable_cae: bool = False,
-        **kwargs: Any,
-    ) -> AccessToken:
-        if self._credential is None:
-            self._credential = await self._initialize_async(self._env_client)
+    async def get_credential(self) -> TokenCredential:
+        """Return the synchronous credential that supplies this provider's tokens, creating it on first use.
+
+        Concurrent first requests all use the first credential created, so they can share its token.
+
+        :return: The credential.
+        :rtype: ~azure.core.credentials.TokenCredential
+        """
+        credential = self._credential
+        if credential is None:
+            created = await self._initialize_async(self._env_client)
+            with self._credential_lock:
+                closed = self._env_client is None
+                # A provider closed while the credential was created keeps no credential.
+                if self._credential is None and not closed:
+                    self._credential = created
+                credential = self._credential
+            if credential is not created:
+                close_created = getattr(created, "close", None)
+                if close_created is not None:
+                    close_created()
+            if closed:
+                raise EvaluationException(
+                    f"{AsyncAzureTokenProvider.__name__} instance has already been closed.",
+                    target=ErrorTarget.UNKNOWN,
+                    category=ErrorCategory.INVALID_VALUE,
+                    blame=ErrorBlame.USER_ERROR,
+                )
 
-        if self._credential is None:
+        if credential is None:
             raise EvaluationException(
                 f"{self.__class__.__name__} could not determine the credential to use.",
                 target=ErrorTarget.UNKNOWN,
                 category=ErrorCategory.INVALID_VALUE,
                 blame=ErrorBlame.SYSTEM_ERROR,
             )
+        return credential
 
-        return self._credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, enable_cae=enable_cae, **kwargs)
+    async def get_token(
+        self,
+        *scopes: str,
+        claims: Optional[str] = None,
+        tenant_id: Optional[str] = None,
+        enable_cae: bool = False,
+        **kwargs: Any,
+    ) -> AccessToken:
+        credential = await self.get_credential()
+        return credential.get_token(*scopes, claims=claims, tenant_id=tenant_id, enable_cae=enable_cae, **kwargs)
 
     async def __aenter__(self) -> "AsyncAzureTokenProvider":
         try:
-            if self._credential is None:
-                self._credential = await self._initialize_async(self._env_client)
+            await self.get_credential()
         except BaseException:
             try:
                 await self.close()