Preserve provider model catalogs when refreshes fail (#55958)

## Why

A failed provider catalog refresh could discard model metadata and disable tool search on subsequent turns, even after a successful discovery.

## What changed

Keep the last successful catalog, ETag, and original freshness when refreshes fail without an access rejection. Continue invalidating cached catalogs for HTTP `401`/`403`, rejected refresh tokens, and gateway credentials that require login. Distinguish `GatewayLoginRequired` from login in progress so only the former invalidates the catalog.

Successful empty responses remain authoritative, and cached catalogs remain scoped to the provider and authentication identity.

## Testing

Add regression coverage for catalog preservation across refresh errors and restarts, access rejection invalidation, identity isolation, and tool search and instruction stability on the next turn after a failed refresh.

GitOrigin-RevId: 9dd7c73ac2cc1fd49664efd93b839baa94e05c49
This commit is contained in:
Abhinav
2026-10-21 18:43:15 +00:00
committed by copyberry
parent 9e12b14891
commit 067249a283
6 changed files with 388 additions and 47 deletions
@@ -1589,3 +1589,81 @@ async fn model_catalog_url_supplies_conversation_model_and_instructions() -> Res
assert_eq!(request["text"].get("verbosity"), None);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn failed_catalog_refresh_preserves_tool_search_on_the_next_turn() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = MockServer::start().await;
let mut model = codex_models_manager::model_info::model_info_from_slug("gateway-model");
model.visibility = ModelVisibility::List;
model.supports_search_tool = true;
model.used_fallback_model_metadata = false;
let catalog = ModelsResponse {
models: vec![model],
};
Mock::given(method("GET"))
.and(path("/codex/models"))
.respond_with(ResponseTemplate::new(200).set_body_json(&catalog))
.up_to_n_times(1)
.expect(1)
.mount(&server)
.await;
let responses = mount_sse_sequence(
&server,
vec![
sse(vec![ev_response_created("first"), ev_completed("first")]),
sse(vec![ev_response_created("second"), ev_completed("second")]),
],
)
.await;
let catalog_url = format!("{}/codex/models", server.uri());
let test = test_codex()
.with_auth(CodexAuth::from_api_key("gateway-api-key"))
.with_model("gateway-model")
.with_config(move |config| {
config.model_provider.model_catalog_url = Some(catalog_url.into());
config
.features
.enable(Feature::ApiKeyModelDiscovery)
.unwrap();
config.features.enable(Feature::Collab).unwrap();
config.features.disable(Feature::MultiAgentV2).unwrap();
})
.build_with_auto_env(&server)
.await?;
test.submit_turn("first turn").await?;
let first_tools = responses.single_request().body_json()["tools"].clone();
assert!(
first_tools
.as_array()
.unwrap()
.iter()
.any(|tool| tool["type"] == "tool_search"),
"the initial catalog must enable deferred tool search"
);
Mock::given(method("GET"))
.and(path("/codex/models"))
.respond_with(ResponseTemplate::new(503))
.expect(1..)
.mount(&server)
.await;
let manager = test.thread_manager.get_models_manager();
manager
.refresh_if_new_etag(
"changed-etag".into(),
codex_core::test_support::default_http_client_factory(),
)
.await;
assert_eq!(manager.get_remote_models().await, catalog.models);
test.submit_turn("second turn").await?;
let requests = responses.requests();
assert_eq!(requests.len(), 2);
assert_eq!(requests[1].body_json()["tools"], first_tools);
assert_eq!(
requests[1].instructions_text(),
requests[0].instructions_text()
);
Ok(())
}
+9 -1
View File
@@ -14,6 +14,7 @@ use codex_model_provider_info::GatewayOAuthConfig;
use codex_model_provider_info::GatewayOAuthDelivery;
use codex_model_provider_info::ModelProviderInfo;
use codex_protocol::error::CodexErr;
use codex_protocol::error::CodexErrorDetails;
use codex_protocol::error::Result;
use http::HeaderMap;
use http::HeaderName;
@@ -39,7 +40,14 @@ pub(crate) async fn compose_auth(
.get_ref()
.and_then(|source| source.downcast_ref::<GatewayAuthError>())
{
return CodexErr::InvalidRequest(login_error.to_string());
return match login_error {
GatewayAuthError::LoginRequired => {
CodexErrorDetails::GatewayLoginRequired(login_error.to_string()).into()
}
GatewayAuthError::LoginInProgress => {
CodexErr::InvalidRequest(login_error.to_string())
}
};
}
// Issuer errors may echo arbitrary credentials from configured URLs. Keep the
// diagnostic safe and bounded, preserving retries for transient failures.
@@ -21,6 +21,8 @@ use codex_login::default_client::create_client_for_route;
use codex_model_provider_info::GatewayOAuthDelivery;
use codex_model_provider_info::ModelProviderInfo;
use codex_models_manager::manager::RefreshStrategy;
use codex_protocol::error::CodexErrorDetails;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::SessionSource;
use http::HeaderMap;
use http::HeaderName;
@@ -528,6 +530,91 @@ async fn gateway_setup_errors_are_reported_when_authentication_is_requested() {
assert!(server.received_requests().await.unwrap().is_empty());
}
#[tokio::test]
async fn gateway_login_required_invalidates_the_last_successful_catalog() {
let server = MockServer::start().await;
let home = tempfile::tempdir().unwrap();
let primary = AuthManager::from_auth_for_testing_with_home(
CodexAuth::from_api_key("primary-token"),
home.path().to_path_buf(),
);
let info = ModelProviderInfo {
model_catalog_url: Some(format!("{}/models", server.uri()).into()),
gateway_oauth: Some(GatewayOAuthConfig {
authorization_url: format!("{}/authorize", server.uri()),
token_url: format!("{}/token", server.uri()),
client_id: "client".into(),
resource: None,
scopes: vec![],
redirect_port: None,
delivery: GatewayOAuthDelivery::Header {
name: "x-gateway-auth".into(),
scheme: "Bearer".into(),
},
}),
..ModelProviderInfo::create_openai_provider(Some(server.uri()))
};
let login_control = codex_login::GatewayLoginControl::for_runtime(&primary.runtime_config());
login_control.require_explicit_login();
let gateway = seed_gateway_auth(
&info,
&primary,
json!({"access_token": "first", "refresh_token": "revoked", "expires_at": i64::MAX}),
);
let provider = create_model_provider(info, Some(primary));
let models = provider.models_manager(
home.path().to_path_buf(),
/*config_model_catalog*/ None,
);
models.set_api_key_model_discovery_enabled(/*enabled*/ true);
let factory = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault);
let catalog = codex_models_manager::bundled_models_response().unwrap();
Mock::given(method("GET"))
.and(path("/models"))
.respond_with(ResponseTemplate::new(200).set_body_json(&catalog))
.expect(1)
.mount(&server)
.await;
assert_eq!(
models
.raw_model_catalog(RefreshStrategy::Online, factory.clone())
.await,
catalog
);
Mock::given(method("POST"))
.and(path("/token"))
.respond_with(ResponseTemplate::new(400).set_body_json(json!({"error": "invalid_grant"})))
.expect(1)
.mount(&server)
.await;
let error = gateway.refresh_access_token("first").await.unwrap_err();
assert!(matches!(
error
.get_ref()
.and_then(|source| source.downcast_ref::<codex_login::GatewayAuthError>()),
Some(codex_login::GatewayAuthError::LoginRequired)
));
assert_eq!(
models
.raw_model_catalog(RefreshStrategy::Online, factory.clone())
.await,
codex_protocol::openai_models::ModelsResponse::default()
);
// Recreate the manager to verify persistent invalidation as well as the in-memory result.
let restarted = provider.models_manager(
home.path().to_path_buf(),
/*config_model_catalog*/ None,
);
restarted.set_api_key_model_discovery_enabled(/*enabled*/ true);
assert_eq!(
restarted
.raw_model_catalog(RefreshStrategy::Offline, factory)
.await,
codex_protocol::openai_models::ModelsResponse::default()
);
}
#[tokio::test]
async fn gateway_login_states_stop_request_retries() {
let server = MockServer::start().await;
@@ -556,11 +643,20 @@ async fn gateway_login_states_stop_request_retries() {
let provider = create_model_provider(info, Some(primary));
let gateway = provider.gateway_auth_manager().unwrap().unwrap();
let error = provider.api_auth().await.err().expect("login required");
assert!(matches!(
error.details(),
CodexErrorDetails::GatewayLoginRequired(_)
));
assert_eq!(
(error.to_string(), error.retry_delay(/*retry_count*/ 1)),
(
error.to_string(),
error.retry_delay(/*retry_count*/ 1),
error.to_codex_protocol_error()
),
(
codex_login::GatewayAuthError::LoginRequired.to_string(),
None
None,
CodexErrorInfo::Other
),
);
let cancel = tokio::sync::Notify::new();
@@ -571,6 +667,10 @@ async fn gateway_login_states_stop_request_retries() {
let check = async {
url_rx.await.unwrap();
let error = provider.api_auth().await.err().expect("login in progress");
assert!(matches!(
error.details(),
CodexErrorDetails::InvalidRequest(_)
));
assert_eq!(
(error.to_string(), error.retry_delay(/*retry_count*/ 1)),
(
+15 -2
View File
@@ -10,6 +10,7 @@ use codex_http_client::HttpClientFactory;
use codex_login::AuthManager;
use codex_protocol::auth::AuthMode;
use codex_protocol::config_types::CollaborationModeMask;
use codex_protocol::error::CodexErrorDetails;
use codex_protocol::error::Result as CoreResult;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ModelPreset;
@@ -594,7 +595,19 @@ impl OpenAiModelsManager {
{
Ok(response) => response,
Err(error) => {
if matches!(&self.catalog_source, CatalogSource::ExplicitProvider(_)) {
// A failed fetch is not an authoritative empty catalog. Keep the last
// successful snapshot (and its original freshness) unless access was
// explicitly rejected. Provider/auth identity checks still govern reuse.
let is_provider_catalog =
matches!(&self.catalog_source, CatalogSource::ExplicitProvider(_));
let access_denied = matches!(error.http_status_code_value(), Some(401 | 403));
let credentials_rejected = matches!(
error.details(),
CodexErrorDetails::RefreshTokenFailed(_)
| CodexErrorDetails::GatewayLoginRequired(_)
);
if is_provider_catalog && (access_denied || credentials_rejected) {
{
let mut current = self.remote_models.write().await;
if request_identity != self.endpoint_client.identity() {
@@ -724,7 +737,7 @@ impl OpenAiModelsManager {
};
{
let current = self.remote_models.read().await;
// A failed fetch must not be replaced by stale cache data even if invalidation failed.
// An access rejection must not be replaced by stale cache data if invalidation failed.
if current.fetched_at == DateTime::<Utc>::MIN_UTC
&& current.identity.as_ref() == Some(&identity)
{
+180 -42
View File
@@ -1120,59 +1120,197 @@ async fn authoritative_catalog_preserves_hidden_models_and_matches_exact_ids() {
}
#[tokio::test]
async fn authoritative_catalog_failure_invalidates_cache_until_refresh_succeeds() {
// Like a native TTL backend, this cache may return an invalidated stored entry.
let cache = Arc::new(TestModelsCache::default());
let models = vec![remote_model(
"provider-model",
"Provider Model",
/*priority*/ 0,
)];
let endpoint = TestModelsEndpoint::with_command_auth(vec![
Ok(models.clone()),
Err(CodexErr::RequestTimeout),
Ok(models.clone()),
]);
let manager = OpenAiModelsManager::new_with_cache(
cache.clone(),
endpoint.clone(),
/*auth_manager*/ None,
)
.with_provider_catalog();
assert_eq!(
manager
.raw_model_catalog(RefreshStrategy::Online, DEFAULT_HTTP_CLIENT_FACTORY)
.await
.models,
models
);
for strategy in [RefreshStrategy::Online, RefreshStrategy::Offline] {
async fn authoritative_catalog_access_rejection_invalidates_cache_until_refresh_succeeds() {
for error in [
catalog_http_error(/*status*/ 401),
catalog_http_error(/*status*/ 403),
CodexErr::RefreshTokenFailed(codex_protocol::error::RefreshTokenFailedError::new(
codex_protocol::error::RefreshTokenFailedReason::Revoked,
"credentials revoked",
)),
CodexErrorDetails::GatewayLoginRequired(
codex_login::GatewayAuthError::LoginRequired.to_string(),
)
.into(),
] {
// Like a native TTL backend, this cache may return an invalidated stored entry.
let cache = Arc::new(TestModelsCache::default());
let models = vec![remote_model(
"provider-model",
"Provider Model",
/*priority*/ 0,
)];
let endpoint = TestModelsEndpoint::with_command_auth(vec![
Ok(models.clone()),
Err(error),
Ok(models.clone()),
]);
let manager = OpenAiModelsManager::new_with_cache(
cache.clone(),
endpoint.clone(),
/*auth_manager*/ None,
)
.with_provider_catalog();
assert_eq!(
manager
.raw_model_catalog(strategy, DEFAULT_HTTP_CLIENT_FACTORY)
.raw_model_catalog(RefreshStrategy::Online, DEFAULT_HTTP_CLIENT_FACTORY)
.await
.models,
models
);
for strategy in [RefreshStrategy::Online, RefreshStrategy::Offline] {
assert_eq!(
manager
.raw_model_catalog(strategy, DEFAULT_HTTP_CLIENT_FACTORY)
.await,
ModelsResponse::default()
);
}
let restarted =
OpenAiModelsManager::new_with_cache(cache, endpoint, /*auth_manager*/ None)
.with_provider_catalog();
assert_eq!(
restarted
.raw_model_catalog(RefreshStrategy::Offline, DEFAULT_HTTP_CLIENT_FACTORY)
.await,
ModelsResponse::default()
);
assert_eq!(
restarted
.raw_model_catalog(
RefreshStrategy::OnlineIfUncached,
DEFAULT_HTTP_CLIENT_FACTORY,
)
.await
.models,
models
);
}
let restarted =
OpenAiModelsManager::new_with_cache(cache, endpoint, /*auth_manager*/ None)
}
fn catalog_http_error(status: u16) -> CodexErr {
CodexErr::UnexpectedStatus(codex_protocol::error::UnexpectedResponseError {
status: status.try_into().expect("valid status"),
body: String::new(),
user_message: None,
url: None,
cf_ray: None,
request_id: None,
identity_authorization_error: None,
identity_error_code: None,
})
}
#[tokio::test]
async fn authoritative_catalog_refresh_failure_preserves_snapshot_and_freshness() {
for error in [
CodexErr::RequestTimeout,
CodexErr::Stream("network unavailable".to_string()),
catalog_http_error(/*status*/ 429),
catalog_http_error(/*status*/ 503),
catalog_http_error(/*status*/ 404),
CodexErr::InvalidRequest("malformed catalog".to_string()),
CodexErr::InvalidRequest(codex_login::GatewayAuthError::LoginInProgress.to_string()),
] {
let cache = Arc::new(TestModelsCache::default());
let mut model = remote_model("provider-model", "Provider Model", /*priority*/ 0);
model.supports_search_tool = true;
model.context_window = Some(1_000_000);
let models = vec![model];
let endpoint = Arc::new(TestModelsEndpoint {
has_command_auth: true,
etag: Some("last-successful-etag".into()),
responses: Mutex::new(vec![Ok(models.clone()), Err(error), Ok(Vec::new())].into()),
..TestModelsEndpoint::default()
});
let manager = OpenAiModelsManager::new_with_cache(
cache.clone(),
endpoint.clone(),
/*auth_manager*/ None,
)
.with_provider_catalog();
let initial = manager
.raw_model_catalog(RefreshStrategy::Online, DEFAULT_HTTP_CLIENT_FACTORY)
.await;
let initial_info = manager
.get_model_info("provider-model", &ModelsManagerConfig::default())
.await;
let stored = cache.stored_entries();
assert_eq!(initial.models, models);
manager
.refresh_if_new_etag("new-etag".to_string(), DEFAULT_HTTP_CLIENT_FACTORY)
.await;
assert_eq!(manager.get_remote_models().await, models);
assert_eq!(
manager
.get_model_info("provider-model", &ModelsManagerConfig::default())
.await,
initial_info
);
assert_eq!(manager.remote_models.read().await.clone(), stored[0]);
assert_eq!(cache.stored_entries(), stored);
let restarted = OpenAiModelsManager::new_with_cache(
cache.clone(),
endpoint.clone(),
/*auth_manager*/ None,
)
.with_provider_catalog();
assert_eq!(
restarted
.raw_model_catalog(
RefreshStrategy::OnlineIfUncached,
DEFAULT_HTTP_CLIENT_FACTORY,
)
.await,
initial
);
assert_eq!(endpoint.fetch_count(), 2);
// A successful empty response remains authoritative and is persisted.
assert_eq!(
manager
.raw_model_catalog(RefreshStrategy::Online, DEFAULT_HTTP_CLIENT_FACTORY)
.await,
ModelsResponse::default()
);
assert_eq!(
cache.stored_entries().last().unwrap().models,
Vec::<ModelInfo>::new()
);
}
}
#[tokio::test]
async fn failed_discovery_does_not_reuse_a_catalog_from_another_identity() {
let stored = ModelsCacheEntry {
fetched_at: Utc::now(),
etag: Some("previous-etag".into()),
client_version: Some(crate::client_version_to_whole()),
identity: Some("previous-provider-or-account".into()),
models: vec![remote_model(
"private-model",
"Private Model",
/*priority*/ 0,
)],
};
let cache = TestModelsCache::with_entry(stored.clone());
let endpoint = TestModelsEndpoint::with_command_auth(vec![Err(CodexErr::RequestTimeout)]);
let manager =
OpenAiModelsManager::new_with_cache(cache.clone(), endpoint, /*auth_manager*/ None)
.with_provider_catalog();
assert_eq!(
restarted
.raw_model_catalog(RefreshStrategy::Offline, DEFAULT_HTTP_CLIENT_FACTORY)
manager
.raw_model_catalog(
RefreshStrategy::OnlineIfUncached,
DEFAULT_HTTP_CLIENT_FACTORY
)
.await,
ModelsResponse::default()
);
assert_eq!(
restarted
.raw_model_catalog(
RefreshStrategy::OnlineIfUncached,
DEFAULT_HTTP_CLIENT_FACTORY,
)
.await
.models,
models
);
assert_eq!(cache.entry.lock().unwrap().as_ref(), Some(&stored));
assert_eq!(manager.get_remote_models().await, Vec::<ModelInfo>::new());
}
#[tokio::test]
+4
View File
@@ -201,6 +201,9 @@ pub enum CodexErrorDetails {
UnsupportedOperation(String),
#[error("{0}")]
RefreshTokenFailed(RefreshTokenFailedError),
/// Gateway credentials require user action before requests can resume.
#[error("{0}")]
GatewayLoginRequired(String),
#[error("Fatal error: {0}")]
Fatal(String),
// -----------------------------------------------------------------
@@ -422,6 +425,7 @@ impl CodexErr {
| CodexErrorDetails::InvalidPrompt { .. }
| CodexErrorDetails::ToolCollision(_)
| CodexErrorDetails::RefreshTokenFailed(_)
| CodexErrorDetails::GatewayLoginRequired(_)
| CodexErrorDetails::UnsupportedOperation(_)
| CodexErrorDetails::Sandbox(_)
| CodexErrorDetails::LandlockSandboxExecutableNotProvided