fix(storage): 优化路径锁与语义刷新并发 (#2029)

* fix(storage): refine path lock semantic refresh concurrency

Use exact path locks for source commits, tree locks only for lifecycle and schema scopes, and coalesce derived semantic writes to avoid stale summary overwrites under concurrent resource and memory updates.

* chore: split benchmark changes into separate PR

* chore: keep semantic refresh design notes out of docs

* style: format lock changes

* fix: preserve resource lifecycle locks

* Revert "fix: preserve resource lifecycle locks"

This reverts commit d2fb274.

* fix(resource): simplify lifecycle locking

* fix(queuefs): consolidate semantic sidecar writes

Author: qin-ctx

docs/en/api/03-filesystem.md

@@ -497,7 +497,7 @@ openviking stat viking://resources/my-project/docs
 }
 ```
 
-The `isLocked` field reports whether the path is currently held by a path lock — either the path itself has a valid `.path.ovlock`, or any ancestor directory holds a SUBTREE lock. Returns `false` when the LockManager is unavailable or the lookup fails, so callers can avoid attempting a write only to observe `ResourceBusyError`.
+The `isLocked` field reports whether the path is currently held by a path lock: the path itself has a valid lock (including an exact-path lock for the target), or any ancestor directory holds a TreeLock. Returns `false` when the LockManager is unavailable or the lookup fails, so callers can avoid attempting a write only to observe `ResourceBusyError`.
 
 The `count` field (directories only) contains the estimated number of items (files and subdirectories) under this directory (from vector index).
 

docs/en/concepts/09-transaction.md

@@ -497,7 +497,7 @@ openviking stat viking://resources/my-project/docs
 }
 ```
 
-`isLocked` 字段反映路径当前是否被 path lock 持有：路径自身存在有效的 `.path.ovlock`，或者任一祖先目录持有 SUBTREE 锁。当 LockManager 不可用或查询失败时返回 `false`，调用方可据此避免先写入再观察到 `ResourceBusyError`。
+`isLocked` 字段反映路径当前是否被路径锁持有：路径自身存在有效锁（包括目标路径对应的 exact-path lock），或者任一祖先目录持有 TreeLock。当 LockManager 不可用或查询失败时返回 `false`，调用方可据此避免先写入再观察到 `ResourceBusyError`。
 
 `count` 字段（仅目录）包含该目录下的项目（文件和子目录）估计数量（来自向量索引）。
 

docs/zh/api/03-filesystem.md

@@ -65,4 +65,3 @@ function formatRepoLine(item) {
   const abstract = item.abstract || item.overview
   return abstract ? `- **${name}** (${item.uri})\n  ${abstract}` : `- **${name}** (${item.uri})`
 }
-

docs/zh/concepts/09-transaction.md

@@ -251,7 +251,7 @@ async def add_resource(
             reason: Context/reason for adding this resource.
             instruction: Specific instruction for processing.
             wait: If True, wait for processing to complete.
-            to: Exact target URI (must not exist yet).
+            to: Exact target URI. Existing targets keep the add_resource incremental-update behavior.
             parent: Target parent URI (must already exist).
             build_index: Whether to build vector index immediately (default: True).
             summarize: Whether to generate summary (default: False).

examples/opencode-plugin/lib/repo-context.mjs

@@ -8,12 +8,13 @@
 
 v5.0 Architecture:
 1. Parser: parse + create directory structure in temp VikingFS
-2. TreeBuilder: move to AGFS + enqueue to SemanticQueue + create Resources
-3. SemanticProcessor: async generate L0/L1 + vectorize
+2. TreeBuilder: build final URI metadata and keep temp references
+3. ResourceProcessor: source commit from temp to final VikingFS path
+4. SemanticProcessor: async generate L0/L1 + vectorize
 
 IMPORTANT (v5.0 Architecture):
 - Parser creates directory structure directly, no LLM calls
-- TreeBuilder moves files and enqueues to SemanticQueue
+- TreeBuilder does not move files; source commit is handled after URI metadata is built
 - SemanticProcessor handles all semantic generation asynchronously
 - Temporary directory approach eliminates memory pressure and enables concurrency
 - Resource objects are lightweight (no content fields)
@@ -40,18 +41,20 @@ class TreeBuilder:
 
     New v5.0 Architecture:
     - Parser creates directory structure in temp VikingFS (no LLM calls)
-    - TreeBuilder moves to AGFS + enqueues to SemanticQueue + creates Resources
+    - TreeBuilder builds final URI metadata while preserving temp URIs
+    - ResourceProcessor commits temp content to the final source path
     - SemanticProcessor handles semantic generation asynchronously
 
     Process flow:
     1. Parser creates directory structure with files in temp VikingFS
-    2. TreeBuilder.finalize_from_temp() moves to AGFS, enqueues to SemanticQueue, creates Resources
-    3. SemanticProcessor generates .abstract.md and .overview.md asynchronously
-    4. SemanticProcessor directly vectorizes and inserts to collection
+    2. TreeBuilder.finalize_from_temp() returns final URI and temp URI metadata
+    3. ResourceProcessor performs source commit with short path locks
+    4. SemanticProcessor generates .abstract.md and .overview.md asynchronously
+    5. SemanticProcessor directly vectorizes and inserts to collection
 
     Key changes from v4.0:
     - Semantic generation moved from Parser to SemanticQueue
-    - TreeBuilder enqueues directories for async processing
+    - ResourceProcessor enqueues directories for async processing
     - Direct vectorization in SemanticProcessor (no EmbeddingQueue)
     """
 
@@ -75,31 +78,6 @@ def _get_base_uri(
         # Agent scope
         return "viking://agent"
 
-    async def _resolve_unique_uri(self, uri: str, max_attempts: int = 100) -> str:
-        """Return a URI that does not collide with an existing resource.
-
-        If *uri* is free, return it unchanged.  Otherwise append ``_1``,
-        ``_2``, ... until a free name is found.
-        """
-        viking_fs = get_viking_fs()
-
-        async def _exists(u: str) -> bool:
-            try:
-                await viking_fs.stat(u)
-                return True
-            except Exception:
-                return False
-
-        if not await _exists(uri):
-            return uri
-
-        for i in range(1, max_attempts + 1):
-            candidate = f"{uri}_{i}"
-            if not await _exists(candidate):
-                return candidate
-
-        raise FileExistsError(f"Cannot resolve unique name for {uri} after {max_attempts} attempts")
-
     # ============================================================================
     # v5.0 Methods (temporary directory + SemanticQueue architecture)
     # ============================================================================
@@ -116,10 +94,10 @@ async def finalize_from_temp(
         create_parent: bool = False,
     ) -> "BuildingTree":
         """
-        Finalize processing by moving from temp to AGFS.
+        Finalize URI metadata for a temp parse result.
 
         Args:
-            to_uri: Exact target URI (must not exist)
+            to_uri: Exact target URI, or resources root to import under
             parent_uri: Target parent URI (must exist unless create_parent is True)
             create_parent: Whether to automatically create parent directory if it doesn't exist
         """
@@ -171,6 +149,8 @@ def is_resources_root(uri: Optional[str]) -> bool:
             candidate_uri = to_uri
         else:
             effective_parent_uri = parent_uri or to_uri if use_to_as_parent else parent_uri
+            if effective_parent_uri:
+                effective_parent_uri = effective_parent_uri.rstrip("/")
             if effective_parent_uri:
                 # Parent URI must exist and be a directory, or create it if requested
                 try:
@@ -202,26 +182,16 @@ def is_resources_root(uri: Optional[str]) -> bool:
                 base_uri = effective_parent_uri
             candidate_uri = VikingURI(base_uri).join(final_doc_name).uri
 
-        if to_uri and not use_to_as_parent:
-            final_uri = candidate_uri
-        elif use_to_as_parent:
-            # Treat an explicit resources root target as "import under this
-            # directory" while preserving the child URI so downstream logic can
-            # incrementally update viking://resources/<child> when it exists.
-            final_uri = candidate_uri
-        else:
-            final_uri = await self._resolve_unique_uri(candidate_uri)
-
         tree = BuildingTree(
             source_path=source_path,
             source_format=source_format,
         )
-        tree._root_uri = final_uri
+        tree._root_uri = candidate_uri
         if not to_uri or use_to_as_parent:
             tree._candidate_uri = candidate_uri
 
         # Create a minimal Context object for the root so that tree.root is not None
-        root_context = Context(uri=final_uri, temp_uri=temp_doc_uri)
+        root_context = Context(uri=candidate_uri, temp_uri=temp_doc_uri)
         tree.add_context(root_context)
 
         return tree

openviking/async_client.py

@@ -11,7 +11,7 @@
     AGFSHTTPError,
     AGFSTimeoutError,
 )
-from openviking.storage.errors import ResourceBusyError
+from openviking.storage.errors import LockAcquisitionError, ResourceBusyError
 from openviking_cli.exceptions import (
     ConflictError,
     FailedPreconditionError,
@@ -404,7 +404,21 @@ def map_exception(
     if isinstance(exc, OpenVikingError):
         return exc
     if isinstance(exc, ResourceBusyError):
-        return ConflictError(str(exc), resource=resource)
+        details: dict[str, Any] = {
+            "resource": exc.uri or resource,
+            "uri": exc.uri or resource,
+            "conflict_type": exc.conflict_type,
+            "retryable": exc.retryable,
+        }
+        return OpenVikingError(str(exc), code="CONFLICT", details=details)
+    if isinstance(exc, LockAcquisitionError):
+        details = {
+            "resource": resource,
+            "uri": resource,
+            "conflict_type": "path_busy",
+            "retryable": True,
+        }
+        return OpenVikingError(str(exc), code="CONFLICT", details=details)
     if isinstance(exc, PermissionError):
         return PermissionDeniedError(str(exc), resource=resource)
     if isinstance(exc, FileNotFoundError):

openviking/parse/tree_builder.py

@@ -207,4 +207,9 @@ async def mv(
         if "not found" in err_msg or "no such file or directory" in err_msg:
             raise NotFoundError(from_uri, "file")
         raise
+    except Exception as exc:
+        mapped = map_exception(exc, resource=from_uri)
+        if mapped is not None:
+            raise mapped from exc
+        raise
     return Response(status="ok", result={"from": from_uri, "to": to_uri})

openviking/server/error_mapping.py

@@ -318,7 +318,7 @@ async def _resolve_shared(
             ctx=internal_ctx,
         )
         handle = get_lock_manager().create_handle()
-        acquired = await get_lock_manager().acquire_subtree(handle, lock_path, timeout=0.0)
+        acquired = await get_lock_manager().acquire_tree(handle, lock_path, timeout=0.0)
         if not acquired:
             raise PermissionDeniedError("Temporary upload is being consumed.")