Skip to content

fix(fs): canonicalize URIs before move #3219

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
qin-ctx merged 3 commits into from
Jul 13, 2026
+60 −295
Merged

fix(fs): canonicalize URIs before move #3219

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8984294
fix(fs): canonicalize URIs before move
qin-ctx Jul 13, 2026
175d3b6
fix(uri): canonicalize grep and ovpack inputs
qin-ctx Jul 13, 2026
13c4ff4
fix(mv): preserve destination on vector update failure
qin-ctx Jul 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions openviking/service/pack_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

from typing import Optional

from openviking.core.namespace import canonicalize_uri
from openviking.core.uri_validation import validate_viking_uri
from openviking.server.identity import RequestContext
from openviking.storage.ovpack.operations import backup_ovpack as local_backup_ovpack
Expand Down Expand Up @@ -60,7 +61,7 @@ async def export_ovpack(
Exported file path
"""
viking_fs = self._ensure_initialized()
uri = validate_viking_uri(uri)
uri = canonicalize_uri(validate_viking_uri(uri), ctx)
return await local_export_ovpack(
viking_fs,
uri,
Expand Down Expand Up @@ -105,7 +106,7 @@ async def import_ovpack(
Imported root resource URI
"""
viking_fs = self._ensure_initialized()
parent = validate_viking_uri(parent, field_name="parent")
parent = canonicalize_uri(validate_viking_uri(parent, field_name="parent"), ctx)
return await local_import_ovpack(
viking_fs,
file_path,
Expand Down
94 changes: 19 additions & 75 deletions openviking/storage/viking_fs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -712,6 +712,10 @@ async def mv(
details={"from_uri": old_uri, "to_uri": new_uri},
)

real_ctx = self._ctx_or_default(ctx)
old_uri = canonicalize_uri(old_uri, real_ctx)
new_uri = canonicalize_uri(new_uri, real_ctx)

lock_context = (
LockContext(
get_lock_manager(),
Expand Down Expand Up @@ -913,6 +917,10 @@ async def grep(
Returns:
Dict with matches, count, match_count, files_scanned
"""
real_ctx = self._ctx_or_default(ctx)
uri = canonicalize_uri(uri, real_ctx)
if exclude_uri:
exclude_uri = canonicalize_uri(exclude_uri, real_ctx)
self._ensure_access(uri, ctx)
# Skip vector_store.count() — the count field is not needed for grep,
# and avoiding it saves one VikingDB API call.
Expand Down Expand Up @@ -1088,7 +1096,7 @@ async def _grep_vikingdb_then_fs(
filter_expr = PathScope("uri", uri, depth=level_limit)
excluded_prefix = None
if exclude_uri:
excluded_prefix = self._normalize_uri(exclude_uri).rstrip("/")
excluded_prefix = exclude_uri.rstrip("/")
self._ensure_access(excluded_prefix, ctx)
filter_expr = And(
[
Expand Down Expand Up @@ -1234,7 +1242,7 @@ async def _grep_with_agfs(

excluded_path = None
if exclude_uri:
normalized_excluded_uri = self._normalize_uri(exclude_uri).rstrip("/")
normalized_excluded_uri = exclude_uri.rstrip("/")
self._ensure_access(normalized_excluded_uri, ctx)
excluded_path = self._uri_to_path(normalized_excluded_uri, ctx=ctx)

Expand Down Expand Up @@ -1331,7 +1339,7 @@ async def _grep_encrypted(
compiled_pattern = re.compile(pattern, flags)
excluded_prefix = None
if exclude_uri:
excluded_prefix = self._normalize_uri(exclude_uri).rstrip("/")
excluded_prefix = exclude_uri.rstrip("/")
self._ensure_access(excluded_prefix, ctx)
file_uris = await self._collect_grep_files(
uri,
Expand Down Expand Up @@ -1366,7 +1374,7 @@ async def search_recursive(current_uri: str, current_depth: int) -> None:
if current_depth > level_limit:
return

normalized_current_uri = self._normalize_uri(current_uri)
normalized_current_uri = current_uri
if excluded_prefix and (
normalized_current_uri == excluded_prefix
or normalized_current_uri.startswith(excluded_prefix + "/")
Expand All @@ -1392,7 +1400,7 @@ async def search_recursive(current_uri: str, current_depth: int) -> None:
else:
file_uris.append(entry_uri)

normalized_uri = self._normalize_uri(uri)
normalized_uri = uri
if excluded_prefix and (
normalized_uri == excluded_prefix or normalized_uri.startswith(excluded_prefix + "/")
):
Expand Down Expand Up @@ -2944,7 +2952,6 @@ async def _update_vector_store_uris(
old_base: str,
new_base: str,
ctx: Optional[RequestContext] = None,
levels: Optional[List[int]] = None,
) -> None:
"""Update URIs in vector store (when moving files).

Expand All @@ -2954,83 +2961,20 @@ async def _update_vector_store_uris(
if not vector_store:
return

old_base_uri = self._path_to_uri(old_base, ctx=ctx)
new_base_uri = self._path_to_uri(new_base, ctx=ctx)
real_ctx = self._ctx_or_default(ctx)

for uri in uris:
try:
new_uri = uri.replace(old_base_uri, new_base_uri, 1)

await vector_store.update_uri_mapping(
ctx=self._ctx_or_default(ctx),
new_uri = new_base + uri[len(old_base) :]
if await vector_store.update_uri_mapping(
ctx=real_ctx,
uri=uri,
new_uri=new_uri,
levels=levels,
)
logger.debug(f"[VikingFS] Updated URI: {uri} -> {new_uri}")
):
logger.debug(f"[VikingFS] Updated URI: {uri} -> {new_uri}")
except Exception as e:
logger.warning(f"[VikingFS] Failed to update {uri} in vector store: {e}")

async def _mv_vector_store_l0_l1(
self,
old_uri: str,
new_uri: str,
ctx: Optional[RequestContext] = None,
lock_handle: Optional["LockHandle"] = None,
) -> None:
from openviking.storage.errors import LockAcquisitionError, ResourceBusyError
from openviking.storage.transaction import LockContext, get_lock_manager

self._ensure_access(old_uri, ctx)
self._ensure_access(new_uri, ctx)

real_ctx = self._ctx_or_default(ctx)
old_dir = VikingURI.normalize(old_uri).rstrip("/")
new_dir = VikingURI.normalize(new_uri).rstrip("/")
if old_dir == new_dir:
return

for uri in (old_dir, new_dir):
if uri.endswith(("/.abstract.md", "/.overview.md")):
raise ValueError(f"mv_vector_store expects directory URIs, got: {uri}")

try:
old_stat = await self.stat(old_dir, ctx=real_ctx)
except Exception as e:
raise FileNotFoundError(f"mv_vector_store old_uri not found: {old_dir}") from e
try:
new_stat = await self.stat(new_dir, ctx=real_ctx)
except Exception as e:
raise FileNotFoundError(f"mv_vector_store new_uri not found: {new_dir}") from e

if not (isinstance(old_stat, dict) and old_stat.get("isDir", False)):
raise ValueError(f"mv_vector_store expects old_uri to be a directory: {old_dir}")
if not (isinstance(new_stat, dict) and new_stat.get("isDir", False)):
raise ValueError(f"mv_vector_store expects new_uri to be a directory: {new_dir}")

old_path = self._uri_to_path(old_dir, ctx=real_ctx)
new_path = self._uri_to_path(new_dir, ctx=real_ctx)

try:
async with LockContext(
get_lock_manager(),
[old_path],
lock_mode="mv",
mv_dst_path=new_path,
src_is_dir=True,
handle=lock_handle,
):
await self._update_vector_store_uris(
uris=[old_dir],
old_base=old_dir,
new_base=new_dir,
ctx=real_ctx,
levels=[0, 1],
)

except LockAcquisitionError:
raise ResourceBusyError(f"Resource is being processed: {old_dir}", uri=old_dir)

def _get_vector_store(self) -> Optional["VikingVectorIndexBackend"]:
"""Get vector store instance."""
return self.vector_store
Expand Down
Loading