Fix/semantic target sync (volcengine#2207)

* fix(storage): sync semantic target before DAG

* Update CONTRIBUTING_CN.md (volcengine#2206)

* Update CONTRIBUTING_CN.md

---------

Co-authored-by: qin-ctx <qinhaojie.exe@bytedance.com>

Author: qin-ptr

openviking/storage/queuefs/semantic_dag.py

@@ -4,7 +4,7 @@
 
 import asyncio
 from dataclasses import dataclass, field
-from typing import Awaitable, Callable, Dict, List, Optional
+from typing import Dict, List, Optional
 
 from openviking.server.identity import RequestContext
 from openviking.storage.queuefs.semantic_sidecar import write_semantic_sidecars
@@ -82,7 +82,6 @@ def __init__(
         recursive: bool = True,
         lock: LockLease = NO_LOCK,
         is_code_repo: bool = False,
-        sync_to_target: bool = False,
         changes: Optional[Dict[str, List[str]]] = None,
         skip_vectorization: bool = False,
         coalesce_key: str = "",
@@ -99,7 +98,6 @@ def __init__(
         self._recursive = recursive
         self._lock = lock
         self._is_code_repo = is_code_repo
-        self._sync_to_target = sync_to_target
         self._changes = changes or {}
         self._skip_vectorization = skip_vectorization
         self._coalesce_key = coalesce_key
@@ -123,50 +121,6 @@ def __init__(
         self._overview_cache: Dict[str, Dict[str, str]] = {}
         self._overview_cache_lock = asyncio.Lock()
 
-    def _create_on_complete_callback(self) -> Callable[[], Awaitable[None]]:
-        """Create on_complete callback for incremental update or full update."""
-
-        async def noop_callback() -> None:
-            return
-
-        if not self._target_uri or not self._root_uri:
-            return noop_callback
-
-        if self._target_uri == self._root_uri:
-            return noop_callback
-
-        # Full resource updates may still need a deferred temp -> target sync
-        # when semantic generation owns the resource lock.
-        if not self._incremental_update and not self._sync_to_target:
-            return noop_callback
-
-        async def sync_diff_callback() -> None:
-            try:
-                diff = await self._processor._sync_topdown_recursive(
-                    self._root_uri,
-                    self._target_uri,
-                    ctx=self._ctx,
-                    file_change_status=self._file_change_status,
-                    lock=self._lock,
-                )
-                logger.info(
-                    f"[SyncDiff] Diff computed: "
-                    f"added_files={len(diff.added_files)}, "
-                    f"deleted_files={len(diff.deleted_files)}, "
-                    f"updated_files={len(diff.updated_files)}, "
-                    f"added_dirs={len(diff.added_dirs)}, "
-                    f"deleted_dirs={len(diff.deleted_dirs)}"
-                )
-            except Exception as e:
-                logger.error(
-                    f"[SyncDiff] Error in sync_diff_callback: "
-                    f"root_uri={self._root_uri}, target_uri={self._target_uri} "
-                    f"error={e}",
-                    exc_info=True,
-                )
-
-        return sync_diff_callback
-
     async def run(self, root_uri: str) -> None:
         """Run DAG execution starting from root_uri."""
         self._root_uri = root_uri
@@ -179,13 +133,9 @@ async def run(self, root_uri: str) -> None:
             await self._lock.close()
             raise
 
-        original_on_complete = self._create_on_complete_callback()
-
         # Release owned semantic locks after downstream vectorization finishes.
         async def wrapped_on_complete() -> None:
             try:
-                if original_on_complete:
-                    await original_on_complete()
                 if self._telemetry_id and self._semantic_msg_id:
                     get_request_wait_tracker().mark_semantic_done(
                         self._telemetry_id, self._semantic_msg_id
@@ -325,13 +275,13 @@ def _get_target_file_path(self, current_uri: str) -> Optional[str]:
                 f"Invalid target_uri or root_uri for incremental update: target_uri={self._target_uri}, root_uri={self._root_uri}"
             )
             return None
-        try:
-            relative_path = current_uri[len(self._root_uri) :]
-            if relative_path.startswith("/"):
-                relative_path = relative_path[1:]
-            return f"{self._target_uri}/{relative_path}" if relative_path else self._target_uri
-        except Exception:
+        if self._target_uri != self._root_uri:
+            logger.warning(
+                "Incremental semantic update expects target_uri == root_uri: "
+                f"target_uri={self._target_uri}, root_uri={self._root_uri}"
+            )
             return None
+        return current_uri
 
     def _is_direct_incremental_update(self) -> bool:
         return (

openviking/storage/queuefs/semantic_processor.py

@@ -62,6 +62,13 @@ class DiffResult:
     added_dirs: List[str] = field(default_factory=list)
     deleted_dirs: List[str] = field(default_factory=list)
 
+    def to_changes(self) -> Dict[str, List[str]]:
+        return {
+            "added": self.added_files + self.added_dirs,
+            "modified": self.updated_files,
+            "deleted": self.deleted_files + self.deleted_dirs,
+        }
+
 
 class RequestQueueStats:
     processed: int = 0
@@ -352,24 +359,36 @@ async def on_dequeue(
                         else:
                             is_incremental = False
                             target_uri = msg.target_uri
+                            run_uri = msg.uri
+                            changes = msg.changes
                             viking_fs = get_viking_fs()
                             if msg.target_uri:
                                 target_exists = await viking_fs.exists(
                                     msg.target_uri, ctx=self._current_ctx
                                 )
-                                # Check if target URI exists and is not the same as the source URI（避免重复处理）
                                 if msg.uri != msg.target_uri:
-                                    if msg.target_preexisting is None:
-                                        target_preexisting = target_exists
-                                    else:
-                                        target_preexisting = bool(msg.target_preexisting)
-                                else:
-                                    target_preexisting = target_exists
-                                if target_preexisting and msg.uri != msg.target_uri:
-                                    is_incremental = True
                                     logger.info(
-                                        f"Target URI exists, using incremental update: {msg.target_uri}"
+                                        "Syncing semantic source into target before processing: "
+                                        f"{msg.uri} -> {msg.target_uri}"
                                     )
+                                    diff = await self._sync_topdown_recursive(
+                                        msg.uri,
+                                        msg.target_uri,
+                                        ctx=self._current_ctx,
+                                        lock=semantic_lock.lock,
+                                    )
+                                    logger.info(
+                                        "[SyncDiff] Diff computed: "
+                                        f"added_files={len(diff.added_files)}, "
+                                        f"deleted_files={len(diff.deleted_files)}, "
+                                        f"updated_files={len(diff.updated_files)}, "
+                                        f"added_dirs={len(diff.added_dirs)}, "
+                                        f"deleted_dirs={len(diff.deleted_dirs)}"
+                                    )
+                                    changes = diff.to_changes()
+                                    is_incremental = True
+                                    target_uri = msg.target_uri
+                                    run_uri = msg.target_uri
                                 elif target_exists and msg.changes and msg.uri == msg.target_uri:
                                     is_incremental = True
                                     logger.info(
@@ -394,18 +413,17 @@ async def on_dequeue(
                                 recursive=msg.recursive,
                                 lock=semantic_lock.lock,
                                 is_code_repo=msg.is_code_repo,
-                                sync_to_target=bool(target_uri and msg.uri != target_uri),
-                                changes=msg.changes,
+                                changes=changes,
                                 skip_vectorization=msg.skip_vectorization,
                                 coalesce_key=msg.coalesce_key,
                                 coalesce_version=msg.coalesce_version,
                             )
                             self._dag_executor = executor
                             lock_transferred = True
-                            await executor.run(msg.uri)
+                            await executor.run(run_uri)
                             self._cache_dag_stats(
                                 msg.telemetry_id,
-                                msg.uri,
+                                run_uri,
                                 executor.get_stats(),
                             )
                             if not executor.stale:
@@ -723,7 +741,7 @@ async def list_children(dir_uri: str) -> Tuple[Dict[str, str], Dict[str, str]]:
                 name = entry.get("name", "")
                 if not name or name in [".", ".."]:
                     continue
-                if name.startswith(".") and name not in [".abstract.md", ".overview.md"]:
+                if name.startswith("."):
                     continue
                 item_uri = VikingURI(dir_uri).join(name).uri
                 if entry.get("isDir", False):
@@ -736,18 +754,6 @@ async def sync_dir(root_dir: str, target_dir: str) -> None:
             root_files, root_dirs = await list_children(root_dir)
             target_files, target_dirs = await list_children(target_dir)
 
-            try:
-                await viking_fs._mv_vector_store_l0_l1(
-                    root_dir,
-                    target_dir,
-                    ctx=ctx,
-                    lock_handle=lock_handle,
-                )
-            except Exception as e:
-                logger.error(
-                    f"[SyncDiff] Failed to move L0/L1 index: {root_dir} -> {target_dir}, error={e}"
-                )
-
             file_names = set(root_files.keys()) | set(target_files.keys())
             for name in sorted(file_names):
                 root_file = root_files.get(name)
@@ -804,7 +810,7 @@ async def sync_dir(root_dir: str, target_dir: str) -> None:
                             )
                             changed = False
                     if changed:
-                        diff.updated_files.append(root_file)
+                        diff.updated_files.append(target_file)
                         try:
                             await viking_fs.rm(target_file, ctx=ctx, lock_handle=lock_handle)
                         except Exception as e:
@@ -825,8 +831,8 @@ async def sync_dir(root_dir: str, target_dir: str) -> None:
                     continue
 
                 if root_file and not target_file:
-                    diff.added_files.append(root_file)
                     target_file_uri = VikingURI(target_dir).join(name).uri
+                    diff.added_files.append(target_file_uri)
                     try:
                         await viking_fs.mv(
                             root_file,
@@ -876,8 +882,8 @@ async def sync_dir(root_dir: str, target_dir: str) -> None:
                     continue
 
                 if root_subdir and not target_subdir:
-                    diff.added_dirs.append(root_subdir)
                     target_subdir_uri = VikingURI(target_dir).join(name).uri
+                    diff.added_dirs.append(target_subdir_uri)
                     try:
                         await viking_fs.mv(
                             root_subdir,
@@ -899,7 +905,7 @@ async def sync_dir(root_dir: str, target_dir: str) -> None:
             parent_uri = VikingURI(target_uri).parent
             if parent_uri:
                 await viking_fs.mkdir(parent_uri.uri, exist_ok=True, ctx=ctx)
-            diff.added_dirs.append(root_uri)
+            diff.added_dirs.append(target_uri)
             await viking_fs.mv(root_uri, target_uri, ctx=ctx, lock_handle=lock_handle)
             return diff
 

openviking/utils/resource_processor.py

@@ -255,6 +255,7 @@ async def process_resource(
             candidate_uri = getattr(context_tree, "_candidate_uri", None) if context_tree else None
             resource_lock: LockLease = NO_LOCK
             target_preexisting = False
+            source_committed = False
 
             if root_uri and temp_uri:
                 from openviking.storage.transaction import get_lock_manager
@@ -276,6 +277,11 @@ async def process_resource(
                         resource_lock = await self._acquire_resource_lock(
                             lock_manager, dst_path, uri=root_uri
                         )
+                    if not target_preexisting:
+                        await viking_fs.persist_temp_tree(temp_uri, root_uri, ctx=ctx)
+                        await viking_fs.delete_temp(parse_result.temp_dir_path, ctx=ctx)
+                        temp_uri = root_uri
+                        source_committed = True
                 except Exception:
                     stage_status = "error"
                     raise
@@ -334,7 +340,7 @@ async def process_resource(
                         pass
 
             if resource_lock.active:
-                if not should_summarize and temp_uri:
+                if not should_summarize and temp_uri and not source_committed:
                     viking_fs = get_viking_fs()
                     await viking_fs.persist_temp_tree(temp_uri, root_uri, ctx=ctx)
                     await viking_fs.delete_temp(parse_result.temp_dir_path, ctx=ctx)

tests/misc/test_resource_processor_mv.py

@@ -90,17 +90,13 @@ def get_handle(self, handle_id):
 class _FakeVikingFS:
     def __init__(self, *, exists_result=False, existing_uris=None):
         self.agfs = SimpleNamespace(
-            cat=MagicMock(return_value=b"content"),
-            ls=MagicMock(return_value=[{"name": "content.md", "isDir": False}]),
-            mkdir=MagicMock(return_value={"status": "ok"}),
-            mv=MagicMock(return_value={"status": "ok"}),
-            stat=MagicMock(return_value={"isDir": True}),
             write=MagicMock(return_value={"status": "ok"}),
         )
-        self.mv = AsyncMock(return_value={})
         self._exists_result = exists_result
         self._existing_uris = set(existing_uris or [])
         self.exists_calls = []
+        self.persist_calls = []
+        self.delete_temp_calls = []
 
     def bind_request_context(self, ctx):
         return _CtxMgr()
@@ -115,21 +111,24 @@ async def mkdir(self, uri, exist_ok=False, ctx=None):
         return None
 
     async def delete_temp(self, temp_dir_path, ctx=None):
+        self.delete_temp_calls.append(temp_dir_path)
         return None
 
     async def persist_temp_tree(self, temp_uri, target_uri, ctx=None):
+        self.persist_calls.append((temp_uri, target_uri))
         self.agfs.write(self._uri_to_path(target_uri, ctx=ctx), b"content")
 
     def _uri_to_path(self, uri, ctx=None):
         return f"/mock/{uri.replace('viking://', '')}"
 
 
 @pytest.mark.asyncio
-async def test_resource_processor_first_add_persist_does_not_await_agfs_mv(monkeypatch):
+async def test_resource_processor_first_add_summarizes_from_committed_uri(monkeypatch):
     from openviking.utils.resource_processor import ResourceProcessor
 
     fake_fs = _FakeVikingFS()
     fake_lock_manager = _FakeLockManager()
+    summarize_calls = []
 
     monkeypatch.setattr(
         "openviking.utils.resource_processor.get_current_telemetry",
@@ -152,22 +151,27 @@ async def test_resource_processor_first_add_persist_does_not_await_agfs_mv(monke
             warnings=[],
         )
     )
-
-    context_tree = SimpleNamespace(
-        root=SimpleNamespace(uri="viking://resources/root", temp_uri="viking://temp/root_tmp")
+    rp.tree_builder.finalize_from_temp = AsyncMock(
+        return_value=SimpleNamespace(
+            root=SimpleNamespace(uri="viking://resources/root", temp_uri="viking://temp/root_tmp")
+        )
+    )
+    rp._summarizer = SimpleNamespace(
+        summarize=AsyncMock(
+            side_effect=lambda *args, **kwargs: (
+                summarize_calls.append(kwargs) or {"status": "success"}
+            )
+        )
     )
-    rp.tree_builder.finalize_from_temp = AsyncMock(return_value=context_tree)
 
-    ctx = object()
-    result = await rp.process_resource(path="x", ctx=ctx, build_index=False, summarize=False)
+    result = await rp.process_resource(path="x", ctx=object(), build_index=True)
 
     assert result["status"] == "success"
     assert result["root_uri"] == "viking://resources/root"
-    fake_fs.agfs.mv.assert_not_called()
-    fake_fs.agfs.write.assert_called_once()
-    fake_fs.mv.assert_not_awaited()
-    assert fake_lock_manager.acquired_exact_paths == []
-    assert fake_lock_manager.acquired_tree_paths == ["/mock/resources/root"]
+    assert fake_fs.persist_calls == [("viking://temp/root_tmp", "viking://resources/root")]
+    assert fake_fs.delete_temp_calls == ["viking://temp/tmpdir"]
+    assert summarize_calls[0]["temp_uris"] == ["viking://resources/root"]
+    assert summarize_calls[0]["target_preexisting"] is False
 
 
 @pytest.mark.asyncio
@@ -218,11 +222,7 @@ async def test_resource_processor_second_add_preserves_temp_uri_for_incremental(
     assert result["root_uri"] == "viking://resources/root"
     assert summarize_calls[0]["temp_uris"] == ["viking://temp/root_tmp"]
     assert summarize_calls[0]["target_preexisting"] is True
-    fake_fs.agfs.mv.assert_not_called()
-    fake_fs.agfs.write.assert_not_called()
-    fake_fs.mv.assert_not_awaited()
-    assert fake_lock_manager.acquired_exact_paths == []
-    assert fake_lock_manager.acquired_tree_paths == ["/mock/resources/root"]
+    assert fake_fs.persist_calls == []
 
 
 @pytest.mark.asyncio
@@ -283,5 +283,6 @@ async def test_resource_processor_auto_candidate_skips_existing_and_busy(monkeyp
     ]
     assert fake_lock_manager.acquired_exact_paths == []
     assert fake_lock_manager.acquired_tree_paths == ["/mock/resources/root_2"]
-    assert summarize_calls[0]["temp_uris"] == ["viking://temp/root_tmp"]
+    assert summarize_calls[0]["temp_uris"] == ["viking://resources/root_2"]
     assert summarize_calls[0]["target_preexisting"] is False
+    assert fake_fs.persist_calls == [("viking://temp/root_tmp", "viking://resources/root_2")]