feat: 持久化 Sync 收件箱并在拉取时保留离线冲突

2026-09-08 15:38:10 +08:00
parent 2bc9ef84b9
commit 251d3bac7d
7 changed files with 705 additions and 17 deletions
@@ -10,6 +10,7 @@ mod runtime_compat;
 pub mod session_lock;
 #[cfg(feature = "desktop")]
 pub mod sync_client;
+pub mod sync_inbox;
 pub mod sync_state;
 pub mod workspace;
 pub mod workspace_broker;
@@ -259,6 +259,141 @@ impl SyncClient {
        workspace.access(|ws| ws.sync_ack(&job, &revision))?;
        Ok(true)
    }
+    pub async fn pull_page(
+        &self,
+        workspace: &impl WorkspaceAccess,
+        binding: &Binding,
+    ) -> Result<usize> {
+        if Url::parse(&binding.endpoint)
+            .ok()
+            .is_none_or(|url| url != self.endpoint)
+        {
+            return Err(SyncError::new("SYNC_BINDING_CHANGED"));
+        }
+        identifier(&binding.remote_vault)?;
+        workspace.access(|ws| {
+            while ws.sync_apply_pending(&binding.id)? {}
+            Ok(())
+        })?;
+        let (cursor, boundary) = workspace.access(|ws| {
+            ws.check_binding(&binding.id)?;
+            Ok((
+                ws.sync_binding()?
+                    .ok_or_else(|| crate::workspace::HostError::new("SYNC_BINDING_CHANGED"))?
+                    .cursor,
+                ws.sync_boundary(&binding.id)?,
+            ))
+        })?;
+        let mut path = format!(
+            "sync/v1/vaults/{}/changes?cursor={cursor}&limit=100",
+            binding.remote_vault
+        );
+        if let Some(end) = boundary {
+            path.push_str(&format!("&boundary={end}"));
+        }
+        let page = self.json(Method::GET, &path, None).await?;
+        let end = page["boundary"]
+            .as_i64()
+            .ok_or_else(|| SyncError::new("SYNC_RESPONSE_INVALID"))?;
+        let items = page["items"]
+            .as_array()
+            .filter(|items| items.len() <= 100)
+            .ok_or_else(|| SyncError::new("SYNC_RESPONSE_INVALID"))?;
+        if items.is_empty() {
+            if end != cursor {
+                return Err(SyncError::new("SYNC_RESPONSE_INVALID"));
+            }
+            return Ok(0);
+        }
+        workspace.access(|ws| ws.sync_set_boundary(&binding.id, end))?;
+        for (index, item) in items.iter().enumerate() {
+            let revision: crate::sync_inbox::RemoteRevision = serde_json::from_value(item.clone())
+                .map_err(|_| SyncError::new("SYNC_RESPONSE_INVALID"))?;
+            revision.validate(binding)?;
+            if revision.sequence != cursor + index as i64 + 1 || revision.sequence > end {
+                return Err(SyncError::new("SYNC_RESPONSE_INVALID"));
+            }
+            if revision.operation == "put" {
+                self.download(workspace, binding, &revision).await?;
+            }
+            workspace.access(|ws| {
+                ws.sync_stage(&binding.id, &revision)?;
+                ws.sync_apply_pending(&binding.id)?;
+                Ok(())
+            })?;
+        }
+        Ok(items.len())
+    }
+    async fn download(
+        &self,
+        workspace: &impl WorkspaceAccess,
+        binding: &Binding,
+        revision: &crate::sync_inbox::RemoteRevision,
+    ) -> Result<()> {
+        let digest = revision
+            .hash
+            .as_deref()
+            .ok_or_else(|| SyncError::new("SYNC_RESPONSE_INVALID"))?;
+        let target = workspace.access(|ws| {
+            ws.check_binding(&binding.id)?;
+            ws.sync_spool(digest)
+        })?;
+        if target.exists() {
+            let bytes = std::fs::read(&target)?;
+            if bytes.len() as i64 == revision.size && crate::workspace::hash(&bytes) == digest {
+                return Ok(());
+            }
+            return Err(SyncError::new("SYNC_SPOOL_CORRUPT"));
+        }
+        let url = self
+            .endpoint
+            .join(&format!(
+                "sync/v1/vaults/{}/objects/{digest}",
+                binding.remote_vault
+            ))
+            .map_err(|_| SyncError::new("SYNC_PATH_INVALID"))?;
+        let mut response = self
+            .client
+            .get(url)
+            .bearer_auth(self.token.as_str())
+            .send()
+            .await
+            .map_err(|_| SyncError::new("SYNC_NETWORK_ERROR"))?;
+        if !response.status().is_success() {
+            return Err(SyncError {
+                code: "SYNC_DOWNLOAD_FAILED".into(),
+                status: response.status().as_u16(),
+                retry_after: None,
+            });
+        }
+        let mut file = tempfile::NamedTempFile::new_in(
+            target
+                .parent()
+                .ok_or_else(|| SyncError::new("SYNC_SPOOL_FAILED"))?,
+        )?;
+        let mut hasher = Sha256::new();
+        let mut length = 0u64;
+        while let Some(chunk) = response
+            .chunk()
+            .await
+            .map_err(|_| SyncError::new("SYNC_NETWORK_ERROR"))?
+        {
+            length += chunk.len() as u64;
+            if length > revision.size as u64 {
+                return Err(SyncError::new("SYNC_OBJECT_CORRUPT"));
+            }
+            workspace.access(|ws| ws.check_binding(&binding.id))?;
+            std::io::Write::write_all(&mut file, &chunk)?;
+            hasher.update(&chunk);
+        }
+        if length != revision.size as u64 || format!("{:x}", hasher.finalize()) != digest {
+            return Err(SyncError::new("SYNC_OBJECT_CORRUPT"));
+        }
+        file.as_file().sync_all()?;
+        file.persist_noclobber(target)
+            .map_err(|_| SyncError::new("SYNC_SPOOL_FAILED"))?;
+        Ok(())
+    }
    async fn upload(
        &self,
        workspace: &impl WorkspaceAccess,
@@ -0,0 +1,397 @@
+//! Persist received revisions before Workspace writes; cursor advancement follows application.
+use crate::{
+    sync_state::{Binding, Job},
+    workspace::{hash, HostError, Result, Workspace},
+};
+use rusqlite::{params, OptionalExtension};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::{fs, io::Write};
+use uuid::Uuid;
+
+#[derive(Clone, Serialize, Deserialize)]
+pub struct RemoteRevision {
+    pub vault_id: String,
+    pub sequence: i64,
+    pub file_id: String,
+    pub base_revision: i64,
+    pub path: String,
+    pub operation: String,
+    pub hash: Option<String>,
+    pub size: i64,
+    pub operation_id: String,
+}
+
+impl RemoteRevision {
+    pub fn validate(&self, binding: &Binding) -> Result<()> {
+        if self.vault_id != binding.remote_vault
+            || self.sequence <= 0
+            || self.base_revision < 0
+            || self.base_revision >= self.sequence
+            || !(0..=104857600).contains(&self.size)
+            || self.file_id.len() < 16
+            || self.file_id.len() > 80
+            || !self
+                .file_id
+                .bytes()
+                .all(|b| b.is_ascii_alphanumeric() || b == b'-')
+            || !matches!(self.operation.as_str(), "put" | "delete")
+            || (self.operation == "delete" && (self.hash.is_some() || self.size != 0))
+            || (self.operation == "put"
+                && self.hash.as_ref().is_none_or(|h| {
+                    h.len() != 64
+                        || !h
+                            .bytes()
+                            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
+                }))
+        {
+            return Err(HostError::new("SYNC_RESPONSE_INVALID"));
+        }
+        Ok(())
+    }
+}
+
+impl Workspace {
+    pub fn sync_bind_download(
+        &mut self,
+        endpoint: &str,
+        remote_vault: &str,
+        account: &str,
+    ) -> Result<Binding> {
+        if self.sync_binding()?.is_some() {
+            return Err(HostError::new("SYNC_ALREADY_BOUND"));
+        }
+        if self
+            .scan()?
+            .iter()
+            .any(|entry| !entry.is_folder && !entry.deleted)
+        {
+            return Err(HostError::new("SYNC_RECONCILIATION_REQUIRED"));
+        }
+        let id = Uuid::new_v4().to_string();
+        let tx = self.db.transaction()?;
+        tx.execute(
+            "UPDATE outbox SET state='archived' WHERE state IN ('pending','queued')",
+            [],
+        )?;
+        tx.execute(
+            "INSERT INTO sync_bindings VALUES (?1,?2,?3,?4,'active',0)",
+            params![id, endpoint, remote_vault, account],
+        )?;
+        tx.commit()?;
+        self.sync_binding()?
+            .ok_or_else(|| HostError::new("DATABASE_ERROR"))
+    }
+    pub fn sync_boundary(&self, binding: &str) -> Result<Option<i64>> {
+        self.check_binding(binding)?;
+        Ok(self
+            .db
+            .query_row(
+                "SELECT boundary FROM sync_windows WHERE binding=?1",
+                [binding],
+                |r| r.get(0),
+            )
+            .optional()?)
+    }
+    pub fn sync_set_boundary(&self, binding: &str, boundary: i64) -> Result<()> {
+        self.check_binding(binding)?;
+        let cursor = self
+            .sync_binding()?
+            .ok_or_else(|| HostError::new("SYNC_BINDING_CHANGED"))?
+            .cursor;
+        if boundary < cursor
+            || self
+                .sync_boundary(binding)?
+                .is_some_and(|old| old != boundary)
+        {
+            return Err(HostError::new("SYNC_RESPONSE_INVALID"));
+        }
+        self.db.execute(
+            "INSERT OR IGNORE INTO sync_windows VALUES (?1,?2)",
+            params![binding, boundary],
+        )?;
+        Ok(())
+    }
+    pub fn sync_store_bytes(&self, bytes: &[u8]) -> Result<String> {
+        let digest = hash(bytes);
+        let path = self.sync_spool(&digest)?;
+        if path.exists() {
+            if fs::symlink_metadata(&path)?.file_type().is_symlink()
+                || hash(&fs::read(&path)?) != digest
+            {
+                return Err(HostError::new("SYNC_SPOOL_CORRUPT"));
+            }
+        } else {
+            let mut temp = tempfile::NamedTempFile::new_in(path.parent().unwrap())?;
+            temp.write_all(bytes)?;
+            temp.as_file().sync_all()?;
+            temp.persist_noclobber(path)
+                .map_err(|_| HostError::new("SYNC_SPOOL_FAILED"))?;
+        }
+        Ok(digest)
+    }
+    pub fn sync_stage(&self, binding: &str, revision: &RemoteRevision) -> Result<()> {
+        self.check_binding(binding)?;
+        let active = self
+            .sync_binding()?
+            .ok_or_else(|| HostError::new("SYNC_BINDING_CHANGED"))?;
+        revision.validate(&active)?;
+        self.resolve(&revision.path)?;
+        if revision.sequence != active.cursor + 1
+            || self
+                .sync_boundary(binding)?
+                .is_none_or(|end| revision.sequence > end)
+        {
+            return Err(HostError::new("SYNC_CURSOR_INVALID"));
+        }
+        if let Some(digest) = &revision.hash {
+            let bytes = fs::read(self.sync_spool(digest)?)?;
+            if bytes.len() as i64 != revision.size || hash(&bytes) != *digest {
+                return Err(HostError::new("SYNC_SPOOL_CORRUPT"));
+            }
+        }
+        let encoded =
+            serde_json::to_string(revision).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?;
+        let existing: Option<String> = self
+            .db
+            .query_row(
+                "SELECT revision FROM sync_inbox WHERE binding=?1 AND sequence=?2",
+                params![binding, revision.sequence],
+                |r| r.get(0),
+            )
+            .optional()?;
+        if existing.is_some_and(|value| value != encoded) {
+            return Err(HostError::new("SYNC_REVISION_CHANGED"));
+        }
+        self.db.execute(
+            "INSERT OR IGNORE INTO sync_inbox VALUES (?1,?2,?3,?4,?5,'pending')",
+            params![
+                binding,
+                revision.sequence,
+                encoded,
+                Uuid::new_v4().to_string(),
+                Uuid::new_v4().to_string()
+            ],
+        )?;
+        Ok(())
+    }
+    fn sync_finish(&mut self, binding: &str, revision: &RemoteRevision, state: &str) -> Result<()> {
+        self.check_binding(binding)?;
+        let tx = self.db.transaction()?;
+        let changed = tx.execute(
+            "UPDATE sync_bindings SET cursor=?2 WHERE id=?1 AND state='active' AND cursor=?3",
+            params![binding, revision.sequence, revision.sequence - 1],
+        )?;
+        if changed != 1 {
+            return Err(HostError::new("SYNC_CURSOR_INVALID"));
+        }
+        tx.execute("INSERT INTO sync_heads VALUES (?1,?2,?3,?4,?5) ON CONFLICT(binding,file_id) DO UPDATE SET revision=excluded.revision,path=excluded.path,hash=excluded.hash WHERE sync_heads.revision<excluded.revision", params![binding,revision.file_id,revision.sequence,revision.path,revision.hash.as_deref().unwrap_or("")])?;
+        tx.execute(
+            "UPDATE sync_inbox SET state=?3 WHERE binding=?1 AND sequence=?2",
+            params![binding, revision.sequence, state],
+        )?;
+        tx.execute(
+            "DELETE FROM sync_windows WHERE binding=?1 AND boundary=?2",
+            params![binding, revision.sequence],
+        )?;
+        tx.commit()?;
+        Ok(())
+    }
+    fn sync_preserve_conflict(
+        &mut self,
+        binding: &str,
+        revision: &RemoteRevision,
+        local_path: &str,
+    ) -> Result<()> {
+        let path = self.resolve(local_path)?;
+        let digest = if path.is_file() {
+            self.sync_store_bytes(&fs::read(path)?)?
+        } else {
+            String::new()
+        };
+        let remote =
+            serde_json::to_string(revision).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?;
+        let tx = self.db.transaction()?;
+        tx.execute("UPDATE sync_conflicts SET state='superseded' WHERE binding=?1 AND file_id=?2 AND state='open' AND sequence<?3", params![binding,revision.file_id,revision.sequence])?;
+        tx.execute(
+            "INSERT OR IGNORE INTO sync_conflicts VALUES (?1,?2,?3,?4,?5,?6,'open')",
+            params![
+                binding,
+                revision.sequence,
+                revision.file_id,
+                local_path,
+                digest,
+                remote
+            ],
+        )?;
+        tx.execute("UPDATE sync_jobs SET state='conflict' WHERE binding=?1 AND file_id=?2 AND state!='acked'", params![binding,revision.file_id])?;
+        tx.commit()?;
+        self.sync_finish(binding, revision, "conflict")
+    }
+    pub fn sync_apply_pending(&mut self, binding: &str) -> Result<bool> {
+        self.check_binding(binding)?;
+        let pending: Option<(String,String,String)> = self.db.query_row("SELECT revision,operation_id,rename_id FROM sync_inbox WHERE binding=?1 AND state='pending' ORDER BY sequence LIMIT 1", [binding], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?))).optional()?;
+        let Some((encoded, operation_id, rename_id)) = pending else {
+            return Ok(false);
+        };
+        let revision: RemoteRevision =
+            serde_json::from_str(&encoded).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?;
+        let own: Option<Job> = self.db.query_row("SELECT binding,operation_id,file_id,path,hash,size,operation,state,base_revision,upload_id FROM sync_jobs WHERE binding=?1 AND operation_id=?2", params![binding,revision.operation_id], |r| {
+            Ok(Job { binding:r.get(0)?,operation_id:r.get(1)?,file_id:r.get(2)?,path:r.get(3)?,hash:r.get(4)?,size:r.get(5)?,operation:r.get(6)?,state:r.get(7)?,base_revision:r.get(8)?,upload_id:r.get(9)? })
+        }).optional()?;
+        if let Some(job) = own {
+            self.sync_ack(
+                &job,
+                &serde_json::to_value(&revision)
+                    .map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?,
+            )?;
+            self.sync_finish(binding, &revision, "applied")?;
+            return Ok(true);
+        }
+        if self
+            .operation(&operation_id)?
+            .is_some_and(|value| value["state"] == "committed")
+        {
+            self.sync_finish(binding, &revision, "applied")?;
+            return Ok(true);
+        }
+        let local_path = self.path_for_id(&revision.file_id).ok();
+        let path = local_path.as_deref().unwrap_or(&revision.path);
+        let local = self.resolve(path)?;
+        let current = if local.is_file() {
+            hash(&fs::read(&local)?)
+        } else {
+            String::new()
+        };
+        let queued: bool = self.db.query_row("SELECT EXISTS(SELECT 1 FROM outbox WHERE file_id=?1 AND state IN ('pending','queued'))", [&revision.file_id], |r| r.get(0))?;
+        let head: Option<String> = self
+            .db
+            .query_row(
+                "SELECT hash FROM sync_heads WHERE binding=?1 AND file_id=?2",
+                params![binding, revision.file_id],
+                |r| r.get(0),
+            )
+            .optional()?;
+        let conflict = queued
+            || (local_path.is_none() && local.exists())
+            || (local_path.is_some() && head.as_deref() != Some(current.as_str()))
+            || (path != revision.path && self.resolve(&revision.path)?.exists());
+        if conflict {
+            self.sync_preserve_conflict(binding, &revision, path)?;
+            return Ok(true);
+        }
+        if revision.operation == "delete" {
+            if local_path.is_some() && local.exists() {
+                self.mutate_with_origin("delete", path, "", &current, &operation_id, "remote")?;
+            }
+        } else {
+            if let Some(previous) = local_path
+                .as_deref()
+                .filter(|previous| *previous != revision.path)
+            {
+                self.mutate_with_origin(
+                    "rename",
+                    previous,
+                    &revision.path,
+                    &current,
+                    &rename_id,
+                    "remote",
+                )?;
+            }
+            let content = fs::read(
+                self.sync_spool(
+                    revision
+                        .hash
+                        .as_deref()
+                        .ok_or_else(|| HostError::new("SYNC_RESPONSE_INVALID"))?,
+                )?,
+            )?;
+            self.write_with_identity(
+                &revision.path,
+                &current,
+                &content,
+                "remote",
+                &operation_id,
+                Some(&revision.file_id),
+            )?;
+        }
+        self.sync_finish(binding, &revision, "applied")?;
+        Ok(true)
+    }
+    pub fn sync_conflicts(&self, binding: &str) -> Result<Vec<Value>> {
+        self.check_binding(binding)?;
+        let mut statement = self.db.prepare("SELECT sequence,file_id,local_path,local_hash,remote FROM sync_conflicts WHERE binding=?1 AND state='open' ORDER BY sequence")?;
+        let rows = statement
+            .query_map([binding], |r| {
+                Ok((
+                    r.get::<_, i64>(0)?,
+                    r.get::<_, String>(1)?,
+                    r.get::<_, String>(2)?,
+                    r.get::<_, String>(3)?,
+                    r.get::<_, String>(4)?,
+                ))
+            })?
+            .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}))
+        }).collect()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn inbox_reopen_before_and_after_file_commit_never_advances_cursor_early() {
+        for committed in [false, true] {
+            let root = tempfile::tempdir().unwrap();
+            let mut ws = Workspace::open(root.path()).unwrap();
+            let binding = ws
+                .sync_bind_download("https://sync.example", "remote-vault", "account")
+                .unwrap();
+            let file_id = Uuid::new_v4().to_string();
+            let digest = ws.sync_store_bytes(b"remote-content").unwrap();
+            let revision = RemoteRevision {
+                vault_id: "remote-vault".into(),
+                sequence: 1,
+                file_id: file_id.clone(),
+                base_revision: 0,
+                path: "nested/a.md".into(),
+                operation: "put".into(),
+                hash: Some(digest),
+                size: 14,
+                operation_id: Uuid::new_v4().to_string(),
+            };
+            ws.sync_set_boundary(&binding.id, 1).unwrap();
+            ws.sync_stage(&binding.id, &revision).unwrap();
+            if committed {
+                let operation: String = ws
+                    .db
+                    .query_row("SELECT operation_id FROM sync_inbox", [], |r| r.get(0))
+                    .unwrap();
+                ws.write_with_identity(
+                    "nested/a.md",
+                    "",
+                    b"remote-content",
+                    "remote",
+                    &operation,
+                    Some(&file_id),
+                )
+                .unwrap();
+            }
+            assert_eq!(ws.sync_binding().unwrap().unwrap().cursor, 0);
+            drop(ws);
+            let mut ws = Workspace::open(root.path()).unwrap();
+            assert_eq!(ws.sync_binding().unwrap().unwrap().cursor, 0);
+            assert!(ws.sync_apply_pending(&binding.id).unwrap());
+            assert_eq!(ws.sync_binding().unwrap().unwrap().cursor, 1);
+            let document = ws.read("nested/a.md").unwrap();
+            assert_eq!(document.content, "remote-content");
+            assert_eq!(document.entry.file_id, file_id);
+            assert_eq!(document.entry.revision, 1);
+            assert_eq!(ws.pending_count().unwrap(), 0);
+            assert!(!ws.sync_apply_pending(&binding.id).unwrap());
+        }
+    }
+}
@@ -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 > 3 {
+        if version > 4 {
            return Err(HostError::new("SCHEMA_INCOMPATIBLE"));
        }
-        if (1..3).contains(&version) {
+        if (1..4).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()])?;
@@ -154,7 +154,21 @@ impl Workspace {
            CREATE UNIQUE INDEX IF NOT EXISTS sync_active ON sync_bindings(state) WHERE state='active';
            CREATE TABLE IF NOT EXISTS sync_jobs (binding TEXT NOT NULL,operation_id TEXT NOT NULL,file_id TEXT NOT NULL,path TEXT NOT NULL,hash TEXT NOT NULL,size INTEGER NOT NULL,operation TEXT NOT NULL,state TEXT NOT NULL,base_revision INTEGER,upload_id TEXT,remote_revision INTEGER,error TEXT,PRIMARY KEY(binding,operation_id));
            CREATE TABLE IF NOT EXISTS sync_heads (binding TEXT NOT NULL,file_id TEXT NOT NULL,revision INTEGER NOT NULL,path TEXT NOT NULL,hash TEXT NOT NULL,PRIMARY KEY(binding,file_id));
-            PRAGMA user_version=3; COMMIT;")?;
+            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));")?;
+        let has_origin: bool = db.query_row(
+            "SELECT EXISTS(SELECT 1 FROM pragma_table_info('file_ops') WHERE name='origin')",
+            [],
+            |r| r.get(0),
+        )?;
+        if !has_origin {
+            db.execute(
+                "ALTER TABLE file_ops ADD COLUMN origin TEXT NOT NULL DEFAULT 'local'",
+                [],
+            )?;
+        }
+        db.execute_batch("PRAGMA user_version=4; COMMIT;")?;
        let vault_id: String = db
            .query_row("SELECT id FROM identity", [], |r| r.get(0))
            .optional()?
@@ -359,6 +373,17 @@ impl Workspace {
        content: &[u8],
        origin: &str,
        operation_id: &str,
+    ) -> Result<Entry> {
+        self.write_with_identity(path, expected, content, origin, operation_id, None)
+    }
+    pub(crate) fn write_with_identity(
+        &mut self,
+        path: &str,
+        expected: &str,
+        content: &[u8],
+        origin: &str,
+        operation_id: &str,
+        identity: Option<&str>,
    ) -> Result<Entry> {
        if Uuid::parse_str(operation_id).is_err() {
            return Err(HostError::new("OPERATION_ID_INVALID"));
@@ -404,9 +429,18 @@ impl Workspace {
        if current != expected {
            return Err(HostError::new("REVISION_CONFLICT"));
        }
-        let file_id = self
-            .entry(path)?
-            .map_or_else(|| Uuid::new_v4().to_string(), |e| e.file_id);
+        let previous = self.entry(path)?;
+        if identity.is_some_and(|id| previous.as_ref().is_some_and(|entry| entry.file_id != id)) {
+            return Err(HostError::new("PATH_CONFLICT"));
+        }
+        let file_id = previous.map_or_else(
+            || {
+                identity
+                    .map(str::to_owned)
+                    .unwrap_or_else(|| Uuid::new_v4().to_string())
+            },
+            |entry| entry.file_id,
+        );
        let tx = self.db.transaction()?;
        tx.execute(
            "INSERT INTO operations VALUES (?1,?2,'pending',NULL)",
@@ -577,6 +611,7 @@ impl Workspace {
            destination,
            expected,
            &Uuid::new_v4().to_string(),
+            "local",
        )
    }

@@ -588,7 +623,19 @@ impl Workspace {
        expected: &str,
        operation_id: &str,
    ) -> Result<serde_json::Value> {
-        let id = self.prepare_file_op_with_id(kind, path, destination, expected, operation_id)?;
+        self.mutate_with_origin(kind, path, destination, expected, operation_id, "local")
+    }
+    pub(crate) fn mutate_with_origin(
+        &mut self,
+        kind: &str,
+        path: &str,
+        destination: &str,
+        expected: &str,
+        operation_id: &str,
+        origin: &str,
+    ) -> Result<serde_json::Value> {
+        let id =
+            self.prepare_file_op_with_id(kind, path, destination, expected, operation_id, origin)?;
        if self
            .operation(&id)?
            .is_some_and(|v| v["state"] == "committed")
@@ -609,12 +656,16 @@ impl Workspace {
        destination: &str,
        expected: &str,
        id: &str,
+        origin: &str,
    ) -> Result<String> {
-        if !matches!(kind, "rename" | "delete") || Uuid::parse_str(id).is_err() {
+        if !matches!(origin, "local" | "remote")
+            || !matches!(kind, "rename" | "delete")
+            || Uuid::parse_str(id).is_err()
+        {
            return Err(HostError::new("INVALID_OPERATION"));
        }
        let fingerprint = hash(
-            &serde_json::to_vec(&(kind, path, destination, expected))
+            &serde_json::to_vec(&(kind, path, destination, expected, origin))
                .map_err(|_| HostError::new("INVALID_OPERATION"))?,
        );
        let previous: Option<String> = self
@@ -657,24 +708,34 @@ impl Workspace {
            params![id, fingerprint],
        )?;
        tx.execute(
-            "INSERT INTO file_ops VALUES (?1,?2,?3,?4,?5,?6,'pending')",
-            params![id, kind, path, destination, expected, content],
+            "INSERT INTO file_ops VALUES (?1,?2,?3,?4,?5,?6,'pending',?7)",
+            params![id, kind, path, destination, expected, content, origin],
        )?;
        tx.commit()?;
        Ok(id.to_owned())
    }

    fn apply_file_op(&mut self, id: &str) -> Result<()> {
-        let (kind, path, destination, expected, content): (
+        let (kind, path, destination, expected, content, origin): (
            String,
            String,
            String,
            String,
            Vec<u8>,
+            String,
        ) = self.db.query_row(
-            "SELECT kind,path,destination,hash,content FROM file_ops WHERE id=?1",
+            "SELECT kind,path,destination,hash,content,origin FROM file_ops WHERE id=?1",
            [id],
-            |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)),
+            |r| {
+                Ok((
+                    r.get(0)?,
+                    r.get(1)?,
+                    r.get(2)?,
+                    r.get(3)?,
+                    r.get(4)?,
+                    r.get(5)?,
+                ))
+            },
        )?;
        let source = self.resolve(&path)?;
        let previous = self
@@ -724,13 +785,17 @@ impl Workspace {
                "UPDATE files SET path=?1,revision=revision+1 WHERE id=?2",
                params![destination, previous.file_id],
            )?;
-            tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,hash,'put',?2,'pending' FROM files WHERE id=?3", params![id,content,previous.file_id])?;
+            if origin == "local" {
+                tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,hash,'put',?2,'pending' FROM files WHERE id=?3", params![id,content,previous.file_id])?;
+            }
        } else {
            tx.execute(
                "UPDATE files SET deleted=1,revision=revision+1 WHERE id=?1",
                [&previous.file_id],
            )?;
-            tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,'','delete',X'','pending' FROM files WHERE id=?2", params![id,previous.file_id])?;
+            if origin == "local" {
+                tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,'','delete',X'','pending' FROM files WHERE id=?2", params![id,previous.file_id])?;
+            }
        }
        tx.execute("DELETE FROM file_ops WHERE id=?1", [id])?;
        let mut result = previous;