diff --git a/apps/aether-gateway/src/lib.rs b/apps/aether-gateway/src/lib.rs index 20c2ba715..606ac990b 100644 --- a/apps/aether-gateway/src/lib.rs +++ b/apps/aether-gateway/src/lib.rs @@ -90,6 +90,7 @@ mod upstream_admission; mod usage; mod video_tasks; mod wallet_runtime; +mod xai_profile; pub use self::ai_serving::api::{codex_client_originator, codex_client_user_agent}; pub(crate) use self::ai_serving::api::{ diff --git a/apps/aether-gateway/src/main.rs b/apps/aether-gateway/src/main.rs index c130fb53b..b90319e34 100644 --- a/apps/aether-gateway/src/main.rs +++ b/apps/aether-gateway/src/main.rs @@ -2527,6 +2527,20 @@ async fn run() -> Result<(), Box> { ); } } + match state.prewarm_xai_client_profile().await { + Ok(version) => { + info!( + xai_client_version = %version, + "prewarmed Grok CLI client profile" + ); + } + Err(err) => { + warn!( + error = %err, + "failed to refresh Grok CLI client profile; built-in or cached profile remains active" + ); + } + } match prewarm_direct_h2c_sender_cache_from_env_for_startup().await { Ok(Some(report)) => { if report.failed_targets > 0 { diff --git a/apps/aether-gateway/src/state/core.rs b/apps/aether-gateway/src/state/core.rs index 52420dfaf..6ee302ad6 100644 --- a/apps/aether-gateway/src/state/core.rs +++ b/apps/aether-gateway/src/state/core.rs @@ -75,6 +75,7 @@ use crate::maintenance::spawn_stats_hourly_aggregation_worker; use crate::maintenance::spawn_usage_cleanup_worker; use crate::maintenance::spawn_usage_counter_flush_worker; use crate::maintenance::spawn_wallet_daily_usage_aggregation_worker; +use crate::xai_profile::spawn_worker as spawn_xai_client_profile_worker; const SYSTEM_CONFIG_CACHE_TTL: Duration = Duration::from_secs(30); // Requests may use a stale value after the fresh window until the entry reaches @@ -153,6 +154,10 @@ impl AppState { crate::codex_profile::prewarm(self.runtime_state()).await } + pub async fn prewarm_xai_client_profile(&self) -> Result { + crate::xai_profile::prewarm(self.runtime_state()).await + } + pub async fn prewarm_chat_pii_redaction_runtime_config(&self) -> Result { crate::privacy::read_chat_pii_redaction_runtime_config(self) .await @@ -2351,6 +2356,10 @@ impl AppState { crate::task_runtime::TASK_KEY_CODEX_CLIENT_PROFILE, Some(spawn_codex_client_profile_worker(background_state.clone())), ); + supervise_worker( + crate::task_runtime::TASK_KEY_XAI_CLIENT_PROFILE, + Some(spawn_xai_client_profile_worker(background_state.clone())), + ); supervise_worker( crate::task_runtime::TASK_KEY_VIDEO_TASK_POLLER, spawn_video_task_poller(background_state.clone()), diff --git a/apps/aether-gateway/src/task_runtime/mod.rs b/apps/aether-gateway/src/task_runtime/mod.rs index 226579c97..a607fd0cd 100644 --- a/apps/aether-gateway/src/task_runtime/mod.rs +++ b/apps/aether-gateway/src/task_runtime/mod.rs @@ -25,6 +25,7 @@ pub(crate) const TASK_KEY_USAGE_COUNTER_FLUSH: &str = "usage.counter.flush.worke pub(crate) const TASK_KEY_VIDEO_TASK_POLLER: &str = "video.task.poller"; pub(crate) const TASK_KEY_MODEL_FETCH_WORKER: &str = "model.fetch.worker"; pub(crate) const TASK_KEY_CODEX_CLIENT_PROFILE: &str = "maintenance.codex.client.profile"; +pub(crate) const TASK_KEY_XAI_CLIENT_PROFILE: &str = "maintenance.xai.client.profile"; pub(crate) const TASK_KEY_PROVIDER_QUOTA_RESET: &str = "provider.quota.reset.worker"; pub(crate) const TASK_KEY_ACCOUNT_SELF_CHECK: &str = "account.self_check.worker"; pub(crate) const TASK_KEY_POOL_SCORE_REBUILD: &str = "pool.score.rebuild.worker"; @@ -211,6 +212,14 @@ const TASK_DEFINITIONS: &[TaskDefinition] = &[ true, RETRY_ONCE, ), + TaskDefinition::new( + TASK_KEY_XAI_CLIENT_PROFILE, + TaskKind::Scheduled, + "interval", + true, + true, + RETRY_ONCE, + ), TaskDefinition::new( TASK_KEY_PROVIDER_QUOTA_RESET, TaskKind::Scheduled, diff --git a/apps/aether-gateway/src/xai_profile.rs b/apps/aether-gateway/src/xai_profile.rs new file mode 100644 index 000000000..dad4e9be7 --- /dev/null +++ b/apps/aether-gateway/src/xai_profile.rs @@ -0,0 +1,592 @@ +//! Grok CLI 客户端版本的运行时发布与官方版本刷新。 +//! +//! cli-chat-proxy.grok.com 会对低于最低版本的 `x-grok-client-version` 直接返回 426， +//! 因此网关定期读取官方发布渠道并原子替换传输层使用的版本号。 + +use std::collections::BTreeMap; +use std::future::Future; +use std::time::Duration; + +use aether_runtime_state::RuntimeState; +use futures_util::StreamExt as _; +use reqwest::{redirect::Policy, Client}; +use semver::Version; +use serde::{Deserialize, Serialize}; +use tracing::{info, warn}; + +use crate::provider_transport::{set_xai_client_version, xai_client_version}; +use crate::AppState; + +/// 官方安装脚本读取的 stable 渠道，响应体是纯文本版本号。 +const CLI_STABLE_CHANNEL_ENDPOINT: &str = "https://x.ai/cli/stable"; +/// stable 渠道不可达时（部分部署地区无法直连 x.ai）退回 npm 发布元数据。 +const CLI_NPM_RELEASE_ENDPOINT: &str = "https://registry.npmjs.org/@xai-official%2Fgrok/latest"; +const CLI_NPM_PACKAGE: &str = "@xai-official/grok"; +const PROFILE_CACHE_KEY: &str = "aether:xai:client-profile:v1"; +const PROFILE_CACHE_TTL: Duration = Duration::from_secs(30 * 24 * 60 * 60); +/// xAI 会在发布后很快抬高最低版本，刷新间隔比 Codex 更短。 +const PROFILE_REFRESH_INTERVAL: Duration = Duration::from_secs(3 * 60 * 60); +const RELEASE_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +const RELEASE_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); +const MAX_RELEASE_BYTES: usize = 256 * 1024; +const CLI_TARGETS: [&str; 6] = [ + "darwin-arm64", + "darwin-x64", + "linux-arm64", + "linux-x64", + "win32-arm64", + "win32-x64", +]; + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct NpmRelease { + name: String, + version: String, + optional_dependencies: BTreeMap, +} + +#[derive(Debug, Deserialize, Serialize)] +struct CachedProfile { + version: String, + verified_at_unix_secs: u64, +} + +#[derive(Debug, thiserror::Error)] +enum ProfileRefreshError { + #[error("Grok CLI release client initialization failed: {0}")] + Client(#[from] reqwest::Error), + #[error("Grok CLI release request returned HTTP {0}")] + HttpStatus(u16), + #[error("Grok CLI release response exceeded {MAX_RELEASE_BYTES} bytes")] + ResponseTooLarge, + #[error("Grok CLI release metadata is invalid")] + InvalidMetadata, + #[error("Grok CLI release version is older than the active profile")] + Rollback, + #[error("Grok CLI profile cache operation failed: {0}")] + Cache(String), + #[error("Grok CLI stable channel failed ({stable}); npm fallback failed ({npm})")] + AllSourcesFailed { stable: String, npm: String }, +} + +fn version_sequence(version: &str) -> Result { + let parsed = Version::parse(version).map_err(|_| ProfileRefreshError::InvalidMetadata)?; + if !parsed.pre.is_empty() + || !parsed.build.is_empty() + || parsed.major > 999 + || parsed.minor > 999 + || parsed.patch > 999 + { + return Err(ProfileRefreshError::InvalidMetadata); + } + Ok(1 + parsed.major * 1_000_000 + parsed.minor * 1_000 + parsed.patch) +} + +/// stable 渠道只返回一行版本号；任何多余内容都视为异常响应（例如被劫持的 HTML 页面）。 +fn parse_stable_channel(bytes: &[u8]) -> Result { + if bytes.len() > MAX_RELEASE_BYTES { + return Err(ProfileRefreshError::ResponseTooLarge); + } + let text = std::str::from_utf8(bytes).map_err(|_| ProfileRefreshError::InvalidMetadata)?; + let version = text.trim(); + version_sequence(version)?; + Ok(version.to_owned()) +} + +/// 校验 npm latest 标签及六个平台二进制包来自同一版本发布。 +fn parse_npm_release(bytes: &[u8]) -> Result { + if bytes.len() > MAX_RELEASE_BYTES { + return Err(ProfileRefreshError::ResponseTooLarge); + } + let release = serde_json::from_slice::(bytes) + .map_err(|_| ProfileRefreshError::InvalidMetadata)?; + version_sequence(&release.version)?; + if release.name != CLI_NPM_PACKAGE + || CLI_TARGETS.iter().any(|target| { + release + .optional_dependencies + .get(&format!("{CLI_NPM_PACKAGE}-{target}")) + != Some(&release.version) + }) + { + return Err(ProfileRefreshError::InvalidMetadata); + } + Ok(release.version) +} + +fn refresh_enabled_from(value: Option<&str>) -> bool { + !value.is_some_and(|value| { + matches!( + value.trim().to_ascii_lowercase().as_str(), + "0" | "false" | "off" + ) + }) +} + +fn refresh_enabled() -> bool { + refresh_enabled_from( + std::env::var("AETHER_XAI_CLIENT_PROFILE_REFRESH") + .ok() + .as_deref(), + ) +} + +fn fixed_version_from(value: Option<&str>) -> Option { + let value = value?.trim(); + if value.is_empty() || version_sequence(value).is_err() { + None + } else { + Some(value.to_owned()) + } +} + +fn fixed_version_override() -> Option { + let value = std::env::var("AETHER_XAI_CLIENT_VERSION").ok()?; + let version = fixed_version_from(Some(&value)); + if version.is_none() { + warn!( + event_name = "xai_client_profile_fixed_version_invalid", + "AETHER_XAI_CLIENT_VERSION is invalid; using cached or built-in profile" + ); + } + version +} + +fn build_release_client() -> Result { + Client::builder() + .https_only(true) + .no_proxy() + .redirect(Policy::none()) + .connect_timeout(RELEASE_CONNECT_TIMEOUT) + .timeout(RELEASE_REQUEST_TIMEOUT) + .build() + .map_err(ProfileRefreshError::Client) +} + +async fn fetch_bounded(client: &Client, url: &str) -> Result, ProfileRefreshError> { + let response = client + .get(url) + .send() + .await + .map_err(ProfileRefreshError::Client)?; + if !response.status().is_success() { + return Err(ProfileRefreshError::HttpStatus(response.status().as_u16())); + } + if response + .content_length() + .is_some_and(|length| length > MAX_RELEASE_BYTES as u64) + { + return Err(ProfileRefreshError::ResponseTooLarge); + } + + let mut bytes = Vec::new(); + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(ProfileRefreshError::Client)?; + if bytes.len().saturating_add(chunk.len()) > MAX_RELEASE_BYTES { + return Err(ProfileRefreshError::ResponseTooLarge); + } + bytes.extend_from_slice(&chunk); + } + Ok(bytes) +} + +async fn fetch_latest_with_fallback( + fetch_stable: S, + fetch_npm: N, +) -> Result +where + S: FnOnce() -> SFut, + SFut: Future>, + N: FnOnce() -> NFut, + NFut: Future>, +{ + let stable_error = match fetch_stable().await { + Ok(version) => return Ok(version), + Err(error) => error, + }; + fetch_npm() + .await + .map_err(|npm_error| ProfileRefreshError::AllSourcesFailed { + stable: stable_error.to_string(), + npm: npm_error.to_string(), + }) +} + +async fn fetch_latest_cli_version(client: &Client) -> Result { + fetch_latest_with_fallback( + || async { + let bytes = fetch_bounded(client, CLI_STABLE_CHANNEL_ENDPOINT).await?; + parse_stable_channel(&bytes) + }, + || async { + let bytes = fetch_bounded(client, CLI_NPM_RELEASE_ENDPOINT).await?; + parse_npm_release(&bytes) + }, + ) + .await +} + +fn publish_version(version: &str) -> Result<(), ProfileRefreshError> { + set_xai_client_version(version) + .map(|_| ()) + .map_err(|_| ProfileRefreshError::InvalidMetadata) +} + +async fn restore_cached_profile(runtime: &RuntimeState) -> Result<(), ProfileRefreshError> { + let Some(raw) = runtime + .kv_get(PROFILE_CACHE_KEY) + .await + .map_err(|err| ProfileRefreshError::Cache(err.to_string()))? + else { + return Ok(()); + }; + let cached = serde_json::from_str::(&raw) + .map_err(|_| ProfileRefreshError::InvalidMetadata)?; + if let Some(version) = cached_version_to_restore(&cached, &xai_client_version())? { + publish_version(&version)?; + info!( + event_name = "xai_client_profile_restored", + version = %version, + verified_at_unix_secs = cached.verified_at_unix_secs, + "restored cached Grok CLI profile" + ); + } + Ok(()) +} + +fn cached_version_to_restore( + cached: &CachedProfile, + active_version: &str, +) -> Result, ProfileRefreshError> { + let cached_sequence = version_sequence(&cached.version)?; + let active_sequence = version_sequence(active_version)?; + Ok((cached_sequence > active_sequence).then(|| cached.version.clone())) +} + +async fn refresh_once_with_fetch( + runtime: &RuntimeState, + fixed_version: Option<&str>, + refresh_is_enabled: bool, + fetch_latest: F, +) -> Result +where + F: FnOnce() -> Fut, + Fut: Future>, +{ + if let Some(version) = fixed_version { + publish_version(version)?; + return Ok(version.to_owned()); + } + + if let Err(error) = restore_cached_profile(runtime).await { + // 缓存损坏或暂时不可用不应阻断官方版本检查；当前进程继续使用旧画像。 + warn!( + event_name = "xai_client_profile_cache_restore_failed", + error = %error, + "could not restore cached Grok CLI profile" + ); + } + if !refresh_is_enabled { + return Ok(xai_client_version()); + } + + let version = fetch_latest().await?; + let current = xai_client_version(); + if version_sequence(&version)? < version_sequence(¤t)? { + return Err(ProfileRefreshError::Rollback); + } + + let cached = CachedProfile { + version: version.clone(), + verified_at_unix_secs: chrono::Utc::now().timestamp().max(0) as u64, + }; + let serialized = + serde_json::to_string(&cached).map_err(|_| ProfileRefreshError::InvalidMetadata)?; + publish_version(&version)?; + if let Err(error) = runtime + .kv_set(PROFILE_CACHE_KEY, serialized, Some(PROFILE_CACHE_TTL)) + .await + { + // 本地版本已经完成原子替换；缓存写失败只影响下次进程启动的恢复。 + warn!( + event_name = "xai_client_profile_cache_write_failed", + error = %error, + "published Grok CLI profile locally but could not persist the cache" + ); + } + Ok(version) +} + +async fn refresh_once(runtime: &RuntimeState) -> Result { + let fixed_version = fixed_version_override(); + refresh_once_with_fetch( + runtime, + fixed_version.as_deref(), + refresh_enabled(), + || async { + let client = build_release_client()?; + fetch_latest_cli_version(&client).await + }, + ) + .await +} + +pub(crate) async fn prewarm(runtime: &RuntimeState) -> Result { + refresh_once(runtime).await.map_err(|err| err.to_string()) +} + +pub(crate) fn spawn_worker(app: AppState) -> tokio::task::JoinHandle<()> { + crate::task_runtime::spawn_singleton_worker( + app, + crate::task_runtime::TASK_KEY_XAI_CLIENT_PROFILE, + |app| async move { + let mut interval = tokio::time::interval(PROFILE_REFRESH_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + // 启动阶段由 prewarm 完成一次检查；后台任务只负责后续定时刷新，避免重复建连。 + interval.tick().await; + loop { + interval.tick().await; + match refresh_once(app.runtime_state()).await { + Ok(version) => info!( + event_name = "xai_client_profile_refreshed", + version = %version, + "refreshed Grok CLI profile" + ), + Err(error) => warn!( + event_name = "xai_client_profile_refresh_failed", + error = %error, + "keeping the previous Grok CLI profile after refresh failure" + ), + } + } + }, + ) +} + +#[cfg(test)] +mod tests { + use std::sync::{ + atomic::{AtomicBool, Ordering}, + Mutex, OnceLock, + }; + use std::time::Duration; + + use aether_runtime_state::{MemoryRuntimeStateConfig, RuntimeState}; + + use super::{ + cached_version_to_restore, fetch_latest_with_fallback, fixed_version_from, + parse_npm_release, parse_stable_channel, refresh_enabled_from, refresh_once_with_fetch, + CachedProfile, ProfileRefreshError, PROFILE_CACHE_KEY, + }; + use crate::provider_transport::{set_xai_client_version, xai_client_version}; + + static PROFILE_TEST_LOCK: OnceLock> = OnceLock::new(); + + struct VersionRestore(String); + + impl Drop for VersionRestore { + fn drop(&mut self) { + let _ = set_xai_client_version(&self.0); + } + } + + fn version_restore_guard() -> (std::sync::MutexGuard<'static, ()>, VersionRestore) { + let lock = PROFILE_TEST_LOCK.get_or_init(|| Mutex::new(())); + let guard = lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let restore = VersionRestore(xai_client_version()); + (guard, restore) + } + + fn npm_release(version: &str) -> serde_json::Value { + let mut deps = serde_json::Map::new(); + for target in super::CLI_TARGETS { + deps.insert( + format!("@xai-official/grok-{target}"), + serde_json::Value::String(version.to_string()), + ); + } + serde_json::json!({ + "name": "@xai-official/grok", + "version": version, + "optionalDependencies": deps, + }) + } + + #[test] + fn stable_channel_accepts_only_a_bare_release_version() { + assert_eq!(parse_stable_channel(b"1.0.46\n").unwrap(), "1.0.46"); + assert!(parse_stable_channel(b"1.0.46").is_err()); + assert!(parse_stable_channel(b"1.0.47-alpha.1").is_err()); + assert!(parse_stable_channel(b"").is_err()); + } + + #[test] + fn npm_release_requires_every_platform_binary_at_the_same_version() { + let body = npm_release("1.0.46"); + assert_eq!( + parse_npm_release(&serde_json::to_vec(&body).unwrap()).unwrap(), + "1.0.46" + ); + + let mut mismatched = npm_release("1.0.46"); + mismatched["optionalDependencies"]["@xai-official/grok-linux-x64"] = + serde_json::Value::String("1.0.45".to_string()); + assert!(parse_npm_release(&serde_json::to_vec(&mismatched).unwrap()).is_err()); + + let mut wrong_package = npm_release("1.0.46"); + wrong_package["name"] = serde_json::Value::String("grok".to_string()); + assert!(parse_npm_release(&serde_json::to_vec(&wrong_package).unwrap()).is_err()); + } + + #[test] + fn refresh_and_fixed_version_environment_policies_are_strict() { + assert!(!refresh_enabled_from(Some("off"))); + assert!(!refresh_enabled_from(Some(" FALSE "))); + assert!(refresh_enabled_from(None)); + assert_eq!( + fixed_version_from(Some(" 1.0.46 ")).as_deref(), + Some("1.0.46") + ); + assert!(fixed_version_from(Some("1.0.46-beta.1")).is_none()); + assert!(fixed_version_from(Some("1.0")).is_none()); + } + + #[test] + fn cached_profile_never_rewinds_active_profile() { + let cached = CachedProfile { + version: "1.0.50".to_string(), + verified_at_unix_secs: 1, + }; + assert_eq!( + cached_version_to_restore(&cached, "1.0.46").unwrap(), + Some("1.0.50".to_string()) + ); + assert_eq!(cached_version_to_restore(&cached, "1.1.0").unwrap(), None); + } + + #[tokio::test] + async fn npm_is_used_only_when_the_stable_channel_fails() { + let npm_called = AtomicBool::new(false); + let version = fetch_latest_with_fallback( + || async { Ok("1.0.46".to_string()) }, + || async { + npm_called.store(true, Ordering::SeqCst); + Ok("1.0.45".to_string()) + }, + ) + .await + .unwrap(); + assert_eq!(version, "1.0.46"); + assert!(!npm_called.load(Ordering::SeqCst)); + + let version = fetch_latest_with_fallback( + || async { Err(ProfileRefreshError::HttpStatus(503)) }, + || async { Ok("1.0.46".to_string()) }, + ) + .await + .unwrap(); + assert_eq!(version, "1.0.46"); + + let result = fetch_latest_with_fallback( + || async { Err(ProfileRefreshError::HttpStatus(503)) }, + || async { Err(ProfileRefreshError::InvalidMetadata) }, + ) + .await; + assert!(matches!( + result, + Err(ProfileRefreshError::AllSourcesFailed { .. }) + )); + } + + #[tokio::test] + async fn cache_hit_is_restored_without_network_when_refresh_is_disabled() { + let (_lock, _restore) = version_restore_guard(); + set_xai_client_version("1.0.46").unwrap(); + let runtime = RuntimeState::memory(MemoryRuntimeStateConfig::default()); + runtime + .kv_set( + PROFILE_CACHE_KEY, + serde_json::to_string(&CachedProfile { + version: "1.0.50".to_string(), + verified_at_unix_secs: 1, + }) + .unwrap(), + Some(Duration::from_secs(60)), + ) + .await + .unwrap(); + + let result = refresh_once_with_fetch(&runtime, None, false, || async { + Err(ProfileRefreshError::HttpStatus(599)) + }) + .await + .unwrap(); + + assert_eq!(result, "1.0.50"); + assert_eq!(xai_client_version(), "1.0.50"); + } + + #[tokio::test] + async fn refresh_failure_keeps_previous_profile() { + let (_lock, _restore) = version_restore_guard(); + let runtime = RuntimeState::memory(MemoryRuntimeStateConfig::default()); + let before = xai_client_version(); + let result = refresh_once_with_fetch(&runtime, None, true, || async { + Err(ProfileRefreshError::HttpStatus(503)) + }) + .await; + + assert!(matches!(result, Err(ProfileRefreshError::HttpStatus(503)))); + assert_eq!(xai_client_version(), before); + } + + #[tokio::test] + async fn successful_refresh_publishes_and_caches_version() { + let (_lock, _restore) = version_restore_guard(); + set_xai_client_version("1.0.46").unwrap(); + let runtime = RuntimeState::memory(MemoryRuntimeStateConfig::default()); + let result = + refresh_once_with_fetch(&runtime, None, true, || async { Ok("1.0.51".to_string()) }) + .await + .unwrap(); + + assert_eq!(result, "1.0.51"); + assert_eq!(xai_client_version(), "1.0.51"); + let cached = runtime.kv_get(PROFILE_CACHE_KEY).await.unwrap().unwrap(); + assert!(cached.contains("\"1.0.51\"")); + } + + #[tokio::test] + async fn fixed_version_override_skips_network_and_publishes_version() { + let (_lock, _restore) = version_restore_guard(); + let runtime = RuntimeState::memory(MemoryRuntimeStateConfig::default()); + let fetch_called = AtomicBool::new(false); + let result = refresh_once_with_fetch(&runtime, Some("1.0.60"), true, || async { + fetch_called.store(true, Ordering::SeqCst); + Ok("1.0.61".to_string()) + }) + .await + .unwrap(); + + assert_eq!(result, "1.0.60"); + assert!(!fetch_called.load(Ordering::SeqCst)); + assert_eq!(xai_client_version(), "1.0.60"); + } + + #[tokio::test] + async fn rollback_is_rejected_without_replacing_profile() { + let (_lock, _restore) = version_restore_guard(); + set_xai_client_version("1.0.60").unwrap(); + let runtime = RuntimeState::memory(MemoryRuntimeStateConfig::default()); + let result = + refresh_once_with_fetch(&runtime, None, true, || async { Ok("1.0.59".to_string()) }) + .await; + + assert!(matches!(result, Err(ProfileRefreshError::Rollback))); + assert_eq!(xai_client_version(), "1.0.60"); + } +} diff --git a/crates/aether-provider/transport/src/lib.rs b/crates/aether-provider/transport/src/lib.rs index 0f12952e7..17dfe421c 100644 --- a/crates/aether-provider/transport/src/lib.rs +++ b/crates/aether-provider/transport/src/lib.rs @@ -200,7 +200,8 @@ pub use windsurf::{ pub use xai::{ extract_xai_user_id_from_auth_config, extract_xai_user_id_from_value, insert_cli_identity_headers, insert_cli_identity_headers_if_needed, is_xai_provider_transport, - resolved_xai_request_base_url, resolved_xai_upstream_base_url, - should_attach_cli_identity_headers, xai_auth_uses_api, xai_uses_official_api, XAI_API_BASE_URL, - XAI_CHAT_PROXY_BASE_URL, XAI_PROVIDER_TYPE, + resolved_xai_request_base_url, resolved_xai_upstream_base_url, set_xai_client_version, + should_attach_cli_identity_headers, xai_auth_uses_api, xai_client_version, + xai_uses_official_api, XAI_API_BASE_URL, XAI_CHAT_PROXY_BASE_URL, XAI_DEFAULT_CLIENT_VERSION, + XAI_PROVIDER_TYPE, }; diff --git a/crates/aether-provider/transport/src/xai.rs b/crates/aether-provider/transport/src/xai.rs index 5dbeae020..7da488600 100644 --- a/crates/aether-provider/transport/src/xai.rs +++ b/crates/aether-provider/transport/src/xai.rs @@ -1,6 +1,7 @@ pub mod video; use std::collections::BTreeMap; +use std::sync::{OnceLock, RwLock}; use aether_ai_formats::normalize_api_format_alias; use serde_json::Value; @@ -10,7 +11,10 @@ use crate::snapshot::GatewayProviderTransportSnapshot; pub const XAI_PROVIDER_TYPE: &str = "xai"; pub const XAI_CHAT_PROXY_BASE_URL: &str = "https://cli-chat-proxy.grok.com/v1"; pub const XAI_API_BASE_URL: &str = "https://api.x.ai/v1"; -pub const XAI_CLIENT_VERSION: &str = "0.2.120"; +/// 内置的 Grok CLI 版本；网关后台任务会用官方发布版本覆盖它。 +/// +/// cli-chat-proxy 会对过旧的版本直接返回 426，因此这里只作为发布检查不可用时的兜底。 +pub const XAI_DEFAULT_CLIENT_VERSION: &str = "1.0.46"; pub const XAI_TOKEN_AUTH_HEADER: &str = "x-xai-token-auth"; pub const XAI_TOKEN_AUTH_VALUE: &str = "xai-grok-cli"; pub const XAI_CLIENT_VERSION_HEADER: &str = "x-grok-client-version"; @@ -19,8 +23,37 @@ pub const XAI_CLIENT_IDENTIFIER_VALUE: &str = "grok-shell"; pub const XAI_AUTHENTICATE_RESPONSE_HEADER: &str = "x-authenticateresponse"; pub const XAI_AUTHENTICATE_RESPONSE_VALUE: &str = "authenticate-response"; +static ACTIVE_CLIENT_VERSION: OnceLock> = OnceLock::new(); + +fn active_client_version() -> &'static RwLock { + ACTIVE_CLIENT_VERSION.get_or_init(|| RwLock::new(XAI_DEFAULT_CLIENT_VERSION.to_owned())) +} + +/// 返回当前发布的 Grok CLI 版本快照。 +pub fn xai_client_version() -> String { + active_client_version() + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() +} + +/// 原子替换当前 Grok CLI 版本，返回替换前的版本；版本校验由发布检查器负责，这里只拒绝明显非法值。 +pub fn set_xai_client_version(version: &str) -> Result { + let version = version.trim(); + if version.is_empty() + || version.len() > 64 + || !version.bytes().all(|byte| (33..=126).contains(&byte)) + { + return Err("invalid Grok CLI version"); + } + let mut current = active_client_version() + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + Ok(std::mem::replace(&mut *current, version.to_owned())) +} + pub fn xai_cli_user_agent() -> String { - format!("xai-grok-workspace/{XAI_CLIENT_VERSION}") + format!("xai-grok-workspace/{}", xai_client_version()) } pub fn is_xai_provider_transport(transport: &GatewayProviderTransportSnapshot) -> bool { @@ -91,10 +124,11 @@ pub fn should_attach_cli_identity_headers( } pub fn insert_cli_identity_headers(headers: &mut BTreeMap) { + let client_version = xai_client_version(); let user_agent = xai_cli_user_agent(); for (name, value) in [ (XAI_TOKEN_AUTH_HEADER, XAI_TOKEN_AUTH_VALUE), - (XAI_CLIENT_VERSION_HEADER, XAI_CLIENT_VERSION), + (XAI_CLIENT_VERSION_HEADER, client_version.as_str()), ("user-agent", user_agent.as_str()), (XAI_CLIENT_IDENTIFIER_HEADER, XAI_CLIENT_IDENTIFIER_VALUE), ( @@ -441,4 +475,31 @@ mod tests { Some(r#"{"api_key":"xai-key","using_api":true}"#) )); } + + #[test] + fn cli_identity_headers_follow_published_client_version() { + use super::{ + insert_cli_identity_headers, set_xai_client_version, xai_client_version, + XAI_CLIENT_VERSION_HEADER, + }; + + let previous = xai_client_version(); + assert!(set_xai_client_version("").is_err()); + assert!(set_xai_client_version("1.0 .1").is_err()); + assert_eq!(xai_client_version(), previous); + + set_xai_client_version(" 9.8.7 ").expect("valid version"); + let mut headers = BTreeMap::new(); + insert_cli_identity_headers(&mut headers); + set_xai_client_version(&previous).expect("restore version"); + + assert_eq!( + headers.get(XAI_CLIENT_VERSION_HEADER).map(String::as_str), + Some("9.8.7") + ); + assert_eq!( + headers.get("user-agent").map(String::as_str), + Some("xai-grok-workspace/9.8.7") + ); + } }