From 41f97ac836b5e11194d95de268d04f4acd439c6e Mon Sep 17 00:00:00 2001 From: David Lee <88446549+davidwushi1145@users.noreply.github.com> Date: Sat, 14 Nov 2026 16:34:39 +0800 Subject: [PATCH] feat(openai): check client profile releases hourly (#711) --- .../src/transport/profile/cli_release.rs | 142 +++++++++----- .../openai/src/transport/profile/mod.rs | 29 ++- .../transport/profile/platform_release/mod.rs | 33 +++- .../openai/src/transport/profile/selection.rs | 2 +- .../tests/transport/profile/cli_release.rs | 177 +++++++++++++++++- .../openai/tests/transport/profile/mod.rs | 48 +++-- .../transport/profile/platform_release/mod.rs | 95 ++++++++-- docs/api.md | 6 +- docs/architecture.md | 4 + 9 files changed, 452 insertions(+), 84 deletions(-) diff --git a/backend/crates/providers/openai/src/transport/profile/cli_release.rs b/backend/crates/providers/openai/src/transport/profile/cli_release.rs index 4493f50c9..08988edcc 100644 --- a/backend/crates/providers/openai/src/transport/profile/cli_release.rs +++ b/backend/crates/providers/openai/src/transport/profile/cli_release.rs @@ -1,18 +1,22 @@ //! CLI 稳定发布来自官方 npm 包及其平台依赖，不与 Desktop 的内嵌版本混用 use std::collections::BTreeMap; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Duration; use chrono::Utc; -use futures::StreamExt as _; +use futures::{StreamExt as _, future::BoxFuture}; use gateway_core::provider_ports::{ProviderArtifactProfile, ProviderArtifactProfileCachePort}; use gateway_core::routing::ProviderKind; use reqwest::{Client, redirect::Policy}; use serde::Deserialize; +use tokio::time::Instant; use super::selection::{ClientKind, ClientPlatform, ClientRelease, object}; -use super::{ARTIFACT_PROFILE_CACHE_TTL, CodexWireProfileState, baseline}; +use super::{ + ARTIFACT_PROFILE_CACHE_RENEW_INTERVAL, ARTIFACT_PROFILE_CACHE_TTL, CodexWireProfileState, + baseline, +}; const ENDPOINT: &str = "https://registry.npmjs.org/@openai%2Fcodex/latest"; const MAX_BYTES: usize = 256 * 1024; @@ -67,11 +71,21 @@ fn sequence(version: &str) -> Result { Ok(1 + version.major * 1_000_000 + version.minor * 1_000 + version.patch) } -pub struct CliReleaseService { +/// 官方 npm 发布读取边界 +pub trait CliReleaseTransport: Send + Sync { + fn fetch(&self) -> BoxFuture<'_, Result>; +} + +struct OfficialCliReleaseTransport { client: Client, +} + +pub struct CliReleaseService { + transport: Arc, provider: ProviderKind, state: CodexWireProfileState, cache: Arc, + cache_writes: Mutex>, } impl CliReleaseService { @@ -88,12 +102,27 @@ impl CliReleaseService { .timeout(Duration::from_secs(30)) .build() .map_err(|_| CliReleaseError::Fetch)?; - Ok(Self { - client, + Ok(Self::with_transport( provider, state, cache, - }) + Arc::new(OfficialCliReleaseTransport { client }), + )) + } + + pub fn with_transport( + provider: ProviderKind, + state: CodexWireProfileState, + cache: Arc, + transport: Arc, + ) -> Self { + Self { + transport, + provider, + state, + cache, + cache_writes: Mutex::default(), + } } pub async fn restore(&self) { @@ -135,7 +164,7 @@ impl CliReleaseService { } pub async fn refresh(&self) -> Result<(), CliReleaseError> { - let result = self.fetch().await; + let result = self.transport.fetch().await; for (platform, arch, target) in TARGETS { let outcome = match &result { Ok(version) => self.publish(platform, arch, target, version).await, @@ -155,16 +184,29 @@ impl CliReleaseService { version: &str, ) -> Result { let next = sequence(version).map_err(|error| error.to_string())?; - if self - .state - .client_release(ClientKind::Cli, platform, arch) - .is_some_and(|current| { - sequence(¤t.codex_version).is_ok_and(|current| current > next) - }) - { + let current = self.state.client_release(ClientKind::Cli, platform, arch); + if current.as_ref().is_some_and(|current| { + sequence(¤t.codex_version).is_ok_and(|current| current > next) + }) { return Err("官方 CLI 发布版本低于当前已核验版本".to_owned()); } - let verified_at = Utc::now(); + let unchanged = current.filter(|current| current.codex_version == version); + if let Some(current) = &unchanged { + let fresh = self + .cache_writes + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(target) + .is_some_and(|written| written.elapsed() < ARTIFACT_PROFILE_CACHE_RENEW_INTERVAL); + if fresh { + return Ok(current.clone()); + } + } + // 同版探测和缓存续期保留实际核验时间，只有新版才产生新的核验时间 + let verified_at = unchanged + .as_ref() + .and_then(|release| release.verified_at) + .unwrap_or_else(Utc::now); let payload = object(&serde_json::json!({"version": version})).map_err(|error| error.to_string())?; let profile = ProviderArtifactProfile::new( @@ -179,41 +221,51 @@ impl CliReleaseService { .replace_if_newer(profile, ARTIFACT_PROFILE_CACHE_TTL) .await { - Ok(true) => Ok(ClientRelease { - codex_version: version.to_owned(), - desktop_version: None, - desktop_build: None, - verified_at: Some(verified_at), - }), + Ok(true) => { + self.cache_writes + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(target.to_owned(), Instant::now()); + Ok(ClientRelease { + codex_version: version.to_owned(), + desktop_version: None, + desktop_build: None, + verified_at: Some(verified_at), + }) + } Ok(false) => Err("CLI 发布缓存已更新，等待下次同步".to_owned()), Err(_) => Err("CLI 发布版本缓存写入失败".to_owned()), } } +} - async fn fetch(&self) -> Result { - let response = self - .client - .get(ENDPOINT) - .send() - .await - .map_err(|_| CliReleaseError::Fetch)?; - if !response.status().is_success() - || response - .content_length() - .is_some_and(|length| length > MAX_BYTES as u64) - { - return Err(CliReleaseError::Fetch); - } - let mut bytes = Vec::new(); - let mut stream = response.bytes_stream(); - while let Some(chunk) = stream.next().await { - let chunk = chunk.map_err(|_| CliReleaseError::Fetch)?; - if bytes.len().saturating_add(chunk.len()) > MAX_BYTES { - return Err(CliReleaseError::Invalid); +impl CliReleaseTransport for OfficialCliReleaseTransport { + fn fetch(&self) -> BoxFuture<'_, Result> { + Box::pin(async move { + let response = self + .client + .get(ENDPOINT) + .send() + .await + .map_err(|_| CliReleaseError::Fetch)?; + if !response.status().is_success() + || response + .content_length() + .is_some_and(|length| length > MAX_BYTES as u64) + { + return Err(CliReleaseError::Fetch); } - bytes.extend_from_slice(&chunk); - } - parse_cli_release(&bytes) + let mut bytes = Vec::new(); + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|_| CliReleaseError::Fetch)?; + if bytes.len().saturating_add(chunk.len()) > MAX_BYTES { + return Err(CliReleaseError::Invalid); + } + bytes.extend_from_slice(&chunk); + } + parse_cli_release(&bytes) + }) } } diff --git a/backend/crates/providers/openai/src/transport/profile/mod.rs b/backend/crates/providers/openai/src/transport/profile/mod.rs index d9c4aef24..3034631a2 100644 --- a/backend/crates/providers/openai/src/transport/profile/mod.rs +++ b/backend/crates/providers/openai/src/transport/profile/mod.rs @@ -34,11 +34,12 @@ use selection::{ClientKind, ClientPlatform, ClientRelease}; pub const CODEX_DESKTOP_APPCAST_URL: &str = "https://persistent.oaistatic.com/codex-app-prod/appcast.xml"; /// 官方发布元数据检查周期 -pub const APPCAST_POLL_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60); +pub const APPCAST_POLL_INTERVAL: Duration = Duration::from_secs(60 * 60); const APPCAST_TIMEOUT: Duration = Duration::from_secs(30); const MAX_APPCAST_BYTES: usize = 1024 * 1024; const ARTIFACT_PROFILE_CACHE_TTL: Duration = Duration::from_secs(30 * 24 * 60 * 60); +const ARTIFACT_PROFILE_CACHE_RENEW_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60); const ARTIFACT_PROFILE_SCHEMA_VERSION: u64 = 1; /// 保留已有配置类型入口；位置规则由出口领域统一维护 @@ -634,6 +635,7 @@ pub struct CodexDesktopReleaseService { profile: CodexWireProfileState, cache: CodexArtifactProfileCache, verified: Arc>>, + cache_written_at: Arc>>, } impl CodexDesktopReleaseService { @@ -668,6 +670,7 @@ impl CodexDesktopReleaseService { profile, cache, verified: Arc::new(RwLock::new(verified)), + cache_written_at: Arc::default(), } } @@ -732,10 +735,24 @@ impl CodexDesktopReleaseService { { return Err(CodexDesktopReleaseError::InconsistentRelease); } - if !self.cache.replace_if_newer(&verified).await? { - return Err(CodexDesktopReleaseError::ReleaseRollback); + // 轻量检查只推进观察时间，缓存单独按天续期 + let renewal_due = self + .cache_written_at + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_none_or(|written| { + written.elapsed() >= ARTIFACT_PROFILE_CACHE_RENEW_INTERVAL + }); + if renewal_due { + if !self.cache.replace_if_newer(&verified).await? { + return Err(CodexDesktopReleaseError::ReleaseRollback); + } + *self + .cache_written_at + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = + Some(tokio::time::Instant::now()); } - self.profile.update_bundled_release(&verified); return Ok(release); } } @@ -766,6 +783,10 @@ impl CodexDesktopReleaseService { .verified .write() .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(published); + *self + .cache_written_at + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(tokio::time::Instant::now()); Ok(release) } } diff --git a/backend/crates/providers/openai/src/transport/profile/platform_release/mod.rs b/backend/crates/providers/openai/src/transport/profile/platform_release/mod.rs index ffb78cfef..75d3653eb 100644 --- a/backend/crates/providers/openai/src/transport/profile/platform_release/mod.rs +++ b/backend/crates/providers/openai/src/transport/profile/platform_release/mod.rs @@ -3,7 +3,10 @@ use super::selection::{ ClientKind, ClientPlatform, ClientProfileSelection, ClientRelease, VersionMode, object, }; -use super::{ARTIFACT_PROFILE_CACHE_TTL, CodexWireProfileState, baseline}; +use super::{ + ARTIFACT_PROFILE_CACHE_RENEW_INTERVAL, ARTIFACT_PROFILE_CACHE_TTL, CodexWireProfileState, + baseline, +}; use chrono::Utc; use futures::future::BoxFuture; use gateway_core::provider_ports::{ProviderArtifactProfile, ProviderArtifactProfileCachePort}; @@ -135,6 +138,7 @@ pub struct PlatformDesktopReleaseService { cache: Arc, transport: Arc, verified: Mutex>, + cache_written_at: Mutex>, } impl PlatformDesktopReleaseService { pub fn new( @@ -149,6 +153,7 @@ impl PlatformDesktopReleaseService { cache, transport, verified: Mutex::new(baseline_artifacts()), + cache_written_at: Mutex::default(), } } pub async fn restore(&self) { @@ -206,14 +211,32 @@ impl PlatformDesktopReleaseService { .cloned(); let mut artifact = self .transport - .fetch(target, previous) + .fetch(target, previous.clone()) .await .map_err(|error| error.to_string())?; let sequence = sequence(target, &artifact.release).map_err(|error| error.to_string())?; if !self.accepts(target, &artifact.release) { return Err("官方 Desktop 发布低于当前版本或同构建版本不一致".to_owned()); } - let verified_at = Utc::now(); + let unchanged = previous + .as_ref() + .is_some_and(|previous| previous.identity == artifact.identity); + if unchanged + && self + .cache_written_at + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(&target) + .is_some_and(|written| written.elapsed() < ARTIFACT_PROFILE_CACHE_RENEW_INTERVAL) + { + return Ok(artifact.release); + } + // 相同制品只续期缓存，保留实际提取版本时的核验时间 + let verified_at = if unchanged { + artifact.release.verified_at.unwrap_or_else(Utc::now) + } else { + Utc::now() + }; // 核验时间属于缓存信封，避免同构建重复检查被判定为不同制品 artifact.release.verified_at = None; let payload = @@ -235,6 +258,10 @@ impl PlatformDesktopReleaseService { Ok(false) => return Err("Desktop 发布缓存已更新，等待下次同步".to_owned()), Err(_) => return Err("Desktop 发布缓存写入失败".to_owned()), } + self.cache_written_at + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(target, tokio::time::Instant::now()); artifact.release.verified_at = Some(verified_at); self.verified .lock() diff --git a/backend/crates/providers/openai/src/transport/profile/selection.rs b/backend/crates/providers/openai/src/transport/profile/selection.rs index db6d4caff..36e7cc67e 100644 --- a/backend/crates/providers/openai/src/transport/profile/selection.rs +++ b/backend/crates/providers/openai/src/transport/profile/selection.rs @@ -282,7 +282,7 @@ impl ClientProfileSelection { verified_at: release.verified_at.unwrap_or(DateTime::UNIX_EPOCH), }; if let Some(entry) = self.cli_entry { - // 入口后缀使用同一次解析得到的 Core 版本，避免每日更新后头部与后缀混用 + // 入口后缀使用同一次解析得到的 Core 版本，避免版本更新后头部与后缀混用 // originator 可单独覆盖；入口名与官方 clientInfo.name 的语义保持一致 profile.exact_user_agent = Some(format!( "{} ({}; {})", diff --git a/backend/crates/providers/openai/tests/transport/profile/cli_release.rs b/backend/crates/providers/openai/tests/transport/profile/cli_release.rs index 98329dd0d..543534505 100644 --- a/backend/crates/providers/openai/tests/transport/profile/cli_release.rs +++ b/backend/crates/providers/openai/tests/transport/profile/cli_release.rs @@ -82,7 +82,7 @@ async fn official_cli_cache_updates_entry_headers_together_and_preserves_frozen_ )) .unwrap(); - // 复用官方版本服务的恢复入口，模拟每日检查已核验并写入的下一份发布资料 + // 复用官方版本服务的恢复入口，模拟版本检查已核验并写入的下一份发布资料 cache .replace_if_newer( ProviderArtifactProfile::new( @@ -142,3 +142,178 @@ async fn official_cli_cache_updates_entry_headers_together_and_preserves_frozen_ automatic.resolve(&state).unwrap() ); } + +struct CliTransport { + version: std::sync::Mutex, +} + +impl provider_openai::transport::profile::cli_release::CliReleaseTransport for CliTransport { + fn fetch( + &self, + ) -> futures::future::BoxFuture< + '_, + Result, + > { + Box::pin(async move { Ok(self.version.lock().unwrap().clone()) }) + } +} + +#[derive(Default)] +struct CliCache { + writes: std::sync::Mutex>, + failing_target: std::sync::Mutex>, +} + +impl gateway_core::provider_ports::ProviderArtifactProfileCachePort for CliCache { + fn replace_if_newer( + &self, + profile: gateway_core::provider_ports::ProviderArtifactProfile, + _ttl: std::time::Duration, + ) -> futures::future::BoxFuture< + '_, + Result, + > { + Box::pin(async move { + *self + .writes + .lock() + .unwrap() + .entry(profile.artifact_key().to_owned()) + .or_default() += 1; + if self.failing_target.lock().unwrap().as_deref() == Some(profile.artifact_key()) { + return Err(gateway_core::provider_ports::ProviderStoreError::new( + gateway_core::provider_ports::ProviderStoreErrorKind::Unavailable, + "write CLI fixture", + )); + } + Ok(true) + }) + } + + fn read<'a>( + &'a self, + _provider: &'a gateway_core::routing::ProviderKind, + _key: &'a str, + ) -> futures::future::BoxFuture< + 'a, + Result< + Option, + gateway_core::provider_ports::ProviderStoreError, + >, + > { + Box::pin(async { Ok(None) }) + } +} + +#[tokio::test(start_paused = true)] +async fn unchanged_cli_release_only_renews_cache_daily_and_preserves_verification_time() { + use provider_openai::transport::profile::{ + CodexWireProfileState, + cli_release::CliReleaseService, + selection::{ClientKind, ClientPlatform}, + }; + use std::{sync::Arc, time::Duration}; + let state = CodexWireProfileState::new(super::wire_profile()); + let cache = Arc::new(CliCache::default()); + let transport = Arc::new(CliTransport { + version: std::sync::Mutex::new("1.0.0".to_owned()), + }); + let service = CliReleaseService::with_transport( + gateway_core::routing::ProviderKind::new("openai").unwrap(), + state.clone(), + cache.clone(), + transport, + ); + service.refresh().await.unwrap(); + let verified = state + .client_release(ClientKind::Cli, ClientPlatform::Linux, "x86_64") + .unwrap(); + tokio::time::advance(Duration::from_secs(3600)).await; + service.refresh().await.unwrap(); + assert!( + cache + .writes + .lock() + .unwrap() + .values() + .all(|count| *count == 1) + ); + tokio::time::advance(Duration::from_secs(23 * 3600)).await; + service.refresh().await.unwrap(); + assert!( + cache + .writes + .lock() + .unwrap() + .values() + .all(|count| *count == 2) + ); + assert_eq!( + state + .client_release(ClientKind::Cli, ClientPlatform::Linux, "x86_64") + .unwrap(), + verified + ); +} + +#[tokio::test(start_paused = true)] +async fn cli_partial_cache_failure_retries_the_same_release_without_republishing_other_targets() { + use provider_openai::transport::profile::{ + CodexWireProfileState, + cli_release::CliReleaseService, + selection::{ClientKind, ClientPlatform}, + }; + use std::{sync::Arc, time::Duration}; + let state = CodexWireProfileState::new(super::wire_profile()); + let original = state + .client_release(ClientKind::Cli, ClientPlatform::Linux, "x86_64") + .unwrap(); + let cache = Arc::new(CliCache::default()); + *cache.failing_target.lock().unwrap() = Some("cli-linux-x64".to_owned()); + let transport = Arc::new(CliTransport { + version: std::sync::Mutex::new("1.0.0".to_owned()), + }); + let service = CliReleaseService::with_transport( + gateway_core::routing::ProviderKind::new("openai").unwrap(), + state.clone(), + cache.clone(), + transport, + ); + service.refresh().await.unwrap(); + assert_eq!( + state + .client_release(ClientKind::Cli, ClientPlatform::Linux, "x86_64") + .unwrap(), + original + ); + assert!( + state + .client_release_status(ClientKind::Cli, ClientPlatform::Linux, "x86_64") + .1 + .is_some() + ); + *cache.failing_target.lock().unwrap() = None; + tokio::time::advance(Duration::from_secs(3600)).await; + service.refresh().await.unwrap(); + assert_eq!( + state + .client_release(ClientKind::Cli, ClientPlatform::Linux, "x86_64") + .unwrap() + .codex_version, + "1.0.0" + ); + assert!( + state + .client_release_status(ClientKind::Cli, ClientPlatform::Linux, "x86_64") + .1 + .is_none() + ); + let writes = cache.writes.lock().unwrap(); + assert_eq!(writes["cli-linux-x64"], 2); + assert!( + writes + .iter() + .filter(|(key, _)| key.as_str() != "cli-linux-x64") + .all(|(_, count)| *count == 1) + ); +} diff --git a/backend/crates/providers/openai/tests/transport/profile/mod.rs b/backend/crates/providers/openai/tests/transport/profile/mod.rs index 2d3331007..dbf118313 100644 --- a/backend/crates/providers/openai/tests/transport/profile/mod.rs +++ b/backend/crates/providers/openai/tests/transport/profile/mod.rs @@ -1,6 +1,7 @@ //! 客户端画像测试入口，以及 Desktop 发行解析和原子更新测试 use std::collections::VecDeque; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -75,14 +76,16 @@ impl CodexDesktopReleaseTransport for ReleaseTransport { #[derive(Default)] struct ArtifactProfiles { profile: Mutex>, - fail_writes: bool, + fail_writes: AtomicBool, + writes: AtomicUsize, } impl ArtifactProfiles { fn failing() -> Self { Self { profile: Mutex::new(None), - fail_writes: true, + fail_writes: AtomicBool::new(true), + writes: AtomicUsize::new(0), } } } @@ -94,7 +97,8 @@ impl ProviderArtifactProfileCachePort for ArtifactProfiles { _ttl: Duration, ) -> BoxFuture<'_, Result> { Box::pin(async move { - if self.fail_writes { + self.writes.fetch_add(1, Ordering::SeqCst); + if self.fail_writes.load(Ordering::SeqCst) { return Err(ProviderStoreError::new( ProviderStoreErrorKind::Unavailable, "replace test artifact profile", @@ -272,19 +276,34 @@ async fn bundled_release_update_should_change_core_and_desktop_identity_atomical assert_eq!(transport.core_probes(), 1); } -#[tokio::test] -async fn unchanged_verified_release_should_refresh_cache_without_downloading_core_again() { +#[tokio::test(start_paused = true)] +async fn unchanged_verified_release_renews_cache_daily_without_downloading_core_again() { let state = CodexWireProfileState::new(wire_profile()); let latest = release("26.810.41047", "6570"); let transport = Arc::new(ReleaseTransport::new( - [Ok(latest.clone()), Ok(latest)], + (0..5).map(|_| Ok(latest.clone())), [Ok("0.148.0-alpha.9".to_owned())], )); let cache = Arc::new(ArtifactProfiles::default()); - let service = service(state, transport.clone(), cache, None); + let service = service(state.clone(), transport.clone(), cache.clone(), None); service.refresh().await.expect("first release refresh"); + let verified = state.snapshot(); + tokio::time::advance(Duration::from_secs(60 * 60)).await; service.refresh().await.expect("cached release refresh"); + assert_eq!(cache.writes.load(Ordering::SeqCst), 1); + + tokio::time::advance(Duration::from_secs(23 * 60 * 60)).await; + cache.fail_writes.store(true, Ordering::SeqCst); + service.refresh().await.expect_err("renewal failure"); + assert_eq!(state.snapshot(), verified); + cache.fail_writes.store(false, Ordering::SeqCst); + service.refresh().await.expect("retry renewal"); + assert_eq!(cache.writes.load(Ordering::SeqCst), 3); + service.refresh().await.expect("renewed cache"); + assert_eq!(cache.writes.load(Ordering::SeqCst), 3); + assert_eq!(state.snapshot(), verified); + assert!(service.status().snapshot().last_error.is_none()); assert_eq!(transport.core_probes(), 1); } @@ -329,15 +348,11 @@ async fn cache_failure_should_not_publish_a_partially_verified_profile() { let original = wire_profile(); let state = CodexWireProfileState::new(original.clone()); let transport = Arc::new(ReleaseTransport::new( - [Ok(release("26.810.41047", "6570"))], - [Ok("0.148.0-alpha.9".to_owned())], + (0..2).map(|_| Ok(release("26.810.41047", "6570"))), + (0..2).map(|_| Ok("0.148.0-alpha.9".to_owned())), )); - let service = service( - state.clone(), - transport, - Arc::new(ArtifactProfiles::failing()), - None, - ); + let cache = Arc::new(ArtifactProfiles::failing()); + let service = service(state.clone(), transport, cache.clone(), None); service .refresh() @@ -346,6 +361,9 @@ async fn cache_failure_should_not_publish_a_partially_verified_profile() { assert_eq!(state.snapshot(), original); assert!(service.status().snapshot().latest.is_none()); + cache.fail_writes.store(false, Ordering::SeqCst); + service.refresh().await.expect("retry same release"); + assert_eq!(state.snapshot().desktop_build, "6570"); } /// 滞后档位按发布观察序列解析：启动种子、历次核验发布与失败观察共同构成历史。 diff --git a/backend/crates/providers/openai/tests/transport/profile/platform_release/mod.rs b/backend/crates/providers/openai/tests/transport/profile/platform_release/mod.rs index 188c611c7..f8b12c941 100644 --- a/backend/crates/providers/openai/tests/transport/profile/platform_release/mod.rs +++ b/backend/crates/providers/openai/tests/transport/profile/platform_release/mod.rs @@ -14,6 +14,7 @@ use provider_openai::transport::profile::selection::{ClientKind, ClientProfileSe use serde_json::json; use std::collections::BTreeMap; use std::io; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -24,7 +25,8 @@ mod xz; #[derive(Default)] struct Cache { profiles: Mutex>, - reject: bool, + reject: AtomicBool, + writes: AtomicUsize, } impl ProviderArtifactProfileCachePort for Cache { fn read<'a>( @@ -40,7 +42,8 @@ impl ProviderArtifactProfileCachePort for Cache { _ttl: Duration, ) -> BoxFuture<'_, Result> { Box::pin(async move { - if self.reject { + self.writes.fetch_add(1, Ordering::Relaxed); + if self.reject.load(Ordering::Relaxed) { return Ok(false); } self.profiles @@ -69,6 +72,9 @@ impl DesktopArtifactTransport for Transport { if round == 1 && target == TARGETS[0] { return Err(io::Error::other("fixture outage")); } + if round == 3 { + return Ok(previous); + } let build = if round == 2 { 1 } else { @@ -136,22 +142,85 @@ async fn releases_are_independent_atomic_cached_and_preserved_on_failures_or_rol .await; assert_eq!(selection.resolve(&restored).unwrap(), frozen); } +#[tokio::test(start_paused = true)] +async fn unchanged_artifacts_renew_daily_and_keep_verification_time() { + let state = CodexWireProfileState::new(super::wire_profile()); + let cache = Arc::new(Cache::default()); + let transport = Arc::new(Transport::default()); + *transport.round.lock().unwrap() = 3; + let service = service(state.clone(), cache.clone(), transport); + let target = TARGETS[0]; + let original = state.client_release(ClientKind::Desktop, target.platform, target.arch); + service.refresh().await; + assert_eq!(cache.writes.load(Ordering::Relaxed), 4); + tokio::time::advance(Duration::from_secs(60 * 60)).await; + service.refresh().await; + assert_eq!(cache.writes.load(Ordering::Relaxed), 4); + for day in 1..=31 { + tokio::time::advance(Duration::from_secs(24 * 60 * 60)).await; + service.refresh().await; + assert_eq!(cache.writes.load(Ordering::Relaxed), 4 * (day + 1)); + assert_eq!( + state.client_release(ClientKind::Desktop, target.platform, target.arch), + original + ); + } + cache.reject.store(true, Ordering::Relaxed); + tokio::time::advance(Duration::from_secs(24 * 60 * 60)).await; + service.refresh().await; + assert!( + state + .client_release_status(ClientKind::Desktop, target.platform, target.arch) + .1 + .is_some() + ); + cache.reject.store(false, Ordering::Relaxed); + tokio::time::advance(Duration::from_secs(60 * 60)).await; + service.refresh().await; + assert_eq!(cache.writes.load(Ordering::Relaxed), 4 * 34); + assert!( + state + .client_release_status(ClientKind::Desktop, target.platform, target.arch) + .1 + .is_none() + ); + assert_eq!( + state.client_release(ClientKind::Desktop, target.platform, target.arch), + original + ); +} + #[tokio::test] -async fn cache_rejection_keeps_the_previous_release() { +async fn cache_rejection_preserves_previous_release_and_retries_same_artifact() { let state = CodexWireProfileState::new(super::wire_profile()); let before = state.client_release(ClientKind::Desktop, TARGETS[0].platform, TARGETS[0].arch); - service( - state.clone(), - Arc::new(Cache { - reject: true, - ..Default::default() - }), - Arc::new(Transport::default()), - ) - .refresh() - .await; + let cache = Arc::new(Cache { + reject: AtomicBool::new(true), + ..Default::default() + }); + let transport = Arc::new(Transport::default()); + let service = service(state.clone(), cache.clone(), transport); + service.refresh().await; assert_eq!( state.client_release(ClientKind::Desktop, TARGETS[0].platform, TARGETS[0].arch), before ); + cache.reject.store(false, Ordering::Relaxed); + service.refresh().await; + assert_eq!(cache.writes.load(Ordering::Relaxed), 8); + for target in TARGETS { + assert_eq!( + state + .client_release(ClientKind::Desktop, target.platform, target.arch) + .unwrap() + .codex_version, + "0.200.0" + ); + assert!( + state + .client_release_status(ClientKind::Desktop, target.platform, target.arch) + .1 + .is_none() + ); + } } diff --git a/docs/api.md b/docs/api.md index 4ed314b1d..71f8294ca 100644 --- a/docs/api.md +++ b/docs/api.md @@ -2468,13 +2468,14 @@ TUI 默认标识为 `codex-tui`，Exec 为 `codex_exec`，入口后缀使用同 因此 Desktop 预设保持默认即可；CLI 预设建议配置真实值，多 Key 可通过各自的画像覆盖差异化 `versionLag` 让 `latest` 模式跟随官方发布但滞后 N 个已观察版本采用，省略时采用最新版本。 -档位基于网关观察到的发布序列解析：启动种子与历次 24 小时检查的核验版本按时间倒序构成历史， +档位基于网关观察到的发布序列解析：启动种子与历次检查的核验版本按时间倒序构成历史， 取第 N 项；序列不足时回退最旧已核验版本，不会因此解析失败。历史仅保留在进程内， 重启后从启动种子与制品缓存重新积累 六套预设均支持自动更新：macOS Desktop 支持 arm64，Windows/Linux Desktop 及三套 CLI 支持 arm64、x86_64。 预设接口的 `automaticAvailable`、`reason` 表示当前组合的可用性；自定义架构可能使自动解析不可用。 -每 24 小时后台检查官方稳定发布，失败保留同组合上次有效版本；固定值不受后台更新影响。 +启动时立即检查官方稳定发布，之后每轮检查完成等待 1 小时再检查。 +版本变化后提取并更新画像，失败保留同组合上次有效版本，下一轮重试；固定值不受后台更新影响。 Desktop 的应用版本、Core 和构建号来自同一平台、架构的官方制品：macOS ZIP、Windows MSIX、Linux DEB。 Windows/Linux 通过 ETag 检查更新，未变化时复用已核验版本；CLI 依据官方 npm 稳定标签和对应平台依赖 @@ -2485,6 +2486,7 @@ Windows/Linux 通过 ETag 检查更新，未变化时复用已核验版本；CLI 完整自定义配置不携带官方制品核验时间。 客户端画像配置控制应用层请求字段，不切换操作系统的 TLS 实现。默认 HTTP 使用 native TLS， WebSocket 使用 rustls；配置自定义 CA 时 HTTP 也使用 rustls。TLS 指纹需按实际部署平台与传输路径核验。 +版本未变化时只更新 `checkedAt`，保留 `verifiedAt`；缓存续期不改写核验时间。 未完成本次启动检查时 `checkedAt` 为 `null`。非法或当前不可用的选择返回 `400`，保存失败不提交其他修改 Responses 的自动 Desktop Git 上下文与请求头使用同一画像快照：已有 `appServerVersion` 同步为 Core 版本， diff --git a/docs/architecture.md b/docs/architecture.md index 14347b603..b0ea32dff 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1297,6 +1297,10 @@ Worker 由各 Bundle 贡献、由 Host 统一监督： 以及账号错误保护恢复 worker（到期解除或探测恢复）、重置卡自动使用和公共 CPR 互联同步 - Provider：credential refresh、quota/catalog 健康、账号预热和官方版本/etag 检查 +OpenAI 客户端画像由 Provider 每小时检查发布元数据，发布标识变化后才提取对应制品版本并发布画像。 +同版本的有效制品缓存每 24 小时续期，保留 30 天；续期保留原核验时间，失败不推进成功写入游标。 +版本及缓存更新失败保留当前画像，下轮继续尝试；Host 只监督调度与取消。 + 互联同步使用账号出口的受管 HTTP 客户端，不跟随跳转；网络等待不持有数据库事务。 运行期排程、并发预算、连接失效和观测提交见 [CPR 互联观测](#cpr-互联观测)，失败遵守退避及上游 `Retry-After`