diff --git a/backend/crates/gateway-admin/src/model/provider_credentials.rs b/backend/crates/gateway-admin/src/model/provider_credentials.rs index 85882ae15..9a6dc2d8b 100644 --- a/backend/crates/gateway-admin/src/model/provider_credentials.rs +++ b/backend/crates/gateway-admin/src/model/provider_credentials.rs @@ -683,6 +683,8 @@ pub struct PreparedCredentialRotationFacts { pub plan_type: Option, /// Token 刷新保留提交时的资料，不以准备阶段的副本覆盖新套餐。 pub preserve_profile: bool, + /// 连接配置变更保留当前凭据健康状态与错误事实。 + pub preserve_credential_state: bool, pub provider_material: ProviderDocument, pub has_refresh_token: bool, pub access_token_expires_at: Option>, @@ -704,6 +706,13 @@ impl PreparedCredentialRotation { Self { facts, guard } } + /// 仅修改连接配置时，不把配置更新视为凭据恢复成功。 + #[must_use] + pub fn preserving_credential_state(mut self) -> Self { + self.facts.preserve_credential_state = true; + self + } + #[must_use] pub const fn facts(&self) -> &PreparedCredentialRotationFacts { &self.facts diff --git a/backend/crates/gateway-admin/tests/use_case/accounts.rs b/backend/crates/gateway-admin/tests/use_case/accounts.rs index 39b81306d..aecfb1d9e 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts.rs @@ -246,6 +246,7 @@ impl FakeProviderAdmin { email: account.email.clone(), plan_type: account.plan_type.clone(), preserve_profile: false, + preserve_credential_state: false, provider_material: document(), has_refresh_token: account.has_refresh_token, access_token_expires_at: account @@ -3088,6 +3089,7 @@ fn plugin_rotation_facts(account: &AccountRecord) -> PreparedCredentialRotationF email: account.email.clone(), plan_type: account.plan_type.clone(), preserve_profile: true, + preserve_credential_state: false, provider_material: document(), has_refresh_token: account.has_refresh_token, access_token_expires_at: account.access_token_expires_at, diff --git a/backend/crates/gateway-api/src/admin/accounts/credentials.rs b/backend/crates/gateway-api/src/admin/accounts/credentials.rs index 9397b98f9..0c5480a68 100644 --- a/backend/crates/gateway-api/src/admin/accounts/credentials.rs +++ b/backend/crates/gateway-api/src/admin/accounts/credentials.rs @@ -259,18 +259,20 @@ impl UpdateAccountRequest { } } -/// 编辑 OpenAI API Key 账号的连接设置；省略密钥时保留已保存的值。 +/// 编辑 OpenAI 账号的连接设置；OAuth 仅接受传输方式。 #[derive(Clone, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct AccountConnectionUpdateRequest { - pub base_url: String, + pub base_url: Option, pub transport: String, pub api_key: Option, } impl AccountConnectionUpdateRequest { fn validate(&self) -> Result<(), WireValidationError> { - require_text(&self.base_url, 2048, "connection.baseUrl")?; + if let Some(base_url) = &self.base_url { + require_text(base_url, 2048, "connection.baseUrl")?; + } if !matches!(self.transport.as_str(), "http" | "prefer_websocket") { return Err(WireValidationError::new("connection.transport")); } @@ -285,10 +287,11 @@ impl AccountConnectionUpdateRequest { } fn into_document(self) -> ProviderDocument { - let mut material = Map::from_iter([ - ("base_url".to_owned(), Value::String(self.base_url)), - ("transport".to_owned(), Value::String(self.transport)), - ]); + let mut material = + Map::from_iter([("transport".to_owned(), Value::String(self.transport))]); + if let Some(base_url) = self.base_url { + material.insert("base_url".to_owned(), Value::String(base_url)); + } if let Some(key) = self.api_key { material.insert("api_key".to_owned(), Value::String(key)); } diff --git a/backend/crates/gateway-api/tests/admin/accounts/handlers.rs b/backend/crates/gateway-api/tests/admin/accounts/handlers.rs index c89b8d3a6..1f584635a 100644 --- a/backend/crates/gateway-api/tests/admin/accounts/handlers.rs +++ b/backend/crates/gateway-api/tests/admin/accounts/handlers.rs @@ -39,14 +39,21 @@ async fn connection_update_requires_admin_and_validates_before_calling_the_servi "groupIds":[], "connection":{"baseUrl":"https://api.example.invalid/v1", "transport":"http"} }); - for (authenticated, transport, expected) in [ - (false, "http", StatusCode::UNAUTHORIZED), - (true, "invalid", StatusCode::BAD_REQUEST), + for (authenticated, transport, oauth, expected) in [ + (false, "http", false, StatusCode::UNAUTHORIZED), + (true, "invalid", false, StatusCode::BAD_REQUEST), // 夹具没有凭据 Store，合法输入必须进入服务，不能按普通设置静默保存。 - (true, "http", StatusCode::SERVICE_UNAVAILABLE), + (true, "http", false, StatusCode::SERVICE_UNAVAILABLE), + (true, "http", true, StatusCode::SERVICE_UNAVAILABLE), ] { let mut input = input.clone(); input["connection"]["transport"] = serde_json::json!(transport); + if oauth { + input["connection"] + .as_object_mut() + .unwrap() + .remove("baseUrl"); + } let mut request = Request::builder() .method("POST") .uri("/api/admin/accounts/update") diff --git a/backend/crates/gateway-plugin/runtime/src/callback/accounts.rs b/backend/crates/gateway-plugin/runtime/src/callback/accounts.rs index cdc99d8ac..c21c9fb93 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/accounts.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/accounts.rs @@ -250,6 +250,7 @@ fn prepare_replace( email: facts.email, plan_type: facts.plan_type, preserve_profile: false, + preserve_credential_state: false, provider_material: ProviderDocument::new(OpaqueProviderData::new(facts.material)), has_refresh_token: facts.has_refresh_token, access_token_expires_at: credential_timestamp(facts.access_token_expires_at_ms) diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs index 0bb927fcc..8537dde38 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs @@ -1068,6 +1068,7 @@ pub(super) fn prepare_rotation( access_token_expires_at: prepared.access_token_expires_at, next_refresh_at: prepared.next_refresh_at, preserve_profile: prepared.preserve_profile, + preserve_credential_state: prepared.preserve_credential_state, }, audit: mutation_audit( context, diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/repository.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/repository.rs index ae19f9543..0c76371c5 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/repository.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/repository.rs @@ -834,16 +834,16 @@ pub(crate) async fn rotate_provider_account_in_transaction( upstream_user_id = case when $11::boolean then $12::text else upstream_user_id end, upstream_account_id = case when $11::boolean then $13::text else upstream_account_id end, credential_state = case - when not enabled then credential_state + when $15::boolean or not enabled then credential_state when $11::boolean or credential_state <> 'unknown' then 'ready' else 'unknown' end, credential_observed_at = case - when not enabled then credential_observed_at + when $15::boolean or not enabled then credential_observed_at else now() end, - last_error_reason = case when enabled then null else last_error_reason end, - last_error_message = case when enabled then null else last_error_message end, + last_error_reason = case when enabled and not $15::boolean then null else last_error_reason end, + last_error_message = case when enabled and not $15::boolean then null else last_error_message end, updated_at = greatest(now(), updated_at) where id = $1 and provider_kind = $2 and credential_revision = $3 @@ -863,6 +863,7 @@ pub(crate) async fn rotate_provider_account_in_transaction( .bind(upstream_user_id) .bind(upstream_account_id) .bind(update.preserve_profile) + .bind(update.preserve_credential_state) .fetch_optional(&mut **transaction) .await .map_err(|error| { diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/rows.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/rows.rs index babadcf85..1795eecb7 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/rows.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/rows.rs @@ -361,6 +361,7 @@ pub struct ProviderCredentialUpdate { pub access_token_expires_at: Option>, pub next_refresh_at: Option>, pub preserve_profile: bool, + pub preserve_credential_state: bool, } impl fmt::Debug for ProviderCredentialUpdate { diff --git a/backend/crates/gateway-store/tests/postgres/provider_accounts/authorization.rs b/backend/crates/gateway-store/tests/postgres/provider_accounts/authorization.rs index 0d8787ff1..084355b2f 100644 --- a/backend/crates/gateway-store/tests/postgres/provider_accounts/authorization.rs +++ b/backend/crates/gateway-store/tests/postgres/provider_accounts/authorization.rs @@ -157,6 +157,7 @@ async fn reauthorization_replay_keeps_the_committed_revision_and_later_profile_e email: None, plan_type: None, preserve_profile: true, + preserve_credential_state: false, provider_material: ProviderDocument::new(OpaqueProviderData::new( json!({"access_token":"rotated-test-only"}) .as_object() diff --git a/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs b/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs index 6ed23453f..21e1e994d 100644 --- a/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs @@ -3032,6 +3032,7 @@ pub(super) fn account(id: &str, upstream_user_id: &str) -> NewProviderAccount { fn credential_update(account_id: &str, revision: u64, marker: &str) -> ProviderCredentialUpdate { ProviderCredentialUpdate { preserve_profile: false, + preserve_credential_state: false, account_id: account_id.to_owned(), expected_revision: Revision::new(revision).expect("credential revision"), provider_credentials_json: credential_json(marker), @@ -3547,3 +3548,60 @@ async fn adaptive_concurrency_handles_unlimited_and_latest_locked_settings_witho ); database.close().await; } + +#[tokio::test] +async fn connection_configuration_update_preserves_credential_health() { + let Some(database) = TestDatabase::create("connection_configuration_health").await else { + return; + }; + let repository = PgProviderAccountRepository::new(database.pool.clone()); + const ID: &str = "acct_transport_health"; + repository + .insert_provider_account(account(ID, "transport-health-user")) + .await + .unwrap(); + sqlx::query("update provider_accounts set credential_state = 'expired', last_error_reason = 'credential_expired', last_error_message = 'test expiration' where id = $1") + .bind(ID).execute(&database.pool).await.unwrap(); + let before: serde_json::Value = + sqlx::query_scalar("select to_jsonb(a) from provider_accounts a where id = $1") + .bind(ID) + .fetch_one(&database.pool) + .await + .unwrap(); + let mut credential = credential_update(ID, 1, "same-secret"); + credential.preserve_credential_state = true; + credential.preserve_profile = true; + repository + .rotate_provider_account(RotateProviderAccount { + scope: ProviderAccountAdminScope { + provider_kind: "openai".to_owned(), + }, + profile: profile(ID, "must not replace name"), + replacement_identity: None, + credential, + settings: None, + audit: audit("audit_transport_health", "rotate", ID), + }) + .await + .unwrap(); + let after: serde_json::Value = + sqlx::query_scalar("select to_jsonb(a) from provider_accounts a where id = $1") + .bind(ID) + .fetch_one(&database.pool) + .await + .unwrap(); + for field in [ + "name", + "credential_state", + "credential_observed_at", + "last_error_reason", + "last_error_message", + ] { + assert_eq!( + after[field], before[field], + "configuration update changed {field}" + ); + } + assert_eq!(after["credential_revision"], 2); + database.close().await; +} diff --git a/backend/crates/providers/openai/src/admin.rs b/backend/crates/providers/openai/src/admin.rs index 762ef39b4..eca52d1aa 100644 --- a/backend/crates/providers/openai/src/admin.rs +++ b/backend/crates/providers/openai/src/admin.rs @@ -553,6 +553,21 @@ impl ProviderAdmin for OpenAiAdminProvider { .map_err(map_credential_admin_error)?; return prepared_rotation(prepared, command.account.provider_kind); } + if command + .provider_material + .expose_to_provider() + .expose_to_provider() + .contains_key("transport") + { + let prepared = CodexCredentialAdmin + .prepare_transport_update( + current, + Value::Object(command.provider_material.into_provider_data().into_inner()), + ) + .map_err(map_credential_admin_error)?; + return prepared_rotation(prepared, command.account.provider_kind) + .map(PreparedCredentialRotation::preserving_credential_state); + } let mut secret = rotation_secret(command.provider_material)?; if secret.id_token.is_none() { let runtime = CodexCredentialCodec::decode(¤t.credential) @@ -602,23 +617,23 @@ impl ProviderAdmin for OpenAiAdminProvider { &self, account_id: &ProviderAccountId, ) -> Result, ProviderAdminError> { - let account = self.account(account_id).await?; - if account.authentication_kind() != crate::credential::CODEX_AUTHENTICATION_KIND_API_KEY { - return Ok(None); - } + self.account(account_id).await?; let current = self .accounts .load_current_credential(account_id) .await .map_err(map_store_error)?; - let crate::credential::CodexCredentialData::ApiKey(data) = - CodexCredentialCodec::decode_complete(¤t.credential) - .map_err(|_| provider_admin_error(ProviderAdminErrorKind::Invalid))? - else { - return Err(provider_admin_error(ProviderAdminErrorKind::Invalid)); + let data = CodexCredentialCodec::decode_complete(¤t.credential) + .map_err(|_| provider_admin_error(ProviderAdminErrorKind::Invalid))?; + let value = match data { + crate::credential::CodexCredentialData::ApiKey(data) => { + serde_json::to_value(data.configuration()) + .map_err(|_| provider_admin_error(ProviderAdminErrorKind::Internal))? + } + crate::credential::CodexCredentialData::OAuth(data) => { + serde_json::json!({"transport": data.transport}) + } }; - let value = serde_json::to_value(data.configuration()) - .map_err(|_| provider_admin_error(ProviderAdminErrorKind::Internal))?; let object = value .as_object() .cloned() @@ -966,6 +981,7 @@ fn prepared_rotation( email: profile.email, plan_type: profile.plan_type, preserve_profile, + preserve_credential_state: false, provider_material: ProviderDocument::new(OpaqueProviderData::new( credential.into_inner(), )), diff --git a/backend/crates/providers/openai/src/credential/admin.rs b/backend/crates/providers/openai/src/credential/admin.rs index 667188fc7..2cb0287ad 100644 --- a/backend/crates/providers/openai/src/credential/admin.rs +++ b/backend/crates/providers/openai/src/credential/admin.rs @@ -24,7 +24,7 @@ use serde::Serialize; use serde_json::Value; use thiserror::Error; -use super::api_key::{ApiKeyCredentialData, ApiKeyTransport, CODEX_AUTHENTICATION_KIND_API_KEY}; +use super::api_key::{ApiKeyCredentialData, CODEX_AUTHENTICATION_KIND_API_KEY}; use super::recovery_log::{CodexOAuthRecoveryOperation, record_oauth_recovery}; use super::security::CodexCredentialCodec; use super::token_client::{ @@ -32,8 +32,8 @@ use super::token_client::{ }; use super::types::{ CODEX_AUTHENTICATION_KIND_OAUTH, CodexAccountProfile, CodexCredentialData, - CodexCredentialPrincipal, CodexOAuthMetadata, CodexOAuthSecret, parse_access_token_expiration, - parse_chatgpt_jwt_claims, + CodexCredentialPrincipal, CodexOAuthMetadata, CodexOAuthSecret, ResponsesTransport, + parse_access_token_expiration, parse_chatgpt_jwt_claims, }; const PROVIDER_NAME: &str = "openai"; @@ -257,7 +257,7 @@ struct CodexCprApiKeyExportAccount { authentication_kind: &'static str, base_url: String, api_key: String, - transport: ApiKeyTransport, + transport: ResponsesTransport, } #[derive(Serialize)] @@ -453,7 +453,7 @@ impl CodexCredentialAdmin { #[serde(deny_unknown_fields)] struct Rotation { base_url: String, - transport: ApiKeyTransport, + transport: ResponsesTransport, api_key: Option, } let rotation: Rotation = serde_json::from_value(material) @@ -496,6 +496,50 @@ impl CodexCredentialAdmin { }) } + pub(crate) fn prepare_transport_update( + &self, + current: LoadedCredential, + material: Value, + ) -> Result { + #[derive(serde::Deserialize)] + #[serde(deny_unknown_fields)] + struct Connection { + transport: ResponsesTransport, + } + let connection: Connection = serde_json::from_value(material) + .map_err(|_| CodexCredentialAdminError::InvalidInput)?; + let mut data = CodexCredentialCodec::decode_complete(¤t.credential) + .map_err(|_| CodexCredentialAdminError::InvalidCredential)?; + data.oauth_mut() + .ok_or(CodexCredentialAdminError::InvalidCredential)? + .transport = connection.transport; + let credential = CodexCredentialCodec::encode_complete(data) + .map_err(|_| CodexCredentialAdminError::InvalidCredential)?; + let profile = ProviderAccountUpdate { + account_id: current.account.id().clone(), + name: current.account.name().to_owned(), + email: current.account.email().map(str::to_owned), + plan_type: current.account.plan_type().map(str::to_owned), + }; + let credential = CredentialCasUpdate::new( + current.account.id().clone(), + current.account.revision(), + profile.clone(), + credential, + current.account.has_refresh_token(), + current.account.access_token_expires_at(), + current.account.next_refresh_at(), + ) + .map_err(|_| CodexCredentialAdminError::InvalidCredential)? + .preserving_profile(); + Ok(PreparedCodexCredentialRotation { + profile, + credential, + replacement_identity: None, + refresh_guards: None, + }) + } + pub fn prepare_import( &self, input: ImportCodexOAuthCredential, diff --git a/backend/crates/providers/openai/src/credential/api_key.rs b/backend/crates/providers/openai/src/credential/api_key.rs index 091d3d5e2..1921a8829 100644 --- a/backend/crates/providers/openai/src/credential/api_key.rs +++ b/backend/crates/providers/openai/src/credential/api_key.rs @@ -7,13 +7,7 @@ use serde::{Deserialize, Serialize}; pub const CODEX_AUTHENTICATION_KIND_API_KEY: &str = "api_key"; -#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum ApiKeyTransport { - #[default] - Http, - PreferWebsocket, -} +use super::types::ResponsesTransport; /// 可在管理端展示的上游设置，不包含密钥。 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -21,7 +15,7 @@ pub enum ApiKeyTransport { pub struct ApiKeyConfiguration { pub base_url: String, #[serde(default)] - pub transport: ApiKeyTransport, + pub transport: ResponsesTransport, } impl ApiKeyConfiguration { @@ -40,7 +34,7 @@ pub struct ApiKeyCredentialData { pub api_key: String, pub base_url: String, #[serde(default)] - pub transport: ApiKeyTransport, + pub transport: ResponsesTransport, } impl ApiKeyCredentialData { diff --git a/backend/crates/providers/openai/src/credential/mod.rs b/backend/crates/providers/openai/src/credential/mod.rs index 7e9ce07e4..b8cc9db69 100644 --- a/backend/crates/providers/openai/src/credential/mod.rs +++ b/backend/crates/providers/openai/src/credential/mod.rs @@ -25,7 +25,7 @@ pub(crate) use oauth::oauth_owner_ref; pub(crate) use types::parse_access_token_expiration; pub use api_key::{ - ApiKeyAuthentication, ApiKeyConfiguration, ApiKeyCredentialData, ApiKeyTransport, + ApiKeyAuthentication, ApiKeyConfiguration, ApiKeyCredentialData, CODEX_AUTHENTICATION_KIND_API_KEY, }; @@ -71,5 +71,5 @@ pub(crate) use selector::{CodexCyberPolicyScope, SelectCodexProviderEndpointCred pub use types::{ CODEX_AUTHENTICATION_KIND_OAUTH, CodexAccountProfile, CodexCookie, CodexCookieCaptureOutcome, CodexCredentialData, CodexCredentialPrincipal, CodexOAuthCredentialData, CodexOAuthSecret, - RuntimeCodexCookie, UpsertCodexCookie, + ResponsesTransport, RuntimeCodexCookie, UpsertCodexCookie, }; diff --git a/backend/crates/providers/openai/src/credential/security.rs b/backend/crates/providers/openai/src/credential/security.rs index db44410a1..156444dda 100644 --- a/backend/crates/providers/openai/src/credential/security.rs +++ b/backend/crates/providers/openai/src/credential/security.rs @@ -17,6 +17,7 @@ const MAX_COOKIES: usize = 128; /// 已解析且只在 Provider 内可见的认证材料。 pub struct CodexRuntimeCredential { + pub transport: super::ResponsesTransport, pub authentication: CodexRuntimeAuthentication, pub principal: Option, pub installation_id: String, @@ -122,6 +123,7 @@ impl CodexCredentialCodec { ) -> Result { Self::encode_complete(CodexCredentialData::OAuth(CodexOAuthCredentialData { schema_version: CODEX_CREDENTIAL_SCHEMA_VERSION, + transport: super::ResponsesTransport::oauth_default(), principal, installation_id, access_token: secret.access_token.expose_secret().to_owned(), @@ -172,6 +174,10 @@ impl CodexCredentialCodec { let data = serde_json::from_value::(value) .map_err(|_| CodexCredentialDataError::Invalid)?; validate(&data)?; + let transport = match &data { + CodexCredentialData::OAuth(data) => data.transport, + CodexCredentialData::ApiKey(data) => data.transport, + }; let (authentication, principal, installation_id, cookies, oauth_client_id, oauth_scope) = match data { CodexCredentialData::ApiKey(data) => ( @@ -199,6 +205,7 @@ impl CodexCredentialCodec { ), }; Ok(CodexRuntimeCredential { + transport, authentication, principal, installation_id, @@ -238,6 +245,7 @@ impl CodexCredentialCodec { match (&mut incoming, existing) { (CodexCredentialData::OAuth(incoming), CodexCredentialData::OAuth(existing)) => { incoming.installation_id = existing.installation_id; + incoming.transport = existing.transport; } (CodexCredentialData::ApiKey(incoming), CodexCredentialData::ApiKey(existing)) => { incoming.installation_id = existing.installation_id; diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index 7c2f7b01a..ce7069695 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -393,6 +393,28 @@ impl CodexCredentialSelector { request.attempt.deadline(), request.attempt.concurrency_wait_budget(), ); + let continuation_account = match request.attempt.continuation_attempt() { + ContinuationAttempt::Native => request + .attempt + .continuation() + .and_then(gateway_core::engine::continuation::ContinuationBinding::pinned) + .map(|continuation| continuation.account().clone()), + ContinuationAttempt::ReplayOwner => request + .attempt + .account_state_owner() + .filter(|owner| owner.provider() == &self.provider_kind) + .map(|owner| owner.account().clone()), + ContinuationAttempt::None | ContinuationAttempt::ReplayAny => None, + }; + let required_account = request.attempt.required_account().cloned(); + if required_account + .as_ref() + .zip(continuation_account.as_ref()) + .is_some_and(|(required, continuation)| required != continuation) + { + return Err(CredentialSelectionError::NoEligibleCredential); + } + let pinned_account = required_account.or(continuation_account); let mut snapshot_retries = 0; 'capacity: loop { let diagnostic = request.attempt.is_diagnostic_required_account(); @@ -437,7 +459,7 @@ impl CodexCredentialSelector { let mut eligible = Vec::with_capacity(accounts.len()); for account in accounts { if request.requires_websocket - && account.authentication_kind() == super::CODEX_AUTHENTICATION_KIND_API_KEY + && pinned_account.as_ref().is_none_or(|id| id == account.id()) { let runtime = match self.repository.load_runtime_credential(&account).await { Ok(runtime) => runtime, @@ -449,11 +471,15 @@ impl CodexCredentialSelector { )?; continue 'capacity; } + // 非固定账号的损坏凭据不能阻断其余账号的传输资格检查。 + Err(CredentialRepositoryError::InvalidCredentialData) + if pinned_account.is_none() => + { + continue; + } Err(error) => return Err(error.into()), }; - if !matches!(runtime.authentication, CodexRuntimeAuthentication::ApiKey(ref auth) - if auth.configuration.transport == super::ApiKeyTransport::PreferWebsocket) - { + if runtime.transport == super::ResponsesTransport::Http { continue; } } @@ -514,28 +540,6 @@ impl CodexCredentialSelector { AccountCandidate { account, signals } }) .collect::>(); - let continuation_account = match request.attempt.continuation_attempt() { - ContinuationAttempt::Native => request - .attempt - .continuation() - .and_then(gateway_core::engine::continuation::ContinuationBinding::pinned) - .map(|continuation| continuation.account().clone()), - ContinuationAttempt::ReplayOwner => request - .attempt - .account_state_owner() - .filter(|owner| owner.provider() == &self.provider_kind) - .map(|owner| owner.account().clone()), - ContinuationAttempt::None | ContinuationAttempt::ReplayAny => None, - }; - let required_account = request.attempt.required_account().cloned(); - if required_account - .as_ref() - .zip(continuation_account.as_ref()) - .is_some_and(|(required, continuation)| required != continuation) - { - return Err(CredentialSelectionError::NoEligibleCredential); - } - let pinned_account = required_account.or_else(|| continuation_account.clone()); let mut affinity = if diagnostic { AffinitySelection::default() } else { @@ -800,6 +804,7 @@ impl CodexCredentialSelector { } return Ok(CodexCredentialLease { installation_id: runtime.installation_id, + transport: runtime.transport, account, authentication: runtime.authentication, cookies, @@ -1444,6 +1449,7 @@ impl fmt::Debug for CodexCredentialSelector { } pub struct CodexCredentialLease { + transport: super::ResponsesTransport, account: ProviderAccount, authentication: CodexRuntimeAuthentication, cookies: Vec, @@ -1457,6 +1463,10 @@ pub struct CodexCredentialLease { } impl CodexCredentialLease { + pub(crate) const fn transport(&self) -> super::ResponsesTransport { + self.transport + } + #[must_use] pub const fn account(&self) -> &ProviderAccount { &self.account diff --git a/backend/crates/providers/openai/src/credential/types.rs b/backend/crates/providers/openai/src/credential/types.rs index 38703b580..2edce2d94 100644 --- a/backend/crates/providers/openai/src/credential/types.rs +++ b/backend/crates/providers/openai/src/credential/types.rs @@ -279,12 +279,36 @@ impl fmt::Debug for CodexCookie { } } +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ResponsesTransport { + #[default] + Http, + PreferWebsocket, +} + +impl ResponsesTransport { + pub(crate) const fn oauth_default() -> Self { + Self::PreferWebsocket + } + + fn is_oauth_default(&self) -> bool { + *self == Self::oauth_default() + } +} + pub const CODEX_AUTHENTICATION_KIND_OAUTH: &str = "oauth"; /// Codex OAuth 对 `provider_credentials_json` 的完整明文 schema。 #[derive(Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct CodexOAuthCredentialData { + // 默认值不扩展旧凭据 JSON，恢复 WS 优先后仍可由旧版本读取。 + #[serde( + default = "ResponsesTransport::oauth_default", + skip_serializing_if = "ResponsesTransport::is_oauth_default" + )] + pub transport: ResponsesTransport, pub schema_version: u32, #[serde(default, skip_serializing_if = "Option::is_none")] pub principal: Option, diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index e889109e9..505079abe 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -709,15 +709,14 @@ impl CodexProvider { ); } let requirement = transport_requirement(&upstream_request); - let api_http = matches!(lease.authentication(), crate::credential::CodexRuntimeAuthentication::ApiKey(auth) - if auth.configuration.transport == crate::credential::ApiKeyTransport::Http); - if api_http && requirement.requires_websocket() { + let http_only = lease.transport() == crate::credential::ResponsesTransport::Http; + if http_only && requirement.requires_websocket() { return Err(provider_error( ProviderErrorKind::Unsupported, UpstreamSendState::NotSent, )); } - let requested_transport = if api_http { + let requested_transport = if http_only { CodexProviderTransport::HttpOnly } else { selected_transport(&upstream_request) diff --git a/backend/crates/providers/openai/tests/admin.rs b/backend/crates/providers/openai/tests/admin.rs index 61e967fab..0893aac11 100644 --- a/backend/crates/providers/openai/tests/admin.rs +++ b/backend/crates/providers/openai/tests/admin.rs @@ -2280,7 +2280,7 @@ async fn api_key_admin_exposes_only_configuration_and_preserves_key_when_rotatin .seed_api_key( "acct_api_admin", "https://first.example/v1".to_owned(), - provider_openai::credential::ApiKeyTransport::Http, + provider_openai::credential::ResponsesTransport::Http, ) .await; let account = store.account("acct_api_admin").unwrap(); @@ -2347,3 +2347,99 @@ async fn api_key_admin_exposes_only_configuration_and_preserves_key_when_rotatin ProviderAdminErrorKind::Unsupported ); } + +#[tokio::test] +async fn oauth_transport_settings_preserve_tokens_refresh_schedule_and_health() { + let store = Arc::new(MemoryAccountStore::default()); + store + .seed_oauth_credential(ImportCodexOAuthCredential { + account_id: "acct_oauth_transport".to_owned(), + name: "OAuth transport".to_owned(), + secret: secret("test-oauth-transport"), + verified_account: profile("chatgpt-oauth-transport"), + next_refresh_at: Some(Utc::now() + chrono::Duration::minutes(30)), + enabled: true, + }) + .await; + let account = store.account("acct_oauth_transport").unwrap(); + let current = store.load_current_credential(account.id()).await.unwrap(); + let config = valid_config(); + let bundle = provider_openai::initialize( + config.config.clone(), + provider_ports_with(Arc::clone(&store), Arc::new(TestOAuthPending::default())), + ) + .await + .unwrap(); + let admin = bundle.admin_provider(); + let configuration = admin + .account_configuration(account.id()) + .await + .unwrap() + .unwrap(); + assert_eq!( + configuration.expose_to_provider().expose_to_provider(), + json!({"transport":"prefer_websocket"}).as_object().unwrap() + ); + for transport in ["http", "prefer_websocket"] { + let prepared = admin + .prepare_rotation(PrepareCredentialRotation { + account: account_record(&account), + provider_material: ProviderDocument::new(OpaqueProviderData::new( + json!({"transport":transport}).as_object().unwrap().clone(), + )), + }) + .await + .unwrap(); + let facts = prepared.facts(); + let mut expected = current.credential.expose_to_provider().clone(); + if transport == "http" { + expected.insert("transport".to_owned(), json!(transport)); + } + assert_eq!( + facts + .provider_material + .expose_to_provider() + .expose_to_provider(), + &expected + ); + assert_eq!( + facts.next_refresh_at, + account.next_refresh_at().map(DateTime::::from) + ); + assert_eq!( + facts.access_token_expires_at, + account.access_token_expires_at().map(DateTime::::from) + ); + assert_eq!(facts.has_refresh_token, account.has_refresh_token()); + assert!(facts.preserve_profile && facts.preserve_credential_state); + } + for invalid in [ + json!({"transport":"invalid"}), + json!({"transport":"http", "base_url":"https://other.example"}), + json!({"transport":"http", "api_key":"test-key"}), + ] { + let error = admin + .prepare_rotation(PrepareCredentialRotation { + account: account_record(&account), + provider_material: ProviderDocument::new(OpaqueProviderData::new( + invalid.as_object().unwrap().clone(), + )), + }) + .await + .unwrap_err(); + assert_eq!(error.kind(), ProviderAdminErrorKind::Invalid); + } + store.set_oauth_transport( + "acct_oauth_transport", + provider_openai::credential::ResponsesTransport::Http, + ); + let configuration = admin + .account_configuration(account.id()) + .await + .unwrap() + .unwrap(); + assert_eq!( + configuration.expose_to_provider().expose_to_provider(), + json!({"transport":"http"}).as_object().unwrap() + ); +} diff --git a/backend/crates/providers/openai/tests/credential/admin.rs b/backend/crates/providers/openai/tests/credential/admin.rs index 04aff4871..e1cc07f36 100644 --- a/backend/crates/providers/openai/tests/credential/admin.rs +++ b/backend/crates/providers/openai/tests/credential/admin.rs @@ -891,7 +891,7 @@ async fn api_key_import_export_preserves_target_without_oauth_exchange() { assert_eq!(data.api_key, "sk-test-only"); assert_eq!( data.transport, - provider_openai::credential::ApiKeyTransport::Http + provider_openai::credential::ResponsesTransport::Http ); } diff --git a/backend/crates/providers/openai/tests/credential/catalog.rs b/backend/crates/providers/openai/tests/credential/catalog.rs index 177dfcf88..df3d5d903 100644 --- a/backend/crates/providers/openai/tests/credential/catalog.rs +++ b/backend/crates/providers/openai/tests/credential/catalog.rs @@ -783,17 +783,17 @@ fn invalid_response_etag_is_rejected_without_touching_the_catalog() { #[tokio::test] async fn api_key_catalogs_are_isolated_and_join_oauth_without_claiming_native_metadata() { - use provider_openai::credential::{ApiKeyTransport, CodexCatalogScope}; + use provider_openai::credential::{CodexCatalogScope, ResponsesTransport}; let oauth = MockServer::start().await; let first = MockServer::start().await; let second = MockServer::start().await; let store = Arc::new(MemoryAccountStore::default()); let oauth_account = seed_account(&store, "acct_oauth").await; store - .seed_api_key("acct_first", first.uri(), ApiKeyTransport::Http) + .seed_api_key("acct_first", first.uri(), ResponsesTransport::Http) .await; store - .seed_api_key("acct_second", second.uri(), ApiKeyTransport::Http) + .seed_api_key("acct_second", second.uri(), ResponsesTransport::Http) .await; let a = store.account("acct_first").unwrap(); let b = store.account("acct_second").unwrap(); @@ -872,13 +872,13 @@ async fn api_key_catalogs_are_isolated_and_join_oauth_without_claiming_native_me #[tokio::test] async fn api_key_client_catalog_negotiates_and_preserves_versioned_native_objects() { use gateway_core::routing::ProviderModelContent; - use provider_openai::credential::ApiKeyTransport; + use provider_openai::credential::ResponsesTransport; use serde_json::json; let upstream = MockServer::start().await; let store = Arc::new(MemoryAccountStore::default()); store - .seed_api_key("acct_api", upstream.uri(), ApiKeyTransport::Http) + .seed_api_key("acct_api", upstream.uri(), ResponsesTransport::Http) .await; let account = store.account("acct_api").expect("API account"); // 先填满普通 ID 目录，确保客户端不会把已有套餐缓存误当成原生元数据。 @@ -960,14 +960,14 @@ async fn api_key_client_catalog_negotiates_and_preserves_versioned_native_object #[tokio::test] async fn api_key_native_catalog_preserves_unknown_reasoning_and_background_entitlements() { use gateway_core::routing::ProviderModelContent; - use provider_openai::credential::ApiKeyTransport; + use provider_openai::credential::ResponsesTransport; use provider_openai::transport::CodexCatalogCapabilityEvidence; use serde_json::json; let upstream = MockServer::start().await; let store = Arc::new(MemoryAccountStore::default()); store - .seed_api_key("acct_api", upstream.uri(), ApiKeyTransport::Http) + .seed_api_key("acct_api", upstream.uri(), ResponsesTransport::Http) .await; let original = json!({"slug":"z-unknown-reasoning", "display_name":"Unknown reasoning"}); let second = json!({"slug":"vendor/model", "display_name":"Second upstream model"}); @@ -1017,7 +1017,7 @@ async fn api_key_native_catalog_preserves_unknown_reasoning_and_background_entit #[tokio::test] async fn api_key_client_catalog_keeps_native_sources_stable_and_account_scoped() { use gateway_core::routing::ProviderModelContent; - use provider_openai::credential::ApiKeyTransport; + use provider_openai::credential::ResponsesTransport; use serde_json::json; let upstream = MockServer::start().await; @@ -1040,7 +1040,7 @@ async fn api_key_client_catalog_keeps_native_sources_stable_and_account_scoped() .seed_api_key( id, format!("{}/{id}", upstream.uri()), - ApiKeyTransport::Http, + ResponsesTransport::Http, ) .await; accounts.push(store.account(id).unwrap()); @@ -1078,12 +1078,12 @@ async fn api_key_client_catalog_keeps_native_sources_stable_and_account_scoped() #[tokio::test] async fn api_key_client_catalog_cache_is_invalidated_by_credential_revision() { use gateway_core::account::{CredentialCasOutcome, CredentialCasUpdate, ProviderAccountUpdate}; - use provider_openai::credential::ApiKeyTransport; + use provider_openai::credential::ResponsesTransport; let upstream = MockServer::start().await; let store = Arc::new(MemoryAccountStore::default()); store - .seed_api_key("acct_api", upstream.uri(), ApiKeyTransport::Http) + .seed_api_key("acct_api", upstream.uri(), ResponsesTransport::Http) .await; let account = store.account("acct_api").unwrap(); let scope = client_scope(std::slice::from_ref(&account)); @@ -1134,13 +1134,13 @@ async fn api_key_client_catalog_cache_is_invalidated_by_credential_revision() { #[tokio::test] async fn api_key_catalog_accepts_namespaced_model_ids_without_inventing_capabilities() { - use provider_openai::credential::ApiKeyTransport; + use provider_openai::credential::ResponsesTransport; use provider_openai::transport::CodexCatalogCapabilityEvidence; let upstream = MockServer::start().await; let store = Arc::new(MemoryAccountStore::default()); store - .seed_api_key("acct_api", upstream.uri(), ApiKeyTransport::Http) + .seed_api_key("acct_api", upstream.uri(), ResponsesTransport::Http) .await; let ids = [ "gpt-5.4", @@ -1195,12 +1195,12 @@ async fn api_key_catalog_accepts_namespaced_model_ids_without_inventing_capabili #[tokio::test] async fn api_key_catalog_rejects_invalid_ids_and_duplicates() { - use provider_openai::credential::ApiKeyTransport; + use provider_openai::credential::ResponsesTransport; let upstream = MockServer::start().await; let store = Arc::new(MemoryAccountStore::default()); store - .seed_api_key("acct_api", upstream.uri(), ApiKeyTransport::Http) + .seed_api_key("acct_api", upstream.uri(), ResponsesTransport::Http) .await; let service = service_with_catalog_cache(&store, upstream.uri(), catalog_cache()); for invalid in [ @@ -1230,7 +1230,7 @@ async fn api_key_catalog_rejects_invalid_ids_and_duplicates() { #[tokio::test] async fn api_key_catalog_does_not_replace_native_metadata_for_shared_models() { use gateway_core::routing::ProviderModelContent; - use provider_openai::credential::ApiKeyTransport; + use provider_openai::credential::ResponsesTransport; let oauth = MockServer::start().await; let upstream = MockServer::start().await; @@ -1245,7 +1245,7 @@ async fn api_key_catalog_does_not_replace_native_metadata_for_shared_models() { let service = service_with_catalog_cache(&store, oauth.uri(), catalog_cache()); let before = service.synchronize().await.expect("OAuth directory"); store - .seed_api_key("acct_api", upstream.uri(), ApiKeyTransport::Http) + .seed_api_key("acct_api", upstream.uri(), ResponsesTransport::Http) .await; Mock::given(method("GET")) .and(path("/models")) @@ -1298,7 +1298,7 @@ async fn api_key_catalog_does_not_replace_native_metadata_for_shared_models() { #[tokio::test] async fn slow_api_key_catalogs_share_a_deadline_and_preserve_healthy_catalogs() { - use provider_openai::credential::ApiKeyTransport; + use provider_openai::credential::ResponsesTransport; use std::time::Duration; let upstream = MockServer::start().await; @@ -1332,7 +1332,7 @@ async fn slow_api_key_catalogs_share_a_deadline_and_preserve_healthy_catalogs() .seed_api_key( "acct_fast", format!("{}/fast", upstream.uri()), - ApiKeyTransport::Http, + ResponsesTransport::Http, ) .await; accounts.push(store.account("acct_fast").expect("fast API account")); @@ -1342,7 +1342,7 @@ async fn slow_api_key_catalogs_share_a_deadline_and_preserve_healthy_catalogs() .seed_api_key( &id, format!("{}/slow", upstream.uri()), - ApiKeyTransport::Http, + ResponsesTransport::Http, ) .await; accounts.push(store.account(&id).expect("slow API account")); diff --git a/backend/crates/providers/openai/tests/credential/contract.rs b/backend/crates/providers/openai/tests/credential/contract.rs index 694784985..17c85894d 100644 --- a/backend/crates/providers/openai/tests/credential/contract.rs +++ b/backend/crates/providers/openai/tests/credential/contract.rs @@ -2020,3 +2020,26 @@ fn model_access_queue_waits_for_allowed_account_without_using_free_forbidden_acc .all(|lease| lease.account_id() == &allowed) ); } + +#[test] +fn legacy_oauth_defaults_to_websocket_and_reimport_preserves_http() { + use gateway_core::account::PlaintextCredential; + use provider_openai::credential::ResponsesTransport; + let incoming = CodexCredentialCodec::encode_new( + &secret("new-token"), + &profile("chatgpt-transport"), + Vec::new(), + ) + .unwrap(); + let mut legacy = incoming.expose_to_provider().clone(); + legacy.remove("transport"); + let runtime = CodexCredentialCodec::decode(&PlaintextCredential::new(legacy.clone())).unwrap(); + assert_eq!(runtime.transport, ResponsesTransport::PreferWebsocket); + legacy.insert("transport".to_owned(), json!("http")); + let existing = PlaintextCredential::new(legacy); + let preserved = CodexCredentialCodec::preserve_installation_id(&incoming, &existing).unwrap(); + assert_eq!( + CodexCredentialCodec::decode(&preserved).unwrap().transport, + ResponsesTransport::Http + ); +} diff --git a/backend/crates/providers/openai/tests/credential/refresh.rs b/backend/crates/providers/openai/tests/credential/refresh.rs index 46a07804a..fcb0f2e35 100644 --- a/backend/crates/providers/openai/tests/credential/refresh.rs +++ b/backend/crates/providers/openai/tests/credential/refresh.rs @@ -644,6 +644,10 @@ async fn scheduled_refresh_preserves_the_stored_token_set_when_rotation_fields_a .expect("test expiry"); seed_refreshable_account(&store, account_id, expires_at, None).await; + store.set_oauth_transport( + account_id, + provider_openai::credential::ResponsesTransport::Http, + ); service.refresh_due().await.expect("refresh cycle"); let account = store.account(account_id).expect("refreshed account"); @@ -653,6 +657,10 @@ async fn scheduled_refresh_preserves_the_stored_token_set_when_rotation_fields_a .await .expect("refreshed credential"); let runtime = CodexCredentialCodec::decode(&loaded.credential).expect("runtime credential"); + assert_eq!( + runtime.transport, + provider_openai::credential::ResponsesTransport::Http + ); let oauth = runtime.authentication.oauth().expect("OAuth credential"); assert_eq!( oauth.access_token.expose_secret(), diff --git a/backend/crates/providers/openai/tests/credential/types.rs b/backend/crates/providers/openai/tests/credential/types.rs index ab2051da2..05a7fd2e3 100644 --- a/backend/crates/providers/openai/tests/credential/types.rs +++ b/backend/crates/providers/openai/tests/credential/types.rs @@ -39,6 +39,7 @@ fn account_profile_debug_redacts_identity_fields() { #[test] fn plaintext_provider_schema_round_trips_dynamic_cookie_data() { let data = CodexCredentialData::OAuth(CodexOAuthCredentialData { + transport: provider_openai::credential::ResponsesTransport::PreferWebsocket, schema_version: 1, principal: Some(CodexCredentialPrincipal { oauth_subject: "subject-private".to_owned(), diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index 17bad3396..e1878c807 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -209,7 +209,7 @@ async fn native_openai_revalidates_translated_transport_without_reselecting() { .seed_api_key( "acct_provider_contract", upstream.uri(), - provider_openai::credential::ApiKeyTransport::Http, + provider_openai::credential::ResponsesTransport::Http, ) .await; let source = Operation::Generate(GenerateRequest::from_protocol_payload( @@ -827,7 +827,7 @@ async fn grok_downstream_identity_should_only_change_selected_codex_oauth_reques .seed_api_key( "acct_provider_contract", server.uri(), - provider_openai::credential::ApiKeyTransport::Http, + provider_openai::credential::ResponsesTransport::Http, ) .await; } else { @@ -932,9 +932,9 @@ async fn capture_replay_compatibility_request( "acct_provider_contract", base_url.clone(), if websocket { - provider_openai::credential::ApiKeyTransport::PreferWebsocket + provider_openai::credential::ResponsesTransport::PreferWebsocket } else { - provider_openai::credential::ApiKeyTransport::Http + provider_openai::credential::ResponsesTransport::Http }, ) .await; @@ -2392,7 +2392,7 @@ async fn api_websocket_precheck_and_selection_share_the_snapshot_retry_budget() .seed_api_key( "acct_provider_contract", upstream.uri(), - provider_openai::credential::ApiKeyTransport::PreferWebsocket, + provider_openai::credential::ResponsesTransport::PreferWebsocket, ) .await; store.on_credential_load(Arc::new(|store, id, count| { @@ -9832,7 +9832,7 @@ async fn api_key_native_endpoints_preserve_bodies_headers_and_own_base_url() { .seed_api_key( "acct_provider_contract", format!("{}{prefix}", upstream.uri()), - provider_openai::credential::ApiKeyTransport::Http, + provider_openai::credential::ResponsesTransport::Http, ) .await; let provider = provider_with_base_url(&store, oauth.uri()); @@ -9925,7 +9925,7 @@ async fn api_key_responses_forward_lite_memgen_and_native_compaction_without_cat .seed_api_key( "acct_provider_contract", upstream.uri(), - provider_openai::credential::ApiKeyTransport::Http, + provider_openai::credential::ResponsesTransport::Http, ) .await; Mock::given(method("GET")) @@ -10039,7 +10039,7 @@ async fn api_key_default_http_uses_own_prefix_plain_json_and_only_own_authentica .seed_api_key( "acct_provider_contract", format!("{}{prefix}", upstream.uri()), - provider_openai::credential::ApiKeyTransport::Http, + provider_openai::credential::ResponsesTransport::Http, ) .await; // 后台发现不协商客户端版本；客户端目录独立请求并按实际版本缓存。 @@ -10166,7 +10166,7 @@ async fn disabled_api_key_diagnostic_preserves_authentication_and_transport_cons .seed_api_key( account_id, upstream.uri(), - provider_openai::credential::ApiKeyTransport::Http, + provider_openai::credential::ResponsesTransport::Http, ) .await; let account = store.account(account_id).expect("API account"); @@ -10264,7 +10264,7 @@ async fn api_key_http_account_is_rejected_before_websocket_warmup_or_old_revisio .seed_api_key( "acct_provider_contract", upstream.uri(), - provider_openai::credential::ApiKeyTransport::Http, + provider_openai::credential::ResponsesTransport::Http, ) .await; let provider = provider(&store); @@ -10312,7 +10312,7 @@ async fn api_key_websocket_uses_api_path_and_bearer_without_oauth_identity() { .seed_api_key( "acct_provider_contract", base, - provider_openai::credential::ApiKeyTransport::PreferWebsocket, + provider_openai::credential::ResponsesTransport::PreferWebsocket, ) .await; let server = tokio::spawn(async move { @@ -10977,3 +10977,171 @@ async fn fast_policy_changes_preserve_the_websocket_continuation_and_meter_each_ } server.await.unwrap(); } + +#[tokio::test] +async fn oauth_http_transport_overrides_websocket_preference_and_delivers_terminal_event() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + store.set_oauth_transport( + "acct_provider_contract", + provider_openai::credential::ResponsesTransport::Http, + ); + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer at-acct_provider_contract")) + .respond_with( + ResponseTemplate::new(200).set_body_raw(CAPTURE_COMPLETED_SSE, "text/event-stream"), + ) + .expect(1) + .mount(&upstream) + .await; + let operation = Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object( + "openai", + json!({"model":"gpt-5.4", "input":"hello", "stream":true}) + .as_object() + .unwrap() + .clone(), + ) + .unwrap() + .with_context(Map::from_iter([("use_websocket".to_owned(), json!(true))])), + )); + let mut stream = provider_with_base_url(&store, upstream.uri()) + .execute( + planned_request("openai", operation), + context("req_oauth_http", CancellationToken::new()), + ) + .await + .unwrap(); + assert_eq!(stream.metadata().transport().as_str(), "http_sse"); + let mut completed = false; + while let Some(event) = stream.next().await { + completed |= event + .unwrap() + .wire_event() + .is_some_and(|wire| wire.data()["type"] == "response.completed"); + } + assert!(completed, "HTTP/SSE must deliver a terminal response"); + assert_eq!( + upstream.received_requests().await.unwrap().len(), + 1, + "no WebSocket attempt or replay" + ); +} + +#[tokio::test] +async fn oauth_http_transport_rejects_websocket_only_warmup_without_sending() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + store.set_oauth_transport( + "acct_provider_contract", + provider_openai::credential::ResponsesTransport::Http, + ); + let upstream = MockServer::start().await; + let warmup = Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object( + "openai", + json!({"model":"gpt-5.4", "input":[], "store":false, "generate":false}) + .as_object() + .unwrap() + .clone(), + ) + .unwrap(), + )); + assert!( + provider_with_base_url(&store, upstream.uri()) + .execute( + planned_request("openai", warmup), + diagnostic_context("req_oauth_warmup", "acct_provider_contract") + ) + .await + .is_err() + ); + assert!(upstream.received_requests().await.unwrap().is_empty()); +} + +#[tokio::test] +async fn oauth_http_transport_rejects_downstream_websocket_chain_without_sending() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + store.set_oauth_transport( + "acct_provider_contract", + provider_openai::credential::ResponsesTransport::Http, + ); + let upstream = MockServer::start().await; + let operation = Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object( + "openai", + json!({"model":"gpt-5.4", "input":"hello", "store":false, "stream":true}) + .as_object() + .unwrap() + .clone(), + ) + .unwrap() + .with_context(Map::from_iter([ + ("use_websocket".to_owned(), json!(true)), + ( + "downstream_websocket_connection_id".to_owned(), + json!("client-connection"), + ), + ])), + )); + let result = provider_with_base_url(&store, upstream.uri()) + .execute( + planned_request("openai", operation), + context("req_oauth_http_ws_chain", CancellationToken::new()), + ) + .await; + let Err(error) = result else { + panic!("connection-local chain needs an upstream WebSocket") + }; + assert_eq!(error.kind(), ProviderErrorKind::NoEligibleAccount); + assert_eq!(error.send_state(), UpstreamSendState::NotSent); + assert!(upstream.received_requests().await.unwrap().is_empty()); +} + +#[tokio::test] +async fn invalid_oauth_candidate_does_not_block_another_websocket_account() { + use gateway_core::account::{NewProviderAccount, PlaintextCredential}; + + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_affinity").await; + create_account(&store, "acct_provider_contract").await; + let invalid = store.account("acct_affinity").unwrap(); + store.delete_account(invalid.id()).await.unwrap(); + store + .create_account(NewProviderAccount { + account: invalid, + credential: PlaintextCredential::new(Map::new()), + model_access: None, + }) + .await + .unwrap(); + + let operation = Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object( + "openai", + json!({"model":"gpt-5.4", "input":"hello", "store":false, "stream":true}) + .as_object() + .unwrap() + .clone(), + ) + .unwrap() + .with_context(Map::from_iter([( + "downstream_websocket_connection_id".to_owned(), + json!("client-connection"), + )])), + )); + let stream = provider_with_base_url(&store, OFFICIAL_CODEX_BASE_URL.to_owned()) + .execute( + planned_request("openai", operation), + context("req_oauth_ws_valid_pool", CancellationToken::new()), + ) + .await + .expect("healthy account remains eligible"); + assert_eq!( + stream.metadata().provider_account_id().as_str(), + "acct_provider_contract" + ); +} diff --git a/backend/crates/providers/openai/tests/support.rs b/backend/crates/providers/openai/tests/support.rs index 07ec6454f..e0e1244d9 100644 --- a/backend/crates/providers/openai/tests/support.rs +++ b/backend/crates/providers/openai/tests/support.rs @@ -72,11 +72,26 @@ impl MemoryAccountStore { .expect("seed test OAuth credential"); } + pub(crate) fn set_oauth_transport( + &self, + id: &str, + transport: provider_openai::credential::ResponsesTransport, + ) { + use provider_openai::credential::CodexCredentialCodec; + let mut accounts = self.accounts.lock().unwrap(); + let stored = accounts + .get_mut(&ProviderAccountId::new(id).unwrap()) + .unwrap(); + let mut data = CodexCredentialCodec::decode_complete(&stored.credential).unwrap(); + data.oauth_mut().unwrap().transport = transport; + stored.credential = CodexCredentialCodec::encode_complete(data).unwrap(); + } + pub(crate) async fn seed_api_key( &self, id: &str, base_url: String, - transport: provider_openai::credential::ApiKeyTransport, + transport: provider_openai::credential::ResponsesTransport, ) { let credential = provider_openai::credential::CodexCredentialCodec::encode_complete( provider_openai::credential::CodexCredentialData::ApiKey( diff --git a/backend/crates/providers/xai/src/admin.rs b/backend/crates/providers/xai/src/admin.rs index 110036872..3fbb27a6c 100644 --- a/backend/crates/providers/xai/src/admin.rs +++ b/backend/crates/providers/xai/src/admin.rs @@ -1049,6 +1049,7 @@ fn prepared_rotation( email: profile.email, plan_type: profile.plan_type, preserve_profile, + preserve_credential_state: false, provider_material: ProviderDocument::new(OpaqueProviderData::new( credential.into_inner(), )), diff --git a/frontend/src/api/modules/accounts.ts b/frontend/src/api/modules/accounts.ts index 15e28e234..ce64d66b8 100644 --- a/frontend/src/api/modules/accounts.ts +++ b/frontend/src/api/modules/accounts.ts @@ -391,7 +391,7 @@ interface AccountResetCreditConsumeParam extends AccountIdParam { } interface AccountUpdateParam { - connection?: { baseUrl: string, transport: ApiKeyConfiguration['transport'], apiKey?: string } + connection?: { baseUrl?: string, transport: ApiKeyConfiguration['transport'], apiKey?: string } outboundProxyUrl?: string outboundProxyId?: string accountId: string diff --git a/frontend/src/views/accounts/components/AccountEditModal.vue b/frontend/src/views/accounts/components/AccountEditModal.vue index 1c0c4f854..f5d2bab74 100644 --- a/frontend/src/views/accounts/components/AccountEditModal.vue +++ b/frontend/src/views/accounts/components/AccountEditModal.vue @@ -3,9 +3,9 @@ import type { AccountRow } from '../constants' import type { ApiKeyAccountForm } from '../utils/upstreamApiKey' import type { AccountGroup, AccountModelAccess } from '@/api' -import { BaseButton, BaseFormItem, BaseModal, BaseTextarea } from '@codex-proxy/ui' +import { BaseButton, BaseFormItem, BaseModal, BaseSegmented, BaseTextarea } from '@codex-proxy/ui' import ProviderIconGroup from '@/components/ProviderIconGroup.vue' -import { isOpenAiApiKeyAccount } from '../utils/upstreamApiKey' +import { isOpenAiApiKeyAccount, isOpenAiOAuthAccount } from '../utils/upstreamApiKey' import AccountApiKeyFields from './AccountApiKeyFields.vue' import AccountIdentityCell from './AccountIdentityCell.vue' import AccountPlanBadge from './AccountPlanBadge.vue' @@ -26,6 +26,11 @@ const emit = defineEmits<{ const open = defineModel({ required: true }) const apiKey = defineModel('apiKey', { required: true }) +const oauthTransport = defineModel('oauthTransport', { required: true }) +const transportOptions = [ + { label: 'WS', value: 'prefer_websocket' }, + { label: 'SSE', value: 'http' }, +] const notes = defineModel('notes', { required: true }) const enabled = defineModel('enabled', { required: true }) const concurrencyLimit = defineModel('concurrencyLimit', { required: true }) @@ -61,6 +66,16 @@ const selectedGroupIds = defineModel('selectedGroupIds', { required: t + +

+ 正在读取上游设置… +

+ + + +

上游连接 @@ -85,6 +100,7 @@ const selectedGroupIds = defineModel('selectedGroupIds', { required: t :groups="groups" :groups-loading="groupsLoading" :disabled="saving" + :show-scheduling="false" :endpoint="account.outboundProxyEndpoint" :account-id="account.id" /> diff --git a/frontend/src/views/accounts/components/AccountSettingsFields.vue b/frontend/src/views/accounts/components/AccountSettingsFields.vue index 1d01fc7e1..6f65f65e9 100644 --- a/frontend/src/views/accounts/components/AccountSettingsFields.vue +++ b/frontend/src/views/accounts/components/AccountSettingsFields.vue @@ -13,8 +13,9 @@ withDefaults(defineProps<{ accountId?: string preserveProxy?: boolean preserveModelAccess?: boolean + showScheduling?: boolean proxyError?: string -}>(), { preserveProxy: true }) +}>(), { preserveProxy: true, showScheduling: true }) const modelAccess = defineModel('modelAccess', { required: true }) const enabled = defineModel('enabled', { required: true }) @@ -28,7 +29,7 @@ const selectedGroupIds = defineModel('selectedGroupIds', { required: t 测试连接 + + + {{ account.enabled ? '停用调度' : '启用调度' }} + >['items'][number] @@ -34,6 +34,8 @@ export function useAccountEditor(options: { const configurationLoading = configurationRequest.loading const configurationReady = shallowRef(false) const savedConfiguration = shallowRef() + const oauthTransport = shallowRef('prefer_websocket') + const savedOAuthTransport = shallowRef('prefer_websocket') async function loadConfiguration(accountId: string) { const requestId = configurationRequest.start() @@ -41,6 +43,15 @@ export function useAccountEditor(options: { const detail = await getAccountDetail({ accountId }, { signal: configurationRequest.signal }) if (!configurationRequest.isCurrent(requestId)) return + if (isOpenAiOAuthAccount(detail.account)) { + const transport = detail.credentialConfiguration?.transport + if (transport !== 'http' && transport !== 'prefer_websocket') + throw new Error('该账号没有 OAuth 上游设置') + oauthTransport.value = transport + savedOAuthTransport.value = transport + configurationReady.value = true + return + } const configuration = parseApiKeyConfiguration(detail.credentialConfiguration) if (!configuration) throw new Error('该账号没有 API Key 上游设置') @@ -75,10 +86,12 @@ export function useAccountEditor(options: { modelAccess.value = { ...account.modelAccess, models: [...account.modelAccess.models] } selectedGroupIds.value = account.groups.map(group => group.id) apiKey.value = emptyApiKeyAccountForm() + oauthTransport.value = 'prefer_websocket' + savedOAuthTransport.value = 'prefer_websocket' savedConfiguration.value = undefined configurationReady.value = false showEditModal.value = true - if (isOpenAiApiKeyAccount(account)) + if (isOpenAiApiKeyAccount(account) || isOpenAiOAuthAccount(account)) void loadConfiguration(account.id) } @@ -87,9 +100,10 @@ export function useAccountEditor(options: { if (!accountId || saving.value) return const isApiKey = isOpenAiApiKeyAccount(editingAccount.value) + const isOAuth = isOpenAiOAuthAccount(editingAccount.value) + if (isApiKey && !configurationReady.value) + return if (isApiKey) { - if (!configurationReady.value) - return const error = apiKeyAccountError(apiKey.value, true) if (error) { toast.warning(error) @@ -131,7 +145,9 @@ export function useAccountEditor(options: { ...settings, connection: connectionChanged ? { baseUrl: apiKey.value.base_url.trim(), transport: apiKey.value.transport, apiKey: apiKey.value.apiKey || undefined } - : undefined, + : isOAuth && configurationReady.value && oauthTransport.value !== savedOAuthTransport.value + ? { transport: oauthTransport.value } + : undefined, }) showEditModal.value = false toast.success('账号已更新') @@ -144,6 +160,8 @@ export function useAccountEditor(options: { return configurationRequest.invalidate() apiKey.value = emptyApiKeyAccountForm() + oauthTransport.value = 'prefer_websocket' + savedOAuthTransport.value = 'prefer_websocket' savedConfiguration.value = undefined configurationReady.value = false editingAccountId.value = null @@ -159,6 +177,7 @@ export function useAccountEditor(options: { return { apiKey, + oauthTransport, configurationLoading, configurationReady, showEditModal, diff --git a/frontend/src/views/accounts/composables/useAccountMutations.ts b/frontend/src/views/accounts/composables/useAccountMutations.ts index 9ba81a33c..7cd7a34f1 100644 --- a/frontend/src/views/accounts/composables/useAccountMutations.ts +++ b/frontend/src/views/accounts/composables/useAccountMutations.ts @@ -5,6 +5,7 @@ import { toast } from '@codex-proxy/ui' import dayjs from 'dayjs' import { computed, ref, shallowReactive, watch } from 'vue' import { + batchUpdateAccounts, deleteAccounts, exportAccounts, getAccountModelCatalog, @@ -43,6 +44,7 @@ export function useAccountMutations(options: { const refreshingAccounts = useIdSet() const refreshingQuotaAccounts = useIdSet() const downloadingCatalogAccounts = useIdSet() + const togglingSchedulingAccounts = useIdSet() const deletingAccountAction = useAsyncAction() const batchDeletingAction = useAsyncAction() const exportingAccountsAction = useAsyncAction() @@ -50,6 +52,7 @@ export function useAccountMutations(options: { const refreshingAccountIds = refreshingAccounts.ids const refreshingQuotaAccountIds = refreshingQuotaAccounts.ids const downloadingCatalogAccountIds = downloadingCatalogAccounts.ids + const togglingSchedulingAccountIds = togglingSchedulingAccounts.ids const deletingAccount = deletingAccountAction.loading const batchDeleting = batchDeletingAction.loading const exportingAccounts = exportingAccountsAction.loading @@ -203,6 +206,17 @@ export function useAccountMutations(options: { }) } + async function handleToggleScheduling(account: AccountRow) { + await togglingSchedulingAccounts.run(account.id, async () => { + try { + await batchUpdateAccounts({ accountIds: [account.id], enabled: !account.enabled }) + await loadAccounts() + toast.success(account.enabled ? '调度已停用' : '调度已启用') + } + catch {} + }) + } + async function handleRefreshQuota(accountId: string) { await refreshingQuotaAccounts.run(accountId, async () => { try { @@ -296,6 +310,7 @@ export function useAccountMutations(options: { refreshingAccountIds, refreshingQuotaAccountIds, downloadingCatalogAccountIds, + togglingSchedulingAccountIds, deletingAccount, batchDeleting, exportingAccounts, @@ -309,5 +324,6 @@ export function useAccountMutations(options: { handleRefresh, handleRefreshQuota, handleQuotaReset, + handleToggleScheduling, } } diff --git a/frontend/src/views/accounts/index.vue b/frontend/src/views/accounts/index.vue index 722dad7c8..f3d43198a 100644 --- a/frontend/src/views/accounts/index.vue +++ b/frontend/src/views/accounts/index.vue @@ -79,6 +79,7 @@ const { refreshingAccountIds, refreshingQuotaAccountIds, downloadingCatalogAccountIds, + togglingSchedulingAccountIds, deletingAccount, creatingAccount, authorizingOAuth, @@ -103,6 +104,7 @@ const { handleRefresh, handleRefreshQuota, handleQuotaReset, + handleToggleScheduling, } = useAccountMutations({ onImportTaskCreated: importTasks.created, accounts, @@ -166,6 +168,7 @@ const { const { apiKey: editingApiKey, + oauthTransport: editingOAuthTransport, configurationLoading, configurationReady, showEditModal, @@ -325,6 +328,7 @@ const { :recovering="recoveringAccountIds.has(row.id)" :refreshing="refreshingAccountIds.has(row.id)" :testing="testingConnectionIds.has(row.id)" + :toggling-scheduling="togglingSchedulingAccountIds.has(row.id)" @edit="openAccountEdit" @delete="requestDeleteAccount" @download-model-catalog="handleDownloadModelCatalog" @@ -332,6 +336,7 @@ const { @refresh="handleRefresh" @reauthorize="openReauthorizeAccount" @test="openConnectionTest" + @toggle-scheduling="handleToggleScheduling" /> @@ -412,6 +417,7 @@ const { | null | undefined): boolean { + return account?.provider === 'openai' && account.authenticationKind === 'oauth' +} + export interface ApiKeyAccountForm extends ApiKeyConfiguration { name: string apiKey: string