fix(auth): refresh provider credentials per request (ENG-2116)

anton/cli.py

@@ -133,7 +133,7 @@ def _reexec() -> None:
 # Core dependencies from pyproject.toml that anton needs at runtime
 _REQUIRED_PACKAGES: dict[str, str] = {
     "anthropic": "anthropic>=0.42.0",
-    "openai": "openai>=1.0",
+    "openai": "openai>=2.21.0",
     "pydantic": "pydantic>=2.0",
     "pydantic_settings": "pydantic-settings>=2.0",
     "prompt_toolkit": "prompt-toolkit>=3.0",

anton/core/llm/anthropic.py

@@ -13,6 +13,7 @@
     ContextOverflowError,
     LLMProvider,
     LLMResponse,
+    ProviderAuthError,
     ProviderConnectionInfo,
     StreamComplete,
     StreamEvent,
@@ -47,7 +48,7 @@ def _raise_for_status_error(
     failures are classified first; only what's left is offered to the transient
     classifier, and the generic "unavailable" copy is the last resort.
 
-    - 401 → ConnectionError (invalid-key copy; cowork-server keys on this phrase).
+    - 401 → ProviderAuthError (canonical provider-credential refusal).
     - 429 WITH a quota ``detail`` → TokenLimitExceeded (keeps its own card).
     - 402/429 with an M3 gate wallet code (``wallet_empty`` /
       ``included_allowance_exhausted``, body or X-MindsHub-Reason header)
@@ -60,7 +61,7 @@ def _raise_for_status_error(
     - anything else → the generic "temporarily unavailable" ConnectionError.
     """
     if exc.status_code == 401:
-        raise ConnectionError(
+        raise ProviderAuthError(
             "Invalid API key — check your ANTHROPIC_API_KEY environment variable."
         ) from exc
 

anton/core/llm/client.py

@@ -1,14 +1,75 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import AsyncIterator
-from typing import TYPE_CHECKING
+from collections.abc import AsyncIterator, Awaitable, Callable
+from typing import TYPE_CHECKING, TypeVar
 
-from .provider import LLMProvider, LLMResponse, StreamComplete, StreamEvent
+from .provider import (
+    LLMProvider,
+    LLMResponse,
+    ProviderAuthError,
+    StreamComplete,
+    StreamEvent,
+)
 
 if TYPE_CHECKING:
     from anton.config.settings import AntonSettings
 
+_T = TypeVar("_T")
+
+
+class _ProviderAuthConfirmation:
+    """One immediate retry budget for a single logical provider call."""
+
+    def __init__(self) -> None:
+        self._available = True
+
+    def take(self) -> bool:
+        if not self._available:
+            return False
+        self._available = False
+        return True
+
+
+async def _call_with_auth_confirmation(
+    operation: Callable[[], Awaitable[_T]],
+    *,
+    role: str,
+) -> _T:
+    """Retry one typed provider-auth refusal, then propagate the second."""
+    confirmation = _ProviderAuthConfirmation()
+    while True:
+        try:
+            return await operation()
+        except ProviderAuthError as exc:
+            if not confirmation.take():
+                exc.role = role
+                raise
+
+
+async def _stream_with_auth_confirmation(
+    operation: Callable[[], AsyncIterator[_T]],
+    *,
+    role: str,
+) -> AsyncIterator[_T]:
+    """Retry a pre-stream auth refusal without replaying emitted events."""
+    confirmation = _ProviderAuthConfirmation()
+    while True:
+        yielded = False
+        try:
+            async for event in operation():
+                yielded = True
+                yield event
+            return
+        except ProviderAuthError as exc:
+            # ProviderAuthError is produced from an HTTP 401 before a response
+            # stream starts. Keep the guard anyway: if a future provider maps a
+            # mid-stream event to this type, replaying would duplicate visible
+            # text or tool-use events.
+            if yielded or not confirmation.take():
+                exc.role = role
+                raise
+
 
 def _resolve_openai_compatible_flavor(settings: AntonSettings) -> str:
     """Distinguish MindsHub/mdb.ai passthrough from a generic openai-compatible
@@ -63,6 +124,7 @@ def __init__(
         # Defaults to the coding role so hosts that construct LLMClient
         # directly (cowork-server) get behavior-preserving summarization
         # with no changes.
+        self._router_auth_role = "router" if router_provider is not None else "coding"
         self._router_provider = router_provider or coding_provider
         self._router_model = router_model or coding_model
         self._max_tokens = max_tokens
@@ -113,13 +175,16 @@ async def plan(
         native_web_tools: set[str] | None = None,
     ) -> LLMResponse:
         listener = self.usage_listener
-        response = await self._planning_provider.complete(
-            model=self._planning_model,
-            system=system,
-            messages=messages,
-            tools=tools,
-            max_tokens=max_tokens or self._max_tokens,
-            native_web_tools=native_web_tools,
+        response = await _call_with_auth_confirmation(
+            lambda: self._planning_provider.complete(
+                model=self._planning_model,
+                system=system,
+                messages=messages,
+                tools=tools,
+                max_tokens=max_tokens or self._max_tokens,
+                native_web_tools=native_web_tools,
+            ),
+            role="planning",
         )
         self._notify_usage("planning", self._planning_model, response.usage, listener)
         self._record_served(response)
@@ -135,13 +200,16 @@ async def plan_stream(
         native_web_tools: set[str] | None = None,
     ) -> AsyncIterator[StreamEvent]:
         listener = self.usage_listener
-        async for event in self._planning_provider.stream(
-            model=self._planning_model,
-            system=system,
-            messages=messages,
-            tools=tools,
-            max_tokens=max_tokens or self._max_tokens,
-            native_web_tools=native_web_tools,
+        async for event in _stream_with_auth_confirmation(
+            lambda: self._planning_provider.stream(
+                model=self._planning_model,
+                system=system,
+                messages=messages,
+                tools=tools,
+                max_tokens=max_tokens or self._max_tokens,
+                native_web_tools=native_web_tools,
+            ),
+            role="planning",
         ):
             if isinstance(event, StreamComplete):
                 self._notify_usage(
@@ -214,13 +282,16 @@ async def code(
         native_web_tools: set[str] | None = None,
     ) -> LLMResponse:
         listener = self.usage_listener
-        response = await self._coding_provider.complete(
-            model=self._coding_model,
-            system=system,
-            messages=messages,
-            tools=tools,
-            max_tokens=max_tokens or self._max_tokens,
-            native_web_tools=native_web_tools,
+        response = await _call_with_auth_confirmation(
+            lambda: self._coding_provider.complete(
+                model=self._coding_model,
+                system=system,
+                messages=messages,
+                tools=tools,
+                max_tokens=max_tokens or self._max_tokens,
+                native_web_tools=native_web_tools,
+            ),
+            role="coding",
         )
         self._notify_usage("coding", self._coding_model, response.usage, listener)
         return response
@@ -239,11 +310,14 @@ async def summarize(
         this is behavior-preserving unless a distinct model is selected.
         """
         listener = self.usage_listener
-        response = await self._router_provider.complete(
-            model=self._router_model,
-            system=system,
-            messages=messages,
-            max_tokens=max_tokens or self._max_tokens,
+        response = await _call_with_auth_confirmation(
+            lambda: self._router_provider.complete(
+                model=self._router_model,
+                system=system,
+                messages=messages,
+                max_tokens=max_tokens or self._max_tokens,
+            ),
+            role=self._router_auth_role,
         )
         self._notify_usage("router", self._router_model, response.usage, listener)
         return response
@@ -263,13 +337,16 @@ async def gate(
         only answer from context or delegate.
         """
         listener = self.usage_listener
-        response = await self._router_provider.complete(
-            model=self._router_model,
-            system=system,
-            messages=messages,
-            tools=tools,
-            tool_choice=tool_choice,
-            max_tokens=max_tokens or self._max_tokens,
+        response = await _call_with_auth_confirmation(
+            lambda: self._router_provider.complete(
+                model=self._router_model,
+                system=system,
+                messages=messages,
+                tools=tools,
+                tool_choice=tool_choice,
+                max_tokens=max_tokens or self._max_tokens,
+            ),
+            role=self._router_auth_role,
         )
         self._notify_usage("router", self._router_model, response.usage, listener)
         return response
@@ -304,13 +381,16 @@ async def _generate_object_with(
         budget = max_tokens or self._max_tokens
 
         listener = self.usage_listener
-        response = await provider.complete(
-            model=model,
-            system=system,
-            messages=messages,
-            tools=[tool],
-            tool_choice={"type": "tool", "name": tool["name"]},
-            max_tokens=budget,
+        response = await _call_with_auth_confirmation(
+            lambda: provider.complete(
+                model=model,
+                system=system,
+                messages=messages,
+                tools=[tool],
+                tool_choice={"type": "tool", "name": tool["name"]},
+                max_tokens=budget,
+            ),
+            role=role,
         )
         # Count BEFORE the no-tool-call raise below: a structured call that
         # failed (and its bigger-budget retry) still spent real tokens

anton/core/llm/openai.py

@@ -3,7 +3,7 @@
 import json
 import logging
 import os
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Awaitable, Callable
 from typing import NoReturn
 
 import openai
@@ -19,6 +19,7 @@
     LLMProvider,
     LLMResponse,
     ModelUnavailableError,
+    ProviderAuthError,
     ProviderConnectionInfo,
     StreamComplete,
     StreamEvent,
@@ -42,6 +43,8 @@
 
 logger = logging.getLogger(__name__)
 
+AsyncAPIKeyProvider = Callable[[], Awaitable[str]]
+
 
 def _raise_for_status_error(exc: "openai.APIStatusError", model: str) -> NoReturn:
     """Map a provider HTTP error onto anton's typed/curated exceptions.
@@ -51,8 +54,7 @@ def _raise_for_status_error(exc: "openai.APIStatusError", model: str) -> NoRetur
     previous four copy-pasted blocks had already diverged in wording.
 
     Mapping policy:
-      - 401 → ConnectionError with the invalid-key copy (cowork-server's
-        provider_auth detection keys on this exact phrase).
+      - 401 → ProviderAuthError with the invalid-key copy.
       - 403 with a structured gateway code (``model_access_denied`` /
         ``model_disabled``) → ModelUnavailableError carrying the code +
         model, with actionable copy. Detection is code-exact on purpose:
@@ -74,8 +76,8 @@ def _raise_for_status_error(exc: "openai.APIStatusError", model: str) -> NoRetur
     ``code`` sits at top level. The gateway's 429 is FastAPI-style
     (``{"detail": …}``, no envelope) and passes through untouched. Both
     fields are therefore read from the top level first, with an envelope
-    fallback for clients that deliver the wire shape unmodified (anton's
-    pyproject allows ``openai>=1.0``, and proxies exist that re-wrap).
+    fallback for clients that deliver the wire shape unmodified (anton requires
+    ``openai>=2.21.0``, and proxies exist that re-wrap).
     The originally shipped ENG-598 mapper read only
     ``body["error"]["code"]`` — a key the SDK had already peeled off —
     which left the model-403 card dead in production.
@@ -93,7 +95,7 @@ def _raise_for_status_error(exc: "openai.APIStatusError", model: str) -> NoRetur
     _foreign = origin_is_known_third_party(exc)
 
     if exc.status_code == 401:
-        raise ConnectionError(
+        raise ProviderAuthError(
             "Invalid API key — check your OpenAI API key configuration."
         ) from exc
 
@@ -827,7 +829,12 @@ def __init__(
         vision_format: str = "openai",
         flavor: str = FLAVOR_OPENAI_COMPATIBLE_GENERIC,
         reasoning_effort: str | None = None,
+        api_key_provider: AsyncAPIKeyProvider | None = None,
     ) -> None:
+        # Keep the construction-time value separate from the live supplier.
+        # ChatSession passes export_connection_info() into ScratchpadManager,
+        # whose subprocess/env boundary cannot carry a callable. The main-process
+        # async OpenAI client can ask the supplier for a fresh bearer per request.
         self._api_key = api_key
         self._base_url = base_url
         self._ssl_verify = ssl_verify
@@ -860,21 +867,35 @@ def __init__(
 
         import httpx
 
+        client_api_key = api_key_provider if api_key_provider is not None else api_key
         if api_version and _is_azure_endpoint(base_url):
             # Azure OpenAI: use the dedicated client which handles deployment
             # URL construction and api-version automatically.
+            if api_key_provider is not None:
+                # AsyncAzureOpenAI._prepare_options fully overrides the base
+                # hook and never chains to super(), so AsyncOpenAI's
+                # _refresh_api_key never runs and the supplier is never
+                # awaited. The base __init__ has already replaced a callable
+                # api_key with "", so the client would send an empty api-key
+                # header on every request and 401 forever. Refuse at
+                # construction instead of failing on the first call.
+                raise EndpointConfigurationError(
+                    "Azure OpenAI cannot refresh credentials per request: "
+                    "AsyncAzureOpenAI ignores a callable api_key. Pass a "
+                    "static api_key for Azure endpoints."
+                )
             azure_kwargs: dict = {"api_version": api_version}
-            if api_key:
-                azure_kwargs["api_key"] = api_key
+            if client_api_key:
+                azure_kwargs["api_key"] = client_api_key
             if base_url:
                 azure_kwargs["azure_endpoint"] = base_url
             if not ssl_verify:
                 azure_kwargs["http_client"] = httpx.AsyncClient(verify=False)
             self._client = AsyncAzureOpenAI(**azure_kwargs)
         else:
             kwargs: dict = {}
-            if api_key:
-                kwargs["api_key"] = api_key
+            if client_api_key:
+                kwargs["api_key"] = client_api_key
             if base_url:
                 kwargs["base_url"] = base_url
             if not ssl_verify:

anton/core/llm/provider.py

@@ -482,6 +482,21 @@ class TokenLimitExceeded(Exception):
     """Raised when the LLM returns 429 due to billing/token limits."""
 
 
+class ProviderAuthError(ConnectionError):
+    """Raised when a provider rejects its credential with HTTP 401.
+
+    The type distinguishes an authentication refusal from unrelated
+    ``ConnectionError`` failures. It remains a ``ConnectionError`` subclass so
+    in-process callers written against the previous 401 mapping keep working.
+    The client stamps ``role`` on a confirmed refusal so hosts can attribute
+    the recovery action to the provider that actually failed.
+    """
+
+    def __init__(self, message: str, *, role: str | None = None):
+        super().__init__(message)
+        self.role = role
+
+
 class StructuredOutputError(ValueError):
     """Raised when a forced-tool-call structured-output call yields no usable call.