feat: 暂存工作区载荷并捕获外部 Sync 变更

docs/development/OpenNexus生产化实施进度-2026-09-08.md

@@ -68,3 +68,14 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写
 - Host 提供每 5 秒串行同步，先完成固定拉取窗口再发送本地作业；SQLite schema 6 保存暂停偏好。网络失败使用有界退避并尊重 Retry-After；认证、权限、协议等错误停止自动重试并显示。绑定变化/暂停取消旧轮次。
 - 真实本地 HTTP 服务验证 Stronghold 会话存储、强制刷新、关闭重开保险库后复用、退出后旧令牌 401。新增 UI 测试验证密码清空、测试 HTTP 显式传参、确认取消和冲突 CAS 参数。Rust 全目标 43 项与前端全量 95 文件/508 项通过，正式 TypeScript 两项目检查通过。前端测试仍输出既有 KaTeX quirks/localhost:3000 连接警告，没有失败项。
 - 尚未完成：后台轮询真实桌面实机验收、初始非空合并预览、外部文件变化捕获、附件流式持久化及全故障矩阵、数据分类和不同身份同路径解决。不能据此标记所有 S/E 验收 ID 已通过。
+
+
+## 增量：外部编辑捕获与大附件崩溃续传
+
+- Workspace schema 7 将新 write/rename/delete journal 和 outbox 的正文替换为不可变 payload 引用，先 fsync 文件再记录摘要/长度；兼容旧内联正文恢复与上传迁移。100 MiB 不再写入 SQLite 正文列。暂不执行历史 payload GC。
+- 独立 sync_observed 保存上次提交快照；文件树/编辑器读取更新 files 缓存不会吞掉外部变更。后台捕获外部新增、编辑、删除，采用快照校验且不向外部编辑器写回。重复扫描不新增 outbox；删除捕获支持重启恢复。
+- 当前文件分类允许 Markdown 和 attachments 下常用附件类型，排除隐藏目录、已知缓存/日志/模型目录和可执行包扩展名；普通 JSON/配置不会直接复制。旧/新待发作业均检查分类，未支持的远端类型拒绝应用并保留游标。默认任务/Skill/配置等逻辑记录的白名单尚待实现，不作为 S-08 完整证据。
+- 真实本地 HTTP 100 MiB 测试每 10 MiB 在服务端写入后、响应前强杀独立 Rust 客户端，共 10 次。恢复进程复用同一设备会话、查询服务端 offset，最终上传/下载长度、SHA-256、file_id 一致且远端落盘无回流；两个临时 Vault 的元数据文件均小于 5 MiB。最初测试错误地重新登录生成不同设备，按协议无法复用原上传，现已改为通过 stdin 传递同一受控测试会话。
+- 此测试尚未覆盖完整 S-02 的每个 pull 边界 kill 20 轮，也不代表四并发服务 RSS 或生产 MinIO 性能验收。外部 rename 的稳定身份识别和目录大规模扫描优化继续实施。
+
+- 本增量 Rust desktop 全目标 46 项通过、1 个受父测试驱动的独立进程辅助入口标记 ignored（父测试实际运行并强杀该入口 10 次）；Clippy `-D warnings` 通过。

frontend/src-tauri/src/lib.rs

@@ -2,6 +2,7 @@
 
 pub mod core;
 pub mod credentials;
+mod payloads;
 pub mod recent;
 #[cfg(feature = "desktop")]
 pub mod request_lifecycle;
@@ -12,6 +13,7 @@ pub mod session_lock;
 pub mod sync_auth;
 #[cfg(feature = "desktop")]
 pub mod sync_client;
+pub mod sync_discovery;
 pub mod sync_inbox;
 pub mod sync_resolution;
 pub mod sync_state;

frontend/src-tauri/src/payloads.rs

@@ -0,0 +1,87 @@
+//! Immutable payloads are fsynced before any SQLite reference becomes visible.
+use crate::workspace::{hash, HostError, Result, Workspace};
+use rusqlite::{params, OptionalExtension};
+use sha2::{Digest, Sha256};
+use std::{
+    fs,
+    io::{Read, Write},
+    path::Path,
+};
+impl Workspace {
+    pub(crate) fn store_payload(&self, operation: &str, content: &[u8]) -> Result<()> {
+        let digest = hash(content);
+        let target = self.sync_spool(&digest)?;
+        if target.exists() {
+            verify(&target, &digest, content.len() as u64)?;
+        } else {
+            let mut temp = tempfile::NamedTempFile::new_in(target.parent().unwrap())?;
+            temp.write_all(content)?;
+            temp.as_file().sync_all()?;
+            temp.persist_noclobber(&target)
+                .map_err(|_| HostError::new("SYNC_SPOOL_FAILED"))?;
+            #[cfg(unix)]
+            fs::File::open(target.parent().unwrap())?.sync_all()?;
+        }
+        let old: Option<(String, i64)> = self
+            .db
+            .query_row(
+                "SELECT hash,size FROM payloads WHERE operation_id=?1",
+                [operation],
+                |r| Ok((r.get(0)?, r.get(1)?)),
+            )
+            .optional()?;
+        if old.is_some_and(|v| v != (digest.clone(), content.len() as i64)) {
+            return Err(HostError::new("OPERATION_PAYLOAD_CONFLICT"));
+        }
+        self.db.execute(
+            "INSERT OR IGNORE INTO payloads VALUES (?1,?2,?3)",
+            params![operation, digest, content.len() as i64],
+        )?;
+        Ok(())
+    }
+    pub(crate) fn payload(&self, operation: &str, legacy: &[u8]) -> Result<Vec<u8>> {
+        let Some((digest, size)) = self.payload_ref(operation)? else {
+            return Ok(legacy.to_vec());
+        };
+        let target = self.sync_spool(&digest)?;
+        verify(&target, &digest, size as u64)?;
+        Ok(fs::read(target)?)
+    }
+    pub(crate) fn payload_ref(&self, operation: &str) -> Result<Option<(String, i64)>> {
+        Ok(self
+            .db
+            .query_row(
+                "SELECT hash,size FROM payloads WHERE operation_id=?1",
+                [operation],
+                |r| Ok((r.get(0)?, r.get(1)?)),
+            )
+            .optional()?)
+    }
+}
+pub(crate) fn verify(path: &Path, digest: &str, size: u64) -> Result<()> {
+    let meta = fs::symlink_metadata(path)?;
+    if meta.file_type().is_symlink() || !meta.is_file() || meta.len() != size {
+        return Err(HostError::new("SYNC_SPOOL_CORRUPT"));
+    }
+    #[cfg(windows)]
+    {
+        use std::os::windows::fs::MetadataExt;
+        if meta.file_attributes() & 0x400 != 0 {
+            return Err(HostError::new("UNSAFE_PATH"));
+        }
+    }
+    let mut stream = fs::File::open(path)?;
+    let mut hasher = Sha256::new();
+    let mut buffer = vec![0; 1024 * 1024];
+    loop {
+        let count = stream.read(&mut buffer)?;
+        if count == 0 {
+            break;
+        }
+        hasher.update(&buffer[..count]);
+    }
+    if format!("{:x}", hasher.finalize()) != digest {
+        return Err(HostError::new("SYNC_SPOOL_CORRUPT"));
+    }
+    Ok(())
+}

frontend/src-tauri/src/sync_commands.rs

@@ -321,6 +321,7 @@ async fn cycle(host: &Host, binding: &Binding) -> Result<(), SyncError> {
         .await?;
         let work = async {
             client.handshake().await?;
+            host.workspace.access(|ws| ws.sync_discover(&binding.id))?;
             for _ in 0..10 {
                 if client.pull_page(&host.workspace, binding).await? == 0 {
                     break;

frontend/src-tauri/src/sync_discovery.rs

@@ -0,0 +1,247 @@
+//! Reconcile externally edited files against committed snapshots, never against UI read caches.
+use crate::workspace::{hash, HostError, Result, Workspace};
+use rusqlite::{params, OptionalExtension};
+use std::{collections::HashSet, fs, io::Read, path::Path};
+use uuid::Uuid;
+/// File transport only. Logical records receive their own versioned whitelist separately.
+pub fn allowed(path: &str) -> bool {
+    let parts: Vec<_> = path.split('/').collect();
+    if parts.iter().any(|part| {
+        part.starts_with('.')
+            || matches!(
+                part.to_ascii_lowercase().as_str(),
+                "node_modules" | "target" | "__pycache__" | "cache" | "logs" | "models"
+            )
+    }) {
+        return false;
+    }
+    let extension = Path::new(path)
+        .extension()
+        .and_then(|v| v.to_str())
+        .unwrap_or("")
+        .to_ascii_lowercase();
+    extension == "md"
+        || (parts
+            .first()
+            .is_some_and(|v| v.eq_ignore_ascii_case("attachments"))
+            && matches!(
+                extension.as_str(),
+                "png"
+                    | "jpg"
+                    | "jpeg"
+                    | "gif"
+                    | "webp"
+                    | "svg"
+                    | "pdf"
+                    | "mp3"
+                    | "wav"
+                    | "m4a"
+                    | "ogg"
+                    | "flac"
+                    | "mp4"
+                    | "webm"
+                    | "mov"
+                    | "txt"
+                    | "csv"
+                    | "docx"
+                    | "xlsx"
+                    | "pptx"
+                    | "bin"
+            ))
+}
+impl Workspace {
+    pub(crate) fn sync_paths(&self) -> Result<Vec<String>> {
+        fn walk(ws: &Workspace, directory: &Path, paths: &mut Vec<String>) -> Result<()> {
+            for item in fs::read_dir(directory)? {
+                let item = item?;
+                let name = item.file_name().to_string_lossy().into_owned();
+                if name.starts_with('.')
+                    || matches!(
+                        name.to_ascii_lowercase().as_str(),
+                        "node_modules" | "target" | "__pycache__" | "cache" | "logs" | "models"
+                    )
+                {
+                    continue;
+                }
+                let path = item
+                    .path()
+                    .strip_prefix(&ws.root)
+                    .map_err(|_| HostError::new("UNSAFE_PATH"))?
+                    .to_string_lossy()
+                    .replace('\\', "/");
+                let resolved = ws.resolve(&path)?;
+                if item.file_type()?.is_dir() {
+                    walk(ws, &resolved, paths)?;
+                } else if allowed(&path) && item.file_type()?.is_file() {
+                    paths.push(path);
+                }
+            }
+            Ok(())
+        }
+        let mut paths = Vec::new();
+        walk(self, &self.root, &mut paths)?;
+        paths.sort();
+        Ok(paths)
+    }
+    pub fn sync_discover(&mut self, binding: &str) -> Result<usize> {
+        self.check_binding(binding)?;
+        let paths = self.sync_paths()?;
+        let mut seen = HashSet::new();
+        let mut changes = 0;
+        for path in paths {
+            let target = self.resolve(&path)?;
+            if fs::metadata(&target)?.len() > 104857600 {
+                return Err(HostError::new("FILE_TOO_LARGE"));
+            }
+            let mut bytes = Vec::new();
+            fs::File::open(&target)?
+                .take(104857601)
+                .read_to_end(&mut bytes)?;
+            if bytes.len() > 104857600 {
+                return Err(HostError::new("FILE_TOO_LARGE"));
+            }
+            let digest = hash(&bytes);
+            let previous = self.entry(&path)?;
+            let observed: Option<(String, String, bool)> = if let Some(entry) = &previous {
+                self.db
+                    .query_row(
+                        "SELECT path,hash,deleted FROM sync_observed WHERE file_id=?1",
+                        [&entry.file_id],
+                        |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
+                    )
+                    .optional()?
+            } else {
+                None
+            };
+            if let Some(entry) = &previous {
+                seen.insert(entry.file_id.clone());
+            }
+            if observed
+                .as_ref()
+                .is_some_and(|v| v == &(path.clone(), digest.clone(), false))
+            {
+                continue;
+            }
+            // Confirm the snapshot without writing back over an external editor.
+            crate::payloads::verify(&target, &digest, bytes.len() as u64)?;
+            let operation = Uuid::new_v4().to_string();
+            self.store_payload(&operation, &bytes)?;
+            let file_id = previous.map_or_else(|| Uuid::new_v4().to_string(), |e| e.file_id);
+            let tx = self.db.transaction()?;
+            tx.execute("INSERT INTO files VALUES (?1,?2,?3,1,0) ON CONFLICT(path) DO UPDATE SET hash=excluded.hash,revision=files.revision+1,deleted=0",params![file_id,path,digest])?;
+            tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,hash,'put',X'','pending' FROM files WHERE id=?2",params![operation,file_id])?;
+            tx.execute("INSERT INTO sync_observed VALUES (?1,?2,?3,0) ON CONFLICT(file_id) DO UPDATE SET path=excluded.path,hash=excluded.hash,deleted=0",params![file_id,path,digest])?;
+            tx.commit()?;
+            seen.insert(file_id);
+            changes += 1;
+        }
+        let previous = {
+            let mut statement = self
+                .db
+                .prepare("SELECT file_id,path FROM sync_observed WHERE deleted=0")?;
+            let rows = statement
+                .query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?
+                .collect::<std::result::Result<Vec<_>, _>>()?;
+            rows
+        };
+        for (file_id, path) in previous {
+            if seen.contains(&file_id) || !allowed(&path) || self.resolve(&path)?.exists() {
+                continue;
+            }
+            let operation = Uuid::new_v4().to_string();
+            let tx = self.db.transaction()?;
+            tx.execute(
+                "UPDATE files SET deleted=1,revision=revision+1 WHERE id=?1",
+                [&file_id],
+            )?;
+            tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,'','delete',X'','pending' FROM files WHERE id=?2",params![operation,file_id])?;
+            tx.execute(
+                "UPDATE sync_observed SET deleted=1 WHERE file_id=?1",
+                [&file_id],
+            )?;
+            tx.commit()?;
+            changes += 1;
+        }
+        Ok(changes)
+    }
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn external_edits_after_ui_reads_are_queued_once_and_deletions_survive_restart() {
+        let root = tempfile::tempdir().unwrap();
+        let mut ws = Workspace::open(root.path()).unwrap();
+        let binding = ws
+            .sync_bind_empty("https://sync.example", "remote", "account")
+            .unwrap();
+        fs::write(root.path().join("a.md"), b"one").unwrap();
+        ws.read("a.md").unwrap();
+        ws.scan().unwrap();
+        assert_eq!(ws.sync_discover(&binding.id).unwrap(), 1);
+        assert_eq!(ws.sync_discover(&binding.id).unwrap(), 0);
+        fs::write(root.path().join("a.md"), b"two").unwrap();
+        ws.read("a.md").unwrap();
+        assert_eq!(ws.sync_discover(&binding.id).unwrap(), 1);
+        fs::remove_file(root.path().join("a.md")).unwrap();
+        drop(ws);
+        let mut ws = Workspace::open(root.path()).unwrap();
+        assert_eq!(ws.sync_discover(&binding.id).unwrap(), 1);
+        assert_eq!(ws.sync_discover(&binding.id).unwrap(), 0);
+        assert_eq!(ws.pending_count().unwrap(), 3);
+        let inline: i64 = ws
+            .db
+            .query_row(
+                "SELECT COALESCE(SUM(length(content)),0) FROM outbox",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(inline, 0);
+    }
+    #[test]
+    fn file_classification_excludes_credentials_caches_and_executable_packages() {
+        for path in [
+            ".env",
+            ".ainote/credentials.md",
+            ".git/readme.md",
+            "models/model.md",
+            "logs/trace.md",
+            "node_modules/package/readme.md",
+            "attachments/tool.exe",
+            "attachments/plugin.zip",
+            "settings.json",
+        ] {
+            assert!(!allowed(path), "{path}");
+        }
+        for path in [
+            "notes/a.md",
+            "attachments/movie.mp4",
+            "attachments/image.png",
+            "attachments/fixture.bin",
+        ] {
+            assert!(allowed(path), "{path}");
+        }
+    }
+    #[test]
+    fn prohibited_workspace_outbox_is_retained_but_never_sent() {
+        let root = tempfile::tempdir().unwrap();
+        let mut ws = Workspace::open(root.path()).unwrap();
+        ws.write("provider-settings.json", "", b"fixture-secret", "local")
+            .unwrap();
+        let binding = ws
+            .sync_bind_empty("https://sync.example", "remote", "account")
+            .unwrap();
+        assert!(ws.sync_next(&binding.id).unwrap().is_none());
+        assert_eq!(ws.pending_count().unwrap(), 0);
+        assert_eq!(
+            ws.read("provider-settings.json").unwrap().content,
+            "fixture-secret"
+        );
+        let state: String = ws
+            .db
+            .query_row("SELECT state FROM outbox", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(state, "excluded");
+    }
+}

frontend/src-tauri/src/sync_inbox.rs

@@ -47,6 +47,9 @@ impl RemoteRevision {
         {
             return Err(HostError::new("SYNC_RESPONSE_INVALID"));
         }
+        if !crate::sync_discovery::allowed(&self.path) {
+            return Err(HostError::new("SYNC_CLASS_UNSUPPORTED"));
+        }
         Ok(())
     }
 }
@@ -61,11 +64,7 @@ impl Workspace {
         if self.sync_binding()?.is_some() {
             return Err(HostError::new("SYNC_ALREADY_BOUND"));
         }
-        if self
+        if !self.sync_paths()?.is_empty() {
-            .scan()?
-            .iter()
-            .any(|entry| !entry.is_folder && !entry.deleted)
-        {
             return Err(HostError::new("SYNC_RECONCILIATION_REQUIRED"));
         }
         let id = Uuid::new_v4().to_string();

frontend/src-tauri/src/sync_resolution.rs

@@ -184,6 +184,10 @@ impl Workspace {
             "UPDATE files SET deleted=1,revision=revision+1 WHERE id=?1",
             [&revision.file_id],
         )?;
+        tx.execute(
+            "UPDATE sync_observed SET deleted=1 WHERE file_id=?1",
+            [&revision.file_id],
+        )?;
         let entry: Entry = tx.query_row(
             "SELECT id,path,hash,revision FROM files WHERE id=?1",
             [&revision.file_id],

frontend/src-tauri/src/sync_state.rs

@@ -1,8 +1,8 @@
 //! Durable queue state. Network code never invents a remote base from a local revision.
-use crate::workspace::{hash, HostError, Result, Workspace};
+use crate::workspace::{HostError, Result, Workspace};
 use rusqlite::{params, OptionalExtension};
 use serde::{Deserialize, Serialize};
-use std::{fs, io::Write, path::PathBuf};
+use std::{fs, path::PathBuf};
 use uuid::Uuid;
 
 #[derive(Clone, Serialize, Deserialize)]
@@ -147,28 +147,31 @@ impl Workspace {
             let Some((operation_id, file_id, path, digest, operation, content)) = pending else {
                 break;
             };
-            if operation == "put" {
+            if !crate::sync_discovery::allowed(&path) {
-                if hash(&content) != digest {
+                self.db.execute(
-                    return Err(HostError::new("SYNC_SPOOL_CORRUPT"));
+                    "UPDATE outbox SET state='excluded' WHERE operation_id=?1",
-                }
+                    [&operation_id],
-                let target = self.sync_spool(&digest)?;
+                )?;
-                if target.exists() {
+                continue;
-                    if fs::symlink_metadata(&target)?.file_type().is_symlink()
+            }
-                        || hash(&fs::read(&target)?) != digest
+            let size = if operation == "put" {
-                    {
+                if self.payload_ref(&operation_id)?.is_none() {
+                    self.store_payload(&operation_id, &content)?;
+                }
+                let (stored, size) = self
+                    .payload_ref(&operation_id)?
+                    .ok_or_else(|| HostError::new("SYNC_SPOOL_CORRUPT"))?;
+                if stored != digest {
                     return Err(HostError::new("SYNC_SPOOL_CORRUPT"));
                 }
+                crate::payloads::verify(&self.sync_spool(&digest)?, &digest, size as u64)?;
+                size
             } else {
-                    let mut temp = tempfile::NamedTempFile::new_in(target.parent().unwrap())?;
+                0
-                    temp.write_all(&content)?;
+            };
-                    temp.as_file().sync_all()?;
-                    temp.persist_noclobber(target)
-                        .map_err(|_| HostError::new("SYNC_SPOOL_FAILED"))?;
-                }
-            }
             let tx = self.db.transaction()?;
             tx.execute("INSERT OR IGNORE INTO sync_jobs VALUES (?1,?2,?3,?4,?5,?6,?7,'pending',NULL,NULL,NULL,NULL)",
-                params![binding,operation_id,file_id,path,digest,content.len() as i64,operation])?;
+                params![binding,operation_id,file_id,path,digest,size,operation])?;
             tx.execute(
                 "UPDATE outbox SET state='queued',content=X'' WHERE operation_id=?1",
                 [&operation_id],
@@ -179,9 +182,16 @@ impl Workspace {
     }
     pub fn sync_next(&self, binding: &str) -> Result<Option<Job>> {
         self.check_binding(binding)?;
-        Ok(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 state NOT IN ('acked','archived','conflict') AND NOT EXISTS (SELECT 1 FROM sync_conflicts c WHERE c.binding=sync_jobs.binding AND c.file_id=sync_jobs.file_id AND c.state='open') ORDER BY rowid LIMIT 1", [binding], |r| {
+        let 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 state NOT IN ('acked','archived','conflict') AND NOT EXISTS (SELECT 1 FROM sync_conflicts c WHERE c.binding=sync_jobs.binding AND c.file_id=sync_jobs.file_id AND c.state='open') ORDER BY rowid LIMIT 1", [binding], |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()?)
+        }).optional()?;
+        if job
+            .as_ref()
+            .is_some_and(|job| !crate::sync_discovery::allowed(&job.path))
+        {
+            return Err(HostError::new("SYNC_CLASS_UNSUPPORTED"));
+        }
+        Ok(job)
     }
     fn check_job(&self, job: &Job) -> Result<()> {
         self.check_binding(&job.binding)?;
@@ -258,6 +268,7 @@ impl Workspace {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::workspace::hash;
     #[test]
     fn queue_uses_remote_bases_and_keeps_retry_payload_across_restart() {
         let root = tempfile::tempdir().unwrap();

frontend/src-tauri/src/workspace.rs

@@ -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 > 6 {
+        if version > 7 {
             return Err(HostError::new("SCHEMA_INCOMPATIBLE"));
         }
-        if (1..6).contains(&version) {
+        if (1..7).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,8 @@ 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 payloads (operation_id TEXT PRIMARY KEY,hash TEXT NOT NULL,size INTEGER NOT NULL);
+            CREATE TABLE IF NOT EXISTS sync_observed (file_id TEXT PRIMARY KEY,path TEXT NOT NULL,hash TEXT NOT NULL,deleted INTEGER NOT NULL);
             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(
@@ -170,7 +172,10 @@ impl Workspace {
                 [],
             )?;
         }
-        db.execute_batch("PRAGMA user_version=6; COMMIT;")?;
+        if version < 7 {
+            db.execute_batch("INSERT OR IGNORE INTO sync_observed SELECT f.id,COALESCE((SELECT o.path FROM outbox o WHERE o.file_id=f.id AND o.state IN ('pending','queued') ORDER BY rowid DESC LIMIT 1),(SELECT h.path FROM sync_heads h JOIN sync_bindings b ON h.binding=b.id WHERE h.file_id=f.id AND b.state='active'),f.path),COALESCE((SELECT o.hash FROM outbox o WHERE o.file_id=f.id AND o.state IN ('pending','queued') ORDER BY rowid DESC LIMIT 1),(SELECT h.hash FROM sync_heads h JOIN sync_bindings b ON h.binding=b.id WHERE h.file_id=f.id AND b.state='active'),f.hash),f.deleted FROM files f;")?;
+        }
+        db.execute_batch("PRAGMA user_version=7; COMMIT;")?;
         let vault_id: String = db
             .query_row("SELECT id FROM identity", [], |r| r.get(0))
             .optional()?
@@ -222,7 +227,7 @@ impl Workspace {
         Ok(path)
     }
 
-    fn entry(&self, path: &str) -> Result<Option<Entry>> {
+    pub(crate) fn entry(&self, path: &str) -> Result<Option<Entry>> {
         Ok(self
             .db
             .query_row(
@@ -477,6 +482,7 @@ impl Workspace {
             },
             |entry| entry.file_id,
         );
+        self.store_payload(operation_id, content)?;
         let tx = self.db.transaction()?;
         tx.execute(
             "INSERT INTO operations VALUES (?1,?2,'pending',NULL)",
@@ -484,7 +490,14 @@ impl Workspace {
         )?;
         tx.execute(
             "INSERT INTO journal VALUES (?1,?2,?3,?4,?5,?6,'pending')",
-            params![operation_id, file_id, path, expected, content, origin],
+            params![
+                operation_id,
+                file_id,
+                path,
+                expected,
+                b"".as_slice(),
+                origin
+            ],
         )?;
         tx.commit()?;
         self.apply_journal(operation_id, &file_id, path, expected, content, origin)?;
@@ -534,11 +547,14 @@ impl Workspace {
             #[cfg(unix)]
             File::open(parent)?.sync_all()?;
         }
+        // Upgrade legacy inline journal payloads before publishing an outbox reference.
+        self.store_payload(operation_id, content)?;
         // 文件成功但 DB 未提交时，重启凭 journal 补齐同一 operation_id，避免丢 outbox。
         let tx = self.db.transaction()?;
         tx.execute("INSERT INTO files VALUES (?1,?2,?3,1,0) ON CONFLICT(path) DO UPDATE SET hash=excluded.hash,revision=files.revision+1,deleted=0", params![file_id,path,digest])?;
+        tx.execute("INSERT INTO sync_observed VALUES (?1,?2,?3,0) ON CONFLICT(file_id) DO UPDATE SET path=excluded.path,hash=excluded.hash,deleted=0",params![file_id,path,digest])?;
         if origin == "local" {
-            tx.execute("INSERT OR IGNORE INTO outbox SELECT ?1,id,revision,path,hash,'put',?2,'pending' FROM files WHERE path=?3", params![operation_id,content,path])?;
+            tx.execute("INSERT OR IGNORE INTO outbox SELECT ?1,id,revision,path,hash,'put',?2,'pending' FROM files WHERE path=?3", params![operation_id,b"".as_slice(),path])?;
         }
         let entry = tx.query_row(
             "SELECT id,path,hash,revision,deleted FROM files WHERE path=?1",
@@ -597,6 +613,7 @@ impl Workspace {
             result
         };
         for (op, id, path, expected, content, origin) in pending {
+            let content = self.payload(&op, &content)?;
             match self.apply_journal(&op, &id, &path, &expected, &content, &origin) {
                 Err(e) if e.code == "RECOVERY_CONFLICT" => {}
                 result => result?,
@@ -738,6 +755,7 @@ impl Workspace {
         }
         self.entry(path)?
             .ok_or_else(|| HostError::new("FILE_NOT_FOUND"))?;
+        self.store_payload(id, &content)?;
         let tx = self.db.transaction()?;
         tx.execute(
             "INSERT INTO operations VALUES (?1,?2,'pending',NULL)",
@@ -745,7 +763,15 @@ impl Workspace {
         )?;
         tx.execute(
             "INSERT INTO file_ops VALUES (?1,?2,?3,?4,?5,?6,'pending',?7)",
-            params![id, kind, path, destination, expected, content, origin],
+            params![
+                id,
+                kind,
+                path,
+                destination,
+                expected,
+                b"".as_slice(),
+                origin
+            ],
         )?;
         tx.commit()?;
         Ok(id.to_owned())
@@ -773,6 +799,8 @@ impl Workspace {
                 ))
             },
         )?;
+        let content = self.payload(id, &content)?;
+        self.store_payload(id, &content)?;
         let source = self.resolve(&path)?;
         let previous = self
             .entry(&path)?
@@ -822,7 +850,7 @@ impl Workspace {
                 params![destination, 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])?;
+                tx.execute("INSERT INTO outbox SELECT ?1,id,revision,path,hash,'put',?2,'pending' FROM files WHERE id=?3", params![id,b"".as_slice(),previous.file_id])?;
             }
         } else {
             tx.execute(
@@ -833,6 +861,7 @@ impl Workspace {
                 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("INSERT INTO sync_observed SELECT id,path,hash,deleted FROM files WHERE id=?1 ON CONFLICT(file_id) DO UPDATE SET path=excluded.path,hash=excluded.hash,deleted=excluded.deleted",[&previous.file_id])?;
         tx.execute("DELETE FROM file_ops WHERE id=?1", [id])?;
         let mut result = previous;
         result.revision += 1;

frontend/src-tauri/tests/sync_push.rs

@@ -303,6 +303,140 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate
             );
         }
     }
+
+    // Kill the actual client process after each durable 10 MiB server offset,
+    // before its response reaches the client. The next process must query offset.
+    use sha2::{Digest, Sha256};
+    use std::io::Write;
+    let large_remote = client
+        .json(
+            reqwest::Method::POST,
+            "sync/v1/vaults",
+            Some(json!({"name":"100 MiB resumable"})),
+        )
+        .await
+        .unwrap();
+    let large_remote = large_remote["vault_id"].as_str().unwrap();
+    let large_root = tempfile::tempdir().unwrap();
+    let mut large_ws = Workspace::open(large_root.path()).unwrap();
+    let large_binding = large_ws
+        .sync_bind_empty(&endpoint, large_remote, "rust-fixture")
+        .unwrap();
+    std::fs::create_dir(large_root.path().join("attachments")).unwrap();
+    let mut attachment =
+        std::fs::File::create(large_root.path().join("attachments/large.bin")).unwrap();
+    let block = vec![42u8; 1024 * 1024];
+    let mut hasher = Sha256::new();
+    for _ in 0..100 {
+        attachment.write_all(&block).unwrap();
+        hasher.update(&block);
+    }
+    attachment.sync_all().unwrap();
+    drop(attachment);
+    let expected_hash = format!("{:x}", hasher.finalize());
+    assert_eq!(large_ws.sync_discover(&large_binding.id).unwrap(), 1);
+    large_ws.sync_capture(&large_binding.id).unwrap();
+    let large_id = large_ws
+        .sync_next(&large_binding.id)
+        .unwrap()
+        .unwrap()
+        .file_id;
+    drop(large_ws);
+    std::fs::write(root.path().join("interrupt-upload"), b"controlled-fixture").unwrap();
+    for boundary in 1..=10 {
+        let mut worker = Server(
+            Command::new(std::env::current_exe().unwrap())
+                .args(["--ignored", "--exact", "resumable_upload_worker"])
+                .env("OPENNEXUS_SYNC_WORKER_ROOT", large_root.path())
+                .stdin(Stdio::piped())
+                .stdout(Stdio::null())
+                .stderr(Stdio::null())
+                .spawn()
+                .unwrap(),
+        );
+        worker
+            .0
+            .stdin
+            .take()
+            .unwrap()
+            .write_all(session.access_token.as_bytes())
+            .unwrap();
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(45);
+        let marker = root.path().join("upload-boundary");
+        while !marker.exists() {
+            assert!(
+                worker.0.try_wait().unwrap().is_none(),
+                "upload worker exited before boundary {boundary}"
+            );
+            assert!(
+                std::time::Instant::now() < deadline,
+                "upload boundary timeout"
+            );
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        }
+        assert_eq!(
+            std::fs::read_to_string(&marker)
+                .unwrap()
+                .parse::<usize>()
+                .unwrap(),
+            boundary * 10 * 1024 * 1024
+        );
+        worker.0.kill().unwrap();
+        worker.0.wait().unwrap();
+        std::fs::remove_file(marker).unwrap();
+    }
+    std::fs::remove_file(root.path().join("interrupt-upload")).unwrap();
+    let large_workspace = Arc::new(Mutex::new(Workspace::open(large_root.path()).unwrap()));
+    assert!(client
+        .push_one(&large_workspace, &large_binding)
+        .await
+        .unwrap());
+    assert!(!client
+        .push_one(&large_workspace, &large_binding)
+        .await
+        .unwrap());
+    let download_root = tempfile::tempdir().unwrap();
+    let download = Arc::new(Mutex::new(Workspace::open(download_root.path()).unwrap()));
+    let download_binding = download
+        .lock()
+        .unwrap()
+        .sync_bind_download(&endpoint, large_remote, "rust-fixture")
+        .unwrap();
+    assert_eq!(
+        client_b
+            .pull_page(&download, &download_binding)
+            .await
+            .unwrap(),
+        1
+    );
+    let received = std::fs::read(download_root.path().join("attachments/large.bin")).unwrap();
+    assert_eq!(received.len(), 104857600);
+    assert_eq!(format!("{:x}", Sha256::digest(&received)), expected_hash);
+    assert_eq!(
+        download.lock().unwrap().path_for_id(&large_id).unwrap(),
+        "attachments/large.bin"
+    );
+    assert_eq!(download.lock().unwrap().pending_count().unwrap(), 0);
+    assert_eq!(
+        download
+            .lock()
+            .unwrap()
+            .sync_discover(&download_binding.id)
+            .unwrap(),
+        0
+    );
+    // SQLite stores metadata, never the 100 MiB body.
+    for directory in [large_root.path(), download_root.path()] {
+        let managed = directory.join(".ainote");
+        for item in std::fs::read_dir(managed).unwrap().flatten() {
+            if item.file_type().unwrap().is_file() {
+                assert!(
+                    item.metadata().unwrap().len() < 5 * 1024 * 1024,
+                    "large body leaked into metadata storage"
+                );
+            }
+        }
+    }
     // Host sessions survive encrypted storage reopen and refresh on the actual service.
     use notesagent_host::{credentials::CredentialBroker, sync_auth};
     let credential_root = tempfile::tempdir().unwrap();
@@ -359,3 +493,20 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate
         401
     );
 }
+
+#[tokio::test]
+#[ignore = "helper process driven and killed by the parent fault test"]
+async fn resumable_upload_worker() {
+    let root = std::env::var("OPENNEXUS_SYNC_WORKER_ROOT").expect("controlled fixture root");
+    let ws = Arc::new(Mutex::new(Workspace::open(Path::new(&root)).unwrap()));
+    let binding = ws.lock().unwrap().sync_binding().unwrap().unwrap();
+    assert!(binding.endpoint.starts_with("http://127.0.0.1:"));
+    use std::io::Read;
+    let mut token = Zeroizing::new(String::new());
+    std::io::stdin()
+        .take(4096)
+        .read_to_string(&mut token)
+        .unwrap();
+    let client = SyncClient::new(&binding.endpoint, token, true).unwrap();
+    client.push_one(&ws, &binding).await.unwrap();
+}

server sync/tests/host_fixture.py

@@ -18,6 +18,18 @@ def main():
     database = Database('sqlite:///' + str(root / 'sync.sqlite3'))
     app = create_app(database, DiskObjects(root / 'objects'), root / 'staging')
     database.add_user('rust-fixture', 'controlled-fixture-password')
+    @app.middleware('http')
+    async def interrupt_upload(request, call_next):
+        response = await call_next(request)
+        if (root / 'interrupt-upload').exists() and request.method == 'PUT' and '/uploads/' in request.url.path and response.status_code == 200:
+            offset = int(request.query_params.get('offset', '0')) + int(request.headers.get('content-length', '0'))
+            if offset and offset % (10 * 1024 * 1024) == 0:
+                marker = root / 'upload-boundary'
+                marker.write_text(str(offset), encoding='ascii')
+                for _ in range(600):
+                    if not marker.exists(): break
+                    await asyncio.sleep(.05)
+        return response
     sock = socket.socket()
     sock.bind(('127.0.0.1', 0))
     sock.listen(128)