Add provider-owned authentication recovery (#39274)

## What changed

- Let model providers identify recoverable authentication errors and refresh their own authentication state before falling back to `AuthManager` recovery.
- Apply provider recovery to both HTTP and WebSocket response requests, with at most one provider-owned recovery attempt per request.
- Preserve existing `401 Unauthorized` handling as the default for providers that do not customize recovery.

## Testing

- Add coverage for successful and failed provider recovery, including the single-attempt bound and preservation of the original unauthorized error.

GitOrigin-RevId: 9a737e3be96354699586f97dd88e544a45082969
This commit is contained in:
alexsong-oai authored and copyberry committed 2026-08-18 20:14:17 +00:00
1 parent f950a1ba0b
commit 22b860e80b
4 files changed
+193 -9

No files matched your search

+46 -9
View File
@@ -126,6 +126,7 @@ use codex_login::auth_env_telemetry::AuthEnvTelemetry;
use codex_login::auth_env_telemetry::collect_auth_env_telemetry; use codex_login::auth_env_telemetry::collect_auth_env_telemetry;
use codex_model_provider::AgentIdentitySessionFallback; use codex_model_provider::AgentIdentitySessionFallback;
use codex_model_provider::ProviderAuthScope; use codex_model_provider::ProviderAuthScope;
use codex_model_provider::ProviderUnauthorizedRecovery;
use codex_model_provider::SharedModelProvider; use codex_model_provider::SharedModelProvider;
use codex_model_provider::create_model_provider; use codex_model_provider::create_model_provider;
#[cfg(test)] #[cfg(test)]
@@ -1451,6 +1452,7 @@ impl ModelClientSession {
let mut auth_recovery = auth_manager let mut auth_recovery = auth_manager
.as_ref() .as_ref()
.map(AuthManager::unauthorized_recovery); .map(AuthManager::unauthorized_recovery);
let mut provider_auth_recovery_attempted = false;
let mut pending_retry = PendingUnauthorizedRetry::default(); let mut pending_retry = PendingUnauthorizedRetry::default();
loop { loop {
let client_setup = self.client.current_client_setup().await?; let client_setup = self.client.current_client_setup().await?;
@@ -1520,9 +1522,13 @@ impl ModelClientSession {
); );
return Ok(stream); return Ok(stream);
} }
Err(ApiError::Transport( Err(ApiError::Transport(unauthorized_transport))
unauthorized_transport @ TransportError::Http { status, .. }, if self
)) if status == StatusCode::UNAUTHORIZED => { .client
.state
.provider
.is_recoverable_auth_error(&unauthorized_transport) =>
{
let response_debug_context = let response_debug_context =
extract_response_debug_context(&unauthorized_transport); extract_response_debug_context(&unauthorized_transport);
inference_trace_attempt.record_failed( inference_trace_attempt.record_failed(
@@ -1534,6 +1540,7 @@ impl ModelClientSession {
handle_unauthorized( handle_unauthorized(
unauthorized_transport, unauthorized_transport,
&mut auth_recovery, &mut auth_recovery,
&mut provider_auth_recovery_attempted,
session_telemetry, session_telemetry,
&self.client.state.provider, &self.client.state.provider,
) )
@@ -1584,11 +1591,13 @@ impl ModelClientSession {
request_trace: Option<W3cTraceContext>, request_trace: Option<W3cTraceContext>,
inference_trace: &InferenceTraceContext, inference_trace: &InferenceTraceContext,
) -> Result<WebsocketStreamOutcome> { ) -> Result<WebsocketStreamOutcome> {
let auth_manager = self.client.state.provider.auth_manager(); let provider = Arc::clone(&self.client.state.provider);
let auth_manager = provider.auth_manager();
let mut auth_recovery = auth_manager let mut auth_recovery = auth_manager
.as_ref() .as_ref()
.map(AuthManager::unauthorized_recovery); .map(AuthManager::unauthorized_recovery);
let mut provider_auth_recovery_attempted = false;
let mut pending_retry = PendingUnauthorizedRetry::default(); let mut pending_retry = PendingUnauthorizedRetry::default();
loop { loop {
let client_setup = self.client.current_client_setup().await?; let client_setup = self.client.current_client_setup().await?;
@@ -1643,21 +1652,22 @@ impl ModelClientSession {
{ {
return Ok(WebsocketStreamOutcome::FallbackToHttp); return Ok(WebsocketStreamOutcome::FallbackToHttp);
} }
Err(ApiError::Transport( Err(ApiError::Transport(unauthorized_transport))
unauthorized_transport @ TransportError::Http { status, .. }, if provider.is_recoverable_auth_error(&unauthorized_transport) =>
)) if status == StatusCode::UNAUTHORIZED => { {
pending_retry = PendingUnauthorizedRetry::from_recovery( pending_retry = PendingUnauthorizedRetry::from_recovery(
handle_unauthorized( handle_unauthorized(
unauthorized_transport, unauthorized_transport,
&mut auth_recovery, &mut auth_recovery,
&mut provider_auth_recovery_attempted,
session_telemetry, session_telemetry,
&self.client.state.provider, &provider,
) )
.await?, .await?,
); );
continue; continue;
} }
Err(err) => return Err(self.client.state.provider.map_api_error(err)), Err(err) => return Err(provider.map_api_error(err)),
} }
let (incremental_request, previous_response_id_from_untraced_warmup) = let (incremental_request, previous_response_id_from_untraced_warmup) =
@@ -2225,10 +2235,37 @@ struct WebsocketConnectParams<'a> {
async fn handle_unauthorized( async fn handle_unauthorized(
transport: TransportError, transport: TransportError,
auth_recovery: &mut Option<UnauthorizedRecovery>, auth_recovery: &mut Option<UnauthorizedRecovery>,
provider_auth_recovery_attempted: &mut bool,
session_telemetry: &SessionTelemetry, session_telemetry: &SessionTelemetry,
provider: &SharedModelProvider, provider: &SharedModelProvider,
) -> Result<UnauthorizedRecoveryExecution> { ) -> Result<UnauthorizedRecoveryExecution> {
let debug = extract_response_debug_context(&transport); let debug = extract_response_debug_context(&transport);
if !*provider_auth_recovery_attempted {
*provider_auth_recovery_attempted = true;
match provider.recover_from_unauthorized().await {
Ok(ProviderUnauthorizedRecovery::Recovered) => {
return Ok(UnauthorizedRecoveryExecution {
mode: "provider",
phase: "provider_refresh",
});
}
Ok(ProviderUnauthorizedRecovery::NotConfigured) => {}
Err(error) => {
let original = provider.map_api_error(ApiError::Transport(transport));
warn!(
error = %error,
original_error = %original,
"provider authentication recovery failed"
);
return Err(if error.is_retryable() {
original
} else {
error
});
}
}
}
if let Some(recovery) = auth_recovery if let Some(recovery) = auth_recovery
&& recovery.has_next() && recovery.has_next()
{ {
+118
View File
@@ -27,19 +27,27 @@ use codex_login::AuthManager;
use codex_login::CodexAuth; use codex_login::CodexAuth;
use codex_login::auth::AgentIdentityAuthPolicy; use codex_login::auth::AgentIdentityAuthPolicy;
use codex_model_provider::BearerAuthProvider; use codex_model_provider::BearerAuthProvider;
use codex_model_provider::ModelProvider;
use codex_model_provider::ModelProviderFuture;
use codex_model_provider::ProviderAccountResult;
use codex_model_provider::ProviderUnauthorizedRecovery;
use codex_model_provider::SharedModelProvider; use codex_model_provider::SharedModelProvider;
use codex_model_provider::create_model_provider; use codex_model_provider::create_model_provider;
use codex_model_provider_info::CHATGPT_CODEX_BASE_URL; use codex_model_provider_info::CHATGPT_CODEX_BASE_URL;
use codex_model_provider_info::ModelProviderInfo; use codex_model_provider_info::ModelProviderInfo;
use codex_model_provider_info::WireApi; use codex_model_provider_info::WireApi;
use codex_model_provider_info::create_oss_provider_with_base_url; use codex_model_provider_info::create_oss_provider_with_base_url;
use codex_models_manager::manager::SharedModelsManager;
use codex_otel::SessionTelemetry; use codex_otel::SessionTelemetry;
use codex_protocol::ThreadId; use codex_protocol::ThreadId;
use codex_protocol::auth::AuthMode; use codex_protocol::auth::AuthMode;
use codex_protocol::error::CodexErr;
use codex_protocol::error::CodexErrorDetails;
use codex_protocol::models::BaseInstructions; use codex_protocol::models::BaseInstructions;
use codex_protocol::models::ContentItem; use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem; use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ModelInfo; use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ModelsResponse;
use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::InternalSessionSource; use codex_protocol::protocol::InternalSessionSource;
use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SessionSource;
@@ -57,6 +65,7 @@ use pretty_assertions::assert_eq;
use serde_json::json; use serde_json::json;
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::collections::VecDeque; use std::collections::VecDeque;
use std::path::PathBuf;
use std::pin::Pin; use std::pin::Pin;
use std::sync::Arc; use std::sync::Arc;
use std::sync::Mutex; use std::sync::Mutex;
@@ -659,6 +668,7 @@ async fn bedrock_unauthorized_error_uses_provider_mapping() {
/*auth_manager*/ None, /*auth_manager*/ None,
); );
let mut auth_recovery = None; let mut auth_recovery = None;
let mut provider_auth_recovery_attempted = false;
let url = "https://bedrock-mantle.us-east-2.api.aws/openai/v1/responses"; let url = "https://bedrock-mantle.us-east-2.api.aws/openai/v1/responses";
let error = super::handle_unauthorized( let error = super::handle_unauthorized(
TransportError::Http { TransportError::Http {
@@ -671,6 +681,7 @@ async fn bedrock_unauthorized_error_uses_provider_mapping() {
), ),
}, },
&mut auth_recovery, &mut auth_recovery,
&mut provider_auth_recovery_attempted,
&test_session_telemetry(), &test_session_telemetry(),
&provider, &provider,
) )
@@ -685,6 +696,113 @@ async fn bedrock_unauthorized_error_uses_provider_mapping() {
); );
} }
#[derive(Debug)]
struct TestRecoveryProvider {
inner: SharedModelProvider,
should_fail: bool,
attempts: Arc<AtomicUsize>,
}
impl ModelProvider for TestRecoveryProvider {
fn info(&self) -> &ModelProviderInfo {
self.inner.info()
}
fn auth_manager(&self) -> Option<Arc<AuthManager>> {
None
}
fn auth(&self) -> ModelProviderFuture<'_, Option<CodexAuth>> {
self.inner.auth()
}
fn account_state(&self) -> ProviderAccountResult {
self.inner.account_state()
}
fn recover_from_unauthorized(
&self,
) -> ModelProviderFuture<'_, codex_protocol::error::Result<ProviderUnauthorizedRecovery>> {
self.attempts.fetch_add(1, Ordering::Relaxed);
Box::pin(async move {
if self.should_fail {
Err(CodexErr::Io(std::io::Error::other(
"provider recovery failed",
)))
} else {
Ok(ProviderUnauthorizedRecovery::Recovered)
}
})
}
fn models_manager(
&self,
codex_home: PathBuf,
config_model_catalog: Option<ModelsResponse>,
) -> SharedModelsManager {
self.inner.models_manager(codex_home, config_model_catalog)
}
}
#[tokio::test]
async fn provider_owned_auth_recovery_is_bounded_and_preserves_unauthorized_failures() {
for should_fail in [false, true] {
let attempts = Arc::new(AtomicUsize::new(0));
let provider: SharedModelProvider = Arc::new(TestRecoveryProvider {
inner: test_model_provider(),
should_fail,
attempts: Arc::clone(&attempts),
});
assert!(provider.auth_manager().is_none());
let unauthorized = || TransportError::Http {
status: http::StatusCode::UNAUTHORIZED,
url: Some("https://example.com/v1/responses".to_string()),
headers: None,
body: Some("unauthorized".to_string()),
};
let mut auth_recovery = None;
let mut provider_auth_recovery_attempted = false;
let telemetry = test_session_telemetry();
let result = super::handle_unauthorized(
unauthorized(),
&mut auth_recovery,
&mut provider_auth_recovery_attempted,
&telemetry,
&provider,
)
.await;
let error = if should_fail {
result.expect_err("failed provider recovery should return the original error")
} else {
let recovered = result.expect("provider recovery should succeed without AuthManager");
assert_eq!(
(recovered.mode, recovered.phase),
("provider", "provider_refresh")
);
super::handle_unauthorized(
unauthorized(),
&mut auth_recovery,
&mut provider_auth_recovery_attempted,
&telemetry,
&provider,
)
.await
.expect_err("provider recovery should not run more than once")
};
match error.details() {
CodexErrorDetails::UnexpectedStatus(response) => {
assert_eq!(response.status, http::StatusCode::UNAUTHORIZED);
assert_eq!(response.body, "unauthorized");
}
other => panic!("unexpected error after provider recovery: {other}"),
}
assert_eq!(attempts.load(Ordering::Relaxed), 1);
}
}
#[tokio::test] #[tokio::test]
async fn dropped_backpressured_response_stream_traces_cancelled_partial_output() async fn dropped_backpressured_response_stream_traces_cancelled_partial_output()
-> anyhow::Result<()> { -> anyhow::Result<()> {
+1
View File
@@ -23,6 +23,7 @@ pub use provider::ProviderAccountError;
pub use provider::ProviderAccountResult; pub use provider::ProviderAccountResult;
pub use provider::ProviderAccountState; pub use provider::ProviderAccountState;
pub use provider::ProviderCapabilities; pub use provider::ProviderCapabilities;
pub use provider::ProviderUnauthorizedRecovery;
pub use provider::RemoteCompactionSupport; pub use provider::RemoteCompactionSupport;
pub use provider::SharedModelProvider; pub use provider::SharedModelProvider;
pub use provider::create_model_provider; pub use provider::create_model_provider;
+28
View File
@@ -7,6 +7,7 @@ use std::sync::Arc;
use codex_api::ApiError; use codex_api::ApiError;
use codex_api::Provider; use codex_api::Provider;
use codex_api::SharedAuthProvider; use codex_api::SharedAuthProvider;
use codex_api::TransportError;
use codex_api::is_azure_responses_provider; use codex_api::is_azure_responses_provider;
use codex_login::AuthManager; use codex_login::AuthManager;
use codex_login::CodexAuth; use codex_login::CodexAuth;
@@ -71,6 +72,15 @@ pub struct ProviderAccountState {
pub requires_openai_auth: bool, pub requires_openai_auth: bool,
} }
/// Outcome of a provider-owned attempt to recover from an authentication failure.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProviderUnauthorizedRecovery {
/// The provider has no provider-specific authentication recovery configured.
NotConfigured,
/// The provider recovered its authentication state and the request can be retried.
Recovered,
}
/// Error returned when a provider cannot construct its app-visible account state. /// Error returned when a provider cannot construct its app-visible account state.
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProviderAccountError { pub enum ProviderAccountError {
@@ -160,6 +170,24 @@ pub trait ModelProvider: fmt::Debug + Send + Sync {
/// manager throughout the codebase; that is a larger refactor than this change. /// manager throughout the codebase; that is a larger refactor than this change.
fn auth_manager(&self) -> Option<Arc<AuthManager>>; fn auth_manager(&self) -> Option<Arc<AuthManager>>;
/// Returns whether this transport failure can be recovered by provider-scoped authentication.
///
/// The default preserves existing unauthorized-response handling. Providers with other
/// authentication failure shapes may recognize additional response or request-signing errors.
fn is_recoverable_auth_error(&self, error: &TransportError) -> bool {
matches!(
error,
TransportError::Http { status, .. } if *status == http::StatusCode::UNAUTHORIZED
)
}
/// Attempts provider-owned authentication recovery before using the auth manager.
fn recover_from_unauthorized(
&self,
) -> ModelProviderFuture<'_, codex_protocol::error::Result<ProviderUnauthorizedRecovery>> {
Box::pin(async { Ok(ProviderUnauthorizedRecovery::NotConfigured) })
}
/// Returns the current provider-scoped auth value, if one is configured. /// Returns the current provider-scoped auth value, if one is configured.
fn auth(&self) -> ModelProviderFuture<'_, Option<CodexAuth>>; fn auth(&self) -> ModelProviderFuture<'_, Option<CodexAuth>>;