fix: preserve custom TUN fields and sync service provider caches

Limit GUI authority to saved TUN fields while preserving custom overrides.

Sync provider caches after service startup, reloads and updates.
Publish copies atomically with session, runtime and cache metadata checks.

src-tauri/src/core/manager/config.rs

@@ -345,6 +345,7 @@ impl CoreManager {
             ConfigApplication::ReloadFrom(staged) => match self.reload_config(&staged).await {
                 Ok(()) => {
                     tracing::Span::current().record("outcome", "staged");
+                    crate::core::service::request_runtime_provider_sync(timing::RUNTIME_PROVIDER_SYNC_DELAY);
                     return Ok(());
                 }
                 Err(err) => logging!(

src-tauri/src/core/manager/state.rs

@@ -276,6 +276,7 @@ impl CoreManager {
                         self.mark_core_ready();
                         self.core_started(RunningMode::Service);
                         self.restore_selected_nodes().await;
+                        service::request_runtime_provider_sync(crate::constants::timing::RUNTIME_PROVIDER_SYNC_DELAY);
                         return Ok(());
                     }
                     Err(e) => {
@@ -300,6 +301,7 @@ impl CoreManager {
             self.mark_core_ready();
             self.core_started(RunningMode::Service);
             self.restore_selected_nodes().await;
+            service::request_runtime_provider_sync(crate::constants::timing::RUNTIME_PROVIDER_SYNC_DELAY);
             Ok(())
         }
     }

src-tauri/src/core/runtime_bundle.rs

@@ -1,7 +1,7 @@
 use anyhow::{Context as _, Result, bail};
 use clash_verge_logging::{Type, logging};
 use clash_verge_service_ipc::{RemoteProvider, RuntimeAsset, RuntimeBundle};
-use serde_yaml_ng::Value;
+use serde_yaml_ng::{Mapping, Value};
 use std::collections::HashSet;
 use std::path::{Component, Path, PathBuf};
 
@@ -13,15 +13,47 @@ pub(crate) const GEO_ASSETS: &[&str] = &[
     "GeoSite.dat",
 ];
 
+/// Provider identity used by the runtime config and core API.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct RemoteProviderRef {
+    pub section: &'static str,
+    pub name: String,
+    pub provider: RemoteProvider,
+}
+
+struct RuntimeParts {
+    config: Value,
+    assets: Vec<RuntimeAsset>,
+    remote_providers: Vec<RemoteProviderRef>,
+}
+
 pub(crate) async fn collect_runtime_bundle(config_file: &Path, core_path: &Path) -> Result<RuntimeBundle> {
     let yaml = tokio::fs::read_to_string(config_file)
         .await
         .with_context(|| format!("failed to read runtime config {config_file:?}"))?;
-    let mut config: Value =
+    let config: Value =
         serde_yaml_ng::from_str(&yaml).with_context(|| format!("failed to parse runtime config {config_file:?}"))?;
     let config_root = config_file
         .parent()
         .ok_or_else(|| anyhow::anyhow!("runtime config has no parent directory"))?;
+    let parts = collect_runtime_parts(config, config_root)?;
+    Ok(RuntimeBundle {
+        yaml: serde_yaml_ng::to_string(&parts.config).context("failed to serialize service runtime config")?,
+        assets: parts.assets,
+        remote_providers: parts
+            .remote_providers
+            .into_iter()
+            .map(|declared| declared.provider)
+            .collect(),
+        core_path: core_path.to_string_lossy().into_owned(),
+    })
+}
+
+pub(crate) fn remote_providers_of(config: &Mapping, config_root: &Path) -> Result<Vec<RemoteProviderRef>> {
+    Ok(collect_runtime_parts(Value::Mapping(config.clone()), config_root)?.remote_providers)
+}
+
+fn collect_runtime_parts(mut config: Value, config_root: &Path) -> Result<RuntimeParts> {
     let config_root = std::fs::canonicalize(config_root)?;
     let mut assets = Vec::new();
     let mut destinations = HashSet::new();
@@ -53,11 +85,10 @@ pub(crate) async fn collect_runtime_bundle(config_file: &Path, core_path: &Path)
         }
     }
 
-    Ok(RuntimeBundle {
-        yaml: serde_yaml_ng::to_string(&config).context("failed to serialize service runtime config")?,
+    Ok(RuntimeParts {
+        config,
         assets,
         remote_providers,
-        core_path: core_path.to_string_lossy().into_owned(),
     })
 }
 
@@ -65,11 +96,11 @@ pub(crate) async fn collect_runtime_bundle(config_file: &Path, core_path: &Path)
 /// Remote URLs let the service decide whether a core download cache remains reusable.
 fn collect_provider_assets(
     config: &mut Value,
-    section: &str,
+    section: &'static str,
     config_root: &Path,
     destinations: &mut HashSet<String>,
     assets: &mut Vec<RuntimeAsset>,
-    remote_providers: &mut Vec<RemoteProvider>,
+    remote_providers: &mut Vec<RemoteProviderRef>,
 ) -> Result<()> {
     let Some(providers) = config
         .as_mapping_mut()
@@ -95,9 +126,9 @@ fn collect_provider_assets(
                 // Reserve remote destinations too; copied and downloaded files must not collide.
                 match remote_providers
                     .iter()
-                    .find(|declared| declared.destination == destination)
+                    .find(|declared| declared.provider.destination == destination)
                 {
-                    Some(declared) if declared.url == url => {}
+                    Some(declared) if declared.provider.url == url => {}
                     Some(_) => {
                         bail!("runtime provider destination {destination:?} is declared for two different sources")
                     }
@@ -105,9 +136,13 @@ fn collect_provider_assets(
                         if !destinations.insert(destination.clone()) {
                             bail!("runtime provider destination {destination:?} is claimed more than once");
                         }
-                        remote_providers.push(RemoteProvider {
-                            destination: destination.clone(),
-                            url,
+                        remote_providers.push(RemoteProviderRef {
+                            section,
+                            name: name.as_str().unwrap_or_default().to_owned(),
+                            provider: RemoteProvider {
+                                destination: destination.clone(),
+                                url,
+                            },
                         });
                     }
                 }

src-tauri/src/core/service.rs

@@ -1,7 +1,7 @@
-#[cfg(any(target_os = "macos", target_os = "windows"))]
 use crate::utils::dirs;
 use crate::{
-    config::Config,
+    config::{Config, runtime::IRuntime},
+    constants,
     core::{
         CoreManager,
         handle::Handle,
@@ -12,27 +12,29 @@ use crate::{
             OwnerRecoveryReason, OwnerSample, OwnerStep, OwnerWatch, PendingAction, RUN_STATE, ReadyWaitError,
             RunState, RunStateEnv, RunStateStore, ServiceHealth,
         },
-        runtime_bundle::collect_runtime_bundle,
+        runtime_bundle::{RemoteProviderRef, collect_runtime_bundle, remote_providers_of},
         tray::Tray,
     },
     process::AsyncHandler,
 };
-use anyhow::{Context as _, Result, bail};
+use anyhow::{Context as _, Result, anyhow, bail};
+use clash_verge_draft::Draft;
 use clash_verge_logging::{Type, logging};
 use clash_verge_service_ipc::{
-    MacosProxyConfig, OwnerSessionProof, ProxyApplyOutcome, RuntimeBundle, ServiceErrorCode, StageRuntimeOutcome,
-    StartClashRequest, WriterConfig,
+    MacosProxyConfig, OwnerCredentials, OwnerSessionProof, ProtocolInfo, ProxyApplyOutcome, RuntimeBundle,
+    RuntimeFileOutcome, RuntimeFileRequest, ServiceErrorCode, StageRuntimeOutcome, StartClashRequest, WriterConfig,
 };
 use once_cell::sync::Lazy;
 use parking_lot::Mutex;
 use std::{
     borrow::Cow,
+    collections::HashMap,
     env::current_exe,
     future::Future,
     path::{Path, PathBuf},
     process::Command as StdCommand,
-    sync::atomic::{AtomicU64, Ordering},
-    time::Duration,
+    sync::atomic::{AtomicBool, AtomicU64, Ordering},
+    time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
 static OWNER_MONITOR_GENERATION: AtomicU64 = AtomicU64::new(0);
@@ -44,6 +46,7 @@ static ACTIVE_SERVICE_SESSION: Lazy<Mutex<Option<ActiveServiceSession>>> = Lazy:
 struct ActiveServiceSession {
     proof: OwnerSessionProof,
     supports_runtime_staging: bool,
+    supports_runtime_file_read: bool,
 }
 
 fn generate_service_session_token() -> Result<String> {
@@ -68,21 +71,43 @@ pub(crate) fn active_service_supports_runtime_staging() -> bool {
         .is_some_and(|session| session.supports_runtime_staging)
 }
 
+pub(crate) fn active_service_supports_runtime_file_read() -> bool {
+    ACTIVE_SERVICE_SESSION
+        .lock()
+        .as_ref()
+        .is_some_and(|session| session.supports_runtime_file_read)
+}
+
 pub(crate) fn clear_active_service_session() {
     ACTIVE_SERVICE_SESSION.lock().take();
 }
 
-/// Probes staging support without failing startup when the fast path is unavailable.
+#[derive(Clone, Copy, Default)]
+struct ServiceCapabilities {
+    runtime_staging: bool,
+    runtime_file_read: bool,
+}
+
+impl ServiceCapabilities {
+    const fn of(info: &ProtocolInfo) -> Self {
+        Self {
+            runtime_staging: info.supports_runtime_staging(),
+            runtime_file_read: info.supports_runtime_file_read(),
+        }
+    }
+}
+
+/// Failed capability probes must not block startup.
 #[tracing::instrument(skip_all, level = "info", fields(supported = tracing::field::Empty))]
-async fn probe_runtime_staging_support() -> bool {
+async fn probe_service_capabilities() -> ServiceCapabilities {
     match clash_verge_service_ipc::get_version().await {
         Ok(response) if response.code == 0 => {
-            let supported = response
+            let capabilities = response
                 .data
                 .as_ref()
-                .is_some_and(clash_verge_service_ipc::ProtocolInfo::supports_runtime_staging);
-            tracing::Span::current().record("supported", supported);
-            supported
+                .map_or_else(ServiceCapabilities::default, ServiceCapabilities::of);
+            tracing::Span::current().record("supported", capabilities.runtime_staging);
+            capabilities
         }
         Ok(response) => {
             tracing::Span::current().record("supported", false);
@@ -93,7 +118,7 @@ async fn probe_runtime_staging_support() -> bool {
                 response.code,
                 response.message
             );
-            false
+            ServiceCapabilities::default()
         }
         Err(error) => {
             tracing::Span::current().record("supported", false);
@@ -102,7 +127,7 @@ async fn probe_runtime_staging_support() -> bool {
                 Type::Service,
                 "无法查询服务协议版本: {error:#}；配置变更将走重启路径"
             );
-            false
+            ServiceCapabilities::default()
         }
     }
 }
@@ -1060,14 +1085,15 @@ pub(super) async fn start_with_existing_service(config_file: &Path) -> Result<()
 
     let result = response.data.context("Clash Verge Service 未返回会话信息")?;
     tracing::Span::current().record("generation", result.session.generation);
-    let supports_runtime_staging = probe_runtime_staging_support().await;
-    tracing::Span::current().record("staging", supports_runtime_staging);
+    let capabilities = probe_service_capabilities().await;
+    tracing::Span::current().record("staging", capabilities.runtime_staging);
     *ACTIVE_SERVICE_SESSION.lock() = Some(ActiveServiceSession {
         proof: OwnerSessionProof {
             generation: result.session.generation,
             token: proposed_session_token,
         },
-        supports_runtime_staging,
+        supports_runtime_staging: capabilities.runtime_staging,
+        supports_runtime_file_read: capabilities.runtime_file_read,
     });
 
     // PAC follows the Running Mode; the caller opens it via `core_started(Service)`.
@@ -1138,16 +1164,424 @@ pub(crate) async fn get_clash_log_snapshot_by_service() -> Result<String> {
         bail!(response.message);
     }
     let encoded = response.data.context("服务未返回核心日志快照")?;
-    if encoded.len() % 2 != 0 {
-        bail!("服务返回了无效的核心日志快照");
-    }
-    let mut content = Vec::with_capacity(encoded.len() / 2);
-    for offset in (0..encoded.len()).step_by(2) {
-        content.push(u8::from_str_radix(&encoded[offset..offset + 2], 16).context("服务返回了无效的核心日志快照")?);
-    }
+    let content = decode_hex(&encoded).context("服务返回了无效的核心日志快照")?;
     Ok(String::from_utf8_lossy(&content).into_owned())
 }
 
+fn decode_hex(encoded: &str) -> Result<Vec<u8>> {
+    if !encoded.len().is_multiple_of(2) {
+        bail!("hex payload has an odd length");
+    }
+    (0..encoded.len())
+        .step_by(2)
+        .map(|offset| u8::from_str_radix(&encoded[offset..offset + 2], 16).context("hex payload is malformed"))
+        .collect()
+}
+
+static PROVIDER_SYNC_QUEUED: AtomicBool = AtomicBool::new(false);
+static PROVIDER_SYNC_SERIAL: Lazy<tokio::sync::Mutex<()>> = Lazy::new(|| tokio::sync::Mutex::new(()));
+static SYNC_TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0);
+const RUNTIME_PROVIDER_SYNC_ATTEMPTS: u32 = 4;
+const CONTENT_COMPARE_CHUNK: usize = 64 * 1024;
+
+pub(crate) fn request_runtime_provider_sync(delay: Duration) {
+    if PROVIDER_SYNC_QUEUED.swap(true, Ordering::AcqRel) {
+        return;
+    }
+    AsyncHandler::spawn(move || async move {
+        tokio::time::sleep(delay).await;
+        for attempt in 1..=RUNTIME_PROVIDER_SYNC_ATTEMPTS {
+            let outcome = {
+                let _serial = PROVIDER_SYNC_SERIAL.lock().await;
+                if attempt == 1 {
+                    PROVIDER_SYNC_QUEUED.store(false, Ordering::Release);
+                }
+                sync_runtime_providers_by_service().await
+            };
+            match outcome {
+                Ok(ProviderSync { pending: 0, .. }) => return,
+                Ok(ProviderSync { pending, .. }) if attempt < RUNTIME_PROVIDER_SYNC_ATTEMPTS => {
+                    logging!(
+                        info,
+                        Type::Service,
+                        "{pending} provider caches are not ready yet; retrying"
+                    );
+                }
+                Ok(ProviderSync { pending, .. }) => {
+                    logging!(warn, Type::Service, "{pending} provider caches were not synced");
+                    return;
+                }
+                Err(error) => {
+                    logging!(
+                        warn,
+                        Type::Service,
+                        "failed to sync provider caches from the service: {error:#}"
+                    );
+                    return;
+                }
+            }
+            tokio::time::sleep(constants::timing::RUNTIME_PROVIDER_SYNC_RETRY_DELAY).await;
+        }
+    });
+}
+
+#[derive(Debug, Default)]
+struct ProviderSync {
+    synced: usize,
+    pending: usize,
+}
+
+struct FetchedCache {
+    declared: RemoteProviderRef,
+    temp: PathBuf,
+    len: u64,
+    mtime_ns: Option<u64>,
+}
+
+/// Core load timestamps keyed by provider section and name.
+type ProviderLoadTimes = HashMap<(&'static str, String), Option<u64>>;
+
+#[tracing::instrument(level = "info", fields(synced = tracing::field::Empty, pending = tracing::field::Empty))]
+async fn sync_runtime_providers_by_service() -> Result<ProviderSync> {
+    if !active_service_supports_runtime_file_read() {
+        return Ok(ProviderSync::default());
+    }
+    let credentials = current_owner_credentials()?;
+    let session = active_service_session()?;
+    let app_dir = dirs::app_home_dir()?;
+    let runtime = Config::runtime().await;
+    let mut outcome = ProviderSync::default();
+
+    let loaded_before = provider_load_times().await;
+    let mut fetched = Vec::new();
+    for declared in applied_remote_providers(&runtime, &app_dir)? {
+        let (temp, file) = match create_sync_temp(&app_dir.join(&declared.provider.destination)).await {
+            Ok(created) => created,
+            Err(error) => {
+                logging!(warn, Type::Service, "{error:#}");
+                outcome.pending += 1;
+                continue;
+            }
+        };
+        match fetch_runtime_file(&credentials, &session, &declared.provider.destination, file).await {
+            Ok(Fetch::Complete { len, mtime_ns }) => fetched.push(FetchedCache {
+                declared,
+                temp,
+                len,
+                mtime_ns,
+            }),
+            Ok(Fetch::NotReady) => {
+                outcome.pending += 1;
+                remove_temp(&temp).await;
+            }
+            Ok(Fetch::SessionLost) => {
+                remove_temp(&temp).await;
+                discard_fetched(fetched).await;
+                bail!("the service session ended while provider caches were being read");
+            }
+            Err(error) => {
+                logging!(
+                    warn,
+                    Type::Service,
+                    "provider cache {} was not read: {error:#}",
+                    declared.provider.destination
+                );
+                outcome.pending += 1;
+                remove_temp(&temp).await;
+            }
+        }
+    }
+
+    let (loaded_before, loaded_after) = match (loaded_before, provider_load_times().await) {
+        (Ok(before), Ok(after)) => (before, after),
+        (Err(error), _) | (_, Err(error)) => {
+            logging!(warn, Type::Service, "provider caches were not synced: {error:#}");
+            outcome.pending += fetched.len();
+            discard_fetched(fetched).await;
+            return Ok(outcome);
+        }
+    };
+    let mut ready = Vec::new();
+    for cache in fetched {
+        let key = (cache.declared.section, cache.declared.name.clone());
+        let updated_at = loaded_after.get(&key).copied().flatten();
+        let complete = loaded_before.get(&key) == loaded_after.get(&key)
+            && finished_loading(cache.mtime_ns, updated_at)
+            && match recheck_identity(&credentials, &session, &cache).await {
+                Ok(Fetch::Complete { .. }) => true,
+                Ok(Fetch::NotReady) | Err(_) => false,
+                Ok(Fetch::SessionLost) => {
+                    remove_temp(&cache.temp).await;
+                    discard_fetched(ready).await;
+                    bail!("the service session ended while provider caches were being verified");
+                }
+            };
+        if complete {
+            ready.push(cache);
+        } else {
+            outcome.pending += 1;
+            remove_temp(&cache.temp).await;
+        }
+    }
+
+    outcome.synced = publish_fetched(&app_dir, &runtime, &session, ready).await?;
+    tracing::Span::current().record("synced", outcome.synced);
+    tracing::Span::current().record("pending", outcome.pending);
+    Ok(outcome)
+}
+
+// The on-disk YAML may describe a rejected apply; use the committed runtime.
+fn applied_remote_providers(runtime: &Draft<IRuntime>, app_dir: &Path) -> Result<Vec<RemoteProviderRef>> {
+    let applied = runtime.data_arc();
+    match applied.config.as_ref() {
+        Some(config) => remote_providers_of(config, app_dir),
+        None => Ok(Vec::new()),
+    }
+}
+
+// Hold the lifecycle lock through publication to exclude mode and config switches.
+async fn publish_fetched(
+    app_dir: &Path,
+    runtime: &Draft<IRuntime>,
+    session: &OwnerSessionProof,
+    fetched: Vec<FetchedCache>,
+) -> Result<usize> {
+    let manager = CoreManager::global();
+    let _lifecycle = manager.lifecycle_lock.lock().await;
+    let unchanged = matches!(*manager.get_running_mode(), RunningMode::Service)
+        && active_service_session().is_ok_and(|current| current == *session);
+    let declared = if unchanged {
+        match applied_remote_providers(runtime, app_dir) {
+            Ok(declared) => declared,
+            Err(error) => {
+                discard_fetched(fetched).await;
+                return Err(error);
+            }
+        }
+    } else {
+        Vec::new()
+    };
+    let mut synced = 0;
+    for cache in fetched {
+        let target = app_dir.join(&cache.declared.provider.destination);
+        if !declared.contains(&cache.declared) || same_contents(&cache.temp, &target).await {
+            remove_temp(&cache.temp).await;
+            continue;
+        }
+        match tokio::fs::rename(&cache.temp, &target).await {
+            Ok(()) => synced += 1,
+            Err(error) => {
+                logging!(
+                    warn,
+                    Type::Service,
+                    "provider cache {} was not published: {error}",
+                    cache.declared.provider.destination
+                );
+                remove_temp(&cache.temp).await;
+            }
+        }
+    }
+    Ok(synced)
+}
+
+enum Fetch {
+    Complete { len: u64, mtime_ns: Option<u64> },
+    NotReady,
+    SessionLost,
+}
+
+// mihomo rewrites caches in place; reject changing or recently modified files.
+async fn fetch_runtime_file(
+    credentials: &OwnerCredentials,
+    session: &OwnerSessionProof,
+    destination: &str,
+    mut file: tokio::fs::File,
+) -> Result<Fetch> {
+    use tokio::io::AsyncWriteExt as _;
+
+    let mut written = 0_u64;
+    let mut identity = None;
+    loop {
+        let (hex, len, mtime_ns) = match read_chunk(credentials, session, destination, written).await? {
+            Some(chunk) => chunk,
+            None => return Ok(Fetch::SessionLost),
+        };
+        let Some((hex, len, mtime_ns)) = hex.map(|hex| (hex, len, mtime_ns)) else {
+            return Ok(Fetch::NotReady);
+        };
+        if len == 0 || !has_settled(mtime_ns) || *identity.get_or_insert((len, mtime_ns)) != (len, mtime_ns) {
+            return Ok(Fetch::NotReady);
+        }
+        if written == len {
+            file.flush().await?;
+            return Ok(Fetch::Complete { len, mtime_ns });
+        }
+        let chunk = decode_hex(&hex).context("服务返回了无效的运行时文件")?;
+        if chunk.is_empty() {
+            bail!("the service returned an empty chunk before the end of the file");
+        }
+        file.write_all(&chunk).await?;
+        written += chunk.len() as u64;
+        if written > len {
+            return Ok(Fetch::NotReady);
+        }
+    }
+}
+
+async fn recheck_identity(
+    credentials: &OwnerCredentials,
+    session: &OwnerSessionProof,
+    cache: &FetchedCache,
+) -> Result<Fetch> {
+    let destination = &cache.declared.provider.destination;
+    Ok(match read_chunk(credentials, session, destination, cache.len).await? {
+        None => Fetch::SessionLost,
+        Some((Some(hex), len, mtime_ns)) if hex.is_empty() && (len, mtime_ns) == (cache.len, cache.mtime_ns) => {
+            Fetch::Complete { len, mtime_ns }
+        }
+        Some(_) => Fetch::NotReady,
+    })
+}
+
+/// Outer `None` means session loss; missing hex means an absent file.
+async fn read_chunk(
+    credentials: &OwnerCredentials,
+    session: &OwnerSessionProof,
+    destination: &str,
+    offset: u64,
+) -> Result<Option<(Option<String>, u64, Option<u64>)>> {
+    let request = RuntimeFileRequest {
+        destination: destination.to_owned(),
+        offset,
+    };
+    let response = clash_verge_service_ipc::read_runtime_file(credentials, session, &request)
+        .await
+        .context("无法连接到Clash Verge Service")?;
+    if response.code == ServiceErrorCode::NotActive as u16
+        || response.code == ServiceErrorCode::StaleOwnerSession as u16
+    {
+        return Ok(None);
+    }
+    if response.code > 0 {
+        bail!(response.message);
+    }
+    Ok(Some(match response.data.context("服务未返回运行时文件")? {
+        RuntimeFileOutcome::Absent => (None, 0, None),
+        RuntimeFileOutcome::Chunk { hex, len, mtime_ns } => (Some(hex), len, mtime_ns),
+    }))
+}
+
+// updatedAt changes after cache writes; compare it before and after readback.
+async fn provider_load_times() -> Result<ProviderLoadTimes> {
+    let mihomo = Handle::mihomo();
+    let (rules, proxies) = tokio::join!(mihomo.get_rule_providers(), mihomo.get_proxy_providers());
+    let rules = rules.map_err(|error| anyhow!("failed to query rule providers: {error}"))?;
+    let proxies = proxies.map_err(|error| anyhow!("failed to query proxy providers: {error}"))?;
+    let mut loaded = ProviderLoadTimes::new();
+    for (name, provider) in rules.providers {
+        loaded.insert(("rule-providers", name), epoch_nanos(&provider.updated_at));
+    }
+    for (name, provider) in proxies.providers {
+        loaded.insert(
+            ("proxy-providers", name),
+            provider.updated_at.as_deref().and_then(epoch_nanos),
+        );
+    }
+    Ok(loaded)
+}
+
+fn epoch_nanos(rfc3339: &str) -> Option<u64> {
+    let parsed = chrono::DateTime::parse_from_rfc3339(rfc3339).ok()?;
+    u64::try_from(parsed.timestamp_nanos_opt()?).ok()
+}
+
+// Allow timestamp granularity; missing load times fall back to settling.
+// Windows may retain the old mtime until a writer closes its handle.
+fn finished_loading(mtime_ns: Option<u64>, updated_at_ns: Option<u64>) -> bool {
+    match (mtime_ns, updated_at_ns) {
+        (Some(mtime), Some(updated_at)) => {
+            u128::from(mtime) <= u128::from(updated_at) + constants::timing::RUNTIME_PROVIDER_SETTLE.as_nanos()
+        }
+        _ => true,
+    }
+}
+
+fn has_settled(mtime_ns: Option<u64>) -> bool {
+    let Some(mtime_ns) = mtime_ns else {
+        return true;
+    };
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_or(0, |since_epoch| since_epoch.as_nanos());
+    now.saturating_sub(u128::from(mtime_ns)) >= constants::timing::RUNTIME_PROVIDER_SETTLE.as_nanos()
+}
+
+// Exclusive creation protects existing files from truncation and cleanup.
+async fn create_sync_temp(target: &Path) -> Result<(PathBuf, tokio::fs::File)> {
+    if let Some(parent) = target.parent() {
+        tokio::fs::create_dir_all(parent).await?;
+    }
+    let name = target
+        .file_name()
+        .map(|name| name.to_string_lossy())
+        .unwrap_or_default();
+    for _ in 0..8 {
+        let sequence = SYNC_TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
+        let temp = target.with_file_name(format!(".{name}.sync-{}-{sequence}.tmp", std::process::id()));
+        match tokio::fs::OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .open(&temp)
+            .await
+        {
+            Ok(file) => return Ok((temp, file)),
+            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
+            Err(error) => return Err(error).with_context(|| format!("failed to create {}", temp.display())),
+        }
+    }
+    bail!("no free temporary name beside {}", target.display())
+}
+
+async fn same_contents(temp: &Path, target: &Path) -> bool {
+    use tokio::io::AsyncReadExt as _;
+
+    let (Ok(mut left), Ok(mut right)) = (tokio::fs::File::open(temp).await, tokio::fs::File::open(target).await) else {
+        return false;
+    };
+    let (Ok(left_meta), Ok(right_meta)) = (left.metadata().await, right.metadata().await) else {
+        return false;
+    };
+    if left_meta.len() != right_meta.len() {
+        return false;
+    }
+    let mut remaining = left_meta.len();
+    let mut left_buf = vec![0_u8; CONTENT_COMPARE_CHUNK];
+    let mut right_buf = vec![0_u8; CONTENT_COMPARE_CHUNK];
+    while remaining > 0 {
+        let take = usize::try_from(remaining).map_or(CONTENT_COMPARE_CHUNK, |rest| rest.min(CONTENT_COMPARE_CHUNK));
+        if left.read_exact(&mut left_buf[..take]).await.is_err()
+            || right.read_exact(&mut right_buf[..take]).await.is_err()
+        {
+            return false;
+        }
+        if left_buf[..take] != right_buf[..take] {
+            return false;
+        }
+        remaining -= take as u64;
+    }
+    true
+}
+
+async fn remove_temp(temp: &Path) {
+    let _ = tokio::fs::remove_file(temp).await;
+}
+
+async fn discard_fetched(fetched: Vec<FetchedCache>) {
+    for cache in fetched {
+        remove_temp(&cache.temp).await;
+    }
+}
+
 /// 通过服务停止core
 #[tracing::instrument(skip_all, level = "info", fields(code = tracing::field::Empty, outcome = tracing::field::Empty))]
 pub(super) async fn stop_core_by_service() -> Result<()> {