diff --git a/backend/crates/gateway-api/tests/admin/settings.rs b/backend/crates/gateway-api/tests/admin/settings.rs index c7019c46d..1c889dab6 100644 --- a/backend/crates/gateway-api/tests/admin/settings.rs +++ b/backend/crates/gateway-api/tests/admin/settings.rs @@ -141,7 +141,7 @@ async fn smart_settings_round_trip_and_invalid_updates_leave_the_saved_value_int let fixture = AdminTestFixture::new().await; fixture.auth.insert_session("valid-session"); let mut body = update_body(); - let custom = json!({"loadWeight": 2.0, "quotaWeight": 0.0, "healthWeight": 1.0, "latencyWeight": 0.5, "resetWeight": 1.2, "queueWeight": 2.3, "preferHigherWeight": true}); + let custom = json!({"loadWeight": 2.0, "quotaWeight": 0.0, "healthWeight": 1.0, "latencyWeight": 0.5, "resetWeight": 1.2, "queueWeight": 2.3}); body["smartScheduling"] = custom.clone(); let response = app(fixture.state()) .oneshot(request( @@ -161,8 +161,8 @@ async fn smart_settings_round_trip_and_invalid_updates_leave_the_saved_value_int let mut invalid_values = vec![ json!(null), json!({}), - json!({"loadWeight":0,"quotaWeight":0,"healthWeight":0,"latencyWeight":0,"resetWeight":0,"queueWeight":0,"preferHigherWeight":true}), - json!({"loadWeight":0.01,"quotaWeight":1,"healthWeight":1,"latencyWeight":1,"resetWeight":0,"queueWeight":0,"preferHigherWeight":false}), + json!({"loadWeight":0,"quotaWeight":0,"healthWeight":0,"latencyWeight":0,"resetWeight":0,"queueWeight":0}), + json!({"loadWeight":0.01,"quotaWeight":1,"healthWeight":1,"latencyWeight":1,"resetWeight":0,"queueWeight":0}), ]; for field in ["resetWeight", "queueWeight"] { for value in [json!(-0.1), json!(10.1), json!(0.01), json!(null)] { diff --git a/backend/crates/gateway-core/src/account/selection.rs b/backend/crates/gateway-core/src/account/selection.rs index 28adbab90..ecec5a796 100644 --- a/backend/crates/gateway-core/src/account/selection.rs +++ b/backend/crates/gateway-core/src/account/selection.rs @@ -510,6 +510,8 @@ pub struct AccountSelectionContext { pub policy: AccountSelectionPolicy, pub now: SystemTime, pub excluded_accounts: BTreeSet, + /// 租约争用或队列让位只阻止立即选号,不排除等待资格 + pub busy_accounts: BTreeSet, pub preferred_account: Option, /// 优先账号可调度时,是否优先于其它账号的持久权重 pub preferred_account_overrides_weight: bool, @@ -602,7 +604,7 @@ impl<'a> AccountSelection<'a> { pub struct AccountSelector; impl AccountSelector { - /// 返回调度策略可见的全部合格候选;权重层授权由调用策略的宿主适配器裁剪 + /// 策略同时约束立即选号与等待,候选保留仅因本地容量受阻的账号 #[must_use] pub(crate) fn policy_candidates<'a>( &self, @@ -611,7 +613,12 @@ impl AccountSelector { ) -> Vec<&'a AccountCandidate> { candidates .iter() - .filter(|candidate| self.scheduling_blocker(candidate, context).is_none()) + .filter(|candidate| { + matches!( + self.scheduling_blocker(candidate, context), + None | Some(AccountSchedulingBlocker::ConcurrencyLimit) + ) + }) .collect() } @@ -697,10 +704,7 @@ impl AccountSelector { .max()?; let (mut preferred, preferred_candidate) = self.preferred_decision(candidates, context); if let Some(candidate) = preferred_candidate { - // 权重回切只裁决内置策略的软亲和,插件显式选号按实际选择记录结果 - let prefer_higher_weight = context.policy.strategy() == RotationStrategy::Smart - && context.policy.smart_scheduling().prefer_higher_weight(); - if (!context.preferred_account_overrides_weight || prefer_higher_weight) + if !context.preferred_account_overrides_weight && candidate.account.weight() < highest_weight { preferred = @@ -860,6 +864,9 @@ impl AccountSelector { if context.excluded_accounts.contains(candidate.account.id()) { return Some(AccountSchedulingBlocker::Excluded); } + if context.busy_accounts.contains(candidate.account.id()) { + return Some(AccountSchedulingBlocker::ConcurrencyLimit); + } if context .concurrency_limit(&candidate.account) .limit() diff --git a/backend/crates/gateway-core/src/account/smart_scheduling.rs b/backend/crates/gateway-core/src/account/smart_scheduling.rs index c87206075..3d9ccedad 100644 --- a/backend/crates/gateway-core/src/account/smart_scheduling.rs +++ b/backend/crates/gateway-core/src/account/smart_scheduling.rs @@ -6,7 +6,6 @@ use serde::{Deserialize, Serialize}; #[serde(into = "SmartSchedulingValues")] pub struct SmartSchedulingConfig { weights: [u8; 6], - prefer_higher_weight: bool, } impl<'de> Deserialize<'de> for SmartSchedulingConfig { @@ -24,7 +23,6 @@ impl Default for SmartSchedulingConfig { fn default() -> Self { Self { weights: [10, 8, 10, 5, 0, 0], - prefer_higher_weight: false, } } } @@ -39,10 +37,7 @@ pub enum SmartSchedulingConfigError { impl SmartSchedulingConfig { /// 顺序为负载、剩余额度、健康、首输出延迟、额度重置和排队压力 - pub fn new( - weights: [f64; 6], - prefer_higher_weight: bool, - ) -> Result { + pub fn new(weights: [f64; 6]) -> Result { let mut tenths = [0; 6]; for (index, field) in [ "loadWeight", @@ -67,10 +62,7 @@ impl SmartSchedulingConfig { if tenths == [0; 6] { return Err(SmartSchedulingConfigError::EmptyWeights); } - Ok(Self { - weights: tenths, - prefer_higher_weight, - }) + Ok(Self { weights: tenths }) } #[must_use] @@ -78,11 +70,6 @@ impl SmartSchedulingConfig { self.weights.map(|weight| f64::from(weight) / 10.0) } - #[must_use] - pub const fn prefer_higher_weight(self) -> bool { - self.prefer_higher_weight - } - pub(crate) fn score_tolerance(self) -> f64 { // 与系数同步缩放,避免仅放大相同比例的系数就改变近似最优候选集合 // 排队系数只参与入队选择,不能放大立即选号的候选容差 @@ -105,24 +92,20 @@ struct SmartSchedulingValues { latency_weight: f64, reset_weight: f64, queue_weight: f64, - prefer_higher_weight: bool, } impl TryFrom for SmartSchedulingConfig { type Error = SmartSchedulingConfigError; fn try_from(value: SmartSchedulingValues) -> Result { - Self::new( - [ - value.load_weight, - value.quota_weight, - value.health_weight, - value.latency_weight, - value.reset_weight, - value.queue_weight, - ], - value.prefer_higher_weight, - ) + Self::new([ + value.load_weight, + value.quota_weight, + value.health_weight, + value.latency_weight, + value.reset_weight, + value.queue_weight, + ]) } } @@ -143,7 +126,6 @@ impl From for SmartSchedulingValues { latency_weight, reset_weight, queue_weight, - prefer_higher_weight: value.prefer_higher_weight, } } } diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 9da9e9713..a4e017ef7 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -304,7 +304,6 @@ where "queueTimeoutMs": plan.account_selection_policy().queue_policy().timeout.as_millis(), "sessionAffinityTtlMs": plan.account_selection_policy().openai_session_affinity_ttl().as_millis(), "smartWeights": plan.account_selection_policy().smart_scheduling().weights(), - "preferHigherWeight": plan.account_selection_policy().smart_scheduling().prefer_higher_weight(), }, "requiredAccount": account_selection.required_account().map(|account| account.as_str()), "hasMiddleware": middleware.is_some(), diff --git a/backend/crates/gateway-core/src/engine/mod.rs b/backend/crates/gateway-core/src/engine/mod.rs index 5a52bdf02..d96b74310 100644 --- a/backend/crates/gateway-core/src/engine/mod.rs +++ b/backend/crates/gateway-core/src/engine/mod.rs @@ -829,7 +829,7 @@ impl AttemptContext { provider: &ProviderKind, model: Option<&str>, candidates: &'a [AccountCandidate], - context: &AccountSelectionContext, + context: &mut AccountSelectionContext, ) -> Result>, policy::AccountPolicyError> { match self.request.request_policy() { Some(policy) => { diff --git a/backend/crates/gateway-core/src/engine/policy.rs b/backend/crates/gateway-core/src/engine/policy.rs index 6ee86cef8..cabb2cb34 100644 --- a/backend/crates/gateway-core/src/engine/policy.rs +++ b/backend/crates/gateway-core/src/engine/policy.rs @@ -113,6 +113,7 @@ pub enum ModelRouteDecision { #[derive(Debug, Clone, PartialEq, Eq)] pub struct AccountScheduleCandidate { account_id: ProviderAccountId, + available: bool, weight: u16, in_flight: u32, maximum_concurrency: u32, @@ -124,6 +125,11 @@ pub struct AccountScheduleCandidate { } impl AccountScheduleCandidate { + #[must_use] + pub const fn available(&self) -> bool { + self.available + } + #[must_use] pub const fn account_id(&self) -> &ProviderAccountId { &self.account_id @@ -242,6 +248,7 @@ impl AccountScheduleInput { pub enum AccountScheduleDecision { Pick(ProviderAccountId), Delegate, + DelegateFiltered(BTreeSet), Reject, } @@ -284,6 +291,8 @@ pub struct RequestPolicyFault; /// Provider 选择器需要区分明确拒绝、策略故障和等待期间过期的选择 #[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] pub enum AccountPolicyError { + #[error("account scheduling policy left no selectable account")] + Filtered, #[error("account scheduling policy rejected the request")] Rejected, #[error("account scheduling policy failed")] @@ -414,14 +423,14 @@ impl RequestPolicyContext { .await } - /// 硬绑定已在候选资格中收窄;插件先决定,委托时再采用内置亲和与排序 + /// 插件只处理宿主候选;返回后的上下文保留过滤结果,供亲和重选和等待共用 pub async fn select_account<'a>( &self, attempt_index: NonZeroU32, provider: &ProviderKind, model: Option<&str>, candidates: &'a [AccountCandidate], - context: &AccountSelectionContext, + context: &mut AccountSelectionContext, ) -> Result>, AccountPolicyError> { let eligible = AccountSelector.policy_candidates(candidates, context); if eligible.is_empty() { @@ -431,6 +440,9 @@ impl RequestPolicyContext { .iter() .map(|candidate| AccountScheduleCandidate { account_id: candidate.account.id().clone(), + available: AccountSelector + .scheduling_blocker(candidate, context) + .is_none(), weight: candidate.account.weight().get(), in_flight: candidate.signals.in_flight, maximum_concurrency: context.concurrency_limit(&candidate.account).get(), @@ -458,27 +470,47 @@ impl RequestPolicyContext { .map_err(|_| AccountPolicyError::Fault)?; // 插件调用可能消耗显著时间;无论插件选择还是委托,均用返回时刻重新执行 // 同一资格判断,不能让调用前尚未过期的候选在调用后被内置策略选中 - let mut current = context.clone(); - current.now = SystemTime::now(); match decision { - AccountScheduleDecision::Delegate => AccountSelector - .select(candidates, ¤t) - .map(Some) - .ok_or(AccountPolicyError::StaleCandidate), - AccountScheduleDecision::Reject => Err(AccountPolicyError::Rejected), - AccountScheduleDecision::Pick(account_id) => { - if !eligible + AccountScheduleDecision::Delegate => {} + AccountScheduleDecision::Reject => return Err(AccountPolicyError::Rejected), + AccountScheduleDecision::DelegateFiltered(excluded) => { + let candidate_ids = eligible .iter() - .any(|candidate| candidate.account.id() == &account_id) + .map(|candidate| candidate.account.id()) + .collect::>(); + if excluded.iter().any(|id| !candidate_ids.contains(id)) { + return Err(AccountPolicyError::Fault); + } + let all_filtered = eligible + .iter() + .all(|candidate| excluded.contains(candidate.account.id())); + context.excluded_accounts.extend(excluded); + if all_filtered { + return Err(AccountPolicyError::Filtered); + } + } + AccountScheduleDecision::Pick(account_id) => { + if AccountSelector + .select_policy_candidate(candidates, context, &account_id) + .is_none() { return Err(AccountPolicyError::Fault); } - AccountSelector - .select_policy_candidate(candidates, ¤t, &account_id) + context.now = SystemTime::now(); + return AccountSelector + .select_policy_candidate(candidates, context, &account_id) .map(Some) - .ok_or(AccountPolicyError::StaleCandidate) + .ok_or(AccountPolicyError::StaleCandidate); } } + context.now = SystemTime::now(); + if AccountSelector + .policy_candidates(candidates, context) + .is_empty() + { + return Err(AccountPolicyError::StaleCandidate); + } + Ok(AccountSelector.select(candidates, context)) } } diff --git a/backend/crates/gateway-core/tests/account/mod.rs b/backend/crates/gateway-core/tests/account/mod.rs index fba309e4c..1905bdd53 100644 --- a/backend/crates/gateway-core/tests/account/mod.rs +++ b/backend/crates/gateway-core/tests/account/mod.rs @@ -102,6 +102,7 @@ fn context(strategy: RotationStrategy) -> AccountSelectionContext { policy: AccountSelectionPolicy::new(strategy, NonZeroU32::new(3).expect("positive")), now: SystemTime::now(), excluded_accounts: BTreeSet::new(), + busy_accounts: Default::default(), preferred_account: None, preferred_account_overrides_weight: false, round_robin_cursor: 0, diff --git a/backend/crates/gateway-core/tests/account/smart_scheduling.rs b/backend/crates/gateway-core/tests/account/smart_scheduling.rs index 3ab03f92d..d2cc6dd73 100644 --- a/backend/crates/gateway-core/tests/account/smart_scheduling.rs +++ b/backend/crates/gateway-core/tests/account/smart_scheduling.rs @@ -17,9 +17,9 @@ use super::{candidate_with_concurrency, context, weighted_candidate}; fn reserved_pool_scoring_and_trace_use_its_capacity_instead_of_the_normal_limit() { let mut context = context(RotationStrategy::Smart); context.reserved_concurrency = 5; - context.policy = context.policy.with_smart_scheduling( - SmartSchedulingConfig::new([1.0, 0.0, 0.0, 0.0, 0.0, 0.0], false).unwrap(), - ); + context.policy = context + .policy + .with_smart_scheduling(SmartSchedulingConfig::new([1.0, 0.0, 0.0, 0.0, 0.0, 0.0]).unwrap()); let candidates = [ candidate_with_concurrency("acct_large_normal", 2, 100), candidate_with_concurrency("acct_small_normal", 1, 1), @@ -44,14 +44,14 @@ fn smart_config_rejects_invalid_weights_and_incomplete_wire_values() { for dimension in 0..6 { let mut weights = [1.0; 6]; weights[dimension] = value; - assert!(SmartSchedulingConfig::new(weights, false).is_err()); + assert!(SmartSchedulingConfig::new(weights).is_err()); } } - assert!(SmartSchedulingConfig::new([0.0; 6], true).is_err()); + assert!(SmartSchedulingConfig::new([0.0; 6]).is_err()); for value in [json!({}), json!(null), json!({"loadWeight": 1.0})] { assert!(serde_json::from_value::(value).is_err()); } - let config = SmartSchedulingConfig::new([0.0, 0.1, 9.9, 10.0, 1.2, 2.3], true).unwrap(); + let config = SmartSchedulingConfig::new([0.0, 0.1, 9.9, 10.0, 1.2, 2.3]).unwrap(); let encoded = serde_json::to_value(config).unwrap(); assert_eq!( serde_json::from_value::(encoded).unwrap(), @@ -62,9 +62,9 @@ fn smart_config_rejects_invalid_weights_and_incomplete_wire_values() { #[test] fn reset_weight_rewards_only_future_resets_and_preserves_capacity_checks() { let mut context = context(RotationStrategy::Smart); - context.policy = context.policy.with_smart_scheduling( - SmartSchedulingConfig::new([0.0, 0.0, 0.0, 0.0, 1.0, 0.0], false).unwrap(), - ); + context.policy = context + .policy + .with_smart_scheduling(SmartSchedulingConfig::new([0.0, 0.0, 0.0, 0.0, 1.0, 0.0]).unwrap()); let mut candidates = [ candidate_with_concurrency("acct_unknown", 0, 2), candidate_with_concurrency("acct_past", 0, 2), @@ -137,7 +137,7 @@ fn queue_weight_balances_live_pressure_and_health_only_when_enabled_for_smart() candidates[1].signals.failure_rate_basis_points = Some(6000); let mut context = context(strategy); context.policy = context.policy.with_smart_scheduling( - SmartSchedulingConfig::new([0.0, 0.0, 1.0, 0.0, 0.0, queue_weight], false).unwrap(), + SmartSchedulingConfig::new([0.0, 0.0, 1.0, 0.0, 0.0, queue_weight]).unwrap(), ); let queue = ConcurrencyWaitQueue::default(); let head = queue @@ -197,7 +197,7 @@ fn queue_weight_does_not_widen_immediate_selection_tolerance() { context.round_robin_cursor = 1; for queue_weight in [0.0, 10.0] { context.policy = context.policy.with_smart_scheduling( - SmartSchedulingConfig::new([0.0, 1.0, 0.0, 0.0, 0.0, queue_weight], false).unwrap(), + SmartSchedulingConfig::new([0.0, 1.0, 0.0, 0.0, 0.0, queue_weight]).unwrap(), ); assert_eq!( AccountSelector @@ -210,9 +210,9 @@ fn queue_weight_does_not_widen_immediate_selection_tolerance() { "acct_a" ); } - context.policy = context.policy.with_smart_scheduling( - SmartSchedulingConfig::new([0.0, 0.0, 0.0, 0.0, 0.0, 1.0], false).unwrap(), - ); + context.policy = context + .policy + .with_smart_scheduling(SmartSchedulingConfig::new([0.0, 0.0, 0.0, 0.0, 0.0, 1.0]).unwrap()); // 仅开启排队系数时,立即可用的同权重账号按既有游标轮换 assert_eq!( AccountSelector @@ -244,7 +244,7 @@ fn each_smart_weight_controls_its_signal_without_relaxing_capacity() { let mut context = context(RotationStrategy::Smart); context.policy = context .policy - .with_smart_scheduling(SmartSchedulingConfig::new(weights, false).unwrap()); + .with_smart_scheduling(SmartSchedulingConfig::new(weights).unwrap()); let selected = AccountSelector.select(&candidates, &context).unwrap(); assert_eq!( selected.candidate().account.id().as_str(), @@ -283,8 +283,7 @@ fn proportional_coefficients_preserve_near_best_rotation_and_diagnostics() { for scale in [1.0, 2.0, 5.0] { let mut context = context(RotationStrategy::Smart); context.policy = context.policy.with_smart_scheduling( - SmartSchedulingConfig::new([scale, 0.8 * scale, scale, 0.5 * scale, 0.0, 0.0], false) - .unwrap(), + SmartSchedulingConfig::new([scale, 0.8 * scale, scale, 0.5 * scale, 0.0, 0.0]).unwrap(), ); let mut ids = Vec::new(); for cursor in 0..4 { @@ -313,64 +312,40 @@ fn proportional_coefficients_preserve_near_best_rotation_and_diagnostics() { } #[test] -fn switchback_is_opt_in_and_only_affects_smart_soft_affinity() { - let mut candidates = [ - weighted_candidate("acct_affinity", 10, 0), - weighted_candidate("acct_primary", 100, 0), - ]; +fn higher_weight_recovery_preserves_usable_soft_affinity() { for strategy in [ RotationStrategy::Smart, RotationStrategy::Sticky, RotationStrategy::RoundRobin, RotationStrategy::QuotaResetPriority, ] { + let mut candidates = [ + weighted_candidate("acct_affinity", 10, 0), + weighted_candidate("acct_primary", 100, 3), + ]; let mut context = context(strategy); context.preferred_account = Some(candidates[0].account.id().clone()); context.preferred_account_overrides_weight = true; - for enabled in [false, true] { - context.policy = context.policy.with_smart_scheduling( - SmartSchedulingConfig::new([1.0, 0.8, 1.0, 0.5, 0.0, 0.0], enabled).unwrap(), - ); + for primary_in_flight in [3, 0] { + candidates[1].signals.in_flight = primary_in_flight; let selected = AccountSelector.select(&candidates, &context).unwrap(); - let should_switch = enabled && strategy == RotationStrategy::Smart; assert_eq!( selected.candidate().account.id(), - candidates[usize::from(should_switch)].account.id() + candidates[0].account.id() ); + assert_eq!(selected.preferred(), PreferredAccountSelection::Hit); } + context + .excluded_accounts + .insert(candidates[0].account.id().clone()); + assert_eq!( + AccountSelector + .select(&candidates, &context) + .unwrap() + .candidate() + .account + .id(), + candidates[1].account.id() + ); } - let mut context = context(RotationStrategy::Smart); - context.policy = context.policy.with_smart_scheduling( - SmartSchedulingConfig::new([1.0, 0.8, 1.0, 0.5, 0.0, 0.0], true).unwrap(), - ); - context.preferred_account_overrides_weight = true; - context.preferred_account = Some(candidates[0].account.id().clone()); - candidates[1].signals.in_flight = 3; - assert_eq!( - AccountSelector - .select(&candidates, &context) - .unwrap() - .preferred(), - PreferredAccountSelection::Hit - ); - candidates[1].signals.in_flight = 0; - context - .excluded_accounts - .insert(candidates[1].account.id().clone()); - assert_eq!( - AccountSelector - .select(&candidates, &context) - .unwrap() - .preferred(), - PreferredAccountSelection::Hit - ); - context.excluded_accounts.clear(); - candidates[0] = weighted_candidate("acct_affinity", 100, 2); - assert_eq!( - AccountSelector - .select(&candidates, &context) - .unwrap() - .preferred(), - PreferredAccountSelection::Hit - ); } diff --git a/backend/crates/gateway-core/tests/engine/execution/mod.rs b/backend/crates/gateway-core/tests/engine/execution/mod.rs index dade26db1..1a2169f5e 100644 --- a/backend/crates/gateway-core/tests/engine/execution/mod.rs +++ b/backend/crates/gateway-core/tests/engine/execution/mod.rs @@ -1966,10 +1966,11 @@ impl Provider for PolicySelectingProvider { first_output_latency_ms: None, }, }]; - let selection = AccountSelectionContext { + let mut selection = AccountSelectionContext { policy: attempt.account_selection_policy(), now: SystemTime::now(), excluded_accounts: attempt.excluded_accounts().clone(), + busy_accounts: Default::default(), preferred_account: attempt.required_account().cloned(), preferred_account_overrides_weight: true, round_robin_cursor: 0, @@ -1985,14 +1986,16 @@ impl Provider for PolicySelectingProvider { .upstream_model() .map(UpstreamModelId::as_str), &candidates, - &selection, + &mut selection, ) .await { - Err(AccountPolicyError::Rejected) => Err(ProviderError::new( - ProviderErrorKind::RequestPolicyDenied, - UpstreamSendState::NotSent, - )), + Err(AccountPolicyError::Rejected | AccountPolicyError::Filtered) => { + Err(ProviderError::new( + ProviderErrorKind::RequestPolicyDenied, + UpstreamSendState::NotSent, + )) + } Err(AccountPolicyError::Fault | AccountPolicyError::StaleCandidate) => Err( ProviderError::new(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent), ), diff --git a/backend/crates/gateway-core/tests/engine/policy.rs b/backend/crates/gateway-core/tests/engine/policy.rs index 3f3799985..91f8f5c33 100644 --- a/backend/crates/gateway-core/tests/engine/policy.rs +++ b/backend/crates/gateway-core/tests/engine/policy.rs @@ -23,7 +23,7 @@ use gateway_core::{ }; #[derive(Debug)] -struct FixedScheduler(Option<&'static str>); +struct FixedScheduler(AccountScheduleDecision); impl RequestPolicyPlan for FixedScheduler { fn route_model( @@ -37,9 +37,7 @@ impl RequestPolicyPlan for FixedScheduler { &self, _: AccountScheduleInput, ) -> BoxFuture<'static, Result> { - let decision = self.0.map_or(AccountScheduleDecision::Delegate, |id| { - AccountScheduleDecision::Pick(account_id(id)) - }); + let decision = self.0.clone(); Box::pin(async move { Ok(decision) }) } } @@ -87,8 +85,14 @@ fn candidate(id: &str) -> AccountCandidate { } fn policy(choice: Option<&'static str>) -> RequestPolicyContext { + policy_decision(choice.map_or(AccountScheduleDecision::Delegate, |id| { + AccountScheduleDecision::Pick(account_id(id)) + })) +} + +fn policy_decision(decision: AccountScheduleDecision) -> RequestPolicyContext { RequestPolicyContext::new( - Arc::new(FixedScheduler(choice)), + Arc::new(FixedScheduler(decision)), ExtensionSetReference::new( ExtensionSetId::new("scheduler-test".into()).unwrap(), Arc::new(Lease), @@ -104,6 +108,7 @@ fn context() -> AccountSelectionContext { policy: AccountSelectionPolicy::new(RotationStrategy::Sticky, NonZeroU32::new(2).unwrap()), now: SystemTime::now(), excluded_accounts: BTreeSet::new(), + busy_accounts: Default::default(), preferred_account: Some(account_id("acct_a")), preferred_account_overrides_weight: true, round_robin_cursor: 0, @@ -123,7 +128,7 @@ fn scheduler_can_override_soft_affinity_and_reports_the_override() { &ProviderKind::new("openai").unwrap(), Some("demo-model"), &candidates, - &context(), + &mut context(), ) .await .unwrap() @@ -148,7 +153,7 @@ fn delegated_scheduler_keeps_builtin_affinity() { &ProviderKind::new("openai").unwrap(), None, &candidates, - &context(), + &mut context(), ) .await .unwrap() @@ -164,7 +169,7 @@ fn delegated_scheduler_keeps_builtin_affinity() { } #[test] -fn smart_switchback_does_not_override_explicit_plugin_choices_or_their_telemetry() { +fn explicit_plugin_choices_override_native_soft_affinity() { block_on(async { let mut candidates = [candidate("acct_a"), candidate("acct_b")]; candidates[0].account = candidates[0] @@ -175,10 +180,9 @@ fn smart_switchback_does_not_override_explicit_plugin_choices_or_their_telemetry selection_context.policy = AccountSelectionPolicy::new(RotationStrategy::Smart, NonZeroU32::new(2).unwrap()) .with_smart_scheduling( - gateway_core::account::SmartSchedulingConfig::new( - [1.0, 0.8, 1.0, 0.5, 0.0, 0.0], - true, - ) + gateway_core::account::SmartSchedulingConfig::new([ + 1.0, 0.8, 1.0, 0.5, 0.0, 0.0, + ]) .unwrap(), ); for chosen in [Some("acct_a"), Some("acct_b"), None] { @@ -188,21 +192,19 @@ fn smart_switchback_does_not_override_explicit_plugin_choices_or_their_telemetry &ProviderKind::new("openai").unwrap(), None, &candidates, - &selection_context, + &mut selection_context, ) .await .unwrap() .unwrap(); assert_eq!( selected.candidate().account.id().as_str(), - chosen.unwrap_or("acct_b") + chosen.unwrap_or("acct_a") ); let expected = match chosen { Some("acct_a") => PreferredAccountSelection::Hit, Some(_) => PreferredAccountSelection::OverriddenByPolicy, - None => PreferredAccountSelection::Blocked( - gateway_core::account::AccountSchedulingBlocker::LowerWeight, - ), + None => PreferredAccountSelection::Hit, }; assert_eq!(selected.preferred(), expected); } @@ -221,7 +223,181 @@ fn scheduler_cannot_select_outside_the_hard_bound_candidate_pool() { &ProviderKind::new("openai").unwrap(), None, &candidates, - &context, + &mut context, + ) + .await; + assert!(matches!(result, Err(AccountPolicyError::Fault))); + }); +} + +#[test] +fn filtered_delegation_preserves_native_weights_and_affinity_on_remaining_accounts() { + block_on(async { + let mut candidates = [ + candidate("acct_a"), + candidate("acct_b"), + candidate("acct_c"), + ]; + candidates[1].account = candidates[1] + .account + .clone() + .with_scheduling(None, AccountWeight::new(10).unwrap()); + let policy = policy_decision(AccountScheduleDecision::DelegateFiltered(BTreeSet::from([ + account_id("acct_a"), + ]))); + for strategy in [RotationStrategy::Smart, RotationStrategy::RoundRobin] { + let mut context = context(); + context.policy = AccountSelectionPolicy::new(strategy, NonZeroU32::new(2).unwrap()); + let selected = policy + .select_account( + NonZeroU32::MIN, + &ProviderKind::new("openai").unwrap(), + None, + &candidates, + &mut context, + ) + .await + .unwrap() + .unwrap(); + assert_eq!(selected.candidate().account.id(), &account_id("acct_c")); + assert!(!selected.is_policy_choice()); + assert_eq!( + gateway_core::account::AccountSelector.wait_candidates(&candidates, &context), + [account_id("acct_c"), account_id("acct_b")] + ); + assert_eq!( + gateway_core::account::AccountSelector + .select(&candidates, &context) + .unwrap() + .candidate() + .account + .id(), + &account_id("acct_c") + ); + } + }); +} + +#[test] +fn filtered_delegation_covers_busy_candidates_and_does_not_persist_between_selections() { + block_on(async { + let mut candidates = [candidate("acct_a"), candidate("acct_b")]; + candidates[0].signals.in_flight = 2; + let mut selection = context(); + selection.busy_accounts.insert(account_id("acct_b")); + let filtered = + policy_decision(AccountScheduleDecision::DelegateFiltered(BTreeSet::from([ + account_id("acct_a"), + ]))); + assert!( + filtered + .select_account( + NonZeroU32::MIN, + &ProviderKind::new("openai").unwrap(), + None, + &candidates, + &mut selection + ) + .await + .unwrap() + .is_none() + ); + assert_eq!( + gateway_core::account::AccountSelector.wait_candidates(&candidates, &selection), + [account_id("acct_b")] + ); + candidates[0].signals.in_flight = 0; + let recovered = policy(None) + .select_account( + NonZeroU32::MIN, + &ProviderKind::new("openai").unwrap(), + None, + &candidates, + &mut context(), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(recovered.candidate().account.id(), &account_id("acct_a")); + }); +} + +#[test] +fn filtering_all_candidates_is_terminal_even_when_the_only_candidate_is_busy_or_pinned() { + block_on(async { + for busy in [false, true] { + let candidates = [candidate("acct_a"), candidate("acct_b")]; + let mut context = context(); + context.excluded_accounts.insert(account_id("acct_b")); + if busy { + context.busy_accounts.insert(account_id("acct_a")); + } + let result = + policy_decision(AccountScheduleDecision::DelegateFiltered(BTreeSet::from([ + account_id("acct_a"), + ]))) + .select_account( + NonZeroU32::MIN, + &ProviderKind::new("openai").unwrap(), + None, + &candidates, + &mut context, + ) + .await; + assert!(matches!(result, Err(AccountPolicyError::Filtered))); + } + }); +} + +#[test] +fn filtering_ids_outside_the_host_candidate_pool_is_a_policy_fault() { + block_on(async { + let candidates = [candidate("acct_a"), candidate("acct_b")]; + for id in ["acct_b", "acct_unknown"] { + let mut context = context(); + context.excluded_accounts.insert(account_id("acct_b")); + let result = + policy_decision(AccountScheduleDecision::DelegateFiltered(BTreeSet::from([ + account_id(id), + ]))) + .select_account( + NonZeroU32::MIN, + &ProviderKind::new("openai").unwrap(), + None, + &candidates, + &mut context, + ) + .await; + assert!(matches!(result, Err(AccountPolicyError::Fault))); + } + }); +} + +#[test] +fn empty_filter_preserves_native_affinity_and_busy_candidates_cannot_be_picked() { + block_on(async { + let candidates = [candidate("acct_a"), candidate("acct_b")]; + let selected = policy_decision(AccountScheduleDecision::DelegateFiltered(BTreeSet::new())) + .select_account( + NonZeroU32::MIN, + &ProviderKind::new("openai").unwrap(), + None, + &candidates, + &mut context(), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(selected.preferred(), PreferredAccountSelection::Hit); + let mut context = context(); + context.busy_accounts.insert(account_id("acct_a")); + let result = policy(Some("acct_a")) + .select_account( + NonZeroU32::MIN, + &ProviderKind::new("openai").unwrap(), + None, + &candidates, + &mut context, ) .await; assert!(matches!(result, Err(AccountPolicyError::Fault))); diff --git a/backend/crates/gateway-core/tests/routing/snapshot.rs b/backend/crates/gateway-core/tests/routing/snapshot.rs index 626a6781f..26b7ddbe8 100644 --- a/backend/crates/gateway-core/tests/routing/snapshot.rs +++ b/backend/crates/gateway-core/tests/routing/snapshot.rs @@ -443,7 +443,7 @@ fn compiled_plans_keep_their_scheduling_config_after_a_new_snapshot_is_built() { use gateway_core::account::SmartSchedulingConfig; let configs = [ SmartSchedulingConfig::default(), - SmartSchedulingConfig::new([0.0, 2.0, 1.0, 0.5, 1.2, 2.3], true).unwrap(), + SmartSchedulingConfig::new([0.0, 2.0, 1.0, 0.5, 1.2, 2.3]).unwrap(), ]; let mut plans = Vec::new(); for (index, config) in configs.into_iter().enumerate() { diff --git a/backend/crates/gateway-plugin/runtime/plugin-host-compatibility.json b/backend/crates/gateway-plugin/runtime/plugin-host-compatibility.json index 135d8f695..402a36b54 100644 --- a/backend/crates/gateway-plugin/runtime/plugin-host-compatibility.json +++ b/backend/crates/gateway-plugin/runtime/plugin-host-compatibility.json @@ -4,7 +4,7 @@ "protocol_versions": [2], "capabilities": [ { "capability": "frontend_authentication", "versions": [1] }, - { "capability": "scheduler", "versions": [1] }, + { "capability": "scheduler", "versions": [2] }, { "capability": "model_router", "versions": [1] }, { "capability": "model_catalog", "versions": [1] }, { "capability": "retry_policy", "versions": [1] }, diff --git a/backend/crates/gateway-plugin/runtime/src/adapter/policy/route_schedule.rs b/backend/crates/gateway-plugin/runtime/src/adapter/policy/route_schedule.rs index 795d87ea4..58a27941b 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/policy/route_schedule.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/policy/route_schedule.rs @@ -196,6 +196,7 @@ async fn schedule_with_entry( .iter() .map(|candidate| WireAccountScheduleCandidate { account_id: candidate.account_id().as_str().to_owned(), + available: candidate.available(), weight: candidate.weight(), in_flight: candidate.in_flight(), maximum_concurrency: candidate.maximum_concurrency(), @@ -245,11 +246,27 @@ async fn schedule_with_entry( match decision { WireAccountScheduleDecision::Delegate => Ok(AccountScheduleDecision::Delegate), WireAccountScheduleDecision::Reject => Ok(AccountScheduleDecision::Reject), + WireAccountScheduleDecision::DelegateFiltered { + exclude_account_ids, + } => { + let candidate_ids = candidates + .iter() + .map(|candidate| candidate.account_id()) + .collect::>(); + let excluded = exclude_account_ids + .into_iter() + .map(|id| { + let id = ProviderAccountId::new(id).map_err(|_| ())?; + candidate_ids.contains(&id).then_some(id).ok_or(()) + }) + .collect::>()?; + Ok(AccountScheduleDecision::DelegateFiltered(excluded)) + } WireAccountScheduleDecision::Pick { account_id } => { let account_id = ProviderAccountId::new(account_id).map_err(|_| ())?; if !candidates .iter() - .any(|candidate| candidate.account_id() == &account_id) + .any(|candidate| candidate.account_id() == &account_id && candidate.available()) { return Err(()); } diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs b/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs index c6c406d4a..461a4675c 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs @@ -5,6 +5,7 @@ mod facts; mod http; mod lifecycle; mod mounts; +mod scheduling; mod service; mod websocket; @@ -556,6 +557,7 @@ fn selection_context() -> AccountSelectionContext { ), now: SystemTime::now(), excluded_accounts: BTreeSet::new(), + busy_accounts: Default::default(), preferred_account: None, preferred_account_overrides_weight: true, round_robin_cursor: 0, @@ -582,14 +584,14 @@ async fn scheduler_cannot_pick_outside_the_host_projected_priority_tier() { let generation = prepare(&runtime).await; let context = policy_context(&runtime, generation.clone(), "req_schedule"); let candidates = [candidate("acct_high", 100), candidate("acct_low", 10)]; - let selection = selection_context(); + let mut selection = selection_context(); let selected = context .select_account( NonZeroU32::MIN, &ProviderKind::new("example").unwrap(), Some("upstream-a"), &candidates, - &selection, + &mut selection, ) .await .unwrap() diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/policy/scheduling.rs b/backend/crates/gateway-plugin/runtime/tests/generation/policy/scheduling.rs new file mode 100644 index 000000000..57734a71e --- /dev/null +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/scheduling.rs @@ -0,0 +1,69 @@ +//! 验证真实插件进程的候选过滤、繁忙投影与非法决定边界 + +use super::*; +use gateway_core::engine::policy::AccountPolicyError; + +#[tokio::test] +async fn scheduler_filter_contract_covers_busy_candidates_and_rejects_out_of_scope_ids() { + for (decision, expected) in [ + ( + serde_json::json!({"decision":"delegate_filtered","exclude_account_ids":["acct_high", "acct_high"]}), + "acct_low", + ), + ( + serde_json::json!({"decision":"delegate_filtered","exclude_account_ids":["acct_high", "acct_low"]}), + "filtered", + ), + ( + serde_json::json!({"decision":"delegate_filtered","exclude_account_ids":["acct_unknown"]}), + "fault", + ), + ( + serde_json::json!({"decision":"pick","account_id":"acct_high"}), + "fault", + ), + ] { + let marker = tempfile::NamedTempFile::new().unwrap(); + let (cache, runtime) = setup(vec![InstanceFixture { + id: "scheduler", + configuration: serde_json::json!({"schedule_decision":decision, "schedule_marker": marker.path()}), + bindings: vec![binding(SCHEDULER_CONTRIBUTION, "scheduling", 0, PluginFailurePolicy::Reject)], + }]).await; + let generation = prepare(&runtime).await; + let policy = policy_context(&runtime, generation.clone(), "req_filter"); + let candidates = [candidate("acct_high", 100), candidate("acct_low", 10)]; + let mut selection = selection_context(); + selection + .busy_accounts + .insert(candidates[0].account.id().clone()); + selection.preferred_account = Some(candidates[0].account.id().clone()); + let result = policy + .select_account( + NonZeroU32::MIN, + &ProviderKind::new("example").unwrap(), + None, + &candidates, + &mut selection, + ) + .await; + match expected { + "filtered" => assert!(matches!(result, Err(AccountPolicyError::Filtered))), + "fault" => assert!(matches!(result, Err(AccountPolicyError::Fault))), + id => assert_eq!( + result.unwrap().unwrap().candidate().account.id().as_str(), + id + ), + } + let request: serde_json::Value = + serde_json::from_str(std::fs::read_to_string(marker.path()).unwrap().trim()).unwrap(); + assert_eq!( + request["request"]["candidates"].as_array().unwrap().len(), + 2 + ); + assert_eq!(request["request"]["candidates"][0]["available"], false); + assert_eq!(request["request"]["candidates"][1]["available"], true); + drop(policy); + drop(generation); + super::super::wait_until_empty(cache.path()).await; + } +} diff --git a/backend/crates/gateway-plugin/runtime/tests/support/mod.rs b/backend/crates/gateway-plugin/runtime/tests/support/mod.rs index 436e93cc9..c329c33a4 100644 --- a/backend/crates/gateway-plugin/runtime/tests/support/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/support/mod.rs @@ -59,7 +59,7 @@ pub fn contribution_for_id( id: format!("{plugin_id}.{local_id}"), version: match capability { Capability::Middleware => 4, - Capability::UpstreamAdapter => 2, + Capability::Scheduler | Capability::UpstreamAdapter => 2, _ => 1, }, stages, diff --git a/backend/crates/gateway-plugin/sdk/README.md b/backend/crates/gateway-plugin/sdk/README.md index 69040d388..e1507d234 100644 --- a/backend/crates/gateway-plugin/sdk/README.md +++ b/backend/crates/gateway-plugin/sdk/README.md @@ -9,7 +9,7 @@ SDK 处于实验阶段;能力是否可用取决于宿主支持、清单声明 | SDK 包 | `0.1.0` | Rust 开发依赖,独立于网关版本 | | `manifestVersion` | `2` | 安装清单格式 | | `package.protocolVersion` | `2` | 宿主与插件的进程通信 | -| `contributes.*.version` | 默认 `1`,`middleware` 为 `4`,`upstream_adapter` 为 `2` | 单项能力合同,见[扩展项声明](docs/manifest.md#扩展项简写) | +| `contributes.*.version` | 默认 `1`,`middleware` 为 `4`,`scheduler` / `upstream_adapter` 为 `2` | 单项能力合同,见[扩展项声明](docs/manifest.md#扩展项简写) | 开发顺序:编写[清单](docs/manifest.md) → 实现[能力处理器](docs/capabilities.md) → 用 [CLI](../../../apps/plugin-cli/README.md) 打包 → [安装、配置与启用](../../../../docs/plugins.md)。完整可运行示例位于独立仓库 diff --git a/backend/crates/gateway-plugin/sdk/docs/capabilities.md b/backend/crates/gateway-plugin/sdk/docs/capabilities.md index ad5ed1011..dd0831b73 100644 --- a/backend/crates/gateway-plugin/sdk/docs/capabilities.md +++ b/backend/crates/gateway-plugin/sdk/docs/capabilities.md @@ -469,6 +469,24 @@ middleware 的 `next` 最多调用一次,重试与额外模型执行通过独 ### 路由、调度与观察 `policy.route_model` 和 `policy.schedule_account` 只提出决定,宿主随后复核 Key 规则、模型能力、账号资格和租约。 +`scheduler` 必须声明 `version: 2`。`candidates` 已按 Key 范围、模型权限、账号状态和原生续写绑定收窄, +同时包含暂时繁忙但可以等待的账号;`available` 表示本次可以立即尝试租约,不保证租约一定成功 + +- `Pick { account_id }` 显式选号,只能指定 `available: true` 的候选 +- `Delegate` 将完整候选交给宿主现有调度 +- `DelegateFiltered { exclude_account_ids }` 排除列出的候选,再委托内置权重、Smart 评分和亲和选择;仅接受本次候选 ID,重复 ID 合并,空列表等同于 `Delegate` +- `Reject` 明确拒绝请求 + +过滤结果约束本次评估后的亲和重选、容量判断和等待队列。全部候选被排除,或严格跟随的会话账号被排除时, +请求以 `account_policy_filtered` 结束,不回退到被排除账号或伪装成额度耗尽。租约争用、排队唤醒和后续请求重新评估策略, +过滤结果不写入账号状态。插件决定自身阈值、恢复条件和配置含义;宿主只检查合同、候选边界并执行决定。 +插件设置使用自己的 `configurationSchema` 或管理页面,不向宿主智能调度弹窗注入字段。 +非法输出按绑定的既有故障策略处理,不自动修正插件业务规则 + +```json +{"decision":"delegate_filtered","exclude_account_ids":["acct_example"]} +``` + `observer` v1 通过 `methods::OBSERVE` 接收 `call::observation::Event`,固定绑定 `observation/observe`。 每条绑定以 `event` 选择 `request_completed` 或 `websocket_response`,可分别配置请求范围;同一实例每种事件至多一条绑定,两类绑定使用相同 `order`。 事件选择只决定是否投递,不裁剪字段 diff --git a/backend/crates/gateway-plugin/sdk/docs/manifest.md b/backend/crates/gateway-plugin/sdk/docs/manifest.md index 97062b93a..f9c367903 100644 --- a/backend/crates/gateway-plugin/sdk/docs/manifest.md +++ b/backend/crates/gateway-plugin/sdk/docs/manifest.md @@ -50,7 +50,7 @@ ## 扩展项简写 -作者声明可以省略 `id` 和固定阶段;普通能力的版本默认为 `1`,`middleware` 显式声明版本 `4`,`upstream_adapter` 显式声明版本 `2`: +作者声明可以省略 `id` 和固定阶段;普通能力的版本默认为 `1`,`middleware` 显式声明版本 `4`,`scheduler` 和 `upstream_adapter` 显式声明版本 `2`: ```json { diff --git a/backend/crates/gateway-plugin/sdk/src/call/policy.rs b/backend/crates/gateway-plugin/sdk/src/call/policy.rs index b1ee4932f..9bb162dc1 100644 --- a/backend/crates/gateway-plugin/sdk/src/call/policy.rs +++ b/backend/crates/gateway-plugin/sdk/src/call/policy.rs @@ -77,6 +77,8 @@ pub enum ModelRouteDecision { #[serde(deny_unknown_fields)] pub struct AccountScheduleCandidate { pub account_id: String, + /// false 表示仅可等待,仍可被过滤,但不能作为本轮 Pick 的结果 + pub available: bool, pub weight: u16, pub in_flight: u32, pub maximum_concurrency: u32, @@ -108,7 +110,13 @@ pub struct AccountScheduleRequest { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "decision", rename_all = "snake_case", deny_unknown_fields)] pub enum AccountScheduleDecision { - Pick { account_id: String }, + Pick { + account_id: String, + }, Delegate, + /// 只收窄本轮候选;参数与排除策略由插件拥有,宿主继续执行原生调度 + DelegateFiltered { + exclude_account_ids: Vec, + }, Reject, } diff --git a/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs b/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs index 8b7d6c32a..209cb0d89 100644 --- a/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs +++ b/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs @@ -212,7 +212,6 @@ pub struct SmartSchedulingConfig { pub latency_weight: f64, pub reset_weight: f64, pub queue_weight: f64, - pub prefer_higher_weight: bool, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] diff --git a/backend/crates/gateway-plugin/sdk/src/capability.rs b/backend/crates/gateway-plugin/sdk/src/capability.rs index eb92a77b6..07cb423c3 100644 --- a/backend/crates/gateway-plugin/sdk/src/capability.rs +++ b/backend/crates/gateway-plugin/sdk/src/capability.rs @@ -28,7 +28,6 @@ impl Capability { match self { Self::Middleware => &[4], Self::FrontendAuthentication - | Self::Scheduler | Self::ModelRouter | Self::ModelCatalog | Self::RetryPolicy @@ -36,7 +35,7 @@ impl Capability { | Self::CommandLine | Self::Management | Self::Maintenance => &[1], - Self::UpstreamAdapter => &[2], + Self::Scheduler | Self::UpstreamAdapter => &[2], } } diff --git a/backend/crates/gateway-plugin/sdk/tests/call/policy.rs b/backend/crates/gateway-plugin/sdk/tests/call/policy.rs index e38f4d1de..7a3ec9987 100644 --- a/backend/crates/gateway-plugin/sdk/tests/call/policy.rs +++ b/backend/crates/gateway-plugin/sdk/tests/call/policy.rs @@ -45,6 +45,7 @@ fn route_and_scheduler_decisions_are_explicit_and_strict() { model: Some("gpt-5".into()), candidates: vec![AccountScheduleCandidate { account_id: "acct_candidate".into(), + available: true, weight: 10, in_flight: 1, maximum_concurrency: 4, @@ -68,3 +69,23 @@ fn route_and_scheduler_decisions_are_explicit_and_strict() { json!({"decision":"pick","account_id":"acct_candidate"}) ); } + +#[test] +fn filtered_delegation_has_an_explicit_exclusion_list() { + let decision = AccountScheduleDecision::DelegateFiltered { + exclude_account_ids: vec!["acct_candidate".into()], + }; + let wire = json!({"decision":"delegate_filtered","exclude_account_ids":["acct_candidate"]}); + assert_eq!(serde_json::to_value(&decision).unwrap(), wire); + assert_eq!( + serde_json::from_value::(wire).unwrap(), + decision + ); + for invalid in [ + json!({"decision":"delegate_filtered"}), + json!({"decision":"delegate_filtered","exclude_account_ids":null}), + json!({"decision":"delegate_filtered","exclude_account_ids":[],"threshold":90}), + ] { + assert!(serde_json::from_value::(invalid).is_err()); + } +} diff --git a/backend/crates/gateway-plugin/sdk/tests/manifest.rs b/backend/crates/gateway-plugin/sdk/tests/manifest.rs index 073fa8067..70d154bf6 100644 --- a/backend/crates/gateway-plugin/sdk/tests/manifest.rs +++ b/backend/crates/gateway-plugin/sdk/tests/manifest.rs @@ -438,3 +438,26 @@ fn contribution_declaration_uses_camel_case_description_fields() { }) ); } + +#[test] +fn scheduler_requires_the_current_contract() { + let mut manifest = packaged_manifest(); + manifest.contributes.clear(); + manifest.contributes.insert( + Capability::Scheduler, + ContributionDeclaration { + id: "9acme.request-tags.scheduler".into(), + version: 2, + stages: vec![Stage::Scheduling], + input_formats: vec![], + output_formats: vec![], + }, + ); + assert!(manifest.validate().is_ok()); + manifest + .contributes + .get_mut(&Capability::Scheduler) + .unwrap() + .version = 1; + assert_eq!(manifest.validate(), Err(ManifestError::Invalid)); +} diff --git a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs index 934fd7698..74a8fd546 100644 --- a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs @@ -211,6 +211,11 @@ async fn privacy_steps_migration_preserves_rules_and_other_settings() { .await .unwrap(); expected_row["codex_privacy_policy_json"] = expected.clone(); + // 完整升级还会清理自动回切字段,其余运行设置必须原样保留 + expected_row["smart_scheduling_json"] + .as_object_mut() + .unwrap() + .remove("preferHigherWeight"); super::TEST_MIGRATOR.run(&database.pool).await.unwrap(); let stored: serde_json::Value = @@ -309,7 +314,7 @@ async fn smart_settings_upgrade_preserves_selection_and_publishes_custom_config( ); let mut update = settings_with_margin(3600); update.values.smart_scheduling = - SmartSchedulingConfig::new([0.0, 2.0, 1.0, 0.5, 1.2, 2.3], true).unwrap(); + SmartSchedulingConfig::new([0.0, 2.0, 1.0, 0.5, 1.2, 2.3]).unwrap(); let expected = update.values.smart_scheduling; repository.update_runtime_settings(update).await.unwrap(); let reloaded = repository.load_runtime_settings().await.unwrap(); @@ -1437,3 +1442,44 @@ async fn preferred_affinity_migration_defaults_to_strict_and_preserves_saved_mod database.close().await; } } + +#[tokio::test] +async fn removing_switchback_preserves_custom_weights_and_replaces_database_default() { + let Some(database) = TestDatabase::create_through("remove_switchback", 34).await else { + return; + }; + let expected = + gateway_core::account::SmartSchedulingConfig::new([0.0, 2.0, 1.0, 0.5, 1.2, 2.3]).unwrap(); + let mut legacy = serde_json::to_value(expected).unwrap(); + legacy["preferHigherWeight"] = serde_json::json!(true); + sqlx::query("update runtime_settings set smart_scheduling_json = $1 where id = 1") + .bind(legacy) + .execute(&database.pool) + .await + .unwrap(); + super::TEST_MIGRATOR.run(&database.pool).await.unwrap(); + let repository = PgRuntimeSettingsRepository::new(database.pool.clone()); + assert_eq!( + repository + .load_runtime_settings() + .await + .unwrap() + .values + .smart_scheduling, + expected + ); + sqlx::query("update runtime_settings set smart_scheduling_json = default where id = 1") + .execute(&database.pool) + .await + .unwrap(); + assert_eq!( + repository + .load_runtime_settings() + .await + .unwrap() + .values + .smart_scheduling, + gateway_core::account::SmartSchedulingConfig::default() + ); + database.close().await; +} diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index 107c036fd..405ddad77 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -486,6 +486,7 @@ impl CodexCredentialSelector { error, CredentialSelectionError::SessionBound(_) | CredentialSelectionError::ContinuationOwnerChanged + | CredentialSelectionError::PolicyFiltered | CredentialSelectionError::PolicyRejected | CredentialSelectionError::PolicyUnavailable ) @@ -750,7 +751,7 @@ impl CodexCredentialSelector { } } let mut shortest_retry = None; - let base_excluded = excluded.clone(); + let mut busy_accounts = std::collections::BTreeSet::new(); let policy = request.attempt.account_selection_policy(); loop { @@ -761,6 +762,7 @@ impl CodexCredentialSelector { policy, now: SystemTime::now(), excluded_accounts: excluded.clone(), + busy_accounts: busy_accounts.clone(), preferred_account: preferred.clone(), preferred_account_overrides_weight: true, round_robin_cursor, @@ -772,12 +774,34 @@ impl CodexCredentialSelector { account_scope: request.attempt.account_scope().cloned(), reserved_concurrency, }; - let wait_context = AccountSelectionContext { - excluded_accounts: base_excluded.clone(), - ..context.clone() + for candidate in &candidates { + if !waiting.can_try(candidate.account.id()) { + context.busy_accounts.insert(candidate.account.id().clone()); + } + } + let selection = match request + .attempt + .select_account( + &self.provider_kind, + model.model(), + &candidates, + &mut context, + ) + .await + { + Ok(selection) => selection, + Err(AccountPolicyError::StaleCandidate) => continue 'capacity, + Err(AccountPolicyError::Filtered) => { + return Err(CredentialSelectionError::PolicyFiltered); + } + Err(AccountPolicyError::Rejected) => { + return Err(CredentialSelectionError::PolicyRejected); + } + Err(AccountPolicyError::Fault) => { + return Err(CredentialSelectionError::PolicyUnavailable); + } }; - let mut wait_candidates = - AccountSelector.wait_candidates(&candidates, &wait_context); + let mut wait_candidates = AccountSelector.wait_candidates(&candidates, &context); let preferred_wait = preferred.as_ref().filter(|account| { !diagnostic && (affinity_mode == AccountAffinity::Strict || pinned_account.is_some()) @@ -788,27 +812,6 @@ impl CodexCredentialSelector { wait_candidates.retain(|account| account == preferred); } let capacity = AccountSelector.capacity_snapshot(&candidates, &context); - for candidate in &candidates { - if !waiting.can_try(candidate.account.id()) { - context - .excluded_accounts - .insert(candidate.account.id().clone()); - } - } - let selection = match request - .attempt - .select_account(&self.provider_kind, model.model(), &candidates, &context) - .await - { - Ok(selection) => selection, - Err(AccountPolicyError::StaleCandidate) => continue 'capacity, - Err(AccountPolicyError::Rejected) => { - return Err(CredentialSelectionError::PolicyRejected); - } - Err(AccountPolicyError::Fault) => { - return Err(CredentialSelectionError::PolicyUnavailable); - } - }; // 只约束内置调度;插件的显式选号保留原裁决和完整候选 // 首次绑定由任意会话请求原子认领;后代只在已有绑定时限制为跟随当前账号 let selection = if ((follow_only && binding.is_some()) @@ -816,12 +819,20 @@ impl CodexCredentialSelector { && affinity.preferred_account().is_some())) && selection.is_none_or(|selection| !selection.is_policy_choice()) { - // 优先模式先尝试会话主账号,繁忙或不可用时才采用调度结果;高权重回切不抢占主账号 + // 优先模式先尝试会话主账号,繁忙或不可用时才采用调度结果 let preferred = if follow_only { binding.as_ref().map(ProviderSessionBinding::account_id) } else { preferred.as_ref() }; + if follow_only + && preferred.is_some_and(|account| { + context.excluded_accounts.contains(account) + && !excluded.contains(account) + }) + { + return Err(CredentialSelectionError::PolicyFiltered); + } let mut follow_context = context.clone(); follow_context.now = SystemTime::now(); follow_context.excluded_accounts.extend( @@ -839,7 +850,7 @@ impl CodexCredentialSelector { } else { selection }; - // 可等待集合使用原始排除集;这里的 Excluded 只可能来自本轮租约争用或队列让位 + // 软亲和只等待本地容量,策略排除不能被当作繁忙账号恢复 let selection = selection.filter(|selection| { preferred_wait.is_none() || selection.is_policy_choice() @@ -847,7 +858,6 @@ impl CodexCredentialSelector { selection.preferred(), PreferredAccountSelection::Blocked( AccountSchedulingBlocker::ConcurrencyLimit - | AccountSchedulingBlocker::Excluded ) ) }); @@ -870,7 +880,7 @@ impl CodexCredentialSelector { &mut waiting, &wait_candidates, &candidates, - &wait_context, + &context, ) .await .map_err(|error| { @@ -886,10 +896,12 @@ impl CodexCredentialSelector { continue 'capacity; } // 只判断本次账号范围;空池、认证失效和租约失败不能伪装成额度耗尽 - // 使用最初的排除集合,避免本轮临时跳过的繁忙账号丢失其可恢复语义 + // 策略排除与临时容量阻塞分开,保留繁忙账号的可恢复语义 let mut statuses = candidates .iter() - .filter(|candidate| !base_excluded.contains(candidate.account.id())) + .filter(|candidate| { + !context.excluded_accounts.contains(candidate.account.id()) + }) .map(|candidate| { candidate .account @@ -933,7 +945,7 @@ impl CodexCredentialSelector { }; // 凭据重读和插件选号都可能挂起,取得租约前再次让位于新队首 if !waiting.can_try(account.id()) { - excluded.insert(account.id().clone()); + busy_accounts.insert(account.id().clone()); continue; } let allows_account_state_mutation = !diagnostic || account.enabled(); @@ -957,7 +969,7 @@ impl CodexCredentialSelector { affinity.observe_lease_busy(account.id()); } shortest_retry = minimum_duration(shortest_retry, retry_after); - excluded.insert(account.id().clone()); + busy_accounts.insert(account.id().clone()); } ProviderLeaseAcquisition::Acquired(guard) => { if let Some(key) = binding_key @@ -1766,6 +1778,8 @@ pub enum CredentialSelectionError { Coordinator(#[source] ProviderStoreError), #[error("Codex Cookie policy rejected the value")] CookiePolicy(#[source] super::cookie::CookiePolicyError), + #[error("account scheduling policy left no selectable account")] + PolicyFiltered, #[error("account scheduling policy rejected the request")] PolicyRejected, #[error("account scheduling policy is unavailable")] diff --git a/backend/crates/providers/openai/src/provider/observation.rs b/backend/crates/providers/openai/src/provider/observation.rs index 3ee332064..5772266e2 100644 --- a/backend/crates/providers/openai/src/provider/observation.rs +++ b/backend/crates/providers/openai/src/provider/observation.rs @@ -819,6 +819,7 @@ fn selection_failure(error: &CredentialSelectionError) -> ProviderError { CredentialSelectionError::Coordinator(_) => Some("account_coordinator_unavailable"), CredentialSelectionError::CookiePolicy(_) => Some("account_cookie_policy_rejected"), CredentialSelectionError::PolicyUnavailable => Some("account_policy_unavailable"), + CredentialSelectionError::PolicyFiltered => Some("account_policy_filtered"), _ => None, }; // 这些变体只含固定文案,不格式化可能承载外部正文的其他错误 @@ -871,7 +872,7 @@ fn selection_failure(error: &CredentialSelectionError) -> ProviderError { ProviderErrorKind::ProviderInfrastructureUnavailable, UpstreamSendState::NotSent, ), - CredentialSelectionError::PolicyRejected => provider_error( + CredentialSelectionError::PolicyRejected | CredentialSelectionError::PolicyFiltered => provider_error( ProviderErrorKind::RequestPolicyDenied, UpstreamSendState::NotSent, ), diff --git a/backend/crates/providers/openai/tests/credential/contract/capacity.rs b/backend/crates/providers/openai/tests/credential/contract/capacity.rs index 6eb0117b1..9401fa1da 100644 --- a/backend/crates/providers/openai/tests/credential/contract/capacity.rs +++ b/backend/crates/providers/openai/tests/credential/contract/capacity.rs @@ -321,7 +321,7 @@ async fn queue_enabled_without_affinity_still_uses_a_free_account() { #[derive(Debug)] struct Scheduler { - explicit: bool, + decision: AccountScheduleDecision, } impl RequestPolicyPlan for Scheduler { @@ -336,11 +336,7 @@ impl RequestPolicyPlan for Scheduler { &self, _: AccountScheduleInput, ) -> BoxFuture<'static, Result> { - let decision = if self.explicit { - AccountScheduleDecision::Pick(ProviderAccountId::new("acct_second").unwrap()) - } else { - AccountScheduleDecision::Delegate - }; + let decision = self.decision.clone(); Box::pin(async move { Ok(decision) }) } } @@ -373,7 +369,15 @@ async fn affinity_wait_respects_explicit_policy_choices_but_not_delegated_fallba timeout: Duration::from_secs(2), }), Some(RequestPolicyContext::new( - Arc::new(Scheduler { explicit }), + Arc::new(Scheduler { + decision: if explicit { + AccountScheduleDecision::Pick( + ProviderAccountId::new("acct_second").unwrap(), + ) + } else { + AccountScheduleDecision::Delegate + }, + }), ExtensionSetReference::new( ExtensionSetId::new("capacity-test".into()).unwrap(), Arc::new(ExtensionLease), @@ -402,3 +406,214 @@ async fn affinity_wait_respects_explicit_policy_choices_but_not_delegated_fallba } } } + +fn filtered_policy(ids: &[&str]) -> RequestPolicyContext { + RequestPolicyContext::new( + Arc::new(Scheduler { + decision: AccountScheduleDecision::DelegateFiltered( + ids.iter() + .map(|id| ProviderAccountId::new(*id).unwrap()) + .collect(), + ), + }), + ExtensionSetReference::new( + ExtensionSetId::new("filter-test".into()).unwrap(), + Arc::new(ExtensionLease), + ), + ModelRequestId::new("req_capacity_contract").unwrap(), + ClientApiKeyId::new("key_codex_contract").unwrap(), + vec![], + ) +} + +#[tokio::test] +async fn filtering_affinity_covers_native_reselection_and_lease_contention() { + for blocker in ["none", "concurrency", "lease"] { + let store = Arc::new(MemoryAccountStore::default()); + for id in ["acct_first", "acct_second", "acct_fallback"] { + create_account(&store, id, id); + } + store.set_scheduling("acct_second", None, AccountWeight::new(100).unwrap()); + store.set_scheduling("acct_fallback", None, AccountWeight::new(10).unwrap()); + let first = ProviderAccountId::new("acct_first").unwrap(); + let second = ProviderAccountId::new("acct_second").unwrap(); + let leases = Arc::new(TestLeaseCoordinator::default()); + if blocker == "concurrency" { + leases + .signals + .lock() + .unwrap() + .insert(first.clone(), busy_signal(2)); + } else if blocker == "lease" { + leases.busy_accounts.lock().unwrap().insert(second.clone()); + } + let affinity = Arc::new(MemorySessionAffinity::default()); + let key = bind_first(&affinity).await; + let selector = selector_with_affinity(&store, leases.clone(), affinity); + let request_attempt = capacity_attempt( + account_policy().with_queue(ConcurrencyQueuePolicy { + max_waiting: 1, + timeout: Duration::from_secs(2), + }), + Some(filtered_policy(&["acct_first"])), + ); + let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let selected = selector + .select(&SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &url, + attempt: &request_attempt, + session_affinity_key: Some(&key), + }) + .await + .unwrap(); + assert_eq!( + selected.account_id().as_str(), + if blocker == "lease" { + "acct_fallback" + } else { + "acct_second" + }, + "{blocker}" + ); + assert!( + leases + .requests + .lock() + .unwrap() + .iter() + .all(|request| request.account_id() != &first) + ); + assert!( + block_on(store.get_account(&first)) + .unwrap() + .unwrap() + .enabled() + ); + } +} + +#[tokio::test] +async fn filtered_waiting_never_leases_excluded_affinity_and_next_request_can_recover() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_first", "test-first"); + create_account(&store, "acct_second", "test-second"); + store.set_scheduling("acct_first", None, AccountWeight::new(100).unwrap()); + store.set_scheduling("acct_second", None, AccountWeight::new(10).unwrap()); + let first = ProviderAccountId::new("acct_first").unwrap(); + let second = ProviderAccountId::new("acct_second").unwrap(); + let leases = Arc::new(TestLeaseCoordinator::default()); + leases + .signals + .lock() + .unwrap() + .insert(second.clone(), busy_signal(2)); + let affinity = Arc::new(MemorySessionAffinity::default()); + let key = bind_first(&affinity).await; + let selector = selector_with_affinity(&store, leases.clone(), affinity); + let policy = account_policy().with_queue(ConcurrencyQueuePolicy { + max_waiting: 1, + timeout: Duration::from_secs(2), + }); + let filtered = capacity_attempt(policy, Some(filtered_policy(&["acct_first"]))); + let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let request = SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &url, + attempt: &filtered, + session_affinity_key: Some(&key), + }; + let mut pending = Box::pin(selector.select(&request)); + assert!(pending.as_mut().now_or_never().is_none()); + assert!(leases.requests.lock().unwrap().is_empty()); + leases.signals.lock().unwrap().clear(); + let selected = pending.await.unwrap(); + assert_eq!(selected.account_id(), &second); + drop(selected); + let recovered = capacity_attempt(policy, Some(filtered_policy(&[]))); + let selected = selector + .select(&SelectCodexCredential { + attempt: &recovered, + ..request + }) + .await + .unwrap(); + // 当前会话保留已迁移的亲和,新会话可以重新选到恢复后的高权重账号 + assert_eq!(selected.account_id(), &second); + drop(selected); + let selected = selector + .select(&SelectCodexCredential { + attempt: &recovered, + session_affinity_key: None, + ..request + }) + .await + .unwrap(); + assert_eq!(selected.account_id(), &first); +} + +#[tokio::test] +async fn all_filtered_is_terminal_before_queue_or_lease_even_for_native_pins() { + for pinned in [false, true] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_first", "test-first"); + create_account(&store, "acct_second", "test-second"); + let first = ProviderAccountId::new("acct_first").unwrap(); + let leases = Arc::new(TestLeaseCoordinator::default()); + leases + .signals + .lock() + .unwrap() + .insert(first.clone(), busy_signal(2)); + leases.signals.lock().unwrap().insert( + ProviderAccountId::new("acct_second").unwrap(), + busy_signal(2), + ); + let selector = selector(&store, leases.clone()); + let policy = filtered_policy(if pinned { + &["acct_first"] + } else { + &["acct_first", "acct_second"] + }); + let request_attempt = AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new("req_capacity_contract").unwrap(), + ClientApiKeyId::new("key_codex_contract").unwrap(), + ) + .with_request_policy(Some(policy)), + NonZeroU32::MIN, + SystemTime::now() + Duration::from_secs(5), + account_policy().with_queue(ConcurrencyQueuePolicy { + max_waiting: 1, + timeout: Duration::from_secs(2), + }), + AccountAttemptContext::new(BTreeSet::new(), None, None) + .with_account_scope(contract_account_scope()), + pinned.then(|| { + ContinuationBinding::Pinned(NativeContinuationPin::new( + PreviousResponseId::new("client_response"), + PreviousResponseId::new("upstream_response"), + ClientApiKeyId::new("key_codex_contract").unwrap(), + ProviderKind::new("openai").unwrap(), + first, + )) + }), + CancellationToken::new(), + ); + let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); + let result = selector + .select(&SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &url, + attempt: &request_attempt, + session_affinity_key: None, + }) + .now_or_never() + .expect("filter rejection must not wait"); + assert!( + matches!(result, Err(CredentialSelectionError::PolicyFiltered)), + "{result:?}" + ); + assert!(leases.requests.lock().unwrap().is_empty()); + } +} diff --git a/backend/crates/providers/openai/tests/credential/contract/mod.rs b/backend/crates/providers/openai/tests/credential/contract/mod.rs index 27b1142dc..38e092d0d 100644 --- a/backend/crates/providers/openai/tests/credential/contract/mod.rs +++ b/backend/crates/providers/openai/tests/credential/contract/mod.rs @@ -144,59 +144,46 @@ async fn smart_reselection_respects_native_and_required_account_boundaries() { let selector = selector_with_affinity(&store, Arc::new(TestLeaseCoordinator::default()), affinity); let request_url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap(); - for enabled in [false, true] { - for binding in ["soft", "required", "native"] { - let continuation = (binding == "native").then(|| { - ContinuationBinding::Pinned(NativeContinuationPin::new( - PreviousResponseId::new("client_response"), - PreviousResponseId::new("upstream_response"), - ClientApiKeyId::new("key_codex_contract").unwrap(), - provider.clone(), - original_id.clone(), - )) - }); - let attempt = AttemptContext::new( - RequestAttemptContext::new( - ModelRequestId::new("req_smart_scope").unwrap(), - ClientApiKeyId::new("key_codex_contract").unwrap(), - ), - NonZeroU32::new(1).unwrap(), - SystemTime::now() + Duration::from_secs(30), - account_policy().with_smart_scheduling( - gateway_core::account::SmartSchedulingConfig::new( - [1.0, 0.8, 1.0, 0.5, 1.0, 1.0], - enabled, - ) + for binding in ["soft", "required", "native"] { + let continuation = (binding == "native").then(|| { + ContinuationBinding::Pinned(NativeContinuationPin::new( + PreviousResponseId::new("client_response"), + PreviousResponseId::new("upstream_response"), + ClientApiKeyId::new("key_codex_contract").unwrap(), + provider.clone(), + original_id.clone(), + )) + }); + let attempt = AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new("req_smart_scope").unwrap(), + ClientApiKeyId::new("key_codex_contract").unwrap(), + ), + NonZeroU32::new(1).unwrap(), + SystemTime::now() + Duration::from_secs(30), + account_policy().with_smart_scheduling( + gateway_core::account::SmartSchedulingConfig::new([1.0, 0.8, 1.0, 0.5, 1.0, 1.0]) .unwrap(), - ), - AccountAttemptContext::new( - BTreeSet::new(), - (binding == "required").then(|| original_id.clone()), - None, - ) - .with_account_scope(contract_account_scope()), - continuation, - CancellationToken::new(), - ); - let selected = selector - .select(&SelectCodexCredential { - upstream_model: "gpt-5.4", - request_url: &request_url, - attempt: &attempt, - session_affinity_key: Some(&key), - }) - .await - .unwrap(); - assert_eq!( - selected.account_id().as_str(), - if enabled && binding == "soft" { - "acct_primary" - } else { - "acct_original" - }, - "{binding}, switchback={enabled}" - ); - } + ), + AccountAttemptContext::new( + BTreeSet::new(), + (binding == "required").then(|| original_id.clone()), + None, + ) + .with_account_scope(contract_account_scope()), + continuation, + CancellationToken::new(), + ); + let selected = selector + .select(&SelectCodexCredential { + upstream_model: "gpt-5.4", + request_url: &request_url, + attempt: &attempt, + session_affinity_key: Some(&key), + }) + .await + .unwrap(); + assert_eq!(selected.account_id().as_str(), "acct_original", "{binding}"); } } @@ -1930,11 +1917,8 @@ fn queued_attempt(timeout: Duration) -> AttemptContext { timeout, }) .with_smart_scheduling( - gateway_core::account::SmartSchedulingConfig::new( - [1.0, 0.8, 1.0, 0.5, 1.0, 1.0], - false, - ) - .unwrap(), + gateway_core::account::SmartSchedulingConfig::new([1.0, 0.8, 1.0, 0.5, 1.0, 1.0]) + .unwrap(), ), AccountAttemptContext::new(BTreeSet::new(), None, None) .with_account_scope(contract_account_scope()), @@ -1983,10 +1967,14 @@ fn smart_queue_weight_changes_the_selected_wait_queue() { timeout: Duration::from_secs(2), }) .with_smart_scheduling( - gateway_core::account::SmartSchedulingConfig::new( - [0.0, 0.0, 1.0, 0.0, 0.0, queue_weight], - false, - ) + gateway_core::account::SmartSchedulingConfig::new([ + 0.0, + 0.0, + 1.0, + 0.0, + 0.0, + queue_weight, + ]) .unwrap(), ), AccountAttemptContext::new(BTreeSet::new(), required, None) @@ -2051,10 +2039,9 @@ fn smart_queue_keeps_native_and_required_account_pins_when_another_account_is_fr timeout: Duration::from_secs(2), }) .with_smart_scheduling( - gateway_core::account::SmartSchedulingConfig::new( - [1.0, 0.8, 1.0, 0.5, 10.0, 10.0], - true, - ) + gateway_core::account::SmartSchedulingConfig::new([ + 1.0, 0.8, 1.0, 0.5, 10.0, 10.0, + ]) .unwrap(), ), AccountAttemptContext::new(BTreeSet::new(), (!native).then(|| original.clone()), None) diff --git a/backend/crates/providers/openai/tests/credential/quota/scheduling.rs b/backend/crates/providers/openai/tests/credential/quota/scheduling.rs index 80714b0a6..0a0e45e8d 100644 --- a/backend/crates/providers/openai/tests/credential/quota/scheduling.rs +++ b/backend/crates/providers/openai/tests/credential/quota/scheduling.rs @@ -100,6 +100,7 @@ async fn old_weekly_quota_should_outweigh_one_in_flight_title_request() { ), now, excluded_accounts: BTreeSet::new(), + busy_accounts: Default::default(), preferred_account: None, preferred_account_overrides_weight: true, round_robin_cursor: cursor, diff --git a/backend/crates/providers/openai/tests/provider/contract/session_binding.rs b/backend/crates/providers/openai/tests/provider/contract/session_binding.rs index dcabf0a7a..67c1142ff 100644 --- a/backend/crates/providers/openai/tests/provider/contract/session_binding.rs +++ b/backend/crates/providers/openai/tests/provider/contract/session_binding.rs @@ -647,7 +647,7 @@ async fn child_queue_timeout_does_not_rebind_or_allow_provider_fallback() { #[derive(Debug)] struct SessionScheduler { - explicit: bool, + decision: gateway_core::engine::policy::AccountScheduleDecision, } impl gateway_core::engine::policy::RequestPolicyPlan for SessionScheduler { fn route_model( @@ -672,18 +672,13 @@ impl gateway_core::engine::policy::RequestPolicyPlan for SessionScheduler { gateway_core::engine::policy::RequestPolicyFault, >, > { - use gateway_core::engine::policy::AccountScheduleDecision; assert!( input .candidates() .iter() .any(|candidate| candidate.account_id().as_str() == "acct_subagent_b") ); - let decision = if self.explicit { - AccountScheduleDecision::Pick(ProviderAccountId::new("acct_subagent_b").unwrap()) - } else { - AccountScheduleDecision::Delegate - }; + let decision = self.decision.clone(); Box::pin(async move { Ok(decision) }) } } @@ -695,10 +690,16 @@ impl gateway_core::routing::extensions::ExtensionSetLease for SessionExtensionLe } #[tokio::test] -async fn child_binding_only_constrains_builtin_scheduling_and_preserves_plugin_choices() { +async fn child_binding_respects_filtered_delegation_and_explicit_plugin_choices() { + use gateway_core::account::AccountAffinity; use gateway_core::engine::policy::RequestPolicyContext; use gateway_core::routing::extensions::{ExtensionSetId, ExtensionSetReference}; - for explicit in [false, true] { + for (choice, mode) in [ + ("delegate", AccountAffinity::Strict), + ("pick", AccountAffinity::Strict), + ("filter", AccountAffinity::Strict), + ("filter", AccountAffinity::Preferred), + ] { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_subagent_a").await; let affinity = Arc::new(MemorySessionAffinity::default()); @@ -729,7 +730,19 @@ async fn child_binding_only_constrains_builtin_scheduling_and_preserves_plugin_c let id = ModelRequestId::new("req_plugin_child").unwrap(); let key = ClientApiKeyId::new("key_openai_contract").unwrap(); let policy = RequestPolicyContext::new( - Arc::new(SessionScheduler { explicit }), + Arc::new(SessionScheduler { + decision: match choice { + "pick" => gateway_core::engine::policy::AccountScheduleDecision::Pick( + ProviderAccountId::new("acct_subagent_b").unwrap(), + ), + "filter" => { + gateway_core::engine::policy::AccountScheduleDecision::DelegateFiltered( + BTreeSet::from([ProviderAccountId::new("acct_subagent_a").unwrap()]), + ) + } + _ => gateway_core::engine::policy::AccountScheduleDecision::Delegate, + }, + }), ExtensionSetReference::new( ExtensionSetId::new("session-choice".into()).unwrap(), Arc::new(SessionExtensionLease), @@ -742,7 +755,7 @@ async fn child_binding_only_constrains_builtin_scheduling_and_preserves_plugin_c RequestAttemptContext::new(id, key).with_request_policy(Some(policy)), NonZeroU32::new(1).unwrap(), SystemTime::now() + Duration::from_secs(5), - account_policy(), + account_policy().with_openai_account_affinity(mode), AccountAttemptContext::new(BTreeSet::new(), None, None) .with_account_scope(contract_account_scope()), None, @@ -755,7 +768,18 @@ async fn child_binding_only_constrains_builtin_scheduling_and_preserves_plugin_c ), attempt, )); - if explicit { + if choice == "filter" && mode == AccountAffinity::Strict { + let error = timeout(Duration::from_secs(2), pending) + .await + .unwrap() + .err() + .expect("excluded child owner must be terminal"); + assert_eq!(error.kind(), ProviderErrorKind::RequestPolicyDenied); + assert_eq!( + error.send_state(), + gateway_core::upstream::UpstreamSendState::NotSent + ); + } else if choice == "pick" || choice == "filter" { let stream = timeout(Duration::from_secs(2), pending) .await .unwrap() @@ -1062,7 +1086,7 @@ async fn relaxed_and_preferred_session_requests_apply_distinct_scheduling_polici ); let policy = account_policy() .with_openai_account_affinity(mode) - .with_smart_scheduling(SmartSchedulingConfig::new([1.0; 6], true).unwrap()) + .with_smart_scheduling(SmartSchedulingConfig::new([1.0; 6]).unwrap()) .with_queue(ConcurrencyQueuePolicy { max_waiting, timeout: Duration::from_secs(1), diff --git a/backend/crates/providers/xai/src/credential/selector.rs b/backend/crates/providers/xai/src/credential/selector.rs index afabf0d5e..45d3259e8 100644 --- a/backend/crates/providers/xai/src/credential/selector.rs +++ b/backend/crates/providers/xai/src/credential/selector.rs @@ -239,6 +239,7 @@ impl GrokAccountSessionSelector { policy: request.account_selection_policy(), now: SystemTime::now(), excluded_accounts: request.excluded_accounts().clone(), + busy_accounts: Default::default(), preferred_account: request.required_account().cloned().or(affinity_account), preferred_account_overrides_weight: false, round_robin_cursor: scheduling.round_robin_cursor(), @@ -246,19 +247,15 @@ impl GrokAccountSessionSelector { account_scope: (!diagnostic).then(|| Arc::clone(request.account_scope())), reserved_concurrency: 0, }; - let wait_candidates = AccountSelector.wait_candidates(&candidates, &context); - let capacity_context = context.clone(); for candidate in &candidates { if !waiting.can_try(candidate.account.id()) { - context - .excluded_accounts - .insert(candidate.account.id().clone()); + context.busy_accounts.insert(candidate.account.id().clone()); } } let mut capacity_denied = false; let mut retry_after = None; loop { - let capacity = AccountSelector.capacity_snapshot(&candidates, &capacity_context); + context.excluded_accounts = request.excluded_accounts().clone(); let selection = match request.request_policy() { Some(policy) => { policy @@ -267,7 +264,7 @@ impl GrokAccountSessionSelector { &self.provider_kind, Some(request.upstream_model().as_str()), &candidates, - &context, + &mut context, ) .await } @@ -276,6 +273,9 @@ impl GrokAccountSessionSelector { let Some(selection) = (match selection { Ok(selection) => selection, Err(AccountPolicyError::StaleCandidate) => continue 'refresh, + Err(AccountPolicyError::Filtered) => { + return Err(GrokSessionSelectorError::PolicyFiltered); + } Err(AccountPolicyError::Rejected) => { return Err(GrokSessionSelectorError::PolicyRejected); } @@ -285,6 +285,7 @@ impl GrokAccountSessionSelector { }) else { break; }; + let capacity = AccountSelector.capacity_snapshot(&candidates, &context); let selected = selection.candidate(); let selected_id = selected.account.id().clone(); let selected_revision = selected.account.revision(); @@ -318,7 +319,7 @@ impl GrokAccountSessionSelector { } => { capacity_denied = true; retry_after = minimum_retry_after(retry_after, candidate_retry); - candidates.retain(|candidate| candidate.account.id() != &selected_id); + context.busy_accounts.insert(selected_id); continue; } }; @@ -387,6 +388,7 @@ impl GrokAccountSessionSelector { }); } + let wait_candidates = AccountSelector.wait_candidates(&candidates, &context); if !diagnostic && queue_policy.max_waiting > 0 && !wait_candidates.is_empty() { AccountSelector .wait_for_capacity(&mut waiting, &wait_candidates, &candidates, &context) diff --git a/backend/crates/providers/xai/src/provider/failure.rs b/backend/crates/providers/xai/src/provider/failure.rs index 4c14b3aaa..dc29c4c7a 100644 --- a/backend/crates/providers/xai/src/provider/failure.rs +++ b/backend/crates/providers/xai/src/provider/failure.rs @@ -450,6 +450,18 @@ fn selection_failure(error: &GrokSessionSelectorError) -> ProviderError { UpstreamSendState::NotSent, ); } + GrokSessionSelectorError::PolicyFiltered => { + return provider_error( + ProviderErrorKind::RequestPolicyDenied, + UpstreamSendState::NotSent, + ) + .with_diagnostic( + gateway_core::error::ProviderDiagnostic::new( + "account scheduling policy left no selectable account", + ) + .with_classification("account_selection", "account_policy_filtered"), + ); + } GrokSessionSelectorError::PolicyUnavailable => { return provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent); } diff --git a/backend/crates/providers/xai/src/transport/session.rs b/backend/crates/providers/xai/src/transport/session.rs index c4416a9c4..ea30324c4 100644 --- a/backend/crates/providers/xai/src/transport/session.rs +++ b/backend/crates/providers/xai/src/transport/session.rs @@ -518,6 +518,9 @@ pub enum GrokSessionSelectorError { /// 插件调度策略明确拒绝本次请求 #[error("account scheduling policy rejected the request")] PolicyRejected, + /// 插件排除了本轮全部候选,终止本次调度 + #[error("account scheduling policy left no selectable account")] + PolicyFiltered, /// 插件调度策略调用失败且配置为拒绝 #[error("account scheduling policy is unavailable")] PolicyUnavailable, diff --git a/backend/crates/providers/xai/tests/credential/selector.rs b/backend/crates/providers/xai/tests/credential/selector.rs index c8dc477f7..748a5ee99 100644 --- a/backend/crates/providers/xai/tests/credential/selector.rs +++ b/backend/crates/providers/xai/tests/credential/selector.rs @@ -1092,11 +1092,8 @@ fn queue_request(fixture: &SelectorFixture, timeout: Duration) -> GrokSessionSel timeout, }) .with_smart_scheduling( - gateway_core::account::SmartSchedulingConfig::new( - [1.0, 0.8, 1.0, 0.5, 1.0, 1.0], - false, - ) - .unwrap(), + gateway_core::account::SmartSchedulingConfig::new([1.0, 0.8, 1.0, 0.5, 1.0, 1.0]) + .unwrap(), ), request.deadline(), request.account_scope().clone(), @@ -1135,10 +1132,14 @@ async fn smart_queue_weight_changes_the_selected_wait_queue_and_respects_require timeout: Duration::from_secs(2), }) .with_smart_scheduling( - gateway_core::account::SmartSchedulingConfig::new( - [0.0, 0.0, 1.0, 0.0, 0.0, queue_weight], - false, - ) + gateway_core::account::SmartSchedulingConfig::new([ + 0.0, + 0.0, + 1.0, + 0.0, + 0.0, + queue_weight, + ]) .unwrap(), ), request.deadline(), diff --git a/backend/crates/providers/xai/tests/provider/contract/mod.rs b/backend/crates/providers/xai/tests/provider/contract/mod.rs index f67eda256..c62eeac59 100644 --- a/backend/crates/providers/xai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/xai/tests/provider/contract/mod.rs @@ -1,6 +1,7 @@ //! 验证 xAI 原生执行的协议转换、账号选择与交付状态隔离 mod replay_tool_names; +mod scheduling; mod timing; use std::collections::{BTreeMap, BTreeSet, VecDeque}; diff --git a/backend/crates/providers/xai/tests/provider/contract/scheduling.rs b/backend/crates/providers/xai/tests/provider/contract/scheduling.rs new file mode 100644 index 000000000..8700a7fd9 --- /dev/null +++ b/backend/crates/providers/xai/tests/provider/contract/scheduling.rs @@ -0,0 +1,272 @@ +//! 验证 xAI 真实选择器将插件过滤传递到租约重试、等待与原生续写 + +use super::*; +use futures::FutureExt; +use gateway_core::account::ProviderAccountId; +use gateway_core::concurrency::ConcurrencyQueuePolicy; +use gateway_core::engine::policy::{ + AccountScheduleDecision, AccountScheduleInput, ModelRouteDecision, ModelRouteInput, + RequestPolicyContext, RequestPolicyFault, RequestPolicyPlan, +}; + +#[derive(Debug)] +struct Filter { + all: bool, + seen: Arc>>, +} + +impl RequestPolicyPlan for Filter { + fn route_model( + &self, + _: ModelRouteInput, + ) -> BoxFuture<'static, Result> { + Box::pin(async { Ok(ModelRouteDecision::Unhandled) }) + } + + fn schedule_account( + &self, + input: AccountScheduleInput, + ) -> BoxFuture<'static, Result> { + let excluded = input + .candidates() + .iter() + .filter(|candidate| self.all || candidate.account_id() == &account_id("filter-a")) + .map(|candidate| candidate.account_id().clone()) + .collect(); + self.seen.lock().unwrap().push(input); + Box::pin(async move { Ok(AccountScheduleDecision::DelegateFiltered(excluded)) }) + } +} + +#[derive(Default)] +struct Leases { + busy: Mutex, + deny_second: bool, + acquired: Mutex>, +} + +impl ProviderLeasePort for Leases { + fn load_state<'a>( + &'a self, + key: &'a ClientApiKeyId, + provider: &'a ProviderKind, + ids: &'a [ProviderAccountId], + pool: gateway_core::provider_ports::ProviderConcurrencyPool, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { + let state = DiagnosticLeasePort + .load_state(key, provider, ids, pool) + .await?; + let mut signals = state.signals().clone(); + if *self.busy.lock().unwrap() { + for signal in signals.values_mut() { + signal.in_flight = u32::MAX; + } + } + Ok(ProviderSchedulingState::new(signals, 0)) + }) + } + + fn try_acquire( + &self, + request: ProviderLeaseRequest, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { + let ProviderLeaseRequest::Scheduling(request) = request else { + panic!("scheduling lease") + }; + self.acquired + .lock() + .unwrap() + .push(request.account_id().clone()); + if self.deny_second && request.account_id() == &account_id("filter-b") { + Ok(ProviderLeaseAcquisition::Busy { + retry_after: Some(Duration::from_millis(10)), + }) + } else { + Ok(ProviderLeaseAcquisition::Acquired(Box::new(()))) + } + }) + } +} + +#[tokio::test] +async fn filtered_scheduling_covers_lease_retry_waiting_and_pinned_terminal_rejection() { + for mode in ["retry", "wait", "all", "pin"] { + let store = MemoryProviderAccountStore::shared(); + let repository = GrokCredentialRepository::new(store.clone()); + let mut accounts = BTreeMap::new(); + let kind = ProviderKind::new("xai").unwrap(); + for suffix in ["filter-a", "filter-b", "filter-c"] { + let input = create_input(suffix, suffix); + seed_input(&store, &input).await.unwrap(); + repository + .update_state(&UpdateGrokCredentialState { + account_id: input.account_id.clone(), + expected_revision: CredentialRevision::new(1).unwrap(), + credential_state: CredentialState::Ready, + error_reason: None, + error_message: None, + observed_at: chrono::Utc::now(), + }) + .await + .unwrap(); + accounts.insert( + input.account_id, + RuntimeAccount::new(kind.clone(), BTreeSet::new()), + ); + } + let leases = Arc::new(Leases { + busy: Mutex::new(matches!(mode, "wait" | "all")), + deny_second: mode == "retry", + ..Default::default() + }); + let cache = MemoryGrokCatalogCache::shared(); + let selector = GrokAccountSessionSelector::new( + kind.clone(), + repository.clone(), + cache.clone(), + Arc::new(crate::support::grok_quota_service( + repository.clone(), + Arc::new(UnavailableBillingTransport), + )), + leases.clone(), + Arc::new(MemoryCooldownPort::default()), + Arc::new(AccountFeedbackStats::default()), + ); + let transport = StubInferenceTransport::success(); + let provider = Arc::new( + GrokBuildProvider::new( + Arc::new(selector), + transport.clone(), + Arc::new(crate::support::grok_catalog_service( + repository, + Arc::new(StaticCatalogTransport), + cache, + )), + StubRecovery::new(GrokCredentialRecoveryOutcome::Unavailable), + Arc::new(AccountFeedbackStats::default()), + crate::support::xai_wire_profile(), + ) + .unwrap(), + ); + let id = ModelRequestId::new("req_xai_filter").unwrap(); + let key = ClientApiKeyId::new("key_xai_contract").unwrap(); + let seen = Arc::new(Mutex::new(Vec::new())); + let policy = RequestPolicyContext::new( + Arc::new(Filter { + all: mode == "all", + seen: seen.clone(), + }), + ExtensionSetReference::new( + ExtensionSetId::new("filter-test".into()).unwrap(), + Arc::new(TestExtensionLease), + ), + id.clone(), + key.clone(), + vec![], + ); + let continuation = (mode == "pin").then(|| { + ContinuationBinding::Pinned(NativeContinuationPin::new( + PreviousResponseId::new("client_response"), + PreviousResponseId::new("upstream_response"), + key.clone(), + kind, + account_id("filter-a"), + )) + }); + let scope = Arc::new(FrozenAccountScope::new( + Arc::new(RuntimeAccountDirectory::new(accounts)), + ClientRoutingScope::all_accounts(), + )); + let operation = operation(); + let kind = ProviderKind::new("xai").unwrap(); + let snapshot = RuntimeSnapshot::new( + ConfigRevision::new(1).unwrap(), + gateway_core::settings::SettingsValues::new(2, "smart", Default::default(), None, None), + vec![kind.clone()], + vec![ProviderModel::new( + kind, + UpstreamModelId::new(MODEL).unwrap(), + ModelCapabilities::new(BTreeSet::from([OperationKind::Generate]), Some(131_072)) + .with_feature(Feature::NativeContinuation, SupportLevel::Native), + )], + vec![], + ) + .unwrap(); + let plan = snapshot + .plan( + &PublicModelId::new(MODEL).unwrap(), + &operation, + scope.clone(), + &RoutingContext::default(), + ) + .unwrap(); + let request = ProviderRequest::new(operation, plan.candidates()[0].clone()); + let attempt = AttemptContext::new( + gateway_core::engine::RequestAttemptContext::new(id, key) + .with_request_policy(Some(policy)), + NonZeroU32::MIN, + SystemTime::now() + Duration::from_secs(5), + selection_policy().with_queue(ConcurrencyQueuePolicy { + max_waiting: 1, + timeout: Duration::from_secs(2), + }), + AccountAttemptContext::new(BTreeSet::new(), None, None).with_account_scope(scope), + continuation, + CancellationToken::new(), + ); + let mut pending = Box::pin(provider.execute(request, attempt)); + if mode == "wait" { + assert!(pending.as_mut().now_or_never().is_none()); + assert!(leases.acquired.lock().unwrap().is_empty()); + *leases.busy.lock().unwrap() = false; + } + let result = pending.await; + if matches!(mode, "all" | "pin") { + let error = result.err().expect("terminal filter rejection"); + assert_eq!( + error.kind(), + ProviderErrorKind::RequestPolicyDenied, + "{mode}" + ); + assert_eq!(error.send_state(), UpstreamSendState::NotSent); + assert!(leases.acquired.lock().unwrap().is_empty()); + assert_eq!(transport.calls.load(Ordering::SeqCst), 0); + if mode == "pin" { + assert_eq!(seen.lock().unwrap()[0].candidates().len(), 1); + } + } else { + let stream = result.expect("allowed account"); + assert_ne!( + stream.metadata().provider_account_id(), + &account_id("filter-a") + ); + assert!( + leases + .acquired + .lock() + .unwrap() + .iter() + .all(|id| id != &account_id("filter-a")) + ); + if mode == "retry" { + assert_eq!( + stream.metadata().provider_account_id(), + &account_id("filter-c") + ); + assert!( + seen.lock() + .unwrap() + .iter() + .flat_map(|input| input.candidates()) + .any( + |candidate| candidate.account_id() == &account_id("filter-b") + && !candidate.available() + ) + ); + } + } + assert!(store.account(&account_id("filter-a")).unwrap().enabled()); + } +} diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index 494e06e5f..4c29bba57 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -32,3 +32,4 @@ b49832e205405de7ef078f3b65c0c19f2c030b8984d42411598738b6b6c146bf 0031_proxy_and 8eade5bd4f5c8847366ca951552775605ecf4171a7f0e87fa18574bceb8a3ff1 0032_request_history_retention.sql 2352459469bdd5102e202d335a0258e3c4c615fb88e69341ede8b1c0d655376b 0033_table_naming.sql b85deecd04c6297c0c8d9577794af51aaf35f9e6a99b2c98029c6211b1bcb0eb 0034_account_group_image_generation.sql +0799b008e7241753aaa0637763abfe5eacc4b54e8c3ba67e93954a23dda32dac 0035_remove_smart_switchback.sql diff --git a/backend/migrations/0035_remove_smart_switchback.sql b/backend/migrations/0035_remove_smart_switchback.sql new file mode 100644 index 000000000..7527788ea --- /dev/null +++ b/backend/migrations/0035_remove_smart_switchback.sql @@ -0,0 +1,8 @@ +-- 删除自动回切配置,保留既有智能调度系数 +UPDATE runtime_settings +SET smart_scheduling_json = smart_scheduling_json - 'preferHigherWeight' +WHERE smart_scheduling_json ? 'preferHigherWeight'; + +ALTER TABLE runtime_settings + ALTER COLUMN smart_scheduling_json SET DEFAULT + '{"loadWeight":1.0,"quotaWeight":0.8,"healthWeight":1.0,"latencyWeight":0.5,"resetWeight":0.0,"queueWeight":0.0}'::jsonb; diff --git a/docs/api.md b/docs/api.md index 6b30aed5c..9d53b4e49 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1560,8 +1560,7 @@ Guardian 以 `subagent_kind` 或 `client_metadata.x-openai-subagent` 值 `guardi "healthWeight": 1.0, "latencyWeight": 0.5, "resetWeight": 0.0, - "queueWeight": 0.0, - "preferHigherWeight": false + "queueWeight": 0.0 } ``` @@ -1575,11 +1574,8 @@ Guardian 以 `subagent_kind` 或 `client_metadata.x-openai-subagent` 值 `guardi 设为 `0` 时沿用最短队列规则;启用后结合其余五项评分选择队列。队列人数限于当前进程,只有开启账号排队才会生效, 不影响已入队请求的位置、队内 FIFO、容量上限或等待超时。两项默认均为 `0` -`preferHigherWeight` 默认关闭。开启后,更高权重账号恢复可用时,后续允许重新选号的请求优先回切; -同权重且可用的会话亲和继续保留。OpenAI 优先模式沿用可用的会话主账号,不因权重回切而分流。 -没有可用亲和时,在最高可用权重层内按配置评分。 -原生续写账号绑定仍是硬约束,不因回切而主动换号或触发历史重放。配置随运行设置原子保存和发布, -新请求使用新值,已开始请求及其重试沿用原快照,无需重启 +没有可用亲和时,在最高可用权重层内按配置评分。原生续写账号绑定仍是硬约束。 +配置随运行设置原子保存和发布,新请求使用新值,已开始请求及其重试沿用原快照,无需重启 ### 模型定价 diff --git a/docs/architecture.md b/docs/architecture.md index b5a3040f0..117a9aab8 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -664,9 +664,8 @@ Core 定义、校验并统一拥有评分规则,内置选号与诊断复用同 - 排队系数只在选择等待队列时结合其他评分和进程内实时等待人数,不影响立即选号分数与容差;系数为零时选最短队列。 `ConcurrencyWaitQueue` 在同一锁内读取队长并入队,保持已有位置、同级 FIFO、取消回收、容量和共享等待预算;Provider 传入候选须已限定续写范围 - 健康分按已发送尝试的结果计算账号级平滑失败率并随时间衰减;明确容量拒绝的平滑权重为 0.4,其他可计分失败和成功为 0.2 -- 高权重回切只影响软亲和:先检查候选资格,再允许更高权重层覆盖亲和,同层亲和保留。 - OpenAI 优先模式沿用可用主账号;原生续写账号范围是硬约束。回切只发生在重新选号时,不启动后台任务或主动历史重放 -- 插件显式选号使用候选校验和实际亲和结果,只有委托内置选号才应用智能调度配置 +- 插件显式选号使用候选校验和实际亲和结果;过滤后委托沿用内置评分、权重和亲和,过滤结果同时约束重选与等待。 + 临时容量阻塞与策略排除分开,全部过滤直接拒绝请求,后续评估不继承上次过滤结果 字段、默认值与可配置范围见[智能调度设置](api.md#智能调度) diff --git a/frontend/src/api/modules/settings/index.ts b/frontend/src/api/modules/settings/index.ts index 1e2240cda..01fa75787 100644 --- a/frontend/src/api/modules/settings/index.ts +++ b/frontend/src/api/modules/settings/index.ts @@ -26,7 +26,6 @@ export interface SmartSchedulingConfig { latencyWeight: number resetWeight: number queueWeight: number - preferHigherWeight: boolean } export interface RuntimeSettings { diff --git a/frontend/src/views/settings/components/SmartSchedulingModal.vue b/frontend/src/views/settings/components/SmartSchedulingModal.vue index 22afd613e..39da419f4 100644 --- a/frontend/src/views/settings/components/SmartSchedulingModal.vue +++ b/frontend/src/views/settings/components/SmartSchedulingModal.vue @@ -1,7 +1,7 @@