diff --git a/backend/crates/gateway-admin/src/model/account_groups.rs b/backend/crates/gateway-admin/src/model/account_groups.rs index 9275681ca..65949ed76 100644 --- a/backend/crates/gateway-admin/src/model/account_groups.rs +++ b/backend/crates/gateway-admin/src/model/account_groups.rs @@ -79,6 +79,7 @@ pub struct AccountGroupListQuery { /// Complete account group summary. #[derive(Debug, Clone, PartialEq, Eq)] pub struct AccountGroupRecord { + pub disable_fast: bool, pub id: AccountGroupId, pub name: String, pub description: Option, @@ -107,6 +108,7 @@ pub struct AccountGroupPage { /// Create an account group. #[derive(Debug, Clone, PartialEq, Eq)] pub struct CreateAccountGroup { + pub disable_fast: bool, pub name: String, pub description: Option, pub color: AccountGroupColor, @@ -115,6 +117,7 @@ pub struct CreateAccountGroup { /// Store-ready create command with a generated stable ID. #[derive(Debug, Clone, PartialEq, Eq)] pub struct NewAccountGroup { + pub disable_fast: bool, pub id: AccountGroupId, pub name: String, pub description: Option, @@ -124,6 +127,7 @@ pub struct NewAccountGroup { /// Update an account group's descriptive fields. #[derive(Debug, Clone, PartialEq, Eq)] pub struct UpdateAccountGroup { + pub disable_fast: Option, pub id: AccountGroupId, pub name: String, pub description: Option, diff --git a/backend/crates/gateway-admin/src/model/settings.rs b/backend/crates/gateway-admin/src/model/settings.rs index 0bb8de20b..e42927bc4 100644 --- a/backend/crates/gateway-admin/src/model/settings.rs +++ b/backend/crates/gateway-admin/src/model/settings.rs @@ -17,6 +17,7 @@ pub use gateway_core::account::RotationStrategy; /// 完整运行设置事实。 #[derive(Debug, Clone, PartialEq, Eq)] pub struct RuntimeSettings { + pub disable_fast: bool, pub config_revision: Revision, pub request_location_enabled: bool, pub request_location: gateway_core::account::RequestLocation, @@ -48,6 +49,7 @@ pub struct RuntimeSettings { /// 原子替换运行设置的命令。 #[derive(Debug, Clone, PartialEq, Eq)] pub struct ReplaceRuntimeSettings { + pub disable_fast: Option, pub request_location_enabled: bool, pub request_location: gateway_core::account::RequestLocation, pub model_mappings: ModelMappings, diff --git a/backend/crates/gateway-admin/src/use_case/account_groups.rs b/backend/crates/gateway-admin/src/use_case/account_groups.rs index a278e0640..7cbfbbb0c 100644 --- a/backend/crates/gateway-admin/src/use_case/account_groups.rs +++ b/backend/crates/gateway-admin/src/use_case/account_groups.rs @@ -144,6 +144,7 @@ impl AccountGroupService for DefaultAccountGroupService { id, name: command.name, description: command.description, + disable_fast: command.disable_fast, color: command.color, }, context, diff --git a/backend/crates/gateway-admin/tests/use_case/account_groups.rs b/backend/crates/gateway-admin/tests/use_case/account_groups.rs index 8ba996ec8..70fb3a6b6 100644 --- a/backend/crates/gateway-admin/tests/use_case/account_groups.rs +++ b/backend/crates/gateway-admin/tests/use_case/account_groups.rs @@ -202,6 +202,7 @@ impl AccountRuntimeStore for FakeRuntimeStore { fn group_record() -> AccountGroupRecord { let now = Utc::now(); AccountGroupRecord { + disable_fast: false, id: group_id(), name: "Primary".to_owned(), description: None, diff --git a/backend/crates/gateway-admin/tests/use_case/accounts.rs b/backend/crates/gateway-admin/tests/use_case/accounts.rs index 990e5bf34..6c7faf497 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts.rs @@ -973,6 +973,7 @@ struct StaticSettingsStore; impl SettingsStore for StaticSettingsStore { async fn load_runtime_settings(&self) -> AdminStoreResult { Ok(RuntimeSettings { + disable_fast: false, request_location_enabled: false, request_location: Default::default(), config_revision: revision(1), diff --git a/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs b/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs index 78757ec0b..8a34bb2f7 100644 --- a/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs +++ b/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs @@ -25,6 +25,7 @@ use super::accounts::{FakeAccountStore, FakeProviderAdmin, account_record, event fn runtime_settings(enabled: bool, probe_enabled: bool, adaptive: bool) -> RuntimeSettings { RuntimeSettings { + disable_fast: false, request_location_enabled: false, request_location: Default::default(), config_revision: revision(1), diff --git a/backend/crates/gateway-admin/tests/use_case/observability.rs b/backend/crates/gateway-admin/tests/use_case/observability.rs index a81029269..a16a18647 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability.rs @@ -753,6 +753,7 @@ struct FixtureSettingsStore; impl SettingsStore for FixtureSettingsStore { async fn load_runtime_settings(&self) -> AdminStoreResult { Ok(RuntimeSettings { + disable_fast: false, request_location_enabled: false, request_location: Default::default(), config_revision: Revision::new(1).expect("revision"), diff --git a/backend/crates/gateway-admin/tests/use_case/settings.rs b/backend/crates/gateway-admin/tests/use_case/settings.rs index b3bd6134a..e2b12a440 100644 --- a/backend/crates/gateway-admin/tests/use_case/settings.rs +++ b/backend/crates/gateway-admin/tests/use_case/settings.rs @@ -61,6 +61,7 @@ async fn settings_should_reject_zero_refresh_margin_before_store_call() { request_id: "request-settings".to_owned(), }, ReplaceRuntimeSettings { + disable_fast: None, request_location_enabled: false, request_location: Default::default(), model_mappings: Default::default(), diff --git a/backend/crates/gateway-api/src/admin/account_groups.rs b/backend/crates/gateway-api/src/admin/account_groups.rs index 9742f2fbc..4c0c981d3 100644 --- a/backend/crates/gateway-api/src/admin/account_groups.rs +++ b/backend/crates/gateway-api/src/admin/account_groups.rs @@ -72,6 +72,8 @@ impl ListAccountGroupsQuery { #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] struct CreateAccountGroupRequest { + #[serde(default)] + disable_fast: bool, name: String, description: Option, color: String, @@ -80,6 +82,7 @@ struct CreateAccountGroupRequest { #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] struct UpdateAccountGroupRequest { + disable_fast: Option, id: String, name: String, description: Option, @@ -95,6 +98,7 @@ struct AccountGroupIdRequest { #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] struct AccountGroupView { + disable_fast: bool, id: String, name: String, description: Option, @@ -138,6 +142,7 @@ impl From for AccountGroupView { id: record.id.to_string(), name: record.name, description: record.description, + disable_fast: record.disable_fast, color: record.color.as_str().to_owned(), enabled: record.enabled, member_count: record.member_count, @@ -273,6 +278,7 @@ where CreateAccountGroup { name: request.name, description: request.description, + disable_fast: request.disable_fast, color: group_color(&request.color)?, }, ) @@ -300,6 +306,7 @@ where id: group_id(request.id)?, name: request.name, description: request.description, + disable_fast: request.disable_fast, color: group_color(&request.color)?, }, ) diff --git a/backend/crates/gateway-api/src/admin/settings.rs b/backend/crates/gateway-api/src/admin/settings.rs index ee60c49b4..fb4a02601 100644 --- a/backend/crates/gateway-api/src/admin/settings.rs +++ b/backend/crates/gateway-api/src/admin/settings.rs @@ -34,6 +34,7 @@ pub type ModelMappings = BTreeMap; #[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[serde(rename_all = "camelCase")] pub struct RuntimeSettingsView { + pub disable_fast: bool, pub request_location_enabled: bool, pub request_location: gateway_core::account::RequestLocation, pub model_mappings: ModelMappings, @@ -65,6 +66,7 @@ pub struct RuntimeSettingsView { #[derive(Debug, Clone, PartialEq, Eq, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct UpdateRuntimeSettingsRequest { + pub disable_fast: Option, pub request_location_enabled: bool, pub request_location: gateway_core::account::RequestLocation, pub model_mappings: ModelMappings, @@ -175,6 +177,7 @@ impl UpdateRuntimeSettingsRequest { fn into_command(self) -> Result { self.validate()?; Ok(ReplaceRuntimeSettings { + disable_fast: self.disable_fast, request_location_enabled: self.request_location_enabled, request_location: self .request_location @@ -216,6 +219,7 @@ impl UpdateRuntimeSettingsRequest { impl From for RuntimeSettingsView { fn from(settings: RuntimeSettings) -> Self { Self { + disable_fast: settings.disable_fast, request_location_enabled: settings.request_location_enabled, request_location: settings.request_location, model_mappings: wire_model_mappings(settings.model_mappings), diff --git a/backend/crates/gateway-api/tests/admin/account_groups.rs b/backend/crates/gateway-api/tests/admin/account_groups.rs index dd7b374a8..b0d0ce321 100644 --- a/backend/crates/gateway-api/tests/admin/account_groups.rs +++ b/backend/crates/gateway-api/tests/admin/account_groups.rs @@ -272,3 +272,27 @@ async fn response_json(response: axum::response::Response) -> Value { .expect("account group response body"); serde_json::from_slice(&body).expect("account group response JSON") } + +#[tokio::test] +async fn disable_fast_group_updates_preserve_omitted_values() { + let fixture = authenticated_fixture().await; + for (value, expected) in [(Some(true), true), (None, true), (Some(false), false)] { + let mut body = json!({"id": PRIMARY_GROUP_ID, "name":"Policy", "description":null,"color":"#F43F5ECC"}); + if let Some(value) = value { + body["disableFast"] = json!(value); + } + let response = request( + router(&fixture), + Method::POST, + "/api/admin/account-groups/update", + Some(body), + true, + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response_json(response).await["data"]["record"]["disableFast"], + expected + ); + } +} diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index 017adcdc3..3086e68a1 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -402,6 +402,7 @@ impl SettingsStore for MemorySettingsStore { ) -> AdminStoreResult { let mut settings = self.settings.lock().expect("settings"); let updated = RuntimeSettings { + disable_fast: command.disable_fast.unwrap_or(settings.disable_fast), request_location_enabled: command.request_location_enabled, request_location: command.request_location, config_revision: next_revision(settings.config_revision), @@ -492,6 +493,7 @@ impl MemoryAccountGroupStore { ( primary_id.clone(), AccountGroupRecord { + disable_fast: false, id: primary_id, name: "Alpha routing".to_owned(), description: Some("Primary traffic".to_owned()), @@ -513,6 +515,7 @@ impl MemoryAccountGroupStore { ( secondary_id.clone(), AccountGroupRecord { + disable_fast: false, id: secondary_id, name: "Beta routing".to_owned(), description: None, @@ -628,6 +631,7 @@ impl AccountGroupStore for MemoryAccountGroupStore { let mut state = self.state.lock().expect("account groups"); let now = Utc::now(); let record = AccountGroupRecord { + disable_fast: command.disable_fast, id: command.id.clone(), name: command.name, description: command.description, @@ -659,6 +663,9 @@ impl AccountGroupStore for MemoryAccountGroupStore { record.name = command.name; record.description = command.description; record.color = command.color; + if let Some(disable_fast) = command.disable_fast { + record.disable_fast = disable_fast; + } record.updated_at = Utc::now(); mutation(&mut state, command.id, true) } @@ -1271,6 +1278,7 @@ fn test_runtime_settings() -> RuntimeSettings { ), ]); RuntimeSettings { + disable_fast: false, request_location_enabled: false, request_location: Default::default(), config_revision: Revision::new(7).expect("revision"), diff --git a/backend/crates/gateway-api/tests/admin/settings.rs b/backend/crates/gateway-api/tests/admin/settings.rs index 2e284f634..e19d977bb 100644 --- a/backend/crates/gateway-api/tests/admin/settings.rs +++ b/backend/crates/gateway-api/tests/admin/settings.rs @@ -37,6 +37,7 @@ async fn response_json(response: axum::response::Response) -> Value { fn update_body() -> Value { json!({ + "disableFast": false, "requestLocationEnabled": false, "requestLocation": {"country":"US", "region":"Ohio", "city":"Piketon", "timezone":"America/New_York"}, "modelMappings": { @@ -102,6 +103,7 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() { use gateway_core::routing::{PublicModelId, UpstreamModelId}; let settings = RuntimeSettings { + disable_fast: false, request_location_enabled: false, request_location: Default::default(), config_revision: Revision::new(7).expect("revision"), @@ -146,7 +148,8 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() { assert_eq!( value, json!({ - "requestLocationEnabled": false, + "disableFast": false, + "requestLocationEnabled": false, "requestLocation": {"country":"US", "region":"Ohio", "city":"Piketon", "timezone":"America/New_York"}, "modelMappings": { "gpt-5.4": "gpt-5.5", @@ -199,6 +202,7 @@ fn settings_request_and_response_fields_should_stay_in_lockstep() { .cloned() .collect(); let settings = RuntimeSettings { + disable_fast: false, request_location_enabled: false, request_location: Default::default(), config_revision: Revision::new(7).expect("revision"), @@ -600,3 +604,37 @@ fn decompression_setting_should_reject_invalid_values() { body.as_object_mut().unwrap().remove(field); assert!(serde_json::from_value::(body).is_err()); } + +#[tokio::test] +async fn disable_fast_settings_updates_preserve_omitted_values() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let app = app(fixture.state()); + for (value, expected) in [(Some(true), true), (None, true), (Some(false), false)] { + let mut body = update_body(); + if let Some(value) = value { + body["disableFast"] = json!(value); + } else { + body.as_object_mut().unwrap().remove("disableFast"); + } + let response = app + .clone() + .oneshot(request( + Method::POST, + "/api/admin/settings/update", + Some(body), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let response = app + .clone() + .oneshot(request(Method::GET, "/api/admin/settings", None)) + .await + .unwrap(); + assert_eq!( + response_json(response).await["data"]["disableFast"], + expected + ); + } +} diff --git a/backend/crates/gateway-core/src/account/scope.rs b/backend/crates/gateway-core/src/account/scope.rs index 3aaf4c197..14e4b037b 100644 --- a/backend/crates/gateway-core/src/account/scope.rs +++ b/backend/crates/gateway-core/src/account/scope.rs @@ -247,17 +247,31 @@ impl ClientRoutingScope { /// 一次认证随 RuntimeSnapshot 冻结的账号目录与 Key 权限。 #[derive(Debug, Clone, PartialEq, Eq)] pub struct FrozenAccountScope { + disable_fast: bool, directory: Arc, client_scope: ClientRoutingScope, } impl FrozenAccountScope { + /// Key 绑定分组的冻结 Fast 限制，与账号成员资格无关。 + #[must_use] + pub const fn with_disable_fast(mut self, disable_fast: bool) -> Self { + self.disable_fast = disable_fast; + self + } + + #[must_use] + pub const fn disable_fast(&self) -> bool { + self.disable_fast + } + #[must_use] pub const fn new( directory: Arc, client_scope: ClientRoutingScope, ) -> Self { Self { + disable_fast: false, directory, client_scope, } diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 5ddbbee49..6ff23da5f 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -788,6 +788,7 @@ where })); let context = AttemptContext::new( RequestAttemptContext::new(self.request_id.clone(), self.client_api_key_ref.clone()) + .with_disable_fast(self.plan.disable_fast()) .with_request_location(self.plan.request_location().cloned()) .with_concurrency_wait_budget(self.concurrency_wait_budget.clone()) .with_timing_started_at(self.observation.timing_started_at) diff --git a/backend/crates/gateway-core/src/engine/mod.rs b/backend/crates/gateway-core/src/engine/mod.rs index 89a2c2ba4..6c60b633e 100644 --- a/backend/crates/gateway-core/src/engine/mod.rs +++ b/backend/crates/gateway-core/src/engine/mod.rs @@ -314,6 +314,7 @@ impl ContinuationAttempt { /// Provider 每次执行可见的 request-local context。 #[derive(Debug, Clone)] pub struct RequestAttemptContext { + disable_fast: bool, request_location: Option, request_id: ModelRequestId, client_api_key_ref: ClientApiKeyId, @@ -323,6 +324,12 @@ pub struct RequestAttemptContext { } impl RequestAttemptContext { + #[must_use] + pub const fn with_disable_fast(mut self, disable_fast: bool) -> Self { + self.disable_fast = disable_fast; + self + } + #[must_use] pub fn with_request_location( mut self, @@ -337,6 +344,7 @@ impl RequestAttemptContext { Self { request_id, client_api_key_ref, + disable_fast: false, request_location: None, timing_started_at: Instant::now(), trace: crate::diagnostics::TraceContext::default(), @@ -399,6 +407,11 @@ pub struct AttemptContext { } impl AttemptContext { + #[must_use] + pub const fn disable_fast(&self) -> bool { + self.request.disable_fast + } + #[must_use] pub const fn request_location(&self) -> Option<&crate::account::RequestLocation> { self.request.request_location.as_ref() diff --git a/backend/crates/gateway-core/src/routing/mod.rs b/backend/crates/gateway-core/src/routing/mod.rs index 161a18297..07e0fdc42 100644 --- a/backend/crates/gateway-core/src/routing/mod.rs +++ b/backend/crates/gateway-core/src/routing/mod.rs @@ -536,6 +536,7 @@ impl ProviderCandidate { /// 一次请求冻结的 Provider 尝试顺序。 #[derive(Debug, Clone)] pub struct RoutingPlan { + disable_fast: bool, request_location: Option, config_revision: ConfigRevision, account_selection_policy: AccountSelectionPolicy, @@ -546,6 +547,11 @@ pub struct RoutingPlan { } impl RoutingPlan { + #[must_use] + pub const fn disable_fast(&self) -> bool { + self.disable_fast + } + /// 本次请求冻结的全局位置，重试时沿用同一份配置。 #[must_use] pub const fn request_location(&self) -> Option<&crate::account::RequestLocation> { diff --git a/backend/crates/gateway-core/src/routing/snapshot.rs b/backend/crates/gateway-core/src/routing/snapshot.rs index 3ea7a9878..14f14878c 100644 --- a/backend/crates/gateway-core/src/routing/snapshot.rs +++ b/backend/crates/gateway-core/src/routing/snapshot.rs @@ -28,6 +28,7 @@ const MAXIMUM_CATALOG_STABILITY_ATTEMPTS: usize = 4; /// Store 在一个一致性读取中提供的调度设置事实。 #[derive(Debug, Clone, PartialEq, Eq)] pub struct SnapshotSettingsFacts { + disable_fast: bool, request_location_enabled: bool, request_location: crate::account::RequestLocation, max_concurrent_per_account: u32, @@ -43,6 +44,12 @@ pub struct SnapshotSettingsFacts { } impl SnapshotSettingsFacts { + #[must_use] + pub const fn with_disable_fast(mut self, disable_fast: bool) -> Self { + self.disable_fast = disable_fast; + self + } + #[must_use] pub const fn with_responses_max_decompressed_body_bytes(mut self, bytes: u64) -> Self { self.responses_max_decompressed_body_bytes = bytes; @@ -83,6 +90,7 @@ impl SnapshotSettingsFacts { min_codex_cli_version: Option, ) -> Self { Self { + disable_fast: false, request_location_enabled: false, request_location: crate::account::RequestLocation::default(), max_concurrent_per_account, @@ -128,15 +136,27 @@ impl SnapshotClientPolicyFacts { /// Store 读取到的账号分组事实。 #[derive(Debug, Clone, PartialEq, Eq)] pub struct SnapshotAccountGroupFacts { + disable_fast: bool, id: AccountGroupId, name: String, enabled: bool, } impl SnapshotAccountGroupFacts { + #[must_use] + pub const fn with_disable_fast(mut self, disable_fast: bool) -> Self { + self.disable_fast = disable_fast; + self + } + #[must_use] pub fn new(id: AccountGroupId, name: String, enabled: bool) -> Self { - Self { id, name, enabled } + Self { + id, + name, + enabled, + disable_fast: false, + } } } @@ -442,6 +462,7 @@ async fn compile_runtime_snapshot( }); let mut client_policies = Vec::with_capacity(facts.client_policies.len()); for policy in facts.client_policies { + let mut disable_fast = false; let account_scope = if policy.group_ids.is_empty() { FrozenAccountScope::new( Arc::clone(&account_directory), @@ -458,6 +479,8 @@ async fn compile_runtime_snapshot( let group = groups .get(&group_id) .ok_or(RuntimeSnapshotCompileError::InvalidData)?; + // 禁用分组仅影响选号；Key 仍绑定其 Fast 限制。 + disable_fast |= group.disable_fast; bound_groups.push(RoutingGroupSnapshot::new( group.id.clone(), group.name.clone(), @@ -477,7 +500,7 @@ async fn compile_runtime_snapshot( client_policies.push(ClientPolicy::new( policy.key_id, policy.plaintext_key, - Arc::new(account_scope), + Arc::new(account_scope.with_disable_fast(disable_fast)), true, policy.limits, )); @@ -505,6 +528,7 @@ async fn compile_runtime_snapshot( .map_err(|_| RuntimeSnapshotCompileError::InvalidData) .map(|snapshot| { snapshot + .with_disable_fast(facts.settings.disable_fast) .with_request_location(request_location) .with_responses_max_decompressed_body_bytes(decompressed_body_limit) .with_client_queue_policy(client_queue_policy) @@ -518,6 +542,7 @@ async fn compile_runtime_snapshot( /// 数据面使用的不可变配置快照。 #[derive(Debug, Clone)] pub struct RuntimeSnapshot { + disable_fast: bool, responses_max_decompressed_body_bytes: std::num::NonZeroUsize, request_location: Option, revision: ConfigRevision, @@ -536,6 +561,12 @@ pub struct RuntimeSnapshot { } impl RuntimeSnapshot { + #[must_use] + pub const fn with_disable_fast(mut self, disable_fast: bool) -> Self { + self.disable_fast = disable_fast; + self + } + #[must_use] pub const fn responses_max_decompressed_body_bytes(&self) -> usize { self.responses_max_decompressed_body_bytes.get() @@ -640,6 +671,7 @@ impl RuntimeSnapshot { Ok(Self { responses_max_decompressed_body_bytes: std::num::NonZeroUsize::new(64 * 1024 * 1024) .expect("positive default limit"), + disable_fast: false, request_location: None, revision, account_selection_policy, @@ -951,6 +983,7 @@ impl RuntimeSnapshot { Ok(RoutingPlan { config_revision: self.revision, + disable_fast: self.disable_fast || account_scope.disable_fast(), request_location: self.request_location.clone(), account_selection_policy: self.account_selection_policy, operation: operation.kind(), @@ -993,6 +1026,7 @@ impl RuntimeSnapshot { }; Ok(RoutingPlan { config_revision: self.revision, + disable_fast: self.disable_fast || account_scope.disable_fast(), request_location: self.request_location.clone(), account_selection_policy: self.account_selection_policy, operation: operation.kind(), diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 27038e481..1fa00e2f3 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -594,6 +594,15 @@ fn plan_with_location( operation: &Operation, account_selection_policy: AccountSelectionPolicy, request_location: gateway_core::account::RequestLocation, +) -> RoutingPlan { + plan_with_location_and_fast_policy(operation, account_selection_policy, request_location, false) +} + +fn plan_with_location_and_fast_policy( + operation: &Operation, + account_selection_policy: AccountSelectionPolicy, + request_location: gateway_core::account::RequestLocation, + disable_fast: bool, ) -> RoutingPlan { let provider = ProviderKind::new("openai").expect("provider"); let public_model = PublicModelId::new("gpt-5").expect("public model"); @@ -636,6 +645,7 @@ fn plan_with_location( Vec::new(), ) .expect("snapshot") + .with_disable_fast(disable_fast) .with_request_location(Some(request_location)) .with_account_directory(Arc::clone(&directory)); snapshot @@ -4528,16 +4538,17 @@ fn deadline_before_first_event_records_no_provider_circuit_failure() { } #[test] -fn global_request_location_should_reach_every_account_retry() { +fn global_location_and_fast_policy_reach_every_account_retry() { let operation = generate_operation(); let location = gateway_core::account::RequestLocation { timezone: "Asia/Tokyo".parse().unwrap(), ..Default::default() }; - let route_plan = plan_with_location( + let route_plan = plan_with_location_and_fast_policy( &operation, plan(&operation).account_selection_policy(), location.clone(), + true, ); let (coordinator, _, provider) = coordinator(vec![ Script::Stream { @@ -4570,6 +4581,6 @@ fn global_request_location_should_reach_every_account_retry() { assert!( contexts .iter() - .all(|context| context.request_location() == Some(&location)) + .all(|context| context.request_location() == Some(&location) && context.disable_fast()) ); } diff --git a/backend/crates/gateway-core/tests/engine/execution.rs b/backend/crates/gateway-core/tests/engine/execution.rs index ab4fd35e4..9ca99a051 100644 --- a/backend/crates/gateway-core/tests/engine/execution.rs +++ b/backend/crates/gateway-core/tests/engine/execution.rs @@ -483,7 +483,9 @@ fn detached_early_failure_resumes_cancelled_store_write_and_settles_once_for_all }); } +#[derive(Default)] struct ChargedProvider { + policies: Mutex>, fail: bool, } @@ -510,8 +512,9 @@ impl Provider for ChargedProvider { async fn execute( &self, request: ProviderRequest, - _: AttemptContext, + context: AttemptContext, ) -> Result { + self.policies.lock().unwrap().push(context.disable_fast()); let candidate = request.candidate(); let metadata = ProviderCallMetadata::new( candidate.provider().clone(), @@ -552,7 +555,11 @@ fn charged_service( let service = DefaultExecutionService::new( RuntimeSnapshotHandle::new(start_snapshot()), Arc::new(TrackingExecutionStore::default()), - ProviderRegistry::new([Arc::new(ChargedProvider { fail }) as Arc]).unwrap(), + ProviderRegistry::new([Arc::new(ChargedProvider { + fail, + ..Default::default() + }) as Arc]) + .unwrap(), admissions, Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), @@ -2240,3 +2247,32 @@ fn cancelling_pending_admission_cleans_up_a_slot_acquired_before_the_reply() { drop(pending); assert!(admissions.active.lock().unwrap().is_empty()); } + +#[test] +fn reused_websocket_client_gets_fast_policy_from_each_new_request_snapshot() { + let snapshots = RuntimeSnapshotHandle::new(start_snapshot()); + let admissions = Arc::new(Admissions::default()); + let provider = Arc::new(ChargedProvider::default()); + let service = DefaultExecutionService::new( + snapshots.clone(), + Arc::new(TrackingExecutionStore::default()), + ProviderRegistry::new([provider.clone() as Arc]).unwrap(), + admissions, + Arc::new(UnusedCircuits), + Arc::new(UnusedContinuation), + Arc::new(RecordingClientApiKeyUsage::default()), + ); + let client = service.authenticate("sk_start_test").unwrap(); + for (revision, disable_fast) in [(2, true), (3, false)] { + snapshots.publish( + start_snapshot_with_policy(revision, true, RateLimits::unlimited()) + .with_disable_fast(disable_fast), + ); + let mut next = request(&service, ClientTransport::WebSocket); + next.client = client.clone(); + let mut started = block_on(service.start(next)).unwrap(); + block_on(started.session.collect_uncommitted()).unwrap(); + block_on(started.session.detach_finalize()); + } + assert_eq!(*provider.policies.lock().unwrap(), vec![true, false]); +} diff --git a/backend/crates/gateway-core/tests/routing/snapshot.rs b/backend/crates/gateway-core/tests/routing/snapshot.rs index b3b638d0f..c7e7db148 100644 --- a/backend/crates/gateway-core/tests/routing/snapshot.rs +++ b/backend/crates/gateway-core/tests/routing/snapshot.rs @@ -453,3 +453,84 @@ fn decompression_setting_should_validate_and_remain_frozen_across_publication() 1024 * 1024 ); } + +#[test] +fn disable_fast_uses_bound_groups_and_global_policy_without_changing_account_scope() { + use gateway_core::account::ProviderAccountId; + use gateway_core::routing::AccountGroupId; + for global in [false, true] { + for group_enabled in [false, true] { + for bound in [false, true] { + let group_id = AccountGroupId::new("grp_00000000000000000000000000000001").unwrap(); + let open_group_id = + AccountGroupId::new("grp_00000000000000000000000000000002").unwrap(); + let account_id = ProviderAccountId::new("acct_fast_policy").unwrap(); + let facts = SnapshotFacts::new( + revision(1), + revision(1), + SnapshotSettingsFacts::new(3, 0, "smart", BTreeMap::new(), None, None) + .with_disable_fast(global), + vec![SnapshotClientPolicyFacts::new( + ClientApiKeyId::new("key_fast_policy").unwrap(), + PlaintextClientApiKey::new("sk_fast_policy").unwrap(), + if bound { + vec![group_id.clone(), open_group_id.clone()] + } else { + Vec::new() + }, + RateLimits::unlimited(), + )], + vec![ + SnapshotAccountGroupFacts::new( + group_id.clone(), + "Restricted".to_owned(), + group_enabled, + ) + .with_disable_fast(true), + SnapshotAccountGroupFacts::new( + open_group_id.clone(), + "Open".to_owned(), + true, + ), + ], + vec![SnapshotProviderAccountFacts::new( + account_id.clone(), + "alpha", + )], + vec![ + SnapshotAccountGroupMemberFacts::new(group_id, account_id.clone()), + SnapshotAccountGroupMemberFacts::new(open_group_id, account_id.clone()), + ], + ); + let snapshot = block_on( + RuntimeSnapshotCompiler::new( + Arc::new(TestSnapshotStore::new(Ok(facts))), + Arc::new(TestCatalog::Unavailable), + ) + .compile(), + ) + .unwrap(); + let scope = snapshot + .client_policies() + .next() + .unwrap() + .account_scope() + .clone(); + assert!(scope.allows(&account_id)); + let plan = snapshot + .plan( + &PublicModelId::new("public-model").unwrap(), + &super::operation(), + scope, + &Default::default(), + ) + .unwrap(); + assert_eq!( + plan.disable_fast(), + global || bound, + "global={global}, enabled={group_enabled}, bound={bound}" + ); + } + } + } +} diff --git a/backend/crates/gateway-store/src/admin_adapter.rs b/backend/crates/gateway-store/src/admin_adapter.rs index ee195af34..cf493aeb1 100644 --- a/backend/crates/gateway-store/src/admin_adapter.rs +++ b/backend/crates/gateway-store/src/admin_adapter.rs @@ -43,6 +43,7 @@ impl SettingsStore for AdminSettingsStoreAdapter { refresh_margin_seconds: command.refresh_margin_seconds, refresh_concurrency: command.refresh_concurrency, max_concurrent_per_account: command.max_concurrent_per_account, + disable_fast: command.disable_fast, request_location_enabled: command.request_location_enabled, request_location: command.request_location, request_interval_ms: command.request_interval_ms, @@ -73,6 +74,7 @@ impl SettingsStore for AdminSettingsStoreAdapter { "runtime_settings", "1", vec![ + "disable_fast".to_owned(), "request_location_enabled".to_owned(), "request_location_json".to_owned(), "model_mappings_json".to_owned(), @@ -183,6 +185,7 @@ pub(crate) fn admin_runtime_settings( .collect::>()?; Ok(AdminRuntimeSettings { config_revision: admin_revision(settings.config_revision)?, + disable_fast: settings.disable_fast, request_location_enabled: settings.request_location_enabled, request_location: settings.request_location, model_mappings, diff --git a/backend/crates/gateway-store/src/postgres/account_groups.rs b/backend/crates/gateway-store/src/postgres/account_groups.rs index 05c306fd2..394ff76d1 100644 --- a/backend/crates/gateway-store/src/postgres/account_groups.rs +++ b/backend/crates/gateway-store/src/postgres/account_groups.rs @@ -234,20 +234,25 @@ impl AccountGroupStore for PgAccountGroupRepository { "create", "account_group", id.as_str(), - vec!["name".to_owned(), "description".to_owned()], + vec![ + "name".to_owned(), + "description".to_owned(), + "disable_fast".to_owned(), + ], ); let revision = self .mutate(audit, |transaction| { Box::pin(async move { sqlx::query( "insert into account_groups - (id, name, description, color, enabled, created_at, updated_at) - values ($1, $2, $3, $4, true, now(), now())", + (id, name, description, color, disable_fast, enabled, created_at, updated_at) + values ($1, $2, $3, $4, $5, true, now(), now())", ) .bind(command.id.as_str()) .bind(command.name) .bind(command.description) .bind(command.color.as_str()) + .bind(command.disable_fast) .execute(&mut **transaction) .await .map_err(|error| map_group_write_error(error, command.id.as_str()))?; @@ -274,20 +279,25 @@ impl AccountGroupStore for PgAccountGroupRepository { "update", "account_group", id.as_str(), - vec!["name".to_owned(), "description".to_owned()], + vec![ + "name".to_owned(), + "description".to_owned(), + "disable_fast".to_owned(), + ], ); let revision = self .mutate(audit, |transaction| { Box::pin(async move { let result = sqlx::query( "update account_groups - set name = $2, description = $3, color = $4, updated_at = now() + set name = $2, description = $3, color = $4, disable_fast = coalesce($5, disable_fast), updated_at = now() where id = $1", ) .bind(command.id.as_str()) .bind(command.name) .bind(command.description) .bind(command.color.as_str()) + .bind(command.disable_fast) .execute(&mut **transaction) .await .map_err(|error| map_group_write_error(error, command.id.as_str()))?; @@ -386,7 +396,7 @@ impl AccountGroupStore for PgAccountGroupRepository { fn group_select() -> QueryBuilder { QueryBuilder::new( - "select g.id, g.name, g.description, g.color, g.enabled, g.created_at, g.updated_at, + "select g.id, g.name, g.description, g.color, g.enabled, g.disable_fast, g.created_at, g.updated_at, coalesce(members.member_count, 0)::bigint as member_count, coalesce(keys.client_key_count, 0)::bigint as client_key_count, coalesce(members.provider_counts, '{}'::jsonb) as provider_counts @@ -463,6 +473,9 @@ fn group_record(row: &sqlx::postgres::PgRow) -> StoreResult let provider_counts = serde_json::from_value::>(provider_counts) .map_err(|_| invalid("invalid provider counts"))?; Ok(AccountGroupRecord { + disable_fast: row + .try_get("disable_fast") + .map_err(|_| invalid("invalid disable_fast"))?, id: AccountGroupId::new( row.try_get::("id") .map_err(|_| invalid("invalid id"))?, diff --git a/backend/crates/gateway-store/src/postgres/runtime_settings.rs b/backend/crates/gateway-store/src/postgres/runtime_settings.rs index e4ddff743..a3f25d5da 100644 --- a/backend/crates/gateway-store/src/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/src/postgres/runtime_settings.rs @@ -19,6 +19,7 @@ use crate::{Revision, StoreError, StoreResult, postgres_unavailable}; #[derive(Clone, PartialEq, Eq)] pub struct RuntimeSettings { + pub disable_fast: bool, pub config_revision: Revision, pub admin_api_key: Option, pub refresh_margin_seconds: u64, @@ -64,6 +65,7 @@ impl fmt::Debug for RuntimeSettings { &self.max_concurrent_per_account, ) .field("request_interval_ms", &self.request_interval_ms) + .field("disable_fast", &self.disable_fast) .field("rotation_strategy", &self.rotation_strategy) .field("request_location_enabled", &self.request_location_enabled) .field("request_location", &self.request_location) @@ -108,6 +110,7 @@ impl fmt::Debug for RuntimeSettings { #[derive(Clone)] pub struct RuntimeSettingsUpdate { + pub disable_fast: Option, pub admin_api_key: Option, pub refresh_margin_seconds: u64, pub refresh_concurrency: u32, @@ -143,6 +146,7 @@ impl fmt::Debug for RuntimeSettingsUpdate { "admin_api_key", &self.admin_api_key.as_ref().map(|_| "[REDACTED]"), ) + .field("disable_fast", &self.disable_fast) .field("rotation_strategy", &self.rotation_strategy) .field("request_location_enabled", &self.request_location_enabled) .field("request_location", &self.request_location) @@ -229,7 +233,7 @@ impl RuntimeSettingsRepository for PgRuntimeSettingsRepository { pub(crate) async fn load_runtime_settings_from_pool(pool: &PgPool) -> StoreResult { let row = sqlx::query_as::<_, RuntimeSettingsRow>( - "select config_revision, admin_api_key, refresh_margin_seconds, request_location_json, request_location_enabled, + "select config_revision, admin_api_key, refresh_margin_seconds, request_location_json, request_location_enabled, disable_fast, refresh_concurrency, max_concurrent_per_account, request_interval_ms, rotation_strategy, model_mappings_json, usage_retention_days, ops_event_retention_days, audit_retention_days, min_codex_desktop_version, @@ -291,7 +295,7 @@ pub(crate) async fn load_runtime_settings_in_transaction( transaction: &mut Transaction<'_, Postgres>, ) -> StoreResult { let row = sqlx::query_as::<_, RuntimeSettingsRow>( - "select config_revision, admin_api_key, refresh_margin_seconds, request_location_json, request_location_enabled, + "select config_revision, admin_api_key, refresh_margin_seconds, request_location_json, request_location_enabled, disable_fast, refresh_concurrency, max_concurrent_per_account, request_interval_ms, rotation_strategy, model_mappings_json, usage_retention_days, ops_event_retention_days, audit_retention_days, min_codex_desktop_version, @@ -347,6 +351,7 @@ pub(crate) async fn update_runtime_settings_in_transaction( request_location_json = $23, request_location_enabled = $24, responses_max_decompressed_body_bytes = $25, + disable_fast = coalesce($26, disable_fast), updated_at = now() where id = 1 returning config_revision", @@ -388,6 +393,7 @@ pub(crate) async fn update_runtime_settings_in_transaction( i64::try_from(update.responses_max_decompressed_body_bytes) .map_err(|_| invalid_numeric())?, ) + .bind(update.disable_fast) .fetch_optional(&mut **transaction) .await .map_err(|_| postgres_unavailable("update runtime settings in transaction"))? @@ -437,6 +443,7 @@ pub(crate) async fn update_admin_api_key_in_transaction( #[derive(sqlx::FromRow)] struct RuntimeSettingsRow { + disable_fast: bool, config_revision: i64, admin_api_key: Option, refresh_margin_seconds: i64, @@ -475,6 +482,7 @@ fn runtime_settings_from_row(row: RuntimeSettingsRow) -> StoreResult, bool, i64, + bool, ), >( "select config_revision, refresh_margin_seconds, refresh_concurrency, max_concurrent_per_account, request_interval_ms, rotation_strategy, model_mappings_json, min_codex_desktop_version, - min_codex_cli_version, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, request_location_json, request_location_enabled, responses_max_decompressed_body_bytes + min_codex_cli_version, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, request_location_json, request_location_enabled, responses_max_decompressed_body_bytes, disable_fast from runtime_settings where id = 1", ) .fetch_optional(&mut **transaction) @@ -272,6 +279,7 @@ async fn load_settings( Ok(( revision_from_i64(row.0)?, SnapshotRuntimeSettings { + disable_fast: row.15, responses_max_decompressed_body_bytes: to_u64(row.14)?, request_location_enabled: row.13, request_location: row.12.0, @@ -315,15 +323,16 @@ async fn load_client_keys( async fn load_account_groups( transaction: &mut Transaction<'_, Postgres>, ) -> StoreResult> { - let rows = sqlx::query_as::<_, (String, String, bool)>( - "select id, name, enabled from account_groups order by id", + let rows = sqlx::query_as::<_, (String, String, bool, bool)>( + "select id, name, enabled, disable_fast from account_groups order by id", ) .fetch_all(&mut **transaction) .await .map_err(|_| postgres_unavailable("load snapshot account groups"))?; rows.into_iter() - .map(|(id, name, enabled)| { + .map(|(id, name, enabled, disable_fast)| { Ok(SnapshotAccountGroupData { + disable_fast, id: AccountGroupId::new(id).map_err(|_| invalid("invalid account group id"))?, name, enabled, diff --git a/backend/crates/gateway-store/tests/postgres/account_groups.rs b/backend/crates/gateway-store/tests/postgres/account_groups.rs index 03df67fa7..2d063cea9 100644 --- a/backend/crates/gateway-store/tests/postgres/account_groups.rs +++ b/backend/crates/gateway-store/tests/postgres/account_groups.rs @@ -49,6 +49,7 @@ async fn groups_aggregate_cross_provider_members_and_key_bindings_without_multip groups .create_account_group( NewAccountGroup { + disable_fast: false, id: mixed_group.clone(), name: "Mixed Production".to_owned(), description: Some("cross-provider".to_owned()), @@ -61,6 +62,7 @@ async fn groups_aggregate_cross_provider_members_and_key_bindings_without_multip groups .create_account_group( NewAccountGroup { + disable_fast: false, id: empty_group.clone(), name: "Empty Pool".to_owned(), description: None, @@ -255,6 +257,7 @@ async fn group_costs_should_include_statusless_websocket_but_reject_statusless_h groups .create_account_group( NewAccountGroup { + disable_fast: false, id: group_id(EMPTY_GROUP), name: "Statusless Costs".to_owned(), description: None, @@ -430,3 +433,61 @@ async fn audit_count(pool: &sqlx::PgPool) -> u64 { .expect("count audit rows"); u64::try_from(value).expect("non-negative audit count") } + +#[tokio::test] +async fn disable_fast_group_updates_preserve_omitted_values_and_publish_snapshot_facts() { + use gateway_admin::model::account_groups::UpdateAccountGroup; + use gateway_store::postgres::{PgRuntimeSnapshotRepository, RuntimeSnapshotRepository}; + let Some(database) = TestDatabase::create("disable_fast_group").await else { + return; + }; + let repository = PgAccountGroupRepository::new(database.pool.clone()); + let id = group_id(MIXED_GROUP); + repository + .create_account_group( + NewAccountGroup { + id: id.clone(), + name: "Fast policy".to_owned(), + description: None, + color: group_color("#2563EBFF"), + disable_fast: true, + }, + &context("create-fast"), + ) + .await + .unwrap(); + for (value, expected) in [(None, true), (Some(false), false), (Some(true), true)] { + let mutation = repository + .update_account_group( + UpdateAccountGroup { + id: id.clone(), + name: "Renamed policy".to_owned(), + description: None, + color: group_color("#2563EBFF"), + disable_fast: value, + }, + &context("update-fast"), + ) + .await + .unwrap(); + assert_eq!(mutation.record.unwrap().disable_fast, expected); + let snapshot = PgRuntimeSnapshotRepository::new(database.pool.clone()) + .load_runtime_snapshot() + .await + .unwrap(); + assert_eq!( + snapshot.config_revision.get(), + mutation.config_revision.get() + ); + assert_eq!( + snapshot + .account_groups + .iter() + .find(|group| group.id == id) + .unwrap() + .disable_fast, + expected + ); + } + database.close().await; +} diff --git a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs index bb6256075..4c53b8e2f 100644 --- a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs @@ -9,6 +9,7 @@ use super::TestDatabase; fn settings_with_margin(refresh_margin_seconds: u64) -> RuntimeSettingsUpdate { RuntimeSettingsUpdate { + disable_fast: None, request_location_enabled: false, request_location: Default::default(), admin_api_key: None, @@ -429,3 +430,29 @@ async fn decompression_setting_should_persist_and_reach_snapshot_facts() { } database.close().await; } + +#[tokio::test] +async fn disable_fast_persists_and_omitted_updates_preserve_the_restriction() { + let Some(database) = TestDatabase::create("disable_fast_settings").await else { + return; + }; + let repository = PgRuntimeSettingsRepository::new(database.pool.clone()); + let initial = repository.load_runtime_settings().await.unwrap(); + assert!(!initial.disable_fast); + for (value, expected) in [(Some(true), true), (None, true), (Some(false), false)] { + let mut update = settings_with_margin(1_800); + update.disable_fast = value; + let revision = repository.update_runtime_settings(update).await.unwrap(); + let reloaded = repository.load_runtime_settings().await.unwrap(); + assert_eq!(reloaded.disable_fast, expected); + assert_eq!(reloaded.config_revision, revision); + use gateway_store::postgres::{PgRuntimeSnapshotRepository, RuntimeSnapshotRepository}; + let snapshot = PgRuntimeSnapshotRepository::new(database.pool.clone()) + .load_runtime_snapshot() + .await + .unwrap(); + assert_eq!(snapshot.settings.disable_fast, expected); + assert_eq!(snapshot.config_revision, revision); + } + database.close().await; +} diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index 0ddf6490d..10dbf4ad7 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -361,6 +361,8 @@ impl Provider for CodexProvider { let continuation_requested = generate.native_continuation_requested(); let mut upstream_request = encode_generate_request(generate, upstream_model.as_str(), None) .map_err(map_request_error)?; + // 编码已生成独立请求；HTTP、WS 与重试在头部和计量之前共用此策略。 + upstream_request.apply_fast_policy(context.disable_fast()); if let Some(conversation_id) = previous_session .as_ref() .and_then(|state| state.conversation_id.as_ref()) diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index f7107d0e8..493a7fb50 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -611,6 +611,21 @@ impl CodexResponsesRequest { self.body.get("service_tier").and_then(Value::as_str) } + /// 只覆盖顶层 Fast 请求，显式使用官方标准档退出值。 + pub(crate) fn apply_fast_policy(&mut self, disable_fast: bool) { + if disable_fast + && self.service_tier().is_some_and(|tier| { + let tier = tier.trim(); + tier.eq_ignore_ascii_case("priority") || tier.eq_ignore_ascii_case("fast") + }) + { + self.body.insert( + "service_tier".to_owned(), + Value::String("default".to_owned()), + ); + } + } + /// 前一个 response ID。 pub fn previous_response_id(&self) -> Option<&str> { self.body diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index a0246886e..8f5cca856 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -719,11 +719,20 @@ fn global_request_location() -> gateway_core::account::RequestLocation { } fn context(request_id: &str, cancellation: CancellationToken) -> AttemptContext { + context_with_fast_policy(request_id, cancellation, false) +} + +fn context_with_fast_policy( + request_id: &str, + cancellation: CancellationToken, + disable_fast: bool, +) -> AttemptContext { AttemptContext::new( RequestAttemptContext::new( ModelRequestId::new(request_id).expect("request id"), ClientApiKeyId::new("key_openai_contract").expect("client key id"), ) + .with_disable_fast(disable_fast) .with_request_location(Some(global_request_location())), NonZeroU32::new(1).expect("attempt"), SystemTime::now() + Duration::from_secs(30), @@ -5325,11 +5334,47 @@ async fn completed_response_persists_session_affinity_before_stream_consumer_sto #[tokio::test] async fn responses_should_observe_and_bill_the_outbound_service_tier_on_both_transports() { for use_websocket in [false, true] { - for (requested, reported, expected_cost) in [ - (Some("priority"), Some("default"), 6_875_000), - (Some("priority"), None, 6_875_000), - (Some("default"), Some("priority"), 3_437_500), - (None, Some("priority"), 3_437_500), + for (disable_fast, requested, reported, expected_tier, expected_cost) in [ + ( + false, + Some("priority"), + Some("default"), + Some("priority"), + Some(6_875_000), + ), + ( + false, + Some("priority"), + None, + Some("priority"), + Some(6_875_000), + ), + ( + false, + Some("default"), + Some("priority"), + Some("default"), + Some(3_437_500), + ), + (false, None, Some("priority"), None, Some(3_437_500)), + ( + true, + Some("priority"), + Some("priority"), + Some("default"), + Some(3_437_500), + ), + (true, Some("fast"), None, Some("default"), Some(3_437_500)), + ( + true, + Some("default"), + None, + Some("default"), + Some(3_437_500), + ), + (true, None, None, None, Some(3_437_500)), + (true, Some("flex"), None, Some("flex"), Some(1_720_000)), + (true, Some("ultrafast"), None, Some("ultrafast"), None), ] { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_provider_contract").await; @@ -5354,7 +5399,21 @@ async fn responses_should_observe_and_bill_the_outbound_service_tier_on_both_tra let base_url = format!("http://{}", listener.local_addr().expect("address")); let server = tokio::spawn(async move { let (socket, _) = listener.accept().await.expect("accept WebSocket"); - let mut websocket = accept_codex_test_websocket(socket).await; + let mut websocket = crate::transport::accept_codex_test_websocket_with( + socket, + |request, response| { + let expected_hint = expected_tier.map_or_else( + || "model=gpt-5.4".to_owned(), + |tier| format!("model=gpt-5.4;tier={tier}"), + ); + assert_eq!(request.headers()["x-codex-routing-hint"], expected_hint); + response.headers_mut().insert( + "sec-websocket-extensions", + "permessage-deflate".parse().unwrap(), + ); + }, + ) + .await; let request = websocket.next().await.expect("request").expect("frame"); let request: Value = serde_json::from_str(request.to_text().expect("text")) .expect("request JSON"); @@ -5383,23 +5442,30 @@ async fn responses_should_observe_and_bill_the_outbound_service_tier_on_both_tra let mut body = Map::from_iter([ ("model".to_owned(), json!("gpt-5.4")), ("input".to_owned(), json!("hello")), + ( + "metadata".to_owned(), + json!({"service_tier":"priority","text":"fast priority"}), + ), ]); if let Some(requested) = requested { body.insert("service_tier".to_owned(), json!(requested)); } - let payload = ProtocolPayload::json_object("openai", body) + let payload = ProtocolPayload::json_object("openai", body.clone()) .expect("payload") .with_context(Map::from_iter([( "use_websocket".to_owned(), json!(use_websocket), )])); + let operation = Operation::Generate(GenerateRequest::from_protocol_payload(payload)); + let original = operation.clone(); let mut stream = provider_with_base_url(&store, base_url) .execute( - planned_request( - "openai", - Operation::Generate(GenerateRequest::from_protocol_payload(payload)), + planned_request("openai", operation.clone()), + context_with_fast_policy( + "req_service_tier", + CancellationToken::new(), + disable_fast, ), - context("req_service_tier", CancellationToken::new()), ) .await .expect("provider stream"); @@ -5423,6 +5489,11 @@ async fn responses_should_observe_and_bill_the_outbound_service_tier_on_both_tra let outbound = if let Some(server) = http_server { server.verify().await; let requests = server.received_requests().await.expect("upstream requests"); + let expected_hint = expected_tier.map_or_else( + || "model=gpt-5.4".to_owned(), + |tier| format!("model=gpt-5.4;tier={tier}"), + ); + assert_eq!(requests[0].headers["x-codex-routing-hint"], expected_hint); captured_request_body(&requests[0]) } else { websocket_server @@ -5432,13 +5503,15 @@ async fn responses_should_observe_and_bill_the_outbound_service_tier_on_both_tra }; assert_eq!( outbound.get("service_tier").and_then(Value::as_str), - requested + expected_tier ); + assert_eq!(operation, original); + assert_eq!(outbound["metadata"], body["metadata"]); assert!(!observations.is_empty()); for observation in &observations { assert_eq!( observation.service_tier(), - requested, + expected_tier, "WebSocket={use_websocket}" ); } @@ -5453,7 +5526,7 @@ async fn responses_should_observe_and_bill_the_outbound_service_tier_on_both_tra .expect("metadata JSON"); assert_eq!( metadata.get("requestedServiceTier").and_then(Value::as_str), - requested + expected_tier ); assert_eq!( metadata.get("upstreamServiceTier").and_then(Value::as_str), @@ -5461,7 +5534,7 @@ async fn responses_should_observe_and_bill_the_outbound_service_tier_on_both_tra ); assert_eq!( costs, - vec![expected_cost], + expected_cost.into_iter().collect::>(), "WebSocket={use_websocket}, requested={requested:?}, reported={reported:?}" ); assert_eq!(raw_response, response_frames.as_bytes()); @@ -8934,3 +9007,173 @@ async fn quota_continuation_full_client_replay_selects_another_account() { .unwrap() .unwrap(); } + +#[tokio::test] +async fn fast_policy_changes_preserve_the_websocket_continuation_and_meter_each_outbound_tier() { + const ACCOUNT: &str = "acct_provider_contract"; + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, ACCOUNT).await; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut websocket = + crate::transport::accept_codex_test_websocket_with(socket, |request, response| { + // 提示头只在首次握手发送；后续档位由各自 response.create 正文指定。 + assert_eq!( + request.headers()["x-codex-routing-hint"], + "model=gpt-5.4;tier=priority" + ); + response.headers_mut().insert( + "sec-websocket-extensions", + "permessage-deflate".parse().unwrap(), + ); + }) + .await; + for (index, tier) in ["priority", "default", "priority"].into_iter().enumerate() { + let message = websocket.next().await.unwrap().unwrap(); + let frame: Value = serde_json::from_str(message.to_text().unwrap()).unwrap(); + assert_eq!(frame["type"], "response.create"); + assert_eq!(frame["service_tier"], tier); + assert_eq!( + frame.get("previous_response_id").cloned(), + index + .checked_sub(1) + .map(|previous| json!(format!("resp_fast_turn_{previous}"))) + ); + websocket.send(Message::Text(json!({ + "type":"response.created", "response":{"id":format!("resp_fast_turn_{index}"),"model":"gpt-5.4"} + }).to_string().into())).await.unwrap(); + websocket.send(Message::Text(json!({ + "type":"response.completed", "response":{ + "id":format!("resp_fast_turn_{index}"), "model":"gpt-5.4", "status":"completed", "output":[], + "service_tier":"priority", "usage":{"input_tokens":100,"output_tokens":10,"input_tokens_details":{"cached_tokens":25,"cache_write_tokens":0},"total_tokens":110} + } + }).to_string().into())).await.unwrap(); + } + assert!( + timeout(Duration::from_millis(100), listener.accept()) + .await + .is_err(), + "all three turns must use the same upstream connection" + ); + }); + let provider = provider_with_base_url(&store, base_url); + let mut session_state = ProviderSessionState::new( + "openai", + Map::from_iter([ + ("account_id".to_owned(), json!(ACCOUNT)), + ( + "conversation_id".to_owned(), + json!("fast-policy-continuation"), + ), + ("continuation_scope".to_owned(), json!("connection_local")), + ]), + ) + .unwrap(); + for (index, disable_fast) in [false, true, false].into_iter().enumerate() { + let previous = index + .checked_sub(1) + .map(|index| format!("resp_fast_turn_{index}")); + let mut body = + json!({"model":"gpt-5.4","input":"next turn","service_tier":"priority","store":false}); + if let Some(previous) = &previous { + body["previous_response_id"] = json!(previous); + } + let operation = Operation::Generate( + GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", body.as_object().unwrap().clone()) + .unwrap() + // 与客户端 WebSocket 一致，首轮也禁止按快路径预算降级到 HTTP。 + .with_context(Map::from_iter([ + ("use_websocket".to_owned(), json!(true)), + ( + "downstream_websocket_connection_id".to_owned(), + json!("ws_fast_policy_continuation"), + ), + ])), + ) + .with_provider_session_state(session_state.clone()), + ); + let account = ProviderAccountId::new(ACCOUNT).unwrap(); + let provider_kind = ProviderKind::new("openai").unwrap(); + let key = ClientApiKeyId::new("key_openai_contract").unwrap(); + let binding = previous.as_ref().map(|previous| { + ContinuationBinding::Pinned(NativeContinuationPin::new( + PreviousResponseId::new(previous), + PreviousResponseId::new(previous), + key.clone(), + provider_kind.clone(), + account.clone(), + )) + }); + let context = AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new(format!("req_fast_turn_{index}")).unwrap(), + key, + ) + .with_disable_fast(disable_fast), + NonZeroU32::new(1).unwrap(), + SystemTime::now() + Duration::from_secs(30), + account_policy(), + AccountAttemptContext::new( + BTreeSet::new(), + None, + Some(ProviderAccountStateOwner::new(provider_kind, account)), + ) + .with_account_scope(contract_account_scope()), + binding, + CancellationToken::new(), + ) + .with_continuation_attempt(if previous.is_some() { + ContinuationAttempt::Native + } else { + ContinuationAttempt::None + }); + let mut stream = provider + .execute(planned_request("openai", operation), context) + .await + .unwrap(); + let mut costs = Vec::new(); + let mut pool = None; + let mut observed_tier = None; + while let Some(event) = timeout(Duration::from_secs(5), stream.next()) + .await + .unwrap() + { + let event = event.unwrap(); + if let Some(update) = event.session_update() { + session_state = update.clone(); + } + if let Some(observation) = event.response_observation() { + pool = observation.websocket_pool().or(pool); + observed_tier = observation + .service_tier() + .map(str::to_owned) + .or(observed_tier); + } + for fact in event.canonical_facts() { + if let GatewayEvent::CalculatedCost(cost) = fact { + costs.push(cost.total().amount().scaled()); + } + } + } + assert_eq!( + pool, + Some(if index == 0 { + WebSocketPoolKind::New + } else { + WebSocketPoolKind::Reuse + }) + ); + assert_eq!( + observed_tier.as_deref(), + Some(if disable_fast { "default" } else { "priority" }) + ); + assert_eq!( + costs, + vec![if disable_fast { 3_437_500 } else { 6_875_000 }] + ); + } + server.await.unwrap(); +} diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index da7aa4bf6..b2811a2b2 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -11,3 +11,4 @@ b7d3df7b48d835a00a3b74cbefc958a039d38fea5529cdad04bfbc703dca72aa 0001_initial.s ae7f7acb3014f0a60d979dd9c04b3065322b206e42eb6256505baf78caf9dd22 0011_proxy_request_location.sql b20e425c0b062b72d35fdebe294e023eb17ceb44ac93083ff3d48400715bf430 0012_upstream_response_model.sql 6e23fceef1096b50c1af3584cb1687eded2ad37e31e65391397ec984bf78c7e3 0013_responses_decompression_limit.sql +12a540be31d65c4925cbe63dab42ea2d4ec2864fecf71a6dd4ee942e71726857 0014_disable_fast.sql diff --git a/backend/migrations/0014_disable_fast.sql b/backend/migrations/0014_disable_fast.sql new file mode 100644 index 000000000..9197c97d5 --- /dev/null +++ b/backend/migrations/0014_disable_fast.sql @@ -0,0 +1,3 @@ +-- Fast 限制默认关闭；旧客户端省略更新字段时保留已有策略。 +ALTER TABLE runtime_settings ADD COLUMN disable_fast boolean NOT NULL DEFAULT false; +ALTER TABLE account_groups ADD COLUMN disable_fast boolean NOT NULL DEFAULT false; diff --git a/docs/api.md b/docs/api.md index 88f5b4241..8a159db01 100644 --- a/docs/api.md +++ b/docs/api.md @@ -833,12 +833,15 @@ PostgreSQL 或 Redis。 ## 6. 账号分组 分组是 Provider-neutral 的账号集合；一个组可包含任意 Provider 账号，一个账号也可属于多个组。 +分组详情和列表返回 `disableFast`，创建时省略默认为 `false`，更新时省略或 `null` 保留现值。 +Client Key 绑定的任一分组开启此限制（包括已禁用分组）时，该 Key 的 OpenAI Responses 请求关闭 Fast； +未绑定分组的 Key 只受全局限制，不按最终所选账号的分组判断。 | 方法 | 路由 | 主要 query/body | 说明 | | --- | --- | --- | --- | | `GET` | `/api/admin/account-groups` | `page`、`pageSize`、`search`、`enabled` | 分页查询分组；返回账号可用性、并发槽位（Redis 不可用时 `usedSlots=null`）及成功请求 USD 用量 | -| `POST` | `/api/admin/account-groups/create` | `{ name, description, color }` | 创建空分组；`color` 严格为 `#RRGGBBAA`，返回时统一大写 | -| `POST` | `/api/admin/account-groups/update` | `{ id, name, description, color }` | 更新名称、描述和颜色 | +| `POST` | `/api/admin/account-groups/create` | `{ name, description, color, disableFast? }` | 创建空分组；`color` 严格为 `#RRGGBBAA`，返回时统一大写 | +| `POST` | `/api/admin/account-groups/update` | `{ id, name, description, color, disableFast? }` | 更新名称、描述、颜色和 Fast 限制 | | `POST` | `/api/admin/account-groups/enable` | `{ id }` | 启用 | | `POST` | `/api/admin/account-groups/disable` | `{ id }` | 禁用；已绑定 Key 保持受限，不回退到全部账号 | | `POST` | `/api/admin/account-groups/delete` | `{ id }` | 删除未被 Client Key 引用的组 | @@ -935,6 +938,7 @@ HTTP 返回 `429`，`error.code` 为 `key_daily_budget_exceeded` 或 `key_weekly 设置更新字段包括： ```text +disableFast requestLocationEnabled requestLocation modelMappings @@ -961,6 +965,13 @@ accountAutoFreezeProbeModel accountAutoFreezeAdaptiveConcurrency ``` +`disableFast` 默认 `false`，更新时省略或 `null` 保留现值。全局开启时，所有 Key 的 OpenAI Responses 请求关闭 Fast； +全局关闭时仍应用 Key 绑定分组的限制。关闭 Fast 只将顶层 `service_tier` 的 `priority`（含 `fast` 别名） +改为显式 `default`，继续处理请求；不改变 `flex`、`ultrafast`、缺失值、默认档、嵌套字段或其他 Provider。 +HTTP 和每个 WebSocket `response.create` 均使用请求开始时的配置，同一请求重试保持该配置； +HTTP 请求头及新建 WS 的握手提示按当时的最终出站档位构造；复用 WS 时不重发握手头， +每个 `response.create` 仍独立应用档位策略，请求档位统计与本地费用估算使用各帧的最终出站档位。 + `requestLocationEnabled` 是必填布尔值，默认 `false`：关闭时不覆盖客户端原有位置和时区；开启时使用已保存的 `requestLocation`。关闭不会清空自定义值，代理自定义位置仍优先。 `requestLocation` 是必填的完整对象 `{ country, region, city, timezone }`，不接受 `null`；初始保存值为 diff --git a/docs/architecture.md b/docs/architecture.md index 45a356a86..37eb552a6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -133,6 +133,13 @@ service；客户端原生对象保存在有界进程缓存中，与套餐 Redis `RuntimeSnapshot → RoutingPlan → AttemptContext` 冻结传递，关闭时保留客户端原有字段； Provider 在选定账号后应用代理位置覆盖。请求期间不额外查询全局设置，配置发布不改变已开始请求的全局值。 +Fast 限制由同一快照链路冻结：全局开关与 Client Key 所有绑定分组的开关取逻辑或； +禁用分组仍贡献限制，无分组 Key 只受全局开关影响，不按所选账号的分组重新解释。 +OpenAI Provider 在独立编码请求上、生成上游头与观测前统一应用顶层档位覆盖，HTTP、WS 与重试共用。 +该策略不改变选号、会话亲和或共享原始请求；每个新 WS 请求重新取得当前策略。 +WS 路由提示属于握手，连接复用时不重发；档位变化不重建连接或打断 `previous_response_id` 续写， +各帧正文及计费仍使用本次请求的最终档位。 + `engine::observation` 统一维护单次响应的用量、费用、时间和响应 ID，并负责重试前清理；协调器继续 独占发送、提交、重试和终结顺序。Provider 上报费用优先于本地估算，丢弃的 attempt 不得污染最终计量。 Provider 本地估算按当前 attempt 实际发送的上游模型查价，响应声明的模型只作观测；费用明细复用同一口径。 diff --git a/frontend/src/api/modules/account-groups.ts b/frontend/src/api/modules/account-groups.ts index e32619753..3bee61f24 100644 --- a/frontend/src/api/modules/account-groups.ts +++ b/frontend/src/api/modules/account-groups.ts @@ -25,6 +25,7 @@ export interface AccountGroupUsage { } export interface AccountGroup extends AccountGroupRef { + disableFast: boolean description: string | null memberCount: number providerCounts: Record @@ -63,12 +64,14 @@ interface AccountGroupListParams { } interface AccountGroupCreateParam { + disableFast?: boolean name: string description: string | null color: string } interface AccountGroupUpdateParam { + disableFast?: boolean id: string name: string description: string | null diff --git a/frontend/src/api/modules/settings.ts b/frontend/src/api/modules/settings.ts index a41f8ebe5..acc38f467 100644 --- a/frontend/src/api/modules/settings.ts +++ b/frontend/src/api/modules/settings.ts @@ -5,6 +5,7 @@ import request from '../request' export type RotationStrategy = 'smart' | 'quota_reset_priority' | 'round_robin' | 'sticky' export interface RuntimeSettings { + disableFast: boolean requestLocationEnabled: boolean requestLocation: RequestLocation modelMappings: Record diff --git a/frontend/src/views/groups/components/AccountGroupFormModal.vue b/frontend/src/views/groups/components/AccountGroupFormModal.vue index d89b1a148..13e110cf5 100644 --- a/frontend/src/views/groups/components/AccountGroupFormModal.vue +++ b/frontend/src/views/groups/components/AccountGroupFormModal.vue @@ -9,6 +9,7 @@ import BaseFormItem from '@/components/base/BaseForm/FormItem.vue' import BaseForm from '@/components/base/BaseForm/index.vue' import BaseInput from '@/components/base/BaseInput.vue' import BaseModal from '@/components/base/BaseModal/index.vue' +import BaseSwitch from '@/components/base/BaseSwitch.vue' import BaseTextarea from '@/components/base/BaseTextarea.vue' import { ACCOUNT_GROUP_COLOR_PRESETS } from '../constants' @@ -23,7 +24,7 @@ const open = defineModel({ required: true }) const form = defineModel('form', { required: true }) const title = computed(() => props.group ? '编辑分组' : '创建分组') const description = computed(() => props.group - ? '修改分组名称和用途说明。' + ? '修改分组名称、用途说明和 Fast 限制。' : '创建后，可在账号管理中将账号加入这个分组。') @@ -61,6 +62,14 @@ const description = computed(() => props.group :disabled="saving" /> + + +