feat: 通过加密会话和冲突设置连接桌面 Sync

frontend/src-tauri/src/credentials.rs

@@ -506,6 +506,7 @@ impl CredentialBroker {
         self.unlocked.as_ref().ok_or("CREDENTIALS_LOCKED".into())
     }
     pub fn lock(&mut self) {
+        self.lock_epoch.fetch_add(1, Ordering::SeqCst);
         self.unlocked.take();
         self.ownership.take();
     }

frontend/src-tauri/src/lib.rs

@@ -9,6 +9,8 @@ mod runtime_compat;
 #[cfg(windows)]
 pub mod session_lock;
 #[cfg(feature = "desktop")]
+pub mod sync_auth;
+#[cfg(feature = "desktop")]
 pub mod sync_client;
 pub mod sync_inbox;
 pub mod sync_resolution;

frontend/src-tauri/src/main.rs

@@ -2,6 +2,9 @@
 
 //! 预览 Host 只开放本地文件命令；未接通的 AI / 同步 / 凭据能力明确返回不可用。
 
+mod sync_commands;
+use sync_commands::*;
+
 use base64::{engine::general_purpose::STANDARD as BASE64_STANDARD, Engine as _};
 use notesagent_host::core::CoreSupervisor;
 use notesagent_host::credentials::CredentialBroker;
@@ -18,6 +21,7 @@ use zeroize::Zeroizing;
 #[derive(Default)]
 struct Host {
     requests: Requests,
+    sync: Arc<sync_commands::Runtime>,
     workspace: Arc<Mutex<Option<Workspace>>>,
     recent: Mutex<Option<RecentVaultStore>>,
     core: Arc<Mutex<Option<CoreSupervisor>>>,
@@ -57,7 +61,7 @@ fn host_capabilities(host: State<'_, Host>) -> serde_json::Value {
         .ok()
         .and_then(|mut core| core.as_mut().map(|c| c.available()))
         .unwrap_or(false);
-    serde_json::json!({"protocol":1,"workspace":true,"core":ready,"sync":false,"credentials":true,"extensions":false,"release":"preview","product":"OpenNexus"})
+    serde_json::json!({"protocol":1,"workspace":true,"core":ready,"sync":true,"credentials":true,"extensions":false,"release":"preview","product":"OpenNexus"})
 }
 
 #[derive(serde::Serialize)]
@@ -568,6 +572,7 @@ fn workspace_choose(host: State<'_, Host>) -> Result<Option<RecentVault>, String
         .as_mut()
         .ok_or("HOST_NOT_READY")?
         .remember(&result)?;
+    host.sync.cancel();
     *guard = Some(workspace);
     Ok(Some(result))
 }
@@ -592,6 +597,7 @@ fn workspace_open(host: State<'_, Host>, path: String) -> Result<RecentVault, St
     }
     let workspace = Workspace::open(Path::new(&authorized.path)).map_err(|e| e.code)?;
     let result = info(&workspace);
+    host.sync.cancel();
     *guard = Some(workspace);
     Ok(result)
 }
@@ -617,6 +623,7 @@ fn workspace_revoke(host: State<'_, Host>) -> Result<(), String> {
             .ok_or("HOST_NOT_READY")?
             .revoke(&active.root)?;
     }
+    host.sync.cancel();
     *workspace = None;
     Ok(())
 }
@@ -800,6 +807,14 @@ fn main() {
                     }
                 }
             });
+            let sync_handle = app.handle().clone();
+            tauri::async_runtime::spawn(async move {
+                let mut interval = tokio::time::interval(Duration::from_secs(5));
+                loop {
+                    interval.tick().await;
+                    let _ = sync_commands::run(&sync_handle.state::<Host>(), false).await;
+                }
+            });
             Ok(())
         })
         .on_menu_event(|app, event| {
@@ -816,6 +831,16 @@ fn main() {
         })
         .invoke_handler(tauri::generate_handler![
             host_capabilities,
+            sync_login,
+            sync_vaults,
+            sync_create_vault,
+            sync_bind,
+            sync_unbind,
+            sync_pause,
+            sync_status,
+            sync_resolve,
+            sync_logout,
+            sync_run,
             credentials_status,
             credentials_unlock,
             credentials_lock,

frontend/src-tauri/src/sync_auth.rs

@@ -0,0 +1,221 @@
+//! Device-local Sync sessions. The serialized record never crosses IPC.
+use crate::{
+    credentials::{CredentialBroker, CredentialId, Scope},
+    sync_client::{Session, SyncClient, SyncError},
+    workspace::hash,
+};
+use serde::{Deserialize, Serialize};
+use std::{
+    future::Future,
+    sync::{atomic::Ordering, Arc, Mutex},
+    time::{Duration, SystemTime, UNIX_EPOCH},
+};
+use zeroize::Zeroizing;
+pub type Credentials = Arc<Mutex<Option<CredentialBroker>>>;
+type Result<T> = std::result::Result<T, SyncError>;
+#[derive(Serialize, Deserialize)]
+struct SavedSession {
+    endpoint: String,
+    account: String,
+    allow_test_http: bool,
+    expires_at: u64,
+    session: Session,
+}
+fn now() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs()
+}
+fn identity(endpoint: &str, account: &str) -> CredentialId {
+    CredentialId {
+        scope: Scope::Sync(hash(format!("{endpoint}\n{account}").as_bytes())),
+        id: "session".into(),
+    }
+}
+fn broker<T>(
+    credentials: &Credentials,
+    action: impl FnOnce(&mut CredentialBroker) -> std::result::Result<T, String>,
+) -> Result<T> {
+    let mut guard = credentials
+        .lock()
+        .map_err(|_| SyncError::new("HOST_BUSY"))?;
+    action(
+        guard
+            .as_mut()
+            .ok_or_else(|| SyncError::new("CREDENTIALS_UNAVAILABLE"))?,
+    )
+    .map_err(|e| SyncError::new(&e))
+}
+fn save(credentials: &Credentials, saved: &SavedSession) -> Result<()> {
+    let bytes = Zeroizing::new(
+        serde_json::to_vec(saved).map_err(|_| SyncError::new("SYNC_SESSION_INVALID"))?,
+    );
+    broker(credentials, |b| {
+        b.put(&identity(&saved.endpoint, &saved.account), bytes)
+    })
+}
+pub fn available(credentials: &Credentials, endpoint: &str, account: &str) -> Result<bool> {
+    broker(credentials, |b| {
+        b.resolve(
+            &identity(endpoint, account).scope,
+            &identity(endpoint, account),
+        )
+        .map(|v| v.is_some())
+    })
+}
+/// Dropping a guarded HTTP future closes the in-flight operation on any lock epoch change.
+pub async fn guarded<T>(
+    credentials: &Credentials,
+    future: impl Future<Output = Result<T>>,
+) -> Result<T> {
+    let (signal, epoch) = broker(credentials, |b| {
+        if b.is_locked() {
+            return Err("CREDENTIALS_LOCKED".into());
+        }
+        let signal = b.lock_signal();
+        let epoch = signal.load(Ordering::SeqCst);
+        Ok((signal, epoch))
+    })?;
+    tokio::pin!(future);
+    let mut interval = tokio::time::interval(Duration::from_millis(50));
+    loop {
+        tokio::select! {
+            biased;
+            _=interval.tick()=>{if signal.load(Ordering::SeqCst)!=epoch {return Err(SyncError::new("CREDENTIALS_LOCKED"));}},
+            result=&mut future=>{if signal.load(Ordering::SeqCst)!=epoch {return Err(SyncError::new("CREDENTIALS_LOCKED"));} return result;}
+        }
+    }
+}
+pub async fn login(
+    credentials: &Credentials,
+    endpoint: &str,
+    account: &str,
+    password: Zeroizing<String>,
+    device: &str,
+    allow_test_http: bool,
+) -> Result<String> {
+    if account.is_empty()
+        || account.len() > 128
+        || account.contains(['\n', '\r'])
+        || password.len() > 1024
+        || device.is_empty()
+        || device.len() > 100
+    {
+        return Err(SyncError::new("SYNC_LOGIN_INVALID"));
+    }
+    guarded(credentials, async {
+        let public = SyncClient::new(endpoint, Zeroizing::new(String::new()), allow_test_http)?;
+        public.handshake().await?;
+        let session = public.login(account, password, device).await?;
+        let endpoint = public.endpoint().to_owned();
+        save(
+            credentials,
+            &SavedSession {
+                endpoint: endpoint.clone(),
+                account: account.into(),
+                allow_test_http,
+                expires_at: now().saturating_add(session.expires_in),
+                session,
+            },
+        )?;
+        Ok(endpoint)
+    })
+    .await
+}
+/// The caller serializes refreshes with the coordinator gate.
+pub async fn client(
+    credentials: &Credentials,
+    endpoint: &str,
+    account: &str,
+    force_refresh: bool,
+) -> Result<SyncClient> {
+    let id = identity(endpoint, account);
+    let encoded = broker(credentials, |b| b.resolve(&id.scope, &id))?
+        .ok_or_else(|| SyncError::new("SYNC_LOGIN_REQUIRED"))?;
+    let mut saved: SavedSession =
+        serde_json::from_slice(&encoded).map_err(|_| SyncError::new("SYNC_SESSION_INVALID"))?;
+    if saved.endpoint != endpoint || saved.account != account {
+        return Err(SyncError::new("SYNC_SESSION_INVALID"));
+    }
+    if force_refresh || saved.expires_at <= now().saturating_add(30) {
+        let public = SyncClient::new(
+            endpoint,
+            Zeroizing::new(String::new()),
+            saved.allow_test_http,
+        )?;
+        let refreshed = guarded(credentials, public.refresh(&saved.session.refresh_token)).await?;
+        if refreshed.device_id != saved.session.device_id {
+            return Err(SyncError::new("SYNC_SESSION_INVALID"));
+        }
+        saved.expires_at = now().saturating_add(refreshed.expires_in);
+        saved.session = refreshed;
+        save(credentials, &saved)?;
+    }
+    SyncClient::new(
+        endpoint,
+        Zeroizing::new(saved.session.access_token.clone()),
+        saved.allow_test_http,
+    )
+}
+pub async fn logout(credentials: &Credentials, endpoint: &str, account: &str) -> Result<()> {
+    let client = client(credentials, endpoint, account, false).await?;
+    // A failed server revocation is reported; the encrypted record remains available for retry.
+    guarded(
+        credentials,
+        client.json(reqwest::Method::DELETE, "sync/v1/auth/sessions", None),
+    )
+    .await?;
+    broker(credentials, |b| b.delete(&identity(endpoint, account)))
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[tokio::test]
+    async fn lock_cancels_inflight_and_scopes_do_not_expose_tokens() {
+        let root = tempfile::tempdir().unwrap();
+        let mut b = CredentialBroker::new(root.path().join("credentials"));
+        b.unlock(Zeroizing::new(b"test-password-12345".to_vec()))
+            .unwrap();
+        let credentials = Arc::new(Mutex::new(Some(b)));
+        let saved = SavedSession {
+            endpoint: "https://sync.example/".into(),
+            account: "account".into(),
+            allow_test_http: false,
+            expires_at: now() + 900,
+            session: Session {
+                access_token: "private-access".into(),
+                refresh_token: "private-refresh".into(),
+                expires_in: 900,
+                device_id: "device".into(),
+            },
+        };
+        save(&credentials, &saved).unwrap();
+        assert!(available(&credentials, &saved.endpoint, &saved.account).unwrap());
+        assert!(!available(&credentials, "https://other.example/", &saved.account).unwrap());
+        assert!(!available(&credentials, &saved.endpoint, "other").unwrap());
+        let bytes = std::fs::read(root.path().join("credentials")).unwrap();
+        assert!(!bytes.windows(14).any(|v| v == b"private-access"));
+        let task_credentials = credentials.clone();
+        let pending = tokio::spawn(async move {
+            guarded(&task_credentials, async {
+                tokio::time::sleep(Duration::from_secs(30)).await;
+                Ok(())
+            })
+            .await
+        });
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        credentials.lock().unwrap().as_mut().unwrap().lock();
+        let result = tokio::time::timeout(Duration::from_secs(1), pending)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(result.unwrap_err().code, "CREDENTIALS_LOCKED");
+        assert_eq!(
+            available(&credentials, &saved.endpoint, &saved.account)
+                .unwrap_err()
+                .code,
+            "CREDENTIALS_LOCKED"
+        );
+    }
+}

frontend/src-tauri/src/sync_client.rs

@@ -21,7 +21,7 @@ pub struct SyncError {
     pub retry_after: Option<u64>,
 }
 impl SyncError {
-    fn new(code: &str) -> Self {
+    pub(crate) fn new(code: &str) -> Self {
         Self {
             code: code.into(),
             status: 0,
@@ -197,6 +197,19 @@ impl SyncClient {
         let value = self.json(Method::POST, "sync/v1/auth/sessions", Some(json!({"username":username,"password":password.as_str(),"device_name":device_name}))).await?;
         serde_json::from_value(value).map_err(|_| SyncError::new("SYNC_RESPONSE_INVALID"))
     }
+    pub fn endpoint(&self) -> &str {
+        self.endpoint.as_str()
+    }
+    pub async fn refresh(&self, refresh_token: &str) -> Result<Session> {
+        let value = self
+            .json(
+                Method::POST,
+                "sync/v1/auth/refresh",
+                Some(json!({"refresh_token":refresh_token})),
+            )
+            .await?;
+        serde_json::from_value(value).map_err(|_| SyncError::new("SYNC_RESPONSE_INVALID"))
+    }
     pub async fn handshake(&self) -> Result<()> {
         let result = self
             .json(Method::GET, "sync/v1/handshake?protocol=1", None)

frontend/src-tauri/src/sync_commands.rs

@@ -0,0 +1,358 @@
+//! Main-window commands. Every ongoing run is bound to one Workspace and one account.
+use super::{with_workspace, Host};
+use notesagent_host::{
+    sync_auth,
+    sync_client::{SyncError, WorkspaceAccess},
+    sync_state::Binding,
+};
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+use std::{
+    collections::HashMap,
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Mutex,
+    },
+    time::{Duration, Instant},
+};
+use tauri::State;
+use zeroize::Zeroizing;
+#[derive(Default)]
+pub struct Runtime {
+    gate: tokio::sync::Mutex<()>,
+    epoch: AtomicU64,
+    status: Mutex<HashMap<String, Progress>>,
+}
+#[derive(Default)]
+struct Progress {
+    running: bool,
+    error: Option<String>,
+    failures: u32,
+    retry: Option<Instant>,
+    halted: bool,
+}
+impl Runtime {
+    pub fn cancel(&self) {
+        self.epoch.fetch_add(1, Ordering::SeqCst);
+    }
+}
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Login {
+    endpoint: String,
+    account: String,
+    password: String,
+    device_name: String,
+    allow_test_http: bool,
+}
+#[derive(Serialize)]
+pub struct Connection {
+    endpoint: String,
+    account: String,
+}
+#[tauri::command]
+pub async fn sync_login(host: State<'_, Host>, request: Login) -> Result<Connection, String> {
+    let _guard = host.sync.gate.lock().await;
+    let endpoint = sync_auth::login(
+        &host.credentials,
+        &request.endpoint,
+        &request.account,
+        Zeroizing::new(request.password),
+        &request.device_name,
+        request.allow_test_http,
+    )
+    .await
+    .map_err(|e| e.code)?;
+    host.sync.status.lock().map_err(|_| "HOST_BUSY")?.clear();
+    Ok(Connection {
+        endpoint,
+        account: request.account,
+    })
+}
+#[tauri::command]
+pub async fn sync_vaults(
+    host: State<'_, Host>,
+    endpoint: String,
+    account: String,
+) -> Result<Value, String> {
+    let _guard = host.sync.gate.lock().await;
+    let client = sync_auth::client(&host.credentials, &endpoint, &account, false)
+        .await
+        .map_err(|e| e.code)?;
+    sync_auth::guarded(
+        &host.credentials,
+        client.json(reqwest::Method::GET, "sync/v1/vaults", None),
+    )
+    .await
+    .map_err(|e| e.code)
+}
+#[tauri::command]
+pub async fn sync_create_vault(
+    host: State<'_, Host>,
+    endpoint: String,
+    account: String,
+    name: String,
+) -> Result<Value, String> {
+    if name.trim().is_empty() || name.len() > 100 {
+        return Err("SYNC_NAME_INVALID".into());
+    }
+    let _guard = host.sync.gate.lock().await;
+    let client = sync_auth::client(&host.credentials, &endpoint, &account, false)
+        .await
+        .map_err(|e| e.code)?;
+    sync_auth::guarded(
+        &host.credentials,
+        client.json(
+            reqwest::Method::POST,
+            "sync/v1/vaults",
+            Some(json!({"name":name})),
+        ),
+    )
+    .await
+    .map_err(|e| e.code)
+}
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Bind {
+    vault_id: String,
+    endpoint: String,
+    account: String,
+    remote_vault: String,
+    mode: String,
+}
+#[tauri::command]
+pub async fn sync_bind(host: State<'_, Host>, request: Bind) -> Result<Binding, String> {
+    let _guard = host.sync.gate.lock().await;
+    let client = sync_auth::client(
+        &host.credentials,
+        &request.endpoint,
+        &request.account,
+        false,
+    )
+    .await
+    .map_err(|e| e.code)?;
+    if request.mode == "upload" {
+        sync_auth::guarded(
+            &host.credentials,
+            client.verify_empty(&request.remote_vault),
+        )
+        .await
+        .map_err(|e| e.code)?;
+    } else if request.mode == "download" {
+        // Verify account ownership before creating the durable binding.
+        let vaults = sync_auth::guarded(
+            &host.credentials,
+            client.json(reqwest::Method::GET, "sync/v1/vaults", None),
+        )
+        .await
+        .map_err(|e| e.code)?;
+        if !vaults["items"]
+            .as_array()
+            .is_some_and(|items| items.iter().any(|v| v["id"] == request.remote_vault))
+        {
+            return Err("SYNC_VAULT_DENIED".into());
+        }
+    } else {
+        return Err("SYNC_RECONCILIATION_REQUIRED".into());
+    }
+    with_workspace(&host, |ws| {
+        if ws.vault_id != request.vault_id {
+            return Err(notesagent_host::workspace::HostError::new("VAULT_CHANGED"));
+        }
+        if request.mode == "upload" {
+            ws.sync_bind_empty(&request.endpoint, &request.remote_vault, &request.account)
+        } else {
+            ws.sync_bind_download(&request.endpoint, &request.remote_vault, &request.account)
+        }
+    })
+}
+#[tauri::command]
+pub fn sync_unbind(host: State<'_, Host>, binding_id: String) -> Result<(), String> {
+    host.sync.cancel();
+    with_workspace(&host, |ws| ws.sync_unbind(&binding_id))
+}
+#[tauri::command]
+pub fn sync_pause(host: State<'_, Host>, binding_id: String, paused: bool) -> Result<(), String> {
+    host.sync.cancel();
+    with_workspace(&host, |ws| ws.sync_pause(&binding_id, paused))?;
+    host.sync
+        .status
+        .lock()
+        .map_err(|_| "HOST_BUSY")?
+        .remove(&binding_id);
+    Ok(())
+}
+#[tauri::command]
+pub fn sync_status(host: State<'_, Host>) -> Result<Value, String> {
+    let snapshot = with_workspace(&host, |ws| {
+        let binding = ws.sync_binding()?;
+        let paused = binding
+            .as_ref()
+            .map(|b| ws.sync_paused(&b.id))
+            .transpose()?
+            .unwrap_or(false);
+        let conflicts = binding
+            .as_ref()
+            .map(|b| ws.sync_conflicts(&b.id))
+            .transpose()?
+            .unwrap_or_default();
+        Ok((
+            ws.vault_id.clone(),
+            binding,
+            paused,
+            conflicts,
+            ws.pending_count()?,
+        ))
+    })?;
+    let (vault_id, binding, paused, conflicts, pending) = snapshot;
+    let credential_state = if let Some(b) = &binding {
+        sync_auth::available(&host.credentials, &b.endpoint, &b.account)
+            .map(|exists| {
+                if exists {
+                    "ready"
+                } else {
+                    "SYNC_LOGIN_REQUIRED"
+                }
+                .to_owned()
+            })
+            .unwrap_or_else(|e| e.code)
+    } else {
+        "unbound".into()
+    };
+    let statuses = host.sync.status.lock().map_err(|_| "HOST_BUSY")?;
+    let status = binding.as_ref().and_then(|b| statuses.get(&b.id));
+    Ok(
+        json!({"vault_id":vault_id,"binding":binding,"paused":paused,"pending":pending,"conflicts":conflicts,"credential_state":credential_state,"running":status.is_some_and(|s|s.running),"error":status.and_then(|s|s.error.as_ref()),"retry_in":status.and_then(|s|s.retry).map(|time|time.saturating_duration_since(Instant::now()).as_secs())}),
+    )
+}
+#[tauri::command]
+pub fn sync_resolve(
+    host: State<'_, Host>,
+    binding_id: String,
+    sequence: i64,
+    choice: String,
+    destination: String,
+    expected: String,
+) -> Result<(), String> {
+    with_workspace(&host, |ws| {
+        ws.sync_resolve(&binding_id, sequence, &choice, &destination, &expected)
+    })
+}
+#[tauri::command]
+pub async fn sync_logout(
+    host: State<'_, Host>,
+    endpoint: String,
+    account: String,
+) -> Result<(), String> {
+    host.sync.cancel();
+    let _guard = host.sync.gate.lock().await;
+    sync_auth::logout(&host.credentials, &endpoint, &account)
+        .await
+        .map_err(|e| e.code)
+}
+#[tauri::command]
+pub async fn sync_run(host: State<'_, Host>) -> Result<(), String> {
+    run(&host, true).await
+}
+pub async fn run(host: &Host, manual: bool) -> Result<(), String> {
+    let Ok(_guard) = host.sync.gate.try_lock() else {
+        return if manual {
+            Err("SYNC_BUSY".into())
+        } else {
+            Ok(())
+        };
+    };
+    let binding = with_workspace(host, |ws| ws.sync_binding())?.ok_or("SYNC_NOT_BOUND")?;
+    if with_workspace(host, |ws| ws.sync_paused(&binding.id))? {
+        return Err("SYNC_PAUSED".into());
+    }
+    {
+        let mut statuses = host.sync.status.lock().map_err(|_| "HOST_BUSY")?;
+        let state = statuses.entry(binding.id.clone()).or_default();
+        if !manual && (state.halted || state.retry.is_some_and(|v| v > Instant::now())) {
+            return Ok(());
+        }
+        state.running = true;
+        state.error = None;
+    }
+    let result = cycle(host, &binding).await;
+    let mut statuses = host.sync.status.lock().map_err(|_| "HOST_BUSY")?;
+    let state = statuses.entry(binding.id.clone()).or_default();
+    state.running = false;
+    match result {
+        Ok(()) => {
+            *state = Progress::default();
+            Ok(())
+        }
+        Err(error) => {
+            state.failures = if error.code == "CREDENTIALS_LOCKED" {
+                0
+            } else {
+                state.failures.saturating_add(1)
+            };
+            state.error = Some(error.code.clone());
+            state.halted = matches!(error.status, 401 | 403 | 413 | 426 | 507)
+                || matches!(
+                    error.code.as_str(),
+                    "PROTOCOL_INCOMPATIBLE" | "SYNC_LOGIN_REQUIRED"
+                );
+            state.retry = Some(
+                Instant::now()
+                    + Duration::from_secs(
+                        error
+                            .retry_after
+                            .unwrap_or(2u64.saturating_pow(state.failures.min(8)))
+                            .clamp(1, 3600),
+                    ),
+            );
+            Err(error.code)
+        }
+    }
+}
+async fn cycle(host: &Host, binding: &Binding) -> Result<(), SyncError> {
+    let epoch = host.sync.epoch.load(Ordering::SeqCst);
+    let work = async {
+        let client = sync_auth::client(
+            &host.credentials,
+            &binding.endpoint,
+            &binding.account,
+            false,
+        )
+        .await?;
+        let work = async {
+            client.handshake().await?;
+            for _ in 0..10 {
+                if client.pull_page(&host.workspace, binding).await? == 0 {
+                    break;
+                }
+            }
+            // Finish the fixed incoming window before freezing any new remote base.
+            if host
+                .workspace
+                .access(|ws| ws.sync_boundary(&binding.id))?
+                .is_some()
+            {
+                return Ok(());
+            }
+            for _ in 0..20 {
+                if !client.push_one(&host.workspace, binding).await? {
+                    break;
+                }
+            }
+            client.pull_page(&host.workspace, binding).await?;
+            Ok(())
+        };
+        sync_auth::guarded(&host.credentials, work).await
+    };
+    tokio::pin!(work);
+    let mut tick = tokio::time::interval(Duration::from_millis(50));
+    loop {
+        tokio::select! {biased;
+            _=tick.tick()=>{
+                if host.sync.epoch.load(Ordering::SeqCst)!=epoch {return Err(SyncError{code:"SYNC_CANCELLED".into(),status:0,retry_after:None});}
+                host.workspace.access(|ws|ws.sync_paused(&binding.id).map(|_|()))?;
+            },
+            result=&mut work=>return result,
+        }
+    }
+}

frontend/src-tauri/src/sync_inbox.rs

@@ -335,7 +335,10 @@ impl Workspace {
             .collect::<std::result::Result<Vec<_>, _>>()?;
         rows.into_iter().map(|(sequence,file_id,local_path,local_hash,remote)| {
             let remote: Value = serde_json::from_str(&remote).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?;
-            Ok(serde_json::json!({"sequence":sequence,"file_id":file_id,"local_path":local_path,"local_hash":local_hash,"remote":remote}))
+            let current_path=self.path_for_id(&file_id).unwrap_or_else(|_|local_path.clone());
+            let source=self.resolve(&current_path)?;
+            let current_hash=if source.is_file() {hash(&fs::read(source)?)} else {String::new()};
+            Ok(serde_json::json!({"sequence":sequence,"file_id":file_id,"local_path":local_path,"local_hash":local_hash,"current_path":current_path,"current_hash":current_hash,"remote":remote}))
         }).collect()
     }
 }

frontend/src-tauri/src/sync_resolution.rs

@@ -35,6 +35,7 @@ impl Workspace {
             let (path,remote): (String,String) = self.db.query_row("SELECT local_path,remote FROM sync_conflicts WHERE binding=?1 AND sequence=?2 AND state='open'",params![binding,sequence],|r| Ok((r.get(0)?,r.get(1)?)))?;
             let revision: RemoteRevision = serde_json::from_str(&remote)
                 .map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?;
+            let path = self.path_for_id(&revision.file_id).unwrap_or(path);
             let source = self.resolve(&path)?;
             let current = if source.is_file() {
                 self.sync_store_bytes(&fs::read(source)?)?

frontend/src-tauri/src/sync_state.rs

@@ -28,6 +28,23 @@ pub struct Job {
 }
 
 impl Workspace {
+    pub fn sync_paused(&self, binding: &str) -> Result<bool> {
+        self.check_binding(binding)?;
+        Ok(self
+            .db
+            .query_row(
+                "SELECT paused FROM sync_preferences WHERE binding=?1",
+                [binding],
+                |r| r.get(0),
+            )
+            .optional()?
+            .unwrap_or(false))
+    }
+    pub fn sync_pause(&mut self, binding: &str, paused: bool) -> Result<()> {
+        self.check_binding(binding)?;
+        self.db.execute("INSERT INTO sync_preferences VALUES (?1,?2) ON CONFLICT(binding) DO UPDATE SET paused=excluded.paused",params![binding,paused])?;
+        Ok(())
+    }
     pub fn sync_binding(&self) -> Result<Option<Binding>> {
         Ok(self.db.query_row("SELECT id,endpoint,remote_vault,account,cursor FROM sync_bindings WHERE state='active'", [], |r| {
             Ok(Binding { id:r.get(0)?, endpoint:r.get(1)?, remote_vault:r.get(2)?, account:r.get(3)?, cursor:r.get(4)? })

frontend/src-tauri/src/workspace.rs

@@ -18,7 +18,7 @@ pub struct HostError {
 pub type Result<T> = std::result::Result<T, HostError>;
 
 impl HostError {
-    pub(crate) fn new(code: &str) -> Self {
+    pub fn new(code: &str) -> Self {
         Self {
             code: code.into(),
             message: code.into(),
@@ -135,10 +135,10 @@ impl Workspace {
         let db = Connection::open(db_path)?;
         db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;")?;
         let version: i64 = db.query_row("PRAGMA user_version", [], |r| r.get(0))?;
-        if version > 5 {
+        if version > 6 {
             return Err(HostError::new("SCHEMA_INCOMPATIBLE"));
         }
-        if (1..5).contains(&version) {
+        if (1..6).contains(&version) {
             // Independent, complete SQLite backup before the schema ownership change.
             let backup = managed.join(format!("host-schema{version}-{}.sqlite3", Uuid::new_v4()));
             db.execute("VACUUM INTO ?1", [backup.to_string_lossy().as_ref()])?;
@@ -157,6 +157,7 @@ impl Workspace {
             CREATE TABLE IF NOT EXISTS sync_windows (binding TEXT PRIMARY KEY,boundary INTEGER NOT NULL);
             CREATE TABLE IF NOT EXISTS sync_inbox (binding TEXT NOT NULL,sequence INTEGER NOT NULL,revision TEXT NOT NULL,operation_id TEXT NOT NULL,rename_id TEXT NOT NULL,state TEXT NOT NULL,PRIMARY KEY(binding,sequence));
             CREATE TABLE IF NOT EXISTS sync_conflicts (binding TEXT NOT NULL,sequence INTEGER NOT NULL,file_id TEXT NOT NULL,local_path TEXT NOT NULL,local_hash TEXT NOT NULL,remote TEXT NOT NULL,state TEXT NOT NULL,PRIMARY KEY(binding,sequence));
+            CREATE TABLE IF NOT EXISTS sync_preferences (binding TEXT PRIMARY KEY,paused INTEGER NOT NULL DEFAULT 0);
             CREATE TABLE IF NOT EXISTS sync_resolutions (binding TEXT NOT NULL,sequence INTEGER NOT NULL,choice TEXT NOT NULL,destination TEXT NOT NULL,expected TEXT NOT NULL,operation_id TEXT NOT NULL,rename_id TEXT NOT NULL,copy_id TEXT NOT NULL,state TEXT NOT NULL,PRIMARY KEY(binding,sequence));")?;
         let has_origin: bool = db.query_row(
             "SELECT EXISTS(SELECT 1 FROM pragma_table_info('file_ops') WHERE name='origin')",
@@ -169,7 +170,7 @@ impl Workspace {
                 [],
             )?;
         }
-        db.execute_batch("PRAGMA user_version=5; COMMIT;")?;
+        db.execute_batch("PRAGMA user_version=6; COMMIT;")?;
         let vault_id: String = db
             .query_row("SELECT id FROM identity", [], |r| r.get(0))
             .optional()?