test(sync): 登记 S-01 生产验收

docs/development/OpenNexus生产验收Runner.md

@@ -24,4 +24,4 @@ python scripts/phase3-production-acceptance.py `
 
 报告目录包含 `summary.json`、`case-manifest.json`、`junit.xml`、`cases/<ID>.json` 和脱敏的 `logs/<ID>.log`。摘要记录 commit、各锁文件 SHA-256、配置摘要与已提供安装产物摘要。日志将仓库、数据根、报告根、用户主目录和配置声明的秘密值替换为占位符，并限制为 10 MiB。报告目录必须为空，避免单例复跑覆盖原始证据。
 
-当前 runner 与失败闭合行为已实现，B-02 Fernet 迁移 driver 已登记；其余 29 个生产验收 ID 尚未登记，运行时会生成 `NOT_IMPLEMENTED` 证据并退出 1。这用于阻止误报，不是这些用例的验收通过。
+当前 runner 与失败闭合行为已实现，B-02 Fernet 迁移和 S-01 Sync 客户端 driver 已登记；其余 28 个生产验收 ID 尚未登记，运行时会生成 `NOT_IMPLEMENTED` 证据并退出 1。这用于阻止误报，不是这些用例的验收通过。

frontend/src-tauri/tests/sync_push.rs

@@ -18,7 +18,7 @@ impl Drop for Server {
 }
 
 #[tokio::test]
-async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicates() {
+async fn s01_actual_service_preserves_offline_chains_and_response_loss_idempotency() {
     let root = tempfile::tempdir().unwrap();
     std::fs::write(root.path().join(".opennexus-test"), b"fixture").unwrap();
     let service = Path::new(env!("CARGO_MANIFEST_DIR"))
@@ -100,13 +100,55 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate
         ws.sync_bind_empty(&endpoint, remote, "rust-fixture")
             .unwrap()
     };
+    assert_eq!(workspace.lock().unwrap().pending_count().unwrap(), 20);
     let first = workspace
         .lock()
         .unwrap()
         .sync_next(&binding.id)
         .unwrap()
         .unwrap();
-    assert!(client.push_one(&workspace, &binding).await.unwrap());
+    // Commit the first revision, then kill the client before the response can
+    // acknowledge the local journal. Reopen must keep all pending operations.
+    std::fs::write(
+        root.path().join("interrupt-revision"),
+        b"controlled-fixture",
+    )
+    .unwrap();
+    drop(workspace);
+    let mut lost_response = Server(
+        Command::new(std::env::current_exe().unwrap())
+            .args(["--ignored", "--exact", "revision_response_loss_worker"])
+            .env("OPENNEXUS_SYNC_WORKER_ROOT", local.path())
+            .stdin(Stdio::piped())
+            .stdout(Stdio::inherit())
+            .stderr(Stdio::inherit())
+            .spawn()
+            .unwrap(),
+    );
+    use std::io::Write;
+    lost_response
+        .0
+        .stdin
+        .take()
+        .unwrap()
+        .write_all(session.access_token.as_bytes())
+        .unwrap();
+    let committed = root.path().join("revision-committed");
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
+    while !committed.exists() {
+        assert!(lost_response.0.try_wait().unwrap().is_none());
+        assert!(
+            std::time::Instant::now() < deadline,
+            "revision commit timeout"
+        );
+        tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+    }
+    lost_response.0.kill().unwrap();
+    lost_response.0.wait().unwrap();
+    std::fs::remove_file(root.path().join("interrupt-revision")).unwrap();
+    std::fs::remove_file(committed).unwrap();
+    let workspace = Arc::new(Mutex::new(Workspace::open(local.path()).unwrap()));
+    assert_eq!(workspace.lock().unwrap().pending_count().unwrap(), 20);
     let payload = workspace
         .lock()
         .unwrap()
@@ -123,6 +165,9 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate
             .unwrap();
         assert_eq!(replay["sequence"], 1);
     }
+    assert_eq!(workspace.lock().unwrap().pending_count().unwrap(), 20);
+    assert!(client.push_one(&workspace, &binding).await.unwrap());
+    assert_eq!(workspace.lock().unwrap().pending_count().unwrap(), 19);
     for _ in 1..20 {
         assert!(client.push_one(&workspace, &binding).await.unwrap());
     }
@@ -182,6 +227,22 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate
         "fixture-19"
     );
     assert_eq!(workspace_b.lock().unwrap().pending_count().unwrap(), 0);
+    assert_eq!(
+        workspace_b
+            .lock()
+            .unwrap()
+            .read("note.md")
+            .unwrap()
+            .entry
+            .hash,
+        workspace
+            .lock()
+            .unwrap()
+            .read("note.md")
+            .unwrap()
+            .entry
+            .hash
+    );
     assert_eq!(
         workspace_b
             .lock()
@@ -207,9 +268,20 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate
     {
         let mut ws = workspace_b.lock().unwrap();
         let current = ws.read("note.md").unwrap();
-        ws.write("note.md", &current.entry.hash, b"offline-b", "local")
-            .unwrap();
+        let mut digest = current.entry.hash;
+        for index in 0..20 {
+            digest = ws
+                .write(
+                    "note.md",
+                    &digest,
+                    format!("offline-b-{index}").as_bytes(),
+                    "local",
+                )
+                .unwrap()
+                .hash;
+        }
     }
+    assert_eq!(workspace_b.lock().unwrap().pending_count().unwrap(), 20);
     client.push_one(&workspace, &binding).await.unwrap();
     assert_eq!(
         client_b.pull_page(&workspace_b, &binding_b).await.unwrap(),
@@ -217,7 +289,7 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate
     );
     assert_eq!(
         workspace_b.lock().unwrap().read("note.md").unwrap().content,
-        "offline-b"
+        "offline-b-19"
     );
     let conflicts = workspace_b
         .lock()
@@ -291,7 +363,7 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate
         client.pull_page(&workspace, &binding).await.unwrap();
         client_b.pull_page(&workspace_b, &binding_b).await.unwrap();
         let expected = if choice == "local" {
-            "offline-b"
+            "offline-b-19"
         } else {
             "next-a"
         };
@@ -781,7 +853,6 @@ 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,
@@ -1069,3 +1140,20 @@ async fn resumable_upload_worker() {
     let client = SyncClient::new(&binding.endpoint, token, true).unwrap();
     client.push_one(&ws, &binding).await.unwrap();
 }
+
+#[tokio::test]
+#[ignore = "helper process killed after the server commits but before response delivery"]
+async fn revision_response_loss_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();
+}

scripts/acceptance_cases/s01_sync_client.py

@@ -0,0 +1,96 @@
+"""S-01 two-client offline chain and idempotent response-loss driver."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+TEST = "s01_actual_service_preserves_offline_chains_and_response_loss_idempotency"
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def result(case_id: str, status: str, reason: str, assertions: list[dict]) -> dict:
+    evidence = f"cargo exact integration test {TEST}"
+    for assertion in assertions:
+        assertion.update({"status": status, "evidence": evidence})
+    files = []
+    for relative in ("frontend/src-tauri/tests/sync_push.rs", "server sync/tests/host_fixture.py"):
+        files.append({"path": relative, "sha256": sha256(ROOT / relative)})
+    return {
+        "schema": 1,
+        "case_id": case_id,
+        "status": status,
+        "reason": reason,
+        "assertions": assertions,
+        "metrics": {"peak_rss_bytes": None, "max_process_count": None, "denied_access_count": None},
+        "files": files,
+        "revisions": [
+            {"scope": "isolated ephemeral Vault", "offline_chain_count": 20, "idempotent_replays": 100},
+        ],
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--config", required=True)
+    parser.add_argument("--output", required=True)
+    args = parser.parse_args()
+    case_id = os.environ.get("OPENNEXUS_ACCEPTANCE_CASE_ID", "")
+    output = Path(args.output)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    assertions = [
+        {"name": "both Rust clients retain twenty offline edits without silent pending loss"},
+        {"name": "server revisions form a contiguous base and sequence chain"},
+        {"name": "client kill after commit preserves pending work for reopen"},
+        {"name": "one hundred identical operation retries return the original revision"},
+        {"name": "both clients converge on content hash and stable file identity"},
+    ]
+    cargo = shutil.which("cargo")
+    if case_id != "S-01" or cargo is None:
+        payload = result(case_id, "FAILED", "S-01 requires the registered case ID and Cargo.", assertions)
+        output.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+        return 1
+    command = [
+        cargo,
+        "test",
+        "--manifest-path",
+        str(ROOT / "frontend" / "src-tauri" / "Cargo.toml"),
+        "--locked",
+        "--features",
+        "desktop",
+        "--test",
+        "sync_push",
+        TEST,
+        "--",
+        "--exact",
+        "--nocapture",
+    ]
+    completed = subprocess.run(command, cwd=ROOT, capture_output=True, text=True, check=False)
+    print(completed.stdout, end="")
+    print(completed.stderr, end="")
+    passed = completed.returncode == 0 and "1 passed; 0 failed" in completed.stdout
+    payload = result(
+        case_id,
+        "PASSED" if passed else "FAILED",
+        "" if passed else "The exact Rust S-01 acceptance oracle failed.",
+        assertions,
+    )
+    output.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    return 0 if passed else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/phase3_acceptance.py

@@ -37,6 +37,11 @@ CASE_DRIVERS: dict[str, dict[str, Any]] = {
         "timeout_seconds": 900,
         "required_metrics": (),
     },
+    "S-01": {
+        "driver": "scripts/acceptance_cases/s01_sync_client.py",
+        "timeout_seconds": 900,
+        "required_metrics": (),
+    },
 }
 ENV_NAME = re.compile(r"[A-Z][A-Z0-9_]{2,127}")
 RUN_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{2,63}")

server sync/tests/host_fixture.py

@@ -29,6 +29,12 @@ def main():
                 for _ in range(600):
                     if not marker.exists(): break
                     await asyncio.sleep(.05)
+        if (root / 'interrupt-revision').exists() and request.method == 'POST' and request.url.path.endswith('/revisions') and response.status_code == 200:
+            marker = root / 'revision-committed'
+            marker.write_text('committed', encoding='ascii')
+            for _ in range(600):
+                if not (root / 'interrupt-revision').exists(): break
+                await asyncio.sleep(.05)
         return response
     sock = socket.socket()
     sock.bind(('127.0.0.1', 0))